Skip to content

Commit 1b4e757

Browse files
bentshermanpditommaso
bentsherman
and
pditommaso
authored
Report the task directives read while the command is rendered (#7506)
* Report the task directives read while the command is rendered Rendering the task command reads the directives it interpolates off the task config; record those reads so that an executor adjusting the resources at schedule time can tell whether the rendered command carries a value it is about to change. Being observed rather than inferred, this covers the script, a `shell` block, a `template` file and a dynamic directive value the command interpolates, in both parsers, without any of them being known to the check. A directive resolved after the command has been rendered is not covered. nf-seqera submits the affected tasks with prediction model `none`. Signed-off-by: Ben Sherman <bentshermann@gmail.com> * Address the review of the directive access tracking [ci fast] - carry the accessed directives over to a task copy: the copy keeps the command that was rendered from them, therefore it depends on the same directives. TaskConfig#newCache no longer resets them, since the value cache belongs to a context while the access log belongs to a rendered command - rename the tracking to the `access` vocabulary: `record` reads as a noun in this codebase (TraceRecord, ProgressRecord, RecordMap, and the Java keyword) and `read` is ambiguous in a config class - note the paths that bypass TaskConfig#get, and the dynamic top-level directive whose nested access the value cache can hide - document that the check answers false before the command has been rendered, for a native `exec` task and for a task array - cover the absence of false positives, the task copy, and the config-file `ext.args` idiom through both config parsers Assisted-by: Claude Opus 5 (Claude Code) Signed-off-by: Paolo Di Tommaso <paolo.ditommaso@gmail.com> * Clarify the prediction model precedence [ci fast] State the three cases as a list, so the nested ternary reads as precedence rather than as a check: an explicit hint, then the automatic `task.memory` check, then null to inherit the run-level model. Assisted-by: Claude Opus 5 (Claude Code) Signed-off-by: Paolo Di Tommaso <paolo.ditommaso@gmail.com> --------- Signed-off-by: Ben Sherman <bentshermann@gmail.com> Signed-off-by: Paolo Di Tommaso <paolo.ditommaso@gmail.com> Co-authored-by: Paolo Di Tommaso <paolo.ditommaso@gmail.com>
1 parent 5400da6 commit 1b4e757

6 files changed

Lines changed: 451 additions & 6 deletions

File tree

‎modules/nextflow/src/main/groovy/nextflow/processor/TaskConfig.groovy‎

Lines changed: 49 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -48,6 +48,11 @@ class TaskConfig extends LazyMap implements Cloneable {
4848

4949
private transient Map cache = new LinkedHashMap(20)
5050

51+
/** The directive names accessed while {@link #trackingAccess} is enabled */
52+
private transient Set<String> accessedDirectives = new HashSet<>(10)
53+
54+
private transient boolean trackingAccess
55+
5156
TaskConfig() { }
5257

5358
TaskConfig( Map<String,Object> entries ) {
@@ -58,13 +63,46 @@ class TaskConfig extends LazyMap implements Cloneable {
5863
def copy = (TaskConfig)super.clone()
5964
copy.setTarget(new HashMap<>(this.getTarget()))
6065
copy.newCache()
66+
// copied, not reset: the copy carries over the command rendered from them
67+
// -- see TaskRun#clone -- therefore it depends on the same directives
68+
copy.accessedDirectives = new HashSet<>(this.accessedDirectives)
6169
return copy
6270
}
6371

72+
/**
73+
* Discard the resolved directive values. Note it does *not* touch the accessed names: the
74+
* value cache belongs to a context, the access log to a rendered command.
75+
*/
6476
private void newCache() {
6577
cache = [:]
6678
}
6779

80+
/**
81+
* Track the directives accessed while the given action runs.
82+
*
83+
* The task command is rendered by accessing the directives it interpolates off this object,
84+
* therefore tracking the accesses while it happens tells which directives the rendered
85+
* command depends on. It is scoped to that action because the directives are accessed all
86+
* the time by the rest of the engine e.g. the executor asking for the memory to request.
87+
*
88+
* The caller must disable it once the command is rendered, including on failure, since
89+
* a flag left enabled would report every later access as a dependency of the command.
90+
*
91+
* @see nextflow.processor.TaskRun#resolve
92+
* @param value Whether the directive accesses must be tracked
93+
*/
94+
void trackDirectiveAccess(boolean value) {
95+
trackingAccess = value
96+
}
97+
98+
/**
99+
* @param directive The directive name e.g. {@code memory}
100+
* @return {@code true} when the given directive was accessed while the accesses were tracked
101+
*/
102+
boolean isDirectiveAccessed(String directive) {
103+
return accessedDirectives.contains(directive)
104+
}
105+
68106
/**
69107
* Assign the context map for dynamic evaluation of task config properties
70108
* @param context A {@link TaskContext} object that holds the task evaluation context
@@ -137,6 +175,14 @@ class TaskConfig extends LazyMap implements Cloneable {
137175
}
138176

139177
def get( String key ) {
178+
// note this is the funnel for a directive *property* access, either directly or via the
179+
// matching getter e.g. #getMemory. Only #eval (used by the task hasher) and #getRawValue
180+
// bypass it -- a directive read through those while tracking would go unnoticed
181+
if( trackingAccess )
182+
accessedDirectives.add(key)
183+
184+
// note the access is tracked before the cache is consulted, so a directive already
185+
// resolved outside the tracked window is still reported
140186
if( cache.containsKey(key) )
141187
return cache.get(key)
142188

@@ -152,6 +198,9 @@ class TaskConfig extends LazyMap implements Cloneable {
152198
else
153199
result = super.get(key)
154200

201+
// note a dynamic top-level directive is cached by its resolved value, so a directive its
202+
// closure accesses in turn is only seen on the first resolution -- resolving one before
203+
// the command is rendered would hide it. `ext` is unaffected: what is cached is its map
155204
cache.put(key,result)
156205
return result
157206
}

‎modules/nextflow/src/main/groovy/nextflow/processor/TaskRun.groovy‎

Lines changed: 42 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -853,9 +853,48 @@ class TaskRun implements Cloneable {
853853
* @param body A {@code BodyDef} object instance
854854
*/
855855
void resolve(BodyDef body) {
856-
processor.session.stubRun && config.getStubBlock()
857-
? resolveStub(config.getStubBlock())
858-
: resolveBody(body)
856+
// track the directives accessed while the command is rendered -- see #isDirectiveReferenced
857+
// note the null guards are load-bearing -- the access below is behind a short-circuit
858+
config?.trackDirectiveAccess(true)
859+
try {
860+
processor.session.stubRun && config.getStubBlock()
861+
? resolveStub(config.getStubBlock())
862+
: resolveBody(body)
863+
}
864+
finally {
865+
config?.trackDirectiveAccess(false)
866+
}
867+
}
868+
869+
/**
870+
* Report whether the rendered task command depends on the value of the given
871+
* {@code task} directive e.g. {@code memory} for a script interpolating
872+
* {@code "-Xmx${task.memory.toGiga()}g"}.
873+
*
874+
* The command is rendered *before* the task is scheduled, therefore an executor that
875+
* adjusts the requested resources at schedule time needs to know whether the command
876+
* carries a value it is about to change.
877+
*
878+
* The reference is *observed*, not inferred: rendering the command accesses the directive
879+
* off the task config, and {@link #resolve} tracks the accesses while it happens. That
880+
* covers every path the command can be rendered through -- the script, a {@code shell}
881+
* block, a {@code template} file, and a dynamic directive value the command interpolates,
882+
* whether declared in the process or in the config file -- without any of them being
883+
* known here.
884+
*
885+
* Note it reports the *last* rendering of this task, hence {@code false} until
886+
* {@link #resolve} has run, for an {@code exec} task, and for a task array -- which
887+
* {@code TaskArrayCollector} assembles without resolving it.
888+
*
889+
* ponytail: a directive resolved *after* the command has been rendered is not observed
890+
* e.g. `beforeScript = { "-Xmx${task.memory}" }`, which the wrapper builder resolves at
891+
* submit time. Widen the tracked action to cover the wrapper if that case shows up.
892+
*
893+
* @param directive The directive name e.g. {@code memory}
894+
* @return {@code true} when rendering the command accessed the given directive
895+
*/
896+
boolean isDirectiveReferenced(String directive) {
897+
return config != null && config.isDirectiveAccessed(directive)
859898
}
860899

861900
protected void resolveBody(BodyDef body) {

‎modules/nextflow/src/test/groovy/nextflow/processor/TaskDirectiveReadsTest.groovy‎

Lines changed: 209 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,209 @@
1+
/*
2+
* Copyright 2013-2026, Seqera Labs
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
package nextflow.processor
18+
19+
import java.nio.file.Files
20+
21+
import ch.artecat.grengine.Grengine
22+
import nextflow.Session
23+
import nextflow.config.parser.v1.ConfigParserV1
24+
import nextflow.config.parser.v2.ConfigParserV2
25+
import nextflow.script.BodyDef
26+
import nextflow.script.ProcessConfig
27+
import nextflow.script.dsl.ProcessConfigBuilder
28+
import spock.lang.Specification
29+
import spock.lang.Unroll
30+
31+
/**
32+
* Verify the directives read while the task command is rendered are reported by
33+
* {@link TaskRun#isDirectiveReferenced}.
34+
*
35+
* @author Ben Sherman <ben.sherman@seqera.io>
36+
*/
37+
class TaskDirectiveReadsTest extends Specification {
38+
39+
private TaskRun taskWith(Map directives, BodyDef body) {
40+
return taskWith(new TaskConfig(directives), body)
41+
}
42+
43+
private TaskRun taskWith(TaskConfig config, BodyDef body) {
44+
final task = unresolvedTaskWith(config)
45+
task.resolve(body)
46+
return task
47+
}
48+
49+
private TaskRun unresolvedTaskWith(TaskConfig config) {
50+
final processor = new TaskProcessor()
51+
processor.@session = new Session()
52+
processor.@grengine = new Grengine()
53+
54+
final task = new TaskRun(processor: processor, config: config)
55+
task.context = new TaskContext(holder: [:])
56+
task.config.setContext(task.context)
57+
return task
58+
}
59+
60+
/**
61+
* Build the task config as the engine does, by applying a parsed config file to the process
62+
* config -- so the directive values are the closures the parser produced.
63+
*/
64+
private TaskConfig taskConfigFrom(parser, String text, String processName) {
65+
final config = parser.parse(text)
66+
final processConfig = new ProcessConfig([:])
67+
new ProcessConfigBuilder(processConfig)
68+
.applyConfig(config.process as Map, processName, processName, processName)
69+
return processConfig.createTaskConfig()
70+
}
71+
72+
def 'should report a directive read by the task script' () {
73+
when:
74+
def task = taskWith(
75+
[memory: '8 GB', cpus: 4],
76+
new BodyDef({-> "java -Xmx${task.memory.toGiga()}g -jar app.jar"}, 'java ...', 'script') )
77+
78+
then:
79+
task.script == 'java -Xmx8g -jar app.jar'
80+
and:
81+
task.isDirectiveReferenced('memory')
82+
!task.isDirectiveReferenced('cpus')
83+
}
84+
85+
def 'should report a directive read behind a dynamic ext directive' () {
86+
when:
87+
// the script mentions `task.ext.args`, the memory reference is inside the closure --
88+
// this is the common nf-core config idiom `ext.args = { ... }`
89+
def task = taskWith(
90+
[memory: '8 GB', ext: [args: {"-Xmx${task.memory.toGiga()}g"}]],
91+
new BodyDef({-> "java ${task.ext.args} -jar app.jar"}, 'java ...', 'script') )
92+
93+
then:
94+
task.script == 'java -Xmx8g -jar app.jar'
95+
and:
96+
task.isDirectiveReferenced('memory')
97+
}
98+
99+
def 'should report a directive read by a shell block' () {
100+
when:
101+
def task = taskWith(
102+
[memory: '8 GB'],
103+
new BodyDef({-> 'java -Xmx!{task.memory.toGiga()}g -jar app.jar'}, 'java ...', 'shell') )
104+
105+
then:
106+
task.script == 'java -Xmx8g -jar app.jar'
107+
and:
108+
task.isDirectiveReferenced('memory')
109+
}
110+
111+
def 'should report a directive read by a template file' () {
112+
given:
113+
def file = Files.createTempDirectory('test').resolve('foo.sh')
114+
file.text = 'java -Xmx${task.memory.toGiga()}g -jar app.jar'
115+
116+
when:
117+
def task = taskWith(
118+
[memory: '8 GB'],
119+
new BodyDef({-> template(file)}, 'template(file)', 'script') )
120+
121+
then:
122+
task.script == 'java -Xmx8g -jar app.jar'
123+
and:
124+
task.isDirectiveReferenced('memory')
125+
}
126+
127+
def 'should not report a directive accessed outside the tracked action' () {
128+
given:
129+
// the engine accesses the directives all the time e.g. the executor asking for the
130+
// memory to request -- only the accesses made while rendering the command count
131+
def task = taskWith(
132+
[memory: '8 GB'],
133+
new BodyDef({-> 'echo hello'}, 'echo hello', 'script') )
134+
135+
when:
136+
task.config.getMemory()
137+
138+
then:
139+
!task.isDirectiveReferenced('memory')
140+
}
141+
142+
def 'should not report any directive when the command interpolates none' () {
143+
when:
144+
def task = taskWith(
145+
[memory: '8 GB', cpus: 4],
146+
new BodyDef({-> 'echo hello'}, 'echo hello', 'script') )
147+
148+
then:
149+
// pins the absence of false positives: what the engine itself touches while the
150+
// command is rendered must not be reported as a dependency of it
151+
!task.isDirectiveReferenced('memory')
152+
!task.isDirectiveReferenced('cpus')
153+
}
154+
155+
def 'should answer false before the command has been rendered' () {
156+
given:
157+
def task = unresolvedTaskWith(new TaskConfig(memory: '8 GB'))
158+
159+
expect:
160+
!task.isDirectiveReferenced('memory')
161+
}
162+
163+
def 'should carry the accessed directives over to a task copy' () {
164+
given:
165+
// a copy keeps the rendered command, so it depends on the same directives -- the
166+
// retryable/spot path in TaskProcessor copies the task without resolving it again
167+
def task = taskWith(
168+
[memory: '8 GB'],
169+
new BodyDef({-> "java -Xmx${task.memory.toGiga()}g -jar app.jar"}, 'java ...', 'script') )
170+
171+
when:
172+
def copy = task.makeCopy()
173+
174+
then:
175+
copy.script == 'java -Xmx8g -jar app.jar'
176+
and:
177+
copy.isDirectiveReferenced('memory')
178+
}
179+
180+
@Unroll
181+
def 'should report a directive accessed behind a dynamic ext value from the config file [#parser.class.simpleName, #processName]' () {
182+
given:
183+
// the nf-core idiom: the script only mentions `task.ext.args`, the memory reference
184+
// sits inside the closure the config parser produced
185+
def config = taskConfigFrom(parser, '''
186+
process {
187+
memory = '8 GB'
188+
withName: FOO {
189+
ext.args = { "-Xmx${task.memory.toGiga()}g" }
190+
}
191+
}
192+
''', processName)
193+
194+
when:
195+
def task = taskWith(config, new BodyDef({-> "java ${task.ext.args} -jar app.jar"}, 'java ...', 'script'))
196+
197+
then:
198+
task.script == expected
199+
and:
200+
task.isDirectiveReferenced('memory') == referenced
201+
202+
where:
203+
parser | processName | expected | referenced
204+
new ConfigParserV1() | 'FOO' | 'java -Xmx8g -jar app.jar' | true
205+
new ConfigParserV2() | 'FOO' | 'java -Xmx8g -jar app.jar' | true
206+
new ConfigParserV1() | 'BAR' | 'java null -jar app.jar' | false
207+
new ConfigParserV2() | 'BAR' | 'java null -jar app.jar' | false
208+
}
209+
}

‎plugins/nf-seqera/README.md‎

Lines changed: 41 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -107,6 +107,47 @@ seqera {
107107
}
108108
```
109109

110+
#### Processes depending on `task.memory`
111+
112+
When a prediction model is enabled the scheduler can allocate less memory than the task requested.
113+
The task script however is rendered *before* the task is scheduled, therefore a script referencing
114+
`task.memory` carries the memory that was *requested*, not the one that was allocated e.g.
115+
116+
```groovy
117+
process FOO {
118+
memory 8.GB
119+
120+
script:
121+
"""
122+
java -Xmx${task.memory.toGiga()}g -jar app.jar
123+
"""
124+
}
125+
```
126+
127+
Here `-Xmx8g` is baked into the command even when the scheduler allocates less, and the task fails
128+
with an out-of-memory error. To prevent this the executor submits the affected tasks with prediction
129+
model `none` and reports a warning.
130+
131+
The reference is observed while the command is rendered rather than inferred from the source, so the
132+
check covers every way the value can reach the command — the script, a `shell` block, a `template`
133+
file, and a dynamic directive the command interpolates, including the common config `ext.args` idiom:
134+
135+
```groovy
136+
process { withName: FOO { ext.args = { "-Xmx${task.memory.toGiga()}g" } } }
137+
```
138+
139+
A directive resolved *after* the command has been rendered is not covered, e.g. `beforeScript`.
140+
141+
Set the `seqera/predictionModel` hint explicitly on the process to override this behaviour:
142+
143+
```groovy
144+
process FOO {
145+
hints 'seqera/predictionModel': 'qr/v1'
146+
}
147+
```
148+
149+
Note that `task.cpus` is not subject to this check.
150+
110151
## Resources
111152

112153
- [Seqera Platform Documentation](https://docs.seqera.io/)

0 commit comments

Comments
 (0)