Idea was that if a `SortedSet` of unrunnable tasks is too expensive, then insertion order is probably _approximately_ priority order, since higher-priority (root) tasks will be scheduled first. This would give us O(1) for all necessary operations, instead of O(logn) for adding and removing.

Interestingly, the SortedSet implementation could be hacked to support O(1) `pop` and `popleft`, and inserting a min/max value. In the most common case (root tasks), we're always inserting a value that's greater than the max. Something like this might be the best tradeoff, since it gives us O(1) in the common case but still maintains the sorted gaurantee, which is easier to reason about.

Now task states on the dashboard are listed in the logical order that tasks transition through.

Simpler, though I think basically just an int of 1 may be the most reasonable.

This is just a hack currently, but maybe it would actually be useful?

This reverts commit df11f719b59aad11f39a27ccae7b2fd4dfd9243a.

When there were multiple root task groups, we were just re-using the last worker for every batch because it had nothing processing on it.

Unintentionally this also fixes dask#6597 in some cases (because the first task goes to processing, but we measure queued, so we pick the same worker for both task groups)

This reverts commit fdd5fd9.

1. The family metric itself is flawed. Added linear chain traversal, but it's still not good. The maxsize is problematic and probably the wrong way to think about it? a) there's quite likely no maxsize parameter that will ever be right, because you could always have multiple independent crazy substructures that are each maxsize+1. b) even when every task would be in the same family because they're all interconnected, there's still benefit to scheduling subsequent things together, even if you do partition. Minimizing priority partitions is always what you want. Maybe there's something where maxsize is not a hard cutoff, but a cutoff for where to split up interconnected structures?
2. Families probably need to be data structures? When a task completes, you'd like to know if it belongs to a family that actually has more tasks to run on that worker, vs the task just happens to look like it belongs to a family but was never scheduled as a rootish task.

Overall I like the family structure for scheduling up/down scaling, but figuring out how to identify them is tricky. Partitioning priority order is great because it totally avoids this problem, of course at the expense of scaling. Can we combine priority and graph structure to identify isolated families when reasonable, partition on priority when not?

Update docstring and add back logic for queuing disabled case

Just easier to explain this way

I think this fix is reasonable? I wonder if occupancy should include queued tasks though?

Tasks shouldn't be both `no-worker` and in the queue. If all workers are paused, tasks will currently to go `no-worker`, even if they're queued. If we then try to schedule them (because a slot opens up from task completion, tasks released, new worker joining, etc.) we find an invalid state.

This was logging the actual end state, instead of the recommended end state

If all workers were paused, we would put tasks in the `no-worker` state. Now that `queued` is a thing, we want queued tasks in this case to just stay on the queue, and not be added to `unrunnable`.

This commit takes the opposite of @crusaderky's view in https://github.com/dask/distributed/pull/5665/files#r787886583, and makes `idle` always a subset of `running`. Even if pedantically, the name `idle` isn't quite accurate, `idle` is typically _used_ as the set of "prime candidate for new tasks", so we make it that way.

We do this to maintain the invariant that `valid_workers` always returns None if the task doesn't have restrictions. Our root task detection logic relied on this, as did the `not ts.loose_restrictions` check. Otherwise, when some workers are paused, root tasks will no longer be scheduled in the typical way.

There are other approaches here which might be simpler, which I'll explore in following commits.

A way more minimal fix than 5b9d825afb9ab3a61ab22afef3b047dde238bc5f, but not ideal because if only some workers are paused, we'll get root task overproduction on the others (because having `valid_workers` bypasses the root task detection logic).

`valid_workers` will return a set if some workers are paused, even if the task doesn't have restrictions. This is anoying and a bit misleading, but possibly less intrusive of a change than 5b9d825afb9ab3a61ab22afef3b047dde238bc5f?

I think overall excluding paused workers from idle is just more sensible. As is having `valid_workers` not concern itself with `running` workers. `valid_workers` should only deal in _task-specific_ restrictions, not restrictions that would apply to all tasks.

This reverts commits 65735f8, ffbb53b.

Workers seem to be running out of memory on CI. Probably different base unmanaged memory sizes than my machine. This is tricky.

…no-co-assign

Bit of a hack, but closes dask#6597. I'd like to have a better metric for the batch size, but I think this is about as good as we can get. Any reasonably large number will do here.

Want to see if CI passes. This would be retaining current scheduling behavior. Task withholding would be behind a feature flag.

…no-co-assign

Already covered by `if p < nc` in `check_idle_saturated`. But the one removed here didn't check for `status == Status.running`

Not sure why these numbers changed

Apparently they're just unpredictable

TODO this is one of the main things unhandled in this PR: how do we address occupancy? Do queued tasks contribute to total occupancy or not? In either case, how is that implemented?? (I don't really want to make a `queued_occ` dict tracking per-task occupancy, like we have for processing; that feels like overkill.)

I mistakenly thought that in the transitions loop, new recommendations were processed after old ones. I believe it's the opposite (`dict.update` will add the new items at the end, `dict.popitem` will pop those new items off the end).

It wouldn't be too hard to sort all the recommendations here, just some extra allocations and copies.

Co-authored-by: crusaderky <[email protected]>

The previous naming and docstring was just wrong.

This overhauls `decide_worker` into separate methods for different
cases.

More importantly, it explicitly turns `transition_waiting_processing`
into the primary dispatch mechanism for ready tasks.

All ready tasks (deps in memory) now always get recommended to
processing, regardless of whether there are any workers in the cluster,
whether the have restrictions, whether they're root-ish, etc.

`transition_waiting_processing` then decides how to handle them
(depending on whether they're root-ish or not), and calls the
appropriate `decide_worker` method to search for a worker.

If a worker isn't available, then it recommends them off to `queued` or
`no-worker` (depending, again, on whether they're root-ish and the
WORKER_SATURATION setting).

This also updates the `no-worker` state to better match `queued`.
Before, `bulk_schedule_after_adding_worker` would send `no-worker` tasks
to `waiting`, which would then send them to `processing`. This was
weird, because in order to be in `no-worker`, they should already be ready
to run (just in need of a worker). So going straight to `processing` makes
more sense than sending a ready task back to waiting.

Finally, this adds a `SchedulerState.is_rootish` helper. Not quite the
static field on a task @fjetter wants in dask#6922, but a step in that
direction.

The only valid way I can imagine any of these happening is `client.scatter` within a worker. If this is actually needed, I guess I should add an equivalent for queued?

…no-co-assign

This reverts commit 2cc8631.

This reverts commits dd88b0d 842ee71 100118a

Still maybe not a test that should run in CI, I just like how real-world it is. Let's see if picking the task size based on available memory helps on windows.

will add it back when we actually implement co-assignment

…no-co-assign

Co-authored-by: crusaderky <[email protected]>

I don't understand why it's flaking on windows, but I imagine it's just because memory measurement and process memory overhead behaves differently. It could really just run on linux, but leaving it un-skipped for macOS right now out of convenience for macOS developers to run locally.

…no-co-assign

don't even need to pass it in right now; it's not used

feeling pretty good about just `test_graph_execution_width`

Using it as an API saves having to manage `running` and `idle` in multiple places

…no-co-assign

hesitant on this, but I don't want to introduce a flaky test