Zum Inhalt
wannacode

Thread-Pool und Aufgaben

Kurz gesagt: Ein Thread-Pool sind ein paar Worker vor einer Aufgabenwarteschlange statt eines neuen Threads pro Aufgabe; in Python ist das ThreadPoolExecutor, und ProcessPoolExecutor zum Rechnen.

Problem

Ein Thread für jedes Bild

Der Shop startet einen Sale. Verkäufer laden Bilder ihrer Produkte hoch, und jedes Bild braucht ein Vorschaubild für die Produktliste. Ein Vorschaubild zu erzeugen ist Arbeit, die rechnet: Sie geht durch jeden Pixel (siehe Arbeit, die wartet, und Arbeit, die rechnet).

Die erste Version startet für jedes hochgeladene Bild einen Thread:

threads = []
for image in uploads:
    thread = threading.Thread(target=save_thumbnail, args=(image,))
    thread.start()
    threads.append(thread)

Mit 20 Bildern funktioniert alles auf deinem Rechner: „Fertig: 20 Vorschaubilder“. In der ersten Stunde des Sales kommen 10.000 auf einmal.

  • Jeder Thread braucht einen eigenen Stack und einen Platz im Betriebssystem. Tausende Threads fressen den Speicher, und nahe der Grenze des Systems scheitert start() mit RuntimeError: can't start new thread.
  • Selbst wenn Speicher da ist, rechnet wegen des GIL immer nur ein Thread. Die anderen 9.999 warten und wechseln sich ab.

Der Server wird nicht langsamer, er fällt um. Und ein Server, der down ist, bedient auch die Kunden nicht.

Die Idee

Ein paar Worker und eine Warteschlange

Ein Thread-Pool sind ein paar Worker, einmal erzeugt, und davor eine Warteschlange von Aufgaben. Eine Aufgabe ist „verkleinere dieses Bild“. Ein freier Worker nimmt die nächste Aufgabe aus der Warteschlange, erledigt sie und kommt für eine neue zurück.

  • Die Zahl der Worker ist konstant. 10.000 Bilder ergeben nicht 10.000 Threads, sondern 10.000 Aufgaben in der Warteschlange. Eine Aufgabe ist ein kleines Objekt; ein Thread ist teuer.
  • Die Worker werden wiederverwendet. Für jedes Bild wird kein Thread erzeugt und zerstört.
  • Die Warteschlange fängt die Spitze ab. Kommen Uploads schneller, als die Worker sie bearbeiten, warten die Aufgaben, statt den Server zu Fall zu bringen.

Das ist Producer und Consumer: Der Code, der Uploads annimmt, legt Aufgaben in die Warteschlange, und die Worker sind die Consumer.

Den Pool hast du schon in der Lektion über Threads kennengelernt: ThreadPoolExecutor, submit und Future.result(). Hier sehen wir, was darin passiert, wie man die Ergebnisse ordnet, was mit Fehlern geschieht und wie viele Worker du brauchst.

Sieh es dir an

Drei Worker

Ein Vorschaubild zu erzeugen ist hier eine Pause, 2 ms für jedes KB, und die Größe kommt aus dem Namen des Bildes. So ist die Ausgabe bei jedem Lauf gleich:

# Vorschaubild erzeugen: Hier ist es eine Pause, und die Größe kommt aus dem Namen.
def thumbnail(image):
    if image.startswith("defekt"):
        raise ValueError("defekte Datei")
    kb = 20 + 3 * len(image)
    time.sleep(2 * kb / 1000)
    return kb

Acht Bilder, ein Pool aus drei Workern:

with ThreadPoolExecutor(max_workers=3) as pool:
    sizes = list(pool.map(thumbnail, images))
total = 0
for image, kb in zip(images, sizes):
    print(f"{image}: {kb} KB")
    total += kb
