Python Testing & Debugging Beyond unittestAdvanced Python testing playbooks
Async & Concurrency

Testing Thread Safety with Barriers and Events

The difference between a concurrency test that is useful and one that is theatre is whether the threads were actually contending. A test that starts eight threads and asserts a total is measuring the scheduler's mood: on an idle machine they run almost serially, the assertion passes, and the bug ships. Coordination primitives fix that by making the contention a precondition rather than a hope.

This guide covers the four primitives worth knowing for test code — Barrier, Event, Semaphore and Condition — with the coordination shape each one fits and the failure modes each one introduces when misused.

Prerequisites

  • Python 3.9+; everything here is threading from the standard library.
  • pytest >= 8.0 with pytest-timeout, so a coordination mistake fails the test instead of hanging the suite.
  • The reasoning behind deterministic reproduction in testing threads and race conditions.
  • An invariant worth asserting — a total, a count, a uniqueness property — because none of this is useful without one.

Solution

Match the primitive to the shape of the coordination, and give every wait a timeout.

Python
import threading

import pytest


@pytest.mark.timeout(10)
def test_counter_is_atomic_under_contention():
    counter = Counter()
    workers = 8
    # Barrier: nobody proceeds until all eight have arrived, so the increment
    # loops overlap rather than running one after another.
    gate = threading.Barrier(workers, timeout=5)
    errors: list[BaseException] = []

    def worker():
        try:
            gate.wait()
            for _ in range(2_000):
                counter.increment()
        except BaseException as exc:      # noqa: BLE001 — re-raised below
            errors.append(exc)

    threads = [threading.Thread(target=worker) for _ in range(workers)]
    for thread in threads:
        thread.start()
    for thread in threads:
        thread.join(timeout=5)
        assert not thread.is_alive(), "a worker never finished"

    assert not errors, errors[0]
    assert counter.value == workers * 2_000
Python
import threading


def test_consumer_sees_every_item():
    produced = threading.Event()          # Event: one signal, many listeners
    queue: list[int] = []
    lock = threading.Lock()

    def producer():
        with lock:
            queue.extend(range(100))
        produced.set()                    # tell everyone the data is ready

    def consumer(seen):
        assert produced.wait(timeout=5), "producer never signalled"
        with lock:
            seen.extend(queue)

    seen_a, seen_b = [], []
    threads = [
        threading.Thread(target=producer),
        threading.Thread(target=consumer, args=(seen_a,)),
        threading.Thread(target=consumer, args=(seen_b,)),
    ]
    for thread in threads:
        thread.start()
    for thread in threads:
        thread.join(timeout=5)

    assert seen_a == seen_b == list(range(100))
Four coordination primitives and the shapes they fit Four cards. A barrier holds N peers until all arrive, then releases them together. An event lets one thread signal many waiters once. A semaphore bounds how many threads hold a resource at once. A condition wakes waiters when a predicate over shared state becomes true. Each card names the test scenario it suits. Pick the primitive from the coordination, not from habit Barrier(n) n peers arrive, all released together use for: contention between identical operations breaks if one crashes Event() one set(), many waiters released use for: "I am inside the window now" latching: stays set Semaphore(k) at most k holders at any moment use for: asserting a pool's concurrency limit counts, not signals Condition() wait until a predicate holds use for: state-dependent handoffs always loop on wait
Barriers coordinate peers, events coordinate a signaller and listeners, semaphores bound occupancy, and conditions wait on state. Using the wrong one produces coordination that works by accident.

Why this works

A Barrier blocks every caller of wait() until the configured number have arrived, then releases them all. In a test that means the increment loops start simultaneously rather than in start-up order, which is what produces genuine overlap on a multi-core machine. Without it, thread one typically finishes its loop before thread eight has been scheduled at all, and the lost-update window is never entered.

An Event is a latch: once set, every current and future wait() returns immediately. That asymmetry is exactly right for "the producer has reached the point I care about" and exactly wrong for a repeated handoff, where the second iteration finds the event already set. For repeated signalling, a Condition with an explicit predicate, or a queue.Queue, is the correct primitive.

Edge cases and failure modes

  • A barrier participant that raises before arriving. The barrier breaks and every other waiter gets BrokenBarrierError, masking the original exception. Wrap worker bodies so the real error is recorded and reported.
  • An event used for repeated handoffs. The second wait returns instantly because the event is still set. clear() between iterations reintroduces a race; use a Condition or a queue instead.
  • Condition.wait() without a loop. Spurious wakeups and other waiters mean the predicate must be re-checked. cv.wait_for(predicate, timeout=5) does this correctly and should be the default.
  • No timeouts. Every Barrier, Event.wait, acquire and join takes one. Without them a coordination mistake becomes a hung suite instead of a failing test — see failing fast with pytest-timeout.
  • Threading primitives inside coroutines. threading.Event().wait() blocks the event loop rather than yielding, so a coroutine waiting on one deadlocks everything. Use the asyncio equivalents.

Making worker failures visible

Every pattern above depends on one thing that threading does not give you: an exception in a worker reaching the test. A raw Thread prints the traceback to stderr and then discards it, so the test carries on and usually fails later with a message about the wrong thing.

Python
import concurrent.futures
import threading


def run_workers(target, count, *, timeout=10):
    """Run `count` copies of `target` concurrently and re-raise any failure."""
    gate = threading.Barrier(count, timeout=timeout)

    def wrapped():
        gate.wait()
        return target()

    with concurrent.futures.ThreadPoolExecutor(max_workers=count) as pool:
        futures = [pool.submit(wrapped) for _ in range(count)]
        # result() re-raises in the calling thread, so worker failures fail
        # the test with their own traceback rather than vanishing.
        return [future.result(timeout=timeout) for future in futures]
