-module(taskle). -compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch]). -define(FILEPATH, "src/taskle.gleam"). -export([async/1, await/2, cancel/1, async_unlinked/1, pid/1, parallel_map/3]). -export_type([task/1, error/0, task_message/1, await_msg/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. -type task_message(FFM) :: {task_result, FFM}. -type await_msg(FFN) :: {task_msg, task_message(FFN)} | {down_msg, gleam@erlang@process:down()}. -file("src/taskle.gleam", 46). ?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(() -> FFO)) -> task(FFO). 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", 76). ?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(FFQ), integer()) -> {ok, FFQ} | {error, error()}. await(Task, Timeout) -> {task, Task_pid, Monitor, Owner, Subject} = Task, gleam@bool:guard( erlang:self() /= Owner, {error, not_owner}, fun() -> Selector = begin _pipe = gleam_erlang_ffi:new_selector(), _pipe@1 = gleam@erlang@process:select_specific_monitor( _pipe, Monitor, fun(Field@0) -> {down_msg, Field@0} end ), gleam@erlang@process:select_map( _pipe@1, Subject, fun(Field@0) -> {task_msg, Field@0} end ) 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, {process_down, _, _, Reason}}} -> case Reason of normal -> {error, {crashed, <<"normal"/utf8>>}}; killed -> {error, {crashed, <<"killed"/utf8>>}}; {abnormal, Reason@1} -> {error, {crashed, gleam@string:inspect(Reason@1)}} end; {ok, {down_msg, {port_down, _, _, _}}} -> {error, {crashed, <<"port_down"/utf8>>}}; {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, Owner, _} = Task, gleam@bool:guard( erlang:self() /= Owner, {error, not_owner}, fun() -> gleam@erlang@process:kill(Pid), gleam@erlang@process:demonitor_process(Monitor), {ok, nil} end ). -file("src/taskle.gleam", 220). ?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. Use this for fire-and-forget\n" " scenarios where you don't want the parent process to be affected if the task crashes.\n" "\n" " ## Examples\n" "\n" " ```gleam\n" " let fire_and_forget = taskle.async_unlinked(fn() {\n" " // This task won't affect the parent process if it crashes\n" " risky_operation()\n" " })\n" " ```\n" ). -spec async_unlinked(fun(() -> FGR)) -> task(FGR). 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", 246). ?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", 177). -spec await_all_loop(list(task(FGK)), list(FGK), integer(), integer()) -> {ok, list(FGK)} | {error, error()}. await_all_loop(Tasks, Results, Timeout, Start_time) -> case Tasks of [] -> {ok, lists:reverse(Results)}; [Task | Rest] -> Elapsed = erlang:system_time() - Start_time, Remaining_timeout = Timeout - (Elapsed div 1000000), gleam@bool:lazy_guard( Remaining_timeout =< 0, fun() -> gleam@list:each(Rest, fun cancel/1), {error, timeout} end, fun() -> case await(Task, Remaining_timeout) of {ok, Result} -> await_all_loop( Rest, [Result | Results], Timeout, Start_time ); {error, Err} -> gleam@list:each(Rest, fun cancel/1), {error, Err} end end ) end. -file("src/taskle.gleam", 167). -spec await_all(list(task(FGE)), integer()) -> {ok, list(FGE)} | {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", 157). ?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(FFY), fun((FFY) -> FGA), integer()) -> {ok, list(FGA)} | {error, error()}. parallel_map(List, Fun, Timeout) -> Tasks = gleam@list:map(List, fun(Item) -> async(fun() -> Fun(Item) end) end), await_all(Tasks, Timeout).