[pipeline] support async ops in fused subprocess stages - #1582
Merged
Conversation
`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
enabled auto-merge (squash)
July 2, 2026 08:32
moto-meta
pushed a commit
that referenced
this pull request
Jul 6, 2026
Follow-up to the `.to()` region API (1/4–4/4; see #1584 for the overall design and rationale). Removes the executor-identity fusion path now that `.to()` regions provide the same capability with an explicit, statically-configurable surface. Both removed features were added in the still-unreleased 0.6.0 cycle, so no deprecation shim is warranted. Removed: - The `fuse_subprocess_stages` keyword from `PipelineBuilder.build`, `build_pipeline`, and `run_pipeline_in_subprocess`. - The executor-identity fusion machinery in `_fuse.py` (`_find_fusable_runs`, `_scan_run`, `_FusableRun`, `_fusable_*`, `_fuse_subprocess_stages`, `_pool_params`, the identity `_build_fused_stage`) and the async-op-as-fusion-tag pass (`_strip_async_executor_tags`). The marker path (`_fuse_marked_regions` and friends) is kept. Reverted (the async-op executor relaxation from #1582): an async op may no longer be given an `executor`. `PipeConfig.__post_init__` again rejects any executor on an async op, and `convert_to_async` asserts it is `None`. In a `.to()` region an async op is placed by the marker and needs no per-stage executor tag, so the relaxation is obsolete. `run_pipeline_in_subprocess` now fuses `.to()` regions in the main process (via `_fuse_marked_regions`), preserving the main-ownership of region worker pools that the old flag provided. Docs: the parallelism guide's "Multi-processing (fused)" section is rewritten as "Multi-processing (region)" using `.to()`.
moto-meta
pushed a commit
that referenced
this pull request
Jul 10, 2026
Follow-up to the `.to()` region API (1/4–4/4; see #1584 for the overall design and rationale). Removes the executor-identity fusion path now that `.to()` regions provide the same capability with an explicit, statically-configurable surface. Both removed features were added in the still-unreleased 0.6.0 cycle, so no deprecation shim is warranted. Removed: - The `fuse_subprocess_stages` keyword from `PipelineBuilder.build`, `build_pipeline`, and `run_pipeline_in_subprocess`. - The executor-identity fusion machinery in `_fuse.py` (`_find_fusable_runs`, `_scan_run`, `_FusableRun`, `_fusable_*`, `_fuse_subprocess_stages`, `_pool_params`, the identity `_build_fused_stage`) and the async-op-as-fusion-tag pass (`_strip_async_executor_tags`). The marker path (`_fuse_marked_regions` and friends) is kept. Reverted (the async-op executor relaxation from #1582): an async op may no longer be given an `executor`. `PipeConfig.__post_init__` again rejects any executor on an async op, and `convert_to_async` asserts it is `None`. In a `.to()` region an async op is placed by the marker and needs no per-stage executor tag, so the relaxation is obsolete. `run_pipeline_in_subprocess` now fuses `.to()` regions in the main process (via `_fuse_marked_regions`), preserving the main-ownership of region worker pools that the old flag provided. Docs: the parallelism guide's "Multi-processing (fused)" section is rewritten as "Multi-processing (region)" using `.to()`.
moto-meta
pushed a commit
that referenced
this pull request
Jul 10, 2026
Follow-up to the `.to()` region API (1/4–4/4; see #1584 for the overall design and rationale). Removes the executor-identity fusion path now that `.to()` regions provide the same capability with an explicit, statically-configurable surface. Both removed features were added in the still-unreleased 0.6.0 cycle, so no deprecation shim is warranted. Removed: - The `fuse_subprocess_stages` keyword from `PipelineBuilder.build`, `build_pipeline`, and `run_pipeline_in_subprocess`. - The executor-identity fusion machinery in `_fuse.py` (`_find_fusable_runs`, `_scan_run`, `_FusableRun`, `_fusable_*`, `_fuse_subprocess_stages`, `_pool_params`, the identity `_build_fused_stage`) and the async-op-as-fusion-tag pass (`_strip_async_executor_tags`). The marker path (`_fuse_marked_regions` and friends) is kept. Reverted (the async-op executor relaxation from #1582): an async op may no longer be given an `executor`. `PipeConfig.__post_init__` again rejects any executor on an async op, and `convert_to_async` asserts it is `None`. In a `.to()` region an async op is placed by the marker and needs no per-stage executor tag, so the relaxation is obsolete. `run_pipeline_in_subprocess` now fuses `.to()` regions in the main process (via `_fuse_marked_regions`), preserving the main-ownership of region worker pools that the old flag provided. Docs: the parallelism guide's "Multi-processing (fused)" section is rewritten as "Multi-processing (region)" using `.to()`.
moto-meta
pushed a commit
that referenced
this pull request
Jul 10, 2026
Follow-up to the `.to()` region API (1/4–4/4; see #1584 for the overall design and rationale). Removes the executor-identity fusion path now that `.to()` regions provide the same capability with an explicit, statically-configurable surface. Both removed features were added in the still-unreleased 0.6.0 cycle, so no deprecation shim is warranted. Removed: - The `fuse_subprocess_stages` keyword from `PipelineBuilder.build`, `build_pipeline`, and `run_pipeline_in_subprocess`. - The executor-identity fusion machinery in `_fuse.py` (`_find_fusable_runs`, `_scan_run`, `_FusableRun`, `_fusable_*`, `_fuse_subprocess_stages`, `_pool_params`, the identity `_build_fused_stage`) and the async-op-as-fusion-tag pass (`_strip_async_executor_tags`). The marker path (`_fuse_marked_regions` and friends) is kept. Reverted (the async-op executor relaxation from #1582): an async op may no longer be given an `executor`. `PipeConfig.__post_init__` again rejects any executor on an async op, and `convert_to_async` asserts it is `None`. In a `.to()` region an async op is placed by the marker and needs no per-stage executor tag, so the relaxation is obsolete. `run_pipeline_in_subprocess` now fuses `.to()` regions in the main process (via `_fuse_marked_regions`), preserving the main-ownership of region worker pools that the old flag provided. Docs: the parallelism guide's "Multi-processing (fused)" section is rewritten as "Multi-processing (region)" using `.to()`.
moto-meta
pushed a commit
that referenced
this pull request
Jul 10, 2026
Follow-up to the `.to()` region API (1/4–4/4; see #1584 for the overall design and rationale). Removes the executor-identity fusion path now that `.to()` regions provide the same capability with an explicit, statically-configurable surface. Both removed features were added in the still-unreleased 0.6.0 cycle, so no deprecation shim is warranted. Removed: - The `fuse_subprocess_stages` keyword from `PipelineBuilder.build`, `build_pipeline`, and `run_pipeline_in_subprocess`. - The executor-identity fusion machinery in `_fuse.py` (`_find_fusable_runs`, `_scan_run`, `_FusableRun`, `_fusable_*`, `_fuse_subprocess_stages`, `_pool_params`, the identity `_build_fused_stage`) and the async-op-as-fusion-tag pass (`_strip_async_executor_tags`). The marker path (`_fuse_marked_regions` and friends) is kept. Reverted (the async-op executor relaxation from #1582): an async op may no longer be given an `executor`. `PipeConfig.__post_init__` again rejects any executor on an async op, and `convert_to_async` asserts it is `None`. In a `.to()` region an async op is placed by the marker and needs no per-stage executor tag, so the relaxation is obsolete. `run_pipeline_in_subprocess` now fuses `.to()` regions in the main process (via `_fuse_marked_regions`), preserving the main-ownership of region worker pools that the old flag provided. Docs: the parallelism guide's "Multi-processing (fused)" section is rewritten as "Multi-processing (region)" using `.to()`.
moto-meta
pushed a commit
that referenced
this pull request
Jul 10, 2026
Follow-up to the `.to()` region API (1/4–4/4; see #1584 for the overall design and rationale). Removes the executor-identity fusion path now that `.to()` regions provide the same capability with an explicit, statically-configurable surface. Both removed features were added in the still-unreleased 0.6.0 cycle, so no deprecation shim is warranted. Removed: - The `fuse_subprocess_stages` keyword from `PipelineBuilder.build`, `build_pipeline`, and `run_pipeline_in_subprocess`. - The executor-identity fusion machinery in `_fuse.py` (`_find_fusable_runs`, `_scan_run`, `_FusableRun`, `_fusable_*`, `_fuse_subprocess_stages`, `_pool_params`, the identity `_build_fused_stage`) and the async-op-as-fusion-tag pass (`_strip_async_executor_tags`). The marker path (`_fuse_marked_regions` and friends) is kept. Reverted (the async-op executor relaxation from #1582): an async op may no longer be given an `executor`. `PipeConfig.__post_init__` again rejects any executor on an async op, and `convert_to_async` asserts it is `None`. In a `.to()` region an async op is placed by the marker and needs no per-stage executor tag, so the relaxation is obsolete. `run_pipeline_in_subprocess` now fuses `.to()` regions in the main process (via `_fuse_marked_regions`), preserving the main-ownership of region worker pools that the old flag provided. Docs: the parallelism guide's "Multi-processing (fused)" section is rewritten as "Multi-processing (region)" using `.to()`.
mthrok
added a commit
that referenced
this pull request
Jul 10, 2026
Follow-up to the `.to()` region API (1/4–4/4; see #1584 for the overall design and rationale). Removes the executor-identity fusion path now that `.to()` regions provide the same capability with an explicit, statically-configurable surface. Both removed features were added in the still-unreleased 0.6.0 cycle, so no deprecation shim is warranted. Removed: - The `fuse_subprocess_stages` keyword from `PipelineBuilder.build`, `build_pipeline`, and `run_pipeline_in_subprocess`. - The executor-identity fusion machinery in `_fuse.py` (`_find_fusable_runs`, `_scan_run`, `_FusableRun`, `_fusable_*`, `_fuse_subprocess_stages`, `_pool_params`, the identity `_build_fused_stage`) and the async-op-as-fusion-tag pass (`_strip_async_executor_tags`). The marker path (`_fuse_marked_regions` and friends) is kept. Reverted (the async-op executor relaxation from #1582): an async op may no longer be given an `executor`. `PipeConfig.__post_init__` again rejects any executor on an async op, and `convert_to_async` asserts it is `None`. In a `.to()` region an async op is placed by the marker and needs no per-stage executor tag, so the relaxation is obsolete. `run_pipeline_in_subprocess` now fuses `.to()` regions in the main process (via `_fuse_marked_regions`), preserving the main-ownership of region worker pools that the old flag provided. Docs: the parallelism guide's "Multi-processing (fused)" section is rewritten as "Multi-processing (region)" using `.to()`.
mthrok
added a commit
that referenced
this pull request
Jul 10, 2026
Follow-up to the `.to()` region API (1/4–4/4; see #1584 for the overall design and rationale). Removes the executor-identity fusion path now that `.to()` regions provide the same capability with an explicit, statically-configurable surface. Both removed features were added in the still-unreleased 0.6.0 cycle, so no deprecation shim is warranted. Removed: - The `fuse_subprocess_stages` keyword from `PipelineBuilder.build`, `build_pipeline`, and `run_pipeline_in_subprocess`. - The executor-identity fusion machinery in `_fuse.py` (`_find_fusable_runs`, `_scan_run`, `_FusableRun`, `_fusable_*`, `_fuse_subprocess_stages`, `_pool_params`, the identity `_build_fused_stage`) and the async-op-as-fusion-tag pass (`_strip_async_executor_tags`). The marker path (`_fuse_marked_regions` and friends) is kept. Reverted (the async-op executor relaxation from #1582): an async op may no longer be given an `executor`. `PipeConfig.__post_init__` again rejects any executor on an async op, and `convert_to_async` asserts it is `None`. In a `.to()` region an async op is placed by the marker and needs no per-stage executor tag, so the relaxation is obsolete. `run_pipeline_in_subprocess` now fuses `.to()` regions in the main process (via `_fuse_marked_regions`), preserving the main-ownership of region worker pools that the old flag provided. Docs: the parallelism guide's "Multi-processing (fused)" section is rewritten as "Multi-processing (region)" using `.to()`.
mthrok
added a commit
that referenced
this pull request
Jul 11, 2026
Follow-up to the `.to()` region API (1/4–4/4; see #1584 for the overall design and rationale). Removes the executor-identity fusion path now that `.to()` regions provide the same capability with an explicit, statically-configurable surface. Both removed features were added in the still-unreleased 0.6.0 cycle, so no deprecation shim is warranted. Removed: - The `fuse_subprocess_stages` keyword from `PipelineBuilder.build`, `build_pipeline`, and `run_pipeline_in_subprocess`. - The executor-identity fusion machinery in `_fuse.py` (`_find_fusable_runs`, `_scan_run`, `_FusableRun`, `_fusable_*`, `_fuse_subprocess_stages`, `_pool_params`, the identity `_build_fused_stage`) and the async-op-as-fusion-tag pass (`_strip_async_executor_tags`). The marker path (`_fuse_marked_regions` and friends) is kept. Reverted (the async-op executor relaxation from #1582): an async op may no longer be given an `executor`. `PipeConfig.__post_init__` again rejects any executor on an async op, and `convert_to_async` asserts it is `None`. In a `.to()` region an async op is placed by the marker and needs no per-stage executor tag, so the relaxation is obsolete. `run_pipeline_in_subprocess` now fuses `.to()` regions in the main process (via `_fuse_marked_regions`), preserving the main-ownership of region worker pools that the old flag provided. Docs: the parallelism guide's "Multi-processing (fused)" section is rewritten as "Multi-processing (region)" using `.to()`.
mthrok
added a commit
that referenced
this pull request
Jul 13, 2026
Follow-up to the `.to()` region API (1/4–4/4; see #1584 for the overall design and rationale). Removes the executor-identity fusion path now that `.to()` regions provide the same capability with an explicit, statically-configurable surface. Both removed features were added in the still-unreleased 0.6.0 cycle, so no deprecation shim is warranted. Removed: - The `fuse_subprocess_stages` keyword from `PipelineBuilder.build`, `build_pipeline`, and `run_pipeline_in_subprocess`. - The executor-identity fusion machinery in `_fuse.py` (`_find_fusable_runs`, `_scan_run`, `_FusableRun`, `_fusable_*`, `_fuse_subprocess_stages`, `_pool_params`, the identity `_build_fused_stage`) and the async-op-as-fusion-tag pass (`_strip_async_executor_tags`). The marker path (`_fuse_marked_regions` and friends) is kept. Reverted (the async-op executor relaxation from #1582): an async op may no longer be given an `executor`. `PipeConfig.__post_init__` again rejects any executor on an async op, and `convert_to_async` asserts it is `None`. In a `.to()` region an async op is placed by the marker and needs no per-stage executor tag, so the relaxation is obsolete. `run_pipeline_in_subprocess` now fuses `.to()` regions in the main process (via `_fuse_marked_regions`), preserving the main-ownership of region worker pools that the old flag provided. Docs: the parallelism guide's "Multi-processing (fused)" section is rewritten as "Multi-processing (region)" using `.to()`.
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 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.Builds on the parent diff, which moves the
_is_isolating_pool/_is_interpreter_poolhelpers into_common/_convert; this diff adds their new call sites (notably indefs).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._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.