Synchronize Threads with a Barrier in Python
Demonstrates using threading.Barrier to synchronize multiple threads at phase boundaries, ensuring all workers wait for each other before proceeding.
Python code
21 linesimport threading
import time
from random import randint
def worker(barrier, worker_id):
for phase in range(3):
time.sleep(randint(1, 3))
print(f"Worker {worker_id} finished phase {phase} at {time.time():.2f}")
barrier.wait()
print(f"Worker {worker_id}: all phases complete")
if __name__ == "__main__":
num_workers = 3
barrier = threading.Barrier(num_workers)
threads = []
for i in range(num_workers):
t = threading.Thread(target=worker, args=(barrier, i))
threads.append(t)
t.start()
for t in threads:
t.join()Output
Worker 0 finished phase 0 at 1710000000.12
Worker 1 finished phase 0 at 1710000001.45
Worker 2 finished phase 0 at 1710000002.78
Worker 0 finished phase 1 at 1710000003.90
Worker 1 finished phase 1 at 1710000004.56
Worker 2 finished phase 1 at 1710000005.23
Worker 0 finished phase 2 at 1710000006.01
Worker 1 finished phase 2 at 1710000007.34
Worker 2 finished phase 2 at 1710000008.67
Worker 0: all phases complete
Worker 1: all phases complete
Worker 2: all phases completeHow it works
threading.Barrier creates a synchronization point that blocks until a fixed number of threads call wait(). When the required count arrives, all threads are released simultaneously, allowing them to proceed to the next phase. The loop runs three phases, and each worker sleeps a random amount to simulate variable work, but the barrier ensures they all start the next phase together. After all phases, each thread prints a completion message independently. This pattern is useful for dividing work into iterative rounds where each round must finish before the next starts.
Common mistakes
- Creating a barrier with a mismatched count (e.g., more threads than barriers) causing BlockingIOError or deadlock.
- Calling barrier.wait() from the main thread instead of within the worker threads.
- Not joining threads after starting them, leading to incomplete execution order.
- Using a timeout with barrier.wait() and not handling BrokenBarrierError when a thread fails.
Variations
- Use a `threading.Barrier` with a `timeout` parameter to avoid infinite blocking if a thread crashes.
- Replace the manual loop with `for phase in range(3): barrier.wait()` if the barrier is reused for each phase.
Real-world use cases
- Coordinating multiple worker processes in a data pipeline where each stage must complete before the next begins.
- Synchronizing parallel simulation steps in scientific computing to ensure consistent time-step evolution.
- Orchestrating concurrent API load tests where all clients should start hitting the endpoint at the same moment.
Sponsored
More from Concurrency & performance
- Benchmark list.append vs deque.append in Python medium
- Build a Python Performance Profiler That Generates Readable Reports medium
- Graceful Shutdown Executor Context Manager in Python medium
- How to Build a Producer-Consumer Pattern with asyncio.Queue in Python medium
- How to Cancel an asyncio Task with Graceful Cleanup in Python medium
- How to Convert Data in Parallel with ThreadPoolExecutor in Python easy
Keep learning
Related tutorials and quizzes for this topic.