|$ curl https://forge-ai.dev/api/markdown?path=docs/python/threading

$cat docs/python-—-threading.md

updated Recently·20 min read·published

Python — Threading

◆Python◆Advanced

Introduction

The threading module provides a high-level API for working with threads in Python. Threads are lightweight concurrent units within a single process, sharing the same memory space. They are ideal for I/O-bound tasks where the bottleneck is waiting for external resources (network, disk, database).

In CPython, the Global Interpreter Lock (GIL) prevents multiple threads from executing Python bytecode simultaneously, limiting CPU-bound parallelism. However, threads release the GIL during blocking I/O operations, making them effective for concurrent I/O.

Feature	Threading	Multiprocessing
Memory	Shared (same process)	Separate (each process)
GIL	Affected by GIL	Bypasses GIL
Overhead	Low (lightweight)	Higher (spawn process)
Best for	I/O-bound tasks	CPU-bound tasks
Sharing data	Direct (needs locks)	IPC (Queue, Pipe)

Creating Threads

The Thread class represents a thread of execution. Pass a target callable and args tuple, then call start() to begin execution and join() to wait for completion. You can also name threads for debugging and set the daemon flag.

creating_threads.py

Python

1	import threading
2	import time
3	from typing import Callable, Any
4
5	# Basic thread — target function + args
6	def download(url: str, timeout: int = 10) -> None:
7	print(f"[{threading.current_thread().name}] fetching {url}")
8	time.sleep(1) # simulate network I/O
9	print(f"[{threading.current_thread().name}] done {url}")
10
11	t = threading.Thread(target=download, args=("https://example.com",))
12	t.start() # run the thread
13	t.join() # wait for it to finish
14	print("main thread continues")
15
16	# Multiple threads
17	threads = []
18	for i in range(5):
19	t = threading.Thread(
20	target=download,
21	args=(f"https://site-{i}.com",),
22	name=f"DL-{i}", # meaningful name
23	)
24	threads.append(t)
25	t.start()
26
27	for t in threads:
28	t.join()
29
30	print(f"all {len(threads)} threads complete")
31
32	# Custom Thread subclass
33	class WorkerThread(threading.Thread):
34	def __init__(self, name: str, work: Callable[[], Any]):
35	super().__init__(name=name)
36	self._work = work
37	self.result: Any = None
38
39	def run(self) -> None:
40	self.result = self._work()
41
42	def compute() -> int:
43	time.sleep(0.5)
44	return 42
45
46	worker = WorkerThread("Worker-1", compute)
47	worker.start()
48	worker.join()
49	print(worker.result) # → 42
50
51	# Thread identity
52	print(threading.current_thread().name) # "MainThread"
53	print(threading.current_thread().ident) # integer thread ID
54	print(threading.active_count()) # number of alive threads

Daemon Threads

Daemon threads run in the background and are automatically killed when all non-daemon threads exit. They are useful for background monitoring, periodic cleanup, or heartbeats. Set daemon=True or set the daemon property before calling start().

daemon.py

Python

1	import threading
2	import time
3
4	def background_logger():
5	"""Daemon thread — prints every second."""
6	while True:
7	time.sleep(1)
8	print(f"[bg] alive at {time.time():.0f}")
9
10	# Set daemon flag before start
11	daemon = threading.Thread(
12	target=background_logger,
13	daemon=True,
14	name="BG-Logger",
15	)
16	daemon.start()
17
18	# Main thread runs for 5 seconds, then exits.
19	# The daemon thread is killed automatically.
20	time.sleep(5)
21	print("main exiting — daemon will be killed")
22	# → daemon thread stops even if its loop is infinite
23
24	# Daemon check
25	print(daemon.daemon) # → True
26	print(threading.main_thread().daemon) # → False
27
28	# Non-daemon thread keeps process alive
29	def keep_alive():
30	time.sleep(10)
31	print("still alive")
32
33	t = threading.Thread(target=keep_alive, daemon=False)
34	t.start()
35	# Process won't exit until t completes

⚠

warning

Daemon threads are abruptly terminated when the process exits. Do not use them for critical operations (file writes, DB commits, resource cleanup). Use join() on non-daemon threads instead.

Synchronization Primitives

Python's threading module provides several synchronization primitives to coordinate threads and protect shared state. Choosing the right primitive depends on the access pattern and contention level.

