Move invocation of output callback for workflows (#396)

* Move invocation of output callback for workflows from the job() method to the
receive_output() method.

This closes a gap whereby sometimes all steps of a subworkflow have completed,
but it has not yet called its output callback.  This can lead to the parent
workflow incorrectly believing no progress has been made and returning
'None' (which indicates a workflow deadlock if there are no jobs currently
executing) instead of re-evaluating pending steps to see if they can be run.

* Restore call to output on workflow completion, for cases where workflow exits early and returns partial results.

Author: tetron

cwltool/workflow.py

@@ -245,6 +245,9 @@ def job(self, joborder, output_callback, **kwargs):
         # type: (Dict[Text, Text], functools.partial[None], **Any) -> Generator
         kwargs["part_of"] = self.name
         kwargs["name"] = shortname(self.id)
+
+        _logger.info(u"[%s] start", self.name)
+
         for j in self.step.job(joborder, output_callback, **kwargs):
             yield j
 
@@ -257,6 +260,8 @@ def __init__(self, workflow, **kwargs):
         self.steps = [WorkflowJobStep(s) for s in workflow.steps]
         self.state = None  # type: Dict[Text, WorkflowStateItem]
         self.processStatus = None  # type: Text
+        self.did_callback = False
+
         if "outdir" in kwargs:
             self.outdir = kwargs["outdir"]
         elif "tmp_outdir_prefix" in kwargs:
@@ -270,8 +275,27 @@ def __init__(self, workflow, **kwargs):
         _logger.debug(u"[%s] initialized from %s", self.name,
                       self.tool.get("id", "workflow embedded in %s" % kwargs.get("part_of")))
 
-    def receive_output(self, step, outputparms, jobout, processStatus):
-        # type: (WorkflowJobStep, List[Dict[Text,Text]], Dict[Text,Text], Text) -> None
+    def do_output_callback(self, final_output_callback):
+        # type: (Callable[[Any, Any], Any]) -> None
+
+        supportsMultipleInput = bool(self.workflow.get_requirement("MultipleInputFeatureRequirement")[0])
+
+        try:
+            wo = object_from_state(self.state, self.tool["outputs"], True, supportsMultipleInput, "outputSource",
+                                   incomplete=True)
+        except WorkflowException as e:
+            _logger.error(u"[%s] Cannot collect workflow output: %s", self.name, e)
+            wo = {}
+            self.processStatus = "permanentFail"
+
+        _logger.info(u"[%s] completed %s", self.name, self.processStatus)
+
+        self.did_callback = True
+
+        final_output_callback(wo, self.processStatus)
+
+    def receive_output(self, step, outputparms, final_output_callback, jobout, processStatus):
+        # type: (WorkflowJobStep, List[Dict[Text,Text]], Callable[[Any, Any], Any], Dict[Text,Text], Text) -> None
 
         for i in outputparms:
             if "id" in i:
@@ -295,8 +319,12 @@ def receive_output(self, step, outputparms, jobout, processStatus):
         step.completed = True
         self.made_progress = True
 
-    def try_make_job(self, step, **kwargs):
-        # type: (WorkflowJobStep, **Any) -> Generator
+        completed = sum(1 for s in self.steps if s.completed)
+        if completed == len(self.steps):
+            self.do_output_callback(final_output_callback)
+
+    def try_make_job(self, step, final_output_callback, **kwargs):
+        # type: (WorkflowJobStep, Callable[[Any, Any], Any], **Any) -> Generator
         inputparms = step.tool["inputs"]
         outputparms = step.tool["outputs"]
 
@@ -315,7 +343,7 @@ def try_make_job(self, step, **kwargs):
 
             _logger.debug(u"[%s] starting %s", self.name, step.name)
 
-            callback = functools.partial(self.receive_output, step, outputparms)
+            callback = functools.partial(self.receive_output, step, outputparms, final_output_callback)
 
             valueFrom = {
                 i["id"]: i["valueFrom"] for i in step.tool["inputs"]
@@ -394,7 +422,7 @@ def valueFromFunc(k, v):  # type: (Any, Any) -> Any
             step.completed = True
 
     def run(self, **kwargs):
-        _logger.debug(u"[%s] workflow starting", self.name)
+        _logger.info(u"[%s] start", self.name)
 
     def job(self, joborder, output_callback, **kwargs):
         # type: (Dict[Text, Any], Callable[[Any, Any], Any], **Any) -> Generator
@@ -429,7 +457,7 @@ def job(self, joborder, output_callback, **kwargs):
 
                 if not step.submitted:
                     try:
-                        step.iterable = self.try_make_job(step, **kwargs)
+                        step.iterable = self.try_make_job(step, output_callback, **kwargs)
                     except WorkflowException as e:
                         _logger.error(u"[%s] Cannot make job: %s", step.name, e)
                         _logger.debug("", exc_info=True)
@@ -458,19 +486,8 @@ def job(self, joborder, output_callback, **kwargs):
                 else:
                     yield None
 
-        supportsMultipleInput = bool(self.workflow.get_requirement("MultipleInputFeatureRequirement")[0])
-
-        try:
-            wo = object_from_state(self.state, self.tool["outputs"], True, supportsMultipleInput, "outputSource",
-                                   incomplete=True)
-        except WorkflowException as e:
-            _logger.error(u"[%s] Cannot collect workflow output: %s", self.name, e)
-            wo = {}
-            self.processStatus = "permanentFail"
-
-        _logger.info(u"[%s] outdir is %s", self.name, self.outdir)
-
-        output_callback(wo, self.processStatus)
+        if not self.did_callback:
+            self.do_output_callback(output_callback)
 
 
 class Workflow(Process):