🏗️ Case Studies: Putting It Together

How to Read These

Every earlier page taught one move in isolation. Real systems need a combination, and the combination is dictated by the bottleneck. Each case study runs the same loop from Optimizations 101: Measure → find the bottleneck → pick the model → apply techniques → protect it.

Case Study 1 — High-Throughput Web Scraper

Find the Bottleneck

A fetch is ~200ms of waiting (DNS + TCP + TLS + server think time) and ~1ms of CPU (parse a small JSON body). That's ~99.5% I/O wait. The CPU is idle; the constraint is how many requests we can have in flight at once without melting sockets or memory. Classic I/O-bound.

Which Model & Why → ASYNCIO

Threads would work but a thread per in-flight request costs ~8MB of stack and real context-switch overhead; scaling to thousands is wasteful. Multiprocessing gives CPU parallelism we don't need — there's no CPU to parallelize. Asyncio holds tens of thousands of waiting sockets on one core for almost nothing. See the model comparison; for pushing async hard see async at scale.

Techniques Applied

• Connection pooling — one AsyncClient for the whole run.
• Concurrency limits — a global Semaphore caps in-flight requests.
• Backpressure — a bounded results Queue; producers block when the writer falls behind.
• Rate limiting — per-host politeness so we don't hammer one server.
• Resilience — per-request timeout + retry with exponential backoff and jitter.
• Memory optimization — stream URLs in, stream results out; nothing full-materializes.

Runnable Skeleton

import asyncio
import json
import random
from collections import defaultdict
from urllib.parse import urlsplit

import httpx

# --- tunables (size from Little's Law: throughput * latency) ---
GLOBAL_CONCURRENCY = 600     # ~3000 req/s * 0.2s
PER_HOST_CONCURRENCY = 4     # politeness: never flood one host
RESULT_QUEUE_MAX = 10_000    # backpressure bound on unwritten results
REQUEST_TIMEOUT = 10.0
MAX_RETRIES = 3

limits = httpx.Limits(max_connections=GLOBAL_CONCURRENCY,
                      max_keepalive_connections=200)


def url_source(path: str):
    """Stream URLs from disk — never load 1M into a list (memory-bound trap)."""
    with open(path) as fh:
        for line in fh:
            url = line.strip()
            if url:
                yield url


async def fetch_one(client: httpx.AsyncClient, url: str) -> dict | None:
    """One fetch with timeout + retry/backoff. Returns a record or None on give-up."""
    for attempt in range(MAX_RETRIES):
        try:
            r = await client.get(url, timeout=REQUEST_TIMEOUT)
            r.raise_for_status()
            # tiny CPU cost — fine to do inline; if it grew, offload (see async_at_scale)
            return {"url": url, "status": r.status_code, "len": len(r.content)}
        except (httpx.HTTPError, httpx.TimeoutException):
            if attempt == MAX_RETRIES - 1:
                return {"url": url, "error": True}
            # exponential backoff + full jitter — avoid retry stampedes
            await asyncio.sleep((2 ** attempt) * 0.5 * (0.5 + random.random()))
    return None


async def writer(queue: "asyncio.Queue[dict | None]", out_path: str) -> None:
    """Single consumer: the only thing that touches the file. Bounds memory."""
    with open(out_path, "w") as fh:
        while True:
            record = await queue.get()
            if record is None:          # sentinel = done
                queue.task_done()
                return
            fh.write(json.dumps(record) + "\n")
            queue.task_done()


async def scrape(url_path: str, out_path: str) -> None:
    global_sem = asyncio.Semaphore(GLOBAL_CONCURRENCY)
    host_sems: dict[str, asyncio.Semaphore] = defaultdict(
        lambda: asyncio.Semaphore(PER_HOST_CONCURRENCY)
    )
    results: asyncio.Queue = asyncio.Queue(maxsize=RESULT_QUEUE_MAX)

    async with httpx.AsyncClient(limits=limits, follow_redirects=True) as client:
        writer_task = asyncio.create_task(writer(results, out_path))

        async def worker(url: str) -> None:
            host = urlsplit(url).netloc
            # two nested caps: global budget AND per-host politeness
            async with global_sem, host_sems[host]:
                record = await fetch_one(client, url)
            if record is not None:
                await results.put(record)   # blocks here if writer is behind = backpressure

        # Spawn in bounded waves so we never hold 1M Task objects in memory at once.
        pending: set[asyncio.Task] = set()
        for url in url_source(url_path):
            pending.add(asyncio.create_task(worker(url)))
            if len(pending) >= GLOBAL_CONCURRENCY * 2:
                done, pending = await asyncio.wait(
                    pending, return_when=asyncio.FIRST_COMPLETED
                )
        if pending:
            await asyncio.gather(*pending)

        await results.put(None)   # tell writer to stop
        await writer_task


