🚧 Concurrency Limits

The Problem: Unbounded Fan-Out Kills

# ❌ Anti-pattern: unbounded fan-out — 10,000 concurrent calls at once
import asyncio, httpx

async def enrich_all(records, client):
    # fires ALL 10k requests immediately: DDoSes the downstream, exhausts the pool
    return await asyncio.gather(*(client.get(f"/enrich/{r}") for r in records))

Concurrency Limit vs Rate Limit — Not the Same Thing

These get conflated constantly, and they protect against different failures. This page is about concurrency.

Sizing the Limit with Little's Law

Don't pick the number by vibes. Little's Law gives you the right ceiling from measurements you already have.

# Little's Law:   concurrency = throughput × latency
#
#   L (in-flight)  =  λ (arrival/target rate)  ×  W (avg service latency)
#
# Example: downstream comfortably serves 2,000 req/s, each call averages 50ms:
#   L = 2000 req/s × 0.050 s = 100 concurrent requests
#
# → a semaphore / pool of ~100 keeps the downstream at its sweet spot.
#   Go higher and you just build a queue (latency ↑, throughput flat).
#   Go lower and you leave throughput on the table.

How a Semaphore Gate Works

A semaphore is just a counter of N permits. Acquire to enter, release to leave; when all permits are out, the next arrival waits. That's the entire mechanism — a gate that lets at most N through at a time.

In Threading — Semaphore(N) and sized pools

Two ways to cap threads: wrap the critical section in a threading.Semaphore, or — simpler and usually better — just size the ThreadPoolExecutor. The pool's max_workers is a concurrency limit.

import threading, requests
from concurrent.futures import ThreadPoolExecutor

session = requests.Session()
sem = threading.Semaphore(50)          # at most 50 downstream calls in flight

def call(record):
    with sem:                          # blocks the thread when 50 are already out
        return session.get(f"https://api/enrich/{record}", timeout=5).json()

# Even without the semaphore, the pool size caps concurrency at 50 by itself:
with ThreadPoolExecutor(max_workers=50) as pool:
    results = list(pool.map(call, records))

In Asyncio — Semaphore(N) around each task

This is the canonical async pattern, and the fix for the 10k-gather problem above: still gather everything, but make every coroutine acquire a shared semaphore first, so only N run concurrently.

import asyncio, httpx

async def enrich_all(records):
    sem = asyncio.Semaphore(50)                 # ✅ at most 50 in flight
    async with httpx.AsyncClient(timeout=5) as client:
        async def call(record):
            async with sem:                     # gate: 51st coroutine awaits here
                r = await client.get(f"/enrich/{record}")
                return r.json()
        return await asyncio.gather(*(call(r) for r in records))

# vs the ❌ naive version that fires all 10,000 at once and DDoSes the downstream:
#   return await asyncio.gather(*(client.get(f"/enrich/{r}") for r in records))

In Multiprocessing — Pool(processes=N)

For CPU-bound fan-out, the concurrency limit is simply the process count. Because of the GIL, real parallelism tops out at your core count anyway — so N = os.cpu_count() is usually the right ceiling. More processes than cores just adds context-switching and memory, not speed.

import os
from multiprocessing import Pool

def crunch(chunk):
    return heavy_cpu_work(chunk)          # e.g. hashing, parsing, numeric work

if __name__ == "__main__":
    n = os.cpu_count()                    # cap parallelism at the core count
    with Pool(processes=n) as pool:       # ✅ Pool IS the concurrency limit
        results = pool.map(crunch, chunks)
        # chunks beyond `n` queue internally and start as workers free up

Where to Put the Limit

Match the limit to the actual bottleneck — capping the wrong resource does nothing useful.

Key Takeaways