Training Job Checkpoints and Distributed Failure

Consider a hypothetical training job on 64 GPUs that has been running for three days. At 02:10 one node loses its network link, and the process on it stops responding. The other 63 workers are not told. Each of them is inside a collective operation, such as the gradient all-reduce, waiting for a contribution that will never arrive. They hold all 64 GPUs and make no progress until a timeout fires and tears the processes down. The scheduler then requeues the job. When it is admitted again, it loads the most recent checkpoint, which was written two hours before the failure. Two hours of work on 64 GPUs are gone, plus the time spent hung, plus the time to start again. If the previous checkpoint had been half-written when the node died, the restart would have found a corrupted file and had to go back further.

A training job is a long computation that runs on hardware that fails, so the question is not whether the job will be interrupted but how much each interruption costs and whether the restarted job is the same computation as the one that was interrupted. This article follows a training job through its life. It describes the states a job passes through and who is responsible at each step, how distributed jobs fail, what a checkpoint has to contain for a restart to be a continuation and not a different run, how to write checkpoints so that a crash cannot corrupt them, how to verify that resumption is exact, how to choose how often to checkpoint, and how to measure the waste that remains. After reading it you should be able to design a checkpoint and restart protocol for a job, test that it works, and estimate its cost.

The article follows Scheduling GPU Workloads and Managing Resource Isolation, which explained gang admission and why a preempted job loses all work since its last saved state, and Versioning Models, Data, Prompts, and Evaluation Artifacts, which treats a checkpoint as one more artifact with a lineage. It assumes you know what gradient descent with momentum does at each step, what a process signal is, and what a file system rename is. Distributed training is explained as needed: data parallelism means every worker holds a copy of the model, computes gradients on its own batch, and averages gradients with the others each step.

The experiments use a small NumPy model so that they can run on a laptop with no GPU: Python 3.12.14 and NumPy 2.2.3 on macOS 26.7.1. They test the logic of checkpointing, restarting, and failure injection. They cannot test GPU behavior, NCCL, or power loss, and where the article describes PyTorch features it relies on the PyTorch documentation (version 2.14 at the time of reading) and not on runs.

The life of a training job

A training job passes through a small number of states, and recovery is clearest when each transition has an owner.

State	What happens	Who handles it
Queued	The job waits for quota and for GPUs that can host all its workers together	Queueing layer and scheduler, as in the previous article
Starting	Workers start, find each other (rendezvous), receive ranks, load the latest valid checkpoint if there is one	Launcher, such as torchrun, and the training code
Running	Steps execute; at intervals the job writes a checkpoint	Training code
Interrupted	A worker fails, a node is lost, the job is preempted, or maintenance evicts it	Detected by the launcher, node controller, or scheduler
Restarting	The surviving workers are stopped, the group is re-formed, and the job loads a checkpoint	Launcher within its restart budget, otherwise the platform by requeueing
Finished or failed	The job completes, or exhausts its restarts and is reported as failed	Platform

The launcher in PyTorch is torchrun, which runs under the TorchElastic agent. Its documentation describes the failure behavior precisely. If workers fail in a job with n workers, then all workers are stopped and restarted, up to a limit set by --max-restarts. A failure of the agent that manages a node’s workers is a local worker-group failure, and the documentation leaves it to the job manager either to fail the entire job (gang semantics) or to try to replace the node. With elastic launching, written as --nnodes=1:4 for a range of one to four nodes, a node departure or arrival stops all workers and starts a new worker group with new RANK and WORLD_SIZE values; the documentation warns not to hard-code assumptions about WORLD_SIZE for that reason. The environment variable TORCHELASTIC_RESTART_COUNT tells a worker how many times the group has restarted. The documentation also describes the structure it recommends for a training script: load a checkpoint at the start, then train, saving the checkpoint when it is time. The launcher restarts processes, and everything about continuing the computation is the training code’s job.

That division of labor is the important one for the platform. The launcher can restart processes. The scheduler can re-admit the gang. Neither knows what the training job has computed. Only the checkpoint carries that across a restart, so the checkpoint protocol is the contract between a training job and everything underneath it.

How distributed training fails

A data-parallel training step ends with a collective operation in which every worker contributes its gradients and receives the average. All workers must reach it. This synchronous structure is what makes failures expensive: a stall in one worker stalls all workers.

