The group booking falls through: structured concurrency

Half past ten, and a group fills six tables at once. Lou takes the orders and passes them to the kitchen, and the kitchen starts all six courses together. Ten minutes later, a hand goes up at table 3. Someone forgot to mention a peanut allergy. The head of the group says, "In that case, we'll all call it a night."

Minho doesn't hear and keeps the stew for table 5 bubbling. Sua turns off his burner. "Nobody ever made a rule that if one table drops out, everything stops."

Chef Yoon Seoyeon presses her timer. "Work that starts together has to end together. Tonight's job is making that promise in code."

10-1. The gap in gather: one fails, the rest keep boiling (5 min)

Let's handle the group booking with Lesson 9's asyncio.gather. Table 2 fails after 0.1 seconds.

import asyncio

async def table(n, sec, fail=False):
    await asyncio.sleep(sec)
    if fail:
        raise ValueError(f"T{n} allergy")
    print(f"T{n} served")

async def main():
    try:
        await asyncio.gather(table(1, 0.3), table(2, 0.1, fail=True), table(3, 0.5))
    except ValueError as e:
        print("caught:", e)
    await asyncio.sleep(0.6)   # keep the place open a little longer

asyncio.run(main())
# caught: T2 allergy
# T1 served
# T3 served

• gather passes the first exception straight up. That's why caught: prints first.
• But the other two tables are not cancelled. Dishes nobody is waiting for keep boiling in the background and print served.
• In a real program, that means serving food to guests who already left, or requests whose results nobody will ever read still running.

gather(return_exceptions=True) just puts the exceptions in the result list; it doesn't stop the rest either. If you want "one fails, everything stops," you need a different tool.

10-2. TaskGroup: work that starts together ends together (12 min)

asyncio.TaskGroup (3.11) handles tasks as one bundle. Managing tasks in bundles like this is called structured concurrency.

import asyncio

async def table(n, sec):
    await asyncio.sleep(sec)
    return f"T{n}"

async def main():
    async with asyncio.TaskGroup() as tg:
        tasks = [tg.create_task(table(n, 0.1 * n)) for n in range(1, 4)]
    # once you're here, all three tasks are done
    print([t.result() for t in tasks])

asyncio.run(main())
# ['T1', 'T2', 'T3']

A TaskGroup makes three promises.

1. When you leave the async with block, every task created inside it is already finished. No separate await needed.
2. If one task raises an exception, it cancels all the others still running.
3. Once the cancelling is done, it gathers the exceptions that failed into a single ExceptionGroup and raises that.

Since tg.create_task keeps hold of each task, you don't have to worry about Lesson 9's "task that vanishes because nobody kept a reference," either. Now, the group booking where table 3 fails.

import asyncio

async def table(n, sec, fail=False):
    try:
        await asyncio.sleep(sec)
        if fail:
            raise ValueError(f"T{n} allergy")
        print(f"T{n} served")
    except asyncio.CancelledError:
        print(f"T{n} cancelled")
        raise                      # re-raise the cancellation you received

async def main():
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(table(1, 0.05))
            tg.create_task(table(2, 0.3))
            tg.create_task(table(3, 0.1, fail=True))
            tg.create_task(table(4, 0.5))
    except* ValueError as eg:
        print("group failed:", eg.exceptions)
    await asyncio.sleep(0.6)

asyncio.run(main())
# T1 served
# T2 cancelled      (T2 and T4 can swap order from run to run)
# T4 cancelled
# group failed: (ValueError('T3 allergy'),)

• T1 finished before the failure, so it's served as usual.
• The moment T3 fails at 0.1 seconds, cancellation goes out to T2 and T4, which were still boiling. The order in which the cancel messages print can differ between runs.
• Even after waiting another 0.6 seconds, no more served lines appear. Unlike gather, no pots are secretly boiling in the background.
• Inside a cancelled task, asyncio.CancelledError is raised at the await. That's why you can catch it with except and clean up. Always re-raise a cancellation you catch (10-5).

