Under the Hood
Book ↗
Projects/Training at Scale
Project 12

Distributed Training: FSDP and ZeRO (single-box proxy).

The Concept

Think of a library where every member owns three or four books, not the whole collection. When someone needs to read a book, they ask the member who owns it and they pass it over. The reader reads, hands the book back, and the owner puts it on their shelf again. Nobody hoards the catalog. Storage is cheap because each member only carries their share, and the only cost is that books move around when they are needed.

That picture is the entire idea behind Fully Sharded Data Parallel training, usually shortened to FSDP. The model's parameters get cut into N pieces, one per worker process, and so do the gradients and the optimizer state. Each worker owns exactly 1/N of each, and only 1/N. To run a forward pass, the workers gather the pieces they need from each other just in time, run the layer, and then drop the gathered copies. The full weight tensor exists for a few milliseconds during forward and a few more during backward, and then it disappears again. The rest of the time, each worker carries only its slice.

A few terms before we go further, because we are about to use them constantly.

A worker or rank is one process running one copy of the training script. In multi-GPU training, one rank usually owns one GPU. In our single-box proxy, two ranks share one GPU (or two ranks each own a CPU). The set of ranks participating in a training run is called the world, and the total count is the world size. Rank 0 through rank N-1 each have a unique integer identity.

A shard is a slice of a parameter tensor. If a weight matrix has shape (1024, 4096) and the world size is 4, the simplest way to shard is to flatten the tensor to length 1024 * 4096 = 4_194_304 and give each rank a contiguous slice of length 1_048_576. Rank 0 owns the first slice, rank 1 the second, and so on. Sharding works on the flat view because flat slicing always divides cleanly with padding, while slicing a multi-dimensional tensor along, say, the row axis runs into edge cases when the row count is not divisible by the world size.

The first time I worked on a sharded system at this scale was nothing like an LLM. It was the Aspira AWS migration. Customer data spread across 28 parks and roughly 5,000 sites, six years of campground reservations, ported into Redshift without losing a record at 99.99% uptime. The math of FSDP is different, the stakes are different, the bytes are different. The mental discipline is the same one. Stop thinking about the data as "one big object that you address by its full shape." Start thinking about it as "N consistent slices that have to agree about who owns what." Half of distributed engineering is just internalizing that shift.

All-gather is the operation that collects the full tensor from all ranks. Every rank starts with its 1/N slice. Every rank ends holding the full concatenated tensor. The cost is communication: each rank sends its slice to every other rank, or equivalently the system routes the slices through whatever interconnect is fastest. The dual operation is reduce-scatter, which goes the other direction. Every rank starts with a full tensor (typically a full gradient), the system sums those tensors element-wise across all ranks, and then each rank ends up with a 1/N slice of the summed result. All-gather is "spread the shards out into the whole." Reduce-scatter is "collapse the whole into shards while summing."

The reason to know the names is that these two operations are the entire vocabulary of FSDP. Forward pass uses all-gather to materialize each layer's weights just before computing the layer. Backward pass uses reduce-scatter to fold per-rank gradients back into sharded gradients. That is it. Everything else is bookkeeping.

Zero Redundancy Optimizer, almost always called ZeRO, is the same idea told as a three-step ladder. ZeRO Stage 1 shards only the optimizer state, that big chunk of fp32 Adam moments that lives next to the parameters and is twice as large as the parameters themselves. ZeRO Stage 2 adds gradients to what gets sharded. ZeRO Stage 3 adds the parameters themselves. ZeRO-3 and FSDP are the same algorithm wearing different brand names. DeepSpeed shipped ZeRO first, PyTorch shipped FSDP later, and the two communities use both names for the same picture.

The savings curve is the part that makes this worth the engineering. If you have N workers and you fully shard parameters, gradients, and optimizer state, each worker holds 1/N of the storage. With AdamW, a model of P parameters in mixed precision needs roughly 2P + 2P + 12P = 16P bytes: two bytes per parameter for the bf16 weights, two bytes per parameter for the gradients, four bytes for the fp32 master weights, and eight bytes for the two fp32 Adam moments. For a 70-billion-parameter model that is about 1.1 terabytes. With 64 workers fully sharded, each worker holds about 17 gigabytes, which fits in one H100. The model that did not fit on any single GPU fits on every single GPU at the same time, with room to spare.

The first time I sat with these numbers, I expected the storage savings to be the interesting part. They are not. The interesting part is the second-order consequence: once each worker only holds 1/N, the cost of starting and ending each layer goes up, because somebody has to materialize the full tensor for the few microseconds the kernel needs it. The savings are real. They are also paid for. The rest of this chapter is mostly about reading both sides of that ledger at once.

