如何查看asyncio任务当前处于等待状态的位置？

我运行着一个复杂系统，其中包含名为“automation”的异步任务，希望查看该任务当前处于等待状态的具体位置（类似async-await的调用栈）。以下示例代码创建了该automation任务，其调用do_something，后者又调用sleep；运行时打印任务栈，但print_stack仅指向顶层协程automation中的await do_something()行，无法展示更深层级。

#!/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())

我尝试过使用asyncio.BaseEventLoop._scheduled，但示例中该值始终为[]；由于生产环境使用uvloop，我查阅了相关的issue和PR，但这些内容已停滞约4年。请问还有其他可行的方法吗？

方法1：利用inspect模块遍历协程帧栈 #

Python的inspect模块可直接获取协程的底层帧信息，通过递归遍历协程的cr_await属性，能完整输出整个调用链：

#!/usr/bin/env python3
import asyncio
import inspect

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()

def print_full_task_stack(task):
    print(f"Task stack for {task.get_name()}:")
    current = task._coro
    while current:
        frame = inspect.getframe(current)
        print(f"  File {frame.f_code.co_filename}, line {frame.f_lineno}, in {frame.f_code.co_name}")
        # 输出当前行代码（可选）
        line = inspect.getsource(frame).strip().split('\n')[frame.f_lineno - frame.f_code.co_firstlineno]
        print(f"    {line}")
        # 遍历下一层等待的协程
        current = getattr(current, 'cr_await', None)

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

asyncio.run(main())

运行后会输出完整调用栈，示例结果：

Task stack for automation:
  File test.py, line 14, in automation
    await do_something()
  File test.py, line 10, in do_something
    await sleep()
  File test.py, line 6, in sleep
    await asyncio.sleep(0.1)

方法2：开启asyncio调试模式增强栈信息 #

开启asyncio调试模式后，print_stack会输出更详细的上下文，也可通过PYTHONASYNCIODEBUG=1环境变量全局启用：

修改main函数启用调试：

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

注意：调试模式会带来一定性能损耗，建议仅在排查问题时使用。

方法3：兼容uvloop的通用栈遍历方案 #

uvloop的协程属性与标准asyncio略有差异，可通过兼容属性访问实现跨环境的栈遍历：

def print_uvloop_task_stack(task):
    print(f"Task stack for {task.get_name()}:")
    current = task._coro
    while current:
        # 适配uvloop的ag_frame属性与标准库的帧获取方式
        frame = current.ag_frame if hasattr(current, 'ag_frame') else inspect.getframe(current)
        print(f"  File {frame.f_code.co_filename}, line {frame.f_lineno}, in {frame.f_code.co_name}")
        line = inspect.getsource(frame).strip().split('\n')[frame.f_lineno - frame.f_code.co_firstlineno]
        print(f"    {line}")
        # 适配uvloop的ag_await与标准库的cr_await属性
        current = getattr(current, 'ag_await', getattr(current, 'cr_await', None))

该方法同时兼容标准asyncio和uvloop环境，可直接替换方法1中的print_full_task_stack函数使用。

内容的提问来源于stack exchange，提问作者Falko