Primitive	Purpose	Reentrant
Lock	Exclusive access (mutex)	No
RLock	Reentrant exclusive access	Yes
Semaphore	Limit concurrent access count	N/A
Event	Signal between threads	N/A
Condition	Wait/notify pattern	Uses Lock/RLock
Barrier	Synchronize N threads at a point	N/A
Timer	Run a function after a delay	N/A

Lock & RLock

A Lock (mutex) ensures only one thread accesses a critical section at a time. Always use a context manager (with lock:) to guarantee release. An RLock (reentrant lock) can be acquired multiple times by the same thread, which is necessary when a function that acquires a lock calls another function that needs the same lock.

lock.py

Python

1	import threading
2
3	# ----- Lock (non-reentrant mutex) -----
4	lock = threading.Lock()
5	counter = 0
6
7	def increment():
8	global counter
9	for _ in range(100_000):
10	with lock: # acquire — blocks if held
11	counter += 1
12	# released automatically on block exit
13
14	threads = [
15	threading.Thread(target=increment)
16	for _ in range(4)
17	]
18	for t in threads: t.start()
19	for t in threads: t.join()
20
21	print(counter) # → 400_000 (correct with lock)
22
23	# Manually acquire/release (not recommended)
24	lock.acquire()
25	try:
26	counter += 1
27	finally:
28	lock.release()
29
30	# Lock is NOT reentrant — this would deadlock
31	lock.acquire()
32	# lock.acquire() ← would block forever (deadlock!)
33
34	# ----- RLock (reentrant) -----
35	rlock = threading.RLock()
36	data = {"value": 0}
37
38	def outer():
39	with rlock:
40	inner() # same thread re-acquires RLock
41	data["value"] += 1
42
43	def inner():
44	with rlock: # OK — same thread, RLock allows it
45	temp = data["value"]
46
47	outer() # works fine with RLock
48	print(data["value"]) # → 1
49
50	# Deadlock example (Lock version)
51	lock_a = threading.Lock()
52	lock_b = threading.Lock()
53
54	def deadlock_risk():
55	"""Swap lock order to avoid deadlock."""
56	with lock_a:
57	with lock_b:
58	pass
59
60	def opposite_order():
61	with lock_b: # always acquire in same order
62	with lock_a:
63	pass
64
65	# try-except with timeout
66	if not lock.acquire(timeout=2.0):
67	print("could not acquire lock in 2s")

ℹ

info

Prefer Lock over RLock unless you specifically need reentrancy (same thread re-acquiring). RLock has slightly more overhead and can mask design issues. Always acquire multiple locks in a consistent global order to avoid deadlocks.

Semaphore

A Semaphore limits the number of threads that can access a resource concurrently. It maintains an internal counter that decrements on acquire() and increments on release(). BoundedSemaphore raises ValueError if released more times than acquired, catching bugs.

semaphore.py

Python

1	import threading
2	import time
3
4	# Semaphore — limit to 3 concurrent connections
5	connection_pool = threading.Semaphore(3)
6
7	def make_request(n: int) -> None:
8	with connection_pool:
9	print(f"[T{n}] acquired connection")
10	time.sleep(1) # simulate I/O
11	print(f"[T{n}] releasing connection")
12	# semaphore released here
13
14	threads = [
15	threading.Thread(target=make_request, args=(i,))
16	for i in range(10)
17	]
18	for t in threads: t.start()
19	for t in threads: t.join()
20	# At most 3 threads run concurrently
21
22	# BoundedSemaphore — catches extra releases
23	bounded = threading.BoundedSemaphore(2)
24	bounded.acquire() # counter → 1
25	bounded.acquire() # counter → 0
26	# bounded.release() # counter → 1
27	# bounded.release() # counter → 2 (max)
28	# bounded.release() # ValueError! too many releases
29
30	# Semaphore as a latch (start with 0)
31	latch = threading.Semaphore(0)
32
33	def waiter(n: int) -> None:
34	print(f"[T{n}] waiting for latch")
35	latch.acquire() # blocks until released
36	print(f"[T{n}] passed latch")
37
38	waiters = [
39	threading.Thread(target=waiter, args=(i,))
40	for i in range(3)
41	]
42	for t in waiters: t.start()
43	time.sleep(0.5)
44	print("releasing latch 3 times")
45	for _ in range(3):
46	latch.release() # unblocks one waiter each time
47	for t in waiters: t.join()

Event

