[pipeline] 1/4: Add pipeline executor and placement config types

This is part 1/4 of a series (plus a follow-up) adding the `.to()` region API to the SPDL `PipelineBuilder` — a declarative way to designate where a run of pipeline stages executes: the main process, a subprocess pool, or a subinterpreter pool. This diff carries the overview for the whole series; the later diffs point back here for the rationale.

Running stages off the main process previously meant attaching a shared `ProcessPoolExecutor` instance to consecutive `pipe()` calls and passing `fuse_subprocess_stages=True` to `build()`. The engine fused adjacent stages that shared the same executor object into one subprocess pipeline, so the value handed from one stage to the next stayed in the worker instead of being pickled back to the main process between stages. That worked but had real limitations: it was implicit (it keyed off executor object identity, which is easy to get wrong and cannot be expressed as static config), it required a live `Executor` (so a pipeline could not be fully described as serializable data), it could not pull `aggregate`/`disaggregate`/`path_variants` stages into a region (a fusion run was bounded at those stages), and it had no subinterpreter support.

The `.to()` API makes the region explicit and declarative. `to(ProcessPoolExecutorConfig(...))` or `to(InterpreterPoolExecutorConfig(...))` opens a region; every stage after it runs in that worker pool until `to(MAIN_PROCESS)` closes the region. The target is a serializable spec rather than a live executor, so a pipeline stays expressible as static config; a region may contain `aggregate`/`disaggregate`/`path_variants`; the worker-pool configuration has a single home; and subinterpreter pools (Python 3.14+) are supported alongside subprocess pools. The value passed between stages inside a region never leaves the worker and need not be picklable — only the region's inputs and outputs cross the boundary.

- 1/4 — this PR (#1584): the config-layer building blocks (the `ExecutorConfig` spec hierarchy — `ProcessPoolExecutorConfig` and `InterpreterPoolExecutorConfig` — plus `MAIN_PROCESS` and the `PlacementConfig` region marker), and widening `PipelineConfig.pipes` to carry region markers. Additive and inert.
- 2/4 — #1585: the engine pass `_fuse_marked_regions` that consumes the markers, replacing each maximal span of stages under a region target with one stage that runs the span as a nested pipeline in a worker pool. Dormant until markers exist.
- 3/4 — #1586: the subinterpreter worker-pool backend, behind a `_PoolBackend` seam, so a region can run in subinterpreters as well as subprocesses.
- 4/4 — #1587: the public surface (`PipelineBuilder.to()` and its validation), which makes the feature usable end to end.
- follow-up — #1588 (BC-breaking): removes the superseded `fuse_subprocess_stages` executor-identity fusion path now that `.to()` provides the same capability with an explicit, statically-configurable surface.

Adds the config-layer types, additive only — no behavior change yet:

- `ExecutorConfig`: the base spec for a worker-pool executor — a serializable description of a pool (worker count, initializer, initargs) that the pipeline later materializes into a live executor. Kept as a base so more executor targets (e.g. a thread pool) can be added without touching the placement machinery, and so the specs can be reused wherever a pool is configured.
- `ProcessPoolExecutorConfig` / `InterpreterPoolExecutorConfig`: the two concrete `ExecutorConfig` subclasses, for a subprocess or subinterpreter worker-pool region. The process variant adds `mp_context`; the subinterpreter variant adds nothing (no new process is started). They describe the pool without holding a live `Executor`.
- `MAIN_PROCESS`: a sentinel target that closes a region and brings subsequent stages back to the main process. It reprs as `MAIN_PROCESS` and, unlike `None`, is unambiguous and serializes cleanly.
- `PlacementConfig`: a region-marker node carried in `PipelineConfig.pipes`; stages after a marker (until the next one) run on its target (an `ExecutorConfig` or `MAIN_PROCESS`).

`PipelineConfig.pipes` now accepts `PlacementConfig`. Nothing produces these markers yet (`PipelineBuilder.to()` lands in 4/4), so this is inert; the fusion pass that consumes them lands in 2/4.

Note: a region's worker thread count and stats interval are pipeline/build settings, not properties of a placement, so they are not fields of these specs — the engine derives the region's thread count from the concurrency of the stages it fuses and inherits the stats interval from `build_pipeline`.

Author: mthrok

src/spdl/pipeline/_components/_node.py

@@ -26,6 +26,7 @@
     PathVariantsConfig,
     PipeConfig,
     PipelineConfig,
+    PlacementConfig,
     SinkConfig,
     SourceConfig,
 )
