passing rerun to all tasks

Author: djarecka

pydra/engine/core.py

@@ -811,7 +811,7 @@ async def _run(self, submitter=None, rerun=False, **kwargs):
             self.hooks.pre_run_task(self)
             try:
                 self.audit.monitor()
-                await self._run_task(submitter)
+                await self._run_task(submitter, rerun=rerun)
                 result.output = self._collect_outputs()
             except Exception as e:
                 record_error(self.output_dir, e)
@@ -825,11 +825,11 @@ async def _run(self, submitter=None, rerun=False, **kwargs):
         self.hooks.post_run(self, result)
         return result
 
-    async def _run_task(self, submitter):
+    async def _run_task(self, submitter, rerun=False):
         if not submitter:
             raise Exception("Submitter should already be set.")
         # at this point Workflow is stateless so this should be fine
-        await submitter._run_workflow(self)
+        await submitter._run_workflow(self, rerun=rerun)
 
     def set_output(self, connections):
         """

pydra/engine/tests/test_workflow.py

@@ -2358,7 +2358,8 @@ def test_wf_nostate_cachelocations_updated(plugin, tmpdir):
 def test_wf_nostate_cachelocations_recompute(plugin, tmpdir):
     """
     Two wfs with the same inputs but slightly different graph;
-    the second wf should recompute the results
+    the second wf should recompute the results,
+    but the second node should use the results from the first wf (has the same input)
     """
     cache_dir1 = tmpdir.mkdir("test_wf_cache3")
     cache_dir2 = tmpdir.mkdir("test_wf_cache4")
@@ -2401,14 +2402,14 @@ def test_wf_nostate_cachelocations_recompute(plugin, tmpdir):
     results2 = wf2.result()
     assert 8 == results2.output.out
 
-    # checking execution time
-    assert t1 > 3
-    assert t2 > 3
-
     # checking if both dir exists
     assert wf1.output_dir.exists()
     assert wf2.output_dir.exists()
 
+    # checking execution time (second task shouldn't be recompute, t2 should be small)
+    assert t1 > 3
+    assert t2 < 0.5
+
 
 @pytest.mark.parametrize("plugin", Plugins)
 def test_wf_ndstate_cachelocations(plugin, tmpdir):