Evaluate CWL when conditionals before scheduling to avoid wasted batch-system submissions (#3990)

src/toil/cwl/cwltoil.py

@@ -403,6 +403,10 @@ def __init__(
         self.requirements = requirements or []
         self.container_engine = container_engine
 
+    def is_empty(self) -> bool:
+        """Whether this is an empty Conditional."""
+        return self.expression is None
+
     def is_false(self, job: CWLObjectType) -> bool:
         """
         Determine if expression evaluates to False given completed step inputs.
@@ -2890,10 +2894,11 @@ def run(self, file_store: AbstractFileStore) -> CWLObjectType:
 
 class CWLJobWrapper(CWLNamedJob):
     """
-    Wrap a CWL job that uses dynamic resources requirement.
+    Determines how and whether to run the wrapped CWL job.
 
-    When executed, this creates a new child job which has the correct resource
-    requirement set.
+    Wraps a CWL job that uses a dynamic resource requirement or has a
+    conditional. This job is responsible for creating a CWLJob child with
+    the right resource requirements, when the job should not be skipped.
     """
 
     def __init__(
@@ -2921,7 +2926,7 @@ def run(self, file_store: AbstractFileStore) -> Any:
         """Create a child job with the correct resource requirements set."""
         cwljob = resolve_dict_w_promises(self.cwljob, file_store)
 
-        # Check confitional to license full evaluation of job inputs.
+        # Check conditional to license full evaluation of job inputs.
         if self.conditional.is_false(cwljob):
             return self.conditional.skipped_outputs()
 
@@ -2950,11 +2955,9 @@ def __init__(
         cwljob: CWLObjectType,
         runtime_context: cwltool.context.RuntimeContext,
         parent_name: str | None = None,
-        conditional: Conditional | None = None,
     ):
         """Store the context for later execution."""
         self.cwltool = tool
-        self.conditional = conditional or Conditional()
 
         if runtime_context.builder:
             self.builder = runtime_context.builder
@@ -3159,9 +3162,6 @@ def run(self, file_store: AbstractFileStore) -> Any:
         # Deletes duplicate listings
         remove_redundant_mounts(cwljob)
 
-        if self.conditional.is_false(cwljob):
-            return self.conditional.skipped_outputs()
-
         fill_in_defaults(
             self.step_inputs, cwljob, self.runtime_context.make_fs_access("")
         )
@@ -3507,35 +3507,42 @@ def makeJob(
         wfjob.addFollowOn(followOn)
         return wfjob, followOn
     else:
-        # Decied if we have any requirements we care about that are dynamic
+        # Decide if we have any requirements we care about that are dynamic
         REQUIREMENT_TYPES = [
             "ResourceRequirement",
             "http://commonwl.org/cwltool#CUDARequirement",
         ]
+        has_dynamic_resource_requirement = False
         for requirement_type in REQUIREMENT_TYPES:
             req, _ = tool.get_requirement(requirement_type)
             if req:
                 for r in req.values():
                     if isinstance(r, str) and ("$(" in r or "${" in r):
                         # One of the keys in this requirement has a text substitution in it.
                         # TODO: This is not a real lex!
+                        has_dynamic_resource_requirement = True
 
-                        # Found a dynamic resource requirement so use a job wrapper
-                        job_wrapper = CWLJobWrapper(
-                            cast(ToilCommandLineTool, tool),
-                            jobobj,
-                            runtime_context,
-                            parent_name=parent_name,
-                            conditional=conditional,
-                        )
-                        return job_wrapper, job_wrapper
-        # Otherwise, all requirements are known now.
+        if has_dynamic_resource_requirement or (
+            conditional is not None and not conditional.is_empty()
+        ):
+            # Resource requirements and the `when` conditional can depend on
+            # promises from upstream steps that only resolve once the job
+            # runs, so check them in a cheap local wrapper first.
+            job_wrapper = CWLJobWrapper(
+                cast(ToilCommandLineTool, tool),
+                jobobj,
+                runtime_context,
+                parent_name=parent_name,
+                conditional=conditional,
+            )
+            return job_wrapper, job_wrapper
+        # Otherwise, all requirements are known now, and the step is
+        # unconditional, so it can be scheduled directly.
         job = CWLJob(
             tool,
             jobobj,
             runtime_context,
             parent_name=parent_name,
-            conditional=conditional,
         )
         return job, job
 

src/toil/test/cwl/conditional_step_depends_on_step.cwl

@@ -0,0 +1,37 @@
+# `consume`'s `when` references `produce`'s output, so the condition can only
+# be evaluated once that output's promise has resolved.
+# See <https://github.com/DataBiosphere/toil/issues/3990>.
+cwlVersion: v1.2
+class: Workflow
+requirements:
+  InlineJavascriptRequirement: {}
+inputs:
+  number: int
+outputs: []
+steps:
+  produce:
+    in:
+      number: number
+    out: [result]
+    run:
+      cwlVersion: v1.2
+      class: ExpressionTool
+      requirements:
+        InlineJavascriptRequirement: {}
+      inputs:
+        number: int
+      outputs:
+        result: int
+      expression: "$({'result': inputs.number})"
+  consume:
+    in:
+      result: produce/result
+    when: $(inputs.result > 1)
+    run:
+      cwlVersion: v1.2
+      class: CommandLineTool
+      inputs:
+        result: int
+      baseCommand: "true"
+      outputs: []
+    out: []

src/toil/test/cwl/cwlTest.py

@@ -2080,6 +2080,89 @@ def test_pick_value_with_one_null_value(
                     )
 
 
+@needs_cwl
+@pytest.mark.cwl
+@pytest.mark.cwl_small
+def test_when_false_not_scheduled(
+    caplog: pytest.LogCaptureFixture, tmp_path: Path
+) -> None:
+    """
+    A step skipped by its `when` condition must only run the CWLJobWrapper not the CWLJob. See: #3990.
+    """
+    from toil.cwl import cwltoil
+
+    with get_data("test/cwl/conditional_wf.cwl") as cwl_file:
+        with get_data("test/cwl/conditional_wf.yaml") as job_file:
+            with caplog.at_level(logging.DEBUG, logger="toil.leader"):
+                cwltoil.main(
+                    ["--logDebug", f"--outdir={tmp_path}", str(cwl_file), str(job_file), "--disableChaining=True"]
+                )
+                assert any(
+                    "Finished toil run successfully" in record.getMessage()
+                    for record in caplog.records
+                ), "Toil run didn't finish"
+                assert any(
+                        "Issued job 'CWLJobWrapper'" in record.getMessage()
+                        for record in caplog.records
+                ), "'CWLJobWrapper' not issued"
+                assert not any(
+                        "Issued job 'CWLJob'" in record.getMessage()
+                        for record in caplog.records
+                ), "'CWLJob' issued"
+
+
+@needs_cwl
+@pytest.mark.cwl
+@pytest.mark.cwl_small
+def test_when_on_step_output_scheduled(
+    caplog: pytest.LogCaptureFixture, tmp_path: Path
+) -> None:
+    """
+    A step whose `when` references an upstream step's output must still only
+    run the CWLJobWrapper when skipped, and the CWLJob when not. See: #3990.
+    """
+    from toil.cwl import cwltoil
+
+    with get_data("test/cwl/conditional_step_depends_on_step.cwl") as cwl_file:
+        # produce/result (1) is not > 1: consume is skipped and only its
+        # CWLJobWrapper should run.
+        with caplog.at_level(logging.DEBUG, logger="toil.leader"):
+            cwltoil.main(
+                ["--logDebug", f"--outdir={tmp_path}", str(cwl_file), "--number", "1", "--disableChaining=True"]
+            )
+            assert any(
+                "Finished toil run successfully" in record.getMessage()
+                for record in caplog.records
+            ), "Toil run didn't finish"
+            assert any(
+                "Issued job 'CWLJobWrapper'" in record.getMessage()
+                and "consume" in record.getMessage()
+                for record in caplog.records
+            ), "consume's 'CWLJobWrapper' not issued"
+            assert not any(
+                "Issued job 'CWLJob'" in record.getMessage()
+                and "consume" in record.getMessage()
+                for record in caplog.records
+            ), "consume's real 'CWLJob' issued despite being skipped"
+
+        caplog.clear()
+
+        # produce/result (2) is > 1: consume should actually run.
+        with caplog.at_level(logging.DEBUG, logger="toil.leader"):
+            cwltoil.main(
+                ["--logDebug", f"--outdir={tmp_path}", str(cwl_file), "--number", "2", "--disableChaining=True"]
+            )
+            assert any(
+                "Finished toil run successfully" in record.getMessage()
+                for record in caplog.records
+            ), "Toil run didn't finish"
+            assert any(
+                "Issued job 'CWLJob'" in record.getMessage()
+                and "consume" in record.getMessage()
+                for record in caplog.records
+            ), "consume's real 'CWLJob' not issued"
+
+
 @needs_cwl
 @pytest.mark.cwl
 @pytest.mark.cwl_small