The timeline below runs a real TaskGroup on the server. The higher the table number, the slightly longer the dish takes (T1 0.3 s, T2 0.345 s, … T6 about 0.53 s). The table you pick as "Table that fails" fails right around when its own dish would have finished.

• With the starting setting (table 3 fails), T1 and T2 finish first, and the moment T3 fails at about 0.4 seconds, the three bars for T4 to T6 are cut off in gray at the same instant. T1 and T2, already done, stay as they are.
• Make table 1 fail and the other five are all cut off at 0.3 seconds; make table 6 fail and only T6 fails, after the other five have all finished.
• Set the failing table to "None" and all six bars run to the end. The slowest, T6, finishes at about 0.53 seconds.
• Drop the time limit to about 0.4 seconds and only the first few tables finish; whichever tables are still going at the limit are cut off in gray. Dishes that already finished are not undone.

10-3. ExceptionGroup and except* (7 min)

Several tasks can also fail at the same time. That's why a TaskGroup raises a bundle of exceptions rather than a single one.

import asyncio

async def boom(exc, sec):
    await asyncio.sleep(sec)
    raise exc

async def main():
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(boom(ValueError("T2 allergy"), 0.1))
            tg.create_task(boom(KeyError("T5"), 0.1))
    except Exception as e:
        print(type(e).__name__, "|", e)
        print(e.exceptions)

asyncio.run(main())
# ExceptionGroup | unhandled errors in a TaskGroup (2 sub-exceptions)
# (ValueError('T2 allergy'), KeyError('T5'))

An ExceptionGroup (3.11) is an exception that holds a list of exceptions in .exceptions. To pick them out of the bundle by type, use except*.

eg = ExceptionGroup("kitchen", [ValueError("T2 allergy"), KeyError("T5"), ValueError("T6 salt")])

try:
    raise eg
except* ValueError as g:
    print("ValueError:", [str(e) for e in g.exceptions])
except* KeyError as g:
    print("KeyError:", g.exceptions)
# ValueError: ['T2 allergy', 'T6 salt']
# KeyError: (KeyError('T5'),)

• With ordinary except, only one block runs. With except*, every matching block runs. Above, both blocks run.
• g is always an ExceptionGroup: a smaller bundle holding just the matching exceptions.
• Any exception that no except* matches is raised again as a leftover bundle.

eg = ExceptionGroup("kitchen", [ValueError("T2"), KeyError("T5")])
try:
    try:
        raise eg
    except* ValueError:
        print("handled ValueError")
except ExceptionGroup as rest:
    print("left:", rest.exceptions)
print(isinstance(eg, Exception), eg.subgroup(ValueError).exceptions)
# handled ValueError
# left: (KeyError('T5'),)
# True (ValueError('T2'),)

except* comes with two rules. In 3.14, breaking either one is a syntax error, caught before anything runs.

src1 = '''
def f():
    try:
        pass
    except* ValueError:
        return 1
'''
src2 = '''
try:
    pass
except ValueError:
    pass
except* KeyError:
    pass
'''
for src in (src1, src2):
    try:
        compile(src, "<menu>", "exec")
    except SyntaxError as e:
        print("SyntaxError:", e.msg)
# SyntaxError: 'break', 'continue' and 'return' cannot appear in an except* block
# SyntaxError: cannot have both 'except' and 'except*' on the same 'try'

• You can't use return, break or continue inside an except* block. Several blocks have to run one after another, and jumping out halfway would break that. Store the result in a variable and return it outside the try.
• You can't mix except and except* on one try. If you need both, split it into two nested trys. The engine behind this course's labs does exactly that.

10-4. Time limits: asyncio.timeout and wait_for (8 min)

Tonight's second rule is "no dish runs past its time limit." asyncio.timeout() (3.11) puts a time limit on a whole block.

import asyncio, time

async def stew(sec):
    await asyncio.sleep(sec)
    return "stew"

