Different stimulus_id's success/failure in Worker.execute

dask · Mar 31, 2022 · 7bf0117 · 7bf0117
1 parent 6d6a0ee
commit 7bf0117
Showing 1 changed file with 4 additions and 3 deletions.
diff --git a/distributed/worker.py b/distributed/worker.py
@@ -3507,7 +3507,8 @@ async def execute(self, key: str, *, stimulus_id: str) -> None:
             assert ts, self.story(key)
             ts.done = True
             result["key"] = ts.key
-            result["stimulus_id"] = stimulus_id = f"{result['op']}-{time()}"
+            result["stimulus_id"] = new_stimulus_id = f"{result['op']}-{time()}"
+            del stimulus_id
             value = result.pop("result", None)
             ts.startstops.append(
                 {"action": "compute", "start": result["start"], "stop": result["stop"]}
@@ -3544,7 +3545,7 @@ async def execute(self, key: str, *, stimulus_id: str) -> None:
                     result["traceback_text"],
                 )
 
-            self.transitions(recommendations, stimulus_id=stimulus_id)
+            self.transitions(recommendations, stimulus_id=new_stimulus_id)
 
             logger.debug("Send compute response to scheduler: %s, %s", ts.key, result)
 
@@ -3563,7 +3564,7 @@ async def execute(self, key: str, *, stimulus_id: str) -> None:
                 ts,
                 "error",
                 **emsg,
-                stimulus_id=stimulus_id,
+                stimulus_id=f"execute-erred-{time}",
             )
         finally:
             self.ensure_computing()