Testing concurrent code

Code in:JavaC#PythonJavaScript

In short: Concurrent code is tested with a stress test of many runs, a start barrier, forced interleavings and an invariant check, because one green run proves nothing.

Problem

It passes on my machine

The sales counter from the lesson on race conditions has no lock. A colleague writes a test: four Workers sell 1000 times each, and at the end the number sold must be 4000:

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("Sold (expected 4000):");
  console.log(sold[0]);
} else {
  const { sold } = workerData;
  for (let k = 0; k < 1_000; k++) {
    sold[0]++;
  }
}

Output

Sold (expected 4000):
4000

For them the test passes. It passes in CI too. The code goes into the sale, and in the first hour the shop has sold more items than the counter shows.

Workers start slowly (each one is a new V8), so the first one has often finished its 1000 sales before the next one has started. On a 4-core machine under the load of the sale, it is different.

One green test of concurrent code proves almost nothing: in this run the threads just did not meet exactly between the read and the write. How do you write a test that catches such a bug?

The idea

A stress test: many runs and a barrier

Three things make the bug visible:

• Many runs. The test is not one run but 20 (or 1000 in a nightly CI job). You count how many are red.
• Many threads at once. More threads than cores, so that they switch in the middle of the work.
• A barrier at the start. Each thread waits until all have arrived, and they start together. That way they overlap the whole time, not only at the end.

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]++;
  }
}

Step by step, in words

1. The test starts 4 Workers, each sells 1000 times without a lock. Without a barrier, each Worker starts as soon as it is created.
2. 20 runs: in this example 6 of 20 red. A Worker often finishes its 1000 sales before the next one has started.
3. A barrier: each Worker waits at the start until all four have arrived. Then they run together.
4. 20 runs with the barrier: 20 of 20 red. The same broken counter, but now the test catches it every time.

The test with a barrier:

// A barrier for Workers: each one waits until all have arrived (at most 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("Runs that lost sales (of 20):");
  console.log(failed);
} else {
  worker();
}

Output

Runs that lost sales (of 20):
20

The output is "at most 20": how many runs turn red depends on the machine. On my machine without a barrier it is about 6 of 20, with a barrier 20 of 20.

The barrier is a counter in a SharedArrayBuffer: each Worker increases it with Atomics.add, wakes the others with Atomics.notify and waits with Atomics.wait until the counter reaches 4. Atomics.wait is allowed in a Worker; in the browser's main thread it is not. The wait is at most 1 second.

The same test against the counter with a 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);
  }
}

Output

Runs that lost sales: 0 of 20

More precisely, if you are curious

A stress test finds bugs, but it does not prove there are none. "0 of 20" means "we did not see it this time". That is why in CI such tests run thousands of times, on machines with many cores, and code review asks separately: what is shared and which lock guards it.

How it is in the other languages: Java has a ready barrier (CyclicBarrier), so does C# (Barrier), and so does Python (threading.Barrier). In Node we write it ourselves, because Atomics is the lowest level.

Try it

Find the bug

Another colleague wrote a stress test without a barrier. It passes 20 of 20 times, on every run, on every machine, even though the counter has no lock:

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]++;
  }
}

Output

Runs that lost sales: 0 of 20

Pick the line with the bug.

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}

Check

Try it

Fill in

Which line makes the threads start together?

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]++;
  }
}

Check

The idea

Force the interleaving

A stress test finds the bug sometimes. For a test that fails every time, the test chooses the order of the threads itself. For that the code has a seam: a place where the test can stop the thread. In the real program the seam does nothing.

The scenario is the bug from the lesson on race conditions: 10 items are left, two buyers buy one each.

1. The first buyer reads 10 and stops at the seam.
2. The second buyer goes all the way: reads 10, writes 9.
3. The first one goes on and writes its 10 − 1 = 9.

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

  async buy(who) {
    const left = this.count;
    await this.pause(who);   // a seam for the test: here the test can stop the thread
    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 has read 10 and waits
  const second = stock.buy("second");
  await Promise.race([second, sleep(200)]);   // second buys to the end or waits for the lock
  go.open();   // first writes
  await Promise.all([first, second]);
  const verdict = stock.count === 8 ? "PASS" : "FAIL";
  return `Stock: ${stock.count} (expected 8): ${verdict}`;
}

Output

Stock: 9 (expected 8): FAIL

In Node the shared state in ordinary code is shared between awaits (the async lesson), so the seam is await this.pause(who). The latch is a promise: open() resolves it, and wait(ms) waits for it or for the timer, whichever comes first.

The stock is 9 instead of 8, on every run. Now the fix: the read and the write with no await between them:

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

  async buy(who) {
    this.count--;   // no await between the read and the write
    await this.pause(who);   // a seam for the test: here the test can stop the thread
  }
}

Output

Stock: 8 (expected 8): PASS

The test is the same. The first buyer has already decreased the stock before stopping at the seam, so the second one sees 9. A test that failed every time before the fix and passes every time after it is exactly what you want in CI.

