[Pipeline] Support async ops in fused subprocess stages - #1579
Merged
Conversation
mthrok
added a commit
that referenced
this pull request
Jul 2, 2026
Pure refactor, no behavior change. Moves the executor-classification helpers `_is_isolating_pool` and `_is_interpreter_pool` — and the `_INTERPRETER_POOL_CLASS` version guard they depend on — from `spdl/pipeline/_fuse.py` into `spdl/pipeline/_common/_convert.py`, next to the existing `_is_process_pool` they build on. `_common/_convert` sits low in the import graph, so hosting the helpers there lets other modules (notably `spdl/pipeline/defs`) reuse them without importing `_fuse` and creating an import cycle. `_fuse` now imports `_is_isolating_pool` from `_common/_convert` instead of defining it. Extracted from #1579
Contributor
|
This pull request has been imported. If you are a Meta employee, you can view this in D110397738. (Because this pull request was imported automatically, there will not be any future comments.) |
mthrok
added a commit
that referenced
this pull request
Jul 2, 2026
…1580) Pure refactor, no behavior change. Moves the executor-classification helpers `_is_isolating_pool` and `_is_interpreter_pool` — and the `_INTERPRETER_POOL_CLASS` version guard they depend on — from `spdl/pipeline/_fuse.py` into `spdl/pipeline/_common/_convert.py`, next to the existing `_is_process_pool` they build on. `_common/_convert` sits low in the import graph, so hosting the helpers there lets other modules (notably `spdl/pipeline/defs`) reuse them without importing `_fuse` and creating an import cycle. `_fuse` now imports `_is_isolating_pool` from `_common/_convert` instead of defining it. Extracted from #1579
`fuse_subprocess_stages=True` fuses runs of adjacent pipe stages that share the same isolating-pool (process/interpreter) executor instance into one nested `Pipeline` that runs inside a worker process, eliminating the per-stage IPC round-trip. Previously an async op could never be part of a fused run: `PipeConfig` rejected any `executor` on an async op, and fusion groups stages purely by executor-instance identity. An async op between two pool stages therefore split the fusable run in two, forcing the intermediate value to round-trip (and be pickled) through the main process. This lets an async op join a fused run by tagging it with the same isolating-pool executor as its neighbours. The executor is never used to run the coroutine (an async op always runs on the event loop) — it is only the fusion-group key. When fused, the tag is stripped and the op runs on the worker's own event loop, exactly as fused `path_variants` async branches already do. When not fused (fusion off, or a lone async op), the tag is ignored and the op runs on the main loop as before. The execution engine is unchanged: each worker already rebuilds the sub-config with the normal `build_pipeline`, which runs a full event loop. Builds on the parent diff, which moves the `_is_isolating_pool`/`_is_interpreter_pool` helpers into `_common/_convert`; this diff adds their new call sites (notably in `defs`). Details: - `PipeConfig.__post_init__` now allows an isolating-pool executor on an async op and rejects only non-isolating executors (e.g. a thread pool), which have no effect on an async op. - `convert_to_async` ignores the executor for async ops instead of asserting it is `None`. - Fusion detection: an async op that now carries an executor flows through `_fusable_pool_executor` automatically. Inside a `path_variants` stage, `_scan_variant_pool_executors` ignores an async op's pool tag instead of treating it as an input-ordered pool-pipe — an async op's executor is only a fusion-group tag (it runs on the loop, not the pool), so its `output_order="input"` cannot be broken by pool parallelism and must not block fusing the same-pool stage. Only a sync input-ordered pool-pipe still blocks fusion. `_stage_concurrency` counts async ops as zero worker threads (they run on the loop, not the thread pool). - `run_pipeline_in_subprocess` strips any executor tag left on an unfused async op before the op-agnostic executor-hoisting pass, so a tag never spawns an idle worker pool the op will not use. A fused async op must be picklable, like any fused stage. This is documented on `PipelineBuilder.pipe` and in the parallelism guide.
mthrok
marked this pull request as ready for review
July 2, 2026 02:38
mthrok
added a commit
that referenced
this pull request
Jul 2, 2026
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
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.
fuse_subprocess_stages=Truefuses runs of adjacent pipe stages that share the same isolating-pool (process/interpreter) executor instance into one nestedPipelinethat runs inside a worker process, eliminating the per-stage IPC round-trip. Previously an async op could never be part of a fused run:PipeConfigrejected anyexecutoron an async op, and fusion groups stages purely by executor-instance identity. An async op between two pool stages therefore split the fusable run in two, forcing the intermediate value to round-trip (and be pickled) through the main process.This commit lets an async op join a fused run by tagging it with the same isolating-pool executor as its neighbours. The executor is never used to run the coroutine (an async op always runs on the event loop) — it is only the fusion-group key. When fused, the tag is stripped and the op runs on the worker's own event loop, exactly as fused
path_variantsasync branches already do. When not fused (fusion off, or a lone async op), the tag is ignored and the op runs on the main loop as before. The execution engine is unchanged: each worker already rebuilds the sub-config with the normalbuild_pipeline, which runs a full event loop.Details:
PipeConfig.__post_init__now allows an isolating-pool executor on an async op and rejects only non-isolating executors (e.g. a thread pool), which have no effect on an async op.convert_to_asyncignores the executor for async ops instead of asserting it isNone._is_isolating_pool/_is_interpreter_poolmoved from_fuseto_common/_convertsodefscan reuse them without an import cycle._fusable_pool_executorautomatically. Inside apath_variantsstage,_scan_variant_pool_executorsignores an async op's pool tag instead of treating it as an input-ordered pool-pipe — an async op's executor is only a fusion-group tag (it runs on the loop, not the pool), so itsoutput_order="input"cannot be broken by pool parallelism and must not block fusing the same-pool stage. Only a sync input-ordered pool-pipe still blocks fusion._stage_concurrencycounts async ops as zero worker threads (they run on the loop, not the thread pool).run_pipeline_in_subprocessstrips any executor tag left on an unfused async op before the op-agnostic executor-hoisting pass, so a tag never spawns an idle worker pool the op will not use.A fused async op must be picklable, like any fused stage. This is documented on
PipelineBuilder.pipeand in the parallelism guide.