Skip to content

Async Generators vs Queues for Streaming Pipelines

Both shapes appear in every async codebase that processes a stream of items, and they are not interchangeable. An async generator chains stages with async for, giving lock-step backpressure and readable code: async for row in parse(read(path)). An asyncio.Queue decouples stages into independent tasks, allowing several consumers to work concurrently on the same stream. Choosing wrongly produces one of two predictable failures — a generator pipeline that cannot use more than one CPU-second of concurrency because every stage waits for the one before it, or a queue pipeline whose stages silently accumulate unbounded backlog because nobody set maxsize.

The decision comes down to one question: does any stage need concurrency within itself? This guide walks through both shapes, their backpressure and cleanup semantics, the hybrid that most real pipelines end up using, and the specific traps each one hides.

Prerequisites

Two ways to build a pipeline Two columns contrasting generator and queue pipeline shapes. Two ways to build a pipeline async generators pull-based, sequential Stages chain with async for Backpressure is implicit Memory is O(1) per stage One item in flight, end to end queues + tasks push-based, concurrent Stages are independent tasks Backpressure needs maxsize Memory is O(maxsize) N workers per stage The deciding question: does any stage need concurrency within itself?

1. The generator pipeline: lock-step and readable

Each stage pulls from the one before it. Backpressure is implicit and exact.

import asyncio
from collections.abc import AsyncIterator


async def read_lines(path: str) -> AsyncIterator[bytes]:
    with open(path, "rb") as fh:
        for line in fh:
            yield line
            await asyncio.sleep(0)                 # stay cooperative


async def parse(source: AsyncIterator[bytes]) -> AsyncIterator[dict]:
    async for line in source:
        if line.strip():
            yield json.loads(line)


async def enrich(source: AsyncIterator[dict]) -> AsyncIterator[dict]:
    async for record in source:
        record["region"] = REGIONS.get(record["zone"], "unknown")
        yield record


async def run(path: str) -> None:
    async for record in enrich(parse(read_lines(path))):
        await sink.write(record)

Nothing buffers: a stage produces one item only when the next stage asks for it, so memory is O(1) regardless of stream length and the slowest stage sets the pace naturally. The cost is that everything is sequential — while sink.write() awaits, no parsing happens, even though the two are entirely independent. Verify: measure RSS across a multi-gigabyte input; it should stay flat.

2. The queue pipeline: concurrency between and within stages

Stages become tasks connected by bounded queues, so several consumers can work at once.

import asyncio


async def producer(path: str, out: asyncio.Queue) -> None:
    async for line in read_lines(path):
        await out.put(line)                        # suspends when full: backpressure
    for _ in range(WORKERS):
        await out.put(None)                        # one sentinel per consumer


async def worker(inp: asyncio.Queue, out: asyncio.Queue) -> None:
    while (line := await inp.get()) is not None:
        record = json.loads(line)
        record["extra"] = await lookup(record["id"])    # concurrent across workers
        await out.put(record)
    await out.put(None)


async def run(path: str) -> None:
    raw: asyncio.Queue = asyncio.Queue(maxsize=1000)
    done: asyncio.Queue = asyncio.Queue(maxsize=1000)
    async with asyncio.TaskGroup() as tg:
        tg.create_task(producer(path, raw), name="produce")
        for i in range(WORKERS):
            tg.create_task(worker(raw, done), name=f"work.{i}")
        tg.create_task(writer(done), name="write")

The gain is real concurrency: with WORKERS=16, sixteen lookup() calls are in flight while the producer keeps reading. The costs are the bounded buffers (memory you must size deliberately) and the shutdown protocol — one sentinel per consumer, or a separate mechanism, because a queue has no natural end-of-stream. Verify: throughput should scale with WORKERS up to the point where lookup()'s dependency saturates.

A slow sink, seen from each shape Three lanes showing backpressure behaviour under a slow sink. A slow sink, seen from each shape generator read parse sink — everything waits queue, unbounded producer races ahead: backlog grows queue, maxsize set fills to maxsize producer suspends on put() time, with a sink slower than the source Both shapes can apply backpressure — only one does so without being configured.

3. Compare the properties that actually differ

Five differences decide almost every case.

Property Async generator Queue + tasks
Backpressure Implicit, exact (pull-based) Explicit, via maxsize
Memory O(1) per stage O(maxsize) per queue
Concurrency within a stage None — strictly sequential N workers per stage
End of stream StopAsyncIteration, automatic Sentinels or join(), manual
Cleanup finally runs on aclose() Task cancellation, explicit
Error propagation Raises into the consumer Must be routed deliberately

The single decisive question is the third row: if a stage awaits I/O that could usefully run concurrently across items — an HTTP lookup, a database query — the generator shape leaves that concurrency on the floor. If every stage is CPU-light and sequential, the generator is simpler, safer and uses less memory. Verify: measure the per-item wait in your slowest stage; if it is dominated by I/O, the queue shape will beat the generator by roughly the concurrency factor.

