Async Programming with asyncio

The event loop mental model

In M37 you saw that threads overlap I/O waits, but they bring preemptive switching and therefore races at any instruction. asyncio takes a different approach called cooperative multitasking. All your tasks run in one thread. A task runs until it reaches an await on something that isn't ready yet (a network reply, a timer); at that point it voluntarily suspends and hands control back to the event loop. The loop then runs another task that is ready. When the reply arrives, the loop marks the waiting task ready again and resumes it exactly where it left off.

Conceptually, the loop is just this:

while tasks_exist:
    run every callback in the READY queue        # each runs until its next await
    wait for I/O or the next timer (selectors / epoll / kqueue / IOCP)
    move tasks whose I/O or timer completed to the READY queue

Compare the three ways of doing three 1-second waits. Step through each scenario and watch what the CPU is doing during the waits:

Three consequences follow from "one thread, switch only at await":

• Races are rarer and visible. Code between two awaits runs without interruption, so self.count += 1 needs no lock. But any check → await → act sequence can still interleave with other tasks (you'll see this in Project 5).
• Blocking is catastrophic. A task that calls time.sleep(1) or a blocking HTTP library freezes every task for that second, because the loop can't run anything else until it returns.
• No parallelism. asyncio is for I/O-bound concurrency. CPU-heavy work stalls the loop just like blocking I/O; offload it (see Mixing in blocking code).

Coroutines, await, and Tasks

Precise vocabulary prevents most asyncio confusion:

The crucial difference: await coro() runs the coroutine now, inside the current task, and waits for it. Concurrency only happens when you create Tasks, because each Task is a separate entry the loop can switch between. Under the hood, a coroutine is a generator-like object: await works much like yield from, passing suspension requests up to the Task that drives it.

gather vs TaskGroup: structured concurrency

asyncio.gather(*aws) runs awaitables concurrently and returns their results in argument order. Its error handling is the problem: by default, the first exception propagates to the awaiting code, but the other tasks keep running, unsupervised, in the background. With return_exceptions=True, exceptions are returned as values in the result list, which is handy for "try everything, report failures" but easy to forget to check.

asyncio.TaskGroup (3.11+) implements structured concurrency: tasks are created inside an async with block and cannot outlive it. The block waits for all of them. If any task fails, the group cancels all the others, waits for them to finish cancelling, and raises an ExceptionGroup containing every failure. Lifetimes are visible in the code's indentation, like variables in a scope.

The except* syntax (3.11+) matches exceptions inside an ExceptionGroup: each except* clause handles the matching subgroup, and anything unmatched is re-raised. Several clauses can run for one group.

Timeouts

Every network operation needs a timeout; a peer that never answers otherwise hangs your task forever. async with asyncio.timeout(seconds): (3.11+) cancels the enclosed code if it runs too long and converts that cancellation into TimeoutError at the async with boundary. Because it scopes a block, one deadline can cover several awaits (connect, send, read). Related tools:

• asyncio.timeout_at(loop.time() + n): an absolute deadline, ideal for propagating one budget through nested calls.
• cm.reschedule(new_deadline) on the context manager to extend or shorten it; timeout(None) means "no deadline yet".
• asyncio.wait_for(aw, timeout): the older single-awaitable form, now implemented with timeout.

Cancellation semantics

Cancellation is asyncio's mechanism for stopping work: timeouts, TaskGroup failures, and shutdown all use it. task.cancel() doesn't kill anything immediately. It schedules a CancelledError to be raised inside the task at its current (or next) await. The task can then run cleanup in finally blocks and async with exits, and must let the exception propagate.

• CancelledError inherits from BaseException, not Exception, so a generic except Exception: doesn't swallow it. Never catch it without re-raising (unless you are the code that requested the cancellation, and even then prefer the structured tools).
• Cleanup code may await (closing a connection), but keep it short: a second cancel can interrupt it.
• asyncio.shield(aw) protects an inner awaitable from cancellation of the outer task: the outer await is cancelled, but the inner operation keeps running. Use it rarely (for example, finishing a commit).
• Internally, tasks count cancellation requests (task.cancelling(), task.uncancel()); this is how timeout() and TaskGroup distinguish their own cancellations from yours.

Producer/consumer with asyncio.Queue

asyncio.Queue mirrors queue.Queue but its put/get are coroutines that suspend instead of blocking a thread. It's not thread-safe; it's for tasks on one loop. The standard worker-pool shape:

1. Create the queue (optionally with maxsize for back-pressure) and N worker tasks.
2. Each worker loops: item = await queue.get(), process, and queue.task_done() in a finally.
3. The producer puts items; then await queue.join() waits until every item is done.
4. Stop workers: cancel them, put one sentinel per worker, or (3.13+) call queue.shutdown(), which makes pending and future get() calls raise asyncio.QueueShutDown once the queue is empty.

Limiting concurrency and rate

Creating 10,000 tasks is cheap; opening 10,000 simultaneous connections to one server is rude and usually fails. Two different limits are often needed:

• Concurrency limit (how many at once): asyncio.Semaphore(n). Each task does async with sem: around the request; at most n are inside.
• Rate limit (how many per second): a token bucket or a minimum interval between starts. A semaphore alone doesn't bound rate: fast responses let tasks cycle through it quickly.

Async iterators, generators, and context managers

Some protocols need to await in the middle: fetching the next page of results, or closing a connection. Python has async versions of the iteration and context-manager protocols:

Mixing in blocking code

Not every library is async. A synchronous database driver, requests, urllib, or a CPU-heavy function will block the loop. The bridges:

• await asyncio.to_thread(func, *args) (3.9+): run a blocking function in the loop's default thread pool; perfect for blocking I/O (the GIL is released while it waits). This is how Project 5's real fetcher uses urllib.
• await loop.run_in_executor(executor, func, *args): the same with an executor of your choice; use a ProcessPoolExecutor (or InterpreterPoolExecutor) for CPU-bound work.
• asyncio.run_coroutine_threadsafe(coro, loop): the reverse direction, submitting a coroutine to a loop running in another thread (returns a concurrent.futures.Future).

Inspecting running async programs

Inside a program, the same information is available as an API. asyncio.capture_call_graph(task) returns a FutureCallGraph (the task, its coroutine call stack, and who awaits it); asyncio.print_call_graph(task) and format_call_graph render it. Omit the argument to describe the current task.

Preview: TaskGroup.cancel() in 3.15

In 3.14 you can get the same behavior portably with asyncio.wait(tasks, return_when=FIRST_COMPLETED) followed by cancelling the pending set, or by the "raise a private exception" pattern:

Common bugs and how to spot them

Practice

import asyncio

async def fetch_all(urls, fetch):
    async with asyncio.TaskGroup() as tg:
        tasks = {url: tg.create_task(fetch(url)) for url in dict.fromkeys(urls)}
    return {url: task.result() for url, task in tasks.items()}

import asyncio

async def bounded_map(func, items, limit):
    if limit < 1:
        raise ValueError("limit must be >= 1")
    sem = asyncio.Semaphore(limit)

    async def one(item):
        async with sem:
            return await func(item)

    async with asyncio.TaskGroup() as tg:
        tasks = [tg.create_task(one(item)) for item in items]
    return [t.result() for t in tasks]

import asyncio

async def with_fallback(aw, seconds, default):
    try:
        async with asyncio.timeout(seconds):
            return await aw
    except TimeoutError:
        return default

_TIMED_OUT = object()

async def first_ok(coros, seconds):
    for aw in coros:
        result = await with_fallback(aw, seconds, _TIMED_OUT)
        if result is not _TIMED_OUT:
            return result
    raise TimeoutError("every attempt timed out")

import asyncio
from contextlib import aclosing

async def countdown(n, log):
    try:
        for i in range(n, 0, -1):
            await asyncio.sleep(0)
            yield i
    finally:
        log.append("closed")

async def take(agen, k):
    out = []
    async with aclosing(agen) as g:
        async for item in g:
            out.append(item)
            if len(out) == k:
                break
    return out

import asyncio

async def process_all(items, handler, workers):
    queue = asyncio.Queue()
    results = [None] * len(items)
    for pair in enumerate(items):
        queue.put_nowait(pair)

    async def worker():
        while True:
            job = await queue.get()
            try:
                if job is None:
                    return
                index, item = job
                results[index] = await handler(item)
            finally:
                queue.task_done()

    async with asyncio.TaskGroup() as tg:
        for _ in range(workers):
            tg.create_task(worker())
        await queue.join()
        for _ in range(workers):
            queue.put_nowait(None)
    return results

Mini-project: an async rate-limited fetcher simulator

pixi init fetchsim
cd fetchsim
pixi add python=3.14 pytest
pixi run python fetchsim.py --requests 200 --concurrency 10 --rate 50
pixi run python -m pytest -q

Quiz

Recap

• asyncio runs many tasks on one thread, switching only at await; never block the loop.
• Coroutine functions create coroutine objects; Tasks schedule them concurrently. Keep references, or better, use a TaskGroup.
• TaskGroup gives structured concurrency and ExceptionGroups; handle them with except*.
• Use asyncio.timeout for deadlines; cancellation is an exception raised at an await, so clean up in finally and re-raise.
• asyncio.Queue (with task_done/join, sentinels, or 3.13's shutdown()) builds worker pools; Semaphore bounds concurrency; rate needs its own limiter.
• async for, async generators, and async with handle protocols that need to await; close generators with aclosing.
• to_thread and executors bridge blocking code; 3.14 adds python -m asyncio ps/pstree and capture_call_graph.

Going further

• asyncio documentation, especially Coroutines and Tasks and Developing with asyncio
• Call graph introspection (3.14)
• Notes on structured concurrency (Nathaniel J. Smith)
• PEP 654: Exception Groups and except*
• Next: M39 Performance & Profiling