In asyncio.tasks.py sleep() does not handle asyncio.exceptions.CancelledError correctly

[data2code]

Bug report

Bug description:

The example code below always throws an exception, when the dask computation is completed successfully:

from time import sleep
import dask
from dask.distributed import Client,progress
from dask_jobqueue import SGECluster
from dask import compute, persist, delayed

@dask.delayed
def f():
    sleep(1)
    return 1

cluster = SGECluster(cores=1, memory='1GB')
cluster.scale(1)
c = Client(cluster)
out=[ f() ]*3

try:
    x=persist(*out)
    progress(x)
    out=compute(x)
except:
    print("error")

c.close()
print(out[0])

Output:

(1, 1, 1)
2023-11-30 21:27:09,396 - distributed.scheduler - ERROR -
Traceback (most recent call last):
File "/miniconda3/envs/X/lib/python3.9/site-packages/distributed/utils.py", line 832, in wrapper
return await func(*args, **kwargs)
File "/miniconda3/envs/X/lib/python3.9/site-packages/distributed/scheduler.py", line 7246, in feed
await asyncio.sleep(interval)
File "/miniconda3/envs/X/lib/python3.9/asyncio/tasks.py", line 652, in sleep
return await future
asyncio.exceptions.CancelledError

The root cause was shown in the last three lines of the above output. The future was canceled, which I believe was the expected behavior, but sleep() did not handle this well.

A possible fix is to add two bond lines after line 652 as shown below.

Although my system in 3.9, the bug seems to be there in the latest cpython version. Could this be patched?

Thank you!

    try:
        return await future
### add two lines below
    except exceptions.CancelledError:
        pass
### end of added lines
    finally:
        h.cancel()

CPython versions tested on:

3.9

Operating systems tested on:

Linux

[itamaro]

The issue description involves dask, distributed, and miniconda.
Could you provide a reproducer that doesn't involve any 3rd party libraries?

Note that time.sleep isn't asyncio-aware. I don't know how this works with dask, but with Python's asyncio you'd need to await asyncio.sleep instead.

[hugovk]

Closing due to lack of response.