Try it

Three at the seam

The same Stock without a lock. Now the seam waits until all three buyers have read the stock:

What will the program print?

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(`Stock: ${stock.count}`);

Answer

Stock: 9

Stock: 8

Stock: 7

Check

The idea

Check the invariant, not the order

With 8 buyers, who bought what depends on the run. A test that expects "the first buyer bought 13" fails even with correct code. So the test checks an invariant: something that must be true in every order.

For the shop it is: the stock plus the number sold is what there was at the start, the stock is not below zero, and the buyers' purchases added up equal the number sold. 8 buyers make 50 attempts each for 100 items:

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(`Stock: ${shop[0]}, sold: ${shop[1]}`);
  console.log(check("stock + sold = 100", shop[0] + shop[1] === 100));
  console.log(check("stock >= 0", shop[0] >= 0));
  console.log(check("bought = sold", 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]++;
    }
  }
}

Output

Stock: 0, sold: 100
stock + sold = 100: PASS
stock >= 0: PASS
bought = sold: PASS

The shop:

// shop[0] is the stock, shop[1] is the number sold.
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;
    }
  }
}

Here tryBuy has no lock: Atomics.compareExchange decreases the stock only if it is still the value we read (the lesson on atomic operations). Each Worker writes to its own slot in bought, so nothing is needed there.

400 attempts for 100 items means the stock ends at 0 in every order. That is an invariant too, which is why the output is the same on every run.

The idea

A timeout: the deadlock fails the test

Two warehouses transfer goods to each other. Each transfer locks the warehouse it takes from, then the one it puts into, in opposite order. That is a deadlock: each thread holds one lock and waits for the other forever.

A test without a timeout does not fail: it hangs. CI waits an hour until it kills it, and nobody knows which test it was. So a test with threads always waits with a timeout:

// A lock in a SharedArrayBuffer: 0 is free, 1 is taken.
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);   // a 50 ms pause
  lock(locks, to);
  // move the goods
}

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 "transfer between warehouses": ' + (ok ? "PASS" : "did not finish in 1 s: FAIL"));
} else {
  transfer(workerData);
}

Output

Test "transfer between warehouses": did not finish in 1 s: FAIL

The locks are in a SharedArrayBuffer: Atomics.compareExchange takes the lock if it is free, and Atomics.wait waits. Promise.race waits either for both Workers or for the timer. Then terminate() stops both, otherwise the program would never end.

More precisely, if you are curious

Test frameworks have the timeout ready: { timeout: 1000 } in node:test. How to write a test in the language, see the lesson on testing.

There are also tools that look for these bugs for you. Here we only name them, without using them:

• jcstress (Java): runs small pieces of code millions of times in every order the JVM allows and counts the results.
• ThreadSanitizer (C, C++, Go, Rust) and Go's race detector (go test -race): watch every read and write during the test and say "these two threads touch the same thing without a lock", even if this time the result was right.
• Microsoft.Coyote (.NET): runs async and multithreaded code under its own scheduler, tries many orders, and can replay the order that failed.
• faulthandler (Python): when a test is stuck, prints where every thread is.

Exercise

A stress test for the register

Solve it in your own editor.

The sale's register has Atomics.add(register, 0, euros). Write a stress test:

• 20 runs. In each, 4 Workers with a barrier at the start; each one sells 500 times for €2. The expected total is €4000.
• A run in which the Workers do not finish within 1 second counts as a timeout, and the test goes on.
• At the end print how many runs had a wrong total and how many a timeout, and whether the register passes.

Output

Runs: 20
With a wrong total: 0
With a timeout: 0
The register: PASS

More tasks

• Break the register. Remove the lock (register[0] += euros) and see how many runs turn red. Run it a few times: the number changes.
• A forced interleaving. Add a seam in sell and write a test that fails every time without the lock and passes with it.
• A deadlock in the register. Add a second lock (the register and the warehouse) in opposite order and check that your test reports a timeout instead of hanging.

Test

Check yourself

1. The test of the counter without a lock passes. What does that mean?

The counter is safe on this machineThe counter is safeAlmost nothing: in this run the threads did not overlap

2. What is the barrier at the start of the stress test for?

To make the test fasterSo that all threads start together and overlapSo that the threads run one after another

3. The stress test passes 20 of 20 times. Is the code safe?

Yes, if the barrier was onYes, 20 runs are enoughNot proven: the test shows bugs, it does not prove there are none

4. How do you make a test fail every time on the same race condition?

More runsLatches in a seam of the code: the first thread stops between the read and the write until the second one is throughThread.sleep with different pauses

5. Why must a test with threads have a timeout?

To make it fasterOtherwise a deadlock makes the test hang forever and stops the whole CIBecause threads do not finish on their own

Check

Start the lesson over

Related lessons

• Builds on: Atomic operations and immutable objects · Deadlock · A pool of threads and tasks · async is not a thread
• Tasks on this topic: Concurrency and race conditions