Python-Multiprocessing: Mehrere Prozesse richtig einsetzen

Python-Multiprocessing nutzt mehrere Prozesse statt Threads. Der Grund ist der Global Interpreter Lock (GIL): In einem CPython-Prozess kann immer nur ein Thread Python-Bytecode ausführen. Für rechenintensive Aufgaben bringt Threading deshalb nichts – Prozesse schon.

Wann welches Verfahren

# Threading - sinnvoll bei WARTEN auf Ein-/Ausgabe
#   Datei lesen, HTTP-Anfrage, Datenbankabfrage
#   Waehrend ein Thread wartet, rechnet ein anderer.
#   Der GIL wird beim Warten freigegeben.

# Multiprocessing - sinnvoll bei RECHNEN
#   Bildverarbeitung, Kryptografie, Datenanalyse, Kompression
#   Jeder Prozess hat seinen eigenen Interpreter und eigenen GIL.
#   Echte Parallelitaet auf mehreren Kernen.

# asyncio - sinnvoll bei VIELEN gleichzeitigen Verbindungen
#   Tausende Netzwerkverbindungen in einem Prozess/Thread.

# Entscheidungshilfe:
#   Rechenzeit dominiert?        -> multiprocessing
#   Wartezeit dominiert?         -> threading oder asyncio
#   Beides gemischt?             -> Prozesse, die intern asyncio nutzen
#   Kleine Aufgaben (< 1 ms)?   -> gar keine Parallelitaet,
#                                   der Verwaltungsaufwand frisst alles

# Faustregel: Der Aufwand fuer einen neuen Prozess liegt bei etwa
# 1 ms bis einigen ms (je nach Datenmenge, die uebergeben wird).
# Aufgaben, die weniger als etwa 10 ms dauern, NICHT parallelisieren.

Der einfache Weg: Prozesspool

from concurrent.futures import ProcessPoolExecutor
import time

def rechenintensiv(n: int) -> int:
    # Primzahlen bis n zaehlen - reine Rechenarbeit.
    treffer = 0
    for k in range(2, n):
        if all(k % t for t in range(2, int(k ** 0.5) + 1)):
            treffer += 1
    return treffer

werte = [20_000, 25_000, 30_000, 35_000,
         20_000, 25_000, 30_000, 35_000]

# --- Seriell ---
start = time.perf_counter()
ergebnis_seriell = [rechenintensiv(n) for n in werte]
dauer_seriell = time.perf_counter() - start

# --- Parallel ---
if __name__ == "__main__":              # WICHTIG, siehe unten!
    with ProcessPoolExecutor(max_workers=4) as pool:
        start = time.perf_counter()
        ergebnis_parallel = list(pool.map(rechenintensiv, werte))
        dauer_parallel = time.perf_counter() - start

    assert ergebnis_seriell == ergebnis_parallel   # gleiches Ergebnis!
    print(f"seriell:  {dauer_seriell:.2f}s")
    print(f"parallel: {dauer_parallel:.2f}s")
    print(f"Beschleunigung: {dauer_seriell / dauer_parallel:.1f}x")

# max_workers: Standard ist seit Python 3.13 os.process_cpu_count(),
# in aelteren Versionen os.cpu_count().
# Bei rechenintensiven Aufgaben: Anzahl der Kerne.
# Bei Aufgaben mit viel Speicher: weniger, sonst droht Speichermangel.

Ergebnisse immer vergleichen. Ein Parallelisierungsfehler, der zu leicht abweichenden Zahlen führt, ist besonders tückisch: Er fällt erst in Produktion auf. assert auf Gleichheit mit der seriellen Version ist die einfachste und wirksamste Absicherung.

Der `main`-Block

# Auf Linux (gabelnde Startmethode) funktioniert es oft ohne,
# auf Windows und macOS (spawn) NICHT:
#   Jeder neue Prozess importiert das Hauptmodul erneut.
#   Fehlt der Schutz, startet jedes Kind wieder einen Pool
#   -> Prozesslawine bis zum Speichermangel.

