Die Bonschiene: queue

Sechste Nacht, Freitag. Die Servicekräfte stürmen mit den Bons in die Küche und drücken sie einfach irgendwem in die Hand. Minho und Sua schnappen sich gleichzeitig denselben Bon, und dieselbe Nudelsuppe geht zweimal raus. Herr Tak steht derweil mit dem Messer in der Hand da und wartet auf Arbeit.

Chefin Yoon Seoyeon schraubt über dem Küchendurchgang eine lange Metallleiste an. Die Bonschiene. „Der Service steckt hier nur Bons rein. Die Köche ziehen hier nur Bons raus. Ihr redet nicht miteinander.“

6-1. Wer reinsteckt und wer rauszieht (8 Min.)

Die Seite, die Arbeit erzeugt (der Service), und die Seite, die sie abarbeitet (die Köche), werden getrennt, und dazwischen kommt eine Warteschlange. Dieses Design heißt Erzeuger-Verbraucher-Muster (Producer-Consumer). In Python ist queue.Queue diese Warteschlange.

import queue

rail = queue.Queue()        # die Bonschiene
for n in range(1, 5):
    rail.put(f"#{n}")       # reinstecken
print(rail.qsize(), rail.get(), rail.get(), rail.qsize())
# 4 #1 #2 2

Der Bon, der zuerst reingesteckt wurde, kommt auch zuerst raus (FIFO, first in, first out). Jetzt lassen wir eine Servicekraft und zwei Köche als Threads laufen.

import threading
import time

rail = queue.Queue()
served = []

def hall():                          # Erzeuger: nimmt Bestellungen auf und steckt sie rein
    for n in range(1, 7):
        time.sleep(0.05)
        rail.put(f"#{n}")

def cook(name):                      # Verbraucher: zieht Bons raus und kocht
    for _ in range(3):
        order = rail.get()           # ist die Schiene leer, wird hier gewartet
        time.sleep(0.1)
        served.append((name, order))

ts = [threading.Thread(target=hall),
      threading.Thread(target=cook, args=("minho",)),
      threading.Thread(target=cook, args=("sua",))]
for t in ts: t.start()
for t in ts: t.join()
print(sorted(order for _, order in served))
# ['#1', '#2', '#3', '#4', '#5', '#6']

• Alle sechs Bestellungen wurden genau einmal gekocht. Dass zwei Köche denselben Bon greifen, kommt nicht vor.
• Nirgends im Code steht ein Lock. In Queue stecken schon ein Lock und die Condition aus Lektion 4, deshalb kümmern sich put und get selbst darum, dass immer nur einer zum Zug kommt. Die Queue ersetzt den Lock.
• Ist die Schiene leer, wartet get(), bis ein Bon reingesteckt wird. Kein Koch muss geschäftig im Kreis laufen und ständig fragen: „Gibt's was?“

Wenn Threads Daten austauschen müssen, denk zuerst daran, sie über eine Queue weiterzureichen, statt eine gemeinsame Variable mit einem Lock zu schützen. Je weniger Code Locks direkt anfasst, desto seltener passieren Unfälle wie in Lektion 3 und 5.

6-2. Wissen, dass alle Bestellungen fertig sind: task_done und join (8 Min.)

Im Beispiel eben hatten wir jedem Koch gesagt: „Drei Gerichte, dann Feierabend.“ In einer echten Küche weiß man nicht, wie viele Bestellungen kommen. Die Köche behalten die Schiene ständig im Blick, und die Chefin will nur eins wissen: Sind alle reingesteckten Bestellungen fertig?

rail = queue.Queue()

def cook(name):
    while True:
        order = rail.get()
        time.sleep(0.1)
        rail.task_done()          # „dieser Bon ist erledigt“

for name in ("minho", "sua", "tak"):
    threading.Thread(target=cook, args=(name,), daemon=True).start()

start = time.perf_counter()
for n in range(1, 10):
    rail.put(f"#{n}")
rail.join()                       # bis alle reingesteckten Bons erledigt sind
print(f"all 9 done: {time.perf_counter() - start:.2f}s")
# all 9 done: 0.30s

