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 threads sell 1000 times each, and at the end the number sold must be 4000:

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("Sold (expected 4000):")
print(sell_once())

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.

In Python this is especially easy: because of the GIL, threads switch about every 5 ms, and self.sold += 1 is a few instructions. The chance that the switch lands exactly between the read and the write is small. Small, but not zero: on a server with many requests an hour, it happens.

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.

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

    def sell(self):
        left = self.sold
        time.sleep(0)  # another thread gets in here
        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

Step by step, in words

1. The test starts 4 threads, each sells 1000 times without a lock. Without a barrier, each thread starts as soon as it is created.
2. 20 runs: 20 of 20 red. Here time.sleep(0) opens a wide gap, so it fails even without a barrier. In real code the gap is narrow.
3. A barrier: each thread waits at the start until all four have arrived. Then they run together.
4. With the barrier: 20 of 20 red, as without it: sleep(0) keeps the gap wide. With a narrow gap the barrier decides.

The test with a barrier:

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

    def sell(self):
        left = self.sold
        time.sleep(0)  # another thread gets in here
        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("Runs that lost sales (of 20):")
print(failed)

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 it is 20 of 20 both with and without a barrier.

threading.Barrier(4, timeout=1): wait() waits until 4 threads have called it, then lets them all go. If one does not arrive within 1 second, it raises BrokenBarrierError instead of hanging.

Here sell releases the GIL with time.sleep(0) between the read and the write (as in the lesson on race conditions). The gap is so wide that the test fails even without a barrier. In real code the gap is a few instructions, and there the barrier and the many runs decide whether you see it at all.

The same test against the counter with a 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)  # another thread gets in here
            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

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: in Java the barrier is CyclicBarrier, in C# Barrier, in Node you write it with Atomics.wait and Atomics.notify. Instead of a barrier, a threading.Event that the main thread sets is also common.

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:

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

    def sell(self):
        left = self.sold
        time.sleep(0)  # another thread gets in here
        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

Output

Runs that lost sales: 0 of 20

Pick the line with the bug.

1class Counter:2 def __init__(self):3 self.sold = 045 def sell(self):6 left = self.sold7 time.sleep(0) # another thread gets in here8 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

Check

Try it

Fill in

Which line makes the threads start together?

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

    def sell(self):
        left = self.sold
        time.sleep(0)  # another thread gets in here
        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

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:
    def __init__(self, count):
        self.count = count
        self.pause = lambda: None

    def buy(self):
        left = self.count
        self.pause()  # a seam for the test: here the test can stop the thread
        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 has read 10 and waits
    second.start()
    second.join(timeout=0.2)  # second buys to the end or waits for the lock
    go.set()  # first writes
    first.join(timeout=1)
    second.join(timeout=1)
    verdict = "PASS" if stock.count == 8 else "FAIL"
    return f"Stock: {stock.count} (expected 8): {verdict}"

Output

Stock: 9 (expected 8): FAIL

The seam is the field pause. The test puts a function in it that stops only the thread named first: it sets read and waits for go. A threading.Event is a latch: set() opens it, wait(timeout=1) waits for it to open, at most 1 second.

The stock is 9 instead of 8, on every run. Now the fix: with self.lock: around the read and the write:

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()  # a seam for the test: here the test can stop the thread
            self.count = left - 1

Output

Stock: 8 (expected 8): PASS

The test is the same. The first buyer holds the lock while it is stopped at the seam. The second one waits for the lock, so second.join(timeout=0.2) runs out before it has bought. Only when the first one writes 9 and releases the lock does the second one read 9 and write 8. 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?

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"Stock: {stock.count}")

Answer

Stock: 8

Stock: 9

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:

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"Stock: {shop.stock}, sold: {shop.sold}")
print(check("stock + sold = 100", shop.stock + shop.sold == 100))
print(check("stock >= 0", shop.stock >= 0))
print(check("bought = sold", sum(bought) == shop.sold))

Output

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

The 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

Each thread writes to its own slot bought[me], so the list needs no lock. The total is right in every order; who bought how much is not fixed.

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:

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


def transfer(source, target):
    with source:
        time.sleep(0.05)
        with target:
            pass  # move the goods


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

Output

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

join(timeout=...) waits at most that long. Then is_alive() tells whether the thread is still running. daemon=True lets Python stop even though the two threads are locked forever.

More precisely, if you are curious

Test frameworks have the timeout ready: pytest-timeout (@pytest.mark.timeout(1)) or faulthandler.dump_traceback_later, which prints the stack of every thread when the test is stuck. 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 sell(euros). Write a stress test:

• 20 runs. In each, 4 threads with a barrier at the start; each one sells 500 times for €2. The expected total is €4000.
• A run in which the threads 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 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