Python asyncio — 47-Second Freeze from Sync Calls

asyncio solves a specific and important scalability problem: how do you handle thousands of concurrent I/O operations without spawning thousands of OS threads? The answer is cooperative multitasking — coroutines voluntarily yield control at await points, allowing a single-threaded event loop to multiplex work across all of them efficiently. No thread management, no locking, no context switch overhead from the OS.

The distinction senior engineers must genuinely internalise is that asyncio is concurrency, not parallelism. One thread handles all scheduling. This means any blocking call — a synchronous HTTP request, a CPU-heavy computation, even time.sleep() — freezes the entire loop and every coroutine waiting on it. Not some of them. All of them. This is the property that makes asyncio both powerful and dangerous: the performance gains from concurrency and the catastrophic failure mode from a single blocking call live right next to each other in the same codebase.

This guide covers production-grade patterns: orchestrating concurrent tasks with gather(), building fault-tolerant batch operations against unreliable dependencies, understanding the timeout and cancellation model deeply enough to use it correctly under pressure, and the operational mistakes I have seen repeatedly bring down async services. The examples are written for Python 3.11+ but the core patterns apply from 3.8 onward.

Why Your Async Code Freezes for 47 Seconds

asyncio is Python's cooperative concurrency model: a single-threaded event loop that multiplexes I/O-bound tasks by yielding control at explicit await points. The core mechanic is that only one coroutine runs at a time, and it must voluntarily suspend itself before another can proceed. This means any synchronous blocking call — time.sleep(), a CPU-bound loop, or a blocking I/O operation — stalls the entire event loop for its full duration. In practice, a single sync call of 47 milliseconds can cascade into a 47-second freeze under load because the loop cannot schedule other coroutines. Use asyncio when your workload is I/O-bound with many concurrent operations (network requests, file reads, database queries) and you need to maximize throughput without the overhead of threads. It is not a solution for CPU-bound work — that requires multiprocessing or thread pools. The real-world impact: a single sync call in a web server's async handler can drop thousands of requests per second.

Coroutines and the Event Loop: The Engine Room

A coroutine is a specialised Python generator with async/await syntax. When you define a function with async def, calling it does not execute a single line of the function body — it returns a coroutine object. To actually run it, you must either await it inside another coroutine or schedule it on the event loop via asyncio.create_task() or asyncio.run(). This is not a convention or a style choice — it is how the object model works.

A common misconception is that async def makes a function asynchronous in some broad sense. It does not. It makes the function coroutine-returning. The function body executes zero lines until the coroutine is awaited or scheduled. Forgetting this distinction leads to one of the most common asyncio bugs in production codebases: creating a coroutine object, not awaiting it, and then wondering why the operation never happened. Python 3.11+ will emit a RuntimeWarning about this, but only if something holds a reference long enough to trigger garbage collection — in high-throughput code where objects are short-lived, this warning is sometimes never emitted.

The event loop is the heartbeat of every asyncio application. It maintains a queue of ready callbacks, a selector watching registered I/O file descriptors (using epoll on Linux, kqueue on macOS, IOCP on Windows), and a heap of scheduled callbacks ordered by their scheduled time. When a coroutine awaits an I/O operation, it registers a callback with the selector and suspends — control returns to the loop, which picks the next ready callback and runs it. When the OS signals that I/O is complete, the selector delivers the event, the callback is scheduled, and the original coroutine resumes from exactly where it yielded. This entire mechanism happens in a single thread. No OS context switches. No lock contention. No stack per concurrent operation beyond the coroutine frame itself.

import asyncio
import time

# Production-grade coroutine with proper type hints.
# Note: calling fetch_service_status() without await returns a coroutine object.
# Calling it with await runs the body and returns the string result.
async def fetch_service_status(service_name: str, delay: float) -> str:
    print(f"[io.thecodeforge] Requesting status for {service_name}...")
    # await suspends this coroutine and yields control back to the event loop.
    # The loop is free to run other coroutines during this delay.
    # asyncio.sleep() is non-blocking — it registers a timer callback, not a thread sleep.
    await asyncio.sleep(delay)
    return f"{service_name}: UP"


