Worker-pool shutdown is where async services deadlock. The shape is always the
same: consumers awaiting get() on a queue nobody will feed again, held alive
by a join() that will never return because one consumer crashed between
get() and task_done().
asyncio.Queue is not queue.Queue
It is documented as not thread-safe, and that is not a footnote. If you
reach for asyncio.Queue to pass work between a thread and the event loop, you
have a data race. The tools for that direction are
loop.call_soon_threadsafe and asyncio.run_coroutine_threadsafe.
maxsize is the other thing people leave at its default. An unbounded queue
does not remove backpressure; it moves the failure from “the producer waits” to
“the process runs out of memory”. If your producer is faster than your
consumers — reading a file, draining a socket, iterating a cursor — an unbounded
queue is a slow-motion OOM. maxsize is the single most important argument.
join() and task_done()
Queue.join() blocks until every item that was put has had a matching
task_done(). The counter is incremented by put and decremented by
task_done — not by get. That asymmetry is the whole API, and it is
deliberate: the queue is tracking work completed, not items removed.
Which means the only correct shape is:
item = await queue.get()
try:
await handle(item)
finally:
queue.task_done()
Without the finally, a handler that raises skips its task_done(), the
counter never reaches zero, and join() waits forever. The service does not
crash — it hangs, on shutdown, with no error, and every log line looks normal.
💡A consumer calls task_done() twice for one item. What happens?
click to reveal
ValueError: task_done() called too many times, raised immediately.
That is a better outcome than the opposite mistake, and worth knowing because
it tells you the counter is validated rather than merely decremented. The
double call usually comes from calling task_done() inside the try and
in the finally after a refactor, or from a continue path that was added
later and did not notice the finally already handled it.
The asymmetric failure modes are the practical point. Too few task_done()
calls gives you a silent hang at shutdown, discovered in production. Too many
gives you a loud ValueError at the call site, discovered in the first test.
So when in doubt, structure the code so task_done() happens exactly once in a
finally — and let the loud failure catch you if you got it wrong.
Shutdown: 3.13 fixed this properly
Before 3.13, the only way to tell consumers “no more work” was a sentinel,
one per consumer, pushed after the last real item. It works, and it costs you
something specific: the queue’s element type becomes Queue[Item | None] (or
Queue[Item | Sentinel]), and now every consumer must narrow before it can
touch an item. That union propagates into every function the item is passed to
until you narrow it away.
Python 3.13 added Queue.shutdown(immediate=False) and QueueShutDown:
queue.shutdown() # no more puts; gets drain what is left, then raise
queue.shutdown(immediate=True) # discard the backlog; unblock join() now
After shutdown(), put raises QueueShutDown immediately, and get keeps
returning the remaining items and then raises QueueShutDown — so a consumer
loop becomes try: while True: item = await queue.get() ... except QueueShutDown: return, and the queue stays Queue[Item]. The type stops carrying a shutdown
protocol it has nothing to do with.
immediate=True is the emergency version: it drops queued items and unblocks
join() while violating join’s invariant — join() returns without every
item having been processed. That is a legitimate thing to want during an
abandoned shutdown, and it is not what you want during a normal one.
💡A worker raises. Your producer is blocked on await queue.put(item) because the queue is full. What breaks, and what makes it not break?
click to reveal
Everything stops, and nothing reports it — unless the producer and the workers
are inside the same TaskGroup.
The mechanics: the worker’s task ends with an exception. The producer is
suspended in put, waiting for space that will never appear because one fewer
consumer is draining. If the remaining consumers also finish or fail, the
producer waits forever on a queue nobody will ever drain. If the producer is
the coroutine that was going to push the sentinels, the surviving consumers
also wait forever on get(). Complete deadlock, no exception, no log line.
TaskGroup is what fixes it, and the placement is the trick: put the producer
in the body of the async with, not in a task. When a consumer task fails,
the group cancels the body — so the await queue.put(...) is cancelled rather
than hanging — then cancels the remaining consumers, waits for all of them to
unwind, and raises an ExceptionGroup containing the real failure.
That is the concrete payoff of structured concurrency here: the deadlock becomes an exception, and it becomes one without you having written any code to detect it.