async def main():
    start = time.perf_counter()
    try:
        async with asyncio.timeout(0.2):
            await stew(1.0)
    except TimeoutError:
        print(f"timeout after {time.perf_counter() - start:.2f}s")

    try:
        print(await asyncio.wait_for(stew(0.1), timeout=0.2))
        print(await asyncio.wait_for(stew(1.0), timeout=0.2))
    except TimeoutError:
        print("wait_for: timeout")

    print(asyncio.TimeoutError is TimeoutError)

asyncio.run(main())
# timeout after 0.20s
# stew
# wait_for: timeout
# True

• When time runs out, the work inside the block is cancelled, and the built-in TimeoutError comes out of the block. Since 3.11, asyncio.TimeoutError is the same thing as the built-in TimeoutError.
• asyncio.wait_for(coroutine, timeout=) is the older way to put a limit on a single coroutine. It's handy when you want the result back directly.
• asyncio.timeout can wrap several awaits in one block, as in "20 minutes total, from taking the order to serving it."

You can also set the limit as a point in time ("at 10:45") instead of "in so many seconds," and you can change it after entering the block.

import asyncio, time

async def main():
    loop = asyncio.get_running_loop()
    start = time.perf_counter()
    deadline = loop.time() + 0.3            # a point in time: 0.3 s from now
    try:
        async with asyncio.timeout_at(deadline):
            await asyncio.sleep(1)
    except TimeoutError:
        print(f"timeout_at: {time.perf_counter() - start:.1f}s")

    start = time.perf_counter()
    try:
        async with asyncio.timeout(None) as cm:   # no limit at first
            cm.reschedule(loop.time() + 0.2)      # set it to 0.2 s later on
            print(cm.when() is not None)
            await asyncio.sleep(1)
    except TimeoutError:
        print(f"reschedule: {time.perf_counter() - start:.1f}s, expired={cm.expired()}")

asyncio.run(main())
# timeout_at: 0.3s
# True
# reschedule: 0.2s, expired=True

• timeout_at(when) takes an absolute time on the loop.time() clock. It's good when several steps share a single deadline.
• timeout(None) starts with no limit. With cm.reschedule(when) you can set the limit later or push it back.

10-5. Cancellation: cancel, CancelledError, finally (6 min)

TaskGroup and time limits end up doing the same thing: sending a cancellation to a task. To send one yourself, use task.cancel().

import asyncio

async def braise():
    print("stove on")
    try:
        await asyncio.sleep(10)
    finally:
        print("stove off")          # even if cancelled, turn the burner off

async def main():
    task = asyncio.create_task(braise())
    await asyncio.sleep(0.1)
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print("cancelled:", task.cancelled())

asyncio.run(main())
# stove on
# stove off
# cancelled: True

• cancel() doesn't kill the task on the spot. It only makes CancelledError go off at the next await where the task is paused.
• Put cleanup, such as turning off the burner or closing a connection, in try/finally. It always runs, whether the task was cancelled, succeeded or failed.
• CancelledError inherits from BaseException, not Exception. That's by design, so an except Exception: doesn't swallow a cancellation by accident.

What happens if you swallow the cancellation?

import asyncio, time

async def stubborn():
    try:
        await asyncio.sleep(0.5)
    except asyncio.CancelledError:
        pass                          # swallowed the cancellation
    await asyncio.sleep(0.5)          # and kept on cooking
    return "late"

async def main():
    start = time.perf_counter()
    try:
        async with asyncio.timeout(0.1):
            print(await stubborn())
    except TimeoutError:
        print("timeout")
    print(f"{time.perf_counter() - start:.1f}s")

asyncio.run(main())
# late
# 0.6s

We set a 0.1-second limit, yet it took 0.6 seconds and no TimeoutError was raised. A time limit works through cancellation, and the coroutine ate that cancellation. A TaskGroup's "cancel the rest" falls apart in exactly the same way. So if you catch CancelledError, do your cleanup and raise it again (raise).

10-6. shield blocks cancellation, cancelling counts it (2 min)

Some work, like saving a receipt, must not be cut off halfway. asyncio.shield() keeps a cancellation from outside from reaching the task inside.

import asyncio

async def save_receipt():
    await asyncio.sleep(0.3)
    print("receipt saved")
    return "ok"

