Der Gastraum steht still: blockierende Aufrufe und Kombinieren

Elf Uhr nachts. Tisch 3 bestellt ein Gratin. Gratin lässt sich nur in dem alten Ofen backen, den die Chefin gebraucht gekauft hat. Ist die Tür dieses Ofens einmal zu, bleibt der Griff verriegelt, bis alles fertig gebacken ist, und jemand muss die ganze Zeit danebenstehen und den Griff festhalten.

Lou stellt sich vor den Ofen. In diesem Moment steht der Gastraum still. Tisch 1, der Wasser möchte, und Tisch 5, der zahlen will: Alle warten nur darauf, dass Lou zurückkommt.

Herr Tak schnalzt mit der Zunge. „Vor dem Ofen stehen ist keine Arbeit. Das ist bloß Rumstehen.“ Die Chefin nickt. „Lou darf nicht warten. Das Warten überlässt man jemand anderem.“

11-1. Steht Lou still, steht der Gastraum still (7 Min.)

Die Event-Loop von asyncio läuft in einem einzigen Thread. Erst wenn eine Koroutine bei await den Platz freigibt, kann die Loop zum nächsten Task wechseln. Ruft man aber eine Funktion auf, die await nicht kennt, etwa time.sleep oder eine alte synchrone Bibliothek, kann sich die Loop nicht rühren, bis diese Funktion fertig ist. Solche Aufrufe nennt man blockierende Aufrufe.

import asyncio, time

start = time.perf_counter()

async def lu_rounds():                 # Lou dreht alle 0,1 s eine Runde durch den Gastraum
    for _ in range(5):
        print(f"round {time.perf_counter() - start:.1f}s")
        await asyncio.sleep(0.1)

def old_oven():                        # eine alte Funktion, die await nicht kennt
    time.sleep(0.35)
    return "bread"

async def bake():
    await asyncio.sleep(0.15)
    print("oven", old_oven())          # hier steht die ganze Loop 0,35 s still

async def main():
    await asyncio.gather(lu_rounds(), bake())

asyncio.run(main())
# round 0.0s
# round 0.1s
# oven bread
# round 0.5s
# round 0.6s
# round 0.7s

• Die Runden bei 0,2, 0,3 und 0,4 Sekunden sind komplett verschwunden. Während Lou von 0,15 bis 0,5 Sekunden vor dem Ofen stand, konnte kein einziger Task laufen.
• Ein blockierender Aufruf bremst nicht nur seine eigene Koroutine. Er hält alle Tasks auf der Loop gleichzeitig an. Das ist der größte Unterschied zu mehreren Threads.
• Die üblichen Verdächtigen: time.sleep, synchrone Netzwerkbibliotheken wie requests.get, das Lesen großer Dateien, schwere Berechnungen.

In der Zeitleiste unten benutzt ein langsamer Tisch den alten Ofen (time.sleep, 0,6 Sekunden). Die anderen Tische warten in Häppchen von je 0,1 Sekunden mit asyncio.sleep.

• Mit ausgeschaltetem Schalter dauert es etwa 0,9 Sekunden. T1 und T2 bräuchten nur 0,3 Sekunden, ziehen sich aber bis etwa 0,76 Sekunden, und T4 bis T6 fangen erst bei 0,6 Sekunden an, wenn der Ofen fertig ist. Lou war bei diesen Tischen noch nicht einmal vorbeigekommen.
• Stellst du den langsamen Tisch auf 1, fangen die anderen fünf alle erst nach 0,6 Sekunden an.
• Mit eingeschaltetem Schalter wandert der Ofen in einen Thread; die anderen Tische sind nach etwa 0,3 Sekunden fertig, und auch die Gesamtzeit schrumpft auf die Ofenzeit von etwa 0,6 Sekunden.

11-2. Den Täter mit dem Debug-Modus finden (6 Min.)

In einem großen Programm ist schwer zu erkennen, welche Koroutine die Loop festhält. Der Debug-Modus von asyncio meldet Callbacks und Tasks, die am Stück lange gelaufen sind, im Log.

import asyncio, logging, time

logging.basicConfig(level=logging.WARNING, format="%(name)s %(levelname)s: %(message)s")

def old_oven():
    time.sleep(0.3)

async def bake():
    old_oven()

async def main():
    await asyncio.create_task(bake(), name="T3-oven")

asyncio.run(main(), debug=True)
# z. B.: asyncio WARNING: Executing <Task finished name='T3-oven' coro=<bake() done, defined at ...oven.py:8>
#     result=None created at ...\asyncio\tasks.py:395> took 0.301 seconds

