Orchestration and Data Infrastructure

A distributed training program is not a run until an operational system can start it, identify its state, recover it, and prove what it consumed. The control plane holds the desired run and reconciles it with the observed run. The data plane supplies identified training examples under a declared order and mixture contract. Checkpointing joins them: it commits model state and data progress at one logical boundary so recovery does not combine incompatible histories. The practical questions are how often to save, how data order and coverage are defined, what survives a restart, and how operators distinguish a slow step from a damaged run.

These mechanisms have separate histories. Young derived a first-order checkpoint-interval approximation in 1974, long before accelerator training (Young 1974). Cluster managers such as Borg, Omega, and Kubernetes later made desired-state reconciliation a standard control-plane pattern (Burns et al. 2016). Machine-learning input runtimes such as tf.data then treated loading, transformation, parallelism, buffering, and iterator recovery as one dataflow problem (Murray et al. 2021). Modern training infrastructure combines those ideas, but each run must still declare its own semantics and failure budget.

This chapter specifies how checkpointing, the data plane, and observability fit inside an orchestration protocol with explicit identities, state transitions, commit rules, and recovery objectives. The parallel algorithms being controlled come from Chapter 10; the accelerators and interconnect are described in Chapter 62; compiler and kernel artifacts come from Chapter 64; and source selection and mixture design begin in Chapter 6.

A run is a reconciled state machine

The durable input to orchestration is an immutable run specification, not a shell command remembered by one scheduler process. It should bind at least:

The control plane keeps a durable run record beside this specification. The record contains the current run generation, state, admission decision, worker membership, last progress heartbeat, latest committed checkpoint, data cursor, failure history, and terminal result. A controller repeatedly compares the desired run with the observed run and attempts to reconcile the difference. Retries are safe only when controller effects are idempotent or guarded by a unique operation identifier.

Figure 65.1 shows the state machine. “Running” is not the absence of an error; it means membership, health, and progress invariants hold for the current generation.

Leases are not enough without fencing

A worker may pause, lose its lease, and later resume after a replacement has started. Stopping its heartbeat does not stop its process. Every effect that can mutate shared state therefore carries a monotonically increasing fencing token, usually the run generation. Checkpoint publication, progress updates, and final artifact writes are accepted only when the token equals the generation in the durable run record. A stale worker can still compute, but its writes are rejected.

The same rule applies to cancellation. The controller first records a cancelled terminal state and advances or revokes the generation, then asks workers to stop. Cleanup can be retried because outputs are generation-scoped and garbage collection ignores committed artifacts.

Scheduling and worker membership are separate contracts

Synchronous training commonly benefits from gang scheduling: the scheduler admits the resources for a worker group together so a partial allocation does not sit idle waiting for the rest. Gang scheduling is an admission policy, not a property of a collective. Once launched, a collective blocks because every member of its process group must issue compatible operations. Topology-aware placement adds another condition: enough devices may be free in aggregate while no valid placement fits the required local or scale-up domain (Gu et al. 2019; Kubernetes SIG Scheduling 2026).

The run specification must choose a membership contract:

• A fixed world size treats any lost member as a group failure. The controller aborts communication, replaces or requeues capacity, rebuilds the same-sized group, and restores a checkpoint.
• An elastic range permits a membership change, but only if model sharding, optimizer state, global-batch semantics, data assignment, and schedules all support the new size.
• Role-specific or local repair is a separate system capability. It must state which communicator and state can be rebuilt without restarting the rest.

For example, TorchElastic stops all surviving workers after a worker failure or membership change and forms a new group; it does not hot-swap one rank into a live collective. The system must restart the worker group, rendezvous again, and restore committed state. Rank is not a stable identity across that rendezvous, and an elastic world-size change can assign both a new rank and a new world size (PyTorch Contributors 2026). Stable sample ownership, checkpoint keys, and output identity must therefore use logical identifiers rather than a rank number alone.

A checkpoint is a distributed commit

A parameter file is useful for inference, but it is not necessarily a training checkpoint. A recoverable checkpoint captures every state variable that changes future updates:

