Structured concurrency

create_task and gather from asyncio fundamentals have a weakness. When one task fails, the others keep running, unsupervised, and their errors can go unnoticed. Structured concurrency fixes this with one rule: tasks live inside a block, and the block does not end until every task in it has finished. In Python that block is asyncio.TaskGroup, added in 3.11.

async with

A TaskGroup is an asynchronous context manager. It works like the context managers from the intermediate level, except that entering and leaving can pause, so you write async with. It is only allowed inside async def. Advanced contextlib shows how to write your own with @asynccontextmanager.

TaskGroup

tg.create_task() starts a task in the group. Leaving the async with block waits for all of them:

import asyncio


async def fetch(name, delay):
    await asyncio.sleep(delay)
    return f"{name}: 200 OK"


async def main():
    async with asyncio.TaskGroup() as tg:
        home = tg.create_task(fetch("home", 0.2))
        about = tg.create_task(fetch("about", 0.1))
    # Leaving the block waited for both tasks.
    print(home.result())
    print(about.result())


asyncio.run(main())

home: 200 OK
about: 200 OK

After the block, every task is done, so result() is safe to call. No task can outlive the code that started it.

When a task fails

If any task raises, the group cancels the others, waits for them to finish cancelling, and raises an ExceptionGroup containing the failures. Exception notes and groups taught except* for exactly this:

import asyncio


async def slow_download():
    try:
        await asyncio.sleep(10)
        return "big file"
    except asyncio.CancelledError:
        print("download cancelled, cleaning up")
        raise


async def broken_parser():
    await asyncio.sleep(0)
    raise ValueError("bad header")

async def main():
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(slow_download())
            tg.create_task(broken_parser())
    except* ValueError as group:
        print("failed:", [str(e) for e in group.exceptions])


asyncio.run(main())

download cancelled, cleaning up
failed: ['bad header']

The program ends at once, not after ten seconds. The download received asyncio.CancelledError at its await, ran its cleanup and re-raised. The group always raises an ExceptionGroup, even for a single failure, so catch it with except*, not plain except ValueError.

Cancellation

Cancelling a task throws asyncio.CancelledError into it at the await where it is paused. task.cancel() requests it:

import asyncio


async def heartbeat():
    beats = 0
    try:
        while True:
            beats += 1
            await asyncio.sleep(0)
    finally:
        print("heartbeat stopped after", beats, "beats")


async def main():
    task = asyncio.create_task(heartbeat())
    for _ in range(3):
        await asyncio.sleep(0)
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print("cancelled:", task.cancelled())


asyncio.run(main())

heartbeat stopped after 3 beats
cancelled: True

Use finally or except CancelledError: ... raise to clean up. Never swallow CancelledError without re-raising it: TaskGroup and timeout both rely on it reaching them. CancelledError derives from BaseException, so a plain except Exception does not catch it by accident.

Timeouts

asyncio.timeout(delay) is an async context manager that cancels the code inside it when the time runs out, and turns that cancellation into a TimeoutError as the block exits:

import asyncio


async def slow_query():
    await asyncio.sleep(10)
    return "rows"


async def main():
    try:
        async with asyncio.timeout(0.1) as deadline:
            print(await slow_query())
    except TimeoutError:
        print("query timed out")
    print("expired:", deadline.expired())


asyncio.run(main())

query timed out
expired: True

The try goes around the async with. Inside the block the code sees CancelledError. Only when the block exits does it become TimeoutError. asyncio.timeout(None) means no limit, and deadline.reschedule() can move the deadline while it runs. Both timeout and TaskGroup arrived in 3.11. In older code you will see asyncio.wait_for(coro, timeout), which still works.

A timeout around a TaskGroup puts one deadline on all its tasks. A timeout inside each task gives each its own deadline, as the first exercise shows.

Producers and consumers with asyncio.Queue

asyncio.Queue is the async counterpart of the queue.Queue from Threads: await queue.put(item) and await queue.get() pause instead of blocking a thread. With maxsize, put waits while the queue is full, so a fast producer cannot run far ahead of slow consumers. That is backpressure. A TaskGroup holds the producer and a fixed number of workers, which also caps how many fetches run at once:

import asyncio

PAGES = {"/": 13, "/about": 15, "/blog": 40, "/contact": 18, "/shop": 27}
WORKERS = 2


async def producer(queue):
    for url in PAGES:
        await queue.put(url)  # waits while the queue is full
    for _ in range(WORKERS):
        await queue.put(None)  # one stop signal per worker


async def worker(queue, sizes):
    while (url := await queue.get()) is not None:
        await asyncio.sleep(0.01)  # fetch the page
        sizes[url] = PAGES[url]


async def main():
    queue = asyncio.Queue(maxsize=2)
    sizes = {}
    async with asyncio.TaskGroup() as tg:
        tg.create_task(producer(queue))
        for _ in range(WORKERS):
            tg.create_task(worker(queue, sizes))
    print(sorted(sizes.items()))
    print("total:", sum(sizes.values()))


asyncio.run(main())

[('/', 13), ('/about', 15), ('/blog', 40), ('/contact', 18), ('/shop', 27)]
total: 113

The producer finishes with one None per worker, the same sentinel pattern as with threads, and the group ends when every worker has returned. If a worker raised, the group would cancel the producer and the other workers. 3.13+Requires version 3.13 or later queue.shutdown() is the alternative to sentinels: after it, get on an empty queue raises asyncio.QueueShutDown. To cap concurrency without a queue, wrap the work in async with asyncio.Semaphore(n):.

Stopping a group early

3.15+Requires version 3.15 or later Sometimes the group’s job is done before every task is: you asked three mirrors for the same file and one answered. TaskGroup.cancel() cancels all remaining tasks and leaves the block normally, without raising:

import asyncio


async def download(mirror, delay, tg, log):
    try:
        await asyncio.sleep(delay)
    except asyncio.CancelledError:
        log["cancelled"].append(mirror)
        raise
    log["winner"] = mirror
    tg.cancel()  # we have what we need: stop the whole group


async def main():
    log = {"winner": None, "cancelled": []}
    async with asyncio.TaskGroup() as tg:
        tg.create_task(download("eu", 0.5, tg, log))
        tg.create_task(download("us", 0.1, tg, log))
        tg.create_task(download("asia", 0.9, tg, log))
    print("winner:", log["winner"])
    print("cancelled:", sorted(log["cancelled"]))


asyncio.run(main())

winner: us
cancelled: ['asia', 'eu']

Before 3.15 the usual workaround was to raise a private exception from the winning task and suppress it outside the group. The cancelled list is sorted because the order in which the group cancels its tasks is not specified.

Seeing what is running

3.14+Requires version 3.14 or later When an async program hangs, python -m asyncio ps PID and python -m asyncio pstree PID show its tasks from another terminal, without changing the program. Here pstree inspects a program whose TaskGroup is still waiting for a task called stuck (paths shortened):

python3.15 -m asyncio pstree 19348

└── (T) Task-1
    └──  main stuck.py:11
        └──  TaskGroup.__aexit__ .../asyncio/taskgroups.py:75
            └──  TaskGroup._aexit .../asyncio/taskgroups.py:125
                └── (T) stuck
                    └──  fetch stuck.py:6
                        └──  sleep .../asyncio/tasks.py:721

Each (T) is a task, and below it is the stack where it is paused. Naming tasks, with tg.create_task(coro, name="stuck"), makes this output far easier to read. ps prints the same information as a flat table. These tools may need extra permissions to inspect another process. 3.15+Requires version 3.15 or later A --retries N option retries when the target changes while it is being read.

When not to use a TaskGroup

Use a TaskGroup whenever tasks belong together and one failure should stop the rest. When you want every result, failures included, asyncio.gather(..., return_exceptions=True) collects them without cancelling anything. A background task that should outlive the request that started it, such as a cache refresher, belongs in a group owned by something longer-lived, like the application’s main coroutine, not in the request handler. And for one coroutine, plain await is enough.

Common pitfalls

Catching TimeoutError inside the timeout block. It is only raised at the block’s exit. See the second exercise.

Swallowing CancelledError. A task that catches it and carries on breaks timeouts and task groups. Clean up, then re-raise.

Plain except around a TaskGroup. Failures arrive as an ExceptionGroup. Use except*.

Fire-and-forget tasks. A task created with asyncio.create_task and never awaited can fail silently. Put it in a TaskGroup instead.

/blog  timed out
/home  status 200
/news  status 200
/shop  error: connection refused

import asyncio

FAKE_SERVERS = {  # url: (seconds to answer, what happens)
    "/home": (0.1, "ok"),
    "/news": (0.2, "ok"),
    "/shop": (0.1, "refused"),
    "/blog": (5.0, "ok"),
}


async def fetch(url):
    delay, outcome = FAKE_SERVERS[url]
    await asyncio.sleep(delay)
    if outcome == "refused":
        raise ConnectionRefusedError("connection refused")
    return 200

async def checked_fetch(url, report):
    try:
        async with asyncio.timeout(0.5):
            report[url] = f"status {await fetch(url)}"
    except TimeoutError:
        report[url] = "timed out"
    except ConnectionError as error:
        report[url] = f"error: {error}"


async def main():
    report = {}
    async with asyncio.TaskGroup() as tg:
        for url in FAKE_SERVERS:
            tg.create_task(checked_fetch(url, report))
    for url in sorted(report):
        print(f"{url:6} {report[url]}")


asyncio.run(main())

/blog  timed out
/home  status 200
/news  status 200
/shop  error: connection refused

import asyncio


async def main():
    async with asyncio.timeout(0.1):
        try:
            await asyncio.sleep(10)
        except TimeoutError:
            print("timed out, using cached data")
    print("carrying on")


try:
    asyncio.run(main())
except TimeoutError:
    print("TimeoutError escaped from main()")

TimeoutError escaped from main()

import asyncio


async def main():
    try:
        async with asyncio.timeout(0.1):
            await asyncio.sleep(10)
    except TimeoutError:
        print("timed out, using cached data")
    print("carrying on")


try:
    asyncio.run(main())
except TimeoutError:
    print("TimeoutError escaped from main()")

timed out, using cached data
carrying on

import asyncio


async def check_disk():
    await asyncio.sleep(0)
    raise ValueError("disk full")


async def check_config():
    raise KeyError("port")


async def main():
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(check_disk())
            tg.create_task(check_config())
    except* ValueError as group:
        print("ValueError:", len(group.exceptions))
    except* KeyError as group:
        print("KeyError:", len(group.exceptions))


asyncio.run(main())

ValueError: 1
KeyError: 1

Summary

• async with asyncio.TaskGroup() as tg: starts tasks with tg.create_task() and does not exit until all of them are done.
• If a task fails, the group cancels the rest and raises an ExceptionGroup. Handle it with except*.
• Cancellation arrives as CancelledError at an await. Clean up in finally, and re-raise.
• asyncio.timeout(delay) raises TimeoutError at the block’s exit, so put the try around the block.
• 3.15+Requires version 3.15 or later tg.cancel() stops a whole group early without an error.
• A bounded asyncio.Queue connects producers and consumers and applies backpressure.
• 3.14+Requires version 3.14 or later python -m asyncio ps and pstree show the tasks of a running program.
• Next: free-threaded Python, threads without the GIL.