Figure 12.1. A single parameter tensor sharded across N ranks. The flat view is sliced into N contiguous pieces of equal length; each rank owns one piece and nothing else.

The catch is communication. The model still has to assemble itself layer by layer to compute. If your interconnect is fast (NVLink between GPUs in one node, NVSwitch between nodes), the cost is small enough that the trade pays off. If your interconnect is slow (generic ethernet between cloud machines, with millisecond round-trip times), the cost can dominate, and the model spends more time waiting on the network than computing. Real FSDP deployments live on this knife edge, and a serious part of the job is reasoning about where the time goes.

A small caveat on terminology, because the literature will confuse you otherwise. People sometimes say "data parallel" to mean what we used to do, where you replicate the model on every worker and split the batch. They say "fully sharded data parallel" to mean what this chapter teaches. The word "data" in both names refers to splitting the input batch across workers, which is still happening here. The "fully sharded" part is the new piece: even though every worker processes a different batch slice, the model parameters themselves are no longer replicated. They are sharded.

The mental shift the rest of the chapter pushes is this. In ordinary single-GPU training, every parameter lives in one place at all times, and computation is local. In FSDP, every parameter lives in one place most of the time and in every place during the few microseconds when its layer is computing. Memory is the resource you are saving. Communication is the resource you are paying. The skill is reading both at once.

Most introductions to distributed training I have read are honestly bad on this point. They walk you through the API. They leave you fluent in syntax and blind to the budget you are spending. The whole job is the budget.

Why It Matters

Without FSDP or ZeRO, training models larger than a single GPU's memory becomes either impossible or absurdly expensive. You can rent a GPU with 80 gigabytes of memory, but you cannot rent one with 1.1 terabytes, because no such product exists. Above a certain size, sharding stops being a clever optimization and starts being the only way the run happens at all. Every frontier model trained in the last three years used some form of this idea.

The second reason this chapter matters is debugging. Distributed bugs are different from single-GPU bugs. On one GPU, when something is wrong, it crashes or produces garbage immediately. On multiple ranks, it can crash on one rank and not the others, hang forever because two ranks are waiting on a collective the third rank never reached, or, worst of all, silently diverge, where each rank trains a slightly different model and the loss curves look fine on each rank but the eventual checkpoint is nonsense. We will see one of those silent failures up close in the BREAK IT section.

There is a third reason, less direct but worth naming. Reading distributed training code is a skill the field has decided is optional, which means most engineers cannot do it. If you can look at a 200-line FSDP wrapping function and tell which calls are all-gathers and which are reduce-scatters, you have a leg up on most of the people working in this area. The mechanics are not hard. The notation is unfamiliar. This chapter fixes the notation problem.

I will say it plainly because I felt this for years. The barrier to entry on distributed training is almost entirely vocabulary, and the vocabulary is poorly taught.

The skill you build here also feeds directly into later chapters. Project 17: Production Serving — vLLM, Continuous Batching, Quantization uses tensor parallelism, a sibling technique that shards along a different axis. Project 18: Mixture of Experts runs expert-parallel layers, which is yet another sharding pattern. Project 23: Reward Models and RLHF trains a policy and a reward model concurrently, sometimes on the same hardware, and the memory-balancing math comes straight out of the ZeRO ladder. The vocabulary you learn here (rank, world size, shard, all-gather, reduce-scatter) recurs in every one of those chapters.

A final reason: the things that go wrong in distributed training are exactly the things instrumentation from Project 11: Training Debugging — Spikes, NaNs, Profiling lets you catch. The four vital signs from that chapter (loss, gradient norm, activation distribution, and update ratio) all behave a little differently across ranks during a failure, and a per-rank dashboard is the single best diagnostic when distributed training goes wrong. We will sketch what that looks like in Step 8.

Some of this is grounded in our own work. The KALAVAI cooperative-training paper (arXiv:2603.22755) covers fusion across 4 model families from 410M to 6.9B, more than 7,550 individual runs, with +7.7% perplexity improvement averaged across families. Every one of those runs lived or died on per-rank sanity checks. Without that hygiene, the loss curves look believable on each contributor's machine and the fused model is garbage. The reason this chapter exists in its current form is that the lessons from those runs are not in the textbooks.

The Build

build.py

$python projects/12_distributed-training-fsdp-and-zero-single-box-proxy/build.py --tiny

"""
Project 12: Distributed Training: FSDP and ZeRO (Single-Box Proxy)

Canonical runnable build for this project. Generated by tools/extract_code.py
from the book's chapter markdown — re-run that tool to regenerate.
"""

# === Step 1: Launch two PyTorch processes on one box ===

import os
import torch
import torch.distributed as dist
import torch.multiprocessing as mp