• Eingeschaltet wird er mit asyncio.run(main(), debug=True). Auch die Umgebungsvariable PYTHONASYNCIODEBUG=1 oder python -X dev schalten ihn ein.
• Eine Warnzeile enthält den Namen des Tasks (T3-oven), Datei und Zeile, in der die Koroutine definiert ist, und wie lange sie die Loop festgehalten hat (took 0.301 seconds). Gibst du deinen Tasks ein name=, findest du den Täter leichter.
• Die Meldung kommt als WARNING über den Logger asyncio. Die Pfade sind auf jedem PC anders.

Die Schwelle ist loop.slow_callback_duration, Standard 0,1 Sekunden. Willst du empfindlicher messen, senkst du sie.

import asyncio, time

async def main():
    loop = asyncio.get_running_loop()
    print(loop.get_debug(), loop.slow_callback_duration)
    loop.slow_callback_duration = 0.05     # schon ab 0,05 s melden
    time.sleep(0.07)

asyncio.run(main(), debug=True)
# True 0.1
# z. B.: Executing <Task finished name='Task-1' coro=<main() done, defined at ...> result=None created at ...> took 0.070 seconds

Darüber hinaus zeigt der Debug-Modus, wo Koroutinen erzeugt wurden, die ohne await liegen geblieben sind, und meldet einen Fehler, wenn ein anderer Thread die Loop falsch anfasst. In der Entwicklung einschalten, im Betrieb ausschalten (er macht alles langsamer).

11-3. Das Warten übernimmt ein Thread: to_thread und run_in_executor (8 Min.)

Wenn sich der alte Ofen nicht reparieren lässt, stellt man eben eine eigene Person davor. asyncio.to_thread() (3.9) lässt eine blockierende Funktion in einem anderen Thread laufen und gibt der Loop ein Objekt zurück, auf das sie warten kann.

import asyncio, time

start = time.perf_counter()

async def lu_rounds():
    for _ in range(5):
        print(f"round {time.perf_counter() - start:.1f}s")
        await asyncio.sleep(0.1)

def old_oven(item, *, minutes):
    time.sleep(0.35)
    return f"{item} {minutes}m"

async def bake():
    await asyncio.sleep(0.15)
    bread = await asyncio.to_thread(old_oven, "bread", minutes=40)
    print("oven", bread)

async def main():
    await asyncio.gather(lu_rounds(), bake())

asyncio.run(main())
# round 0.0s
# round 0.1s
# round 0.2s
# round 0.3s
# round 0.4s
# oven bread 40m

• Lou dreht lückenlos alle 0,1 Sekunden ihre Runde. Den Ofen bewacht ein Thread, Lou holt nur das Ergebnis mit await ab.
• to_thread(Funktion, *args, **kwargs) reicht die Argumente unverändert weiter. Es läuft im Thread-Pool aus Lektion 7 (dem Standard-Executor der Loop).
• Es ist ein Thread, also gelten die GIL-Regeln aus Lektion 8 weiter. Bei blockierendem Code, der wartet, wirkt es gut, schwere Berechnungen werden dadurch aber nicht schneller.

Schwere Berechnungen schickst du in die Filialküche, also in einen Prozess-Pool. Mit asyncio.get_running_loop() holst du dir die gerade laufende Loop und benutzt loop.run_in_executor(executor, Funktion, *args). Weil Prozesse im Spiel sind, speicherst du den Code als Datei und führst ihn aus.

import asyncio
from concurrent.futures import ProcessPoolExecutor

def knead(n):                          # Arbeit mit vollen Händen (CPU)
    return sum(i * i for i in range(n))

async def main():
    loop = asyncio.get_running_loop()
    with ProcessPoolExecutor(max_workers=2) as pool:
        doughs = await asyncio.gather(
            loop.run_in_executor(pool, knead, 2_000_000),
            loop.run_in_executor(pool, knead, 3_000_000),
        )
    print(doughs)

if __name__ == "__main__":
    asyncio.run(main())
# [2666664666667000000, 8999995500000500000]

• Übergibst du run_in_executor als erstes Argument None, wird der Standard-Thread-Pool verwendet. Genau das macht to_thread bequem für dich.
• run_in_executor nimmt keine Schlüsselwortargumente. Wenn du welche brauchst, verpackst du sie mit functools.partial(knead, n=...).
• Es ist ein Prozess-Pool, also gelten auch die Regeln aus Lektion 8: Du brauchst if __name__ == "__main__":, und die übergebenen Funktionen und Werte müssen sich picklen lassen.

