How to Debug Asyncio

You can debug asyncio programs by enabling debug-level logging, enabling warnings, and running the asyncio event loop in debug mode.

This will report additional messages from the asyncio module in the standard library, and perform additional checks, such as calls to thread-unsafe functions and reporting when asyncio tasks take too long to complete.

In this tutorial, you will discover how to debug asyncio programs.

Let's get started.

We Need to Debug Asyncio

Sometimes things do go as planned with our asyncio program.

We need insight into what is going on.

There are two levels of insight we might need.

1. What is happening inside the asyncio module.
2. What is happening inside our program.

This can be achieved by logging debug messages.

How can we debug asyncio?

How to Debug Asyncio

There are perhaps three considerations when debugging asyncio, from a high level.

1. Accessing debug messages from the asyncio module.
2. Running the asyncio event loop in debug mode.
3. Accessing warning messages.
4. Adding logging messages (like debug messages) to our program code.

Let's consider each of these aspects in turn.

Access Asyncio Debug Messages

The asyncio module internally does a lot of logging.

You can see this if you search the code files for "logger" in the module:

• Asyncio source code

All asyncio module logging is performed under the context of the module name, e.g. "asyncio".

We can therefore enable all debug messages (and higher) for the asyncio module and all of our programs by setting the log level to DEBUG.

For example:

...
# enable debugging
logging.basicConfig(level=logging.DEBUG)

This alone will report basic information from the asyncio module internals.

Running the Event Loop in Debug Mode

If debug log messages for the asyncio module are permitted (see above), we can run the asyncio event loop in debug mode in order to see more information.

This can be achieved in a few ways:

1. Via an argument to asyncio.run()
2. Via a method on the event loop itself.
3. Via an environment variable.

The first approach involves passing the "debug" argument set to True to the asyncio.run() function used to start the event loop for our program.

For example:

...
# start the event loop
asyncio.run(main(), debug=True)

The second is to acquire the event loop object, such as via the asyncio.get_running_loop() function and then calling the set_debug() method.

For example:

...
# enable debug mode on the current event loop
loop = asyncio.get_running_loop()
loop.set_debug()

The third approach is to set the PYTHONASYNCIODEBUG environment variable.

This could be set in the shell, command line environment, or IDE environment.

For example

PYTHONASYNCIODEBUG=1

It may also be set programmatically via the os.environ() function.

For example:

...
# set debug mode
os.environ["PYTHONASYNCIODEBUG"] = "1"

Once set, the event loop will do extra work to calculate interesting debug information and present it via log message, such as at the DEBUG and WARNING levels.

Callbacks taking longer than 100 milliseconds are logged.

-- Developing with asyncio

Specifically, asyncio debug mode provides:

1. Coroutines that were not awaited are logged, e.g. the stack trace.
2. Thread-unsafe methods raise exceptions if called from the wrong threads.
3. Execution of I/O is logged, if it takes too long.
4. Execution time of tasks are logged, if they take too long.

As such, these additional checks and logging can add overhead, making execution time slower.

How to Configure How Long is "Too Long"

When in debug mode, the asyncio event loop will report warning messages for coroutines that take too long to complete.

These messages are reported after the coroutines and tasks are completed.

The "too long" duration by default is 0.1 seconds, e.g. 100 milliseconds.

It can be configured via the slow_callback_duration attribute on the event loop object.

The loop.slow_callback_duration attribute can be used to set the minimum execution duration in seconds that is considered “slow”.

-- Developing with asyncio

The attribute is measured in seconds, including fractions of a second.

For example:

...
# change the how long limit for reporting warnings
loop = asyncio.get_running_loop()
loop.slow_callback_duration = 0.01 # 10 ms

Accessing Warning Messages

By default, the asyncio will emit warnings in stations that represent common problems.

The most common is when a coroutine is created and not awaited.

For example:

...
RuntimeWarning: coroutine '...' was never awaited
RuntimeWarning: Enable tracemalloc to get the object allocation traceback

You can learn more about this specific warning in the tutorial:

• RuntimeWarning: Coroutine Was Never Awaited

When the event loop is run in debug mode, additional warnings may be emitted.

These can be accessed without enabling debug logging.

This can be achieved by enabling the warning messages alone, via the warnings module.

For example, we can set the filter on warning messages to be 'always' for ResoruceWarnings via the warnings.simplefilter() function.

For example:

...
# enable warnings
warnings.simplefilter('always', ResourceWarning)

Or all warnings:

...
# enable all warnings
warnings.simplefilter('always')

This can help expose warning messages, such as when coroutines take too long to complete.

Adding Debug Logging to Our Program

