04 - Async/Await Advanced Python: asyncio Internals, TaskGroup and Robusta Cancellation

The problem: one task fails, and the others?

You have a service that queries three sources together to build a page. One answers with an error. What happens to the other two? With asyncio.gather, the default answer is surprising: they keep running, with nobody waiting for their result. With asyncio.TaskGroup they are cancelled and the error reaches you packed in an ExceptionGroup. This article explains the difference, its costs, and how to write coroutines that let themselves be cancelled cleanly.

All the listings were run with Python 3.14.4 and use only the standard library: where the original called a real network, there is an asyncio.sleep in its place, which suspends the coroutine in the same way. TaskGroup, asyncio.timeout and except* arrived in Python 3.11, so the code requires at least that version. For a comparison with JavaScript's event loop model, see the article on the event loop.

How asyncio works: one thread, many coroutines

asyncio has a single event loop running on one thread. A coroutine is an async def function that can suspend itself at every await: at that point it yields control to the loop, which starts another. When the awaited operation is ready (a readable socket, an expired timer) the loop resumes the coroutine from where it stopped. "Simultaneous" work is therefore a well-made illusion: it is the waits that overlap, not the computations.

import asyncio
import time

async def fetch(nome: str) -> str:
    await asyncio.sleep(1)          # suspends the coroutine, not the thread
    return f"data from {nome}"

async def main():
    t0 = time.monotonic()
    risultati = await asyncio.gather(fetch("a"), fetch("b"), fetch("c"))
    print(risultati)
    print(f"total time: {time.monotonic() - t0:.1f} s")   # ~1 s, not 3 s

asyncio.run(main())

# Output:
# ['data from a', 'data from b', 'data from c']
# total time: 1.0 s

Three one-second coroutines finish in one second and not in three, because while one sleeps the other two have already started. asyncio.gather wraps each coroutine in a task, that is, it schedules it on the loop, and returns the results in the order in which you passed them. The measured time (1.0 s) holds for this machine and for simulated network waits: with a CPU computation in place of sleep the three coroutines would run one after the other, because none of them would ever yield the thread.

gather: the error that leaves work lying around

The documentation of gather is explicit: with return_exceptions=False, the default value, the first exception is propagated immediately to whoever is waiting, and the other awaitables are not cancelled and keep running. The listing makes this visible: three tasks, the second fails at 0.2 s, the third would have finished at 0.3 s.

import asyncio

async def lavoro(n: int) -> int:
    try:
        await asyncio.sleep(n * 0.1)
        if n == 2:
            raise ValueError(f"task {n} failed")
        print(f"task {n}: completed")
        return n * 10
    except asyncio.CancelledError:
        print(f"task {n}: cancelled")
        raise

async def main():
    try:
        await asyncio.gather(lavoro(1), lavoro(2), lavoro(3))
    except ValueError as e:
        print("gather raised:", e)
    await asyncio.sleep(0.5)        # give task 3 time to show itself
    print("end main")

asyncio.run(main())

# Output:
# task 1: completed
# gather raised: task 2 failed
# task 3: completed
# end main

The exception reaches main before task 3 has finished, but the line "task 3: completed" appears anyway, afterwards: the task worked toward a result that nobody will use. The try/except inside lavoro is there to print any cancellation; here it never triggers. In a real service that task could be a database write or a paid call, and every failed request would leave one alive.

The partial remedy is return_exceptions=True: the exceptions become elements of the list of results and gather waits for all of them.

import asyncio

async def lavoro(n: int) -> int:
    await asyncio.sleep(n * 0.1)
    if n == 2:
        raise ValueError(f"task {n} failed")
    return n * 10

async def main():
    risultati = await asyncio.gather(
        lavoro(1), lavoro(2), lavoro(3), return_exceptions=True
    )
    print(risultati)
    for r in risultati:
        print("error:" if isinstance(r, BaseException) else "ok:", r)

asyncio.run(main())

# Output:
# [10, ValueError('task 2 failed'), 30]
# ok: 10
# error: task 2 failed
# ok: 30

No task is left orphaned, but the price is that the error no longer interrupts anything: you have to walk the list and tell results from exceptions with isinstance, and if you forget to do so the error disappears silently. It is a legitimate choice when you want every result you can get (a price aggregator that shows the sources it reached), and a wrong one when a failure makes the rest useless.

