SpecCluster + Dask JobQueue does not retire workers when scaling down.
- Dominant language
- Python
- Stars
- 1.7k
- Forks
- 778
- Avg merge
- 2h 50m
- Merged PRs (30d)
- 3
Description
I am currently chasing what appears to be a bug in the combination of `distributed` and `dask-jobqueue`.
Particularly, I find that this piece of code almost never completes when running a dask-jobqueue `HTCondorCluster` with adaptive scaling enabled:
```
import time
def square(x):
time.sleep(0.5)
return x ** 2
def neg(x):
return -x
A = client.map(square, range(1000))
B = client.map(neg, A)
total = client.submit(sum, B)
total.result()
```
What happens is that, whenever the tasks get almost finished, one of the workers is killed. This results in lost keys, sets the progress back a few hundred tasks, and repeats indefinitely.
Putting some debugging in the decision making, it appears that the adaptive code is deciding to scale down whenever it decides there is insufficient work to support all the workers. This results in the invocation of the following code in `SpecCluster`:
https://github.com/dask/distributed/blob/2.24.0/distributed/deploy/spec.py#L327-L329
Particularly, the list `to_close` comes from `self.workers` and appears to be integers (e.g., `0`) and invokes the `scheduler.retire_workers` method. Adding a print statement, I see the following worker dictionary:
```
{0: , 1: ,
2: , 3: ,
4: , 5: }
```
However, that method filters passed workers on a different set of keys:
https://github.com/dask/distributed/blob/93a8d28933ebe4b82d715e9f6948be3c38c49953/distributed/scheduler.py#L3502
Examples from adding a print statement:
```
SortedDict({'tls://129.93.X.Y:32806': , 'tls://129.93.X.Y:32883': , 'tls://129.93.X.Y:32768': , 'tls://129.93.X.Y:32769': , 'tls://129.93.X.Y:32815': })
```
(IP addresses lightly redacted)
Because, for example, the key `0` does not match `tls://129.93.182.193:32806`, the set of workers to retire becomes the empty set and nothing occurs.
Next, the selected workers are immediately killed. Because the unique data is not replicated elsewhere, the intermediate task output is lost and has to be re-run. If you time it right, this will get into an loop until a given task has been killed enough times and becomes terminal because of a KilledWorker exception.
I don't see any immediately obvious solution as there's no standard naming scheme between the worker name and the job ID. However, I'm not overly familiar with the code.
**Environment**:
- Dask version: 0.240.
- Python version: 3.8
- Operating System: Linux
- Install method (conda, pip, source): conda
Contributor guide
Assessment
This issue has not been assessed yet.