Reinstate: AMM ReduceReplicas to iterate only on replicated tasks

[crusaderky]

Reinstate #5297 after it was reverted in #5335.
DO NOT MERGE - this PR has been flagged to cause a scheduler deadlock; further investigation is needed.

XREFS

• AMM ReduceReplicas to iterate only on replicated tasks #5297
• Run multiple AMMs in parallel #5315
• Revert AMM ReduceReplicas and parallel AMMs updates #5335
• Reinstate: Run multiple AMMs in parallel #5339
• Reinstate: Enhance AMM docstrings #5340
• Reinstate: AMM ReduceReplicas to iterate only on replicated tasks #5341

[crusaderky]

Cherry picked candidate fix from #5337

[crusaderky]

I tried adding

            parent._transitions({key: "released"}, client_msgs, worker_msgs)
+           assert not client_msgs and not worker_msgs
            self.send_all(client_msgs, worker_msgs)

but it doesn't trip any unit tests.
By reading the code I understand that the dicts should be filled by transition_processing_released, specifically at https://github.com/crusaderky/distributed/blob/3c2bcfd66a49bb7e092f9840c2438d9343bbdd94/distributed/scheduler.py#L3006-L3010

transition_erred_released and transition_memory_released also write to the msgs dicts, but the two states are not in worker.PROCESSING (since the message originates from here: https://github.com/crusaderky/distributed/blob/3c2bcfd66a49bb7e092f9840c2438d9343bbdd94/distributed/worker.py#L2748-L2750) so they would only happen in case of grave misalignment between worker and scheduler (I can't think of a legit use case where the worker thinks the task is still processing while the scheduler thinks it's already erred or in memory?)

@fjetter I'm out of ideas... help is appreciated

[crusaderky]

@fjetter correction: the below does not trip any unit tests at all!

+           assert False
            parent._transitions({key: "released"}, client_msgs, worker_msgs)
            self.send_all(client_msgs, worker_msgs)

It's a whole code path that is only triggered in production and this makes me extremely uncomfortable. However, it's been tested to work by our reporter and I think at this point it may be out of scope for this PR? What is your opinion about merging this as it is?

[fjetter]

the below does not trip any unit tests at all!

exceptions in our transition state machine are not properly propagated. you might actually hit it but it is logged and lost. if you are "lucky" a unit test would deadlock but the unit test will not fail due to the exception. I'll take another look and see if I can provide any guidance.

What is your opinion about merging this as it is?

The problem is that we're removing any call to this handler in #5046 and we have also reports for deadlocking with that PR. However this is triggered, if we go on with the worker state refactoring, we'd introduce the regression without noticing

[crusaderky]

Split out #5359. Note that, as of this PR, stimulus_missing_data (which is dead code) still invokes the old, non-encapsulated changes to who_has/has_what.

[crusaderky]

All test failures unrelated. The code involved in the deadlock is no longer in this PR. Merging tomorrow unless anybody objects.