已取消任务调用result()抛出InvalidStateError而非CancelledError问题咨询

我构建了如下结构，旨在调用一系列异步操作，在任一操作完成或超时到期时立即返回。但运行时发现，读取已取消任务的结果时抛出的是InvalidStateError而非预期的CancelledError。

import asyncio

async def async_operation():
        while True:
                await asyncio.sleep(1)

async def main():
    task1 = None
    task2 = None
    try:
        task1 = asyncio.create_task(async_operation())
        task2 = asyncio.create_task(async_operation())

        await asyncio.wait([task1, task2], timeout=2, return_when=asyncio.FIRST_COMPLETED)
    finally:
            if task1 and not task1.done():
                print("Cancelling task1!") # this is printed!
                task1.cancel()
            if task2 and not task2.done():
                task2.cancel()
    try:
            # This throws:
            #   asyncio.exceptions.InvalidStateError: Result is not set.
            res = task1.result()
            # I would expect one of the following two cases to be true:
            #
            #   1. the task completes and thus the result is set or
            #   2. it is cancelled in the finally block and thus calling .result()
            #      should raise CancelledError and not InvalidStateError
            print(f"Got result: ${res}")
    except asyncio.CancelledError:
            print(f"Caught CancelledError!")

asyncio.run(main())

原因解释 #

task.cancel()仅会给任务添加一个取消标记，并不会立即终止任务执行。任务只有在下次遇到await这类可等待对象时，才会触发CancelledError；如果任务内部没有捕获这个异常，异常会向上传播，任务才会进入已取消完成的状态。

在你的代码中，调用task1.cancel()后，没有等待任务处理取消流程就直接调用result()——此时任务仍处于pending状态（虽然被标记为取消，但还没执行到await asyncio.sleep(1)这个节点触发异常）。这种情况下调用result()，因为任务既没有正常完成（无结果），也没有因取消而完成（未触发异常），所以会抛出InvalidStateError。

规范实现建议 #

要让任务真正完成取消流程，需要在调用cancel()后等待任务结束，同时正确处理任务状态：

改进后的代码示例 #

import asyncio

async def async_operation():
    while True:
        await asyncio.sleep(1)

async def main():
    task1 = None
    task2 = None
    try:
        task1 = asyncio.create_task(async_operation())
        task2 = asyncio.create_task(async_operation())

        await asyncio.wait([task1, task2], timeout=2, return_when=asyncio.FIRST_COMPLETED)
    finally:
        # 收集需要取消的任务
        pending_tasks = []
        if task1 and not task1.done():
            print("Cancelling task1!")
            task1.cancel()
            pending_tasks.append(task1)
        if task2 and not task2.done():
            task2.cancel()
            pending_tasks.append(task2)
        
        # 等待任务完成取消流程，return_exceptions=True避免异常传播
        if pending_tasks:
            await asyncio.gather(*pending_tasks, return_exceptions=True)
    
    # 处理任务结果
    if task1.done():
        try:
            res = task1.result()
            print(f"Got result: {res}")
        except asyncio.CancelledError:
            print(f"Caught CancelledError for task1!")
        except Exception as e:
            print(f"Task1 raised exception: {e}")

asyncio.run(main())

关键要点 #

• 调用cancel()后，必须await任务，让它完成取消流程，进入done状态。
• 使用return_exceptions=True调用asyncio.gather()，可以避免取消任务的异常向上传播导致程序崩溃。
• 调用result()前，先通过task.done()检查任务状态，确保任务已完成（无论正常还是异常）。

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