**A crashed or unreachable worker.** Peers wait inside the collective. PyTorch’s init_process_group documents a timeout for operations on the group, with a default of 10 minutes for the NCCL backend and 30 minutes for other backends. After the timeout, collectives are aborted asynchronously and the process crashes, because CUDA execution is asynchronous and it is no longer safe to continue running user code after a failed NCCL operation, which might leave later CUDA operations running on corrupted data. During the wait, every GPU in the job is allocated and idle. The timeout is therefore a direct multiplier on the cost of every failure: with 64 GPUs and a 10-minute timeout, each hang burns about ten GPU-hours before recovery starts. Shortening it detects failures sooner and risks aborting legitimate long operations. The PyTorch environment variable reference lists controls around this: TORCH_NCCL_ASYNC_ERROR_HANDLING selects what happens when the watchdog observes an error (the documented default of 3 tears down the process without aborting the communicator), TORCH_NCCL_ENABLE_MONITORING enables a monitoring thread that aborts the process if the watchdog thread itself stops making heartbeats within TORCH_NCCL_HEARTBEAT_TIMEOUT_SEC (described as preventing jobs from sitting stuck and tying up cluster resources), and TORCH_NCCL_DESYNC_DEBUG and the flight recorder buffer (TORCH_NCCL_TRACE_BUFFER_SIZE) help to identify which rank diverged.

**A straggler.** A worker that is slow, not dead, delays every step, because the step ends when the slowest worker finishes. No error is raised, so no restart is triggered. A small calculation shows the scale effect. If each of n workers is independently slow on any given step with probability \( p \), the chance that at least one is slow is \( 1-(1-p)^n \), and if a slow worker takes twice as long, the expected step time is \( 1 + (1-(1-p)^n) \) times the normal step time. The program below confirms the formula by simulation.

import random, statistics
# A synchronous step finishes when the slowest of n ranks finishes.
# Each rank is independently slow (2.0x the normal step time) with probability p on any step.
def mean_step(n, p, slow=2.0, steps=20000, seed=1):
    rng = random.Random(seed)
    return statistics.mean(max(slow if rng.random() < p else 1.0 for _ in range(n)) for _ in range(steps))
print(f"{'ranks':>6}{'p(slow)':>9}{'P(>=1 slow)':>13}{'simulated step time':>21}{'formula 1+(1-(1-p)^n)':>24}")
for n in (8, 64, 512, 4096):
    for p in (0.0005, 0.002):
        q = 1 - (1 - p) ** n
        print(f"{n:>6}{p:>9}{q:>13.3f}{mean_step(n, p, steps=4000 if n >= 4096 else 20000):>21.3f}{1 + q:>24.3f}")

 ranks  p(slow)  P(>=1 slow)  simulated step time   formula 1+(1-(1-p)^n)
     8   0.0005        0.004                1.004                   1.004
     8    0.002        0.016                1.017                   1.016
    64   0.0005        0.032                1.031                   1.032
    64    0.002        0.120                1.119                   1.120
   512   0.0005        0.226                1.228                   1.226
   512    0.002        0.641                1.644                   1.641
  4096   0.0005        0.871                1.873                   1.871
  4096    0.002        1.000                2.000                   2.000

With 8 workers, per-worker slow-step probabilities of 0.05 and 0.2 percent cost 0.4 and 1.7 percent of step time. With 512 workers the same probabilities cost 23 and 64 percent, and with 4,096 workers they cost 87 percent and, at the higher probability, a full doubling, since at least one worker is slow on essentially every step. The simulation and the formula agree within 0.003 in every row. The practical consequence is that at scale a rare per-worker slowdown becomes a constant cluster-level slowdown, and the detection tooling has to find the slow worker by comparing timing across ranks, since nothing crashes.

**Failures in the surrounding system.** The storage that checkpoints are written to can be slow or unavailable, the data source can stall, and the scheduler can preempt the job. Kubernetes terminates a Pod by sending the container’s main process a TERM signal, waiting for a grace period (30 seconds by default), and then sending KILL. A job can use that window to write an emergency checkpoint, but the window is short compared with the time to write a large model’s state, and a checkpoint that is not finished when the KILL arrives is a torn checkpoint, which the next section addresses.

