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:

static int sold = 0;

static boolean lostSales() throws InterruptedException {
    sold = 0;
    List<Thread> threads = new ArrayList<>();
    for (int i = 0; i < 4; i++) {
        Thread t = new Thread(() -> {
            for (int k = 0; k < 1_000; k++) {
                sold++;
            }
        });
        t.start();
        threads.add(t);
    }
    for (Thread t : threads) {
        t.join();
    }
    return sold != 4_000;
}

lostSales();
System.out.println("Verkauft (erwartet 4000):");
System.out.println(sold);

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.

Einen Thread zu erzeugen dauert länger als 1.000 Erhöhungen. Der erste Thread ist oft fertig, bevor der letzte gestartet ist, und die Threads überschneiden sich gar nicht.

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.

static int sold = 0;

static boolean lostSales() throws InterruptedException {
    sold = 0;
    List<Thread> threads = new ArrayList<>();
    for (int i = 0; i < 4; i++) {
        Thread t = new Thread(() -> {
            for (int k = 0; k < 1_000; k++) {
                sold++;
            }
        });
        t.start();
        threads.add(t);
    }
    for (Thread t : threads) {
        t.join();
    }
    return sold != 4_000;
}

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: In diesem Beispiel 0 von 20 rot. Ein Thread ist mit seinen 1.000 Verkäufen oft fertig, bevor der nächste gestartet ist.
3. Eine Barriere: Jeder Thread wartet am Start, bis alle vier angekommen sind. Dann laufen sie gemeinsam.
4. 20 Läufe mit Barriere: 11 von 20 rot. Derselbe kaputte Zähler, aber jetzt fängt der Test ihn immer wieder ab.

Der Test mit einer Barriere:

static int sold = 0;

static boolean lostSales() throws InterruptedException {
    sold = 0;
    CyclicBarrier start = new CyclicBarrier(4);
    List<Thread> threads = new ArrayList<>();
    for (int i = 0; i < 4; i++) {
        Thread t = new Thread(() -> {
            waitAt(start);
            for (int k = 0; k < 1_000; k++) {
                sold++;
            }
        });
        t.start();
        threads.add(t);
    }
    for (Thread t : threads) {
        t.join();
    }
    return sold != 4_000;
}

// Wartet an der Barriere, bis alle da sind (höchstens 1 s).
static void waitAt(CyclicBarrier start) {
    try {
        start.await(1, TimeUnit.SECONDS);
    } catch (Exception e) {
        throw new IllegalStateException(e);
    }
}