Die Queue zählt intern, wie viele Bestellungen noch nicht erledigt sind.

• Mit get() rausgezogen heißt noch nicht erledigt. Erst wenn du nach dem Kochen task_done() aufrufst, sinkt die Zahl.
• Die Koch-Threads beobachten mit while True für immer die Schiene, deshalb haben sie daemon=True. Sind alle Bestellungen fertig, löst sich join(), und wenn das Programm endet, machen die Köche gleich mit Feierabend.
• Rufst du task_done() öfter auf, als du Bons rausgezogen hast, gibt es ValueError: task_done() called too many times.

6-3. Wenn die Schiene voll ist (10 Min.)

Du kannst festlegen, wie viele Plätze die Schiene hat. Bei Queue(maxsize=2) passen nur zwei Bons drauf. Rufst du auf einer vollen Schiene put() auf, wartet die Seite, die reinsteckt. Gerät die Küche in Verzug, nimmt der Service eben langsamer Bestellungen auf.

rail = queue.Queue(maxsize=2)     # nur zwei Bons passen drauf
rail.put("#1")
rail.put("#2")
print(rail.full())
# True

try:
    rail.put("#3", timeout=0.5)   # wenn nach 0,5 s noch kein Platz frei ist
except queue.Full:
    print("Full after 0.5s")
# Full after 0.5s

try:
    rail.put_nowait("#3")         # = put("#3", block=False)
except queue.Full:
    print("Full right away")
# Full right away

print(rail.get_nowait(), rail.get(timeout=0.5))
# #1 #2

try:
    rail.get(timeout=0.5)         # die Schiene ist leer
except queue.Empty:
    print("Empty after 0.5s")
# Empty after 0.5s

Gibst du kein maxsize an oder setzt es auf 0, wird die Schiene beliebig lang. Ist der Service viel schneller als die Küche, stapeln sich die Bons ohne Ende und füllen den Speicher. Legst du eine Größe fest, wartet die schnelle Seite auf die langsame. Das nennt man Backpressure (Gegendruck).

rail = queue.Queue(maxsize=2)

def cook():                           # ein einziger, langsamer Koch
    while True:
        rail.get()
        time.sleep(0.1)
        rail.task_done()

threading.Thread(target=cook, daemon=True).start()

start = time.perf_counter()
for n in range(1, 7):
    rail.put(f"#{n}")                 # schneller Service: kein Platz frei, also wartet put
print(f"hall done at {time.perf_counter() - start:.1f}s")
rail.join()
print(f"kitchen done at {time.perf_counter() - start:.1f}s")
# hall done at 0.3s
# kitchen done at 0.6s

Der Service wollte die sechs Bons im Handumdrehen reinstecken, aber die Schiene hat nur zwei Plätze, also hing er 0,3 Sekunden lang in put. Lässt du maxsize weg, ist der Service sofort fertig, und auf der Schiene stapeln sich die Bons.

Probier es in der Zeitleiste unten selbst aus. Die oberen Zeilen sind die Servicekräfte, die unteren die Köche. Eine Servicekraft braucht 0,1 Sekunden, um eine Bestellung aufzunehmen (orange), plus die Zeit, in der sie wartet, bis auf der Schiene ein Platz frei wird (gelb).

• Mit den Startwerten (2 Köche, 3 Plätze) dauert es etwa 1,6 Sekunden, und auf der Schiene hängen höchstens 3 Bons. Die Servicekräfte fangen ab etwa 0,3 Sekunden an zu warten.
• Mit 8 Plätzen warten die Servicekräfte gar nicht und sind nach etwa 0,5 Sekunden fertig. Die Gesamtzeit bleibt trotzdem bei etwa 1,6 Sekunden. Eine langsame Küche wird durch eine längere Schiene nicht schneller.
• Mit 4 Köchen sinkt die Gesamtzeit auf etwa 1,0 Sekunden. Wo der Engpass liegt, erkennst du daran, in welchen Zeilen die gelben Balken lang werden.

6-4. Das Signal für Feierabend: Wächterwert und shutdown (9 Min.)