The cleanest snapshot boundary is after a global optimizer step has committed and before the next step consumes data. At that step boundary, no partial gradient accumulation remains. If a system saves mid-step, it must also persist the accumulated gradients, microbatch cursor, pipeline state, and any pending collective semantics.

Sharding changes where bytes live, not the consistency rule. Each rank or storage worker may write part of the state, but all parts must describe one checkpoint generation and one logical step. PyTorch Distributed Checkpoint, for example, coordinates distributed state and can load supported sharded tensors into a different layout; custom state still has to supply its own compatible save and restore contract (PyTorch Contributors 2026).

Publish the manifest last

Figure 65.2 gives a storage-independent commit protocol. The word “atomic” refers to checkpoint visibility: a reader sees the previous complete generation or the new complete generation, never a mixture.

One concrete protocol is:

1. Allocate a new checkpoint generation under the current fencing token.
2. At one step boundary, stage an immutable copy of the local state. Training must not mutate the bytes being persisted.
3. Write every temporary shard with its logical keys, byte count, and checksum.
4. Verify all expected shards and the global key coverage. A checksum detects storage or transfer damage; it cannot prove that a tensor was numerically correct before hashing.
5. Publish one immutable manifest and commit record last. Update “latest” only after that record is durable and only if the fencing token is current.
6. Restore only committed manifests. Treat every incomplete checkpoint as staging debris and remove it through bounded garbage collection.

The manifest is the transaction boundary. Directory listing, the presence of one rank's file, or a timestamp is not evidence that a checkpoint is complete. The consistent-global-state problem predates machine learning; Chandy and Lamport's distributed snapshot work formalized why locally plausible pieces do not automatically form one valid global state (Chandy and Lamport 1985).

Cadence is a model with assumptions

Young's first-order model balances visible checkpoint cost against expected recomputation (Young 1974). Here, including detection and restoration as terms that do not change the first-order optimum, the expected waste fraction is

$$W(T)=\frac{C}{T}+\frac{T}{2M}+\frac{R+D}{M},$$

and the minimizing interval is

$$T_{\text{Young}}=\sqrt{2CM}.$$

where:

• $W(T)$ is the expected fraction of elapsed time lost to checkpoint exposure, replay, detection, and recovery;
• $T$ is useful computation time between completed checkpoints, in seconds;
• $C$ is the measured exposed cost of one blocking checkpoint, in seconds;
• $M$ is mean time between interruptions for the synchronized job, in seconds;
• $R$ is rendezvous, restore, and validation time per interruption, in seconds;
• $D$ is detection, fencing, and requeue delay per interruption, in seconds;
• $T_{\text{Young}}$ is the first-order minimizing interval.

The model assumes a long-running job, independent failures with an exponential interruption-time distribution, constant costs, successful restart, and a failure uniformly distributed within each interval. It also assumes a blocking checkpoint, $C\ll T\ll M$, and no failure while checkpointing. Daly derived a higher-order Poisson model rather than turning the approximation into a universal law (Daly 2006). A measured job-level interruption rate should replace a sum of component rates when software failures, preemption, storage incidents, or correlated failure violate independence.

The runnable uses only the Python standard library. With a 30-second exposed cost and a six-hour job MTBF, the first-order interval is about nineteen minutes.

from math import sqrt

checkpoint_cost_s = 30.0
job_mtbf_s = 6 * 60 * 60
detect_and_restore_s = 180.0

young_s = sqrt(2 * checkpoint_cost_s * job_mtbf_s)
print(f"Young interval: {young_s / 60:.2f} minutes")

for interval_min in (5, 10, 20, 40, 80):
    interval_s = interval_min * 60
    waste = (
        checkpoint_cost_s / interval_s
        + interval_s / (2 * job_mtbf_s)
        + detect_and_restore_s / job_mtbf_s
    )
    print(f"{interval_min:>2} min -> expected waste {100 * waste:5.2f}%")