async def main():
    inner = asyncio.create_task(save_receipt())
    try:
        async with asyncio.timeout(0.1):
            await asyncio.shield(inner)
    except TimeoutError:
        print("waiter gave up")
    print(await inner)              # the inner task kept running

asyncio.run(main())
# waiter gave up
# receipt saved
# ok

The side that was waiting gave up at 0.1 seconds, but the inner task ran to the end. You have to keep a reference to the inner task to collect its result later.

Task.cancelling() returns the number of cancellation requests the task has received, and Task.uncancel() lowers that number by one. asyncio.timeout and TaskGroup use them internally to tell "is this cancellation one I sent?" You'll rarely need them yourself.

import asyncio

async def worker():
    try:
        await asyncio.sleep(1)
    except asyncio.CancelledError:
        me = asyncio.current_task()
        print("cancelling:", me.cancelling())
        me.uncancel()
        print("after uncancel:", me.cancelling())
        return "rescued"

async def main():
    t = asyncio.create_task(worker())
    await asyncio.sleep(0.05)
    t.cancel()
    print(await t)

asyncio.run(main())
# cancelling: 1
# after uncancel: 0
# rescued

Closing time (10 min)

Now both of tonight's rules go into one kitchen. Each table gets a 0.6-second limit, and all six tables are bundled into one TaskGroup.

import asyncio, time

async def table(n, sec, allergy, stopped):
    try:
        async with asyncio.timeout(0.6):     # no dish runs past 0.6 s
            await asyncio.sleep(sec)
            if n == allergy:
                raise ValueError(f"T{n} allergy")
            return f"T{n}"
    except asyncio.CancelledError:
        stopped.append(f"T{n}")             # burner off, hands up
        raise

async def service(menu, allergy=None):
    start, stopped = time.perf_counter(), []
    try:
        async with asyncio.TaskGroup() as tg:
            tasks = [tg.create_task(table(n, s, allergy, stopped)) for n, s in menu.items()]
        print("served:", [t.result() for t in tasks])
    except* ValueError as eg:
        print("cancelled by:", [str(e) for e in eg.exceptions], "stopped:", sorted(stopped))
    except* TimeoutError as eg:
        print("too slow:", len(eg.exceptions))
    print(f"{time.perf_counter() - start:.1f}s")

MENU = {1: 0.30, 2: 0.35, 3: 0.20, 4: 0.40, 5: 0.45, 6: 0.50}
asyncio.run(service(MENU))
asyncio.run(service(MENU, allergy=3))
asyncio.run(service({**MENU, 6: 0.9}))
# served: ['T1', 'T2', 'T3', 'T4', 'T5', 'T6']
# 0.5s
# cancelled by: ['T3 allergy'] stopped: ['T1', 'T2', 'T4', 'T5', 'T6']
# 0.2s
# too slow: 1
# 0.6s

• On a normal night, all six tables boil together and finish in about 0.5 seconds, set by the longest one, T6.
• When table 3 fails at 0.2 seconds, the other five turn off their burners on the spot.
• When one dish runs past 0.6 seconds, that table's TimeoutError comes up in the bundle, and again the rest stop.

Near closing, the guest by the window finishes her bowl of noodles and closes her small notebook. As she pays, she says a few words to the chef: "I'll be back next Friday."

The chef nods and turns to the sous-chef. "Cancellations and time limits both happen at an await. But the old oven in the corner doesn't know what await is. If Lou stands in front of it and waits, what happens to the floor?"

Practice

1. In the timeline, change the failing table from 1 through 6. Explain why only the tables numbered lower than the failing one survive. Then set the failing table to "None" and raise the time limit from 0.3 to 0.55 seconds, watching more and more tables make it to the end.
2. Rewrite the gather example from 10-1 with a TaskGroup so that when table 2 fails, tables 1 and 3 print cancelled.
3. In table from the closing-time code, delete raise and put await asyncio.sleep(0.3) in its place (ignoring the cancellation and boiling on), then run it with allergy=3. See how the elapsed time changes, and write down what went wrong.