diff --git a/src/toil/cwl/cwltoil.py b/src/toil/cwl/cwltoil.py index 112dc06c57..60d45e1390 100644 --- a/src/toil/cwl/cwltoil.py +++ b/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,11 +3510,12 @@ 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: @@ -3519,17 +3523,24 @@ def makeJob( 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, diff --git a/src/toil/test/cwl/conditional_step_depends_on_step.cwl b/src/toil/test/cwl/conditional_step_depends_on_step.cwl new file mode 100644 index 0000000000..04b3cd2519 --- /dev/null +++ b/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 . +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: [] diff --git a/src/toil/test/cwl/cwlTest.py b/src/toil/test/cwl/cwlTest.py index a61a22284a..9d38b9bdd4 100644 --- a/src/toil/test/cwl/cwlTest.py +++ b/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