Should I call Task.await if Task.yield returns {:ok, value}?

Please clarify your use case? What are you doing exactly and how is Task.yield better in your scenario, compared to using Task.async + Task.await?

I need to send the task result in a 2500ms timeframe. If the result is not ready yet, I have to send a dummy message, and I have another way to send the result back.

So basically

params = some_data()
ref = make_ref()
parent = self()

spawn(fn ->
  t = Task.async(fn -> mod.run(params) end)

  case Task.yield(t, @max_wait_ms) do
    {:ok, result} ->
      send(parent, {:result, ref, result})

    nil ->
      send(parent, {:result, ref, :still_running})
      result = Task.await(t)
      send_delayed_response(params, result)

    {:exit, reason} ->
      exit(reason)
  end
end)

receive do
  {:result, ^ref, result} -> {:ok, result}
end

So I did a simple test in the shell and I have my answer anyway : after calling Task.yield, no more messages from the task are coming (the monitor is also cleaned).

iex(1)> t = Task.async(fn -> :test end)
%Task{
  owner: #PID<0.104.0>,
  pid: #PID<0.106.0>,
  ref: #Reference<0.1716671456.3384016898.195927>
}
iex(2)> Task.yield t
{:ok, :test}
iex(3)> flush
:ok

I’ve started build a library around oban a few month ago, which does what you’re doing here, but with the persistent queue of oban. I just didn’t finish it properly at the time.

… provided the task has already terminated