4. Get cleanup right in both shapes

Each has one cleanup trap, and they are different.

import asyncio
from contextlib import aclosing


# Generators: an early break leaves the generator suspended with its finally pending
async def consume_some(source) -> list:
    out = []
    async with aclosing(source) as stream:         # guarantees aclose() -> finally runs
        async for item in stream:
            out.append(item)
            if len(out) >= 10:
                break                              # safe now
    return out


# Queues: cancellation must not lose the in-flight item or skip task_done()
async def worker(queue: asyncio.Queue) -> None:
    while True:
        item = await queue.get()
        try:
            await handle(item)
        finally:
            queue.task_done()                      # every path, including cancellation

For generators, contextlib.aclosing() is the fix for the most common leak: a break (or an exception) out of an async for leaves the generator parked at its yield with an open file or cursor. For queues, a missing task_done() in a finally means join() never returns, which turns into a shutdown that hangs until the orchestrator kills it. Verify: break early from a generator and confirm its finally prints immediately; cancel a queue worker mid-item and confirm join() still completes.

5. Combine them: generators inside, queues between

Most production pipelines end up hybrid, and deliberately so.

import asyncio


async def stage_worker(inp: asyncio.Queue, out: asyncio.Queue) -> None:
    """Concurrency comes from N of these; readability from the generator inside."""
    while (batch := await inp.get()) is not None:
        async with aclosing(transform(batch)) as stream:      # generator per batch
            async for record in stream:
                await out.put(record)
        inp.task_done()


async def transform(batch: list[bytes]):
    for line in batch:                              # a clean, testable generator
        record = json.loads(line)
        record["ts"] = normalise(record["ts"])
        yield record

The rule of thumb that produces this shape: use queues where you need parallelism, and generators where you need composition. Queues at the stage boundaries give you worker counts you can tune per stage; generators inside each stage keep the per-item logic linear, unit-testable without any loop plumbing, and free of sentinel handling. Verify: each generator can be tested on a plain list with no queues, and the throughput of the whole pipeline still scales with the worker counts.

The shape most pipelines converge on A hybrid pipeline of queues between stages and generators within them. The shape most pipelines converge on bounded queue stage boundary N workers parallelism lives here generator inside per-item logic, testable bounded queue next boundary Queues where you need parallelism, generators where you need composition.

Verification

  • Memory matches the shape. The generator pipeline is flat regardless of input size; the queue pipeline peaks near sum(maxsize × item size).
  • Concurrency is real. In the queue pipeline, in-flight downstream calls equal the worker count under load.
  • Nothing hangs on shutdown. join() completes and all generators run their finally blocks, including on early termination.

Pitfalls and edge cases

  • asyncio.Queue() with no maxsize. The default is unbounded, which is a deferred out-of-memory rather than a pipeline.
  • Early break from an async for. The generator stays suspended holding its resources; use contextlib.aclosing.
  • Expecting concurrency from generators. They are strictly sequential; two stages never run at the same time.
  • One sentinel for N consumers. Only one worker sees it; send one per consumer or use join() plus cancellation.
  • task_done() outside a finally. A cancelled or failing worker then leaves join() waiting forever.
  • Sharing one async generator between tasks. Concurrent __anext__ calls raise RuntimeError; give each consumer its own.
  • Forgetting shutdown_asyncgens(). In a hand-managed loop, suspended generators never run their cleanup.

Frequently Asked Questions

When should I use an async generator instead of asyncio.Queue?

Use a generator when stages are sequential and composition matters: it gives exact pull-based backpressure, constant memory, automatic end-of-stream, and code that reads linearly. Switch to a queue when any stage would benefit from processing several items concurrently — typically because it awaits I/O — since a generator pipeline processes exactly one item at a time end to end.

Do async generators provide backpressure?

Yes, and it is the strictest kind. A generator produces an item only when its consumer asks for the next one, so a slow consumer directly throttles the producer with no buffering at all. A queue provides backpressure too, but only if maxsize is set: awaiting put() on a full queue suspends the producer, whereas the default unbounded queue provides none.

How do I signal the end of a queue-based pipeline?

Either send one sentinel value per consumer, so every worker sees exactly one and exits, or use queue.join() to wait for all work to be marked done and then cancel the workers. The sentinel approach is explicit and easy to reason about; the join approach avoids sentinel bookkeeping but requires task_done() on every path, including exceptions and cancellation.

Why does my async generator's cleanup never run?

Because it was abandoned while suspended. Breaking out of an async for, or raising inside the loop, leaves the generator parked at its yield with the finally block pending, and relying on garbage collection to close it is unsafe once the loop is shutting down. Wrap iteration in contextlib.aclosing() so aclose() is always called.