[EDIT] this comment refers to a previous version of the PR. The condition described below is still there, but only triggered by multiple events in short sequence.

This PR introduces a O(n^2*logn) condition where

1. task y has 100 dependencies x1, ... x100 all from the same few workers (nworkers << ntasks)
2. x1 transitions released->fetch->_ensure_communicating->flight. len(data_needed) == 0.
3. x2 transitions released->fetch->_ensure_communicating-> remain in fetch. len(data_needed) == 1; we had to skip 0 tasks before it in data_needed.
4. ...x100 transitions released->fetch->_ensure_communicating-> remain in fetch. len(data_needed) == 99; we had to skip 98 tasks before it in data_needed.
   So we had ~100^2/2 ~=5000 pure-CPU iterations that get out of data_needed, write to skipped_worker_in_flight_or_busy, and then add back to data_needed at the end of _ensure_communicating. As data_needed is a heap, each push to it is O(logn), where n is the number of tasks contained within.

This should be negligible most of the time. I'll write another PR later on to make running _ensure_communicating twice in a row truly negligible (blocked by #6388).

To clarify: this condition is already there when you have many compute-task and acquire-replicas requests, which cause the data_needed queue to expand rapidly. This PR specifically extends the condition to tasks fetched within the same event.

Unit Test Results

       15 files  +       12         15 suites  +12   6h 40m 57s ⏱️ + 5h 53m 11s
  2 831 tests +  1 635    2 748 ✔️ +  1 586    81 💤 +  47  2 ❌ +2 
20 979 runs  +17 394  20 032 ✔️ +16 549  945 💤 +843  2 ❌ +2 

For more details on these failures, see this check.

Results for commit e1058fe. ± Comparison against base commit 69b798d.

♻️ This comment has been updated with latest results.

[EDIT] this comment refers to a previous version of the PR and is now obsolete.

The transition log has changed from

           - ('x', 'ensure-task-exists', 'released')
           - ('x', 'released', 'fetch', 'fetch', {})
           - ('gather-dependencies', 'tcp://127.0.0.1:53985', {'x'})
           - ('x', 'fetch', 'flight', 'flight', {})

to

           - ('x', 'ensure-task-exists', 'released'),
           - ('gather-dependencies', 'tcp://127.0.0.1:53985', {'x'}),
           - ('x', 'released', 'fetch', 'fetch', {'x': ('flight', 'tcp://127.0.0.1:53985')}),
           - ('x', 'fetch', 'flight', 'flight', {}),

This... is correct, but it's very counter-intuitive.
What's happening is that

1. transition_released_fetch starts. It sets ts.state = "fetch", adds it to data_needed, and internally invokes _ensure_communicating.
2. _ensure_communicating removes ts from data_needed, prints its own log line gather-dependencies, and returns a recommendation to transition to flight, together with a GatherDep instruction
3. transition_released_fetch returns
4. _transition logs the released->fetch transition and returns the recommendations, instructions generated by _ensure_communicating
5. _transitions calls _transition again for fetch->flight and logs the outcome.

Again, all this is correct, but it's confusing; it took me an unhealthy amount of time to figure out why the gather-dependencies log line appeared before ('x', 'released', 'fetch', 'fetch'). Not sure if we can/want to do anything about this?

This... is correct, but it's very counter-intuitive.
What's happening is that

As I'm arguing in #6442 (comment) I believe this log should simply be removed

I could reproduce the issue about the assumptions in the test about which keys to be fetched. This is not entirely obvious from the tests and I consider this a bit concerning.

It's also not about any priorities, ordering, etc. but rather that we're using an unordered set for ts.dependencies which then triggers transitions here

This randomness was previously buffered by the delayed _ensure_communicating. I don't feel great about introducing randomness to our scheduling (e.g. there is not even a key-based tie breaker). This entire refactoring effort is supposed to help us make things more deterministic.

I opened #6497 with a suggestion on how to move forward with ensure_communicating. Maybe it's worth putting this PR on ice until #6497 is settled

This has been parked until a new design is agreed upon in #6497.

Unit Test Results

See test report for an extended history of previous test failures. This is useful for diagnosing flaky tests.

       15 files  ±  0         15 suites  ±0   10h 10m 51s ⏱️ +27s
  2 896 tests +  2    2 811 ✔️ +  3    84 💤 ±0  1 ❌  - 1 
21 451 runs  +14  20 486 ✔️ +15  964 💤 ±0  1 ❌  - 1 

For more details on these failures, see this check.

Results for commit 9051170. ± Comparison against base commit 88e1fe0.

♻️ This comment has been updated with latest results.

The PR has been rewritten from scratch and is now ready for review and merge.
It exacerbates the O(n^2*logn) condition described above; this is fixed in #6587.