From ddf10a005f6ade07020a77775077388999af9b5e Mon Sep 17 00:00:00 2001 From: Anna Giroti Date: Mon, 17 Aug 2026 23:45:32 -0700 Subject: [PATCH 1/2] Evaluate CWL `when` conditionals before scheduling to avoid wasted batch-system submissions Steps with a `when` conditional were always scheduled onto the real batch system as a fully-resourced job, even when the condition was going to evaluate false and the step would be skipped. On a busy HPC system, a skipped step that asks for large resources can sit queued for hours before finding out it had nothing to do. CWLJobWrapper already solves this problem for dynamic resource requirements: it runs as a cheap local job that resolves inputs before spawning the real job. Route `when`-conditional steps through the same wrapper, so the condition is checked with resolved inputs before ever creating the real job, and skipped steps never touch the batch system. --- src/toil/cwl/cwltoil.py | 41 ++++++---- .../cwl/conditional_step_depends_on_step.cwl | 37 +++++++++ src/toil/test/cwl/cwlTest.py | 77 +++++++++++++++++++ 3 files changed, 140 insertions(+), 15 deletions(-) create mode 100644 src/toil/test/cwl/conditional_step_depends_on_step.cwl 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..30e6875eaf 100644 --- a/src/toil/test/cwl/cwlTest.py +++ b/src/toil/test/cwl/cwlTest.py @@ -2080,6 +2080,83 @@ def test_pick_value_with_one_null_value( ) +@needs_cwl +@pytest.mark.cwl +@pytest.mark.cwl_small +def test_skipped_step_runs_locally( + 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_depends_on_step_output( + caplog: pytest.LogCaptureFixture, tmp_path: Path +) -> None: + """ + A step's `when` can reference an upstream step's output, still an + unresolved promise at job-construction time. See: #3990. + """ + from toil.cwl import cwltoil + + with get_data("test/cwl/conditional_step_depends_on_step.cwl") as cwl_file: + with caplog.at_level(logging.DEBUG, logger="toil.leader"): + cwltoil.main( + ["--logDebug", f"--outdir={tmp_path}", str(cwl_file), "--sleep", "10"] + ) + assert any( + # Look for return values instead of a message + "Finished toil run successfully" in record.getMessage() + for record in caplog.records + ) + """ + for record in caplog.records: + if ( + record.name == "toil.leader" + and "Issued job" in record.getMessage() + and "consume" in record.getMessage() + ): + assert record.levelno == logging.DEBUG + + caplog.clear() + with caplog.at_level(logging.DEBUG, logger="toil.leader"): + cwltoil.main( + ["--logDebug", f"--outdir={tmp_path}", str(cwl_file), "--sleep", "2"] + ) + issuances = [ + record + for record in caplog.records + if record.name == "toil.leader" + and "Issued job" in record.getMessage() + and "consume" in record.getMessage() + ] + assert any(record.levelno == logging.INFO for record in issuances)""" + + @needs_cwl @pytest.mark.cwl @pytest.mark.cwl_small From 12ba6ba6385c4686950b2d6db21264f3ef571b79 Mon Sep 17 00:00:00 2001 From: Anna Giroti Date: Thu, 20 Aug 2026 00:51:42 -0700 Subject: [PATCH 2/2] Scope conditional test assertions to the right step, rename for clarity --- src/toil/test/cwl/cwlTest.py | 50 ++++++++++++++++++++---------------- 1 file changed, 28 insertions(+), 22 deletions(-) diff --git a/src/toil/test/cwl/cwlTest.py b/src/toil/test/cwl/cwlTest.py index 30e6875eaf..9d38b9bdd4 100644 --- a/src/toil/test/cwl/cwlTest.py +++ b/src/toil/test/cwl/cwlTest.py @@ -2083,7 +2083,7 @@ def test_pick_value_with_one_null_value( @needs_cwl @pytest.mark.cwl @pytest.mark.cwl_small -def test_skipped_step_runs_locally( +def test_when_false_not_scheduled( caplog: pytest.LogCaptureFixture, tmp_path: Path ) -> None: """ @@ -2114,47 +2114,53 @@ def test_skipped_step_runs_locally( @needs_cwl @pytest.mark.cwl @pytest.mark.cwl_small -def test_when_depends_on_step_output( +def test_when_on_step_output_scheduled( caplog: pytest.LogCaptureFixture, tmp_path: Path ) -> None: """ - A step's `when` can reference an upstream step's output, still an - unresolved promise at job-construction time. See: #3990. + 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", "10"] + ["--logDebug", f"--outdir={tmp_path}", str(cwl_file), "--sleep", "1", "--disableChaining=True"] ) assert any( - # Look for return values instead of a message "Finished toil run successfully" in record.getMessage() for record in caplog.records - ) - """ - for record in caplog.records: - if ( - record.name == "toil.leader" - and "Issued job" in record.getMessage() - and "consume" in record.getMessage() - ): - assert record.levelno == logging.DEBUG + ), "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"] + ["--logDebug", f"--outdir={tmp_path}", str(cwl_file), "--sleep", "2", "--disableChaining=True"] ) - issuances = [ - record + assert any( + "Finished toil run successfully" in record.getMessage() for record in caplog.records - if record.name == "toil.leader" - and "Issued job" in record.getMessage() + ), "Toil run didn't finish" + assert any( + "Issued job 'CWLJob'" in record.getMessage() and "consume" in record.getMessage() - ] - assert any(record.levelno == logging.INFO for record in issuances)""" + for record in caplog.records + ), "consume's real 'CWLJob' not issued" @needs_cwl