Mastering asyncio Patterns: Queues, Semaphores, Timeouts, and Cancellation

In modern Python applications, asyncio is the go-to library for writing concurrent code using async/await syntax. While basic coroutines and tasks are straightforward, real-world systems demand more sophisticated patterns to manage work distribution, control concurrency, handle slow operations, and shut down gracefully. This tutorial dives deep into four essential asyncio patterns: Queues for producer-consumer workflows, Semaphores for limiting concurrent access, Timeouts for bounding operation duration, and Cancellation for cooperative task termination. You'll learn not only the APIs but also the pitfalls and best practices that separate production-ready code from toy examples. By the end, you'll be able to build robust, scalable async applications that handle backpressure, resource limits, and shutdowns with confidence.

1. asyncio.Queue: Coordinating Producers and Consumers

The asyncio.Queue is a FIFO queue designed for async tasks. It allows producers to put items and consumers to get items, with built-in support for backpressure via maxsize. When the queue is full, put() blocks until space is available. When empty, get() blocks until an item arrives. This pattern is ideal for web scrapers, data pipelines, or any workload where you want to decouple work generation from processing. Here's a basic producer-consumer example:

import asyncio

async def producer(queue, n):
    for i in range(n):
        await asyncio.sleep(0.1)  # simulate work
        await queue.put(i)
        print(f"Produced {i}")
    await queue.put(None)  # sentinel to signal end

async def consumer(queue):
    while True:
        item = await queue.get()
        if item is None:
            break
        print(f"Consumed {item}")
        queue.task_done()

async def main():
    queue = asyncio.Queue(maxsize=5)
    prod = asyncio.create_task(producer(queue, 10))
    cons = asyncio.create_task(consumer(queue))
    await asyncio.gather(prod, cons)

asyncio.run(main())

2. asyncio.Semaphore: Controlling Concurrency

When you have many tasks that need to access a limited resource (e.g., database connections, API rate limits), asyncio.Semaphore limits the number of concurrent coroutines. It works like a mutex but allows multiple acquisitions up to a count. Use it as an async context manager: async with semaphore: ... This ensures that only a fixed number of tasks enter the critical section at once. Example:

import asyncio

async def fetch_url(sem, url):
    async with sem:
        print(f"Fetching {url}")
        await asyncio.sleep(1)  # simulate network
        return f"Data from {url}"

async def main():
    sem = asyncio.Semaphore(3)
    urls = [f"http://example.com/{i}" for i in range(10)]
    tasks = [fetch_url(sem, url) for url in urls]
    results = await asyncio.gather(*tasks)
    print(results)

asyncio.run(main())

3. Timeouts with asyncio.wait_for

Network calls, file I/O, or user input can hang indefinitely. asyncio.wait_for wraps a coroutine and raises asyncio.TimeoutError if it doesn't complete within a given timeout. This is crucial for responsiveness. Example:

import asyncio

async def slow_operation():
    await asyncio.sleep(10)
    return "Done"

async def main():
    try:
        result = await asyncio.wait_for(slow_operation(), timeout=2)
        print(result)
    except asyncio.TimeoutError:
        print("Operation timed out!")

asyncio.run(main())

4. Cancellation: Graceful Shutdown and Task Cleanup

Cancellation in asyncio is cooperative. Calling task.cancel() schedules a CancelledError to be raised at the next await point inside the task. The task can catch this exception to perform cleanup. For graceful shutdown, you typically cancel all tasks and then gather them with return_exceptions=True to avoid propagating CancelledError. Example:

import asyncio

async def worker(name, duration):
    try:
        print(f"{name} starting")
        await asyncio.sleep(duration)
        print(f"{name} completed")
    except asyncio.CancelledError:
        print(f"{name} cancelled")
        raise  # re-raise to acknowledge cancellation

async def main():
    tasks = [
        asyncio.create_task(worker("A", 5)),
        asyncio.create_task(worker("B", 2)),
    ]
    await asyncio.sleep(1)
    for t in tasks:
        t.cancel()
    await asyncio.gather(*tasks, return_exceptions=True)
    print("All tasks done")

asyncio.run(main())

5. Combining Patterns: A Production-Ready Worker Pool

In real applications, you often combine these patterns. For example, a web scraper might use a queue to hold URLs, a semaphore to limit concurrent requests, timeouts on each fetch, and cancellation for shutdown. Here's a complete example:

import asyncio

async def fetch(sem, url, timeout):
    async with sem:
        try:
            result = await asyncio.wait_for(
                asyncio.sleep(1), timeout=timeout
            )
            return f"Fetched {url}"
        except asyncio.TimeoutError:
            return f"Timeout {url}"

async def producer(queue, urls):
    for url in urls:
        await queue.put(url)
    for _ in range(10):  # sentinels for consumers
        await queue.put(None)

async def consumer(queue, sem, timeout, results):
    while True:
        url = await queue.get()
        if url is None:
            queue.task_done()
            break
        result = await fetch(sem, url, timeout)
        results.append(result)
        queue.task_done()

async def main():
    queue = asyncio.Queue(maxsize=20)
    sem = asyncio.Semaphore(5)
    timeout = 2
    urls = [f"http://example.com/{i}" for i in range(50)]
    results = []
    prod = asyncio.create_task(producer(queue, urls))
    consumers = [asyncio.create_task(consumer(queue, sem, timeout, results)) for _ in range(10)]
    await asyncio.gather(prod, *consumers)
    print(f"Fetched {len(results)} URLs")

asyncio.run(main())

6. Best Practices and Common Pitfalls

1. Always set maxsize on queues to prevent memory blowup. 2. Use asyncio.Semaphore as a context manager to ensure release. 3. Prefer asyncio.wait_for over manual timeout logic. 4. Handle CancelledError properly: do cleanup, then re-raise. 5. Use asyncio.shield to protect critical sections from cancellation. 6. Avoid mixing asyncio with blocking calls; use loop.run_in_executor for blocking I/O. 7. Test cancellation and timeout behavior with unit tests. Common pitfalls include forgetting to call queue.task_done(), using semaphore incorrectly with create_task, and not re-raising CancelledError.