Day 5 Checkpoint: Backpressure and the Leak

A semaphore stops you doing too much at once. It doesn’t stop you producing faster than anything downstream can consume — and when the producer is a fast network and the consumer is a slow database, that difference is measured in gigabytes.

A bounded queue is not enough on its own

asyncio.Queue(maxsize=n) makes put wait once there are n items waiting, and that wait is backpressure. But it only bounds what the queue has accepted. It does nothing about work that has been produced and is waiting to be put — and that is where the memory actually goes.

The trap is spawning one task per input:

    async with asyncio.TaskGroup() as group:
        for name in names:
            group.create_task(produce(name))     # one task per input — the leak

Every task reads its source (bounded three at a time by the semaphore), then lines up at queue.put. A task blocked on put has already finished reading and is holding its result. So with ten thousand inputs you get a bounded queue and ten thousand results sitting in blocked tasks. Counting only what the queue accepted hides it completely.

The fix is a fixed pool of workers that pull inputs lazily from a shared iterator. A worker reads the next input only after it has handed off the current one, so it can’t run ahead of the consumer — and the total in flight is bounded by workers + queue_size + 1, whatever the input count. (The + 1 is the value the consumer is holding.) Because the workers advance the iterator only as they become free, a generator or other lazy stream is never drained up front.

import asyncio


async def naive(queue_size, items, consume_delay=0.02):
    queue = asyncio.Queue(maxsize=queue_size)
    outstanding = peak = 0

    async def produce(index):
        nonlocal outstanding, peak
        await asyncio.sleep(0.001)              # a fast read
        outstanding += 1                        # the value exists now, before the hand-off
        peak = max(peak, outstanding)
        await queue.put(index)                  # ...and the task keeps holding it here

    async def consume():
        nonlocal outstanding
        while True:
            await queue.get()
            await asyncio.sleep(consume_delay)
            outstanding -= 1

    consumer = asyncio.create_task(consume())
    async with asyncio.TaskGroup() as group:
        for index in range(items):
            group.create_task(produce(index))
    consumer.cancel()
    return peak


async def pooled(workers, queue_size, items, consume_delay=0.02):
    sources = iter(range(items))                # pulled lazily, one at a time
    queue = asyncio.Queue(maxsize=queue_size)
    outstanding = peak = 0

    async def produce():
        nonlocal outstanding, peak
        while True:
            try:
                index = next(sources)           # no await between asking and owning it
            except StopIteration:
                return
            await asyncio.sleep(0.001)
            outstanding += 1
            peak = max(peak, outstanding)
            await queue.put(index)              # blocks until taken: no reading ahead

    async def consume():
        nonlocal outstanding
        taken = 0
        while taken < items:
            await queue.get()
            await asyncio.sleep(consume_delay)
            outstanding -= 1
            taken += 1

    async with asyncio.TaskGroup() as group:
        for _ in range(workers):
            group.create_task(produce())
        group.create_task(consume())
    return peak


async def main():
    for items in (10, 40):
        n = await naive(queue_size=2, items=items)
        p = await pooled(workers=3, queue_size=2, items=items)
        print(f"{items:3} inputs:  one task each -> peak {n:3}    worker pool -> peak {p}")


asyncio.run(main())

outstanding is counted the moment a value exists — before the hand-off — so it measures the real backlog, not just what the queue holds. The per-task version’s peak tracks the input count: 10, then 40. The worker pool stays flat at workers + queue_size + 1 — six here — whatever you throw at it. And because next(sources) is synchronous, no two workers ever grab the same input, and a generator is advanced only as far as the workers have reached.

Shutting down, and supervising the consumer

run_pipeline runs the workers and the consumer inside one TaskGroup, so no task can outlive the call and a failure anywhere cancels the rest. The consumer loops forever, so a coordinator tells it when to stop:

    async def coordinate() -> None:
        async with asyncio.TaskGroup() as producers:
            for _ in range(workers):
                producers.create_task(produce())
        await results.put(_STOP)            # producers done: release the consumer

    async with asyncio.TaskGroup() as group:
        group.create_task(coordinate())
        group.create_task(consume())        # a failure here cancels the producers