if __name__ == "__main__":
    asyncio.run(scrape("urls.txt", "results.jsonl"))

Case Study 2 — High-QPS JSON API

Find the Bottleneck

Profiling one request: ~5ms DB, ~40ms downstream call (I/O wait), ~8ms JSON serialization (CPU, GIL-held), ~2ms framework. So per request it's ~85% I/O-bound, but with a real CPU tax on serialization that the GIL serializes across all concurrent requests. And there's a downstream-bound failure mode: when pricing slows to 2s, naive code lets requests pile up until the queue eats all memory.

Which Model & Why → THREADS (per-process pool) behind multiple processes

The honest tradeoff:

We pick threads behind N worker processes (e.g. gunicorn -w 4 -k gthread): it uses the team's existing sync DB and client code, the GIL-releasing I/O lets threads overlap the downstream wait, and the 4 processes give the CPU parallelism serialization needs. Each process owns its own sized SQLAlchemy pool. (Async is the better pick if the whole stack is already non-blocking and concurrency is enormous — see async at scale.)

Techniques Applied

• Thread pool tuning — size the pool for the I/O wait, one pool per process.
• Connection pooling — a bounded SQLAlchemy engine per process; total conns under Postgres max_connections.
• Caching — TTL cache with single-flight so a cache miss under load doesn't stampede the DB (thundering herd).
• Resilience — hard timeout on the downstream, plus a circuit breaker so a slow pricing service fails fast instead of tying up threads.
• Backpressure / load shedding — when the work queue is full, return 503 immediately rather than queueing into a latency spiral.

Runnable Handler Sketch

import threading
import time

import requests
from sqlalchemy import create_engine, text

# --- one sized pool per worker PROCESS (see connection_pooling) ---
engine = create_engine(
    "postgresql+psycopg://app@db/prod",
    pool_size=20, max_overflow=10, pool_pre_ping=True,
)
downstream = requests.Session()   # pooled keep-alive conns to pricing service


# ---------- single-flight TTL cache (beats cache stampede) ----------
class SingleFlightCache:
    def __init__(self, ttl: float):
        self._ttl = ttl
        self._store: dict[str, tuple[float, object]] = {}
        self._locks: dict[str, threading.Lock] = {}
        self._guard = threading.Lock()

    def get(self, key: str, loader):
        now = time.monotonic()
        hit = self._store.get(key)
        if hit and now - hit[0] < self._ttl:
            return hit[1]
        # only ONE thread loads a given key; the rest wait on its lock
        with self._guard:
            lock = self._locks.setdefault(key, threading.Lock())
        with lock:
            hit = self._store.get(key)          # re-check: someone may have filled it
            if hit and time.monotonic() - hit[0] < self._ttl:
                return hit[1]
            value = loader()
            self._store[key] = (time.monotonic(), value)
            return value


cache = SingleFlightCache(ttl=5.0)


# ---------- circuit breaker for the downstream ----------
class CircuitBreaker:
    def __init__(self, fail_max: int, reset_after: float):
        self._fail_max, self._reset_after = fail_max, reset_after
        self._fails = 0
        self._opened_at = 0.0
        self._lock = threading.Lock()

    def allow(self) -> bool:
        with self._lock:
            if self._fails < self._fail_max:
                return True
            if time.monotonic() - self._opened_at > self._reset_after:
                self._fails = 0          # half-open: let one probe through
                return True
            return False                 # OPEN — fail fast, don't tie up a thread

    def record(self, ok: bool) -> None:
        with self._lock:
            if ok:
                self._fails = 0
            else:
                self._fails += 1
                self._opened_at = time.monotonic()


breaker = CircuitBreaker(fail_max=20, reset_after=5.0)


# ---------- the request handler (runs on a pool thread) ----------
class Overloaded(Exception):
    """Signals the framework to return 503 (load shedding)."""


def handle(product_id: str, queue_depth: int, queue_limit: int) -> dict:
    # 1. LOAD SHEDDING: if the work queue is saturated, reject NOW, don't queue.
    if queue_depth >= queue_limit:
        raise Overloaded()

    # 2. CACHE + single-flight: collapse duplicate concurrent misses into one DB read.
    def load_product():
        with engine.connect() as conn:
            row = conn.execute(
                text("SELECT id, name, base FROM products WHERE id = :id"),
                {"id": product_id},
            ).one()
            return dict(row._mapping)

    product = cache.get(product_id, load_product)

    # 3. DOWNSTREAM behind a circuit breaker + hard timeout.
    price = product["base"]
    if breaker.allow():
        try:
            r = downstream.get(f"http://pricing/quote/{product_id}", timeout=0.1)
            r.raise_for_status()
            price = r.json()["price"]
            breaker.record(ok=True)
        except requests.RequestException:
            breaker.record(ok=False)      # fall back to base price, stay up
    # else: breaker OPEN -> skip the call entirely, serve base price fast

    # 4. Serialization is the GIL-bound CPU cost — parallelized by running N processes.
    return {"id": product["id"], "name": product["name"], "price": price}

