Parallel Processing

Watch a single worker fall behind a hundred tasks, then see how spreading the same workload across a hundred workers — and eventually thousands of GPU cores — is the idea that makes modern AI possible.

What is the difference between data parallelism and model parallelism? Data parallelism replicates the whole model on every device, gives each a different slice of the batch, and synchronises gradients by averaging them. Model parallelism splits the model itself across devices, either inside each layer (tensor parallelism) or by groups of layers (pipeline parallelism), and synchronises by passing activations between the pieces.

Modern AI workloads are dominated by arithmetic that can be cut into pieces with no dependence between them: multiplying matrices, applying a function to every element, processing different examples of a batch. Parallel computing executes those pieces at the same moment on many arithmetic units, many devices, or both. Training a large model is possible only because the work is spread across thousands of units inside one accelerator and then across many accelerators; serving a model at low latency is possible only because each token's matrix arithmetic is spread the same way.

Parallelism has hard limits, and they are the real subject of this chapter. A step that must wait for the previous step's result cannot be overlapped with it. Work that is split must be combined, which costs communication. The slowest participant decides when everyone is done. Speedup from N workers is therefore almost always below N, often far below, and the engineering skill is in knowing which limit you are hitting. We will build that skill in order: what can run simultaneously, how much a serial remainder costs, the three ways to partition a model (data, tensor, pipeline), the communication that stitches the partitions back together, how they combine on 16 devices, and how all of it fails.

Take a nightly job in the support assistant's knowledge pipeline. A policy update arrives as 100 text chunks. Each chunk is turned into an embedding vector (see Embeddings), the 100 vectors are normalised and merged into one batch, and an index is then built over that batch. Say each embedding takes 2 s, the merge takes 10 s and the index build takes 20 s (illustrative timings on one worker). Run serially the job takes 100 x 2 + 10 + 20 = 230 s.

The shape of the dependencies decides everything that follows. Embedding chunk 37 needs nothing from chunk 36, so all 100 embeddings may run at once. The merge needs all of them. The index build needs the merge. A diagram of such "needs the result of" relations is a directed acyclic graph (DAG): nodes are operations, edges are data dependencies, and any two nodes with no path between them may run simultaneously.

Two quantities describe any such graph. Work T₁ is the total time on one worker: 230 s. Span T_∞ is the length of the longest chain of dependent operations, the critical path: one embedding, then the merge, then the index, 2 + 10 + 20 = 32 s. No number of workers can finish faster than the span, so the best conceivable speedup is T₁/T_∞ = 230/32 = 7.19, even with 100 workers and free communication. With 8 workers, 100 tasks take ceil(100/8) = 13 rounds of 2 s = 26 s, then 30 s of serial tail: 56 s, a speedup of 230/56 = 4.11 and a parallel efficiency of 4.11/8 = 0.51. We will return to why this is below even the 7.19 bound.

Backend engineers already use two different ideas that are easy to blur. Concurrency is structuring a program so that several tasks are in progress during overlapping time periods, whether or not any two of them execute at the same instant. Parallelism is several computations actually executing at the same instant on separate execution resources. An async server handling 10,000 connections on one thread is highly concurrent and not parallel at all: it is multiplexing waiting.

The distinction matters because the two accelerate different bottlenecks. Concurrency hides latency: while one request waits on a network call, the thread serves another. Parallelism adds throughputfor compute: more multiply-adds per second. When the support assistant answers a question it needs a policy search (about 120 ms in this illustration) and, for order questions, an order-lookup tool call (about 200 ms). The two are independent, so issuing them together costs max(120, 200) = 200 ms instead of 320 ms. That saving needs no extra CPU cores; it needs only that the program does not sit idle while it waits.

import asyncio
import time

async def retrieve_policy():      # stand-in for a vector-store call (~120 ms)
    await asyncio.sleep(0.120)
    return ["POL-EU-009#c0002"]

async def lookup_order():         # stand-in for an authorised order-lookup tool (~200 ms)
    await asyncio.sleep(0.200)
    return {"order": "A-1042", "status": "shipped"}

async def main():
    t0 = time.perf_counter()
    docs, order = await asyncio.gather(retrieve_policy(), lookup_order())
    print(f"{(time.perf_counter() - t0) * 1000:.0f} ms", docs, order)   # about 200 ms, not 320

asyncio.run(main())

Parallel execution comes in several concrete forms that you will meet again below. Processes give separate address spaces and, on a multicore CPU, separate cores; in CPython, CPU-bound threads are constrained by the interpreter lock so CPU parallelism usually means processes or native extensions. Threadsshare memory and suit I/O waiting and native code that releases the lock. A GPU kernel launches thousands of lightweight threads, grouped so that neighbouring threads execute the same instruction on different data elements; this is the unit of parallelism inside one accelerator (see CPUs, GPUs and TPUs for why the hardware is built that way). A batch is parallelism over examples: the same computation applied to many inputs, which is what makes a matrix multiply out of what would otherwise be many vector operations. Finally, multiple devicesare parallelism across separate chips, which is where communication appears and where the rest of this chapter spends its time.

Three kinds of dependency structure appear in AI workloads, and they sit at three points on the spectrum from "free parallelism" to "inherently serial".

Elementwise operations: no dependencies at all

Applying an activation function to a tensor with a million elements is a million independent computations: output element i reads only input element i. The dependency graph has no edges between elements, so the span is one element's work. The limit is not dependency but memory traffic: each element must be read and written, and for cheap functions the memory system, not the arithmetic units, decides the speed.

Matrix multiplication: independent outputs, shared inputs

For C = A x B with A of size m x k and B of size k x n, every output element C[i][j] is a dot product of row i of A and column j of B. The m x n outputs are mutually independent: that is m · n tasks, each of length k. Different outputs share inputs, but sharing a read is not a dependency. So the graph is wide and shallow, nearly ideal for parallel hardware, and a transformer layer is mostly a few large matrix multiplications, which is why accelerators dedicate their silicon to them. A concrete size: a 4096 x 4096 weight matrix applied to 1,000 token vectors produces 4,096,000 independent dot products, each of 4,096 multiply-adds, about 2 x 4096 x 4096 x 1000 = 3.4 x 10¹⁰ floating-point operations.

Autoregressive generation: a chain through time

A language model generates token t+1 from tokens up to t. Sampling token t+1 needs the model's output for position t, which needs token t, which had to be chosen first. The token-to-token dependency is a chain: its span equals its work in tokens, and no number of devices can generate token 50 before token 49 exists. (Techniques such as speculative decoding try to propose several tokens and verify them together, but verification still respects the same dependency; they change the constant, not the chain.)

This is why a request splits into two phases with opposite parallel character. Prefillprocesses the whole prompt: all prompt tokens are already known, so their computations for a layer can run together as large matrix-matrix products. Decode then produces output tokens one at a time, each pass handling one new token per sequence. The arithmetic intensity (floating-point operations per byte moved from memory) shows the consequence. For an illustrative 7-billion-parameter model in 16-bit weights, weights occupy 7 x 10⁹ x 2 bytes = 14 GB (decimal). A forward pass costs roughly 2 operations per parameter per token. A 1,000-token prefill is about 2 x 7 x 10⁹ x 1000 = 1.4 x 10¹³ operations while reading those 14 GB once: about 1,000 operations per byte, which keeps arithmetic units busy. One decode step for one sequence is 1.4 x 10¹⁰ operations while reading the same 14 GB: about 1 operation per byte. Prefill is limited by compute; single-sequence decode is limited by memory bandwidth. (Attention and KV-cache traffic are ignored here; they add to both.) Serving systems fight the decode chain by batching many differentrequests into one pass, since requests are independent of each other even though tokens within one are not; that is developed in Inference Infrastructure.

Suppose a job takes time T₁ on one worker, and a fraction p of that time is work that parallelises perfectly while the remaining 1 - p is strictly serial. With N workers the parallel part takes pT₁/N and the serial part is unchanged, so

T(N) = T1 * [ (1 - p) + p / N ]

speedup    S(N) = T1 / T(N) = 1 / [ (1 - p) + p / N ]
efficiency E(N) = S(N) / N
ceiling    S(infinity) = 1 / (1 - p)

For the illustrative value p = 0.9 and N = 8: the bracket is 0.1 + 0.9/8 = 0.1 + 0.1125 = 0.2125, so S = 1/0.2125 = 4.706, not 8, and the efficiency is 4.706/8 = 0.588. Eight workers deliver the throughput of 4.7 and 3.3 of them are, in effect, wasted. The full computation for other worker counts:

The ceiling 1/(1 - 0.9) = 10 is reached only as N grows without bound. Read the table backwards for a planning rule: to get 90% efficiency on 8 workers you need S = 7.2, and solving 1/7.2 = (1 - p) + p/8 gives p = 0.984. A serial fraction of 1.6% is the most you can tolerate. Parallel scaling is won by removing serial time, not by adding workers.

Change p and N below; the Karp-Flatt value recovers 1 - p exactly, which is why the diagnostic works.

Three assumptions sit under the formula, and real systems break each of them. Amdahl fixes the problem size, so it describes strong scaling; if you enlarge the problem with the cluster (weak scaling), the serial fraction shrinks as a share and Gustafson's scaled speedup N - (1 - p)(N - 1) applies: with p = 0.9 and N = 8, 8 - 0.1 x 7 = 7.3. It treats the serial part as constant, though in practice serial work such as launch overhead, tokenisation, input loading, the optimiser's global bookkeeping and checkpoint writing is exactly what grows with scale. And it assumes the parallel part divides perfectly with no overhead, which the next section shows is optimistic. In AI systems typical contributors to the serial fraction are a single-threaded data loader that cannot feed the accelerators fast enough, a Python launch path that issues kernels one at a time, the aggregation step after a gradient computation, and the autoregressive chain itself.

Applied to the embedding job: the parallel fraction is 200/230 = 0.870, so Amdahl predicts 1/(0.130 + 0.870/8) = 4.18 on 8 workers, slightly above the 4.11 computed from the DAG. The gap is load imbalance: 100 tasks over 8 workers is 12.5 tasks each, but tasks are indivisible so the busiest workers do 13 and the others idle for one round. Amdahl does not model granularity; the DAG schedule does.

Speedup S(N) = T(1)/T(N) compares wall-clock time for the same job. Parallel efficiency E(N) = S(N)/N says what fraction of the added hardware did useful work: 1.0 is ideal linear scaling, and an efficiency below roughly 0.7 usually deserves investigation because you are paying for capacity that is waiting. Always specify the baseline: speedup relative to one device of the parallel implementation can differ from speedup relative to the best single-device implementation, and only the latter is honest.

Amdahl's law accounts for one cause of lost efficiency, the serial fraction. Four others show up in practice.

1. 1

   Synchronisation

   Workers meet at barriers (for example at the gradient aggregation in each training step). Every step ends at the pace of the last arrival, so the time spent waiting is the sum over workers of the gap between their finish time and the slowest one's.
2. 2

   Load imbalance

   Work that does not divide evenly, such as variable-length sequences or the 13-versus-12.5 task split above, leaves some workers finished while others run. Total time is the maximum, not the average.
3. 3

   Kernel and launch overhead

   Each operation has a fixed cost to launch and schedule. When the per-device slice of work becomes small (a small matrix, a small batch), that fixed cost no longer amortises and efficiency falls even before communication is counted.
4. 4

   Communication

   Partial results must move between devices. Time on the wire is time not computing unless it can be overlapped with computation, and its cost depends on message size, link bandwidth and latency, covered in the communication section.

A single number that folds all of these together is the experimentally determined serial fraction(the Karp-Flatt metric), computed from a measured speedup: e = (1/S - 1/N) / (1 - 1/N). It is useful for diagnosis because Amdahl predicts that e stays constant as N grows. If measured e is flat, the loss is a fixed serial piece; if it climbs with N, something that scales with worker count (communication, synchronisation, contention) is eating the gain. The experiment section applies it to a table of measurements.

If the model fits in one device's memory, the simplest way to use N devices for training is to give each device a complete copy of the model and a different slice of the batch. Every replica runs the forward and backward pass on its own examples and produces its own gradient. Those gradients are averaged across replicas, and every replica applies the same optimiser update to its copy, so all copies stay identical. One training step therefore has four phases: local forward and backward, gradient aggregation, identical optimiser update, next batch.

Why averaging is correct: the loss of a batch is the mean of per-example losses, so its gradient is the mean of per-example gradients. If each worker computes the mean gradient over its equal-sized shard, then the mean of the worker means is exactly the full-batch mean. Data parallelism with equal shards is therefore mathematically the same computation as one device processing the whole batch (up to floating-point rounding order), only done in 1/N of the time per step. The effective batch size is the per-worker batch times the worker count times any gradient-accumulation steps: 8 workers x 4 sequences x 2 accumulation steps = 64 sequences per optimiser update. Enlarging the effective batch changes optimisation behaviour (learning-rate and warm-up settings usually have to be revisited), so "more workers" is not purely a systems decision.

import numpy as np

rng = np.random.default_rng(0)
X = rng.normal(size=(64, 5))
y = rng.normal(size=64)
w = rng.normal(size=5)

def grad(Xb, yb, w):
    # gradient of mean squared error 0.5 * mean((Xb @ w - yb) ** 2)
    return Xb.T @ (Xb @ w - yb) / len(yb)

full = grad(X, y, w)
shards = np.array_split(np.arange(64), 8)          # 8 workers, 8 examples each
local = [grad(X[i], y[i], w) for i in shards]      # each worker's local gradient
averaged = np.mean(local, axis=0)                  # what all-reduce(avg) returns
print(np.allclose(full, averaged))                 # True: equal shard sizes

uneven = np.array_split(np.arange(64), [10, 20])    # shards of 10, 10, 44 examples
naive = np.mean([grad(X[i], y[i], w) for i in uneven], axis=0)
weighted = sum(len(i) * grad(X[i], y[i], w) for i in uneven) / 64
print(np.allclose(full, naive), np.allclose(full, weighted))  # False True

What data parallelism buys is throughput: the per-step compute is divided by N while each replica's memory use is unchanged. What it costs is twofold. Every replica stores the whole model, so it cannot train a model whose weights, gradients and optimiser state exceed one device. And each step ends with a gradient aggregation of one gradient per parameter, a message the size of the model, whose cost is derived in the next section. A standard mitigation is to start communicating the gradients of late layers while the backward pass is still computing early layers, overlapping communication with compute.

Partitioned work has to be reassembled, and distributed training does that with collectiveoperations in which all devices in a group take part. Three matter most, and the fourth way devices exchange data, point-to-point send/receive, is a plain transfer from one device to one other.

All-reduce combines one value per device with an operation such as sum and leaves the result on every device. It is what data parallelism uses on gradients. Reduce-scatter performs the same reduction but leaves each device with only one distinct piece of the result. All-gatherconcatenates one piece per device so that everyone ends with the whole. The relation shown in the figure is exact: an all-reduce can be implemented as a reduce-scatter followed by an all-gather, and that composition is the bandwidth-optimal ring all-reduce. Reduce-scatter and all-gather also appear on their own when state is sharded: parameters are all-gathered just before a layer needs them, and gradients are reduce-scattered so each device holds only the gradient shard it owns. Point-to-point transfers carry activations from one pipeline stage to the next.

The figure below steps through a 4-device ring all-reduce using real numbers, so you can watch reduce-scatter and then all-gather happen.

Cost of a ring all-reduce

In a ring over N devices with a message of S bytes, the message is cut into N chunks. Reduce-scatter takes N - 1 steps, each device sending one chunk of size S/N per step; all-gather takes another N - 1. Each device therefore sends 2(N - 1)/N x S bytes in total, a number that is almost independent of N: about 2S for large rings. A simple time model, with per-step latency α and per-link bandwidth β, is

T = 2 (N - 1) * alpha  +  2 (N - 1)/N * S / beta

Example (illustrative values):
  gradients of a 1B-parameter model in 16-bit  ->  S = 2 GB (decimal)
  N = 8 devices, beta = 50 GB/s per link, alpha = 10 microseconds

  bytes sent per device = 2 * 7/8 * 2 GB = 3.5 GB
  bandwidth term        = 3.5 GB / 50 GB/s = 0.0700 s   = 70.0 ms
  latency term          = 2 * 7 * 10 us    = 0.00014 s  =  0.14 ms
  T                     = 70.14 ms

If one step's compute takes 400 ms (illustrative) and the all-reduce is not overlapped, the step is 470 ms and the communication share is 70/470 = 15%. If the backward pass can hide 60 ms of it, only 10 ms is exposed. The engineering question is always exposed communication, the part not hidden behind computation.

Topology matters because links are not equal. Devices inside one server are commonly joined by a much faster interconnect than the network between servers; the ratio is a property of the specific hardware, and that hardware is the subject of AI Accelerators, so no specific figures are claimed here. The consequence is a placement rule that the hybrid section uses: put the parallelism dimension that communicates most often and in the largest volume, tensor parallelism, on the fastest links, and the dimension that communicates least often per unit of compute on the slowest. A ring that crosses a slow link runs at the speed of that link, because every step waits for the slowest hop.

When a single layer's weights, or its activations at the batch size you want, do not fit on one device, the layer itself is divided. The idea follows directly from the matrix multiplication structure above. Take Y = X W with X = [1, 2] and W = [[1, 2], [3, 4]]. Two splits are possible:

X = [1, 2]        W = [[1, 2],
                       [3, 4]]       Y = X W = [1*1 + 2*3, 1*2 + 2*4] = [7, 10]

Column split (each device holds some OUTPUT columns, X is replicated):
  device 0: W0 = [[1],[3]]   Y0 = X W0 = [7]
  device 1: W1 = [[2],[4]]   Y1 = X W1 = [10]
  result = concat(Y0, Y1) = [7, 10]            needs: all-gather (or keep sharded)

Row split (each device holds some INPUT rows, X is split to match):
  device 0: X0 = [1], W0 = [[1, 2]]   P0 = X0 W0 = [1, 2]
  device 1: X1 = [2], W1 = [[3, 4]]   P1 = X1 W1 = [6, 8]
  result = P0 + P1 = [7, 10]                   needs: all-reduce (sum of partials)

A transformer's feed-forward block is two matrix multiplications with an elementwise nonlinearity between them. The standard arrangement splits the first by columns and the second by rows. After the column split each device holds its own slice of the hidden activations; the nonlinearity is elementwise so it runs locally with no communication; the row-split second multiplication consumes exactly the slice the device already has and produces a partial sum of the final output, so a single all-reduce per block finishes it. Attention is divided the same way by assigning whole attention heads to devices. Per transformer layer, this typically costs two all-reduces in the forward pass (one after attention, one after the feed-forward block) and two in the backward pass.

The trade is memory for communication. Weights and the matmul work of the layer shrink by the tensor-parallel degree T, but an all-reduce of the layer's activations sits on the critical path of every layer, at every step, and cannot be deferred or overlapped as easily as a once-per-step gradient reduction. Quantify it for an illustrative layer: 8,192 tokens in the batch, hidden size 4,096, 16-bit activations. The activation tensor is 8192 x 4096 x 2 = 67.1 MB (decimal). A ring all-reduce among T = 4 moves 2 x 3/4 x 67.1 = 100.7 MB per device; at an assumed 400 GB/s intra-node link that is 0.25 ms. Two per layer, 32 layers: 64 x 0.25 ms = 16.1 ms per forward pass of pure communication. Whether that is acceptable depends on the layer compute it accompanies; the point is that the volume scales with the batch and the number of layers, so tensor parallelism is practical only over very fast links, in practice inside a single server, and its degree rarely grows large.

Pipeline parallelism assigns consecutive groups of layers to different devices, called stages: layers 1-8 to stage 0, 9-16 to stage 1, and so on. Only the activations at the stage boundary cross between devices, and only once per boundary per microbatch, so communication is a modest point-to-point transfer rather than a collective. That makes pipelining the usual choice for spanning slower links between servers. The cost is idleness: stage 1 cannot start on a batch until stage 0 has produced its output.

If you send the whole batch through, only one stage works at any moment: with p stages, utilisation is 1/p. The fix is to split the batch into m microbatches and stream them: as soon as stage 0 finishes microbatch 1 it hands it on and starts microbatch 2 while stage 1 works on microbatch 1. The pipeline fills, runs in steady state, then drains.

With equal time per stage per microbatch, the whole pass takes m + p - 1 slots rather than the m a perfectly overlapped schedule would need: the first p - 1 slots are pipeline fill and the last p - 1 are drain. Each stage is busy m slots out of m + p - 1, so the fraction of time lost to the pipeline bubble is

bubble fraction = (p - 1) / (m + p - 1)          p = stages, m = microbatches

Check with p = 4, m = 4 (figure):
  slots           = m + p - 1 = 7
  busy cells      = p * m     = 16
  total cells     = p * slots = 28
  idle cells      = 28 - 16   = 12   ->  12/28 = 3/7 = 0.4286 = (4-1)/(4+4-1)   OK

More microbatches (p = 4):
  m = 1 : 3/4  = 75.0%        m = 8  : 3/11 = 27.3%
  m = 4 : 3/7  = 42.9%        m = 16 : 3/19 = 15.8%
                              m = 32 : 3/35 =  8.6%

Two forces pull against each other here. More microbatches shrink the bubble, but each microbatch is then smaller, so its matrix operations are less efficient per device (the kernel-overhead effect from earlier), and activations for more in-flight microbatches must be held for the backward pass. Schedules differ in how they manage that. The simple fill-then-drain schedule keeps activations for all m microbatches alive. A one-forward-one-backward (1F1B) schedule interleaves a backward step as early as possible, so that in the standard analysis at most about p microbatches are in flight; with equal forward and backward times it has the same bubble fraction but a lower activation-memory peak. Interleaving several small, non-adjacent stage chunks per device can shrink the bubble further at the price of more communication. Scheduling, not just stage count, is therefore the design lever, and unequal stages are a separate hazard: the pipeline runs at the speed of its slowest stage, so layer boundaries are chosen to balance time, not layer count.

The strategies so far partition examples, weights or layers. A different pressure appears when a single sequence is long. Self-attention compares every token with every other, so its work grows with the square of sequence length n. For n = 128,000 tokens there are 128,000² = 1.64 x 10¹⁰token pairs per attention head per layer; materialising that score matrix in 16-bit form would be about 32.8 GB (decimal) for a single head of a single layer. Optimised attention kernels avoid materialising it, but the arithmetic and the key/value state still grow with length, and activation memory at large length can exceed one device even if the weights fit.

The response is to partition the sequence dimension: with 8 devices each holds a contiguous 16,000-token block. Operations that act on tokens independently (feed-forward layers, normalisation) then need no communication, because each token's computation is local. Attention is the exception, since a query on device 0 needs keys and values from all other blocks. Designs for this exchange key/value blocks around a ring while computing, so transfer overlaps with arithmetic, or redistribute data so that each device holds all tokens for a subset of heads. These add communication that scales with the sequence length and are one more reason context length has a systems cost, not only a modelling one. The terminology is not uniform: in some frameworks "sequence parallelism" refers to a narrower trick that shards sequence-wise activations inside a tensor-parallel group to save memory, and "context parallelism" to the attention-spanning scheme above. This chapter makes no claim about a single universal implementation; check the definition used by whatever framework you adopt. For the application-level view of long contexts see Context Window.

Frontier-scale training rarely uses one strategy. Each has a different resource profile: data parallelism multiplies throughput but not capacity; tensor parallelism divides memory but needs the fastest links; pipeline parallelism divides memory with light communication but adds bubbles. The practical recipe is to compose them so each is applied where its cost is cheapest. The device count is the product of the degrees: devices = DP x TP x PP.

Take a toy cluster of 16 devices arranged as four servers of four devices (illustrative), and choose DP = 2, TP = 4, PP = 2, so 2 x 4 x 2 = 16. Give every device a coordinate (dp, pp, tp) and the rank 8 x dp + 4 x pp + tp.

Read the figure as three overlapping groupings of the same 16 devices. Tensor-parallel groups are the four-device sets that share a (dp, pp): 16/4 = 4 groups, each living inside one server so their per-layer all-reduces use the fastest links. Pipeline groups are pairs sharing (dp, tp), for example GPU 0 and GPU 4: 16/2 = 8 groups, each pair exchanging stage-boundary activations across servers. Data-parallel groups are pairs sharing (pp, tp), for example GPU 0 and GPU 8: 16/2 = 8 groups, and each pair all-reduces only the gradients of the one weight shard they have in common, so each all-reduce moves 1/8 of the model, not all of it.

Now the arithmetic that justifies the layout, for an illustrative 40-billion-parameter model with 32 layers and 32 attention heads. Layers are divided by PP: 32/2 = 16 per stage. Heads are divided by TP: 32/4 = 8 per device. Each device therefore holds 1/(4 x 2) = 1/8 of the weights. With 16 bytes per parameter of training state the model is 640 GB, so each device would hold 80 GB before activations: no room on an illustrative 80 GB device. Sharding the optimiser state across the two data-parallel replicas (a ZeRO-style technique, see Training Infrastructure) gives weights and gradients 4 x 40 x 10⁹ / 8 = 20 GB plus optimiser state 12 x 40 x 10⁹ / 16 = 30 GB: 50 GB, leaving 30 GB for activations. Batch geometry: with 8 microbatches of 4 sequences per replica, the global batch is 2 x 8 x 4 = 64 sequences, and the pipeline bubble with p = 2, m = 8 is 1/9 = 11.1%.

The constraints are as important as the multiplication. Every degree must divide the quantity it partitions: layers by PP, attention heads and hidden dimensions by TP, the global batch by DP times the microbatch size. The product must equal the device count, which makes some cluster sizes awkward (a prime number of devices permits only trivial layouts). Tensor-parallel groups should not span the slow network, which caps TP at the number of devices per server in this toy example. Larger PP lowers per-device memory but increases bubble fraction unless microbatches grow with it. Finding a good point in this space is part analysis (as above) and part measurement, because the real trade-offs depend on delivered link speeds and kernel efficiency.

Inference uses the same ideas with different priorities, because the objective shifts from finishing a long job to answering each request quickly. Three mechanisms apply, and they address different quantities.

Data-parallel replicas put identical copies of the model behind a load balancer. Requests are independent, so this scales throughput almost linearly with replica count while the queueing delay falls; it does not shorten a single request, whose per-token time is set by one device. Adding replicas is the correct response to "requests wait too long in the queue" and the wrong response to "each answer streams too slowly". If your assistant calls a hosted LLM API, this replication happens inside the provider; what you control is client-side concurrency within the provider's rate limits.

Tensor-parallel serving spreads one replica over several devices. It has two motivations: the weights may not fit on one device, and it lowers per-token latency. Single-sequence decode is memory-bandwidth bound as computed earlier, so time per token is roughly weights bytes divided by memory bandwidth. With the illustrative 14 GB model and a 1,000 GB/s device, that is 14 ms; with TP 4 each device reads 3.5 GB, taking 3.5 ms. But each layer now ends with two all-reduces. A decode step moves tiny messages (one token's hidden vector, about 8 KB at hidden size 4,096 in 16 bits), so the cost is dominated by latency α, not bandwidth: with an assumed 15 µs per all-reduce and 32 layers x 2 = 64 all-reduces, communication adds 0.96 ms. The result is 3.5 + 0.96 = 4.46 ms, a speedup of 14/4.46 = 3.14 and an efficiency of 0.78 on 4 devices. Communication is why 4 devices give 3.1x rather than 4x, and why latency-bound all-reduces limit how far TP degree can rise.

Request-level concurrency is the third mechanism, and it is where parallelism and the autoregressive chain meet. Different requests have no dependence on each other, so a serving engine batches their decode steps into one pass and, with continuous batching, admits new requests and retires finished ones every step rather than waiting for the whole batch. This raises hardware utilisation during memory-bound decode, at the price of some added latency per request; it is covered in depth in Inference Infrastructure, and the endpoint, routing and autoscaling side in Model Serving. Pipeline parallelism is used for serving too, mainly when a model must span more devices than one server holds, and there it adds hop latency to every token.

Formulas predict; only measurement tells you what your system does. A good scaling experiment answers three questions: how does time fall as workers are added, how much of the remaining time is communication or waiting, and where is the point of diminishing returns. Design it so the answers are not artefacts.

1. 1

   Fix the workload and choose the scaling mode

   Strong scaling holds the total work constant (same data, same batch) and varies workers 1, 2, 4, 8. Weak scaling holds work per worker constant. Report which you ran; they answer different questions and use different formulas (Amdahl versus Gustafson).
2. 2

   Control the environment

   Warm up and discard the first run (caches, lazy initialisation, JIT or compilation). Repeat each setting at least five times and report the median plus spread, not the best run. Pin the hardware and note what else shares the machine.
3. 3

   Instrument phases separately

   Time start-up, the parallel compute phase, the communication or combine phase and any serial remainder with their own timers, and record per-worker times so imbalance is visible. For accelerators, use a profiler to separate kernel time from time inside collectives.
4. 4

   Compute the derived metrics

   Speedup against the best single-worker implementation, efficiency, and the Karp-Flatt serial fraction e at each N, so a flat e suggests a fixed serial part and a rising e suggests overhead growing with workers.

import statistics
import time
from multiprocessing import Pool

def work(n):
    s = 0
    for i in range(n):
        s += i * i % 7
    return s

def run(workers, tasks=64, size=300_000):
    t0 = time.perf_counter()
    with Pool(workers) as pool:
        t1 = time.perf_counter()
        results = pool.map(work, [size] * tasks, chunksize=1)
        t2 = time.perf_counter()
    reduced = sum(results)
    t3 = time.perf_counter()
    return t1 - t0, t2 - t1, t3 - t2          # start-up, parallel phase, combine

if __name__ == "__main__":
    run(2)                                      # warm-up, discarded
    base = None
    for n in (1, 2, 4, 8):
        trials = [run(n) for _ in range(5)]
        startup, parallel, combine = (statistics.median(t[k] for t in trials) for k in range(3))
        total = startup + parallel + combine
        base = base or total
        s = base / total
        print(f"{n}: startup={startup:.3f}s parallel={parallel:.3f}s combine={combine:.4f}s "
              f"speedup={s:.2f} efficiency={s / n:.2f}")

Now read a table of measurements, here hypothetical numbers to demonstrate the analysis: a job takes T(1) = 100 s, and 56, 33 and 22 s on 2, 4 and 8 workers.

The serial fraction e is flat at about 0.11 across worker counts. That is the signature of a fixed serial portion (an Amdahl ceiling near 1/0.11 = 9), not of overhead growing with N, so the next step is to find and shrink the serial part (data loading, combine step, start-up) rather than to tune communication. If the same table had shown e rising from 0.11 to 0.20 as N grew, you would instead suspect synchronisation or a saturated shared resource such as memory bandwidth or network. Diminishing returns are visible directly: the step from 4 to 8 workers costs twice the hardware for a 1.5x gain (3.03 to 4.55).

A single-process program fails by crashing; a parallel job has additional failure modes caused by the very coupling that makes it fast. Because collectives are synchronisation points, the health of the whole job depends on the worst participant.

Stragglers

One slow worker slows everyone, since every synchronous step ends when the last worker arrives. Causes include a thermally throttled or failing device, a noisy neighbour, a slow disk, or a data shard that happens to hold more work. Consider four workers where three finish compute in 100 ms and the fourth takes 150 ms; the all-reduce takes 20 ms.

The step takes 170 ms instead of 120 ms, an efficiency of 120/170 = 0.71 for that step, and the idle time is 3 x 50 = 150 worker-ms out of 4 x 170 = 680 (22%). The mean per-worker compute was 112.5 ms; the slowest was 150, and the slowest sets the pace. Mitigation begins with detection: per-worker step-time histograms make the outlier obvious, and replacing or excluding a persistently slow node is often the real fix. Algorithmic responses exist (asynchronous or bounded-staleness updates, backup workers) but change the optimisation and are used sparingly.

Uneven microbatches

Sequences of different lengths produce different compute times. Within a data-parallel group this creates stragglers; inside a pipeline it makes stage times unequal, widening bubbles beyond the formula's idealised value; and unequal shard sizes break the plain-average gradient shown earlier. Remedies are length-bucketed batching, packing multiple short sequences into one fixed-length sample, and weighting gradients by token count.

Worker crashes and distributed hangs

When a worker dies, the others are usually inside a collective waiting for its contribution. They do not crash; they block. A job that stops making progress without any error message is the characteristic symptom of a distributed hang, and a dead worker, a stuck network link or a mismatched collective order (two ranks executing different collectives) all produce it. The defence is a watchdog: collective timeouts, a launcher that monitors worker exit status, and a policy to kill the whole group and restart it from the last checkpoint, because a partially alive group cannot make progress.

Recovery cost is lost work. If failures are independent and each device has an illustrative mean time between failures of 50,000 hours, a job on 1,024 devices sees a failure about every 50,000/1,024 = 48.8 hours on average, and the rate grows linearly with job size. Each failure discards on average half a checkpoint interval of work plus restart time, while each checkpoint itself costs time to write. A first-order optimum for the interval (Young's approximation) is sqrt(2 x C x MTBF); with a 300 s checkpoint cost and the 48.8 h (175,781 s) cluster MTBF that is sqrt(2 x 300 x 175,781) = 10,270 s, about 2.85 hours. This is only a planning estimate under assumptions (independent failures, constant rate), but it shows why checkpoint cadence is a computed trade-off, not a habit; what a checkpoint must contain and how it is written asynchronously is the subject of Training Infrastructure.

A last subtlety is reproducibility. Floating-point addition is not associative, and a collective may sum contributions in an order that depends on topology and group size. Two runs that are mathematically identical can therefore differ in the last bits, and a restart on a different number of workers will not reproduce the previous trajectory exactly. Treat bitwise-identical resumption as a property you must engineer and test, not one you get for free.

The pieces above combine into a sequence of questions, asked in order of increasing cost. First, does the model, with its training state and a useful micro-batch of activations, fit on one device? If yes, use data parallelism alone: it has the simplest communication and the best-understood scaling, and its efficiency is a function of how well the gradient all-reduce overlaps with backward compute. If no, first reduce the footprint without adding new communication patterns (shard optimiser state across the data-parallel group, recompute activations rather than store them). If it still does not fit, add tensor parallelism inside a server up to the number of fast-linked devices, since its per-layer collectives need the best links. If the model must still span more than a server, add pipeline stages across servers and raise the microbatch count until the bubble formula (p - 1)/(m + p - 1) is acceptably small. Add data-parallel replicas on top to use the remaining devices. Treat long sequences as an additional, separate axis when activation memory at the target context length forces it.

For the assistant in this course, most of this is not your decision. A team calling a hosted LLM API owns only the client-side concurrency shown earlier, and the provider owns the parallel layout. The decisions on this page become yours when you fine-tune or self-host a model, and then they must be made with measured, not assumed, numbers: run the scaling experiment, find the serial fraction, and compute the exposed communication before adding hardware. For agents that fan out several tool calls at once, the same dependency-graph reasoning applies; see Tools and Tool Calling.

Parallelism explains why thousands of simple arithmetic units beat a few powerful ones for this workload, and what stops them from scaling perfectly. What it does not explain is how a particular chip is organised: the memory hierarchy, the matrix units, the interconnect and the numeric formats that decide how much of the parallel work actually runs at hardware speed. That is the next layer of the stack.

1. 1

   Draw the dependency graph before choosing a strategy: work T1 and span T∞ bound what any number of workers can achieve, and a workload is only parallel to the extent the graph allows.

2. 2

   Amdahl's law sets the ceiling at 1 / (1 − p): with p = 0.9 the limit is 10× however many devices you add, and the experimentally measured Karp–Flatt serial fraction tells you whether lost speedup is a serial part or overhead.

3. 3

   Data parallelism replicates the model and splits the batch, so it only works when the model plus its training state fits on one device (about 16 bytes per parameter under standard mixed-precision Adam); gradients are averaged with an all-reduce.

4. 4

   Collectives cost real time: a non-overlapped all-reduce adds directly to the step (400 ms of compute becomes 470 ms in the page's example), and topology matters because intra-server links are much faster than inter-server links.

5. 5

   Tensor parallelism splits a layer and needs frequent communication, so keep it inside a fast-linked server; pipeline parallelism splits by layers and pays a bubble that more microbatches shrink at the price of smaller matmuls and more in-flight activations.

6. 6

   In a hybrid layout every parallel degree must divide the quantity it partitions, and sharding optimiser state (ZeRO-style) can be what makes a 40-billion-parameter model fit.

7. 7

   Serving uses the same ideas with different priorities (replicas for throughput, tensor parallelism to fit or speed up a replica), and a scaling experiment should report efficiency and the measured serial fraction, not just speedup.