Questions & Help>Questions

Saša Jurić's "Beyond Task.Async" blog

samba6

samba6

Saša Jurić's "Beyond Task.Async" blog

I am trying to follow this blog Beyond Task.Async. So far I have got the following code working:

defmodule Aggregator do
  @moduledoc """
  Testing async tasks
  """

  def new, do: 0

  def aggregate(result), do: result
  def aggregate(aggregator, result) do
    Process.sleep(500)
    aggregator + result
  end
end

defmodule Calc do
  @moduledoc """
  Testing async tasks
  """

  @spec run(Integer.t) :: Integer.t
  def run(id) when id > 0 do
    IO.puts "\n\n sleeping: #{id}"
    Process.sleep(id)
    IO.puts "\n\n waking from slumber: #{id}"
    id
  end
end

defmodule Async do
  @moduledoc """
  Testing async tasks
  """

  def run(diff \\ 0, timeout \\ 900) do
    1..10
    |> Enum.map(fn _ -> Enum.random(1..1000) - diff end)
    |> Enum.map(&Task.async(fn ->
          try do
            {:ok, Calc.run(&1)}
          rescue _ ->
            :error
          end
       end))
    |> collect_result timeout
  end

  defp collect_result(tasks, timeout) do
    ref = make_ref()
    timer = Process.send_after(self(), {:timeout, ref}, timeout)

    try do
      collect_result(tasks, Aggregator.new, ref)
    after
      :erlang.cancel_timer(timer)
      receive do
        {:timeout, ^ref} ->
          :ok

      after 0 ->
        :ok
      end
    end
  end
  defp collect_result([], aggregator, _), do: {:ok, Aggregator.aggregate(aggregator)}
  defp collect_result(tasks, aggregator, ref) do
    receive do
      {:timeout, ^ref} ->
        {:timeout, Aggregator.aggregate(aggregator)}

      msg ->
        case Task.find(tasks, msg) do
          {{:ok, result}, task} ->
              collect_result(
                List.delete(tasks, task),
                Aggregator.aggregate(aggregator, result),
                ref
              )

          # if task errors, stop monitoring task and ignore its result
          {:error, task} ->
            collect_result(List.delete(tasks, task), aggregator, ref)

          nil ->
            collect_result(tasks, aggregator, ref)
        end
    end
  end
end

So when I run: iex(295)> Async.run 700, I get {:ok, 489}

However, I read that Task.find was deprecated in favour of explicit message matching. So I refactored collect_result/3 to:

  defp collect_result(tasks, aggregator, ref) do
    receive do
      {:timeout, ^ref} ->
        {:timeout, Aggregator.aggregate(aggregator)}

      {task, {:ok, result}} ->
        collect_result(
          List.delete(tasks, task),
          Aggregator.aggregate(aggregator, result),
          ref
        )

      # if task errors, stop monitoring task and ignore its result
      {task, :error} ->
        collect_result(List.delete(tasks, task), aggregator, ref)
    end
  end

Now Async.run times out no matter the value of diff. For examples, iex(306)> Async.run 700 returned {:timeout, 260}, which is ridiculous since Calc.run(260) was the only successful task.

I can’t seem to figure out how to refactor the code using explicit message matching and still get same behaviour as first iteration of the code. Any help from the community will be greatly appreciated.

7 comments
3245 7

First 7 of 7 Posts Switch mode

samba6

samba6 OP

Ok, I got it working. I forgot that Task.async returns %Task{owner: owner, ref: ref, ...} so that List.delete(tasks, task) did not delete the task since tasks = [%Task{}, %Task{}, ...], but task is a reference. So I got the tasks element corresponding to task using the code below:

defp get_matched_task(task, tasks) do
    Enum.find(tasks, nil, fn task_struct ->
      case task_struct do
        %Task{ref: ^task} ->
          true
        _ ->
          false
      end
    end)
  end

And so collect_result/3 becomes:

defp collect_result(tasks, aggregator, timer_ref) do
    receive do
      {:timeout, ^timer_ref} ->
        {:timeout, Aggregator.aggregate(aggregator)}

      {task, val} ->
        task_ = get_matched_task(task, tasks)
        case val do
          {:ok, result} ->
            collect_result(
              List.delete(tasks, task_),
              Aggregator.aggregate(aggregator, result),
              timer_ref
            )

          _ ->
            collect_result(List.delete(tasks, task_), aggregator, timer_ref)
        end

      _ ->
        collect_result(tasks, aggregator, timer_ref)
    end
  end