Asynchronous checkpointing changes the measurement, not the obligation. It first stages immutable state, often into host memory, then drains it in the background. The staging pause and training interference belong in $C$; the background duration creates durability lag between the staged step and the latest recoverable step. Host-memory use, network traffic, and storage traffic remain real costs. A bounded writer backlog is essential: if a new snapshot arrives faster than the writer drains the old one, the system must delay or skip the new snapshot under a declared policy rather than allocate buffers without limit. CheckFreq demonstrates a two-phase, resumable design in its evaluated settings, not a free write (Mohan et al. 2021).

Sharded checkpointing can reduce bytes handled by one writer and exploit parallel storage bandwidth, but total durable bytes remain the full logical state. Scaling stops when storage bandwidth, metadata, network contention, or a coordinator becomes the bottleneck. Cadence must therefore consider state size, step duration, staging memory, sustained and burst bandwidth, checkpoint failure, writer lag, and retention, not just hardware MTBF.

Storage tiers implement recovery objectives

A tier is useful only relative to a protected failure domain. Local memory may survive a worker-process crash but not host loss. A peer copy may survive one host loss but not a rack outage. Remote durable storage may protect against a cluster loss yet still share a regional, credential, metadata, or control-plane failure.

The policy starts from two objectives:

• Recovery point objective (RPO): the maximum acceptable gap between the current committed training step and the latest recoverable step.
• Recovery time objective (RTO): the maximum acceptable time from an interruption to validated resumed progress.

Replication, retention, and cadence must meet the RPO across the named failure domains; admission, diagnosis, replacement, load bandwidth, and validation must meet the RTO. Recent multi-level systems explore different tradeoffs, including in-memory copies (Wang et al. 2023), but the design space is not a binary contest. A restore drill under the intended failure domain is stronger evidence than a successful write benchmark.

Recovery begins with classifying the fault

Checkpoint/restart repairs lost process state only when the fault's cause has been removed or bypassed. Different failures require different action:

silent data corruption (SDC) lets the run continue while producing subtly wrong numbers without an explicit crash; it can also cause a nonfinite spike, a different optimization trajectory, or persistent divergence (Ma et al. 2025). Ordinary checkpointing neither detects nor corrects silent data corruption and may faithfully preserve it. Recovery is possible when independent detection occurs within the rollback window and a retained known-good checkpoint still exists. Checksums provide byte integrity; semantic or numerical SDC requires end-to-end integrity checks such as replicated computation, invariant checks, active hardware diagnostics, or evaluated canary steps.

A fixed-membership recovery normally follows this order:

1. Detect and classify lack of progress or invalid state.
2. Abort communicators, stop the worker group, and fence its run generation.
3. Capture diagnostics and quarantine a suspect node, image, data shard, or storage generation.
4. Allocate and preflight replacement capacity, or requeue with bounded backoff. Warm spare capacity is an optional cost policy, not a scheduler invariant.
5. Rendezvous, rebuild the role and rank map, and choose the latest committed, validated checkpoint within the rollback window.
6. Restore or supportedly reshard every state family, restore the data cursor, and validate one or more steps before marking the run healthy.

The data plane begins with immutable identity

A path such as corpus/latest is a location, not a dataset identity. Before launch, resolve mutable aliases into an immutable dataset manifest. The manifest should record:

• a dataset and schema version;
• each object's immutable version or digest, byte size, record count, and logical range;
• a stable sample identifier or derivation rule;
• format, compression, split, filtering, deduplication, and provenance;
• tokenizer version, vocabulary digest, transforms, and packing policy;
• mixture policy, replacement and source-exhaustion behavior;
• the access policy and authorization scope used to read the data.

The dataset ID can be the digest of a canonical manifest. Cache keys should use content identity rather than a mutable URI, and a cache fill should verify the digest before publication. General provenance models represent artifacts, executions, and derivations explicitly; that same lineage is what lets a later checkpoint name the exact data and transformation graph it consumed (Moreau and Missier 2013).

