Distributed Training: Data vs Tensor vs Pipeline Parallelism, AllReduce, and NCCL

A single training step has four phases: forward pass, loss computation, backward pass producing gradients, and an optimizer step applying those gradients to the weights. On one GPU this is straightforward. On N GPUs, phase four becomes a problem.

Suppose you have 4 GPUs and a batch of 128 examples. Data parallelism gives each GPU 32 examples. Each GPU runs a forward and backward pass and ends up holding a gradient tensor — the same shape as the model's parameters — computed from its own 32 examples. Those four gradient tensors are all different, because they came from different data.

If each GPU now applies its own gradient to its own copy of the weights, you have created four divergent models. After one step they differ slightly; after a thousand steps they are four unrelated fine-tunes, and there is no meaningful checkpoint to save. The training run has silently turned into four smaller, worse training runs.

The fix is to make every GPU apply the same update. Since all four gradients are unbiased estimates of the gradient over the full 128-example batch, the correct combined estimate is their average. So before the optimizer step, every GPU must obtain the average of all four gradient tensors. That is precisely the operation AllReduce provides: every participant contributes a tensor, and every participant receives the reduced result. After AllReduce, all four GPUs hold identical averaged gradients, apply identical updates, and remain bit-identical replicas of one model.

This gives you the single sentence to remember: data-parallel training is mathematically equivalent to single-GPU training on the full batch, and AllReduce is what makes that equivalence true.

The corollary is the one that bites in practice. Adding GPUs under data parallelism does not just make training faster; it increases the effective batch size:

effective batch size = per_device_batch × number_of_devices × gradient_accumulation_steps

Change the GPU count and you have changed a hyperparameter. The learning rate that was tuned for the old effective batch is no longer the right one, which is why "we added GPUs and the loss curve changed" is a routine and entirely explicable event — a fact 12-03 needs and this lesson supplies.

L1 — Intuition: three ways to split, three different constraints

Ask what is running out and the answer names the strategy.

• Time is running out. The model fits on one GPU; you just want to get through more data per second. → Data parallelism. Replicate the model, split the batch.
• Memory is running out because one layer is too big. A single weight matrix or attention block cannot fit. → Tensor parallelism. Split the matrix itself across GPUs, keep them tightly coupled.
• Memory is running out because there are too many layers. Any individual layer fits, but the stack does not. → Pipeline parallelism. Put layers 1–8 on GPU 0, 9–16 on GPU 1, and so on.

These are not mutually exclusive. Large-scale pretraining routinely combines all three — often described as 3D parallelism — plus optimizer-state sharding. That composition is beyond this exam's depth ceiling; the three identities are not.

L2 — Mechanism: what each strategy moves across the wire

Data parallelism. Every GPU holds the full model, the full optimizer state, and a slice of the batch. Communication happens once per step (or overlapped with the backward pass), and the volume is proportional to the parameter count — you are exchanging a full gradient tensor. Scaling is excellent as long as compute per step is large relative to that exchange. The memory cost is the catch: every GPU pays the full weights + gradients + optimizer state bill computed in 11-04. Four GPUs give you 4× the throughput and 1× the model capacity.

Tensor parallelism (also called intra-layer or model parallelism). A weight matrix is partitioned — for example, a feed-forward projection split column-wise across two GPUs — so each GPU computes a partial result on the same input, and a collective combines the partials before the layer's output is complete. Communication happens multiple times per layer, inside the forward and backward passes, and it is latency-sensitive. That is why tensor parallelism is normally confined to GPUs inside a single node connected by high-bandwidth NVLink, and rarely stretched across a network. It reduces per-GPU memory for weights, activations, and optimizer state proportionally.

Pipeline parallelism (inter-layer). The layer stack is cut into stages, one per GPU, and activations are passed from stage to stage. Communication volume is small — only the activations at the stage boundaries — but the naive version wastes most of the hardware: while GPU 0 works on the batch, GPUs 1–3 wait, and that idle time is the pipeline bubble. The standard remedy is micro-batching: split each batch into micro-batches and keep them in flight simultaneously so every stage is busy on a different micro-batch. More micro-batches means a smaller bubble and more in-flight activations to store.

L3 — Collectives, ring-allreduce, and NCCL

A collective operation is a communication pattern involving a whole group of participants at once, as opposed to a point-to-point send between two. The ones worth knowing by name:

Two identities are worth internalizing: AllReduce = ReduceScatter followed by AllGather, and Broadcast is what makes all replicas identical at step 0 while AllReduce is what keeps them identical at every step after.