Now I the timeout message is sent only when a task runs for longer than the timeout. But I’m thinking may be there is a better way of achieving same result. Any help in this regard will be appreciated.

2
Post #1
peerreynders

peerreynders

Might as well use a Map

defmodule Async do
  @moduledoc """
  Testing async tasks
  """

  def run(diff \\ 0, timeout \\ 900) do
    1..10
    |> Enum.map(fn _ -> Enum.random(1..1000) - diff end)
    |> Enum.map(&Task.async(fn ->
          try do
            {:ok, Calc.run(&1)}
          rescue _ ->
            :error
          end
        end))
    |> Enum.reduce(%{}, &(Map.put &2, &1.ref, &1)) # 1. Put tasks into a map indexed by monitor reference
    |> collect_result(timeout)
  end

  defp collect_result(tasks, timeout) do
    ref = make_ref()
    timer = Process.send_after(self(), {:timeout, ref}, timeout)

    try do
      collect_result(tasks, Aggregator.new, ref)
    after
      :erlang.cancel_timer(timer)
      receive do
        {:timeout, ^ref} ->
          :ok
      after 0 ->
        :ok
      end
    end
  end
  defp collect_result(tasks , aggregator, _) when tasks == %{} do
    {:ok, Aggregator.aggregate(aggregator)} # 5. no more tasks to process
  end
  defp collect_result(tasks, aggregator, ref) do
    receive do
      {:timeout, ^ref} ->
        (Map.values tasks) # 4. clean up remaining tasks
        |> Enum.each(&(Task.shutdown &1, :brutal_kill))
        |> Enum.each(&(Process.demonitor &1.ref, [:flush]))

        {:timeout, Aggregator.aggregate(aggregator)}

      {task_ref, reply} ->
        cond do
          Map.has_key? tasks, task_ref ->        # 2. verify source
            Process.demonitor task_ref, [:flush] # 3. purge :DOWN message
            new_aggregator =
              case reply do
                {:ok, result} ->
                  Aggregator.aggregate aggregator, result
                :error ->
                  aggregator
              end
            tasks
            |> Map.delete(task_ref)
            |> collect_result(new_aggregator, ref)

          true ->
            collect_result(tasks, aggregator, ref)
        end
    end
  end
end

What particular avenue of improvement did you have in mind?

At this point I’m wondering whether

  1. A GenServer could clean this up
  2. Once GenServers are in the picture whether I can actually get rid of Task entirely

The more I’m exposed to Task the more I’m predisposed to shoving it over into the corner with Agent as a nice curiosity which ultimately isn’t all that useful.

1
Post #2
OvermindDL1

OvermindDL1

It has a few limited uses, I do not think the use with GenServer is such a use though and GenServer should just be used. :slight_smile:

I’ve legit used Task only a couple of times total, and I could have easily done them another way (which really might have been shorter code too).

0
Post #3
samba6

samba6 OP

Thanks @peerreynders. Your corrections worked. I asked if there was a better way because I happened to look at Task.find after I had written get_matched_task/2 and they looked similar. Since Task.find was deprecated, I thought a function which does almost the same thing had to be defective.

I really like the cleanups you introduced. It solved a headache I was having where I’d get {:DOWN, _, _, _, _} messages and couldn’t figure out where they were coming from. I tried using observer.start to see if I could catch them, but I guess since those processes had exited before I turned to observer, I wouldn’t find them.

I am truly grateful. Thanks

0
Post #4
peerreynders

peerreynders

So here is an approach where you essentially handle the spawned processes by yourself.

 # File: calc.ex
defmodule Calc do
  @spec run(Integer.t) :: Integer.t
  def run(id) when id > 0 do
    desc = "#{id} (#{inspect self()})"
    IO.puts (" sleeping: " <> desc)
    Process.sleep(id)
    IO.puts (" waking from slumber: " <> desc)
    id
  end
end

# File: accumulator.ex
defmodule Accumulator do
  def sum(addend, augend) do
    Process.sleep(500) # NOTE: This is responsible for some tasks
    addend + augend    # running beyond the timeout.
  end                  # The point is that only messages
end                    # up to the timeout message are actually processed
                       # See what happens when you comment "sleep" out