# RICHTIG - immer so schreiben:
def main():
    with ProcessPoolExecutor() as pool:
        ergebnisse = list(pool.map(aufgabe, daten))
    return ergebnisse

if __name__ == "__main__":
    main()

# Startmethode explizit waehlen (ab Python 3.8 eine Auswahl):
import multiprocessing as mp
if __name__ == "__main__":
    mp.set_start_method("spawn")        # sauber, aber langsamer
    # "fork"    - schnell, aber unsicher mit Threads und Bibliotheken
    # "spawn"   - neuer Interpreter, sicher, hoehere Startkosten
    # "forkserver" - Kompromiss, auf Linux verfuegbar
    # Fuer numerische Bibliotheken (BLAS, OpenMP, CUDA) ist "spawn"
    # die sichere Wahl - "fork" fuehrt dort zu Verklemmungen.

Verschiedene Arbeitsverteilungen

from concurrent.futures import ProcessPoolExecutor, as_completed

# 1. map - gleiche Funktion auf alle Elemente, Ergebnis in Reihenfolge
with ProcessPoolExecutor() as pool:
    ergebnisse = list(pool.map(verarbeite, dateien))
#   Nachteil: alles wartet, bis das erste Ergebnis fertig ist
#   (kein Problem bei list(), bei Iteration ueber das Ergebnis schon)

# 2. submit - einzelne Aufgaben, Ergebnisse nach Fertigstellung
with ProcessPoolExecutor(max_workers=8) as pool:
    zukuenftig = {pool.submit(verarbeite, d): d for d in dateien}
    for f in as_completed(zukuenftig):
        pfad = zukuenftig[f]
        try:
            print(pfad, "->", f.result())
        except Exception as fehler:
            print(pfad, "FEHLER:", fehler)
#   Vorteil: Ergebnisse erscheinen, sobald sie fertig sind.
#   Ein Fehler in einer Aufgabe killt nicht den ganzen Lauf.

# 3. Ungleiche Aufgabengroessen - Stapel bilden
#   map() verteilt der Reihe nach; bei stark unterschiedlichen
#   Laufzeiten arbeitet ein Prozess lange allein weiter.
def in_stapel(daten, groesse):
    for i in range(0, len(daten), groesse):
        yield daten[i:i + groesse]

with ProcessPoolExecutor() as pool:
    ergebnisse = []
    for teil in pool.map(verarbeite_stapel, in_stapel(daten, 50)):
        ergebnisse.extend(teil)

# 4. Fortschritt anzeigen
from tqdm import tqdm
with ProcessPoolExecutor() as pool:
    for _ in tqdm(pool.map(verarbeite, dateien), total=len(dateien)):
        pass

Speicher und Datenübergabe

Zwischen Prozessen werden Daten kopiert (serialisiert und übertragen). Das ist der häufigste Grund, warum parallele Läufe langsamer sind als erwartet.

# SCHLECHT: Ein grosses Objekt fuer jeden Aufruf uebergeben
def verarbeite(grosses_dict, schluessel):   # bei jedem Aufruf kopiert!
    return grosses_dict[schluessel] * 2

with ProcessPoolExecutor() as pool:
    ergebnisse = list(pool.map(lambda k: verarbeite(daten, k), schluessel))

# BESSER: Initialisierung je Prozess - einmal je Arbeiter, nicht je Aufgabe
import os
_globale_daten = None

def initialisiere(pfad):
    global _globale_daten
    _globale_daten = lade_daten(pfad)

def verarbeite(schluessel):
    return _globale_daten[schluessel] * 2

with ProcessPoolExecutor(initializer=initialisiere, initargs=("daten.pkl",)) as pool:
    ergebnisse = list(pool.map(verarbeite, schluessel))
#   Der Aufwand fuer das Laden entsteht einmal je Prozess.
#   Bei 4 Prozessen also 4-mal statt 1000-mal.

