Testing Threads & Race Conditions

Concurrency bugs have a characteristic lifecycle: they are found in production, cannot be reproduced locally, get a defensive lock added on the suspicion that it might help, and reappear eighteen months later. The step that breaks the cycle is a test that fails every single run while the bug is present. That is achievable far more often than teams assume, because most Python races have a window an interleaving can be forced into rather than waited for.

Prerequisites

• Python 3.9+; threading.Barrier and concurrent.futures are standard library, and nothing here needs a third-party plugin.
• pytest >= 8.0, plus pytest-timeout so a deadlocked test fails rather than hanging — see failing fast with pytest-timeout.
• Familiarity with the difference between a lock, an event and a barrier, and with why time.sleep is not a synchronisation primitive.
• For async code, the cooperative equivalent is in testing async and concurrent Python — the techniques differ because the scheduler differs.

Core concept: races have windows, and windows can be held open

A race condition is a window between two operations that the code assumes are atomic. The bug is not that threads run concurrently; it is that the invariant is false for a few nanoseconds and something observed it during that time. Testing for it means holding the window open deliberately.

There are two ways to hold it open, and they suit different code. The first is a barrier: every thread blocks until all of them have arrived, so they enter the contended region simultaneously rather than by luck. The second is a seam: a hook inside the window that the test can block on, which lets one thread be frozen mid-sequence while another runs to completion. Barriers need no production change and catch races between identical operations; seams need a small injectable callback and catch races between different operations.

The GIL does not save you

The most persistent misconception about Python concurrency is that the global interpreter lock makes shared state safe. It makes bytecode atomic, which is a much weaker guarantee than it sounds. counter += 1 compiles to a load, a binary add and a store, and since Python 3.10 the interpreter may switch threads between any two bytecodes at a granularity controlled by sys.setswitchinterval. Three bytecodes is three opportunities for another thread to interleave.

import dis


def increment(counter):
    counter.value += 1


dis.dis(increment)

  2           0 LOAD_FAST                0 (counter)
              2 DUP_TOP
              4 LOAD_ATTR                0 (value)     ← read
              6 LOAD_CONST               1 (1)
              8 INPLACE_ADD                             ← modify
             10 ROT_TWO
             12 STORE_ATTR               0 (value)     ← write

The read at offset 4 and the write at offset 12 are separated by four bytecodes. Every check-then-act pattern has the same shape: if key not in cache: cache[key] = compute(key), if self._conn is None: self._conn = connect(), if os.path.exists(p): os.remove(p). Under free-threaded builds (PEP 703, available as an official variant from 3.13) the window widens further because there is no interpreter lock serialising the bytecodes at all — code that was accidentally safe becomes reliably unsafe, which makes these tests worth writing now rather than at the migration.

Two practical consequences follow. First, sys.setswitchinterval(0.000001) in a conftest.py makes the interpreter switch threads far more aggressively during the test session, which widens every window in the code under test at no cost to correctness. It is a blunt instrument and it slows the suite slightly, but for a module full of concurrency tests it turns several probabilistic failures into reliable ones. Second, any invariant that spans more than one attribute access needs a lock regardless of how short the code looks — brevity is not atomicity.

Step-by-step implementation

1. Write the failing test with a barrier

import threading

import pytest


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

    def increment(self):
        current = self.value          # read
        current += 1                  # modify
        self.value = current          # write — the window is between read and here


@pytest.mark.timeout(5)               # a deadlock fails; it does not hang
def test_increment_is_atomic():
    counter = Counter()
    workers = 8
    # Every thread blocks here until all 8 have arrived, then all proceed.
    gate = threading.Barrier(workers, timeout=5)

    def worker():
        gate.wait()
        for _ in range(1000):
            counter.increment()

    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(), "worker did not finish"

    assert counter.value == workers * 1000

2. Make the fix, and watch the same test pass

import threading


class Counter:
    def __init__(self):
        self.value = 0
        self._lock = threading.Lock()

    def increment(self):
        with self._lock:              # the whole read-modify-write is now atomic
            self.value += 1

3. Use a seam when the racing operations differ

A barrier works when every thread runs the same code. When the race is between two different operations — a reader and a writer, an initialiser and a consumer — the test needs to freeze one of them at a chosen point:

import threading
from unittest.mock import patch


