In diesem Kapitel schlagen wir die Brücke von einfachen Funktionen über Generator-Ausdrücke hin zu modernen asynchronen Programmierkonzepten. Wir betrachten, wie das Pausieren von Funktionen (durch yield) die Grundlage für kooperatives Multitasking und effiziente I/O-Operationen mit async und await bildet.
Funktionen und Generator-Ausdrücke¶
Eine normale Python-Funktion wird von oben nach unten ausgeführt und gibt einen einzigen Wert zurück. Generatoren liefern potentiell mehrfach Werte.
Generator-Ausdrücke (Generator Expressions) werten Daten lazy (bedarfsausgelöst) aus. Anstatt alle Werte auf einmal im Speicher zu berechnen, geben sie diese sequenziell zurück (“yield”) und behalten ihren internen Zustand bei. Sie lassen sich auch hervorragend aneinanderreihen (Chaining), um speichereffiziente Berechnungspipelines zu erstellen.
# Eine Pipeline: Quadrate generieren, dann nach geraden Zahlen filtern
squares = (x * x for x in range(5))
even_squares = (sq for sq in squares if sq % 2 == 0)
print(list(even_squares)) # Output: [0, 4, 16][0, 4, 16]
Das Schlüsselwort yield und Kontrollfluss¶
Das Schlüsselwort yield erlaubt es einer Funktion, ihre Ausführung zu pausieren, optional einen Wert zurückzugeben und ihren lokalen Zustand (sowie den Instruction Pointer) beizubehalten. Ein Aufruf von next() setzt die Ausführung direkt nach dem yield fort.
def squares():
for i in range(1, 5):
yield i * i
sq_gen = squares()
print(next(sq_gen)) # Output: 1
print(next(sq_gen)) # Output: 41
4
Ein “nacktes” yield gibt die Kontrolle ab, ohne einen spezifischen Wert zurückzugeben (es liefert implizit None). Dies ist für kooperatives Multitasking essenziell. Stößt ein Generator auf ein return, wird eine StopIteration-Exception geworfen, welche den Rückgabewert enthält.
def task():
yield "Arbeite in task()..."
yield "Arbeite weiter in task()..."
yield # wie yield None
return "Fertig!"
gen = task()
print(next(gen))
print("Kontrollfluss wieder weg von task()")
print(next(gen))
assert next(gen) is None
try:
next(gen)
except StopIteration as e:
print(e.value) # Output: Fertig!Arbeite in task()...
Kontrollfluss wieder weg von task()
Arbeite weiter in task()...
Fertig!
Früher hat man so etwas, wo der Kontrollfluss wechseln musste, meist mit Callbacks gelöst. Das kennt man aus UI-Frameworks, wo bei einem Button eine onClick-Methode registriert wird. Spätestens wenn man im Callback wieder einen Callback braucht (etwa bei Netzwerkaufrufen, deren Ergebnisse Netzwerkaufrufe auslösen), wird das aber sehr unübersichtlich, und lässt auch kein Exception Handling mit try..except mehr zu. So gibt es heute das Konzept der expliziten Callbacks noch im UI-Bereich, nicht so sehr im Netzwerk-Bereich.
Das Schlüsselwort yield from¶
Um Generatoren zu verschachteln, stellt yield from eine Verbindung her. Entscheidend ist: Wenn der Sub-Generator am Ende einen Wert mit return zurückgibt, fängt yield from dieses Ergebnis auf.
def sub_task():
print("Arbeite...")
yield from (i * i for i in range(1, 5))
return "Endergebnis gut!"
def main_task():
result = yield from sub_task()
print(f"Sub-Task beendet mit: {result}")
print(
f"main_task is an instance of {type(main_task)} but main_task() is an instance of {type(main_task())}"
)
for step in main_task():
print(step)main_task is an instance of <class 'function'> but main_task() is an instance of <class 'generator'>
Arbeite...
1
4
9
16
Sub-Task beendet mit: Endergebnis gut!
Allgemein Nachrichten weiterreichen¶
Während wir bei einer Generator Expression und yield xyz einen Generator beschreiben, der eben Werte generiert, so kann man einer Methode auch Werte übergeben, und das nicht nur beim Instantiieren des Generators, sonder mittendrin mit send bzw. xyz = yield:
def flexible_methode(startwerte):
print(startwerte)
altdaten = yield 42
print("Mittendrin")
neudaten = yield 23
print("Altdaten:", altdaten)
print("Neudaten:", neudaten)
yield "Kurz vor Schluss"
yield "Verdammt kurz vor Schluss"
return "Ende"
def arbeit():
meth = flexible_methode("Anfang")
print("Jetzt geht's los")
assert next(meth) == 42
print("Jetzt geht's weiter")
result = meth.send("Daten von sonstwo")
assert result == 23
print("Jetzt geht's noch etwas weiter")
result = yield from meth
print(result)
g = arbeit()
for i, item in enumerate(g):
print(i, item)Jetzt geht's los
Anfang
Jetzt geht's weiter
Mittendrin
Jetzt geht's noch etwas weiter
Altdaten: Daten von sonstwo
Neudaten: None
0 Kurz vor Schluss
1 Verdammt kurz vor Schluss
Ende
Dabei ist next(meth) semantisch das gleiche wie meth.send(None).
Achtung: das Codebeispiel ist etwas, was man eher nicht so hinprogrammieren würde, und dient nur dem Zweck, in Ruhe den Kontrollfluss und Datenfluss nachzuvollziehen. Die Zeile Neudaten: None ist zunächst verwirrend, aber gerade dadurch wird klar, wie/wo die Werte eingefüllt werden.
Koroutinen: Das Konzept¶
Während eine Standard-Subroutine (Funktion) genau einen Einstiegs- und einen Austrittspunkt hat (beim return), ist eine Koroutine (Coroutine) eine Verallgemeinerung, die es erlaubt, die Ausführung an mehreren Punkten zu pausieren und fortzusetzen.
In Python nutzen Koroutinen die Pausieren/Fortsetzen-Mechanik von yield, um in einer Planungs-Umgebung (Scheduling) zusammenzuarbeiten, während Generatoren primär dafür gedacht sind, Daten zu produzieren. Indem eine Koroutine freiwillig die Kontrolle abgibt, bevor ihre eigentliche Aufgabe vollständig abgeschlossen ist, erlaubt sie es anderen Aufgaben, auf demselben CPU-Thread zu laufen.
Koroutinen bei I/O, GUIs und Netzwerken¶
Bei Operationen, die durch Input/Output limitiert sind (I/O-bound) – wie etwa das Abfragen einer Datenbank oder das Warten auf ein Netzwerkpaket –, verbringt die CPU die meiste Zeit im Leerlauf.
Ein klassisches synchrones Skript blockiert während dieser Wartezeit den gesamten Thread. Wenn wir diese Operationen als Koroutinen strukturieren, kann eine Funktion, die auf Netzwerkdaten wartet, die Ausführungskontrolle an einen zentralen Planer (die Event Loop) abgeben (yield). Die Event Loop treibt in dieser Zeit andere, bereite Koroutinen voran (z.B. das Rendern einer GUI). Sobald die I/O-Daten eintreffen, setzt die Event Loop die ursprüngliche Koroutine exakt an der Stelle fort, an der sie unterbrochen wurde.
Von yield from zu async / await¶
Wir nehmen unsere vorherigen Generator-Konzepte und schreiben sie mit moderner asynchroner Syntax neu. Python formalisiert das Muster des kooperativen Abgebens von Kontrolle, indem yield from durch await ersetzt wird und Koroutinen explizit mit async def markiert werden.
import asyncio
import timeasync def async_square(i):
print(f"Starte Task {i}...")
await asyncio.sleep(
1
) # Simuliert eine I/O-Pause, gibt Kontrolle an die Event Loop zurück
print(f"Beende Task {i}")
return i * iasync def main_task():
# 'await' fängt den Rückgabewert auf wie 'yield from'
for i in range(3):
result = await async_square(i)
print(f"Ergebnis ist {result}")
start_time = time.time()
# In einem reinen Python-Skript würde man die Event Loop so starten:
# asyncio.run(main_task())
# In Jupyter Notebooks läuft bereits eine Event Loop,
# daher können wir direkt 'await' auf der obersten Ebene nutzen:
await main_task()
end_time = time.time()
print(f"Es dauerte {end_time - start_time:.2f} Sekunden.")Starte Task 0...
Beende Task 0
Ergebnis ist 0
Starte Task 1...
Beende Task 1
Ergebnis ist 1
Starte Task 2...
Beende Task 2
Ergebnis ist 4
Es dauerte 3.01 Sekunden.
Außerhalb einer async-Methode darf man gar kein await nutzen, sodass dieser Code einen SyntaxError wirft:
def problematisch():
await main_task()Außerdem lassen sich async-Methoden nicht ohne await nutzen. Das Ergebnis ist eine coroutine-Instanz, von der sich nicht next() aufrufen lässt (TypeError: coroutine object is not an iterator). Versuchen wir es doch, bekommen wir ein RuntimeWarning coroutine 'main_task' was never awaited. Die Klasse coroutine ist Teil des Moduls asyncio.
ohne_await_aufgerufen = main_task()
print(type(ohne_await_aufgerufen))<class 'coroutine'>
Also: async-Methoden nur mit await aufrufen, und wenn man await in einer Methode benutzen möchte, muss die async sein. Das bedeutet, sobald man Kontakt mit der async-Welt hat, ist man in der async-Welt. Man spricht in Analogie auch davon, dass die Methode ‘gefärbt’ ist (und auf andere Methoden abfärbt).
Während wir eben gesehen haben, dass das Programm nicht schneller ablief als ein entsprechendes ‘synchrones’ Programm, können wir auch erzwingen, dass tatsächliche Nebenläufigkeit stattfindet. Dazu müssen wir aber aus der Logik der for-Schleife raus und explizit die Nebenläufigkeit mit asyncio.gather erzwingen:
async def main_concurrent():
start_time = time.time()
# asyncio.gather startet alle übergebenen Koroutinen quasi-gleichzeitig
# und wartet, bis alle abgeschlossen sind.
results = await asyncio.gather(async_square(1), async_square(2), async_square(3))
end_time = time.time()
print(f"\nAlle Ergebnisse: {results}")
print(f"Benötigte Gesamtzeit: {end_time - start_time:.2f} Sekunden")
await main_concurrent()Starte Task 1...
Starte Task 2...
Starte Task 3...
Beende Task 1
Beende Task 2
Beende Task 3
Alle Ergebnisse: [1, 4, 9]
Benötigte Gesamtzeit: 1.00 Sekunden
Structured Concurrency mit TaskGroups¶
Die Funktion asyncio.gather sammelt Koroutinen, bietet aber wenig Kontrolle, wenn eine der Aufgaben fehlschlägt. Moderne asynchrone Python-Programme (ab Python 3.11) nutzen daher das Konzept der strukturierten Nebenläufigkeit (Structured Concurrency) via asyncio.TaskGroup.
Futures und create_task()¶
Bisher haben wir meist direkt auf Koroutinen gewartet (await async_square(2)). Unter der Haube arbeitet asyncio jedoch mit sogenannten Futures. Ein Future ist ein Platzhalter-Objekt für ein Ergebnis, das in der Zukunft berechnet wird. Es kapselt den Zustand einer asynchronen Operation (ausstehend, fertig oder fehlgeschlagen).
Wenn wir eine Koroutine nicht sofort mit await blockierend abwarten wollen, sondern sie im Hintergrund starten möchten, nutzen wir asyncio.create_task(). Diese Funktion nimmt die Koroutine, verpackt sie in einen Task (eine spezielle Unterklasse von Future), plant sie in der Event Loop ein und gibt das Task-Objekt sofort zurück. Future verhält sich wie ein JavaScript Promise.
start_time = time.time()
async def background_demo():
# Startet sofort im Hintergrund, blockiert den aktuellen Fluss nicht
mein_task = asyncio.create_task(asyncio.sleep(5))
print(f"Mache etwas anderes bei {time.time() - start_time:.2f} Sekunden")
# Später können wir auf das Ergebnis des Futures/Tasks warten
await mein_task
print(f"Auf geht's bei {time.time() - start_time:.2f} Sekunden")
await background_demo()
print(f"Fertig nach {time.time() - start_time:.2f} Sekunden")Auf geht's bei 0.00 Sekunden
Mache etwas anderes bei 0.00 Sekunden
Fertig nach 5.01 Sekunden
Ein TaskGroup-Kontextmanager stellt sicher, dass alle darin gestarteten Aufgaben entweder gemeinsam erfolgreich beendet werden, oder - falls eine Aufgabe eine Exception wirft - alle anderen noch laufenden Aufgaben sauber abgebrochen werden. Dies verhindert verwaiste Tasks.
async def main_taskgroup():
start_time = time.time()
# Der Kontextmanager wartet am Ende des Blocks automatisch auf alle Tasks
async with asyncio.TaskGroup() as tg:
# tg.create_task() plant die Koroutine sofort in der Event Loop ein
tasks = [tg.create_task(async_square(i)) for i in range(3)]
# Hier sind wir erst, wenn der with-Block komplett abgearbeitet ist
results = [task.result() for task in tasks]
end_time = time.time()
print(f"\nAlle Ergebnisse (TaskGroup): {results}")
print(f"Benötigte Gesamtzeit: {end_time - start_time:.2f} Sekunden")
await main_taskgroup()Starte Task 0...
Starte Task 1...
Starte Task 2...
Beende Task 0
Beende Task 1
Beende Task 2
Alle Ergebnisse (TaskGroup): [0, 1, 4]
Benötigte Gesamtzeit: 1.00 Sekunden
Dabei haben wir async with gesehen (es gibt auch async for). Dieses Konstrukt ist notwendig, weil with bzw. for syntactic sugar sind für die Dunder-Methoden __enter__/__exit__/__iter__/__next__, die somit also auch async sein dürfen.
Das Nebenläufigkeitsmodell: Präemptiv vs. Kooperativ¶
Wenn wir mehrere Koroutinen planen, laufen diese nebenläufig (concurrent), aber nicht parallel. In einer Standard-Python asyncio-Umgebung hat das Nebenläufigkeitsmodell spezifische technische Randbedingungen:
Single Process & Single Thread: Alles läuft innerhalb eines einzigen Betriebssystem-Prozesses und auf exakt demselben physischen CPU-Kern.
Kooperatives Multitasking (Cooperative Multitasking): Aufgaben müssen die Kontrolle freiwillig über das Schlüsselwort
awaitan die Event Loop zurückgeben.
Dies unterscheidet sich fundamental vom präemptiven Multitasking (Preemptive Multitasking), das moderne Desktop-Betriebssysteme wie Linux, Windows oder macOS verwenden. In einem präemptiven System unterbricht der OS-Kernel einen laufenden Thread gewaltsam, um einem anderen Programm Rechenzeit zuzuteilen. In Pythons kooperativem Modell gibt es diese erzwungene Unterbrechung nicht. Wenn eine async-Funktion eine aufwändige mathematische Berechnung ohne await-Anweisung ausführt, blockiert sie die CPU vollständig (“hogging”) und verhindert, dass alle anderen Aufgaben auf der Event Loop jemals ausgeführt werden.
Geschichte¶
Die Idee von Koroutinen geht auf Melvin Conway im Jahre 1958 für Assembler-Programme zurück und wurde auch schon in Simula 67 in die Sprache integriert. Statt dem kooperativen Multitasking wurde aber mit präemptiven Multitasking und Hardware-Interrrupts gearbeitet. Die neuere Beschäftigung mit kooperativem Multitasking kam erst mit der Anforderung, dass ein Webserver 10000 Anfragen gleichzeitig bearbeiten können sollte. Der Apache Webserver z.B. hat für jede Verbindung einen eigenen Thread verwendet, was schnell problematisch wird. Das erklärt, wieso die Industrie weigehend zu Nginx oder Node.js gewechselt ist. Tatsächlich war es Microsoft, die in C# 5.0 mit async/await erstmals wieder Sprachunterstützung für kooperatives Multitasking eingeführt haben. Python hat 2015 in Python 3.5 nachgezogen (PEP 492) und später auch z.B. Javascript und Kotlin.
Die Async-REPL¶
Normalerweise kann await nur innerhalb einer async def-Funktion verwendet werden. Auf der obersten Ebene (Top-Level) eines Skripts muss asyncio.run(main()) verwendet werden, um die Event Loop initial zu starten.
Python bietet jedoch eine Async-REPL, um interaktives Testen zu erleichtern. Wenn man den Python-Interpreter im Terminal mit python -m asyncio bzw. uv run python -m asyncio startet, öffnet sich eine spezielle Konsole, in der im Hintergrund bereits eine Event Loop läuft. In dieser REPL kann await direkt auf der obersten Ebene verwendet werden, ohne eine Wrapper-Funktion schreiben zu müssen:
# Innerhalb von 'python -m asyncio' im Terminal
>>> import asyncio
>>> await asyncio.sleep(1) # Funktioniert direkt!
>>> Jupyter Notebooks (die im Hintergrund von IPython betrieben werden) unterstützen Top-Level await ebenfalls nativ von Haus aus und fungieren im Grunde als eine persistente, interaktive Async-REPL.
Anwendungsgebiete¶
Immer, wenn Code warten muss, etwa auf eine langsame Festplatte, Netzwerkverbindung oder Nutzereingaben, kann währenddessen anderer Code laufen. Das sind genau die Situationen, in denen wir mit async/await arbeiten können. Wenn wir das mit Hilfe einer Bibliothek tun, müssen wir uns aber klar machen, dass das Programm sich einer Event Loop unterordnen muss, also selbst auch mit async/await strukturiert werden muss. Das Modul httpx z.B. bietet sowohl eine synchrone Schnittstelle (bei der eine HTTP-Anfrage den Kontrollfluss aufhält, bis das Ergebnis da ist) als auch eine asynchrone Schnittstelle (die den Kontrollfluss abgibt und wieder aufnimmt, sobald Ergebnisse da sind), allerdings erzwingt die asynchrone Schnittstelle, das Ergebnis zu awaiten, muss also aus einer async-Umgebung gestartet werden. Übrigens ist die synchrone Schnittstelle aus der asynchronen generiert indem alle asynchronen Befehle durch synchrone Pendants ersetzt werden. An dieser Stelle sei noch erwähnt, dass httpx dort zum Einsatz kommt, wo früher requests genutzt wurde, gerade auch weil httpx im Gegensatz zu requests eine asynchrone Schnittstelle bereitstellt. Außerdem ist seit kurzem httpx2 der aktuell von der Community gepflegte Fork.
Asynchron im synchronen Code¶
Wenn man asynchronen Code geschrieben hat (gut!), den man nun in synchronem Code nutzen möchte, kann man einfach asyncio.run als Wrapper verwenden. Das klappt nur im synchronen Kontext - läuft bereits eine Event Loop, kann man kein asyncio.run mehr starten (muss man aber auch nicht mehr).
async def _internal_async_fetch():
await asyncio.sleep(1)
return "data"
def fetch_data():
return asyncio.run(_internal_async_fetch())
# print(fetch_data()) # <-- schlägt fehl im Jupyter Notebook, weil bereits asynchronProzesse, Threads und der GIL¶
Um asynchrone Programmierung richtig einzuordnen, müssen wir die alternativen Nebenläufigkeitsmodelle (Concurrency Models) betrachten, die das Betriebssystem bietet:
Prozesse: Ein Prozess ist eine eigenständige Programminstanz mit eigenem, isoliertem Speicherbereich. Auf Unix-Systemen werden neue Prozesse oft durch einen fork()-Aufruf erstellt, der den Speicher des Elternprozesses kopiert (dank Copy-on-Write-Mechanik meistens recht effizient). Prozesse bieten maximale Isolation, aber die Interprozesskommunikation (IPC) ist aufwändig. In Python nutzen wir dafür das
multiprocessing-Modul.Threads: Threads existieren innerhalb eines Prozesses und teilen sich denselben Arbeitsspeicher. Das macht den Datenaustausch trivial, birgt aber die massive Gefahr von Race Conditions bei gleichzeitigem Zugriff auf Variablen, weshalb Locks und Mutexe zur Synchronisation nötig sind.
Blockierenden Code auslagern: asyncio.to_thread()¶
Was passiert, wenn wir innerhalb einer asynchronen Anwendung eine synchrone, blockierende Funktion aufrufen müssen (etwa einen alten Datenbanktreiber oder rechenintensive Datenverarbeitung)? Ein Aufruf wie time.sleep(2) ohne await würde die gesamte Event Loop anhalten, kein anderer Code könnte mehr ausgeführt werden.
Um das zu verhindern, können wir solche synchronen Funktionen mit asyncio.to_thread() an einen separaten Thread-Pool übergeben. Die Event Loop läuft währenddessen ungehindert weiter:
def legacy_blocking_io(task_id):
"""Eine klassische synchrone Funktion, die den Thread blockiert."""
print(f"[Thread] Starte langsames I/O für Aufgabe {task_id}...")
time.sleep(1) # Blockiert, gibt Kontrolle nicht kooperativ ab!
return f"Daten {task_id}"
async def main_to_thread():
print("Starte main_to_thread...")
# Lagere den blockierenden Aufruf in einen Hintergrund-Thread aus
# und erhalte ein Awaitable zurück, auf das die Event Loop warten kann
result1 = asyncio.to_thread(legacy_blocking_io, 1)
result2 = asyncio.to_thread(legacy_blocking_io, 2)
# Wir können diese Threads nun nebenläufig abwarten
results = await asyncio.gather(result1, result2)
print(f"Ergebnisse aus den Threads: {results}")
start_time = time.time()
await main_to_thread()
end_time = time.time()
print(f"Benötigte Gesamtzeit: {end_time - start_time:.2f} Sekunden")Starte main_to_thread...
[Thread] Starte langsames I/O für Aufgabe 1...
[Thread] Starte langsames I/O für Aufgabe 2...
Ergebnisse aus den Threads: ['Daten 1', 'Daten 2']
Benötigte Gesamtzeit: 1.00 Sekunden
Der Global Interpreter Lock (GIL)¶
Die Standard-Python-Implementierung (CPython) nutzt historisch bedingt den GIL. Dieser Mutex stellt sicher, dass zu jedem Zeitpunkt immer nur ein einziger OS-Thread Python-Bytecode ausführt. Das war notwendig, um die Garbage Collection sicher durchzuführen.
Das bedeutet: Selbst auf einer CPU mit 16 physischen Kernen läuft klassischer multithreaded Python-Code, der reine Rechenoperationen ausführt (CPU-bound), nicht schneller als auf einem Kern. Für I/O-bound Aufgaben (wie Web-Requests oder Festplattenzugriffe) ist Multithreading dennoch nützlich, da Python den GIL während des Wartens auf die externe Operation freigibt.
noGIL (Freethreading)¶
Eine der massivsten architektonischen Änderungen in der Geschichte von Python ist die Einführung von “Freethreading” (PEP 703) ab Python 3.13 (zunächst als Experiment). In Python 3.15 gilt der noGIL-Build 3.15t als stabil, ist aber nicht der normale Python-Build 3.15 (vielleicht ab Python 3.16, abhängig vom Community-Feedback). Durch den vollständigen Verzicht auf den GIL wird echtes paralleles Ausführen von Threads auf mehreren CPU-Kernen ermöglicht. Dies erfordert jedoch Anpassungen im gesamten C-Extension-Ökosystem (wie etwa Numpy), damit es noch ‘threadsafe’ bleibt.
Man kann das ausprobieren mit
uv run --python 3.15t programm.pyund von Innen sehen, ob der GIL noch aktiviert ist:
import sys
sys._is_gil_enabled()TrueHigh-Level Threading (concurrent.futures)¶
Mit der Einführung von Freethreading wird concurrent.futures zum neuen Standard für rechenintensive (CPU-bound) Parallelisierung. Der ThreadPoolExecutor abstrahiert für uns.
Entscheidend ist hierbei das funktionale Programmierparadigma: Eingaben werden auf reine Funktionen (pure functions) abgebildet, die keinen globalen Zustand verändern.
def heavy_computation(x):
# Simuliert eine rechenintensive Aufgabe
time.sleep(0.5)
return x * xfrom concurrent.futures import ThreadPoolExecutor
start_time = time.time()
with ThreadPoolExecutor(max_workers=4) as executor:
# map() verteilt die Eingaben automatisch auf die Worker-Threads
results = list(executor.map(heavy_computation, [1, 2, 3, 4]))
end_time = time.time()
print(f"Ergebnisse aus Threads: {results}")
print(f"Dauer: {end_time - start_time:.2f} Sekunden")Ergebnisse aus Threads: [1, 4, 9, 16]
Dauer: 0.50 Sekunden
Green Threads¶
Als grüne Threads bezeichnet man Ansätze, mit einer Art VM Threads am Betriebssystem vorbei zu verwalten. Das gab es etwa in frühen Java-Versionen, ist im Wesentlichen wie bei Goroutinen in Go und es ist das Design Pattern hinter der Python-Bibliothek gevent, was zum Beispiel bei früheren Versionen von Flask und Django genutzt wurde, als es noch kein async/await gab. Interessant an gevent, auch wenn es heute eher nicht in neuen Projekten zum Einsatz kommt, ist die Vermeidung von ‘function coloring’ durch Monkey Patching. So kann man synchronen Code schreiben, der dann aber asynchron ausgeführt wird.
Heute benutzt man statt green Threads eher explizite asynchrone I/O.
uvloop und Granian¶
Eine Möglichkeit, asynchrone Schnittstellen noch schneller zu bekommen, ist wenn man die Event Loop von asyncio ersetzt durch eine, die in C geschrieben ist, etwa uvloop, das ist eine Schnittstelle zu libuv, ursprünglich die I/O Engine von Node.js.
Wenn man etwa eine Webapp mit fastapi bastelt und diese auf einem uvicorn-Server startet, nutzt man implizit uvloop. Das ist der wesentliche Grund, wieso Webanwendungen mit Python vergleichsweise performant sind. Ein weiterer ist, dass man nun uvicorn durch Granian ersetzen kann, das ist ein in Rust geschriebener, noch etwas schnellerer Server im ASGI-Standard (asynchrones Server Gateway Interface).
Systemressourcen und echtes Multiprocessing¶
Während asyncio und kooperatives Multitasking perfekt für I/O-lastige Aufgaben (Netzwerk, Festplatte) geeignet sind, helfen sie bei rechenintensiven Aufgaben (CPU-bound) nicht weiter. Wegen des Global Interpreter Locks (GIL) kann in CPython standardmäßig immer nur ein Thread echten Python-Bytecode ausführen.
Um alle Kerne einer modernen CPU voll auszulasten, müssen wir auf Prozess-Ebene parallelisieren.
Ressourcen prüfen mit psutil¶
Bevor wir Prozesse starten, sollten wir wissen, über welche Ressourcen das System verfügt. Das externe Paket psutil (Python System and Process Utilities) bietet eine plattformübergreifende Schnittstelle dafür:
import psutil
# Anzahl der logischen CPU-Kerne (inkl. Hyperthreading)
cores = psutil.cpu_count(logical=True)
# Physischer Arbeitsspeicher in GB
ram_gb = psutil.virtual_memory().total / (1024**3)
print(f"System: {cores} Kerne, {ram_gb:.1f} GB RAM")System: 12 Kerne, 14.9 GB RAM
Der fork-Systemaufruf und multiprocessing¶
Auf Unix-basierten Systemen werden neue Prozesse typischerweise durch einen fork()-Systemaufruf erstellt. Dabei wird der gesamte Speicher des Elternprozesses für den Kindprozess geklont. Dank der “Copy-on-Write”-Mechanik des Betriebssystems passiert dies schnell und benötigt zunächst kaum zusätzlichen Speicher, da Speicherseiten erst dupliziert werden, wenn ein Prozess sie verändert.
Das Python-Standardmodul multiprocessing abstrahiert diesen Prozess. Im Gegensatz zu Threads (die sich den Speicher teilen) hat jeder Prozess seinen eigenen Python-Interpreter und eigenen Speicher.
import multiprocessingstart_time = time.time()
# Einen Pool mit 4 Worker-Prozessen starten
with multiprocessing.Pool(processes=4) as pool:
# Die Daten werden serialisiert, an die Prozesse verteilt und parallel verarbeitet
results = pool.map(heavy_computation, [1, 2, 3, 4])
print(results)
end_time = time.time()
print(f"Dauer: {end_time - start_time:.2f} Sekunden")[1, 4, 9, 16]
Dauer: 0.55 Sekunden
Datenaustausch zwischen Prozessen (IPC)¶
Da Prozesse streng isolierten Speicher haben, können sie nicht wie Threads einfach auf dieselbe globale Variable zugreifen. Die Kommunikation muss explizit über sogenannte Inter-Process Communication (IPC) erfolgen.
Das multiprocessing-Modul bietet dafür unter anderem Queues und Pipes. Hierbei kümmert sich Python unter der Haube darum, die Objekte zu serialisieren (pickle), in einen Bytestrom zu verwandeln, an den anderen Prozess zu senden und dort wieder zu deserialisieren. Ein klassisches Beispiel für IPC ist das Erzeuger-Verbraucher-Muster. Ein Prozess generiert Daten und schiebt sie in die Warteschlange, ein anderer Prozess holt sie ab. Wichtig ist dabei das Konzept der Sentinel Value, ein spezielles Signal wie None, um dem Verbraucher mitzuteilen, dass keine Daten mehr kommen:
SLEEPTIME = 0.000001 # Simuliert Dauer der Datengenerierung
def producer(queue):
"""Dieser Worker generiert Daten."""
for i in range(1, 4):
item = f"Datenpaket {i}"
print(f"[Produzent] Erstelle und sende: {item}")
queue.put(item)
time.sleep(SLEEPTIME)
queue.put(None)
print("[Produzent] Arbeit beendet.")
def consumer(queue):
"""Dieser Worker wartet auf Daten und verarbeitet sie."""
while True:
# get() blockiert solange, bis ein neues Element verfügbar ist
item = queue.get()
# Prüfen auf Beendigungssignal
if item is None:
print("[Konsument] Signal zum Beenden empfangen.")
break
print(f"[Konsument] Empfange und verarbeite: {item}")
def main():
# Erstelle eine prozessübergreifende, threadsichere Queue
ipc_queue = multiprocessing.Queue()
# Starte zwei völlig unabhängige Prozesse, die sich nur die Queue teilen
p1 = multiprocessing.Process(target=producer, args=(ipc_queue,))
p2 = multiprocessing.Process(target=consumer, args=(ipc_queue,))
p1.start()
p2.start()
# Hauptprogramm wartet, bis beide Prozesse fertig sind
p1.join()
p2.join()
main()[Produzent] Erstelle und sende: Datenpaket 1
[Produzent] Erstelle und sende: Datenpaket 2
[Konsument] Empfange und verarbeite: Datenpaket 1
[Produzent] Erstelle und sende: Datenpaket 3[Konsument] Empfange und verarbeite: Datenpaket 2
[Produzent] Arbeit beendet.
[Konsument] Empfange und verarbeite: Datenpaket 3
[Konsument] Signal zum Beenden empfangen.
Hier fällt auf, dass der Zeilenumbruch nicht immer ‘korrekt’ aussieht. Das liegt daran, dass das Python-print den Zeilenumbruch separat auf den Output-Stream schreibt. Dieser Vorgang kann zwischendurch unterbrochen werden, sodass ein print(xyz) tatsächlich durch ein anderes print(abc) unterbrochen wird und somit der Zeilenumbruch des ersten erst nach dem zweiten ausgedruckt wird. Wenn man dieses Verhalten beobachten möchte, kann man den Parameter SLEEPTIME anpassen.
Embarrasingly parallel¶
Viele Probleme erfordern gar keine IPC, sondern sind von der Natur einer for-Schleife, bei der es nicht so sehr auf die Reihenfolge ankommt, und die Instanzen parallel verarbeitet werden könnten. Das lässt sich sehr einfach mit joblib umsetzen (was übrigens auch bei scikit-learn zum Einsatz kommt):
from joblib import Parallel, delayed
def analyze_dataset(dataset_id):
"""
Simuliert eine rechenintensive, isolierte Aufgabe pro Datensatz.
"""
print(f"Starte Analyse von Datensatz {dataset_id}...")
# Simuliere Rechenzeit
time.sleep(1)
result = dataset_id**2
print(f"-> Datensatz {dataset_id} abgeschlossen.")
return result
# Wir haben 8 Datensätze, die alle Rechenzeit benötigen
datasets = [1, 2, 3, 4, 5, 6, 7, 8]
start_time = time.time()
# Parallele Ausführung
# n_jobs=-1 nutzt alle verfügbaren Kerne
# n_jobs=4 nutzt exakt 4 Kerne
results = Parallel(n_jobs=-1)(delayed(analyze_dataset)(ds_id) for ds_id in datasets)
end_time = time.time()
print(f"\nAlle Ergebnisse: {results}")
print(f"Benötigte Gesamtzeit: {end_time - start_time:.2f} Sekunden Walltime")
Alle Ergebnisse: [1, 4, 9, 16, 25, 36, 49, 64]
Benötigte Gesamtzeit: 1.81 Sekunden Walltime
Ausblick¶
Mit Ray lassen sich Aufgaben ähnlich wie mit joblib (und mit dem identischen Interface) auf mehrere Rechner verteilen (ohne dass man den Unterschied zu Threads deutlich merkt).
!export RAY_ACCEL_ENV_VAR_OVERRIDE_ON_ZERO=0
# Um eine Warnung zu unterdrücken ;-)import ray
# Ray initialisieren (startet lokale Worker oder verbindet sich mit einem Cluster)
ray.init()
# Der Decorator macht aus einer normalen Funktion einen verteilten "Task"
@ray.remote
def analyze_dataset(dataset_id):
time.sleep(1)
return dataset_id * 1000
# Paralleler Aufruf (gibt sofort "ObjectRefs" (Futures) zurück, ohne zu blockieren)
futures = [analyze_dataset.remote(i) for i in range(8)]
# ray.get() sammelt die Ergebnisse ein, sobald sie berechnet wurden
results = ray.get(futures)
print(results)
ray.shutdown()2026-06-17 17:33:22,692 INFO worker.py:2012 -- Started a local Ray instance.
/home/voelkel/.local/lib/python3.13/site-packages/ray/_private/worker.py:2051: FutureWarning: Tip: In future versions of Ray, Ray will no longer override accelerator visible devices env var if num_gpus=0 or num_gpus=None (default). To enable this behavior and turn off this error message, set RAY_ACCEL_ENV_VAR_OVERRIDE_ON_ZERO=0
warnings.warn(
[0, 1000, 2000, 3000, 4000, 5000, 6000, 7000]
Die Data-Science-Bibliothek dask stellt einen Data Frame zur Verfügung, der verteilt auf mehreren Rechnern abgespeichert ist, aber ein Interface anbietet, was dies wegabstrahiert. Damit lassen sich dann auch direkt Machine-Learning-Aufgaben, etwa mit scikit-learn parallelisiert ausführen.
Exkurs: WebAssembly Pyodide¶
Wenn wir Python-Code im Browser ausführen (z. B. über Pyodide in JupyterLite), ändert sich die Laufzeitumgebung drastisch. Der Browser ist eine streng isolierte Sandbox. Betriebssystem-Aufrufe wie fork(), auf denen das multiprocessing-Modul basiert, existieren dort nicht.
In solchen Umgebungen ist die Event Loop unsere einzige Möglichkeit, Programme responsiv zu halten. Wenn eine synchrone Python-Funktion in Pyodide läuft und lange rechnet, friert der gesamte Haupt-Thread des Browsers (inklusive der grafischen Oberfläche) ein. Angenehm ist, dass man in Pyodide JavaScript-Promises in Python-Futures umgewandelt bekommt, und so eine Brücke zwischen JavaScript-Web-APIs und Python-Code hat.