An Event lets one thread signal one or more other threads that something has happened. It has an internal boolean flag: threads wait for the flag to be set with wait(), and a signalling thread sets it with set(). Unlike Condition, Event does not require holding a lock.

event.py

Python

1	import threading
2	import time
3
4	# Event — one-shot signal
5	start_event = threading.Event()
6
7	def worker(n: int) -> None:
8	print(f"[T{n}] ready, waiting for start signal")
9	start_event.wait() # blocks until set()
10	print(f"[T{n}] running!")
11
12	workers = [
13	threading.Thread(target=worker, args=(i,))
14	for i in range(4)
15	]
16	for t in workers: t.start()
17	time.sleep(1)
18	print("main: releasing all workers")
19	start_event.set() # unblocks all waiters
20	for t in workers: t.join()
21
22	# Event with timeout
23	event = threading.Event()
24	result = event.wait(timeout=2.0) # returns False if timed out
25	if not result:
26	print("timed out waiting for event")
27
28	# Reusable pattern — clear + set (not safe with multiple waiters)
29	def producer():
30	"""Readies data and signals consumer."""
31	time.sleep(1)
32	data.append("ready")
33	event.set()
34
35	def consumer():
36	event.wait()
37	event.clear() # reset for next round
38	print(f"got: {data[-1]}")
39
40	data = []
41	event = threading.Event()
42
43	t1 = threading.Thread(target=producer)
44	t2 = threading.Thread(target=consumer)
45	t1.start(); t2.start()
46	t1.join(); t2.join()
47
48	# Periodic wakeup pattern
49	stop_event = threading.Event()
50
51	def periodic(interval: float):
52	while not stop_event.is_set():
53	print(f"tick at {time.time():.0f}")
54	stop_event.wait(timeout=interval)
55
56	p = threading.Thread(target=periodic, args=(1.0,), daemon=True)
57	p.start()
58	time.sleep(3)
59	stop_event.set() # clean shutdown
60	print("stopped periodic")

Thread-Safe Queues

The queue.Queue class provides a thread-safe FIFO queue with built-in locking. It is the preferred way to exchange data between threads — no manual lock management needed. The queue blocks on get() when empty and on put() when full (with optional timeout).

queue.py

Python

1	import threading
2	import queue
3	import time
4	from dataclasses import dataclass, field
5
6	# Basic producer-consumer
7	q: queue.Queue = queue.Queue(maxsize=20)
8
9	def producer(name: str, count: int):
10	for i in range(count):
11	item = f"{name}-item-{i}"
12	q.put(item) # blocks if queue full
13	print(f"[P:{name}] put {item}")
14	time.sleep(0.1)
15	q.put(None) # sentinel — signals "done"
16
17	def consumer(name: str):
18	while True:
19	item = q.get() # blocks if queue empty
20	if item is None: # sentinel check
21	q.task_done()
22	break
23	print(f"[C:{name}] got {item}")
24	q.task_done() # signal task completion
25
26	# Start one producer and two consumers
27	prod = threading.Thread(target=producer, args=("A", 5))
28	cons = [
29	threading.Thread(target=consumer, args=("X",)),
30	threading.Thread(target=consumer, args=("Y",)),
31	]
32	prod.start()
33	for c in cons: c.start()
34	prod.join()
35	q.join() # wait until all tasks processed
36	for c in cons: c.join()
37	print("all done")
38
39	# Structured data
40	@dataclass
41	class Task:
42	url: str
43	retries: int = 3
44	priority: int = field(default=0)
45
46	task_queue: queue.Queue[Task] = queue.Queue()
47
48	def dispatcher():
49	for _ in range(5):
50	task_queue.put(Task(url="https://example.com"))
51	task_queue.put(None)
52
53	# Queue methods
54	q = queue.Queue(maxsize=10)
55	q.put("item", block=True, timeout=1.0) # raises Full after 1s
56	item = q.get(block=True, timeout=2.0) # raises Empty after 2s
57	print(q.qsize()) # approximate size
58	print(q.empty()) # True if empty
59	print(q.full()) # True if full
60
61	# Variants: LifoQueue (stack), PriorityQueue
62	from queue import LifoQueue, PriorityQueue
63
64	stack = LifoQueue() # last-in, first-out
65	pq: PriorityQueue[tuple[int, str]] = PriorityQueue()
66	pq.put((2, "medium"))
67	pq.put((1, "high"))
68	pq.put((3, "low"))
69	while not pq.empty():
70	print(pq.get()) # (1, high), (2, medium), (3, low)

