How to tell where an asyncio task is waiting?

Falko 0
Viewed 47

I have rather complex system running an asynchronous task called "automation". Meanwhile I would like to inspect where the task is currently waiting. Something like a callstack for async-await.

The following example creates such an automation task, stepping into do_something which in turn calls sleep. While this task is running, its stack is printed. I'd wish to see something like "automation → do_something → sleep". But print_stack only points to the line await do_something() in the top-level coroutine automation, but nothing more.

#!/usr/bin/env python3
import asyncio

async def sleep():
    await asyncio.sleep(0.1)

async def do_something():
    print('...')
    await sleep()

async def automation():
    for _ in range(10):
        await do_something()

async def main():
    task = asyncio.create_task(automation(), name='automation')
    while not task.done():
        task.print_stack()
        await asyncio.sleep(0.1)

asyncio.run(main())

I thought about using _scheduled from asyncio.BaseEventLoop, but this seems to be always [] in my example. And since my production code runs uvloop I looked into https://github.com/MagicStack/uvloop/issues/135, https://github.com/MagicStack/uvloop/issues/163 and https://github.com/MagicStack/uvloop/pull/171, all of which are stale for about 4 years.

Is there something else I could try?

coroutine python-asyncio python
1 Answers

I found something: When running asyncio with debug=True and using a different interval of 0.11s (to avoid both task being "in sync"), we can access asyncio.get_running_loop()._scheduled. This contains a list of asyncio.TimerHandle. In debug mode each TimerHandle has a proper _source_traceback containing a list of traceback.FrameSummary with information like filename, lineno, name and line.

...

async def main():
    task = asyncio.create_task(automation(), name='automation')
    while not task.done():
        for timer_handle in asyncio.get_running_loop()._scheduled:
            for frame_summary in timer_handle._source_traceback:
                print(f'{frame_summary.filename}:{frame_summary.lineno} {frame_summary.name}')
        await asyncio.sleep(0.11)

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

The output looks something like this:

/opt/homebrew/Cellar/python@3.10/3.10.6_1/Frameworks/Python.framework/Versions/3.10/lib/python3.10/asyncio/base_events.py:600 run_forever
/opt/homebrew/Cellar/python@3.10/3.10.6_1/Frameworks/Python.framework/Versions/3.10/lib/python3.10/asyncio/base_events.py:1888 _run_once
/opt/homebrew/Cellar/python@3.10/3.10.6_1/Frameworks/Python.framework/Versions/3.10/lib/python3.10/asyncio/events.py:80 _run
/Users/falko/./test.py:17 automation
/Users/falko/./test.py:12 do_something
/Users/falko/./test.py:7 sleep
/opt/homebrew/Cellar/python@3.10/3.10.6_1/Frameworks/Python.framework/Versions/3.10/lib/python3.10/asyncio/tasks.py:601 sleep

The main drawback for my application is the fact that uvloop doesn't come with the field _scheduled. That's why I'm still looking for an alternative approach.

Falko0
Related