async def main():
    start_time = time.perf_counter()

    # SEQUENTIAL ANTI-PATTERN: awaiting independent tasks one at a time.
    # Each await suspends main() until that coroutine finishes.
    # Auth-Service completes, then Payment-Gateway starts. Never concurrent.
    # Total time = sum of all delays = 1.0 + 1.0 = ~2.0 seconds.
    status_a = await fetch_service_status("Auth-Service", 1.0)
    status_b = await fetch_service_status("Payment-Gateway", 1.0)

    total_time = time.perf_counter() - start_time
    print(f"Sequential results: {status_a}, {status_b}")
    print(f"Total Sequential Time: {total_time:.2f}s")
    # Output: ~2.00s — we are paying the full cost of both delays in series.
    # The next section shows how gather() eliminates this waste.


if __name__ == "__main__":
    asyncio.run(main())

asyncio.gather() — Orchestrating True Concurrency

When you have independent I/O operations, awaiting them sequentially one by one is leaving performance on the table. If you have three health checks that each take one second, awaiting them in series costs three seconds. Running them with gather() costs one second — the duration of the slowest one. That is the entire value proposition of gather(), and it is substantial.

asyncio.gather() takes an iterable of coroutines (or awaitables), wraps each one into a Task internally via create_task(), and schedules them all onto the event loop simultaneously. It then suspends the calling coroutine until every task completes, and returns a list of results in the same order as the input arguments — regardless of which tasks finished first. This ordering guarantee is important and worth relying on: you can safely unpack results positionally.

One nuance worth understanding: gather() creates tasks at the moment it is called, not at the moment the await resolves. This means all tasks start running as soon as the event loop gets control after the gather() call, which is immediately when the caller awaits gather(). If you are constructing a list of coroutines before calling gather(), those coroutines have not started yet — they are still inert coroutine objects. Only gather() turns them into running Tasks.

The performance model is straightforward: gather() converts sequential wait time into concurrent wait time. The total duration is bounded by max(all task durations) rather than sum(all task durations). For workloads involving many small network calls — health checks, fanout requests to microservices, parallel database lookups — this can reduce latency by an order of magnitude.

import asyncio
import time


async def fetch_service_status(service_name: str, delay: float) -> str:
    print(f"[io.thecodeforge] Requesting status for {service_name}...")
    await asyncio.sleep(delay)
    return f"{service_name}: UP"


async def main():
    start_time = time.perf_counter()

    # CONCURRENT PATTERN: gather() wraps each coroutine into a Task
    # and schedules all three onto the event loop simultaneously.
    # The loop runs them interleaved: Database yields at its await,
    # Cache runs until its await, Search-Index runs until its await,
    # and so on until all three complete.
    #
    # Total time = max(1.5, 0.5, 1.2) = ~1.5 seconds, not 1.5+0.5+1.2 = 3.2s.
    results = await asyncio.gather(
        fetch_service_status("Database", 1.5),
        fetch_service_status("Cache", 0.5),
        fetch_service_status("Search-Index", 1.2),
        # In production, always include return_exceptions=True.
        # Omitted here for clarity — see the Fault Tolerance section.
    )

    total_time = time.perf_counter() - start_time

    # Results are in input order regardless of completion order.
    # Cache finished first (0.5s) but results[1] is Cache — positional, guaranteed.
    print(f"Concurrent Results: {results}")
    print(f"Total Concurrent Time: {total_time:.2f}s")


asyncio.run(main())

Fault Tolerance: Exceptions, Timeouts, and Cancellation

In production, external APIs fail, network partitions happen, and upstream services degrade. The question is not whether these events will occur — it is whether your async code handles them gracefully or cascades them into wider outages.

The two primary tools for fault tolerance in asyncio are gather(return_exceptions=True) for batch resilience and asyncio.wait_for() for individual operation timeouts. They address different failure modes and are commonly used together.

gather(return_exceptions=True) is the production standard for any batch operation against multiple dependencies. Instead of letting the first exception short-circuit the entire batch, it captures all exceptions as return values — Exceptions are just results that happen to be error objects. You iterate the results list, check isinstance(result, Exception) for each entry, and handle successes and failures individually. Every task gets a chance to complete. No orphaned tasks. Full visibility into which dependencies failed and which succeeded.

