Debugging & Performance

Debugging Async Code and Event Loops

Async bugs rarely raise a clean traceback at the point of failure. A coroutine silently never runs, a blocking call freezes the whole loop, a task dies with its exception swallowed, or teardown trips RuntimeError: Event loop is closed long after the real mistake. The reason is structural: in asyncio, scheduling is decoupled from execution, exceptions live on Task objects until someone retrieves them, and one event loop multiplexes everything. This guide turns those invisible failures into observable ones — debug mode, slow-callback detection, never-awaited warnings, task introspection, and stepping through coroutines with pdb.

Prerequisites

  • Python 3.8+ (asyncio.all_tasks, asyncio.current_task, asyncio.run are the modern, loop-agnostic APIs; the pre-3.7 get_event_loop patterns are deprecated).
  • Python 3.11+ for the async-aware REPL (python -m asyncio) that lets you await at the prompt.
  • aiomonitor >= 0.7 for attaching to a live loop over a console (pip install aiomonitor); aiodebug for slow-callback logging hooks.
  • tracemalloc (stdlib) for allocation tracebacks on never-awaited coroutines.

Core concept

The event loop runs one callback at a time. A coroutine becomes a Task only when it is awaited or scheduled (create_task, gather, ensure_future); a bare some_coro() call just builds a coroutine object that does nothing until awaited — forget the await and it is garbage-collected unrun, producing the "coroutine was never awaited" warning. Because the loop is single-threaded, any synchronous blocking call (a requests GET, time.sleep, a CPU-bound loop) freezes every task, which debug mode surfaces as a slow-callback warning. Exceptions raised inside a task do not propagate to the caller; they are stored on the task and only re-raised when its result is awaited, which is why a crashed background task can vanish without a trace.

asyncio debug mode is the single most valuable switch. Enable it via PYTHONASYNCIODEBUG=1, asyncio.run(main(), debug=True), or loop.set_debug(True). It logs callbacks slower than loop.slow_callback_duration (default 0.1s), warns on never-awaited coroutines with their origin, and checks that loop calls happen on the owning thread.

Task lifecycle on the event loop A coroutine becomes a scheduled task, runs and suspends on awaits, then finishes or stores an exception; forgetting to await leaves it unrun. Where async tasks go wrong coroutine object created by some_coro() scheduled Task await / create_task running on loop one task at a time done: result awaited and returned exception stored silent until awaited never awaited GC'd, never ran Debug mode flags the red paths unawaited coroutines, slow callbacks, lost errors
A coroutine only runs once awaited or scheduled as a task; an exception sits on the task until its result is retrieved, and a forgotten await leaves the coroutine garbage-collected and unrun. Debug mode makes these red paths visible.

Step-by-step implementation

1. Enable debug mode

The cleanest switch is the debug flag on asyncio.run, which sets the loop into debug mode for the whole run:

Python
import asyncio

async def main() -> None:
    await asyncio.sleep(0.01)

# debug=True: log slow callbacks, warn on unawaited coroutines, check thread safety.
asyncio.run(main(), debug=True)

For a loop you manage yourself, call loop.set_debug(True); to enable it globally without code changes, export PYTHONASYNCIODEBUG=1 before launching.

2. Catch slow callbacks (blocked loop)

Any synchronous blocking call stalls the loop. Debug mode logs it:

Python
import asyncio, time

async def handler() -> None:
    time.sleep(0.5)            # BUG: synchronous sleep blocks the whole loop

async def main() -> None:
    loop = asyncio.get_running_loop()
    loop.slow_callback_duration = 0.05    # lower threshold to catch shorter stalls
    await handler()

asyncio.run(main(), debug=True)
# WARNING:asyncio:Executing <Handle ...> took 0.500 seconds

The fix is to move blocking work off the loop with await loop.run_in_executor(None, blocking_fn) or an async-native client. aiodebug.log_slow_callbacks can route these warnings into structured logging in production.

3. Find never-awaited coroutines

A forgotten await produces a RuntimeWarning at garbage-collection time — far from the bug. Promote it to an error and enable tracemalloc so the warning carries the allocation traceback:

Python
import asyncio, tracemalloc, warnings

tracemalloc.start()                                   # capture where coroutines are created
warnings.simplefilter("error", RuntimeWarning)        # turn the warning into a raised error

async def fetch() -> int:
    return 42

async def main() -> None:
    fetch()                  # BUG: missing await -> coroutine never runs
    await asyncio.sleep(0)

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