✓

best practice

Using queue.Queue with sentinel values (None) is the cleanest pattern for producer-consumer threading. It eliminates explicit lock management and naturally handles back-pressure via maxsize.

ThreadPoolExecutor

concurrent.futures.ThreadPoolExecutor provides a high-level interface for managing a pool of threads. You submit callables and receive Future objects. The executor handles thread lifecycle, work distribution, and result collection automatically.

thread_pool.py

Python

1	from concurrent.futures import (
2	ThreadPoolExecutor,
3	as_completed,
4	wait,
5	FIRST_COMPLETED,
6	ALL_COMPLETED,
7	)
8	import time
9
10	# Basic usage — submit + result
11	def fetch(url: str) -> tuple[str, int]:
12	time.sleep(0.5) # simulate network I/O
13	return url, len(url) * 10
14
15	with ThreadPoolExecutor(max_workers=4) as executor:
16	future = executor.submit(fetch, "https://example.com")
17	url, size = future.result() # blocks until done
18	print(f"{url}: {size}")
19
20	# Submit multiple, process as completed
21	urls = [
22	"https://python.org",
23	"https://github.com",
24	"https://stackoverflow.com",
25	"https://news.ycombinator.com",
26	]
27
28	with ThreadPoolExecutor(max_workers=4) as executor:
29	futures = {
30	executor.submit(fetch, url): url
31	for url in urls
32	}
33	for future in as_completed(futures):
34	url, size = future.result()
35	print(f"[done] {url}: {size} bytes")
36
37	# map — results in submission order
38	with ThreadPoolExecutor(max_workers=4) as executor:
39	results = executor.map(fetch, urls, timeout=10)
40	for url, size in results:
41	print(f"{url}: {size}")
42
43	# wait — block until condition met
44	with ThreadPoolExecutor(max_workers=4) as executor:
45	futures = [executor.submit(fetch, url) for url in urls]
46	done, pending = wait(futures, timeout=3.0,
47	return_when=FIRST_COMPLETED)
48	print(f"{len(done)} done, {len(pending)} pending")
49
50	# Exception handling
51	def might_fail(url: str) -> str:
52	if "bad" in url:
53	raise ValueError(f"bad url: {url}")
54	return f"ok: {url}"
55
56	with ThreadPoolExecutor(max_workers=2) as executor:
57	future = executor.submit(might_fail, "bad.com")
58	try:
59	result = future.result()
60	except ValueError as e:
61	print(f"caught: {e}")
62
63	# Context manager ensures shutdown
64	executor = ThreadPoolExecutor(max_workers=2)
65	executor.submit(fetch, "https://example.com")
66	# executor.shutdown(wait=True) # called automatically by 'with'
67
68	# Adjust max_workers based on workload
69	# I/O-bound: higher than CPU count (e.g., 4× CPUs)
70	# Memory-bound: keep lower to avoid OOM

The GIL & Its Impact

The Global Interpreter Lock (GIL) is a mutex in CPython that prevents multiple threads from executing Python bytecode simultaneously. This means CPU-bound threads cannot run in parallel — only one thread holds the GIL at any moment. However, the GIL is released during I/O operations, making threading effective for I/O-bound workloads.

gil.py

Python