asyncio.wait_for(coroutine, timeout=seconds) enforces a maximum duration on a single coroutine. If the coroutine does not complete within the timeout, asyncio raises TimeoutError and cancels the wrapped coroutine. This is the mechanism for implementing SLAs on individual downstream calls — if your upstream health check should complete in under 3 seconds, wrap it in wait_for() with timeout=3.0 and handle TimeoutError explicitly.

Cancellation is where most engineers get tripped up. When wait_for() fires, it sends a CancelledError to the wrapped coroutine at its current await point. If the coroutine has cleanup logic that also awaits — closing a database connection, writing an audit log, releasing a lock — that cleanup will only run if it is inside a try/finally block. Code after a cancelled await does not execute. This is cooperative cancellation, and respecting it correctly is what separates async code that is safe from async code that merely appears to work in testing.

import asyncio
from typing import Any


async def api_call(name: str, should_fail: bool = False, delay: float = 0.2) -> str:
    """Simulates an external API call with configurable failure and latency."""
    await asyncio.sleep(delay)
    if should_fail:
        raise RuntimeError(f"Upstream failure in {name}")
    return f"{name}_data"


async def main():
    # --- Pattern 1: Defensive Gathering ---
    # return_exceptions=True captures all outcomes.
    # No task is abandoned. Every failure is inspectable.
    tasks = [
        api_call("Payment-API"),
        api_call("Inventory-API", should_fail=True),  # this one will fail
        api_call("Shipping-API"),
    ]
    results: list[Any] = await asyncio.gather(*tasks, return_exceptions=True)

    successes = []
    failures = []
    for i, result in enumerate(results):
        if isinstance(result, Exception):
            failures.append((i, result))
            print(f"[FAILED] Task {i}: {result}")
        else:
            successes.append(result)
            print(f"[OK]     Task {i}: {result}")

    print(f"\nSucceeded: {len(successes)}, Failed: {len(failures)}")

    # --- Pattern 2: Enforcing Timeouts with wait_for ---
    # The slow API takes 2 seconds. Our SLA is 0.5 seconds.
    print("\n--- Timeout enforcement ---")
    try:
        result = await asyncio.wait_for(
            api_call("Slow-Upstream-API", delay=2.0),
            timeout=0.5
        )
    except asyncio.TimeoutError:
        # TimeoutError means the coroutine was cancelled after 0.5 seconds.
        # The downstream call may still be in flight on the remote server —
        # we just stopped waiting for it.
        print("Slow-Upstream-API exceeded 500ms SLA — request cancelled.")

    # --- Pattern 3: Protecting cleanup with try/finally ---
    # Cleanup runs even if the coroutine is cancelled mid-execution.
    async def careful_operation() -> str:
        try:
            await asyncio.sleep(5.0)  # this will be cancelled
            return "completed"
        finally:
            # This runs even on CancelledError — use it for cleanup.
            print("[Cleanup] Releasing resources on cancellation.")

    print("\n--- Cancellation with cleanup ---")
    task = asyncio.create_task(careful_operation())
    await asyncio.sleep(0.1)  # let the task start
    task.cancel()             # trigger cancellation
    try:
        await task
    except asyncio.CancelledError:
        print("Task was cancelled cleanly.")


asyncio.run(main())

The Golden Rule: Never Block the Event Loop

This is the number one cause of production performance degradation in Python async services, and it is also the most insidious because it does not fail loudly. A blocking call does not raise an exception. It does not log a warning by default. It simply holds the event loop thread for its entire duration, during which every other coroutine in the process is frozen. One synchronous HTTP call taking 200ms freezes 10,000 concurrent connections for 200ms. Under load, these micro-freezes compound into p99 latency spikes that look like intermittent upstream degradation but are entirely self-inflicted.

The most common offenders in codebases I have reviewed: the requests library (always synchronous, even for simple GET calls), time.sleep() used as a delay inside handlers, CPU-heavy operations like image resizing or report generation, and third-party SDK clients that were written before async was widespread. The pattern is usually introduced by someone who understood the application's sync codebase well and did not fully internalise the async execution model.

