Drain In-Process LLM Job Queues Before Worker Exit

Last reviewed: November 5, 2026 (Asia/Shanghai).

Direct answer

Close queue admission before waiting for queued LLM work to finish. On Python 3.13 or later, call queue.shutdown() without immediate=True, let consumers finish accepted items, await queue.join(), and await the consumer tasks themselves. Close shared HTTP clients and result stores after their borrowers end. Put this sequence inside a finite deployment shutdown budget, with a separate recovery path for work that cannot finish.

The Python Queues reference documents that ordinary shutdown rejects future puts and wakes blocked producers with QueueShutDown. Already queued items remain available to consumers. This closes an important race: a producer waiting for capacity must not add another model job after the drain has started.

An empty queue is insufficient evidence of completion. A worker removes an item before running its inference, so qsize() can be zero during an active call. Likewise, join() tracks task_done() accounting; it cannot inspect whether a result was valid or saved. Verify local task termination and application outcomes separately.

Who this is for

This guide is for Python service maintainers using an in-process asyncio.Queue to hand LLM jobs to long-lived consumers. Its API examples require Python 3.13 or later. The checked references are versioned Python 3.14 documentation and Uvicorn’s current server-behavior guide.

An in-memory queue does not preserve accepted work across a process crash. If the application promises eventual completion, record acceptance durably and use the queue as a scheduling aid, or use a durable broker. The procedure below improves orderly retirement; recovery still needs an independent record of jobs and their outcomes.

Key takeaways

• Stop new submissions and close the queue before waiting for a drain.
• Keep accepted items, active handlers, and completed outcomes as separate counts.
• Pair each successful get() with one task_done() in the consumer.
• Await consumer termination before closing resources they borrow.
• Treat immediate shutdown as discarding waiting work, and reconcile those jobs.
• Reserve time for cancellation, cleanup, and recovery evidence inside the deployment deadline.

Contract details to verify

Define the acceptance boundary

Choose exactly when the application accepts responsibility for a submission. A coroutine beginning await queue.put(job) has not necessarily enqueued anything: it can be blocked by a full queue. A producer rejected by QueueShutDown must return the application’s stable unavailable outcome and must not dispatch the model directly as a fallback around that rejection.

For a durable job service, the acknowledgment contract may require an accepted-job record before responding to the submitter. Coordinate that record with local scheduling so a shutdown between those steps leaves a recoverable job. Do not describe an in-memory put() as durable acceptance.

Use application admission state to stop ordinary submission paths, then invoke queue.shutdown() to reject puts that were already waiting or arrive through a racing path. Join the producer tasks through their owner as well. A readiness change alone does not prevent a handler already running in the process from attempting an enqueue.

asyncio.Queue is not thread-safe. Run these operations in its owning event loop; thread and process handoffs need their own supported interface. In a deployment with several worker processes, inventory and drain every process’s queue. One empty queue does not establish deployment-wide completion.

Distinguish the four completion boundaries

The Queues reference increments unfinished-task accounting for each put and decrements it through task_done(). Keep the decrement in a finally block around processing after a successful get. Do not call it when get() raises, and do not call it twice to make a stuck join disappear.

Save this small consumer as queue_worker.py:

import asyncio


async def worker(queue, handle):
    while True:
        try:
            job = await queue.get()
        except asyncio.QueueShutDown:
            return
        try:
            await handle(job)
        finally:
            queue.task_done()

The injected handler owns inference, response validation, response closure, and the application’s outcome recording. It must await its job-owned children before returning. Bind the job’s approved context at this boundary using the site’s worker context guidance .

An unexpected handler exception escapes this consumer after the accounting decrement. Its supervisor must observe that failure and reconcile accepted jobs; a zero unfinished count must not override it. Map expected job failures to explicit recorded outcomes inside the handler. Preserve cancellation rather than converting it into another generation attempt.

Give the queue a resource owner

Retain consumer task references and give their lifecycle to a supervisor. A TaskGroup can own them, but it does not instruct an idle consumer to stop: without queue closure or another termination protocol, the group can wait forever for workers blocked in get().

For the normal path, close admission, call queue.shutdown(), wait for queue accounting, and then await every consumer. Consumers exit when the shut-down queue has no more items. An owned task group also propagates unexpected consumer failures and cancels siblings according to its documented rules. Keep that failure outcome visible.

Only then close the resources used by those consumers. The HTTP client lifespan guide explains client ownership. For example, an outer HTTP-client context can surround an inner worker-task scope so worker teardown runs before client closure.

