dask / dask/distributed

Bug: putative race condition when scaling down LocalCluster

Open
#3,526 7 comments 0 reactions 0 assignees View on GitHub
Dominant language
Python
Stars
1.7k
Forks
778
Avg merge
2h 50m
Merged PRs (30d)
3

Description

@jcrist
```python
import dask.array as da
from distributed import LocalCluster, Client
def test_cluster(num_workers):
cl = Client(LocalCluster(n_workers=8, threads_per_worker=1))
cl.cluster.scale(num_workers)

result = cl.compute(da.zeros((10,100), chunks=(1,-1)),sync=True)
return result

test_cluster(1)
```
This produces this following error:
```
---------------------------------------------------------------------------
KilledWorker Traceback (most recent call last)
in
----> 1 test_cluster(1)

in test_cluster(num_workers)
5 cl.cluster.scale(num_workers)
6
----> 7 result = cl.compute(da.zeros((10,100), chunks=(1,-1)),sync=True)
8 return result

~/miniconda3/lib/python3.7/site-packages/distributed/client.py in compute(self, collections, sync, optimize_graph, workers, allow_other_workers, resources, retries, priority, fifo_timeout, actors, traverse, **kwargs)
2798
2799 if sync:
-> 2800 result = self.gather(futures)
2801 else:
2802 result = futures

~/miniconda3/lib/python3.7/site-packages/distributed/client.py in gather(self, futures, errors, direct, asynchronous)
1891 direct=direct,
1892 local_worker=local_worker,
-> 1893 asynchronous=asynchronous,
1894 )
1895

~/miniconda3/lib/python3.7/site-packages/distributed/client.py in sync(self, func, asynchronous, callback_timeout, *args, **kwargs)
778 else:
779 return sync(
--> 780 self.loop, func, *args, callback_timeout=callback_timeout, **kwargs
781 )
782

~/miniconda3/lib/python3.7/site-packages/distributed/utils.py in sync(loop, func, callback_timeout, *args, **kwargs)
346 if error[0]:
347 typ, exc, tb = error[0]
--> 348 raise exc.with_traceback(tb)
349 else:
350 return result[0]

~/miniconda3/lib/python3.7/site-packages/distributed/utils.py in f()
330 if callback_timeout is not None:
331 future = asyncio.wait_for(future, callback_timeout)
--> 332 result[0] = yield future
333 except Exception as exc:
334 error[0] = sys.exc_info()

~/miniconda3/lib/python3.7/site-packages/tornado/gen.py in run(self)
733
734 try:
--> 735 value = future.result()
736 except Exception:
737 exc_info = sys.exc_info()

~/miniconda3/lib/python3.7/site-packages/distributed/client.py in _gather(self, futures, errors, direct, local_worker)
1750 exc = CancelledError(key)
1751 else:
-> 1752 raise exception.with_traceback(traceback)
1753 raise exc
1754 if errors == "skip":

KilledWorker: ("('zeros-1f6fef5a3ac954e8a1a222adcbe5ba9c', 3, 0)", )
```
A reasonable guess seems to be that a "doomed" worker is assigned work before it gets removed as an effect of the `scale` operation. Adding a `time.sleep(4)` call between scaling and computing prevents the error.

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.