Leihköche: concurrent.futures

Siebte Nacht, halb elf. Das Telefon klingelt, die Chefin legt den Hörer auf und sagt: „Firmenfeier, 30 Portionen.“

Minho bindet seine Schürze neu und fragt: „Soll ich 30 Threads anlegen?“ Die Chefin schüttelt den Kopf. „Wenn wir Leute einstellen, müssen wir uns um alles kümmern: Unterkunft, Lohn, und am Ende auch die Verabschiedung. Heute rufen wir bei einer Personalagentur an. Die kümmert sich um die Leute, wir werfen nur die Arbeit rüber.“

7-1. Arbeit an die Agentur abgeben: ThreadPoolExecutor (10 Min.)

concurrent.futures.ThreadPoolExecutor ist eine Personalagentur, die ein paar Threads in Bereitschaft hält und jede neue Aufgabe an einen freien Thread weitergibt. Sie nimmt dir ab, was du in Lektion 2 noch selbst gemacht hast: Thread anlegen, start, join und eine Liste für die Ergebnisse vorbereiten.

import time
from concurrent.futures import ThreadPoolExecutor

def cook(n):
    time.sleep(0.2)
    return f"#{n} ready"

with ThreadPoolExecutor(max_workers=3) as agency:   # drei Leihköche
    fut = agency.submit(cook, 1)        # wer Arbeit abgibt, bekommt eine „Abholmarke“ (Future)
    print(fut.done())                   # wird noch gekocht
    print(fut.result())                 # wartet, bis es fertig ist, und holt das Ergebnis ab
    print(fut.done())
# False
# #1 ready
# True

• submit(Funktion, Argumente...) gibt die Arbeit ab und liefert sofort ein Future zurück. Das ist nicht das Essen, sondern die „Abholmarke für das Essen“. Mit print(fut) siehst du den Zustand, etwa <Future at 0x... state=running>.
• fut.result() wartet, bis das Gericht fertig ist, und liefert dann den Rückgabewert der Funktion. Das Problem aus Lektion 2, „Threads geben keinen Rückgabewert zurück“, weshalb wir Ergebnisse in Listen gesteckt haben, ist damit gelöst.
• Beim Verlassen des with-Blocks wird shutdown(wait=True) aufgerufen: Es wird gewartet, bis alle abgegebenen Aufgaben fertig sind, und dann werden die Threads aufgeräumt.

Jetzt geben wir 30 Stück ab. Es gibt nur fünf Köche, also übernimmt jeder nacheinander sechs.

import threading

def cook(n):
    time.sleep(0.1)
    return threading.current_thread().name

start = time.perf_counter()
with ThreadPoolExecutor(max_workers=5, thread_name_prefix="agency") as agency:
    futures = [agency.submit(cook, n) for n in range(1, 31)]   # 30 Bestellungen
# nach dem with-Block sind alle Aufgaben erledigt
names = {f.result() for f in futures}
print(len(futures), sorted(names), f"{time.perf_counter() - start:.2f}s")
# 30 ['agency_0', 'agency_1', 'agency_2', 'agency_3', 'agency_4'] 0.60s

30 Bestellungen, aber nur fünf Threads wurden angelegt. 30 Bestellungen à 0,1 Sekunden, aufgeteilt auf fünf Leute: 0,6 Sekunden. Die Kosten fürs Anlegen und Abräumen von Threads fallen nur einmal an, und dieselben Köche nehmen immer wieder die nächste Bestellung. Das nennt man einen Thread-Pool.

7-2. Fertig-Reihenfolge gegen Eingabe-Reihenfolge: as_completed und map (10 Min.)

Es gibt zwei Wege, die Ergebnisse abzuholen. Diesmal sorgen wir dafür, dass Bestellung #1 am längsten dauert.

from concurrent.futures import as_completed

def cook(n):
    time.sleep(0.1 * (5 - n))     # #1 dauert am längsten (0,4 s … 0,1 s)
    return f"#{n}"

