:- pred master_loop_normal(
    par_bits(MasterCustomMsg, FeederCustomMsg, WorkerCustomMsg)::in,

    % Master process plugin
    master_process(MasterCustomMsg, FeederCustomMsg, WorkerCustomMsg, MasterData)::in(master_process),

    % Master finish plugin
    master_finish(MasterCustomMsg, FeederCustomMsg, WorkerCustomMsg, MasterData, Result)
    ::in(master_finish),

    % Number of running worker threads.
    int::in, int::out,

    % If the feeder has exited.
    bool::in, bool::out,

    % custom data that's being maintained, by the three master plugins.
    MasterData::in, MasterData::out,

    % End result.
    Result::out,

    io::di, io::uo
) is cc_multi.

:- pragma require_tail_recursion(master_loop_normal/12, [error/*, in_all_grades*/]).

master_loop_normal(
    ParBits, Process, Finish,
    !NumWorkers, !FeederExited, !MasterData,
    Result,
    !IO
) :-
    log_format_mfwpar("master_loop_normal: Workers remaining = %i, feeder exited = %s",
        [i(!.NumWorkers), s(string(!.FeederExited))], !IO),

    (
        if
            !.NumWorkers = 0,
            !.FeederExited = yes
        then
            % No more workers and no more feeder. Work off the rest of the master messages. Then
            % call the finish plugin and exit the master loop.
            master_loop_rest(
                ParBits, Process, Finish,
                !NumWorkers, !FeederExited, !MasterData,
                Result, !IO)
        else
            % Get next master message.
            priority_channel.take(ParBits ^ master_channel, Msg, !IO),
            log_format_mfwpar("master_loop_normal: Received message %s", [s(string(Msg))], !IO),

            (
                % A worker has terminated. Record that.
                Msg = worker_terminated,
                !:NumWorkers = !.NumWorkers - 1
            ;
                % The feeder has terminated. Record that.
                Msg = feeder_terminated,
                !:FeederExited = yes
            ;
                % The feeder or a worker have initiated abortion, and sent "abort" messages to all
                % threads.
                Msg = abort(Exc),

                % Wait for all the other threads to exit, call the finish plugin and then rethrow
                % the exception. This discards all messages, except for "feeder_termianted" and
                % "worker_terminated".
                master_loop_abort(
                    ParBits, Finish,
                    !NumWorkers, !FeederExited, !MasterData,
                    Exc, !IO
                )
            ;
                % Custom message - call the master process plugin with that message.
                Msg = custom(Custom),

                log_format("Calling master process plugin with %s.", [s(string(Msg))], !IO),
                (
                    try [io(!IO)]
                        Process(ParBits, !MasterData, Custom, !IO)

                    then
                        true

                    catch_any Exc ->
                        % An exception from the process plugin. Go into abortion mode. First, send
                        % the exception as "abort" messages to all threads.
                        send_abort(ParBits, Exc, !IO),

                        % Wait for all the other threads to exit, then call the finish plugin and
                        % rethrow the exception. This discards all messages, except for
                        % "feeder_termianted" and "worker_terminated".
                        master_loop_abort(
                            ParBits, Finish,
                            !NumWorkers, !FeederExited, !MasterData,
                            Exc, !IO
                        )
                )
            ),

            master_loop_normal(
                ParBits, Process, Finish,
                !NumWorkers, !FeederExited, !MasterData,
                Result, !IO)
    ).