**How often this happens.** The Llama 3 technical report gives a public data point for a very large job. For a 54-day snapshot of pre-training on 16,000 GPUs, it reports 466 job interruptions, of which 47 were planned maintenance and 419 were unexpected; about 78 percent of the unexpected ones were attributed to confirmed or suspected hardware issues, GPU problems being the largest category. It notes that because training is synchronous, a single GPU failure may require restarting the entire job, and that the team achieved more than 90 percent effective training time, with automation handling nearly all interruptions. Dividing 54 days (1,296 hours) by 419 gives about one unexpected interruption every 3.1 hours for that job. The number belongs to that system and scale, and it should not be carried over to a different cluster. It does show the order of magnitude to plan for when a job spans thousands of GPUs: failures are routine events, and recovery speed determines the efficiency of the whole run.

What a checkpoint must contain

A checkpoint is useful only if resuming from it continues the same computation. The model weights are the obvious part, and they are far from enough. A training step is a function of the whole training state: the parameters, the optimizer’s internal buffers, the position in the learning-rate schedule, the state of every random number generator the step consumes, and the position in the data. Losing any one changes what the next step computes.

The trainer used in this article is deliberately small. It is a 10-16-1 network with a tanh hidden layer, dropout on the hidden layer, SGD with momentum, a learning rate that halves every 200 steps, and mini-batches drawn in an order determined by the epoch. Its complete state is written out explicitly:

def fresh_state():
    r = np.random.default_rng(INIT_SEED)
    p = {"W1": r.normal(size=(D, H)) * 0.3, "b1": np.zeros(H), "W2": r.normal(size=(H, 1)) * 0.3, "b2": np.zeros(1)}
    return {"params": p, "mom": {k: np.zeros_like(v) for k, v in p.items()},
            "step": 0, "epoch": 0, "pos": 0, "drop_rng": np.random.Generator(np.random.PCG64(DROP_SEED)).bit_generator.state}

def train_step(s, X, y):
    p, m = s["params"], s["mom"]
    idx = perm_for(s["epoch"])[s["pos"] * BATCH:(s["pos"] + 1) * BATCH]
    rng = np.random.Generator(np.random.PCG64()); rng.bit_generator.state = s["drop_rng"]
    xb, yb = X[idx], y[idx][:, None]
    h = np.tanh(xb @ p["W1"] + p["b1"]); mask = (rng.random(h.shape) > 0.2) / 0.8; hd = h * mask
    out = 1 / (1 + np.exp(-(hd @ p["W2"] + p["b2"]))); g = (out - yb) / len(xb)
    gW2 = hd.T @ g; gb2 = g.sum(0); gh = (g @ p["W2"].T) * mask * (1 - h ** 2)
    grads = {"W1": xb.T @ gh, "b1": gh.sum(0), "W2": gW2, "b2": gb2}
    for k in p:
        m[k] = 0.9 * m[k] + grads[k]; p[k] = p[k] - lr_at(s["step"]) * m[k]
    s["drop_rng"] = rng.bit_generator.state
    s["step"] += 1; s["pos"] += 1
    if s["pos"] == STEPS_PER_EPOCH: s["pos"] = 0; s["epoch"] += 1

The state has parameters (params), momentum buffers (mom), the global step that selects the learning rate, the epoch and position pos within it that select the batch, and drop_rng, the dropout generator’s internal state. A production job adds to this list: the optimizer’s second-moment estimates for Adam, mixed-precision loss-scaler state, the random states of Python, NumPy, and the framework on every rank including the CUDA generators, the data sampler’s epoch and offset, and metadata recording the world size and parallelism layout, the code version, and the lineage identifiers from the versioning article.

The first experiment tests the claim that each piece matters. A reference run trains for 600 steps without interruption. Then the job is crashed at step 275 and resumed from the checkpoint at step 250, but with one piece of the saved state discarded each time:

D. resume after a crash at step 275, but discard one piece of saved state
   discard mom : {'final_step': 600, 'digest': '182abce4091f0efd', 'loss': 0.035514} | matches reference digest: False
   discard rng : {'final_step': 600, 'digest': 'fdfaec1a0c45672d', 'loss': 0.032197} | matches reference digest: False
   discard pos : {'final_step': 600, 'digest': '645aa776d37d5a36', 'loss': 0.032948} | matches reference digest: False
   discard step: {'final_step': 600, 'digest': '244a3b8986b3e221', 'loss': 0.014005} | matches reference digest: False

