Performance Patterns — Choosing the Right Concurrency Model
Python · Concurrency · Run with python3
Explanation
Python offers three concurrency models: `threading` (I/O-bound, shared memory, GIL limited), `multiprocessing` (CPU-bound, process isolation, higher overhead), and `asyncio` (I/O-bound, single-thread, highest concurrency per memory). The rules: I/O-bound + simple → `asyncio`; I/O-bound + sync libraries → `threading`; CPU-bound → `multiprocessing` or C extensions (NumPy, Cython). Mix: use `asyncio.to_thread()` to run blocking code in a thread pool from an async context.
Real-World Scenario
A FastAPI endpoint makes several database queries and an HTTP call. `async def` with `asyncpg` (async Postgres driver) and `httpx.AsyncClient` keeps it on a single thread with high throughput. If the endpoint also needs to run a CPU-heavy image resize, `await asyncio.to_thread(resize_image, img)` offloads it to a thread pool without blocking the event loop. ```python import asyncio import time import concurrent.futures def cpu_heavy(n: int) -> int: """Blocks the thread — not suitable to call directly in async.""" return sum(i * i for i in range(n)) async def mixed_workload() -> None: # I/O-bound: run concurrently in the event loop io_tasks = [asyncio.sleep(0.1) for _ in range(5)] await asyncio.gather(*io_tasks) # CPU-bound: offload to a thread pool loop = asyncio.get_event_loop() with concurrent.futures.ThreadPoolExecutor() as pool: result = await loop.run_in_executor(pool, cpu_heavy, 1_000_000) print(f"CPU result: {result}") # Python 3.9+ shorthand result2 = await asyncio.to_thread(cpu_heavy, 500_000) print(f"to_thread result: {result2}") asyncio.run(mixed_workload()) ```
Quick Snippet
import asyncio
import time
import concurrent.futures
def cpu_heavy(n: int) -> int:
"""Blocks the thread — not suitable to call directly in async."""
return sum(i * i for i in range(n))
async def mixed_workload() -> None:
# I/O-bound: run concurrently in the event loop
io_tasks = [asyncio.sleep(0.1) for _ in range(5)]
await asyncio.gather(*io_tasks)
# CPU-bound: offload to a thread pool
loop = asyncio.get_event_loop()
with concurrent.futures.ThreadPoolExecutor() as pool:
result = await loop.run_in_executor(pool, cpu_heavy, 1_000_000)
print(f"CPU result: {result}")
# Python 3.9+ shorthand
result2 = await asyncio.to_thread(cpu_heavy, 500_000)
print(f"to_thread result: {result2}")
asyncio.run(mixed_workload())Full Example
# Compile: python3 concurrency_patterns.py
# Output:
# CPU result printed
# to_thread result printed
import asyncio
import concurrent.futures
def cpu_heavy(n: int) -> int:
return sum(i * i for i in range(n))
async def mixed_workload() -> None:
io_tasks = [asyncio.sleep(0.01) for _ in range(5)]
await asyncio.gather(*io_tasks)
loop = asyncio.get_event_loop()
with concurrent.futures.ThreadPoolExecutor() as pool:
result = await loop.run_in_executor(pool, cpu_heavy, 100_000)
print(f"CPU result: {result}")
result2 = await asyncio.to_thread(cpu_heavy, 50_000)
print(f"to_thread result: {result2}")
asyncio.run(mixed_workload())
# Summary
print("\nConcurrency model guide:")
print(" I/O-bound + simple → asyncio")
print(" I/O-bound + sync lib → threading")
print(" CPU-bound → multiprocessing")
print(" CPU-bound in async → asyncio.to_thread")
python3 concurrency_patterns.py