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

src/toil/cwl/cwltoil.py

@@ -2890,10 +2890,13 @@ def run(self, file_store: AbstractFileStore) -> CWLObjectType:
 
 class CWLJobWrapper(CWLNamedJob):
     """
-    Wrap a CWL job that uses dynamic resources requirement.
+    Wrap a CWL job that uses a dynamic resources requirement, or that may be
+    skipped by a `when` conditional that can't be safely evaluated until the
+    step's inputs are resolved.
 
-    When executed, this creates a new child job which has the correct resource
-    requirement set.
+    When executed, this runs on the leader with minimal resources, resolves
+    the job's inputs, and then either reports the step as skipped or creates
+    a new child job which has the correct resource requirement set.
     """
 
     def __init__(
@@ -2921,7 +2924,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()
 
@@ -3507,29 +3510,37 @@ 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 conditional.expression is not None
+        ):
+            # 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,

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:
+  sleep: int
+outputs: []
+steps:
+  produce:
+    in:
+      sleep: sleep
+    out: [result]
+    run:
+      cwlVersion: v1.2
+      class: ExpressionTool
+      requirements:
+        InlineJavascriptRequirement: {}
+      inputs:
+        sleep: int
+      outputs:
+        result: int
+      expression: "$({'result': inputs.sleep})"
+  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), "--sleep", "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), "--sleep", "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