Python asyncio Shutdown Hangs — CancelledError Swallowing

Every modern Python backend eventually hits the same wall: your code spends 90% of its time waiting — waiting for a database to respond, waiting for an API call to return, waiting for a file to be read off disk. Threads are the classic answer, but they carry a tax: memory overhead, the GIL playing referee, and race conditions that appear only in production at 3am. There's a better tool for I/O-bound concurrency, and it's been in the standard library since Python 3.4, matured dramatically in 3.7, and is now the backbone of frameworks like FastAPI, aiohttp, and Starlette.

asyncio solves the 'waiting problem' through cooperative multitasking. Instead of the OS forcibly switching between threads, your coroutines voluntarily yield control back to an event loop whenever they'd otherwise block. One thread, one event loop, potentially thousands of concurrent operations — all without a mutex in sight. The catch is that everything in your call stack must understand this contract, which is what makes asyncio feel like learning a new dialect of Python at first.

By the end of this article you'll understand how the event loop actually schedules work, how coroutines differ from generators at the bytecode level, why blocking the event loop is the cardinal sin of async Python, and how to structure real production services — including proper cancellation, timeouts, error propagation, and the patterns that separate async code that scales from async code that silently serializes everything.

Why asyncio Shutdown Hangs — CancelledError Swallowing

asyncio is Python's built-in library for writing concurrent code using the async/await syntax. It implements cooperative multitasking on a single thread via an event loop that schedules coroutines — functions defined with async def that can suspend execution at await points. The core mechanic is explicit yielding: a coroutine voluntarily pauses, allowing the loop to run other coroutines until the awaited operation completes.

In practice, asyncio works by wrapping I/O-bound operations (network requests, file reads, sleep) into awaitable objects. The event loop polls registered file descriptors and timers, resuming coroutines when data arrives or time passes. Key properties: all coroutines share one thread, so no locks are needed for shared state, but a long-running CPU-bound coroutine blocks the entire loop. Cancellation is cooperative — a CancelledError is raised at the next await, but if a coroutine catches and ignores it, shutdown hangs.

Use asyncio when your application is I/O-bound with many concurrent connections — web servers, API gateways, data pipelines. It matters because it enables handling tens of thousands of connections per process without the overhead of threads or processes. The critical nuance: asyncio does not make Python faster; it makes waiting faster. If your workload is CPU-bound, use multiprocessing or threads.

How the Event Loop Actually Works — Not the Simplified Version

Most explanations stop at 'the event loop runs coroutines'. That's not enough when something breaks in production. The event loop is essentially a tight loop around a system call — select, epoll, or kqueue depending on your OS — that asks the kernel: 'which of these file descriptors are ready?'. When one is ready, the loop wakes up the coroutine that was waiting on it and resumes execution from the exact point it yielded.

A coroutine is a function defined with async def. Under the hood it compiles to a code object whose frame can be suspended and resumed. When you await something, Python calls __await__ on the awaitable, which ultimately bottoms out in a Future object. That Future registers a callback with the event loop. The coroutine's frame is frozen — local variables and all — until the Future resolves, at which point the loop schedules its resumption.

This is why you can have 10,000 concurrent HTTP connections on a single thread: no connection holds the thread while waiting. Each coroutine's frame costs roughly 1-2 KB of heap memory — orders of magnitude cheaper than an OS thread's 1-8 MB stack.

Understanding this model is what makes the rule 'never block the event loop' feel obvious rather than arbitrary: if your coroutine calls time.sleep(5), the entire event loop freezes for five seconds because it's still on the same thread. Every other 'concurrent' coroutine is stuck waiting.

import asyncio
import time

# ---------------------------------------------------------------------------
# This example shows TWO tasks running concurrently on ONE thread.
# Watch the timestamps — both tasks overlap in wall time.
# ---------------------------------------------------------------------------

async def fetch_user_profile(user_id: int) -> dict:
    """Simulates a DB round-trip with a network delay."""
    print(f"[{time.perf_counter():.3f}s] START  fetch_user_profile({user_id})")
    # asyncio.sleep yields control back to the event loop.
    # The loop can run OTHER coroutines while this one 'waits'.
    await asyncio.sleep(1.0)  # pretend this is: await db.fetch_one(query)
    print(f"[{time.perf_counter():.3f}s] FINISH fetch_user_profile({user_id})")
    return {"id": user_id, "name": f"User_{user_id}"}