TaskGroup: the block that leaves nobody behind

asyncio.TaskGroup, new in Python 3.11, applies structured concurrency: tasks are born inside an async with block and cannot outlive it. The documentation says that on the first exception other than CancelledError the group's remaining tasks are cancelled, and that the block exits only when all of them have finished. The errors leave the block grouped in an ExceptionGroup, hence the new except*.

import asyncio

async def lavoro(n: int) -> int:
    try:
        await asyncio.sleep(n * 0.1)
        if n == 2:
            raise ValueError(f"task {n} failed")
        print(f"task {n}: completed")
        return n * 10
    except asyncio.CancelledError:
        print(f"task {n}: cancelled")
        raise

async def main():
    try:
        async with asyncio.TaskGroup() as tg:
            t1 = tg.create_task(lavoro(1))
            tg.create_task(lavoro(2))
            tg.create_task(lavoro(3))
    except* ValueError as eg:
        print("group:", type(eg).__name__, [str(e) for e in eg.exceptions])
    print("t1.result() =", t1.result())
    await asyncio.sleep(0.5)
    print("end main: no orphan task")

asyncio.run(main())

# Output:
# task 1: completed
# task 3: cancelled
# group: ExceptionGroup ['task 2 failed']
# t1.result() = 10
# end main: no orphan task

Same scenario as before, opposite outcome. Task 1 finishes normally and you keep its result; task 2 fails; task 3 receives a CancelledError (the "cancelled" line) instead of carrying on. After the block, t1.result() is 10 and the extra half-second wait shows no residual task. Note that the error is an ExceptionGroup even though only one task failed: for the first time the code that catches it must write except*.

ExceptionGroup and except*: several errors in one

If two tasks fail in the same turn of the loop, a single exception is not enough to tell what happened. PEP 654 introduced ExceptionGroup to collect unrelated errors, and except* to handle them by type: each clause takes the subgroups of the indicated type and lets the rest pass.

import asyncio

async def solleva(tipo, msg):
    await asyncio.sleep(0.1)
    raise tipo(msg)

async def main():
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(solleva(ValueError, "A"))
            tg.create_task(solleva(ValueError, "B"))
            tg.create_task(solleva(TypeError, "C"))
    except* ValueError as eg:
        print("ValueError:", sorted(str(e) for e in eg.exceptions))
    except* TypeError as eg:
        print("TypeError:", sorted(str(e) for e in eg.exceptions))

asyncio.run(main())

# A nested group: eg.exceptions in turn contains a group.
try:
    raise ExceptionGroup("outer", [
        ValueError("v1"),
        ExceptionGroup("inner", [KeyError("k1"), KeyError("k2")]),
    ])
except* KeyError as eg:
    print("exceptions:", eg.exceptions)
    print("split:", eg.split(KeyError)[0].exceptions[0].exceptions)
except* ValueError as eg:
    print("ValueError:", eg.exceptions)

# Output:
# ValueError: ['A', 'B']
# TypeError: ['C']
# exceptions: (ExceptionGroup('inner', [KeyError('k1'), KeyError('k2')]),)
# split: (KeyError('k1'), KeyError('k2'))
# ValueError: (ValueError('v1'),)

In the first part, the two ValueErrors land in the except* ValueError clause and the TypeError in the other; both clauses run, unlike the normal except, which stops at the first match. The second part corrects an easy mistake to make: with nested groups, eg.exceptions is not the flat list of the KeyErrors, but still contains the inner group. To get to the individual exceptions you need to go down one level or use split, which returns a group with the same structure but only the exceptions of the requested type.

Cancellation: cooperative, and easy to break

Task.cancel() kills nothing: it asks the task to receive an asyncio.CancelledError at the next await. The documentation recommends using try/finally for cleanup and, if the CancelledError is caught, re-raising it once cleanup is done. The reason is structural: TaskGroup and asyncio.timeout are built on cancellation and, in the words of the documentation, may misbehave if a coroutine swallows it. The listing compares the two versions.

import asyncio

async def corretto():
    try:
        await asyncio.sleep(10)
    except asyncio.CancelledError:
        print("correct: cleanup, then re-raise")
        raise

