Eine Servicechefin für alle Tische: erste Schritte mit asyncio

Ab heute Nacht ist der Gastraum anders organisiert. Es gibt eine einzige Servicekraft: Lou. Und zehn Tische. Die Chefin hat niemanden zusätzlich eingestellt.

Lou dreht ohne Pause ihre Runden durch den Gastraum. Sie nimmt eine Bestellung auf, gibt sie in die Küche und geht, statt auf das Essen zu warten, gleich zum nächsten Tisch. Klingelt die Glocke in der Küche, holt sie den Teller und bringt ihn raus. Eine Runde, noch eine Runde. Weil sie auf Glockenzeichen (Events) reagiert und dabei immer denselben Weg abläuft, haben ihr die Köche schon am ersten Abend einen Spitznamen gegeben: die Event-Loop.

Lou bindet sich die Schürze zu. „Am meisten hasse ich es, vor einem Tisch zu stehen und einfach nur zu warten. Während ich warte, warten die anderen Tische auf mich.“

Die Chefin nickt. „Bis gestern haben wir mehr Leute eingestellt. Heute geht es darum, wie eine einzige Person ihre Wartezeit nutzt. In Python heißt das asyncio.“

9-1. async def erzeugt eine Koroutine (7 Min.)

Setzt du async vor eine Funktion, wird sie zur Koroutinenfunktion. Fangen wir mit dem Seltsamen an: Wenn du sie aufrufst, läuft sie nicht.

import asyncio

async def take_order(table):
    print("order", table)
    return f"T{table}"

c = take_order(1)
print(type(c).__name__)
print(c)
# coroutine
# <coroutine object take_order at 0x...>
# <sys>:0: RuntimeWarning: coroutine 'take_order' was never awaited

• take_order(1) hat nicht "order 1" ausgegeben. Vom Funktionsrumpf ist keine einzige Zeile gelaufen.
• Zurück kam ein Koroutinenobjekt. Das ist wie ein Zettel, auf dem steht: „Bestellung von Tisch 1 aufnehmen“.
• Wirfst du den Zettel weg, ohne ihn je auszuführen, gibt Python am Ende die Warnung was never awaited aus. Siehst du diese Warnung, hast du fast immer ein await vergessen.

Es gibt zwei Wege, den Zettel auszuführen. Auf oberster Ebene gibst du ihn an asyncio.run(), innerhalb einer anderen Koroutine machst du await.

print(asyncio.run(take_order(1)))
# order 1
# T1

c = take_order(2)
asyncio.run(c)
asyncio.run(c)
# order 2
# RuntimeError: cannot reuse already awaited coroutine

Ein Koroutinenobjekt lässt sich nur einmal verwenden. Für dieselbe Bestellung noch einmal rufst du take_order(2) erneut auf und bekommst einen neuen Zettel.

9-2. await und asyncio.run (7 Min.)

await heißt: „Ich warte hier, bis diese Sache fertig ist. Die Loop darf in der Zwischenzeit etwas anderes machen.“ await darf nur innerhalb von async def stehen. In einem normalen def gibt es SyntaxError: 'await' outside async function.

import asyncio
import time

async def table(n, wait):
    print(f"T{n} order")
    await asyncio.sleep(wait)   # während sie auf die Küche wartet, ist Lou frei
    print(f"T{n} served")
    return f"T{n}"

async def one_by_one():
    a = await table(1, 0.3)
    b = await table(2, 0.3)
    return [a, b]

start = time.perf_counter()
print(asyncio.run(one_by_one()), f"{time.perf_counter() - start:.2f}s")
# T1 order
# T1 served
# T2 order
# T2 served
# ['T1', 'T2'] 0.62s

• asyncio.sleep(0.3) ist die asyncio-Version von time.sleep(0.3). Während des Wartens gibt sie den Platz an die Loop ab.
• Trotzdem dauert es 0,3 + 0,3 = etwa 0,6 Sekunden. Die nächste Zeile kommt erst dran, wenn await table(1, ...) fertig ist. Als würde Lou vor Tisch 1 stehen bleiben, bis das Essen kommt.
• await heißt nicht „gleichzeitig“, sondern „warten“. Um gleichzeitig zu warten, brauchst du die Werkzeuge aus dem nächsten Abschnitt.

asyncio.run(main()) ist der Eingang zum Programm.