async def fetch_user_orders(user_id: int) -> list:
    """Simulates a second DB query running at the same time."""
    print(f"[{time.perf_counter():.3f}s] START  fetch_user_orders({user_id})")
    await asyncio.sleep(1.5)  # slightly longer query
    print(f"[{time.perf_counter():.3f}s] FINISH fetch_user_orders({user_id})")
    return [{"order_id": 101, "item": "Widget"}, {\"order_id\": 102, \"item\": \"Gadget\"}]\n\n\nasync def build_user_dashboard(user_id: int) -> None:\n    wall_start = time.perf_counter()\n\n    # asyncio.gather schedules BOTH coroutines as Tasks immediately.\n    # They run concurrently — the loop switches between them at each 'await'.\n    profile, orders = await asyncio.gather(\n        fetch_user_profile(user_id),\n        fetch_user_orders(user_id),\n    )\n\n    wall_elapsed = time.perf_counter() - wall_start\n    print(f\"\\nDashboard ready in {wall_elapsed:.3f}s  (would be 2.5s if sequential)\")\n    print(f\"Profile : {profile}\")\n    print(f\"Orders  : {orders}\")\n\n\nif __name__ == \"__main__\":\n    # asyncio.run() creates a fresh event loop, runs the coroutine to\n    # completion, then closes and cleans up the loop. Use this as your\n    # single entry point — never nest asyncio.run() calls.\n    asyncio.run(build_user_dashboard(user_id=42))",
        "output": "[0.001s] START  fetch_user_profile(42)\n[0.001s] START  fetch_user_orders(42)\n[1.002s] FINISH fetch_user_profile(42)\n[1.502s] FINISH fetch_user_orders(42)\n\nDashboard ready in 1.503s  (would be 2.5s if sequential)\nProfile : {'id': 42, 'name': 'User_42'}\nOrders  : [{'order_id': 101, 'item': 'Widget'}, {'order_id': 102, 'item': 'Gadget'}]"
      }

Event Loop Architecture — Visualising the Lifecycle

The event loop is not a black box. It follows a predictable cycle, and understanding each phase helps diagnose hangs and latency. At its core, the loop continuously runs three phases:

1. Polling – asks the OS (via epoll/kqueue/select) which file descriptors are ready for reading or writing. This is where the loop blocks when there is truly nothing to do – but it blocks with a timeout so it can wake periodically to run scheduled callbacks.
2. Running ready callbacks – for each ready FD, the loop invokes the associated callback (e.g., resuming a coroutine that was waiting on a socket). Callbacks are run until the ready list is empty or a maximum number is processed (to avoid starving other tasks).
3. Scheduling – after callbacks, the loop runs any scheduled callbacks (e.g., from asyncio.sleep or call_later). These are stored in a heap and sorted by deadline. The loop also checks for cancelled tasks and prepares the next poll timeout.

The loop also maintains internal data structures: a ready deque, a heap of scheduled timers, and a mapping of FDs to callbacks. The debug mode exposes these with the PYTHONASYNCIODEBUG=1 environment variable, which logs each phase’s duration.

A common performance win is tuning the loop.slow_callback_duration: set it to something aggressive (e.g., 10ms) in development to catch when a callback takes too long. In production, use structured logging to emit metric-like timestamps for each loop iteration.

Here’s a simplified flow of the event loop’s main loop:

Tasks, Futures, and Awaitable Contracts — The Object Model Behind await

There are three things you can await in Python: coroutines, Tasks, and Futures. Understanding the difference is critical for writing correct async code.

A coroutine object (what you get when you call an async def function without await) is a lazy generator-like object. Nothing runs until the event loop drives it. If you write fetch_user_profile(42) without awaiting it, Python will create the object and immediately warn you it was never awaited.

A Future is a low-level promise object. It starts in a pending state and transitions to done (with a result or an exception) exactly once. You almost never create Futures manually in application code — they live inside the networking and I/O layers.

A Task wraps a coroutine and schedules it to run on the event loop immediately via asyncio.create_task(). This is the key difference from a bare await: await coroutine() runs that coroutine sequentially from your perspective, while asyncio.create_task(coroutine()) schedules it concurrently and returns a handle you can await later.

The await keyword calls __await__() on the right-hand side object, which must return an iterator. For Tasks and Futures that iterator suspends the current coroutine and resumes it when the Future resolves. This is the same protocol yield from used in Python 3.4 generators — asyncio is built on top of that generator machinery.

import asyncio
import time


async def slow_api_call(endpoint: str

Cancellation, Timeouts, and Error Handling — The Production Minefield

Happy-path async code is easy. Production async code is defined by how it handles failure. Three scenarios trip up even experienced developers: task cancellation, timeouts, and exception propagation through gather.

Cancellation in asyncio is cooperative, not forcible. When you call task.cancel(), Python injects a CancelledError into the coroutine at its next await point. If the coroutine catches CancelledError and doesn't re-raise it, the cancellation is silently swallowed — a serious bug. Always re-raise CancelledError or use finally blocks for cleanup.

Timeouts are best handled with asyncio.timeout() (Python 3.11+) or asyncio.wait_for() on earlier versions. Both wrap a CancelledError in a TimeoutError so you can distinguish 'took too long' from 'was cancelled by a parent task'.

Exception propagation through gather has a sharp edge: by default, if one task raises, gather cancels the remaining tasks and re-raises the first exception. You lose the results of tasks that succeeded. Pass return_exceptions=True in production so you can inspect every result individually.

asyncio.TaskGroup (Python 3.11+) is now the preferred pattern — it provides structured concurrency where all child tasks are cancelled if any one fails, and all exceptions are surfaced together via an ExceptionGroup.

import asyncio
import time


async def database_query(query_name: str

Blocking the Event Loop — How to Detect It and What to Do Instead

Blocking the event loop is the cardinal sin of async Python and the most common source of 'asyncio isn't faster than sync code' complaints. Any call that holds the thread without yielding — a requests.get(), a time.sleep(), a CPU-heavy loop, even a naively-called json.loads() on a 50MB payload — freezes every other coroutine in your application.

The asyncio event loop has a built-in slow-callback detector: set loop.slow_callback_duration to a low threshold (e.g. 50ms) and enable debug mode. Python will log a warning whenever a callback holds the loop longer than that threshold. This is invaluable in production profiling.

For blocking I/O you can't make async (a synchronous DB driver, a legacy library), use loop.run_in_executor() to offload work to a thread pool. For CPU-bound work, use ProcessPoolExecutor — threads won't help here because of the GIL.

asyncio.to_thread() (Python 3.9+) is a clean shorthand for run_in_executor with the default thread pool. It's idiomatic for wrapping synchronous file I/O, synchronous HTTP calls, or any legacy synchronous function you can't replace yet.

The mental model: the event loop is the single thread. Think of it as a very important person's personal assistant. Every synchronous call is a task that physically occupies the assistant. Every await is handing a task to someone else while the assistant handles the next thing.

import asyncio
import time
import httpx          # async HTTP client — pip install httpx
import requests       # synchronous HTTP client — pip install requests
from concurrent.futures import ThreadPoolExecutor


# ---------------------------------------------------------------------------
# Simulating CPU-heavy work (e.g. image processing, parsing, encryption)
# ---------------------------------------------------------------------------

def cpu_bound_resize(image_id: int) -> str:
    """Pretend this is PIL image processing — pure Python, GIL-bound."""
    # Simulate work with a tight loop — this BLOCKS the event loop if called
    # directly from a coroutine.
    result = sum(i * i for i in range(500_000))  # ~50ms on typical hardware
    return f"image_{image_id}_resized (checksum={result % 9999})"


async def bad_blocking_example() -> None:
    """This coroutine BLOCKS the event loop for ~50ms per call."""
    print("[BAD]  Calling CPU-bound work directly in a coroutine...")
    start = time.perf_counter()

    # These run SEQUENTIALLY and BLOCK the loop — no other coroutine can run.
    results = [cpu_bound_resize(img_id) for img_id in range(5)]

    elapsed = time.perf_counter() - start
    print(f"[BAD]  Done in {elapsed:.3f}s — event loop was blocked the whole time")


async def good_thread_executor_example() -> None:
    """Offload blocking work to a thread pool — loop stays free."""
    print("[GOOD] Offloading CPU-bound work to thread pool...")
    loop = asyncio.get_running_loop()
    start = time.perf_counter()

    # run_in_executor submits work to a ThreadPoolExecutor.
    # The event loop is FREE while the thread runs — other coroutines proceed.
    with ThreadPoolExecutor(max_workers=4) as pool:
        tasks = [
            loop.run_in_executor(pool, cpu_bound_resize, img_id)
            for img_id in range(5)
        ]
        results = await asyncio.gather(*tasks)

    elapsed = time.perf_counter() - start
    print(f"[GOOD] Done in {elapsed:.3f}s — ran 4 tasks in parallel via threads")
    for r in results:
        print(f"  {r}")


async def good_asyncio_to_thread_example() -> None:
    """asyncio.to_thread (Python 3.9+) — cleaner syntax for the same pattern."""
    print("\n[BEST] Using asyncio.to_thread for legacy sync functions...")
    start = time.perf_counter()

    # to_thread wraps the sync function in the default executor automatically.
    # This is the idiomatic way to call synchronous code from async code.
    tasks = [
        asyncio.to_thread(cpu_bound_resize, img_id)
        for img_id in range(5)
    ]
    results = await asyncio.gather(*tasks)

    elapsed = time.perf_counter() - start
    print(f"[BEST] Done in {elapsed:.3f}s")
    for r in results:
        print(f"  {r}")


async def demonstrate_slow_callback_detection() -> None:
    """Enable asyncio debug mode to catch blocking calls automatically."""
    loop = asyncio.get_running_loop()
    loop.slow_callback_duration = 0.05  # warn if any callback takes > 50ms
    # In production: set PYTHONASYNCIODEBUG=1 environment variable OR
    # asyncio.run(main(), debug=True)
    # Python will log: 'Executing <Task ...> took 0.123 seconds'
    print("Slow callback detection threshold set to 50ms")


if __name__ == "__main__":
    asyncio.run(bad_blocking_example())
    asyncio.run(good_thread_executor_example())
    asyncio.run(good_asyncio_to_thread_example())
    asyncio.run(demonstrate_slow_callback_detection())

Handling Blocking Code — A Strategy Guide for run_in_executor and to_thread

When you inevitably encounter a blocking call inside a coroutine, you have three options: replace it with an async alternative, offload it to a thread pool, or offload it to a process pool. The right choice depends on the nature of the work.

Decision Strategy:

1. Is there an async-native library? Use it directly (e.g., httpx.AsyncClient instead of requests, aiofiles instead of file I/O). This is the fastest path — zero context switching overhead.
2. Is the work I/O-bound but synchronous? (e.g., legacy database driver, filesystem with open()). Use asyncio.to_thread() (Python 3.9+) or loop.run_in_executor(None, func, ...). Both submit the call to the default ThreadPoolExecutor, freeing the event loop. The default thread pool has min(32, os.cpu_count() + 4) workers, which is sufficient for most I/O workloads. Tune max_workers via a custom executor if you see thread starvation (e.g., many slow synchronous calls).
3. Is the work CPU-bound? (e.g., parsing JSON, image processing, cryptography). Use ProcessPoolExecutor via loop.run_in_executor(pool, func, ...). Threads won’t help because of the GIL. A process pool gives true parallelism. Be mindful of the overhead: each call pickles the function and arguments, so only use it for work that takes at least several hundred milliseconds.
4. Is the work a blocking C extension that releases the GIL? (e.g., some numpy routines). You can use a thread pool — the GIL is released during the operation, so threads provide parallelism. Profile to confirm.

Here’s a quick reference table:

Always wrap blocking calls with a timeout using asyncio.wait_for on the offloaded future, so you can detect when the executor worker is stuck. This prevents your whole application from hanging on a blocked worker.

import asyncio
import time
from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor


def cpu_heavy_sort(n: int) -> list:
    """CPU-bound: sorts a large list. Should use ProcessPoolExecutor."""
    import random
    data = [random.random() for _ in range(n)]
    data.sort()
    return data[:10]  # first 10 sorted values


def sync_file_read(path: str) -> str:
    """I/O-bound: reads a file. Use to_thread."""
    with open(path) as f:
        return f.read()


async def handle_blocking_code_example():
    loop = asyncio.get_running_loop()

    # Strategy 1: CPU-bound → ProcessPoolExecutor
    with ProcessPoolExecutor(max_workers=2) as pool:
        tasks = [
            loop.run_in_executor(pool, cpu_heavy_sort, 1_000_000),
            loop.run_in_executor(pool, cpu_heavy_sort, 2_000_000),
        ]
        results = await asyncio.gather(*tasks)
        print(f"CPU results: {results}")

    # Strategy 2: I/O-bound sync → to_thread
    file_results = await asyncio.gather(
        asyncio.to_thread(sync_file_read, "/etc/hostname"),
        asyncio.to_thread(sync_file_read, "/etc/resolv.conf"),
    )
    print(f"File results: {[r[:50] for r in file_results]}")

    # Strategy 3: Use a custom ThreadPoolExecutor for heavy I/O
    with ThreadPoolExecutor(max_workers=10) as custom_pool:
        heavy_io_tasks = [
            loop.run_in_executor(custom_pool, sync_file_read, "/var/log/syslog"),
            loop.run_in_executor(custom_pool, sync_file_read, "/var/log/dmesg"),
        ]
        # Add a timeout to prevent hang
        try:
            syslog, dmesg = await asyncio.wait_for(
                asyncio.gather(*heavy_io_tasks), timeout=5.0
            )
            print("Read heavy files successfully")
        except asyncio.TimeoutError:
            print("Timed out reading heavy files")


asyncio.run(handle_blocking_code_example())

Production Patterns: Structured Concurrency with TaskGroup vs gather

asyncio.gather() and asyncio.TaskGroup both run multiple tasks concurrently, but they differ fundamentally in failure handling. gather() by default cancels all tasks if one raises, and it swallows the CancelledError that was injected into surviving tasks — you'll never see them. TaskGroup (Python 3.11+) provides structured concurrency: every task is a child of the group, and if any one fails, all children are cancelled. When all finish, exceptions are merged into an ExceptionGroup.

The critical distinction: with gather(return_exceptions=False), you get the first exception and the rest are cancelled silently. With gather(return_exceptions=True), you get a list with mixed results and exceptions, but the tasks that finished after the first failure are already cancelled by the time you inspect the results. TaskGroup avoids this surprise: you commit to either all succeed or all are cancelled, and you handle exceptions after the context manager exits.

In high-throughput services, prefer gather with return_exceptions=True when you want to preserve successful results despite some failures — for example, fetching data from multiple caches where one miss is acceptable. Use TaskGroup when you want atomicity: if any part of the workflow fails, the whole operation should be abandoned.

Another pattern: asyncio.wait() gives fine-grained control over FIRST_COMPLETED, FIRST_EXCEPTION, ALL_COMPLETED. It's useful for race patterns (e.g., fetch from primary, fallback to secondary on timeout).

import asyncio


async def fetch_cache(key: str

Asyncio Batching Strategies — gather() vs TaskGroup vs as_completed

Choosing the right batching primitive is critical for both performance and correctness. Below is a comparison of the three main approaches. Use this as a quick reference when designing production async workflows.

When to use each:

• gather: Best for most cases where you have a fixed set of I/O operations and want to collect all results at once. Use return_exceptions=True for fault tolerance.
• TaskGroup: Use when you need structured concurrency and atomicity — if one task fails, you want the entire group to abort cleanly. Excellent for complex workflows where cleanup matters.
• as_completed: Use when you want to process results as soon as they arrive, rather than waiting for the slowest task. Useful for progressive UI updates or streaming pipelines.

All three can be combined: for example, use TaskGroup for a batch of related operations, and inside a task use gather to collect sub-results. The key is to match the semantic to your failure tolerance.

The async/await Keyword Contract — What Really Happens Under the Hood

Stop treating async def as magic. It's a compiler transformation that rewrites your function into a state machine. When you call an async function, Python doesn't execute a single line of your code. It returns a coroutine object — a frozen generator that holds your function's frame and local variables.

The await keyword is the yield point. It suspends execution by returning control to the event loop, passing a Future or awaitable that signals when the coroutine should resume. The event loop tracks these awaitables in its run queue. When the underlying I/O completes or a timer fires, the loop calls .send(None) on your coroutine, resuming exactly where it paused.

This is not threads. There's no preemption. Your coroutine runs until it hits an await that blocks, then it parks itself. That's why blocking inside a coroutine kills concurrency — the loop can't switch to another task until your coroutine voluntarily yields.

// io.thecodeforge — python tutorial

import asyncio
import inspect

async def fetch_data(url):
    print(f"Fetching {url}")
    await asyncio.sleep(0.5)  # yield point
    print(f"Done {url}")
    return f"data from {url}"

async def main():
    # Calling async function returns a coroutine object
    coro = fetch_data("https://api.billing.io")
    print(f"Type: {type(coro)}")
    print(f"Is coroutine: {asyncio.iscoroutine(coro)}")
    print(f"Frame locals before await: {coro.cr_frame.f_locals}")
    
    # Only here does execution begin
    result = await coro
    print(f"Result: {result}")

asyncio.run(main())

Coroutine Chaining — Why Your Pipeline Must Be an Awaitable Stack

Real async applications chain coroutines. Your entry point awaits a fetch_payment(), which awaits validate_card(), which awaits authorize_gateway(). Each await peels back one layer of the call stack. This is not free — every suspension and resume costs a microsecond of loop overhead.

The performance trap: if you chain too many shallow coroutines that just delegate to the next, you create a cascading series of mini-suspensions. The event loop switches between runnable tasks, but each chain switch requires a context switch in the C implementation. Benchmark with asyncio.get_running_loop().slow_callback_duration to detect when your chain latency exceeds 100ms.

Design rule: flatten your chains. Combine multiple small async operations into one coroutine if they share I/O context. Use asyncio.gather() only when tasks are genuinely independent — false parallelism through chains that serialize on the same resource is slower than synchronous code.

// io.thecodeforge — python tutorial

import asyncio
import time

async def validate_payment(payment_id):
    await asyncio.sleep(0.1)  # simulate DB lookup
    return {"id": payment_id, "valid": True}

async def process_refund(payment):
    if not payment["valid"]:
        raise ValueError("Invalid payment")
    await asyncio.sleep(0.2)  # simulate API call
    return {"refund_id": f"ref_{payment['id']}"}

async def main():
    start = time.monotonic()
    
    # Chained awaits — each suspension adds overhead
    payment = await validate_payment(42)
    refund = await process_refund(payment)
    
    elapsed = time.monotonic() - start
    print(f"Chained: {elapsed:.3f}s — refund_id: {refund['refund_id']}")

asyncio.run(main())

The asyncio Event Loop — Multitasking Without a Scheduler (and Why It Matters)

The event loop is not a scheduler. There's no preemptive time-slicing. It's a select()-style reactor loop that polls registered file descriptors and callbacks. When you await asyncio.sleep(5), the loop registers a timer callback in its internal heap. When the timer fires, the loop calls .send() on your coroutine to resume it.

The critical difference from threads: the loop can only run one task at a time. If you have 10,000 concurrent I/O operations, the loop cycles through them, checking readiness. No context switch overhead. No GIL contention. But the moment any single coroutine does CPU work without yielding — parsing a JSON response, hashing a password — the entire loop freezes. All 10,000 connections stall.

Design your tasks to yield frequently. For CPU work, use loop.run_in_executor(None, cpu_bound_func). The default executor uses a ThreadPoolExecutor with min(32, os.cpu_count() + 4) workers. Tune that number based on your I/O vs CPU ratio. Blocking for more than 50ms in a single coroutine is a production incident waiting to happen.

// io.thecodeforge — python tutorial

import asyncio
import time

def cpu_intensive_task(n):
    # Simulates parsing a large payload
    total = 0
    for i in range(n):
        total += i ** 2
    return total

async def good_coroutine(task_id, n):
    # Yields to loop; allows other tasks to run
    start = time.monotonic()
    result = await asyncio.get_event_loop().run_in_executor(
        None, cpu_intensive_task, n
    )
    elapsed = time.monotonic() - start
    print(f"Task {task_id}: {elapsed:.3f}s — result {result}")

async def bad_coroutine(task_id, n):
    # Blocks the loop — no other task runs
    start = time.monotonic()
    result = cpu_intensive_task(n)
    elapsed = time.monotonic() - start
    print(f"Task {task_id}: {elapsed:.3f}s — blocked loop")

async def main():
    tasks = [good_coroutine(1, 10_000_000), good_coroutine(2, 10_000_000)]
    await asyncio.gather(*tasks)

asyncio.run(main())

Rate Limiting with Semaphores — Controlling Concurrency in Production

Without rate limiting, async tasks can overwhelm external APIs, databases, or rate-limited services. asyncio.Semaphore caps the number of concurrent coroutines executing a critical section. Create a semaphore with the max concurrency count, then await it before each protected operation. The semaphore blocks when full, releasing a slot after the context exits. This prevents 429 errors and kernel resource exhaustion. Use it with TaskGroup or gather to batch parallel work. Always set an upper bound based on the downstream service limits, never guess. Combine with exponential backoff for retries to build resilient pipelines.

// io.thecodeforge — python tutorial

import asyncio

async def fetch_url(sem: asyncio.Semaphore, url: str):
    async with sem:
        print(f"Fetching {url}")
        await asyncio.sleep(1)  # simulate I/O
        return url

async def main():
    sem = asyncio.Semaphore(3)  # max 3 concurrent
    tasks = [fetch_url(sem, f"https://api.example/{i}") for i in range(10)]
    results = await asyncio.gather(*tasks)
    print(f"Completed: {len(results)}")

asyncio.run(main())

Retry Logic with Exponential Backoff — Resilient Async Calls

Network calls fail. A retry strategy with exponential backoff prevents cascading failures and respects server recovery time. The classic pattern: on failure, wait 2^attempt seconds (plus jitter), then retry up to a max retry count. In asyncio, wrap the call in a loop: try the operation, catch expected exceptions, sleep asynchronously with asyncio.sleep (never time.sleep—that blocks the event loop). Add random jitter to avoid thundering herd. Use a timeout per attempt to avoid hanging. Integrate with semaphore rate limiting for production-grade reliability.

// io.thecodeforge — python tutorial

import asyncio
import random

async def fetch_with_retry(url: str, max_retries: int = 3):
    for attempt in range(max_retries):
        try:
            # simulate flaky fetch
            if random.random() < 0.7:
                raise ConnectionError("timeout")
            return f"OK: {url}"
        except ConnectionError as e:
            if attempt == max_retries - 1:
                raise
            wait = (2 ** attempt) + random.uniform(0, 1)
            print(f"Retry {attempt+1} in {wait:.2f}s")
            await asyncio.sleep(wait)

async def main():
    result = await fetch_with_retry("https://api.example/data")
    print(result)

asyncio.run(main())

Async Iterators and Async Comprehensions — Streaming Data Efficiently

Async iterators let you consume data as it arrives, not all at once. Define an __aiter__ returning self and __anext__ raising StopAsyncIteration when done. Use async for in loops to process chunks—ideal for streaming API responses or file lines. Async comprehensions (async for inside list/dict/set creation) build collections from async sources in one expression. Both require an async context (inside an async def). They avoid buffering entire datasets, reducing memory pressure. Combine with async generators (yield in async def) for clean, lazy pipelines.

// io.thecodeforge — python tutorial

import asyncio

class AsyncCounter:
    def __init__(self, limit: int):
        self.limit = limit
        self.current = 0

    def __aiter__(self):
        return self

    async def __anext__(self):
        if self.current >= self.limit:
            raise StopAsyncIteration
        await asyncio.sleep(0.1)  # simulate I/O
        self.current += 1
        return self.current

async def main():
    async for num in AsyncCounter(5):
        print(f"Got: {num}")
    squares = [x async for x in AsyncCounter(3)]
    print(f"Squares: {squares}")

asyncio.run(main())

Async I/O Isn’t Simple

Many developers treat async I/O as a magic performance switch: just add async/await and everything speeds up. The reality is far more subtle. Async Python runs all coroutines on a single thread, interleaving them via cooperative yielding. If any coroutine blocks — even for 50ms — the entire event loop stalls. Unlike threading, the OS won’t preempt a misbehaving task. The cognitive load skyrockets when debugging race conditions in shared state because asyncio gives you no locks for free. Mixing synchronous libraries (requests, time.sleep) with async code silently serializes your program. Profiling async bottlenecks requires specialized tools like asyncio.Task.cancel() or loop.slow_callback_duration. The paradox: async I/O simplifies throughput for I/O-bound problems but adds complexity for every other concern — error propagation, cancellation, and resource cleanup. Before adopting asyncio, ask whether your bottleneck is genuinely I/O latency, not CPU work or developer productivity.

// io.thecodeforge — python tutorial
import asyncio

def sync_block():
    import time
    time.sleep(0.1)  # blocks event loop

async def bad():
    loop = asyncio.get_running_loop()
    print(loop.time())
    sync_block()  # Never do this — stalls all tasks
    print(loop.time())

asyncio.run(bad())

Libraries Supporting Async I/O

The async ecosystem is fragmented but essential for production systems. For HTTP, aiohttp and httpx (with async support) replace requests. Databases: asyncpg (PostgreSQL) and aiosqlite (SQLite) give true non-blocking queries. Redis: aioredis (merged into redis-py as redis.asyncio). For task queues, use aio-pika (RabbitMQ) or nats-py. If you need HTTP servers, aiohttp or FastAPI with uvicorn’s async workers provide high throughput. The tricky part is that many popular libraries (Django ORM, SQLAlchemy 1.x, boto3) lack native async support, forcing you to run them in executors. Newer tools like SQLAlchemy 2.0 async, beanie (MongoDB ODM), and motor (MongoDB driver) fill gaps. Always verify the library’s async support by checking for async def methods or look for an async submodule. Pro tip: use anyio’s backends to write library-agnostic async code that works with both asyncio and trio, future-proofing your stack against event loop lock-in.

// io.thecodeforge — python tutorial
import asyncio
import httpx

async def fetch(url: str) -> str:
    async with httpx.AsyncClient() as client:
        resp = await client.get(url)
        return resp.text

async def main():
    data = await fetch("https://httpbin.org/get")
    print(data[:50])

asyncio.run(main())

asyncio.sleep() vs time.sleep() — The One That Kills Your Event Loop

You call time.sleep(2) inside an async function thinking you're just pausing. What you've actually done is freeze the entire event loop — all your other coroutines, their network calls, your semaphores, everything — stops dead for 2 seconds. time.sleep() is a blocking system call that suspends the whole Python thread. The event loop can't switch to another task because it never gets a chance to run. asyncio.sleep() yields control back to the event loop, which then schedules other pending coroutines during that 2-second pause. The fundamental distinction: blocking vs yielding. One starves the loop, the other feeds it. Never import time.sleep in async code unless you enjoy watching your concurrent system collapse into sequential misery.

// io.thecodeforge — python tutorial

import asyncio
import time

async def block_the_loop():
    print("Blocking starts")
    time.sleep(2)  # Freezes EVERYTHING
    print("Blocking done")

async def async_good():
    print("Async sleep starts")
    await asyncio.sleep(2)  # Yields to loop
    print("Async sleep done")

async def main():
    await asyncio.gather(block_the_loop(), async_good())

asyncio.run(main())

Eager Task Factory — Fire Coroutines Now, Schedule Later

You write task = asyncio.create_task(my_coro()) and think it starts immediately. It doesn't — the coroutine gets scheduled on the event loop's run queue and won't execute until the current task yields control via an await. That's lazy scheduling overhead. When you need a coroutine to begin execution right now, not after you finish your current await, you use the eager task factory: asyncio.Task(my_coro(), eager_start=True). This creates the task and runs it until its first suspension point before returning. Perfect for kicking off background work like health checks or cache warming that must start immediately while your main coroutine continues. The factory bypasses the queue and executes directly. Use it when latency of first execution matters. Overusing it on trivial coroutines adds unnecessary stack overhead — reserve for operations where starting instantly beats waiting for the next event loop iteration.

// io.thecodeforge — python tutorial

import asyncio

async def speed_demon():
    print("I started immediately!")
    await asyncio.sleep(0)
    return 42

async def slow_poke():
    await asyncio.sleep(0.1)
    print("I had to wait for the loop")

async def main():
    asyncio.get_event_loop().set_task_factory(
        asyncio.eager_task_factory
    )
    eager = asyncio.Task(speed_demon(), eager_start=True)
    lazy = asyncio.create_task(slow_poke())
    print("Main continues...")
    await asyncio.sleep(0.2)
    print(f"Eager result: {eager.result()}")

asyncio.run(main())