The reference run ends with digest 5347f33c6149b3bc, a hash of the final weights, and a loss of 0.034869. Every ablated run completes 600 steps and produces a model with a plausible loss, and every one has a different digest. Discarding the momentum buffers (mom) changes the next steps’ updates. Replacing the dropout generator state (rng) changes which units are dropped. Resetting the position in the epoch (pos) feeds batches the model has already seen at the wrong times. Resetting the step counter (step) restarts the learning-rate schedule at its largest value and, because the loop runs until the counter reaches 600, trains for far longer than the reference, which explains the lower loss. None of these failures raises an error. A job that resumes with partial state looks healthy, and the damage appears only as a result that cannot be reproduced. This is why the versioning article treated resumption as part of reproducibility.

Writing checkpoints that survive a crash

The next failure is a crash during the write. A checkpoint of a large model consists of many files, one or more per rank, and writing takes long enough that a failure during it is likely. The protocol must ensure that a reader never mistakes a partial checkpoint for a complete one.

The naive protocol is to overwrite the previous checkpoint in place, shard by shard. This is the code:

def save_naive(root, s, crash_after_shard=None):
    os.makedirs(root, exist_ok=True)
    for i, k in enumerate(GROUPS):
        np.savez(os.path.join(root, f"shard-{k}.npz"), **shard_arrays(s, k))
        if crash_after_shard is not None and i + 1 == crash_after_shard: os._exit(137)
    json.dump(meta_of(s), open(os.path.join(root, "meta.json"), "w"))

If the process dies after writing two of the four shard files, the directory holds the first two shards from the new step and the last two from the old one, and the metadata file still describes the old step. Nothing signals that anything is wrong, since every file is individually valid.

The robust protocol builds on three rules. Write the whole checkpoint into a temporary location first. Record what a complete checkpoint consists of (file names and content hashes) in a manifest written last. Publish with a single atomic operation, which on a POSIX file system is rename of a directory or a file within one file system. A checkpoint then either exists completely, with a manifest that verifies, or it does not exist. The loader checks the manifest and the hashes, and quarantines anything that fails. The implementation:

def save_atomic(root, s, crash_after_shard=None):
    final = os.path.join(root, f"step-{s['step']:06d}"); tmp = final + ".tmp"
    shutil.rmtree(tmp, ignore_errors=True); os.makedirs(tmp)
    files = {}
    for i, k in enumerate(GROUPS):
        path = os.path.join(tmp, f"shard-{k}.npz")
        with open(path, "wb") as f:
            np.savez(f, **shard_arrays(s, k)); f.flush(); os.fsync(f.fileno())
        files[f"shard-{k}.npz"] = sha_file(path)
        if crash_after_shard is not None and i + 1 == crash_after_shard: os._exit(137)    # simulated crash mid-write
    mpath = os.path.join(tmp, "MANIFEST.json")
    with open(mpath, "w") as f:
        json.dump({"meta": meta_of(s), "files": files}, f); f.flush(); os.fsync(f.fileno())
    fsync_dir(tmp)
    os.rename(tmp, final)                                    # the directory appears completely or not at all
    fsync_dir(root)
    lp = os.path.join(root, "LATEST.tmp")
    with open(lp, "w") as f: f.write(os.path.basename(final)); f.flush(); os.fsync(f.fileno())
    os.replace(lp, os.path.join(root, "LATEST")); fsync_dir(root)

def load_atomic(root):
    """Return the newest checkpoint whose manifest and shard checksums verify, else None."""
    cands = sorted([d for d in os.listdir(root) if d.startswith("step-") and not d.endswith(".tmp")], reverse=True) if os.path.isdir(root) else []
    for d in cands:
        try:
            man = json.load(open(os.path.join(root, d, "MANIFEST.json")))
            if any(sha_file(os.path.join(root, d, fn)) != h for fn, h in man["files"].items()):
                os.rename(os.path.join(root, d), os.path.join(root, d + ".corrupt")); continue   # quarantine, keep for diagnosis
            s = {"params": {}, "mom": {}, **{k: man["meta"][k] for k in ("step", "epoch", "pos", "drop_rng")}}
            for k in GROUPS:
                z = np.load(os.path.join(root, d, f"shard-{k}.npz")); s["params"][k] = z["param"]; s["mom"][k] = z["mom"]
            return s
        except (OSError, ValueError, KeyError): continue
    return None

