[FIXED] asyncio : with gather, a custom exception seems to trigger exceptions from other coroutines, with a strange python trace printed in the console

Issue

I have a Python 3.9 code like that :

    # do async dask stuff
    async def a(self):
      client = Client(address=self.cluster, asynchronous=True)
      ...

    # do async other stuff (not dask related)
    async def b(self):
      ...
      raise MyException("my custom exception")
      ...

    # main
    async def run(self):

        # create tasks
        task_a = asyncio.create_task(self.a())
        task_b = asyncio.create_task(self.b())

        # gather
        await asyncio.gather(
            task_a,
            task_b,
        )

Sometimes, when my custom exception arise, I have the following logs :

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 

I suppose that if I raise an exception in coroutine b, it is normal that the coroutine a (dask stuff) may failed abrutly.

However, I don't understand why in the Python Exception message, I have a label that tells me about a Dask Exception (distributed.scheduler - ERROR - Couldn't gather keys), but with a root cause labelled as my custom exception (raise MyException('my custom exception'))

How it is possible ? It is like two exceptions that have nothing to do, are merged into one strange exception, and printed by Python ..

Solution

I believe only one of these is an exception, and dask is only doing logging. Because of the way that async works, these things happen concurrently, and the output is intermingled.

As to why dask is failing here, I am not certain. It may simply be because your whole async program has come to a fast finish following the exception. Using return_exceptions=True in gather may solve that.

Answered By - mdurant