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 Worker verkaufen je 1.000-mal, und am Ende müssen 4.000 verkauft sein:

if (isMainThread) {
  const sold = new Int32Array(new SharedArrayBuffer(4));
  const done = [];
  for (let i = 0; i < 4; i++) {
    const w = new Worker(new URL(import.meta.url), { workerData: { sold } });
    done.push(once(w, "exit"));
  }
  await Promise.all(done);
  console.log("Verkauft (erwartet 4000):");
  console.log(sold[0]);
} else {
  const { sold } = workerData;
  for (let k = 0; k < 1_000; k++) {
    sold[0]++;
  }
}

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.

Worker starten langsam (jeder ist ein neues V8), deshalb hat der erste oft seine 1.000 Verkäufe beendet, bevor der nächste gestartet ist. Auf einem Rechner mit 4 Kernen unter der Last des Sales ist es anders.

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.

async function lostSales() {
  const sold = new Int32Array(new SharedArrayBuffer(4));
  const done = [];
  for (let i = 0; i < 4; i++) {
    const w = new Worker(new URL(import.meta.url), {
      workerData: { sold },
    });
    done.push(once(w, "exit"));
  }
  await Promise.all(done);
  return sold[0] !== 4_000;
}

function worker() {
  const { sold } = workerData;
  for (let k = 0; k < 1_000; k++) {
    sold[0]++;
  }
}

Schritt für Schritt, in Worten

1. Der Test startet 4 Worker, jeder verkauft 1.000-mal ohne Lock. Ohne Barriere startet jeder Worker, sobald er erzeugt ist.
2. 20 Läufe: In diesem Beispiel 6 von 20 rot. Ein Worker ist mit seinen 1.000 Verkäufen oft fertig, bevor der nächste gestartet ist.
3. Eine Barriere: Jeder Worker wartet am Start, bis alle vier angekommen sind. Dann laufen sie gemeinsam.
4. 20 Läufe mit Barriere: 20 von 20 rot. Derselbe kaputte Zähler, aber jetzt fängt der Test ihn jedes Mal ab.

Der Test mit einer Barriere:

// Eine Barriere für Worker: Jeder wartet, bis alle angekommen sind (höchstens 1 s).
function arrive(gate, parties) {
  Atomics.add(gate, 0, 1);
  Atomics.notify(gate, 0);
  const end = Date.now() + 1000;
  let n;
  while ((n = Atomics.load(gate, 0)) < parties && Date.now() < end) {
    Atomics.wait(gate, 0, n, 50);
  }
}

async function lostSales() {
  const sold = new Int32Array(new SharedArrayBuffer(4));
  const gate = new Int32Array(new SharedArrayBuffer(4));
  const done = [];
  for (let i = 0; i < 4; i++) {
    const w = new Worker(new URL(import.meta.url), {
      workerData: { sold, gate },
    });
    done.push(once(w, "exit"));
  }
  await Promise.all(done);
  return sold[0] !== 4_000;
}

function worker() {
  const { sold, gate } = workerData;
  arrive(gate, 4);
  for (let k = 0; k < 1_000; k++) {
    sold[0]++;
  }
}

if (isMainThread) {
  let failed = 0;
  for (let run = 0; run < 20; run++) {
    if (await lostSales()) {
      failed++;
    }
  }
  console.log("Läufe mit verlorenen Verkäufen (von 20):");
  console.log(failed);
} else {
  worker();
}

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 etwa 6 von 20, mit Barriere 20 von 20.

Die Barriere ist ein Zähler in einem SharedArrayBuffer: Jeder Worker erhöht ihn mit Atomics.add, weckt die anderen mit Atomics.notify und wartet mit Atomics.wait, bis der Zähler 4 erreicht. Atomics.wait ist in einem Worker erlaubt; im Hauptthread des Browsers nicht. Das Warten dauert höchstens 1 Sekunde.

Derselbe Test gegen den Zähler mit einem Lock:

async function lostSales() {
  const sold = new Int32Array(new SharedArrayBuffer(4));
  const gate = new Int32Array(new SharedArrayBuffer(4));
  const done = [];
  for (let i = 0; i < 4; i++) {
    const w = new Worker(new URL(import.meta.url), {
      workerData: { sold, gate },
    });
    done.push(once(w, "exit"));
  }
  await Promise.all(done);
  return sold[0] !== 4_000;
}

