🚦 Pattern 5: Synchronization Primitives

Beyond Locks

A Lock answers one question: "can I touch this shared state right now?" But threads often need to answer richer questions — "has startup finished yet?", "is there an item in the buffer?", "are all 5 workers ready to begin?". Busy-waiting in a while loop with time.sleep works but burns CPU and adds latency. The threading module ships four primitives built for exactly this.

threading.Event — a One-Bit Signal

An Event is a boolean flag that threads can wait on. One thread flips it with set(); every thread blocked in wait() wakes up at once. It's the simplest way to broadcast "go!" or "we're shutting down" to many workers.

• event.set() — flip the flag to true, wake all current waiters.
• event.clear() — flip back to false; future wait() calls block again.
• event.wait(timeout=None) — block until set; returns the flag's value (handy to detect timeouts).
• event.is_set() — check without blocking.

import threading, time

ready = threading.Event()

def worker(n):
    print(f"worker {n} waiting for go signal")
    ready.wait()                      # blocks here until ready.set()
    print(f"worker {n} running!")

threads = [threading.Thread(target=worker, args=(i,)) for i in range(3)]
for t in threads:
    t.start()

time.sleep(1)                         # let all workers reach wait()
print("main: GO")
ready.set()                           # one-to-many wakeup: all 3 fire at once
for t in threads:
    t.join()

threading.Condition — Wait for a Predicate

An Event is a single bit. A Condition lets a thread wait for an arbitrary predicate over shared state — "the queue is non-empty", "the buffer has room" — and bundles a lock so you can inspect that state safely. It's the classic producer/consumer tool.

• cond.wait() — atomically release the lock and sleep; re-acquire the lock on wakeup.
• cond.notify(n=1) — wake up to n waiters.
• cond.notify_all() — wake every waiter.

import threading, time, collections

cond = threading.Condition()          # has its own internal lock
buffer = collections.deque()

def producer():
    for i in range(5):
        time.sleep(0.2)
        with cond:                    # must hold the lock to notify
            buffer.append(i)
            print(f"produced {i}")
            cond.notify()             # wake one waiting consumer

def consumer():
    while True:
        with cond:                    # must hold the lock to wait
            while not buffer:         # WHILE, not if — see below
                cond.wait()
            item = buffer.popleft()
        print(f"consumed {item}")
        if item == 4:
            return

t1 = threading.Thread(target=consumer)
t2 = threading.Thread(target=producer)
t1.start(); t2.start()
t1.join(); t2.join()

Semaphore & BoundedSemaphore — Limit Concurrency to N

A Semaphore is a counter. acquire() decrements it (blocking at zero) and release() increments it. Initialize it to N and only N threads can be inside the guarded section at once — the perfect throttle for "at most 5 concurrent connections to this flaky API."

import threading, time

MAX_CONCURRENT = 3
gate = threading.Semaphore(MAX_CONCURRENT)   # 3 permits

def call_api(n):
    with gate:                    # acquire a permit (blocks if 0 left)
        print(f"request {n} in flight")
        time.sleep(1)             # only 3 of these overlap at any moment
    # permit released on exit

threads = [threading.Thread(target=call_api, args=(i,)) for i in range(10)]
for t in threads: t.start()
for t in threads: t.join()

threading.Barrier — Rendezvous of N Threads

A Barrier makes exactly N threads wait for each other. Each calls barrier.wait(); the first N-1 block, and when the N-th arrives, all are released together. Ideal for phased computation where no thread may start phase 2 until every thread finished phase 1.

import threading, time, random

N = 3
barrier = threading.Barrier(N)

def worker(n):
    time.sleep(random.random())       # each finishes phase 1 at a different time
    print(f"worker {n} done phase 1, waiting at barrier")
    barrier.wait()                    # blocks until all N arrive
    print(f"worker {n} starting phase 2")   # all print together

threads = [threading.Thread(target=worker, args=(i,)) for i in range(N)]
for t in threads: t.start()
for t in threads: t.join()

Which One Do I Reach For?