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, max_queue_length/2, cancel/2, new_with_initialiser/4, new_with_state/3, new/2, do_call/3, call/3, call_parallel/3, submit_all/4, submit/4, start/1, supervised/1]).
-export_type([builder/3, pool_msg/2, state/2, worker/2, active_worker/2, caller/1, channel/2, request/2, work/2]).
-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/erlang/process\n"
"\n"
" pub fn main() {\n"
" // Create a pool name\n"
" let pool_name = process.new_name(\"image_downloader\")\n"
"\n"
" // Start an unsupervised pool\n"
" let assert Ok(_) =\n"
" crew.new(pool_name, download_image)\n"
" |> crew.fixed_size(4)\n"
" |> crew.start\n"
"\n"
" // Execute work on the pool\n"
" let result = crew.run(pool_name, 5000, \"tasteful-ramen.jpeg\")\n"
" }\n"
" ```\n"
).
-opaque builder(FKL, FKM, FKN) :: {builder,
gleam@erlang@process:name(pool_msg(FKM, FKN)),
integer(),
gleam@option:option(integer()),
fun(() -> FKL),
integer(),
fun((FKL, FKM) -> FKN)}.
-opaque pool_msg(FKO, FKP) :: {worker_started,
gleam@erlang@process:pid_(),
gleam@erlang@process:subject(work(FKO, FKP))} |
{worker_idle, gleam@erlang@process:pid_()} |
{monitored_process_exited, gleam@erlang@process:down()} |
{get_worker_count, gleam@erlang@process:subject(integer())} |
{enqueue,
gleam@erlang@process:subject({ok, FKP} |
{error, gleam@erlang@process:exit_reason()}),
list(FKO),
gleam@option:option(gleam@erlang@process:subject(nil))} |
{cancel,
gleam@erlang@process:subject({ok, FKP} |
{error, gleam@erlang@process:exit_reason()})}.
-type state(FKQ, FKR) :: {state,
gleam@erlang@process:name(pool_msg(FKQ, FKR)),
integer(),
list(worker(FKQ, FKR)),
gleam@dict:dict(gleam@erlang@process:pid_(), active_worker(FKQ, FKR)),
gleam@dict:dict(gleam@erlang@process:pid_(), caller(FKR)),
gleam@dict:dict(gleam@erlang@process:subject({ok, FKR} |
{error, gleam@erlang@process:exit_reason()}), channel(FKQ, FKR)),
gleam@deque:deque(work(FKQ, FKR)),
gleam@deque:deque(request(FKQ, FKR))}.
-type worker(FKS, FKT) :: {worker,
gleam@erlang@process:pid_(),
gleam@erlang@process:subject(work(FKS, FKT)),
gleam@erlang@process:monitor()}.
-type active_worker(FKU, FKV) :: {active_worker,
worker(FKU, FKV),
work(FKU, FKV)}.
-type caller(FKW) :: {caller,
gleam@erlang@process:pid_(),
gleam@erlang@process:monitor(),
gleam@set:set(gleam@erlang@process:subject({ok, FKW} |
{error, gleam@erlang@process:exit_reason()}))}.
-type channel(FKX, FKY) :: {channel,
gleam@erlang@process:pid_(),
gleam@set:set(gleam@erlang@process:pid_()),
gleam@erlang@process:subject({ok, FKY} |
{error, gleam@erlang@process:exit_reason()})} |
{gleam_phantom, FKX}.
-type request(FKZ, FLA) :: {request,
gleam@erlang@process:pid_(),
gleam@erlang@process:subject({ok, FLA} |
{error, gleam@erlang@process:exit_reason()}),
list(FKZ),
gleam@option:option(gleam@erlang@process:subject(nil))}.
-type work(FLB, FLC) :: {work,
FLB,
gleam@erlang@process:pid_(),
gleam@erlang@process:subject({ok, FLC} |
{error, gleam@erlang@process:exit_reason()})}.
-file("src/crew.gleam", 131).
?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(FMH, FMI, FMJ), integer()) -> builder(FMH, FMI, FMJ).
fixed_size(Builder, Size) ->
{builder,
erlang:element(2, Builder),
Size,
erlang:element(4, Builder),
erlang:element(5, Builder),
erlang:element(6, Builder),
erlang:element(7, Builder)}.
-file("src/crew.gleam", 151).
?DOC(
" To avoid overloading the system, the internal work queue is limited.\n"
"\n"
" If this limit is reached, callers of `enqueue` have to wait until enough\n"
" work has been done before continuing.\n"
"\n"
" By default, the `max_queue_size` depends on the number of workers in the pool.\n"
"\n"
" ## Example\n"
"\n"
" ```gleam\n"
" crew.new(pool_name, worder)\n"
" |> crew.max_queue_length(100)\n"
" ```\n"
).
-spec max_queue_length(builder(FMQ, FMR, FMS), integer()) -> builder(FMQ, FMR, FMS).
max_queue_length(Builder, Max_queue_length) ->
{builder,
erlang:element(2, Builder),
erlang:element(3, Builder),
{some, Max_queue_length},
erlang:element(5, Builder),
erlang:element(6, Builder),
erlang:element(7, Builder)}.
-file("src/crew.gleam", 376).
?DOC(
" The first time `enqueue*` is called from a process, the pool starts to monitor\n"
" that process and cancels all ongoing work in case it goes down.\n"
"\n"
" Sometimes it is useful to manually unsubscribe and cancel all ongoing work\n"
" for a subject. Doing so will also remove the monitor added in the pool once\n"
" the last `receive` subject got cancelled.\n"
"\n"
" Note that finished work might still arrive on this selector after\n"
" `cancel` got called.\n"
).
-spec cancel(
gleam@erlang@process:name(pool_msg(any(), FPA)),
gleam@erlang@process:subject({ok, FPA} |
{error, gleam@erlang@process:exit_reason()})
) -> nil.
cancel(Pool, Receive) ->
gleam@otp@actor:send(
gleam@erlang@process:named_subject(Pool),
{cancel, Receive}
).
-file("src/crew.gleam", 748).
-spec enqueue_loop2(
gleam@erlang@process:pid_(),
gleam@erlang@process:subject({ok, FRF} |
{error, gleam@erlang@process:exit_reason()}),
list(FRH),
gleam@deque:deque(work(FRH, FRF)),
integer()
) -> {gleam@deque:deque(work(FRH, FRF)), list(FRH), integer()}.
enqueue_loop2(Caller, Receive, Work, Queue, Capacity) ->
case Work of
[Work@1 | Rest] when Capacity > 0 ->
Queue@1 = gleam@deque:push_back(
Queue,
{work, Work@1, Caller, Receive}
),
Capacity@1 = Capacity - 1,
enqueue_loop2(Caller, Receive, Rest, Queue@1, Capacity@1);
_ ->
{Queue, Work, Capacity}
end.
-file("src/crew.gleam", 894).
-spec worker_loop(
gleam@erlang@process:subject(pool_msg(any(), any())),
gleam@erlang@process:pid_(),
gleam@erlang@process:subject(work(FXY, FYC)),
FTA,
fun((FTA, FXY) -> FYC)
) -> nil.
worker_loop(Pool, Self, Subject, State, Do_work) ->
{work, Work, _, Receive} = gleam_erlang_ffi:'receive'(Subject),
Result = Do_work(State, Work),
gleam@erlang@process:send(Pool, {worker_idle, Self}),
gleam@erlang@process:send(Receive, {ok, Result}),
worker_loop(Pool, Self, Subject, State, Do_work).
-file("src/crew.gleam", 882).
-spec worker(builder(any(), any(), any()), gleam@erlang@process:subject(nil)) -> nil.
worker(Builder, Initialised) ->
Self = erlang:self(),
Subject = gleam@erlang@process:new_subject(),
Pool_subject = gleam@erlang@process:named_subject(
erlang:element(2, Builder)
),
State = (erlang:element(5, Builder))(),
gleam@erlang@process:send(Initialised, nil),
gleam@erlang@process:send(Pool_subject, {worker_started, Self, Subject}),
worker_loop(Pool_subject, Self, Subject, State, erlang:element(7, Builder)).
-file("src/crew.gleam", 855).
-spec start_worker(builder(any(), any(), any())) -> {ok,
gleam@otp@actor:started(nil)} |
{error, gleam@otp@actor:start_error()}.
start_worker(Builder) ->
Initialised = gleam@erlang@process:new_subject(),
Pid = proc_lib:spawn_link(fun() -> worker(Builder, Initialised) end),
Monitor = gleam@erlang@process:monitor(Pid),
Selector = begin
_pipe = gleam_erlang_ffi:new_selector(),
_pipe@1 = gleam@erlang@process:select_map(
_pipe,
Initialised,
fun(Field@0) -> {ok, Field@0} end
),
gleam@erlang@process:select_specific_monitor(
_pipe@1,
Monitor,
fun(Field@0) -> {error, Field@0} end
)
end,
Result = case gleam_erlang_ffi:select(Selector, erlang:element(6, Builder)) of
{ok, {ok, nil}} ->
{ok, {started, Pid, nil}};
{ok, {error, Down}} ->
{error, {init_exited, erlang:element(4, Down)}};
{error, nil} ->
gleam@erlang@process:unlink(Pid),
gleam@erlang@process:kill(Pid),
{error, init_timeout}
end,
gleam@erlang@process:demonitor_process(Monitor),
Result.
-file("src/crew.gleam", 907).
-spec repeat(integer(), FTC, fun((FTC) -> FTC)) -> FTC.
repeat(Times, State, F) ->
case Times > 0 of
true ->
repeat(Times - 1, F(State), F);
false ->
State
end.
-file("src/crew.gleam", 918).
-spec 'try'({ok, FTD} | {error, FTE}, fun((FTE) -> FTH), fun((FTD) -> FTH)) -> FTH.
'try'(Result, Or, Then) ->
case Result of
{ok, X} ->
Then(X);
{error, X@1} ->
Or(X@1)
end.
-file("src/crew.gleam", 925).
-spec try_({ok, FTJ} | {error, any()}, FTN, fun((FTJ) -> FTN)) -> FTN.
try_(Result, Or, Then) ->
case Result of
{ok, X} ->
Then(X);
{error, _} ->
Or
end.
-file("src/crew.gleam", 111).
?DOC(
" Create a new worker pool builder with the given name and initialiser.\n"
"\n"
" The name is used to register the pool so that work can be sent to it.\n"
" The initialiser will run on the worker process before it registers itself\n"
" as a worker.\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"
).
-spec new_with_initialiser(
gleam@erlang@process:name(pool_msg(FLY, FLZ)),
integer(),
fun(() -> FMD),
fun((FMD, FLY) -> FLZ)
) -> builder(FMD, FLY, FLZ).
new_with_initialiser(Name, Init_timeout, Init, Work) ->
Size = crew_ffi:scheduler_count(),
{builder, Name, Size, none, Init, Init_timeout, Work}.
-file("src/crew.gleam", 95).
?DOC(
" Create a new worker pool builder with the given name and state.\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"
).
-spec new_with_state(
gleam@erlang@process:name(pool_msg(FLP, FLQ)),
FLU,
fun((FLU, FLP) -> FLQ)
) -> builder(FLU, FLP, FLQ).
new_with_state(Name, State, Work) ->
new_with_initialiser(Name, 1000, fun() -> State end, Work).
-file("src/crew.gleam", 82).
?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, worker)\n"
" ```\n"
).
-spec new(gleam@erlang@process:name(pool_msg(FLH, FLI)), fun((FLH) -> FLI)) -> builder(nil, FLH, FLI).
new(Name, Work) ->
new_with_state(Name, nil, fun(_, Input) -> Work(Input) end).
-file("src/crew.gleam", 346).
-spec receive_loop(
list(any()),
gleam@erlang@process:selector(FOT),
integer(),
list(FOT)
) -> {ok, list(FOT)} | {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", 300).
?DOC(false).
-spec do_call(
gleam@erlang@process:name(pool_msg(FOK, FOL)),
integer(),
list(FOK)
) -> list(FOL).
do_call(Pool, Timeout, Work) ->
Timeout_end = crew_ffi:system_time() + Timeout,
Pool_pid@1 = case gleam_erlang_ffi:process_named(Pool) 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 => <<"do_call"/utf8>>,
line => 306,
value => _assert_fail,
start => 9444,
'end' => 9489,
pattern_start => 9455,
pattern_end => 9467})
end,
Monitor = gleam@erlang@process:monitor(Pool_pid@1),
Receive = gleam@erlang@process:new_subject(),
gleam@otp@actor:send(
gleam@erlang@process:named_subject(Pool),
{enqueue, Receive, Work, none}
),
Selector = begin
_pipe = gleam_erlang_ffi:new_selector(),
_pipe@1 = gleam@erlang@process:select_specific_monitor(
_pipe,
Monitor,
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 => <<"do_call"/utf8>>,
line => 321})
end
),
gleam@erlang@process:select_map(
_pipe@1,
Receive,
fun(Result) -> case Result of
{ok, Value} ->
Value;
{error, Down@1} ->
gleam@erlang@process:demonitor_process(Monitor),
Msg@1 = <<"Worker exited: "/utf8,
(gleam@string:inspect(Down@1))/binary>>,
erlang:error(#{gleam_error => panic,
message => Msg@1,
file => <<?FILEPATH/utf8>>,
module => <<"crew"/utf8>>,
function => <<"do_call"/utf8>>,
line => 330})
end end
)
end,
Result@1 = receive_loop(Work, Selector, Timeout_end, []),
gleam@erlang@process:demonitor_process(Monitor),
cancel(Pool, Receive),
case Result@1 of
{ok, Value@1} ->
Value@1;
{error, _} ->
erlang:error(#{gleam_error => panic,
message => <<"Pool did not complete work in time"/utf8>>,
file => <<?FILEPATH/utf8>>,
module => <<"crew"/utf8>>,
function => <<"do_call"/utf8>>,
line => 342})
end.
-file("src/crew.gleam", 266).
?DOC(
" Send a single piece of work to one of the workers and wait for the result.\n"
"\n"
" This function blocks until the work is completed or the timeout is reached.\n"
" If no worker is available, the message will be queued and sent to the next\n"
" free worker.\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` - The message to send to the worker function\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"
).
-spec call(gleam@erlang@process:name(pool_msg(FNY, FNZ)), integer(), FNY) -> FNZ.
call(Pool, Timeout, Work) ->
Result@1 = case do_call(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 => <<"call"/utf8>>,
line => 271,
value => _assert_fail,
start => 8365,
'end' => 8417,
pattern_start => 8376,
pattern_end => 8384})
end,
Result@1.
-file("src/crew.gleam", 291).
?DOC(
" Send multiple pieces of work concurrently to the pool workers, 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 messages describing the work to be done\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"
).
-spec call_parallel(
gleam@erlang@process:name(pool_msg(FOD, FOE)),
integer(),
list(FOD)
) -> list(FOE).
call_parallel(Pool, Timeout, Work) ->
lists:reverse(do_call(Pool, Timeout, Work)).
-file("src/crew.gleam", 416).
?DOC(
" Submit multiple pieces of work to the pool using a subscription channel.\n"
" The work will be done asynchronously and the result will be sent back\n"
" to the provided subject. The timeout controls how long to wait for the\n"
" work to be added to the queue successfully.\n"
"\n"
" This is a lower-level function for submitting work. It is the callers\n"
" responsibility to handle timeouts, submission order and failures.\n"
" Most users should prefer the `call_parallel` function.\n"
"\n"
" The first time you `enqueue` is called from a process, the pool sets up\n"
" a monitor making sure work is cancelled when the process no longer exists\n"
" to receive a result. You can clean up this monitor early by using `cancel`.\n"
).
-spec submit_all(
gleam@erlang@process:name(pool_msg(FPP, FPQ)),
integer(),
gleam@erlang@process:subject({ok, FPQ} |
{error, gleam@erlang@process:exit_reason()}),
list(FPP)
) -> nil.
submit_all(Pool, Timeout, Receive, Work) ->
Counter = crew_ffi:get_counter(Pool),
Subject = gleam@erlang@process:named_subject(Pool),
case Work of
[] ->
nil;
[_ | _] when Counter > 0 ->
gleam@otp@actor:send(Subject, {enqueue, Receive, Work, none});
[_ | _] ->
gleam@otp@actor:call(
Subject,
Timeout,
fun(Enqueued) -> {enqueue, Receive, Work, {some, Enqueued}} end
)
end.
-file("src/crew.gleam", 395).
?DOC(
" Submit a single piece of work to the pool using a subscription channel.\n"
" The work will be done asynchronously and the result will be sent back\n"
" to the provided subject. The timeout controls how long to wait for the\n"
" work to be added to the queue successfully.\n"
"\n"
" This is a lower-level function for submitting work. It is the callers\n"
" responsibility to handle timeouts, submission order and failures.\n"
" Most users should prefer `call`.\n"
"\n"
" The first time you `enqueue` is called from a process, the pool sets up\n"
" a monitor making sure work is cancelled when the process no longer exists\n"
" to receive a result. You can clean up this monitor early by using `cancel`.\n"
).
-spec submit(
gleam@erlang@process:name(pool_msg(FPH, FPI)),
integer(),
gleam@erlang@process:subject({ok, FPI} |
{error, gleam@erlang@process:exit_reason()}),
FPH
) -> nil.
submit(Pool, Timeout, Receive, Work) ->
submit_all(Pool, Timeout, Receive, [Work]).
-file("src/crew.gleam", 793).
-spec pop_work(state(FRV, FRW)) -> {ok, {work(FRV, FRW), state(FRV, FRW)}} |
{error, nil}.
pop_work(State) ->
gleam@result:'try'(
gleam@deque:pop_front(erlang:element(8, State)),
fun(_use0) ->
{Work, Queue} = _use0,
'try'(
gleam@deque:pop_front(erlang:element(9, State)),
fun(_) ->
crew_ffi:increment_counter(erlang:element(2, State), 1),
{ok,
{Work,
{state,
erlang:element(2, State),
erlang:element(3, State),
erlang:element(4, State),
erlang:element(5, State),
erlang:element(6, State),
erlang:element(7, State),
Queue,
erlang:element(9, State)}}}
end,
fun(_use0@1) ->
{{request, Caller, Receive, Request, Enqueued},
Overflow_queue} = _use0@1,
case Request of
[First | Rest] ->
Queue@1 = gleam@deque:push_back(
Queue,
{work, First, Caller, Receive}
),
Overflow_queue@1 = case Rest of
[_ | _] ->
Request@1 = {request,
Caller,
Receive,
Rest,
Enqueued},
gleam@deque:push_front(
Overflow_queue,
Request@1
);
[] ->
case Enqueued of
{some, Enqueued@1} ->
gleam@erlang@process:send(
Enqueued@1,
nil
);
none ->
nil
end,
Overflow_queue
end,
{ok,
{Work,
{state,
erlang:element(2, State),
erlang:element(3, State),
erlang:element(4, State),
erlang:element(5, State),
erlang:element(6, State),
erlang:element(7, State),
Queue@1,
Overflow_queue@1}}};
[] ->
case Enqueued of
{some, Enqueued@2} ->
gleam@erlang@process:send(Enqueued@2, nil);
none ->
nil
end,
pop_work(
{state,
erlang:element(2, State),
erlang:element(3, State),
erlang:element(4, State),
erlang:element(5, State),
erlang:element(6, State),
erlang:element(7, State),
erlang:element(8, State),
Overflow_queue}
)
end
end
)
end
).
-file("src/crew.gleam", 766).
-spec try_dequeue_work(state(FRN, FRO), worker(FRN, FRO)) -> state(FRN, FRO).
try_dequeue_work(State, Worker) ->
'try'(
pop_work(State),
fun(_) ->
{state,
erlang:element(2, State),
erlang:element(3, State),
[Worker | erlang:element(4, State)],
erlang:element(5, State),
erlang:element(6, State),
erlang:element(7, State),
erlang:element(8, State),
erlang:element(9, State)}
end,
fun(_use0) ->
{Work, State@1} = _use0,
'try'(
gleam_stdlib:map_get(
erlang:element(7, State@1),
erlang:element(4, Work)
),
fun(_) -> try_dequeue_work(State@1, Worker) end,
fun(Channel) ->
gleam@erlang@process:send(erlang:element(3, Worker), Work),
Active_worker = {active_worker, Worker, Work},
Active_workers = gleam@dict:insert(
erlang:element(5, State@1),
erlang:element(2, Worker),
Active_worker
),
Channel@1 = {channel,
erlang:element(2, Channel),
gleam@set:insert(
erlang:element(3, Channel),
erlang:element(2, Worker)
),
erlang:element(4, Channel)},
Channels = gleam@dict:insert(
erlang:element(7, State@1),
erlang:element(4, Work),
Channel@1
),
{state,
erlang:element(2, State@1),
erlang:element(3, State@1),
erlang:element(4, State@1),
Active_workers,
erlang:element(6, State@1),
Channels,
erlang:element(8, State@1),
erlang:element(9, State@1)}
end
)
end
).
-file("src/crew.gleam", 682).
-spec enqueue_loop(
state(FQU, FQV),
channel(FQU, FQV),
list(FQU),
gleam@option:option(gleam@erlang@process:subject(nil))
) -> state(FQU, FQV).
enqueue_loop(State, Channel, Work, Enqueued) ->
{channel, Caller, Workers, Receive} = Channel,
case {Work, erlang:element(4, State)} of
{[Work@1 | Rest], [Worker | Idle_workers]} ->
Work@2 = {work, Work@1, Caller, Receive},
gleam@erlang@process:send(erlang:element(3, Worker), Work@2),
Active_worker = {active_worker, Worker, Work@2},
Active_workers = gleam@dict:insert(
erlang:element(5, State),
erlang:element(2, Worker),
Active_worker
),
Channel@1 = {channel,
erlang:element(2, Channel),
gleam@set:insert(Workers, erlang:element(2, Worker)),
erlang:element(4, Channel)},
_pipe = {state,
erlang:element(2, State),
erlang:element(3, State),
Idle_workers,
Active_workers,
erlang:element(6, State),
erlang:element(7, State),
erlang:element(8, State),
erlang:element(9, State)},
enqueue_loop(_pipe, Channel@1, Rest, Enqueued);
{[_ | _], []} ->
Capacity = crew_ffi:get_counter(erlang:element(2, State)),
{Queue, Overflow, New_capacity} = enqueue_loop2(
Caller,
Receive,
Work,
erlang:element(8, State),
Capacity
),
crew_ffi:decrement_counter(
erlang:element(2, State),
Capacity - New_capacity
),
Overflow_queue = case Overflow of
[] ->
case Enqueued of
{some, Enqueued@1} ->
gleam@erlang@process:send(Enqueued@1, nil);
none ->
nil
end,
erlang:element(9, State);
Work@3 ->
Request = {request, Caller, Receive, Work@3, Enqueued},
gleam@deque:push_back(erlang:element(9, State), Request)
end,
Channels = gleam@dict:insert(
erlang:element(7, State),
Receive,
Channel
),
{state,
erlang:element(2, State),
erlang:element(3, State),
erlang:element(4, State),
erlang:element(5, State),
erlang:element(6, State),
Channels,
Queue,
Overflow_queue};
{[], _} ->
case Enqueued of
{some, Enqueued@2} ->
gleam@erlang@process:send(Enqueued@2, nil);
none ->
nil
end,
{state,
erlang:element(2, State),
erlang:element(3, State),
erlang:element(4, State),
erlang:element(5, State),
erlang:element(6, State),
gleam@dict:insert(erlang:element(7, State), Receive, Channel),
erlang:element(8, State),
erlang:element(9, State)}
end.
-file("src/crew.gleam", 528).
-spec pool(state(FQI, FQJ), pool_msg(FQI, FQJ)) -> gleam@otp@actor:next(state(FQI, FQJ), pool_msg(FQI, FQJ)).
pool(State, Msg) ->
Next_state = case Msg of
{worker_started, Pid, Send} ->
Monitor = gleam@erlang@process:monitor(Pid),
Worker = {worker, Pid, Send, Monitor},
_pipe = {state,
erlang:element(2, State),
erlang:element(3, State) + 1,
erlang:element(4, State),
erlang:element(5, State),
erlang:element(6, State),
erlang:element(7, State),
erlang:element(8, State),
erlang:element(9, State)},
try_dequeue_work(_pipe, Worker);
{monitored_process_exited, {process_down, _, Pid@1, _}} ->
case {gleam_stdlib:map_get(erlang:element(6, State), Pid@1),
gleam_stdlib:map_get(erlang:element(5, State), Pid@1)} of
{{ok, _}, _} ->
try_(
gleam_stdlib:map_get(erlang:element(6, State), Pid@1),
State,
fun(Caller) ->
Channels@1 = gleam@set:fold(
erlang:element(4, Caller),
erlang:element(7, State),
fun(Channels, Receive) ->
try_(
gleam_stdlib:map_get(
erlang:element(7, State),
Receive
),
Channels,
fun(Channel) ->
gleam@set:each(
erlang:element(3, Channel),
fun gleam@erlang@process:kill/1
),
gleam@dict:delete(Channels, Receive)
end
)
end
),
Callers = gleam@dict:delete(
erlang:element(6, State),
Pid@1
),
{state,
erlang:element(2, State),
erlang:element(3, State),
erlang:element(4, State),
erlang:element(5, State),
Callers,
Channels@1,
erlang:element(8, State),
erlang:element(9, State)}
end
);
{{error, _}, {ok, _}} ->
Active_workers = gleam@dict:delete(
erlang:element(5, State),
Pid@1
),
Worker_count = erlang:element(3, State) - 1,
{state,
erlang:element(2, State),
Worker_count,
erlang:element(4, State),
Active_workers,
erlang:element(6, State),
erlang:element(7, State),
erlang:element(8, State),
erlang:element(9, State)};
{{error, _}, {error, _}} ->
Idle_workers = gleam@list:filter(
erlang:element(4, State),
fun(Worker@1) ->
erlang:element(2, Worker@1) /= Pid@1
end
),
Worker_count@1 = erlang:element(3, State) - 1,
{state,
erlang:element(2, State),
Worker_count@1,
Idle_workers,
erlang:element(5, State),
erlang:element(6, State),
erlang:element(7, State),
erlang:element(8, State),
erlang:element(9, State)}
end;
{monitored_process_exited, _} ->
State;
{worker_idle, Pid@2} ->
try_(
gleam_stdlib:map_get(erlang:element(5, State), Pid@2),
State,
fun(_use0) ->
{active_worker, Worker@2, {work, _, _, Receive@1}} = _use0,
Active_workers@1 = gleam@dict:delete(
erlang:element(5, State),
Pid@2
),
case gleam_stdlib:map_get(
erlang:element(7, State),
Receive@1
) of
{ok, Channel@1} ->
Channel@2 = {channel,
erlang:element(2, Channel@1),
gleam@set:delete(
erlang:element(3, Channel@1),
Pid@2
),
erlang:element(4, Channel@1)},
Channels@2 = gleam@dict:insert(
erlang:element(7, State),
Receive@1,
Channel@2
),
_pipe@1 = {state,
erlang:element(2, State),
erlang:element(3, State),
erlang:element(4, State),
Active_workers@1,
erlang:element(6, State),
Channels@2,
erlang:element(8, State),
erlang:element(9, State)},
try_dequeue_work(_pipe@1, Worker@2);
{error, _} ->
{state,
erlang:element(2, State),
erlang:element(3, State) - 1,
erlang:element(4, State),
Active_workers@1,
erlang:element(6, State),
erlang:element(7, State),
erlang:element(8, State),
erlang:element(9, State)}
end
end
);
{get_worker_count, Reply_to} ->
gleam@erlang@process:send(Reply_to, erlang:element(3, State)),
State;
{enqueue, _, [], none} ->
State;
{enqueue, _, [], {some, Enqueued}} ->
gleam@erlang@process:send(Enqueued, nil),
State;
{enqueue, Receive@2, Work, Enqueued@1} ->
try_(
gleam@erlang@process:subject_owner(Receive@2),
State,
fun(Pid@3) ->
Channel@4 = case gleam_stdlib:map_get(
erlang:element(7, State),
Receive@2
) of
{ok, Channel@3} ->
Channel@3;
{error, _} ->
{channel, Pid@3, gleam@set:new(), Receive@2}
end,
Caller@2 = case gleam_stdlib:map_get(
erlang:element(6, 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@2 = gleam@set:new(),
gleam@set:insert(_pipe@2, Receive@2)
end}
end,
_pipe@3 = {state,
erlang:element(2, State),
erlang:element(3, State),
erlang:element(4, State),
erlang:element(5, State),
gleam@dict:insert(
erlang:element(6, State),
Pid@3,
Caller@2
),
erlang:element(7, State),
erlang:element(8, State),
erlang:element(9, State)},
enqueue_loop(_pipe@3, Channel@4, Work, Enqueued@1)
end
);
{cancel, Receive@3} ->
try_(
gleam_stdlib:map_get(erlang:element(7, State), Receive@3),
State,
fun(Channel@5) ->
Channels@3 = gleam@dict:delete(
erlang:element(7, State),
Receive@3
),
gleam@set:each(
erlang:element(3, Channel@5),
fun gleam@erlang@process:kill/1
),
try_(
gleam_stdlib:map_get(
erlang:element(6, State),
erlang:element(2, Channel@5)
),
{state,
erlang:element(2, State),
erlang:element(3, State),
erlang:element(4, State),
erlang:element(5, State),
erlang:element(6, State),
Channels@3,
erlang:element(8, State),
erlang:element(9, 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(6, State),
erlang:element(2, Caller@4)
);
false ->
gleam@dict:insert(
erlang:element(6, State),
erlang:element(2, Caller@4),
Caller@4
)
end,
{state,
erlang:element(2, State),
erlang:element(3, State),
erlang:element(4, State),
erlang:element(5, State),
Callers@1,
Channels@3,
erlang:element(8, State),
erlang:element(9, State)}
end
)
end
)
end,
gleam@otp@actor:continue(Next_state).
-file("src/crew.gleam", 494).
-spec init_pool(
builder(any(), FPZ, FQA),
gleam@erlang@process:subject(pool_msg(FPZ, FQA))
) -> {ok,
gleam@otp@actor:initialised(state(FPZ, FQA), pool_msg(FPZ, FQA), gleam@erlang@process:subject(pool_msg(FPZ, FQA)))} |
{error, any()}.
init_pool(Builder, 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,
Max_queue_length = begin
_pipe@2 = erlang:element(4, Builder),
gleam@option:unwrap(_pipe@2, erlang:element(3, Builder) * 100)
end,
crew_ffi:set_counter(erlang:element(2, Builder), Max_queue_length),
State = {state,
erlang:element(2, Builder),
0,
[],
maps:new(),
maps:new(),
maps:new(),
gleam@deque:new(),
gleam@deque:new()},
_pipe@3 = gleam@otp@actor:initialised(State),
_pipe@4 = gleam@otp@actor:selecting(_pipe@3, Selector),
_pipe@5 = gleam@otp@actor:returning(_pipe@4, Self),
{ok, _pipe@5}.
-file("src/crew.gleam", 210).
-spec start_tree(builder(any(), any(), any())) -> {ok,
gleam@otp@actor:started(gleam@otp@static_supervisor:supervisor())} |
{error, gleam@otp@actor:start_error()}.
start_tree(Builder) ->
gleam@bool:guard(
erlang:element(3, Builder) =< 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(_capture) -> init_pool(Builder, _capture) end
),
_pipe@1 = gleam@otp@actor:named(
_pipe,
erlang:element(2, Builder)
),
_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() -> start_worker(Builder) end)
end,
Worker_supervisor_spec = begin
gleam@otp@supervision:supervisor(
fun() -> _pipe@3 = Worker_supervisor,
_pipe@4 = repeat(
erlang:element(3, Builder),
_pipe@3,
fun(_capture@1) ->
gleam@otp@static_supervisor:add(
_capture@1,
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", 177).
?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(any(), any(), any())) -> {ok,
gleam@otp@static_supervisor:supervisor()} |
{error, gleam@otp@actor:start_error()}.
start(Builder) ->
gleam@result:map(
start_tree(Builder),
fun(Result) -> erlang:element(3, Result) end
).
-file("src/crew.gleam", 203).
?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(any(), any(), any())) -> gleam@otp@supervision:child_specification(gleam@otp@static_supervisor:supervisor()).
supervised(Builder) ->
gleam@otp@supervision:supervisor(fun() -> start_tree(Builder) end).