~/adam.log
Back to posts
elixir concurrency otp task supervisor fault-tolerance async task-async-stream task-supervisor processes process-linking let-it-crash supervision-trees iex concurrent-data-processing-in-elixir book-notes learning-in-public beginner

Concurrent Data Processing in Elixir - Chapter 1

Published 2026-09-20

1. Easy Concurrency with the Task Module

From the dawn of time we have tried to make tasks easier and faster. Everything was just trying to make the single processor faster and not running things at the same time. When we got multi-core processors everything changed. We now have the ability to have 2 processors running different tasks at the same time.

Before there was a way of doing things that was start and stop of different tasks, this is like threading. Where you can start A and then start B and then process A then process B then finish A then Finish B. When we get to true multi-core we can do both at the same time.

Introducing the Task Module

In order to do most anything in Elixir you need to spawn a process and then execute the code. spawn/1 and receive will be used to start and get data from the processes. There is also a Task module that you can use to reduce the repetitive code.

Creating Our Playground

Okay here is where the rubber meets the road let’s create a new project.

mix new sender --sup

We now created a simple project with a supervision tree. Go ahead and get into the new project.

cd sender
mix -s mix

Once inside we have a few things that we can do but 

recompile()
# can be used to pull any changes to the file.

We will be using some sudo code and Process.sleep/1 in order to simulate the process of sending emails. Okay let’s get into the sender.ex file and add in some logic.

  def send_email(email) do
    Process.sleep(3000)
    IO.puts("Email to #{email} sent")
    {:ok, "email_sent"}
  end

If we where to go into our process and then run the function we will “send and email” and then get an output and then a return of {:ok, “email_sent”}

Starting Tasks and Retrieving Results

Okay so start up the application or recompile() and try this command.

Sender.send_email("hello@world.com")
{:ok, "email_sent"}

Was there a delay? There should have been. Now we can try and do more than 1 email and see what happens. There should be message and delays for each one. Add this code to the sender.ex

  def notify_all(emails) do
    Enum.each(emails, &send_email/1)
  end

Let us all add in a new file to the main directory called .iex.ex. Add the following to the file.

emails = [
    "hello@world.com",
    "hola@world.com",
    "nihao@world.com",
    "konnichiwa@world.com",
]

We can now exit and restart the iex session and then run these commands.

iex(1)> emails
["hello@world.com", "hola@world.com", "nihao@world.com", "konnichiwa@world.com"]

This will save a lot of typing as anything that is within the .iex.exs will be run when we start the application. So let’s try out the new list of emails and the new notify_all/1

iex(2)> Sender.notify_all(emails)
Email to hello@world.com sent
Email to hola@world.com sent
Email to nihao@world.com sent
Email to konnichiwa@world.com sent
:ok

There should have been a delay before each email that was sent… That is not fast so let’s make it so.

Synchronous and Asynchronous Code

Okay so there is a few bit of terminology that we should go over here so we can all be on the same page.

  1. synchronously or blocking where you can only do one thing at at time.
  2. asynchronously or non-blocking where things can be done in the background.

There is a little image that does into the baking of a cake, you can mix the batter and then preheat the oven at the same time. Or you can mix the batter and then preheat the oven.

Starting Processes

We will start to use the Task module with the iex session. We will eventually start to use the functions and more to automate these things but let’s get back into our iex session.

