Nur drei Herdplatten: Synchronisationswerkzeuge

Vierte Nacht, zwanzig vor zehn. Wegen einer Gruppenreservierung sind für einen Tag drei zusätzliche Köche gekommen. Sechs Köche, drei Herdplatten.

Minho stellt sich als Erster an eine Platte und dreht das Feuer auf. „Wenn wir vorheizen, geht's schneller.“ Die Chefin dreht wortlos das Gasventil zu. „Vor dem Öffnen wird kein Feuer gemacht. Und wenn sich sechs Leute vor den Platten drängen, kann sich keiner mehr bewegen.“

Sua hebt die Hand. „Und wenn wir an die Herdplatten nur drei Nummernmarken hängen? Nur wer eine Marke hat, darf ans Feuer.“ Zum ersten Mal lächelt die Chefin. „Das ist das erste Werkzeug heute Nacht.“

4-1. Drei Herdplatten: Semaphore (12 Min.)

Der Lock aus Lektion 3 war ein Stück Kreide, also immer nur eine Person. Ein Semaphore(n) sind n Nummernmarken. Bis zu n Leute gleichzeitig dürfen rein, die Person Nummer n+1 wartet, bis jemand eine Marke zurückgibt.

import threading
import time

stoves = threading.Semaphore(3)         # drei Herdplatten
start = time.perf_counter()

def cook(name):
    with stoves:                        # warten, bis eine Platte frei ist, dann zugreifen
        print(f"{time.perf_counter() - start:.1f}s {name} on")
        time.sleep(0.3)

names = [f"cook{i}" for i in range(1, 7)]
threads = [threading.Thread(target=cook, args=(n,)) for n in names]
for th in threads:
    th.start()
for th in threads:
    th.join()
print(f"total {time.perf_counter() - start:.2f}s")
# zum Beispiel:
# 0.0s cook1 on
# 0.0s cook2 on
# 0.0s cook3 on
# 0.3s cook4 on
# 0.3s cook5 on
# 0.3s cook6 on
# total 0.60s

• Beim Betreten von with stoves: wird eine Marke genommen (acquire), beim Verlassen zurückgegeben (release). Sind null Marken übrig, wartet, wer rein will.
• Drei bei 0,0 Sekunden, drei bei 0,3 Sekunden. Mit drei Platten werden sechs Bestellungen in zwei Runden erledigt, das dauert 0,6 Sekunden. Wer zuerst an eine Platte kommt, kann sich von Lauf zu Lauf ändern.
• Semaphore(1) lässt wie Lock immer nur einen rein. Aber wer bei einem Lock einen schon freigegebenen Lock noch einmal freigibt, bekommt einen RuntimeError; ein Semaphor dagegen zählt bei einem zusätzlichen release() stillschweigend eine Marke dazu (siehe BoundedSemaphore unten).
• In der Praxis begrenzt man damit die Zahl von Ressourcen, die nicht alle auf einmal belastet werden dürfen, etwa „höchstens fünf gleichzeitige Verbindungen“ oder „höchstens drei Downloads gleichzeitig“.

Ändere in der Zeitleiste unten die Zahl der Herdplatten. Gelbe Balken sind Wartezeit auf eine Platte, orange Balken die Zeit, in der wirklich gekocht wird.

• 3 Herdplatten: Drei Reihen sind sofort orange, die anderen drei warten 0,3 Sekunden gelb und werden dann orange. Insgesamt etwa 0,6 Sekunden.
• 1 Herdplatte: Selbst mit sechs Köchen kocht einer nach dem anderen. Der letzte Koch wartet 1,5 Sekunden, insgesamt etwa 1,8 Sekunden. Genauso wie allein.
• 6 Herdplatten: kaum Wartezeit, etwa 0,3 Sekunden. Sobald es so viele Platten wie Köche gibt, wird es nicht mehr schneller.
• Mehr Köche bei gleich vielen Platten bringen nur mehr gelbe Balken. Der Engpass ist nicht die Zahl der Leute, sondern die Zahl der Platten.

BoundedSemaphore: Fehler abfangen, bei denen zu viele Marken aufgehängt werden

Ein Semaphore zählt nicht nach, wie viele Marken zurückkommen. Ruft ein Bug release() einmal zu oft auf, sind es heimlich vier Marken. Drei Platten, aber vier Leute am Feuer.