Finally, we can add debug (and other levels), log messages to our program by using the logging module.

For example:

# custom coroutine
async def task(value):
    # do work
    # ...
    # log a debug message
    logging.debug(f'Task done with value={value}')

Note, additional debug information is provided in the asyncio API documentation:

• Developing with asyncio

Now that we know the basics of debugging asyncio programs, let's look at some worked examples.

Example of Enabling Asyncio Debug Logging

We can explore an example of enabling the logging of debug messages in the "asyncio" module in the standard library.

In this example we will have a task that takes an argument, reports a message, sleeps a moment, then reports a final message. The main coroutine awaits this task.

To enable logging in the asyncio module, we can set the log level for the entire program to "DEBUG" before starting the asyncio event loop.

...
# enable debug logging
logging.basicConfig(level=logging.DEBUG)
# start the event loop
asyncio.run(main())

By default, this will log all messages to standard output (the console).

Tying this together, the complete example is listed below.

# SuperFastPython.com
# example of enabling the asyncio module debugging
import logging
import asyncio

# simple task that takes a moment
async def task(value):
    # report a message
    print(f'Task is running, value={value}')
    # block a moment
    await asyncio.sleep(2)
    # report a message
    print(f'Task is done, value={value}')

# main coroutine
async def main():
    # execute and await the task
    await task(99)

# enable debug logging
logging.basicConfig(level=logging.DEBUG)
# start the event loop
asyncio.run(main())

Running the example first sets the logging level to DEBUG.

It then starts the asyncio event loop and executes the main() coroutine.

The main() coroutine runs the task() coroutine and awaits it. The task() coroutine runs, reports a message, sleeps, and then reports a final message.

Then the program terminates.

We can see that in this case that only a single debug message is reported by the asyncio module.

The message is about the IO selector that was chosen by the asyncio event loop, in this case, the KqueueSelector.

DEBUG:asyncio:Using selector: KqueueSelector
Task is running, value=99
Task is done, value=99

Next, let's look at an example of enabling debug mode for the event loop.

Example of Running the Event Loop in Debug Mode

We can explore an example of enabling debug mode in the asyncio event loop.

This will cause additional checking and logging to be performed, which may slow down large programs with many coroutines.

To enable debug mode, we will set the "debug" argument to True when starting the asyncio event loop with asyncio.run(). We will also enable debug-level logging so that we can see any additional log messages reported by the event loop.

...
# enable debug logging
logging.basicConfig(level=logging.DEBUG)
# start the event loop
asyncio.run(main(), debug=True)

Tying this together, the complete example is listed below.

# SuperFastPython.com
# example of enabling debug logging and run in debug mode
import logging
import asyncio

# simple task that takes a moment
async def task(value):
    # report a message
    print(f'Task is running, value={value}')
    # block a moment
    await asyncio.sleep(2)
    # report a message
    print(f'Task is done, value={value}')

# main coroutine
async def main():
    # execute and await the task
    await task(99)

# enable debug logging
logging.basicConfig(level=logging.DEBUG)
# start the event loop
asyncio.run(main(), debug=True)

Running the program first sets debug level logging, then starts the event loop in debug mode.

The main() coroutine runs, executes the task() coroutine and the program terminates.

We can see that as before, the asyncio module reports the selector that is being used, e.g. KqueueSelector.

When the program closes, it then reports that the event loop has closed (e.g. is no longer running) and other details of its state (e.g. it was run in debug mode)

DEBUG:asyncio:Using selector: KqueueSelector
Task is running, value=99
Task is done, value=99
DEBUG:asyncio:Close <_UnixSelectorEventLoop running=False closed=False debug=True>

Next, let's look at the reporting of long-running tasks.

Example of Task Took Too Long in Debug Mode (with logging)

We can explore an example of reporting log messages running asyncio tasks when running the asyncio event loop in debug mode.

A task takes too long if it has not been executed by the event loop in more than 100 milliseconds (e.g. 0.1 seconds).

Typically, this occurs when an asyncio performs some blocking function call (like reading from a file) or some CPU-intensive task (such as parsing a file or calculating something).

We can simulate a blocking call by calling the time.sleep() function in our task() coroutine.

# simple task that takes a moment
async def task(value):
    # report a message
    print(f'Task is running, value={value}')
    # block a moment
    time.sleep(2)
    # report a message
    print(f'Task is done, value={value}')

This will cause the asyncio event loop to block while executing our task.

We will run the event loop in debug mode and use logging configured to the DEBUG-level. The task will block for 2 seconds (more than 0.1 seconds required to trigger a warning), and a warning log message will be reported.