print(f"Gesamt: {total} KB")
Schritt für Schritt, in Worten
  1. 8 Bilder warten. Der Pool besteht aus 3 Threads: ThreadPoolExecutor(max_workers=3).
  2. map macht aus jedem Bild eine Aufgabe in der Warteschlange. Die drei Threads nehmen die ersten drei.
  3. Thread 3 wird als Erstes fertig: Hut ist das dritte Bild, und sein Ergebnis kommt an dritter Stelle. Der Worker hält nicht an: Er nimmt die nächste Aufgabe.
  4. Die Warteschlange ist leer. Thread 1 ist fertig und hat nichts mehr zu nehmen, also wartet er. Die anderen beenden ihre letzten Bilder.
  5. map liefert die Größen in der Reihenfolge der Bilder, nicht in der Reihenfolge des Fertigwerdens. Gesamt 379 KB; das Ende von with schließt den Pool.
images = ["Kleid.jpg", "Mantel.jpg", "Hut.jpg", "Jacke.jpg", "Rucksack.jpg", "Gürtel.jpg", "Mütze.jpg", "Uhr.jpg"]
with ThreadPoolExecutor(max_workers=3) as pool:
    sizes = list(pool.map(thumbnail, images))
total = 0
for image, kb in zip(images, sizes):
    print(f"{image}: {kb} KB")
    total += kb
print(f"Gesamt: {total} KB")
Ausgabe
Kleid.jpg: 47 KB
Mantel.jpg: 50 KB
Hut.jpg: 41 KB
Jacke.jpg: 47 KB
Rucksack.jpg: 56 KB
Gürtel.jpg: 50 KB
Mütze.jpg: 47 KB
Uhr.jpg: 41 KB
Gesamt: 379 KB
  • ThreadPoolExecutor(max_workers=3) erzeugt drei Threads und davor eine Warteschlange.
  • map macht für jedes Bild eine Aufgabe und liefert die Ergebnisse in der Reihenfolge der Bilder, egal, was zuerst fertig wurde. Wartest du nicht auf alle, nimmst du sie einzeln aus map: Es wartet genau auf das nächste.
  • with schließt den Pool am Ende: Er nimmt keine neuen Aufgaben an, wartet auf die gestarteten und stoppt die Threads.

Threads helfen hier, weil ein Vorschaubild zu erzeugen eine Pause ist. Ein echtes Vorschaubild rechnet, und wegen des GIL rechnen Threads nicht gleichzeitig. Dafür gibt es Prozesse, etwas weiter unten.

Genauer, für Neugierige

Ohne max_workers hat der Pool min(32, os.cpu_count() + 4) Threads. Seine Warteschlange ist endlos: Kommen Uploads immer schneller, wächst sie, bis der Speicher ausgeht.

Wie es in den anderen Sprachen ist: In Java ist der Pool ein ExecutorService (dort rechnen Threads wirklich parallel), in C# gibt es einen Pool für das ganze Programm, und Node hat keinen fertigen Pool von Workern.

Probier es aus

Finde den Fehler

Ein Kollege hat die Vorschaubilder mit einem Pool umgeschrieben, und der Sale wurde nicht schneller. Sechs Bilder brauchen so lange wie ohne Pool:

start = time.monotonic()
sizes = []
with ThreadPoolExecutor(max_workers=3) as pool:
    for image in images:
        size = pool.submit(thumbnail, image)
        sizes.append(size.result())
print(f"Dauerte über 0,5 s: {time.monotonic() - start > 0.5}")
Ausgabe
Dauerte über 0,5 s: True

Wähle die Zeile mit dem Fehler.

Die Idee

Ergebnisse in Eingabereihenfolge

Aufgaben werden in einer Reihenfolge fertig, die du nicht kennst: Sie hängt von den Bildern ab, von der Last, davon, welcher Worker frei war. Gib also nie „in der Reihenfolge des Fertigwerdens“ aus, wenn du jedes Mal dieselbe Ausgabe willst.

Zwei Wege liefern bei jedem Lauf dasselbe Ergebnis:

  • In Eingabereihenfolge. Du hältst die Promises in einer Liste in der Reihenfolge der Aufgaben und wartest in dieser Reihenfolge darauf. So funktioniert der Code oben.
  • Eine Summe oder eine Anzahl. Brauchst du nur das Ergebnis insgesamt, ist die Reihenfolge egal: 47 + 50 ist dasselbe wie 50 + 47.

