Persist raises error when Future is cancelled
- Dominant language
- Python
- Stars
- 1.7k
- Forks
- 778
- Avg merge
- 2h 50m
- Merged PRs (30d)
- 3
Description
**Describe the issue**: Cancelling a future from `futures_of`, will cause subsequent `persist` calls to return a cancelled future error. Adding a small `asyncio.sleep` gets around the issue for a single event loop, but if there's another event loop or `Client` requesting the same future then this work-around doesn't fix the issue (as is the case if you have multiple machines each connecting with their own `Client`.
**Minimal Complete Verifiable Example**:
```python
from distributed.client import Client
import dask.array as da
from distributed import as_completed, futures_of
#the scheduler_address routes to a Dask Kubernetes deployment
scheduler_address = "my-scheduler.com:8786"
client = await Client(scheduler_address, direct_to_workers=True, asynchronous=True)
for i in range(2):
x = da.from_array([1,2,3,4,5])
x_persist = x.persist()
futures = futures_of(x_persist)
async for future, chunk_data in as_completed(futures, with_results=True):
print(future)
if issubclass(chunk_data.__class__, Exception):
raise chunk_data
await future.cancel()`
```

**Anything else we need to know?**: A similar issue exists for the non-asynchronous `Client`. Also, we need to cancel the future so it isn't retained in memory. The `asyncio.sleep` work-around seems to indicate there might be a race condition. Maybe we can update `persist` to force the future to be computed if the future is in a cancelled state?
**Environment**:
- Dask version: 2023.3.1
- Python version:
- Operating System: Linux
- Install method (conda, pip, source): pip/ poetry
Contributor guide
Assessment
This issue has not been assessed yet.