Enable asyncio Debug Mode

See the Python docs for all the ways to turn on async debug mode.

Warnings about blocking async code

In debug mode, you will get warnings about async code that blocks for too long as well as other warnings.

Example code:

import asyncio
import time

async def blocks():
    print("START")

    # Synchronous blocking code
    time.sleep(0.35)

    await asyncio.sleep(0.5)

    # Synchronous blocking code
    time.sleep(0.15)

    await asyncio.sleep(0.1)
    print("DONE")

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

Output:

START
Executing <Task pending name='Task-804' coro=<blocks() running at <ipython-input-7-1388d45e4cc5>:4> wait_for=<Future pending cb=[<TaskWakeupMethWrapper object at 0x7f71d70e7910>()] created at /usr/lib/python3.9/asyncio/base_events.py:424> cb=[_run_until_complete_cb() at /usr/lib/python3.9/asyncio/base_events.py:184] created at /usr/lib/python3.9/asyncio/base_events.py:621> took 0.351 seconds
Executing <Task pending name='Task-804' coro=<blocks() running at <ipython-input-7-1388d45e4cc5>:6> wait_for=<Future pending cb=[<TaskWakeupMethWrapper object at 0x7f71d8450430>()] created at /usr/lib/python3.9/asyncio/base_events.py:424> cb=[_run_until_complete_cb() at /usr/lib/python3.9/asyncio/base_events.py:184] created at /usr/lib/python3.9/asyncio/base_events.py:621> took 0.151 seconds
DONE

The time.sleep function simulates running blocking synchronous code. Notice we get warnings about how long the coroutine blocks every time the async function spends too long blocking the async loop.

Solutions

1. Optimizing the blocking code.
2. Breaking up the blocking code into multiple async tasks.
3. Running the code in a thread with asyncio.to_thread or with a custom executor and loop.run_in_executor.

Change the warning threshold

If you find that you need a higher or lower threshold for warnings about async code that blocks the main loop for too long, you can set it by changing the slow_callback_duration attribute of the async loop:

loop.slow_callback_duration = 0.15

If you don’t have access to the loop, you are likely using the default loop. In this case use:

asyncio.get_event_loop().slow_callback_duration = 0.15

This attribute has been available since Python 3.6, and maybe before that.

Default warning threshold value

The default as of Python 3.10 (in 2021) is 0.1 seconds.

Change the log handler

You can adjust the asyncio logger by getting the "asyncio" logging.Logger.

The default level of logging.DEBUG may be too verbose, so you can set its logging level to logging.WARNING.

import logging

logger: logging.Logger = logging.getLogger("asyncio")
logger.setLevel(logging.WARNING)

json_logging_handler = ...
logger.addHandler(json_logging_handler)

Use aiodebug

If you don’t want to turn on debug mode in production, use aiodebug to log warnings when callbacks block the event loop:

aiodebug.log_slow_callbacks.enable(0.15)

This works by wrapping asyncio.events.Handler._run with a timing and logging function.

To modify the logger’s output, you need to use the logger named after a specific file in the aiodebug project.

This may work:

import aiodebug.log_slow_callbacks

logger = logging.getLogger(aiodebug.log_slow_callbacks.__name__)