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:

int sold = 0;

bool LostSales()
{
    sold = 0;
    var threads = new List<Thread>();
    for (int i = 0; i < 4; i++)
    {
        var t = new Thread(() =>
        {
            for (int k = 0; k < 1_000; k++)
            {
                sold++;
            }
        });
        t.Start();
        threads.Add(t);
    }
    foreach (var t in threads)
    {
        t.Join();
    }
    return sold != 4_000;
}

LostSales();
Console.WriteLine("Verkauft (erwartet 4000):");
Console.WriteLine(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.

int sold = 0;

bool LostSales()
{
    sold = 0;
    var threads = new List<Thread>();
    for (int i = 0; i < 4; i++)
    {
        var t = new Thread(() =>
        {
            for (int k = 0; k < 1_000; k++)
            {
                sold++;
            }
        });
        t.Start();
        threads.Add(t);
    }
    foreach (var t in 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: 13 von 20 rot. Derselbe kaputte Zähler, aber jetzt fängt der Test ihn immer wieder ab.

Der Test mit einer Barriere:

int sold = 0;

bool LostSales()
{
    sold = 0;
    var start = new Barrier(4);
    var threads = new List<Thread>();
    for (int i = 0; i < 4; i++)
    {
        var t = new Thread(() =>
        {
            start.SignalAndWait(TimeSpan.FromSeconds(1));
            for (int k = 0; k < 1_000; k++)
            {
                sold++;
            }
        });
        t.Start();
        threads.Add(t);
    }
    foreach (var t in threads)
    {
        t.Join();
    }
    return sold != 4_000;
}

int failed = 0;
for (int run = 0; run < 20; run++)
{
    if (LostSales())
    {
        failed++;
    }
}
Console.WriteLine("Läufe mit verlorenen Verkäufen (von 20):");
Console.WriteLine(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 (im Container, in dem die Programme geprüft werden) sind es ohne Barriere 0 von 20, mit Barriere etwa 13.

Barrier(4): SignalAndWait wartet, bis 4 Threads es aufgerufen haben, und lässt dann alle los. Das Timeout von 1 Sekunde verhindert, dass der Test hängt, wenn ein Thread nie ankommt.

Derselbe Test gegen den Zähler mit einem Lock:

int sold = 0;
object sync = new object();

bool LostSales()
{
    sold = 0;
    var start = new Barrier(4);
    var threads = new List<Thread>();
    for (int i = 0; i < 4; i++)
    {
        var t = new Thread(() =>
        {
            start.SignalAndWait(TimeSpan.FromSeconds(1));
            for (int k = 0; k < 1_000; k++)
            {
                lock (sync)
                {
                    sold++;
                }
            }
        });
        t.Start();
        threads.Add(t);
    }
    foreach (var t in 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 Java ist die Barriere CyclicBarrier (oder ein CountDownLatch-Latch), in Python threading.Barrier, in Node schreibst du sie mit Atomics.wait und Atomics.notify. Statt einer Barriere nimmt C#-Code oft einen Latch: ein ManualResetEventSlim, das der Hauptthread öffnet.

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:

int sold = 0;

bool LostSales()
{
    sold = 0;
    for (int i = 0; i < 4; i++)
    {
        var 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.

1int sold = 0;23bool LostSales()4{5 sold = 0;6 for (int i = 0; i < 4; i++)7 {8 var t = new Thread(() =>9 {10 for (int k = 0; k < 1_000; k++)11 {12 sold++;13 }14 });15 t.Start();16 t.Join();17 }18 return sold != 4_000;19}

Prüfen

Probier es aus

Ergänze

Welche Zeile lässt die Threads gemeinsam starten?

int sold = 0;

bool LostSales()
{
    sold = 0;
    var start = new Barrier(4);
    var threads = new List<Thread>();
    for (int i = 0; i < 4; i++)
    {
        var t = new Thread(() =>
        {
            
            for (int k = 0; k < 1_000; k++)
            {
                sold++;
            }
        });
        t.Start();
        threads.Add(t);
    }
    foreach (var t in 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.

string ForcedTest(Stock stock)
{
    var read = new ManualResetEventSlim();
    var go = new ManualResetEventSlim();
    stock.Pause = () =>
    {
        if (Thread.CurrentThread.Name == "first")
        {
            read.Set();
            go.Wait(TimeSpan.FromSeconds(1));
        }
    };
    var first = new Thread(stock.Buy) { Name = "first" };
    var second = new Thread(stock.Buy) { Name = "second" };
    first.Start();
    read.Wait(TimeSpan.FromSeconds(1));   // first hat 10 gelesen und wartet
    second.Start();
    second.Join(200);   // second kauft alles oder wartet auf den Lock
    go.Set();           // first schreibt
    first.Join(1000);
    second.Join(1000);
    string verdict = stock.Count == 8 ? "BESTANDEN" : "DURCHGEFALLEN";
    return $"Bestand: {stock.Count} (erwartet 8): {verdict}";
}

class Stock
{
    public int Count;
    public Action Pause = () => { };

    public Stock(int count) => Count = count;

    public void Buy()
    {
        int left = Count;
        Pause();   // eine Nahtstelle für den Test: Hier kann der Test den Thread anhalten
        Count = left - 1;
    }
}

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 read und wartet auf go. Ein ManualResetEventSlim ist ein Latch: Set() öffnet ihn, Wait wartet, bis er offen ist, hier höchstens 1 Sekunde.

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

class Stock
{
    public int Count;
    public Action Pause = () => { };
    private readonly object sync = new object();

    public Stock(int count) => Count = count;

    public void Buy()
    {
        lock (sync)
        {
            int left = Count;
            Pause();   // 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?

var stock = new Stock(10);
var allRead = new CountdownEvent(3);
stock.Pause = () =>
{
    allRead.Signal();
    allRead.Wait(TimeSpan.FromSeconds(1));
};
var buyers = new List<Thread>();
for (int i = 0; i < 3; i++)
{
    var t = new Thread(stock.Buy);
    t.Start();
    buyers.Add(t);
}
foreach (var t in buyers)
{
    t.Join();
}
Console.WriteLine($"Bestand: {stock.Count}");

Antwort

Bestand: 8

Bestand: 9

Bestand: 7

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:

var shop = new Shop(100);
int[] bought = new int[8];
var start = new Barrier(8);
var buyers = new List<Thread>();
for (int i = 0; i < 8; i++)
{
    int me = i;
    var t = new Thread(() =>
    {
        start.SignalAndWait(TimeSpan.FromSeconds(1));
        for (int k = 0; k < 50; k++)
        {
            if (shop.TryBuy())
            {
                bought[me]++;
            }
        }
    });
    t.Start();
    buyers.Add(t);
}
foreach (var t in buyers)
{
    t.Join();
}
Console.WriteLine($"Bestand: {shop.Stock}, verkauft: {shop.Sold}");
Console.WriteLine(Check("Bestand + verkauft = 100", shop.Stock + shop.Sold == 100));
Console.WriteLine(Check("Bestand >= 0", shop.Stock >= 0));
Console.WriteLine(Check("gekauft = verkauft", bought.Sum() == 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;
    private readonly object sync = new object();

    public Shop(int stock) => this.stock = stock;

    public bool TryBuy()
    {
        lock (sync)
        {
            if (stock == 0) return false;
            stock--;
            sold++;
            return true;
        }
    }

    public int Stock { get { lock (sync) return stock; } }
    public int Sold { get { lock (sync) 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:

object sofia = new object();
object plovdiv = new object();

void Transfer(object from, object to)
{
    lock (from)
    {
        Thread.Sleep(50);
        lock (to)
        {
            // die Ware umbuchen
        }
    }
}

bool FinishesWithin(int ms, params ThreadStart[] parts)
{
    var threads = parts.Select(part => new Thread(part) { IsBackground = true }).ToList();
    threads.ForEach(t => t.Start());
    var end = DateTime.Now.AddMilliseconds(ms);
    foreach (var t in threads)
    {
        var left = end - DateTime.Now;
        t.Join(left > TimeSpan.Zero ? left : TimeSpan.FromMilliseconds(1));
    }
    return threads.All(t => !t.IsAlive);
}

bool ok = FinishesWithin(1000,
    () => Transfer(sofia, plovdiv),
    () => Transfer(plovdiv, sofia));
Console.WriteLine("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(TimeSpan) wartet höchstens so lange und gibt zurück, ob der Thread fertig wurde. IsBackground = true lässt das Programm auch dann enden, wenn die beiden Threads für immer gesperrt sind.

Genauer, für Neugierige

Test-Frameworks haben das Timeout fertig: [Fact(Timeout = 1000)] in xUnit (für async-Tests) oder [Timeout] in NUnit. 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