Current section
Files
Jump to
Current section
Files
src/taskle.erl
-module(taskle).
-compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch]).
-define(FILEPATH, "src/taskle.gleam").
-export([async/1, start/1, async_unlinked/1, pid/1, await/2, cancel/1, shutdown/2, parallel_map/3, try_await_all/2, await_forever/1, yield/1, await2/3, await3/4, await4/5, await5/6, race/2, all_settled/2]).
-export_type([task/1, error/0, settled_result/1, task_message/1, await_msg/1, race_message/1, all_settled_message/1]).
-if(?OTP_RELEASE >= 27).
-define(MODULEDOC(Str), -moduledoc(Str)).
-define(DOC(Str), -doc(Str)).
-else.
-define(MODULEDOC(Str), -compile([])).
-define(DOC(Str), -compile([])).
-endif.
-opaque task(FFL) :: {task,
gleam@erlang@process:pid_(),
gleam@erlang@process:monitor(),
gleam@erlang@process:pid_(),
gleam@erlang@process:subject(task_message(FFL))}.
-type error() :: timeout | {crashed, binary()} | not_owner | not_ready.
-type settled_result(FFM) :: {fulfilled, FFM} | {rejected, error()}.
-type task_message(FFN) :: {task_result, FFN}.
-type await_msg(FFO) :: {task_msg, task_message(FFO)} |
{down_msg, gleam@erlang@process:down()}.
-type race_message(FFP) :: {race_result,
{ok, FFP} | {error, error()},
gleam@erlang@process:pid_()}.
-type all_settled_message(FFQ) :: {all_settled_result,
integer(),
settled_result(FFQ)}.
-file("src/taskle.gleam", 58).
?DOC(
" Creates an asynchronous task that runs the given function in a separate process.\n"
"\n"
" The task is unlinked from the calling process, meaning that if the task crashes,\n"
" it won't cause the calling process to crash. Only the process that creates the\n"
" task can await its result or cancel it.\n"
"\n"
" ## Examples\n"
"\n"
" ```gleam\n"
" let task = taskle.async(fn() {\n"
" // This runs in a separate process\n"
" process.sleep(1000)\n"
" 42\n"
" })\n"
" ```\n"
).
-spec async(fun(() -> FFR)) -> task(FFR).
async(Fun) ->
Owner = erlang:self(),
Subject = gleam@erlang@process:new_subject(),
Pid = proc_lib:spawn(
fun() ->
Result = Fun(),
gleam@erlang@process:send(Subject, {task_result, Result})
end
),
Monitor = gleam@erlang@process:monitor(Pid),
{task, Pid, Monitor, Owner, Subject}.
-file("src/taskle.gleam", 307).
?DOC(
" Creates an asynchronous task for side effects that doesn't need to be awaited.\n"
"\n"
" This is useful for fire-and-forget operations where you don't need the result.\n"
" The task runs in an unlinked process and won't affect the parent process if it crashes.\n"
"\n"
" **Key difference from `async_unlinked`:** Tasks created with `start` do NOT send\n"
" their result back, making them truly fire-and-forget. While you can technically\n"
" call `await` on them, it will never receive a result. Use `start` for side effects\n"
" like logging, cleanup, or background processing where you don't care about the return value.\n"
"\n"
" ## Examples\n"
"\n"
" ```gleam\n"
" taskle.start(fn() {\n"
" // Log something or perform side effects\n"
" io.println(\"Background task completed\")\n"
" cleanup_temp_files()\n"
" })\n"
" ```\n"
).
-spec start(fun(() -> FHG)) -> task(FHG).
start(Fun) ->
Owner = erlang:self(),
Subject = gleam@erlang@process:new_subject(),
Pid = proc_lib:spawn(
fun() ->
_ = Fun(),
nil
end
),
Monitor = gleam@erlang@process:monitor(Pid),
{task, Pid, Monitor, Owner, Subject}.
-file("src/taskle.gleam", 347).
?DOC(
" Creates an asynchronous task that is not linked to the calling process.\n"
"\n"
" This is identical to `async`, but provided for clarity when you specifically\n"
" want to emphasize that the task is unlinked. The task DOES send its result back\n"
" and can be awaited, unlike tasks created with `start`.\n"
"\n"
" **Key difference from `start`:** Tasks created with `async_unlinked` send their\n"
" result back and can be awaited. Use this when you need the result but want to\n"
" emphasize the unlinked nature, or when you might await the task later.\n"
"\n"
" ## Examples\n"
"\n"
" ```gleam\n"
" let task = taskle.async_unlinked(fn() {\n"
" // This task won't affect the parent process if it crashes\n"
" expensive_computation()\n"
" })\n"
"\n"
" // You can still await the result\n"
" case taskle.await(task, 5000) {\n"
" Ok(result) -> use_result(result)\n"
" Error(_) -> handle_error()\n"
" }\n"
" ```\n"
).
-spec async_unlinked(fun(() -> FHI)) -> task(FHI).
async_unlinked(Fun) ->
Owner = erlang:self(),
Subject = gleam@erlang@process:new_subject(),
Pid = proc_lib:spawn(
fun() ->
Result = Fun(),
gleam@erlang@process:send(Subject, {task_result, Result})
end
),
Monitor = gleam@erlang@process:monitor(Pid),
{task, Pid, Monitor, Owner, Subject}.
-file("src/taskle.gleam", 436).
?DOC(
" Returns the process ID of the task's underlying BEAM process.\n"
"\n"
" Useful for debugging or process monitoring.\n"
"\n"
" ## Examples\n"
"\n"
" ```gleam\n"
" let task = taskle.async(fn() { 42 })\n"
" let process_id = taskle.pid(task)\n"
" io.debug(process_id)\n"
" ```\n"
).
-spec pid(task(any())) -> gleam@erlang@process:pid_().
pid(Task) ->
{task, Pid, _, _, _} = Task,
Pid.
-file("src/taskle.gleam", 879).
-spec down_to_error(gleam@erlang@process:down()) -> error().
down_to_error(Down) ->
case Down of
{process_down, _, _, Reason} ->
case Reason of
normal ->
{crashed, <<"normal"/utf8>>};
killed ->
{crashed, <<"killed"/utf8>>};
{abnormal, Reason@1} ->
{crashed, gleam@string:inspect(Reason@1)}
end;
{port_down, _, _, _} ->
{crashed, <<"port_down"/utf8>>}
end.
-file("src/taskle.gleam", 507).
-spec build_race_selector(
list(task(FIE)),
gleam@erlang@process:selector(race_message(FIE))
) -> gleam@erlang@process:selector(race_message(FIE)).
build_race_selector(Tasks, Selector) ->
gleam@list:fold(
Tasks,
Selector,
fun(Acc_selector, Task) ->
{task, Task_pid, Monitor, _, Subject} = Task,
_pipe = Acc_selector,
_pipe@1 = gleam@erlang@process:select_map(
_pipe,
Subject,
fun(Msg) -> case Msg of
{task_result, Value} ->
gleam@erlang@process:demonitor_process(Monitor),
{race_result, {ok, Value}, Task_pid}
end end
),
gleam@erlang@process:select_specific_monitor(
_pipe@1,
Monitor,
fun(Down) ->
{race_result, {error, down_to_error(Down)}, Task_pid}
end
)
end
).
-file("src/taskle.gleam", 620).
-spec build_all_settled_selector(
list({integer(), task(FJI)}),
gleam@erlang@process:selector(all_settled_message(FJI))
) -> gleam@erlang@process:selector(all_settled_message(FJI)).
build_all_settled_selector(Indexed_tasks, Selector) ->
gleam@list:fold(
Indexed_tasks,
Selector,
fun(Acc_selector, Indexed_task) ->
{Index, Task} = Indexed_task,
{task, _, Monitor, _, Subject} = Task,
_pipe = Acc_selector,
_pipe@1 = gleam@erlang@process:select_map(
_pipe,
Subject,
fun(Msg) -> case Msg of
{task_result, Value} ->
gleam@erlang@process:demonitor_process(Monitor),
{all_settled_result, Index, {fulfilled, Value}}
end end
),
gleam@erlang@process:select_specific_monitor(
_pipe@1,
Monitor,
fun(Down) ->
{all_settled_result, Index, {rejected, down_to_error(Down)}}
end
)
end
).
-file("src/taskle.gleam", 892).
-spec remaining_timeout_ms(integer(), integer()) -> integer().
remaining_timeout_ms(Start_time, Total_timeout) ->
Elapsed = erlang:system_time() - Start_time,
Total_timeout - (Elapsed div 1000000).
-file("src/taskle.gleam", 581).
-spec all_settled_collect_loop(
gleam@erlang@process:selector(all_settled_message(FIZ)),
integer(),
list({integer(), settled_result(FIZ)}),
integer(),
integer()
) -> {ok, list(settled_result(FIZ))} | {error, error()}.
all_settled_collect_loop(
Selector,
Remaining_count,
Results,
Timeout,
Start_time
) ->
case Remaining_count of
0 ->
_pipe = Results,
_pipe@1 = gleam@list:sort(
_pipe,
fun(A, B) ->
gleam@int:compare(
erlang:element(1, A),
erlang:element(1, B)
)
end
),
_pipe@2 = gleam@list:map(
_pipe@1,
fun(Pair) -> erlang:element(2, Pair) end
),
{ok, _pipe@2};
_ ->
Remaining = remaining_timeout_ms(Start_time, Timeout),
gleam@bool:guard(
Remaining =< 0,
{error, timeout},
fun() -> case gleam_erlang_ffi:select(Selector, Remaining) of
{ok, {all_settled_result, Index, Result}} ->
all_settled_collect_loop(
Selector,
Remaining_count - 1,
[{Index, Result} | Results],
Timeout,
Start_time
);
{error, nil} ->
{error, timeout}
end end
)
end.
-file("src/taskle.gleam", 897).
-spec is_timeout_exceeded(integer(), integer()) -> boolean().
is_timeout_exceeded(Start_time, Total_timeout) ->
remaining_timeout_ms(Start_time, Total_timeout) =< 0.
-file("src/taskle.gleam", 174).
-spec shutdown_wait_loop(gleam@erlang@process:pid_(), integer(), integer()) -> {ok,
nil} |
{error, error()}.
shutdown_wait_loop(Pid, Timeout, Start_time) ->
gleam@bool:guard(
is_timeout_exceeded(Start_time, Timeout),
{error, timeout},
fun() -> case erlang:is_process_alive(Pid) of
false ->
{ok, nil};
true ->
gleam_erlang_ffi:sleep(10),
shutdown_wait_loop(Pid, Timeout, Start_time)
end end
).
-file("src/taskle.gleam", 901).
-spec build_task_selector(
task(FKZ),
fun((task_message(FKZ)) -> FLC),
fun((gleam@erlang@process:down()) -> FLC)
) -> gleam@erlang@process:selector(FLC).
build_task_selector(Task, Result_mapper, Down_mapper) ->
{task, _, Monitor, _, Subject} = Task,
_pipe = gleam_erlang_ffi:new_selector(),
_pipe@1 = gleam@erlang@process:select_map(_pipe, Subject, Result_mapper),
gleam@erlang@process:select_specific_monitor(_pipe@1, Monitor, Down_mapper).
-file("src/taskle.gleam", 912).
-spec validate_ownership(task(any())) -> {ok, nil} | {error, error()}.
validate_ownership(Task) ->
{task, _, _, Owner, _} = Task,
gleam@bool:guard(
erlang:self() /= Owner,
{error, not_owner},
fun() -> {ok, nil} end
).
-file("src/taskle.gleam", 88).
?DOC(
" Waits for a task to complete with a timeout in milliseconds.\n"
"\n"
" Only the process that created the task can await its result. If called from\n"
" a different process, returns `Error(NotOwner)`.\n"
"\n"
" ## Examples\n"
"\n"
" ```gleam\n"
" case taskle.await(task, 5000) {\n"
" Ok(value) -> io.println(\"Success: \" <> int.to_string(value))\n"
" Error(taskle.Timeout) -> io.println(\"Task timed out after 5 seconds\")\n"
" Error(taskle.Crashed(reason)) -> io.println(\"Task failed: \" <> reason)\n"
" Error(taskle.NotOwner) -> io.println(\"Cannot await task from different process\")\n"
" }\n"
" ```\n"
).
-spec await(task(FFT), integer()) -> {ok, FFT} | {error, error()}.
await(Task, Timeout) ->
{task, Task_pid, Monitor, _, _} = Task,
gleam@result:'try'(
validate_ownership(Task),
fun(_) ->
Selector = build_task_selector(
Task,
fun(Field@0) -> {task_msg, Field@0} end,
fun(Field@0) -> {down_msg, Field@0} end
),
case gleam_erlang_ffi:select(Selector, Timeout) of
{ok, {task_msg, {task_result, Value}}} ->
gleam@erlang@process:demonitor_process(Monitor),
{ok, Value};
{ok, {down_msg, Down}} ->
{error, down_to_error(Down)};
{error, nil} ->
gleam@erlang@process:kill(Task_pid),
gleam@erlang@process:demonitor_process(Monitor),
{error, timeout}
end
end
).
-file("src/taskle.gleam", 128).
?DOC(
" Cancels a running task.\n"
"\n"
" Only the process that created the task can cancel it. If called from\n"
" a different process, returns `Error(NotOwner)`.\n"
"\n"
" ## Examples\n"
"\n"
" ```gleam\n"
" let task = taskle.async(fn() {\n"
" process.sleep(10_000)\n"
" \"done\"\n"
" })\n"
"\n"
" // Cancel the task\n"
" case taskle.cancel(task) {\n"
" Ok(Nil) -> io.println(\"Task cancelled\")\n"
" Error(taskle.NotOwner) -> io.println(\"Cannot cancel task from different process\")\n"
" }\n"
" ```\n"
).
-spec cancel(task(any())) -> {ok, nil} | {error, error()}.
cancel(Task) ->
{task, Pid, Monitor, _, _} = Task,
gleam@result:'try'(
validate_ownership(Task),
fun(_) ->
gleam@erlang@process:kill(Pid),
gleam@erlang@process:demonitor_process(Monitor),
{ok, nil}
end
).
-file("src/taskle.gleam", 162).
?DOC(
" Shuts down a task gracefully with a timeout.\n"
"\n"
" This function attempts to shut down a task more gracefully than `cancel`.\n"
" It first removes monitoring of the process, then kills it. If the task doesn't shut down\n"
" within the timeout, it returns an error.\n"
"\n"
" Only the process that created the task can shut it down. If called from\n"
" a different process, returns `Error(NotOwner)`.\n"
"\n"
" ## Examples\n"
"\n"
" ```gleam\n"
" let task = taskle.async(fn() {\n"
" process.sleep(10_000)\n"
" \"done\"\n"
" })\n"
"\n"
" // Shutdown the task with a 1 second timeout\n"
" case taskle.shutdown(task, 1000) {\n"
" Ok(Nil) -> io.println(\"Task shut down gracefully\")\n"
" Error(taskle.Timeout) -> io.println(\"Task didn't shut down in time\")\n"
" Error(taskle.NotOwner) -> io.println(\"Cannot shutdown task from different process\")\n"
" }\n"
" ```\n"
).
-spec shutdown(task(any()), integer()) -> {ok, nil} | {error, error()}.
shutdown(Task, Timeout) ->
{task, Pid, Monitor, _, _} = Task,
gleam@result:'try'(
validate_ownership(Task),
fun(_) ->
gleam@erlang@process:demonitor_process(Monitor),
gleam@erlang@process:kill(Pid),
Start_time = erlang:system_time(),
shutdown_wait_loop(Pid, Timeout, Start_time)
end
).
-file("src/taskle.gleam", 256).
-spec await_all_loop(list(task(FGZ)), list(FGZ), integer(), integer()) -> {ok,
list(FGZ)} |
{error, error()}.
await_all_loop(Tasks, Results, Timeout, Start_time) ->
case Tasks of
[] ->
{ok, lists:reverse(Results)};
[Task | Rest] ->
Remaining = remaining_timeout_ms(Start_time, Timeout),
gleam@bool:lazy_guard(
Remaining =< 0,
fun() ->
gleam@list:each(Rest, fun(T) -> _ = cancel(T) end),
{error, timeout}
end,
fun() -> case await(Task, Remaining) of
{ok, Result} ->
await_all_loop(
Rest,
[Result | Results],
Timeout,
Start_time
);
{error, Err} ->
gleam@list:each(
Rest,
fun(T@1) -> _ = cancel(T@1) end
),
{error, Err}
end end
)
end.
-file("src/taskle.gleam", 246).
-spec await_all(list(task(FGT)), integer()) -> {ok, list(FGT)} |
{error, error()}.
await_all(Tasks, Timeout) ->
case Tasks of
[] ->
{ok, []};
_ ->
Start_time = erlang:system_time(),
await_all_loop(Tasks, [], Timeout, Start_time)
end.
-file("src/taskle.gleam", 209).
?DOC(
" Processes a list of items in parallel, applying the given function to each item.\n"
"\n"
" Returns when all tasks complete or when any task fails/times out. If any task\n"
" fails, all remaining tasks are cancelled.\n"
"\n"
" ## Examples\n"
"\n"
" ```gleam\n"
" let numbers = [1, 2, 3, 4, 5]\n"
"\n"
" case taskle.parallel_map(numbers, fn(x) { x * x }, 5000) {\n"
" Ok(results) -> {\n"
" // results = [1, 4, 9, 16, 25]\n"
" io.debug(results)\n"
" }\n"
" Error(taskle.Timeout) -> io.println(\"Some tasks timed out\")\n"
" Error(taskle.Crashed(reason)) -> io.println(\"A task crashed: \" <> reason)\n"
" }\n"
" ```\n"
).
-spec parallel_map(list(FGH), fun((FGH) -> FGJ), integer()) -> {ok, list(FGJ)} |
{error, error()}.
parallel_map(List, Fun, Timeout) ->
Tasks = gleam@list:map(List, fun(Item) -> async(fun() -> Fun(Item) end) end),
await_all(Tasks, Timeout).
-file("src/taskle.gleam", 239).
?DOC(
" Waits for all tasks to complete with a timeout.\n"
"\n"
" Returns when all tasks complete or when any task fails/times out. If any task fails, all remaining tasks are cancelled.\n"
"\n"
" ## Examples\n"
"\n"
" ```gleam\n"
" let task1 = taskle.async(fn() { 42 })\n"
" let task2 = taskle.async(fn() { \"hello\" })\n"
" let task3 = taskle.async(fn() { True })\n"
"\n"
" case taskle.try_await_all([task1, task2, task3], 5000) {\n"
" Ok([a, b, c]) -> {\n"
" // a = 42, b = \"hello\", c = True\n"
" io.debug([a, b, c])\n"
" }\n"
" Error(taskle.Timeout) -> io.println(\"Some tasks timed out\")\n"
" Error(taskle.Crashed(reason)) -> io.println(\"A task crashed: \" <> reason)\n"
" }\n"
" ```\n"
).
-spec try_await_all(list(task(FGN)), integer()) -> {ok, list(FGN)} |
{error, error()}.
try_await_all(Tasks, Timeout) ->
await_all(Tasks, Timeout).
-file("src/taskle.gleam", 377).
?DOC(
" Waits for a task to complete without a timeout.\n"
"\n"
" Only the process that created the task can await its result. If called from\n"
" a different process, returns `Error(NotOwner)`. Will wait indefinitely until\n"
" the task completes or crashes.\n"
"\n"
" ## Examples\n"
"\n"
" ```gleam\n"
" case taskle.await_forever(task) {\n"
" Ok(value) -> io.println(\"Success: \" <> int.to_string(value))\n"
" Error(taskle.Crashed(reason)) -> io.println(\"Task failed: \" <> reason)\n"
" Error(taskle.NotOwner) -> io.println(\"Cannot await task from different process\")\n"
" }\n"
" ```\n"
).
-spec await_forever(task(FHK)) -> {ok, FHK} | {error, error()}.
await_forever(Task) ->
{task, _, Monitor, _, _} = Task,
gleam@result:'try'(
validate_ownership(Task),
fun(_) ->
Selector = build_task_selector(
Task,
fun(Field@0) -> {task_msg, Field@0} end,
fun(Field@0) -> {down_msg, Field@0} end
),
case gleam_erlang_ffi:select(Selector) of
{task_msg, {task_result, Value}} ->
gleam@erlang@process:demonitor_process(Monitor),
{ok, Value};
{down_msg, Down} ->
{error, down_to_error(Down)}
end
end
).
-file("src/taskle.gleam", 408).
?DOC(
" Checks if a task has completed without blocking.\n"
"\n"
" Returns `Ok(value)` if the task has completed, `Error(NotReady)` if it's still\n"
" running, or other errors if the task crashed or ownership check fails.\n"
"\n"
" ## Examples\n"
"\n"
" ```gleam\n"
" case taskle.yield(task) {\n"
" Ok(value) -> io.println(\"Task completed: \" <> int.to_string(value))\n"
" Error(taskle.NotReady) -> io.println(\"Task still running\")\n"
" Error(taskle.Crashed(reason)) -> io.println(\"Task failed: \" <> reason)\n"
" Error(taskle.NotOwner) -> io.println(\"Cannot check task from different process\")\n"
" }\n"
" ```\n"
).
-spec yield(task(FHO)) -> {ok, FHO} | {error, error()}.
yield(Task) ->
{task, _, Monitor, _, _} = Task,
gleam@result:'try'(
validate_ownership(Task),
fun(_) ->
Selector = build_task_selector(
Task,
fun(Field@0) -> {task_msg, Field@0} end,
fun(Field@0) -> {down_msg, Field@0} end
),
case gleam_erlang_ffi:select(Selector, 0) of
{ok, {task_msg, {task_result, Value}}} ->
gleam@erlang@process:demonitor_process(Monitor),
{ok, Value};
{ok, {down_msg, Down}} ->
{error, down_to_error(Down)};
{error, nil} ->
{error, not_ready}
end
end
).
-file("src/taskle.gleam", 663).
?DOC(
" Waits for two tasks to complete and returns a tuple of their results.\n"
"\n"
" Both tasks must complete successfully within the timeout, otherwise an error is returned.\n"
" If any task fails, all remaining tasks are cancelled.\n"
"\n"
" ## Examples\n"
"\n"
" ```gleam\n"
" let task1 = taskle.async(fn() { 42 })\n"
" let task2 = taskle.async(fn() { \"hello\" })\n"
"\n"
" case taskle.await2(task1, task2, 5000) {\n"
" Ok(#(num, str)) -> {\n"
" // num = 42, str = \"hello\"\n"
" io.debug(#(num, str))\n"
" }\n"
" Error(taskle.Timeout) -> io.println(\"Tasks timed out\")\n"
" Error(taskle.Crashed(reason)) -> io.println(\"A task crashed: \" <> reason)\n"
" }\n"
" ```\n"
).
-spec await2(task(FJP), task(FJR), integer()) -> {ok, {FJP, FJR}} |
{error, error()}.
await2(Task1, Task2, Timeout) ->
Start_time = erlang:system_time(),
gleam@result:'try'(
begin
_pipe = await(Task1, Timeout),
gleam@result:map_error(
_pipe,
fun(Err) ->
_ = cancel(Task2),
Err
end
)
end,
fun(Result1) ->
Remaining = remaining_timeout_ms(Start_time, Timeout),
gleam@result:'try'(
await(Task2, Remaining),
fun(Result2) -> {ok, {Result1, Result2}} end
)
end
).
-file("src/taskle.gleam", 704).
?DOC(
" Waits for three tasks to complete and returns a tuple of their results.\n"
"\n"
" All tasks must complete successfully within the timeout, otherwise an error is returned.\n"
" If any task fails, all remaining tasks are cancelled.\n"
"\n"
" ## Examples\n"
"\n"
" ```gleam\n"
" let task1 = taskle.async(fn() { 42 })\n"
" let task2 = taskle.async(fn() { \"hello\" })\n"
" let task3 = taskle.async(fn() { True })\n"
"\n"
" case taskle.await3(task1, task2, task3, 5000) {\n"
" Ok(#(num, str, bool)) -> {\n"
" // num = 42, str = \"hello\", bool = True\n"
" io.debug(#(num, str, bool))\n"
" }\n"
" Error(taskle.Timeout) -> io.println(\"Tasks timed out\")\n"
" Error(taskle.Crashed(reason)) -> io.println(\"A task crashed: \" <> reason)\n"
" }\n"
" ```\n"
).
-spec await3(task(FJV), task(FJX), task(FJZ), integer()) -> {ok,
{FJV, FJX, FJZ}} |
{error, error()}.
await3(Task1, Task2, Task3, Timeout) ->
Start_time = erlang:system_time(),
gleam@result:'try'(
begin
_pipe = await(Task1, Timeout),
gleam@result:map_error(
_pipe,
fun(Err) ->
_ = cancel(Task2),
_ = cancel(Task3),
Err
end
)
end,
fun(Result1) ->
Remaining = remaining_timeout_ms(Start_time, Timeout),
gleam@result:'try'(
begin
_pipe@1 = await(Task2, Remaining),
gleam@result:map_error(
_pipe@1,
fun(Err@1) ->
_ = cancel(Task3),
Err@1
end
)
end,
fun(Result2) ->
Remaining2 = remaining_timeout_ms(Start_time, Timeout),
gleam@result:'try'(
await(Task3, Remaining2),
fun(Result3) -> {ok, {Result1, Result2, Result3}} end
)
end
)
end
).
-file("src/taskle.gleam", 756).
?DOC(
" Waits for four tasks to complete and returns a tuple of their results.\n"
"\n"
" All tasks must complete successfully within the timeout, otherwise an error is returned.\n"
" If any task fails, all remaining tasks are cancelled.\n"
"\n"
" ## Examples\n"
"\n"
" ```gleam\n"
" let task1 = taskle.async(fn() { 42 })\n"
" let task2 = taskle.async(fn() { \"hello\" })\n"
" let task3 = taskle.async(fn() { True })\n"
" let task4 = taskle.async(fn() { 3.14 })\n"
"\n"
" case taskle.await4(task1, task2, task3, task4, 5000) {\n"
" Ok(#(num, str, bool, float)) -> {\n"
" // num = 42, str = \"hello\", bool = True, float = 3.14\n"
" io.debug(#(num, str, bool, float))\n"
" }\n"
" Error(taskle.Timeout) -> io.println(\"Tasks timed out\")\n"
" Error(taskle.Crashed(reason)) -> io.println(\"A task crashed: \" <> reason)\n"
" }\n"
" ```\n"
).
-spec await4(task(FKD), task(FKF), task(FKH), task(FKJ), integer()) -> {ok,
{FKD, FKF, FKH, FKJ}} |
{error, error()}.
await4(Task1, Task2, Task3, Task4, Timeout) ->
Start_time = erlang:system_time(),
gleam@result:'try'(
begin
_pipe = await(Task1, Timeout),
gleam@result:map_error(
_pipe,
fun(Err) ->
_ = cancel(Task2),
_ = cancel(Task3),
_ = cancel(Task4),
Err
end
)
end,
fun(Result1) ->
Remaining = remaining_timeout_ms(Start_time, Timeout),
gleam@result:'try'(
begin
_pipe@1 = await(Task2, Remaining),
gleam@result:map_error(
_pipe@1,
fun(Err@1) ->
_ = cancel(Task3),
_ = cancel(Task4),
Err@1
end
)
end,
fun(Result2) ->
Remaining2 = remaining_timeout_ms(Start_time, Timeout),
gleam@result:'try'(
begin
_pipe@2 = await(Task3, Remaining2),
gleam@result:map_error(
_pipe@2,
fun(Err@2) ->
_ = cancel(Task4),
Err@2
end
)
end,
fun(Result3) ->
Remaining3 = remaining_timeout_ms(
Start_time,
Timeout
),
gleam@result:'try'(
await(Task4, Remaining3),
fun(Result4) ->
{ok, {Result1, Result2, Result3, Result4}}
end
)
end
)
end
)
end
).
-file("src/taskle.gleam", 821).
?DOC(
" Waits for five tasks to complete and returns a tuple of their results.\n"
"\n"
" All tasks must complete successfully within the timeout, otherwise an error is returned.\n"
" If any task fails, all remaining tasks are cancelled.\n"
"\n"
" ## Examples\n"
"\n"
" ```gleam\n"
" let task1 = taskle.async(fn() { 42 })\n"
" let task2 = taskle.async(fn() { \"hello\" })\n"
" let task3 = taskle.async(fn() { True })\n"
" let task4 = taskle.async(fn() { 3.14 })\n"
" let task5 = taskle.async(fn() { [1, 2, 3] })\n"
"\n"
" case taskle.await5(task1, task2, task3, task4, task5, 5000) {\n"
" Ok(#(num, str, bool, float, list)) -> {\n"
" // num = 42, str = \"hello\", bool = True, float = 3.14, list = [1, 2, 3]\n"
" io.debug(#(num, str, bool, float, list))\n"
" }\n"
" Error(taskle.Timeout) -> io.println(\"Tasks timed out\")\n"
" Error(taskle.Crashed(reason)) -> io.println(\"A task crashed: \" <> reason)\n"
" }\n"
" ```\n"
).
-spec await5(task(FKN), task(FKP), task(FKR), task(FKT), task(FKV), integer()) -> {ok,
{FKN, FKP, FKR, FKT, FKV}} |
{error, error()}.
await5(Task1, Task2, Task3, Task4, Task5, Timeout) ->
Start_time = erlang:system_time(),
gleam@result:'try'(
begin
_pipe = await(Task1, Timeout),
gleam@result:map_error(
_pipe,
fun(Err) ->
_ = cancel(Task2),
_ = cancel(Task3),
_ = cancel(Task4),
_ = cancel(Task5),
Err
end
)
end,
fun(Result1) ->
Remaining = remaining_timeout_ms(Start_time, Timeout),
gleam@result:'try'(
begin
_pipe@1 = await(Task2, Remaining),
gleam@result:map_error(
_pipe@1,
fun(Err@1) ->
_ = cancel(Task3),
_ = cancel(Task4),
_ = cancel(Task5),
Err@1
end
)
end,
fun(Result2) ->
Remaining2 = remaining_timeout_ms(Start_time, Timeout),
gleam@result:'try'(
begin
_pipe@2 = await(Task3, Remaining2),
gleam@result:map_error(
_pipe@2,
fun(Err@2) ->
_ = cancel(Task4),
_ = cancel(Task5),
Err@2
end
)
end,
fun(Result3) ->
Remaining3 = remaining_timeout_ms(
Start_time,
Timeout
),
gleam@result:'try'(
begin
_pipe@3 = await(Task4, Remaining3),
gleam@result:map_error(
_pipe@3,
fun(Err@3) ->
_ = cancel(Task5),
Err@3
end
)
end,
fun(Result4) ->
Remaining4 = remaining_timeout_ms(
Start_time,
Timeout
),
gleam@result:'try'(
await(Task5, Remaining4),
fun(Result5) ->
{ok,
{Result1,
Result2,
Result3,
Result4,
Result5}}
end
)
end
)
end
)
end
)
end
).
-file("src/taskle.gleam", 918).
-spec validate_multiple_ownership(list(task(any()))) -> {ok, nil} |
{error, error()}.
validate_multiple_ownership(Tasks) ->
Current_pid = erlang:self(),
gleam@bool:guard(
not gleam@list:all(
Tasks,
fun(Task) ->
{task, _, _, Owner, _} = Task,
Owner =:= Current_pid
end
),
{error, not_owner},
fun() -> {ok, nil} end
).
-file("src/taskle.gleam", 472).
-spec race_with_selector(list(task(FHZ)), integer()) -> {ok, FHZ} |
{error, error()}.
race_with_selector(Tasks, Timeout) ->
case validate_multiple_ownership(Tasks) of
{error, Err} ->
gleam@list:each(Tasks, fun(T) -> _ = cancel(T) end),
{error, Err};
{ok, nil} ->
Selector = build_race_selector(
Tasks,
gleam_erlang_ffi:new_selector()
),
case gleam_erlang_ffi:select(Selector, Timeout) of
{ok, {race_result, Result, Winning_task_pid}} ->
_pipe = Tasks,
_pipe@1 = gleam@list:filter(
_pipe,
fun(Task) -> pid(Task) /= Winning_task_pid end
),
gleam@list:each(
_pipe@1,
fun(Task@1) -> _ = cancel(Task@1) end
),
Result;
{error, nil} ->
gleam@list:each(Tasks, fun(T@1) -> _ = cancel(T@1) end),
{error, timeout}
end
end.
-file("src/taskle.gleam", 464).
?DOC(
" Waits for the first task to complete (analog of Promise.race).\n"
"\n"
" Returns the result of the first task to complete, whether successful or failed.\n"
" All other tasks are cancelled when the first one completes.\n"
"\n"
" ## Examples\n"
"\n"
" ```gleam\n"
" let task1 = taskle.async(fn() {\n"
" process.sleep(1000)\n"
" \"slow\"\n"
" })\n"
" let task2 = taskle.async(fn() {\n"
" process.sleep(100)\n"
" \"fast\"\n"
" })\n"
"\n"
" case taskle.race([task1, task2], 5000) {\n"
" Ok(\"fast\") -> io.println(\"Task2 won the race\")\n"
" Error(taskle.Timeout) -> io.println(\"All tasks timed out\")\n"
" Error(taskle.Crashed(reason)) -> io.println(\"First task to complete crashed: \" <> reason)\n"
" }\n"
" ```\n"
).
-spec race(list(task(FHU)), integer()) -> {ok, FHU} | {error, error()}.
race(Tasks, Timeout) ->
case Tasks of
[] ->
{error, timeout};
[Single_task] ->
await(Single_task, Timeout);
_ ->
race_with_selector(Tasks, Timeout)
end.
-file("src/taskle.gleam", 560).
-spec all_settled_concurrent(list(task(FIS)), integer()) -> {ok,
list(settled_result(FIS))} |
{error, error()}.
all_settled_concurrent(Tasks, Timeout) ->
gleam@result:'try'(
validate_multiple_ownership(Tasks),
fun(_use0) ->
nil = _use0,
Indexed_tasks = gleam@list:index_map(
Tasks,
fun(Task, Index) -> {Index, Task} end
),
Task_count = erlang:length(Tasks),
Selector = build_all_settled_selector(
Indexed_tasks,
gleam_erlang_ffi:new_selector()
),
all_settled_collect_loop(
Selector,
Task_count,
[],
Timeout,
erlang:system_time()
)
end
).
-file("src/taskle.gleam", 550).
?DOC(
" Waits for all tasks to complete regardless of success or failure (analog of Promise.allSettled).\n"
"\n"
" Returns the results of all tasks, whether they succeeded or failed. Unlike `try_await_all`,\n"
" this function never cancels tasks early - it waits for all tasks to complete.\n"
"\n"
" ## Examples\n"
"\n"
" ```gleam\n"
" let task1 = taskle.async(fn() { 42 })\n"
" let task2 = taskle.async(fn() {\n"
" // This will crash\n"
" panic as \"oops\"\n"
" })\n"
" let task3 = taskle.async(fn() { \"hello\" })\n"
"\n"
" case taskle.all_settled([task1, task2, task3], 5000) {\n"
" Ok([Fulfilled(42), Rejected(Crashed(\"oops\")), Fulfilled(\"hello\")]) -> {\n"
" io.println(\"All tasks completed\")\n"
" }\n"
" Error(taskle.Timeout) -> io.println(\"Some tasks timed out\")\n"
" }\n"
" ```\n"
).
-spec all_settled(list(task(FIL)), integer()) -> {ok, list(settled_result(FIL))} |
{error, error()}.
all_settled(Tasks, Timeout) ->
case Tasks of
[] ->
{ok, []};
_ ->
all_settled_concurrent(Tasks, Timeout)
end.