-module(crew@task_pool). -compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch, inline]). -define(FILEPATH, "src/crew/task_pool.gleam"). -export([fixed_size/2, max_queue_length/2, start/1, supervised/1, new/1, run/3, run_all/3, parallel_map/4, run_sorted/3, submit_all/4, submit/4]). -export_type([any_/0, builder/0, pool_msg/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( " The `task_pool` module implements a simple, generic task pool that can run\n" " arbitrary functions passed to it using the generic crew pool under the hood.\n" ). -type any_() :: any(). -opaque builder() :: {builder, crew:builder(nil, fun(() -> any_()), any_())}. -type pool_msg() :: any(). -file("src/crew/task_pool.gleam", 53). ?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" " task_pool.new(pool_name)\n" " |> task_pool.fixed_size(8) // Use 8 workers regardless of CPU count\n" " ```\n" ). -spec fixed_size(builder(), integer()) -> builder(). fixed_size(Builder, Size) -> {builder, Builder@1} = Builder, {builder, crew:fixed_size(Builder@1, Size)}. -file("src/crew/task_pool.gleam", 70). ?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 workers in the pool.\n" " ## Example\n" "\n" " ```gleam\n" " task_pool.new(pool_name)\n" " |> task_pool.max_queue_length(100)\n" " ```\n" ). -spec max_queue_length(builder(), integer()) -> builder(). max_queue_length(Builder, Max_queue_length) -> {builder, Builder@1} = Builder, {builder, crew:max_queue_length(Builder@1, Max_queue_length)}. -file("src/crew/task_pool.gleam", 92). ?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" " task_pool.new(pool_name)\n" " |> task_pool.start\n" " ```\n" ). -spec start(builder()) -> {ok, gleam@otp@static_supervisor:supervisor()} | {error, gleam@otp@actor:start_error()}. start(Builder) -> {builder, Builder@1} = Builder, crew:start(Builder@1). -file("src/crew/task_pool.gleam", 116). ?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" " task_pool.new(pool_name)\n" " |> task_pool.fixed_size(4)\n" " |> task_pool.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, Builder@1} = Builder, crew:supervised(Builder@1). -file("src/crew/task_pool.gleam", 39). ?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 = task_pool.new(pool_name)\n" " ```\n" ). -spec new(gleam@erlang@process:name(pool_msg())) -> builder(). new(Name) -> {builder, crew:new(crew_ffi:identity(Name), fun(F) -> F() end)}. -file("src/crew/task_pool.gleam", 143). ?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 = task_pool.run(pool_name, 5000, fn() {\n" " httpc.send(...)\n" " })\n" " ```\n" ). -spec run(gleam@erlang@process:name(pool_msg()), integer(), fun(() -> HHZ)) -> HHZ. run(Pool, Timeout, Work) -> crew:call(crew_ffi:identity(Pool), Timeout, Work). -file("src/crew/task_pool.gleam", 177). ?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 = task_pool.run_all(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 run_all( gleam@erlang@process:name(pool_msg()), integer(), list(fun(() -> HIB)) ) -> list(HIB). run_all(Pool, Timeout, Work) -> crew:call_parallel(crew_ffi:identity(Pool), Timeout, Work). -file("src/crew/task_pool.gleam", 247). ?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 = task_pool.parallel_map(user_ids, pool_name, 5000, fetch_user_data)\n" " ```\n" ). -spec parallel_map( list(HII), gleam@erlang@process:name(pool_msg()), integer(), fun((HII) -> HIL) ) -> list(HIL). 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 ), crew:do_call(crew_ffi:identity(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/task_pool.gleam", 211). ?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] = task_pool.run_sorted(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 run_sorted( gleam@erlang@process:name(pool_msg()), integer(), list(fun(() -> HIF)) ) -> list(HIF). run_sorted(Pool, Timeout, Work) -> parallel_map(Work, Pool, Timeout, fun(F) -> F() end). -file("src/crew/task_pool.gleam", 299). ?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 `run*` functions.\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()), integer(), gleam@erlang@process:subject({ok, HIT} | {error, gleam@erlang@process:exit_reason()}), list(fun(() -> HIT)) ) -> nil. submit_all(Pool, Timeout, Receive, Work) -> crew:submit_all(crew_ffi:identity(Pool), Timeout, Receive, Work). -file("src/crew/task_pool.gleam", 278). ?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 the `run*` functions.\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()), integer(), gleam@erlang@process:subject({ok, HIO} | {error, gleam@erlang@process:exit_reason()}), fun(() -> HIO) ) -> nil. submit(Pool, Timeout, Receive, Work) -> submit_all(Pool, Timeout, Receive, [Work]).