Mehr Köche einstellen: Grundlagen von threading

Zweite Nacht, halb zehn. Die drei Postenköche binden ihre Schürzen. Minho hat schon das Messer in der Hand, Sua sortiert die Bons der Reihe nach, und Herr Tak wischt in aller Ruhe die Herdplatten ab.

Um zehn kommen die Bestellungen, und jeder setzt seinen Topf auf. Doch eine halbe Stunde später fragt die Chefin: „Wer hat die Udon für Tisch 3 gemacht?“, und niemand kann antworten.

Die Chefin tippt auf den Timer an ihrem Hals. „Eingestellt haben wir die Köche. Heute lernen wir, wie man ihnen Namensschilder gibt, Arbeit zuteilt und wartet, bis sie fertig sind.“

2-1. Einen Koch reinholen: Thread, start, join (10 Min.)

In Python ist ein Koch ein Thread. Alle Threads teilen sich dieselbe Küche (den Speicher), aber jeder erledigt seine eigene Arbeit. Threads erzeugst du mit threading.Thread.

import threading
import time

def cook(name, dish):
    print(f"{name}: start {dish}")
    time.sleep(0.3)          # warten, bis die Suppe kocht
    print(f"{name}: done {dish}")

minho = threading.Thread(target=cook, args=("minho", "ramen"))
minho.start()                # Minho fängt an
print("chef: next order")    # die Chefin wartet nicht, sondern macht weiter
minho.join()                 # warten, bis Minho fertig ist
print("chef: served")
# minho: start ramen
# chef: next order
# minho: done ramen
# chef: served

• target= bekommt die Arbeit (die Funktion), die der Thread erledigen soll, und args= die Argumente für diese Funktion, und zwar als Tupel. Du rufst die Funktion nicht auf, sondern übergibst sie selbst, also cook und nicht cook().
• Ein Thread(...) anzulegen bewirkt noch gar nichts. Der Koch ist nur eingestellt. Erst wenn du start() aufrufst, läuft cook in einem neuen Thread los.
• start() kehrt sofort zurück. Während Minho also Ramen kocht, geht die Chefin (der Haupt-Thread, der das Programm gestartet hat) schon zur nächsten Zeile weiter.
• join() wartet, bis dieser Thread fertig ist. Sozusagen: „Am Pass warten, bis Minho den Teller herüberreicht.“

Bekommt jeder der drei eine Bestellung, sieht das so aus:

def cook(name, dish):
    time.sleep(0.3)
    print(f"{name}: done {dish}")

crew = [("minho", "ramen"), ("sua", "udon"), ("tak", "soba")]
threads = [threading.Thread(target=cook, args=c) for c in crew]

start = time.perf_counter()
for th in threads:
    th.start()        # erst alle drei losschicken
for th in threads:
    th.join()         # dann auf alle warten
print(f"{time.perf_counter() - start:.2f}s")
# zum Beispiel:
# sua: done udon
# minho: done ramen
# tak: done soba
# 0.30s

In welcher Reihenfolge die drei fertig werden, ändert sich bei jedem Lauf. Alle drei sind nach 0,3 Sekunden fast gleichzeitig fertig, und es hängt vom Betriebssystem ab, wen es zuerst aufweckt. In nebenläufigem Code ist die „Ausgabereihenfolge“ kein Versprechen.

start() und join() gehören nicht in dieselbe Schleife. Wechselst du th.start(); th.join() ab, startet der Nächste erst, wenn der Vorige fertig ist, und am Ende dauert es 0,9 Sekunden.

Du kannst dem Warten auch eine Grenze setzen. join(timeout=sekunden) wartet nur so lange und kehrt dann zurück; ob der Thread noch arbeitet, prüfst du mit is_alive().

slow = threading.Thread(target=time.sleep, args=(1.0,))
slow.start()
slow.join(timeout=0.2)
print(slow.is_alive())   # True: kocht noch
slow.join()
print(slow.is_alive())   # False