If we dig into the asyncio module code base on GitHub, we can see that in the code for running tasks, the event loop operates differently when in the debug mode (as we are), recording the start and end time of a task, and reporting the warning message only if the task takes longer than the minimum allowed time (defined by slow_callback_duration).

Here is a snippet from the _run_once() method on the event loop from the "asyncio/base_events.py" file that shows this:

...
if self._debug:
    try:
        self._current_handle = handle
        t0 = self.time()
        handle._run()
        dt = self.time() - t0
        if dt >= self.slow_callback_duration:
            logger.warning('Executing %s took %.3f seconds',
                           _format_handle(handle), dt)

Note: we cannot trigger a warning by calling asyncio.sleep(). This is because asyncio.sleep() is a coroutine that is awaited, allowing the asyncio event loop to operate normally. A call to time.sleep() will block the entire event loop and prevent it from progressing.

You can learn more about asyncio.sleep() vs time.sleep() in the tutorial:

• Asyncio sleep() in Python

Tying this together, the complete example is listed below.

# SuperFastPython.com
# example of task took too long in debug mode with logging
import logging
import time
import asyncio

# simple task that takes a moment
async def task(value):
    # report a message
    print(f'Task is running, value={value}')
    # block a moment
    time.sleep(2)
    # report a message
    print(f'Task is done, value={value}')

# main coroutine
async def main():
    # execute and await the task
    await task(99)

# enable debug logging
logging.basicConfig(level=logging.DEBUG)
# start the event loop
asyncio.run(main(), debug=True)

Running the program first sets debug level logging, then starts the event loop in debug mode.

The main() coroutine runs, executes the task() coroutine and the program terminates.

We can see that as before, the asyncio module reports the selector that is being used, e.g. KqueueSelector.

The task() coroutine also reports its print message as before.

In this case, we can see the warning message of the task taking too long. Specifically, we can see a WARNING level message was reported that includes the details of the task and how long it took.

Finally, we can see a DEBUG log message on the final state of the event loop.

DEBUG:asyncio:Using selector: KqueueSelector
Task is running, value=99
Task is done, value=99
WARNING:asyncio:Executing <Task finished name='Task-1' coro=<main() done, defined at ...> result=None created at ...> took 2.004 seconds
DEBUG:asyncio:Close <_UnixSelectorEventLoop running=False closed=False debug=True>

Next, let's explore the same example with warnings enabled instead of logging.

Example of Task Took Too Long in Debug Mode (without logging)

We can explore the reporting of tasks that take long when running the asyncio event loop in debug mode, by using the logging infrastructure.

Tasks that take too long are reported as a warning. We can allow just warning messages to be reported (e.g. to stdout) via the warnings module.

This is preferable when debugging our programs and not wanting to interfere with existing logging infrastructure.

We can enable all warnings in our asyncio program before starting the event loop in debug mode.

...
# enable all warnings
warnings.simplefilter('always')
# start the event loop
asyncio.run(main(), debug=True)

Tying this together, the complete example is listed below.

# SuperFastPython.com
# example of task took too long in debug mode with warnings
import warnings
import time
import asyncio

# simple task that takes a moment
async def task(value):
    # report a message
    print(f'Task is running, value={value}')
    # block a moment
    time.sleep(2)
    # report a message
    print(f'Task is done, value={value}')

# main coroutine
async def main():
    # execute and await the task
    await task(99)

# enable all warnings
warnings.simplefilter('always')
# start the event loop
asyncio.run(main(), debug=True)

Running the program first sets all warnings to be visible, then starts the event loop in debug mode.

The main() coroutine runs, executes the task() coroutine and the program terminates.

We can see that no debug log messages are shown, as we did not configure the logging infrastructure.

Instead, in this case, we can see that the details of long-running blocking tasks are reported directly on standard output.

This highlights how we can run the asyncio event loop in debug mode and see helpful warning messages, without enabling or changing the logging infrastructure of our programming.

Task is running, value=99
Task is done, value=99
Executing <Task finished name='Task-1' coro=<main() done, defined at ...> result=None created at ...> took 2.004 seconds

Next, let's explore an example of configuring how long is "too long" for long-running tasks.

Example of Configuring slow_callback_duration in Debug Mode

We can configure how long is "too long" reporting a warning message for long-running tasks in the event loop.

This can be achieved by setting the slow_callback_duration attribute on the asyncio event loop object.

In this case, we will decrease the duration from 100 milliseconds down to 10 milliseconds. This can be done in our main() coroutine before awaiting any additional coroutines.

