Debugging and Troubleshooting Asyncio

How to find blocking calls, hangs, deadlocks, lost exceptions and leaks in async code, plus the observability that makes them visible in production before users notice.

#Contents

• Triage table
• Error messages decoded
• Debug mode
• Finding blocking calls
• Event-loop lag monitor
• Debugging hangs
• Deadlock patterns
• Lost exceptions and never-awaited coroutines
• Resource leaks
• Observability in production
• Profiling

#Triage table

Run python scripts/async_lint.py <path> first: it finds most of these causes statically.

#Error messages decoded

#Debug mode

Enable any of:

• PYTHONASYNCIODEBUG=1 python app.py
• python -X dev app.py (dev mode: asyncio debug + more warnings)
• asyncio.run(main(), debug=True) or loop.set_debug(True)

What it adds:

• Logs every callback/task step slower than loop.slow_callback_duration (default 0.1 s) on the asyncio logger: Executing <Task ...> took 0.512 seconds.
• Never-awaited coroutine warnings include where the coroutine was created.
• Raises RuntimeError when non-threadsafe loop methods are called from another thread.
• Logs unclosed transports and slow DNS/IO selector calls; tasks keep their creation traceback.

It costs performance; use it in development, CI and when reproducing a production issue.

import asyncio
import logging
import time


class Collect(logging.Handler):
    def __init__(self) -> None:
        super().__init__()
        self.messages: list[str] = []

    def emit(self, record: logging.LogRecord) -> None:
        self.messages.append(record.getMessage())


async def blocking_handler() -> None:
    time.sleep(0.2)               # the bug: blocks the loop for 200 ms
    await asyncio.sleep(0)


async def main() -> None:
    loop = asyncio.get_running_loop()
    loop.slow_callback_duration = 0.05
    await asyncio.create_task(blocking_handler(), name="handle-request-42")


handler = Collect()
logging.getLogger("asyncio").addHandler(handler)
logging.getLogger("asyncio").setLevel(logging.WARNING)
asyncio.run(main(), debug=True)
slow = [m for m in handler.messages if m.startswith("Executing")]
print("slow step logged:", bool(slow), "| names the task:", "handle-request-42" in slow[0])
# Output: slow step logged: True | names the task: True

Naming tasks (create_task(..., name=...), tg.create_task(..., name=...)) makes these logs and task dumps far easier to read.

#Finding blocking calls

1. Static: python scripts/async_lint.py src/ --select AIO001,AIO002,AIO003,AIO004,AIO005,AIO006,AIO016.
2. Debug mode slow-callback logs (above) name the task; the culprit runs between two of its awaits.
3. py-spy dump --pid <PID> while the service stalls shows the Python stack of the loop thread; a frame inside requests, socket.recv, time.sleep or a C extension is the blocker. py-spy top for CPU hot spots.
4. Third-party wrappers sometimes hide sync calls inside async def (an "async" SDK method that calls a sync client internally). The lag monitor below catches those in production.

#Event-loop lag monitor

A heartbeat task that measures how late it wakes up. Lag above ~50-100 ms means something is blocking the loop. Export it as a metric (gauge or histogram).

import asyncio
import time


async def monitor_loop_lag(samples: list[float], interval: float = 0.02) -> None:
    loop = asyncio.get_running_loop()
    while True:
        start = loop.time()
        await asyncio.sleep(interval)
        samples.append(loop.time() - start - interval)     # how late we woke up


async def main() -> None:
    samples: list[float] = []
    monitor = asyncio.create_task(monitor_loop_lag(samples), name="loop-lag-monitor")
    await asyncio.sleep(0.1)
    time.sleep(0.15)                 # simulate a blocking call somewhere in a handler
    await asyncio.sleep(0.1)
    monitor.cancel()
    try:
        await monitor
    except asyncio.CancelledError:
        pass
    print("max lag over 100 ms:", max(samples) > 0.1)


asyncio.run(main())
# Output: max lag over 100 ms: True

#Debugging hangs

Dump every task's stack: where each one is waiting tells you who is waiting on whom. Dump before cancelling: a timeout that cancels the stuck task also removes it from asyncio.all_tasks().

import asyncio
import io


async def waits_forever(event: asyncio.Event) -> None:
    await event.wait()                       # nobody ever calls event.set()


