Python — Concurrency

◆Python◆Advanced

Introduction

Python offers multiple concurrency models, each suited for different workloads. The Global Interpreter Lock (GIL) limits CPU-bound threading, but asyncio and multiprocessing provide effective alternatives.

Model	Best For	GIL Impact	When to Use
Threading	I/O-bound	Limited by GIL	File I/O, network requests, DB queries
Asyncio	I/O-bound	No GIL issue	Web servers, async APIs, streaming
Multiprocessing	CPU-bound	Bypasses GIL	Data processing, ML training, CPU crunching
Concurrent Futures	Both	Depends on executor	High-level pool management

Threading

Threads are lightweight concurrent units within a single process. In CPython, the GIL prevents true parallel execution of Python bytecode, but I/O-bound threads release the GIL during blocking operations.

threading.py

Python

1	import threading
2	import time
3
4	# Basic thread
5	def worker(name: str, delay: float):
6	print(f"{name} starting")
7	time.sleep(delay)
8	print(f"{name} done")
9
10	threads = []
11	for i in range(3):
12	t = threading.Thread(target=worker, args=(f"T{i}", i))
13	threads.append(t)
14	t.start()
15
16	for t in threads:
17	t.join() # wait for all threads
18
19	print("all threads done")
20
21	# Thread with return value (use concurrent.futures instead)
22	class ResultThread(threading.Thread):
23	def __init__(self, target, args=()):
24	super().__init__()
25	self._target = target
26	self._args = args
27	self.result = None
28
29	def run(self):
30	self.result = self._target(*self._args)
31
32	# Thread synchronization — Lock
33	lock = threading.Lock()
34	shared_counter = 0
35
36	def safe_increment():
37	global shared_counter
38	for _ in range(100_000):
39	with lock: # acquire/release automatically
40	shared_counter += 1
41
42	t1 = threading.Thread(target=safe_increment)
43	t2 = threading.Thread(target=safe_increment)
44	t1.start()
45	t2.start()
46	t1.join()
47	t2.join()
48	print(shared_counter) # → 200_000 (without lock: race condition)
49
50	# Thread-local data
51	local_data = threading.local()
52
53	def set_and_print(value):
54	local_data.value = value
55	time.sleep(0.1)
56	print(f"Thread {threading.current_thread().name}: {local_data.value}")
57
58	threads = [
59	threading.Thread(target=set_and_print, args=("A",)),
60	threading.Thread(target=set_and_print, args=("B",)),
61	]
62	for t in threads: t.start()
63	for t in threads: t.join()
64
65	# Semaphore — limit concurrent access
66	semaphore = threading.Semaphore(3) # max 3 at once
67
68	def limited_worker(n):
69	with semaphore:
70	print(f"Worker {n} running")
71	time.sleep(1)
72
73	threads = [threading.Thread(target=limited_worker, args=(i,)) for i in range(10)]
74	for t in threads: t.start()
75	for t in threads: t.join()

ℹ

info

Threading is good for I/O-bound tasks where threads spend most of their time waiting (network, disk, database). For CPU-bound work, use multiprocessing. For high-concurrency I/O, consider asyncio.

Asyncio

Asyncio uses cooperative multitasking with an event loop. Tasks voluntarily yield control at await points, enabling high concurrency without threads.

asyncio.py

Python

