Elixir Task.async_stream: Bounded Concurrency for I/O-Bound Enumerables

The problem: waiting on I/O without drowning the VM

You have a list of items—user IDs, file paths, or API endpoints—and each one waits on something outside the BEAM. Mapping sequentially is simple but slow. Spawning one Task.async per item is faster in theory, but it can overwhelm schedulers, exhaust file descriptors, or trip remote rate limits. The practical challenge is finding the middle ground: run enough work in parallel to save time, but never so much that the node degrades.

The useful takeaway is that Task.async_stream processes an enumerable concurrently while capping the number of live tasks, reports success or failure for every item without aborting the whole run, and preserves input order unless you turn ordering off.

How Task.async_stream bounds work and backpressures the source

Task.async_stream takes an enumerable and a function. It starts a separate task for each element, but no more than max_concurrency tasks run at the same time. As each task finishes, another is started.

The result is a lazy Stream. It only pulls the next element from the source when a task slot is free. That laziness is the backpressure mechanism: if the consumer is slow or the workers are saturated, the source is not drained faster than the cap allows. This limits memory and scheduler pressure for large inputs.

Results arrive as tuples: {:ok, value} for success and {:exit, reason} for failure. Because failures are captured per item, one crashing worker does not bring down the enumeration. You decide how to handle each tuple downstream.

Worked example: bounded batch with per-item failures

Run this in iex -S mix inside a Mix project, or plain iex for a quick experiment. No special permissions are required. The placeholders below are your input list and worker logic.

items = 1..10

worker = fn id ->
  Process.sleep(:rand.uniform(100))

  if rem(id, 3) == 0 do
    raise "simulated failure for #{id}"
  else
    {:ok, id * 2}
  end
end

Task.async_stream(items, worker, max_concurrency: 3, timeout: 5_000)
|> Stream.each(fn
  {:ok, result} -> IO.puts("ok: #{result}")
  {:exit, reason} -> IO.puts("exit: #{inspect(reason)}")
end)
|> Stream.run()

Expect results in input order because ordered commonly defaults to true. Failures appear as {:exit, reason} tuples rather than crashing the caller. To verify the concurrency bound, run the same stream with max_concurrency: 1 and then with max_concurrency: 10 and compare total wall-clock time. Switching to ordered: false returns results as they complete, which reduces latency but loses sequence guarantees.

Trade-offs: ordering, timeouts, and supervision

Three options change the behavior significantly, and their defaults are version-sensitive. Verify them in your environment with h Task.async_stream before relying on them.

Option	Common default	Effect and trade-off
max_concurrency	Often System.schedulers_online()	Caps simultaneously running tasks. Set explicitly for predictable bounds.
timeout	Often 5000 ms	Tasks exceeding this are terminated and reported as {:exit, :timeout}. Increase for slow I/O.
ordered	Often true	Preserves input order by buffering results. Set to false to stream results as they finish and avoid buffering cost.

When you need clean shutdown guarantees, pass a Task.Supervisor via the task_supervisor option. The tasks become dynamically supervised processes under that supervisor, so stopping the supervisor terminates in-flight work cleanly. This differs from static child specifications in a supervision tree.

Keep these limitations in mind:

• The default timeout is commonly 5,000 milliseconds, but this is version-sensitive. Long-running tasks are killed and reported as {:exit, :timeout} unless you increase timeout:.
• ordered: true buffers results so sequence is preserved. A fast producer and slow consumer can accumulate buffered items in memory.
• This is best for I/O-bound work. CPU-bound tasks can saturate BEAM schedulers and may not benefit from higher concurrency.
• While the lazy stream backpressures the source, it does not provide the multi-stage demand signaling of GenStage or Flow. Use those when you need complex multi-consumer pipelines.

A practical check before you ship it

1. Start iex -S mix.
2. Run h Task.async_stream to confirm the options and defaults available in your Elixir version.
3. Run your worker with a mix of fast and slow items, then compare ordered: true against ordered: false for output order and latency.
4. Observe the node with :erlang.system_info(:process_count) or Observer while increasing max_concurrency to confirm the live task count stays near your cap.

Start with a modest max_concurrency, such as 10 to 20 for typical I/O work, and tune upward only if the node has headroom. Match on {:ok, _} and {:exit, _} for every item, and rely on the lazy stream to keep the source from being drained faster than your cap allows.