Processes and Message Passing

What you'll build

A price checker that queries several (simulated) suppliers at the same time in separate processes, collects replies with receive, applies a timeout, and then does the same with Task.async_stream.

The idea

A BEAM process is not an OS thread. It's tiny (a few kilobytes), starts in microseconds, has its own memory, and communicates only by messages. A single machine can run millions of them; the scheduler spreads them across all CPU cores.

• spawn(fn -> ... end) starts a process and returns its pid.
• send(pid, message) puts a message in that process's mailbox.
• receive do pattern -> ... after ms -> ... end takes the first matching message, or times out.
• self() is the current process's pid (include it so the other side can reply).
• Task wraps the common patterns: Task.async/await for one concurrent job, Task.async_stream to process a collection concurrently with a limit on how many run at once.

Because processes share nothing, a crash in one can't corrupt another's memory, which is the foundation of Elixir's fault tolerance.

The code

defmodule Supplier do
  # Simulated suppliers: name -> {price, response time in ms}
  @catalog %{"acme" => {12.40, 40}, "globex" => {11.90, 80}, "initech" => {12.10, 500}}

  def quote(name) do
    {price, delay} = Map.fetch!(@catalog, name)
    Process.sleep(delay)
    price
  end

  def names, do: Map.keys(@catalog) |> Enum.sort()
end

parent = self()

for name <- Supplier.names() do
  spawn(fn -> send(parent, {:quote, name, Supplier.quote(name)}) end)
end

collect = fn collect, acc ->
  receive do
    {:quote, name, price} -> collect.(collect, [{name, price} | acc])
  after
    200 -> acc   # stop waiting: slow suppliers miss the deadline
  end
end

quotes = collect.(collect, []) |> Enum.sort()
IO.inspect(quotes, label: "quotes within 200 ms")
{best, price} = Enum.min_by(quotes, &elem(&1, 1))
IO.puts("best: #{best} at #{price}")

results =
  Supplier.names()
  |> Task.async_stream(&{&1, Supplier.quote(&1)}, max_concurrency: 3, timeout: 200, on_timeout: :kill_task)
  |> Enum.map(fn
    {:ok, {name, price}} -> "#{name}: #{price}"
    {:exit, :timeout} -> "a supplier timed out"
  end)

IO.inspect(results, label: "async_stream")

pids = for _ <- 1..10_000, do: spawn(fn -> receive do: (:stop -> :ok) end)
IO.puts("started #{length(pids)} processes; all alive: #{Enum.all?(pids, &Process.alive?/1)}")
Enum.each(pids, &send(&1, :stop))

quotes within 200 ms: [{"acme", 12.4}, {"globex", 11.9}]
best: globex at 11.9
async_stream: ["acme: 12.4", "globex: 11.9", "a supplier timed out"]
started 10000 processes; all alive: true

The 10,000 processes start in a few milliseconds. That's why Elixir apps happily use one process per user connection, per chat room or per background job.

Common mistakes

• Forgetting after in receive, so a missing reply blocks forever.
• Not including self() in a request message, so the other process can't reply.
• Unbounded mailboxes. If messages arrive faster than they're processed, memory grows; apply back-pressure or limits.
• Spawning without supervision for long-lived work; use supervised Tasks or GenServers.

Key points

• Processes are lightweight, isolated and communicate by messages.
• spawn, send, receive ... after (always add a timeout).
• Task.async/await and Task.async_stream for concurrent work with limits.
• Isolation means one crash can't corrupt another process's state.

Further reading: Processes, Task.