Async & Concurrency for Throughput

Python concurrency models target different bottlenecks: asyncio for I/O parallelism, multiprocessing for CPU parallelism, and threading for I/O with blocking libraries.

Recipe

import asyncio
import httpx
 
async def fetch_all(urls: list[str]) -> list[int]:
    async with httpx.AsyncClient() as client:
        responses = await asyncio.gather(*[client.get(u) for u in urls])
        return [r.status_code for r in responses]

When to reach for this:

• Many concurrent HTTP/database requests (I/O-bound)
• CPU-bound work on multiple cores (multiprocessing)
• Throughput matters more than single-request latency

Working Example

import asyncio
import httpx
from concurrent.futures import ProcessPoolExecutor
 
# I/O-bound: asyncio
async def fetch_urls(urls: list[str]) -> list[str]:
    async with httpx.AsyncClient(timeout=10.0) as client:
        tasks = [client.get(url) for url in urls]
        responses = await asyncio.gather(*tasks, return_exceptions=True)
        return [r.text[:100] if isinstance(r, httpx.Response) else str(r) for r in responses]
 
# CPU-bound: process pool
def compute(n: int) -> int:
    return sum(i * i for i in range(n))
 
async def run_cpu_bound(items: list[int]) -> list[int]:
    loop = asyncio.get_running_loop()
    with ProcessPoolExecutor() as pool:
        return list(await asyncio.gather(*[
            loop.run_in_executor(pool, compute, n) for n in items
        ]))
 
async def main():
    urls = ["https://httpbin.org/get"] * 20
    io_results = await fetch_urls(urls)
    cpu_results = await run_cpu_bound([100_000] * 4)
    print(len(io_results), sum(cpu_results))
 
asyncio.run(main())

What this demonstrates:

• asyncio.gather runs I/O concurrently on one thread
• ProcessPoolExecutor bypasses GIL for CPU work
• return_exceptions=True prevents one failure from canceling all
• Combine models: async orchestration + process pool for CPU stages

Deep Dive

Model Selection

Gotchas

• asyncio for CPU work - no speedup, adds complexity. Fix: multiprocessing or vectorization.
• Unbounded concurrency - overwhelms server or hits connection limits. Fix: asyncio.Semaphore(50) to cap parallel requests.
• Shared mutable state across threads - data races. Fix: queues, locks, or process isolation.
• Creating processes per task - fork overhead. Fix: reuse ProcessPoolExecutor.
• Blocking call in async function - freezes event loop. Fix: await asyncio.to_thread(blocking_fn) or async library.

Alternatives

FAQs

Related

• 30 Async Rules - asyncio pitfalls
• Concurrency - GIL and models
• Caching & Memoization - reduce I/O
• Performance Best Practices - model selection

Stack versions: This page was written for Python 3.14.0, FastAPI 0.115+, Django 5.2, Flask 3.1, Pydantic 2, PyTorch 2.6+, pandas 2.2+, Polars 1.x, ruff 0.9+, and uv 0.6+.