For unavoidable synchronous code — a legacy library that cannot be replaced, a CPU-intensive operation that has no async equivalent — the correct approach is loop.run_in_executor(), which offloads the blocking call to a thread pool and returns an awaitable that the event loop can wait on without blocking itself. The thread pool runs in parallel with the event loop thread. The event loop remains free to process other coroutines while the thread pool handles the blocking work. This adds thread overhead, but it is categorically better than blocking the loop.

For CPU-bound work that needs true parallelism, ProcessPoolExecutor bypasses the GIL by running work in separate processes. The inter-process communication overhead makes this appropriate for coarse-grained work (process this batch) rather than fine-grained work (transform this value).

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


# This is a synchronous function — it calls time.sleep() which blocks.
# Calling this directly inside an async handler would freeze the event loop.
def legacy_report_generator(report_id: str) -> str:
    """Simulates a legacy synchronous library call that cannot be rewritten."""
    time.sleep(2)  # 2-second blocking operation
    return f"Report-{report_id}: generated"


def cpu_bound_compression(data: str) -> str:
    """Simulates CPU-intensive work — no I/O, pure computation."""
    # In reality this would be image compression, encryption, ML inference, etc.
    result = sum(ord(c) for c in data * 10000)  # burn some CPU
    return f"Compressed({result})"


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

    # --- Pattern 1: Offload blocking I/O to a thread pool ---
    # ThreadPoolExecutor keeps the event loop free while the thread runs.
    # Other coroutines continue executing during the await.
    print("[1] Offloading blocking sync call to thread pool...")
    with ThreadPoolExecutor(max_workers=4) as thread_pool:
        result = await loop.run_in_executor(
            thread_pool,
            legacy_report_generator,
            "Q4-2026"
        )
    print(f"[1] Result: {result}")

    # --- Pattern 2: Offload CPU-bound work to a process pool ---
    # ProcessPoolExecutor creates separate processes with separate GILs.
    # True CPU parallelism — the event loop thread is not blocked.
    print("\n[2] Offloading CPU-bound work to process pool...")
    with ProcessPoolExecutor(max_workers=2) as process_pool:
        result = await loop.run_in_executor(
            process_pool,
            cpu_bound_compression,
            "sensitive_payload_data"
        )
    print(f"[2] Result: {result}")

    # --- Pattern 3: The async-native approach (preferred) ---
    # For new code, use async libraries that never block the loop.
    # import httpx
    # async with httpx.AsyncClient(timeout=5.0) as client:
    #     response = await client.get("https://api.thecodeforge.io/data")
    print("\n[3] Async-native approach: use httpx, motor, aiofiles — never requests, pymongo, open()")


asyncio.run(main())

What Is Asyncio — And What It Absolutely Isn't

Asyncio is a concurrency framework for I/O-bound work running in a single thread. It's not parallelism. It won't make your CPU-bound loops faster. What it does is let you juggle a thousand open network connections without breaking a sweat.

The event loop sits in the middle. It polls file descriptors, schedules coroutines when data arrives, and yields control back to waiting tasks. No threads, no GIL contention — just cooperative multitasking with explicit yield points.

When you call asyncio.run(main()), that creates a new event loop, runs main() as a coroutine, and blocks until done. Inside that loop, every await is a handshake: "I'm waiting on something — go run someone else's code while I do." If you forget to await, you get a coroutine object, not execution. If you block the loop with a time.sleep() instead of asyncio.sleep(), you freeze the entire show for every other task.

The hard truth: asyncio is powerful, but it demands discipline. One synchronous call to requests.get() inside a coroutine and your async web server is now serving one request at a time.

// io.thecodeforge — python tutorial

import asyncio
import time

async def fetch_data(url: str) -> str:
    print(f"Fetching {url}")
    # TRAP: This blocks the entire event loop
    time.sleep(2)  # should be asyncio.sleep(2)
    return f"Data from {url}"

async def main():
    urls = ["https://api.service1.com", "https://api.service2.com"]
    tasks = [asyncio.create_task(fetch_data(url)) for url in urls]
    results = await asyncio.gather(*tasks)
    for r in results:
        print(r)

asyncio.run(main())

Asyncio vs Threading: When to Use a Sledgehammer vs a Scalpel

Threading gives you preemptive multitasking. The OS decides when to swap threads. Asyncio gives you cooperative multitasking. You decide when to yield. The difference isn't academic — it dictates what kind of work you can do.