@@ -312,6 +313,7 @@ def _convert_pipes(
         | DisaggregateConfig
         | PathVariantsConfig
         | _SubprocessPipelineConfig
+        | PlacementConfig
     ],
     n: _TNodes,
     q_class: type[AsyncQueue],
@@ -340,6 +342,12 @@ def _convert_pipes(
         match cfg:
             case PathVariantsConfig():
                 n = _convert_path_variants(cfg, n, q_class, pipeline_id, stage_id, idx)
+            case PlacementConfig():
+                # Region markers are build-time directives resolved before the pipeline is
+                # built; they must never reach node construction.
+                raise ValueError(
+                    "PlacementConfig region markers must be resolved before building nodes."
+                )
             case _:
                 in_q = _get_output_queue(n, idx)
                 info = _get_stage_info(cfg, pipeline_id, stage_id)

src/spdl/pipeline/_profile.py

@@ -25,6 +25,7 @@
     PathVariantsConfig,
     PipeConfig,
     PipelineConfig,
+    PlacementConfig,
     SinkConfig,
     SourceConfig,
 )
@@ -263,7 +264,9 @@ def _profile_pipeline(
         raise ValueError(f"Unexpected source type {type(cfg.src)}")
 
     for pipe in cfg.pipes:
-        if isinstance(pipe, (PathVariantsConfig, _SubprocessPipelineConfig)):
+        if isinstance(
+            pipe, (PathVariantsConfig, _SubprocessPipelineConfig, PlacementConfig)
+        ):
             _LG.warning(
                 "Skipping %s stage in profiling (not supported).", type(pipe).__name__
             )

src/spdl/pipeline/defs/__init__.py

@@ -26,6 +26,7 @@
 """
 
 from ._defs import (
+    _MainProcess,
     _PipeArgs,
     _PipeType,
     _SubprocessPipelineConfig,
@@ -36,18 +37,24 @@
     Collate,
     Disaggregate,
     DisaggregateConfig,
+    ExecutorConfig,
+    InterpreterPoolExecutorConfig,
+    MAIN_PROCESS,
     Merge,
     MergeConfig,
     PathVariants,
     PathVariantsConfig,
     Pipe,
     PipeConfig,
     PipelineConfig,
+    PlacementConfig,
+    ProcessPoolExecutorConfig,
     SinkConfig,
     SourceConfig,
 )
 
 __all__ = [
+    "_MainProcess",
     "_PipeArgs",
     "_PipeType",
     "_SubprocessPipelineConfig",
@@ -62,13 +69,18 @@
     "Collate",
     "Disaggregate",
     "DisaggregateConfig",
+    "ExecutorConfig",
+    "InterpreterPoolExecutorConfig",
+    "MAIN_PROCESS",
     "Merge",
     "MergeConfig",
     "PathVariants",
     "PathVariantsConfig",
     "Pipe",
     "PipeConfig",
     "PipelineConfig",
+    "PlacementConfig",
+    "ProcessPoolExecutorConfig",
     "SinkConfig",
     "SourceConfig",
 ]

src/spdl/pipeline/defs/_defs.py

@@ -40,6 +40,12 @@
     "Collate",
     "DisaggregateConfig",
     "_SubprocessPipelineConfig",
+    "_MainProcess",
+    "ExecutorConfig",
+    "ProcessPoolExecutorConfig",
+    "InterpreterPoolExecutorConfig",
+    "MAIN_PROCESS",
+    "PlacementConfig",
     "PipelineConfig",
     "SinkConfig",
     "SourceConfig",
@@ -436,6 +442,110 @@ class _SubprocessPipelineConfig:
     """The picklable submit-side handle (``_SubprocessPipelineHandle``) for the worker pool."""
 
 
+################################################################################
+# Executor placement (region markers for `.to()`)
+################################################################################
+
+
+@dataclass(frozen=True)
+class ExecutorConfig:
+    """**[Experimental]** Base spec for a worker-pool executor.
+
+    A serializable description of a pool of workers, materialized into a live executor by
+    the pipeline. Subclassed by :py:class:`ProcessPoolExecutorConfig` and
+    :py:class:`InterpreterPoolExecutorConfig`; those subclasses are used as ``.to()``
+    placement targets (via :py:class:`PlacementConfig`).
+
+    .. versionadded:: 0.6.0
+    """
+
+    max_workers: int | None = None
+    """Number of workers. If ``None``, defaults to the number of CPUs."""
+
+    initializer: Callable[..., object] | None = None
+    """Callable run once in each worker before it processes any work."""
+
+    initargs: tuple[Any, ...] = ()
+    """Positional arguments passed to ``initializer``."""
+
+
+@dataclass(frozen=True)
+class ProcessPoolExecutorConfig(ExecutorConfig):
+    """**[Experimental]** A worker-pool executor backed by subprocesses.
+
+    Used as a ``.to()`` target to run a region's stages in a pool of worker *processes*
+    as one nested pipeline — the op->op handoff stays in the worker, so intermediate
+    values are not copied back to the main process and need not be picklable. The region
+    ends at the next ``.to()`` target (see :py:data:`MAIN_PROCESS`).
+
+    .. versionadded:: 0.6.0
+    """
+
+    mp_context: str | None = None
+    """Multiprocessing start method (e.g. ``"spawn"``, ``"fork"``, ``"forkserver"``),
+    as accepted by :py:func:`multiprocessing.get_context`. ``None`` uses the default
+    context."""
+
+
+@dataclass(frozen=True)
+class InterpreterPoolExecutorConfig(ExecutorConfig):
+    """**[Experimental]** A worker-pool executor backed by subinterpreters.
+
+    Like :py:class:`ProcessPoolExecutorConfig`, but the workers are Python
+    *subinterpreters* (:py:mod:`concurrent.interpreters`) sharing the process rather than
+    separate processes. There is no ``mp_context`` because no new process is started.
+
+    .. note::
+
+       Requires Python 3.14 or later; using this target on an older interpreter is an
+       error. NumPy and PyTorch cannot be imported inside a subinterpreter, so a region
+       whose stages need them must use :py:class:`ProcessPoolExecutorConfig` instead.
+
+    .. versionadded:: 0.6.0
+    """
+
+
+@dataclass(frozen=True)
+class _MainProcess:
+    """Sentinel ``.to()`` target for the main process. Use the :py:data:`MAIN_PROCESS`
+    singleton rather than instantiating this type."""
+
+    def __repr__(self) -> str:
+        return "MAIN_PROCESS"
+
+
+MAIN_PROCESS: _MainProcess = _MainProcess()
+"""**[Experimental]** The main-process execution target. Pass to
+:py:meth:`spdl.pipeline.PipelineBuilder.to`
+to close a worker-pool region and bring subsequent stages back to the main process.
+
+.. versionadded:: 0.6.0
+"""
+
+
+@dataclass(frozen=True)
+class PlacementConfig:
+    """**[Experimental]** A region marker designating where the subsequent stages execute.
+
+    Sits among the stage configs in :py:attr:`PipelineConfig.pipes`: every stage after
+    this marker (until the next :py:class:`PlacementConfig`) runs on :py:attr:`target`.
+    :py:meth:`spdl.pipeline.PipelineBuilder.to` appends one of these. A pipeline
+    implicitly starts on the main process, so a marker is needed only to enter a
+    worker-pool region and (with :py:data:`MAIN_PROCESS`) to leave it.
+
+    .. versionadded:: 0.6.0
+    """
+
+    target: "ProcessPoolExecutorConfig | InterpreterPoolExecutorConfig | _MainProcess"
+    """Where the stages following this marker execute."""
+
+    name: str = "placement"
+    """Name of the marker (used only for display)."""
+
+    def __repr__(self) -> str:
+        return f"{self.name}({self.target!r})"
+
+
 ################################################################################
 # PathVariants
 ################################################################################
@@ -607,6 +717,7 @@ class PipelineConfig(Generic[U]):
         | DisaggregateConfig[Any]
         | PathVariantsConfig[Any]
         | _SubprocessPipelineConfig
+        | PlacementConfig
     ]
     """Pipe configurations."""
 

tests/pipeline/executor_config_test.py

@@ -0,0 +1,64 @@
+# Copyright (c) Meta Platforms, Inc. and affiliates.
+# All rights reserved.
+#
+# This source code is licensed under the BSD-style license found in the
+# LICENSE file in the root directory of this source tree.
+
+# pyre-strict
+
+import unittest
+
+from spdl.pipeline.defs import (
+    InterpreterPoolExecutorConfig,
+    MAIN_PROCESS,
+    PlacementConfig,
+    ProcessPoolExecutorConfig,
+)
+
+# NOTE: The behavior of these config types (how a region's worker pool is built and run from a
+# spec), including that a ``PipelineConfig`` accepts region markers in its ``pipes``, is covered
+# end-to-end by ``marked_region_fuse_test`` and ``builder_to_test``. The tests here cover only what
+# those cannot: the hand-written ``__repr__`` methods and the deliberate structural contract of
+# ``InterpreterPoolExecutorConfig`` (that it rejects an ``mp_context``). Plain-dataclass mechanics
+# (default values, field storage, ``frozen``, auto-generated ``__eq__``) are intentionally not
+# re-tested.
+
+
+class TestInterpreterPoolExecutorConfig(unittest.TestCase):
+    """Verify the InterpreterPoolExecutorConfig spec."""
+
+    def test_rejects_mp_context(self) -> None:
+        """InterpreterPoolExecutorConfig rejects ``mp_context``; ProcessPoolExecutorConfig honors it.
+
+        A subinterpreter shares the process, so there is no multiprocessing start method to choose.
+        Passing ``mp_context`` must raise rather than be silently accepted and ignored -- unlike
+        :py:class:`ProcessPoolExecutorConfig`, which starts real processes and stores the chosen
+        start method.
+        """
+        self.assertEqual(
+            ProcessPoolExecutorConfig(mp_context="spawn").mp_context, "spawn"
+        )
+        with self.assertRaises(TypeError):
+            InterpreterPoolExecutorConfig(mp_context="spawn")  # pyre-ignore[28]
+
+
+class TestMainProcess(unittest.TestCase):
+    """Verify the MAIN_PROCESS sentinel target."""
+
+    def test_repr(self) -> None:
+        """The sentinel's custom ``__repr__`` renders its public name for readable configs."""
+        self.assertEqual(repr(MAIN_PROCESS), "MAIN_PROCESS")
+
+
+class TestPlacementConfig(unittest.TestCase):
+    """Verify the PlacementConfig region marker."""
+
+    def test_repr_includes_target(self) -> None:
+        """The marker's custom ``__repr__`` surfaces its target for readable pipeline dumps."""
+        self.assertEqual(
+            repr(PlacementConfig(target=MAIN_PROCESS)), "placement(MAIN_PROCESS)"
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/pipeline/shutdown_hook_test.py

@@ -214,13 +214,29 @@ def _child_exit_code(
 class InterpreterExitTopologyTest(unittest.TestCase):
     """Held, unstopped pipelines exit cleanly at interpreter exit, across topologies."""
 
+    @unittest.skipUnless(
+        "forkserver" in mp.get_all_start_methods(),
+        "forkserver start method is unavailable on this platform (e.g. Windows)",
+    )
     def test_plain_forkserver(self) -> None:
         """Plain thread-only pipeline exits cleanly (forkserver)."""
-        self.assertEqual(_child_exit_code(_scenario_plain, "forkserver"), 0)
+        exit_code = _child_exit_code(_scenario_plain, "forkserver")
+        self.assertIsNotNone(
+            exit_code,
+            "Child process hung (did not exit within timeout); the pipeline's "
+            "non-daemon event-loop thread likely blocked interpreter shutdown.",
+        )
+        self.assertEqual(exit_code, 0)
 
     def test_plain_spawn(self) -> None:
         """Plain thread-only pipeline exits cleanly (spawn)."""
-        self.assertEqual(_child_exit_code(_scenario_plain, "spawn"), 0)
+        exit_code = _child_exit_code(_scenario_plain, "spawn")
+        self.assertIsNotNone(
+            exit_code,
+            "Child process hung (did not exit within timeout); the pipeline's "
+            "non-daemon event-loop thread likely blocked interpreter shutdown.",
+        )
+        self.assertEqual(exit_code, 0)
 
 
 # Note: the `.to(ProcessPoolExecutorConfig)` region topology's interpreter-exit teardown is