Distributed Training Is a Scheduling Problem

Collectives set the step time. An all_reduce across four ranks measured about 277 GB/s, against an H100 NVLink peak near 900. That gap is normal. The communication in a parallel plan is a number you measure, not a number you assume.

Keep the work near the data

A node usually holds 8 GPUs. Inside each GPU the streaming multiprocessors do the math, and a ladder of memory feeds them. Every rung down is larger, slower, and further away.

• L1 and shared memory: very small, very fast, one per SM.
• HBM: larger and slower than L1, one per GPU.
• NVLink: fast, and it connects the GPUs inside one node.
• NVSwitch: slower, and it connects nodes to each other.
• PCIe and Ethernet: older, slower, and heavier on overhead.

In the best case the data already sits in L1. Otherwise it comes from HBM. On a multi-GPU node it can come from another GPU, and that rung is where a fast kernel waits.

The bandwidth note set the ceiling by the bytes you move on one GPU. The same ceiling exists one rung up, where the bytes cross NVLink or NVSwitch instead. Keep the arithmetic intensity high. Cross a link only when the math needs data that lives on the other side.

Seven collectives cover everything you will write

Collectives are the standard communication patterns across devices, and every parallel strategy is assembled from them. world_size is the number of devices in the group. rank is the device ID, and it runs from 0 to world_size - 1.

• broadcast: one rank sends a tensor to every rank.
• scatter: one rank sends a different slice to each rank.
• gather: one rank collects the values from every rank.
• reduce: combine the values across ranks with sum, min, or max, and write the result on one rank.
• all_gather: gather, with the full result landing on every rank.
• reduce_scatter: reduce across ranks, then send slice i to rank i.
• all_reduce: reduce, then all_gather, so every rank holds the reduced result.

The names decode themselves. Reduce combines values across ranks and gather collects them onto one rank. Scatter splits the outputs across ranks, and all means every rank ends holding the result.

Four ranks each build t = [0, 1, 2, 3] + rank.

t = [0, 1, 2, 3] + rank

Rank 0: [0, 1, 2, 3]
Rank 1: [1, 2, 3, 4]
Rank 2: [2, 3, 4, 5]
Rank 3: [3, 4, 5, 6]

dist.all_reduce(t, op=SUM)

Every rank: [6, 10, 14, 18]

The call changes the tensor in place, and every rank ends with the same vector.

reduce_scatter takes an input whose first dimension is world_size, so with four ranks the shape is [4, N]. It reduces across ranks, then hands slice i to rank i. Each rank keeps a tensor of shape [N]. Run an all_gather after it and you rebuild what all_reduce produces.

NCCL implements the collectives as ring and tree algorithms over NVLink, NVSwitch, and PCIe. When the program starts, the ranks discover the topology and pick their paths. The collectives then launch CUDA kernels that move data GPU to GPU with little CPU work.

torch.distributed exposes all_reduce, reduce_scatter, all_gather, broadcast, and barrier over two backends. Use nccl for GPU collectives and gloo for CPU collectives. Start one Python process for each rank. Call dist.init_process_group with world_size, rank, backend, and an init method. The collectives then run on ordinary tensors.

Measure the collective the way you measure a kernel

The kernels note put a measurement in front of every optimization. The same discipline applies here, with links in place of SMs, and the steps barely change.

• Run a warm-up pass so the kernels load before the clock starts.
• Synchronize the devices before and after the timed region.
• Use large tensors so that fixed overhead does not own the number.
• Divide the bytes moved by the elapsed time to get effective bandwidth.

Estimate the traffic before you read the clock. During an all_reduce, each rank moves about this much.

The factor of 2 counts two passes. The data leaves once for the reduction, then leaves again to reach every rank.

The benchmark ran four ranks with 100,000,000 float32 values on each, which is 400 MB per rank. Those tensors produced the 277 GB/s. Distance from the H100 NVLink peak moves with tensor size, algorithm, overlap, and topology.

One skipped collective hangs every rank

Every rank must call the same collectives in the same order, with shapes that match. reduce_scatter reads its destinations from the first dimension, so slice i goes to rank i. A shape that differs on one rank breaks the call.

