Current section
Files
Jump to
Current section
Files
src/crew.erl
-module(crew).
-compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch, inline]).
-define(FILEPATH, "src/crew.gleam").
-export([fixed_size/2, subscribe/1, unsubscribe/2, start/1, supervised/1, new/1, select_map/3, enqueue_many/2, work_many/3, work/3, parallel_map/4, work_ordered/3, enqueue/2]).
-export_type([builder/0, channel/1, pool_msg/0, state/0, worker/0, queue_item/0, active_worker/0, caller/0, request/0, any_/0, work/0, work_result/0]).
-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(
" A worker pool for Gleam that distributes work across a limited number of\n"
" worker processes. Workers are pooled to avoid the overhead of\n"
" spawning and killing processes for each task.\n"
"\n"
" The pool manages a queue of work items and distributes them to idle workers.\n"
" When no workers are available, work is queued until a worker becomes free.\n"
" The pool handles worker crashes gracefully and automatically manages the\n"
" worker lifecycle.\n"
"\n"
" ## Example\n"
"\n"
" ```gleam\n"
" import crew\n"
" import gleam/otp/static_supervisor as supervisor\n"
"\n"
" pub fn main() {\n"
" // Create a pool name\n"
" let pool_name = process.new_name(\"my_crew\")\n"
"\n"
" // Start an unsupervised pool\n"
" let assert Ok(_) =\n"
" crew.new(pool_name)\n"
" |> crew.fixed_size(4)\n"
" |> crew.start\n"
"\n"
" // Execute work on the pool\n"
" let result = crew.work(pool_name, 5000, fn() {\n"
" // Some expensive computation\n"
" expensive_computation()\n"
" })\n"
" }\n"
" ```\n"
).
-opaque builder() :: {builder, gleam@erlang@process:name(pool_msg()), integer()}.
-opaque channel(FKL) :: {channel,
gleam@erlang@process:subject(pool_msg()),
gleam@erlang@process:subject(work_result()),
gleam@erlang@process:monitor()} |
{gleam_phantom, FKL}.
-opaque pool_msg() :: {worker_started,
gleam@erlang@process:pid_(),
gleam@erlang@process:subject(work())} |
{worker_idle, gleam@erlang@process:pid_()} |
{monitored_process_exited, gleam@erlang@process:down()} |
{subscribe, gleam@erlang@process:subject(work_result())} |
{unsubscribe, gleam@erlang@process:subject(work_result())} |
{enqueue,
gleam@erlang@process:subject(work_result()),
list(fun(() -> any_()))}.
-type state() :: {state,
list(worker()),
gleam@dict:dict(gleam@erlang@process:pid_(), active_worker()),
gleam@dict:dict(gleam@erlang@process:pid_(), caller()),
gleam@dict:dict(gleam@erlang@process:subject(work_result()), request()),
gleam@deque:deque(queue_item())}.
-type worker() :: {worker,
gleam@erlang@process:pid_(),
gleam@erlang@process:subject(work()),
gleam@erlang@process:monitor()}.
-type queue_item() :: {queue_item,
fun(() -> any_()),
gleam@erlang@process:pid_(),
gleam@erlang@process:subject(work_result())}.
-type active_worker() :: {active_worker,
worker(),
gleam@erlang@process:pid_(),
gleam@erlang@process:subject(work_result())}.
-type caller() :: {caller,
gleam@erlang@process:pid_(),
gleam@erlang@process:monitor(),
gleam@set:set(gleam@erlang@process:subject(work_result()))}.
-type request() :: {request,
gleam@erlang@process:pid_(),
gleam@set:set(gleam@erlang@process:pid_()),
gleam@erlang@process:subject(work_result())}.
-type any_() :: any().
-type work() :: {work,
fun(() -> any_()),
gleam@erlang@process:subject(work_result())}.
-type work_result() :: {done, any_()} |
{worker_exited, gleam@erlang@process:exit_reason()}.
-file("src/crew.gleam", 92).
?DOC(
" Set the number of worker processes in the pool to a fixed number.\n"
"\n"
" When set to less than 1 starting the pool will fail.\n"
"\n"
" ## Example\n"
"\n"
" ```gleam\n"
" crew.new(pool_name)\n"
" |> crew.fixed_size(8) // Use 8 workers regardless of CPU count\n"
" ```\n"
).
-spec fixed_size(builder(), integer()) -> builder().
fixed_size(Builder, Size) ->
{builder, erlang:element(2, Builder), Size}.
-file("src/crew.gleam", 410).
?DOC(
" Create a typed subscription channel to the worker pool for work submission.\n"
"\n"
" This function provides a lower-level interface for submitting work to the\n"
" pool. In many cases calling the `work*` function from a separate process\n"
" will be easier.\n"
"\n"
" Subscribing to the pool creates a reference to the current process in the pool\n"
" that must be cleaned up using `unsubscribe`. The returned channel can be used\n"
" with `select_map` and `enqueue` for custom message handling.\n"
"\n"
" ## Panics\n"
" - If the pool is not running\n"
"\n"
" ## Example\n"
"\n"
" ```gleam\n"
" let channel = crew.subscribe(pool_name)\n"
" crew.enqueue(channel, fn() { some_work() })\n"
" // Handle results with selector...\n"
" crew.unsubscribe(channel, selector)\n"
" ```\n"
).
-spec subscribe(gleam@erlang@process:name(pool_msg())) -> channel(any()).
subscribe(Pool) ->
Pool_subject = gleam@erlang@process:named_subject(Pool),
Pool_pid@1 = case gleam@erlang@process:subject_owner(Pool_subject) of
{ok, Pool_pid} -> Pool_pid;
_assert_fail ->
erlang:error(#{gleam_error => let_assert,
message => <<"Pool is not running"/utf8>>,
file => <<?FILEPATH/utf8>>,
module => <<"crew"/utf8>>,
function => <<"subscribe"/utf8>>,
line => 412,
value => _assert_fail,
start => 12419,
'end' => 12480,
pattern_start => 12430,
pattern_end => 12442})
end,
Monitor = gleam@erlang@process:monitor(Pool_pid@1),
Receive = gleam@erlang@process:new_subject(),
gleam@otp@actor:send(Pool_subject, {subscribe, Receive}),
{channel, Pool_subject, Receive, Monitor}.
-file("src/crew.gleam", 463).
?DOC(
" Remove a subscription channel from the pool and clean up resources.\n"
"\n"
" This function should be called when you're done using a channel obtained\n"
" from `subscribe`. It stops any in-progress work for this channel and\n"
" removes the channel's handlers from the selector.\n"
).
-spec unsubscribe(channel(any()), gleam@erlang@process:selector(FMG)) -> gleam@erlang@process:selector(FMG).
unsubscribe(Channel, Selector) ->
gleam@otp@actor:send(
erlang:element(2, Channel),
{unsubscribe, erlang:element(3, Channel)}
),
gleam@erlang@process:demonitor_process(erlang:element(4, Channel)),
_pipe = Selector,
_pipe@1 = gleam@erlang@process:deselect_specific_monitor(
_pipe,
erlang:element(4, Channel)
),
gleam@erlang@process:deselect(_pipe@1, erlang:element(3, Channel)).
-file("src/crew.gleam", 536).
-spec init_pool(gleam@erlang@process:subject(pool_msg())) -> {ok,
gleam@otp@actor:initialised(state(), pool_msg(), gleam@erlang@process:subject(pool_msg()))} |
{error, any()}.
init_pool(Self) ->
Selector = begin
_pipe = gleam_erlang_ffi:new_selector(),
_pipe@1 = gleam@erlang@process:select(_pipe, Self),
gleam@erlang@process:select_monitors(
_pipe@1,
fun(Field@0) -> {monitored_process_exited, Field@0} end
)
end,
State = {state, [], maps:new(), maps:new(), maps:new(), gleam@deque:new()},
_pipe@2 = gleam@otp@actor:initialised(State),
_pipe@3 = gleam@otp@actor:selecting(_pipe@2, Selector),
_pipe@4 = gleam@otp@actor:returning(_pipe@3, Self),
{ok, _pipe@4}.
-file("src/crew.gleam", 683).
-spec enqueue_loop(state(), request(), list(fun(() -> any_()))) -> state().
enqueue_loop(State, Request, Work) ->
{request, Caller, Workers, Receive} = Request,
case {Work, erlang:element(2, State)} of
{[Work@1 | Rest], [Worker | Idle_workers]} ->
gleam@erlang@process:send(
erlang:element(3, Worker),
{work, Work@1, Receive}
),
Active_worker = {active_worker, Worker, Caller, Receive},
Active_workers = gleam@dict:insert(
erlang:element(3, State),
erlang:element(2, Worker),
Active_worker
),
Request@1 = {request,
erlang:element(2, Request),
gleam@set:insert(Workers, erlang:element(2, Worker)),
erlang:element(4, Request)},
_pipe = {state,
Idle_workers,
Active_workers,
erlang:element(4, State),
erlang:element(5, State),
erlang:element(6, State)},
enqueue_loop(_pipe, Request@1, Rest);
{_, _} ->
Queue@1 = gleam@list:fold(
Work,
erlang:element(6, State),
fun(Queue, Work@2) ->
gleam@deque:push_back(
Queue,
{queue_item, Work@2, Caller, Receive}
)
end
),
Requests = gleam@dict:insert(
erlang:element(5, State),
Receive,
Request
),
{state,
erlang:element(2, State),
erlang:element(3, State),
erlang:element(4, State),
Requests,
Queue@1}
end.
-file("src/crew.gleam", 791).
-spec worker_loop(
gleam@erlang@process:subject(pool_msg()),
gleam@erlang@process:pid_(),
gleam@erlang@process:subject(work())
) -> any().
worker_loop(Pool, Self, Subject) ->
{work, Work, Receive} = gleam_erlang_ffi:'receive'(Subject),
Result = Work(),
gleam@erlang@process:send(Pool, {worker_idle, Self}),
gleam@erlang@process:send(Receive, {done, Result}),
worker_loop(Pool, Self, Subject).
-file("src/crew.gleam", 782).
-spec worker(gleam@erlang@process:name(pool_msg())) -> nil.
worker(Pool) ->
Self = erlang:self(),
Subject = gleam@erlang@process:new_subject(),
Pool_subject = gleam@erlang@process:named_subject(Pool),
gleam@erlang@process:send(Pool_subject, {worker_started, Self, Subject}),
worker_loop(Pool_subject, Self, Subject).
-file("src/crew.gleam", 803).
-spec repeat(integer(), FMX, fun((FMX) -> FMX)) -> FMX.
repeat(Times, State, F) ->
case Times > 0 of
true ->
repeat(Times - 1, F(State), F);
false ->
State
end.
-file("src/crew.gleam", 814).
-spec 'try'({ok, FMY} | {error, FMZ}, fun((FMZ) -> FNC), fun((FMY) -> FNC)) -> FNC.
'try'(Result, Or, Then) ->
case Result of
{ok, X} ->
Then(X);
{error, X@1} ->
Or(X@1)
end.
-file("src/crew.gleam", 718).
-spec try_dequeue_work(state(), worker()) -> state().
try_dequeue_work(State, Worker) ->
'try'(
gleam@deque:pop_front(erlang:element(6, State)),
fun(_) ->
{state,
[Worker | erlang:element(2, State)],
erlang:element(3, State),
erlang:element(4, State),
erlang:element(5, State),
erlang:element(6, State)}
end,
fun(_use0) ->
{{queue_item, Work, Caller, Receive}, Queue} = _use0,
'try'(
gleam_stdlib:map_get(erlang:element(5, State), Receive),
fun(_) ->
try_dequeue_work(
{state,
erlang:element(2, State),
erlang:element(3, State),
erlang:element(4, State),
erlang:element(5, State),
Queue},
Worker
)
end,
fun(Request) ->
gleam@erlang@process:send(
erlang:element(3, Worker),
{work, Work, Receive}
),
Active_worker = {active_worker, Worker, Caller, Receive},
Active_workers = gleam@dict:insert(
erlang:element(3, State),
erlang:element(2, Worker),
Active_worker
),
Request@1 = {request,
erlang:element(2, Request),
gleam@set:insert(
erlang:element(3, Request),
erlang:element(2, Worker)
),
erlang:element(4, Request)},
Requests = gleam@dict:insert(
erlang:element(5, State),
Receive,
Request@1
),
{state,
erlang:element(2, State),
Active_workers,
erlang:element(4, State),
Requests,
Queue}
end
)
end
).
-file("src/crew.gleam", 821).
-spec try_({ok, FNE} | {error, any()}, FNI, fun((FNE) -> FNI)) -> FNI.
try_(Result, Or, Then) ->
case Result of
{ok, X} ->
Then(X);
{error, _} ->
Or
end.
-file("src/crew.gleam", 743).
-spec abort_caller(
state(),
gleam@erlang@process:pid_(),
gleam@erlang@process:exit_reason()
) -> state().
abort_caller(State, Caller, Reason) ->
try_(
gleam_stdlib:map_get(erlang:element(4, State), Caller),
State,
fun(Caller@1) ->
gleam@erlang@process:demonitor_process(erlang:element(3, Caller@1)),
Requests@1 = gleam@set:fold(
erlang:element(4, Caller@1),
erlang:element(5, State),
fun(Requests, Receive) ->
try_(
gleam_stdlib:map_get(erlang:element(5, State), Receive),
Requests,
fun(Request) ->
gleam@set:each(
erlang:element(3, Request),
fun gleam@erlang@process:kill/1
),
gleam@erlang@process:send(
Receive,
{worker_exited, Reason}
),
gleam@dict:delete(Requests, Receive)
end
)
end
),
Callers = gleam@dict:delete(
erlang:element(4, State),
erlang:element(2, Caller@1)
),
{state,
erlang:element(2, State),
erlang:element(3, State),
Callers,
Requests@1,
erlang:element(6, State)}
end
).
-file("src/crew.gleam", 557).
-spec pool(state(), pool_msg()) -> gleam@otp@actor:next(state(), pool_msg()).
pool(State, Msg) ->
Next_state = case Msg of
{worker_started, Pid, Send} ->
Monitor = gleam@erlang@process:monitor(Pid),
Worker = {worker, Pid, Send, Monitor},
try_dequeue_work(State, Worker);
{monitored_process_exited, {process_down, _, Pid@1, Reason}} ->
case {gleam_stdlib:map_get(erlang:element(4, State), Pid@1),
gleam_stdlib:map_get(erlang:element(3, State), Pid@1)} of
{{ok, _}, _} ->
abort_caller(State, Pid@1, Reason);
{{error, _}, {ok, {active_worker, _, Caller, _}}} ->
Active_workers = gleam@dict:delete(
erlang:element(3, State),
Pid@1
),
_pipe = {state,
erlang:element(2, State),
Active_workers,
erlang:element(4, State),
erlang:element(5, State),
erlang:element(6, State)},
abort_caller(_pipe, Caller, Reason);
{{error, _}, {error, _}} ->
Idle_workers = gleam@list:filter(
erlang:element(2, State),
fun(Worker@1) ->
erlang:element(2, Worker@1) /= Pid@1
end
),
{state,
Idle_workers,
erlang:element(3, State),
erlang:element(4, State),
erlang:element(5, State),
erlang:element(6, State)}
end;
{monitored_process_exited, _} ->
State;
{worker_idle, Pid@2} ->
try_(
gleam_stdlib:map_get(erlang:element(3, State), Pid@2),
State,
fun(_use0) ->
{active_worker, Worker@2, _, Receive} = _use0,
Active_workers@1 = gleam@dict:delete(
erlang:element(3, State),
Pid@2
),
case gleam_stdlib:map_get(erlang:element(5, State), Receive) of
{ok, Request} ->
Request@1 = {request,
erlang:element(2, Request),
gleam@set:delete(
erlang:element(3, Request),
Pid@2
),
erlang:element(4, Request)},
Requests = gleam@dict:insert(
erlang:element(5, State),
Receive,
Request@1
),
_pipe@1 = {state,
erlang:element(2, State),
Active_workers@1,
erlang:element(4, State),
Requests,
erlang:element(6, State)},
try_dequeue_work(_pipe@1, Worker@2);
{error, _} ->
{state,
erlang:element(2, State),
Active_workers@1,
erlang:element(4, State),
erlang:element(5, State),
erlang:element(6, State)}
end
end
);
{enqueue, _, []} ->
State;
{enqueue, Receive@1, Work} ->
try_(
gleam_stdlib:map_get(erlang:element(5, State), Receive@1),
State,
fun(Request@2) -> enqueue_loop(State, Request@2, Work) end
);
{subscribe, Receive@2} ->
try_(
gleam@erlang@process:subject_owner(Receive@2),
State,
fun(Pid@3) ->
Request@3 = begin
_pipe@2 = gleam_stdlib:map_get(
erlang:element(5, State),
Receive@2
),
gleam@result:unwrap(
_pipe@2,
{request, Pid@3, gleam@set:new(), Receive@2}
)
end,
Caller@2 = case gleam_stdlib:map_get(
erlang:element(4, State),
Pid@3
) of
{ok, Caller@1} ->
{caller,
erlang:element(2, Caller@1),
erlang:element(3, Caller@1),
gleam@set:insert(
erlang:element(4, Caller@1),
Receive@2
)};
{error, _} ->
Monitor@1 = gleam@erlang@process:monitor(Pid@3),
{caller,
Pid@3,
Monitor@1,
begin
_pipe@3 = gleam@set:new(),
gleam@set:insert(_pipe@3, Receive@2)
end}
end,
Callers = gleam@dict:insert(
erlang:element(4, State),
Pid@3,
Caller@2
),
Requests@1 = gleam@dict:insert(
erlang:element(5, State),
Receive@2,
Request@3
),
{state,
erlang:element(2, State),
erlang:element(3, State),
Callers,
Requests@1,
erlang:element(6, State)}
end
);
{unsubscribe, Receive@3} ->
try_(
gleam_stdlib:map_get(erlang:element(5, State), Receive@3),
State,
fun(Request@4) ->
Requests@2 = gleam@dict:delete(
erlang:element(5, State),
Receive@3
),
gleam@set:each(
erlang:element(3, Request@4),
fun gleam@erlang@process:kill/1
),
try_(
gleam_stdlib:map_get(
erlang:element(4, State),
erlang:element(2, Request@4)
),
{state,
erlang:element(2, State),
erlang:element(3, State),
erlang:element(4, State),
Requests@2,
erlang:element(6, State)},
fun(Caller@3) ->
Caller@4 = {caller,
erlang:element(2, Caller@3),
erlang:element(3, Caller@3),
gleam@set:delete(
erlang:element(4, Caller@3),
Receive@3
)},
Callers@1 = case gleam@set:is_empty(
erlang:element(4, Caller@4)
) of
true ->
gleam@erlang@process:demonitor_process(
erlang:element(3, Caller@4)
),
gleam@dict:delete(
erlang:element(4, State),
erlang:element(2, Caller@4)
);
false ->
gleam@dict:insert(
erlang:element(4, State),
erlang:element(2, Caller@4),
Caller@4
)
end,
{state,
erlang:element(2, State),
erlang:element(3, State),
Callers@1,
Requests@2,
erlang:element(6, State)}
end
)
end
)
end,
gleam@otp@actor:continue(Next_state).
-file("src/crew.gleam", 146).
-spec start_tree(gleam@erlang@process:name(pool_msg()), integer()) -> {ok,
gleam@otp@actor:started(gleam@otp@static_supervisor:supervisor())} |
{error, gleam@otp@actor:start_error()}.
start_tree(Name, Size) ->
gleam@bool:guard(
Size =< 0,
{error, {init_failed, <<"pool size must be greater than zero"/utf8>>}},
fun() ->
Main_supervisor = gleam@otp@static_supervisor:new(rest_for_one),
Worker_supervisor = gleam@otp@static_supervisor:new(one_for_one),
Pool_spec = begin
gleam@otp@supervision:worker(
fun() ->
_pipe = gleam@otp@actor:new_with_initialiser(
1000,
fun init_pool/1
),
_pipe@1 = gleam@otp@actor:named(_pipe, Name),
_pipe@2 = gleam@otp@actor:on_message(
_pipe@1,
fun pool/2
),
gleam@otp@actor:start(_pipe@2)
end
)
end,
Worker_spec = begin
gleam@otp@supervision:worker(
fun() ->
Pid = proc_lib:spawn_link(fun() -> worker(Name) end),
{ok, {started, Pid, nil}}
end
)
end,
Worker_supervisor_spec = begin
gleam@otp@supervision:supervisor(
fun() -> _pipe@3 = Worker_supervisor,
_pipe@4 = repeat(
Size,
_pipe@3,
fun(_capture) ->
gleam@otp@static_supervisor:add(
_capture,
Worker_spec
)
end
),
gleam@otp@static_supervisor:start(_pipe@4) end
)
end,
_pipe@5 = Main_supervisor,
_pipe@6 = gleam@otp@static_supervisor:add(_pipe@5, Pool_spec),
_pipe@7 = gleam@otp@static_supervisor:add(
_pipe@6,
Worker_supervisor_spec
),
gleam@otp@static_supervisor:start(_pipe@7)
end
).
-file("src/crew.gleam", 115).
?DOC(
" Start an unsupervised worker pool from the given builder.\n"
"\n"
" Returns a supervisor that manages the pool and its workers. In most cases,\n"
" you should use `supervised` instead to get a child specification that can\n"
" be added to your application's supervision tree.\n"
"\n"
" ## Panics\n"
" This function will exit the process if any workers fail to start, similar\n"
" to `static_supervisor.start`.\n"
"\n"
" ## Example\n"
"\n"
" ```gleam\n"
" let assert Ok(pool_supervisor) =\n"
" crew.new(pool_name)\n"
" |> crew.start\n"
" ```\n"
).
-spec start(builder()) -> {ok, gleam@otp@static_supervisor:supervisor()} |
{error, gleam@otp@actor:start_error()}.
start(Builder) ->
{builder, Name, Size} = Builder,
gleam@result:map(
start_tree(Name, Size),
fun(Result) -> erlang:element(3, Result) end
).
-file("src/crew.gleam", 140).
?DOC(
" Create a child specification for a supervised worker pool.\n"
"\n"
" This is the recommended way to start a worker pool as part of your\n"
" application's supervision tree. The returned child specification can be\n"
" added to a supervisor using `static_supervisor.add`.\n"
"\n"
" ## Example\n"
"\n"
" ```gleam\n"
" let pool_spec =\n"
" crew.new(pool_name)\n"
" |> crew.fixed_size(4)\n"
" |> crew.supervised\n"
"\n"
" let assert Ok(_) =\n"
" supervisor.new(supervisor.OneForOne)\n"
" |> supervisor.add(pool_spec)\n"
" |> supervisor.start\n"
" ```\n"
).
-spec supervised(builder()) -> gleam@otp@supervision:child_specification(gleam@otp@static_supervisor:supervisor()).
supervised(Builder) ->
{builder, Name, Size} = Builder,
gleam@otp@supervision:supervisor(fun() -> start_tree(Name, Size) end).
-file("src/crew.gleam", 78).
?DOC(
" Create a new worker pool builder with the given name.\n"
"\n"
" The name is used to register the pool so that work can be sent to it.\n"
"\n"
" By default, the pool will have a number of workers equal to the number of\n"
" scheduler threads available on the system (typically the number of CPU cores).\n"
"\n"
" ## Example\n"
"\n"
" ```gleam\n"
" let pool_name = process.new_name(\"pool\")\n"
" let builder = crew.new(pool_name)\n"
" ```\n"
).
-spec new(gleam@erlang@process:name(pool_msg())) -> builder().
new(Name) ->
{builder, Name, crew_ffi:scheduler_count()}.
-file("src/crew.gleam", 358).
-spec receive_loop(
list(fun(() -> FLP)),
gleam@erlang@process:selector(FLP),
integer(),
list(FLP)
) -> {ok, list(FLP)} | {error, nil}.
receive_loop(Work, Selector, Timeout_end, State) ->
case Work of
[] ->
{ok, State};
[_ | Work@1] ->
Timeout = Timeout_end - crew_ffi:system_time(),
case gleam_erlang_ffi:select(Selector, Timeout) of
{ok, Result} ->
receive_loop(
Work@1,
Selector,
Timeout_end,
[Result | State]
);
{error, nil} ->
{error, nil}
end
end.
-file("src/crew.gleam", 435).
?DOC(
" Add a worker pool channel to a selector for receiving work results.\n"
"\n"
" This allows you to receive work results as part of the larger message-handling\n"
" loop. Used in conjunction with `subscribe` and `enqueue` for lower-level\n"
" pool usage.\n"
"\n"
" Work results arrive in completion order. It is your responsibility to handle\n"
" the lifecycle and ordering.\n"
"\n"
" ## Panics\n"
" - If a worker or the pool crashes while executing work, the selector will\n"
" panic with details about the crash.\n"
).
-spec select_map(
gleam@erlang@process:selector(FLZ),
channel(FMB),
fun((FMB) -> FLZ)
) -> gleam@erlang@process:selector(FLZ).
select_map(Selector, Channel, Tagger) ->
_pipe = Selector,
_pipe@1 = gleam@erlang@process:select_specific_monitor(
_pipe,
erlang:element(4, Channel),
fun(Down) ->
Msg = <<"Pool exited while waiting for work to complete: "/utf8,
(gleam@string:inspect(Down))/binary>>,
erlang:error(#{gleam_error => panic,
message => Msg,
file => <<?FILEPATH/utf8>>,
module => <<"crew"/utf8>>,
function => <<"select_map"/utf8>>,
line => 445})
end
),
gleam@erlang@process:select_map(
_pipe@1,
erlang:element(3, Channel),
fun(Result) -> case Result of
{done, Value} ->
Tagger(crew_ffi:identity(Value));
{worker_exited, Reason} ->
Msg@1 = <<"Worker exited: "/utf8,
(gleam@string:inspect(Reason))/binary>>,
erlang:error(#{gleam_error => panic,
message => Msg@1,
file => <<?FILEPATH/utf8>>,
module => <<"crew"/utf8>>,
function => <<"select_map"/utf8>>,
line => 452})
end end
).
-file("src/crew.gleam", 490).
?DOC(
" Submit multiple pieces of work to the pool using a subscription channel.\n"
"\n"
" This is a lower-level function for submitting work. Results must be handled\n"
" using a selector with `select_map`. It is the callers responsibility to handle\n"
" timeouts and submission order. Most users should prefer the `work*` functions.\n"
).
-spec enqueue_many(channel(FML), list(fun(() -> FML))) -> nil.
enqueue_many(Channel, Work) ->
gleam@otp@actor:send(
erlang:element(2, Channel),
{enqueue, erlang:element(3, Channel), crew_ffi:identity(Work)}
).
-file("src/crew.gleam", 332).
-spec do_work(
gleam@erlang@process:name(pool_msg()),
integer(),
list(fun(() -> FLM))
) -> list(FLM).
do_work(Pool, Timeout, Work) ->
Channel = subscribe(Pool),
enqueue_many(Channel, Work),
Selector = begin
_pipe = gleam_erlang_ffi:new_selector(),
select_map(_pipe, Channel, fun(X) -> X end)
end,
Timeout_end = crew_ffi:system_time() + Timeout,
Result = receive_loop(Work, Selector, Timeout_end, []),
unsubscribe(Channel, Selector),
case Result of
{ok, Value} ->
Value;
{error, _} ->
erlang:error(#{gleam_error => panic,
message => <<"Pool did not complete work in the alloted time"/utf8>>,
file => <<?FILEPATH/utf8>>,
module => <<"crew"/utf8>>,
function => <<"do_work"/utf8>>,
line => 354})
end.
-file("src/crew.gleam", 246).
?DOC(
" Execute multiple pieces of work concurrently on the pool, without\n"
" ordering guarantees.\n"
"\n"
" Work is distributed among the available workers and executed concurrently.\n"
" Results are returned in the order they complete, not the order they were\n"
" submitted.\n"
"\n"
" ## Parameters\n"
" - `pool` - The name of the pool to execute work on\n"
" - `timeout` - Maximum time to wait for all work to complete in milliseconds\n"
" - `work` - A list of functions containing work to be executed\n"
"\n"
" ## Panics\n"
" - If the pool does not complete all work within the specified timeout\n"
" - If the pool is not running\n"
" - If any worker crashes while executing work\n"
"\n"
" ## Example\n"
"\n"
" ```gleam\n"
" let users = crew.work_many(pool_name, 10000, [\n"
" fn() { fetch_user_data(user1) },\n"
" fn() { fetch_user_data(user2) },\n"
" fn() { fetch_user_data(user3) },\n"
" ])\n"
" ```\n"
).
-spec work_many(
gleam@erlang@process:name(pool_msg()),
integer(),
list(fun(() -> FKZ))
) -> list(FKZ).
work_many(Pool, Timeout, Work) ->
lists:reverse(do_work(Pool, Timeout, Work)).
-file("src/crew.gleam", 211).
?DOC(
" Execute a single piece of work on the pool and wait for the result.\n"
"\n"
" This function blocks until the work is completed or the timeout is reached.\n"
" The work function is executed on one of the worker processes in the pool.\n"
"\n"
" ## Parameters\n"
" - `pool` - The name of the pool to execute work on\n"
" - `timeout` - Maximum time to wait for completion in milliseconds\n"
" - `work` - A function containing the work to be executed\n"
"\n"
" ## Panics\n"
" - If the pool does not complete the work within the specified timeout\n"
" - If the pool is not running\n"
" - If the worker crashes while executing the work\n"
"\n"
" ## Example\n"
"\n"
" ```gleam\n"
" let response = crew.work(pool_name, 5000, fn() {\n"
" httpc.send(...)\n"
" })\n"
" ```\n"
).
-spec work(gleam@erlang@process:name(pool_msg()), integer(), fun(() -> FKX)) -> FKX.
work(Pool, Timeout, Work) ->
Result@1 = case work_many(Pool, Timeout, [Work]) of
[Result] -> Result;
_assert_fail ->
erlang:error(#{gleam_error => let_assert,
message => <<"Pattern match failed, no pattern matched the value."/utf8>>,
file => <<?FILEPATH/utf8>>,
module => <<"crew"/utf8>>,
function => <<"work"/utf8>>,
line => 216,
value => _assert_fail,
start => 6392,
'end' => 6446,
pattern_start => 6403,
pattern_end => 6411})
end,
Result@1.
-file("src/crew.gleam", 316).
?DOC(
" Apply a function to each element of a list concurrently using the worker pool.\n"
"\n"
" This is similar to `list.map` but executes the mapping function\n"
" concurrently across all workers. Results are returned in the same\n"
" order as the input list.\n"
"\n"
" There's a bunch of extra overhead involved with spawning a work item\n"
" per list element and making sure the order matches. Depending on your\n"
" workload it might make sense to split your list into chunks first to reduce\n"
" work queue pressure.\n"
"\n"
" ## Parameters\n"
" - `list` - The list of items to map over\n"
" - `pool` - The name of the pool to execute work on\n"
" - `timeout` - Maximum time to wait for all mappings to complete in milliseconds\n"
" - `fun` - The function to apply to each item\n"
"\n"
" ## Panics\n"
" - If the pool does not complete all mappings within the specified timeout\n"
" - If the pool is not running\n"
" - If any worker crashes while executing a mapping\n"
"\n"
" ## Example\n"
"\n"
" ```gleam\n"
" let user_ids = [1, 2, 3, 4, 5]\n"
" let users = crew.parallel_map(user_ids, pool_name, 5000, fetch_user_data)\n"
" ```\n"
).
-spec parallel_map(
list(FLG),
gleam@erlang@process:name(pool_msg()),
integer(),
fun((FLG) -> FLJ)
) -> list(FLJ).
parallel_map(List, Pool, Timeout, Fun) ->
Unordered = begin
_pipe = List,
_pipe@1 = gleam@list:index_map(
_pipe,
fun(Item, Index) -> fun() -> {Index, Fun(Item)} end end
),
do_work(Pool, Timeout, _pipe@1)
end,
_pipe@2 = Unordered,
_pipe@3 = gleam@list:sort(
_pipe@2,
fun(A, B) ->
gleam@int:compare(erlang:element(1, A), erlang:element(1, B))
end
),
gleam@list:map(_pipe@3, fun(X) -> erlang:element(2, X) end).
-file("src/crew.gleam", 280).
?DOC(
" Execute multiple pieces of work concurrently on the pool and return results\n"
" in submission order.\n"
"\n"
" Work functions are distributed across available workers and executed\n"
" concurrently, but results are reordered to match the original submission\n"
" order before being returned.\n"
"\n"
" ## Parameters\n"
" - `pool` - The name of the pool to execute work on\n"
" - `timeout` - Maximum time to wait for all work to complete in milliseconds\n"
" - `work` - A list of functions containing work to be executed\n"
"\n"
" ## Panics\n"
" - If the pool does not complete all work within the specified timeout\n"
" - If the pool is not running\n"
" - If any worker crashes while executing work\n"
"\n"
" ## Example\n"
"\n"
" ```gleam\n"
" let assert [user1, user2, user3] = crew.work_ordered(pool_name, 10000, [\n"
" fn() { fetch_user_data(user1) },\n"
" fn() { fetch_user_data(user2) },\n"
" fn() { fetch_user_data(user3) },\n"
" ])\n"
" ```\n"
).
-spec work_ordered(
gleam@erlang@process:name(pool_msg()),
integer(),
list(fun(() -> FLD))
) -> list(FLD).
work_ordered(Pool, Timeout, Work) ->
parallel_map(Work, Pool, Timeout, fun(F) -> F() end).
-file("src/crew.gleam", 481).
?DOC(
" Submit a single piece of work to the pool using a subscription channel.\n"
"\n"
" This is a lower-level function for submitting work. Results must be handled\n"
" using a selector with `select_map`. It is the callers responsibility to handle\n"
" timeouts and submission order. Most users should prefer the `work*` functions.\n"
).
-spec enqueue(channel(FMJ), fun(() -> FMJ)) -> nil.
enqueue(Channel, Work) ->
enqueue_many(Channel, [Work]).