• Mit 1 Koch dauert es genauso lange wie „Allein, eins nach dem anderen“, etwa 1,8 Sekunden. Mit nur einem Thread wird nichts schneller.
• Mit 2 Köchen sind es etwa 0,9 Sekunden, mit 3 etwa 0,6 und mit 6 etwa 0,3 Sekunden.
• 4 und 5 Köche brauchen beide etwa 0,6 Sekunden. Verteilt man 6 Bestellungen auf vier Leute, muss einer zwei machen. Die Dauer bestimmt immer derjenige, der am meisten übernimmt.
• Eine Reihe ist ein Thread. Die Köche in dieser Zeitleiste legen genau wie im Code oben einen Thread an und rufen start() und join() auf.

2-2. Wer run() aufruft, ruft nur eine Funktion auf (5 Min.)

Thread hat auch eine Methode namens run(). Dem Namen nach heißt sie „ausführen“, aber wer sie direkt aufruft, tappt in eine Falle.

def cook(name, dish):
    time.sleep(0.3)
    print(f"{threading.current_thread().name}: done {dish}")

threads = [threading.Thread(target=cook, args=c) for c in crew]

start = time.perf_counter()
for th in threads:
    th.run()          # run() statt start()
print(f"{time.perf_counter() - start:.2f}s")
# MainThread: done ramen
# MainThread: done udon
# MainThread: done soba
# 0.90s

• run() ist eine ganz normale Methode, die target aufruft. Sie erzeugt keinen neuen Thread, sondern führt alles der Reihe nach im aktuellen Thread aus. Deshalb hat jedes Mal MainThread gekocht, und es dauert 0,3 × 3 = 0,9 Sekunden.
• Einen neuen Thread startet start(). start() erzeugt den neuen Thread und ruft darin run() auf.

Ein Thread kann nur einmal starten. Rufst du start() zweimal für denselben Thread auf, gibt es einen Fehler. Für dieselbe Arbeit noch einmal legst du einen neuen Thread an.

th = threading.Thread(target=time.sleep, args=(0.1,))
th.start()
th.start()
# RuntimeError: threads can only be started once

2-3. Namensschilder: name, current_thread, enumerate (7 Min.)

Das Problem heute Nacht war: „Keiner weiß, wer es war.“ Threads kann man ein Namensschild geben.

def cook(dish):
    me = threading.current_thread()       # der Thread, der diesen Code gerade ausführt
    print(me.name, dish, threading.get_native_id())
    time.sleep(0.2)

a = threading.Thread(target=cook, args=("ramen",), name="minho")
b = threading.Thread(target=cook, args=("udon",))      # und ohne Namen?
a.start(); b.start()
print([t.name for t in threading.enumerate()])
print(threading.active_count())
a.join(); b.join()
print(threading.current_thread() is threading.main_thread())
print([t.name for t in threading.enumerate()])
# zum Beispiel:
# minho ramen 25680
# Thread-1 (cook) udon 38072
# ['MainThread', 'minho', 'Thread-1 (cook)']
# 3
# True
# ['MainThread']

• Dass b ohne Namen Thread-1 heißt und nicht Thread-2, liegt daran, dass nur Threads ohne Namen durchnummeriert werden.
• Die Zahl von get_native_id() ist bei jedem Lauf anders. Die Nummer, die Python intern verwendet, ist th.ident (oder threading.get_ident()); beide stehen erst fest, wenn der Thread gestartet ist.
• Ein Thread, der seine Arbeit erledigt hat, verschwindet aus enumerate(). Wer Feierabend hat, steht nicht mehr auf der Liste.

2-4. Threads geben den Teller nicht zurück (6 Min.)

Eine Funktion liefert ihr Ergebnis mit return, aber ein Thread nimmt diesen Wert nicht entgegen. Der Rückgabewert der target-Funktion wird einfach verworfen, und auch join() liefert nichts zurück.

def make(dish):
    time.sleep(0.1)
    return f"{dish} ok"

th = threading.Thread(target=make, args=("ramen",))
th.start()
print(th.join())     # join liefert nichts zurück
# None

Also stellst du vorher ein Gefäß für die Ergebnisse bereit, in das der Thread sie legt. Am einfachsten bekommt jeder Koch sein eigenes Fach.

dishes = ["ramen", "udon", "soba"]
results = [None] * len(dishes)          # die Plätze vorher anlegen

def cook(i, dish):
    results[i] = make(dish)             # nur ins eigene Fach schreiben

threads = [threading.Thread(target=cook, args=(i, d)) for i, d in enumerate(dishes)]
for th in threads:
    th.start()
