Elixir: Tasks & Agents

Most concurrent work in Elixir does not need a hand-rolled spawn/receive loop. Task wraps one-off and pooled computation with timeouts and error propagation; Agent wraps a simple state holder. Both are thin, well-tested abstractions over the process model.

Task.async / await

# Fire concurrent HTTP fetches; await returns results in order.
results =
  ["https://elixir-lang.org", "https://hex.pm"]
  |> Enum.map(&Task.async(fn -> fetch(&1) end))   # starts immediately
  |> Enum.map(&Task.await(&1, 5_000))             # 5s timeout each

# async/await errors are LINKED: if the task crashes, the caller dies
# too. For "do not take me down with you", use Task.Supervised:
task = Task.Supervisor.async(MyApp.TaskSup, fn -> risky() end)
case Task.yield(task, 5_000) || Task.shutdown(task) do
  {:ok, result} -> {:ok, result}
  {:exit, reason} -> {:error, reason}
  nil -> {:error, :timeout}        # shutdown() returns nil after a timeout
end

Task.async_stream: Concurrency with a Limit

Processing 10,000 jobs needs bounded concurrency — spawning all of them at once would overwhelm the CPU or the remote service. Task.async_stream runs a pool of N tasks with timeouts, lazily, with back-pressure.

urls
|> Task.async_stream(&fetch/1,
      max_concurrency: 20,          # at most 20 in flight
      timeout: 10_000,              # per-task timeout
      on_timeout: :kill_task,       # the slow one dies, the run continues
      order: false)                 # results arrive AS COMPLETED (faster)
|> Enum.reduce({[], []}, fn
      {:ok, result}, {ok, err} -> {[result | ok], err}
      {:exit, reason}, {ok, err} -> {ok, [reason | err]}
    end)

Agents: Simple State

When the whole job is "hold a value, update it atomically", an Agent is a GenServer with a friendlier API. The functions run inside the agent process, so reads and writes serialize naturally.

{:ok, counter} = Agent.start_link(fn -> 0 end)

Agent.get(counter, & &1)                        #=> 0
Agent.update(counter, &(&1 + 1))
Agent.get_and_update(counter, fn n -> {n, n * 2} end)  # get & set in one
Agent.get(counter, & &1)                        #=> 2

Choosing the Tool

Need	Use	Why
One computation, concurrent with others	`Task.async/await`	results as values, timeouts included
Map a job list over N workers	`Task.async_stream`	bounded concurrency + back-pressure built in
Hold and update a small value	`Agent`	atomic get/update without writing a server
State with several operations, invariants, or timers	`GenServer`	full control of the message protocol
Shared read-mostly data	ETS / `:persistent_term`	no process bottleneck — see Performance

Supervising Background Work

Raw Task.async tasks are linked to the caller — right for "answer me", wrong for "fire and forget". Background work belongs under a Task.Supervisor so a crash is observed and restartable rather than propagated into a request.

# In the application supervision tree (next lessons):
# {Task.Supervisor, name: MyApp.TaskSup}

# Fire-and-forget with a supervisor watching:
Task.Supervisor.start_child(MyApp.TaskSup, fn ->
  Logger.info("expensive cleanup starting")
  cleanup()
end)

Practice

Fetch three URLs concurrently with Task.async, then again with Task.async_stream(max_concurrency: 2); compare the timing.
Build an Agent-backed key/value counter and add a reset/0 using Agent.get_and_update/2.
Prove that a crash inside an unsupervised Task.async kills the caller, then repeat it under Task.Supervisor and observe the difference.

Next: GenServer