Extract TaskRunner from TaskProcessor - #7036
Draft
bentsherman wants to merge 3 commits into
Draft
Conversation
bentsherman
force-pushed
the
task-output-resolver-v1
branch
from
July 29, 2026 23:33
0678fc3 to
52d90b1
Compare
✅ Deploy Preview for nextflow-docs ready!
To edit notification comments on pull requests, go to your Netlify project configuration. |
bentsherman
force-pushed
the
task-output-resolver-v1
branch
from
July 29, 2026 23:55
52d90b1 to
fba6d59
Compare
bentsherman
changed the title
bentsherman
force-pushed
the
task-output-resolver-v1
branch
from
August 3, 2026 16:19
5daa309 to
5e59d3f
Compare
Signed-off-by: Ben Sherman <bentshermann@gmail.com>
Rename `finalizeTask0` to `taskCompleted` and document it as the boundary between the execution of a task and the dataflow interface of the process, that is the only point at which task results are emitted to the process output channels. All execution paths converge here: the `when` guard, a store directory hit, a cached result and the normal completion of a task. No behaviour change, this only names the seam in preparation for extracting the task execution logic to a separate class. Assisted-by: Claude Opus 5 (1M context) Signed-off-by: Ben Sherman <bentshermann@gmail.com>
At its core a process is a function that accepts an input, executes a task and returns an asynchronous result. TaskProcessor implements this functionality together with the dataflow interface of the process, so that a task can only be executed by feeding values through the process input channels. Move the execution of a task to a separate `TaskRunner` class: - the resolution of the task inputs and outputs - the store directory and cache lookup - the delegation to the executor, including task arrays - the task error strategy and retries TaskProcessor keeps the dataflow interface -- the operator, the process state, the output binding and the publishing -- as well as the process scoped services shared by all task executions, such as the name, the config, the executor and the task environment. The processor creates a runner with `taskCompleted` as the completion handler, so that the runner does not depend on the dataflow interface: a task can be executed by invoking `TaskRunner.submit()` directly, without the process being part of a dataflow network (see the test 'should execute a task without a dataflow network'). Note that `resumeOrDie` is now synchronized on the runner rather than on the processor, so it no longer excludes `bindOutputs` and `sendPoisonPill`. These methods do not share mutable state, the coarse lock was incidental. Assisted-by: Claude Opus 5 (1M context) Signed-off-by: Ben Sherman <bentshermann@gmail.com>
bentsherman
force-pushed
the
task-output-resolver-v1
branch
from
August 6, 2026 13:26
5e59d3f to
21eb11a
Compare
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
At its core, a process is a function that accepts an input, executes a
task and returns an asynchronous result. TaskProcessor implements this functionality together with the dataflow interface, so that a task can only be executed by feeding values through the process input channels.
This PR extracts the core task execution logic in
TaskProcessorto a separateTaskRunnerclass:TaskProcessor keeps the dataflow interface -- the operator, the process
state, the output channel emission and the publishing -- as well as the process scoped services shared by all task executions, such as the name, the config, the executor and the task environment.
The processor creates a runner with
taskCompletedas the completion handler, so that the runner does not depend on the dataflow interface: a task can be executed by invokingTaskRunner.submit()directly, without the process being part of a dataflow network.(NOTE: This PR also extracts the legacy process output resolution logic to a separate
TaskOutputResolverV1class, mirroringTaskOutputResolverfor typed outputs).Side benefit: reduces TaskProcessor from ~2200 lines to ~1200 lines, making it easier for agents to read and modify