iex(1)> f =
...(1)>   fn ->
...(1)>     Process.sleep(500)
...(1)>     :test
...(1)>   end
#Function<21.126501267/0 in :erl_eval.expr/5>
iex(2)> t = Task.async(f)
%Task{
  owner: #PID<0.104.0>,
  pid: #PID<0.111.0>,
  ref: #Reference<0.3116447749.1782841345.230907>
}
iex(3)> Task.yield(t,300)
nil
iex(4)> flush
:ok
iex(5)> Process.sleep(300)
:ok
iex(6)> flush
{#Reference<0.3116447749.1782841345.230907>, :test}
{:DOWN, #Reference<0.3116447749.1782841345.230907>, :process, #PID<0.111.0>,
 :normal}
:ok
iex(7)>

iex(1)> f =
...(1)>   fn ->
...(1)>     Process.sleep(500)
...(1)>     :test
...(1)>   end
#Function<21.126501267/0 in :erl_eval.expr/5>
iex(2)> t = Task.async(f)
%Task{
  owner: #PID<0.104.0>,
  pid: #PID<0.111.0>,
  ref: #Reference<0.4284780453.977797126.180625>
}
iex(3)> case Task.yield(t, 300) || Task.shutdown(t) do
...(3)>   {:ok, result} = result ->
...(3)>     result
...(3)> 
...(3)>   nil ->
...(3)>     "Timed out"
...(3)> end
"Timed out"
iex(4)> flush
:ok
iex(5)> Process.sleep(300)
:ok
iex(6)> flush
:ok
iex(7)> 

And the documentation points this out.

I looked into it but it lack docs indeed.

Here I have 3 processes :

• The main code block I posted (it is a Phoenix controller). It has to return in the 2500ms timeframe.
• The spawn process (will be a supervised task that you do not have to await). This one yields from the task and return the data, but if the data is not ready, it will await it forever (I forgot to set infinity timeout on await). That is why I don not use spawn_link.
• The task itself, a simple worker

The trick is that if the yield is succesful, we do not want to call send_delayed_response as we will just return the data.

So your on_task_gone could be a good fit but I would have to use only Task.await then ? and catch the timeout exit on my own ? I did not plan to use oban or other libraries because my problem is solved with this single block. But if you want to separate that from Omen and create a tiny lib that just redirects the task result to another destination after a special yield I’d use it.

Absolutely, that is why I said if Task.yield returns {:ok, value}?. Of course if yield returns nil you have to await or yield more.

But the documentation does not tell that if yield is successful you can forget about the task entirely, and on the other hand stresses very much about the need to await at all costs.

The point is that in general a timed out yield should be accompanied by a Task.shutdown/2.

It seems a peculiar decision to use the timeout and then not terminate the task - by calling await you are giving the task an additional 5000 ms. If the task still doesn’t respond the process will exit.

Yeah, I created it specifically to add the functionality you have with Task to jobs in Oban, because I don’t like the task being silently dropped if e.g. the machine goes down after your 2500ms timeframe, but before the Task itself actually finished.

I don’t know if it is peculiar, it is just what those who call my code expect : a response within 2500ms or a response elsewhere, later. In either case the task has to be completed. shutdown does not fit in this scheme.

My code above lacks the timeout for await, which will be :infinity actually.

Thank you

Your approach still lacks resilience as the parent may get the still_running message but there never is a follow up because the Task goes zombie or crashes after being late.

I’d look into using Task.Supervisor instead.

Example:

# file: my_app/lib/my_app/application.ex
#
# created with "mix new my_app --sup"
# 
defmodule MyApp.Application do
  use Application

  def start(_type, _args) do
    name = MyApp.TaskSupervisor

    children = [
      {Task.Supervisor, name: name}
    ]

    opts = [strategy: :rest_for_one, name: MyApp.Supervisor]
    Supervisor.start_link(children, opts)
  end
end

# file: my_app/lib/demo.ex
#
defmodule Demo do
  @name MyApp.TaskSupervisor
  @max_wait_ms 500
  @timely_timeout div(@max_wait_ms, 2)
  @late_timeout @max_wait_ms + @timely_timeout
  @too_late_timeout 3 * @max_wait_ms

  def launch(arg) do
    task = Task.Supervisor.async_nolink(@name, __MODULE__, :some_work, [arg])

    task
    |> Task.yield(@max_wait_ms)
    |> handle_task_yield(task)
  end

  defp handle_task_yield({:ok, value}, _) do
    IO.puts("result: #{inspect(value)}")
    {:ok, value}
  end

  defp handle_task_yield(nil, task) do
    IO.puts("Timed out")
    await_late_result(task)
  end

  defp handle_task_yield({:exit, reason}, _) do
    IO.puts("Task exit: #{inspect(reason)}")
    {:error, reason}
  end

  defp await_late_result(%Task{ref: mon, pid: pid}) do
    # in GenServer these messages would go through 
    # handle_info
    receive do
      {^mon, value} ->
        Process.demonitor(mon, [:flush])
        IO.puts("Better late than never: #{inspect(value)}")
        {:ok, value}

      {:DOWN, ^mon, _, ^pid, reason} when reason != :normal ->
        Process.demonitor(mon, [:flush])
        IO.puts("Task LATE exit: #{inspect(reason)}")
        {:error, reason}
    after
      @max_wait_ms ->
        IO.puts("I'm not waiting forever!")
        Process.demonitor(mon, [:flush])
        Process.exit(pid, :kill)
        {:error, :far_too_late}
    end
  end

  def some_work(:timely = type) do
    Process.sleep(@timely_timeout)
    type
  end

  def some_work(:late = type) do
    Process.sleep(@late_timeout)
    type
  end

  def some_work(:too_late = type) do
    Process.sleep(@too_late_timeout)
    type
  end

  def some_work(:crash = type) do
    exit(type)
  end

  def some_work(:late_crash = type) do
    Process.sleep(@late_timeout)
    exit(type)
  end

  def some_work(:too_late_crash = type) do
    Process.sleep(@too_late_timeout)
    exit(type)
  end
end

$ iex -S mix
Erlang/OTP 22 [erts-10.5] [source] [64-bit] [smp:8:8] [ds:8:8:10] [async-threads:1] [hipe] [dtrace]

Compiling 1 file (.ex)
Interactive Elixir (1.9.1) - press Ctrl+C to exit (type h() ENTER for help)
iex(1)> Demo.launch(:timely)
result: :timely
{:ok, :timely}
iex(2)> Demo.launch(:late)
Timed out
Better late than never: :late
{:ok, :late}
iex(3)> Demo.launch(:too_late)
Timed out
I'm not waiting forever!
{:error, :far_too_late}
iex(4)> Demo.launch(:crash)

11:17:17.617 [error] Task #PID<0.153.0> started from #PID<0.145.0> terminating
** (stop) :crash
    (my_app) lib/demo.ex:71: Demo.some_work/1
    (elixir) lib/task/supervised.ex:90: Task.Supervised.invoke_mfa/2
    (elixir) lib/task/supervised.ex:35: Task.Supervised.reply/5
    (stdlib) proc_lib.erl:249: :proc_lib.init_p_do_apply/3
Function: &Demo.some_work/1
    Args: [:crash]
Task exit: :crash
{:error, :crash}
iex(5)> Demo.launch(:late_crash)
Timed out
Task LATE exit: :late_crash

11:17:18.370 [error] Task #PID<0.155.0> started from #PID<0.145.0> terminating
** (stop) :late_crash
    (my_app) lib/demo.ex:76: Demo.some_work/1
    (elixir) lib/task/supervised.ex:90: Task.Supervised.invoke_mfa/2
    (elixir) lib/task/supervised.ex:35: Task.Supervised.reply/5
    (stdlib) proc_lib.erl:249: :proc_lib.init_p_do_apply/3
Function: &Demo.some_work/1
    Args: [:late_crash]
{:error, :late_crash}
iex(6)> Demo.launch(:too_late_crash)
Timed out
I'm not waiting forever!
{:error, :far_too_late}
iex(7)> flush
:ok
iex(8)>