with ThreadPoolExecutor(max_workers=4) as agency:
    start = time.perf_counter()
    futures = [agency.submit(cook, n) for n in range(1, 5)]
    for f in as_completed(futures):                 # in der Reihenfolge, in der sie fertig werden
        print(f.result(), f"{time.perf_counter() - start:.1f}s", end="  ")
    print()
    start = time.perf_counter()
    for dish in agency.map(cook, range(1, 5)):      # in der Reihenfolge der Eingabe
        print(dish, f"{time.perf_counter() - start:.1f}s", end="  ")
    print()
# #4 0.1s  #3 0.2s  #2 0.3s  #1 0.4s
# #1 0.4s  #2 0.4s  #3 0.4s  #4 0.4s

Auch wenn #4 schon nach 0,1 Sekunden fertig ist, rückt map es erst bei 0,4 Sekunden heraus, wenn #1 fertig ist. Die Eingabe-Reihenfolge muss ja eingehalten werden. Die Gesamtzeit ist gleich, aber wann der erste Teller rausgeht, unterscheidet sich.

as_completed liefert nur Futures. Um zu wissen, zu welcher Bestellung ein Ergebnis gehört, ordnet man sie üblicherweise in einem dict zu.

with ThreadPoolExecutor(max_workers=4) as agency:
    table_of = {agency.submit(cook, n): f"T{n}" for n in range(1, 5)}   # Future → Tisch
    for fut in as_completed(table_of):
        print(table_of[fut], fut.result())
# T4 #4
# T3 #3
# T2 #2
# T1 #1

Auch die „Leihköche“ in der Zeitleiste unten laufen mit einem echten ThreadPoolExecutor. Dieses Szenario ist so gebaut, dass Bestellungen umso schneller fertig werden, je höher ihre Nummer ist.

• Beide Varianten dauern etwa 1,0 Sekunden, und die Balken sehen fast gleich aus. Die Köche arbeiten genauso wie vorher, nur die Reihenfolge, in der die Ergebnisse abgeholt werden, ist anders.
• Bei as_completed ist die Rückgabe-Reihenfolge durcheinander, zum Beispiel #3 → #2 → #1 → #6 → #5 → #4 → …. Von den Bestellungen, mit denen die drei gleichzeitig angefangen haben, wird die kürzeste zuerst fertig.
• Bei map ist sie immer #1 → #2 → … → #10.
• Erhöhst du auf das Maximum von 8 Köchen, ist alles nach etwa 0,4 Sekunden fertig, und die Rückgabe-Reihenfolge bei as_completed ist fast umgedreht, etwa #8 → #7 → … → #1. Die acht Bestellungen starten alle gleichzeitig, und die kurzen mit den hohen Nummern werden zuerst fertig.

7-3. Ein verbrannter Teller zeigt sich erst, wenn du die Marke einlöst (7 Min.)

Fliegt in einer abgegebenen Aufgabe eine Exception, wird sie still im Future aufbewahrt. In dem Moment, in dem du result() aufrufst, fliegt sie genau dort erneut.

def cook(n):
    if n == 3:
        raise ValueError(f"#{n} burnt")    # Bestellung 3 ist verbrannt
    return f"#{n}"

with ThreadPoolExecutor(max_workers=2) as agency:
    futures = [agency.submit(cook, n) for n in range(1, 5)]
# bis hierhin ist alles ruhig, als wäre nichts passiert

for f in futures:
    try:
        print(f.result())
    except ValueError as e:                # fliegt beim Aufruf von result()
        print("error:", e)
# #1
# #2
# error: #3 burnt
# #4

print(repr(futures[2].exception()), futures[0].exception())
# ValueError('#3 burnt') None

• Rufst du result() nie auf, verschwindet die Exception, ohne irgendwo ausgegeben zu werden. Als wäre ein verbrannter Teller zwar nicht zum Gast gegangen, aber niemand hätte es gemerkt.
• exception() holt die Exception heraus, ohne sie zu werfen. Bei Erfolg ist es None.
• map wirft die Exception genau an der Stelle, an der es das Ergebnis herausgibt, und alle Ergebnisse danach bekommst du nicht mehr. Wenn du trotz eines Fehlschlags alle übrigen Ergebnisse brauchst, nimm submit und as_completed.

