`Scheduler was unaware of this worker` message during worker shutdown
- Dominant language
- Python
- Stars
- 1.7k
- Forks
- 778
- Avg merge
- 2h 50m
- Merged PRs (30d)
- 3
Description
@bnaul reported seeing frequent messages like `Scheduler was unaware of this worker 'tcp://10.124.34.24:45585'. Shutting down` on an adaptive cluster with 2k workers. I think this message is likely to show up somewhat often during normal worker shutdown, due to the inconsistency in how the scheduler defines closed workers: https://github.com/dask/distributed/issues/6390. It's not actually a sign of anything being wrong in this case, but is noisy and misleading.
Here's a possible flow for a worker being closed:
* Scheduler calls `remove_worker` on the worker and _instantly_ [removes its state](https://github.com/dask/distributed/blob/16748b7e32aaa498c3ccb9006ea7237b154d146f/distributed/scheduler.py#L4342) tracking that worker’s existence. It also [queues an `{"op": "close"}`](https://github.com/dask/distributed/blob/16748b7e32aaa498c3ccb9006ea7237b154d146f/distributed/scheduler.py#L4325) message to send to that worker (but does not send it yet).
* Before that `{"op": "close"}` message has [reached the worker](https://github.com/dask/distributed/blob/16748b7e32aaa498c3ccb9006ea7237b154d146f/distributed/worker.py#L1467), the worker sends another heartbeat (or potentially, a heartbeat was even already on the wire before the scheduler ran `remove_worker`)
* Scheduler receives the heartbeat
* Scheduler [thinks the worker doesn’t exist](https://github.com/dask/distributed/blob/16748b7e32aaa498c3ccb9006ea7237b154d146f/distributed/scheduler.py#L3577-L3580), causing this response
* The worker is already shutting down, so the `{"status": "missing"}` response doesn’t actually change anything worker-side (it still shuts down as usual)
Eliminating this message on the worker side would be easy:
```diff
diff --git a/distributed/worker.py b/distributed/worker.py
index 86e1fba5..1173abc3 100644
--- a/distributed/worker.py
+++ b/distributed/worker.py
@@ -1177,12 +1177,13 @@ class Worker(BaseWorker, ServerNode):
self._update_latency(end - start)
if response["status"] == "missing":
- # Scheduler thought we left. Reconnection is not supported, so just shut down.
- logger.error(
- f"Scheduler was unaware of this worker {self.address!r}. Shutting down."
- )
- # Something is out of sync; have the nanny restart us if possible.
- await self.close(nanny=False)
+ if self.status not in (Status.closing, Status.closed):
+ # Scheduler thought we left. Reconnection is not supported, so just shut down.
+ logger.error(
+ f"Scheduler was unaware of this worker {self.address!r}. Shutting down."
+ )
+ # Something is out of sync; have the nanny restart us if possible.
+ await self.close(nanny=False)
return
self.scheduler_delay = response["time"] - middle
```
On the scheduler side though, it's reflective of a larger problem around how we represent the state of a closed worker: https://github.com/dask/distributed/issues/6390.
The problem is that the scheduler puts things in a "closed" state (by deleting `self.workers[address]`) while the worker may still be alive and connected. In the case of a clean shutdown, it would probably make sense if worker closure followed a request-response:
1. Scheduler asks worker to close. It keeps `workers[address]` around, but the `WorkerState.status` is set to `closing`, and any tasks it's running or storing are transitioned off immediately (like currently happens).
2. Worker responds with a confirmation of the closure, which would be guaranteed to be its last message to the scheduler.
3. Scheduler removes `workers[address]`.
This confirmation wouldn't have to be a message or RPC, per se—it could simply take the form of the worker closing the batched stream to the scheduler. The point is though, there needs to be some way of representing gray area where we've asked a worker to close, but haven't confirmed that it's gone yet.
Contributor guide
Assessment
This issue has not been assessed yet.