Worker addresses are treated as unique identifiers, but may not be
- Dominant language
- Python
- Stars
- 1.7k
- Forks
- 778
- Avg merge
- 2h 50m
- Merged PRs (30d)
- 3
Description
The `__hash__` of a `WorkerState` object is just its address: https://github.com/dask/distributed/blob/33fc50ca9817216bb4105b68f5e0859ebfb80fdb/distributed/scheduler.py#L480
As is the equality check (https://github.com/dask/distributed/pull/3321 https://github.com/dask/distributed/pull/3483):
https://github.com/dask/distributed/blob/33fc50ca9817216bb4105b68f5e0859ebfb80fdb/distributed/scheduler.py#L501-L504
And in general, there are a number of places where we store things in dicts keyed by worker address, and assume that `if ws.address in self.workers`, then `ws is self.workers[ws.address]`. (`stealing.py` is especially guilty—most of its logic is basically built around this.)
However, it's completely valid for a worker to disconnect, then for a new worker to connect from the same address. (Even with reconnection removed https://github.com/dask/distributed/pull/6361, a Nanny https://github.com/dask/distributed/issues/6387 or a user script could do this.) These are logically different workers, though they happen to have the same address.
This can cause:
* bad decisions: a scheduling or work-stealing decision is made about the old worker at that address; when it's enacted, there's a different worker at that address and the decision may no longer be appropriate
* deadlocks: a `WorkerState` object is updated which is no longer in `self.workers` (though its _address_ is), a `TaskState` is made to point at a `WorkerState` which has been removed, etc.
Outcomes:
* `WorkerState` objects should be uniquely identifiable. `WorkerState` objects referring to logically different `dask-worker` invocations must not be equal or have the same hash, even if they happen to have the same address.
* Any logic which gives up control flow (via `await`, or storing some state in a dict to be used later, etc.) must verify, each time it regains control, that the worker it's dealing with still exists in the cluster (not just that its address exists).
Alternatives:
* If this is too much of a change to make, we could instead maintain a monotonically-increasing set of worker addresses, and prohibit address reuse. The scheduler would just reject a worker trying to connect if it had an address we'd already seen before. Of course, this would eliminate the possibility of worker reconnection https://github.com/dask/distributed/issues/6391, and maybe break nannies too.
Causes https://github.com/dask/distributed/issues/6356, https://github.com/dask/distributed/issues/3256, https://github.com/dask/distributed/issues/6263, maybe https://github.com/dask/distributed/issues/3892
cc @crusaderky @fjetter @bnaul
Contributor guide
Assessment
This issue has not been assessed yet.