如何统计使用Trio框架时Python协程的挂起次数？

你可以利用Trio官方提供的Instrument监控接口，实现自定义的挂起次数统计工具，完全匹配你示例中的track_suspensions()用法。

实现代码 #

import trio

class TrackSuspensions:
    def __init__(self):
        self.suspension_count = 0
        self._instrument = None
        self._target_task = None

    def __enter__(self):
        # 记录当前要监控的目标任务
        self._target_task = trio.lowlevel.current_task()
        
        # 定义自定义Instrument，仅统计目标任务的挂起
        class SuspensionTracker(trio.Instrument):
            def task_suspended(self, task):
                if task is self._target_task:
                    self.suspension_count += 1
        
        self._instrument = SuspensionTracker()
        trio.lowlevel.add_instrument(self._instrument)
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        # 退出上下文时移除监控器
        trio.lowlevel.remove_instrument(self._instrument)

# 封装成你想要的函数形式
def track_suspensions():
    return TrackSuspensions()

# 你的测试代码（可直接运行）
async def sleep(count: int):
    for _ in range(count):
        await trio.sleep(1)

async def test():
    with track_suspensions() as ctx:
        await sleep(1)
    assert ctx.suspension_count == 1, f"预期1次挂起，实际{ctx.suspension_count}次"

    with track_suspensions() as ctx:
        await sleep(3)
    assert ctx.suspension_count == 3, f"预期3次挂起，实际{ctx.suspension_count}次"

trio.run(test)

关键说明 #

• trio.Instrument是Trio提供的生命周期监控接口，task_suspended方法会在任意协程挂起时被调用
• 我们通过记录__enter__时的当前任务，确保只统计目标测试代码的挂起次数，避免其他后台协程干扰
• 上下文管理器的设计保证了监控器会自动添加和移除，不会影响其他测试逻辑

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