Case Study 3 — Batch Data Pipeline

Find the Bottleneck

Profile a sample chunk: parsing + scoring is ~95% CPU, all of it pure-Python, all of it GIL-held. Reading the file is trivial by comparison. This is textbook CPU-bound — and because of the GIL, threads and async give zero speedup here: they'd all fight for one core.

Which Model & Why → MULTIPROCESSING

Multiprocessing is the only model that gives real CPU parallelism in CPython — each process has its own GIL, so 16 processes genuinely use 16 cores (see CPU-bound scaling and choosing the model). Threads/async are ruled out by the GIL.

Techniques Applied

• CPU-bound scaling — ProcessPoolExecutor with a tuned chunksize to amortize the per-task pickling cost.
• Memory optimization — stream input in chunks with a generator so 50 GB never lands in RAM; results stream out in order.
• Shared memory — the 2 GB lookup lives in one SharedMemory block mapped read-only into every worker (loaded once, not 16×).
• Per-process init — the initializer= pattern attaches each worker to the shared block once at startup.
• Backpressure — executor.map with a bounded chunk stream keeps only a window of chunks in flight, so producers can't outrun consumers.

Runnable Skeleton

import json
from concurrent.futures import ProcessPoolExecutor
from itertools import islice
from multiprocessing import shared_memory

WORKERS = 16
CHUNK_ROWS = 5_000        # rows per task — big enough to dwarf pickling overhead
LOOKUP_SHM_NAME = "enrich_lookup"

# Each worker attaches to the shared lookup ONCE and reuses it for every chunk.
_lookup: dict | None = None


def build_shared_lookup(lookup: dict) -> shared_memory.SharedMemory:
    """Main process: serialize the 2 GB table into ONE shared block."""
    import pickle
    blob = pickle.dumps(lookup, protocol=5)
    shm = shared_memory.SharedMemory(name=LOOKUP_SHM_NAME, create=True, size=len(blob))
    shm.buf[: len(blob)] = blob
    return shm


def init_worker() -> None:
    """Runs ONCE per worker process — map the shared lookup, don't copy it."""
    import pickle
    global _lookup
    shm = shared_memory.SharedMemory(name=LOOKUP_SHM_NAME)   # attach, no copy
    _lookup = pickle.loads(bytes(shm.buf))                   # deserialize once per proc
    # NOTE: this deserialization is the one unavoidable per-process cost; the raw
    # 2 GB bytes are shared, so we pay RAM for the dict once per core, not the blob.


def score_chunk(rows: list[str]) -> list[str]:
    """Pure-CPU work — parse, enrich against the shared lookup, score. Runs on its own core."""
    assert _lookup is not None
    out: list[str] = []
    for line in rows:
        rec = json.loads(line)
        rec["region"] = _lookup.get(rec["zip"], "unknown")   # enrichment
        rec["score"] = heavy_score(rec)                       # the CPU-bound hotspot
        out.append(json.dumps(rec))
    return out


def heavy_score(rec: dict) -> float:
    # stand-in for the real pure-Python scoring function (the GIL-bound hotspot)
    return sum(ord(c) for c in rec.get("name", "")) / 100.0


def chunked(path: str, size: int):
    """Stream the 50 GB file in fixed-row chunks — RAM stays flat regardless of file size."""
    with open(path) as fh:
        while True:
            block = list(islice(fh, size))
            if not block:
                return
            yield block


def run(in_path: str, out_path: str, lookup: dict) -> None:
    shm = build_shared_lookup(lookup)
    try:
        with ProcessPoolExecutor(WORKERS, initializer=init_worker) as pool, \
                open(out_path, "w") as out:
            # map streams chunks in, yields results IN ORDER as they finish.
            # chunksize=1 here because each `chunk` is already CHUNK_ROWS rows —
            # the batching is done by chunked(), not by the executor.
            for scored_rows in pool.map(score_chunk, chunked(in_path, CHUNK_ROWS),
                                        chunksize=1):
                out.write("\n".join(scored_rows) + "\n")   # stream results out in order
    finally:
        shm.close()
        shm.unlink()      # free the shared block


if __name__ == "__main__":
    big_lookup = {str(z): f"region-{z % 50}" for z in range(2_000_000)}  # stand-in 2GB table
    run("input.jsonl", "output.jsonl", big_lookup)

Decision Recap

The same bottleneck-first logic, three different answers: