diff --git a/src/toil/cwl/cwltoil.py b/src/toil/cwl/cwltoil.py index 112dc06c57..50f35e2634 100644 --- a/src/toil/cwl/cwltoil.py +++ b/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,11 +3507,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,23 +3520,29 @@ 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 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 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..859e44dbf8 --- /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: + 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: [] diff --git a/src/toil/test/cwl/cwlTest.py b/src/toil/test/cwl/cwlTest.py index a61a22284a..6d1f8e93d0 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), "--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