Multiprocessing in Python für CPU-intensive Aufgaben

1. Überblick

Wenn Python-Code viel rechnen muss – etwa Bildverarbeitung, Checksummen berechnen oder numerische Simulationen – wirst du feststellen, dass Threading kaum schneller ist. Der Grund dafür ist der GIL (Global Interpreter Lock). Dieser Artikel erklärt das Problem kurz und prägnant. Danach zeige ich dir, wie du die Module multiprocessing und ProcessPoolExecutor nutzt, um die Arbeit auf mehrere Prozesse zu verteilen. Danach weißt du, wie du CPU-bound-Aufgaben parallelisierst, Tests dafür schreibst und häufige Fehler vermeidest.

Alle Beispiele verwenden Python 3.12+ und nur die Standard-Bibliothek, ohne zusätzliche Installationen.

2. Warum Threading nicht hilft

2.1. Eine CPU-intensive Aufgabe

Wir verwenden eine Funktion, die Primzahlen zählt, in reinem Python geschrieben. Das ist eine typische CPU-intensive Aufgabe:

# cpu_task.py
import math

def count_primes(start: int, end: int) -> int:
    count = 0
    for n in range(max(start, 2), end):
        if all(n % d for d in range(2, math.isqrt(n) + 1)):
            count += 1
    return count

Die Funktion durchläuft jede Zahl im Bereich [start, end) und testet die Teilung durch Zahlen von 2 bis zur Quadratwurzel. Die Funktion liest keine Dateien und macht keine Netzwerkaufrufe. Sie rechnet nur.

2.2. Zeitmessung: Sequenziell und Thread

# bench_threads.py
import time
from concurrent.futures import ThreadPoolExecutor
from cpu_task import count_primes

RANGES = [(i, i + 125_000) for i in range(0, 1_000_000, 125_000)]

def run_sequential() -> int:
    return sum(count_primes(a, b) for a, b in RANGES)

def run_threads() -> int:
    starts, ends = zip(*RANGES)
    with ThreadPoolExecutor(max_workers=8) as executor:
        return sum(executor.map(count_primes, starts, ends))

if __name__ == "__main__":
    for fn in (run_sequential, run_threads):
        t0 = time.perf_counter()
        result = fn()
        print(f"{fn.__name__}: {result} Primzahlen, {time.perf_counter() - t0:.2f}s")

RANGES teilt den Bereich [0, 1_000_000) in 8 Teile auf. run_threads vergibt jeden Teil an einen Thread. Ergebnisse auf einer 8-Kern-Maschine (deine Ergebnisse werden unterschiedlich sein):

run_sequential: 78498 Primzahlen, 6.41s
run_threads: 78498 Primzahlen, 6.58s

Die Thread-Variante ist sogar ein wenig langsamer. In CPython können zu jedem Zeitpunkt nur ein Thread Python-Bytecode ausführen, weil es den GIL halten muss. Acht Threads wechseln sich in der Nutzung eines Kerns ab und verursachen zusätzliche Wechselkosten.

Hinweis: Ab Python 3.13 gibt es einen free-threaded Build (python3.13t), der den GIL entfernt. Dieser Build befindet sich noch in der Experimentalphase, und viele C-Erweiterungen unterstützen ihn noch nicht. Für die meisten aktuellen Projekte ist die Verwendung von Prozessen immer noch der sicherste Weg.

3. Grundlagen von multiprocessing

Jeder Prozess hat seinen eigenen Interpreter und seinen eigenen GIL. Daher können N Prozesse wirklich parallel auf N Kernen laufen.

3.1. Manuelle Process-Erstellung

# basic_process.py
from multiprocessing import Process, Queue
from cpu_task import count_primes

def worker(start: int, end: int, results: Queue) -> None:
    results.put(count_primes(start, end))

if __name__ == "__main__":
    results = Queue()
    ranges = [(0, 500_000), (500_000, 1_000_000)]
    processes = [Process(target=worker, args=(a, b, results)) for a, b in ranges]

    for p in processes:
        p.start()
    total = sum(results.get() for _ in processes)
    for p in processes:
        p.join()

    print("Gesamtzahl Primzahlen:", total)

