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.
- What is happening inside the asyncio module.
- What is happening inside our program.
This can be achieved by logging debug messages.
How can we debug asyncio?
Run loops using all CPUs, download your FREE book to learn how.
How to Debug Asyncio
There are perhaps three considerations when debugging asyncio, from a high level.
- Accessing debug messages from the asyncio module.
- Running the asyncio event loop in debug mode.
- Accessing warning messages.
- 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:
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:
1 2 3 |
... # 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:
- Via an argument to asyncio.run()
- Via a method on the event loop itself.
- 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:
1 2 3 |
... # 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:
1 2 3 4 |
... # 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
1 |
PYTHONASYNCIODEBUG=1 |
It may also be set programmatically via the os.environ() function.
For example:
1 2 3 |
... # 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:
- Coroutines that were not awaited are logged, e.g. the stack trace.
- Thread-unsafe methods raise exceptions if called from the wrong threads.
- Execution of I/O is logged, if it takes too long.
- 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:
1 2 3 4 |
... # 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:
1 2 3 |
... 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:
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:
1 2 3 |
... # enable warnings warnings.simplefilter('always', ResourceWarning) |
Or all warnings:
1 2 3 |
... # 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:
1 2 3 4 5 6 |
# 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:
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.
1 2 3 4 5 |
... # 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.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 |
# 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.
1 2 3 |
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.
Free Python Asyncio Course
Download your FREE Asyncio PDF cheat sheet and get BONUS access to my free 7-day crash course on the Asyncio API.
Discover how to use the Python asyncio module including how to define, create, and run new coroutines and how to use non-blocking I/O.
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.
1 2 3 4 5 |
... # 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.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 |
# 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)
1 2 3 4 |
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.
Overwhelmed by the python concurrency APIs?
Find relief, download my FREE Python Concurrency Mind Maps
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.
1 2 3 4 5 6 7 8 |
# 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:
1 2 3 4 5 6 7 8 9 10 |
... 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:
Tying this together, the complete example is listed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 |
# 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.
1 2 3 4 5 |
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.
1 2 3 4 5 |
... # enable all warnings warnings.simplefilter('always') # start the event loop asyncio.run(main(), debug=True) |
Tying this together, the complete example is listed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 |
# 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.
1 2 3 |
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.
1 2 3 4 |
... # 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.
1 2 3 4 5 6 7 8 |
# 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.
1 2 3 4 5 |
... # 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.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 |
# 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.
1 2 3 4 5 |
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.
- Create a coroutine that is not awaited (without debug mode).
- 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:
The complete example is listed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 |
# 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.
1 2 3 4 5 6 |
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.
1 2 3 |
... # start the event loop asyncio.run(main(), debug=True) |
Tying this together, the complete example is listed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 |
# 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.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 |
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:
In this case, we will change our print() statements to the logging.debug() function.
1 2 3 4 5 6 7 8 |
# 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.
1 2 3 4 5 |
... # enable debug logging logging.basicConfig(level=logging.DEBUG) # start the event loop asyncio.run(main()) |
Tying this together, the complete example is listed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 |
# 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.
1 2 3 |
DEBUG:asyncio:Using selector: KqueueSelector DEBUG:root:Task is running, value=99 DEBUG:root:Task is done, value=99 |
Further Reading
This section provides additional resources that you may find helpful.
Python Asyncio Books
- Python Asyncio Mastery, Jason Brownlee (my book!)
- Python Asyncio Jump-Start, Jason Brownlee.
- Python Asyncio Interview Questions, Jason Brownlee.
- Asyncio Module API Cheat Sheet
I also recommend the following books:
- Python Concurrency with asyncio, Matthew Fowler, 2022.
- Using Asyncio in Python, Caleb Hattingh, 2020.
- asyncio Recipes, Mohamed Mustapha Tahrioui, 2019.
Guides
APIs
- asyncio — Asynchronous I/O
- Asyncio Coroutines and Tasks
- Asyncio Streams
- Asyncio Subprocesses
- Asyncio Queues
- Asyncio Synchronization Primitives
References
Takeaways
You now know how to debug asyncio programs.
Did I make a mistake? See a typo?
I’m a simple humble human. Correct me, please!
Do you have any additional tips?
I’d love to hear about them!
Do you have any questions?
Ask your questions in the comments below and I will do my best to answer.
Photo by Heqi Valiansyah on Unsplash
Do you have any questions?