Threading works for I/O and CPU-bound work, but it's expensive. Each thread carries a 1–8 MB stack, and context switching costs real CPU cycles. Worse, Python's GIL means threads don't parallelise CPU work anyway. For 10k concurrent connections, threads will eat your memory. For 10k connections, asyncio eats maybe 10 MB total.

Asyncio shines when you're waiting on something external: a database query, an HTTP response, a file read. While you wait, other coroutines run on the same thread. Zero context switch overhead. No GIL contention. But if you need to parse a 200 MB JSON blob or run a Monte Carlo simulation, asyncio won't help — you still block the loop. That's when you reach for multiprocessing or push work to a task queue.

Rule of thumb: I/O-bound and many concurrent tasks → asyncio. CPU-bound or complex locking → threads or processes. Mixing both? Carefully. You can offload CPU work to a thread pool with loop.run_in_executor(), but you've just added complexity. Choose the right tool from the start.

// io.thecodeforge — python tutorial

import asyncio
import time
import threading

# Simulate a 2-second I/O wait (e.g., network request)
async def fetch_async(url: str) -> str:
    await asyncio.sleep(2)  # non-blocking
    return f"Async data from {url}"

def fetch_threaded(url: str) -> str:
    time.sleep(2)  # blocking
    return f"Thread data from {url}"

async def run_async():
    start = time.perf_counter()
    tasks = [fetch_async(f"site-{i}.com") for i in range(100)]
    results = await asyncio.gather(*tasks)
    elapsed = time.perf_counter() - start
    print(f"Async: {len(results)} requests in {elapsed:.2f}s")

def run_threaded():
    start = time.perf_counter()
    threads = []
    for i in range(100):
        t = threading.Thread(target=fetch_threaded, args=(f"site-{i}.com",))
        threads.append(t)
        t.start()
    for t in threads:
        t.join()
    elapsed = time.perf_counter() - start
    print(f"Threaded: 100 requests in {elapsed:.2f}s")

asyncio.run(run_async())
run_threaded()

Asyncio Best Practices That Save Your Prod Deploy

Most asyncio code breaks in production because devs treat it like threads with async/await syntax. Stop that. Rule one: never mix asyncio.run() inside a running event loop — you'll get 'RuntimeError: This event loop is already running' and your service crashes at 3 AM. Use asyncio.create_task() instead of low-level loop.create_task() when you're inside a coroutine; the high-level API properly handles cleanup on cancellation.

Second: always wrap your main entry point in asyncio.run(main()). That function creates a fresh event loop, runs your coroutine, and cleans up all pending tasks. Never call loop.close() yourself unless you're writing framework internals. Third: for timeouts, use asyncio.timeout() (Python 3.11+) or asyncio.wait_for() — never implement your own sleep-polling loop. That's how you get 47-second freezes. Fourth: debug mode is your friend. Set PYTHONASYNCIODEBUG=1 or pass debug=True to asyncio.run(). It catches forgotten awaits and slow callbacks.

Fifth: if you're doing CPU-bound work inside a coroutine, you've already lost. Offload to run_in_executor() with a ThreadPoolExecutor. The event loop isn't magic — it's an I/O scheduler.

// io.thecodeforge — python tutorial

import asyncio
import time

async def fetch_data(delay: float) -> str:
    await asyncio.sleep(delay)
    return f"data after {delay}s"

async def main():
    # Use create_task for concurrent execution
    task1 = asyncio.create_task(fetch_data(0.5))
    task2 = asyncio.create_task(fetch_data(1.0))

    # Always use asyncio.timeout for timeouts
    try:
        async with asyncio.timeout(2.0):
            results = await asyncio.gather(task1, task2)
            print(results)
    except asyncio.TimeoutError:
        print("Timed out — check your external API")

if __name__ == "__main__":
    asyncio.run(main())

Real-World Asyncio: Where It Pays and Where It Chokes

Asyncio shines when you're waiting on I/O — network calls, file reads, database queries, API requests. Think web scrapers hitting a hundred endpoints, or a chat server handling ten thousand connections. Each coroutine yields the CPU while waiting, so one thread handles all of them. That's the sweet spot: high-concurrency I/O where latency dominates, not CPU cycles.