Why ring-allreduce matters. The naive implementation of AllReduce sends every rank's tensor to one central rank, which sums and sends the result back. That central rank's link becomes a bottleneck whose traffic grows linearly with the number of participants — a design that stops scaling almost immediately.

Ring-allreduce arranges the N GPUs in a logical ring, each talking only to its neighbour, and runs two phases:

1. Scatter-reduce. Each rank's tensor is conceptually divided into N chunks. Over N−1 steps, each rank sends one chunk to its right neighbour and reduces the chunk it receives from its left. At the end, each rank holds one fully-reduced chunk — a different one per rank.
2. All-gather. Over another N−1 steps, those fully-reduced chunks are passed around the ring until every rank holds every chunk.

The properties that make this the standard algorithm: each rank sends and receives roughly 2 × tensor_size × (N−1)/N bytes, which approaches a constant 2× the tensor size regardless of how many GPUs participate; all links are used simultaneously rather than converging on one hub; and there is no central coordinator. The scaling behaviour is why ring-allreduce, or a bandwidth-optimal variant of it, is the algorithm behind essentially every large data-parallel training job.

NCCL (the NVIDIA Collective Communications Library, pronounced "nickel") is NVIDIA's implementation of these collectives for NVIDIA GPUs. Its identity, which is what the exam wants:

• It provides AllReduce, Broadcast, Reduce, AllGather, ReduceScatter, AllToAll and point-to-point primitives, topology-aware and optimized for NVIDIA interconnects.
• It works within a node over NVLink/NVSwitch or PCIe and across nodes over InfiniBand or Ethernet (RDMA where available).
• It is what PyTorch's DistributedDataParallel and the distributed backends of the major frameworks call underneath when the backend is set to nccl. You almost never write NCCL calls yourself; you select it as a backend.
• It is a communication library for multi-GPU/multi-node collectives — it is not a training framework, not a scheduler, and not a serving component. Distractors love to describe NCCL as a training library or a model-parallel framework. It is neither.

Gradient accumulation is the technique adjacent to all of this and frequently confused with it. Instead of synchronizing after every micro-batch, you run several forward/backward passes, accumulating gradients into the same buffers, and only then AllReduce and step. This raises the effective batch size without more GPUs and without more memory for activations beyond one micro-batch. Its two uses: reaching a large effective batch on limited hardware, and reducing communication frequency. It is a memory/throughput trick, not a parallelism strategy — there is nothing distributed about it, and it works perfectly well on one GPU.

Checkpointing is worth distinguishing carefully, because the word means two different things:

Both appear in distributed-training discussions and they solve unrelated problems. Confusing them is a common error.

The blueprint names data-versus-tensor parallelism as a confusable pair, so this is the table to memorize.

The umbrella-term trap: "model parallelism" is the umbrella covering tensor and pipeline parallelism — both split the model rather than the batch. An answer option offering "model parallelism" alongside "tensor parallelism" and "pipeline parallelism" is offering a category and two of its members, and the correct pick depends on how specific the question's constraint is.

A constructed scenario, all figures derived from the stated assumptions. You are fine-tuning a 7B model. On a single GPU you have established: per-device batch 8, gradient accumulation 2, one step takes 1.0 second of compute, and the dataset is 40,000 examples for one epoch.

Step 1 — the single-GPU baseline.

effective batch  = 8 × 1 GPU × 2 accumulation = 16 examples/step
steps per epoch  = 40,000 / 16               = 2,500 steps
wall-clock       = 2,500 × 1.0 s             = 2,500 s ≈ 41.7 minutes

Step 2 — eight GPUs, data parallel, nothing else changed.

effective batch  = 8 × 8 GPUs × 2            = 128 examples/step
steps per epoch  = 40,000 / 128              ≈ 313 steps
compute per step = 1.0 s (unchanged — each GPU still does 8×2 examples)

Ignoring communication, one epoch now takes 313 s ≈ 5.2 minutes: an 8× speedup. But two things changed that you must account for.

Step 3 — the communication cost. Each step requires an AllReduce over the full gradient tensor. At BF16, gradients for 7.0 × 10⁹ parameters are 14.0 × 10⁹ bytes. Ring-allreduce moves approximately 2 × size × (N−1)/N bytes per rank:

per-rank traffic = 2 × 14.0e9 × (7/8) = 24.5e9 bytes ≈ 24.5 GB per step

