diff --git a/modules/nextflow/src/main/groovy/nextflow/processor/TaskConfig.groovy b/modules/nextflow/src/main/groovy/nextflow/processor/TaskConfig.groovy index 764133c5f7..925ffe0cae 100644 --- a/modules/nextflow/src/main/groovy/nextflow/processor/TaskConfig.groovy +++ b/modules/nextflow/src/main/groovy/nextflow/processor/TaskConfig.groovy @@ -48,6 +48,11 @@ class TaskConfig extends LazyMap implements Cloneable { private transient Map cache = new LinkedHashMap(20) + /** The directive names accessed while {@link #trackingAccess} is enabled */ + private transient Set accessedDirectives = new HashSet<>(10) + + private transient boolean trackingAccess + TaskConfig() { } TaskConfig( Map entries ) { @@ -58,13 +63,46 @@ class TaskConfig extends LazyMap implements Cloneable { def copy = (TaskConfig)super.clone() copy.setTarget(new HashMap<>(this.getTarget())) copy.newCache() + // copied, not reset: the copy carries over the command rendered from them + // -- see TaskRun#clone -- therefore it depends on the same directives + copy.accessedDirectives = new HashSet<>(this.accessedDirectives) return copy } + /** + * Discard the resolved directive values. Note it does *not* touch the accessed names: the + * value cache belongs to a context, the access log to a rendered command. + */ private void newCache() { cache = [:] } + /** + * Track the directives accessed while the given action runs. + * + * The task command is rendered by accessing the directives it interpolates off this object, + * therefore tracking the accesses while it happens tells which directives the rendered + * command depends on. It is scoped to that action because the directives are accessed all + * the time by the rest of the engine e.g. the executor asking for the memory to request. + * + * The caller must disable it once the command is rendered, including on failure, since + * a flag left enabled would report every later access as a dependency of the command. + * + * @see nextflow.processor.TaskRun#resolve + * @param value Whether the directive accesses must be tracked + */ + void trackDirectiveAccess(boolean value) { + trackingAccess = value + } + + /** + * @param directive The directive name e.g. {@code memory} + * @return {@code true} when the given directive was accessed while the accesses were tracked + */ + boolean isDirectiveAccessed(String directive) { + return accessedDirectives.contains(directive) + } + /** * Assign the context map for dynamic evaluation of task config properties * @param context A {@link TaskContext} object that holds the task evaluation context @@ -137,6 +175,14 @@ class TaskConfig extends LazyMap implements Cloneable { } def get( String key ) { + // note this is the funnel for a directive *property* access, either directly or via the + // matching getter e.g. #getMemory. Only #eval (used by the task hasher) and #getRawValue + // bypass it -- a directive read through those while tracking would go unnoticed + if( trackingAccess ) + accessedDirectives.add(key) + + // note the access is tracked before the cache is consulted, so a directive already + // resolved outside the tracked window is still reported if( cache.containsKey(key) ) return cache.get(key) @@ -152,6 +198,9 @@ class TaskConfig extends LazyMap implements Cloneable { else result = super.get(key) + // note a dynamic top-level directive is cached by its resolved value, so a directive its + // closure accesses in turn is only seen on the first resolution -- resolving one before + // the command is rendered would hide it. `ext` is unaffected: what is cached is its map cache.put(key,result) return result } diff --git a/modules/nextflow/src/main/groovy/nextflow/processor/TaskRun.groovy b/modules/nextflow/src/main/groovy/nextflow/processor/TaskRun.groovy index 8fe7c0b24c..ce65a752e1 100644 --- a/modules/nextflow/src/main/groovy/nextflow/processor/TaskRun.groovy +++ b/modules/nextflow/src/main/groovy/nextflow/processor/TaskRun.groovy @@ -853,9 +853,48 @@ class TaskRun implements Cloneable { * @param body A {@code BodyDef} object instance */ void resolve(BodyDef body) { - processor.session.stubRun && config.getStubBlock() - ? resolveStub(config.getStubBlock()) - : resolveBody(body) + // track the directives accessed while the command is rendered -- see #isDirectiveReferenced + // note the null guards are load-bearing -- the access below is behind a short-circuit + config?.trackDirectiveAccess(true) + try { + processor.session.stubRun && config.getStubBlock() + ? resolveStub(config.getStubBlock()) + : resolveBody(body) + } + finally { + config?.trackDirectiveAccess(false) + } + } + + /** + * Report whether the rendered task command depends on the value of the given + * {@code task} directive e.g. {@code memory} for a script interpolating + * {@code "-Xmx${task.memory.toGiga()}g"}. + * + * The command is rendered *before* the task is scheduled, therefore an executor that + * adjusts the requested resources at schedule time needs to know whether the command + * carries a value it is about to change. + * + * The reference is *observed*, not inferred: rendering the command accesses the directive + * off the task config, and {@link #resolve} tracks the accesses while it happens. That + * covers every path the command can be rendered through -- the script, a {@code shell} + * block, a {@code template} file, and a dynamic directive value the command interpolates, + * whether declared in the process or in the config file -- without any of them being + * known here. + * + * Note it reports the *last* rendering of this task, hence {@code false} until + * {@link #resolve} has run, for an {@code exec} task, and for a task array -- which + * {@code TaskArrayCollector} assembles without resolving it. + * + * ponytail: a directive resolved *after* the command has been rendered is not observed + * e.g. `beforeScript = { "-Xmx${task.memory}" }`, which the wrapper builder resolves at + * submit time. Widen the tracked action to cover the wrapper if that case shows up. + * + * @param directive The directive name e.g. {@code memory} + * @return {@code true} when rendering the command accessed the given directive + */ + boolean isDirectiveReferenced(String directive) { + return config != null && config.isDirectiveAccessed(directive) } protected void resolveBody(BodyDef body) { diff --git a/modules/nextflow/src/test/groovy/nextflow/processor/TaskDirectiveReadsTest.groovy b/modules/nextflow/src/test/groovy/nextflow/processor/TaskDirectiveReadsTest.groovy new file mode 100644 index 0000000000..f3118d81ab --- /dev/null +++ b/modules/nextflow/src/test/groovy/nextflow/processor/TaskDirectiveReadsTest.groovy @@ -0,0 +1,209 @@ +/* + * Copyright 2013-2026, Seqera Labs + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package nextflow.processor + +import java.nio.file.Files + +import ch.artecat.grengine.Grengine +import nextflow.Session +import nextflow.config.parser.v1.ConfigParserV1 +import nextflow.config.parser.v2.ConfigParserV2 +import nextflow.script.BodyDef +import nextflow.script.ProcessConfig +import nextflow.script.dsl.ProcessConfigBuilder +import spock.lang.Specification +import spock.lang.Unroll + +/** + * Verify the directives read while the task command is rendered are reported by + * {@link TaskRun#isDirectiveReferenced}. + * + * @author Ben Sherman + */ +class TaskDirectiveReadsTest extends Specification { + + private TaskRun taskWith(Map directives, BodyDef body) { + return taskWith(new TaskConfig(directives), body) + } + + private TaskRun taskWith(TaskConfig config, BodyDef body) { + final task = unresolvedTaskWith(config) + task.resolve(body) + return task + } + + private TaskRun unresolvedTaskWith(TaskConfig config) { + final processor = new TaskProcessor() + processor.@session = new Session() + processor.@grengine = new Grengine() + + final task = new TaskRun(processor: processor, config: config) + task.context = new TaskContext(holder: [:]) + task.config.setContext(task.context) + return task + } + + /** + * Build the task config as the engine does, by applying a parsed config file to the process + * config -- so the directive values are the closures the parser produced. + */ + private TaskConfig taskConfigFrom(parser, String text, String processName) { + final config = parser.parse(text) + final processConfig = new ProcessConfig([:]) + new ProcessConfigBuilder(processConfig) + .applyConfig(config.process as Map, processName, processName, processName) + return processConfig.createTaskConfig() + } + + def 'should report a directive read by the task script' () { + when: + def task = taskWith( + [memory: '8 GB', cpus: 4], + new BodyDef({-> "java -Xmx${task.memory.toGiga()}g -jar app.jar"}, 'java ...', 'script') ) + + then: + task.script == 'java -Xmx8g -jar app.jar' + and: + task.isDirectiveReferenced('memory') + !task.isDirectiveReferenced('cpus') + } + + def 'should report a directive read behind a dynamic ext directive' () { + when: + // the script mentions `task.ext.args`, the memory reference is inside the closure -- + // this is the common nf-core config idiom `ext.args = { ... }` + def task = taskWith( + [memory: '8 GB', ext: [args: {"-Xmx${task.memory.toGiga()}g"}]], + new BodyDef({-> "java ${task.ext.args} -jar app.jar"}, 'java ...', 'script') ) + + then: + task.script == 'java -Xmx8g -jar app.jar' + and: + task.isDirectiveReferenced('memory') + } + + def 'should report a directive read by a shell block' () { + when: + def task = taskWith( + [memory: '8 GB'], + new BodyDef({-> 'java -Xmx!{task.memory.toGiga()}g -jar app.jar'}, 'java ...', 'shell') ) + + then: + task.script == 'java -Xmx8g -jar app.jar' + and: + task.isDirectiveReferenced('memory') + } + + def 'should report a directive read by a template file' () { + given: + def file = Files.createTempDirectory('test').resolve('foo.sh') + file.text = 'java -Xmx${task.memory.toGiga()}g -jar app.jar' + + when: + def task = taskWith( + [memory: '8 GB'], + new BodyDef({-> template(file)}, 'template(file)', 'script') ) + + then: + task.script == 'java -Xmx8g -jar app.jar' + and: + task.isDirectiveReferenced('memory') + } + + def 'should not report a directive accessed outside the tracked action' () { + given: + // the engine accesses the directives all the time e.g. the executor asking for the + // memory to request -- only the accesses made while rendering the command count + def task = taskWith( + [memory: '8 GB'], + new BodyDef({-> 'echo hello'}, 'echo hello', 'script') ) + + when: + task.config.getMemory() + + then: + !task.isDirectiveReferenced('memory') + } + + def 'should not report any directive when the command interpolates none' () { + when: + def task = taskWith( + [memory: '8 GB', cpus: 4], + new BodyDef({-> 'echo hello'}, 'echo hello', 'script') ) + + then: + // pins the absence of false positives: what the engine itself touches while the + // command is rendered must not be reported as a dependency of it + !task.isDirectiveReferenced('memory') + !task.isDirectiveReferenced('cpus') + } + + def 'should answer false before the command has been rendered' () { + given: + def task = unresolvedTaskWith(new TaskConfig(memory: '8 GB')) + + expect: + !task.isDirectiveReferenced('memory') + } + + def 'should carry the accessed directives over to a task copy' () { + given: + // a copy keeps the rendered command, so it depends on the same directives -- the + // retryable/spot path in TaskProcessor copies the task without resolving it again + def task = taskWith( + [memory: '8 GB'], + new BodyDef({-> "java -Xmx${task.memory.toGiga()}g -jar app.jar"}, 'java ...', 'script') ) + + when: + def copy = task.makeCopy() + + then: + copy.script == 'java -Xmx8g -jar app.jar' + and: + copy.isDirectiveReferenced('memory') + } + + @Unroll + def 'should report a directive accessed behind a dynamic ext value from the config file [#parser.class.simpleName, #processName]' () { + given: + // the nf-core idiom: the script only mentions `task.ext.args`, the memory reference + // sits inside the closure the config parser produced + def config = taskConfigFrom(parser, ''' + process { + memory = '8 GB' + withName: FOO { + ext.args = { "-Xmx${task.memory.toGiga()}g" } + } + } + ''', processName) + + when: + def task = taskWith(config, new BodyDef({-> "java ${task.ext.args} -jar app.jar"}, 'java ...', 'script')) + + then: + task.script == expected + and: + task.isDirectiveReferenced('memory') == referenced + + where: + parser | processName | expected | referenced + new ConfigParserV1() | 'FOO' | 'java -Xmx8g -jar app.jar' | true + new ConfigParserV2() | 'FOO' | 'java -Xmx8g -jar app.jar' | true + new ConfigParserV1() | 'BAR' | 'java null -jar app.jar' | false + new ConfigParserV2() | 'BAR' | 'java null -jar app.jar' | false + } +} diff --git a/plugins/nf-seqera/README.md b/plugins/nf-seqera/README.md index 3e7cf8d62a..0839fd8ab8 100644 --- a/plugins/nf-seqera/README.md +++ b/plugins/nf-seqera/README.md @@ -107,6 +107,47 @@ seqera { } ``` +#### Processes depending on `task.memory` + +When a prediction model is enabled the scheduler can allocate less memory than the task requested. +The task script however is rendered *before* the task is scheduled, therefore a script referencing +`task.memory` carries the memory that was *requested*, not the one that was allocated e.g. + +```groovy +process FOO { + memory 8.GB + + script: + """ + java -Xmx${task.memory.toGiga()}g -jar app.jar + """ +} +``` + +Here `-Xmx8g` is baked into the command even when the scheduler allocates less, and the task fails +with an out-of-memory error. To prevent this the executor submits the affected tasks with prediction +model `none` and reports a warning. + +The reference is observed while the command is rendered rather than inferred from the source, so the +check covers every way the value can reach the command — the script, a `shell` block, a `template` +file, and a dynamic directive the command interpolates, including the common config `ext.args` idiom: + +```groovy +process { withName: FOO { ext.args = { "-Xmx${task.memory.toGiga()}g" } } } +``` + +A directive resolved *after* the command has been rendered is not covered, e.g. `beforeScript`. + +Set the `seqera/predictionModel` hint explicitly on the process to override this behaviour: + +```groovy +process FOO { + hints 'seqera/predictionModel': 'qr/v1' +} +``` + +Note that `task.cpus` is not subject to this check. + ## Resources - [Seqera Platform Documentation](https://docs.seqera.io/) diff --git a/plugins/nf-seqera/src/main/io/seqera/executor/SeqeraTaskHandler.groovy b/plugins/nf-seqera/src/main/io/seqera/executor/SeqeraTaskHandler.groovy index c27732f050..b2cb8a6835 100644 --- a/plugins/nf-seqera/src/main/io/seqera/executor/SeqeraTaskHandler.groovy +++ b/plugins/nf-seqera/src/main/io/seqera/executor/SeqeraTaskHandler.groovy @@ -131,10 +131,14 @@ class SeqeraTaskHandler extends TaskHandler implements FusionAwareTask { fusionConfig().snapshotsEnabled(), maxSpotAttempts(baseMachineOpts) ) - // resolve optional per-task prediction model override from the seqera/predictionModel hint; - // when unset the task inherits the run-level model + // per-task prediction model, in order of precedence: + // - an explicit `seqera/predictionModel` hint + // - the automatic `task.memory` check, disabling the prediction + // - null, so the task inherits the run-level model final predictionModelHint = HintHelper.resolvePredictionModel(task.config.getHints()) - final predictionModel = predictionModelHint ? PredictionModel.fromValue(predictionModelHint) : null + final predictionModel = predictionModelHint + ? PredictionModel.fromValue(predictionModelHint) + : (shouldDisablePrediction() ? PredictionModel.NONE : null) // build resource limit from process resourceLimits directive (upper bound for OOM retry scaling) final resourceLim = toResourceLimit() // validate container - Seqera executor requires all processes to specify a container image @@ -169,6 +173,40 @@ class SeqeraTaskHandler extends TaskHandler implements FusionAwareTask { executor.getBatchSubmitter().submit(this, schedTask) } + /** + * Determine whether the resource prediction model must be disabled for this task. + * + * The prediction model can allocate less memory than the task requested. However the task script + * is rendered *before* the task is scheduled, therefore a script referencing {@code task.memory} + * carries the memory that was requested, not the one that has been allocated e.g. a JVM + * {@code -Xmx} setting exceeding the container memory and failing with an out-of-memory error. + * + * The reference is observed while the command is rendered -- see + * {@code TaskRun#isDirectiveReferenced} -- therefore it covers a value reached indirectly + * through a dynamic directive e.g. {@code ext.args = { "-Xmx${task.memory.toGiga()}g" }}, + * but not a directive resolved after the command e.g. {@code beforeScript}. + * + * @return {@code true} when the task should be submitted with prediction model {@code none} + */ + protected boolean shouldDisablePrediction() { + // Nothing to disable unless the run enables a prediction model. Checking this first + // also keeps the warning quiet for the runs where the resources are never adjusted, + // which would otherwise report a problem that cannot happen + final runModel = executor.getSeqeraConfig()?.predictionModel + if( !runModel || runModel == PredictionModel.NONE.getValue() ) + return false + // Note only `memory` is checked. The scheduler can adjust the cpus as well, but a + // stale `task.cpus` costs an over-subscribed thread pool whereas a stale + // `task.memory` fails the task, and `task.cpus` is referenced by nearly every + // process -- disabling the prediction for those would defeat the feature + if( !task.isDirectiveReferenced('memory') ) + return false + // warn once per process rather than once per task: the reference belongs to the + // process definition, so every one of its tasks would otherwise report it + log.warn1("Process `${task.processor?.name ?: task.lazyName()}` depends on the `task.memory` value -- resource prediction has been disabled for this process to prevent an under-allocation of the requested memory", firstOnly: true) + return true + } + protected int maxSpotAttempts(MachineRequirementOpts opts) { final result = opts?.maxSpotAttempts if( result != null && result < 0 ) diff --git a/plugins/nf-seqera/src/test/io/seqera/executor/SeqeraTaskHandlerTest.groovy b/plugins/nf-seqera/src/test/io/seqera/executor/SeqeraTaskHandlerTest.groovy index aa61fa1c93..ec346458b4 100644 --- a/plugins/nf-seqera/src/test/io/seqera/executor/SeqeraTaskHandlerTest.groovy +++ b/plugins/nf-seqera/src/test/io/seqera/executor/SeqeraTaskHandlerTest.groovy @@ -1271,10 +1271,77 @@ class SeqeraTaskHandlerTest extends Specification { [:] | null } + def 'submit disables the prediction model when the process references task.memory'() { + given: + Task captured = null + def handler = createSubmitHandler( + predictionModel: 'qr/v2', + memoryReferenced: true, + onSubmit: { captured = it }, + ) + + when: + handler.submit() + then: + captured.getPredictionModel() == PredictionModel.NONE + } + + def 'submit keeps the run-level prediction model when the process does not reference task.memory'() { + given: + Task captured = null + def handler = createSubmitHandler( + predictionModel: 'qr/v2', + memoryReferenced: false, + onSubmit: { captured = it }, + ) + + when: + handler.submit() + then: + captured.getPredictionModel() == null + } + + @Unroll + def 'submit does not disable the prediction model when the run-level model is #runModel'() { + given: + Task captured = null + def handler = createSubmitHandler( + predictionModel: runModel, + memoryReferenced: true, + onSubmit: { captured = it }, + ) + + when: + handler.submit() + then: + captured.getPredictionModel() == null + + where: + runModel << [null, '', 'none'] + } + + def 'submit lets an explicit predictionModel hint win over the task.memory check'() { + given: + Task captured = null + def handler = createSubmitHandler( + predictionModel: 'qr/v2', + memoryReferenced: true, + hints: ['seqera/predictionModel': 'qr/v1'], + onSubmit: { captured = it }, + ) + + when: + handler.submit() + then: + captured.getPredictionModel() == PredictionModel.QR_V1 + } + private SeqeraTaskHandler createSubmitHandler(Map args) { final hints = args.hints as Map ?: [:] final baseMachineReq = args.baseMachineReq as MachineRequirementOpts final Closure onSubmit = args.onSubmit as Closure ?: {} + final memoryReferenced = args.memoryReferenced as boolean + final runPredictionModel = args.predictionModel as String def taskConfig = Mock(TaskConfig) { getCpus() >> 1 @@ -1289,6 +1356,7 @@ class SeqeraTaskHandlerTest extends Specification { getContainer() >> 'ubuntu:latest' getId() >> TaskId.of(1) getHash() >> HashCode.fromInt(1) + isDirectiveReferenced('memory') >> memoryReferenced lazyName() >> 'sample_task' } def executor = Mock(SeqeraExecutor) { @@ -1298,6 +1366,7 @@ class SeqeraTaskHandlerTest extends Specification { } getSeqeraConfig() >> Mock(ExecutorOpts) { getMachineRequirement() >> baseMachineReq + getPredictionModel() >> runPredictionModel } getRunResourceLabels() >> [:] }