The writer flushes and fsyncs each file before moving on, so the data is on stable storage before the manifest says it is. The manifest is written after all shards, and the final rename is the commit. The LATEST pointer is updated with the same write-then-replace pattern, and the loader does not depend on it: it scans for the newest directory whose files verify, so a crash between the rename and the pointer update loses nothing. A directory ending in .tmp is by definition incomplete, and the writer removes any stale one before starting. A checkpoint that fails verification is renamed with a .corrupt suffix: that keeps it available for diagnosis and prevents a later save at the same step number from colliding with it.

Two experiments test the protocol. In the first, the job is made to crash in the middle of writing the step-300 checkpoint, after two of four shards, once with each writer. In the second, a completed checkpoint is damaged by flipping one byte in one shard.

E. crash while writing the checkpoint at step 300 (after 2 of 4 shards)
   naive  writer: first attempt exit 137; resume says 'start step=250 resumed=True'; final {'final_step': 600, 'digest': 'c949431e53c485e2', 'loss': 0.032807} | matches reference: False
   directory right after the crash: ['step-000150', 'step-000200', 'step-000250', 'step-000300.tmp']
   atomic writer: first attempt exit 137; resume says 'start step=250 resumed=True'; final {'final_step': 600, 'digest': '5347f33c6149b3bc', 'loss': 0.034869} | matches reference: True
F. a completed checkpoint is damaged on disk (one flipped byte in a shard of step 400)
   resume says 'start step=350 resumed=True'; final {'final_step': 600, 'digest': '5347f33c6149b3bc', 'loss': 0.034869} | matches reference: True
   quarantined: ['step-000400.corrupt']

With the naive writer, the first attempt exits with status 137 mid-write, and the resumed job starts from step 250, believing the metadata, but loads two shards from step 300 and two from step 250. It finishes and reports a loss of 0.032807, which looks fine, and its digest does not match the reference. This is silent corruption of the training state: a model has been trained from a state that never existed. With the atomic writer, the crashed attempt leaves a step-000300.tmp directory that the loader ignores, the job resumes from step 250, and the final digest equals the reference exactly. In the second experiment the loader detects the checksum mismatch on step 400, quarantines it, falls back to step 350, and again finishes with the reference digest.

The experiments have limits. The crash is a process exit or a SIGKILL; they do not simulate power loss or a storage system that acknowledges a write before it is durable, and the local disk used here says nothing about the durability semantics of a network file system or object store. On storage without an atomic rename, the same protocol works if the manifest itself is the commit marker: a checkpoint exists if and only if its manifest exists and its listed files verify. The hash check in the loader costs a full read of the checkpoint, which is cheap here and a real cost for a model with hundreds of gigabytes of state.

Sharded checkpoints at scale

In a real distributed job, each rank holds only part of the state, so each rank writes only its part. PyTorch’s torch.distributed.checkpoint (DCP) is the library for this. Its documentation describes it as supporting saving and loading from multiple ranks in parallel, and handling load-time resharding, which allows saving under one cluster topology and loading under another. It differs from torch.save in that it produces multiple files per checkpoint, at least one per rank, and in that it operates in place: the model allocates its storage first and DCP fills it. The entry points are torch.distributed.checkpoint.save and load, with checkpoint_id identifying the checkpoint (a folder path or, for other storage, a key), and get_state_dict and set_state_dict to obtain and apply model and optimizer state in a form that can be resharded with a different number of trainers or parallelism. There is also async_save, which first copies the state to staging storage, by default CPU memory, and then writes in a separate thread; the documentation marks it as experimental. A sketch of how the pieces fit, which was not run because no GPUs or torch were available:

# Configuration sketch, not executed here.
import torch.distributed.checkpoint as dcp
from torch.distributed.checkpoint.state_dict import get_state_dict, set_state_dict

def save(model, optimizer, step, root):
    model_sd, optim_sd = get_state_dict(model, optimizer)
    dcp.save({"model": model_sd, "optim": optim_sd, "step": step},
             checkpoint_id=f"{root}/step-{step:06d}")

def load(model, optimizer, root, step):
    model_sd, optim_sd = get_state_dict(model, optimizer)
    state = {"model": model_sd, "optim": optim_sd, "step": 0}
    dcp.load(state, checkpoint_id=f"{root}/step-{step:06d}")
    set_state_dict(model, optimizer, model_state_dict=state["model"], optim_state_dict=state["optim"])
    return state["step"]