...
# change the how long limit for reporting warnings
loop = asyncio.get_running_loop()
loop.slow_callback_duration = 0.01 # 10 ms

We can then have our coroutine block the event loop for 12 milliseconds.

# simple task that takes a moment
async def task(value):
    # report a message
    print(f'Task is running, value={value}')
    # block a moment
    time.sleep(0.012)
    # report a message
    print(f'Task is done, value={value}')

We will use logging as above to show the warning message and run the event loop in debug mode to ensure that the warning about long-running tasks is emitted.

...
# enable debug logging
logging.basicConfig(level=logging.DEBUG)
# start the event loop
asyncio.run(main(), debug=True)

Tying this together, the complete example is listed below.

# SuperFastPython.com
# example of configuring slow_callback_duration in debug mode
import logging
import time
import asyncio

# simple task that takes a moment
async def task(value):
    # report a message
    print(f'Task is running, value={value}')
    # block a moment
    time.sleep(0.012)
    # report a message
    print(f'Task is done, value={value}')

# main coroutine
async def main():
    # change the how long limit for reporting warnings
    loop = asyncio.get_running_loop()
    loop.slow_callback_duration = 0.01 # 10 ms
    # execute and await the task
    await task(99)

# enable debug logging
logging.basicConfig(level=logging.DEBUG)
# start the event loop
asyncio.run(main(), debug=True)

Running the program first configures the log and sets it to DEBUG level, then starts the event loop in debug mode.

The main() coroutine runs, executes the task() coroutine and the program terminates.

We can see that the warning message reported that the task() coroutine took too long, in this case, longer than 10 milliseconds.

This highlights that we can configure the tolerance for how long is "too long" for blocking tasks in our asyncio programs before a warning message is reported.

DEBUG:asyncio:Using selector: KqueueSelector
Task is running, value=99
Task is done, value=99
WARNING:asyncio:Executing <Task finished name='Task-1' coro=<main() done, defined at ...> result=None created at ...> took 0.015 seconds
DEBUG:asyncio:Close <_UnixSelectorEventLoop running=False closed=False debug=True>

Next, let's explore reporting of coroutines that are not awaited.

Example of Not Awaited Coroutine Reported in Debug Mode

A common problem in asyncio programs is creating coroutines and tasks and then not awaiting them.

This is a problem because the created coroutines and tasks will never be executed. This often leaves developers confused and frustrated, because it looks like the code is being called because creating a coroutine looks like a function call.

Therefore asyncio tries to detect coroutines that are created and not executed and report them.

This behavior is different when running the asyncio event loop in debug mode and not.

Let's take a look at two examples.

1. Create a coroutine that is not awaited (without debug mode).
2. Create a coroutine that is not awaited (with debug mode).

We can then compare the RuntimeWarning messages that are emitted automatically by asyncio, without logging or warnings configured.

Unawaited Coroutine in Normal Mode (Not Debug)

We can explore an example of creating a coroutine and not awaiting it in our asyncio program.

In this case, we will not run the event loop in debug mode and not enable logging or warnings.

By default, asyncio works hard to detect this scenario and reports a RuntimeWarning. The warnings module allows this to be reported to standard error output by default, with a message like "RuntimeWarning: coroutine '...' was never awaited"

You can learn more about this in the tutorial:

• RuntimeWarning: Coroutine Was Never Awaited

The complete example is listed below.

# SuperFastPython.com
# example of unawaited coroutines report a RuntimeWarning
import asyncio

# simple task that takes a moment
async def task(value):
    # report a message
    print(f'Task is running, value={value}')
    # block a moment
    asyncio.sleep(2) # forget to await
    # report a message
    print(f'Task is done, value={value}')

# main coroutine
async def main():
    # execute and await the task
    await task(99)

# start the event loop
asyncio.run(main(), debug=False)

Running the example starts the event loop, not in debug mode.

The main() coroutine runs, then awaits the task() coroutine.

The task() coroutine runs, reports a message then creates the asyncio.sleep() coroutine and does not await it. This means it is not given a chance to run.

A RuntimeWarning is raised and reported "RuntimeWarning: coroutine 'sleep' was never awaited", as controlled by the default configuration of the warnings module.

We can also see a second RuntimeWarning is reported "RuntimeWarning: Enable tracemalloc to get the object allocation traceback". This is just a suggestion on how to learn more about where the fault occurred.

This highlights that even without debug log level, additional warning configuration, or running the event loop in debug mode, we can see RuntimeWarning messages for coroutines that are created and not run.

Task is running, value=99
...
RuntimeWarning: coroutine 'sleep' was never awaited
  asyncio.sleep(2) # forget to await