Manchmal willst du die Ergebnisse in der Reihenfolge des Fertigwerdens: für einen Balken „3 von 8 fertig“ oder um die ersten fertigen Bilder zu zeigen. as_completed liefert sie so. Dann sammelst du nur Dinge, die nicht von der Reihenfolge abhängen:

count = 0
total = 0
with ThreadPoolExecutor(max_workers=3) as pool:
    futures = [pool.submit(thumbnail, image) for image in images]
    for future in as_completed(futures):
        total += future.result()
        count += 1
print(f"Fertig: {count} von {len(images)}, gesamt: {total} KB")
Ausgabe
Fertig: 8 von 8, gesamt: 379 KB
as_completed liefert jedes Future, sobald es fertig ist. result() darauf wartet nicht.

Die Idee

Ein Fehler in einer Aufgabe

Eines der Bilder ist defekt, und thumbnail wirft eine Exception. Sie kommt nicht in dem Thread heraus, der die Aufgabe gestartet hat: Der ist längst weitergezogen. Der Pool fängt sie ab und bewahrt sie im Promise des Ergebnisses auf. Sie kommt heraus, wenn du auf dieses Ergebnis wartest:

images = ["Kleid.jpg", "Mantel.jpg", "defekt.jpg", "Hut.jpg"]
with ThreadPoolExecutor(max_workers=3) as pool:
    futures = [pool.submit(thumbnail, image) for image in images]
for image, future in zip(images, futures):
    try:
        print(f"{image}: {future.result()} KB")
    except ValueError as error:
        print(f"{image}: Fehler: {error}")
Ausgabe
Kleid.jpg: 47 KB
Mantel.jpg: 50 KB
defekt.jpg: Fehler: defekte Datei
Hut.jpg: 41 KB
result() löst dieselbe Exception erneut aus (ValueError). Den anderen Bildern passiert nichts. Bei pool.map kommt der Fehler heraus, wenn die Schleife beim defekten Bild ankommt, und die Ergebnisse danach gehen verloren. Erwartest du Fehler, nimm also submit und result() für jedes einzeln.

Probier es aus

Wohin ist der Fehler gegangen

Das zweite Bild ist defekt. Was gibt das Programm aus?

images = ["Kleid.jpg", "defekt.jpg", "Hut.jpg"]
with ThreadPoolExecutor(max_workers=3) as pool:
    for image in images:
        pool.submit(thumbnail, image)
print(f"Fertig: {len(images)} Bilder")
Antwort

Die Idee

Wie viele Worker

Das hängt davon ab, was sie tun:

  • Arbeit, die rechnet (Vorschaubilder, PDFs, Komprimierung): so viele wie Kerne. Ein neunter Worker auf 8 Kernen hat keinen Platz zum Laufen; er wechselt sich nur mit den anderen ab.
  • Arbeit, die wartet (E-Mail, Zahlung, Datenbank): mehr. Während ein Worker auf das Netzwerk wartet, ist der Kern für einen anderen frei. Wartet eine Aufgabe W ms und rechnet C ms, passen etwa 1 + W / C Worker auf einen Kern.