Erklärung Teil für Teil:

• Process(target=..., args=...) beschreibt einen untergeordneten Prozess. Er läuft noch nicht, bis du start() aufrufst.
• Der untergeordnete Prozess teilt keinen Speicher mit dem übergeordneten Prozess. Um Ergebnisse zurückzugeben, musst du einen Kommunikationskanal verwenden. Hier ist es Queue.
• Das Programm ruft results.get() vor join() auf. Wenn join() zuerst aufgerufen wird, während die Queue noch große Daten enthält, die nicht gelesen wurden, kann das Programm in einen Deadlock geraten.

Gesamtzahl Primzahlen: 78498

Diese Methode ist flexibel, aber ausführlich. In der Praxis verwaltest du Process selten manuell.

3.2. Verwendung von Pool

Pool erstellt vorab eine Gruppe von Workern und verteilt die Arbeit automatisch zwischen ihnen:

# pool_example.py
from multiprocessing import Pool
from cpu_task import count_primes

RANGES = [(i, i + 125_000) for i in range(0, 1_000_000, 125_000)]

if __name__ == "__main__":
    with Pool(processes=8) as pool:
        counts = pool.starmap(count_primes, RANGES)
    print(counts)
    print("Gesamt:", sum(counts))

starmap teilt jedes Tupel in RANGES in die Parameter start, end auf. Die Ergebnisse werden in der gleichen Reihenfolge wie die Eingaben zurückgegeben:

[10746, 9895, 9622, 9480, 9637, 9554, 9454, 9610]
Gesamt: 78498

Wenn du processes nicht angibst, verwendet Pool os.cpu_count() Worker. Der with-Block bereinigt die Worker am Ende.

4. ProcessPoolExecutor: Eine modernere API

concurrent.futures.ProcessPoolExecutor hat die gleiche Schnittstelle wie ThreadPoolExecutor. Deshalb kannst du von Threads zu Prozessen wechseln, indem du einfach den Klassennamen änderst:

# parallel.py
import os
from concurrent.futures import ProcessPoolExecutor
from cpu_task import count_primes

def split_range(start: int, end: int, parts: int) -> list[tuple[int, int]]:
    step = (end - start + parts - 1) // parts
    return [(s, min(s + step, end)) for s in range(start, end, step)]

def parallel_count_primes(start: int, end: int, workers: int | None = None) -> int:
    workers = workers or os.cpu_count() or 1
    chunks = split_range(start, end, workers)
    starts, ends = zip(*chunks)
    with ProcessPoolExecutor(max_workers=workers) as executor:
        return sum(executor.map(count_primes, starts, ends))

if __name__ == "__main__":
    import time
    t0 = time.perf_counter()
    print(parallel_count_primes(0, 1_000_000, workers=8))
    print(f"{time.perf_counter() - t0:.2f}s")

78498
1.27s

Die Zeit sinkt von etwa 6,4 Sekunden auf 1,3 Sekunden. Die Geschwindigkeit steigt nicht um genau 8x aus zwei Gründen. Erstens haben die Segmente unterschiedliche Arbeitslasten: größere Zahlenbereiche brauchen länger zum Überprüfen. Zweitens kostet auch das Starten von Prozessen und die Datenübertragung Zeit.

Wenn du jedes Ergebnis verarbeiten möchtest, sobald es fertig ist, verwende submit zusammen mit as_completed:

from concurrent.futures import ProcessPoolExecutor, as_completed
from cpu_task import count_primes

if __name__ == "__main__":
    with ProcessPoolExecutor() as executor:
        futures = {executor.submit(count_primes, a, a + 250_000): a
                   for a in range(0, 1_000_000, 250_000)}
        for future in as_completed(futures):
            print(f"Segment ab {futures[future]}: {future.result()} Primzahlen")

