crew

-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]).