Bounded concurrency and backpressure

Unbounded fan-out is the most common way a service takes down its own dependency. It is also the most common way an async rewrite makes throughput worse: asyncio.gather(*(fetch(u) for u in urls)) over 100,000 URLs is a memory bug, not a strategy — every coroutine is materialised as a Task up front, before a single one of them runs.

Two limits people conflate:

• Concurrency — how many calls are in flight at once. A counter, or a semaphore. No clock needed.
• Rate — how many calls start per unit time. Needs a clock, and is a different problem.

This item is about concurrency, plus the third thing people skip entirely: backpressure. Bounding in-flight work is not enough if you read the source as fast as it will go. A pipeline that pulls a million rows into a queue in front of a bounded worker pool has moved the memory problem, not solved it. A bounded pipeline stops reading when the consumer stops consuming.

Why not a TaskGroup

Because a TaskGroup cannot be held open across a yield — see py-batched-async-stream-cleanup. Yielding results as they complete means yielding while tasks are still running, so this implementation owns its task set directly: asyncio.wait(..., return_when=FIRST_COMPLETED), explicit cancellation in a finally, and a BaseExceptionGroup raised by hand.

What to write

def bounded_map[T, R](
    fn: Callable[[T], Awaitable[R]],
    items: AsyncIterator[T],
    *,
    concurrency: int,
) -> AsyncGenerator[R, None]

A plain def wrapper that validates eagerly, plus the _bounded_map async generator that does the work.

• concurrency < 1 raises ValueError(f"concurrency must be at least 1, got {concurrency}") at call time.
• Yield results as they complete. When several tasks land in the same asyncio.wait batch, break the tie by submission order — deterministic ordering is part of the contract, not an accident.
• Never more than concurrency calls in flight.
• Never pull more than concurrency items ahead of the consumer. Because the loop only refills after a wait, and only yields between refills, this falls out of the structure — but only if you refill after draining, not before.
• On the first failure: cancel every outstanding task, await them, and raise BaseExceptionGroup("bounded_map failed", failures). Cancelling without awaiting leaves a “Task exception was never retrieved” warning and, worse, a half-cancelled dependency.
• Leave no pending task behind.

Note asyncio.ensure_future, not create_task: the parameter is Awaitable[R], and create_task demands a coroutine specifically.

How the tests stay deterministic without a clock

fn awaits asyncio.sleep(0) exactly hops[i] times. asyncio’s ready queue is FIFO, so hop counts give a reproducible completion order with no wall-clock dependency at all. That is also the general technique for testing async ordering: encode “slower” as “more scheduling turns”, never as “more milliseconds”.

What the report proves

• results — completion order. With hops=[24, 8, 16] the answers arrive as index 1, index 2, index 0. A solution that yields in submission order fails.
• max_inflight — the peak number of fn bodies running at once. Must equal concurrency exactly when there is enough work, and len(hops) when there is not.
• max_lookahead — the peak of items_pulled - results_delivered. This is the backpressure measurement, and it must not exceed concurrency even when the consumer is slow (consumer_lag inserts scheduling turns per result).
• pulled — total items read from the source. After a failure it must stop.
• pending_tasks — outstanding tasks after the stream is closed. Must be 0.

  Loading visualization…