1	import threading
2	import time
3
4	# CPU-bound — threads do NOT speed up due to GIL
5	def cpu_heavy(n: int) -> int:
6	total = 0
7	for i in range(n):
8	total += i * i
9	return total
10
11	N = 50_000_000
12
13	# Sequential version
14	start = time.perf_counter()
15	cpu_heavy(N)
16	print(f"sequential: {time.perf_counter() - start:.2f}s")
17
18	# Parallel (threaded) — no speedup
19	def run_in_threads():
20	threads = [
21	threading.Thread(target=cpu_heavy, args=(N // 2,))
22	for _ in range(2)
23	]
24	start = time.perf_counter()
25	for t in threads: t.start()
26	for t in threads: t.join()
27	print(f"2 threads: {time.perf_counter() - start:.2f}s")
28
29	run_in_threads()
30	# Both ~same time (or threaded is slower due to GIL contention)
31
32	# I/O-bound — threads DO improve throughput
33	def io_task(n: int) -> None:
34	for _ in range(n):
35	time.sleep(0.01) # releases GIL during sleep
36	_ = 1 + 1 # tiny CPU work, re-acquires GIL
37
38	start = time.perf_counter()
39	io_task(100)
40	print(f"sequential I/O: {time.perf_counter() - start:.2f}s")
41
42	def io_threaded():
43	threads = [
44	threading.Thread(target=io_task, args=(50,))
45	for _ in range(2)
46	]
47	start = time.perf_counter()
48	for t in threads: t.start()
49	for t in threads: t.join()
50	print(f"threaded I/O: {time.perf_counter() - start:.2f}s")
51
52	io_threaded()
53	# Threaded I/O is ~2× faster
54
55	# GIL-free C extensions
56	import sys
57	print(sys.version) # CPython has GIL
58	# numpy, pandas, C extensions release GIL during heavy ops
59
60	# GIL visibility
61	import sys
62	try:
63	sys._is_gil_enabled() # Python 3.13+ free-threading
64	except AttributeError:
65	pass
66
67	# PEP 703 (nogil) — experimental free-threaded build
68	# Check: sysconfig.get_config_var("Py_GIL_DISABLED")

ℹ

info

Python 3.13 introduced an experimental free-threaded build (PEP 703) that disables the GIL. In this build, CPU-bound threads can run in parallel. Most existing code works unchanged, but C extensions may need updates for thread safety. The default CPython build still has the GIL.

Best Practices

Threading is powerful but introduces complexity. Following these best practices helps avoid common pitfalls like race conditions, deadlocks, and performance degradation.

best_practices.py

Python

1	import threading
2	import queue
3	from concurrent.futures import ThreadPoolExecutor
4
5	# 1. Prefer ThreadPoolExecutor over raw threads
6	# Manages lifecycle and work distribution.
7	def good():
8	with ThreadPoolExecutor(max_workers=10) as ex:
9	results = list(ex.map(fetch, urls))
10
11	def bad():
12	threads = []
13	for url in urls:
14	t = threading.Thread(target=fetch, args=(url,))
15	t.start()
16	threads.append(t)
17	for t in threads:
18	t.join()
19
20	# 2. Use queue.Queue for data exchange
21	# No manual locks needed — safer and clearer.
22	def producer_consumer():
23	q: queue.Queue = queue.Queue()
24	# producer puts items, consumer gets them
25	# q.get() blocks naturally, no busy-waiting
26
27	# 3. Keep critical sections small
28	# Minimize code under lock to reduce contention.
29	def process_item(item):
30	with cache_lock:
31	cached = cache.get(item)
32	if cached is None:
33	result = expensive_compute(item) # OUTSIDE lock
34	with cache_lock:
35	cache[item] = result
36	else:
37	result = cached
38
39	# 4. Consistent lock ordering
40	# Always acquire locks in the same global order.
41	def transfer(from_acct, to_acct, amount):
42	first, second = sorted(id(from_acct), id(to_acct))
43	# locked: with lock_for(first): with lock_for(second): ...
44	# Prevents deadlock when two threads swap order.
45
46	# 5. Prefer Lock over RLock unless reentrancy needed
47	lock = threading.Lock()
48
49	# 6. Use timeout for lock acquisition
50	if not lock.acquire(timeout=5.0):
51	# handle timeout gracefully
52	pass
53	# Otherwise thread blocks forever on deadlock
54
55	# 7. Know when NOT to thread
56	# CPU-bound → multiprocessing or ProcessPoolExecutor
57	# High-concurrency I/O → asyncio
58	# Simple parallelism → ThreadPoolExecutor
59
60	# 8. Daemon threads for background work only
61	# Not for critical cleanup or writes.
62	bg = threading.Thread(target=heartbeat, daemon=True)
63
64	# 9. Thread-local data for per-thread state
65	local = threading.local()
66
67	def per_thread_work(value):
68	local.session_id = value # each thread has its own copy
69	# No lock needed — data is thread-isolated
70
71	# 10. Test with thread sanitizers
72	# python3 -X faulthandler ...
73	# Use pytest-timeout to catch deadlocks in tests

ℹ

info

When in doubt, start with ThreadPoolExecutor and queue.Queue. They handle most threading scenarios safely. Reach for Lock, Event, and Semaphore only when you have specific coordination needs. Use Condition and Barrier for advanced producer-consumer or phased synchronization.

$Blueprint — Engineering Documentation·Section ID: PYTHON-THREAD·Revision: 1.0

X GitHub YouTube Instagram Facebook Buy me a coffee Sponsor