When building high-throughput asynchronous pipelines in Python—such as streaming tokens from Large Language Models (LLMs) like OpenAI or Anthropic through an asyncio.Queue—managing backpressure and resource cleanup is critical. A common architectural setup involves a producer task reading from an async generator and pushing items into a bounded queue, while a consumer task pulls from it.

However, when the consumer times out or disconnects, cancelling the producer task while it is blocked on await queue.put(token) can raise concerns: Does the async generator actually execute its finally block, or does cancellation leak underlying HTTP connections and network sockets?

Understanding What Happens During Cancellation

When you call prod_task.cancel() while the task is suspended at await queue.put(token), the following sequence occurs:

  1. An asyncio.CancelledError is raised at the exact line where the task was suspended (queue.put()).
  2. The exception unwinds the stack in the producer coroutine.
  3. If the generator is wrapped inside an async for loop or an async with contextlib.aclosing(...) context manager, the unwinding process automatically triggers .aclose() on the async generator.
  4. Calling .aclose() throws a GeneratorExit exception directly into the generator at its current yield point.
  5. The generator's finally block executes, allowing network resources to close cleanly before the CancelledError continues propagating.

The Recommended Pattern: contextlib.aclosing

While Python 3.10+ enhanced async for loops to automatically close async generators upon exception unwinding, explicit management using contextlib.aclosing() is the gold standard pattern. It explicitly guarantees that aclose() is called and awaited as soon as control leaves the block, regardless of Python runtime variations.

import asyncio
import contextlib

async def stream_llm_tokens(n=50):
    try:
        for i in range(n):
            await asyncio.sleep(0.01)
            yield f"token_{i}"
    finally:
        # Guaranteed to run when aclose() is invoked
        print("stream_llm_tokens: resource cleanup completed")

async def producer(queue: asyncio.Queue):
    # Wrap the generator with aclosing to guarantee deterministic cleanup
    async with contextlib.aclosing(stream_llm_tokens()) as token_stream:
        async for token in token_stream:
            await queue.put(token)
    await queue.put(None)

The Hidden Trap: Asynchronous Cleanup During Cancellation

There is one crucial edge case: What if your generator's finally block performs async operations?

In real-world applications, closing an HTTP stream (e.g., using httpx or aiohttp) requires awaiting an asynchronous network close operation (like await response.aclose()). In Python 3.11+, when a task is cancelled, any await statement executed inside a pending cancellation state will immediately raise asyncio.CancelledError again!

If your generator attempts to await resource teardown inside finally while cancelled, that teardown step might be aborted prematurely. To prevent this, wrap the async cleanup in asyncio.shield() or handle the cancellation state during shutdown:

async def stream_llm_tokens_real_world():