-module(eensy@otp@task). -compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch]). -export([async/1, try_await/2, await/2]). -export_type([task/1, await_error/0, 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. ?MODULEDOC( " A task is a kind of process that performs a single task and then shuts\n" " down. Commonly tasks are used to convert sequential code into concurrent\n" " code by performing computation in another process.\n" "\n" " ```gleam\n" " let task = task.async(fn() { do_some_work() })\n" " let value = do_some_other_work()\n" " value + task.await(task, 100)\n" " ```\n" "\n" " Tasks spawned with async can be awaited on by their caller process (and\n" " only their caller) as shown in the example above. They are implemented by\n" " spawning a process that sends a message to the caller once the given\n" " computation is performed.\n" "\n" " There are two important things to consider when using `async`:\n" "\n" " 1. If you are using async tasks, you must await a reply as they are always\n" " sent.\n" "\n" " 2. async tasks link the caller and the spawned process. This means that,\n" " if the caller crashes, the task will crash too and vice-versa. This is\n" " on purpose: if the process meant to receive the result no longer\n" " exists, there is no purpose in completing the computation.\n" "\n" " This module is inspired by Elixir's [Task module][1].\n" "\n" " [1]: https://hexdocs.pm/elixir/master/Task.html\n" "\n" ). -opaque task(HCE) :: {task, gleam@erlang@process:pid_(), gleam@erlang@process:pid_(), gleam@erlang@process:process_monitor(), gleam@erlang@process:selector(message(HCE))}. -type await_error() :: timeout | {exit, gleam@dynamic:dynamic_()}. -type message(HCF) :: {from_monitor, gleam@erlang@process:process_down()} | {from_subject, HCF}. -file("src/eensy/otp/task.gleam", 51). ?DOC( " Spawn a task process that calls a given function in order to perform some\n" " work. The result of this function is send back to the parent and can be\n" " received using the `await` function.\n" "\n" " See the top level module documentation for more information on async/await.\n" ). -spec async(fun(() -> HCG)) -> task(HCG). async(Work) -> Owner = erlang:self(), Subject = gleam@erlang@process:new_subject(), Pid = gleam@erlang@process:start( fun() -> gleam@erlang@process:send(Subject, Work()) end, true ), Monitor = gleam@erlang@process:monitor_process(Pid), Selector = begin _pipe = gleam_erlang_ffi:new_selector(), _pipe@1 = gleam@erlang@process:selecting_process_down( _pipe, Monitor, fun(Field@0) -> {from_monitor, Field@0} end ), gleam@erlang@process:selecting( _pipe@1, Subject, fun(Field@0) -> {from_subject, Field@0} end ) end, {task, Owner, Pid, Monitor, Selector}. -file("src/eensy/otp/task.gleam", 71). -spec assert_owner(task(any())) -> nil. assert_owner(Task) -> Self = erlang:self(), case erlang:element(2, Task) =:= Self of true -> nil; false -> gleam@erlang@process:send_abnormal_exit( Self, <<"awaited on a task that does not belong to this process"/utf8>> ) end. -file("src/eensy/otp/task.gleam", 94). ?DOC( " Wait for the value computed by a task.\n" "\n" " If the a value is not received before the timeout has elapsed or if the\n" " task process crashes then an error is returned.\n" ). -spec try_await(task(HCK), integer()) -> {ok, HCK} | {error, await_error()}. try_await(Task, Timeout) -> assert_owner(Task), case gleam_erlang_ffi:select(erlang:element(5, Task), Timeout) of {ok, {from_subject, X}} -> gleam_erlang_ffi:demonitor(erlang:element(4, Task)), {ok, X}; {ok, {from_monitor, {process_down, _, Reason}}} -> {error, {exit, Reason}}; {error, nil} -> {error, timeout} end. -file("src/eensy/otp/task.gleam", 118). ?DOC( " Wait for the value computed by a task.\n" "\n" " If the a value is not received before the timeout has elapsed or if the\n" " task process crashes then this function crashes.\n" ). -spec await(task(HCO), integer()) -> HCO. await(Task, Timeout) -> _assert_subject = try_await(Task, Timeout), {ok, Value} = case _assert_subject of {ok, _} -> _assert_subject; _assert_fail -> erlang:error(#{gleam_error => let_assert, message => <<"Pattern match failed, no pattern matched the value."/utf8>>, value => _assert_fail, module => <<"eensy/otp/task"/utf8>>, function => <<"await"/utf8>>, line => 119}) end, Value.