RuntimeWarning: Enable tracemalloc to get the object allocation traceback
Task is done, value=99

Next, let's consider the same case with the event loop in debug mode.

Unawaited Coroutine in Debug Mode

We can explore the case of creating coroutines that are not awaited when the asyncio event loop is in debug mode.

Again, we don't need to set DEBUG level logging or change the warning level.

...
# start the event loop
asyncio.run(main(), debug=True)

Tying this together, the complete example is listed below.

# SuperFastPython.com
# example of unawaited coroutines report a RuntimeWarning
import asyncio

# simple task that takes a moment
async def task(value):
    # report a message
    print(f'Task is running, value={value}')
    # block a moment
    asyncio.sleep(2) # forget to await
    # report a message
    print(f'Task is done, value={value}')

# main coroutine
async def main():
    # execute and await the task
    await task(99)

# start the event loop
asyncio.run(main(), debug=True)

Running the example starts the event loop, this time in debug mode.

The main() coroutine runs, then awaits the task() coroutine.

The task() coroutine runs, reports a message then creates the asyncio.sleep() coroutine and does not await it. This means it is not given a chance to run.

As above, a RuntimeWarning is raised and reported "RuntimeWarning: coroutine 'sleep' was never awaited", as controlled by the default configuration of the warnings module.

Then the full stack trace of the coroutine that was not run is reported.

Note, an exception is not raised and does not disrupt the execution of the program. Instead, these are RuntimeWarning messages. This can be confirmed because the next line after the coroutine that was not run is a second print statement, that was executed and reported successful.

This highlights that when we run the event loop in debug mode, we can get a lot more details about coroutines that are created and not run.

Task is running, value=99
...:10: RuntimeWarning: coroutine 'sleep' was never awaited
Coroutine created at (most recent call last)
  File "...", line 20, in <module>
    asyncio.run(main(), debug=True)
  File ".../asyncio/runners.py", line 190, in run
    return runner.run(main)
  File ".../asyncio/runners.py", line 118, in run
    return self._loop.run_until_complete(task)
  File ".../asyncio/base_events.py", line 640, in run_until_complete
    self.run_forever()
  File ".../asyncio/base_events.py", line 607, in run_forever
    self._run_once()
  File ".../asyncio/base_events.py", line 1914, in _run_once
    handle._run()
  File ".../asyncio/events.py", line 80, in _run
    self._context.run(self._callback, *self._args)
  File "...", line 17, in main
    await task(99)
  File "...", line 10, in task
    asyncio.sleep(2) # forget to await
  asyncio.sleep(2) # forget to await
RuntimeWarning: Enable tracemalloc to get the object allocation traceback
Task is done, value=99

Next, let's look at how we can add debug messages to our own asyncio programs.

Example of Custom Debug Messages in Asyncio Program

We can explore an example of adding debug logging to our asyncio program.

In this case, we will update the above example to change our print() statements into debug log messages.

The logging module is comprehensive, and I recommend reading more about it:

• logging — Logging facility for Python

In this case, we will change our print() statements to the logging.debug() function.

# simple task that takes a moment
async def task(value):
    # report a message
    logging.debug(f'Task is running, value={value}')
    # block a moment
    await asyncio.sleep(2)
    # report a message
    logging.debug(f'Task is done, value={value}')

We will then configure logging for the program before starting the event loop.

...
# enable debug logging
logging.basicConfig(level=logging.DEBUG)
# start the event loop
asyncio.run(main())

Tying this together, the complete example is listed below.

# SuperFastPython.com
# example of custom debug logging
import logging
import asyncio

# simple task that takes a moment
async def task(value):
    # report a message
    logging.debug(f'Task is running, value={value}')
    # block a moment
    await asyncio.sleep(2)
    # report a message
    logging.debug(f'Task is done, value={value}')

# main coroutine
async def main():
    # execute and await the task
    await task(99)

# enable debug logging
logging.basicConfig(level=logging.DEBUG)
# start the event loop
asyncio.run(main())

Running the program first sets debug level logging, then starts the event loop in debug mode.

The main() coroutine runs, and executes the task() coroutine.

The task() coroutine reports a debug message, sleeps, then reports a second debug message. These appear on standard out along with the message from the asyncio module.

This highlights how we might add debug logging to our asyncio program directly.

DEBUG:asyncio:Using selector: KqueueSelector
DEBUG:root:Task is running, value=99
DEBUG:root:Task is done, value=99

Takeaways

You now know how to debug asyncio programs.

If you enjoyed this tutorial, you will love my book: Python Asyncio Jump-Start. It covers everything you need to master the topic with hands-on examples and clear explanations.