Python Threading and Synchronization

The essentials

Quick reference

One focused task per row. Jump to the related section for complete, working examples.

Use	Syntax	Examples
Start a thread	`worker.start()`	View examples
Wait with a deadline	`worker.join(timeout=2.0)`	View examples
Name a worker	`worker = threading.Thread(name='indexer', target=build_index)`	View examples
Guard a critical section	`with lock: update_shared_state()`	View examples
Attempt bounded acquisition	`acquired = lock.acquire(timeout=0.5)`	View examples
Create a reentrant lock	`lock = threading.RLock()`	View examples
Wait for a predicate	`ready = condition.wait_for(predicate, timeout=1.0)`	View examples
Notify one waiter	`condition.notify()`	View examples
Request cooperative stop	`stop_event.set()`	View examples
Limit concurrent entries	`limit = threading.Semaphore(4)`	View examples
Detect excess release	`limit = threading.BoundedSemaphore(value=4)`	View examples
Create a phase barrier	`barrier = threading.Barrier(parties=3, timeout=5.0)`	View examples
Arrive at a barrier	`index = barrier.wait()`	View examples
Abort a barrier	`barrier.abort()`	View examples
Create backpressure	`jobs = queue.Queue(maxsize=100)`	View examples
Acknowledge a queued task	`jobs.task_done()`	View examples
Wait for task acknowledgements	`jobs.join()`	View examples
Check completion after join	`if worker.is_alive(): handle_timeout()`	View examples
Install an exception observer	`threading.excepthook = report_thread_failure`	View examples

Threads share a process and its objects, which makes communication inexpensive and races possible. This guide treats ownership, bounded waiting, exception propagation, and shutdown as part of the design. On ordinary CPython builds the GIL does not make compound operations thread-safe; threads are usually best for overlapping blocking I/O. Free-threaded builds can execute Python code in parallel, increasing the importance of explicit synchronization.

Step by step

Detailed examples

Start, identify, and join non-daemon threads

Pass immutable inputs through args, give operational threads useful names, and retain each Thread so the owner can join it. start() may be called once; join() may be called repeatedly, but joining the current thread raises RuntimeError. Daemon threads are stopped abruptly during interpreter shutdown, so they are unsuitable for work that must flush or release resources.

Coordinate a worker with join

import threading

result: list[int] = []
worker = threading.Thread(target=lambda: result.append(21 * 2), name="calculator")
worker.start()
worker.join()
print(worker.name)
print(result[0])

Output

calculator
42

Back to quick reference ↑

Protect invariants with Lock and RLock

A lock protects an invariant, not merely a line of code. Hold it for the smallest complete state transition and use with so exceptions cannot strand it. Lock has no owner and is non-reentrant; RLock tracks its owning thread and recursion depth, which is useful when synchronized methods call one another but can conceal an overly tangled locking design.

Update related values atomically

import threading

lock = threading.Lock()
state = {"accepted": 0, "total": 0}

def record(value: int) -> None:
    with lock:
        state["accepted"] += 1
        state["total"] += value

threads = [threading.Thread(target=record, args=(value,)) for value in (2, 3, 5)]
for thread in threads:
    thread.start()
for thread in threads:
    thread.join()
print(state)

Output

{'accepted': 3, 'total': 10}

Back to quick reference ↑

Wait for state with predicates and events

Condition combines a lock with a wait set. Always wait in a loop or use wait_for(predicate), because wakeups do not prove that the state is ready. Mutate the predicate while holding the condition lock, then notify. Event is a simpler level-triggered flag for readiness or cooperative stop requests; it does not carry data and set() wakes all current waiters.

Publish state before notification

import threading

condition = threading.Condition()
items: list[str] = []

def consumer() -> None:
    with condition:
        condition.wait_for(lambda: bool(items))
        print(items.pop())

thread = threading.Thread(target=consumer)
thread.start()
with condition:
    items.append("ready")
    condition.notify()
thread.join()

Output

ready

Back to quick reference ↑

Bound concurrency with semaphores