iex(3)> Task.start(fn -> IO.puts("Hello async world!") end)
Hello async world!
{:ok, #PID<0.146.0>}

That was instant and it didn’t give the normal {:ok, message} this send back a PID (process identifier). Okay so we now have a way of starting a process and having it do something for us. Let’s update a change to the notify_all/1

  def notify_all(emails) do
    Enum.each(emails, fn email ->
      Task.start(fn ->
        send_email(email)
      end)
    end)
  end

recompile() and then let’s try that notify_all() within the session.

iex(5)> Sender.notify_all(emails)
:ok
Email to hello@world.com sent
Email to hola@world.com sent
Email to nihao@world.com sent
Email to konnichiwa@world.com sent

We still had the same delay to start but man every email fired off at the same time.

Retrieving the Result of a Task

Right now the we have the ability to start a task but we don’t have a way of getting any value from that task. We need a way to give a task a variable that we can match to the task. That is where Task.async/1 comes in we can use this and assign it to a variable. Let’s try it out in a session.

iex(6)> Task.async(fn -> Sender.send_email("hello@world.com") end)
%Task{
  mfa: {:erlang, :apply, 2},
  owner: #PID<0.145.0>,
  pid: #PID<0.168.0>,
  ref: #Reference<0.0.18563.3059740491.4026597379.38657>
}
Email to hello@world.com sent

Let’s go over this.

  • owner is the owner of the PID
  • pid is the id of the PID itself
  • ref process monitor reference

Now that we have that we can work with Task.await/1 or Task.yield/1 these will need to have a Task as an argument. They both will stop a process and try to get information but they will do something different with process timeouts. We can test out what happens with changing the sleep amount with send_email/1

  def send_email(email) do
    Process.sleep(30_000)
    IO.puts("Email to #{email} sent")
    {:ok, "email_sent"}
  end

Now let’s try to use the async() and await()

iex(8)> Task.async(fn -> Sender.send_email("hi@world.com") end) |> Task.await()
** (exit) exited in: Task.await(%Task{mfa: {:erlang, :apply, 2}, owner: #PID<0.145.0>, pid: #PID<0.185.0>, ref: #Reference<0.0.18563.3059740491.4026597394.38751>}, 5000)
    ** (EXIT) time out
    (elixir 1.18.0) lib/task.ex:888: Task.await_receive/3
    iex:8: (file)

This was after 5 seconds it timed out. Now we can test the yield/1 and test more than once while it is processing.

iex(8)> task = Task.async(fn -> Sender.send_email("hi@world.com") end)
%Task{
  mfa: {:erlang, :apply, 2},
  owner: #PID<0.145.0>,
  pid: #PID<0.186.0>,
  ref: #Reference<0.0.18563.3059740491.4026597394.38772>
}
iex(9)> Task.yield(task)
nil
iex(10)> Task.yield(task)
nil
iex(11)> Task.yield(task)
nil
Email to hi@world.com sent
iex(12)> Task.yield(task)
{:ok, {:ok, "email_sent"}}

You can see that it will try for about 5 seconds and then just send back nil if it doesn’t get a response. After it is done it will give you the value of the return statement. Now what happens if the task never actually finishes. Well that is where we can start to use, Task.shutdown(task) this is where we can send it a PID and it will kill it for us.

Let’s start to work with these new functions. Revert the sleep back to 3000 and make these changes to the notify_all/1

  def notify_all(emails) do
    emails
    |> Enum.map(fn email ->
      Task.async(fn ->
        send_email(email)
      end)
    end)
    |> Enum.map(&Task.await/1)
  end

Try it out in the iex session.

iex(15)> Sender.notify_all(emails)
Email to hello@world.com sent
Email to hola@world.com sent
Email to nihao@world.com sent
Email to konnichiwa@world.com sent
[ok: "email_sent", ok: "email_sent", ok: "email_sent", ok: "email_sent"]

Managing Series of Tasks

Okay so now think about if we had 1000_000 users and we need to send an email to everyone of them. The _Task.async/1 would work but it would create a huge amount of strain on the system. We can use Task.async_stream/3 what is great about this new function is that you can do exactly what you could do with async but you can now set a limit on the number of concurrent processes. Let’s test it out in an iex session.

iex(1)> Task.async_stream(emails, &Sender.send_email/1)
#Function<3.119398607/2 in Task.build_stream/3>
iex(2)> # That built a function instead of running the code. Let's look at an other example to see what we mean
nil
iex(3)> Stream.map([1,2,3], & &1 *2)
#Stream<[enum: [1, 2, 3], funs: [#Function<49.118167795/1 in Stream.map/2>]]>

Both of these returned a function and the initial inputs. So if you want to run them you can use Stream.run/1 but that only returns :ok. We can use Enum.to_list in order to put the results into a list.

Okay with that in mind let’s update the notify\all/1 to use the new way of doing things.

  def notify_all(emails) do
    emails
    |> Task.async_stream(&send_email())
    |> Enum.to_list()
  end

And here is us using it in the iex session.

iex(6)> Sender.notify_all(emails)
Email to hello@world.com sent
Email to hola@world.com sent
Email to nihao@world.com sent
Email to konnichiwa@world.com sent
[
  ok: {:ok, "email_sent"},
  ok: {:ok, "email_sent"},
  ok: {:ok, "email_sent"},
  ok: {:ok, "email_sent"}
]

We just talked about how depending on how many processes that we allow it will take more or less time. By default it will limit itself to the number of logical cores in the system you have. Let’s do some tests with limits to see the changes.

|> Task.async_stream(&send_email/1, max_concurrency: 1)
# With this change every email is sent 1 by 1
|> Task.async_stream(&send_email/1, ordered: false)
# This will send the emails in any order. 
|> Task.async_stream(&send_email/1, on_timeout: :kill_task)
# This will kill the process if any one of the tasks takes longer than the standard of 
# 5000ms

Linking Processes

For this we will need a way to link the processes. That way if one process A dies and it would have reported back to an other process B the process B wont be waiting forever if they are linked and B knows what is happening to A.

Now depending on the way the processes are set up you can want the server to terminate depending on the error. Looking at the banking image we can see that if the Financial node breaks we need to stop the children Accounts and Transfers, but if the Sender breaks we don’t need to worry about the Repo node.

You can isolate the crashes but configuring a trap exit. This means that you are setting up a way to receive the exit message of a linked process but you will continue to run anyway waiting for the process to come back online.

Now to go back to our example async and async_stream a process link was crated, Task.start doesn’t create a link, Task.start_link does create a link.

There is also a way to append _nolink in order to make sure that no link is created.

Meeting the Supervisor

Okay here is where the rubber meets the road. We talked about Processes we talked about links we talked about trapping exits. Now we can automate all of this and set up what happens when we have a crash. In Elixir we have a Supervisor and it’s children. We will go to the application.ex and set up the supervisor now.

defmodule Sender.Application do
  # See https://hexdocs.pm/elixir/Application.html
  # for more information on OTP Applications
  @moduledoc false

  use Application

  @impl true
  def start(_type, _args) do
    children = [
      # Starts a worker by calling: Sender.Worker.start_link(arg)
      # {Sender.Worker, arg}
    ]

    # See https://hexdocs.pm/elixir/Supervisor.html
    # for other strategies and supported options
    opts = [strategy: :one_for_one, name: Sender.Supervisor]
    Supervisor.start_link(children, opts)
  end
end

That is the default setup when we create a new supervisor application. Let’s start to tweak it.

Adding a Supervisor

There is some boiler plate supervisors that we can add to the system in order to get off the ground running. The first one is Task.Supervisor This is meant to work with Tasks‘s. Let’s add that one to the children that the application’s supervisor will over see.

  @impl true
  def start(_type, _args) do
    children = [
      {Task.Supervisor, name: Sender.EmailTaskSupervisor}
    ]

    # See https://hexdocs.pm/elixir/Supervisor.html
    # for other strategies and supported options
    opts = [strategy: :one_for_one, name: Sender.Supervisor]
    Supervisor.start_link(children, opts)
  end

This is one of hte smallest definitions of a child that we could use. We could also define it as a map with more information.

  @impl true
  def start(_type, _args) do
    children = [
      %{
        id: Sender.EmailTaskSupervisor,
        start: {
          Task.Supervisor,
          :start_link,
          [[name: Sender.EmailTaskSupervisor]]
        }
      }
    ]

    # See https://hexdocs.pm/elixir/Supervisor.html
    # for other strategies and supported options
    opts = [strategy: :one_for_one, name: Sender.Supervisor]
    Supervisor.start_link(children, opts)
  end

This is different than the last one. This is more verbose and will allow us to build more complex and leverage starting config and more.

Using Task.Supervisor

Before we start into the world of using the Supervisor let’s see what happens when we run into issues without one. Let’s change the sender.ex to the following code.

  def send_email("konnichiwa@world.com" = emai),
    do: raise("Opps, couldn't send email to #{email}!")

  def send_email(email) do
    Process.sleep(3000)
    IO.puts("Email to #{email} sent")
    {:ok, "email_sent"}
  end

Start up the iex session, looking at the code we can see a pattern match for an email that shows that when we get that email we will raise an exception.

iex(10)> self()
#PID<0.145.0>
iex(11)> # That is the pid of the child
nil
iex(12)> Sender.notify_all(emails)

18:53:29.038 [error] Task #PID<0.233.0> started from #PID<0.145.0> terminating
** (RuntimeError) Opps, couldn't send email to konnichiwa@world.com!
    (sender 0.1.0) lib/sender.ex:7: Sender.send_email/1
    (elixir 1.18.0) lib/task/supervised.ex:101: Task.Supervised.invoke_mfa/2
    (elixir 1.18.0) lib/task/supervised.ex:36: Task.Supervised.reply/4
Function: &:erlang.apply/2
    Args: [#Function<0.51129359/1 in Sender.send_email>, ["konnichiwa@world.com"]]
** (EXIT from #PID<0.145.0>) shell process exited with reason: an exception was raised:
    ** (RuntimeError) Opps, couldn't send email to konnichiwa@world.com!
        (sender 0.1.0) lib/sender.ex:7: Sender.send_email/1
        (elixir 1.18.0) lib/task/supervised.ex:101: Task.Supervised.invoke_mfa/2
        (elixir 1.18.0) lib/task/supervised.ex:36: Task.Supervised.reply/4

Interactive Elixir (1.18.0) - press Ctrl+C to exit (type h() ENTER for help)
iex(12)> self()
#PID<0.234.0>
iex(13)> # it crashed and now we have a new pid

Okay let’s update the code so that we have no link to between the processes.

  def notify_all(emails) do
    Sender.EmailTaskSupervisor
    |> Task.Supervisor.async_stream_nolink(emails, &send_email/1)
    |> Enum.to_list()
  end

Now we are leveraging the supervisor in order to have a child and then we are not linking the processes together just allowing the exit message to be propagated up.

ex(1)> Sender.notify_all(emails)

18:57:52.247 [error] Task #PID<0.151.0> started from #PID<0.146.0> terminating
** (RuntimeError) Opps, couldn't send email to konnichiwa@world.com!
    (sender 0.1.0) lib/sender.ex:7: Sender.send_email/1
    (elixir 1.18.0) lib/task/supervised.ex:101: Task.Supervised.invoke_mfa/2
    (elixir 1.18.0) lib/task/supervised.ex:36: Task.Supervised.reply/4
Function: &:erlang.apply/2
    Args: [#Function<0.125847234/1 in Sender.send_email>, ["konnichiwa@world.com"]]
Email to hello@world.com sent
Email to hola@world.com sent
Email to nihao@world.com sent
[
  ok: {:ok, "email_sent"},
  ok: {:ok, "email_sent"},
  ok: {:ok, "email_sent"},
  exit: {%RuntimeError{
     message: "Opps, couldn't send email to konnichiwa@world.com!"
   },
   [
     {Sender, :send_email, 1,
      [file: ~c"lib/sender.ex", line: 7, error_info: %{module: Exception}]},
     {Task.Supervised, :invoke_mfa, 2,
      [file: ~c"lib/task/supervised.ex", line: 101]},
     {Task.Supervised, :reply, 4, [file: ~c"lib/task/supervised.ex", line: 36]}
   ]}
]

Understanding Let It Crash

This is the biggest part of the Erlang and Elixir system. We are not saying that we are not dealing with the errors or that we are trying to make code that doesn’t work right we are allowing the system to fail in order to better understand the issues. You have the ability through logs to really understand the errors that you are getting and still keep the rest of the site up and running.

We have the ability to make sure that processes and nodes are appropriately linked so that only the things we want to go down do.

There are a few different types of restart values that mean something to us.

  • :temporary will never restart child process
  • :transient will restart child process but only when they exit with error
  • :permanent always restarts children keeping them running, even when they try to shut down without an error.

Wrapping Up

You have come a long way and you will still learn a lot for this setup. Keep in mind that Task is meant for single use processes and you will want a robust Supervisor tree for long running applications.