11-4. Die Synchronisationswerkzeuge von asyncio (8 Min.)

Die Werkzeuge aus Lektion 4 und 6 gibt es noch einmal eigens für asyncio. Sie heißen gleich, aber statt beim Warten den Thread anzuhalten, geben sie mit await den Platz frei.

Die Semaphore braucht man am häufigsten, um „gleichzeitige Verbindungen zu begrenzen“. Wir müssen fünfmal beim Lieferanten anrufen, aber der nimmt nur zwei Anrufe gleichzeitig an.

import asyncio, time

start = time.perf_counter()
supplier = asyncio.Semaphore(2)        # höchstens zwei Anrufe gleichzeitig beim Lieferanten

async def call(n):
    async with supplier:
        print(f"call #{n} at {time.perf_counter() - start:.1f}s")
        await asyncio.sleep(0.2)

async def main():
    await asyncio.gather(*(call(n) for n in range(1, 6)))

asyncio.run(main())
# call #1 at 0.0s
# call #2 at 0.0s
# call #3 at 0.2s
# call #4 at 0.2s
# call #5 at 0.4s

asyncio.Queue sieht aus wie die Bonschiene aus Lektion 6. Auch maxsize, task_done() und join() gibt es genauso.

import asyncio

async def hall(q):
    for n in range(1, 6):
        await q.put(f"#{n}")           # ist die Schiene voll, wird hier gewartet
    print("hall done")

async def cook(name, q, done):
    while True:
        ticket = await q.get()
        await asyncio.sleep(0.1)
        done.append((name, ticket))
        q.task_done()

async def main():
    q = asyncio.Queue(maxsize=2)
    done = []
    cooks = [asyncio.create_task(cook(c, q, done)) for c in ("A", "B")]
    await hall(q)
    await q.join()                     # bis für alle Bons task_done aufgerufen wurde
    for c in cooks:
        c.cancel()
    print(len(done), sorted(t for _, t in done))

asyncio.run(main())
# hall done
# 5 ['#1', '#2', '#3', '#4', '#5']

Weil die Loop in einem einzigen Thread läuft, könnte man meinen, es gäbe keine Race Conditions. Aber steht zwischen Lesen und Schreiben ein await, passiert genau der Tafel-Unfall aus Lektion 3.

import asyncio

async def main():
    board = {"count": 0}

    async def cook():
        for _ in range(100):
            v = board["count"]
            await asyncio.sleep(0)     # hier wird zu einem anderen Task gewechselt
            board["count"] = v + 1

    await asyncio.gather(*(cook() for _ in range(4)))
    print(board["count"])

asyncio.run(main())
# 100

Auf der Tafel sollte 400 stehen, es steht aber 100. Alle vier Tasks lesen denselben Wert, geben den Platz frei und schreiben denselben Wert zurück. Fasst du diesen Abschnitt mit asyncio.Lock zusammen und lässt alle mit asyncio.Event auf das Signal „Offen!“ warten, sieht das so aus:

import asyncio

async def main():
    lock = asyncio.Lock()
    opened = asyncio.Event()
    board = {"count": 0}

    async def cook():
        await opened.wait()            # warten bis „Offen!“
        for _ in range(100):
            async with lock:           # steht zwischen Lesen und Schreiben ein await, braucht es einen Lock
                v = board["count"]
                await asyncio.sleep(0)
                board["count"] = v + 1

    tasks = [asyncio.create_task(cook()) for _ in range(4)]
    await asyncio.sleep(0.1)
    opened.set()
    await asyncio.gather(*tasks)
    print(board["count"])

asyncio.run(main())
# 400

Die Werkzeuge der beiden Welten darfst du nicht mischen. Wartet eine Koroutine auf einen threading.Lock, steht die ganze Loop still. Damit verschwindet auch die Chance, dass der Task, der den Lock hält, zurückkommt und ihn freigibt.

import asyncio, threading, time

lock = threading.Lock()                # was, wenn man einen Thread-Lock in einer Koroutine benutzt?
start = time.perf_counter()

async def cook(n):
    if not lock.acquire(timeout=0.2):  # während des Wartens steht die ganze Loop still
        print(f"#{n} gave up at {time.perf_counter() - start:.1f}s")
        return
    try:
        await asyncio.sleep(0.1)       # mit gehaltenem Lock den Platz freigeben
        print(f"#{n} done")
    finally:
        lock.release()

