asyncio中含gather的任务CancelledError处理器未触发问题

问题场景 #

我编写的CancelledError异常处理器在单任务场景下能正常触发，但当任务内部使用asyncio.gather嵌套子任务时，处理器完全不生效。

示例代码 #

import asyncio


async def sleep_func(statement, time, sabotage=False):
    print(f"Starting: {statement}")
    try:
        if sabotage:
            tasks = [
                asyncio.sleep(1000),
                asyncio.sleep(1000),
                asyncio.sleep(1000),
            ]
            await asyncio.gather(*tasks)
        await asyncio.sleep(time)
    except asyncio.CancelledError as e:
        print(f"cancelled {statement}! - {str(e)}")
    except Exception as e:
        print(f"Unhandled exception - {str(e)}")
    print(f"Ending: {statement}")


async def main():
    calls = [
        asyncio.ensure_future(sleep_func("eat", 3)),
        asyncio.ensure_future(sleep_func("pray", 8)),
        asyncio.ensure_future(sleep_func("love", 10, sabotage=True)),
    ]
    print("starting!")
    finished, unfinished = await asyncio.wait(calls, timeout=6)
    for task in unfinished:
        task.cancel("This message should be shown when task is cancelled")


if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())

任务说明 #

主函数启动了三个异步任务：

[
    asyncio.ensure_future(sleep_func("eat", 3)),
    asyncio.ensure_future(sleep_func("pray", 8)),
    asyncio.ensure_future(sleep_func("love", 10, sabotage=True)),
]

结果对比 #

• 预期结果：
  1. 第一个任务正常完成（符合预期）
  2. 第二个任务被取消，打印CancelledError处理器中的提示（符合预期）
  3. 第三个任务被取消，打印CancelledError处理器中的提示（未符合预期）
• 当前实际输出：

starting!
Starting: eat
Starting: pray
Starting: love
Ending: eat
cancelled pray! - This message should be shown when task is cancelled
Ending: pray

• 预期输出：

starting!
Starting: eat
Starting: pray
Starting: love
Ending: eat
cancelled pray! - This message should be shown when task is cancelled
Ending: pray
cancelled love! - This message should be shown when task is cancelled
Ending: love

问题根源 #

你在调用task.cancel()取消未完成的任务后，没有等待这些任务完成取消后的收尾逻辑，main函数就直接结束了。事件循环在main执行完毕后立即退出，第三个任务因为嵌套了asyncio.gather，子任务的取消和异常处理需要更多时间，还没来得及执行CancelledError处理器里的代码就被强制终止了。

第二个任务能正常触发处理器，只是因为它的逻辑简单，取消后的收尾操作在事件循环退出前刚好完成。

解决方案 #

在main函数中，取消未完成的任务后，添加代码等待这些任务完成异常处理逻辑：

修改后的main函数：

async def main():
    calls = [
        asyncio.ensure_future(sleep_func("eat", 3)),
        asyncio.ensure_future(sleep_func("pray", 8)),
        asyncio.ensure_future(sleep_func("love", 10, sabotage=True)),
    ]
    print("starting!")
    finished, unfinished = await asyncio.wait(calls, timeout=6)
    for task in unfinished:
        task.cancel("This message should be shown when task is cancelled")
    # 等待所有未完成任务处理完取消逻辑，return_exceptions避免异常向上抛出
    await asyncio.gather(*unfinished, return_exceptions=True)

修改后的输出 #

运行修改后的代码，就能得到预期的输出结果：

starting!
Starting: eat
Starting: pray
Starting: love
Ending: eat
cancelled pray! - This message should be shown when task is cancelled
Ending: pray
cancelled love! - This message should be shown when task is cancelled
Ending: love

补充说明 #

• return_exceptions=True参数让asyncio.gather将任务的异常（包括CancelledError）作为返回值处理，而非再次抛出，避免main函数因异常中断。
• 任务被取消时，await语句会触发CancelledError，只要该await处于try代码块内就能被捕获处理，但前提是事件循环要给任务足够的时间执行异常处理逻辑。

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