Run the test suite with python -W error::RuntimeWarning -X tracemalloc to fail CI on the warning with a pinpoint traceback. This technique gets a full treatment in tracing "coroutine was never awaited" warnings.

4. Introspect running tasks

When the loop hangs, ask it what is pending. asyncio.all_tasks() returns every live task; get_coro() and get_stack() show what each one is and where it is suspended:

Python
import asyncio

async def slow() -> None:
    await asyncio.sleep(3600)

async def main() -> None:
    asyncio.create_task(slow(), name="slow-worker")
    await asyncio.sleep(0)                      # let the task start and suspend
    for task in asyncio.all_tasks():
        print(task.get_name(), task.get_coro().__qualname__)
        task.print_stack()                      # where the task is parked
    print("current:", asyncio.current_task().get_name())

asyncio.run(main())

For a live, long-running service, attach aiomonitor and inspect tasks over a console without stopping the process:

Python
import aiomonitor, asyncio

async def main() -> None:
    with aiomonitor.start_monitor(asyncio.get_running_loop()):
        await asyncio.sleep(3600)               # `telnet localhost 50101`, then `ps`/`where`

5. Step through with pdb

breakpoint() works inside a coroutine; the loop pauses while you are at the prompt (which can itself trip slow-callback warnings — expected). On Python 3.11+, the async REPL (python -m asyncio) lets you await expressions at the prompt to inspect coroutine results interactively:

Python
import asyncio

async def compute(x: int) -> int:
    result = x * 2
    breakpoint()              # pdb here; `p result`, `await some_coro()` on 3.11+ REPL
    return result

asyncio.run(compute(21))

When the breakpoint must live inside an async test, scope the loop correctly first — see how to scope pytest fixtures for async tests and the pytest-asyncio vs anyio scoping trade-offs so the breakpoint runs on a live loop. General pdb mechanics live in interactive debugging with pdb and ipdb.

Verification

  • Debug mode is on: print(asyncio.get_running_loop().get_debug()) returns True inside main.
  • Slow callbacks are caught: the "Executing ... took N seconds" warning fires for a deliberately blocking call once you lower slow_callback_duration.
  • Unawaited coroutines fail loudly: running under -W error::RuntimeWarning turns a missing await into a raised error rather than a late GC warning.
  • No task leaks at shutdown: after asyncio.run returns, asyncio.all_tasks() is empty (it raises outside a loop, so check inside a final await). Pending tasks at teardown are the classic cause of the next error.

Troubleshooting

SymptomRoot causeFix
RuntimeWarning: coroutine '...' was never awaitedA coroutine was created but never awaited or scheduledawait it, asyncio.create_task(...), or pass to gather; run with tracemalloc to find it
Executing <Handle ...> took N secondsSynchronous blocking call on the loopMove work to run_in_executor or an async client
Loop hangs with no errorA task is awaiting something that never resolvesasyncio.all_tasks() + task.print_stack() to find the parked task
Background task crashed silentlyException stored on the task, never retrievedAdd a done callback or await/gather the task; debug mode surfaces it
RuntimeError: Event loop is closed at teardownReusing or scheduling on a closed loopSee the dedicated guide below
RuntimeError: no running event loopCalling create_task/get_running_loop outside a coroutineCall inside an async def driven by asyncio.run

Frequently Asked Questions

How do I enable asyncio debug mode? Set the environment variable PYTHONASYNCIODEBUG=1, pass debug=True to asyncio.run, or call loop.set_debug(True) on a running loop. Debug mode logs slow callbacks, warns about coroutines that were never awaited, and checks that calls happen on the right thread.

Why do I get a coroutine was never awaited warning? You called an async function but never awaited the coroutine it returned or scheduled it as a task, so it was garbage collected without running. Await it, wrap it in asyncio.create_task, or pass it to asyncio.gather.

How do I see every task currently running on the event loop? Call asyncio.all_tasks(loop) to get the set of pending tasks, and asyncio.current_task() for the one running now. Each task's get_coro and get_stack methods reveal what it is and where it is suspended.

Can I use pdb inside an async function? Yes. breakpoint() works inside a coroutine on Python 3.7+, and on 3.11+ the asyncio REPL plus pdb handles awaits at the prompt. The loop is paused while you are at the breakpoint, so long pauses can trip slow-callback warnings.

← Back to Systematic Debugging & Performance Profiling