1	import asyncio
2
3	# Coroutine function — defined with async def
4	async def fetch_data(url: str) -> str:
5	print(f"fetching {url}")
6	await asyncio.sleep(1) # simulate I/O
7	return f"data from {url}"
8
9	# Running a coroutine
10	async def main():
11	result = await fetch_data("example.com")
12	print(result)
13
14	asyncio.run(main())
15
16	# Running concurrent tasks
17	async def main_concurrent():
18	tasks = [
19	fetch_data(f"url_{i}")
20	for i in range(5)
21	]
22	results = await asyncio.gather(*tasks)
23	print(results)
24
25	asyncio.run(main_concurrent())
26	# All 5 fetch concurrently — takes ~1s total, not 5s
27
28	# asyncio.create_task — fire and forget
29	async def background_work():
30	while True:
31	await asyncio.sleep(1)
32	print("background tick")
33
34	async def main_with_bg():
35	task = asyncio.create_task(background_work())
36	await asyncio.sleep(3)
37	task.cancel() # stop background task
38	print("done")
39
40	asyncio.run(main_with_bg())
41
42	# Timeouts
43	async def slow():
44	await asyncio.sleep(10)
45	return "done"
46
47	async def with_timeout():
48	try:
49	result = await asyncio.wait_for(slow(), timeout=2.0)
50	except asyncio.TimeoutError:
51	print("timed out!")
52
53	# Async context managers
54	class AsyncResource:
55	async def __aenter__(self):
56	print("acquiring")
57	await asyncio.sleep(0.1)
58	return self
59
60	async def __aexit__(self, *args):
61	print("releasing")
62	await asyncio.sleep(0.1)
63
64	async def use_async():
65	async with AsyncResource() as res:
66	print("using resource")
67
68	# Async iteration
69	class AsyncRange:
70	def __init__(self, n):
71	self.n = n
72	self.i = 0
73
74	def __aiter__(self):
75	return self
76
77	async def __anext__(self):
78	if self.i >= self.n:
79	raise StopAsyncIteration
80	await asyncio.sleep(0.1)
81	self.i += 1
82	return self.i
83
84	async def iterate():
85	async for num in AsyncRange(5):
86	print(num)
87
88	# Async queues — producer/consumer
89	async def producer(queue, n):
90	for i in range(n):
91	await queue.put(i)
92	await asyncio.sleep(0.1)
93	await queue.put(None) # sentinel
94
95	async def consumer(queue):
96	while True:
97	item = await queue.get()
98	if item is None:
99	break
100	print(f"consumed {item}")
101
102	async def main_queue():
103	q = asyncio.Queue()
104	await asyncio.gather(
105	producer(q, 5),
106	consumer(q),
107	)

✓

best practice

Asyncio excels at I/O-bound workloads with many concurrent connections (web servers, API gateways, chat servers). It is most effective when all I/O operations use async libraries (aiohttp, asyncpg, httpx).

Multiprocessing

Multiprocessing spawns separate Python processes, each with its own GIL. This enables true parallel execution for CPU-bound tasks.

multiprocessing.py

Python

1	from multiprocessing import Process, Pool, Queue, cpu_count
2	import os
3
4	# Basic process
5	def worker(n: int) -> int:
6	print(f"PID {os.getpid()} working on {n}")
7	return n * n
8
9	processes = []
10	for i in range(4):
11	p = Process(target=worker, args=(i,))
12	processes.append(p)
13	p.start()
14	for p in processes:
15	p.join()
16
17	# Process Pool — easy parallel execution
18	def cpu_heavy(n: int) -> int:
19	return sum(i * i for i in range(n))
20
21	with Pool(processes=cpu_count()) as pool:
22	# map — distribute work across processes
23	results = pool.map(cpu_heavy, [10_000, 20_000, 30_000])
24	print(results)
25
26	# async variants
27	result = pool.apply_async(cpu_heavy, (50_000,))
28	print(result.get()) # blocks until ready
29
30	# Shared memory
31	from multiprocessing import Value, Array
32
33	counter = Value("i", 0) # 'i' = signed int
34	arr = Array("d", [0.0] * 10) # 'd' = double
35
36	def increment(counter):
37	with counter.get_lock(): # synchronize access
38	counter.value += 1
39
40	# Queue — inter-process communication
41	def producer(q):
42	for i in range(5):
43	q.put(i)
44	q.put(None)
45
46	def consumer(q):
47	while True:
48	item = q.get()
49	if item is None:
50	break
51	print(f"got {item}")
52
53	q = Queue()
54	p1 = Process(target=producer, args=(q,))
55	p2 = Process(target=consumer, args=(q,))
56	p1.start()
57	p2.start()
58	p1.join()
59	p2.join()
60
61	# Best practice: use concurrent.futures for simple cases
62	from concurrent.futures import ProcessPoolExecutor
63
64	with ProcessPoolExecutor() as executor:
65	futures = [executor.submit(cpu_heavy, n) for n in [10_000, 20_000]]
66	for f in concurrent.futures.as_completed(futures):
67	print(f.result())