DCP’s documented contract is parallel saving and load-time resharding. What it guarantees about atomic publication depends on the storage writer in use, so the manifest-and-commit protocol above remains the platform’s responsibility unless the chosen writer provides it. Loading with a different world size also changes the order in which gradients are reduced, and floating-point addition is not associative, so a resumed job on a different topology is a valid continuation but not a bit-identical one.

Resuming exactly

Resumption is correct when the resumed job reaches the same state as an uninterrupted one. The platform can test that directly, and the test is cheap enough to run in continuous integration for every training codebase. The orchestrator for these experiments does three things: it runs the job uninterrupted to get a reference digest; it injects crashes at chosen steps; and it sends real SIGKILL signals at random moments and restarts until the job finishes. The results:

A. reference: uninterrupted run
   attempt 1: start step=0 resumed=False -> exit 0
   {'final_step': 600, 'digest': '5347f33c6149b3bc', 'loss': 0.034869}
B. crash injection at steps 130 and 410 (no checkpoint is written at the crash step), then resume
   attempt 1: start step=0 resumed=False -> exit 137
   attempt 2: start step=100 resumed=True -> exit 137
   attempt 3: start step=400 resumed=True -> exit 0
   {'final_step': 600, 'digest': '5347f33c6149b3bc', 'loss': 0.034869} | identical to reference: True
C. real SIGKILL at random moments (each step sleeps 3 ms so that kills land mid-run)
   attempt 1: start step=0 resumed=False -> exit -9
   attempt 2: start step=50 resumed=True -> exit -9
   attempt 3: start step=100 resumed=True -> exit -9
   attempt 4: start step=150 resumed=True -> exit -9
   attempt 5: start step=250 resumed=True -> exit -9
   attempt 6: start step=300 resumed=True -> exit 0
   {'final_step': 600, 'digest': '5347f33c6149b3bc', 'loss': 0.034869} | identical to reference: True

In experiment B, the first attempt crashes at step 130 and the second attempt resumes from step 100, the last checkpoint before the crash, and crashes at step 410. The third attempt resumes from step 400 and finishes. Experiment C does the same with real kills at random moments; attempt boundaries differ from run to run because they depend on timing, which is the point. In both, the final digest and loss equal those of the reference run exactly. Steps 101 to 130 and 401 to 410 were computed twice, and the second computation reproduced the first, so the repeated work cost time and changed nothing else.

This result holds because every source of variation was either fixed or saved: the batch order is a function of the epoch, the dropout generator’s state is in the checkpoint, and the arithmetic is deterministic on one machine. On a GPU cluster, the same test needs the framework’s determinism settings described in the artifact versioning article, and bit-exact equality may not be achievable; the test then compares loss curves within a tolerance derived from seed-to-seed variation. A weaker version still catches the most common mistakes: if the loss jumps at the resume point, or the learning rate printed after resume differs from the value before the crash, some piece of state was lost.

Four more practices round out resumption. The job should log the step it resumed from and the value of TORCHELASTIC_RESTART_COUNT, so that rework can be measured. Side effects tied to steps, such as metric exports and evaluation runs, should be idempotent, since the steps between the checkpoint and the crash run twice. The job should check the checkpoint’s recorded topology and code version against the current ones before using it, and fail loudly on a mismatch. And retention should keep more than the latest checkpoint: the latest may be corrupted, as the second experiment showed, or may have been written after the corruption of the model state itself (a diverged loss, a silent error), so a few older checkpoints should be kept at a spaced interval.

Choosing the checkpoint interval

Checkpointing costs time, and skipping it costs lost work. The question is how often to pay. A first-order model gives an answer. Let \( \delta \) be the time the job is blocked writing one checkpoint, \( T \) the interval of useful compute between checkpoints, \( M \) the mean time between failures of the job, and \( R \) the time to restart and reload. For every unit of useful work, checkpointing adds \( \delta / T \) of overhead. Each failure, which occurs about once per \( M \) units of time, costs the restart time \( R \) plus the work since the last checkpoint, which averages \( T/2 \) if failures arrive at random points in the interval. Per unit of useful work the overhead is therefore approximately

\[ \text{waste}(T) \approx \frac{\delta}{T} + \frac{T/2 + R}{M} \]

The first term falls as \( T \) grows and the second rises. Setting the derivative with respect to \( T \) to zero gives \( -\delta/T^2 + 1/(2M) = 0 \), so

