Nebenläufigen Code testen

Code in:JavaC#PythonJavaScript

Kurz gesagt: Nebenläufiger Code wird mit einem Stresstest aus vielen Läufen, einer Startbarriere, erzwungenen Verzahnungen und einer Prüfung der Invariante getestet, denn ein grüner Lauf beweist nichts.

Problem

Bei mir besteht er

Der Verkaufszähler aus der Lektion Race Condition hat keinen Lock. Ein Kollege schreibt einen Test: Vier Threads verkaufen je 1.000-mal, und am Ende müssen 4.000 verkauft sein:

class Counter:
    def __init__(self):
        self.sold = 0

    def sell(self):
        self.sold += 1


def sell_once():
    counter = Counter()

    def buyer():
        for _ in range(1000):
            counter.sell()

    threads = [threading.Thread(target=buyer) for _ in range(4)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()
    return counter.sold


print("Verkauft (erwartet 4000):")
print(sell_once())

Ausgabe

Verkauft (erwartet 4000):
4000

Bei ihnen besteht der Test. In der CI besteht er auch. Der Code geht in den Sale, und in der ersten Stunde hat der Shop mehr Artikel verkauft, als der Zähler zeigt.

In Python ist das besonders leicht: Wegen des GIL wechseln Threads etwa alle 5 ms, und self.sold += 1 sind ein paar Anweisungen. Die Chance, dass der Wechsel genau zwischen Lesen und Schreiben landet, ist klein. Klein, aber nicht null: Auf einem Server mit vielen Anfragen passiert es in einer Stunde.

Ein grüner Test von nebenläufigem Code beweist fast nichts: In diesem Lauf sind sich die Threads einfach nicht genau zwischen Lesen und Schreiben begegnet. Wie schreibst du einen Test, der so einen Fehler fängt?

Die Idee

Ein Stresstest: viele Läufe und eine Barriere

Drei Dinge machen den Fehler sichtbar:

• Viele Läufe. Der Test ist nicht ein Lauf, sondern 20 (oder 1.000 in einem nächtlichen CI-Job). Du zählst, wie viele rot sind.
• Viele Threads gleichzeitig. Mehr Threads als Kerne, damit sie mitten in der Arbeit wechseln.
• Eine Barriere am Start. Jeder Thread wartet, bis alle angekommen sind, und sie starten gemeinsam. So überschneiden sie sich die ganze Zeit, nicht nur am Ende.

class Counter:
    def __init__(self):
        self.sold = 0

    def sell(self):
        left = self.sold
        time.sleep(0)  # hier kommt ein anderer Thread herein
        self.sold = left + 1


def lost_sales():
    counter = Counter()

    def buyer():
        for _ in range(1000):
            counter.sell()

    threads = []
    for _ in range(4):
        t = threading.Thread(target=buyer)
        t.start()
        threads.append(t)
    for t in threads:
        t.join()
    return counter.sold != 4000

Schritt für Schritt, in Worten

1. Der Test startet 4 Threads, jeder verkauft 1.000-mal ohne Lock. Ohne Barriere startet jeder Thread, sobald er erzeugt ist.
2. 20 Läufe: 20 von 20 rot. Hier öffnet time.sleep(0) eine breite Lücke, deshalb schlägt der Test auch ohne Barriere fehl. In echtem Code ist die Lücke schmal.
3. Eine Barriere: Jeder Thread wartet am Start, bis alle vier angekommen sind. Dann laufen sie gemeinsam.
4. Mit der Barriere: 20 von 20 rot, genauso viele wie ohne: sleep(0) hält die Lücke breit. Bei schmaler Lücke entscheidet die Barriere.

Der Test mit einer Barriere:

class Counter:
    def __init__(self):
        self.sold = 0

    def sell(self):
        left = self.sold
        time.sleep(0)  # hier kommt ein anderer Thread herein
        self.sold = left + 1


def lost_sales():
    counter = Counter()
    start = threading.Barrier(4, timeout=1)

    def buyer():
        start.wait()
        for _ in range(1000):
            counter.sell()

    threads = []
    for _ in range(4):
        t = threading.Thread(target=buyer)
        t.start()
        threads.append(t)
    for t in threads:
        t.join()
    return counter.sold != 4000


failed = 0
for run in range(20):
    if lost_sales():
        failed += 1
print("Läufe mit verlorenen Verkäufen (von 20):")
print(failed)

Ausgabe

Läufe mit verlorenen Verkäufen (von 20):
20

Die Ausgabe ist „höchstens 20“: Wie viele Läufe rot werden, hängt vom Rechner ab. Auf meinem Rechner sind es 20 von 20, mit und ohne Barriere.

threading.Barrier(4, timeout=1): wait() wartet, bis 4 Threads es aufgerufen haben, und lässt dann alle los. Kommt einer innerhalb von 1 Sekunde nicht an, löst es BrokenBarrierError aus, statt zu hängen.

Hier gibt sell das GIL mit time.sleep(0) zwischen Lesen und Schreiben frei (wie in der Lektion über Race Conditions). Die Lücke ist so breit, dass der Test auch ohne Barriere fehlschlägt. In echtem Code ist die Lücke ein paar Anweisungen groß, und dort entscheiden die Barriere und die vielen Läufe, ob du sie überhaupt siehst.

Derselbe Test gegen den Zähler mit einem Lock:

class Counter:
    def __init__(self):
        self.sold = 0
        self.lock = threading.Lock()

    def sell(self):
        with self.lock:
            left = self.sold
            time.sleep(0)  # hier kommt ein anderer Thread herein
            self.sold = left + 1


def lost_sales():
    counter = Counter()
    start = threading.Barrier(4, timeout=1)

    def buyer():
        start.wait()
        for _ in range(1000):
            counter.sell()

    threads = []
    for _ in range(4):
        t = threading.Thread(target=buyer)
        t.start()
        threads.append(t)
    for t in threads:
        t.join()
    return counter.sold != 4000

Ausgabe

Läufe mit verlorenen Verkäufen: 0 von 20

Genauer, für Neugierige

Ein Stresstest findet Fehler, aber er beweist nicht, dass es keine gibt. „0 von 20“ heißt „wir haben es diesmal nicht gesehen“. Deshalb laufen solche Tests in der CI tausendfach, auf Rechnern mit vielen Kernen, und im Code-Review wird getrennt gefragt: Was ist geteilt, und welcher Lock bewacht es.

Wie es in den anderen Sprachen ist: In Java ist die Barriere CyclicBarrier, in C# Barrier, in Node schreibst du sie mit Atomics.wait und Atomics.notify. Statt einer Barriere ist auch ein threading.Event verbreitet, das der Hauptthread setzt.

Probier es aus

Finde den Fehler

Ein anderer Kollege hat einen Stresstest ohne Barriere geschrieben. Er besteht 20 von 20 Mal, bei jedem Lauf, auf jedem Rechner, obwohl der Zähler keinen Lock hat:

class Counter:
    def __init__(self):
        self.sold = 0

    def sell(self):
        left = self.sold
        time.sleep(0)  # hier kommt ein anderer Thread herein
        self.sold = left + 1


def lost_sales():
    counter = Counter()

    def buyer():
        for _ in range(1000):
            counter.sell()

    for _ in range(4):
        t = threading.Thread(target=buyer)
        t.start()
        t.join()
    return counter.sold != 4000

Ausgabe

Läufe mit verlorenen Verkäufen: 0 von 20

Wähle die Zeile mit dem Fehler.

1class Counter:2 def __init__(self):3 self.sold = 045 def sell(self):6 left = self.sold7 time.sleep(0) # hier kommt ein anderer Thread herein8 self.sold = left + 191011def lost_sales():12 counter = Counter()1314 def buyer():15 for _ in range(1000):16 counter.sell()1718 for _ in range(4):19 t = threading.Thread(target=buyer)20 t.start()21 t.join()22 return counter.sold != 4000

Prüfen

Probier es aus

Ergänze

Welche Zeile lässt die Threads gemeinsam starten?

class Counter:
    def __init__(self):
        self.sold = 0

    def sell(self):
        left = self.sold
        time.sleep(0)  # hier kommt ein anderer Thread herein
        self.sold = left + 1


def lost_sales():
    counter = Counter()
    start = threading.Barrier(4, timeout=1)

    def buyer():
        
        for _ in range(1000):
            counter.sell()

    threads = []
    for _ in range(4):
        t = threading.Thread(target=buyer)
        t.start()
        threads.append(t)
    for t in threads:
        t.join()
    return counter.sold != 4000

Prüfen

Die Idee

Die Verzahnung erzwingen

Ein Stresstest findet den Fehler manchmal. Für einen Test, der jedes Mal fehlschlägt, wählt der Test die Reihenfolge der Threads selbst. Dafür hat der Code eine Nahtstelle: eine Stelle, an der der Test den Thread anhalten kann. Im echten Programm tut die Nahtstelle nichts.

Das Szenario ist der Fehler aus der Lektion über Race Conditions: 10 Artikel sind übrig, zwei Kunden kaufen je einen.

1. Der erste Kunde liest 10 und hält an der Nahtstelle an.
2. Der zweite Kunde geht ganz durch: liest 10, schreibt 9.
3. Der erste macht weiter und schreibt sein 10 − 1 = 9.

class Stock:
    def __init__(self, count):
        self.count = count
        self.pause = lambda: None

    def buy(self):
        left = self.count
        self.pause()  # eine Nahtstelle für den Test: Hier kann der Test den Thread anhalten
        self.count = left - 1


def forced_test(stock):
    read = threading.Event()
    go = threading.Event()

    def pause():
        if threading.current_thread().name == "first":
            read.set()
            go.wait(timeout=1)

    stock.pause = pause
    first = threading.Thread(target=stock.buy, name="first")
    second = threading.Thread(target=stock.buy, name="second")
    first.start()
    read.wait(timeout=1)  # first hat 10 gelesen und wartet
    second.start()
    second.join(timeout=0.2)  # second kauft alles oder wartet auf den Lock
    go.set()  # first schreibt
    first.join(timeout=1)
    second.join(timeout=1)
    verdict = "BESTANDEN" if stock.count == 8 else "DURCHGEFALLEN"
    return f"Bestand: {stock.count} (erwartet 8): {verdict}"

Ausgabe

Bestand: 9 (erwartet 8): DURCHGEFALLEN

Die Nahtstelle ist das Feld pause. Der Test legt eine Funktion hinein, die nur den Thread namens first anhält: Sie setzt read und wartet auf go. Ein threading.Event ist ein Latch: set() öffnet ihn, wait(timeout=1) wartet, bis er offen ist, höchstens 1 Sekunde.

Der Bestand ist 9 statt 8, bei jedem Lauf. Jetzt die Korrektur: with self.lock: um Lesen und Schreiben:

class Stock:
    def __init__(self, count):
        self.count = count
        self.pause = lambda: None
        self.lock = threading.Lock()

    def buy(self):
        with self.lock:
            left = self.count
            self.pause()  # eine Nahtstelle für den Test: Hier kann der Test den Thread anhalten
            self.count = left - 1

Ausgabe

Bestand: 8 (erwartet 8): BESTANDEN

Der Test ist derselbe. Der erste Kunde hält den Lock, solange er an der Nahtstelle angehalten ist. Der zweite wartet auf den Lock, deshalb läuft second.join(timeout=0.2) ab, bevor er gekauft hat. Erst wenn der erste 9 schreibt und den Lock freigibt, liest der zweite 9 und schreibt 8. Ein Test, der vor der Korrektur jedes Mal fehlschlug und danach jedes Mal besteht, ist genau das, was du in der CI willst.

Probier es aus

Drei an der Nahtstelle

Dasselbe Stock ohne Lock. Jetzt wartet die Nahtstelle, bis alle drei Kunden den Bestand gelesen haben:

Was gibt das Programm aus?

stock = Stock(10)
all_read = threading.Barrier(3, timeout=1)
stock.pause = all_read.wait
buyers = [threading.Thread(target=stock.buy) for _ in range(3)]
for t in buyers:
    t.start()
for t in buyers:
    t.join()
print(f"Bestand: {stock.count}")

Antwort

Bestand: 8

Bestand: 7

Bestand: 9

Prüfen

Die Idee

Prüfe die Invariante, nicht die Reihenfolge

Bei 8 Kunden hängt es vom Lauf ab, wer was gekauft hat. Ein Test, der „der erste Kunde hat 13 gekauft“ erwartet, schlägt auch bei richtigem Code fehl. Deshalb prüft der Test eine Invariante: etwas, das in jeder Reihenfolge wahr sein muss.

Für den Shop ist das: Der Bestand plus die Zahl der Verkäufe ist das, was am Anfang da war, der Bestand liegt nicht unter null, und die Käufe der Kunden zusammengezählt ergeben die Zahl der Verkäufe. 8 Kunden machen je 50 Versuche für 100 Artikel:

shop = Shop(100)
bought = [0] * 8
start = threading.Barrier(8, timeout=1)


def buyer(me):
    start.wait()
    for _ in range(50):
        if shop.try_buy():
            bought[me] += 1


buyers = [threading.Thread(target=buyer, args=(i,)) for i in range(8)]
for t in buyers:
    t.start()
for t in buyers:
    t.join()
print(f"Bestand: {shop.stock}, verkauft: {shop.sold}")
print(check("Bestand + verkauft = 100", shop.stock + shop.sold == 100))
print(check("Bestand >= 0", shop.stock >= 0))
print(check("gekauft = verkauft", sum(bought) == shop.sold))

Ausgabe

Bestand: 0, verkauft: 100
Bestand + verkauft = 100: BESTANDEN
Bestand >= 0: BESTANDEN
gekauft = verkauft: BESTANDEN

Der Shop:

class Shop:
    def __init__(self, stock):
        self.stock = stock
        self.sold = 0
        self.lock = threading.Lock()

    def try_buy(self):
        with self.lock:
            if self.stock == 0:
                return False
            self.stock -= 1
            self.sold += 1
            return True

Jeder Thread schreibt in seinen eigenen Platz bought[me], die Liste braucht also keinen Lock. Die Summe stimmt in jeder Reihenfolge; wer wie viel gekauft hat, steht nicht fest.

400 Versuche für 100 Artikel bedeuten, dass der Bestand in jeder Reihenfolge bei 0 endet. Das ist auch eine Invariante, deshalb ist die Ausgabe bei jedem Lauf gleich.

Die Idee

Ein Timeout: Der Deadlock lässt den Test fehlschlagen

Zwei Lager buchen Ware gegenseitig um. Jede Umbuchung sperrt das Lager, aus dem sie nimmt, dann das, in das sie legt, in entgegengesetzter Reihenfolge. Das ist ein Deadlock: Jeder Thread hält einen Lock und wartet ewig auf den anderen.

Ein Test ohne Timeout schlägt nicht fehl: Er hängt. Die CI wartet eine Stunde, bis sie ihn abschießt, und niemand weiß, welcher Test es war. Ein Test mit Threads wartet also immer mit einem Timeout:

sofia = threading.Lock()
plovdiv = threading.Lock()


def transfer(source, target):
    with source:
        time.sleep(0.05)
        with target:
            pass  # die Ware umbuchen


def finishes_within(seconds, *parts):
    threads = [threading.Thread(target=part, daemon=True) for part in parts]
    for t in threads:
        t.start()
    end = time.monotonic() + seconds
    for t in threads:
        t.join(timeout=max(0, end - time.monotonic()))
    return not any(t.is_alive() for t in threads)


ok = finishes_within(1, lambda: transfer(sofia, plovdiv), lambda: transfer(plovdiv, sofia))
print("Test „Umbuchung zwischen den Lagern“: " + ("BESTANDEN" if ok else "nicht in 1 s fertig: DURCHGEFALLEN"))

Ausgabe

Test „Umbuchung zwischen den Lagern“: nicht in 1 s fertig: DURCHGEFALLEN

join(timeout=...) wartet höchstens so lange. Danach sagt is_alive(), ob der Thread noch läuft. daemon=True lässt Python auch dann enden, wenn die beiden Threads für immer gesperrt sind.

Genauer, für Neugierige

Test-Frameworks haben das Timeout fertig: pytest-timeout (@pytest.mark.timeout(1)) oder faulthandler.dump_traceback_later, das den Stack jedes Threads ausgibt, wenn der Test festhängt. Wie man in der Sprache einen Test schreibt, siehst du in der Lektion über Tests.

Es gibt auch Werkzeuge, die diese Fehler für dich suchen. Wir nennen sie hier nur, ohne sie zu benutzen:

• jcstress (Java): führt kleine Codestücke millionenfach in jeder Reihenfolge aus, die die JVM erlaubt, und zählt die Ergebnisse.
• ThreadSanitizer (C, C++, Go, Rust) und der Race Detector von Go (go test -race): beobachten jedes Lesen und Schreiben während des Tests und sagen „diese beiden Threads berühren dasselbe ohne Lock“, auch wenn das Ergebnis diesmal richtig war.
• Microsoft.Coyote (.NET): führt asynchronen und mehrfädigen Code unter einem eigenen Scheduler aus, probiert viele Reihenfolgen und kann die Reihenfolge wiederholen, die fehlgeschlagen ist.
• faulthandler (Python): gibt aus, wo jeder Thread steht, wenn ein Test festhängt.

Übung

Ein Stresstest für die Kasse

Löse die Aufgabe in deinem eigenen Editor.

Die Kasse des Sales hat sell(euros). Schreibe einen Stresstest:

• 20 Läufe. In jedem 4 Threads mit einer Barriere am Start; jeder verkauft 500-mal für je 2 €. Die erwartete Summe ist 4.000 €.
• Ein Lauf, in dem die Threads nicht innerhalb von 1 Sekunde fertig werden, zählt als Timeout, und der Test geht weiter.
• Gib am Ende aus, wie viele Läufe eine falsche Summe und wie viele ein Timeout hatten, und ob die Kasse besteht.

Ausgabe

Läufe: 20
Mit falscher Summe: 0
Mit Timeout: 0
Die Kasse: BESTANDEN

Weitere Aufgaben

• Zerstöre die Kasse. Entferne den Lock und sieh nach, wie viele Läufe rot werden. Führe es ein paar Mal aus: Die Zahl ändert sich.
• Eine erzwungene Verzahnung. Füge in sell eine Nahtstelle ein und schreibe einen Test, der ohne den Lock jedes Mal fehlschlägt und mit ihm besteht.
• Ein Deadlock in der Kasse. Füge einen zweiten Lock (die Kasse und das Lager) in entgegengesetzter Reihenfolge hinzu und prüfe, dass dein Test ein Timeout meldet, statt zu hängen.

Test

Teste dich

1. Der Test des Zählers ohne Lock besteht. Was bedeutet das?

Der Zähler ist auf diesem Rechner sicherFast nichts: In diesem Lauf haben sich die Threads nicht überschnittenDer Zähler ist sicher

2. Wozu dient die Barriere am Start des Stresstests?

Damit die Threads nacheinander laufenDamit der Test schneller wirdDamit alle Threads gemeinsam starten und sich überschneiden

3. Der Stresstest besteht 20 von 20 Mal. Ist der Code sicher?

Ja, wenn die Barriere an warNicht bewiesen: Der Test zeigt Fehler, er beweist nicht, dass es keine gibtJa, 20 Läufe reichen

4. Wie sorgst du dafür, dass ein Test bei derselben Race Condition jedes Mal fehlschlägt?

Latches an einer Nahtstelle im Code: Der erste Thread hält zwischen Lesen und Schreiben an, bis der zweite durch istThread.sleep mit verschiedenen PausenMehr Läufe

5. Warum muss ein Test mit Threads ein Timeout haben?

Damit er schneller wirdSonst lässt ein Deadlock den Test ewig hängen und hält die ganze CI anWeil Threads nicht von selbst enden

Prüfen

Lektion von vorn beginnen

Verwandte Lektionen

• Baut auf: Atomare Operationen und unveränderliche Objekte · Deadlock · Thread-Pool und Aufgaben · async ist kein Thread
• Aufgaben zum Thema: Nebenläufigkeit und Race Conditions