Differentiate between compute and network based occupancy
- Dominant language
- Python
- Stars
- 1.7k
- Forks
- 778
- Avg merge
- 2h 50m
- Merged PRs (30d)
- 3
Description
Occupancy is an estimation of work the scheduler assigns to every worker. We compute this value in [`Scheduler._set_duration_estimate`](https://github.com/dask/distributed/blob/b133009cee88fd48c8a345cffde0a8e9163426a6/distributed/scheduler.py#L2784-L2810) which is invoked in a couple of places
- [transition processing->memory](https://github.com/dask/distributed/blob/b133009cee88fd48c8a345cffde0a8e9163426a6/distributed/scheduler.py#L2119-L2126) (iff previously unknown duration)
- [_add_to_processing](https://github.com/dask/distributed/blob/b133009cee88fd48c8a345cffde0a8e9163426a6/distributed/scheduler.py#L7652) (i.e. whenever we assign a task to a worker)
- [`_reevaluate_occupancy_worker`](https://github.com/dask/distributed/blob/b133009cee88fd48c8a345cffde0a8e9163426a6/distributed/scheduler.py#L3043-L3044) (periodically [if scheduler CUP load allows is](https://github.com/dask/distributed/blob/b133009cee88fd48c8a345cffde0a8e9163426a6/distributed/scheduler.py#L7450-L7468))
Occupancy is measured in seconds and is calculated by summing the _expected processing time_ of all tasks assigned to a worker. At all times, the invariant `sum(ws.processing.values()) ~ ws.occupancy` should hold (modulo floating point arithmetic errors).
This _processing time_ is defined as `TaskPrefix.duration_average + get_comm_cost(TaskState, WorkerState)`, i.e. the average _compute_ duration of the `TaskPrefix` (see [`Scheduler.get_task_duration`](https://github.com/dask/distributed/blob/b133009cee88fd48c8a345cffde0a8e9163426a6/distributed/scheduler.py#L2887-L2904)) and the estimated time to transfer all dependencies that are not, yet on that worker, see [`Scheduler.get_comm_cost`](https://github.com/dask/distributed/blob/b133009cee88fd48c8a345cffde0a8e9163426a6/distributed/scheduler.py#L2865-L2885)
Occupancy is used for four purposes
- `Scheduler.total_occupancy` (sum over all workers) is used to define an adaptive target
- `Scheduler.total_occupancy` (sum over all workers) is used to estimate worker saturation
- `WorkerState.processing` to calculate the steal_time ratio in work stealing
- `WorkerState.occupancy` for making a scheduling decision in `Scheduler.worker_objective`
With the exception of the work stealing case, all other examples are very specifically referring to the number of worker threads. Worker threads do not impact network/gather data performance.
Taking `Scheduler.worker_objective` trying to calculate `start_time` as an example, the actual start time should rather be
```python
wait_time_cpu: float = ws.compute_occupancy / ws.nthreads
wait_time_transfer: float = (ws.network_occupancy + comm_nbytes) / bandwidth
start_time = max(wait_time_transfer, wait_time_cpu)
```
This would likely increase the quality of our scheduling decisions and would very clearly avoid double counting problems like https://github.com/dask/distributed/issues/7003
On top, this would add a significant observability component since we would directly visualize how much network vs compute work is expected from a worker. I could also see a ratio of the two values to be an interesting metric to track (similar to what work stealing is trying to do with the steal ratio)
Contributor guide
Assessment
This issue has not been assessed yet.