for th in threads:
    th.join()
print(results)
# ['ramen ok', 'udon ok', 'soba ok']

Willst du nach Namen nachschlagen, nimm ein dict.

done = {}

def cook2(name, dish):
    done[name] = make(dish)

ts = [threading.Thread(target=cook2, args=(n, d)) for n, d in [("minho", "ramen"), ("sua", "udon")]]
for t in ts:
    t.start()
for t in ts:
    t.join()
print(sorted(done.items()))
# [('minho', 'ramen ok'), ('sua', 'udon ok')]

• Solange jeder in ein anderes Fach oder unter einen anderen Schlüssel schreibt, kommen sich die Köche nicht in die Quere. Sobald sie den Wert im selben Fach lesen und überschreiben, sieht die Sache anders aus, und genau das passiert morgen Nacht.
• Lies die Ergebnisse immer erst, wenn join() durch ist. Vorher kann noch ein Fach auf None stehen.
• Wenn dir das Bereitstellen der Gefäße zu umständlich ist: Der Future, den ThreadPoolExecutor aus Lektion 7 zurückgibt, nimmt dir das ab. Mit future.result() bekommst du den Teller direkt.

2-5. daemon=True: Threads, auf die beim Lichtausmachen niemand wartet (6 Min.)

Ein Python-Programm beendet sich nicht schon, wenn der Haupt-Thread fertig ist, sondern wartet, bis alle Nicht-daemon-Threads fertig sind. Normal angelegte Threads sind alle keine daemon-Threads.

def clean():
    for i in range(1, 6):
        print("cleaning", i)
        time.sleep(0.1)

th = threading.Thread(target=clean)      # daemon=False (Standard)
th.start()
time.sleep(0.25)
print("lights off")
# cleaning 1
# cleaning 2
# cleaning 3
# lights off
# cleaning 4
# cleaning 5

Auch wenn die Chefin „Licht aus!“ ruft, wird die Tür erst abgeschlossen, wenn der Koch mit dem Putzen fertig ist. Hängen wir an denselben Code nur daemon=True an:

th = threading.Thread(target=clean, daemon=True)
th.start()
time.sleep(0.25)
print("lights off", th.daemon)
# cleaning 1
# cleaning 2
# cleaning 3
# lights off True

• Auf daemon-Threads wird am Programmende nicht gewartet. Mitten im Putzen geht das Licht aus, und Nummer 4 und 5 werden nie erledigt.
• Willst du trotzdem warten, ruf auch beim daemon-Thread join() auf. join() wartet unabhängig davon, ob ein Thread daemon ist.
• daemon nimmst du nur für Arbeit, die ruhig mittendrin abbrechen darf, etwa einen Überwachungs-Thread, der regelmäßig den Status ausgibt. Gibst du Arbeit, die zu Ende gebracht werden muss, wie Dateien schreiben oder Bestellungen protokollieren, an einen daemon-Thread, bleiben halb geschriebene Dateien zurück.
• Ob ein Thread daemon ist, legst du vor start() fest. Gibst du nichts an, erbt er den Wert des Threads, der ihn erzeugt hat; der Haupt-Thread ist kein daemon, also ist es meistens False.

2-6. Weitere Werkzeuge: von Thread erben, Timer, local (6 Min.)

Von Thread erben und run ersetzen. Das bietet sich an, wenn ein Koch seinen eigenen Zustand (zugeteiltes Gericht, Ergebnis) mit sich tragen soll. Wie in 2-2 gesehen, ruft start() in einem neuen Thread run() auf, also musst du nur run() neu schreiben.

class Cook(threading.Thread):
    def __init__(self, name, dish):
        super().__init__(name=name)     # unbedingt zuerst das __init__ der Elternklasse
        self.dish = dish
        self.result = None

    def run(self):                      # start() ruft das in einem neuen Thread auf
        time.sleep(0.1)
        self.result = f"{self.dish} ok"

tak = Cook("tak", "soba")
tak.start()
tak.join()
print(tak.name, tak.result)
# tak soba ok

threading.Timer: einmal nach n Sekunden. Wie der Nudeltimer: ein Thread, der nach einer festgelegten Zeit einmal eine Funktion aufruft. Solange er noch nicht geklingelt hat, kannst du ihn mit cancel() abstellen.

