Runtime and tuning

How a pipeline executes, what it reports while running, and what it checks before it starts. The guide covers these in Inspecting and Tuning Pipelines and Distributed Training.

Runtime options

Runtime options for Pipeline.options() (RuntimeOptions, IpcTransport).

class RuntimeOptions(runner=None, run_id='default_run_id', per_stage_runner=<factory>, mp_context=<multiprocessing.context.SpawnContext object>, worker_allocation='fit_to_ops', max_workers=None, deterministic=True, prefetch_batches=None, default_stage_prefetch=None, per_stage_prefetch=<factory>, op_queue_capacity=None, ipc_transport='socketpair', ipc_buffer_bytes=524288, mixture_config=None, io_options=<factory>, allow_latency_flush_in_deterministic=True, flush_every_k_chunks=None, shutdown_mode='graceful', auto_validation='strict', mtp_mode=False, mtp_buffer=None, mtp_buffer_bytes=8388608, mtp_prefetch=None, mtp_auto_checkpoint=True, world_size=1, global_rank=0, dp_degree=None, dp_group_id=None, canonical_replicas=None, mapping_strategy=None, shm_enabled=True, shm_coalesce=True, shm_min_item_bytes=4096, shm_min_new_allocation_bytes=1048576, shm_min_forward_bytes=131072, shm_compact_above_ratio=8.0, shm_compact_min_savings_bytes=16777216, shm_max_coalesced_bytes=67108864, max_worker_retries=3, aggregate_dir=None, aggregate_timeout_s=180.0, aggregate_serializer='msgpack', aggregate_compressor='zstd', execution_tracking=ExecutionTrackingMode.OFF, metrics_sink_config=None)

Bases: object

User-tunable knobs that influence how the engine constructs runners.

Numeric tuning knobs default to None, resulting in sensible defaults via resolve_* helpers that form a dependency cascade. Each resolver calls the ones above it when its own field is unset:

max_workers              ← os.cpu_count(), clamped [4, 16]
    └── prefetch_batches             = max(8, 2 × max_workers)
            ├── op_queue_capacity        = max(8, prefetch_batches)
            ├── mtp_buffer               = max(4, prefetch_batches // 2)
            │       └── mtp_prefetch         = max(4, mtp_buffer // 10)
            └── default_stage_prefetch   = max(4, prefetch_batches // 4)  [process]
                                           max(8, prefetch_batches // 2)  [threads/inline]

For example, max_workers=24 naturally bumps the prefetch / queue / MTP buffers without the user having to set them too, while prefetch_batches=64 overrides the tail buffer alone and lets the rest auto-derive from it.

By default, max_workers is derived from the host’s CPU count, but is clamped to a range of [4, 16]. For any large deployment this will hit the upper limit, but the clamp allows Zephon to run reasonably on small machines as well.

runner: str | None = None

run_id: str = 'default_run_id'

per_stage_runner: dict[int, str]

mp_context: Any = <multiprocessing.context.SpawnContext object>

worker_allocation: Literal['fit_to_ops', 'per_stage_fixed', 'global', 'autotune'] = 'fit_to_ops'

max_workers: int | None = None

deterministic: bool = True

prefetch_batches: int | None = None

default_stage_prefetch: int | None = None

per_stage_prefetch: dict[int, int]

op_queue_capacity: int | None = None

ipc_transport: Literal['socketpair', 'pipe'] = 'socketpair'

ipc_buffer_bytes: int = 524288

mixture_config: MixtureReadConfig | None = None

io_options: StoreOptions

allow_latency_flush_in_deterministic: bool = True

flush_every_k_chunks: int | None = None

shutdown_mode: Literal['graceful', 'hard'] = 'graceful'

auto_validation: Literal['strict', 'warn', 'off'] = 'strict'

mtp_mode: bool = False

mtp_buffer: int | None = None

mtp_buffer_bytes: int = 8388608

mtp_prefetch: int | None = None

mtp_auto_checkpoint: bool = True

world_size: int = 1

global_rank: int = 0

dp_degree: int | None = None

dp_group_id: int | None = None

canonical_replicas: int | None = None

mapping_strategy: Literal['contiguous', 'interleaved'] | None = None

shm_enabled: bool = True

shm_coalesce: bool = True

shm_min_item_bytes: int = 4096

shm_min_new_allocation_bytes: int = 1048576

shm_min_forward_bytes: int = 131072

shm_compact_above_ratio: float | None = 8.0

shm_compact_min_savings_bytes: int = 16777216

shm_max_coalesced_bytes: int | None = 67108864

max_worker_retries: int = 3

aggregate_dir: str | None = None

aggregate_timeout_s: float = 180.0

aggregate_serializer: str = 'msgpack'

aggregate_compressor: str = 'zstd'

execution_tracking: ExecutionTrackingMode = 'off'

metrics_sink_config: MetricsSinkConfig | None = None

IpcTransport: Literal['socketpair', 'pipe']

Transport for the process runner’s and MTP’s IPC queues.

Observability

Observability configuration and helpers used across Zephon.

class ExecutionTrackingMode(*values)

Bases: str, Enum

Controls how much execution detail is collected.

OFF = 'off'

STAGES = 'stages'

NODES = 'nodes'

classmethod from_value(value)

Return type:

ExecutionTrackingMode

property collects_nodes: bool

property collects_stages: bool

class FetchStageSummary(index, totals=<factory>, shard_totals=<factory>)

Bases: object

Stage fetch timings keyed by (dataset_id, shard_id).

index: int

totals: FetchTimingTotals

shard_totals: MutableMapping[tuple[int, int], FetchTimingTotals]

apply(delta)

merge(other)

copy()

Return type:

FetchStageSummary

to_records(*, plan_id, stage_name, tracking_mode, include_shards=True)

Return type:

list[dict[str, Any]]

class FetchTimingDelta(stage_index, dataset_id, shard_id, samples, group_ns, resolve_ns, open_ns, read_ns, close_ns, retries, cache_hits, cache_misses, shard_reopens=0)

Bases: object

Incremental fetch metrics emitted per shard group.

stage_index: int

dataset_id: int

shard_id: int

samples: int

group_ns: int

resolve_ns: int

open_ns: int

read_ns: int

close_ns: int

retries: int

cache_hits: int

cache_misses: int

shard_reopens: int

class FetchTimingSummary(plan_id=None, reporting_interval_s=None, tracking_mode=ExecutionTrackingMode.OFF, stages=<factory>)

Bases: object

Aggregated fetch metrics for an entire pipeline.

plan_id: str | None = None

reporting_interval_s: float | None = None

tracking_mode: ExecutionTrackingMode = 'off'

stages: MutableMapping[int, FetchStageSummary]

apply(delta)

merge(other)

iter_stages()

Return type:

Iterator[FetchStageSummary]

to_records(stage_names=None, *, include_shards=True)

Return type:

list[dict[str, Any]]

clone()

Return type:

FetchTimingSummary

has_samples()

Return type:

bool

class FetchTimingTotals(samples=0, groups=0, group_ns=0, resolve_ns=0, open_ns=0, read_ns=0, close_ns=0, retries=0, cache_hits=0, cache_misses=0, shard_reopens=0)

Bases: object

Aggregated timing totals accumulated across deltas.

samples: int

groups: int

group_ns: int

resolve_ns: int

open_ns: int

read_ns: int

close_ns: int

retries: int

cache_hits: int

cache_misses: int

shard_reopens: int

apply(delta)

merge(other)

copy()

Return type:

FetchTimingTotals

property avg_group_ns: float

property avg_resolve_ns: float

property avg_open_ns: float

property avg_read_ns: float

property avg_close_ns: float

property cache_hit_ratio: float

class MTPQueueStats(depth, capacity, staged_bytes, prefetch_depth)

Bases: object

Point-in-time occupancy of the MTP hand-off queue (data_q).

depth counts queued items, which may not have been pickled yet; staged_bytes is the already-pickled data in the kernel transport buffer, readable without the feeder running. Zero staged with depth near capacity means the feeder is starved. prefetch_depth counts items already drained into the main-process prefetch buffer; it is sampled without locking and may be slightly stale.

A field is -1 when unmeasurable: depth on macOS (sem_getvalue is unsupported there), staged_bytes once the queue is closed.

depth: int

Items put by the subprocess but not yet retrieved by the consumer.

capacity: int

Maximum depth — the resolved mtp_buffer.

staged_bytes: int

Pickled bytes staged in the transport buffer, consumer-ready.

prefetch_depth: int

Deserialized items in the prefetch buffer (0 when mtp_prefetch=0).

class MetricsSinkConfig(mode=MetricsSinkMode.LOG, namespace='zephon.pipeline', statsd_host='127.0.0.1', statsd_port=8125, default_tags=<factory>, json_logs=True, max_batch_size=32, flush_interval_s=5.0)

Bases: object

Describe how observability data is exported.

mode: MetricsSinkMode = 'log'

namespace: str = 'zephon.pipeline'

statsd_host: str = '127.0.0.1'

statsd_port: int = 8125

default_tags: Mapping[str, str]

json_logs: bool = True

max_batch_size: int = 32

flush_interval_s: float = 5.0

should_log()

Return type:

bool

should_emit_statsd()

Return type:

bool

with_additional_tags(extra)

Return type:

MetricsSinkConfig

class MetricsSinkMode(*values)

Bases: str, Enum

Where metrics should be emitted.

LOG = 'log'

STATSD = 'statsd'

BOTH = 'both'

classmethod from_value(value)

Return type:

MetricsSinkMode

includes_logs()

Return type:

bool

includes_statsd()

Return type:

bool

class PipelineSummary(plan_id=None, reporting_interval_s=None, tracking_mode=ExecutionTrackingMode.OFF, stages=<factory>, total_wait_ratio=None)

Bases: object

Aggregated metrics for an entire pipeline invocation.

plan_id: str | None = None

reporting_interval_s: float | None = None

tracking_mode: ExecutionTrackingMode = 'off'

stages: MutableMapping[int, StageSummary]

total_wait_ratio: float | None = None

apply(delta)

merge(other)

iter_stages()

Return type:

Iterator[StageSummary]

to_records()

Return type:

list[dict[str, Any]]

clone()

Return type:

PipelineSummary

compute_wait_ratios()

Derive wait ratios for the pipeline based on processing and wait times.

class PrefetchTimingDelta(stage_index, batch_size, prefetch_requests, prefetch_succeeded, prefetch_failed)

Bases: object

Incremental prefetch metrics emitted per batch.

stage_index: int

batch_size: int

prefetch_requests: int

prefetch_succeeded: int

prefetch_failed: int

class PrefetchTimingSummary(plan_id=None, reporting_interval_s=None, tracking_mode=ExecutionTrackingMode.OFF, stages=<factory>)

Bases: object

Aggregated prefetch metrics for an entire pipeline.

plan_id: str | None = None

reporting_interval_s: float | None = None

tracking_mode: ExecutionTrackingMode = 'off'

stages: MutableMapping[int, PrefetchTimingTotals]

apply(delta)

merge(other)

clone()

Return type:

PrefetchTimingSummary

has_samples()

Return type:

bool

property success_rate: float

class PrefetchTimingTotals(batches=0, samples=0, prefetch_requests=0, prefetch_succeeded=0, prefetch_failed=0)

Bases: object

Aggregated prefetch totals accumulated across deltas.

batches: int

samples: int

prefetch_requests: int

prefetch_succeeded: int

prefetch_failed: int

apply(delta)

merge(other)

copy()

Return type:

PrefetchTimingTotals

Validation

Validation harness for user-supplied operators and accumulators.

The framework guarantees deterministic parallel execution only when the callables passed to Pipeline.add_op honor several contracts that the type system cannot enforce: process_many must be stateless across calls, the accumulator must drain on flush() and reset on flush(reset=True), and so on. This module runs a synthetic-sample smoke test that catches the most common violations before the engine starts. See the “Accumulators and Operators” page in the published documentation for the contracts being checked.

The harness is intentionally a smoke test, not a proof: it cannot catch violations that only surface after many iterations or under specific data shapes. Errors here are real bugs; passing this harness is necessary but not sufficient for correctness.

Unlike the rest of zephon (custom exception subclasses + warnings.warn), this module aggregates findings into a ValidationReport of string-coded Issue records. Aggregation lets the user see every problem in their op at once; the tradeoff is identification by string equality on issue.code rather than by isinstance.

class Issue(severity, code, op_name, message, doc_link='https://datologyai.github.io/zephon/understanding/accumulators_operators.html')

Bases: object

A single validation finding.

severity: str

code: str

op_name: str

message: str

doc_link: str = 'https://datologyai.github.io/zephon/understanding/accumulators_operators.html'

exception ValidationError(report)

Bases: RuntimeError

Raised when a strict-mode validation entry point finds at least one error.

Raisers: Pipeline.validate(strict=True), strict auto-validation, and Pipeline.preflight_tokenizers(strict=True).

class ValidationReport(issues=<factory>)

Bases: object

Result of running the validation harness against a pipeline.

issues: list[Issue]

property ok: bool

True when no errors were reported (warnings are allowed).

property errors: list[Issue]

Subset of issues with severity error.

property warnings: list[Issue]

Subset of issues with severity warning.

format()

Human-readable multi-line summary suitable for raising.

Return type:

str

preflight_tokenizers(pipeline)

Return tokenizer initialization failures for a pipeline.

Each tokenizer operator is deep-copied before its lazy initialization runs, leaving pipeline-owned operators unchanged. This may download tokenizer files.

Parameters:

pipeline (Pipeline) – Pipeline to inspect.

Returns:

A ValidationReport containing one TOKENIZER_PREFLIGHT_FAILED error per failure.

Return type:

ValidationReport

Debugging

Diagnostics for a run that leaks or hangs on shutdown.

install_debug_hooks()

Install monkey-patches on resource_tracker.

This should be called early in the program, before any multiprocessing objects are created.

dump_semaphore_registry()

Print debug info about tracked semaphores.

dump_semaphore_leak_report(*, force=False)

Print the semaphore leak analysis report.

Can be called manually at any point to see current state. Also called automatically via weakref finalizer at shutdown.

Parameters:

force (bool) – If True, print even from worker processes with no leaks.