Logical sample assignment and physical delivery are separate. The logical plane maps global positions to samples, transformations, packed sequences, and data-parallel replicas. Tensor-, sequence-, context-, and pipeline-parallel ranks may cooperate on the same logical sample. The physical plane chooses object placement, remote reads, node caches, parallel decoding, and prefetch. Cache hits and worker completion order may change latency, but they should not silently change the declared logical sequence.

Map-style and streaming are access interfaces, not memory-size labels. A map-style dataset is key- or index-addressable even when its bytes live on remote storage. An iterable dataset is useful when random access is unavailable or expensive. Sequential archive formats are one practical streaming design (Aizman et al. 2020), while tf.data shows how interleave, parallel mapping, caching, and prefetch form a composable input dataflow (Murray et al. 2021).

Throughput is a bounded-queue contract

Here the minimum average supply rate needed to hide the input pipeline is

$$R_{\text{req}}=\frac{N_{\text{tok}}}{T_{\text{step}}},$$

where:

• $R_{\text{req}}$ is required delivered throughput in loss-bearing tokens per second;
• $N_{\text{tok}}$ is the number of loss-bearing tokens consumed by one global optimizer step;
• $T_{\text{step}}$ is the target duration of that optimizer step in seconds.

Meeting the average is not sufficient when service time has a long tail. Track time-to-batch quantiles, input wait, read and decode throughput, queue depth and age, cache hit rate, retry count, and prefetch lead. Every queue needs a memory bound and backpressure policy. A producer blocks or sheds only under an explicit rule at the high watermark; adding unbounded prefetch merely converts an input stall into memory pressure and a larger replay window. Plumber's pipeline analysis illustrates why bottlenecks move among I/O, CPU transforms, memory, and parallelism rather than having one fixed “data bandwidth” answer (Kuchnik et al. 2022).

Shuffle and mixture are versioned algorithms

“Shuffle the data” does not specify a reproducible distribution. A full uniform permutation, a seeded shard permutation, and a bounded shuffle buffer have different randomness, locality, memory use, and restart state. A deterministic finite-data design can:

1. derive a versioned pseudorandom permutation over stable sample IDs from the dataset ID, epoch, and run seed;
2. define global batch $k$ as one fixed range of that logical permutation;
3. slice the global batch among data-parallel replicas while sharing samples with cooperating tensor or pipeline ranks;
4. issue physical reads in parallel with sequence numbers, then emit through a bounded reorder buffer;
5. derive stochastic-transform randomness from stable sample context rather than worker arrival order.

A true stream may instead use a seeded shard order plus a bounded shuffle buffer. That is not a uniform global permutation and should not be named as one.

Mixtures need an equally precise contract. For independent source sampling,

$$I_t\sim\operatorname{Categorical}(w_1,\ldots,w_m), \qquad \mathbb{E}[N_i(H)]=Hw_i,$$

where:

• $I_t$ is the source chosen at logical draw $t$;
• $m$ is the number of sources;
• $w_i\geq0$ is source $i$'s target probability and $\sum_{i=1}^{m}w_i=1$;
• $N_i(H)$ is the realized number of draws from source $i$ in a window of $H$ draws;
• $H$ is the declared measurement window;
• $\mathbb{E}$ denotes expectation over the sampler's random draws.

Weights give expected proportions, not exact counts. If exact quotas are required per batch or window, allocate integer quotas with a declared rounding rule, carry rounding debt, and deterministically interleave them. The policy must also define its unit: documents, examples, packed sequences, raw tokens, or loss-bearing tokens. It must state what happens when a source is exhausted: stop, repeat, renormalize, or fail. Sampling weight and loss weight are different controls. Checkpoint both target policy and realized per-source counts.

Resume semantics must be named

“Reproducible” can describe four different promises:

The last row is the distributional resume contract; it is appropriate only when the training claim permits that weaker guarantee. A world-size change may preserve coverage while changing rank-local order, and some samplers can derive an exact global sequence independently of physical worker count. The chosen contract belongs in the run specification and checkpoint schema.

