Refactor busy_worker reinsertion

distributed/worker.py

@@ -126,6 +126,8 @@
     ReleaseWorkerDataMsg,
     RescheduleEvent,
     RescheduleMsg,
+    RetryBusyWorkerEvent,
+    RetryBusyWorkerLater,
     SendMessageToScheduler,
     SerializedTask,
     StateMachineEvent,
@@ -2847,6 +2849,7 @@ def transitions(self, recommendations: Recs, *, stimulus_id: str) -> None:
         else:
             self._handle_instructions(instructions)
 
+    @fail_hard
     @log_errors
     def handle_stimulus(self, stim: StateMachineEvent) -> None:
         self.stimulus_log.append(stim.to_loggable(handled=time()))
@@ -2911,6 +2914,12 @@ def _handle_instructions(self, instructions: Instructions) -> None:
                         name=f"execute({inst.key})",
                     )
 
+                elif isinstance(inst, RetryBusyWorkerLater):
+                    task = asyncio.create_task(
+                        self.retry_busy_worker_later(inst.worker),
+                        name=f"retry_busy_worker_later({inst.worker})",
+                    )
+
                 else:
                     raise TypeError(inst)  # pragma: nocover
 
@@ -3346,7 +3355,9 @@ def done_event():
                 # Avoid hammering the worker. If there are multiple replicas
                 # available, immediately try fetching from a different worker.
                 self.busy_workers.add(worker)
-                self.io_loop.call_later(0.15, self._readd_busy_worker, worker)
+                instructions.append(
+                    RetryBusyWorkerLater(worker=worker, stimulus_id=stimulus_id)
+                )
 
             refresh_who_has = set()
 
@@ -3383,11 +3394,10 @@ def done_event():
                 )
                 self._update_who_has(who_has)
 
-    @log_errors
-    def _readd_busy_worker(self, worker: str) -> None:
-        self.busy_workers.remove(worker)
-        self.handle_stimulus(
-            GatherDepDoneEvent(stimulus_id=f"readd-busy-worker-{time()}")
+    async def retry_busy_worker_later(self, worker: str) -> StateMachineEvent | None:
+        await asyncio.sleep(0.15)
+        return RetryBusyWorkerEvent(
+            worker=worker, stimulus_id=f"retry-busy-worker-{time()}"
         )
 
     @log_errors
@@ -3895,6 +3905,11 @@ def _(self, ev: GatherDepDoneEvent) -> RecsInstrs:
         """Temporary hack - to be removed"""
         return self._ensure_communicating(stimulus_id=ev.stimulus_id)
 
+    @handle_event.register
+    def _(self, ev: RetryBusyWorkerEvent) -> RecsInstrs:
+        self.busy_workers.discard(ev.worker)
+        return self._ensure_communicating(stimulus_id=ev.stimulus_id)
+
     @handle_event.register
     def _(self, ev: CancelComputeEvent) -> RecsInstrs:
         """Scheduler requested to cancel a task"""

distributed/worker_state_machine.py

@@ -280,6 +280,12 @@ class Execute(Instruction):
     key: str
 
 
+@dataclass
+class RetryBusyWorkerLater(Instruction):
+    __slots__ = ("worker",)
+    worker: str
+
+
 @dataclass
 class EnsureCommunicatingAfterTransitions(Instruction):
     __slots__ = ()
@@ -444,6 +450,12 @@ class UnpauseEvent(StateMachineEvent):
     __slots__ = ()
 
 
+@dataclass
+class RetryBusyWorkerEvent(StateMachineEvent):
+    __slots__ = ("worker",)
+    worker: str
+
+
 @dataclass
 class GatherDepDoneEvent(StateMachineEvent):
     """Temporary hack - to be removed"""