-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 => <>, 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 => <>, 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() ).