loose = threading.Semaphore(3)
loose.release()                         # versehentlich einmal zu oft zurückgegeben
print(sum(loose.acquire(blocking=False) for _ in range(10)))   # plötzlich vier Platten
# 4

strict = threading.BoundedSemaphore(3)
strict.release()
# ValueError: Semaphore released too many times

BoundedSemaphore wirft einen ValueError, sobald mehr zurückgegeben wird als anfangs festgelegt. Wenn du einen Semaphor zur Begrenzung einer Anzahl verwendest, ist BoundedSemaphore fast immer die bessere Wahl. Der Fehler rutscht nicht still durch, sondern knallt genau an der Stelle.

4-2. „Offen!“: Event (9 Min.)

Ein Event ist eine Signallampe an der Küchenwand. Am Anfang ist sie aus; schaltet sie jemand ein (set()), legen alle, die auf das Signal gewartet haben (wait()), gleichzeitig los.

opened = threading.Event()              # am Anfang aus
start = time.perf_counter()

def cook(name):
    opened.wait()                       # warten, bis das Signal angeht
    print(f"{time.perf_counter() - start:.1f}s {name} fire")

threads = [threading.Thread(target=cook, args=(n,)) for n in ["minho", "sua", "tak"]]
for th in threads:
    th.start()
time.sleep(0.5)
print(opened.is_set())
opened.set()                            # „Offen!“
for th in threads:
    th.join()
print(opened.is_set())
# False
# 0.5s minho fire     (die Reihenfolge der drei Zeilen ändert sich von Lauf zu Lauf)
# 0.5s sua fire
# 0.5s tak fire
# True

Du kannst wait() auch eine Zeitgrenze geben. Dann ist der Rückgabewert wichtig: True, wenn das Signal angegangen ist, False, wenn die Zeit abgelaufen und es immer noch aus ist.

closing = threading.Event()
print(closing.wait(timeout=0.2))        # 0,2 Sekunden gewartet, kein Signal
closing.set()
print(closing.wait(timeout=0.2))        # ist es schon an: sofort True
closing.clear()                         # wieder ausschalten
print(closing.is_set())
# False
# True
# False

Ein Muster kommt sehr oft vor: Man macht aus dem Signal „Feierabend“ ein Event und lässt den arbeitenden Thread bei jedem Schleifendurchlauf mit if closing.is_set(): break nachsehen. So bricht man ihn nicht mittendrin ab wie den daemon-Thread aus Lektion 2, sondern lässt ihn selbst aufräumen und aufhören.

Die Zeitleiste unten zeigt eine Küche, in der die Chefin nach 0,5 Sekunden „Offen!“ ruft.

• Alle Reihen warten von 0 bis 0,5 Sekunden gelb und werden bei 0,5 Sekunden gleichzeitig orange. Insgesamt etwa 0,8 Sekunden.
• Ob 1 Koch oder 8 Köche: Die Startlinie liegt immer bei 0,5 Sekunden. Ein einziges set() weckt alle, die warten.

4-3. „Weck mich, wenn die Zutaten da sind“: Condition (10 Min.)

Ein Event kennt nur an oder aus. In der Küche wartet man aber meist auf etwas Konkreteres. Minho wartet auf Eier, Sua auf Tofu. Jedes Mal, wenn eine Zutat ins Regal kommt, muss man nachsehen: „Ist meine dabei?“, und wenn nicht, weiter warten. Dafür gibt es Condition.

Eine Condition ist ein Lock + eine „Weck mich“-Liste. Sie schützt den gemeinsamen Zustand (das Zutatenregal) mit dem Lock, und wenn sich der Zustand ändert, weckt sie die Leute auf der Liste.

pantry = []                             # Zutatenregal (gemeinsamer Zustand)
arrived = threading.Condition()         # Lock für das Regal + „Weck mich“-Liste
start = time.perf_counter()

def cook(name, need):
    with arrived:
        ok = arrived.wait_for(lambda: need in pantry, timeout=0.5)
        print(f"{time.perf_counter() - start:.1f}s {name} {need if ok else 'gave up'}")

def supplier():
    for item in ["tofu", "egg"]:        # Fisch kommt heute nicht
        time.sleep(0.1)
        with arrived:
            pantry.append(item)
            arrived.notify_all()        # alle Wartenden wecken

