dask / dask/distributed

Race condition between `Worker.close` and `Worker.heartbeat` causes worker not to close

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

Description

If a worker is asked to shutdown and enters `Worker.close`, but then the `heartbeat` periodic callback occurs and it notices it's missing on the scheduler, the worker will fail to close and will instead reconnect. This could be fixed with an `asyncio.Lock` around `status` changing methods so they can't run concurrently, but there may be another way.

This is resulting in a periodic test failure in the dask-gateway test suite. Here's an example test where we increase the likelihood of the race condition by manually calling `heartbeat` concurrently with `worker.close`:

```python
import asyncio
import distributed

async def main():
scheduler = distributed.Scheduler()
await scheduler
worker = distributed.Worker(scheduler.address)
await worker
client = await distributed.Client(scheduler.address, asynchronous=True)
res = await client.scheduler.identity()
await asyncio.gather(worker.close(), worker.heartbeat(), asyncio.sleep(2))
await worker.finished()

ok = worker.status == "closed"

await client.close()
await scheduler.close()
await scheduler.finished()

assert ok

if __name__ == "__main__":
asyncio.run(main())
```

and a failing output. You can see the logs where the worker revives itself due to the heartbeat:

```
$ python test.py
distributed.scheduler - INFO - Clear task state
distributed.scheduler - INFO - Scheduler at: tcp://10.238.82.44:52349
distributed.worker - INFO - Start worker at: tcp://10.238.82.44:52350
distributed.worker - INFO - Listening to: tcp://10.238.82.44:52350
distributed.worker - INFO - Waiting to connect to: tcp://10.238.82.44:52349
distributed.worker - INFO - -------------------------------------------------
distributed.worker - INFO - Threads: 8
distributed.worker - INFO - Memory: 17.18 GB
distributed.worker - INFO - Local Directory: /Users/jcrist/Code/dask-gateway/dask-worker-space/worker-lxfk56yz
distributed.worker - INFO - -------------------------------------------------
distributed.scheduler - INFO - Register tcp://10.238.82.44:52350
distributed.scheduler - INFO - Starting worker compute stream, tcp://10.238.82.44:52350
distributed.core - INFO - Starting established connection
distributed.worker - INFO - Registered to: tcp://10.238.82.44:52349
distributed.worker - INFO - -------------------------------------------------
distributed.core - INFO - Starting established connection
distributed.scheduler - INFO - Receive client connection: Client-f088b5e2-0bc1-11ea-bdef-24a074f2fd02
distributed.core - INFO - Starting established connection
distributed.worker - INFO - Stopping worker at tcp://10.238.82.44:52350
distributed.scheduler - INFO - Remove worker tcp://10.238.82.44:52350
distributed.core - INFO - Removing comms to tcp://10.238.82.44:52350
distributed.scheduler - INFO - Lost all workers
distributed.worker - INFO - -------------------------------------------------
distributed.scheduler - INFO - Register tcp://10.238.82.44:52350
distributed.scheduler - INFO - Starting worker compute stream, tcp://10.238.82.44:52350
distributed.core - INFO - Starting established connection
distributed.worker - INFO - Registered to: tcp://10.238.82.44:52349
distributed.worker - INFO - -------------------------------------------------
distributed.core - INFO - Starting established connection
distributed.scheduler - INFO - Remove client Client-f088b5e2-0bc1-11ea-bdef-24a074f2fd02
distributed.scheduler - INFO - Remove client Client-f088b5e2-0bc1-11ea-bdef-24a074f2fd02
distributed.scheduler - INFO - Close client connection: Client-f088b5e2-0bc1-11ea-bdef-24a074f2fd02
distributed.scheduler - INFO - Scheduler closing...
distributed.scheduler - INFO - Scheduler closing all comms
distributed.scheduler - INFO - Remove worker tcp://10.238.82.44:52350
distributed.core - INFO - Removing comms to tcp://10.238.82.44:52350
distributed.scheduler - INFO - Lost all workers
distributed.worker - INFO - Stopping worker at tcp://10.238.82.44:52350
Traceback (most recent call last):
File "test.py", line 25, in
asyncio.run(main())
File "/Users/jcrist/miniconda3/envs/dask-gateway/lib/python3.7/asyncio/runners.py", line 43, in run
return loop.run_until_complete(main)
File "/Users/jcrist/miniconda3/envs/dask-gateway/lib/python3.7/asyncio/base_events.py", line 579, in run_until_complete
return future.result()
File "test.py", line 21, in main
assert ok
AssertionError
```

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.