working_actors

-module(working_actors).
-compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch, inline]).
-define(FILEPATH, "src/working_actors.gleam").
-export([spawn_workers/3]).
-export_type([worker_message/2, response_message/1]).

-if(?OTP_RELEASE >= 27).
-define(MODULEDOC(Str), -moduledoc(Str)).
-define(DOC(Str), -doc(Str)).
-else.
-define(MODULEDOC(Str), -compile([])).
-define(DOC(Str), -compile([])).
-endif.

?MODULEDOC(
    " working_actors is a simple library for spawning a group of actors\n"
    " as workers to run a specific function with one of a list of input\n"
    " arguments and collect the function output into a new list.\n"
    "\n"
    " Note that the output list is not necessarily in the same order as\n"
    " the input list, so if it is necessary you should return a record with\n"
    " both the input argument and the output from the worker function.\n"
    "\n"
).

-type worker_message(FFP, FFQ) :: {do_work,
        FFP,
        gleam@erlang@process:subject(response_message(FFQ))} |
    shutdown.

-type response_message(FFR) :: {submit_work, FFR, gleam@erlang@process:pid_()}.

-file("src/working_actors.gleam", 25).
-spec worker_handle_message(fun((FFS) -> FFT), worker_message(FFS, FFT)) -> gleam@otp@actor:next(fun((FFS) -> FFT), worker_message(FFS, FFT)).
worker_handle_message(State, Message) ->
    case Message of
        {do_work, Input, Reply_to} ->
            Reply = State(Input),
            gleam@otp@actor:send(Reply_to, {submit_work, Reply, erlang:self()}),
            gleam@otp@actor:continue(State);

        shutdown ->
            gleam@otp@actor:stop()
    end.

-file("src/working_actors.gleam", 75).
-spec use_workers(
    list(FGE),
    gleam@dict:dict(gleam@erlang@process:pid_(), gleam@erlang@process:subject(worker_message(FGE, FGG))),
    list(gleam@erlang@process:subject(worker_message(FGE, FGG))),
    list(FGG),
    gleam@erlang@process:subject(response_message(FGG))
) -> list(FGG).
use_workers(Tasks, Workers, Idle_workers, Function_responses, Reply_to) ->
    case {Tasks, Idle_workers, maps:size(Workers)} of
        {[First_task | Rest_tasks], [First_worker | Rest_workers], _} ->
            gleam@otp@actor:send(First_worker, {do_work, First_task, Reply_to}),
            use_workers(
                Rest_tasks,
                Workers,
                Rest_workers,
                Function_responses,
                Reply_to
            );

        {[], [First_worker@1 | Rest_workers@1], _} ->
            gleam@otp@actor:send(First_worker@1, shutdown),
            New_workers = gleam@dict:filter(
                Workers,
                fun(_, V) -> V /= First_worker@1 end
            ),
            use_workers(
                [],
                New_workers,
                Rest_workers@1,
                Function_responses,
                Reply_to
            );

        {[First_task@1 | Rest_tasks@1], [], _} ->
            case gleam_erlang_ffi:'receive'(Reply_to) of
                {submit_work, Fn_response, Pid} ->
                    Worker_subject@1 = case gleam_stdlib:map_get(Workers, Pid) of
                        {ok, Worker_subject} -> Worker_subject;
                        _assert_fail ->
                            erlang:error(#{gleam_error => let_assert,
                                        message => <<"Pattern match failed, no pattern matched the value."/utf8>>,
                                        file => <<?FILEPATH/utf8>>,
                                        module => <<"working_actors"/utf8>>,
                                        function => <<"use_workers"/utf8>>,
                                        line => 109,
                                        value => _assert_fail,
                                        start => 3336,
                                        'end' => 3390,
                                        pattern_start => 3347,
                                        pattern_end => 3365})
                    end,
                    gleam@otp@actor:send(
                        Worker_subject@1,
                        {do_work, First_task@1, Reply_to}
                    ),
                    use_workers(
                        Rest_tasks@1,
                        Workers,
                        [],
                        [Fn_response | Function_responses],
                        Reply_to
                    )
            end;

        {[], [], 0} ->
            Function_responses;

        {[], [], _} ->
            case gleam_erlang_ffi:'receive'(Reply_to) of
                {submit_work, Fn_response@1, Pid@1} ->
                    Worker_subject@3 = case gleam_stdlib:map_get(Workers, Pid@1) of
                        {ok, Worker_subject@2} -> Worker_subject@2;
                        _assert_fail@1 ->
                            erlang:error(#{gleam_error => let_assert,
                                        message => <<"Pattern match failed, no pattern matched the value."/utf8>>,
                                        file => <<?FILEPATH/utf8>>,
                                        module => <<"working_actors"/utf8>>,
                                        function => <<"use_workers"/utf8>>,
                                        line => 125,
                                        value => _assert_fail@1,
                                        start => 3803,
                                        'end' => 3857,
                                        pattern_start => 3814,
                                        pattern_end => 3832})
                    end,
                    gleam@otp@actor:send(Worker_subject@3, shutdown),
                    New_workers@1 = gleam@dict:delete(Workers, Pid@1),
                    use_workers(
                        [],
                        New_workers@1,
                        [],
                        [Fn_response@1 | Function_responses],
                        Reply_to
                    )
            end
    end.

-file("src/working_actors.gleam", 54).
?DOC(
    " Spawn `n_workers` worker processes, each ready to run the work\n"
    " function `work_function`. A list of arguments to the `work_function`\n"
    " is provided as the `tasks` list. The functions will be called in\n"
    " parallel until the `work_function` has been run with every argument\n"
    " in the `tasks` list. A new list of the function return values will\n"
    " be returned when all the tasks have run. This list will be in an\n"
    " arbitrary order.\n"
).
-spec spawn_workers(integer(), list(FGA), fun((FGA) -> FGC)) -> list(FGC).
spawn_workers(N_workers, Tasks, Work_function) ->
    Workers = begin
        _pipe = gleam@list:range(1, N_workers),
        _pipe@4 = gleam@list:map(
            _pipe,
            fun(_) -> _pipe@1 = gleam@otp@actor:new(Work_function),
                _pipe@2 = gleam@otp@actor:on_message(
                    _pipe@1,
                    fun worker_handle_message/2
                ),
                _pipe@3 = gleam@otp@actor:start(_pipe@2),
                gleam@result:map(
                    _pipe@3,
                    fun(Started) ->
                        {erlang:element(2, Started), erlang:element(3, Started)}
                    end
                ) end
        ),
        _pipe@5 = gleam@result:values(_pipe@4),
        maps:from_list(_pipe@5)
    end,
    Idle_workers = begin
        _pipe@6 = Workers,
        maps:values(_pipe@6)
    end,
    use_workers(
        Tasks,
        Workers,
        Idle_workers,
        [],
        gleam@erlang@process:new_subject()
    ).