threads = [threading.Thread(target=cook, args=a) for a in [("minho", "egg"), ("sua", "tofu"), ("tak", "fish")]]
threads.append(threading.Thread(target=supplier))
for th in threads:
    th.start()
for th in threads:
    th.join()
# 0.1s sua tofu
# 0.2s minho egg
# 0.5s tak gave up

• Angesehen und geändert wird das Regal nur, während man mit with arrived: den Lock hält. Genau die Regel aus Lektion 3.
• wait_for(bedingung) wartet, bis die Bedingung wahr wird. Während des Wartens legt es den Lock kurz ab, damit der Lieferant Zutaten ins Regal stellen kann. Beim Aufwachen nimmt es den Lock wieder und prüft die Bedingung erneut.
• Mit timeout liefert es nach Ablauf der Zeit den letzten Wert der Bedingung. Der Fisch kam nie, also bekam Herr Tak nach 0,5 Sekunden False und gab auf.
• notify_all() weckt alle auf der Liste. Jeder Geweckte prüft seine eigene Bedingung und schläft wieder ein, wenn sie nicht erfüllt ist. Als der Tofu kam, wurden auch Minho und Herr Tak wach, warteten aber weiter, weil es nicht ihre Zutat war.

notify() weckt nur einen Wartenden. Das passt zu Arbeit, bei der es egal ist, wer sie übernimmt, etwa „irgendeine Servicekraft holt einen fertigen Teller ab“.

plates = 0
counter = threading.Condition()

def cook():
    global plates
    time.sleep(0.1)
    with counter:
        plates += 1
        counter.notify()                # nur einen Wartenden wecken

def waiter(name):
    global plates
    with counter:
        counter.wait_for(lambda: plates > 0)
        plates -= 1
        print(name, "took a plate")

ws = [threading.Thread(target=waiter, args=(n,)) for n in ["w1", "w2"]]
cs = [threading.Thread(target=cook) for _ in range(2)]
for th in ws + cs:
    th.start()
for th in ws + cs:
    th.join()
print(plates)
# zum Beispiel:
# w1 took a plate
# w2 took a plate
# 0

Pro Teller wird nur einer geweckt, also kommt es seltener vor, dass zwei gleichzeitig aufwachen und sich um einen Teller streiten. Wartet dagegen jeder auf eine andere Bedingung, kann notify() den Falschen wecken; dann nimm notify_all(). Im Zweifel ist notify_all() die sichere Wahl.

4-4. Vier Teller für den Gruppentisch auf einmal: Barrier (5 Min.)

An einem Vierertisch der Gruppe müssen alle vier Teller gleichzeitig rausgehen. Kommt nur einer vorneweg, sitzen die anderen drei da und warten, während sein Essen kalt wird. Barrier(n) lässt alle am Pass warten, bis n Leute angekommen sind, und schickt sie dann zusammen los.

start = time.perf_counter()
table = threading.Barrier(4, action=lambda: print("serve 4 plates"))
done = []

def cook(name, sec):
    time.sleep(sec)                     # jeder hat sein eigenes Tempo
    i = table.wait()                    # am Pass warten, bis alle vier da sind
    done.append((round(time.perf_counter() - start, 1), i, name))

crew = [("minho", 0.1), ("sua", 0.2), ("chef", 0.3), ("tak", 0.4)]
threads = [threading.Thread(target=cook, args=c) for c in crew]
for th in threads:
    th.start()
for th in threads:
    th.join()
print(sorted(done))
# serve 4 plates
# [(0.4, 0, 'minho'), (0.4, 1, 'sua'), (0.4, 2, 'chef'), (0.4, 3, 'tak')]

• Minho war nach 0,1 Sekunden fertig, aber alle vier gehen erst bei 0,4 Sekunden, als der langsamste, Herr Tak, ankommt, gemeinsam durch.
• wait() gibt jedem eine andere Nummer von 0 bis n-1. In CPython ist das die Ankunftsreihenfolge: 0, 1, 2, 3. Mit dieser Nummer kannst du festlegen, was „nur einer erledigen soll“ (zum Beispiel if i == 0: einen Logeintrag schreiben).
• Die Funktion bei action= läuft genau einmal, wenn der Letzte ankommt. Das ist das Signal, die Teller rauszuschicken.
• Sind alle durch, kehrt die Barrier in den Ausgangszustand zurück und kann für den nächsten Tisch wiederverwendet werden. Läuft bei wait(timeout=) die Zeit ab, bricht die Barrier, und alle Wartenden bekommen einen BrokenBarrierError.

