如何在Python中分别测量协程的阻塞时间与挂起时间？

要区分协程中的阻塞耗时（占用事件循环线程、无法处理其他任务的时间）和挂起耗时（协程暂停、事件循环可处理其他任务的时间），可以通过以下两种方式实现：

方法一：监控任务状态（简洁高效） #

实现思路 #

将被装饰的协程包装为asyncio.Task对象，通过跟踪任务的状态变化（RUNNING/PENDING）来分别累加阻塞和挂起时间：

• RUNNING状态：协程正在执行同步代码，这段时间属于阻塞耗时
• PENDING状态：协程被挂起，这段时间属于挂起耗时

完整代码 #

import asyncio
import time

def time_this(coro_func):
    async def wrapper(*args, **kwargs):
        # 创建待监控的协程任务
        target_task = asyncio.create_task(coro_func(*args, **kwargs))
        
        blocking_total = 0.0
        suspended_total = 0.0
        last_timestamp = time.perf_counter()
        prev_state = target_task._state

        # 循环监控任务状态直到完成
        while not target_task.done():
            current_state = target_task._state
            current_timestamp = time.perf_counter()

            # 根据状态变化累加对应耗时
            if prev_state == "RUNNING":
                blocking_total += current_timestamp - last_timestamp
            elif prev_state == "PENDING" and current_state == "RUNNING":
                suspended_total += current_timestamp - last_timestamp

            prev_state = current_state
            last_timestamp = current_timestamp
            # 让出事件循环，避免监控逻辑占用过多资源
            await asyncio.sleep(0.001)

        # 处理任务完成前最后一段运行时间
        if prev_state == "RUNNING":
            blocking_total += time.perf_counter() - last_timestamp

        print(f"Time spent blocking IO: {blocking_total:.2f}, suspended: {suspended_total:.2f}")
        return await target_task

    return wrapper

@time_this
async def foo():
    time.sleep(1)
    await asyncio.sleep(1)

async def main():
    await foo()

asyncio.run(main())

说明 #

• target_task._state：CPython环境下的任务状态属性，RUNNING/PENDING分别对应协程的运行/挂起状态，虽为内部属性但在CPython中稳定可用
• 监控逻辑通过await asyncio.sleep(0.001)让出事件循环，确保原协程有足够执行时间，同时避免监控逻辑过度占用CPU

方法二：使用跟踪函数（兼容性更强） #

实现思路 #

利用Python的sys.settrace跟踪协程的执行流程，通过识别协程的调用、返回、挂起（yield）和恢复（resume）事件，精准统计阻塞和挂起时间，不依赖内部属性。

完整代码 #

import asyncio
import time
import sys

def time_this(coro_func):
    async def wrapper(*args, **kwargs):
        blocking_time = 0.0
        suspended_time = 0.0
        last_time = time.perf_counter()
        in_coro_sync = False
        coro_qualname = coro_func.__qualname__

        def trace_callback(frame, event, arg):
            nonlocal blocking_time, suspended_time, last_time, in_coro_sync
            current_time = time.perf_counter()

            # 识别目标协程的同步执行阶段
            if event == "call" and frame.f_code.co_qualname == coro_qualname:
                in_coro_sync = True
                blocking_time += current_time - last_time
            elif event == "return" and frame.f_code.co_qualname == coro_qualname:
                in_coro_sync = False
                blocking_time += current_time - last_time
            elif in_coro_sync and event in ("line", "call", "return"):
                # 协程执行同步代码，累加阻塞时间
                blocking_time += current_time - last_time
            elif event == "yield" and in_coro_sync:
                # 协程即将挂起，截止阻塞时间统计
                blocking_time += current_time - last_time
                in_coro_sync = False
            elif event == "resume" and arg is not None and arg.__qualname__ == coro_qualname:
                # 协程恢复执行，累加挂起时间
                suspended_time += current_time - last_time
                in_coro_sync = True

            last_time = current_time
            return trace_callback

        # 设置跟踪函数并执行协程
        sys.settrace(trace_callback)
        try:
            result = await coro_func(*args, **kwargs)
        finally:
            sys.settrace(None)

        # 补全最后一段阻塞时间
        if in_coro_sync:
            blocking_time += time.perf_counter() - last_time

        print(f"Time spent blocking IO: {blocking_time:.2f}, suspended: {suspended_time:.2f}")
        return result

    return wrapper

@time_this
async def foo():
    time.sleep(1)
    await asyncio.sleep(1)

async def main():
    await foo()

asyncio.run(main())

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