async def main():
    await asyncio.gather(cook(1), cook(2))

asyncio.run(main())
# #2 gave up at 0.2s
# #1 done

Ohne timeout würde #2 ewig warten, und #1 würde nie wieder aufwachen. Umgekehrt sind die Werkzeuge von asyncio nicht threadsicher. Aus einem anderen Thread darfst du nichts in eine asyncio.Queue stecken. Wie ein Thread der Loop Arbeit übergibt, sehen wir in 11-6.

11-5. async with, async for, asynchrone Generatoren (7 Min.)

Auch with und for gibt es in einer Variante, die warten kann.

• async with: Ressourcen, die beim Betreten und Verlassen ein await brauchen (__aenter__, __aexit__). Locks, Semaphoren, Netzwerkverbindungen.
• async for: Schleifen, bei denen man auf den nächsten Wert warten muss (__aiter__, __anext__). Bons, die einzeln eintreffen, Streaming-Antworten.
• Asynchrone Generatoren: Schreibst du yield in ein async def, entsteht ein Iterator, den du mit async for durchläufst.

import asyncio
from contextlib import asynccontextmanager

async def tickets(n):                  # asynchroner Generator: async def + yield
    for i in range(1, n + 1):
        await asyncio.sleep(0.05)      # warten, bis der nächste Bon kommt
        yield f"#{i}"

@asynccontextmanager
async def stove(name):                 # eine Herdplatte für async with
    await asyncio.sleep(0.01)
    print(f"{name} on")
    try:
        yield name
    finally:
        print(f"{name} off")

async def main():
    async with stove("S1"):
        async for t in tickets(3):
            print("cook", t)
    print([t async for t in tickets(2)])

asyncio.run(main())
# S1 on
# cook #1
# cook #2
# cook #3
# S1 off
# ['#1', '#2']

contextlib.asynccontextmanager baut dir mit einem einzigen yield eine Ressource für async with. Auch in List Comprehensions kannst du async for verwenden (allerdings nur innerhalb einer Koroutine).

Auch as_completed aus Lektion 7 gibt es für asyncio. Ab 3.13 kannst du es mit async for durchlaufen und bekommst dann genau die Tasks, die du hineingegeben hast, in der Reihenfolge, in der sie fertig werden.

import asyncio

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

async def main():
    tasks = [asyncio.create_task(dish(n, s)) for n, s in [(1, 0.3), (2, 0.1), (3, 0.2)]]
    async for t in asyncio.as_completed(tasks):      # ab 3.13
        print(t in tasks, await t)

asyncio.run(main())
# True #2
# True #3
# True #1

Du kannst es auch wie früher als for f in asyncio.as_completed(tasks): await f schreiben, aber dann ist das f, das du bekommst, nicht der ursprüngliche Task, sondern ein neues Objekt. Willst du wissen, welche Bestellung fertig ist, ist async for bequemer.

11-6. Arbeit von außerhalb der Loop einwerfen, laufende Tasks ansehen (4 Min.)

Angenommen, die Person, die Reservierungsanrufe annimmt, arbeitet in einem anderen Thread. Will dieser Thread etwas in Lous Plan einschieben, nimmt er asyncio.run_coroutine_threadsafe(Koroutine, loop). Zurück kommt ein concurrent.futures.Future aus Lektion 7, auf das der Thread mit result(timeout=) warten kann.

import asyncio

async def seat(table):
    await asyncio.sleep(0.05)
    return f"{table} seated"

def phone_line(loop):                  # ein Thread außerhalb der Loop (Reservierungstelefon)
    fut = asyncio.run_coroutine_threadsafe(seat("T7"), loop)
    print(fut.result(timeout=1))       # concurrent.futures.Future

async def main():
    loop = asyncio.get_running_loop()
    await asyncio.to_thread(phone_line, loop)

asyncio.run(main())
# T7 seated

Eager Tasks (3.12): Normalerweise setzt create_task einen Task nur auf den Plan und führt ihn erst aus, wenn er an der Reihe ist. Mit asyncio.eager_task_factory läuft er sofort beim Erzeugen bis zum ersten await. Eine Koroutine, die ohne await gleich fertig wird, kommt gar nicht erst auf den Plan, sondern ist auf der Stelle erledigt.

import asyncio

async def greet(n):
    print(f"greet T{n}")               # bis vor das erste await
    await asyncio.sleep(0)
    return n