Willst du sofort etwas tun, wenn eine Aufgabe fertig ist, hängst du mit add_done_callback eine Glocke dran. Du übergibst eine Funktion, die das fertige Future als einziges Argument bekommt.

def cook(n):
    time.sleep(0.1)
    return f"#{n}"

def ring_bell(fut):                      # wird aufgerufen, wenn es fertig ist (Callback); bekommt ein Future
    print("bell:", fut.result(), threading.current_thread().name)

with ThreadPoolExecutor(max_workers=2, thread_name_prefix="agency") as agency:
    for n in (1, 2):
        agency.submit(cook, n).add_done_callback(ring_bell)
# z. B.:
# bell: #1 agency_0
# bell: #2 agency_1

Der Callback wird im Thread des Leihkochs aufgerufen, der die Arbeit erledigt hat (hängst du ihn an ein schon fertiges Future, wird er sofort im Thread desjenigen aufgerufen, der ihn anhängt). Änderst du im Callback gemeinsame Daten, gelten also die Lock-Regeln aus Lektion 3 genauso. Exceptions im Callback landen nur im Log und werden ansonsten ignoriert.

7-4. Abbrechen, und nur auf eins warten (8 Min.)

cancel() kann nur Aufgaben abbrechen, die noch nicht angefangen haben. Was schon auf dem Herd steht, lässt sich nicht zurücknehmen.

with ThreadPoolExecutor(max_workers=1) as agency:     # ein Koch
    futures = [agency.submit(time.sleep, 0.2) for _ in range(3)]
    time.sleep(0.05)                                  # Zeit, damit die erste Bestellung auf den Herd kommt
    print([f.cancel() for f in futures])              # was schon läuft, lässt sich nicht zurücknehmen
    print([f.cancelled() for f in futures], [f.done() for f in futures])
# [False, True, True]
# [False, True, True] [False, True, True]

• cancel() liefert True, wenn das Abbrechen geklappt hat, und False, wenn die Aufgabe schon läuft oder fertig ist.
• done() ist True, wenn die Aufgabe fertig oder abgebrochen ist. Die erste Bestellung wird noch gekocht, also False.

Um den Vertrag mit der Agentur auf einen Schlag zu beenden, nimmst du shutdown(cancel_futures=True). Alles, was noch in der Warteschlange stand, wird abgebrochen, und nur auf die schon begonnenen Aufgaben wird bis zum Ende gewartet.

agency = ThreadPoolExecutor(max_workers=2)
futures = [agency.submit(time.sleep, 0.2) for _ in range(6)]
time.sleep(0.05)
agency.shutdown(cancel_futures=True)     # Wartende abbrechen, was auf dem Herd ist, fertig kochen
print(sum(f.cancelled() for f in futures), sum(not f.cancelled() for f in futures))
# 4 2

Willst du loslegen, sobald irgendeine von mehreren Aufgaben fertig ist, nimmst du wait.

from concurrent.futures import FIRST_COMPLETED, wait

def cook(sec):
    time.sleep(sec)
    return sec

with ThreadPoolExecutor(max_workers=3) as agency:
    futures = [agency.submit(cook, s) for s in (0.3, 0.1, 0.2)]
    done, pending = wait(futures, return_when=FIRST_COMPLETED)   # sobald eine fertig ist
    print([f.result() for f in done], len(pending))
    done, pending = wait(futures, timeout=0.15)                  # Standard ist ALL_COMPLETED
    print(len(done), len(pending))
# [0.1] 2
# 2 1

wait liefert zwei Mengen: die fertigen und die nicht fertigen Aufgaben. Für return_when gibt es FIRST_COMPLETED, FIRST_EXCEPTION und ALL_COMPLETED (Standard), und nach Ablauf von timeout kehrt es zurück, auch wenn noch nicht alles fertig ist. Es wirft dabei keine Exception, du musst also selbst prüfen, ob pending leer ist.

7-5. Gleicher Vertrag, andere Agentur (5 Min.)

