fix: modify self after all processes have finished to avoid competition

Signed-off-by: zjgemi <[email protected]>

src/dflow/step.py

@@ -1585,14 +1585,15 @@ def handle_expr(val, scope):
                         time.sleep(config["debug_batch_interval"])
 
                 failed = []
+                results = {}
                 for future in concurrent.futures.as_completed(futures):
                     j = futures.index(future)
                     try:
                         phase, pars, arts = future.result()
                     except Exception:
                         import traceback
                         traceback.print_exc()
-                        self.parallel_steps[j].phase = "Failed"
+                        results[j] = ("Failed", pars, arts)
                         if not self.continue_on_failed:
                             self.phase = "Failed"
                             if config["debug_failfast"]:
@@ -1601,14 +1602,21 @@ def handle_expr(val, scope):
                             else:
                                 failed.append(self.parallel_steps[j])
                     else:
-                        for name, value in pars.items():
-                            self.parallel_steps[j].outputs.parameters[
-                                name].value = value
-                        for name, path in arts.items():
-                            self.parallel_steps[j].outputs.artifacts[
-                                name].local_path = path
+                        results[j] = (phase, pars, arts)
                         logging.info("Outputs of %s collected" %
                                      self.parallel_steps[j])
+
+                # modify self after all processes have finished to avoid
+                # competition
+                for j in results:
+                    phase, pars, arts = results[j]
+                    self.parallel_steps[j].phase = phase
+                    for name, value in pars.items():
+                        self.parallel_steps[j].outputs.parameters[
+                            name].value = value
+                    for name, path in arts.items():
+                        self.parallel_steps[j].outputs.artifacts[
+                            name].local_path = path
                 if len(failed) > 0:
                     raise RuntimeError("Step %s failed" % failed)
 

src/dflow/steps.py

@@ -268,24 +268,30 @@ def run(self, workflow_id=None, context=None):
                                 "batch" % config["debug_batch_interval"])
                             time.sleep(config["debug_batch_interval"])
 
+                    results = {}
                     for future in concurrent.futures.as_completed(futures):
                         j = futures.index(future)
                         try:
                             phase, pars, arts = future.result()
                         except Exception:
                             import traceback
                             traceback.print_exc()
-                            step[j].phase = "Failed"
+                            results[j] = ("Failed", pars, arts)
                             if not step[j].continue_on_failed:
                                 raise RuntimeError("Step %s failed" % step[j])
                         else:
-                            for name, value in pars.items():
-                                step[j].outputs.parameters[
-                                    name].value = value
-                            for name, path in arts.items():
-                                step[j].outputs.artifacts[
-                                    name].local_path = path
+                            results[j] = (phase, pars, arts)
                             logging.info("Outputs of %s collected" % step[j])
+
+                    # modify self after all processes have finished to avoid
+                    # competition
+                    for j in results:
+                        phase, pars, arts = results[j]
+                        step[j].phase = phase
+                        for name, value in pars.items():
+                            step[j].outputs.parameters[name].value = value
+                        for name, path in arts.items():
+                            step[j].outputs.artifacts[name].local_path = path
             else:
                 step.run(self, context)