Asynchronous worker pools often start with a simple pattern: producers put jobs into an asyncio.Queue, consumers loop over get(), and the application waits for join() before exiting.

The awkward part is shutdown. Older designs commonly put one sentinel value into the queue for each worker, cancel consumers after join(), or maintain a separate stop event. Each approach can work, but each adds a second protocol beside the queue itself.

Python 3.13 added asyncio.Queue.shutdown() and the asyncio.QueueShutDown exception. They let the queue represent its own lifecycle: open for producers, shutting down while existing work drains, and finally closed to consumers.

That makes graceful shutdown easier to reason about, but only if task_done() and join() keep their intended meaning.

Start with the queue lifecycle

A normal asyncio.Queue accepts put() calls and lets consumers wait in get(). A bounded queue also provides backpressure: when it reaches maxsize, await queue.put(item) waits for capacity.

Calling:

queue.shutdown()

changes that contract. The queue can no longer grow. Future put() calls raise QueueShutDown, and producers already blocked in put() are awakened and receive the same exception.

With the default immediate=False, items already in the queue remain available. Consumers can drain them normally. After the queue becomes empty, get() raises QueueShutDown.

That gives a worker a natural exit condition:

import asyncio


async def worker(queue: asyncio.Queue[str]) -> None:
    while True:
        try:
            item = await queue.get()
        except asyncio.QueueShutDown:
            return

        try:
            await process(item)
        finally:
            queue.task_done()

The finally block is important. Once get() succeeds, that item contributes to the queue’s unfinished-task count until exactly one task_done() acknowledges it.

Gracefully drain a worker pool

A small worker pool can use shutdown without sentinel objects:

import asyncio


async def process(item: str) -> None:
    await asyncio.sleep(0.05)
    print(item)


async def worker(name: str, queue: asyncio.Queue[str]) -> None:
    while True:
        try:
            item = await queue.get()
        except asyncio.QueueShutDown:
            return

        try:
            await process(item)
        finally:
            queue.task_done()


async def main() -> None:
    queue: asyncio.Queue[str] = asyncio.Queue(maxsize=100)

    async with asyncio.TaskGroup() as group:
        for number in range(4):
            group.create_task(worker(f"worker-{number}", queue))

        for number in range(1_000):
            await queue.put(f"job-{number}")

        queue.shutdown()
        await queue.join()


asyncio.run(main())

The sequence is deliberate:

  1. Stop admitting new work with shutdown().
  2. Let workers process everything already accepted.
  3. Wait for acknowledgements with join().
  4. Let each worker observe QueueShutDown after the queue drains and return.

No worker-count-dependent sentinel protocol is required.

Treat join() as an acknowledgement barrier

join() does not wait for the queue to become empty. It waits for the unfinished-task count to reach zero.

Every successful put() increases that count. Every matching task_done() decreases it. This distinction matters because a consumer can remove an item from the queue long before it finishes processing the item.

Consider a database writer:

item = await queue.get()
try:
    await write_to_database(item)
finally:
    queue.task_done()

Calling task_done() immediately after get() would let join() return while the database write was still running. During deployment or process shutdown, that can turn a coordination primitive into a false durability signal.

A useful invariant is:

successful get() -> processing finishes or fails -> exactly one task_done()

If failure means the item must be retried durably, design that retry policy separately. task_done() only accounts for the in-memory queue item; it does not prove business-level success.

Handle processing failures explicitly

The finally pattern keeps queue accounting correct, but it can hide an important design question: should one failed job stop the worker pool?

If process() raises and the worker lets the exception escape, a TaskGroup will cancel sibling tasks. That may be exactly what you want for an invariant violation, but it is usually too aggressive for an expected per-item error.

Handle recoverable failures inside the worker:

async def worker(queue, failures):
    while True:
        try:
            item = await queue.get()
        except asyncio.QueueShutDown:
            return

        try:
            await process(item)
        except ExpectedJobError as exc:
            await failures.put((item, exc))
        finally:
            queue.task_done()

For jobs that cannot safely disappear, an in-memory queue should not be the only source of truth. Persist the job or its retry state before acknowledging whatever durable system owns it.

Understand what happens to blocked producers

Shutdown is not only a consumer feature.

Suppose a bounded queue is full and several producers are waiting here:

await queue.put(item)

Once another task calls queue.shutdown(), those blocked producers wake and raise QueueShutDown. This is useful because shutdown does not need to wait for capacity merely to tell producers that no more work is accepted.

Producers should decide whether that exception is expected lifecycle control or an error:

async def producer(queue, source):
    async for item in source:
        try:
            await queue.put(item)
        except asyncio.QueueShutDown:
            return

Do not blindly catch Exception around the entire producer and continue. Once shutdown begins, retrying put() against the same queue cannot reopen it.

Prefer graceful shutdown over immediate shutdown

Queue.shutdown() also accepts immediate=True:

queue.shutdown(immediate=True)

This is a different operation, not merely a faster graceful shutdown. The queue is drained immediately. Blocked getters are awakened and raise QueueShutDown because there are no queued items left to retrieve.

Most importantly, immediate shutdown can unblock join() even though queued work was never processed. That deliberately breaks the normal join() invariant.

Use immediate shutdown only when abandoning queued work is acceptable or when a higher-level failure has already made normal draining impossible.