1. Es erzeugt eine neue Event-Loop (Lou fängt ihre Schicht an).
2. Es lässt die Koroutine main() laufen, bis sie fertig ist.
3. Es bricht übrig gebliebene Tasks ab und schließt die Loop (Lou hat Feierabend).

Normalerweise rufst du es im Programm genau einmal auf, auf oberster Ebene. Rufst du es an einer Stelle auf, an der schon eine Loop läuft, gibt es RuntimeError: asyncio.run() cannot be called from a running event loop.

9-3. Was die Event-Loop macht (6 Min.)

Die Event-Loop ist eine Schleife, die in einem einzigen Thread läuft. Sie tut dasselbe wie Lou auf ihrer Runde durch den Gastraum.

1. Eine Koroutine aussuchen, mit der es gerade weitergehen kann.
2. Diese Koroutine bis zum nächsten await ausführen.
3. Notieren, worauf die Koroutine wartet (Timer, Netzwerkantwort), und zur nächsten Koroutine gehen.
4. Gibt es keine Koroutine, mit der es weitergehen kann, auf die nächste Glocke warten und die zugehörige Koroutine wieder aufwecken.

Übernimmt Lou drei Tische gleichzeitig (mit gather aus dem nächsten Abschnitt), sieht ihr Weg so aus. Tisch 1 braucht 0,3 Sekunden, Tisch 2 0,2 Sekunden, Tisch 3 0,1 Sekunden.

Zwei Dinge solltest du dir merken.

• Es ist nur eine Person. Es gibt nur einen Thread, und zu jedem Zeitpunkt läuft genau eine Koroutine. Schneller wird es, weil sich die Wartezeiten überlappen, nicht weil mehr Hände da sind.
• Gewechselt wird nur bei await. Threads unterbricht das Betriebssystem jederzeit (Lektion 3), Koroutinen räumen ihren Platz nur, wenn sie selbst await machen. Code zwischen zwei await wird deshalb nicht mittendrin unterbrochen.

9-4. gather: gleichzeitig warten (8 Min.)

asyncio.gather(koroutine, koroutine, ...) setzt mehrere Koroutinen auf einmal auf die Loop und wartet, bis alle fertig sind.

async def together():
    return await asyncio.gather(table(1, 0.3), table(2, 0.2), table(3, 0.1))

start = time.perf_counter()
print(asyncio.run(together()), f"{time.perf_counter() - start:.2f}s")
# T1 order
# T2 order
# T3 order
# T3 served
# T2 served
# T1 served
# ['T1', 'T2', 'T3'] 0.31s

• Die Dauer ist nicht die Summe (0,6 Sekunden), sondern die längste Einzelzeit (0,3 Sekunden).
• Fertig wurden sie in der Reihenfolge T3 → T2 → T1, aber die Ergebnisliste hat die Reihenfolge, in der du sie übergeben hast. Genau wie executor.map aus Lektion 7.
• Hast du mehrere Koroutinen in einer Liste, packst du sie mit einem Sternchen aus: gather(*coros).

Schalte im Versuchsaufbau unten zwischen „await Tisch für Tisch“ und „Alles auf einmal mit gather“ hin und her. Eine Reihe ist ein Tisch, und alle Balken haben die Farbe für „Warten“. Lou ist fast nie selbst mit den Händen beschäftigt, die meiste Zeit wartet sie auf die Küche.

• Tisch für Tisch: Die Zeit wächst mit der Zahl der Tische. Bei 10 Tischen etwa 2,6 Sekunden.
• gather: Ob 1 Tisch oder 12, etwa 0,25 Sekunden. Den Effekt, für den wir in Lektion 1 acht Köche eingestellt haben, bekommen wir hier mit einer einzigen Servicekraft.
• Auch hier läuft auf dem Server eine echte asyncio-Loop, und gezeichnet wird, wann jeder Tisch begonnen und geendet hat.

Wenn an einem Tisch etwas schiefgeht: return_exceptions

Wirft eine der an gather übergebenen Koroutinen eine Ausnahme, reicht gather diese Ausnahme direkt weiter. Willst du trotzdem alle Ergebnisse bekommen, gib return_exceptions=True an.

async def table(n):
    await asyncio.sleep(0.1 * n)
    if n == 2:
        raise ValueError(f"T{n} wrong dish")
    return f"T{n}"