def test_double_initialisation_is_prevented(service):
    entered = threading.Event()
    release = threading.Event()
    original = service._build_connection

    def slow_build():
        entered.set()                 # tell the test we are inside the window
        release.wait(timeout=5)       # hold the window open
        return original()

    with patch.object(service, "_build_connection", slow_build):
        first = threading.Thread(target=service.get_connection)
        first.start()
        assert entered.wait(timeout=5), "first thread never entered the builder"

        # Second thread arrives while the first is mid-initialisation.
        second_result = []
        second = threading.Thread(target=lambda: second_result.append(service.get_connection()))
        second.start()

        release.set()
        first.join(timeout=5)
        second.join(timeout=5)

    assert service.build_count == 1, "the connection was built twice"

This is the pattern that catches double-initialisation, cache stampedes and duplicated side effects, and it is deterministic: the second thread is guaranteed to arrive during the first thread's window because the test controls when the window closes. The patching technique it relies on is covered in patching class attributes with patch.object.

4. Surface exceptions raised in worker threads

An exception inside a threading.Thread target is printed to stderr and then discarded; the test does not fail. concurrent.futures fixes this because future.result() re-raises:

from concurrent.futures import ThreadPoolExecutor


def test_workers_do_not_raise():
    with ThreadPoolExecutor(max_workers=4) as pool:
        futures = [pool.submit(do_work, i) for i in range(4)]

    # result() re-raises in the calling thread, so worker failures fail the test.
    results = [future.result(timeout=5) for future in futures]
    assert sorted(results) == [0, 1, 2, 3]

Where threading.Thread is unavoidable, threading.excepthook (3.8+) can collect worker exceptions into a list the test asserts is empty. Silent worker failures are the single most common reason a concurrency test passes while the code is broken.

Verification

A race test has a property ordinary tests do not: it must fail before it can be trusted. Verify it by reverting the fix — comment out the lock, run the test, confirm it fails on every one of ten consecutive runs.

pytest tests/test_counter.py::test_increment_is_atomic --count=10 -q   # pytest-repeat

.........F                                                    [100%]
E       assert 7994 == 8000

Ten passes with the lock and ten failures without it is the evidence that the test is measuring the thing it claims to measure. A test that fails only occasionally without the fix is still probabilistic — the barrier is not tight enough, and usually the reason is that the contended region is too short relative to thread start-up. Increasing the iteration count inside each worker, rather than the number of workers, widens the window without adding scheduling noise.

Troubleshooting

Deadlocks: the failure with no traceback

A race produces a wrong answer; a deadlock produces no answer at all. Two threads each holding a lock the other wants will wait forever, and pytest's default behaviour is to wait with them until the CI platform kills the job.

The first line of defence is ordering discipline in the code: every code path acquires locks in the same global order, which makes a cycle impossible by construction. The second is detection in the tests. faulthandler, in the standard library, can be armed to dump every thread's stack after a timeout:

import faulthandler

# In conftest.py: after 30 seconds of no progress, dump all thread stacks
# and abort. Every stuck thread's frame is printed, which names the locks.
faulthandler.dump_traceback_later(30, exit=True)

Thread 0x00007f2a (most recent call first):
  File "app/cache.py", line 41 in refresh      ← waiting on _write_lock
  File "app/cache.py", line 88 in get
Thread 0x00007f2b (most recent call first):
  File "app/cache.py", line 62 in evict        ← waiting on _index_lock
  File "app/cache.py", line 91 in put

Two threads, two locks, opposite order: the dump contains the complete diagnosis. Without it the same deadlock is a job that timed out with no output at all.

The full setup, including how to combine it with pytest-timeout's thread method so both the dump and a clean failure are produced, is in dumping stacks on deadlock with faulthandler.

When a stress loop is still the right tool

Deterministic tests catch the race you already understand. They cannot find the one nobody has thought of, and for that a stress loop remains the only option — run the operation from many threads, many times, and assert the invariant at the end.

import random
import threading

import pytest


@pytest.mark.slow                      # nightly, not on every push
@pytest.mark.timeout(120)
def test_cache_invariant_under_random_load(cache):
    stop = threading.Event()
    errors: list[BaseException] = []

    def churn(seed: int) -> None:
        rng = random.Random(seed)      # per-thread seed: reproducible from the report
        try:
            while not stop.is_set():
                key = rng.randrange(64)
                if rng.random() < 0.3:
                    cache.evict(key)
                else:
                    cache.get_or_set(key, lambda: key * 2)
        except BaseException as exc:   # noqa: BLE001 - re-raised in the main thread
            errors.append(exc)

    threads = [threading.Thread(target=churn, args=(seed,)) for seed in range(8)]
    for thread in threads:
        thread.start()
    stop.wait(10)                      # ten seconds of contention
    stop.set()
    for thread in threads:
        thread.join(timeout=10)

    assert not errors, errors
    assert cache.size() <= cache.capacity, "eviction lost track of the bound"