async def inghiotte(fermati: asyncio.Event):
    while not fermati.is_set():
        try:
            await asyncio.sleep(10)
        except asyncio.CancelledError:
            print("swallow: cancellation ignored")   # the raise is missing

async def main():
    t = asyncio.create_task(corretto())
    await asyncio.sleep(0.1)
    t.cancel()
    await asyncio.wait([t])
    print("correct: cancelled() =", t.cancelled())

    fermati = asyncio.Event()
    t = asyncio.create_task(inghiotte(fermati))
    await asyncio.sleep(0.1)
    t.cancel()
    await asyncio.wait([t], timeout=1)
    print("swallow: after 1 s done() =", t.done(), "cancelled() =", t.cancelled())
    fermati.set()                  # only to end the example
    t.cancel()
    await asyncio.wait([t])
    print("swallow: now done() =", t.done(), "cancelled() =", t.cancelled())

asyncio.run(main())

# Output:
# correct: cleanup, then re-raise
# correct: cancelled() = True
# swallow: cancellation ignored
# swallow: after 1 s done() = False cancelled() = False
# swallow: cancellation ignored
# swallow: now done() = True cancelled() = False

The corretto task (the correct one) cleans up, re-raises, and ends up cancelled. The inghiotte task (the one that swallows) prints "cancellation ignored" and carries on: after one second done() is still false. Here a while loop makes the problem obvious; I had to use an Event to make the example end, and in a real program the task would have kept the request alive forever. In a TaskGroup it would mean the block never exits. Note the corollary too: CancelledError is a subclass of BaseException, so a plain except Exception does not catch it, and it should be caught only on purpose.

Timeouts: asyncio.timeout and wait_for

The normal way to set a deadline is asyncio.timeout (Python 3.11): a context manager that cancels the code inside it when time runs out and turns it into a TimeoutError outside the block. The older wait_for wraps a single awaitable and does the same. The practical difference is that the context manager covers several statements and combines with other blocks.

import asyncio

async def lenta():
    await asyncio.sleep(5)

async def main():
    try:
        async with asyncio.timeout(0.2):
            await lenta()
    except TimeoutError:
        print("asyncio.timeout: TimeoutError")
    try:
        await asyncio.wait_for(lenta(), timeout=0.2)
    except asyncio.TimeoutError as e:
        print("wait_for: asyncio.TimeoutError is TimeoutError:", asyncio.TimeoutError is TimeoutError)

asyncio.run(main())

# Output:
# asyncio.timeout: TimeoutError
# wait_for: asyncio.TimeoutError is TimeoutError: True

Both deadlines produce TimeoutError. The second check in the listing shows that since Python 3.11 asyncio.TimeoutError is an alias of the built-in exception (the documentation defines it as a "deprecated alias"): I verified the identity with is, so an except TimeoutError is enough everywhere. Code written for 3.10 or earlier, where the two names were different classes, keeps working. The deadline, remember, works only if the coroutine underneath lets itself be cancelled: what was said above applies.

Limiting concurrency: semaphore and queue

A TaskGroup sets no limits: if you hand it ten thousand URLs, one request starts for each, with the risk of exhausting sockets or triggering the server's rate limit. The limit is obtained with an asyncio.Semaphore(N): at most N coroutines can be inside the async with sem block, the others wait their turn. To make it verifiable I counted the peak of simultaneous requests over a hundred URLs with a limit of 10.

import asyncio

attivi = 0
picco = 0

async def scarica(url: str, sem: asyncio.Semaphore) -> dict:
    global attivi, picco
    async with sem:                       # at most N inside this block
        attivi += 1
        picco = max(picco, attivi)
        await asyncio.sleep(0.05)         # in place of the network request
        attivi -= 1
        return {"url": url, "status": 200}

async def main():
    sem = asyncio.Semaphore(10)
    urls = [f"https://example.test/{i}" for i in range(100)]
    async with asyncio.TaskGroup() as tg:
        task = [tg.create_task(scarica(u, sem)) for u in urls]
    print(len(task), "responses, peak of simultaneous requests:", picco)

asyncio.run(main())

# Output:
# 100 responses, peak of simultaneous requests: 10