# File map_reduce.ex
defmodule MapReduce do
  use GenServer

  defp map_spawn(fun, reply_to),
    do: fn elem ->
          Kernel.spawn_monitor(
            fn ->
              reply =
                try do
                  {self(), {:ok, fun.(elem)}}
                rescue _ ->
                  {self(), :error}
                end
              GenServer.cast reply_to, reply
            end
          )
        end

  defp put_task({pid, ref}, tasks),
    do: Map.put tasks, pid, ref

  defp remove_task(tasks, pid),
    do: Map.delete tasks, pid

  defp purge_task({pid, ref}) do
    # No longer interested in the process
    # or the associated :DOWN messages
    Process.demonitor ref, [:flush]

    result =
      cond do
        Process.alive? pid ->
          Process.exit pid, :kill
        true ->
          false # no need
      end

    IO.puts "  purge task #{inspect pid}: #{inspect result}"

    result
  end

  defp cleanup_tasks(tasks) do
    tasks
    |> Map.to_list()
    |> Enum.each(&purge_task/1)
  end

  defp assess_progress({tasks, _, _, _, _} = state) do
    cond do
      tasks == %{} ->
        {:stop, :normal, state}
      true ->
        {:noreply, state}
    end
  end

  defp handle_result({tasks, acc, reduce, reply_to, timer_ref}, {pid, result}) do
    new_acc =
      case result do
        {:ok, value} ->
          reduce.(acc, value)
        :error ->
          acc # TODO account for errors in the reply to creator
      end

    assess_progress {(remove_task tasks, pid), new_acc, reduce, reply_to, timer_ref}
  end

  def handle_down({tasks, acc, reduce, reply_to, timer_ref}, pid, reason) do
    IO.puts "Task #{inspect pid} terminated with reason: #{inspect reason}"
    # TODO: should indicate in reply to creator that there are DOWN tasks
    assess_progress {(remove_task tasks, pid), acc, reduce, reply_to, timer_ref}
  end

  ## callbacks: message handlers
  def handle_cast({pid, _} = reply, {tasks, _, _, _, _} = state) when is_pid pid do
    # process result
    case Map.get tasks, pid, :none do
      :none ->
        {:noreply, state}
      ref ->
        Process.demonitor ref, [:flush] # purge :DOWN message
        handle_result state, reply
    end
  end
  def handle_cast(:init, {data, init, fun, reduce, reply_to, timeout}) do
    tasks =
      data
      |> Enum.map(map_spawn(fun, self()))
      |> Enum.reduce(%{}, &put_task/2)

    timer_ref = Process.send_after self(), :timeout, timeout
    {:noreply, {tasks, init, reduce, reply_to, timer_ref}}
  end

  def handle_info(:timeout, {tasks, acc, reduce, reply_to, _timer_ref}) do
    IO.puts "  !!! timeout !!!"
    {:stop, :normal, {tasks, acc, reduce, reply_to, :none}}
  end
  def handle_info({:DOWN, _ref, :process, pid, reason}, state) do
    handle_down state, pid, reason
  end

  ## callbacks: lifecycle
  def init(arg) do
    GenServer.cast self(), :init   # return ASAP, i.e. delay initialization
    {:ok, arg}
  end

  def terminate(reason, {tasks, acc, _reduce, reply_to, timer_ref}) do
    reply =
      case {reason, timer_ref} do
        {:normal, :none} ->
          {:timeout, self(), acc}
        {:normal, _} ->
          Process.cancel_timer timer_ref
          {:ok, self(), acc}
        _ ->
          {:error, self(), reason}
      end

    cleanup_tasks tasks
    Kernel.send reply_to, reply

    IO.puts "  unprocessed messages - #{inspect (Process.info self(), :message_queue_len)}"
  end

  ## public interface
  def start(data, init, fun, reduce, timeout) do
    args = {data, init, fun, reduce, self(), timeout}
    # Note:  [spawn_opt: :monitor] is not allowed
    # http://erlang.org/doc/man/gen_server.html#start_link-4
    #
    case GenServer.start __MODULE__, args do
      {:ok, pid} ->
        ref = Process.monitor pid
        {pid, ref}
      other ->
        other
    end
  end

  def run_demo(diff \\ 0, timeout \\ 900) do
    data = Enum.map 1..10, fn _ -> Enum.random(1..1000) - diff end
    start data, 0, &Calc.run/1, &Accumulator.sum/2, timeout
  end

end

.

