如何获取asyncio.gather调用的异步函数回溯并定位调用方？

使用asyncio.gather并发调用func_c时，traceback.print_stack()输出的回溯无法区分调用方是func_b还是func_b_v2，所有调用的回溯完全一致。示例代码及输出如下：

示例代码 #

import asyncio
import traceback

async def func_c():
    print("func c")
    traceback.print_stack()

async def func_b():
    print("func b")
    tasks = [func_c() for _ in range(3)]
    await asyncio.gather(*tasks)
    
async def func_b_v2():
    print("func b v2")
    tasks = [func_c() for _ in range(3)]
    await asyncio.gather(*tasks)

async def func_a():
    print("func a")
    await func_b()
    await func_b_v2()
    

async def main():
    await func_a()
    
if __name__ == "__main__":
    asyncio.run(main())

问题回溯输出 #

File "test.py", line 28, in <module>
  asyncio.run(main())
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 1919, in _run_once
  handle._run()
File "asyncio/events.py", line 80, in _run
  self._context.run(self._callback, *self._args)
File "test.py", line 6, in func_c
  traceback.print_stack()

手动在函数间传递回溯的方式不符合代码库中func_c作为通用结果记录/回溯追踪函数的定位，尝试asyncio.create_task加循环await的调用方式，问题依然存在。

核心原因 #

异步任务由事件循环调度执行，当func_c运行时，原调用方func_b/func_b_v2已经退出当前调用栈，traceback.print_stack()只能追踪到事件循环的调度逻辑，无法获取实际调用方的栈信息。必须在创建任务时提前捕获调用上下文并传递。

方法一：手动传递调用栈（简单直接） #

修改func_c接收调用栈参数，在调用方捕获栈信息并传入：

import asyncio
import traceback
import inspect

async def func_c(caller_stack=None):
    print("func c")
    if caller_stack:
        print("调用方回溯:")
        traceback.print_list(caller_stack)
    else:
        traceback.print_stack()

async def func_b():
    print("func b")
    # 捕获当前调用栈，移除当前func_b的调用行
    caller_stack = traceback.extract_stack()[:-1]
    tasks = [func_c(caller_stack) for _ in range(3)]
    await asyncio.gather(*tasks)
    
async def func_b_v2():
    print("func b v2")
    caller_stack = traceback.extract_stack()[:-1]
    tasks = [func_c(caller_stack) for _ in range(3)]
    await asyncio.gather(*tasks)

# 其余代码不变

方法二：使用上下文变量自动传递（无侵入式） #

利用contextvars自动传递调用栈，无需修改func_c的参数：

import asyncio
import traceback
import contextvars

# 定义上下文变量存储调用栈
caller_stack_var = contextvars.ContextVar("caller_stack", default=None)

async def func_c():
    print("func c")
    caller_stack = caller_stack_var.get()
    if caller_stack:
        print("调用方回溯:")
        traceback.print_list(caller_stack)
    else:
        traceback.print_stack()

async def func_b():
    print("func b")
    caller_stack = traceback.extract_stack()[:-1]
    # 设置上下文变量，确保任务能获取到调用栈
    token = caller_stack_var.set(caller_stack)
    try:
        tasks = [func_c() for _ in range(3)]
        await asyncio.gather(*tasks)
    finally:
        # 重置上下文，避免影响其他任务
        caller_stack_var.reset(token)
    
async def func_b_v2():
    print("func b v2")
    caller_stack = traceback.extract_stack()[:-1]
    token = caller_stack_var.set(caller_stack)
    try:
        tasks = [func_c() for _ in range(3)]
        await asyncio.gather(*tasks)
    finally:
        caller_stack_var.reset(token)

# 其余代码不变

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