dask / dask/distributed

Differentiate between compute and network based occupancy

Open
#7,004 0 comments 0 reactions 0 assignees View on GitHub
diagnostics discussion enhancement performance scheduler scheduling stealing
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

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.