Die Gruppenreservierung platzt: strukturierte Nebenläufigkeit

Halb elf in der Nacht. Sechs Tische einer Gruppe nehmen gleichzeitig Platz. Lou nimmt die Bestellungen auf und gibt sie in die Küche, und die Küche beginnt alle sechs Menüs gleichzeitig. Zehn Minuten später hebt jemand an Tisch 3 die Hand: Die Erdnussallergie wurde zu spät erwähnt. Der Organisator der Gruppe sagt: „Dann brechen wir heute alle zusammen auf.“

Minho hat das nicht mitbekommen und lässt den Eintopf für Tisch 5 weiterköcheln. Sua dreht ihm die Platte ab und sagt: „Nirgendwo stand die Regel, dass alles stoppt, wenn ein Tisch rausfällt.“

Chefin Yoon Seoyeon drückt auf ihren Timer. „Was zusammen anfängt, muss auch zusammen aufhören. Das im Code festzuschreiben, ist die Arbeit für heute Nacht.“

10-1. Die Lücke in gather: Einer scheitert, der Rest köchelt weiter (5 Min.)

Wickeln wir die Gruppenreservierung mit asyncio.gather aus Lektion 9 ab. Tisch 2 scheitert schon nach 0,1 Sekunden.

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)   # das Lokal noch ein wenig offen lassen

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

• gather reicht die erste Exception sofort nach oben weiter. Deshalb erscheint caught: zuerst.
• Die anderen beiden Tische werden aber nicht abgebrochen. Gerichte, auf die niemand mehr wartet, köcheln im Hintergrund weiter und geben schließlich served aus.
• In einem echten Programm hieße das: Essen für Gäste, die längst gegangen sind, oder Anfragen, deren Ergebnis niemand mehr abholt, die aber weiterlaufen.

gather(return_exceptions=True) packt Exceptions nur in die Ergebnisliste, stoppt den Rest aber auch nicht. Wenn du „scheitert einer, hören alle auf“ willst, brauchst du ein anderes Werkzeug.

10-2. TaskGroup: Was zusammen anfängt, hört zusammen auf (12 Min.)

asyncio.TaskGroup (3.11) behandelt Tasks als ein Bündel. Diese Art, Tasks gebündelt zu verwalten, nennt man strukturierte Nebenläufigkeit (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)]
    # hier angekommen, sind alle drei Tasks fertig
    print([t.result() for t in tasks])

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

TaskGroup verspricht drei Dinge.

1. Wenn du den async with-Block verlässt, sind alle darin erzeugten Tasks fertig. Ein extra await ist nicht nötig.
2. Wirft ein Task eine Exception, werden alle übrigen, die noch laufen, abgebrochen.
3. Ist das Abbrechen erledigt, werden die aufgetretenen Exceptions in einer einzigen ExceptionGroup gesammelt nach oben gereicht.

Weil tg.create_task den Task festhält, musst du dir auch keine Sorgen um die „Tasks, die verschwinden, weil die Referenz verloren geht“ aus Lektion 9 machen. Jetzt die Gruppenreservierung, bei der Tisch 3 scheitert.

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                      # einen empfangenen Abbruch wieder werfen

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      (die Reihenfolge von T2 und T4 ändert sich von Lauf zu Lauf)
# T4 cancelled
# group failed: (ValueError('T3 allergy'),)

• T1 war vor dem Fehlschlag fertig und wird ganz normal serviert.
• In dem Moment, in dem T3 nach 0,1 Sekunden scheitert, geht an T2 und T4, die noch köcheln, ein Abbruch. In welcher Reihenfolge die Abbruchmeldungen erscheinen, kann sich von Lauf zu Lauf ändern.
• Auch nach weiteren 0,6 Sekunden Warten erscheint kein served mehr. Anders als bei gather köchelt im Hintergrund heimlich kein Topf weiter.
• In einem abgebrochenen Task fliegt an der Stelle des await ein asyncio.CancelledError. Deshalb kannst du ihn mit except abfangen und aufräumen. Einen empfangenen Abbruch wirfst du aber unbedingt wieder (10-5).

Die Zeitleiste unten lässt auf dem Server einen echten TaskGroup laufen. Je höher die Tischnummer, desto ein bisschen länger dauert das Gericht (T1 0,3 s, T2 0,345 s, … T6 etwa 0,53 s). Der Tisch, den du als scheiternden Tisch wählst, scheitert ungefähr dann, wenn sein Gericht fertig wäre.

