lud

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

All is in the title, but to clarify:

It seems that if Task.yield returns {:ok, value} it means that the task result message was received, so there is no risk of receiving an unhandled message later, thus no need to call Task.await (which would give the same value anyway).

But I want to be sure.

Thank you

22 comments

#task #asynchronous

7 2686 22

2019-10-18 23:38:59 UTC

Marked As Solved

lud

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

Post #3

Also Liked

LostKobrakai

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.

Post #4

lud

Well as you want details, there is no phoenix channel, the other way is to post the response to another website via HTTP.

I am implementing a Slack command that will do some stuff on the Gitlab API. Heavy stuff, so it can take a while, but sometimes it will take half a second to work. This is for other devs and they do not really care about the first meaningful paint

Indeed, as I am coding it, I am leaning towards always acknowledging first and always sending the reply through the other way. It is not what is asked though, but it seems that the Slack timeout is too sensitive about network latency.

The exercise was entertaining though.

Post #23

peerreynders

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)>

Post #11

lud

Sorry but I do not get it.

Imagine a task that will never finish for the example.

handle_task_yield is called after Task.yield, which already waited @max_wait_ms

If the arg to handle_task_yield is nil, it will call await_late_result and again wait @max_wait_ms.

So launch has not returned after the first timeout. The result is not delivered.

Thanks for your time

Edit: or your IO.puts("Time out") is actually my send(parent, {:result, :still_running}) in which case we agree and that is what I will do, just using yield twice.

using Task.yield or Task.await directly may not be advisable.

This is why I used send : the parent expects one and only one message from the middle task layer (my spawn’ed process).

I will do a proper implementation soon and ask for your review if you do not mind.

Post #14

peerreynders

From the GenServer’s perspective the intermediate timeout doesn’t really serve any purpose - unless it needs to let someone else know - if it handles the tasks directly.

# 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},
      {Demo, 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
  use GenServer

  def start_link(args) do
    GenServer.start_link(__MODULE__, args, name: __MODULE__)
  end

  def init(args) do
    with {:ok, name} <- Keyword.fetch(args, :name) do
      {:ok, {name, nil}}
    else
      _ ->
        {:stop, :badarg}
    end
  end

  def handle_call({:launch, arg}, _from, {name, nil}) do
    handles = launch_task(name, arg)
    {:reply, :ok, {name, handles}}
  end

  def handle_info({:kill_task, mon}, {name, handles} = state) do
    # timeout expired
    case handles do
      {%Task{ref: ^mon} = task, _} ->
        Process.demonitor(mon, [:flush])
        kill_task(task)
        IO.puts("Timed out: #{inspect(task)}")
        {:noreply, {name, nil}}

      _ ->
        IO.puts("REDUNDANT timeout")
        {:noreply, {name, state}}
    end
  end

  def handle_info({:DOWN, mon, _, down_pid, reason}, {name, handles} = state) do
    # :DOWN message arrives when task exits
    case handles do
      {%Task{ref: ^mon} = task, timeout_ref} ->
        Process.cancel_timer(timeout_ref)
        Process.demonitor(mon, [:flush])
        IO.puts("DOWN: #{inspect(down_pid)} #{inspect(reason)}")
        new_handles = if(reason == :normal, do: {task, nil}, else: nil)
        {:noreply, {name, new_handles}}

      _ ->
        IO.puts("REDUNDANT DOWN: #{inspect(down_pid)} #{inspect(reason)}")
        {:noreply, {name, state}}
    end
  end

  def handle_info({mon, value}, {name, handles} = state) when is_reference(mon) do
    # normal task result arrives here - demonitor/flush :normal :DOWN
    case handles do
      {%Task{ref: ^mon}, timeout_ref} ->
        if timeout_ref do
          Process.cancel_timer(timeout_ref)
          Process.demonitor(mon, [:flush])
        end

        IO.puts("result: #{inspect(value)}")
        {:noreply, {name, nil}}

      _ ->
        IO.puts("REDUNDANT result: #{inspect(value)} with #{inspect(mon)}")
        {:noreply, state}
    end
  end

  def handle_info(msg, state) do
    IO.inspect(msg)
    {:noreply, state}
  end

  # ---

  @max_wait_ms 500
  @timely_timeout div(@max_wait_ms, 2)
  @too_late_timeout 3 * @max_wait_ms

  defp launch_task(name, arg) do
    %Task{ref: mon} = task = Task.Supervisor.async_nolink(name, __MODULE__, :some_work, [arg])
    ref = Process.send_after(self(), {:kill_task, mon}, @max_wait_ms)
    {task, ref}
  end

  defp kill_task(%Task{pid: pid}),
    do: Process.exit(pid, :kill)

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

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

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

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

  # --- Demo API ---

  def launch(arg),
    do: GenServer.call(__MODULE__, {:launch, arg})
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(:too_late)
:ok
iex(2)> Timed out: %Task{owner: #PID<0.145.0>, pid: #PID<0.148.0>, ref: #Reference<0.177062570.1582825475.5428>}

nil
iex(3)> Demo.launch(:crash)   
:ok
iex(4)> 
14:13:26.156 [error] Task #PID<0.151.0> started from Demo terminating
** (stop) :crash
    (my_app) lib/demo.ex:105: 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]
DOWN: #PID<0.151.0> :crash

nil
iex(5)> Demo.launch(:too_late_crash)
:ok
iex(6)> Timed out: %Task{owner: #PID<0.145.0>, pid: #PID<0.154.0>, ref: #Reference<0.177062570.1582825480.4596>}

nil
iex(7)> 
BREAK: (a)bort (c)ontinue (p)roc info (i)nfo (l)oaded
       (v)ersion (k)ill (D)b-tables (d)istribution
a
Peers-MBP:my_app wheatley$ mix format
Peers-MBP:my_app wheatley$ 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)
:ok
iex(2)> result: :timely

nil
iex(3)> Demo.launch(:too_late) 
:ok
iex(4)> Timed out: %Task{owner: #PID<0.145.0>, pid: #PID<0.151.0>, ref: #Reference<0.981638884.3997958147.252725>}

nil
iex(5)> Demo.launch(:crash)
:ok
iex(6)> 
14:15:12.347 [error] Task #PID<0.154.0> started from Demo terminating
** (stop) :crash
    (my_app) lib/demo.ex:105: 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]
DOWN: #PID<0.154.0> :crash

nil
iex(7)> Demo.launch(:too_late_crash)
:ok
iex(8)> Timed out: %Task{owner: #PID<0.145.0>, pid: #PID<0.157.0>, ref: #Reference<0.981638884.3997958147.252801>}

nil
iex(9)>

Post #15