For example, an application may choose it after a fatal dependency failure:

try:
    await run_pipeline(queue)
except FatalPipelineError:
    queue.shutdown(immediate=True)
    raise

Do not interpret a returned join() after immediate shutdown as proof that all accepted jobs completed.

Separate graceful shutdown from cancellation

Queue shutdown and task cancellation solve different problems.

queue.shutdown() changes what producers and consumers can do with the queue. Cancellation interrupts a coroutine at an await point.

For normal service termination, queue shutdown often provides the cleaner first step because workers finish jobs they already own and drain accepted jobs before exiting.

Cancellation is still useful when a deadline expires:

queue.shutdown()

try:
    async with asyncio.timeout(10):
        await queue.join()
except TimeoutError:
    queue.shutdown(immediate=True)
    raise

The surrounding task structure can then cancel workers if the application cannot wait any longer. Keep in mind that cancellation may interrupt process(item) after get() succeeded. The worker should use finally for queue accounting and make external side effects idempotent when retries are possible.

Avoid mixing shutdown protocols casually

A queue using shutdown() usually does not also need None, an object sentinel, and a stop event.

Multiple protocols create ambiguous states. A producer might enqueue a sentinel before another producer has finished. A worker might exit on an event while items remain queued. A sentinel can also consume bounded capacity and requires knowing how many consumers need to receive one.

There are still cases where a data-level marker is meaningful. For example, a stream may contain explicit partition-end records that are part of the business protocol. Keep those markers distinct from queue lifecycle control.

Coordinate multiple producers before closing admission

The task that calls shutdown() must know that no legitimate producer should submit more work.

If several producers run concurrently, do not let the first producer to finish close the shared queue. Instead, coordinate producer completion at a higher level:

async def produce_all(queue, sources):
    async with asyncio.TaskGroup() as group:
        for source in sources:
            group.create_task(produce(queue, source))

    queue.shutdown()

Now shutdown means all producers have finished, rather than merely one producer.

For long-running services, the trigger may instead be a server lifecycle event. The same principle applies: first stop sources that can create new jobs, then close queue admission, then drain accepted work.

Keep backpressure during normal operation

Shutdown support does not replace capacity planning.

An unbounded queue can absorb a temporary burst, but a sustained producer-consumer mismatch becomes memory growth. Give the queue a finite maxsize when producers can safely wait:

queue = asyncio.Queue(maxsize=500)

This bounds queued items, not total resource use. Workers may hold active jobs, and each item may reference large objects. Choose a limit from workload measurements rather than treating the item count as a memory limit.

Shutdown integrates well with bounded queues because blocked producers are explicitly released with QueueShutDown instead of remaining stuck behind a queue that the application no longer intends to drain for new submissions.

Do not share asyncio.Queue across threads

asyncio.Queue is designed for async code and is not thread-safe. If threads must exchange work, use a thread-safe primitive such as queue.Queue, or cross the event-loop boundary with an appropriate thread-safe scheduling mechanism.

The similarly named queue types have related concepts, but do not treat them as interchangeable synchronization objects.

Test lifecycle transitions, not only happy-path output

Queue shutdown bugs are timing bugs, so tests should exercise states around the transition.

Useful cases include:

  • shutdown while the queue is empty;
  • shutdown while items are waiting;
  • shutdown while workers are processing the last items;
  • shutdown while producers are blocked on a full bounded queue;
  • a worker raising before task_done() would normally run;
  • graceful shutdown followed by join();
  • immediate shutdown with unfinished work;
  • repeated attempts to put() after shutdown.

A focused test can verify that blocked producers are released:

async def test_shutdown_releases_blocked_producer():
    queue = asyncio.Queue(maxsize=1)
    await queue.put("first")

    blocked = asyncio.create_task(queue.put("second"))
    await asyncio.sleep(0)

    queue.shutdown()

    try:
        await blocked
    except asyncio.QueueShutDown:
        pass
    else:
        raise AssertionError("producer should observe shutdown")

Also test your application-level guarantees. If a successful join() is supposed to mean all database writes are committed, inject slow writes and failures to prove that task_done() occurs at the correct boundary.

Plan compatibility deliberately

asyncio.Queue.shutdown() and QueueShutDown were added in Python 3.13. Code that must run on Python 3.12 or older cannot use them directly.

For a library supporting older runtimes, keep an established sentinel or cancellation strategy until the minimum supported Python version reaches 3.13. Avoid a partial compatibility shim that imitates the method name without reproducing wake-up and unfinished-task semantics; concurrency behavior is part of the API contract.

Applications already standardized on Python 3.13 or newer can simplify worker lifecycle code by making the queue itself the shutdown boundary.

Build shutdown around explicit invariants

The most reliable design is not the one with the fewest lines. It is the one whose states have clear meanings.

During normal operation, bounded put() provides backpressure. During graceful shutdown, new puts fail while accepted work remains drainable. Each retrieved item is acknowledged exactly once after processing. join() means all accepted work has been acknowledged. Workers exit when an empty, shut-down queue raises QueueShutDown.

Reserve immediate shutdown for cases where you intentionally abandon that completion guarantee.

With those invariants in place, asyncio.Queue.shutdown() removes much of the hand-built signaling that asynchronous worker pools used to need, while making the shutdown path easier to test and explain.

Python Asyncio Concurrency Queues Python 3.13