int failed = 0;
for (int run = 0; run < 20; run++) {
    if (lostSales()) {
        failed++;
    }
}
System.out.println("Läufe mit verlorenen Verkäufen (von 20):");
System.out.println(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 ohne Barriere 0 von 20, mit Barriere etwa 11.

CyclicBarrier(4): await wartet, bis 4 Threads es aufgerufen haben, und lässt dann alle los. waitAt wartet höchstens 1 Sekunde: Ein Test darf nicht hängen, wenn ein Thread nie ankommt.

Derselbe Test gegen den Zähler mit einem Lock:

static int sold = 0;
static final Object lock = new Object();

static boolean lostSales() throws InterruptedException {
    sold = 0;
    CyclicBarrier start = new CyclicBarrier(4);
    List<Thread> threads = new ArrayList<>();
    for (int i = 0; i < 4; i++) {
        Thread t = new Thread(() -> {
            waitAt(start);
            for (int k = 0; k < 1_000; k++) {
                synchronized (lock) {
                    sold++;
                }
            }
        });
        t.start();
        threads.add(t);
    }
    for (Thread t : threads) {
        t.join();
    }
    return sold != 4_000;
}

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 C# ist die Barriere Barrier, in Python threading.Barrier, in Node schreibst du sie mit Atomics.wait und Atomics.notify. Statt einer Barriere ist auch ein Latch verbreitet: ein CountDownLatch(1), den der Hauptthread öffnet, wenn alle Threads gestartet sind.

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:

static int sold = 0;

static boolean lostSales() throws InterruptedException {
    sold = 0;
    for (int i = 0; i < 4; i++) {
        Thread t = new Thread(() -> {
            for (int k = 0; k < 1_000; k++) {
                sold++;
            }
        });
        t.start();
        t.join();
    }
    return sold != 4_000;
}

Ausgabe

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

Wähle die Zeile mit dem Fehler.

1static int sold = 0;23static boolean lostSales() throws InterruptedException {4 sold = 0;5 for (int i = 0; i < 4; i++) {6 Thread t = new Thread(() -> {7 for (int k = 0; k < 1_000; k++) {8 sold++;9 }10 });11 t.start();12 t.join();13 }14 return sold != 4_000;15}

Prüfen

Probier es aus

Ergänze

Welche Zeile lässt die Threads gemeinsam starten?

static int sold = 0;

static boolean lostSales() throws InterruptedException {
    sold = 0;
    CyclicBarrier start = new CyclicBarrier(4);
    List<Thread> threads = new ArrayList<>();
    for (int i = 0; i < 4; i++) {
        Thread t = new Thread(() -> {
            
            for (int k = 0; k < 1_000; k++) {
                sold++;
            }
        });
        t.start();
        threads.add(t);
    }
    for (Thread t : threads) {
        t.join();
    }
    return sold != 4_000;
}

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 {
    int count;
    Runnable pause = () -> {};

    Stock(int count) {
        this.count = count;
    }

    void buy() {
        int left = count;
        pause.run();   // eine Nahtstelle für den Test: Hier kann der Test den Thread anhalten
        count = left - 1;
    }
}

static String forcedTest(Stock stock) throws InterruptedException {
    CountDownLatch read = new CountDownLatch(1);
    CountDownLatch go = new CountDownLatch(1);
    stock.pause = () -> {
        if (Thread.currentThread().getName().equals("first")) {
            read.countDown();
            waitFor(go);
        }
    };
    Thread first = new Thread(stock::buy, "first");
    Thread second = new Thread(stock::buy, "second");
    first.start();
    waitFor(read);      // first hat 10 gelesen und wartet
    second.start();
    second.join(200);   // second kauft alles oder wartet auf den Lock
    go.countDown();     // first schreibt
    first.join(1000);
    second.join(1000);
    String verdict = stock.count == 8 ? "BESTANDEN" : "DURCHGEFALLEN";
    return "Bestand: " + stock.count + " (erwartet 8): " + verdict;
}

static void waitFor(CountDownLatch latch) {
    try {
        latch.await(1, TimeUnit.SECONDS);
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }
}

Ausgabe

Bestand: 9 (erwartet 8): DURCHGEFALLEN

Die Nahtstelle ist das Feld pause. Der Test legt Code hinein, der nur den Thread namens first anhält: Er öffnet den Latch read und wartet auf den Latch go. Ein CountDownLatch(1) ist ein Latch: countDown() öffnet ihn, await wartet, bis er offen ist. waitFor wartet höchstens 1 Sekunde.

Der Bestand ist 9 statt 8, bei jedem Lauf. Jetzt die Korrektur: synchronized auf buy:

class Stock {
    int count;
    Runnable pause = () -> {};

    Stock(int count) {
        this.count = count;
    }

    synchronized void buy() {
        int left = count;
        pause.run();   // eine Nahtstelle für den Test: Hier kann der Test den Thread anhalten
        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(200) 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 = new Stock(10);
CountDownLatch allRead = new CountDownLatch(3);
stock.pause = () -> {
    allRead.countDown();
    waitFor(allRead);
};
List<Thread> buyers = new ArrayList<>();
for (int i = 0; i < 3; i++) {
    Thread t = new Thread(stock::buy);
    t.start();
    buyers.add(t);
}
for (Thread t : buyers) {
    t.join();
}
System.out.println("Bestand: " + stock.count);

Antwort

Bestand: 7

Bestand: 8

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 = new Shop(100);
int[] bought = new int[8];
CyclicBarrier start = new CyclicBarrier(8);
List<Thread> buyers = new ArrayList<>();
for (int i = 0; i < 8; i++) {
    int me = i;
    Thread t = new Thread(() -> {
        waitAt(start);
        for (int k = 0; k < 50; k++) {
            if (shop.tryBuy()) {
                bought[me]++;
            }
        }
    });
    t.start();
    buyers.add(t);
}
for (Thread t : buyers) {
    t.join();
}
int total = Arrays.stream(bought).sum();
System.out.println("Bestand: " + shop.stock() + ", verkauft: " + shop.sold());
System.out.println(check("Bestand + verkauft = 100", shop.stock() + shop.sold() == 100));
System.out.println(check("Bestand >= 0", shop.stock() >= 0));
System.out.println(check("gekauft = verkauft", total == shop.sold()));

Ausgabe

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

Der Shop:

class Shop {
    private int stock;
    private int sold = 0;

    Shop(int stock) {
        this.stock = stock;
    }

    synchronized boolean tryBuy() {
        if (stock == 0) {
            return false;
        }
        stock--;
        sold++;
        return true;
    }

    synchronized int stock() {
        return stock;
    }

    synchronized int sold() {
        return sold;
    }
}

Jeder Thread schreibt in seinen eigenen Platz bought[me], das Array 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:

static final Object sofia = new Object();
static final Object plovdiv = new Object();

static void transfer(Object from, Object to) {
    synchronized (from) {
        sleep(50);
        synchronized (to) {
            // die Ware umbuchen
        }
    }
}

static void sleep(int ms) {
    try {
        Thread.sleep(ms);
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }
}

static boolean finishesWithin(long ms, Runnable... parts) throws InterruptedException {
    List<Thread> threads = new ArrayList<>();
    for (Runnable part : parts) {
        Thread t = new Thread(part);
        t.setDaemon(true);
        t.start();
        threads.add(t);
    }
    long end = System.currentTimeMillis() + ms;
    for (Thread t : threads) {
        t.join(Math.max(1, end - System.currentTimeMillis()));
    }
    return threads.stream().noneMatch(Thread::isAlive);
}

boolean ok = finishesWithin(1000,
    () -> transfer(sofia, plovdiv),
    () -> transfer(plovdiv, sofia));
System.out.println("Test „Umbuchung zwischen den Lagern“: " + (ok ? "BESTANDEN" : "nicht in 1 s fertig: DURCHGEFALLEN"));

Ausgabe

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

join(ms) wartet höchstens ms. Danach sagt isAlive(), ob der Thread noch läuft. setDaemon(true) lässt die JVM auch dann enden, wenn die beiden Threads für immer gesperrt sind.

Genauer, für Neugierige

Test-Frameworks haben das Timeout fertig: @Timeout(1) oder assertTimeoutPreemptively in JUnit 5. 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(int 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