如何在 asyncio 中查找 CancelledError 的原因？答案

【问题标题】：How to find the cause of CancelledError in asyncio?如何在 asyncio 中查找 CancelledError 的原因？
【发布时间】：2022-11-17 20:23:03
【问题描述】：

我有一个依赖一些第三方库的大项目，有时它的执行会被 CancelledError 中断。

为了演示这个问题，让我们看一个小例子：

import asyncio


async def main():
    task = asyncio.create_task(foo())

    # Cancel the task in 1 second.
    loop = asyncio.get_event_loop()
    loop.call_later(1.0, lambda: task.cancel())

    await task


async def foo():
    await asyncio.sleep(999)


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

追溯：

Traceback (most recent call last):
  File "/Users/ss/Library/Application Support/JetBrains/PyCharm2021.2/scratches/async.py", line 19, in <module>
    asyncio.run(main())
  File "/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/asyncio/runners.py", line 43, in run
    return loop.run_until_complete(main)
  File "/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/asyncio/base_events.py", line 579, in run_until_complete
    return future.result()
concurrent.futures._base.CancelledError

如您所见，没有关于 CancelledError 起源位置的信息。我如何找出它的确切原因？

我想出的一种方法是放置大量 try/except 块，这些块将捕获 CancelledError 并缩小它的来源范围。但这很乏味。

【问题讨论】：

这不是我的完整回溯。 During handling of the above exception, another exception occurred:及之前的部分为还追溯的一部分。
@MisterMiyagi，这实际上是我看到的所有回溯。使用 python 3.7 测试。你能发布你看到的回溯吗？
It's rather a bit longer。虽然它是用 Python 3.9 记录的，chaining exists since 3.0。
@MisterMiyagi，谢谢。据我了解，您的回溯准确显示了取消的来源。可惜 python 3.7 没有这样做。
看起来这确实只适用于 3.9； 3.8 也删除了链。不过，我在变更日志中没有找到任何内容。

标签： python python-asyncio

【解决方案1】：

我通过对项目中的每个异步函数应用装饰器来解决它。装饰器的工作很简单——当函数引发 CancelledError 时记录一条消息。通过这种方式，我们将看到哪些功能（更重要的是，以何种顺序）被取消。

这是装饰器代码：

def log_cancellation(f):
    async def wrapper(*args, **kwargs):
        try:
            return await f(*args, **kwargs)
        except asyncio.CancelledError:
            print(f"Cancelled {f}")
            raise
    return wrapper

为了在任何地方添加这个装饰器，我使用了正则表达式。查找：(.*)(async def)。替换为：$1@log_cancellation $1$2。

另外为了避免在每个文件中导入log_cancellation，我修改了内置函数： builtins.log_cancellation = log_cancellation

【讨论】：

【解决方案2】：

如果不在其中调用 call_exception_handler，则无法处理任务的异步异常。这为异常处理程序提供了一个上下文，可以使用 set_exception_handler 对其进行自定义。

如果您正在创建异步任务，则必须在协程中使用 try/except，这可能会发生异常。我已经演示了一些使用set_exception_handler捕获异步任务异常的最小实现

import asyncio
import logging
from asyncio import CancelledError


def async_error_handler(loop, context):
    logger = logging.Logger(__name__)
    logger.error(context.get("message"))


async def main():
    loop = asyncio.get_running_loop()
    
    # Set custom exception handler
    loop.set_exception_handler(async_error_handler)
    task = loop.create_task(foo())

    # Cancel the task after 1 second
    loop.call_later(1, task.cancel)

    await task


async def foo():
    try:
        await asyncio.sleep(999)
    except CancelledError:
        # Catch the case when the coroutine has been canceled
        loop = asyncio.get_running_loop()

        # Emit an event to exception handler with custom context
        loop.call_exception_handler(context={
            "message": "Task has been canceled"
        })


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

context 有更多的属性也可以自定义。阅读有关错误处理的更多信息here。

【讨论】：

【解决方案3】：

rich 包帮助我们确定了 CancelledError 的原因，而无需更改太多代码。

from rich.console import Console

console = Console()

if __name__ == "__main__":
    try:
        asyncio.run(main())  # replace main() with your entrypoint
    except BaseException as e:
        console.print_exception(show_locals=True)

【讨论】：