Bounded Worker Queues and Backpressure: Python Applications

Outcome

Use an asyncio.Queue with a maximum size and a reliable task_done lifecycle.

Why it matters

At this level, code is judged by how safely it changes under load, failure, and team ownership. The technique in this lesson makes an important boundary visible enough to test and review.

Concept

A bounded queue turns overload into waiting instead of memory growth. Producers await put, workers acknowledge in finally, and the coordinator waits for join before cancelling workers.

Read the code

import asyncio

async def worker(queue):
    while True:
        item = await queue.get()
        try:
            await asyncio.sleep(0)
            print(item * 10)
        finally:
            queue.task_done()

async def main():
    queue = asyncio.Queue(maxsize=2)
    task = asyncio.create_task(worker(queue))
    for item in [1, 2, 3]:
        await queue.put(item)
    await queue.join()
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        pass

asyncio.run(main())

Read from the public behavior inward: identify the input boundary, the decision, and the observable result before studying syntax.

Predict the output

Predict the exact output before running the example.

Check your prediction

10
20
30

The third put may wait until the worker creates capacity, which is the intended backpressure. After the queue drains, cancellation is awaited so worker cleanup finishes before main returns.

Two waiting slots are not the same as two unfinished tasks

1. Queue full

Queue: two waiting slots

Worker holds: No item yet

Producer: put(3) waits

Waiting items: 2 of 2
Unfinished work: 2

The third item stays with the producer until a waiting slot becomes available.

Next step in this same run

2. Worker gets item 1

Queue: two waiting slots

Empty

Worker holds: 1

Producer: put(3) can resume

Waiting items: 1 of 2
Unfinished work: 2

get frees a slot. Item 1 is still unfinished while the worker processes it.

Next step in this same run

3. Producer puts item 3

Queue: two waiting slots

Worker holds: 1

Producer: Next: await join()

Waiting items: 2 of 2
Unfinished work: 3

Two queued items plus one worker-held item are unfinished. Capacity still stays at two waiting items.

Next step in this same run

4. Last item still in progress

Queue: two waiting slots

Empty

Worker holds: 3

Producer: join() still waits

Waiting items: 0 of 2
Unfinished work: 1

Acknowledgements for items 1 and 2 reduced unfinished work from 3 to 2 to 1. An empty queue does not mean all work is done.

One valid schedule, not a promised interleaving. get frees waiting capacity; task_done acknowledges one item without certifying processing success. After the final acknowledgement, unfinished work reaches 0: join returns, then the coordinator cancels and awaits the worker.

Modify the code

Change one valid input into the nearest invalid or overloaded case. Write down which layer should reject it and what the caller should observe.

Review the change

Keep the failure at the narrowest boundary that owns the rule. Preserve a stable return value or exception contract so callers do not need to inspect implementation details.

Debug the bug

Calling task_done only on success makes queue.join hang after an exception. Put acknowledgement in finally so every get has exactly one task_done.

Try it yourself

Complete the focused implementation and run its deterministic checks.

Python Builder already has a browser event loop, so this exercise ends with top-level await main(). In a local .py script, use the asyncio.run(main()) entry point shown above instead.

Loading this exercise…

Practical challenge (optional)

Run two workers, collect failures without stopping the queue, and explain how you would expose sustained saturation as a metric.

AI collaboration

Checkpoint

How does maxsize=2 create backpressure?
Why must every successful get pair with exactly one task_done, even after processing fails?
After calling task.cancel(), why does the example await the task?

Answers

Once two unconsumed items are queued, another put waits for a worker to create capacity instead of growing memory without a bound.
join tracks unfinished items; a missing acknowledgement leaves the counter nonzero and can make join wait forever.
Awaiting observes cancellation and lets the worker's cleanup finish before the coordinator returns.

Summary and next step

You made the boundary explicit, predicted its behavior, tested a deterministic implementation, and examined its failure mode. Continue to the next lesson to combine this technique with a wider application or production constraint.