Where does asyncio choke? CPU-bound work. Parsing a 10GB JSON file, image processing, or running ML inference inside a coroutine blocks the event loop. Your 10k connections freeze. Threading or multiprocessing is the right tool there. Also: complex synchronous libraries like some database drivers (looking at you, older MySQL connectors) silently block the loop. Wrap them in run_in_executor() or choose an async-native driver.

Real production pattern: FastAPI or aiohttp for the web layer, async Redis and asyncpg for data, and a separate multiprocessing pool for heavy lifting. The event loop handles thousands of concurrent I/O tasks; the process pool handles the CPU-bound grunt work. Mix them properly, and you scale to thousands of requests per second on a single box.

// io.thecodeforge — python tutorial

import asyncio
import concurrent.futures
import json

def parse_large_json(filepath: str) -> dict:
    # CPU-bound — runs in executor
    with open(filepath) as f:
        return json.load(f)

async def fetch_user(api_url: str) -> dict:
    # I/O-bound — native async
    import aiohttp
    async with aiohttp.ClientSession() as session:
        async with session.get(api_url) as resp:
            return await resp.json()

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

    # Offload CPU work to thread pool
    with concurrent.futures.ThreadPoolExecutor() as pool:
        config_data = await loop.run_in_executor(
            pool, parse_large_json, "config.json"
        )

    # I/O work stays in event loop
    users = await asyncio.gather(
        fetch_user("https://api.example.com/user/1"),
        fetch_user("https://api.example.com/user/2"),
    )
    print(f"Config keys: {list(config_data.keys())}")
    print(f"Users: {len(users)}")

asyncio.run(main())

The Inner Workings of Coroutines

Coroutines are not just async functions. Behind the scenes, Python transforms every async def into a generator-like object with __await__ and send() methods. When you await something, the coroutine suspends itself—saving its local state and instruction pointer—then yields control back to the event loop. The event loop holds a reference to that coroutine, waiting for a signal (like a socket becoming readable) to call send(None) on it, which resumes execution exactly where it paused. This is fundamentally cooperative: no preemption, no OS thread stack. Every await point is a voluntary yield. Understanding this explains why blocking in a coroutine is catastrophic—you aren't giving the loop a chance to call send() on other waiting coroutines. The loop can only resume one coroutine at a time, in a single thread, but it juggles thousands by never blocking at a resume point. This mechanism is why asyncio achieves concurrency without parallelism: each coroutine is a tiny state machine driven by the loop.

// io.thecodeforge — python tutorial

async def demo():
    print("started")
    await asyncio.sleep(1)  # yield to loop
    print("resumed")

# What happens internally:
# 1. demo() returns a coroutine object
# 2. loop creates a Task, wrapping it
# 3. Task.__step() calls coro.send(None)
# 4. coro runs until 'await', then raises StopIteration
#    with a 'yield' inside __await__
# 5. Task captures the yield, registers callback
# 6. After 1s, callback calls coro.send(None) -> resumes
print("Coroutine 'demo' is a state machine")
# Output: started (delayed) resumed

A Homemade asyncio.sleep

Built-in asyncio.sleep(n) seems like magic, but you can build one from scratch to internalize how the event loop schedules work. The key is a future: an object that signals readiness. When you create a future and call loop.call_later(seconds, future.set_result, None), you schedule a callback to mark the future as done after the delay. Then await future suspends the coroutine until that callback fires. Your homemade sleep must return an awaitable that does exactly two things: (1) schedule a callback on the loop with your delay, (2) yield control by awaiting a future that the callback resolves. That's it. No busy waiting, no threads. The event loop holds the timer in its internal heap; when the timer expires, it runs the callback, which marks the future done, which resumes your coroutine. This reveals that asyncio.sleep is a thin wrapper around loop.call_later + future. Understanding this lets you create custom timed waits, cancellable delays, or polling loops that don't block.

// io.thecodeforge — python tutorial

import asyncio

async def my_sleep(delay):
    loop = asyncio.get_running_loop()
    future = loop.create_future()
    loop.call_later(delay, future.set_result, None)
    await future  # suspend here until callback

async def main():
    print("0")
    await my_sleep(1)
    print("1")

asyncio.run(main())
# Output: 0  (pause ~1s)  1