1181 lines
38 KiB
Elixir
1181 lines
38 KiB
Elixir
# SPDX-License-Identifier: Apache-2.0
|
|
# SPDX-FileCopyrightText: 2021 The Elixir Team
|
|
# SPDX-FileCopyrightText: 2012 Plataformatec
|
|
|
|
Code.require_file("test_helper.exs", __DIR__)
|
|
|
|
defmodule TaskTest do
|
|
use ExUnit.Case
|
|
doctest Task
|
|
@moduletag :capture_log
|
|
|
|
def wait_and_send(caller, atom) do
|
|
send(caller, :ready)
|
|
receive do: (true -> true)
|
|
send(caller, atom)
|
|
end
|
|
|
|
defp create_task_in_other_process do
|
|
caller = self()
|
|
spawn(fn -> send(caller, Task.async(fn -> nil end)) end)
|
|
receive do: (task -> task)
|
|
end
|
|
|
|
defp create_dummy_task(reason) do
|
|
{pid, ref} = spawn_monitor(Kernel, :exit, [reason])
|
|
|
|
receive do
|
|
{:DOWN, ^ref, _, _, _} ->
|
|
%Task{ref: ref, pid: pid, owner: self(), mfa: {__MODULE__, :create_dummy_task, 1}}
|
|
end
|
|
end
|
|
|
|
def sleep(number) do
|
|
Process.sleep(number)
|
|
number
|
|
end
|
|
|
|
def wait_until_down(task) do
|
|
ref = Process.monitor(task.pid)
|
|
assert_receive {:DOWN, ^ref, _, _, _}
|
|
end
|
|
|
|
test "can be supervised directly" do
|
|
assert {:ok, _} = Supervisor.start_link([{Task, fn -> :ok end}], strategy: :one_for_one)
|
|
end
|
|
|
|
test "generates child_spec/1" do
|
|
defmodule MyTask do
|
|
use Task
|
|
end
|
|
|
|
assert MyTask.child_spec([:hello]) == %{
|
|
id: MyTask,
|
|
restart: :temporary,
|
|
start: {MyTask, :start_link, [[:hello]]}
|
|
}
|
|
|
|
defmodule CustomTask do
|
|
use Task, id: :id, restart: :permanent, shutdown: :infinity, start: {:foo, :bar, []}
|
|
end
|
|
|
|
assert CustomTask.child_spec([:hello]) == %{
|
|
id: :id,
|
|
restart: :permanent,
|
|
shutdown: :infinity,
|
|
start: {:foo, :bar, []}
|
|
}
|
|
end
|
|
|
|
test "async/1" do
|
|
parent = self()
|
|
fun = fn -> wait_and_send(parent, :done) end
|
|
task = Task.async(fun)
|
|
|
|
# Assert the struct
|
|
assert task.__struct__ == Task
|
|
assert is_pid(task.pid)
|
|
assert is_reference(task.ref)
|
|
assert task.mfa == {:erlang, :apply, 2}
|
|
|
|
# Assert the link
|
|
{:links, links} = Process.info(self(), :links)
|
|
assert task.pid in links
|
|
|
|
receive do: (:ready -> :ok)
|
|
|
|
# Assert the initial call
|
|
{:name, fun_name} = Function.info(fun, :name)
|
|
assert {__MODULE__, fun_name, 0} === :proc_lib.translate_initial_call(task.pid)
|
|
|
|
# Run the task
|
|
send(task.pid, true)
|
|
|
|
# Assert response and monitoring messages
|
|
ref = task.ref
|
|
assert_receive {^ref, :done}
|
|
assert_receive {:DOWN, ^ref, _, _, :normal}
|
|
end
|
|
|
|
test "async/3" do
|
|
task = Task.async(__MODULE__, :wait_and_send, [self(), :done])
|
|
assert task.__struct__ == Task
|
|
assert task.mfa == {__MODULE__, :wait_and_send, 2}
|
|
|
|
{:links, links} = Process.info(self(), :links)
|
|
assert task.pid in links
|
|
|
|
receive do: (:ready -> :ok)
|
|
assert {__MODULE__, :wait_and_send, 2} === :proc_lib.translate_initial_call(task.pid)
|
|
|
|
send(task.pid, true)
|
|
assert Task.await(task) === :done
|
|
assert_receive :done
|
|
end
|
|
|
|
test "async with $callers" do
|
|
grandparent = self()
|
|
|
|
Task.async(fn ->
|
|
parent = self()
|
|
assert Process.get(:"$callers") == [grandparent]
|
|
|
|
Task.async(fn ->
|
|
assert Process.get(:"$callers") == [parent, grandparent]
|
|
end)
|
|
|> Task.await()
|
|
end)
|
|
|> Task.await()
|
|
end
|
|
|
|
test "start/1" do
|
|
parent = self()
|
|
fun = fn -> wait_and_send(parent, :done) end
|
|
{:ok, pid} = Task.start(fun)
|
|
|
|
{:links, links} = Process.info(self(), :links)
|
|
refute pid in links
|
|
|
|
receive do: (:ready -> :ok)
|
|
|
|
{:name, fun_name} = Function.info(fun, :name)
|
|
assert {__MODULE__, fun_name, 0} === :proc_lib.translate_initial_call(pid)
|
|
|
|
send(pid, true)
|
|
assert_receive :done
|
|
end
|
|
|
|
test "start/3" do
|
|
{:ok, pid} = Task.start(__MODULE__, :wait_and_send, [self(), :done])
|
|
|
|
{:links, links} = Process.info(self(), :links)
|
|
refute pid in links
|
|
|
|
receive do: (:ready -> :ok)
|
|
|
|
assert {__MODULE__, :wait_and_send, 2} === :proc_lib.translate_initial_call(pid)
|
|
|
|
send(pid, true)
|
|
assert_receive :done
|
|
end
|
|
|
|
test "completed/1" do
|
|
task = Task.completed(:done)
|
|
assert task.__struct__ == Task
|
|
|
|
refute task.pid
|
|
|
|
assert Task.await(task) == :done
|
|
end
|
|
|
|
test "start_link/1" do
|
|
parent = self()
|
|
fun = fn -> wait_and_send(parent, :done) end
|
|
{:ok, pid} = Task.start_link(fun)
|
|
|
|
{:links, links} = Process.info(self(), :links)
|
|
assert pid in links
|
|
|
|
receive do: (:ready -> :ok)
|
|
|
|
{:name, fun_name} = Function.info(fun, :name)
|
|
assert {__MODULE__, fun_name, 0} === :proc_lib.translate_initial_call(pid)
|
|
|
|
send(pid, true)
|
|
assert_receive :done
|
|
end
|
|
|
|
test "start_link/3" do
|
|
{:ok, pid} = Task.start_link(__MODULE__, :wait_and_send, [self(), :done])
|
|
|
|
{:links, links} = Process.info(self(), :links)
|
|
assert pid in links
|
|
|
|
receive do: (:ready -> :ok)
|
|
|
|
assert {__MODULE__, :wait_and_send, 2} === :proc_lib.translate_initial_call(pid)
|
|
|
|
send(pid, true)
|
|
assert_receive :done
|
|
end
|
|
|
|
test "start_link with $callers" do
|
|
grandparent = self()
|
|
|
|
Task.start_link(fn ->
|
|
parent = self()
|
|
assert Process.get(:"$callers") == [grandparent]
|
|
|
|
Task.start_link(fn ->
|
|
assert Process.get(:"$callers") == [parent, grandparent]
|
|
send(grandparent, :done)
|
|
end)
|
|
end)
|
|
|
|
assert_receive :done
|
|
end
|
|
|
|
describe "ignore/1" do
|
|
test "discards on time replies" do
|
|
task = Task.async(fn -> :ok end)
|
|
wait_until_down(task)
|
|
assert Task.ignore(task) == {:ok, :ok}
|
|
assert catch_exit(Task.await(task, 0)) == {:timeout, {Task, :await, [task, 0]}}
|
|
end
|
|
|
|
test "discards late replies" do
|
|
task = Task.async(fn -> assert_receive(:go) && :ok end)
|
|
assert Task.ignore(task) == nil
|
|
send(task.pid, :go)
|
|
wait_until_down(task)
|
|
assert catch_exit(Task.await(task, 0)) == {:timeout, {Task, :await, [task, 0]}}
|
|
end
|
|
|
|
test "discards on-time failures" do
|
|
Process.flag(:trap_exit, true)
|
|
task = Task.async(fn -> exit(:oops) end)
|
|
wait_until_down(task)
|
|
assert Task.ignore(task) == {:exit, :oops}
|
|
assert catch_exit(Task.await(task, 0)) == {:timeout, {Task, :await, [task, 0]}}
|
|
end
|
|
|
|
test "discards late failures" do
|
|
task = Task.async(fn -> assert_receive(:go) && exit(:oops) end)
|
|
assert Task.ignore(task) == nil
|
|
send(task.pid, :go)
|
|
wait_until_down(task)
|
|
assert catch_exit(Task.await(task, 0)) == {:timeout, {Task, :await, [task, 0]}}
|
|
end
|
|
|
|
test "exits on :noconnection" do
|
|
ref = make_ref()
|
|
task = %Task{ref: ref, pid: self(), owner: self(), mfa: {__MODULE__, :test, 1}}
|
|
send(self(), {:DOWN, ref, self(), self(), :noconnection})
|
|
assert catch_exit(Task.ignore(task)) |> elem(0) == {:nodedown, node()}
|
|
end
|
|
|
|
test "can ignore completed tasks" do
|
|
assert Task.ignore(Task.completed(:done)) == {:ok, :done}
|
|
end
|
|
end
|
|
|
|
describe "await/2" do
|
|
test "demonitors and unalias on timeout" do
|
|
task =
|
|
Task.async(fn ->
|
|
assert_receive :go
|
|
:done
|
|
end)
|
|
|
|
assert catch_exit(Task.await(task, 0)) == {:timeout, {Task, :await, [task, 0]}}
|
|
send(task.pid, :go)
|
|
ref = task.ref
|
|
|
|
wait_until_down(task)
|
|
refute_received {^ref, :done}
|
|
refute_received {:DOWN, ^ref, _, _, _}
|
|
end
|
|
|
|
test "exits on timeout" do
|
|
task = %Task{ref: make_ref(), owner: self(), pid: nil, mfa: {__MODULE__, :test, 1}}
|
|
assert catch_exit(Task.await(task, 0)) == {:timeout, {Task, :await, [task, 0]}}
|
|
end
|
|
|
|
test "exits on normal exit" do
|
|
task = Task.async(fn -> exit(:normal) end)
|
|
assert catch_exit(Task.await(task)) == {:normal, {Task, :await, [task, 5000]}}
|
|
end
|
|
|
|
test "exits on task throw" do
|
|
Process.flag(:trap_exit, true)
|
|
task = Task.async(fn -> throw(:unknown) end)
|
|
|
|
assert {{{:nocatch, :unknown}, _}, {Task, :await, [^task, 5000]}} =
|
|
catch_exit(Task.await(task))
|
|
end
|
|
|
|
test "exits on task error" do
|
|
Process.flag(:trap_exit, true)
|
|
task = Task.async(fn -> raise "oops" end)
|
|
assert {{%RuntimeError{}, _}, {Task, :await, [^task, 5000]}} = catch_exit(Task.await(task))
|
|
end
|
|
|
|
@compile {:no_warn_undefined, :module_does_not_exist}
|
|
|
|
test "exits on task undef module error" do
|
|
Process.flag(:trap_exit, true)
|
|
task = Task.async(&:module_does_not_exist.undef/0)
|
|
|
|
assert {exit_status, mfa} = catch_exit(Task.await(task))
|
|
assert {:undef, [{:module_does_not_exist, :undef, _, _} | _]} = exit_status
|
|
assert {Task, :await, [^task, 5000]} = mfa
|
|
end
|
|
|
|
@compile {:no_warn_undefined, {TaskTest, :undef, 0}}
|
|
|
|
test "exits on task undef function error" do
|
|
Process.flag(:trap_exit, true)
|
|
task = Task.async(&TaskTest.undef/0)
|
|
|
|
assert {{:undef, [{TaskTest, :undef, _, _} | _]}, {Task, :await, [^task, 5000]}} =
|
|
catch_exit(Task.await(task))
|
|
end
|
|
|
|
test "exits on task exit" do
|
|
Process.flag(:trap_exit, true)
|
|
task = Task.async(fn -> exit(:unknown) end)
|
|
assert {:unknown, {Task, :await, [^task, 5000]}} = catch_exit(Task.await(task))
|
|
end
|
|
|
|
test "exits on :noconnection" do
|
|
ref = make_ref()
|
|
task = %Task{ref: ref, pid: self(), owner: self(), mfa: {__MODULE__, :test, 1}}
|
|
send(self(), {:DOWN, ref, :process, self(), :noconnection})
|
|
assert catch_exit(Task.await(task)) |> elem(0) == {:nodedown, node()}
|
|
end
|
|
|
|
test "exits on :noconnection from named monitor" do
|
|
ref = make_ref()
|
|
task = %Task{ref: ref, owner: self(), pid: nil, mfa: {__MODULE__, :test, 1}}
|
|
send(self(), {:DOWN, ref, :process, {:name, :node}, :noconnection})
|
|
assert catch_exit(Task.await(task)) |> elem(0) == {:nodedown, :node}
|
|
end
|
|
|
|
test "raises when invoked from a non-owner process" do
|
|
task = create_task_in_other_process()
|
|
|
|
message =
|
|
"task #{inspect(task)} must be queried from the owner " <>
|
|
"but was queried from #{inspect(self())}"
|
|
|
|
assert_raise ArgumentError, message, fn -> Task.await(task, 1) end
|
|
end
|
|
end
|
|
|
|
describe "await_many/2" do
|
|
test "returns list of replies" do
|
|
tasks = for val <- [1, 3, 9], do: Task.async(fn -> val end)
|
|
assert Task.await_many(tasks) == [1, 3, 9]
|
|
end
|
|
|
|
test "returns replies in input order ignoring response order" do
|
|
refs = [ref_1 = make_ref(), ref_2 = make_ref(), ref_3 = make_ref()]
|
|
|
|
tasks =
|
|
Enum.map(refs, fn ref ->
|
|
%Task{ref: ref, owner: self(), pid: nil, mfa: {__MODULE__, :test, 1}}
|
|
end)
|
|
|
|
send(self(), {ref_2, 3})
|
|
send(self(), {ref_3, 9})
|
|
send(self(), {ref_1, 1})
|
|
assert Task.await_many(tasks) == [1, 3, 9]
|
|
end
|
|
|
|
test "returns an empty list immediately" do
|
|
assert Task.await_many([]) == []
|
|
end
|
|
|
|
test "ignores messages from other processes" do
|
|
other_ref = make_ref()
|
|
tasks = for val <- [:a, :b], do: Task.async(fn -> val end)
|
|
send(self(), other_ref)
|
|
send(self(), {other_ref, :z})
|
|
send(self(), {:DOWN, other_ref, :process, 1, :goodbye})
|
|
assert Task.await_many(tasks) == [:a, :b]
|
|
assert_received ^other_ref
|
|
assert_received {^other_ref, :z}
|
|
assert_received {:DOWN, ^other_ref, :process, 1, :goodbye}
|
|
end
|
|
|
|
test "ignores additional messages after reply" do
|
|
refs = [ref_1 = make_ref(), ref_2 = make_ref()]
|
|
|
|
tasks =
|
|
Enum.map(refs, fn ref ->
|
|
%Task{ref: ref, owner: self(), pid: nil, mfa: {__MODULE__, :test, 1}}
|
|
end)
|
|
|
|
send(self(), {ref_2, :b})
|
|
send(self(), {ref_2, :other})
|
|
send(self(), {ref_1, :a})
|
|
assert Task.await_many(tasks) == [:a, :b]
|
|
assert_received {^ref_2, :other}
|
|
end
|
|
|
|
test "exits on timeout" do
|
|
tasks = [Task.async(fn -> Process.sleep(:infinity) end)]
|
|
assert catch_exit(Task.await_many(tasks, 0)) == {:timeout, {Task, :await_many, [tasks, 0]}}
|
|
end
|
|
|
|
test "exits with same reason when task exits" do
|
|
tasks = [Task.async(fn -> exit(:normal) end)]
|
|
assert catch_exit(Task.await_many(tasks)) == {:normal, {Task, :await_many, [tasks, 5000]}}
|
|
end
|
|
|
|
test "exits immediately when any task exits" do
|
|
tasks = [
|
|
Task.async(fn -> Process.sleep(:infinity) end),
|
|
Task.async(fn -> exit(:normal) end)
|
|
]
|
|
|
|
assert catch_exit(Task.await_many(tasks)) == {:normal, {Task, :await_many, [tasks, 5000]}}
|
|
end
|
|
|
|
test "exits immediately when any task crashes" do
|
|
Process.flag(:trap_exit, true)
|
|
|
|
tasks = [
|
|
Task.async(fn -> Process.sleep(:infinity) end),
|
|
Task.async(fn -> exit(:unknown) end)
|
|
]
|
|
|
|
assert catch_exit(Task.await_many(tasks)) == {:unknown, {Task, :await_many, [tasks, 5000]}}
|
|
|
|
# Make sure all monitors are cleared up afterwards too
|
|
Enum.each(tasks, &Process.exit(&1.pid, :kill))
|
|
refute_received {:DOWN, _, _, _, _}
|
|
end
|
|
|
|
test "exits immediately when any task throws" do
|
|
Process.flag(:trap_exit, true)
|
|
|
|
tasks = [
|
|
Task.async(fn -> Process.sleep(:infinity) end),
|
|
Task.async(fn -> throw(:unknown) end)
|
|
]
|
|
|
|
assert {{{:nocatch, :unknown}, _}, {Task, :await_many, [^tasks, 5000]}} =
|
|
catch_exit(Task.await_many(tasks))
|
|
end
|
|
|
|
test "exits immediately on any task error" do
|
|
Process.flag(:trap_exit, true)
|
|
|
|
tasks = [
|
|
Task.async(fn -> Process.sleep(:infinity) end),
|
|
Task.async(fn -> raise "oops" end)
|
|
]
|
|
|
|
assert {{%RuntimeError{}, _}, {Task, :await_many, [^tasks, 5000]}} =
|
|
catch_exit(Task.await_many(tasks))
|
|
end
|
|
|
|
test "exits immediately on :noconnection" do
|
|
tasks = [
|
|
Task.async(fn -> Process.sleep(:infinity) end),
|
|
%Task{ref: ref = make_ref(), owner: self(), pid: self(), mfa: {__MODULE__, :test, 1}}
|
|
]
|
|
|
|
send(self(), {:DOWN, ref, :process, self(), :noconnection})
|
|
assert catch_exit(Task.await_many(tasks)) |> elem(0) == {:nodedown, node()}
|
|
end
|
|
|
|
test "exits immediately on :noconnection from named monitor" do
|
|
tasks = [
|
|
Task.async(fn -> Process.sleep(:infinity) end),
|
|
%Task{ref: ref = make_ref(), owner: self(), pid: nil, mfa: {__MODULE__, :test, 1}}
|
|
]
|
|
|
|
send(self(), {:DOWN, ref, :process, {:name, :node}, :noconnection})
|
|
assert catch_exit(Task.await_many(tasks)) |> elem(0) == {:nodedown, :node}
|
|
end
|
|
|
|
test "raises when invoked from a non-owner process" do
|
|
tasks = [
|
|
Task.async(fn -> Process.sleep(:infinity) end),
|
|
bad_task = create_task_in_other_process()
|
|
]
|
|
|
|
message =
|
|
"task #{inspect(bad_task)} must be queried from the owner " <>
|
|
"but was queried from #{inspect(self())}"
|
|
|
|
assert_raise ArgumentError, message, fn -> Task.await_many(tasks, 1) end
|
|
end
|
|
end
|
|
|
|
describe "yield/2" do
|
|
test "returns {:ok, result} when reply and :DOWN in message queue" do
|
|
task = %Task{ref: make_ref(), owner: self(), pid: nil, mfa: {__MODULE__, :test, 1}}
|
|
send(self(), {task.ref, :result})
|
|
send(self(), {:DOWN, task.ref, :process, self(), :abnormal})
|
|
assert Task.yield(task, 0) == {:ok, :result}
|
|
refute_received {:DOWN, _, _, _, _}
|
|
end
|
|
|
|
test "returns nil on timeout" do
|
|
task = %Task{ref: make_ref(), pid: nil, owner: self(), mfa: {__MODULE__, :test, 1}}
|
|
assert Task.yield(task, 0) == nil
|
|
end
|
|
|
|
test "return exit on normal exit" do
|
|
task = Task.async(fn -> exit(:normal) end)
|
|
assert Task.yield(task) == {:exit, :normal}
|
|
end
|
|
|
|
test "exits on :noconnection" do
|
|
ref = make_ref()
|
|
task = %Task{ref: ref, pid: self(), owner: self(), mfa: {__MODULE__, :test, 1}}
|
|
send(self(), {:DOWN, ref, self(), self(), :noconnection})
|
|
assert catch_exit(Task.yield(task)) |> elem(0) == {:nodedown, node()}
|
|
end
|
|
|
|
test "raises when invoked from a non-owner process" do
|
|
task = create_task_in_other_process()
|
|
|
|
message =
|
|
"task #{inspect(task)} must be queried from the owner " <>
|
|
"but was queried from #{inspect(self())}"
|
|
|
|
assert_raise ArgumentError, message, fn -> Task.yield(task, 1) end
|
|
end
|
|
end
|
|
|
|
describe "yield_many/2" do
|
|
test "returns {:ok, result} when reply and :DOWN in message queue" do
|
|
task = %Task{ref: make_ref(), owner: self(), pid: nil, mfa: {__MODULE__, :test, 1}}
|
|
send(self(), {task.ref, :result})
|
|
send(self(), {:DOWN, task.ref, :process, self(), :abnormal})
|
|
assert Task.yield_many([task], 0) == [{task, {:ok, :result}}]
|
|
refute_received {:DOWN, _, _, _, _}
|
|
end
|
|
|
|
test "returns nil on timeout by default" do
|
|
task = %Task{ref: make_ref(), owner: self(), pid: nil, mfa: {__MODULE__, :test, 1}}
|
|
assert Task.yield_many([task], 0) == [{task, nil}]
|
|
end
|
|
|
|
test "shuts down on timeout when configured" do
|
|
Process.flag(:trap_exit, true)
|
|
task = Task.async(fn -> Process.sleep(:infinity) end)
|
|
assert Task.yield_many([task], timeout: 0, on_timeout: :kill_task) == [{task, nil}]
|
|
refute Process.alive?(task.pid)
|
|
end
|
|
|
|
test "ignores on timeout when configured" do
|
|
task =
|
|
Task.async(fn ->
|
|
receive do
|
|
:done -> :ok
|
|
end
|
|
end)
|
|
|
|
assert Task.yield_many([task], timeout: 0, on_timeout: :ignore) == [{task, nil}]
|
|
assert Process.alive?(task.pid)
|
|
|
|
ref = Process.monitor(task.pid)
|
|
send(task.pid, :done)
|
|
assert_receive {:DOWN, ^ref, _, _, _}
|
|
|
|
assert Task.yield(task, 0) == nil
|
|
end
|
|
|
|
test "return exit on normal exit" do
|
|
task = Task.async(fn -> exit(:normal) end)
|
|
assert Task.yield_many([task]) == [{task, {:exit, :normal}}]
|
|
end
|
|
|
|
test "exits on :noconnection" do
|
|
ref = make_ref()
|
|
task = %Task{ref: ref, pid: self(), owner: self(), mfa: {__MODULE__, :test, 1}}
|
|
send(self(), {:DOWN, ref, :process, self(), :noconnection})
|
|
assert catch_exit(Task.yield_many([task])) |> elem(0) == {:nodedown, node()}
|
|
end
|
|
|
|
test "raises when invoked from a non-owner process" do
|
|
task = create_task_in_other_process()
|
|
|
|
message =
|
|
"task #{inspect(task)} must be queried from the owner " <>
|
|
"but was queried from #{inspect(self())}"
|
|
|
|
assert_raise ArgumentError, message, fn -> Task.yield_many([task], 1) end
|
|
end
|
|
|
|
test "returns results from multiple tasks" do
|
|
task1 = %Task{ref: make_ref(), owner: self(), pid: nil, mfa: {__MODULE__, :test, 1}}
|
|
task2 = %Task{ref: make_ref(), owner: self(), pid: nil, mfa: {__MODULE__, :test, 1}}
|
|
task3 = %Task{ref: make_ref(), owner: self(), pid: nil, mfa: {__MODULE__, :test, 1}}
|
|
|
|
send(self(), {task1.ref, :result})
|
|
send(self(), {:DOWN, task3.ref, :process, self(), :normal})
|
|
|
|
assert Task.yield_many([task1, task2, task3], 0) ==
|
|
[{task1, {:ok, :result}}, {task2, nil}, {task3, {:exit, :normal}}]
|
|
end
|
|
|
|
test "returns results from multiple tasks with limit" do
|
|
task1 = %Task{ref: make_ref(), owner: self(), pid: nil, mfa: {__MODULE__, :test, 1}}
|
|
task2 = %Task{ref: make_ref(), owner: self(), pid: nil, mfa: {__MODULE__, :test, 1}}
|
|
task3 = %Task{ref: make_ref(), owner: self(), pid: nil, mfa: {__MODULE__, :test, 1}}
|
|
|
|
send(self(), {task1.ref, :result})
|
|
send(self(), {:DOWN, task3.ref, :process, self(), :normal})
|
|
|
|
assert Task.yield_many([task1, task2, task3], limit: 1, timeout: :infinity) ==
|
|
[{task1, {:ok, :result}}, {task2, nil}, {task3, nil}]
|
|
|
|
assert Task.yield_many([task2, task3], limit: 1, timeout: :infinity) ==
|
|
[{task2, nil}, {task3, {:exit, :normal}}]
|
|
end
|
|
|
|
test "returns results from multiple tasks with limit and on timeout" do
|
|
Process.flag(:trap_exit, true)
|
|
task1 = Task.async(fn -> Process.sleep(:infinity) end)
|
|
task2 = Task.async(fn -> :done end)
|
|
|
|
assert Task.yield_many([task1, task2], timeout: :infinity, on_timeout: :kill_task, limit: 1) ==
|
|
[{task1, nil}, {task2, {:ok, :done}}]
|
|
|
|
assert Process.alive?(task1.pid)
|
|
|
|
assert Task.yield_many([task1], timeout: 0, on_timeout: :kill_task, limit: 1) ==
|
|
[{task1, nil}]
|
|
|
|
refute Process.alive?(task1.pid)
|
|
end
|
|
|
|
test "returns results on infinity timeout" do
|
|
task1 = %Task{ref: make_ref(), owner: self(), pid: nil, mfa: {__MODULE__, :test, 1}}
|
|
task2 = %Task{ref: make_ref(), owner: self(), pid: nil, mfa: {__MODULE__, :test, 1}}
|
|
task3 = %Task{ref: make_ref(), owner: self(), pid: nil, mfa: {__MODULE__, :test, 1}}
|
|
|
|
send(self(), {task1.ref, :result})
|
|
send(self(), {task2.ref, :result})
|
|
send(self(), {:DOWN, task3.ref, :process, self(), :normal})
|
|
|
|
assert Task.yield_many([task1, task2, task3], :infinity) ==
|
|
[{task1, {:ok, :result}}, {task2, {:ok, :result}}, {task3, {:exit, :normal}}]
|
|
end
|
|
end
|
|
|
|
describe "shutdown/2" do
|
|
test "returns {:ok, result} when reply and abnormal :DOWN in message queue" do
|
|
task = create_dummy_task(:abnormal)
|
|
send(self(), {task.ref, :result})
|
|
send(self(), {:DOWN, task.ref, :process, task.pid, :abnormal})
|
|
assert Task.shutdown(task) == {:ok, :result}
|
|
refute_received {:DOWN, _, _, _, _}
|
|
end
|
|
|
|
test "returns {:ok, result} when reply and normal :DOWN in message queue" do
|
|
task = create_dummy_task(:normal)
|
|
send(self(), {task.ref, :result})
|
|
send(self(), {:DOWN, task.ref, :process, task.pid, :normal})
|
|
assert Task.shutdown(task) == {:ok, :result}
|
|
refute_received {:DOWN, _, _, _, _}
|
|
end
|
|
|
|
test "returns {:ok, result} when reply and shut down :DOWN in message queue" do
|
|
task = create_dummy_task(:shutdown)
|
|
send(self(), {task.ref, :result})
|
|
send(self(), {:DOWN, task.ref, :process, task.pid, :shutdown})
|
|
assert Task.shutdown(task) == {:ok, :result}
|
|
refute_received {:DOWN, _, _, _, _}
|
|
end
|
|
|
|
test "returns nil on shutting down task" do
|
|
task = Task.async(:timer, :sleep, [:infinity])
|
|
assert Task.shutdown(task) == nil
|
|
end
|
|
|
|
test "returns exit on abnormal :DOWN in message queue" do
|
|
task = create_dummy_task(:abnormal)
|
|
send(self(), {:DOWN, task.ref, :process, task.pid, :abnormal})
|
|
assert Task.shutdown(task) == {:exit, :abnormal}
|
|
end
|
|
|
|
test "returns exit on normal :DOWN in message queue" do
|
|
task = create_dummy_task(:normal)
|
|
send(self(), {:DOWN, task.ref, :process, task.pid, :normal})
|
|
assert Task.shutdown(task) == {:exit, :normal}
|
|
end
|
|
|
|
test "returns nil on shutdown :DOWN in message queue" do
|
|
task = create_dummy_task(:shutdown)
|
|
send(self(), {:DOWN, task.ref, :process, task.pid, :shutdown})
|
|
assert Task.shutdown(task) == nil
|
|
end
|
|
|
|
test "returns exit on killed :DOWN in message queue" do
|
|
task = create_dummy_task(:killed)
|
|
send(self(), {:DOWN, task.ref, :process, task.pid, :killed})
|
|
assert Task.shutdown(task) == {:exit, :killed}
|
|
end
|
|
|
|
test "exits on noconnection :DOWN in message queue" do
|
|
task = create_dummy_task(:noconnection)
|
|
send(self(), {:DOWN, task.ref, :process, task.pid, :noconnection})
|
|
|
|
assert catch_exit(Task.shutdown(task)) ==
|
|
{{:nodedown, node()}, {Task, :shutdown, [task, 5000]}}
|
|
end
|
|
|
|
test "ignores if task PID is nil" do
|
|
ref = make_ref()
|
|
send(self(), {ref, :done})
|
|
|
|
assert Task.shutdown(%Task{ref: ref, owner: self(), pid: nil, mfa: {__MODULE__, :test, 1}}) ==
|
|
{:ok, :done}
|
|
|
|
ref = make_ref()
|
|
send(self(), {:DOWN, ref, :process, self(), :done})
|
|
|
|
assert Task.shutdown(%Task{ref: ref, owner: self(), pid: nil, mfa: {__MODULE__, :test, 1}}) ==
|
|
{:exit, :done}
|
|
end
|
|
|
|
test "raises when invoked from a non-owner process" do
|
|
task = create_task_in_other_process()
|
|
|
|
message =
|
|
"task #{inspect(task)} must be queried from the owner " <>
|
|
"but was queried from #{inspect(self())}"
|
|
|
|
assert_raise ArgumentError, message, fn -> Task.shutdown(task) end
|
|
end
|
|
|
|
test "returns nil on killing task" do
|
|
caller = self()
|
|
|
|
task =
|
|
Task.async(fn ->
|
|
Process.flag(:trap_exit, true)
|
|
wait_and_send(caller, :ready)
|
|
Process.sleep(:infinity)
|
|
end)
|
|
|
|
receive do: (:ready -> :ok)
|
|
|
|
assert Task.shutdown(task, :brutal_kill) == nil
|
|
refute_received {:DOWN, _, _, _, _}
|
|
end
|
|
|
|
test "returns {:exit, :noproc} if task handled" do
|
|
task = create_dummy_task(:noproc)
|
|
assert Task.shutdown(task) == {:exit, :noproc}
|
|
end
|
|
end
|
|
|
|
describe "shutdown/2 with :brutal_kill" do
|
|
test "returns {:ok, result} when reply and abnormal :DOWN in message queue" do
|
|
task = create_dummy_task(:abnormal)
|
|
send(self(), {task.ref, :result})
|
|
send(self(), {:DOWN, task.ref, :process, task.pid, :abnormal})
|
|
assert Task.shutdown(task, :brutal_kill) == {:ok, :result}
|
|
refute_received {:DOWN, _, _, _, _}
|
|
end
|
|
|
|
test "returns {:ok, result} when reply and normal :DOWN in message queue" do
|
|
task = create_dummy_task(:normal)
|
|
send(self(), {task.ref, :result})
|
|
send(self(), {:DOWN, task.ref, :process, task.pid, :normal})
|
|
assert Task.shutdown(task, :brutal_kill) == {:ok, :result}
|
|
refute_received {:DOWN, _, _, _, _}
|
|
end
|
|
|
|
test "returns {:ok, result} when reply and shut down :DOWN in message queue" do
|
|
task = create_dummy_task(:shutdown)
|
|
send(self(), {task.ref, :result})
|
|
send(self(), {:DOWN, task.ref, :process, task.pid, :shutdown})
|
|
assert Task.shutdown(task, :brutal_kill) == {:ok, :result}
|
|
refute_received {:DOWN, _, _, _, _}
|
|
end
|
|
|
|
test "returns nil on killed :DOWN in message queue" do
|
|
task = create_dummy_task(:killed)
|
|
send(self(), {:DOWN, task.ref, :process, task.pid, :killed})
|
|
assert Task.shutdown(task, :brutal_kill) == nil
|
|
end
|
|
|
|
test "returns exit on abnormal :DOWN in message queue" do
|
|
task = create_dummy_task(:abnormal)
|
|
send(self(), {:DOWN, task.ref, :process, task.pid, :abnormal})
|
|
assert Task.shutdown(task, :brutal_kill) == {:exit, :abnormal}
|
|
end
|
|
|
|
test "returns exit on normal :DOWN in message queue" do
|
|
task = create_dummy_task(:normal)
|
|
send(self(), {:DOWN, task.ref, :process, task.pid, :normal})
|
|
assert Task.shutdown(task, :brutal_kill) == {:exit, :normal}
|
|
end
|
|
|
|
test "returns exit on shutdown :DOWN in message queue" do
|
|
task = create_dummy_task(:shutdown)
|
|
send(self(), {:DOWN, task.ref, :process, task.pid, :shutdown})
|
|
assert Task.shutdown(task, :brutal_kill) == {:exit, :shutdown}
|
|
end
|
|
|
|
test "exits on noconnection :DOWN in message queue" do
|
|
task = create_dummy_task(:noconnection)
|
|
send(self(), {:DOWN, task.ref, :process, task.pid, :noconnection})
|
|
|
|
assert catch_exit(Task.shutdown(task, :brutal_kill)) ==
|
|
{{:nodedown, node()}, {Task, :shutdown, [task, :brutal_kill]}}
|
|
end
|
|
|
|
test "returns exit on killing task after shutdown timeout" do
|
|
caller = self()
|
|
|
|
task =
|
|
Task.async(fn ->
|
|
Process.flag(:trap_exit, true)
|
|
wait_and_send(caller, :ready)
|
|
Process.sleep(:infinity)
|
|
end)
|
|
|
|
receive do: (:ready -> :ok)
|
|
assert Task.shutdown(task, 1) == {:exit, :killed}
|
|
end
|
|
|
|
test "returns {:exit, :noproc} if task handled" do
|
|
task = create_dummy_task(:noproc)
|
|
assert Task.shutdown(task, :brutal_kill) == {:exit, :noproc}
|
|
end
|
|
end
|
|
|
|
describe "async_stream/2" do
|
|
test "timeout" do
|
|
assert catch_exit([:infinity] |> Task.async_stream(&sleep/1, timeout: 0) |> Enum.to_list()) ==
|
|
{:timeout, {Task.Supervised, :stream, [0]}}
|
|
|
|
refute_received _
|
|
end
|
|
|
|
test "streams an enumerable with ordered: false" do
|
|
opts = [max_concurrency: 1, ordered: false]
|
|
|
|
assert 4..1//-1
|
|
|> Task.async_stream(&sleep(&1 * 100), opts)
|
|
|> Enum.to_list() == [ok: 400, ok: 300, ok: 200, ok: 100]
|
|
|
|
opts = [max_concurrency: 4, ordered: false]
|
|
|
|
assert 4..1//-1
|
|
|> Task.async_stream(&sleep(&1 * 100), opts)
|
|
|> Enum.to_list() == [ok: 100, ok: 200, ok: 300, ok: 400]
|
|
end
|
|
|
|
test "streams an enumerable with ordered: false, on_timeout: :kill_task" do
|
|
opts = [max_concurrency: 4, ordered: false, on_timeout: :kill_task, timeout: 50]
|
|
|
|
assert [100, 1, 100, 1]
|
|
|> Task.async_stream(&sleep/1, opts)
|
|
|> Enum.to_list() == [ok: 1, ok: 1, exit: :timeout, exit: :timeout]
|
|
|
|
refute_received _
|
|
end
|
|
|
|
test "streams an enumerable with infinite timeout" do
|
|
[ok: :ok] = Task.async_stream([1], fn _ -> :ok end, timeout: :infinity) |> Enum.to_list()
|
|
end
|
|
|
|
test "does not allow streaming with max_concurrency = 0" do
|
|
assert_raise ArgumentError, ":max_concurrency must be an integer greater than zero", fn ->
|
|
Task.async_stream([1], fn _ -> :ok end, max_concurrency: 0)
|
|
end
|
|
end
|
|
|
|
test "does not allow streaming with invalid :on_timeout" do
|
|
assert_raise ArgumentError, ":on_timeout must be either :exit or :kill_task", fn ->
|
|
Task.async_stream([1], fn _ -> :ok end, on_timeout: :unknown)
|
|
end
|
|
end
|
|
|
|
test "does not allow streaming with invalid :timeout" do
|
|
assert_raise ArgumentError, ":timeout must be either a positive integer or :infinity", fn ->
|
|
Task.async_stream([1], fn _ -> :ok end, timeout: :unknown)
|
|
end
|
|
end
|
|
|
|
test "streams with fake down messages on the inbox" do
|
|
parent = self()
|
|
|
|
assert Task.async_stream([:ok], fn :ok ->
|
|
{:links, links} = Process.info(self(), :links)
|
|
|
|
for link <- links do
|
|
send(link, {:DOWN, make_ref(), :process, parent, :oops})
|
|
end
|
|
|
|
:ok
|
|
end)
|
|
|> Enum.to_list() == [ok: :ok]
|
|
end
|
|
|
|
test "with $callers" do
|
|
grandparent = self()
|
|
|
|
Task.async_stream([1], fn 1 ->
|
|
parent = self()
|
|
assert Process.get(:"$callers") == [grandparent]
|
|
|
|
Task.async_stream([1], fn 1 ->
|
|
assert Process.get(:"$callers") == [parent, grandparent]
|
|
send(grandparent, :done)
|
|
end)
|
|
|> Stream.run()
|
|
end)
|
|
|> Stream.run()
|
|
|
|
assert_receive :done
|
|
end
|
|
|
|
test "consuming from another process" do
|
|
parent = self()
|
|
stream = Task.async_stream([1, 2, 3], &send(parent, &1))
|
|
Task.start(Stream, :run, [stream])
|
|
assert_receive 1
|
|
assert_receive 2
|
|
assert_receive 3
|
|
end
|
|
|
|
test "wrapping a flat_map/concat with a haltable stream" do
|
|
result =
|
|
Stream.take([:foo, :bar], 1)
|
|
|> Stream.concat([1, 2])
|
|
|> Task.async_stream(& &1)
|
|
|> Enum.to_list()
|
|
|
|
assert result == [ok: :foo, ok: 1, ok: 2]
|
|
end
|
|
end
|
|
|
|
for {desc, concurrency} <- [==: 4, <: 2, >: 8] do
|
|
describe "async_stream with max_concurrency #{desc} tasks" do
|
|
@opts [max_concurrency: concurrency]
|
|
|
|
test "streams an enumerable with fun" do
|
|
assert 1..4
|
|
|> Task.async_stream(&sleep/1, @opts)
|
|
|> Enum.to_list() == [ok: 1, ok: 2, ok: 3, ok: 4]
|
|
end
|
|
|
|
test "streams an enumerable with mfa" do
|
|
assert 1..4
|
|
|> Task.async_stream(__MODULE__, :sleep, [], @opts)
|
|
|> Enum.to_list() == [ok: 1, ok: 2, ok: 3, ok: 4]
|
|
end
|
|
|
|
test "streams an enumerable without leaking tasks" do
|
|
assert 1..4
|
|
|> Task.async_stream(&sleep/1, @opts)
|
|
|> Enum.to_list() == [ok: 1, ok: 2, ok: 3, ok: 4]
|
|
|
|
refute_received _
|
|
end
|
|
|
|
test "streams an enumerable with slowest first" do
|
|
Process.flag(:trap_exit, true)
|
|
|
|
assert 4..1//-1
|
|
|> Task.async_stream(&sleep/1, @opts)
|
|
|> Enum.to_list() == [ok: 4, ok: 3, ok: 2, ok: 1]
|
|
end
|
|
|
|
test "streams an enumerable with exits" do
|
|
Process.flag(:trap_exit, true)
|
|
|
|
assert 1..4
|
|
|> Task.async_stream(&exit/1, @opts)
|
|
|> Enum.to_list() == [exit: 1, exit: 2, exit: 3, exit: 4]
|
|
|
|
refute_received {:EXIT, _, _}
|
|
end
|
|
|
|
test "shuts down unused tasks" do
|
|
assert [0, :infinity, :infinity, :infinity]
|
|
|> Task.async_stream(&sleep/1, @opts)
|
|
|> Enum.take(1) == [ok: 0]
|
|
|
|
assert Process.info(self(), :links) == {:links, []}
|
|
end
|
|
|
|
test "shuts down unused tasks without leaking messages" do
|
|
assert [0, :infinity, :infinity, :infinity]
|
|
|> Task.async_stream(&sleep/1, @opts)
|
|
|> Enum.take(1) == [ok: 0]
|
|
|
|
refute_received _
|
|
end
|
|
|
|
test "is zippable on success" do
|
|
task = 1..4 |> Task.async_stream(&sleep/1, @opts) |> Stream.map(&elem(&1, 1))
|
|
assert Enum.zip(task, task) == [{1, 1}, {2, 2}, {3, 3}, {4, 4}]
|
|
end
|
|
|
|
test "is zippable on failure" do
|
|
Process.flag(:trap_exit, true)
|
|
task = 1..4 |> Task.async_stream(&exit/1, @opts) |> Stream.map(&elem(&1, 1))
|
|
assert Enum.zip(task, task) == [{1, 1}, {2, 2}, {3, 3}, {4, 4}]
|
|
end
|
|
|
|
test "is zippable with slowest first" do
|
|
task = 4..1//-1 |> Task.async_stream(&sleep/1, @opts) |> Stream.map(&elem(&1, 1))
|
|
assert Enum.zip(task, task) == [{4, 4}, {3, 3}, {2, 2}, {1, 1}]
|
|
end
|
|
|
|
test "with inner halt on success" do
|
|
assert 1..8
|
|
|> Stream.take(4)
|
|
|> Task.async_stream(&sleep/1, @opts)
|
|
|> Enum.to_list() == [ok: 1, ok: 2, ok: 3, ok: 4]
|
|
end
|
|
|
|
test "with inner halt on failure" do
|
|
Process.flag(:trap_exit, true)
|
|
|
|
assert 1..8
|
|
|> Stream.take(4)
|
|
|> Task.async_stream(&exit/1, @opts)
|
|
|> Enum.to_list() == [exit: 1, exit: 2, exit: 3, exit: 4]
|
|
end
|
|
|
|
test "with inner halt and slowest first" do
|
|
assert 8..1//-1
|
|
|> Stream.take(4)
|
|
|> Task.async_stream(&sleep/1, @opts)
|
|
|> Enum.to_list() == [ok: 8, ok: 7, ok: 6, ok: 5]
|
|
end
|
|
|
|
test "with outer halt on success" do
|
|
assert 1..8
|
|
|> Task.async_stream(&sleep/1, @opts)
|
|
|> Enum.take(4) == [ok: 1, ok: 2, ok: 3, ok: 4]
|
|
end
|
|
|
|
test "with outer halt on failure" do
|
|
Process.flag(:trap_exit, true)
|
|
|
|
assert 1..8
|
|
|> Task.async_stream(&exit/1, @opts)
|
|
|> Enum.take(4) == [exit: 1, exit: 2, exit: 3, exit: 4]
|
|
end
|
|
|
|
test "with outer halt and slowest first" do
|
|
assert 8..1//-1
|
|
|> Task.async_stream(&sleep/1, @opts)
|
|
|> Enum.take(4) == [ok: 8, ok: 7, ok: 6, ok: 5]
|
|
end
|
|
|
|
test "terminates inner effect" do
|
|
stream =
|
|
1..4
|
|
|> Task.async_stream(&sleep/1, @opts)
|
|
|> Stream.transform(fn -> :ok end, fn x, acc -> {[x], acc} end, fn _ ->
|
|
Process.put(:stream_transform, true)
|
|
end)
|
|
|
|
Process.put(:stream_transform, false)
|
|
assert Enum.to_list(stream) == [ok: 1, ok: 2, ok: 3, ok: 4]
|
|
assert Process.get(:stream_transform)
|
|
end
|
|
|
|
test "terminates outer effect" do
|
|
stream =
|
|
1..4
|
|
|> Stream.transform(fn -> :ok end, fn x, acc -> {[x], acc} end, fn _ ->
|
|
Process.put(:stream_transform, true)
|
|
end)
|
|
|> Task.async_stream(&sleep/1, @opts)
|
|
|
|
Process.put(:stream_transform, false)
|
|
assert Enum.to_list(stream) == [ok: 1, ok: 2, ok: 3, ok: 4]
|
|
assert Process.get(:stream_transform)
|
|
end
|
|
|
|
test "with :on_timeout set to :kill_task" do
|
|
opts = Keyword.merge(@opts, on_timeout: :kill_task, timeout: 50)
|
|
|
|
assert [100, 1, 100, 1]
|
|
|> Task.async_stream(&sleep/1, opts)
|
|
|> Enum.to_list() == [exit: :timeout, ok: 1, exit: :timeout, ok: 1]
|
|
|
|
refute_received _
|
|
end
|
|
|
|
test "with timeout and :zip_input_on_exit set to true" do
|
|
opts = Keyword.merge(@opts, zip_input_on_exit: true, on_timeout: :kill_task, timeout: 50)
|
|
|
|
assert [1, 100]
|
|
|> Task.async_stream(&sleep/1, opts)
|
|
|> Enum.to_list() == [ok: 1, exit: {100, :timeout}]
|
|
end
|
|
|
|
test "with outer halt on failure and :zip_input_on_exit" do
|
|
Process.flag(:trap_exit, true)
|
|
opts = Keyword.merge(@opts, zip_input_on_exit: true)
|
|
|
|
assert 1..8
|
|
|> Task.async_stream(&exit/1, opts)
|
|
|> Enum.take(4) == [exit: {1, 1}, exit: {2, 2}, exit: {3, 3}, exit: {4, 4}]
|
|
end
|
|
end
|
|
end
|
|
|
|
describe "default :logger reporter" do
|
|
setup do
|
|
translator = :logger.get_primary_config().filters[:logger_translator]
|
|
assert :ok = :logger.remove_primary_filter(:logger_translator)
|
|
on_exit(fn -> :logger.add_primary_filter(:logger_translator, translator) end)
|
|
end
|
|
|
|
test "logs a terminated task" do
|
|
parent = self()
|
|
{:ok, pid} = Task.start_link(__MODULE__, :task, [parent])
|
|
|
|
assert ExUnit.CaptureLog.capture_log(fn ->
|
|
ref = Process.monitor(pid)
|
|
send(pid, :go)
|
|
receive do: ({:DOWN, ^ref, _, _, _} -> :ok)
|
|
end) =~ ~r"""
|
|
\[error\] \*\* Task #PID<\d+\.\d+\.\d+> terminating
|
|
\*\* Started from #PID<\d+\.\d+\.\d+>
|
|
\*\* When function == &TaskTest.task/1
|
|
\*\* arguments == \[#PID<\d+\.\d+\.\d+>\]
|
|
\*\* Reason for termination ==\s
|
|
\*\* {%RuntimeError{message: "oops"},
|
|
"""
|
|
end
|
|
|
|
test "logs a terminated task with a process label" do
|
|
fun = fn ->
|
|
Process.set_label({:any, "term"})
|
|
raise "oops"
|
|
end
|
|
|
|
parent = self()
|
|
{:ok, pid} = Task.start_link(__MODULE__, :task, [parent, fun])
|
|
|
|
assert ExUnit.CaptureLog.capture_log(fn ->
|
|
ref = Process.monitor(pid)
|
|
send(pid, :go)
|
|
receive do: ({:DOWN, ^ref, _, _, _} -> :ok)
|
|
end) =~ ~r"""
|
|
\[error\] \*\* Task #PID<\d+\.\d+\.\d+> terminating
|
|
\*\* Process Label == {:any, "term"}
|
|
\*\* Started from #PID<\d+\.\d+\.\d+>
|
|
\*\* When function == &TaskTest.task/2
|
|
\*\* arguments == \[#PID<\d+\.\d+\.\d+>,
|
|
#Function<.+>\]
|
|
\*\* Reason for termination ==\s
|
|
\*\* {%RuntimeError{message: "oops"},
|
|
"""
|
|
end
|
|
end
|
|
|
|
def task(parent, fun \\ fn -> raise "oops" end) do
|
|
mon = Process.monitor(parent)
|
|
Process.unlink(parent)
|
|
|
|
receive do
|
|
:go ->
|
|
fun.()
|
|
|
|
{:DOWN, ^mon, _, _, _} ->
|
|
exit(:shutdown)
|
|
end
|
|
end
|
|
end
|