def worker(rank: int, world_size: int):
    os.environ["MASTER_ADDR"] = "127.0.0.1"
    os.environ["MASTER_PORT"] = "29500"
    dist.init_process_group(
        backend="gloo",
        rank=rank,
        world_size=world_size,
    )
    x = torch.tensor([float(rank + 1)])
    dist.all_reduce(x, op=dist.ReduceOp.SUM)
    print(f"rank {rank} saw {x.item()}")
    dist.destroy_process_group()

if __name__ == "__main__":
    mp.spawn(worker, args=(2,), nprocs=2, join=True)

# === Step 2: Implement parameter sharding by hand ===

class HandShardedLinear:
    def __init__(self, in_dim: int, out_dim: int, rank: int, world_size: int):
        self.in_dim = in_dim
        self.out_dim = out_dim
        self.rank = rank
        self.world_size = world_size
        total = in_dim * out_dim
        assert total % world_size == 0, "shard must divide evenly"
        shard_size = total // world_size
        self.shard_start = rank * shard_size
        self.shard_end = (rank + 1) * shard_size
        # Each rank initializes only its own slice.
        torch.manual_seed(0)  # deterministic across ranks
        full_init = torch.randn(total) * 0.02
        self.flat_shard = full_init[self.shard_start:self.shard_end].clone()
        self.flat_shard.requires_grad_(True)

def gather_full_weight(self) -> torch.Tensor:
    pieces = [torch.empty_like(self.flat_shard) for _ in range(self.world_size)]
    dist.all_gather(pieces, self.flat_shard)
    full_flat = torch.cat(pieces, dim=0)
    return full_flat.view(self.out_dim, self.in_dim)

def forward(self, x: torch.Tensor) -> torch.Tensor:
    W = self.gather_full_weight()
    y = x @ W.t()
    # Important: in real FSDP the full W is freed immediately after use,
    # so it does not stay in memory beyond this scope.
    return y

# === Step 3: All-gather and reduce-scatter, drawn out ===

def backward_step(self, grad_full_weight: torch.Tensor):
    # grad_full_weight has shape (out_dim, in_dim), same as the full weight.
    grad_flat = grad_full_weight.reshape(-1)
    # Each rank starts with the full gradient (computed locally on its batch).
    # We need reduce-scatter to sum across ranks and keep only this rank's slice.
    out = torch.empty_like(self.flat_shard)
    dist.reduce_scatter_tensor(out, grad_flat, op=dist.ReduceOp.SUM)
    # Now `out` is the summed gradient for this rank's shard.
    self.flat_shard.grad = out

# === Step 5: Wrap the Project 9 GPT in real FSDP ===

import torch
import torch.distributed as dist
from torch.distributed.fsdp import FullyShardedDataParallel as FSDP
from torch.distributed.fsdp.wrap import transformer_auto_wrap_policy
from functools import partial
from nanochat.model import GPT, TransformerBlock  # from Project 9

def setup_fsdp_model(config):
    model = GPT(config)
    wrap_policy = partial(
        transformer_auto_wrap_policy,
        transformer_layer_cls={TransformerBlock},
    )
    sharded = FSDP(
        model,
        auto_wrap_policy=wrap_policy,
        device_id=torch.cuda.current_device() if torch.cuda.is_available() else None,
    )
    return sharded

model = setup_fsdp_model(config)
optimizer = torch.optim.AdamW(model.parameters(), lr=3e-4)

for step, batch in enumerate(loader):
    inputs, targets = batch
    logits = model(inputs)
    loss = compute_loss(logits, targets)
    loss.backward()
    torch.nn.utils.clip_grad_norm_(model.parameters(), 1.0)
    optimizer.step()
    optimizer.zero_grad()

The Steps

4 reference files

Each step is the code for one piece, in isolation — read them as the book introduces them. They assemble into build.py above.

Step 01Launch Two Pytorch Processes On One Boxstep_01_launch-two-pytorch-processes-on-one-box.py
"""
Project 12: Step 1 — Launch two PyTorch processes on one box

Pedagogical reference: this file shows the code for this step in isolation.
For the full assembled, runnable build, use build.py in this same folder.
"""

import os
import torch
import torch.distributed as dist
import torch.multiprocessing as mp

def worker(rank: int, world_size: int):
    os.environ["MASTER_ADDR"] = "127.0.0.1"
    os.environ["MASTER_PORT"] = "29500"
    dist.init_process_group(
        backend="gloo",
        rank=rank,
        world_size=world_size,
    )
    x = torch.tensor([float(rank + 1)])
    dist.all_reduce(x, op=dist.ReduceOp.SUM)
    print(f"rank {rank} saw {x.item()}")
    dist.destroy_process_group()

if __name__ == "__main__":
    mp.spawn(worker, args=(2,), nprocs=2, join=True)