Whether that is negligible or catastrophic depends entirely on the interconnect, and the arithmetic is a ratio: divide 24.5 GB by the achievable per-rank bandwidth. Over a fast intra-node link the transfer is a fraction of the 1.0 s compute and, because gradient AllReduce can be overlapped with the backward pass — each layer's gradients are ready before the layer below finishes, so you can start reducing them immediately — most of it disappears behind compute. Over a slow link it dominates and your 8× becomes 3×. I am deliberately not quoting bandwidth numbers for specific hardware: those are spec-sheet figures, they are generation-dependent, and the exam does not ask for them. What the exam asks is which strategy is bandwidth-sensitive and why, and the formula above is the why.

Step 4 — the hyperparameter you silently changed. Effective batch went from 16 to 128, an 8× increase. A larger batch produces a lower-variance gradient estimate, which generally supports a larger learning rate. Two conventional heuristics exist — scale the LR linearly with batch size, or scale it with the square root — and neither is a law. What is not optional is noticing: if you keep the old LR, the run will underfit relative to what the hardware could achieve, and if you scale aggressively without warmup you may destabilize the early steps. Add or lengthen warmup when you scale up the batch, and expect to re-read the loss curve with 12-03 in hand.

Step 5 — the alternative that needs no extra GPUs. If you wanted effective batch 128 on your single GPU, gradient accumulation gets you there: 8 × 1 × 16 = 128. Same effective batch, same gradient quality, 16× the wall-clock. Gradient accumulation buys batch size with time; data parallelism buys it with hardware. Recognizing that they are interchangeable in their effect on the math and opposite in their effect on the clock is the useful insight.

Step 6 — when this plan fails entirely. Data parallelism requires the model, its gradients, and its optimizer state to fit on one GPU. From 11-04's arithmetic, full fine-tuning a 7B model with Adam at mixed precision needs roughly 16 bytes per parameter — about 112 GB — before activations. No single current GPU holds that. So the realistic options are: use LoRA so the trainable parameter count collapses and data parallelism becomes viable (11-05), shard the optimizer state across the data-parallel ranks, or introduce tensor and pipeline parallelism. The decision about parallelism is downstream of the memory arithmetic, not a substitute for it.

Distributed training is claimed by objectives 4.1 (assist in deployment and evaluation of model scalability, performance, and reliability) and 4.4 (identify system data, hardware, or software components required to meet user needs). More decisively, the official study guide's suggested-reading list names NCCL, AllReduce, and ring-allreduce explicitly — a reading the guide itself asks candidates to do, which makes these terms directly examinable. The course index rates this topic Tier 3 for frequency, so the correct study posture is recognition and identity, not configuration mastery.

There is also a stack-map dimension. NCCL is one of the NVIDIA components whose purpose you should be able to state in one clause: NCCL = multi-GPU and multi-node collective communication. When the exam presents a scenario and asks which NVIDIA component addresses it, the discriminating skill is knowing that NCCL communicates, TensorRT-LLM optimizes, Triton serves, NIM packages, NeMo builds, and RAPIDS processes data — the map 12-13 assembles in full.

Question phrasings:

• "Which operation synchronizes gradients across GPUs in data-parallel training?" — AllReduce.
• "What is NCCL used for?" — optimized collective communication primitives for multi-GPU and multi-node NVIDIA deployments.
• "A model is too large to fit on a single GPU. Which approach addresses this?" — model parallelism (tensor or pipeline), not data parallelism.
• "Why is ring-allreduce preferred over a centralized reduction?" — per-rank communication is independent of the number of GPUs and all links are used, so it scales.
• "What is the effect of adding GPUs under data parallelism, holding per-device batch constant?" — the effective batch size increases proportionally.
• "Which technique increases effective batch size without additional GPUs?" — gradient accumulation.
• "Which parallelism strategy is most affected by the pipeline bubble?" — pipeline parallelism.

Distractor families:

AllReduce is a collective communication operation in which every participant contributes a tensor and every participant receives the combined result — typically the sum, from which an average is derived by dividing by the participant count. In data-parallel training each GPU computes gradients from a different slice of the batch, so the gradients differ; AllReduce averages them so that every GPU applies an identical update and the replicas remain identical copies of one model. Without it, data-parallel training degenerates into N divergent training runs with no coherent checkpoint. AllReduce is distinguished from Reduce, where only one rank receives the result, and from Broadcast, where one rank's tensor is copied out to all. It is also decomposable: AllReduce equals a ReduceScatter followed by an AllGather, which is exactly how the ring implementation is structured.