✓

best practice

For simple parallel mapping, use concurrent.futures.ProcessPoolExecutor — it has a cleaner API than multiprocessing.Pool. For complex inter-process communication, use multiprocessing directly.

Concurrent Futures

The concurrent.futures module provides a high-level interface for asynchronously executing callables using threads or processes.

concurrent_futures.py

Python

1	from concurrent.futures import (
2	ThreadPoolExecutor, ProcessPoolExecutor,
3	as_completed, wait, FIRST_COMPLETED
4	)
5	import time
6	import urllib.request
7
8	# ThreadPoolExecutor — I/O-bound
9	def fetch_url(url: str) -> tuple[str, int]:
10	with urllib.request.urlopen(url, timeout=5) as resp:
11	return url, len(resp.read())
12
13	urls = [
14	"https://python.org",
15	"https://github.com",
16	"https://stackoverflow.com",
17	]
18
19	with ThreadPoolExecutor(max_workers=10) as executor:
20	# Submit individual tasks
21	futures = [executor.submit(fetch_url, url) for url in urls]
22
23	# Process as they complete
24	for future in as_completed(futures):
25	url, size = future.result()
26	print(f"{url}: {size} bytes")
27
28	# Or use map (preserves order)
29	results = executor.map(fetch_url, urls)
30
31	# Wait for first completion
32	with ThreadPoolExecutor() as executor:
33	futures = [executor.submit(fetch_url, url) for url in urls]
34	done, not_done = wait(futures, return_when=FIRST_COMPLETED)
35	first = done.pop().result()
36	print(f"First finished: {first}")
37
38	# ProcessPoolExecutor — CPU-bound
39	def calculate(n: int) -> int:
40	return sum(i ** 2 for i in range(n))
41
42	with ProcessPoolExecutor(max_workers=4) as executor:
43	results = executor.map(calculate, [10_000, 20_000, 30_000])
44	for result in results:
45	print(result)
46
47	# Future API
48	with ThreadPoolExecutor() as executor:
49	future = executor.submit(fetch_url, urls[0])
50	# Check if done (non-blocking)
51	print(future.done()) # → False (likely)
52	# Wait with timeout
53	try:
54	result = future.result(timeout=10)
55	except TimeoutError:
56	print("timed out")
57	# Cancel (if not started)
58	future.cancel()
59	print(future.cancelled())

Choosing the Right Model

choosing.py

Python

1	# Rule of thumb:
2	# - I/O-bound, many concurrent tasks → asyncio (single-threaded)
3	# - I/O-bound, blocking libraries → ThreadPoolExecutor
4	# - CPU-bound, parallel processing → ProcessPoolExecutor
5	# - Complex state sharing → threading + locks
6	# - Pipeline processing → multiprocessing.Queue + Pool
7
8	# Hybrid approach — combine models
9	import asyncio
10	from concurrent.futures import ProcessPoolExecutor
11
12	async def run_cpu_bound():
13	loop = asyncio.get_running_loop()
14	with ProcessPoolExecutor() as pool:
15	# Run CPU-bound work without blocking the event loop
16	result = await loop.run_in_executor(
17	pool, cpu_heavy, 100_000
18	)
19	return result
20
21	async def main_hybrid():
22	# Mix async I/O and CPU-bound work
23	data = await fetch_data("api.example.com")
24	result = await run_cpu_bound()
25	return result

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

X GitHub YouTube Instagram Facebook Buy me a coffee Sponsor