Migrate ensure_communicating transitions to new WorkerState event mechanism

[fjetter]

The ensure_communicating transitions should be migrated to the WorkerState event mechanism as outlined in #5736 (comment)

• The Worker._gather_dep method is broken up into
  • A method Worker._gather_remote_data which accepts an instruction indicating what keys to fetch, the remote worker address, etc. It performs the actual communication and returns a list of events about the outcome of the remote fetch (e.g. Success, DataMissing, RemoteBusy, Error, ...)
  • Handlers, e.g. Worker(State)._handle_gather_remote_busy that deal with the result of the fetch attempt
• The ensure_communicating method is removed from the list of handle_comm every_cycle callbacks
• The logic of ensure_communicating is moved to a private method (perspectively a method of the new class WorkerState)
• The new private method will not perform any transitions but rather return a list of recommendations and/or instructions as necessary
• This private method is then called as part of the transition system, e.g.
  • transition_flight_fetch
  • unpause
• The implicit goal is to remove all invocations of self.loop.add_callback(self.gather_dep, ...) and replace this with the new callback method

Subtasks

• update_who_has can remove workers #6342
• Refactor find_missing and refresh_who_has #6348
• Refactor busy_worker reinsertion #6379
• Cosmetic review of story() #6442
• Refactor gather_dep #6388

Blocked by

• StateMachine event dispatch mechanism #5894

Related work

• Encapsulate spill buffer and memory_monitor #5891
• Migrate ensure_computing transitions to new WorkerState event mechanism #5895

[gjoseph92]

Is cancellation of gather_deps in scope for this? How does that fit in? Just curious.

[crusaderky]

Is cancellation of gather_deps in scope for this? How does that fit in? Just curious.

It's out of scope.
gather_dep will become a task that is explicitly tracked by the worker. It should be straightforward to cancel the task when that happens. Of course many other things need to happen (sender-side cancellation, ignore spurious incoming data, etc. etc.)