使用asyncio.gather时自定义异常触发其他协程异常的问题

我在Python 3.9中有如下代码：

# 执行异步Dask操作
async def a(self):
  client = Client(address=self.cluster, asynchronous=True)
  ...

# 执行其他异步操作（非Dask相关）
async def b(self):
  ...
  raise MyException("my custom exception")
  ...

# 主函数
async def run(self):

    # 创建任务
    task_a = asyncio.create_task(self.a())
    task_b = asyncio.create_task(self.b())

    # 收集任务结果
    await asyncio.gather(
        task_a,
        task_b,
    )

当自定义异常MyException触发时，有时会出现如下日志：

distributed.scheduler - ERROR - Couldn't gather keys {'dask_function-5aaee143b6e5594b8a497e82bda0a8b5': [], 'exec_custom_binary-babfef008702e061894c0fb0a44df016': []} state: ['processing', 'processing'] workers: []
Traceback (most recent call last):
  File "/.../myapp/app.py", line 45, in run_app
    loop.run_until_complete(dask_unit_job_app.run())
  File "/.../lib/python3.9/asyncio/base_events.py", line 642, in run_until_complete
    return future.result()
  File "/.../myapp/app.py", line 614, in run
    await asyncio.gather(
  File "/.../myapp/app.py", line 545, in cancel
    raise MyException('my custom exception')
....exceptions.MyException: my custom exception 

我理解协程b抛出异常时，协程a（Dask操作）可能会意外终止，但疑惑的是：Python异常信息中既包含Dask调度器的错误提示，根因却标注为自定义异常，两个无关异常为何会被合并打印？

回答 #

这俩不是合并的异常，是先后触发的两个独立日志输出，原因如下：

• asyncio.gather默认行为是：只要任意一个任务抛出异常，就会立即取消所有其他未完成的并行任务。协程b抛出MyException后，asyncio会立刻终止协程a的执行。
• 协程a被强制取消时，内部正在运行的Dask异步操作（比如已提交给集群的任务）被迫中断，Dask调度器无法追踪、收集那些处于处理状态的任务键，于是输出了自己的ERROR级日志。
• 最后asyncio.gather会把协程b抛出的MyException正式抛出，触发你看到的完整Traceback。

两者只是因为触发顺序连续，被日志系统先后打印出来，看起来像是合并在一起，但本质是两个独立事件：自定义异常是业务代码抛出的错误，Dask的日志是任务被意外中断导致的内部报错。

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