29. asyncio: the event loop, tasks, timeouts, and cancellation

29.1 Why async exists

A thread per connection costs memory (a stack per thread) and kernel scheduling work. With 10 000 mostly idle connections, most threads just wait. An event loop instead uses one thread and asks the kernel which sockets are ready (epoll on Linux, kqueue on macOS; see OS Ch 78), then resumes only the code waiting on those sockets.

async def functions are coroutines: functions that can pause at an await and let the loop run something else. They are built on the same suspend-and-resume machinery as generators (Ch 13.5).

import asyncio
import time

async def fetch(name: str, delay: float) -> str:
    await asyncio.sleep(delay)          # yields to the event loop while "waiting on I/O"
    return f"{name} done"

async def main() -> None:
    t = time.perf_counter()
    results = await asyncio.gather(fetch("a", 1), fetch("b", 1), fetch("c", 1))
    print(results, f"{time.perf_counter() - t:.1f}s")   # ~1.0s, not 3s

asyncio.run(main())

Calling fetch("a", 1) does not run it; it creates a coroutine object. Only await, a task, or asyncio.run runs it. Forgetting await gives RuntimeWarning: coroutine 'fetch' was never awaited, and the work never happens.

29.2 Structured concurrency with TaskGroup

import asyncio

async def probe(host: str) -> str:
    await asyncio.sleep(0.1)
    if host == "db":
        raise ConnectionError(f"{host}: refused")
    return f"{host}: ok"

async def main() -> None:
    try:
        async with asyncio.TaskGroup() as tg:          # 3.11+
            tasks = [tg.create_task(probe(h)) for h in ["web", "db", "cache"]]
        print([t.result() for t in tasks])
    except* ConnectionError as eg:                     # failures arrive as an ExceptionGroup
        for e in eg.exceptions:
            print("failed:", e)

asyncio.run(main())

TaskGroup guarantees that when the async with block exits, every task it started has finished. If one fails, the others are cancelled and the errors are raised together. Prefer it over bare create_task, which lets tasks outlive the code that started them (and lets their exceptions disappear if nobody awaits them).

29.3 Timeouts and cancellation

import asyncio

async def slow_query() -> str:
    try:
        await asyncio.sleep(10)
        return "rows"
    except asyncio.CancelledError:
        print("query cancelled: releasing the connection")
        raise                                   # always re-raise CancelledError

async def main() -> None:
    try:
        async with asyncio.timeout(0.5):        # 3.11+
            await slow_query()
    except TimeoutError:
        print("timed out")

asyncio.run(main())

Cancellation is delivered as a CancelledError raised at the await where the task is paused. Clean up in except/finally, then re-raise. Swallowing CancelledError breaks timeouts and shutdown. Every network call in async code needs a timeout; the HTTPS guide Ch 21 explains why a missing timeout eventually takes the whole service down.

29.4 Limiting concurrency, and queues

import asyncio
import random

async def download(i: int, sem: asyncio.Semaphore) -> int:
    async with sem:                               # at most N inside at once
        await asyncio.sleep(random.uniform(0.05, 0.2))
        return i

async def producer(q: asyncio.Queue[int]) -> None:
    for i in range(10):
        await q.put(i)                            # blocks when the queue is full: backpressure
    await q.put(-1)

async def consumer(q: asyncio.Queue[int]) -> None:
    while (item := await q.get()) != -1:
        print("processed", item)

async def main() -> None:
    sem = asyncio.Semaphore(5)
    got = await asyncio.gather(*(download(i, sem) for i in range(20)))
    print(len(got), "downloads")
    q: asyncio.Queue[int] = asyncio.Queue(maxsize=3)
    async with asyncio.TaskGroup() as tg:
        tg.create_task(producer(q))
        tg.create_task(consumer(q))

asyncio.run(main())

A semaphore is how you avoid opening 10 000 connections to a server that allows 100. A bounded queue gives backpressure: a fast producer waits for a slow consumer instead of filling memory.

29.5 The one rule: never block the event loop

Everything in an event loop shares one thread. A blocking call (time.sleep, requests.get, a CPU-heavy loop, a synchronous database driver) freezes every connection until it returns.

import asyncio
import hashlib
import time

def blocking_hash(data: bytes) -> str:           # CPU-bound or blocking library call
    time.sleep(0.2)
    return hashlib.sha256(data).hexdigest()[:12]

async def main() -> None:
    # Run blocking work in a thread so the loop stays responsive:
    digest = await asyncio.to_thread(blocking_hash, b"model.bin")
    print(digest)

asyncio.run(main())

Use async libraries in async code (httpx.AsyncClient, asyncpg, aiofiles), asyncio.to_thread for blocking calls, and a process pool (loop.run_in_executor(process_pool, fn)) for CPU-bound work. Run with PYTHONASYNCIODEBUG=1 (or asyncio.run(main(), debug=True)) during development: it logs every callback that blocks the loop for more than 100 ms.

Across the series: Go hides the event loop inside the runtime, so blocking code is fine in a goroutine (OS Ch 78). Rust and Python make it explicit with async/await, and both suffer from the same "function colouring": async functions can call sync ones, but not the other way round without a bridge.