Python
def test_counter_is_atomic_under_contention():
    counter = Counter()
    run_workers(lambda: [counter.increment() for _ in range(2_000)], count=8)
    assert counter.value == 16_000

The helper folds the barrier, the joining and the exception propagation into four lines of test, which is what makes it realistic to write these tests routinely rather than only when a bug forces the issue.

Two details in the helper are deliberate. The barrier is sized to the worker count and shares the same timeout as the futures, so a worker that never arrives produces a BrokenBarrierError rather than a hang. And the executor is used as a context manager, whose __exit__ joins every worker before the block ends — which means an assertion failing after the block cannot tear down state a worker is still using. That ordering is the single most common cause of a concurrency test reporting a confusing secondary error instead of the assertion that actually failed.

Where a worker exception goes, with and without futures With a raw thread, an exception in the worker is printed to standard error and discarded, so the test continues and fails later on an unrelated assertion. With a ThreadPoolExecutor, the exception is stored on the future and re-raised when result is called, so the test fails with the worker's own traceback. A silent worker failure is worse than a loud one threading.Thread worker raises → traceback printed to stderr → discarded → test continues the test fails later, on an assertion that was never the problem ThreadPoolExecutor + future.result() worker raises → stored on the future → re-raised in the test thread the failure names the worker's own line
Where a raw thread is unavoidable, threading.excepthook collects worker exceptions into a list the test can assert is empty.

Asserting on a concurrency limit

A Semaphore is the natural tool for checking that a pool, a rate limiter or a worker group never exceeds its configured concurrency — but the assertion is usually better written with a counter than with the semaphore itself.

Python
import threading


def test_pool_never_exceeds_its_limit(pool):
    concurrent = 0
    peak = 0
    lock = threading.Lock()
    start = threading.Barrier(12, timeout=5)

    def task():
        nonlocal concurrent, peak
        start.wait()
        with lock:
            concurrent += 1
            peak = max(peak, concurrent)      # sampled inside the lock
        try:
            time.sleep(0.01)                  # hold the slot briefly
        finally:
            with lock:
                concurrent -= 1

    threads = [threading.Thread(target=pool.submit_and_wait, args=(task,))
               for _ in range(12)]
    for thread in threads:
        thread.start()
    for thread in threads:
        thread.join(timeout=10)

    assert peak <= pool.max_workers, f"peak concurrency {peak} exceeded the limit"
    assert peak == pool.max_workers, "the pool never reached its configured limit"

Both assertions matter and they pull in opposite directions. The first says the limit was respected; the second says the test actually exercised it, which guards against a false pass where the work finished so quickly that only one task ever ran at a time. A test that only asserts the upper bound passes trivially against a pool with a limit of one.

The brief sleep in the task body is the one place a sleep is defensible in a concurrency test: it is not waiting for a condition, it is deliberately occupying a slot so that occupancy is observable. Ten milliseconds against a five-second timeout is a comfortable margin, and the barrier ensures every task starts together so the peak is real.

Coordinating with a Condition

Condition is the primitive for "wait until the shared state satisfies a predicate", and wait_for makes it safe to use without the classic spurious-wakeup bug.

Python
import threading


def test_worker_processes_items_in_order(worker):
    cv = threading.Condition()
    processed: list[int] = []

    def on_item(item):
        with cv:
            processed.append(item)
            cv.notify_all()               # wake the test, which re-checks

    worker.on_item = on_item
    worker.submit_all([1, 2, 3])

    with cv:
        # wait_for loops on the predicate, so a spurious wakeup is harmless
        # and a partial result does not end the wait early.
        assert cv.wait_for(lambda: len(processed) == 3, timeout=5), processed

    assert processed == [1, 2, 3]

The assert on wait_for's return value is what turns a timeout into a readable failure — wait_for returns False rather than raising, so a bare call silently proceeds to the next assertion and fails there with a confusing message about list contents.

Condition.wait_for compared with a bare wait Two paths. A bare wait returns on any notification, including a spurious one or a partial result, so the test proceeds with incomplete state. wait_for re-evaluates the predicate after every wakeup and only returns true when it holds, returning false on timeout so the failure is attributable. Why the predicate has to be re-checked cv.wait(timeout=5) returns on the first notify — which may follow item 1 of 3 test proceeds with partial state and fails on the wrong assertion cv.wait_for(predicate, timeout=5) re-evaluates after every wakeup; returns True only when the predicate holds returns False on timeout, so the assertion names the real problem
The bare form is correct only inside a while not predicate() loop, which is precisely what wait_for is.

Frequently Asked Questions

When should I use a Barrier rather than an Event? Use a Barrier when N threads must all arrive before any proceeds, which is the right tool for testing contention between identical operations. Use an Event when one thread must signal and others must wait, which suits a producer telling the test it has reached a specific point. Barriers coordinate peers; events coordinate a signaller and its listeners.

What does BrokenBarrierError mean in a test? A participant failed to reach the barrier — it raised, it exited early, or it timed out — so the barrier is broken and every other waiter is released with that exception. It almost always means a worker crashed before arriving, and the underlying exception is the one to chase.

Do these primitives work for asyncio code too? asyncio has its own Event, Semaphore, Condition and Barrier that are not interchangeable with the threading ones. Never use a threading primitive inside a coroutine: its wait blocks the whole loop rather than yielding, which turns a coordination into a deadlock.

← Back to Testing Threads & Race Conditions