Python Asyncio and Blocking Functions

An asyncio program runs every coroutine on one thread. The event loop hands control to a coroutine, and it only gets control back when that coroutine reaches an await that actually suspends. Between those points, nothing else on the loop runs: no other request handler, no heartbeat, no timeout callback. That single fact explains almost every “my async service randomly freezes” bug, and it’s what interviewers are probing when they ask what happens if you call requests.get inside an async def.

What the event loop can and can’t interrupt

Asyncio is cooperative multitasking. The loop can’t preempt a coroutine the way the operating system preempts a thread. It waits for the coroutine to yield. A coroutine yields when it awaits a future that isn’t done yet: asyncio.sleep, a socket read from an async library, a queue get on an empty queue.

So the question to ask about any line inside a coroutine is not “is this function async?” but “does this line give control back to the loop while it waits?” time.sleep(0.5) doesn’t. Neither does requests.get, open(path).read() on a slow network drive, psycopg2 queries, subprocess.run, or a tight loop that parses a 200 MB JSON file. They all keep the thread busy, and the loop sits behind them.

A common misconception is that writing async def in front of a function makes its body concurrent. It doesn’t. An async def function with no await inside runs start to finish without ever letting anything else in. It’s a normal function with extra ceremony.

Watching one sleep freeze a heartbeat

The quickest way to see this is to run a heartbeat next to a handler that blocks. The heartbeat should print every 100 ms.

import asyncio
import time

async def heartbeat():
    start = time.perf_counter()
    for _ in range(5):
        await asyncio.sleep(0.1)
        print(f"tick at {time.perf_counter() - start:.2f}s")

async def handler():
    time.sleep(0.5)          # blocking call inside a coroutine
    return "done"

async def main():
    await asyncio.gather(heartbeat(), handler())

asyncio.run(main())

Running it on Python 3.14 prints:

tick at 0.51s
tick at 0.61s
tick at 0.71s
tick at 0.81s
tick at 0.91s

The first tick should have arrived at 0.10s. It arrived at 0.51s because the heartbeat’s sleep timer had expired long before, but the loop couldn’t run the callback while handler held the thread. In a web service, that half-second is added to the latency of every request that happened to be in flight, including ones that have nothing to do with the slow handler. That’s why these bugs look random in production: the victim isn’t the code that blocks.

Now change one line so the blocking call runs in a worker thread:

async def handler():
    await asyncio.to_thread(time.sleep, 0.5)   # runs in the default thread pool
    return "done"

tick at 0.10s
tick at 0.20s
tick at 0.30s
tick at 0.40s
tick at 0.50s

asyncio.to_thread submits the function to the loop’s default ThreadPoolExecutor and returns an awaitable. The coroutine suspends on that awaitable, so the loop is free to run the heartbeat while the worker thread sleeps.

Finding the blocking call in a real codebase

In a large service you rarely know which call is blocking. Asyncio’s debug mode helps. With asyncio.run(main(), debug=True) (or PYTHONASYNCIODEBUG=1), the loop logs any callback or task step that runs longer than loop.slow_callback_duration, which defaults to 100 ms.

import asyncio
import time

def slow():
    time.sleep(0.3)

async def main():
    loop = asyncio.get_running_loop()
    loop.slow_callback_duration = 0.1
    await asyncio.sleep(0)
    slow()

asyncio.run(main(), debug=True)

The log line it produced (path shortened):

Executing <Task finished name='Task-1' coro=<main() done, defined at .../ab3.py:7> result=None created at .../asyncio/runners.py:110> took 0.302 seconds

It names the task and the duration, which tells you which coroutine to read. Inside it, look for the usual suspects: synchronous HTTP clients, database drivers that aren’t async, file reads, hashlib or bcrypt on large inputs, json.loads on huge payloads, and logging handlers that write to a slow network sink.

Interview tip

Debug mode adds overhead, so turn it on in staging or a load test rather than leaving it on in production. For production, a latency metric on a trivial endpoint (like a health check) is a cheap detector: if /health occasionally takes 400 ms, something on the loop is blocking.

Choosing the fix: async library, thread, or process

There are three fixes, and the right one depends on what the blocking code is doing.

Use an async-native library when one exists. Swapping requests for httpx.AsyncClient or aiohttp, or psycopg2 for asyncpg or psycopg 3’s async API, removes the problem at the source. The I/O waits become real suspension points and you don’t spend a thread per call.

Use asyncio.to_thread for blocking I/O you can’t replace. A vendor SDK, a legacy driver or pathlib file reads are typical. Threads work well here because blocking I/O calls in CPython release the GIL while waiting. Two things to keep in mind. First, the default executor has a limited number of workers (min(32, os.cpu_count() + 4) threads), so a burst of 500 offloaded calls queues up behind them. If the downstream system can only handle 10 concurrent calls anyway, wrap the offload in an asyncio.Semaphore(10) so the limit is explicit. Second, cancellation doesn’t reach the thread:

import asyncio, time

def work():
    time.sleep(0.5)
    print("thread finished anyway")

async def main():
    try:
        async with asyncio.timeout(0.1):
            await asyncio.to_thread(work)
    except TimeoutError:
        print("timed out")
    await asyncio.sleep(0.6)

asyncio.run(main())

timed out
thread finished anyway

The coroutine stopped waiting after 100 ms, but the thread ran to completion. Python has no safe way to kill a thread. If the offloaded function writes to a database, that write still happens after your caller has been told it timed out. Interviewers like this follow-up because it separates people who have run this in production from people who have read the docs once. The answer is to give the blocking call its own timeout (most HTTP and DB clients accept one) so the thread ends on its own.

Use processes for CPU-bound work. Threads don’t help with pure-Python computation because only one thread executes Python bytecode at a time under the GIL (the free-threaded 3.13+ builds change this, but most deployments still use the default build). Use loop.run_in_executor(ProcessPoolExecutor(), fn, *args) for things like image resizing or heavy parsing. Arguments and results are pickled, so this pays off only when the work is large compared with the data you send.

Interview practice: the slow PDF endpoint

An async FastAPI service has an endpoint that generates a PDF report with a synchronous library. It takes about 2 seconds of CPU. Under load, every other endpoint’s p99 latency jumps to several seconds. A colleague proposes wrapping the PDF call in asyncio.to_thread. What do you say?

Answer

The diagnosis is right: the PDF generation runs on the event loop thread, so every other request waits for it, which is why unrelated endpoints slow down. But to_thread is only a partial fix. If the library is pure Python, the worker thread holds the GIL for most of those 2 seconds, so the loop thread still gets starved, just less predictably. I’d first check whether the library releases the GIL (a C-backed renderer might). If it doesn’t, move the work to a ProcessPoolExecutor sized to the CPU cores available, or better, to a background job queue so the endpoint returns a job id and the client polls or receives a webhook. Either way I’d cap concurrency so ten simultaneous report requests can’t consume every core, and I’d give the job a timeout that the worker itself enforces, since cancelling the awaiting coroutine won’t stop it.

Related reading

• Python interview questions
• Python MCQs
• Asyncio tasks and cancellation
• Threads, processes and workload choice
• Official docs: Developing with asyncio