Wenn ein Worker eine Exception wirft, wirft future.result() dieselbe Exception im übergeordneten Prozess. Daher kannst du Fehler wie gewohnt mit try/except abfangen.

5. Viele kleine Aufgaben: chunksize und imap_unordered

Jedes Mal, wenn eine Aufgabe an einen Worker gesendet wird, muss Python den Parameter pickeln, ihn über eine Pipe senden und das Ergebnis unpickeln. Wenn du 100.000 sehr kleine Aufgaben hast, können diese Kosten größer sein als die Berechnung selbst. Du kannst das beheben, indem du Aufgaben in Batches mit chunksize zusammenfasst:

# chunks.py
from multiprocessing import Pool

def is_prime(n: int) -> bool:
    if n < 2:
        return False
    return all(n % d for d in range(2, int(n ** 0.5) + 1))

if __name__ == "__main__":
    numbers = range(1_000_000)
    with Pool() as pool:
        total = sum(pool.imap_unordered(is_prime, numbers, chunksize=10_000))
    print(total)

• chunksize=10_000 sorgt dafür, dass jede Sendung 10.000 Zahlen statt 1 enthält. Die Anzahl der Kommunikationen zwischen Prozessen sinkt drastisch.
• imap_unordered gibt Ergebnisse lazy zurück und ohne Reihenfolge. Dies ist gut, wenn du nur aggregieren möchtest, etwa summieren, und Speicher sparen willst.
• ProcessPoolExecutor.map akzeptiert auch einen Parameter chunksize. Der Standardwert ist 1, also vergiss nicht, ihn anzupassen, wenn du viele kleine Aufgaben hast.

6. Ressourcen einmal pro Worker initialisieren

Angenommen, jede Aufgabe benötigt eine Nachschlagetabelle, die teuer zu erstellen ist. Wenn du die Tabelle bei jedem Funktionsaufruf neu erstellst, ist das sehr verschwendet. Der Parameter initializer ermöglicht es dir, eine Funktion genau einmal auszuführen, wenn der Worker startet:

# initializer_example.py
from concurrent.futures import ProcessPoolExecutor

_price_table: dict[str, float] = {}

def load_prices(currency: str) -> None:
    global _price_table
    rate = 25_000 if currency == "VND" else 1
    _price_table = {f"SKU-{i}": i * 1.5 * rate for i in range(100_000)}

def order_total(skus: list[str]) -> float:
    return sum(_price_table[s] for s in skus)

if __name__ == "__main__":
    orders = [["SKU-1", "SKU-2"], ["SKU-10", "SKU-20", "SKU-30"]]
    with ProcessPoolExecutor(max_workers=2, initializer=load_prices,
                             initargs=("VND",)) as executor:
        print(list(executor.map(order_total, orders)))

[112500.0, 2250000.0]

Die Variable _price_table ist eine globale Variable in jedem Worker. Der übergeordnete Prozess sieht diese Änderungen nicht. Dies ist eine sichere Möglichkeit, dass jeder Worker über eigene Ressourcen verfügt, ohne die Nachschlagetabelle bei jeder Aufgabe zu senden.

7. Tests für parallelen Code schreiben

Du kannst Code mit Prozessen wie normalen Code mit pytest testen. Die Voraussetzung ist, dass die Worker-Funktion auf Modulebene definiert sein muss:

# test_parallel.py
from cpu_task import count_primes
from parallel import parallel_count_primes, split_range

def test_count_primes_known_value():
    assert count_primes(0, 100) == 25

def test_split_range_covers_whole_interval():
    chunks = split_range(0, 10, 3)
    assert chunks == [(0, 4), (4, 8), (8, 10)]

def test_parallel_matches_sequential():
    expected = count_primes(0, 50_000)
    assert parallel_count_primes(0, 50_000, workers=4) == expected

pip install pytest
pytest -q test_parallel.py

...                                                    [100%]
3 bestanden in 0.62s

Der wichtigste Test ist test_parallel_matches_sequential. Er gewährleistet, dass die Segmentierung nicht überspringt oder Elemente doppelt zählt. Das ist ein häufiger Fehler beim Schreiben von split_range.