Statt die Köche mit daemon=True einfach stehen zu lassen, will man ihnen manchmal ordentlich sagen: „Für heute ist Schluss“, damit sie sauber Feierabend machen. Der klassische Weg ist ein Wächterwert (Sentinel), also ein besonderer Bon, auf dem „Feierabend“ steht.

STOP = None                         # Wächterwert: der „Feierabend“-Bon
rail = queue.Queue()
served = []

def cook(name):
    while True:
        order = rail.get()
        if order is STOP:
            break                   # Wächterwert bekommen: Feierabend
        time.sleep(0.05)
        served.append(order)

cooks = [threading.Thread(target=cook, args=(n,)) for n in ("minho", "sua")]
for t in cooks: t.start()
for n in range(1, 7):
    rail.put(f"#{n}")
for _ in cooks:
    rail.put(STOP)                  # einen pro Koch reinstecken
for t in cooks: t.join()
print(len(served), [t.is_alive() for t in cooks])
# 6 [False, False]

Du musst so viele Wächterwerte reinstecken, wie es Köche gibt. Steckst du nur einen rein, macht nur einer Feierabend, und die anderen warten für immer vor der leeren Schiene. Weil die Schiene FIFO ist, kommt der Wächterwert erst raus, wenn alle vorher reingesteckten Bestellungen weg sind.

Ab Python 3.13 kannst du die Schiene selbst schließen. Nach Queue.shutdown() wirft jedes weitere put() queue.ShutDown. get() gibt erst alle übrigen Bons heraus und wirft dann queue.ShutDown. Die Köche musst du nicht mehr zählen.

rail = queue.Queue()
served = []

def cook(name):
    while True:
        try:
            order = rail.get()
        except queue.ShutDown:      # die Schiene ist zu, und es sind keine Bons mehr übrig
            return
        time.sleep(0.05)
        served.append(order)
        rail.task_done()

cooks = [threading.Thread(target=cook, args=(n,)) for n in ("minho", "sua")]
for t in cooks: t.start()
for n in range(1, 7):
    rail.put(f"#{n}")
rail.shutdown()                     # nimmt nichts mehr an; was dran hängt, wird noch gekocht
for t in cooks: t.join()
print(len(served))
# 6

try:
    rail.put("#7")
except queue.ShutDown:
    print("rail is closed")
# rail is closed

Wenn du in aller Eile schließen musst, nimmst du shutdown(immediate=True). Die übrigen Bons auf der Schiene werden alle weggeworfen, und auch ein wartendes get() wirft sofort queue.ShutDown.

rail = queue.Queue()
served = []

def cook(name):
    while True:
        try:
            order = rail.get()
        except queue.ShutDown:
            return
        time.sleep(0.1)
        served.append(order)
        rail.task_done()

cooks = [threading.Thread(target=cook, args=(n,)) for n in ("minho", "sua")]
for t in cooks: t.start()
for n in range(1, 7):
    rail.put(f"#{n}")
time.sleep(0.15)                    # die beiden kochen gerade #3 und #4
rail.shutdown(immediate=True)       # #5 und #6 auf der Schiene wegwerfen und sofort schließen
rail.join()                         # löst sich, sobald #3 und #4 in der Hand fertig sind
for t in cooks: t.join()
print(sorted(served), rail.qsize())
# ['#1', '#2', '#3', '#4'] 0

Auch immediate=True nimmt niemandem die Bestellung weg, die er schon in der Hand hat. Deshalb löst sich join() erst, nachdem für diese Bestellungen task_done() aufgerufen wurde.

6-5. Schienen in anderer Form (5 Min.)

Im Modul queue gibt es noch weitere Schienen, die in anderer Reihenfolge herausgeben.

PriorityQueue: gibt den kleinsten Wert zuerst heraus. Praktisch, wenn Bestellungen von VIP-Tischen zuerst gekocht werden sollen. Üblich ist, Tupel der Form (Priorität, Nummer, Bestellung) hineinzulegen.

import itertools

vip_first = queue.PriorityQueue()
ticket = itertools.count(1)          # bei gleicher Priorität: wer zuerst kam
for prio, dish in [(2, "noodle"), (1, "VIP soup"), (2, "rice"), (1, "VIP noodle")]:
    vip_first.put((prio, next(ticket), dish))