def dump_tasks() -> str:
    out = io.StringIO()
    for task in asyncio.all_tasks():
        out.write(f"--- {task.get_name()} done={task.done()}\n")
        task.print_stack(limit=5, file=out)
    return out.getvalue()


async def main() -> None:
    event = asyncio.Event()
    stuck = asyncio.create_task(waits_forever(event), name="index-builder")
    done, _ = await asyncio.wait({stuck}, timeout=0.05)   # wait() does not cancel on timeout
    if not done:
        report = dump_tasks()                            # dump while the task is still stuck
        print("index-builder" in report, "waits_forever" in report)
        stuck.cancel()


asyncio.run(main())
# Output: True True

Tools for live processes:

• 3.14: python -m asyncio ps <PID> lists all tasks of a running Python process with their names and coroutine stacks; python -m asyncio pstree <PID> shows the await tree (which task awaits which). Both use the remote-debugging support added in 3.14 and may need ptrace permissions (root or /proc/sys/kernel/yama/ptrace_scope = 0).
• 3.14: asyncio.capture_call_graph(task) / asyncio.print_call_graph(task) print the await chain from inside the process (useful in a debug endpoint).
• Any version: install a signal handler that logs dump_tasks() (loop.add_signal_handler(signal.SIGUSR1, ...) on Unix), or expose it on an admin-only debug endpoint.
• faulthandler.dump_traceback_later(timeout, repeat=True) prints all thread stacks periodically; it shows a loop thread stuck in a blocking call even when no asyncio code can run.
• py-spy dump --pid (all threads, native frames with --native).
• aiomonitor (third party) gives a telnet console with task listing and cancellation for running apps.

#Deadlock patterns

Every network await with a timeout turns a silent deadlock into a TimeoutError with a stack trace, which is a large part of why rule 4 ("every network await has a deadline") matters.

#Lost exceptions and never-awaited coroutines

import asyncio
import gc
import warnings


async def save_usage(n: int) -> None:
    await asyncio.sleep(0)


async def main() -> None:
    with warnings.catch_warnings(record=True) as caught:
        warnings.simplefilter("always")
        save_usage(1)                        # bug: missing await
        gc.collect()
    print([str(w.message) for w in caught])


asyncio.run(main())
# Output: ["coroutine 'save_usage' was never awaited"]

• Make warnings visible in tests: -W error::RuntimeWarning or pytest filterwarnings = ["error"] turns never-awaited coroutines into test failures.
• Task exception was never retrieved is logged only when the task object is garbage-collected, which may be much later or at shutdown. Do not rely on it; own tasks and log in a done callback.
• Route the loop's exception handler to your logging (loop.set_exception_handler, see the cancellation reference) and make sure the asyncio logger is not filtered out.

#Resource leaks

• Enable -X dev or -W default::ResourceWarning to see unclosed sockets, transports, files and sessions.
• aiohttp logs Unclosed client session with the creation point in debug mode.
• Count open file descriptors over time (len(os.listdir("/proc/self/fd")) on Linux) as a metric for long-running workers.
• Async generators that are not exhausted keep their resources until finalized; use contextlib.aclosing.

#Observability in production

Logging: put the request id in a contextvars.ContextVar and add it with a logging filter; it propagates into tasks and to_thread automatically. Tracing: OpenTelemetry's context also uses contextvars, so spans follow create_task/gather without extra work; create one span per external call.

import contextvars
import logging

request_id: contextvars.ContextVar[str] = contextvars.ContextVar("request_id", default="-")


class RequestIdFilter(logging.Filter):
    def filter(self, record: logging.LogRecord) -> bool:
        record.request_id = request_id.get()
        return True


handler = logging.StreamHandler()
handler.addFilter(RequestIdFilter())
handler.setFormatter(logging.Formatter("%(asctime)s %(levelname)s [%(request_id)s] %(name)s: %(message)s"))
logging.getLogger().addHandler(handler)

#Profiling

• py-spy record -o profile.svg --pid <PID> samples without code changes; async frames appear as the coroutine functions being resumed.
• yappi with set_clock_type("wall") and coroutine-aware stats attributes wall time correctly across awaits.
• cProfile attributes time to the event loop's select call while waiting; useful for CPU hot spots, not for "where is time spent waiting".
• To measure where a request waits, time each await (stage histograms) rather than profiling.