8. Häufige Fehler und wann man multiprocessing nicht verwenden sollte

if __name__ == "__main__": vergessen.

Mit der Start-Methode spawn (Standard auf Windows und macOS) oder forkserver (Standard auf Linux ab Python 3.14) importiert der untergeordnete Prozess das Hauptmodul erneut. Wenn der Pool-Erstellungscode auf Modulebene liegt, erzeugt jeder untergeordnete Prozess einen neuen Pool. Dann erhältst du einen RuntimeError oder Prozesse werden endlos erzeugt.

Lambda oder verschachtelte Funktionen weitergeben.

Alles, das an Worker gesendet wird, muss pickelbar sein:

from concurrent.futures import ProcessPoolExecutor

if __name__ == "__main__":
    with ProcessPoolExecutor() as executor:
        print(list(executor.map(lambda x: x * x, range(5))))

_pickle.PicklingError: Can't pickle <function <lambda> at 0x...>: attribute lookup <lambda> on __main__ failed

Definiere Funktionen mit def auf Modulebene. Wenn du Parameter festlegen musst, verwende functools.partial.

Zu große Datenmengen weitergeben.

Eine Liste mit Millionen von Elementen, die an jede Aufgabe übergeben wird, wird mehrfach gepickelt. Die Kopierzeit kann die Berechnung überlagern. Es gibt mehrere Abhilfen: Sende nur Indizes oder Dateipfade, verwende initializer, oder nutze multiprocessing.shared_memory mit Array-Daten.

Abhängig von fork-Verhalten.

Code, der auf älterem Linux wegen fork läuft (untergeordnete Prozesse erben globale Variablen), kann auf macOS, Windows oder Python 3.14 fehlschlagen. Um ein konsistentes Verhalten zu haben, wähle die Start-Methode explizit:

import multiprocessing as mp

if __name__ == "__main__":
    mp.set_start_method("spawn")

Hinweis: Verwende multiprocessing nicht für I/O-bound-Aufgaben wie HTTP-Aufrufe oder Datenbankabfragen. Während I/O-Operationen geben Threads weiterhin den GIL frei, daher sind ThreadPoolExecutor oder asyncio viel leichter. Verwende auch nicht multiprocessing, wenn jede Aufgabe nur wenige Mikrosekunden dauert und nicht in Batches erfolgt, oder wenn die Berechnung bereits in einer GIL-freien C-Bibliothek stattfindet (NumPy, Hash-Funktionen auf großen Datenmengen, etc.). In diesen Fällen sind Threads oder Vektorisierung normalerweise ausreichend.

Zu viele Worker erstellen.

Die Anzahl der Worker größer als die Anzahl der CPU-Kerne zu setzen, macht CPU-bound-Code nicht schneller. Es verbraucht nur RAM, da jeder Prozess ein vollständiger Interpreter ist. Mit os.cpu_count() zu beginnen ist sinnvoll.

9. Fazit

Der GIL verhindert, dass Threads reinen Python-Code, der CPU-intensiv ist, beschleunigen können. Da jeder Prozess seinen eigenen GIL hat, ist die Aufteilung der Arbeit auf mehrere Prozesse eine einfache Möglichkeit, alle CPU-Kerne auszulasten. In den meisten Fällen ist ProcessPoolExecutor die kompakteste Wahl. Pool ist nützlich, wenn du imap_unordered oder starmap brauchst. Manuelle Process-Erstellung ist nur für spezielle Situationen.

Wenn du multiprocessing verwendest, solltest du diese Punkte beachten:

• Immer if __name__ == "__main__": verwenden.
• Nur pickelbare Funktionen und Daten weitergeben.
• chunksize erhöhen, wenn viele kleine Aufgaben vorhanden sind.
• initializer für schwere Ressourcen verwenden.
• Schreibe einen Test, der parallele Ergebnisse mit sequenziellen Ergebnissen vergleicht, um die korrekte Segmentierung zu gewährleisten.