function worker() {
  const { sold, gate } = workerData;
  arrive(gate, 4);
  for (let k = 0; k < 1_000; k++) {
    Atomics.add(sold, 0, 1);
  }
}

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: Java hat eine fertige Barriere (CyclicBarrier), C# auch (Barrier), und Python ebenfalls (threading.Barrier). In Node schreiben wir sie selbst, weil Atomics die unterste Ebene ist.

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:

async function lostSales() {
  const sold = new Int32Array(new SharedArrayBuffer(4));
  for (let i = 0; i < 4; i++) {
    const w = new Worker(new URL(import.meta.url), {
      workerData: { sold },
    });
    await once(w, "exit");
  }
  return sold[0] !== 4_000;
}

function worker() {
  const { sold } = workerData;
  for (let k = 0; k < 1_000; k++) {
    sold[0]++;
  }
}

Ausgabe

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

Wähle die Zeile mit dem Fehler.

1async function lostSales() {2 const sold = new Int32Array(new SharedArrayBuffer(4));3 for (let i = 0; i < 4; i++) {4 const w = new Worker(new URL(import.meta.url), {5 workerData: { sold },6 });7 await once(w, "exit");8 }9 return sold[0] !== 4_000;10}1112function worker() {13 const { sold } = workerData;14 for (let k = 0; k < 1_000; k++) {15 sold[0]++;16 }17}

Prüfen

Probier es aus

Ergänze

Welche Zeile lässt die Threads gemeinsam starten?

async function lostSales() {
  const sold = new Int32Array(new SharedArrayBuffer(4));
  const gate = new Int32Array(new SharedArrayBuffer(4));
  const done = [];
  for (let i = 0; i < 4; i++) {
    const w = new Worker(new URL(import.meta.url), {
      workerData: { sold, gate },
    });
    done.push(once(w, "exit"));
  }
  await Promise.all(done);
  return sold[0] !== 4_000;
}

function worker() {
  const { sold, gate } = workerData;
  
  for (let k = 0; k < 1_000; k++) {
    sold[0]++;
  }
}

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 {
  constructor(count) {
    this.count = count;
    this.pause = async () => {};
  }

  async buy(who) {
    const left = this.count;
    await this.pause(who);   // eine Nahtstelle für den Test: Hier kann der Test den Thread anhalten
    this.count = left - 1;
  }
}

async function forcedTest(stock) {
  const read = latch();
  const go = latch();
  stock.pause = async (who) => {
    if (who === "first") {
      read.open();
      await go.wait(1000);
    }
  };
  const first = stock.buy("first");
  await read.wait(1000);   // first hat 10 gelesen und wartet
  const second = stock.buy("second");
  await Promise.race([second, sleep(200)]);   // second kauft alles oder wartet auf den Lock
  go.open();   // first schreibt
  await Promise.all([first, second]);
  const verdict = stock.count === 8 ? "BESTANDEN" : "DURCHGEFALLEN";
  return `Bestand: ${stock.count} (erwartet 8): ${verdict}`;
}

Ausgabe

Bestand: 9 (erwartet 8): DURCHGEFALLEN

In Node ist der gemeinsame Zustand in gewöhnlichem Code zwischen den awaits geteilt (die Lektion async ist kein Thread), deshalb ist die Nahtstelle await this.pause(who). Der Latch ist ein Promise: open() erfüllt es, und wait(ms) wartet darauf oder auf den Timer, je nachdem, was zuerst kommt.

Der Bestand ist 9 statt 8, bei jedem Lauf. Jetzt die Korrektur: Lesen und Schreiben ohne await dazwischen:

class Stock {
  constructor(count) {
    this.count = count;
    this.pause = async () => {};
  }

  async buy(who) {
    this.count--;   // kein await zwischen Lesen und Schreiben
    await this.pause(who);   // eine Nahtstelle für den Test: Hier kann der Test den Thread anhalten
  }
}

Ausgabe

Bestand: 8 (erwartet 8): BESTANDEN

Der Test ist derselbe. Der erste Kunde hat den Bestand schon verringert, bevor er an der Nahtstelle anhält, deshalb sieht der zweite 9. 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?