\[ T_{\text{opt}} = \sqrt{2\,\delta\,M} \]

This is Young’s first-order approximation, published in 1974 and referred to as the Young/Daly formula after Daly’s later analysis, which develops more complete cost functions and higher-order approximations (J. T. Daly, 2006). The restart time \( R \) does not appear in the first-order optimum because it is paid once per failure whatever the interval. The derivation assumes failures arrive randomly at a constant rate, so that the expected loss is half an interval, and that \( \delta \) and \( R \) are small relative to \( M \). It also treats \( \delta \) as the time the job is blocked, so it applies to the exposed cost: a checkpoint that is written in the background after a quick copy to memory has a much smaller \( \delta \).

The formula is only a model, so the next program checks it by simulation. A job needing 200 hours of useful compute runs against randomly arriving failures with a mean gap of 3.1 hours (the figure derived above from the Llama 3 report, used here as a plausible magnitude and not as a property of any real cluster), a restart cost of 10 minutes, and a checkpoint cost of 20 seconds, 2 minutes, or 10 minutes. Failures can strike during compute, during a checkpoint write, or during a restart; a failure during a write loses the chunk being checkpointed. Each configuration is the mean of 400 simulated runs. The output has the goodput (useful time divided by wall-clock time while holding the GPUs), the fractions of wall time spent on checkpoint writes, on work that was lost, and on restarts, and the first-order model’s prediction:

checkpoint cost 20 s, MTBF 3.1 h, restart 10 min -> Young interval sqrt(2*delta*M) = 11.1 min
 interval  goodput    ckpt    lost  restart    model
      5 m    0.876   0.058   0.013    0.052    0.882
     10 m    0.892   0.030   0.026    0.052    0.898
     20 m    0.883   0.015   0.051    0.052    0.890
     30 m    0.864   0.010   0.075    0.052    0.873
     45 m    0.832   0.006   0.110    0.052    0.846
     60 m    0.799   0.004   0.144    0.052    0.819
     90 m    0.736   0.003   0.210    0.052    0.770
    120 m    0.674   0.002   0.272    0.052    0.725
    180 m    0.562   0.001   0.385    0.052    0.650
  Young      0.893   0.027   0.029    0.052   (interval 11.1 min)

checkpoint cost 120 s, MTBF 3.1 h, restart 10 min -> Young interval sqrt(2*delta*M) = 27.3 min
 interval  goodput    ckpt    lost  restart    model
      5 m    0.665   0.267   0.016    0.052    0.682
     10 m    0.765   0.154   0.029    0.052    0.781
     20 m    0.812   0.082   0.054    0.052    0.828
     30 m    0.815   0.054   0.079    0.052    0.833
     45 m    0.799   0.036   0.114    0.052    0.820
     60 m    0.775   0.026   0.147    0.052    0.801
     90 m    0.719   0.016   0.212    0.052    0.759
    120 m    0.662   0.011   0.275    0.052    0.718
    180 m    0.554   0.006   0.388    0.052    0.646
  Young      0.816   0.060   0.072    0.052   (interval 27.3 min)

checkpoint cost 600 s, MTBF 3.1 h, restart 10 min -> Young interval sqrt(2*delta*M) = 61.0 min
 interval  goodput    ckpt    lost  restart    model
      5 m    0.304   0.623   0.021    0.052    0.326
     10 m    0.449   0.461   0.037    0.052    0.481
     20 m    0.583   0.299   0.066    0.052    0.622
     30 m    0.638   0.218   0.092    0.052    0.681
     45 m    0.667   0.152   0.129    0.052    0.716
     60 m    0.671   0.114   0.163    0.052    0.724
     90 m    0.647   0.074   0.228    0.052    0.711
    120 m    0.608   0.051   0.289    0.052    0.685
    180 m    0.519   0.029   0.399    0.052    0.628
  Young      0.671   0.113   0.165    0.052   (interval 61.0 min)

For the 2-minute checkpoint, the formula gives an interval of 27.3 minutes, and the simulated goodput there is 0.816, against 0.815 for the best interval in the sweep (30 minutes). The curve is flat around the optimum: anything between 20 and 45 minutes is within about 2 percentage points. It is steep away from it. At 10 minutes the job spends 15 percent of its wall time writing checkpoints, and at 120 minutes it loses 28 percent of it to rework. The same pattern holds in the other two configurations, with optima that scale with the square root of the checkpoint cost: 11.1 minutes at 20 seconds (goodput 0.893) and 61.0 minutes at 10 minutes (0.671). The model’s prediction in the last column runs a few points high, since it ignores failures during writes and restarts, but it ranks the intervals correctly and puts the optimum in the right place.