Step 02Implement Parameter Sharding By Handstep_02_implement-parameter-sharding-by-hand.py
"""
Project 12: Step 2 — Implement parameter sharding by hand

Pedagogical reference: this file shows the code for this step in isolation.
For the full assembled, runnable build, use build.py in this same folder.
"""

class HandShardedLinear:
    def __init__(self, in_dim: int, out_dim: int, rank: int, world_size: int):
        self.in_dim = in_dim
        self.out_dim = out_dim
        self.rank = rank
        self.world_size = world_size
        total = in_dim * out_dim
        assert total % world_size == 0, "shard must divide evenly"
        shard_size = total // world_size
        self.shard_start = rank * shard_size
        self.shard_end = (rank + 1) * shard_size
        # Each rank initializes only its own slice.
        torch.manual_seed(0)  # deterministic across ranks
        full_init = torch.randn(total) * 0.02
        self.flat_shard = full_init[self.shard_start:self.shard_end].clone()
        self.flat_shard.requires_grad_(True)

def gather_full_weight(self) -> torch.Tensor:
    pieces = [torch.empty_like(self.flat_shard) for _ in range(self.world_size)]
    dist.all_gather(pieces, self.flat_shard)
    full_flat = torch.cat(pieces, dim=0)
    return full_flat.view(self.out_dim, self.in_dim)

def forward(self, x: torch.Tensor) -> torch.Tensor:
    W = self.gather_full_weight()
    y = x @ W.t()
    # Important: in real FSDP the full W is freed immediately after use,
    # so it does not stay in memory beyond this scope.
    return y
Step 03All Gather And Reduce Scatter Drawn Outstep_03_all-gather-and-reduce-scatter-drawn-out.py
"""
Project 12: Step 3 — All-gather and reduce-scatter, drawn out

Pedagogical reference: this file shows the code for this step in isolation.
For the full assembled, runnable build, use build.py in this same folder.
"""

def backward_step(self, grad_full_weight: torch.Tensor):
    # grad_full_weight has shape (out_dim, in_dim), same as the full weight.
    grad_flat = grad_full_weight.reshape(-1)
    # Each rank starts with the full gradient (computed locally on its batch).
    # We need reduce-scatter to sum across ranks and keep only this rank's slice.
    out = torch.empty_like(self.flat_shard)
    dist.reduce_scatter_tensor(out, grad_flat, op=dist.ReduceOp.SUM)
    # Now `out` is the summed gradient for this rank's shard.
    self.flat_shard.grad = out
Step 05Wrap The Project 9 Gpt In Real Fsdpstep_05_wrap-the-project-9-gpt-in-real-fsdp.py
"""
Project 12: Step 5 — Wrap the Project 9 GPT in real FSDP

Pedagogical reference: this file shows the code for this step in isolation.
For the full assembled, runnable build, use build.py in this same folder.
"""

import torch
import torch.distributed as dist
from torch.distributed.fsdp import FullyShardedDataParallel as FSDP
from torch.distributed.fsdp.wrap import transformer_auto_wrap_policy
from functools import partial
from nanochat.model import GPT, TransformerBlock  # from Project 9

def setup_fsdp_model(config):
    model = GPT(config)
    wrap_policy = partial(
        transformer_auto_wrap_policy,
        transformer_layer_cls={TransformerBlock},
    )
    sharded = FSDP(
        model,
        auto_wrap_policy=wrap_policy,
        device_id=torch.cuda.current_device() if torch.cuda.is_available() else None,
    )
    return sharded

model = setup_fsdp_model(config)
optimizer = torch.optim.AdamW(model.parameters(), lr=3e-4)

for step, batch in enumerate(loader):
    inputs, targets = batch
    logits = model(inputs)
    loss = compute_loss(logits, targets)
    loss.backward()
    torch.nn.utils.clip_grad_norm_(model.parameters(), 1.0)
    optimizer.step()
    optimizer.zero_grad()

Break It

break_it.py

$python projects/12_distributed-training-fsdp-and-zero-single-box-proxy/break_it.py --tiny

"""
Project 12: BREAK IT experiment.

Deliberately sabotages one mechanism from build.py to show what happens
when it's removed. Compare outputs to the working version.
"""

shard_size = total // world_size
self.shard_start = rank * shard_size
self.shard_end = (rank + 1) * shard_size

shard_size = total // world_size
self.shard_start = rank * shard_size
if rank == 1:
    self.shard_start += 1  # off-by-one: this rank's shard is shifted by 1
self.shard_end = self.shard_start + shard_size

Read it in the book

This page ships the code. The full derivation — why the chain rule is mechanical not magical, what the computational graph buys you, the long-form reasoning — lives in Chapter 12 of Under the Hood.

Get the book →