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