NCCL, the NVIDIA Collective Communications Library, is NVIDIA's topology-aware implementation of collective communication primitives — AllReduce, Broadcast, Reduce, AllGather, ReduceScatter, AllToAll, and point-to-point sends — for NVIDIA GPUs. It is optimized for the interconnects those GPUs actually use: NVLink and NVSwitch inside a node, PCIe where NVLink is absent, and InfiniBand or Ethernet with RDMA across nodes. In practice you do not write NCCL code. You select it as the communication backend in your training framework and the framework issues the collective calls on your behalf. The exam-relevant identity is one clause: NCCL is the multi-GPU and multi-node collective communication library — not a training framework, not a scheduler, and nothing to do with serving.

Because its per-participant communication cost does not grow with the number of participants. In a centralized scheme every rank sends its full gradient tensor to one coordinator, so that coordinator's link carries traffic proportional to N and becomes the bottleneck as soon as N is more than a handful. Ring-allreduce arranges ranks in a ring and moves data only between neighbours, in two phases — scatter-reduce, then all-gather — each of N−1 steps. Every rank sends and receives about 2 × tensor_size × (N−1)/N bytes, which tends to a constant 2× the tensor size no matter how many GPUs join, and every link in the ring is active at once rather than idle while one hub saturates. There is also no coordinator to fail or to be provisioned specially. Those scaling properties are why ring-allreduce and its bandwidth-optimal relatives underpin large-scale data-parallel training.

Data parallelism splits the batch; tensor parallelism splits the model's layers. Under data parallelism every GPU holds a complete copy of the model and processes different examples, then the gradients are averaged with AllReduce; per-GPU memory for weights and optimizer state is unchanged, so the model must already fit on one GPU. Under tensor parallelism a single weight matrix is partitioned across GPUs, each computing a partial result on the same input, with collectives combining the partials inside every layer; per-GPU memory drops proportionally, so a layer larger than one GPU becomes trainable. The consequences follow: data parallelism communicates once per step and tolerates slower interconnects, while tensor parallelism communicates many times per layer and needs high-bandwidth, low-latency links, which is why it is usually kept inside a single node. The decision rule is simply which resource ran out — time or memory.

It replaces them for the purpose of reaching a large effective batch size, and not at all for the purpose of finishing sooner. Gradient accumulation runs several forward and backward passes, summing gradients into the same buffers, and only applies the optimizer step after the last one. The resulting update is mathematically comparable to one computed on the combined batch, so it gives you the gradient quality of a large batch on hardware that cannot hold a large batch. What it does not give you is throughput: the passes run sequentially, so wall-clock time scales with the accumulation count. Its second, less obvious use is in distributed settings — accumulating for K micro-batches before synchronizing means one AllReduce per K micro-batches instead of one per micro-batch, which is a genuine remedy when the interconnect is the bottleneck.

• Three strategies, three constraints: data parallelism for time, tensor parallelism for a layer too big, pipeline parallelism for a stack too deep.
• Data parallelism replicates the model, so it never solves a model-too-large problem. That is the single most-tested distinction here.
• Gradients from different data slices must be averaged into one identical update, or the replicas diverge into different models. AllReduce does that.
• AllReduce = every rank contributes, every rank receives. Reduce = one receives. Broadcast = one sends. AllReduce = ReduceScatter + AllGather.
• Ring-allreduce keeps per-rank traffic at roughly 2× the tensor size regardless of GPU count, and uses every link at once. That is why it scales.
• NCCL is the collective communication library for NVIDIA GPUs, intra- and inter-node. You select it as a backend; you do not write it.
• Adding GPUs under data parallelism multiplies the effective batch size, which is a hyperparameter change. Re-tune the learning rate and warmup.
• Gradient accumulation buys effective batch size with time rather than hardware, and also reduces AllReduce frequency.
• Tensor parallelism is latency-bound and belongs inside a node; pipeline parallelism tolerates slower links but pays a bubble.
• "Checkpointing" means two different things. Say activation checkpointing for memory and fault-tolerance checkpointing for durability.

Next: 12-05 crosses from training to serving, and it is the pivot of this entire module. Every optimization that follows — batching, PagedAttention, continuous batching, cost per token — depends on one mechanism: the KV cache, the reason an LLM generates tokens one at a time at all, and the reason single-request decoding is limited by memory bandwidth rather than arithmetic.