• Mit der Starteinstellung (scheiternder Tisch 3) werden T1 und T2 zuerst fertig, und wenn T3 nach etwa 0,4 Sekunden scheitert, brechen die drei Balken T4 bis T6 im selben Moment grau ab. T1 und T2, die schon fertig sind, bleiben stehen.
• Wählst du Tisch 1, werden nach 0,3 Sekunden alle fünf anderen abgeschnitten; wählst du 6, scheitert nur T6, nachdem die anderen fünf fertig sind.
• Steht der scheiternde Tisch auf „Keine“, laufen alle sechs Balken bis zum Ende. Der langsamste, T6, ist nach etwa 0,53 Sekunden fertig.
• Senkst du das Zeitlimit auf etwa 0,4 Sekunden, werden nur die vorderen paar Tische fertig, und was zum Zeitpunkt des Limits noch läuft, wird grau abgeschnitten. Schon fertige Gerichte werden nicht rückgängig gemacht.

10-3. ExceptionGroup und except* (7 Min.)

Mehrere Tasks können auch gleichzeitig scheitern. Deshalb wirft TaskGroup nicht eine einzelne Exception, sondern ein Bündel von Exceptions.

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'))

ExceptionGroup (3.11) ist eine Exception, die eine Liste von Exceptions in .exceptions trägt. Um aus dem Bündel nach Art getrennt abzufangen, nimmst du 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'),)

• Beim normalen except läuft nur ein Block. Bei except* laufen alle passenden Blöcke. Oben laufen beide.
• g ist immer eine ExceptionGroup: ein kleineres Bündel, in dem nur die passenden Exceptions stecken.
• Exceptions, die zu keinem except* passen, werden als Restbündel weiter nach oben geworfen.

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'),)

Für except* gelten zwei Regeln. Bei Verstößen gegen beide stoppt 3.14 schon vor der Ausführung mit einem Syntaxfehler.

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'

• In einem except*-Block sind return, break und continue verboten. Mehrere Blöcke sollen nacheinander laufen, da darf niemand mittendrin abhauen. Speichere das Ergebnis in einer Variablen und gib es außerhalb des try zurück.
• In einem try kannst du except und except* nicht mischen. Brauchst du beide, teil das try in zwei Schichten auf. Die Engine hinter den Versuchsaufbauten dieses Kurses macht es genauso.

10-4. Zeitlimits: asyncio.timeout und wait_for (8 Min.)

Die zweite Regel heute Nacht lautet: „Kein Gericht überschreitet die festgelegte Zeit.“ asyncio.timeout() (3.11) legt ein Zeitlimit über einen ganzen 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

• Ist die Zeit um, wird die Arbeit im Block abgebrochen, und außerhalb des Blocks kommt das eingebaute TimeoutError an. Seit 3.11 ist asyncio.TimeoutError dasselbe wie das eingebaute TimeoutError.
• asyncio.wait_for(Koroutine, timeout=) ist der altbekannte Weg, eine einzelne Koroutine zu begrenzen. Praktisch, wenn du das Ergebnis direkt zurückbekommen willst.
• asyncio.timeout kann mehrere await zu einem Block zusammenfassen. Etwa so: „Von der Bestellung bis zum Servieren insgesamt 20 Minuten.“

Das Limit kannst du statt als „in so vielen Sekunden“ auch als „um so und so viel Uhr“ setzen, und du kannst es sogar ändern, nachdem du den Block betreten hast.

import asyncio, time

async def main():
    loop = asyncio.get_running_loop()
    start = time.perf_counter()
    deadline = loop.time() + 0.3            # der „Zeitpunkt“ 0,3 s ab jetzt
    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:   # zuerst ohne Limit
            cm.reschedule(loop.time() + 0.2)      # später auf 0,2 s festgelegt
            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(Zeitpunkt) nimmt einen absoluten Zeitpunkt, gemessen an loop.time(). Gut, wenn sich mehrere Schritte eine einzige „Deadline“ teilen.
• timeout(None) beginnt ohne Limit. Mit cm.reschedule(Zeitpunkt) legst du das Limit später fest oder verschiebst es.

10-5. Abbrechen: cancel, CancelledError, finally (6 Min.)

TaskGroup und Zeitlimits machen am Ende dasselbe: Sie schicken einem Task einen Abbruch. Selbst schickst du ihn mit task.cancel().

import asyncio

