Questions & Help>Questions

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

samba6

samba6

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

Showing Posts 1 to 7

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? Top

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

Trending in Questions Top

katta
Questions & Help>Questions

Phoenix Ecto realistic benchmarks
I having some trouble figuring out if I have set myself too strict of standards for my production server. Currently I can handle 75% of r...
/phoenix#ecto#troubleshooting
2 1863 12
New
achenet
Questions & Help>Questions

No Tailwind binary for FreeBSD
Hello, I’m trying to build a basic Phoenix web-app, and I’d like to use Tailwind. However, when I launch mix phx.server, I get an error...
/phoenix#freebsd#tailwindcss#tailwind
11 1073 14
New
peck
Questions & Help>Questions

Load Testing LiveViews With Tsung (or other tools?)
Hi everyone. My team and I have been working on a fairly modest app based around video streaming and chat, but we’ve landed a customer t...
/phoenix#tsung#load-testing#liveview
7 1213 1
New
bradley
Questions & Help>Questions

Are there any generic KMS libraries out there?
I really like the adapter patterns that ecto, nebulex, waffle, etc. use and would love find something similar for a key management servic...
#libraries#hex-packages
1 953 3
New
unaware8150
Questions & Help>Questions

What could be a clean way to handle constant strings across a codebase?
Hello folks! So at work, we are seeing some situations where we have to define some “fixed” strings that are used across the codebase in...
#learning-elixir
22 559 20
New
Cxx-mlr
Questions & Help>Questions

Using try/catch with update_in/3
I’m working on a small exercise involving update_in/3, and I came up with this solution: data = %{ name: "Periodic Table", category:...
#learning-elixir
9 273 12
New
Alvinkariuki
Questions & Help>Questions

How can I optimise compile time dependencies?
How Can I Optimise Compile Time Dependencies I have been building an elixir application for about 2 years now. Many modules and files ha...
/phoenix#liveview
6 254 5
New

Other Trending Topics Top

GenericJam
News>Announcing

Mob - Native BEAM / Elixir on Native Mobile
Edit: 2026 May 15 - This post is archived. Mob is alive!! Main docs: mob v0.7.11 — Documentation A bit of explanation for the slightly c...
/mob#library#beam#mobile#liveview#react-native#mobile-app#mobile-development
259 8396 111
New
garrison
News>Announcing

Hobbes - a low-level distributed database for the Elixir programming language
Hobbes is a low-level distributed database for the Elixir programming language. Hobbes provides a simple, safe, and scalable storage lay...
#library#database#distributed#hobbes
306 6242 88
New
budgie
Chat & Discussions>Discussions

I genuinely hate programming now
A little off-topic, but I feel like people here have a good head on their shoulders. I used to be quite good at making software. Was luc...
#ai#career#rant
460 3763 113
New
KristerV
Chat & Discussions>AI / LLMs

State of developing agents with Elixir (not coding agents)
Hey. Is there anyone here who creates agents in their apps? Not talking about using agents, but creating them. I’m finding it pretty diff...
#agent#ai#llm#dev-tools#ai-tools
34 4047 22
New
mudasobwa
News>Announcing

Yoke - the DeepSeek Harness (former DSH)
I fully migrated to my own harness from Anthropic/Gemini and I think it’s time to share it. Welcome DSH, the DeepSeek Harness, fully writ...
#library#projects#llm#deepseek#llms
31 4437 36
New
mcass19
News>Announcing

ExRatatui - Elixir bindings for the Rust ratatui terminal UI library
ExRatatui lets you cook up rich terminal UIs in Elixir, powered by Rust’s ratatui via Rustler NIFs. Build interactive terminal applicatio...
#library#rust#terminal#tui
60 3370 16
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.

Code Sync

Elixir, Erlang & BEAM conferences from Code Sync - join us in 2026!

Pragmatic Bookshelf

Practical resources that improve the lives of professional developers.

WyeWorks

**Building the future of software with AI, trust, and human connection.**

Potions

The deployment platform built for Elixir: PaaS ease, VPS control.

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.

Code Sync

Elixir, Erlang & BEAM conferences from Code Sync - join us in 2026!

Pragmatic Bookshelf

Practical resources that improve the lives of professional developers.

WyeWorks

**Building the future of software with AI, trust, and human connection.**

Potions

The deployment platform built for Elixir: PaaS ease, VPS control.

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

Thread Display Mode




Thread Preview

Skip Thread Previews