while not vip_first.empty():
    print(vip_first.get())
# (1, 2, 'VIP soup')
# (1, 4, 'VIP noodle')
# (2, 1, 'noodle')
# (2, 3, 'rice')

Fehlt die Nummer in der Mitte, werden bei gleicher Priorität die dritten Werte miteinander verglichen. Ist die Bestellung etwas Unvergleichbares wie ein dict, knallt es.

pq = queue.PriorityQueue()
pq.put((1, {"dish": "soup"}))
try:
    pq.put((1, {"dish": "rice"}))   # gleiche Priorität, also werden zwei dicts verglichen
except TypeError as e:
    print(e)
# '<' not supported between instances of 'dict' and 'dict'

Außerdem gibt es LifoQueue, die wie ein Tellerstapel das Letzte zuerst herausgibt, und die ganz einfache SimpleQueue.

plates = queue.LifoQueue()           # Tellerstapel: der zuletzt aufgelegte zuerst
for p in ("p1", "p2", "p3"):
    plates.put(p)
print(plates.get(), plates.get(), plates.get())
# p3 p2 p1

simple = queue.SimpleQueue()         # die einfachste Schiene: kein maxsize, task_done oder join
simple.put("#1")
print(simple.get(), hasattr(simple, "task_done"))
# #1 False

Küchenschluss (10 Min.)

Das Ziel für heute Nacht war, dass drei Köche 12 Bestellungen jede genau einmal kochen, ohne eine auszulassen, und dass wir sicher wissen, wann alles fertig ist. Zwei Servicekräfte nehmen abwechselnd Bestellungen auf, die Schiene hat vier Plätze.

rail = queue.Queue(maxsize=4)        # die Schiene hat vier Plätze
served = []

def hall(orders):                    # Servicekraft: nimmt Bestellungen auf und steckt sie rein
    for n in orders:
        time.sleep(0.02)
        rail.put(f"#{n}")

def cook(name):
    while True:
        try:
            order = rail.get()
        except queue.ShutDown:
            return
        time.sleep(0.1)
        served.append((name, order))
        rail.task_done()

start = time.perf_counter()
halls = [threading.Thread(target=hall, args=(range(k, 13, 2),)) for k in (1, 2)]
cooks = [threading.Thread(target=cook, args=(n,)) for n in ("minho", "sua", "tak")]
for t in halls + cooks: t.start()
for t in halls: t.join()             # sind alle Bestellungen aufgenommen,
rail.shutdown()                      # Schiene schließen
rail.join()                          # und warten, bis alle Bons erledigt sind
for t in cooks: t.join()
orders = [o for _, o in served]
print(len(orders), len(set(orders)), f"{time.perf_counter() - start:.2f}s")
# 12 12 0.44s  (Zeit nur ungefähr)

12 Gerichte gekocht, 12 verschiedene Bestellungen. Keine kam doppelt, keine fehlte. Auch Herr Tak hat seinen Teil beigetragen. Und in diesem Code steht nicht eine einzige Zeile Lock.

Die Chefin tippt gegen die Schiene. „Jetzt müssen Gastraum und Küche nicht mehr miteinander reden. Aber was, wenn nächsten Freitag eine Gruppe 30 Gerichte auf einmal bestellt? Stellen wir dann 30 Köche ein?“

Übungen

1. Ändere cook aus 6-2 so, dass nur bei Bestellung #5 ein ValueError fliegt. Löst sich rail.join() dann noch? Damit du nicht ewig wartest, starte es mit waiter = threading.Thread(target=rail.join, daemon=True), gib nach waiter.join(timeout=1) den Wert von waiter.is_alive() aus und repariere es anschließend mit try/finally.
2. Lass die Zeitleiste mit 4 Servicekräften, 1 Koch und 1 Platz auf der Bonschiene laufen. Erklär, in welchen Zeilen die gelben Balken entstehen und wo der Engpass liegt.
3. Bau den Code vom Küchenschluss auf PriorityQueue um, sodass Bestellungen mit gerader Nummer als VIP zuerst gekocht werden. shutdown() funktioniert bei PriorityQueue genauso.