Das Schöne an concurrent.futures: Der Vertrag ist immer derselbe. submit, map, Future und as_completed bleiben, wie sie sind; du tauschst nur den Namen der Agentur aus.

from concurrent.futures import InterpreterPoolExecutor, ProcessPoolExecutor, ThreadPoolExecutor

def knead(n):
    return sum(range(n))

if __name__ == "__main__":               # für den Prozess-Pool unbedingt nötig (Lektion 8)
    for Agency in (ThreadPoolExecutor, ProcessPoolExecutor, InterpreterPoolExecutor):
        with Agency(max_workers=2) as agency:
            print(Agency.__name__, list(agency.map(knead, [10, 100, 1000])))
# ThreadPoolExecutor [45, 4950, 499500]
# ProcessPoolExecutor [45, 4950, 499500]
# InterpreterPoolExecutor [45, 4950, 499500]

Speichere dieses Beispiel als .py-Datei und führ es aus, nicht in der interaktiven Shell. Warum der Prozess-Pool if __name__ == "__main__": verlangt, sehen wir in der nächsten Nacht.

Lässt du max_workers weg, holt der Thread-Pool min(32, os.process_cpu_count() + 4) Köche (bis 3.12 war es os.cpu_count()). Auf diesem PC mit 12 Kernen sind das 16. Für Arbeit mit Warten braucht man mehr Threads als Kerne, deshalb kommen 4 dazu, und damit es nicht zu viele werden, ist bei 32 Schluss. Im echten Betrieb legst du die Zahl besser selbst fest und denkst dabei daran, wie viel der Server auf der anderen Seite verkraftet.

Küchenschluss (10 Min.)

Das Ziel für heute Nacht war, 30 Bestellungen an sechs Leihköche abzugeben, in unter einer Sekunde jedes Gericht rauszuschicken, sobald es fertig ist, und keinen verbrannten Teller zu übersehen. Jede Bestellung braucht 0,1 bis 0,2 Sekunden, und Bestellung 13 brennt an.

def cook(n):
    time.sleep(0.1 + 0.05 * (n % 3))     # 0,1 bis 0,2 s
    if n == 13:
        raise ValueError(f"#{n} burnt")
    return f"#{n}"

served, burnt = [], []
start = time.perf_counter()
with ThreadPoolExecutor(max_workers=6, thread_name_prefix="agency") as agency:
    futures = {agency.submit(cook, n): n for n in range(1, 31)}
    for fut in as_completed(futures):     # rausschicken, sobald fertig
        try:
            served.append(fut.result())
        except ValueError:
            burnt.append(futures[fut])
print(len(served), burnt, f"{time.perf_counter() - start:.2f}s")
# 29 [13] 0.80s  (Zeit nur ungefähr)

29 Teller gingen raus, sobald sie fertig waren, und die verbrannte Nummer 13 kam auf die Liste fürs Nachkochen. 0,8 Sekunden. Kein Code, der Threads anlegt, und keiner, der vorab eine Liste für die Ergebnisse vorbereitet.

Die Frau am Fenster sitzt auch heute allein vor einer Schüssel Nudelsuppe. Während die Teller für die Gruppe nur so hinausfliegen, wirft sie einen Blick auf ihre Armbanduhr und notiert kurz etwas in ihr kleines Notizbuch.

Die Chefin schickt die Leihköche nach Hause und sagt: „Morgen haben wir Berge von Teig. Wenn ich vier Leihköche hole und ihnen den Teig gebe, geht es dann viermal so schnell?“

Übungen

1. Ersetze im Code vom Küchenschluss as_completed durch agency.map(cook, range(1, 31)). Was passiert bei Nummer 13, und wie viele Teller bekommst du?
2. Stell in der Zeitleiste nur 1 Koch ein (dauert etwa 2,8 Sekunden) und vergleich as_completed mit map. Warum ist die Rückgabe-Reihenfolge jetzt gleich?
3. Schreib mit wait(..., return_when=FIRST_COMPLETED) und shutdown(cancel_futures=True) Code, der von 30 Bestellungen nur den ersten fertigen Teller abholt und den Rest abbricht. Wie viele Gerichte waren schon angefangen?