async def main():
    results = await asyncio.gather(table(1), table(2), table(3), return_exceptions=True)
    print(results)
    try:
        await asyncio.gather(table(1), table(2), table(3))
    except ValueError as e:
        print("ValueError:", e)

asyncio.run(main())
# ['T1', ValueError('T2 wrong dish'), 'T3']
# ValueError: T2 wrong dish

• Mit return_exceptions=True landet auch die Ausnahme als Eintrag in der Ergebnisliste. Prüfe die Ergebnisse einzeln mit isinstance(r, Exception).
• Beim Standardwert (False) werden die übrigen Tische nicht abgebrochen, auch wenn die Ausnahme hochkommt; sie laufen weiter. „Wenn einer scheitert, stopp den Rest“ erledigt die TaskGroup aus Lektion 10.

9-5. create_task: schon mal abgeben (6 Min.)

asyncio.create_task(koroutine) verpackt eine Koroutine in einen Task und plant sie sofort auf der Loop ein. Der Task startet, sobald Lou die Hände frei hat (wenn die aktuelle Koroutine await macht). Es wird nicht auf das Ende gewartet, sondern gleich zur nächsten Zeile gegangen; du kannst also Bestellungen schon mal aufgeben und etwas anderes erledigen.

async def table(n, wait):
    print(f"T{n} order")
    await asyncio.sleep(wait)
    print(f"T{n} served")
    return f"T{n}"

async def main():
    start = time.perf_counter()
    t1 = asyncio.create_task(table(1, 0.3))
    t2 = asyncio.create_task(table(2, 0.2))
    print("Lou: 2 tasks created")
    await asyncio.sleep(0.1)          # Lou erledigt kurz etwas anderes
    print("Lou: back", t1.done(), t2.done())
    print(await t1, await t2, f"{time.perf_counter() - start:.2f}s")

asyncio.run(main())
# Lou: 2 tasks created
# T1 order
# T2 order
# Lou: back False False
# T2 served
# T1 served
# T1 T2 0.31s

• "Lou: 2 tasks created" erscheint vor T1 order. Tasks werden gleich beim Erzeugen eingeplant, laufen aber erst los, wenn main mit await seinen Platz räumt.
• await t1 wartet darauf, dass ein bereits laufender Task fertig wird. Beide Tasks haben gemeinsam gewartet, also sind es nicht 0,5, sondern etwa 0,3 Sekunden.
• gather verpackt die übergebenen Koroutinen intern selbst in Tasks. Brauchst du nur die gesammelten Ergebnisse, nimm gather; willst du Tasks einzeln in der Hand behalten, ihren Zustand ansehen oder später auf sie warten, nimm create_task.

Tasks brauchen eine Referenz

Die Event-Loop hält Tasks nur über eine schwache Referenz. Ein Task, den du in keiner Variablen und keiner Collection ablegst, kann von der Garbage Collection eingesammelt werden, bevor er fertig ist. Auch wenn du Arbeit abgibst, deren Ergebnis du nicht brauchst, etwa Wasser nachschenken, musst du eine Referenz festhalten.

background = set()

async def refill_water(table):
    await asyncio.sleep(0.1)
    print(f"T{table} water")

async def main():
    for n in (1, 2, 3):
        task = asyncio.create_task(refill_water(n))
        background.add(task)                         # Referenz festhalten
        task.add_done_callback(background.discard)   # nach dem Ende entfernt er sich selbst
    print("pending", len(background))
    await asyncio.sleep(0.2)
    print("pending", len(background))

asyncio.run(main())
# pending 3
# T1 water
# T2 water
# T3 water
# pending 0

Eine Zeile wie asyncio.create_task(refill_water(n)), die das Ergebnis wegwirft, funktioniert meistens, aber genau dieses „meistens“ ist das Problem. Tasks, die gelegentlich verschwinden, sind die am schwersten zu findenden Bugs.

9-6. asyncio.sleep gegen time.sleep (3 Min.)

Beides heißt „kurz warten“, aber für Lou ist es etwas völlig anderes.

async def polite(n):
    await asyncio.sleep(0.3)   # beim Warten den Platz an die Loop abgeben

async def stubborn(n):
    time.sleep(0.3)            # hält die ganze Loop an

async def main(fn):
    start = time.perf_counter()
    await asyncio.gather(*(fn(n) for n in range(3)))
    print(fn.__name__, f"{time.perf_counter() - start:.2f}s")

