A barrier makes race conditions reproducible
Threads that start staggered rarely collide, so the test passes and production does not. A barrier makes all of them hit the same line at the same instant.
barrier = threading.Barrier(16)
def worker():
barrier.wait()
for _ in range(50_000): record("invoice")Assert on the aggregate. On the free-threaded build this fails against a naive counter every time, which is exactly the point.
pythontestingconcurrency
Longer version: the post this came from.