[m-users.] An example for thread communication

% An inter-thread communication and synchronisation example.
%
% ...

:- module thr3.
:- interface.
:- import_module io.
:- pred main(io::di, io::uo) is cc_multi.

:- implementation.
:- import_module string, list, int, uint64, maybe.
:- import_module thread, thread.channel, thread.semaphore.
:- import_module random, random.sfc64, random.system_rng, require.

:- pragma require_feature_set([concurrency]).


% A message on the worker channel.
:- type workermsg
    ---> work(uint64)
    ;    end.

% A message on the results channel.
:- type mastermsg
    ---> result(uint64)
    ;    end.


% Number of worker threads to spawn.
:- func workercount = int.
workercount = 3.

% Number of work items to send.
:- func workcount = int.
workcount = 10.



main(!IO) :-
    % Set up the two channels and the semaphore.
    thread.channel.init(WorkChannel, !IO),
    thread.channel.init(ResChannel, !IO),
    thread.semaphore.init(0, Sem, !IO),

    % Spawn the worker threads and the feeder thread.
    spawn_workers(0, workercount, WorkChannel, ResChannel, Sem, !IO),
    thread.spawn(feeder(workcount, WorkChannel, Sem), !IO),

    % Collect the work results from the workers. Exit when all worker threads have terminated themselves.
    collector(workercount, 0, ResChannel, !IO).



% Master: Collect the results from the results channel.

:- pred collector(
    int::in,                                    % Number of still active workers
    int::in,                                    % Number of work results received.
    thread.channel.channel(mastermsg)::in,      % Result channel
    io::di, io::uo
) is cc_multi.

collector(Count, ResNum, ResChannel, !IO) :-
    thread.channel.take(ResChannel, Msg, !IO),
    pflush("Collector: received " ++ string(Msg) ++ "\n", !IO),
    (
        Msg = result(_),
        collector(Count, ResNum + 1, ResChannel, !IO)
    ;
        Msg = end,
        (
            if Count - 1 = 0
            then pflush("Collector: All workers finished. Finishing too.\n" ++
                        "Collector: Received " ++ string(ResNum) ++ " work results.\n", 
                        !IO)
            else collector(Count - 1, ResNum, ResChannel, !IO)
        )
    ).



% Feeder: The feeder thread. Send the work to the feeders. Then send "end" messages to the workers, so they
% terminate themselves.

:- pred feeder(
    int::in,                                    % How many work items to send.
    thread.channel.channel(workermsg)::in,      % Work channel
    thread.semaphore.semaphore::in,             % Semaphore
    io::di, io::uo
) is cc_multi.

feeder(Num, WorkChannel, Sem, !IO) :-

    (
        if Num = 0
        then
            pflush("Feeder: done.\n", !IO),
            term(0, workercount, WorkChannel, !IO)
        else
            pflush("Feeder: Waiting for semaphore.\n", !IO),
            thread.semaphore.wait(Sem, !IO),

            rand64(Rand, !IO),
            Wait = Rand `uint64.mod` 1000000000u64,
            Msg = work(Wait),

            pflush("Feeder: sending " ++ string(Msg) ++ "\n", !IO),

            thread.channel.put(WorkChannel, work(Wait), !IO),

            feeder(Num - 1, WorkChannel, Sem, !IO)
    ).



% Feeder: Send an end message for each worker to the work channel. This is called after all work messages have been
% sent.

:- pred term(int::in, int::in,
    thread.channel.channel(workermsg)::in,      % Arbeitskanal
    io::di, io::uo
) is cc_multi.

term(Num, Count, WorkChannel, !IO) :-
    (
        if
            Num = Count
        then
            true
        else
            pflush("Feeder: sending end.\n", !IO),
            thread.channel.put(WorkChannel, end, !IO),
            term(Num + 1, Count, WorkChannel, !IO)
    ).




% The worker thread. 

:- pred worker(
    int::in,
    thread.channel.channel(workermsg)::in,
    thread.channel.channel(mastermsg)::in,
    thread.semaphore.semaphore::in,
    io::di, io::uo
) is cc_multi.

:- pragma no_determinism_warning(pred(worker/6)).

worker(Num, WorkChannel, ResChannel, Sem, !IO) :-
    pflush(string(Num) ++ " Worker-beginning. Incrementing the semaphore.\n", !IO),

    thread.semaphore.signal(Sem, !IO),

    handle_work(Num, WorkChannel, ResChannel, Sem, !IO),

    pflush("Worker " ++ string(Num) ++ ": sending end\n", !IO),

    thread.channel.put(ResChannel, end, !IO),

    pflush(string(Num) ++ " Worker-end. Decrementing the semaphore.\n", !IO),

    thread.semaphore.wait(Sem, !IO).




