Remove EnsureCommunicatingAfterTransitions

[crusaderky]

Partially closes #6497

#6165 introduced this hack, for the sake of being functionally identical to the previous code.
This is a cleaner redesign which is conceptually the same.

[crusaderky]

[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.

[github-actions]

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.

[crusaderky]

[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?

[fjetter]

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

[fjetter]

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

distributed/distributed/worker.py

Lines 2125 to 2129 in 6d85a85

	for dep_ts in ts.dependencies:
	if dep_ts.state != "memory":
	ts.waiting_for_data.add(dep_ts)
	dep_ts.waiters.add(ts)
	recommendations[dep_ts] = "fetch"

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.

[fjetter]

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

[crusaderky]

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

[github-actions]

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.

[crusaderky]

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.

[fjetter]

This is great!

[fjetter]

I love that this is not everywhere anymore ❤️