Three details make the difference between a stress test that is useful and one that is merely slow. Each thread gets its own seeded Random, so a failure can be replayed exactly by re-running with the reported seeds. Exceptions are collected rather than printed, so a worker crash fails the test. And the assertion is an invariant that must hold at any point — a size bound, a conservation law, a monotonic counter — rather than an exact expected value, because under random load there is no single expected value.

Mark these slow and keep them out of the pull-request suite. A ten-second test that fails once a month is valuable in a nightly job and corrosive in a feedback loop, where its occasional failure teaches everyone to re-run the build rather than to read it. The marker-based split is the same one used for integration tests, and the reasoning is identical: a suite people trust is one where red means broken.

For invariant-driven exploration with better shrinking than a random loop can offer, a RuleBasedStateMachine generates operation sequences and then reduces a failure to its minimal reproduction — see modeling a cache with invariants and bundles, which applies exactly this technique to the same problem.

Fixtures that are safe to share across threads

A thread-safety test puts unusual demands on its fixtures, because the fixture's object is now touched by several threads at once. Three rules keep this from becoming a second source of flakiness.

Build the shared object inside the test, not in a session-scoped fixture: the point of the test is that this object is contended, and sharing it with other tests reintroduces order dependence. Where a genuinely shared resource is unavoidable — a database connection pool, a temporary directory — give each thread its own handle from the pool rather than sharing one handle, since most client libraries document per-connection thread safety and nothing stronger. And never let a fixture's teardown run while a worker thread is still alive: join every thread inside the test body, with a timeout, before any assertion that could fail and skip the joins.

That last rule is the one most often broken. An assertion failing mid-test propagates immediately, the fixture tears down the object the workers are still using, and the resulting error — a closed file, a released lock, a dead connection — is reported instead of the assertion that actually failed. Wrapping the thread lifecycle in a try/finally, or using ThreadPoolExecutor as a context manager so its __exit__ joins the workers, keeps the real failure visible. The habit worth forming is to treat every worker thread as a resource with an owner: whoever started it joins it, in a finally, with a timeout, before anything else in the test is allowed to fail.

Frequently Asked Questions

Why can't I just run the test 1000 times to catch a race? Because the scheduler is not adversarial. A race whose window is a few hundred nanoseconds may need millions of iterations on an idle machine and zero on a loaded one, so a loop that passes proves nothing and a loop that fails wastes minutes. Forcing the interleaving with a barrier turns a probabilistic test into a deterministic one that runs in milliseconds.

Does the GIL mean Python code can't have data races? No. The GIL makes individual bytecodes atomic, not statements. x += 1 compiles to a load, an add and a store, and the interpreter can switch threads between any two of them. Anything involving a check followed by an action — a cache miss followed by a fill, a None check followed by an assignment — is a race regardless of the GIL.

How do I test code that uses a ThreadPoolExecutor? Submit work through the executor as production does, but control the workers with a barrier so they arrive at the contended section together. Collect futures and call result() on each so exceptions in workers surface as test failures rather than being swallowed by the pool.

What is the right way to assert that a lock is actually held? Do not inspect the lock. Assert on the invariant the lock exists to protect: run N threads doing M increments and assert the total is exactly N times M. A lock check passes even when the lock protects the wrong region; an invariant check does not.

Should race tests run in the normal suite or separately? In the normal suite, if they are deterministic. A barrier-driven test takes milliseconds and fails every time the bug is present, which is exactly what a regression test should do. Only stress loops and long-running fuzzers belong in a separate nightly job.

Related guides

• Work through a complete reproduction in reproducing a race condition deterministically.
• Learn the primitive-by-primitive patterns in testing thread safety with barriers and events.
• Turn a hang into a diagnosis with dumping stacks on deadlock with faulthandler.
• Explore interleavings you did not think of using stateful and model-based testing.
• For cooperative rather than preemptive concurrency, the equivalent techniques are in pytest-asyncio in depth.

← Back to Testing Async & Concurrent Python