-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(FRY, FRZ) :: {do_work, FRY, gleam@erlang@process:subject(response_message(FRZ))} | shutdown. -type response_message(FSA) :: {submit_work, FSA, gleam@erlang@process:pid_()}. -file("src/working_actors.gleam", 25). -spec worker_handle_message(fun((FSB) -> FSC), worker_message(FSB, FSC)) -> gleam@otp@actor:next(fun((FSB) -> FSC), worker_message(FSB, FSC)). 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(FSN), gleam@dict:dict(gleam@erlang@process:pid_(), gleam@erlang@process:subject(worker_message(FSN, FSP))), list(gleam@erlang@process:subject(worker_message(FSN, FSP))), list(FSP), gleam@erlang@process:subject(response_message(FSP)) ) -> list(FSP). use_workers(Tasks, Workers, Idle_workers, Function_responses, Reply_to) -> case {Tasks, Idle_workers, maps:size(Workers)} of {_, _, 0} -> Function_responses; {[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 => <>, module => <<"working_actors"/utf8>>, function => <<"use_workers"/utf8>>, line => 110, value => _assert_fail, start => 3370, 'end' => 3424, pattern_start => 3381, pattern_end => 3399}) 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; {[], [], _} -> 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 => <>, module => <<"working_actors"/utf8>>, function => <<"use_workers"/utf8>>, line => 125, value => _assert_fail@1, start => 3801, 'end' => 3855, pattern_start => 3812, pattern_end => 3830}) 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(FSJ), fun((FSJ) -> FSL)) -> list(FSL). 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() ).