asyncio.gather() Still Takes 2.5 Seconds? Find the Blocking Call with Debug Mode

You wrapped five calls in asyncio.gather(), expecting them to finish in about the time of the slowest one. Instead the total is the sum of all five, exactly as if you had written a plain for loop.

gather() is almost never the problem. Something inside one of your coroutines is blocking the event loop: a call that does real waiting without ever handing control back with await. While it runs, no other task can make progress, so your "concurrent" code quietly turns sequential.

Here is how to prove it in a minute, find the exact culprit, and fix it.

The 10-line reproduction

import asyncio, time

async def fetch_blocking(i):
    time.sleep(0.5)          # looks harmless, freezes the whole loop
    return i

async def fetch_ok(i):
    await asyncio.to_thread(time.sleep, 0.5)   # waits in a worker thread
    return i

async def main():
    t = time.perf_counter()
    await asyncio.gather(*(fetch_blocking(i) for i in range(5)))
    print(f"blocking: {time.perf_counter() - t:.2f}s")

    t = time.perf_counter()
    await asyncio.gather(*(fetch_ok(i) for i in range(5)))
    print(f"to_thread: {time.perf_counter() - t:.2f}s")

asyncio.run(main())

On Python 3.13 this prints:

blocking: 2.50s
to_thread: 0.51s

Same gather(), same five tasks. The only difference is whether the waiting happens on the event loop thread or off it. In real code time.sleep is rarely the offender. It is usually requests.get(), a sync database driver, open().read() on a big file, subprocess.run(), or a CPU-heavy loop such as parsing a large JSON payload.

Step 1: Turn on asyncio debug mode

You don't need a profiler to find the blocker. asyncio has a built-in detector: in debug mode it logs every step of a task that holds the loop longer than loop.slow_callback_duration (100 ms by default).

Enable it any one of these ways:

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

PYTHONASYNCIODEBUG=1 python app.py
# or
python -X dev app.py

Make sure warnings from the asyncio logger are visible (for a quick script, logging.basicConfig(level=logging.DEBUG) is enough). Running the example above with debug=True prints one line per offending step:

WARNING:asyncio:Executing <Task finished name='Task-2' coro=<fetch_blocking() done, defined at demo.py:4> result=0 created at .../asyncio/tasks.py:748> took 0.500 seconds

That line tells you the coroutine name (fetch_blocking), the file and line where it is defined, and how long it froze the loop. Five of those lines, each about 0.5 s, add up to your mysterious 2.5 s.

If your real blockers are shorter, lower the threshold so they show up:

async def main():
    asyncio.get_running_loop().slow_callback_duration = 0.02  # 20 ms
    ...

Debug mode adds overhead, so use it locally or in staging, not as a permanent production setting.

Step 2: Narrow it down inside the coroutine

The warning points at a coroutine, not a line. If that coroutine is long, time the suspect calls directly:

t = time.perf_counter()
resp = client.get(url)          # suspect
print("get took", time.perf_counter() - t)

The rule of thumb: any call that waits on the network, disk, or a subprocess and is not preceded by await is a blocker. So is any pure-Python loop that runs for more than a few milliseconds.

Step 3: Pick the right fix

What blocks	Fix
time.sleep(n)	await asyncio.sleep(n)
requests	an async client such as httpx.AsyncClient or aiohttp
sync DB driver (psycopg2, pymysql)	an async driver (asyncpg, aiomysql, SQLAlchemy's async engine), or wrap calls in asyncio.to_thread
a sync library with no async version	await asyncio.to_thread(func, *args) (Python 3.9+)
CPU-heavy work (big parsing, hashing, image work)	a process pool via loop.run_in_executor(ProcessPoolExecutor(), func, arg)

asyncio.to_thread is the quickest unblock for I/O-bound sync code, but remember it runs in a thread pool. Because of the GIL, threads will not speed up pure-Python CPU work, which is why that row goes to a process pool instead.

The FastAPI version of this bug

This shows up constantly in FastAPI, and it is sneaky because the "fix" people try first makes it worse.

@app.get("/report")
async def report():
    data = requests.get(UPSTREAM).json()   # blocks the event loop
    return data

With async def, FastAPI runs the endpoint on the event loop. The blocking requests.get() freezes every other request the worker is handling, so under load one slow upstream makes the whole service look dead.

You have two honest options:

1. Keep it sync. Declare it as plain def report():. FastAPI then runs it in a thread pool, and the blocking call only ties up one worker thread.
2. Go fully async. Keep async def and use an async client with await:

@app.get("/report")
async def report():
    async with httpx.AsyncClient(timeout=10) as client:
        r = await client.get(UPSTREAM)
    return r.json()

What you should not do is mix them: async def plus a sync client is the worst of both. If you are still deciding between frameworks for this kind of service, I compared the trade-offs in FastAPI vs Flask: which is better in 2026.

Once it's unblocked: limit concurrency

When the loop is free, gather() fires everything at once, and 500 simultaneous requests can trip rate limits or exhaust a connection pool. Cap it with a semaphore:

sem = asyncio.Semaphore(10)

async def fetch(client, url):
    async with sem:
        r = await client.get(url)
        return r.status_code

async def main(urls):
    async with httpx.AsyncClient(timeout=10) as client:
        return await asyncio.gather(*(fetch(client, u) for u in urls))

On Python 3.11+ you can use asyncio.TaskGroup instead of gather() when you want one failing task to cancel the rest.

Quick checklist

• Total time equals the sum of the tasks? Something is blocking the loop.
• Run with asyncio.run(main(), debug=True) or PYTHONASYNCIODEBUG=1 and read the Executing <Task ...> took X seconds warnings.
• Lower slow_callback_duration if the blockers are short.
• Swap sync clients for async ones, or wrap them in asyncio.to_thread.
• Send CPU-heavy work to a process pool.
• In FastAPI, never put a blocking call inside async def.

If you're newer to async/await and want the fundamentals first (coroutines, create_task, timeouts and cancellation), there's a beginner walkthrough in this Python asyncio tutorial with examples. And when you're debugging the API payloads themselves, the free browser tools on Blogs World, like the JSON formatter, help you inspect responses without pasting them into random sites.