dask / dask/distributed

Client submit with workers doesn’t handle new joining workers correctly

Open
#8,862 1 comment 0 reactions 0 assignees View on GitHub
bug scheduling
Dominant language
Python
Stars
1.7k
Forks
778
Avg merge
2h 50m
Merged PRs (30d)
3

Description

Use `client.submit()` with `workers` argument will run on other workers if they will be joining after you run the submit

Do reproduce it, first run a scheduler and the two workers
```shell
dask scheduler
dask worker tcp://192.0.0.100:8786 --nthreads 1 --name a-0
dask worker tcp://192.0.0.100:8786 --nthreads 1 --name a-1
```
then submit the jobs
```python
import time
from distributed import get_worker
from dask.distributed import Client

client = Client("tcp://192.0.0.100:8786")

def run_me(i):
worker = get_worker()
print(f"worker name {worker.name}, run {i}")
time.sleep(30)

futures = []
for i in range(100):
futures.append(client.submit(run_me, i, workers=["a-0", "b-0"]))

for f in futures:
f.result()
```
This is working properly and only worker `a-0` is getting jobs.
adding one more worker
```shell
dask worker tcp://192.0.0.100:8786 --nthreads 1 --name b-0
```
Now worker a-1 start getting jobs as well
P.S, If instead of adding worker with name `b-0` I add worker with name `b-1`, everything will work correct, so the problem is only happen when I'm adding a worker name that part of `workers` argument.

- Dask version: 2024.8.1
- Python version: 3.10.14
- Operating System: macOS
- Install method (conda, pip, source): pip

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.