The durable cursor advances only after the optimizer step that consumed those examples commits. A producer may have fetched and transformed later examples, but those prefetched items are uncommitted. After failure they are discarded or reconstructed from the committed cursor unless the queue itself is part of the checkpoint. Physical bytes may be read twice; the goal is that recovered training state reflects each logical position according to the declared duplicate and skipped-sample policy.

Figure 65.3 connects the model and data commits. Rolling the model back without rolling the cursor back creates a skip; rolling the cursor back farther than the model creates a duplicate contribution.

Sampler state may be compact when order is a pure function of dataset ID, algorithm version, seed, epoch, and cursor. Stateful shuffle buffers, adaptive mixtures, randomized transforms, and sequence packers require more. In particular, a document cursor is insufficient when a partially filled packed sequence, boundary mask, or residual document will affect the next batch.

Observability verifies each contract

Loss alone cannot localize an operational fault, and a smooth loss curve does not prove data continuity or hardware correctness. Telemetry should retain the run ID, generation, global step, data identity, topology, and software versions needed to join evidence across systems.

Per-rank distributions matter more than a cluster average: one straggler can hide inside a normal mean while setting the collective's pace. A trace should connect admission, input, compute, collective, checkpoint, and recovery spans. Alerts need an owner, threshold or invariant, severity, and runbook action. “No alert” means no configured detector fired; it is not evidence that silent corruption was impossible.

Operational efficiency should include failures rather than reporting only steady-state tokens per second. One useful measure is

$$\operatorname{ETTR}=\frac{T_{\text{productive}}}{T_{\text{allocated}}},$$

where $T_{\text{productive}}$ is elapsed time spent on accepted optimizer steps and $T_{\text{allocated}}$ includes productive time plus checkpoint exposure, replay, recovery, stalls, and idle allocated capacity. Effective useful throughput is accepted loss-bearing tokens divided by end-to-end elapsed time. Cost per accepted token should also charge replay, optional spare capacity, storage, network, and control-plane use.

Operating the run lifecycle

A long run should pass the same recovery contract at small scale before it receives the full allocation:

1. Freeze the specification. Resolve code, images, model and optimizer state, data and tokenizer manifests, topology, resume semantics, RPO/RTO, budgets, retry classes, and cancellation authority.
2. Preflight the path. Verify device and link health, software compatibility, collectives, storage credentials, checkpoint bandwidth, data access, cache integrity, and telemetry on the intended topology.
3. Run a canary. Exercise representative shapes, input transforms, packing, mixture accounting, checkpoint staging and commit, evaluation, and output publication.
4. Use fault injection. Kill a worker during compute and checkpointing; create a hang, straggler, storage timeout, truncated shard, stale worker, missing data object, and supported topology change. Verify fencing, bounded retry, quarantine, fallback, and cursor continuity.
5. Perform a restore drill. Restore the latest and an older generation into every supported layout; reject incomplete or corrupted manifests and measure RPO and RTO.
6. Gate continuous execution. Alert on progress, per-rank tails, input wait, checkpoint lag, mixture deviation, corruption evidence, budget burn, and repeated failure signatures. Escalate deterministic repeats instead of creating a retry storm.
7. Exercise cancel and completion. Fence stale writers, stop workers, preserve committed evidence, release resources, and run bounded garbage collection for temporary shards and caches.
8. Publish a post-run manifest. Bind the terminal state, accepted model artifacts, full provenance, data and token counts, evaluations, incidents, cost, and retained recovery artifacts.

Lower-layer constraint

The operational plane cannot create state that the training program does not expose. If the framework omits a gradient scaler, RNG stream, packer residual, or logical tensor identity, the checkpoint cannot reconstruct it. If the parallelism scheme cannot reshard, the scheduler cannot safely claim elasticity. If storage, host memory, or the interconnect cannot absorb staging and data traffic without interfering with collectives, asynchronous work still lowers training goodput. Those limits come from Chapter 10 and Chapter 62; this chapter turns them into measurable admission, commit, and recovery rules.

The next chapter, Chapter 66, follows the same system downward and forward: it asks which bandwidth, packaging, and fabric constraints are becoming the limiting resources for future runs.

Further reading