v0.6.0

SPDL v0.6.0 Release Notes

Highlights

Declarative execution regions with PipelineBuilder.to(). You can now mark a region of the
pipeline to run together in a worker pool, instead of attaching an executor to individual stages.
Every stage between .to(ProcessPoolExecutorConfig(...)) and .to(MAIN_PROCESS) is fused into one
nested pipeline that runs inside a worker process (or, on Python 3.14+, a subinterpreter), so values
handed from one stage to the next stay in the worker — they are never copied back to the main
process and need not be picklable. Only the region's inputs and outputs cross the boundary. Unlike
per-stage executors, a region can also absorb aggregate, disaggregate, and path_variants
stages, and the target is a serializable spec rather than a live Executor, so a pipeline remains
expressible as static config
(#1584,
#1585,
#1586,
#1587).

from spdl.pipeline import PipelineBuilder
from spdl.pipeline.defs import MAIN_PROCESS, ProcessPoolExecutorConfig

pipeline = (
    PipelineBuilder()
    .add_source(src)
    .to(ProcessPoolExecutorConfig(max_workers=4))
    .pipe(op1, concurrency=2)      # runs in a worker process
    .aggregate(32)                 # runs in a worker process
    .pipe(op2, concurrency=3)      # runs in a worker process
    .to(MAIN_PROCESS)              # data returns to the main process
    .add_sink(4)
    .build(num_threads=4)
)

Note that each stage's concurrency applies within each worker, so effective concurrency is
concurrency × max_workers — size the two together.

Reproducible randomness in worker-based execution. Switching a pipeline between in-process
multi-threading and worker-based execution used to silently change the behavior of the global
RNGs (random, numpy, torch) used inside preprocessing functions: with fork the worker
continued the parent's stream, with spawn/forkserver it seeded from OS entropy. SPDL now
captures the main process' global RNG state at build time and restores it in the worker before
iteration, with no user opt-in — matching the torch.utils.data.DataLoader contract.
numpy/torch state is captured only if the program already imported them, so spdl.pipeline
stays pure Python (#1556).

import random

random.seed(0)

# Draws inside the pipeline's ops now continue from the seeded main-process
# state, regardless of the multiprocessing start method.
for item in run_pipeline_in_subprocess(config, num_threads=4):
    ...

For run_pipeline_in_subinterpreter, only the stdlib random state is copied — NumPy/PyTorch state
objects are not shareable across interpreters, and neither library can be imported in a
subinterpreter at all.

Pipeline iterators are now single-use (one epoch). Pipeline.get_iterator() has always been
documented as covering exactly one epoch, but the old iterator's reuse behavior depended on the
source: with a continuous source, reusing an exhausted iterator silently resumed into the next
epoch; with a finite source it yielded nothing. That mode-dependent behavior is gone — an exhausted
iterator stays exhausted in both modes. See BC-Breaking Changes below
(#1549).

Shared-memory arenas throttle the producer instead of failing it. SharedMemoryRingBuffer and
SharedMemorySegmentPool now park the producer on a process-shared condition variable when the
arena is full and wake it from the consumer's release path, rather than raising immediately (ring)
or busy-polling every 0.5 ms (pool). Measured with no throughput regression
(#1534).

BC-Breaking Changes

• Pipeline.get_iterator() returns a single-use, one-epoch iterator
  (#1549): once the iterator reaches the epoch
  boundary it stays exhausted and yields nothing on reuse, regardless of whether the source is
  continuous. Previously, code that reused one iterator instance across epochs with a continuous
  source advanced into the next epoch. Call get_iterator() once per epoch (or just re-iterate the
  pipeline).

  # Before (v0.5.0) — worked only with a continuous source
  it = pipeline.get_iterator()
  for epoch in range(num_epochs):
      for item in it:      # silently resumed into the next epoch
          ...
  
  # After (v0.6.0)
  for epoch in range(num_epochs):
      for item in pipeline.get_iterator():   # or: for item in pipeline:
          ...

• Shared-memory arenas block on a full arena instead of raising
  (#1534):
  SharedMemoryRingBuffer.write_binary used to raise BufferError the moment an in-progress unit
  did not fit. It now waits up to the new acquire_timeout for the consumer to free space. Pass
  acquire_timeout=0 to restore the previous raise-immediately behavior.

  from spdl.pipeline import SharedMemoryRingBuffer
  
  # Before (v0.5.0): full arena -> immediate BufferError
  ring = SharedMemoryRingBuffer(capacity=1 << 30)
  
  # After (v0.6.0): full arena -> block, then raise BufferError on timeout
  ring = SharedMemoryRingBuffer(capacity=1 << 30)                     # blocking (default)
  ring = SharedMemoryRingBuffer(capacity=1 << 30, acquire_timeout=0)  # legacy behavior

  Both arenas also gain a sticky shutdown flag: shutdown_arena() flips it and broadcasts, so a
  parked producer exits with a clean BufferError at teardown instead of waiting out the full
  timeout. Fast-fail guards remain for the two unsatisfiable cases (a single binary larger than
  capacity, and an in-progress unit whose accumulated bytes exceed capacity).

New Features

• Batched routing in path_variants (#1595):
  with batched=True, the router receives a whole batch and returns one path index per element; the
  batch is partitioned into per-path sub-batches (order preserved), each path processes its
  sub-batch as a list, and the merge concatenates them back into one batch. This amortizes the
  router and fan-out/fan-in cost over the batch instead of paying it per item — useful for bulk
  cache hit/miss splits. Defaults to False, so existing behavior is byte-identical.

  from spdl.pipeline.defs import Aggregate, PathVariants, Pipe
  
  pipes = [
      Aggregate(64),
      PathVariants(
          router=lambda batch: [0 if x in cache else 1 for x in batch],
          paths=[
              [Pipe(load_batch_from_cache)],   # path 0: cache hits
              [Pipe(load_batch_from_source)],  # path 1: cache misses
          ],
          batched=True,
      ),
  ]

  Each path op receives and returns a list, may drop elements by returning a shorter list, and
  must tolerate an empty input list.

• Stdlib executors are supported in run_pipeline_in_subprocess
  (#1548,
  #1550): a pipe stage carrying a
  ThreadPoolExecutor, ProcessPoolExecutor, or (Python 3.14+) InterpreterPoolExecutor can now
  be shipped to the pipeline subprocess. Thread/interpreter pools are lazily reconstructed inside
  the subprocess; process pools are spawned in the main process and driven through a queue-backed
  proxy, so their workers are siblings of the pipeline subprocess and can never be orphaned if it is
  force-killed.

  from concurrent.futures import ProcessPoolExecutor
  
  from spdl.pipeline import run_pipeline_in_subprocess
  from spdl.pipeline.defs import Pipe, PipelineConfig, SinkConfig, SourceConfig
  
  config = PipelineConfig(
      src=SourceConfig(paths),
      pipes=[Pipe(decode, executor=ProcessPoolExecutor(max_workers=4))],
      sink=SinkConfig(buffer_size=8),
  )
  for item in run_pipeline_in_subprocess(config, num_threads=4):
      ...

  The executor must be freshly constructed — one that has already spawned workers or run work is
  rejected with ValueError. max_workers / initializer / initargs are preserved; mp_context
  and max_tasks_per_child are not honored on the hoisted process pool.

• acquire_timeout on SharedMemoryRingBuffer
  (#1534): bounds how long a producer waits
  for the consumer to free space, guarding against a stalled or dead consumer. Already present on
  SharedMemorySegmentPool.

Bug Fixes

• Fix a use-after-free in spdl.io.load_npz: NpzFile stored only a raw pointer into the input
  buffer, so the backing memory could be freed while entries were still being read. The archive is
  now kept alive by the NpzFile, and uncompressed entries keep it alive for the array's own
  lifetime (#1607).
• Fix a hang at interpreter exit for programs that hold a Pipeline reference and never call
  stop(). The pipeline's non-daemon event-loop thread was joined by CPython before the
  weakref.finalize cleanup ran; a shutdown hook registered via threading._register_atexit now
  stops the pipeline first (#1591).
• Fix a stage failure being silently swallowed when an external cancellation raced the stage's
  finalization — the cancellation replaced the in-flight exception, so no PipelineFailure was
  raised. Stage-hook finalization is now shielded from cancellation
  (#1572).
• Fix a deadlock when add_source(..., continuous=True) was combined with a path_variants stage:
  the epoch-end sentinel was passed to the user router and delivered to a single path, so the fan-in
  barrier blocked forever. The sentinel is now broadcast to all paths and collapsed by the merge
  (#1547).
• Fix a Windows crash (ModuleNotFoundError: No module named '_posixsubprocess') in shared-memory
  arena workers, caused by an unconditional resource_tracker.unregister() call on attach. This
  also fixes the resulting parent-side hang
  (#1544).
• Stop logging an AssertionError when a pipeline is built inside a daemon process:
  ProcessGroupStatsMonitor cannot spawn its /proc reader there, and a per-worker monitor would
  be redundant since the main-process monitor already sums over the process group. It now logs a
  single warning and returns (#1590).

Other Changes

• CUDA wheels are now built against CUDA 13 on Linux and Windows
  (#1539,
  #1540,
  #1538).
• Upgrade bundled fmt to 11.0.2, fixing a build failure on recent AppleClang where fmt 10.1.1's
  consteval format-string validation is rejected as non-constant
  (#1562,
  #1561).
• Add a missing <cstdint> include in numpy_support.h
  (#1609).

Documentation

• New Execution Models guide, with the parallelism and GIL material reorganized around it
  (#1593,
  #1592).
• New case study on the cost of inter-process communication in data loading, with a benchmark that
  reproduces the 20+ second startup stall from shipping a 1.2M-path ImageNet dataset to PyTorch
  DataLoader workers (#1594).
• New from-scratch pipeline authoring guide (tools/skills/authoring/building_pipelines.md), plus
  clarification that async functions go straight to .pipe() and must not be wrapped in
  asyncio.run(), and that auto_stop() is obsolete
  (#1583,
  #1578).
• Document that continuous sources have been supported in run_pipeline_in_subprocess since v0.4 —
  the previously assumed continuous=False workaround reintroduces a per-epoch rebuild gap
  (#1546).
• Add free-threaded Python (3.14t) results to the WAV benchmark: soundfile scales with thread
  count instead of degrading (>10x at 16 threads), while spdl.io.load_audio, which already
  releases the GIL, is essentially unchanged
  (#1537).