Python 3.11中TaskGroup遇任务异常冻结是否为已知Bug？

问题确认 #

这是Python 3.11中asyncio.TaskGroup与asyncio.wait_for结合使用时的已知功能性Bug。由于Python 3.11目前处于维护阶段（仅接收安全相关修复，不处理功能性Bug），官方不会针对此问题发布修复补丁。

问题根源 #

当TaskGroup中的某个任务抛出异常时，TaskGroup会尝试取消剩余任务，但Python 3.11中wait_for内部的取消处理逻辑存在缺陷：被取消时，wait_for无法正确唤醒正在等待Event.wait()的任务，导致任务永久阻塞，最终整个事件循环挂起。

规避方案 #

以下几种调整方式可解决该问题，在Python 3.11中实现预期的任务取消逻辑：

方案1：用asyncio.wait替代wait_for #

asyncio.wait的取消逻辑在3.11中不受此Bug影响，可替换原代码中的wait_for调用：

import asyncio

ev = asyncio.Event()

async def t1():
    try:
        while True:
            try:
                print("Waiting")
                # 用wait替代wait_for实现超时等待
                done, pending = await asyncio.wait([ev.wait()], timeout=99)
                if ev.is_set():
                    print("Done waiting")
                    ev.clear()
                else:
                    print("Timeout")
                    raise TimeoutError
            except TimeoutError:
                print("Timeout")
                raise
    except asyncio.CancelledError:
        print("Cancelled -  as expected")
        raise

async def t2():
    ev.set()
    raise RuntimeError()

async def main():
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(t1())
            tg.create_task(t2())
    except* RuntimeError:
        print("RuntimeError -  as expected")

if __name__ == "__main__":
    asyncio.run(main())

方案2：提前检查任务取消状态 #

在进入wait_for等待前，主动检查当前任务是否已被取消，提前触发CancelledError：

import asyncio

ev = asyncio.Event()

async def t1():
    try:
        while True:
            # 让出控制权并检查取消状态
            await asyncio.sleep(0)
            if asyncio.current_task().cancelled():
                raise asyncio.CancelledError()
            
            try:
                print("Waiting")
                await asyncio.wait_for(ev.wait(), 99)
                print("Done waiting")
                ev.clear()
            except TimeoutError:
                print("Timeout")
                raise
    except asyncio.CancelledError:
        print("Cancelled -  as expected")
        raise

async def t2():
    ev.set()
    raise RuntimeError()

async def main():
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(t1())
            tg.create_task(t2())
    except* RuntimeError:
        print("RuntimeError -  as expected")

if __name__ == "__main__":
    asyncio.run(main())

方案3：自定义超时等待函数 #

实现一个不受该Bug影响的自定义超时等待逻辑：

import asyncio

ev = asyncio.Event()

async def wait_with_timeout(fut, timeout):
    """自定义超时等待函数"""
    timeout_task = asyncio.create_task(asyncio.sleep(timeout))
    done, pending = await asyncio.wait(
        [fut, timeout_task],
        return_when=asyncio.FIRST_COMPLETED
    )
    if timeout_task in done:
        fut.cancel()
        raise TimeoutError
    else:
        timeout_task.cancel()
        return await fut

async def t1():
    try:
        while True:
            try:
                print("Waiting")
                await wait_with_timeout(ev.wait(), 99)
                print("Done waiting")
                ev.clear()
            except TimeoutError:
                print("Timeout")
                raise
    except asyncio.CancelledError:
        print("Cancelled -  as expected")
        raise

async def t2():
    ev.set()
    raise RuntimeError()

async def main():
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(t1())
            tg.create_task(t2())
    except* RuntimeError:
        print("RuntimeError -  as expected")

if __name__ == "__main__":
    asyncio.run(main())

验证效果 #

以上三种方案在Python 3.11中运行后，均可得到与Python 3.12一致的预期输出：

Waiting
Done waiting
Waiting
Cancelled -  as expected
RuntimeError -  as expected

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