reverting some of the changes; the state from intermediate tasks are not passed if final splitter is empty

Author: djarecka

pydra/engine/core.py

@@ -222,7 +222,7 @@ def checksum(self):
             self.inputs._graph_checksums = [nd.checksum for nd in self.graph_sorted]
 
         input_hash = self.inputs.hash
-        if self.state is None or self.state.splitter_rpn == []:
+        if self.state is None:
             self._checksum = create_checksum(self.__class__.__name__, input_hash)
         else:
             # including splitter in the hash
@@ -329,7 +329,7 @@ def cache_locations(self, locations):
     @property
     def output_dir(self):
         """Get the filesystem path where outputs will be written."""
-        if self.state and self.state.splitter_rpn:
+        if self.state:
             return [self._cache_dir / checksum for checksum in self.checksum_states()]
         return self._cache_dir / self.checksum
 
@@ -342,7 +342,7 @@ def __call__(self, submitter=None, plugin=None, rerun=False, **kwargs):
         plugin = plugin or self.plugin
         if plugin:
             submitter = Submitter(plugin=plugin)
-        elif self.state and self.state.splitter_rpn:
+        elif self.state:
             submitter = Submitter()
 
         if submitter:
@@ -512,7 +512,7 @@ def done(self):
         # if any of the field is lazy, there is no need to check results
         if is_lazy(self.inputs):
             return False
-        if self.state and self.state.splitter_rpn:
+        if self.state:
             # TODO: only check for needed state result
             if self.result() and all(self.result()):
                 return True
@@ -556,7 +556,7 @@ def result(self, state_index=None):
         """
         # TODO: check if result is available in load_result and
         # return a future if not
-        if self.state and self.state.splitter_rpn:
+        if self.state:
             if state_index is None:
                 # if state_index=None, collecting all results
                 if self.state.combiner:
@@ -753,7 +753,10 @@ def create_connections(self, task):
                     self.graph.add_edges((getattr(self, val.name), task))
                     logger.debug("Connecting %s to %s", val.name, task.name)
 
-                    if getattr(self, val.name).state:
+                    if (
+                        getattr(self, val.name).state
+                        and getattr(self, val.name).state.splitter_rpn_final
+                    ):
                         # adding a state from the previous task to other_states
                         other_states[val.name] = (
                             getattr(self, val.name).state,

pydra/engine/submitter.py

@@ -50,9 +50,7 @@ def __call__(self, runnable, cache_locations=None, rerun=False):
             runnable.inputs._graph_checksums = [
                 nd.checksum for nd in runnable.graph_sorted
             ]
-        if is_workflow(runnable) and (
-            runnable.state is None or runnable.state.splitter_rpn == []
-        ):
+        if is_workflow(runnable) and runnable.state is None:
             self.loop.run_until_complete(self.submit_workflow(runnable, rerun=rerun))
         else:
             self.loop.run_until_complete(self.submit(runnable, wait=True, rerun=rerun))
@@ -93,7 +91,7 @@ async def submit(self, runnable, wait=False, rerun=False):
 
         """
         futures = set()
-        if runnable.state and runnable.state.splitter_rpn:
+        if runnable.state:
             runnable.state.prepare_states(runnable.inputs)
             runnable.state.prepare_inputs()
             logger.debug(
@@ -156,9 +154,7 @@ async def _run_workflow(self, wf, rerun=False):
                 task.inputs.retrieve_values(wf)
                 # checksum has to be updated, so resetting
                 task._checksum = None
-                if is_workflow(task) and (
-                    not task.state or task.state.splitter_rpn == []
-                ):
+                if is_workflow(task) and not task.state:
                     await self.submit_workflow(task, rerun=rerun)
                 else:
                     for fut in await self.submit(task, rerun=rerun):