🌀 Async at Scale

The Cardinal Rule: Never Block the Event Loop

# ❌ The bug: a synchronous call inside a coroutine freezes ALL tasks
import asyncio, time

async def handle(n):
    time.sleep(1)          # SYNC sleep — blocks the loop, not just this task
    return n

async def main():
    # you'd hope for ~1s total; you get ~5s because the loop can't interleave
    await asyncio.gather(*(handle(i) for i in range(5)))

asyncio.run(main())

The Fixes: Await, Async Libs, or Offload

Three ways out, in order of preference:

import asyncio

# ✅ 1. Use the async equivalent — yields control back to the loop
async def handle(n):
    await asyncio.sleep(1)     # ASYNC sleep — 5 of these finish in ~1s total
    return n

async def main():
    await asyncio.gather(*(handle(i) for i in range(5)))   # ~1s, not ~5s

asyncio.run(main())

# ✅ 2. Use an async I/O library instead of a blocking one
import httpx     # not `requests`

async def fetch(client, url):
    r = await client.get(url)      # non-blocking; loop serves others while waiting
    return r.json()

async def main(urls):
    async with httpx.AsyncClient() as client:      # reuse one client + pool
        return await asyncio.gather(*(fetch(client, u) for u in urls))

# ✅ 3. Can't avoid a blocking call? Offload it to a thread — the loop stays free
import asyncio, requests

def blocking_get(url):
    return requests.get(url).json()      # legacy sync lib you can't replace

async def fetch(url):
    # asyncio.to_thread (3.9+) runs it in the default ThreadPoolExecutor
    return await asyncio.to_thread(blocking_get, url)

# older / explicit form: loop.run_in_executor(None, blocking_get, url)

# ✅ CPU-bound work from async: offload to PROCESSES, not threads
import asyncio
from concurrent.futures import ProcessPoolExecutor

def crunch(n):
    return sum(i * i for i in range(n))     # heavy CPU — would stall the loop

async def main():
    loop = asyncio.get_running_loop()
    with ProcessPoolExecutor() as pool:
        # runs in a separate interpreter/core; the loop keeps serving I/O
        return await loop.run_in_executor(pool, crunch, 50_000_000)

asyncio.run(main())

Bounding Fan-Out: Don't gather a Million Coroutines

Cap concurrency with a Semaphore (see concurrency limits), or run a fixed pool of workers off an asyncio.Queue.

# ✅ Semaphore: still create the coroutines, but only N run at once
import asyncio

sem = asyncio.Semaphore(100)          # at most 100 in flight

async def fetch(client, url):
    async with sem:                   # acquire → work → release
        r = await client.get(url)
        return r.json()

async def main(client, urls):
    return await asyncio.gather(*(fetch(client, u) for u in urls))

# ✅ Queue + worker pool: bounded memory even for a truly huge stream of jobs
import asyncio

async def worker(queue, results):
    while True:
        item = await queue.get()
        try:
            results.append(await do_work(item))
        finally:
            queue.task_done()

async def main(items):
    queue = asyncio.Queue(maxsize=1000)        # backpressure: producer waits when full
    results = []
    workers = [asyncio.create_task(worker(queue, results)) for _ in range(50)]
    for item in items:                          # feed lazily — never materialize all
        await queue.put(item)
    await queue.join()                          # wait until every item is processed
    for w in workers:
        w.cancel()
    return results

gather vs as_completed vs TaskGroup

# gather — collect everything, order preserved
import asyncio
results = await asyncio.gather(fetch(a), fetch(b), fetch(c))   # [a, b, c] order

# as_completed — process the fastest responses first (streaming results)
import asyncio
for coro in asyncio.as_completed([fetch(u) for u in urls]):
    result = await coro          # yields in COMPLETION order, not input order
    handle(result)

# TaskGroup (3.11+) — the modern default: structured, auto-cancels on failure
import asyncio

async def main(urls):
    results = []
    async with asyncio.TaskGroup() as tg:        # exits only when all tasks done
        for u in urls:
            tg.create_task(fetch_into(u, results))
    # if any task raised, siblings were cancelled and an ExceptionGroup propagates here
    return results

Deadlines with asyncio.timeout() (3.11+)

Under load, a slow downstream must not tie up a task forever. Wrap any await in a deadline; on expiry it raises TimeoutError and cancels the inner work cleanly.

import asyncio

async def fetch_with_deadline(client, url):
    try:
        async with asyncio.timeout(2.0):        # 2s budget for everything inside
            r = await client.get(url)
            return r.json()
    except TimeoutError:
        return None                             # shed the slow one, keep serving the rest

Faster Loop: uvloop

uvloop is a drop-in event loop built on libuv. Install it and switch with one line — no code changes — for roughly 2–4× higher throughput on network-heavy workloads.

# pip install uvloop   (Linux / macOS)
import asyncio
import uvloop

# Python 3.11+: pass the loop factory to asyncio.run
asyncio.run(main(), loop_factory=uvloop.new_event_loop)

# Older / global install (still widely used):
#   uvloop.install()
#   asyncio.run(main())

Profiling the Loop

The loop can't tell you it's stalled — you have to ask. Two cheap built-in probes catch coroutines that hog the thread:

import asyncio

async def main():
    loop = asyncio.get_running_loop()
    loop.slow_callback_duration = 0.1     # log any callback that runs > 100ms
    loop.set_debug(True)                  # warns on slow callbacks + un-awaited coros
    # ... run your app ...

# Or from the shell, no code change:
#   PYTHONASYNCIODEBUG=1 python app.py

Blocking Offender → Async Fix