Chapter 10 · Section 10.4 · 55 min

Tensor parallelism

ZeRO and FSDP split the storage of a model across GPUs, but every GPU still does all of the math for its own tokens. Before each layer runs, the full weights are gathered onto every GPU, and every GPU stores the full activations of its own sequences. For a 70B model at 8,192 tokens that means 182 GB of activations per sequence, which no amount of sharding removes.

Tensor parallelism takes the opposite approach. Split each weight matrix itself across tt GPUs, have each GPU do 1/t1/t of every matrix multiply on the same tokens, and combine the partial results with a collective. Each GPU then stores 1/t1/t of the weights and computes and stores only part of the activations. Shoeybi et al. (2019) worked out, in the Megatron-LM paper, how to do this for a Transformer with only two all-reduces per layer in the forward pass, using nothing but the algebra of block matrices. This section derives their splits from scratch, checks them in NumPy and PyTorch, counts the communication, and explains why tensor parallelism almost never leaves a single node.

One matrix multiply, split two ways

Take a linear layer Y=XA\mathbf{Y} = \mathbf{X}\mathbf{A} with input X∈Rn×d\mathbf{X} \in \R^{n\times d} (nn tokens, which is bTbT for a micro-batch of bb sequences of TT tokens) and weight A∈Rd×k\mathbf{A} \in \R^{d\times k}, split over tt GPUs. There are two natural ways to cut A\mathbf{A}.

Column parallel. Cut A\mathbf{A} into column blocks, A=[A1∣⋯∣At]\mathbf{A} = [\mathbf{A}_1 \mid \cdots \mid \mathbf{A}_t] with Ai∈Rd×k/t\mathbf{A}_i \in \R^{d\times k/t}. Then

Y=X[A1∣⋯∣At]=[XA1∣⋯∣XAt].\mathbf{Y} = \mathbf{X}[\mathbf{A}_1 \mid \cdots \mid \mathbf{A}_t] = [\mathbf{X}\mathbf{A}_1 \mid \cdots \mid \mathbf{X}\mathbf{A}_t] .

GPU ii holds the full input X\mathbf{X} and its block Ai\mathbf{A}_i, and computes Yi=XAi\mathbf{Y}_i = \mathbf{X}\mathbf{A}_i, the ii-th block of output columns. No communication is needed in the forward pass. The output is left split along its last dimension.

Row parallel. Cut A\mathbf{A} into row blocks Ai∈Rd/t×k\mathbf{A}_i \in \R^{d/t\times k}, and the input correspondingly into column blocks X=[X1∣⋯∣Xt]\mathbf{X} = [\mathbf{X}_1 \mid \cdots \mid \mathbf{X}_t] with Xi∈Rn×d/t\mathbf{X}_i \in \R^{n\times d/t}. Block multiplication gives

Y=[X1∣⋯∣Xt](A1⋮At)=∑i=1tXiAi.\mathbf{Y} = [\mathbf{X}_1 \mid \cdots \mid \mathbf{X}_t]\begin{pmatrix}\mathbf{A}_1\\ \vdots\\ \mathbf{A}_t\end{pmatrix} = \sum_{i=1}^{t} \mathbf{X}_i\mathbf{A}_i .

GPU ii holds Xi\mathbf{X}_i and Ai\mathbf{A}_i and computes a partial sum XiAi\mathbf{X}_i\mathbf{A}_i of the full shape n×kn\times k. The real output is the sum of the partial sums, so the GPUs must all-reduce them.

In plain words, splitting the weight's columns splits the output and needs no communication, while splitting its rows splits the input and leaves partial sums that must be added.

The backward pass mirrors this. With upstream gradient G=∂L/∂Y\mathbf{G} = \partial\mathcal{L}/\partial\mathbf{Y} (§0.2), the column-parallel layer has ∂L/∂Ai=X⊤Gi\partial\mathcal{L}/\partial\mathbf{A}_i = \mathbf{X}^\top\mathbf{G}_i, local to GPU ii, but ∂L/∂X=∑iGiAi⊤\partial\mathcal{L}/\partial\mathbf{X} = \sum_i \mathbf{G}_i\mathbf{A}_i^\top, a sum over GPUs, so the backward pass needs an all-reduce. The row-parallel layer needs the full G\mathbf{G} on every GPU, which it has (the forward all-reduce gave every GPU the full Y\mathbf{Y}), and then ∂L/∂Xi=GAi⊤\partial\mathcal{L}/\partial\mathbf{X}_i = \mathbf{G}\mathbf{A}_i^\top and ∂L/∂Ai=Xi⊤G\partial\mathcal{L}/\partial\mathbf{A}_i = \mathbf{X}_i^\top\mathbf{G} are both local. Each split needs exactly one all-reduce, one in the forward pass and the other in the backward pass.

Python · cell 1
Warm-up

Column- and row-parallel linear layers

Implement column_parallel(X, A, t) returning the list of t output shards [X @ A_1, ..., X @ A_t] for the column blocks of A, and row_parallel(X, A, t) returning (partials, Y), where partials[i] = X_i @ A_i for the matching column blocks of X and row blocks of A, and Y is their sum (the result of the all-reduce). Assume t divides the relevant dimensions, and use np.split so that block i is the i-th contiguous block.

Hidden tests, hints, and AI help, from $1.99 a month.

The Megatron MLP, one all-reduce

Now the feed-forward network of a Transformer block (§5.3), Z=GELU⁡(XA)B\mathbf{Z} = \operatorname{GELU}(\mathbf{X}\mathbf{A})\mathbf{B} with A∈Rd×4d\mathbf{A} \in \R^{d\times 4d} and B∈R4d×d\mathbf{B} \in \R^{4d\times d}. Which way should each matrix be split?

Try splitting A\mathbf{A} by rows first. Then XA=∑iXiAi\mathbf{X}\mathbf{A} = \sum_i \mathbf{X}_i\mathbf{A}_i is a sum of partial products, and we would have to all-reduce it before the GELU, because GELU⁡(U1+U2)≠GELU⁡(U1)+GELU⁡(U2)\operatorname{GELU}(\mathbf{U}_1 + \mathbf{U}_2) \ne \operatorname{GELU}(\mathbf{U}_1) + \operatorname{GELU}(\mathbf{U}_2). That costs a synchronization in the middle of the block.

Split A\mathbf{A} by columns instead. GPU ii gets XAi\mathbf{X}\mathbf{A}_i, a block of columns of the hidden activation, and because GELU is applied elementwise, GELU⁡(XA)=[GELU⁡(XA1)∣⋯∣GELU⁡(XAt)]\operatorname{GELU}(\mathbf{X}\mathbf{A}) = [\operatorname{GELU}(\mathbf{X}\mathbf{A}_1) \mid \cdots \mid \operatorname{GELU}(\mathbf{X}\mathbf{A}_t)]. Each GPU can apply it to its own block with no communication. The hidden activation is now split along its columns, which is exactly the input a row-parallel layer wants. So split B\mathbf{B} by rows, and

Z=∑i=1tGELU⁡(XAi) Bi.\mathbf{Z} = \sum_{i=1}^{t} \operatorname{GELU}(\mathbf{X}\mathbf{A}_i)\,\mathbf{B}_i .

In plain words, each GPU runs a narrow MLP with 4d/t4d/t hidden units on all the tokens, and the narrow MLPs' outputs are added with one all-reduce.

Shoeybi et al. name the two conjugate communication points ff and gg. Operator ff sits where the replicated input enters the parallel region. It is the identity in the forward pass and an all-reduce in the backward pass (the column-parallel ∂L/∂X\partial\mathcal{L}/\partial\mathbf{X} sum). Operator gg sits where the partial sums leave. It is an all-reduce in the forward pass and the identity in the backward pass. So the MLP costs one all-reduce forward and one backward.

Biases need one piece of care. The first layer's bias b1∈R4d\mathbf{b}_1 \in \R^{4d} splits with A\mathbf{A}'s columns, each GPU adding its own slice before the GELU. The second layer's bias b2∈Rd\mathbf{b}_2 \in \R^d is not split, and it must be added once, after the all-reduce, not on every GPU before it, or it would be counted tt times. Gated feed-forward networks such as SwiGLU split the same way, with the gate and up projections column-parallel (with matching column blocks, so that each GPU multiplies the right gate with the right value) and the down projection row-parallel.

How does Megatron split an MLP over t GPUs, and where must they talk?
Colors are GPUs. The first weight matrix is split by columns, so each GPU computes its own block of hidden units and applies GELU locally. The second is split by rows, so each GPU produces a partial sum of the full output.
Loading interactive figure…
Interview

Tensor-parallel MLP on simulated shards

Implement tp_mlp(X, A, b1, B, b2, t) computing GELU⁡(XA+b1)B+b2\operatorname{GELU}(\mathbf{X}\mathbf{A} + \mathbf{b}_1)\mathbf{B} + \mathbf{b}_2 the Megatron way on t simulated GPUs. Shard A and b1 by columns and B by rows into t contiguous blocks, have "GPU" i compute its partial output using only its own shards, all-reduce (sum) the partials, and add b2. Return (Z, partials), the final output and the list of the t partial outputs before the all-reduce. Use the provided gelu.

Hidden tests, hints, and AI help, from $1.99 a month.

Attention, split by heads

Attention splits even more naturally, because its heads are already independent (§4.5). Give each GPU h/th/t of the layer's hh heads. The query, key, and value projections are column-parallel, each GPU holding the columns of WQ\mathbf{W}_Q, WK\mathbf{W}_K, and WV\mathbf{W}_V that belong to its heads. Each GPU then computes full attention for its heads, on all tokens, with no communication at all. The output projection WO\mathbf{W}_O is row-parallel, because the identity of §4.5, [H(1)∣⋯∣H(h)]WO=∑iH(i)WO(i)[\mathbf{H}^{(1)}\mid\cdots\mid\mathbf{H}^{(h)}]\mathbf{W}_O = \sum_i \mathbf{H}^{(i)}\mathbf{W}_O^{(i)}, says that the head outputs times their row blocks of WO\mathbf{W}_O add up. One all-reduce, gg, finishes the attention block, and ff sits at its input.

Two constraints follow. The number of heads must be divisible by tt. With grouped-query attention and nkvn_{kv} key-value heads, each GPU also needs the key-value heads its query heads use, so tt should divide nkvn_{kv}, or the key-value heads are replicated. Llama 3 70B, with 64 query heads and 8 key-value heads, splits cleanly up to t=8t = 8, one key-value head per GPU.

Put together, a Transformer layer under tensor parallelism communicates four times per training step, two all-reduces in the forward pass (gg after attention and after the MLP) and two in the backward pass (ff at their inputs), each of the full activation X∈RbT×d\mathbf{X} \in \R^{bT\times d}. The cell below checks the attention split numerically.

Python · cell 2

Embeddings and the loss

The embedding table and the output layer are the largest single matrices in many models, V×dV\times d each (for Llama 3 70B, 128,256×8192128{,}256 \times 8192, a billion parameters), so they are split too.

Vocabulary-parallel embedding. Split the table by rows, so GPU ii holds the embeddings of token ids in its range of V/tV/t. Each GPU looks up the tokens that fall in its range, writes zeros for the others, and an all-reduce sums the results, so each token's row comes from exactly one GPU.

Vocabulary-parallel cross-entropy. The output layer is column-parallel, so GPU ii ends with logits for its V/tV/t vocabulary entries. Gathering the full bT×VbT\times V logits would be expensive (gigabytes, as §10.1 showed). Instead compute the loss from three small all-reduces. The cross-entropy for a token with logits z\mathbf{z} and target yy is log⁡∑vezv−zy\log\sum_v e^{z_v} - z_y. Each GPU computes its local maximum, and an all-reduce with the max operation gives the global maximum μ\mu for numerical stability. Each GPU computes ∑v∈shardezv−μ\sum_{v\in\text{shard}} e^{z_v - \mu}, and an all-reduce sums them. The GPU that owns yy contributes zyz_y and the others contribute 0, and an all-reduce sums those. Every all-reduce is over bTbT numbers instead of bTVbTV.

Interview

Vocabulary-parallel cross-entropy

Implement vocab_parallel_ce(logit_shards, targets). logit_shards[i] has shape (n, V_i) and holds the logits of vocabulary entries offset_i to offset_i + V_i - 1, where the offsets are the cumulative sizes of the earlier shards. targets has shape (n,) with ids in [0, V). Return the mean cross-entropy over the n tokens. Each shard may only be reduced locally to arrays of shape (n,), and those arrays combined across shards with an elementwise max or sum (the all-reduces). Do not concatenate the shards. It must be stable for logits as large as 10410^4.

Hidden tests, hints, and AI help, from $1.99 a month.

What tensor parallelism costs

Memory. Each GPU holds 1/t1/t of every layer's large matrices, so weights, gradients, and optimizer states all shrink by tt (norm gains and a few biases are replicated, which is negligible). Activations shrink too, but not completely. Korthikanti et al. (2022) redo the count of §10.1 under tensor parallelism. The tensors inside the parallel regions are split, Q\mathbf{Q}, K\mathbf{K}, V\mathbf{V}, the attention matrices, the WO\mathbf{W}_O input, and the two wide MLP tensors, 24Tbd24Tbd bytes plus 5hT2b5hT^2b. The tensors outside them are replicated on every GPU, the two norm inputs, the inputs to the first linear of each block, and the two dropout masks, 10Tbd10Tbd bytes. So

activations per layer=Tbd(10+24t+5hTtd).\text{activations per layer} = Tbd\left(10 + \frac{24}{t} + \frac{5hT}{td}\right) .

The replicated 10Tbd10Tbd is a floor that does not fall with tt, and we remove it below.

Communication. Each all-reduce is of a bT×dbT\times d activation in 16 bits, 2bTd2bTd bytes, so each GPU sends 2(t−1)/t⋅2bTd2(t-1)/t \cdot 2bTd bytes per all-reduce, and four of them per layer per step. Compare with the compute. A layer's forward pass costs about 24bTd224bTd^2 FLOPs for its weight products (§4.7), split over tt GPUs, against two all-reduces. With per-GPU throughput FF and bandwidth B\mathcal{B}, the ratio of communication time to compute time is

2⋅2(t−1)t⋅2bTd / B24bTd2/(tF)=(t−1) F3d B.\frac{2\cdot\frac{2(t-1)}{t}\cdot 2bTd\,/\,\mathcal{B}}{24bTd^2/(tF)} = \frac{(t-1)\,F}{3d\,\mathcal{B}} .

In plain words, the bigger the model's width and the faster the link, the cheaper tensor parallelism is, and every extra GPU in the group makes it more expensive.

The number of tokens has cancelled. Unlike data parallelism, no batch size can hide this communication, because it scales with the tokens exactly as the compute does. Worse, it sits on the critical path. The next layer cannot start until the all-reduce finishes, so it is hard to overlap.

Python · cell 3

For a width-8192 model at t=8t = 8 over NVLink, communication takes about 0.6 times as long as compute at peak throughput (less at realistic throughput, but it adds directly to the step time). Over InfiniBand the same split takes 5.6 times as long as the compute. That is the reason for the rule every practitioner knows. Tensor parallelism stays within a node, usually at t=8t = 8, the number of GPUs sharing an NVSwitch. Larger NVLink domains, such as 72-GPU racks, relax the rule. Splitting matrices finer also makes each GPU's matrix multiplies smaller and less efficient, another reason not to push tt past the point where memory is solved.

Techniques that hide some of this communication do exist. Asynchronous tensor parallelism (or collective matmul, Wang et al. 2023) splits each matrix multiply into chunks and overlaps sending one chunk with computing the next, and fused kernels can overlap the all-reduce with the following computation. They reduce the exposed time but do not change the bytes.

Sequence parallelism removes the replicated activations

Look again at the 10Tbd10Tbd bytes that tensor parallelism leaves replicated. They belong to the LayerNorms, the dropouts, and the residual adds, which work on each token independently. Korthikanti et al. observed that these regions can be split along the sequence instead, each GPU handling T/tT/t of the tokens. They called this sequence parallelism.

The transitions change. Entering a tensor-parallel region, the column-parallel layer needs every token, so the sequence-split activations must be all-gathered. Leaving it, the row-parallel layer's partial sums must be added, and the result is to be split by sequence again. An all-reduce followed by keeping only your own tokens is exactly a reduce-scatter. So the conjugate pair ff, gg is replaced by a new conjugate pair. Korthikanti et al. call them gg, an all-gather in the forward pass and a reduce-scatter in the backward pass, and gˉ\bar g, a reduce-scatter in the forward pass and an all-gather in the backward pass. (They rename Megatron's ff and gg as ff and fˉ\bar f, so watch the letters when you read the two papers side by side.)

Here is the beautiful part. Tensor parallelism did one all-reduce at each point. Sequence parallelism does one all-gather and one reduce-scatter instead, and an all-reduce is a reduce-scatter plus an all-gather (§10.2). So the communication volume is unchanged, and every activation is now split tt ways,

activations per layer=Tbd(34t+5hTtd),\text{activations per layer} = Tbd\left(\frac{34}{t} + \frac{5hT}{td}\right) ,

and with selective recomputation or FlashAttention, 34 Tbd/t34\,Tbd/t. (To remove the whole 10Tbd10Tbd, Korthikanti et al. also store only the local sequence shard of the norm's output, the input of the first column-parallel layer, and all-gather it again during the backward pass when its weight gradient needs it. That one extra all-gather per block is the only added traffic, and they overlap it with computation.)

Python · cell 4
Core

Activation memory and traffic under tensor parallelism

Implement tp_layer_cost(T, b, d, h, t, sp=False, recompute="none") returning (act_bytes, comm_bytes) for one GPT-3-style layer on one GPU of a tensor-parallel group of size t. Activations follow Korthikanti et al.'s Table 2. Without sequence parallelism, "none" gives Tbd(10+24/t+5hT/(td))Tbd(10 + 24/t + 5hT/(td)) and "selective" gives Tbd(10+24/t)Tbd(10 + 24/t). With it, "none" gives Tbd(34/t+5hT/(td))Tbd(34/t + 5hT/(td)) and "selective" gives 34Tbd/t34Tbd/t. With "full" recomputation, keep only 2Tbd2Tbd either way. comm_bytes is what the GPU sends per training step in the layer's tensor-parallel collectives, using 16-bit activations and ring costs. That is four all-reduces of the bT×dbT\times d activation (or the equivalent all-gathers and reduce-scatters with sp), plus two more when recompute="full" reruns the forward pass. With t == 1 there is no communication. Return integers for the activations (round down) and a float for the communication.

Hidden tests, hints, and AI help, from $1.99 a month.

In PyTorch

The ff and gg operators are four lines each as custom autograd functions, which is how Megatron-LM implements them. The code below needs a tensor-parallel process group and is shown for reference.

PyTorch also has tensor parallelism built in, on top of distributed tensors. You describe each submodule's split, and parallelize_module shards the weights and inserts the collectives. With Shard(1) layouts (the sequence dimension) at the boundaries, the same plan gives sequence parallelism. The PrepareModuleInput entries gather the tokens once per block, at its input, rather than once for each of the three projections. This plan follows PyTorch's tensor-parallel tutorial for a Llama-style model, whose attention takes the hidden states and the rotary frequencies.

On a single process we can still check the core claim of this section, that the sharded computation gives exactly the dense gradients, by letting autograd play the role of the collectives. Summing the partial outputs is gg, whose backward is the identity, and giving each shard its own copy of the input makes ff's backward a sum of their gradients.

🔥 In PyTorchfor a machine with PyTorch
data parallel (FSDP)tensor paralleltensor plus sequence parallel
splitsthe batch (and the model states)each matrix, by columns or rowseach matrix, and the sequence outside them
weights per GPU1/ndp1/n_{\text{dp}} with ZeRO-31/t1/t1/t1/t
activations per GPUfull, for its own micro-batchTbd(10+24/t+5hT/(td))Tbd(10 + 24/t + 5hT/(td))Tbd(34+5hT/d)/tTbd(34 + 5hT/d)/t
communicationweights and gradients, about 6N6N bytes per step, overlappable4 all-reduces of activations per layer, on the critical pathsame volume as tensor parallel
scales withtokens per GPUwidth dd and link speedthe same
typical degreehundreds to thousandsup to 8, inside a nodeup to 8, inside a node

Read more

Watch next

🎬 Lecture 48: The Ultra Scale Playbook · GPU MODE · starts at 92:11The tensor parallelism chapter of the playbook, with throughput measured as the degree grows past one node.
🎬 Efficient Large-Scale Language Model Training on GPU Clusters Using Megatron-LM · @ScaleJared Casper of NVIDIA on how Megatron combines tensor, pipeline, and data parallelism.