4-5. Die Werkzeuge des Abends im Überblick (4 Min.)

Alle fünf stecken in threading, und alle sind auf dem Lock aus Lektion 3 aufgebaut. Tatsächlich verwenden Semaphore, Event und Barrier intern eine Condition, und eine Condition trägt intern einen Lock (wenn du keinen angibst, einen RLock).

Küchenschluss (10 Min.)

Das Ziel für heute Nacht war: Vor dem Öffnungssignal macht niemand Feuer, höchstens drei Herdplatten laufen gleichzeitig, und sechs Bestellungen gehen in unter einer Sekunde raus. Ein Event ist die Startlinie, ein Semaphore(3) regelt die Platten, und mit einem Lock zählen wir, „wie viele gerade am Feuer stehen“.

opened = threading.Event()
stoves = threading.Semaphore(3)
lock = threading.Lock()
on_fire = peak = 0
log = []
start = time.perf_counter()

def cook(n):
    global on_fire, peak
    opened.wait()                       # 1. vor „Offen!“ wird kein Feuer gemacht
    with stoves:                        # 2. höchstens drei Platten gleichzeitig
        with lock:
            on_fire += 1
            peak = max(peak, on_fire)
        time.sleep(0.3)
        with lock:
            on_fire -= 1
            log.append((round(time.perf_counter() - start, 1), f"#{n}"))

threads = [threading.Thread(target=cook, args=(n,)) for n in range(1, 7)]
for th in threads:
    th.start()
time.sleep(0.3)                         # alles für die Gäste vorbereiten
opened.set()
for th in threads:
    th.join()
print(sorted(log))
print(f"peak {peak}, total {time.perf_counter() - start:.2f}s")
# zum Beispiel:
# [(0.6, '#1'), (0.6, '#2'), (0.6, '#3'), (0.9, '#4'), (0.9, '#5'), (0.9, '#6')]
# peak 3, total 0.90s

• Bis 0,3 Sekunden kocht niemand. Der erste Teller geht bei 0,6 Sekunden raus, der letzte bei 0,9 Sekunden, also innerhalb des Ziels von einer Sekunde.
• peak ist bei jedem Lauf 3. Welche drei Bestellungen bei 0,6 Sekunden rausgehen, kann sich von Lauf zu Lauf ändern.
• on_fire und peak werden von allen sechs Köchen geändert, deshalb stehen sie in lock. Auch innerhalb des Semaphors.

Kurz vor Küchenschluss leert eine Frau, die allein am Fensterplatz sitzt, eine Schüssel Nudeln. Sie schaut einmal auf ihre Armbanduhr, notiert etwas in ein kleines Notizbuch, zahlt leise und geht.

Die Chefin dreht die Ventile der Platten eins nach dem anderen zu. „Heute habt ihr euch alle gut angestellt. Aber was, wenn es außer den Platten auch Messer und Bretter nur je einmal gibt? Wenn Minho mit dem Messer auf das Brett wartet und Sua mit dem Brett auf das Messer, was passiert dann?“

Übungen

1. Stell in der Herdplatten-Zeitleiste 6 Köche und 6 Bestellungen ein, ändere die Zahl der Platten von 1 bis 6 und trag die Gesamtzeit in eine Tabelle ein. Warum dauert es mit 4 und 5 Platten gleich lang?
2. Bau das Muster mit dem Signal „Feierabend“ aus 4-2 selbst. Leg closing = threading.Event() an und lass einen Koch-Thread alle 0,1 Sekunden eine Schüssel machen, bis closing.is_set() wahr ist. Der Haupt-Thread ruft nach 0,35 Sekunden closing.set() auf und dann join(). Wie viele Schüsseln sind rausgegangen?
3. Überleg dir, was im Barrier-Beispiel aus 4-4 passiert, wenn du Barrier(4) durch Barrier(5) ersetzt. Damit es nicht wirklich für immer hängt, ändere es auf table.wait(timeout=1), lass es laufen und sieh nach, welche Ausnahme kommt.