Allocate one shutdown deadline

The Python task reference defines timeout_at() against the event loop’s clock. Establish one absolute drain deadline and pass the remaining budget through the shutdown stages. Keep each job’s earlier execution deadline effective; retirement does not grant stale jobs a new inference allowance.

Reserve a cleanup interval after the normal drain interval. If jobs exceed the drain deadline, request cancellation, await cleanup within the remaining process budget, and preserve unfinished or uncertain outcomes for recovery. Repeatedly granting a fresh timeout to each worker can exceed the total deployment allowance.

Cancellation is cooperative. The Python reference recommends try/finally cleanup and generally propagating CancelledError afterward. It also documents that wait_for() can exceed its timeout while waiting for cancellation to finish. Neither a timeout nor task.cancel() provides hard preemption of blocking or cancellation-suppressing code. Test the deployment’s final termination boundary separately.

The Uvicorn Server Behavior guide describes waiting for connections and background tasks during graceful process shutdown, subject to configured timeout periods. Its description includes tasks still completing after an HTTP response. Give application-created queue consumers an explicit owner and shutdown hook; verify that hook’s timing in the deployed stack instead of assuming server connection draining finishes the queue.

Happy-path operator workflow

Begin with one active synthetic job, one waiting job, and a producer blocked on a full queue. Trigger the application’s drain transition. Require admission to close and the blocked producer to receive the documented rejection. Confirm that no rejection path silently starts a model call.

Allow the active job to finish, then process the waiting job under its original eligibility and deadline checks. Require each handler to save its valid result or explicit failure outcome before it returns. After ordinary join() completes, await the consumers and inspect their outcomes. Close shared resources after those tasks finish.

Record drain start, admission state, accepted-job count, waiting count, active count, outcome counts, consumer exceptions, and cleanup duration. Use bounded categories in metrics. Keep job references in appropriately restricted diagnostics and exclude prompts, generated text, credentials, and raw request objects.

Error-path operator workflow

If a consumer fails unexpectedly, stop new admission and surface the supervisor failure. Determine which accepted jobs were still waiting, which were dispatched, and which saved a result. Reconcile against the durable record where one exists. Do not restart all accepted jobs just because one local task failed.

If the normal drain budget expires, retain the accepted-job inventory and move to the declared forced-stop policy. queue.shutdown(immediate=True) empties waiting items and reduces unfinished accounting by the number discarded. It does not cancel a handler that already fetched an item. That handler still needs cancellation or completion, cleanup, and its matching task_done().

Immediate shutdown does not return the discarded job identities. Keep those identities in application-owned records before needing this path. A join can return after waiting items were discarded without any inference having run for them; record them as unfinished or recoverable according to the product contract. Never report immediate shutdown itself as successful processing.

For an active provider attempt, local cancellation cannot establish whether remote generation stopped or what usage occurred. Preserve dispatch state and uncertain outcomes, then apply the existing idempotency and reconciliation policy before a replacement attempt. If cleanup cannot finish before process termination, record that limit explicitly rather than marking resources released.

Verify queue shutdown offline

Save this as probe_queue_drain.py beside the consumer and run python3.13 probe_queue_drain.py, or use another Python version with the documented shutdown API. The check uses synthetic strings and synchronization events; it makes no network requests.

import asyncio

from queue_worker import worker


async def graceful():
    queue = asyncio.Queue(maxsize=1)
    entered = asyncio.Event()
    release = asyncio.Event()
    outcomes = []

    async def handle(job):
        if job == "A":
            entered.set()
            await release.wait()
        outcomes.append(job)

    async def blocked_put():
        try:
            await queue.put("C")
        except asyncio.QueueShutDown:
            return "rejected"
        raise AssertionError("blocked producer was admitted")

    async with asyncio.TaskGroup() as owner:
        consumer = owner.create_task(worker(queue, handle))
        queue.put_nowait("A")
        await entered.wait()
        assert queue.empty() and outcomes == []

        joined = owner.create_task(queue.join())
        await asyncio.sleep(0)
        assert not joined.done()

        queue.put_nowait("B")
        producer = owner.create_task(blocked_put())
        await asyncio.sleep(0)
        assert not producer.done()
        queue.shutdown()
        assert await producer == "rejected"

        try:
            queue.put_nowait("late")
        except asyncio.QueueShutDown:
            pass
        else:
            raise AssertionError("late producer was admitted")

        release.set()
        await joined
        await consumer

    assert outcomes == ["A", "B"]