% Worker thread: receive and process work items, and send the result to master, until an "end" is received.

:- pred handle_work(
    int::in,
    thread.channel.channel(workermsg)::in,
    thread.channel.channel(mastermsg)::in,
    thread.semaphore.semaphore::in,
    io::di, io::uo
) is cc_multi.

handle_work(Num, WorkChannel, ResChannel, Sem, !IO) :-

    thread.channel.take(WorkChannel, Msg, !IO),
    pflush("Worker " ++ string(Num) ++ ": received " ++ string(Msg) ++ "\n", !IO),

    (
        Msg = work(Wait),
        nanosleep(0u64, Wait, !IO),

        pflush("Worker " ++ string(Num) ++ ": sending result " ++ string(uint64.cast_to_int(Wait)) ++ "\n", !IO),

        thread.channel.put(ResChannel, result(Wait), !IO),

        thread.semaphore.signal(Sem, !IO),

        handle_work(Num, WorkChannel, ResChannel, Sem, !IO)
    ;
        Msg = end
    ).




% Spawn all the worker threads.

:- pred spawn_workers(
    int::in,                                    % Must be 0, when called from the outside
    int::in,                                    % Number of worker threads to be spawned
    thread.channel.channel(workermsg)::in,      % Work channel
    thread.channel.channel(mastermsg)::in,      % Result channel
    thread.semaphore.semaphore::in,             % Semaphore
    io::di, io::uo
) is cc_multi.

spawn_workers(Num, Count, WorkChannel, ResChannel, Sem, !IO) :-
    (
        if
            Num = Count
        then
            true
        else
            io.flush_output(stdout_stream, !IO),

            thread.spawn(worker(Num, WorkChannel, ResChannel, Sem), !IO),
            spawn_workers(Num + 1, Count, WorkChannel, ResChannel, Sem, !IO)
    ).



% Output some text atomically. We convert the text to a string first, so it can be written out in one piece.

:- pred pflush(string::in, io::di, io::uo).

pflush(Str, !IO) :-
    io.write_string(Str, !IO),
    io.flush_output(stdout_stream, !IO).



%----------------------------------------------------------------------------------------------------
% nanosleep

:- pragma foreign_code("C",
"
#include <time.h>
").


:- pred nanosleep(uint64::in, uint64::in, io::di, io::uo) is det.

:- pragma foreign_proc("C",
    nanosleep(Secs::in, NSecs::in, IOin::di, IOout::uo),
    [ will_not_call_mercury, promise_pure, thread_safe, will_not_throw_exception ],
"
    struct timespec req;
    req.tv_sec = (time_t) Secs;
    req.tv_nsec = (time_t) NSecs;

    nanosleep(&req, 0);

    IOout = IOin;
").




%----------------------------------------------------------------------------------------------------
% Random number generation

:- pred rand64(uint64::out, io::di, io::uo) is det.

rand64(Random, !IO) :-
    get_maybedummy(MaybeDummy, !IO),
    (
        % Already seeded the random generator
        MaybeDummy = yes(Dummy)
    ;
        % Seed the random generator
        MaybeDummy = no,
        open_system_rng(MaybeHandle, !IO),
        (
            MaybeHandle = maybe.ok(Handle),

            random.system_rng.generate_uint64(Handle, Seed1, !IO),
            random.system_rng.generate_uint64(Handle, Seed2, !IO),
            random.system_rng.generate_uint64(Handle, Seed3, !IO),

            random.sfc64.seed(Seed1, Seed2, Seed3, P, RandState),
            make_io_urandom(P, RandState, Dummy, !IO),

            set_maybedummy(yes(Dummy), !IO)
        ;
            MaybeHandle = maybe.error(Msg),
            unexpected($pred, "Error opening system random number generator:\n" ++ Msg)
        )
    ),

    random.generate_uint64(Dummy, Random, !IO).



% Mutable variable for storing the dummy value which is needed for the random.generate_uint64 predicate. So we
% don't need to carry this value around. See section "10.6 Module-local mutable variables" in the Language
% Reference Manual.

:- mutable(
    maybedummy,
    maybe(random.io_urandom(random.sfc64.params, random.sfc64.ustate)),
    no,
    ground,
    [attach_to_io_state, untrailed, thread_local]
).