def pool_size(cores, wait_ms, compute_ms):
    return cores * (1 + wait_ms // compute_ms)




print(f"Vorschaubilder (nur Rechnen): {pool_size(8, 0, 50)}")
print(f"E-Mails (90 ms Warten, 10 ms Rechnen): {pool_size(8, 90, 10)}")
Ausgabe
Vorschaubilder (nur Rechnen): 8
E-Mails (90 ms Warten, 10 ms Rechnen): 80

Die Zahl der Kerne kommt aus os.cpu_count(). Die Formel ist ein Anfang, kein Gesetz: Miss nach. Gehen die E-Mails über einen Server mit einem Limit von 20 Verbindungen, warten 80 Worker nur in einer Schlange davor.

Python hat einen Haken: Wegen des GIL rechnen Threads nicht gleichzeitig. Für Arbeit, die rechnet, besteht der Pool aus Prozessen: ProcessPoolExecutor, jeder Prozess mit eigenem Python und eigenem GIL. Aufgaben und Ergebnisse wandern als Kopien zwischen den Prozessen:

from concurrent.futures import ProcessPoolExecutor




def checksum(invoice):
    return sum((invoice * 31 + i) % 256 for i in range(200_000))




if __name__ == "__main__":
    invoices = [1001, 1002, 1003, 1004]
    with ProcessPoolExecutor(max_workers=4) as pool:
        sums = list(pool.map(checksum, invoices))
    for invoice, total in zip(invoices, sums):
        print(f"Rechnung {invoice}: {total}")
Ausgabe
Rechnung 1001: 25497376
Rechnung 1002: 25499360
Rechnung 1003: 25501344
Rechnung 1004: 25503328

if __name__ == "__main__": verhindert, dass der Hauptcode in jedem Prozess erneut läuft: Auf macOS und Windows beginnt ein Prozess damit, die Datei von vorn zu laden.

Probier es aus

Ergänze

Wie viele Worker für eine Aufgabe, die waitMs wartet und computeMs rechnet?

def pool_size(cores, wait_ms, compute_ms):
    return cores * (1 + )
Genauer, für Neugierige

Für Arbeit, die wartet, hat Python auch asyncio: Ein Thread wartet auf Tausende Anfragen (die Lektion async ist kein Thread). Der Thread-Pool bleibt für Bibliotheken, die warten, ohne dass man sie mit await aufrufen kann.

Übung

Die Rechnungen des Sales

Löse die Aufgabe in deinem Editor.

Nach dem Sale braucht jede Bestellung eine Rechnung. invoice(amount) erzeugt das PDF (hier eine Pause von 50 ms) und gibt den Betrag mit 20 % MwSt. zurück. Eine Bestellung mit einem Betrag von 0 oder weniger ist ungültig, und die Funktion wirft eine Exception mit dem Text „ungültiger Betrag“.

Die Bestellungen sind 1001 bis 1008 mit den Beträgen 50 €, 125 €, 80 €, 0 €, 240 €, 35 €, 90 € und 15 €. Erzeuge die Rechnungen in einem Pool aus 4 Workern. Gib sie in der Reihenfolge der Bestellungen aus, die ungültige mit dem Text des Fehlers. Am Ende: die Summe der guten Rechnungen und wie viele gescheitert sind.

Ausgabe
Bestellung 1001: 60 €
Bestellung 1002: 150 €
Bestellung 1003: 96 €
Bestellung 1004: Fehler: ungültiger Betrag
Bestellung 1005: 288 €
Bestellung 1006: 42 €
Bestellung 1007: 108 €
Bestellung 1008: 18 €
Gesamt: 762 €, mit Fehler: 1

Weitere Aufgaben

  • Eine begrenzte Warteschlange. Lass den Pool höchstens 10 Aufgaben in der Warteschlange warten. Was passiert mit der elften? (threading.BoundedSemaphore um submit)
  • Fortschritt. Gib nach jeder fertigen Rechnung „3 von 8 fertig“ aus, und am Ende noch einmal die geordnete Liste.
  • Ein Timeout. Eine Rechnung, die über 1 Sekunde dauert, gilt als gescheitert, ohne die anderen zu stoppen.

Test

Teste dich

1. Warum ein Pool aus ein paar Threads und nicht ein Thread für jedes hochgeladene Bild?
2. Ein Pool aus 3 Workern verkleinert 8 Bilder. Das dritte wird als Erstes fertig. Wo steht seine Größe, wenn du sie in Listenreihenfolge mit get, result, Promise.all oder sizes[i] liest?
3. Eine Aufgabe im Pool wirft eine Exception. Wann siehst du sie?
4. Der Server hat 8 Kerne. Jede E-Mail wartet 90 ms auf das Netzwerk und rechnet 10 ms. Etwa wie viele Worker halten die Kerne beschäftigt?
5. Wie viele Worker für Vorschaubilder, wenn die Arbeit nur rechnet und der Server 8 Kerne hat?

Erstellt von Ivo Gergov im Rahmen der Kampagne von Evo Tech für zugängliche Programmierausbildung ·