start = time.perf_counter()

def ring(msg):
    print(f"{time.perf_counter() - start:.1f}s", msg)

noodles = threading.Timer(0.5, ring, args=("noodles up",))
noodles.start()                     # klingelt einmal nach 0,5 Sekunden
eggs = threading.Timer(1.0, ring, args=("eggs up",))
eggs.start()
eggs.cancel()                       # vorher abgestellt, passiert gar nichts
noodles.join()
eggs.join()
print("end")
# 0.5s noodles up
# end

threading.local(): die eigene Schürzentasche für jeden Koch. Eine Variable mit demselben Namen, die aber in jedem Thread einen anderen Wert enthält. Welche Tasche es ist, hängt davon ab, wer hineingreift.

pocket = threading.local()

def cook(name, n):
    pocket.name = name
    pocket.count = 0
    for _ in range(n):
        pocket.count += 1
        time.sleep(0.01)
    print(pocket.name, pocket.count)

ts = [threading.Thread(target=cook, args=a) for a in [("minho", 3), ("sua", 5)]]
for t in ts:
    t.start()
for t in ts:
    t.join()
print(hasattr(pocket, "name"))      # die Tasche der Chefin (Haupt-Thread) ist leer
# minho 3
# sua 5
# False

Obwohl pocket eine einzige globale Variable ist, vermischen sich Minhos count und Suas count nicht. So gibt man häufig jedem Thread seine eigene Datenbankverbindung.

Küchenschluss (10 Min.)

Das Ziel für heute Nacht war, sechs Bestellungen auf drei Leute zu verteilen, in unter 0,6 Sekunden fertig zu sein und festzuhalten, wer nach wie vielen Sekunden was fertig hatte. Wir geben allen ein Namensschild (name=) und lassen jeden nach jedem fertigen Gericht eine Zeile in ein gemeinsames Logbuch schreiben.

start = time.perf_counter()
log = []                                # (Fertigzeit, Koch, Bestellung)

def cook(dishes):
    me = threading.current_thread().name
    for dish in dishes:
        time.sleep(0.3)
        log.append((round(time.perf_counter() - start, 1), me, dish))

plan = {"minho": ["#1", "#4"], "sua": ["#2", "#5"], "tak": ["#3", "#6"]}
threads = [threading.Thread(target=cook, args=(d,), name=n) for n, d in plan.items()]
for th in threads:
    th.start()
for th in threads:
    th.join()

for row in sorted(log):
    print(*row)
print(f"total {time.perf_counter() - start:.2f}s, {len(log)} orders")
# 0.3 minho #1
# 0.3 sua #2
# 0.3 tak #3
# 0.6 minho #4
# 0.6 sua #5
# 0.6 tak #6
# total 0.60s, 6 orders

• Das Argument ist eine einzelne Liste, deshalb steht in args=(d,) ein Komma.
• log.append(...) ist eine Listenoperation, die in einem Schritt erledigt ist; auch wenn mehrere Threads sie gleichzeitig aufrufen, geht keine Zeile verloren. Gelesen wird das Logbuch nach join(), sortiert.

Die Chefin überfliegt das Logbuch und nickt. „Gut, jetzt wissen wir, wer was gemacht hat. Dann probieren wir morgen noch eine Sache: Wenn vier Leute jeweils ihre Stückzahl auf eine einzige Zahl an der Tafel addieren, stimmt diese Zahl dann?“

Übungen

1. Ändere das Beispiel mit den drei Köchen aus 2-1 so, dass start() und join() in derselben Schleife stehen. Schätze vorher, wie viele Sekunden es dauert, und miss dann nach.
2. Ändere den Code vom Küchenschluss so, dass nur Herr Tak 0,5 Sekunden pro Bestellung braucht (Tipp: die sleep-Zeit je nach Name des Kochs unterschiedlich setzen). Wer bestimmt jetzt die Gesamtzeit? Vergleiche das mit der Zeitleiste, in der nur eine einzige Reihe länger wird.
3. Baue mit threading.Timer einen Küchentimer, der „dreimal, alle 0,3 Sekunden“ klingelt. Ein Timer klingelt nur einmal, also musst du beim Klingeln den nächsten Timer neu starten.