Putting the consumer in the group is what supervises it. If the consumer raises — a bad row, a closed socket — the group cancels the producers instead of leaving them blocked on a full queue forever. The earlier design, with the consumer created outside the group and a queue.join() waiting on it, would simply hang if the consumer died. A hang is the worst failure mode there is: no error, no exit, nothing to see.

The checkpoint

async def read_all(names, *, limit=3, delay=0.01, failing=frozenset()):
    results: dict[str, str] = {}

    async def worker(name: str) -> None:
        try:
            results[name] = await read_one(name, limit_slots, delay=delay,
                                           fail=name in failing)
        except SourceError:
            pass                            # a missing source is not fatal here

    limit_slots = asyncio.Semaphore(limit)
    async with asyncio.TaskGroup() as group:
        for name in names:
            group.create_task(worker(name))
    return results


async def read_all_strict(names, *, limit=3, delay=0.01, failing=frozenset()):
    results: dict[str, str] = {}
    limit_slots = asyncio.Semaphore(limit)

    async def worker(name: str) -> None:
        results[name] = await read_one(name, limit_slots, delay=delay,
                                       fail=name in failing)

    async with asyncio.TaskGroup() as group:
        for name in names:
            group.create_task(worker(name))
    return results

Bug hunt: the cleanup that skips cancellations

Someone rewrote read_one without the async with, “to make the close explicit”.

Reproduce. Three failures, and the clearest message is:

AssertionError: 4 != 0 : a cancelled read left a source open

All the ordinary tests pass. Reads work, failures work, the limit holds. Only the cancellation tests fail.

Inspect. The new code:

        source = await Source(name, delay=delay, fail=fail).__aenter__()
        try:
            value = await source.read()
        except Exception:
            await source.close()
            raise
        await source.close()
        return value

Narrow. There are three ways out of await source.read(): a value, a SourceError, or a CancelledError from the timeout. The first two close the source. The third doesn’t, because CancelledError inherits from BaseException — so except Exception doesn’t see it, and the close() after the try is skipped by the exception on its way up.

Each cancelled read leaks one source. In a real library that’s a connection, a file handle, or a lock somebody else is waiting for.

Pin. test_a_timeout_closes_every_source and test_cancelling_a_task_still_cleans_up are the regression tests, and between them they cover both ways a cancellation arrives: a deadline and an explicit task.cancel().

Fix. Let the context manager do it:

    async with limit:
        async with Source(name, delay=delay, fail=fail) as source:
            return await source.read()

async with compiles to a try/finally, so every exit closes the source — including the one nobody tests for.

Where Day 5 leaves you

Working on your own machine? Download the Day 5 files.

Two details in there are worth copying into your own work.

Source.open_sources is a counter on the class, incremented on enter and decremented on exit, and the tests assert it’s zero afterwards. Instrumenting a resource like that costs four lines and turns “I think we close everything” into an assertion.

And every test drives asyncio.run from a plain TestCase rather than using IsolatedAsyncioTestCase, because the browser runtime this course uses can’t run the latter. On your own machine either works.

What you learned

• A bounded queue alone doesn’t bound memory: a fixed pool of workers pulling lazily from a shared iterator keeps the backlog to workers + queue_size + 1 and never materialises the input stream, whatever the input count.
• Measure the backlog before the hand-off, or the number flatters you.
• Supervise the consumer inside the same TaskGroup, so its failure cancels the producers instead of hanging them.
• Where you put the try decides whether a failure is local or fatal.
• Cleanup must survive cancellation, and only a cancellation test proves it.

Tomorrow: back to the ORM, with a profiler. The identity map from Day 1 has been quietly growing all week, and Day 6 measures it.