async def main(eager):
    if eager:
        asyncio.get_running_loop().set_task_factory(asyncio.eager_task_factory)
    t = asyncio.create_task(greet(1))
    print("created, done =", t.done())
    await t

asyncio.run(main(False))
asyncio.run(main(True))
# created, done = False
# greet T1
# greet T1
# created, done = False

Laufende Tasks ansehen (3.14): Kennst du die Prozessnummer (PID) eines laufenden asyncio-Programms, kannst du in einem anderen Terminal mit python -m asyncio ps <PID> die Liste der Tasks sehen und mit python -m asyncio pstree <PID> als Baum, wer auf wen wartet. Wartet zum Beispiel der Task hall über einen TaskGroup auf T1 und T2, sieht das so aus (einige Spalten und Pfade weggelassen):

task name   coroutine stack                                   awaiter name
Task-1      main
hall        TaskGroup._aexit -> TaskGroup.__aexit__ -> hall   Task-1
T1          sleep -> table                                    hall
T2          sleep -> table                                    hall

Das ist das Werkzeug für den Moment „Der Gastraum steht still, und ich weiß nicht, wo Lou gerade steht“. Gibst du deinen Tasks ein name=, liest sich die Liste viel leichter.

Küchenschluss (10 Min.)

Nur das Gratin von Tisch 3 kommt in den alten Ofen, und vor den Ofen stellen wir einen Thread. Mit einem einzigen Schalter (fix) vergleichen wir.

import asyncio, time

def old_oven(sec):                     # eine alte Bibliothek, an der wir nichts ändern können
    time.sleep(sec)
    return "gratin"

async def table(n, start, fix):
    if n == 3:
        dish = await asyncio.to_thread(old_oven, 0.6) if fix else old_oven(0.6)
    else:
        await asyncio.sleep(0.3)
        dish = "noodle"
    return f"T{n} {dish} {time.perf_counter() - start:.1f}s"

async def service(fix):
    start = time.perf_counter()
    async with asyncio.TaskGroup() as tg:
        tasks = [tg.create_task(table(n, start, fix)) for n in range(1, 7)]
    print([t.result() for t in tasks])

asyncio.run(service(fix=False))
asyncio.run(service(fix=True))
# ['T1 noodle 0.6s', 'T2 noodle 0.6s', 'T3 gratin 0.6s', 'T4 noodle 0.9s', 'T5 noodle 0.9s', 'T6 noodle 0.9s']
# ['T1 noodle 0.3s', 'T2 noodle 0.3s', 'T3 gratin 0.6s', 'T4 noodle 0.3s', 'T5 noodle 0.3s', 'T6 noodle 0.3s']

• Stellt sich Lou selbst vor den Ofen, geht keine einzige Nudelsuppe, die nur 0,3 Sekunden braucht, nach 0,3 Sekunden raus. T1 und T2 hatten ihre Töpfe schon vor dem Ofen aufgesetzt und werden trotzdem erst nach 0,6 Sekunden serviert; T4 bis T6 setzen ihre Töpfe erst auf, als der Ofen fertig ist, und gehen nach 0,9 Sekunden raus. Genau das Bild aus der Zeitleiste.
• Gibst du den Ofen mit to_thread ab, gehen die Nudelsuppen nach 0,3 Sekunden raus und das Gratin nach der Ofenzeit von 0,6 Sekunden. Das ist das Ziel für heute Nacht.

Nach Küchenschluss tippt die Chefin nacheinander auf die Rezeptkarten an der Küchenwand. Threads, Locks, Queues, Pools, Prozesse, Loop, TaskGroup, to_thread. „So viele Werkzeuge haben wir inzwischen. Aber wenn ein Gast zur Tür hereinkommt: Wie entscheidest du, welches Werkzeug du zuerst in die Hand nimmst?“

Übungen

1. Stell in der Zeitleiste den langsamen Tisch nacheinander auf 1 bis 6. Erklär, wie sich die Balken der Tische vor und nach dem langsamen Tisch unterscheiden und warum.
2. Lass das erste Beispiel aus 11-1 mit asyncio.run(main(), debug=True) laufen und prüf, welcher Task laut Warnmeldung wie viele Sekunden festgehalten hat.
3. Rechne zuerst aus, nach wie vielen Sekunden call #5 startet, wenn du das Semaphoren-Beispiel aus 11-4 auf asyncio.Semaphore(3) umstellst, und lass es dann laufen.