spdl.pipeline.build_pipeline

build_pipeline(pipeline_cfg: PipelineConfig[U], /, *, num_threads: int, max_failures: int | Fraction = -1, report_stats_interval: float = -1, queue_class: type[AsyncQueue] | None = None, task_hook_factory: Callable[[StageInfo], list[TaskHook]] | None = None, stage_id: int = 0, background_tasks: list[Callable[[], BackgroundTask]] | None = None, use_thread_output_queue: bool = False, fuse_subprocess_stages: bool = False) → Pipeline[U][source]

Build a pipeline from the config.

Note

If environment variable SPDL_PIPELINE_DIAGNOSTIC_MODE=1 is set, then this function builds a Pipeline in self-diagnostic mode. In self-diagnostic mode, the pipeline will call profile_pipeline function and benchmark each stage with different concurrency. Once the profiling is done, then the program exits.

See also

Example: Pipeline definitions

Illustrates how to build a complex pipeline.

profile_pipeline()

A function to profile a Pipeline stage by stage.

Parameters:

  • pipeline_cfg – The definition of the pipeline to build.

  • num_threads –

    The number of threads in the thread pool commonly used among Pipeline stages.

    Note

    If a stage in the pipeline has dedicated executor, that stage will use it.

  • max_failures – The maximum number (int) or rate (Fraction) of failures each pipe stage can have before the pipeline is halted. When an int is provided, it specifies the maximum count of failures. Setting -1 (default) disables it. When a Fraction is provided (e.g., Fraction(1, 10) for 10%), it specifies the maximum failure rate (failures / invocations).

  • report_stats_interval –

    When provided, report the pipeline performance stats every given interval. Unit: [sec]

    This is only effective if there is no custom hook or custom AsyncQueue provided for stages. The argument is passed to TaskStatsHook and StatsQueue.

    If a custom stage hook is provided and stats report is needed, you can instantiate TaskStatsHook and include it in the hooks provided to PipelineBuilder.pipe().

    Similarly if you are providing a custom AsyncQueue class, you need to implement the same logic by your self.

  • queue_class – If provided, override the queue class used to connect stages. Must be a class (not an instance) inherits AsyncQueue.

  • task_hook_factory – If provided, used to create task hook objects, given a name of the stage. If None, a default hook, TaskStatsHook is used. To disable hooks, provide a function that returns an empty list.

  • stage_id – The index of the initial stage used for logging.

  • background_tasks – Optional list of BackgroundTaskFactory callables. Each factory is called to create a BackgroundTask whose run() method runs alongside the pipeline stages. Tasks are cancelled when the pipeline completes. Their errors are logged but do not cause the pipeline to fail.

  • use_thread_output_queue – If True, replace the sink’s output queue with a queue.Queue-backed queue for the final handoff from the background event loop to the foreground consumer thread. This bypasses asyncio.run_coroutine_threadsafe, reducing per-batch latency from ~200-400us to ~10us. Default: False.

  • fuse_subprocess_stages –

    If True, fuse runs of two or more adjacent pipe stages that share the same process-pool (or interpreter-pool) executor instance into a single stage that executes the run as one nested pipeline inside a worker pool. This eliminates the inter-stage IPC that otherwise round-trips data back to this process between each stage (so intermediate values need not be picklable), while each fused stage keeps its own concurrency and per-stage stats. A path_variants stage whose branches all use the same pool executor is fused too — the whole routing construct (router and branches) moves into the worker — and fuses on its own even when it is the only such stage. An aggregate/disaggregate between two pool stages is not fused (it keeps its main-process batching) and splits them into separate runs. An async op joins a fused run when tagged with the same executor as its neighbours (see pipe()), running on the worker’s own event loop. Continuous sources are supported (the fused worker sub-pipelines stay warm across epochs and epoch boundaries are propagated across the pool). Default: False.

    Added in version 0.6.0: The fuse_subprocess_stages argument.