dask / dask/distributed

SpecCluster + Dask JobQueue does not retire workers when scaling down.

Open
#4,069 3 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

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

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.