async def immediate():
    queue = asyncio.Queue()
    queue.put_nowait("active")
    assert queue.get_nowait() == "active"
    queue.put_nowait("waiting")
    queue.shutdown(immediate=True)
    assert queue.empty()

    joined = asyncio.create_task(queue.join())
    try:
        await asyncio.sleep(0)
        assert not joined.done()
        queue.task_done()
        await joined
    finally:
        if not joined.done():
            joined.cancel()
            await asyncio.gather(joined, return_exceptions=True)

    try:
        queue.get_nowait()
    except asyncio.QueueShutDown:
        pass
    else:
        raise AssertionError("discarded item was returned")


async def main():
    async with asyncio.timeout(5):
        await graceful()
        await immediate()
    print("admission, graceful drain, and discard accounting: passed")


asyncio.run(main())

A local run on November 5, 2026 in Asia/Shanghai using Python 3.13.9 printed the stated pass line. The graceful trial verified that an empty queue can still have unfinished work, blocked and late puts are rejected, and A then B finish before consumer exit. The immediate trial verified that the waiting item disappears while join still waits for the already fetched item to be acknowledged.

The five-second limit bounds this cooperative fixture; it is not a production drain recommendation. The check does not exercise OS signals, Uvicorn, HTTP cleanup, a durable store, provider cancellation, or billing. Before rollout, repeat admission races and deadline expiry through those deployed components. Include a handler exception, cancellation during result persistence, and forced termination with unresolved accepted jobs.

Failure modes

Shutdown waits while producers keep submitting. Close admission and shut down the queue before joining it. Join producer tasks so rejected submissions reach a stable outcome too.

An empty queue is labeled drained. Consumers already fetched the active jobs. Track their tasks and outcomes alongside queue accounting.

A worker calls task_done() only on success. A failed handler leaves accounting stuck. Use finally, propagate the failure, and preserve recovery responsibility separately.

A task group waits for idle workers forever. The consumers have no exit signal. Ordinary queue shutdown makes empty-queue gets raise QueueShutDown.

Immediate shutdown is presented as completed work. Waiting jobs were removed without processing. Reconcile them from the accepted-job inventory.

The shared client closes before the workers. Active handlers lose a borrowed resource during cleanup. Nest or sequence ownership so workers end first.

FAQ

Does queue shutdown cancel the active inference?

No. It changes queue admission and retrieval behavior. A consumer already processing an item continues until its own operation ends or receives cancellation through its task owner.

Does join() prove the result was saved?

It proves the accounting reached zero under the applicable shutdown mode. Your handler and durable job record establish result acceptance, failure, or recovery responsibility. Inspect consumer errors as well.

Can this run on Python 3.12?

The shown shutdown API and QueueShutDown were added in Python 3.13. On older runtimes, use a separately designed termination protocol, such as coordinated sentinels after producers stop, and verify its full-queue and worker-failure behavior. The code here requires the newer API.

Can an in-memory queue guarantee crash recovery?

No. Orderly draining depends on the process remaining alive long enough to run it. A durable acceptance record or broker is needed when the product promises recovery after abrupt termination.

Reader next step

Inventory one worker’s producers, queue, consumers, and borrowed resources. Write down its acceptance point and recovery record. Run the offline probe, then retire that worker in staging while one job is active and another is waiting. Accept the change when new submissions stop, every accepted job has an explainable outcome, consumers end before shared-resource closure, and forced termination preserves unresolved work for recovery.

Sources checked

• Python 3.14: Queues — fetched November 4 UTC, November 5 in Asia/Shanghai. Supports bounded admission, ordinary and immediate shutdown, blocked producers, queue accounting, non-thread-safety, and the Python 3.13 introduction of the shutdown API.
• Python 3.14: Coroutines and Tasks — fetched at the same time. Supports task-group ownership and failures, cancellation cleanup, event-loop deadlines, and the cancellation wait performed by wait_for().
• Uvicorn: Server Behavior — fetched at the same time. Its Graceful Process Shutdown section supports connection finalization and waiting for background tasks within configured timeout periods.

These maintained primary technical references support an evergreen implementation task. This article does not announce a recent release or incident. The acceptance record, outcome categories, shutdown sequencing, observability, and staging criteria are engineering recommendations. The reported local observations are limited to the Python 3.13.9 standard-library probe.