如何测量Python Trio协程中的阻塞执行耗时？

要实现排除协程挂起状态、仅统计阻塞执行耗时的需求，Trio可以通过自定义Instrument来优雅实现，这比asyncio的方案更贴合Trio的任务调度模型。以下是具体实现：

实现思路 #

Trio的Instrument接口允许我们监听任务的生命周期事件（如恢复、挂起、退出），通过跟踪这些事件的时间点，累加协程实际占用CPU执行的时间段，即可得到阻塞耗时（排除挂起的休眠/等待时间）。

完整代码示例 #

import trio
import time
from typing import Optional

class BlockingTimeInstrument(trio.abc.Instrument):
    def __init__(self):
        self.elapsed: float = 0.0
        self._last_resume_time: Optional[float] = None

    def task_resumed(self, task):
        # 任务恢复执行时，记录当前时间
        self._last_resume_time = time.perf_counter()

    def task_suspended(self, task):
        # 任务挂起时，计算从恢复到挂起的时长并累加
        if self._last_resume_time is not None:
            self.elapsed += time.perf_counter() - self._last_resume_time
            self._last_resume_time = None

    def task_exited(self, task):
        # 任务退出时，补全最后一段执行的时长
        if self._last_resume_time is not None:
            self.elapsed += time.perf_counter() - self._last_resume_time
            self._last_resume_time = None

class measure_blocking_time:
    def __init__(self):
        self.instrument = BlockingTimeInstrument()

    def __enter__(self):
        # 注册Instrument开始监听
        trio.lowlevel.add_instrument(self.instrument)
        return self.instrument

    def __exit__(self, exc_type, exc_val, exc_tb):
        # 移除Instrument结束监听
        trio.lowlevel.remove_instrument(self.instrument)

async def under_test():
    await trio.sleep(2)  # 挂起状态，不统计耗时
    time.sleep(3)        # 阻塞执行，统计耗时

async def measure():
    with measure_blocking_time() as ctx:
        await under_test()
    print(f"阻塞执行耗时: {ctx.elapsed:.2f}秒")
    # 考虑精度误差，用阈值判断而非严格相等
    assert abs(ctx.elapsed - 3) < 0.01, f"期望3秒，实际{ctx.elapsed:.2f}秒"

trio.run(measure)

代码说明 #

• BlockingTimeInstrument：实现Trio的Instrument接口，通过task_resumed和task_suspended记录任务执行的时间片段，task_exited处理任务退出时的收尾计算。
• measure_blocking_time：上下文管理器，负责在目标代码块执行前后注册/移除Instrument，确保测量范围准确。
• 断言时使用误差阈值：由于time.perf_counter存在微小精度误差，避免直接用==判断，改用小范围阈值保证断言可靠性。

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