const stock = new Stock(10);
const allRead = countdown(3);
stock.pause = async () => {
  allRead.arrive();
  await allRead.done;
};
await Promise.all([stock.buy(), stock.buy(), stock.buy()]);
console.log(`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:

if (isMainThread) {
  const shop = new Int32Array(new SharedArrayBuffer(8));
  const bought = new Int32Array(new SharedArrayBuffer(4 * 8));
  const gate = new Int32Array(new SharedArrayBuffer(4));
  shop[0] = 100;
  const done = [];
  for (let me = 0; me < 8; me++) {
    const w = new Worker(new URL(import.meta.url), {
      workerData: { shop, bought, gate, me },
    });
    done.push(once(w, "exit"));
  }
  await Promise.all(done);
  const total = bought.reduce((a, b) => a + b, 0);
  console.log(`Bestand: ${shop[0]}, verkauft: ${shop[1]}`);
  console.log(check("Bestand + verkauft = 100", shop[0] + shop[1] === 100));
  console.log(check("Bestand >= 0", shop[0] >= 0));
  console.log(check("gekauft = verkauft", total === shop[1]));
} else {
  const { shop, bought, gate, me } = workerData;
  arrive(gate, 8);
  for (let k = 0; k < 50; k++) {
    if (tryBuy(shop)) {
      bought[me]++;
    }
  }
}

Ausgabe

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

Der Shop:

// shop[0] ist der Bestand, shop[1] ist die Zahl der Verkäufe.
function tryBuy(shop) {
  while (true) {
    const left = Atomics.load(shop, 0);
    if (left === 0) return false;
    if (Atomics.compareExchange(shop, 0, left, left - 1) === left) {
      Atomics.add(shop, 1, 1);
      return true;
    }
  }
}

Hier hat tryBuy keinen Lock: Atomics.compareExchange verringert den Bestand nur, wenn er noch der Wert ist, den wir gelesen haben (die Lektion Atomare Operationen und unveränderliche Objekte). Jeder Worker schreibt in seinen eigenen Platz in bought, dort braucht es also nichts.

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:

// Ein Lock in einem SharedArrayBuffer: 0 ist frei, 1 ist belegt.
function lock(locks, i) {
  while (Atomics.compareExchange(locks, i, 0, 1) !== 0) {
    Atomics.wait(locks, i, 1, 50);
  }
}

function transfer({ locks, from, to }) {
  lock(locks, from);
  Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, 50);   // eine Pause von 50 ms
  lock(locks, to);
  // die Ware umbuchen
}

async function finishesWithin(ms, jobs) {
  const workers = jobs.map((job) => new Worker(new URL(import.meta.url), { workerData: job }));
  const done = Promise.all(workers.map((w) => once(w, "exit"))).then(() => true);
  const ok = await Promise.race([done, sleep(ms).then(() => false)]);
  await Promise.all(workers.map((w) => w.terminate()));
  return ok;
}

if (isMainThread) {
  const locks = new Int32Array(new SharedArrayBuffer(8));
  const ok = await finishesWithin(1000, [
    { locks, from: 0, to: 1 },
    { locks, from: 1, to: 0 },
  ]);
  console.log("Test „Umbuchung zwischen den Lagern“: " + (ok ? "BESTANDEN" : "nicht in 1 s fertig: DURCHGEFALLEN"));
} else {
  transfer(workerData);
}

Ausgabe

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

Die Locks liegen in einem SharedArrayBuffer: Atomics.compareExchange nimmt den Lock, wenn er frei ist, und Atomics.wait wartet. Promise.race wartet entweder auf beide Worker oder auf den Timer. Dann stoppt terminate() beide, sonst würde das Programm nie enden.

Genauer, für Neugierige

Test-Frameworks haben das Timeout fertig: { timeout: 1000 } in node:test. 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 Atomics.add(register, 0, euros). Schreibe einen Stresstest:

• 20 Läufe. In jedem 4 Worker mit einer Barriere am Start; jeder verkauft 500-mal für je 2 €. Die erwartete Summe ist 4.000 €.
• Ein Lauf, in dem die Worker 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 (register[0] += euros) 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