Bounded Concurrency in Elixir: A Practical Guide to Task.async_stream

The problem: too slow or too wild

You have a list of API endpoints—maybe two thousand of them—and you need to fetch JSON from each one. A simple Enum.map is easy to read, but it is strictly sequential: every request waits for the previous one to finish. If each call takes 200 ms, the whole job takes almost seven minutes.

The obvious fix is to run them all at once. In Elixir that often means Task.async for every URL. That works for ten items. For two thousand, it spawns two thousand processes simultaneously, opens two thousand sockets, and can trip rate limits or exhaust file descriptors. You need something in between: a fixed pool of workers that pull from the list, with results flowing back as they complete. That is bounded concurrency, and Task.async_stream is the idiomatic way to get it.

What Task.async_stream actually does

Task.async_stream/3 takes an enumerable and a function. It returns a Stream that emits one item per input element. Under the hood, it starts up to :max_concurrency linked tasks. When a task finishes, the next element is dispatched. Because it is a stream, you can consume results lazily with Enum or Stream functions, and the producer side respects your consumption rate—this is the backpressure.

Each emitted item is a tuple:

• {:ok, result} when the task completed normally.
• {:exit, reason} when the task crashed, was killed, or exceeded the :timeout.

Key options include :max_concurrency, :timeout (per task), and :on_timeout. The exact set of supported options depends on your OTP version; OTP 24 and newer provide the most complete behavior. Confirm your environment before relying on newer flags.

A worked example: fetching JSON with a worker cap

The following snippet assumes you have an HTTP client such as HTTPoison and a JSON decoder such as Jason available. The pattern applies to any client.

urls = [
  "https://api.example.com/records/1",
  "https://api.example.com/records/2"
  # ... potentially thousands more
]

results =
  Task.async_stream(urls, fn url ->
    case HTTPoison.get(url, [], timeout: 5_000, recv_timeout: 5_000) do
      {:ok, %{status_code: 200, body: body}} ->
        {:ok, Jason.decode!(body)}

      {:ok, %{status_code: code}} ->
        {:error, {:unexpected_status, code}}

      {:error, reason} ->
        {:error, reason}
    end
  end,
    max_concurrency: 5,
    timeout: 10_000
  )
  |> Enum.reduce(%{ok: [], errors: []}, fn
    {:ok, {:ok, data}}, acc ->
      %{acc | ok: [data | acc.ok]}

    {:ok, {:error, reason}}, acc ->
      %{acc | errors: [{:fetch_failed, reason} | acc.errors]}

    {:exit, reason}, acc ->
      %{acc | errors: [{:task_exit, reason} | acc.errors]}
  end)

Here max_concurrency: 5 means at most five HTTP requests are in flight at any moment. The timeout: 10_000 applies to each individual task; if a request hangs longer than ten seconds, the task is terminated and the stream yields {:exit, reason} instead of silently blocking the pipeline.

Trade-offs and tuning

Task.async_stream is not free. It spawns one process per list element. For very large inputs—hundreds of thousands or millions of items—you may approach the BEAM process limit or create unnecessary garbage. If your list is that large, consider chunking the input or moving to a demand-driven pipeline such as GenStage.

Throughput is also bounded by your :max_concurrency setting. If the remote API is the bottleneck, raising the cap may help until you hit rate limits. If the bottleneck is local CPU parsing, a cap above System.schedulers_online() will not magically create more cores. Start with a fraction of available schedulers, measure, and adjust.

Timeout values must match reality. Setting :timeout below the expected request duration will kill healthy work and surface it as an exit tuple. Log those exits so you can distinguish network failures from configuration mistakes.

How to verify it is working

Before you tune, confirm your baseline. Run elixir -v in your terminal to see the installed Elixir and OTP versions. Then instrument your HTTP client to log a timestamp when each request starts and ends; overlapping intervals prove that concurrency is active.

To check that resources are reclaimed, capture the process count before and after the stream finishes:

before = :erlang.system_info(:process_count)
# ... run your async_stream ...
after = :erlang.system_info(:process_count)
IO.puts("Processes before: #{before}, after: #{after}")

If after remains significantly higher than before, tasks may be leaking. In normal operation the count should return close to its starting value once the stream is fully consumed.

Bottom line

Use Task.async_stream when you need to process a list concurrently but cannot afford unbounded fan-out. Set :max_concurrency deliberately, handle both {:ok, result} and {:exit, reason}, and verify that your timeouts match your workload. It turns a fragile "spawn everything" script into a predictable, observable pipeline.