asyncio.run(main(polite))
asyncio.run(main(stubborn))
# polite 0.32s
# stubborn 0.90s

• asyncio.sleep sagt der Loop im Moment des await: „Weck mich in 0,3 Sekunden“, und räumt den Platz.
• time.sleep hält den Thread selbst an. Die Loop läuft in nur einem Thread, also ist das, als würde Lou vor einem Tisch stehen und einfach nur warten. Um die anderen Tische kümmert sich in der Zeit niemand.
• Deshalb kamen die drei Tische nacheinander mit je 0,3 Sekunden dran, insgesamt etwa 0,9 Sekunden. Da hilft auch gather nicht.

Nicht nur time.sleep: Auch lange Berechnungen und alte Bibliotheken, die beim Warten den Platz nicht abgeben, halten die Loop genauso an. Dieses Problem und seine Lösung sind Thema von Lektion 11.

9-7. Koroutinen, Tasks, Futures und ihre Kosten (3 Min.)

In asyncio tauchen drei ähnliche Namen auf.

print(issubclass(asyncio.Task, asyncio.Future))   # True

async def main():
    coro = take_order(1)
    task = asyncio.create_task(coro)
    print(type(coro).__name__, type(task).__name__)   # coroutine Task

Die Idee ist dieselbe wie bei concurrent.futures.Future aus Lektion 7. Nur kann man auf ein asyncio-Future await machen, und man verwendet es nur innerhalb derselben Loop. Futures selbst anzulegen ist selten nötig. Meistens reichen Koroutinen und Tasks.

Tasks sind leicht

Threads erzeugt das Betriebssystem, und jeder bringt Stack-Speicher und Verwaltungsaufwand im Betriebssystem mit. Ein Task ist bloß ein einzelnes Python-Objekt.

async def table(n):
    await asyncio.sleep(0.5)
    return n

async def main():
    return await asyncio.gather(*(table(n) for n in range(10_000)))

# 10,000 tasks: etwa 0.57 s   (10.000 Tische, die 0,5 s warten)
# 1,000 threads: etwa 0.62 s  (1.000 Threads, die 0,5 s warten)
# auf diesem PC gemessen

10.000 Tasks zu starten kostet weniger als 1.000 Threads. Darum passt asyncio so gut, wenn es Tausende oder Zehntausende Wartearbeiten gibt (Chatserver, Anfragen an viele Adressen).

Küchenschluss (10 Min.)

Das Ziel für heute Nacht war, das Essen für alle zehn Tische in unter 0,5 Sekunden rauszubringen, ohne mehr Personal einzustellen. Dafür reichen eine Lou, eine Event-Loop und ein gather.

import asyncio
import time

async def table(n):
    await asyncio.sleep(0.3)   # warten, bis das Essen aus der Küche kommt
    return f"T{n}"

async def hall():
    return await asyncio.gather(*(table(n) for n in range(1, 11)))

start = time.perf_counter()
served = asyncio.run(hall())
print(served)
print(f"{time.perf_counter() - start:.2f}s")
# ['T1', 'T2', 'T3', 'T4', 'T5', 'T6', 'T7', 'T8', 'T9', 'T10']
# 0.30s

Wir haben keinen einzigen Thread angelegt. Zehn Tische zu je 0,3 Sekunden waren in etwa 0,3 Sekunden fertig.

Lou stellt das leere Tablett ab. „Zehn Tische, in einer Runde erledigt.“ Die Chefin schaut auf ihren Timer. „Gut. Aber morgen kommt die Gruppenreservierung. Wenn von sechs Tischen einer absagt, wie stoppst du dann das Essen für die anderen fünf?“

Übungen

1. Führe nacheinander eine Zeile aus, die take_order aus 9-1 ohne await aufruft, und eine, die es mit asyncio.run() aufruft. Vergleiche Ausgabe und Warnung.
2. Wie ändert sich die Reihenfolge der Ausgabe, wenn du in main aus 9-5 das await asyncio.sleep(0.1) löschst? Schreib deine Vermutung zuerst auf und prüfe sie dann.
3. Erhöhe im Code vom Küchenschluss die Zahl der Tische auf 100 und auf 1.000 und miss die Zeit. Vergleiche auch, wie es aussieht, wenn du dasselbe mit 1.000 Threads machst.