Three conclusions follow for a platform. First, reducing the checkpoint cost is worth more than tuning the interval. Going from a 10-minute to a 20-second blocking write raises goodput from 0.67 to 0.89 in this scenario, a gain that no interval choice on the slow writer can reach. That is the case for asynchronous and sharded checkpoints. Second, the optimal interval depends on the failure rate, which differs by cluster and changes with job size, since more GPUs means more components that can fail; the interval should be recomputed from observed interruption counts and not set once. Third, the cost \( \delta \) is not constant. Checkpoint writes from many jobs arrive at the same storage system, and the Llama 3 report describes the highly bursty checkpoint writes that saturate the storage fabric for short periods as a major challenge, so \( \delta \) can grow when other jobs checkpoint at the same time. The platform may need to stagger checkpoint schedules or limit concurrent writers.

Measuring the waste

Choosing an interval requires the parameters, and running a platform requires knowing how much capacity is wasted. The measure to track is goodput: the fraction of allocated GPU time that advanced the training run. The Llama 3 report uses a similar notion, effective training time, defined as time spent on useful training over elapsed time, and reports more than 90 percent for its run. The table breaks the loss down into components that can be measured separately, and each maps to a lever.

Component	How to measure it	Main levers
Queue wait	Time from submission to admission; not GPU time, but delay	Quotas, admission policy, fragmentation (previous article)
Startup and restart	Time from admission to the first training step, per restart	Image and checkpoint load time, rendezvous, a warm pool of nodes
Checkpoint blocking	Time the training loop is blocked per checkpoint, times the number of checkpoints	Asynchronous writes, sharding, faster storage
Rework	Steps recomputed after each restart (step reached before the failure minus step resumed from), times step time	Checkpoint interval, emergency checkpoint on termination
Hang time	Time from the last progress to the abort, per failure	Collective timeout, watchdog and heartbeat monitoring
Straggler slowdown	Median step time compared with the per-rank minimum or an expected value	Detecting and replacing slow nodes

All of these can be computed from three event streams the job already has: restarts (a restart count and the resume step), step completions with timestamps, and checkpoint start and end times. The denominator is GPU-seconds allocated, which the scheduler knows. A dashboard that shows only utilization misses all of this, since a hung job and a healthy one both show allocated GPUs, and a job re-computing lost steps looks busy.

These numbers also feed back into scheduling. A preemption policy that picks victims without regard to how long since their last checkpoint turns an avoidable loss into a certain one: the previous article’s reclaim policy destroyed on average 54 GPU-hours per run because its jobs had no checkpoints, and a job that checkpoints every half hour would lose at most half an hour of its GPUs per preemption. A scheduler can prefer to preempt jobs that have just checkpointed, or request a checkpoint before preempting. Those mechanisms depend on a signal path from the platform to the training code, and on a grace period long enough to use it.

From training jobs to serving

A training job produces an artifact, and the earlier articles in this module defined how that artifact is identified, promoted, and scheduled. Recovery completes the picture for the training side: checkpoints are artifacts too, with their own integrity rules; a restart is a scheduler event whose cost is set by the checkpoint protocol; and a failure is routine at scale, so the interval, the timeout, and the monitoring are design parameters. The next module turns to the other side of the platform, where the model is a running service whose requests arrive continuously and whose cost is measured in latency. Its first article, Model Serving: Batching, Autoscaling, and Tail Latency, starts from the same GPU memory arithmetic used here and asks how many requests a device can serve at once and what happens to the slowest of them.

References

• PyTorch documentation, torchrun (Elastic Launch)
• PyTorch documentation, Distributed Checkpoint (torch.distributed.checkpoint)
• PyTorch documentation, Distributed communication package (torch.distributed)
• PyTorch documentation, Torch Distributed NCCL environment variables
• Kubernetes documentation, Pod Lifecycle
• Grattafiori et al., The Llama 3 Herd of Models
• Optimal Checkpoint Interval with Availability as an Objective Function (arXiv:2410.18124)
• J. T. Daly, A higher order estimate of the optimum checkpoint interval for restart dumps (Future Generation Computer Systems, 2006)