# Groessen, die sich nicht mehr sinnvoll kopieren lassen:
#   Ab etwa 1 GB Kopie je Aufgabe muss die Architektur geaendert werden:
#   - Daten in eine Datei legen und nur den Pfad uebergeben (mmap)
#   - Speicher teilen (multiprocessing.shared_memory)
#   - Datenbank statt Uebergabe
#   - Verarbeitung dort, wo die Daten liegen (Karte/Reduzieren)

Teilen von Speicher

# multiprocessing.shared_memory (ab Python 3.8)
from multiprocessing import shared_memory
import numpy as np

# Erzeuge einen Speicherbereich, den alle Prozesse sehen
shm = shared_memory.SharedMemory(create=True, size=1024 * 1024)
puffer = np.ndarray((1024 * 1024 // 8,), dtype=np.float64, buffer=shm.buf)

puffer[:] = np.arange(puffer.size)        # einmal befuellen

def lies_block(start_stop):
    start, stop = start_stop
    return puffer[start:stop].sum()       # liest ohne Kopie des Ganzen

if __name__ == "__main__":
    with ProcessPoolExecutor() as pool:
        teile = [(i, i + 1000) for i in range(0, puffer.size, 1000)]
        summe = sum(pool.map(lies_block, teile))
    print("Summe:", summe, "erwartet:", puffer.sum())

    shm.close()
    shm.unlink()          # PFLICHT: sonst bleibt der Bereich liegen!

# Alternative bei rein numerischer Arbeit: numpy allein
#   numpy-Operationen geben den GIL frei -> Threads statt Prozesse
#   sind dann oft einfacher UND schneller (kein Kopieren!).

Häufige Fehler

Kein if __name__ == "__main__"-Schutz. Führt bei spawn zu einer Prozesslawine.
Zu kleine Aufgaben parallelisieren. Der Aufwand für Prozessstart und Datenkopie übersteigt den Nutzen. Erst messen, dann parallelisieren.
I/O-lastige Aufgaben mit Prozessen. Für Warten auf Ein-/Ausgabe sind Threads oder asyncio besser geeignet – weniger Aufwand, kein Kopieren.
Globale Zustände erwarten. Jeder Prozess hat seine eigene Kopie. Eine Änderung im Kindprozess ist im Elternprozess nicht sichtbar. Ergebnisse müssen zurückgegeben werden.
Nicht serialisierbare Rückgaben. Offene Dateien, Verbindungen, Sperren und Lambdas lassen sich nicht übertragen. Nur einfache Daten zurückgeben.
Speicher nicht begrenzen. Vier Prozesse mit je 2 GB Bedarf brauchen 8 GB. Ein MemoryError im Kindprozess beendet oft den ganzen Lauf unklar.
Absturz ohne Meldung. Stirbt ein Arbeiter hart (Segmentation fault in einer C-Bibliothek), hebt ProcessPoolExecutor neuerdings einen Fehler – BrokenProcessPool. Das ist ein Hinweis auf ein Problem in einer Bibliothek, nicht auf einen Fehler im Verteiler.
Nicht gemessen. Erst die serielle Laufzeit messen, dann die parallele, dann rechnen. Ohne Messung ist die Parallelisierung geraten.

Kurzbeispiel: Dateiverarbeitung richtig

from concurrent.futures import ProcessPoolExecutor
from pathlib import Path
import hashlib

def pruefe(pfad):
    h = hashlib.sha256()
    with open(pfad, "rb") as f:
        for block in iter(lambda: f.read(1 << 20), b""):
            h.update(block)             # 1 MiB-Bloecke: speicherschonend
    return pfad.name, h.hexdigest()

def main():
    dateien = sorted(Path("/srv/daten").glob("*.bin"))
    # I/O-lastig: Threads sind hier oft die bessere Wahl,
    # aber hashlib gibt den GIL frei -> beides funktioniert.
    with ProcessPoolExecutor(max_workers=4) as pool:
        for name, summe in pool.map(pruefe, dateien):
            print(f"{summe[:16]}  {name}")

if __name__ == "__main__":
    main()

Python: Dateien & IO