A hundred tasks are created at once, but the peak of those "inside" stays at 10: the counter is incremented and decremented around the point where a real request would be in flight. The advantage over a fixed pool of workers is simplicity; the limit is that all hundred tasks exist in memory from the start, so for millions of items the alternative route is better, a bounded queue with a fixed number of consumers.

import asyncio

async def main():
    coda: asyncio.Queue = asyncio.Queue(maxsize=5)   # bounded: the producer slows down
    risultati = []
    massimo = 0

    async def produttore():
        for i in range(30):
            await coda.put(i)             # waits if there are already 5 items
        await coda.put(None)              # sentinel: nothing more is coming

    async def consumatore():
        nonlocal massimo
        while (voce := await coda.get()) is not None:
            massimo = max(massimo, coda.qsize())
            await asyncio.sleep(0.01)     # slow processing
            risultati.append(voce * 2)

    async with asyncio.TaskGroup() as tg:
        tg.create_task(produttore())
        tg.create_task(consumatore())

    print(len(risultati), "processed, queue never above", massimo)

asyncio.run(main())

# Output:
# 30 processed, queue never above 4

The Queue(maxsize=5) is the brake: when it holds five items, await coda.put(i) suspends the producer until the consumer takes one. The queue never exceeded four items in the consumer's observation, and all thirty values arrived in order. The final None is a sentinel: without it, the consumer would wait forever and the TaskGroup would not exit. The topic of the producer faster than the consumer is covered separately in backpressure and flow control.

gather or TaskGroup: how to choose

In new code the default choice is TaskGroup, but gather still has a place. The criterion is what you want to happen when a task fails.

• TaskGroup: the tasks are parts of a single operation and one that fails makes the rest useless; you want no orphans and tasks identifiable by name.
• gather with return_exceptions=True: the tasks are independent and you want the partial results, accepting that you check every element.
• gather by default: almost never; unless you know that, at the first error, the whole program ends anyway.

The third item is the most important: the default behavior of gather is what produces the orphaned tasks seen earlier. gather is not "wrong", but it predates structured concurrency and does not guarantee it. If your minimum Python version is earlier than 3.11, you will have to stay on gather and handle cancellation by hand.

Debugging: naming tasks

When a service has hundreds of tasks, a traceback that says "Task-37" does not help. The name parameter of create_task solves that, and asyncio.all_tasks() lists what is alive.

import asyncio

async def lavoro():
    await asyncio.sleep(1)

async def main():
    async with asyncio.TaskGroup() as tg:
        tg.create_task(lavoro(), name="write-db")
        tg.create_task(lavoro(), name="update-cache")
        await asyncio.sleep(0.1)
        for t in sorted(asyncio.all_tasks(), key=lambda t: t.get_name()):
            print(t.get_name(), "done =", t.done())

asyncio.run(main())

# Output:
# Task-1 done = False
# update-cache done = False
# write-db done = False

The three tasks (two created with a name and the one running main, which Python calls Task-1) are all unfinished at the time of the check. The order is the one I imposed with sorted: all_tasks() returns a set, with no guaranteed order. To look for coroutines that were never awaited, the debug mode (asyncio.run(main(), debug=True) or the PYTHONASYNCIODEBUG=1 variable) logs, according to the documentation, callbacks that take longer than 100 ms and makes the warning about coroutines never awaited more detailed; I did not run it here.

One last warning from the documentation: the event loop keeps only weak references to tasks. A create_task whose result is not stored can be collected by the garbage collector in the middle of execution. Inside a TaskGroup the problem does not exist because the group keeps the references; outside one, keep the tasks in a set.

What you can do now

You know why gather leaves orphaned tasks and how TaskGroup avoids it, you can read and filter an ExceptionGroup with except* without falling into the nested-groups trap, and you can write a coroutine that lets itself be cancelled. You can also limit concurrency with a semaphore or a bounded queue, and name tasks so you can find them again in the logs.

The next article looks at the same problem from the JVM side: virtual threads and structured concurrency in Java, where instead of coroutines you write blocking code on very lightweight threads. For the picture in which asyncio sits beside the other models, see the comparison of concurrency models. The same problem of failing tasks and cancellation, in Rust, is in the article on Rust async and Tokio.

The timings (1.0 s) and the counts (peak 10, maximum queue 4) come from single runs with Python 3.14.4 on a single machine; the exact values change from one run to another.