async def braise():
    print("stove on")
    try:
        await asyncio.sleep(10)
    finally:
        print("stove off")          # auch bei Abbruch die Platte ausmachen

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() tötet den Task nicht auf der Stelle. Es sorgt nur dafür, dass beim nächsten await, an dem der Task wartet, ein CancelledError fliegt.
• Aufräumarbeiten wie Platte ausmachen oder Verbindung schließen gehören in try/finally. Das läuft auf jeden Fall, ob Abbruch, Erfolg oder Fehlschlag.
• CancelledError erbt nicht von Exception, sondern von BaseException. So ist es gedacht, damit except Exception: einen Abbruch nicht aus Versehen verschluckt.

Was passiert, wenn du einen Abbruch verschluckst?

import asyncio, time

async def stubborn():
    try:
        await asyncio.sleep(0.5)
    except asyncio.CancelledError:
        pass                          # den Abbruch verschluckt
    await asyncio.sleep(0.5)          # und einfach weitergekocht
    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

Wir haben ein Limit von 0,1 Sekunden gesetzt, und trotzdem dauert es 0,6 Sekunden, und ein TimeoutError gibt es auch nicht. Das Zeitlimit arbeitet mit einem Abbruch, und diesen Abbruch hat die Koroutine aufgefressen. Das „Abbrechen der übrigen“ bei TaskGroup bricht genauso in sich zusammen. Hast du ein CancelledError empfangen, räumst du also nur auf und wirfst es wieder (raise).

10-6. shield schützt vor Abbruch, cancelling zählt ihn (2 Min.)

Manche Arbeit darf nicht mittendrin abreißen, etwa das Speichern einer Quittung. asyncio.shield() verhindert, dass ein Abbruch von außen bis zum inneren Task durchdringt.

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)              # der innere Task lief einfach weiter

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

Wer gewartet hat, hat nach 0,1 Sekunden aufgegeben, aber der innere Task lief bis zum Ende. Nur wenn du die Referenz auf den inneren Task behältst, kannst du später sein Ergebnis abholen.

Task.cancelling() liefert die Zahl der Abbruchanfragen, die bei diesem Task eingegangen sind, und Task.uncancel() senkt sie um eins. Das sind Werkzeuge, mit denen asyncio.timeout und TaskGroup intern unterscheiden: „Kommt dieser Abbruch von mir?“ Selbst brauchst du sie selten.

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

Küchenschluss (10 Min.)

Wir bringen die beiden Regeln von heute Nacht in einer Küche zusammen. Jeder Tisch bekommt ein Limit von 0,6 Sekunden, und alle sechs Tische werden in einem TaskGroup gebündelt.

import asyncio, time

async def table(n, sec, allergy, stopped):
    try:
        async with asyncio.timeout(0.6):     # kein Gericht dauert länger als 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}")             # Platte aus, Hand heben
        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

• In einer normalen Nacht köcheln alle sechs Tische gemeinsam und sind nach etwa 0,5 Sekunden fertig, im Takt des längsten, T6.
• Scheitert Tisch 3 nach 0,2 Sekunden, machen die anderen fünf sofort ihre Platten aus.
• Dauert ein Gericht länger als 0,6 Sekunden, kommt das TimeoutError dieses Tisches im Bündel nach oben, und auch dann stoppt der Rest.

Gegen Küchenschluss leert die Frau am Fenster ihre Nudelschüssel und klappt ihr kleines Notizbuch zu. Beim Bezahlen sagt sie kurz zur Chefin: „Nächsten Freitag komme ich wieder.“

Die Chefin nickt und dreht sich zu dir um. „Abbruch und Zeitlimit passieren am Ende immer an einem await. Aber der alte Ofen in der Ecke kennt kein await. Was wird aus dem Gastraum, wenn Lou davor steht und wartet?“

Übungen

1. Stell in der Zeitleiste den Tisch, der scheitert, nacheinander auf 1 bis 6. Erklär, warum nur die Tische mit kleinerer Nummer als der gescheiterte überleben. Setz den scheiternden Tisch danach auf „Keine“, erhöh das Zeitlimit von 0,3 auf 0,55 Sekunden und beobachte, wie immer mehr Tische bis zum Ende kommen.
2. Bau das gather-Beispiel aus 10-1 auf TaskGroup um, sodass Tisch 1 und Tisch 3 cancelled ausgeben, wenn Tisch 2 scheitert.
3. Lösch in table aus dem Küchenschluss-Code das raise, setz an seine Stelle await asyncio.sleep(0.3) (Abbruch ignorieren und weiterköcheln) und lass es mit allergy=3 laufen. Schau, wie sich die Laufzeit ändert, und schreib auf, was schiefgegangen ist.