$ iex -S mix
iex(1)> MapReduce.run_demo 700
 sleeping: 6 (#PID<0.120.0>)
 sleeping: 126 (#PID<0.122.0>)
 sleeping: 130 (#PID<0.125.0>)
 sleeping: 152 (#PID<0.127.0>)
 sleeping: 289 (#PID<0.128.0>)
{#PID<0.118.0>, #Reference<0.0.4.573>}
 waking from slumber: 6 (#PID<0.120.0>)
 waking from slumber: 126 (#PID<0.122.0>)
 waking from slumber: 130 (#PID<0.125.0>)
 waking from slumber: 152 (#PID<0.127.0>)
 waking from slumber: 289 (#PID<0.128.0>)
  unprocessed messages - {:message_queue_len, 1}
iex(2)> flush()
{:ok, #PID<0.118.0>, 703}
{:DOWN, #Reference<0.0.4.573>, :process, #PID<0.118.0>, :normal}
:ok
iex(3)> MapReduce.run_demo 0, 500
{#PID<0.131.0>, #Reference<0.0.5.441>}
 sleeping: 203 (#PID<0.132.0>)
 sleeping: 315 (#PID<0.133.0>)
 sleeping: 944 (#PID<0.134.0>)
 sleeping: 959 (#PID<0.135.0>)
 sleeping: 304 (#PID<0.136.0>)
 sleeping: 783 (#PID<0.137.0>)
 sleeping: 407 (#PID<0.138.0>)
 sleeping: 117 (#PID<0.139.0>)
 sleeping: 44 (#PID<0.140.0>)
 sleeping: 564 (#PID<0.141.0>)
 waking from slumber: 44 (#PID<0.140.0>)
 waking from slumber: 117 (#PID<0.139.0>)
 waking from slumber: 203 (#PID<0.132.0>)
 waking from slumber: 304 (#PID<0.136.0>)
 waking from slumber: 315 (#PID<0.133.0>)
 waking from slumber: 407 (#PID<0.138.0>)
 waking from slumber: 564 (#PID<0.141.0>)
 waking from slumber: 783 (#PID<0.137.0>)
 waking from slumber: 944 (#PID<0.134.0>)
 waking from slumber: 959 (#PID<0.135.0>)
  !!! timeout !!!
  kill task #PID<0.134.0>: false
  kill task #PID<0.135.0>: false
  kill task #PID<0.137.0>: false
  kill task #PID<0.141.0>: false
  unprocessed messages - {:message_queue_len, 4}
iex(4)> flush()
{:timeout, #PID<0.131.0>, 1390}
{:DOWN, #Reference<0.0.5.441>, :process, #PID<0.131.0>, :normal}
:ok
iex(5)>

In the end though you shouldn’t spin off processes just because you can. Make sure there is a “good enough reason” for the process to exist. Have a really good, long look at The Erlangelist: To spawn, or not to spawn?

Edit: Added some more IO.puts to reveal more information to help explain behaviour that may seem strange on first blush.

1
Post #5
samba6

samba6 OP

Thanks a lot. That Genserver code looks so nice - something I will be adding to my tool set. And thanks for the advice about spawning processes. I was trying to wrap my head around how concurrency works in elixir, and your advice was timely because I can see myself overdoing it.

Do have a great day!

0
Post #6
samba6

samba6 OP

Thanks @peerreynders, I learnt a lot :joy:

0
Post #7
— All posts loaded —

Where Next?

View thread on forum (has 7 replies!)
Home Questions & Help>Questions
4 3245 7
Last post
REPLY VIEW THREAD WITH 7 POSTS

Trending in Questions Top

jonnycharles
Questions & Help>Questions

Looking for a Prawn-Like PDF Generation Library in Elixir
I’m in search of an Elixir library that offers PDF generation capabilities similar to Ruby’s Prawn. While there have been discussions abo...
#learning-elixir
117 2851 30
New
spammy
Questions & Help>Questions

Simple/light way to self host microapps and services?
I’m looking to build a personal workflow to quickly deploy web applications written in elixir/phoenix, for local consumption (ie not on t...
#deployment
34 1088 21
New
silverdr
Questions & Help>Questions

Using Phoenix.LiveView.TagEngine as an EEx.Engine is deprecated!
Using Phoenix.LiveView.TagEngine as an EEx.Engine is deprecated! To compile HEEx, use Phoenix.LiveView.TagEngine.compile/2 instead. Sta...
/phoenix#liveview
3 821 6
New
dli
Questions & Help>Questions

How to integrate Hologram with existing LiveView app?
Before I dive in myself, did anyone successfully sprinkle Hologram into their existing LiveView app? Looking for hints regarding: Addi...
/hologram#liveview
11 709 6
New
bottlenecked
Questions & Help>Questions

How to do proper post-release steps
Hi all, I wanted to ask how the community is dealing with post-release steps. Today we have Ecto migrations, which make sure that the db...
/phoenix#deployment
13 516 9
New
michallepicki
Questions & Help>Questions

How to start Oban, but delay job processing?
I am using Oban and occasionally, shortly after a deployment, a handful of jobs can fail because of dependency on other parts of the syst...
/oban
6 490 3
New
rahultumpala
Questions & Help>Questions

Help with load testing a custom protocol over TCP
Hello, I have an Elixir backend that implements a custom protocol over TCP. I want to load test the backend and assess the performance o...
#tcp#gen_tcp#loadtesting
1 486 5
New

Other Trending Topics Top

JesseHerrick
News>Announcing

Dexter - A fast, full-featured Elixir LSP optimized for large codebases
Hey, I’m Jesse and I’m the main contributor behind Dexter, a full-featured, lightning-fast Elixir LSP optimized for large codebases. It s...
#library#lsp#dexter
177 5170 56
New
jimsynz
News>Announcing

Beam Bots - Resilient Robotics on the BEAM
Beam Bots (or just BB for short) is a framework for building fault-tolerant robotics applications in Elixir using familiar OTP patterns. ...
/beam-bots#library#robotics#bb
119 2241 27
New
Damirados
News>Announcing

Emerge & Solve - a GUI framework for Elixir
Hello everyone. After busy few months I am happy to announce v0.1.0 of Emerge &amp; Solve. They are GUI (Emerge) and State management (S...
/nerves#library#frameworks#gui#native#state
59 1888 22
New
ausimian
News>Announcing

Emily - A new MLX-based backend for Nx
Emily is an Elixir library that runs Nx computations on Apple’s MLX. Install it as the default Nx backend and Nx, defn, Axon, Nx.Serving,...
#library#libraries#emily
62 1895 19
New
type1fool
Chat & Discussions>Discussions

Elixir-lang.org redesign
I just stumbled on a newly redesigned elixir-lang.org. :tada: It looks like @Software_Mansion did the work, and I think it is generally a...
#marketing
43 1445 11
New
akoutmos
Learning Resources>Books

Programming Nerves (self-published)
@hugobarauna and I (Alex Koutmos) have been hard at work on writing a book on Nerves that takes you from simply blinking LEDs to building...
/nerves/livebook#book#liveview#learning
39 1137 18
New
Back to top

Questions & Help>Questions

Questions Questions ❯

Latest on Elixir Forum

Elixir Forum ❯

Sponsor Spotlight

Dashbit

Build Elixir applications with speed and confidence.

Our Sponsors

Dashbit

Build Elixir applications with speed and confidence.

EEF

Supporting innovation across the BEAM ecosystem.

Coders51

We build reliable cloud platforms for business-critical systems.

AppSignal

Catch errors, track performance, monitor hosts and more.

Honeybadger

Error tracking, logs, uptime, and Just Enough APM. Built for developers.

Pragmatic Bookshelf

Practical resources that improve the lives of professional developers.

ElixirCasts

Producing high quality Elixir screencasts since 2017.

WyeWorks

Enabling companies to succeed by building software people love.

Software Mansion

The team behind Membrane, Popcorn, LiveDebugger. Available for hire.

Pragmatic Studio

Courses that'll move you from confusion to "Aha, now I get it!"

View all our sponsors ❯

Categories:

Sub Categories:

Forums

Popular Tags

Our Sponsors

Dashbit

Build Elixir applications with speed and confidence.

EEF

Supporting innovation across the BEAM ecosystem.

Coders51

We build reliable cloud platforms for business-critical systems.

AppSignal

Catch errors, track performance, monitor hosts and more.

Honeybadger

Error tracking, logs, uptime, and Just Enough APM. Built for developers.

Pragmatic Bookshelf

Practical resources that improve the lives of professional developers.

ElixirCasts

Producing high quality Elixir screencasts since 2017.

WyeWorks

Enabling companies to succeed by building software people love.

Software Mansion

The team behind Membrane, Popcorn, LiveDebugger. Available for hire.

Pragmatic Studio

Courses that'll move you from confusion to "Aha, now I get it!"

View all our sponsors ❯

We're in Beta

About us Mission Statement

Options

Show Best Posts Skip Thread Previews