Semaphore represents a permit count and is appropriate for limiting simultaneous access to a finite resource. BoundedSemaphore additionally raises ValueError when released too many times, catching accounting defects. A semaphore is not a task queue and does not guarantee fair waiter ordering; pair it with explicit work ownership and deadlines.

Use a bounded permit as a context manager

import threading

limit = threading.BoundedSemaphore(1)
with limit:
    print("inside")
print("released")

Output

inside
released

Back to quick reference ↑

Coordinate fixed phases with Barrier

Barrier releases a fixed number of participants after every participant calls wait(). The returned indexes are unique from zero through parties minus one and can elect one thread for housekeeping. A timeout, exception in the action, reset, or abort can break the barrier; every participant must handle BrokenBarrierError rather than wait indefinitely.

Release two phase participants

import threading

barrier = threading.Barrier(2)
indexes: list[int] = []

def arrive() -> None:
    indexes.append(barrier.wait())

thread = threading.Thread(target=arrive)
thread.start()
indexes.append(barrier.wait())
thread.join()
print(sorted(indexes))

Output

[0, 1]

Back to quick reference ↑

Transfer ownership through Queue

queue.Queue provides synchronized put/get operations and optional capacity. Each successful get() representing a task must have exactly one task_done(), normally in finally, so join() can observe completion. Python 3.13 added Queue.shutdown(); use sentinel values when supporting older runtimes. threading.local stores per-thread attributes but should not replace explicit request context at API boundaries.

Finish queued work with a sentinel

from queue import Queue
import threading

jobs: Queue[int | None] = Queue()
answers: list[int] = []

def consume() -> None:
    while True:
        item = jobs.get()
        try:
            if item is None:
                return
            answers.append(item * item)
        finally:
            jobs.task_done()

thread = threading.Thread(target=consume)
thread.start()
for item in (2, 3):
    jobs.put(item)
jobs.put(None)
jobs.join()
thread.join()
print(answers)

Output

[4, 9]

Back to quick reference ↑

Make cancellation and failure observable

Threads cannot be safely killed from the outside. Use an Event plus bounded blocking calls, and join every worker during orderly shutdown. Thread exceptions normally reach threading.excepthook rather than the creating thread; wrap worker entry points to publish failures, or use concurrent.futures when result and exception propagation is the primary need. Never hold a lock while joining a thread that may need that lock.

Return a worker failure to its owner

from queue import Queue
import threading

outcomes: Queue[tuple[str, str]] = Queue()

def run() -> None:
    try:
        raise ValueError("invalid job")
    except Exception as error:
        outcomes.put((type(error).__name__, str(error)))

thread = threading.Thread(target=run)
thread.start()
thread.join()
print(outcomes.get())

Output

('ValueError', 'invalid job')

Back to quick reference ↑

Local code tester

Model cooperative cancellation

Run a deterministic worker loop that obeys an explicit stop predicate, the same ownership model used with threading.Event.

Runs in your browser

Python

def collect_until(values: list[int], should_stop) -> list[int]:
    collected = []
    for value in values:
        if should_stop(value):
            break
        collected.append(value)
    return collected

print(collect_until([2, 4, 7, 8], lambda value: value % 2 == 1))

Output

Press Run to load Python locally.

Sources and further reading

References

Authoritative documentation used to verify and expand this cheat sheet.

Help us improve

Found a typo or missing example?

Tell us what would make this cheat sheet clearer, more complete, or more useful.

Share feedback

Quick reference

Detailed examples

Start, identify, and join non-daemon threads

Protect invariants with Lock and RLock

Wait for state with predicates and events

Bound concurrency with semaphores

Coordinate fixed phases with Barrier

Transfer ownership through Queue

Make cancellation and failure observable

Model cooperative cancellation

More Python cheat sheets

Python asyncio and await Cheat Sheet

Python concurrent.futures and Parallel Work Cheat Sheet

Python Hashing, HMAC, and Secrets Cheat Sheet

References

Found a typo or missing example?