Using Task.async_stream for Bounded Concurrent Processing in Elixir

Use Task.async_stream for Bounded Concurrency

When processing a large or lazy collection—such as URLs to fetch or files to parse—spawning a task per item with Enum.map(&Task.async/1) creates all processes at once and can exhaust resources. Task.async_stream/3 returns a lazy stream that starts only as many tasks as you allow, enforcing a maximum concurrency window and applying per‑task timeouts.

How it works

The function does not start work until you consume the returned stream (e.g., with Enum.to_list/1 or Enum.reduce/3). While consuming, it keeps at most :max_concurrency tasks alive. As a task finishes, its result is emitted and the stream pulls the next item to fill the slot. By default results follow input order; set ordered: false to emit results as soon as each task finishes.

Example: limited HTTP fetches

defmodule Fetcher do
  def fetch_all(urls) do
    urls
    |> Task.async_stream_nolink(&fetch_one/1, max_concurrency: 4, timeout: 6_000, ordered: false)
    |> Enum.reduce(%{ok: %{}, errors: []}, fn
      {:ok, resp}, acc -> %{acc | ok: Map.put(acc.ok, resp.url, resp.body)}
      {:exit, reason}, acc -> %{acc | errors: [reason | acc.errors]}
    end)
  end

  defp fetch_one(url) do
    # placeholder: use Req or HTTPoison in real code
    Process.sleep(:rand.uniform(4000))
    %{url: url, body: ""}
  end
end

The example uses Task.async_stream_nolink/3 so a failing task returns {:exit, reason} instead of crashing the caller. max_concurrency: 4 limits simultaneous requests, and timeout: 6_000 kills any request longer than six seconds, yielding {:exit, :timeout}.

Checking the limit

Add IO.puts("Starting #{url}") inside fetch_one/1 and run iex -S mix. You should observe no more than four “Starting” lines before any “Finished”‑style output, showing that new tasks start only when a slot frees up.

Limits

• Default timeout is 5000 ms; omitting :timeout gives {:exit, :timeout} after five seconds.
• The linked variant Task.async_stream/3 ties each task to the caller; an unhandled exit crashes the calling process. Use the _nolink version or trap exits to isolate failures.
• The stream is lazy: no work happens until you enumerate it.
• With ordered: true (default) results follow input order, which can cause head‑of‑line blocking; ordered: false improves throughput when task times vary.

Common mistakes

1. Matching only {:ok, _} in the reducer. A {:exit, reason} tuple causes a MatchError and crashes the process.
2. Assuming completion‑order results. ordered: true preserves input order; ordered: false yields finish order.
3. Hand‑rolling unbounded concurrency with Enum.map(&Task.async/1) followed by Task.await_many—this spawns all processes at once.
4. Setting max_concurrency higher than the downstream service can handle.
5. Doing heavy work inside the reducer; it runs in the caller’s process and stalls the stream.

When to pick another tool

• Two to five known calls: Task.async/await is simpler.
• Continuous event streams with back‑pressure: consider GenStage or Broadway.
• Long‑lived stateful workers: a pool of GenServer processes under a Supervisor.

Verification

Run h Task.async_stream in IEx to see the exact option names and defaults for your Elixir version.