If one rank skips a call, the others wait forever. The job does not crash and it does not print a trace. The ranks that did their part sit inside the collective, and the GPUs go quiet.

all_reduce does two jobs at once. It moves the gradients and it synchronizes the ranks. Skip it on one rank during a training step and every other rank blocks at that line.

Point-to-point carries the same contract. send(tensor, dst_rank) names its destination and recv(tensor, src_rank) names its source. Every send needs a recv that matches it, and sends between one pair of ranks keep their order. A send with no matching recv deadlocks the program. recv writes into a tensor you supply, and both calls block in simple code. isend returns a handle for an asynchronous send.

Two tools make a hang findable. barrier holds every process until all reach the same point. Use it to print in rank order, and sometimes for correctness. The gloo backend runs the same collectives on CPU, where debugging costs less.

Three splits, three costs

Data parallelism splits the batch and is the most common choice. Every rank holds a full copy of the model and its own optimizer. local_batch_size is batch_size divided by world_size, and rank r takes its own slice.

Take a deep MLP whose layers each do a [hidden_dim × hidden_dim] matmul and a nonlinearity. The input batch has shape [batch_size, hidden_dim]. Each rank runs the forward pass on its own slice, so the local loss differs across ranks. The backward pass gives local gradients.

An all_reduce with SUM over each parameter gradient, divided by world_size, makes the gradients identical again. Then every rank takes the same optimizer step. The parameters start equal because every rank uses the same init and the same RNG seed. They stay equal because the gradients synchronize at every step.

The cost is a full copy of the model and its optimizer state on every rank. One all_reduce for every gradient rides on top of that, at every step.

Tensor parallelism splits the model itself. You need it when the model does not fit on one GPU at batch size 1. local_num_dim is hidden_dim divided by world_size, and each rank holds a [hidden_dim × local_num_dim] shard of every layer.

The activations x start identical on every rank with shape [batch_size × hidden_dim]. Each rank computes local_x = x @ local_W, applies the nonlinearity, and holds a [batch_size × local_num_dim] piece. An all_gather concatenates the pieces back into x, and the next layer repeats the pattern. The backward pass runs the same idea in reverse with reduce_scatter and all_reduce.

Each rank stores 1/world_size of the parameters, which is how the width grows past one GPU. The bill is an all_gather inside every layer. The parallelism note kept tensor parallelism inside a node, and this bill is the reason. Only NVLink is fast enough to carry a collective at every layer.

Pipeline parallelism splits the layers by depth. With world_size = 2, a 4-layer MLP puts layers 0 and 1 on rank 0. Layers 2 and 3 go to rank 1. Rank 0 runs its layers and sends the activations forward. Rank 1 receives them, runs its layers, and stores the outputs.

Run that naively and rank 1 waits while rank 0 processes the whole batch. Then rank 0 waits while rank 1 finishes. Those gaps are the bubble.

Microbatches fill the bubble. A batch of 128 becomes 4 microbatches of 32. Rank 0 works on microbatch k+1 while rank 1 works on microbatch k. The last stage computes the loss and sends the gradients back. Each earlier stage backpropagates through its layers and passes them further back.

Other axes exist too, such as the sequence length inside attention. Each of the three splits turns a memory problem into a communication problem.

The Builder Test

Count the bytes one training step sends. In data parallel that is one all_reduce for each parameter gradient. The estimate above gives the bytes for each. Sum them.

Then time the step with the devices synchronized and divide. Compare the effective bandwidth against the link your plan assumed. A rate far under that link points at small messages, a bad algorithm choice, or communication that never overlapped compute.

Idle bubbles, all_reduce time, uneven shards, and small microbatches each erase the gain. Fix the one your measurement names before you add GPUs.

What Carries

More GPUs do not make training faster on their own. A cluster is worth what its schedule keeps in motion, and collectives are what the schedule is made of.

Name the wall before you choose the split. If memory is the wall, shard the state. A matrix too large for one GPU means splitting the operation. A batch that divides cleanly starts with data parallelism.

The cluster runs now. The next question is how large a model and how many tokens to buy with it.