Distributed Data-Parallel Training

Synchronous data parallelism repeats one pattern each step. Every worker begins from the same replicated parameters. A distributed sampler assigns each worker a distinct micro-batch. Each replica runs forward and backward on its own evidence. An all-reduce combines the local gradients into a shared gradient. Every worker then applies an identical update to parameters and optimizer state.

The fourth step is the one with a price attached, and the price is proved rather than estimated. Patarasuk and Yuan derived it in 2009. Their Lemma 3 states that “The minimum number of partial results to be communicated to complete a one-item all-reduce operation on N processes over all algorithms is 2 x (N - 1)”, which forces at least one process to move ⌈2·((N−1)/N)·X⌉·itsize bytes. Then they draw the consequence: “This indicates that the best of any all-reduce algorithm can do is to achieve roughly 2 times the time to send the msize-byte data.”

Their ring algorithm reaches that floor. Theorem 1 reads “When X is divisible by N, the proposed algorithm is bandwidth optimal.” Nothing goes below it. Read the bound the way a scaling budget has to read it. The reduction costs about two full sends of the gradient buffer, however many workers share the step. Adding workers shrinks the local computation. It leaves that cost exactly where it was.

These failures may reduce data diversity or change the objective silently. The first two are not hypothetical. They are the documented default behaviour of the sampler most PyTorch jobs use. PyTorch's own documentation for DistributedSampler says what drop_last=False does: “the sampler will add extra indices to make the data evenly divisible across the replicas”. So the default distortion is duplication of real examples, not truncation.

The duplication is concrete enough that another framework ships a method to undo it. Hugging Face Accelerate documents Accelerator.gather_for_metrics as one that “Gathers input_data and potentially drops duplicates in the last batch if on a distributed system. Should be used for gathering the inputs and targets for metric calculation.” Its worked example is annotated “Assuming two processes, with a batch size of 5 on a dataset with 9 samples” and asserts len(gathered_items) == 9. Nine items back from nine samples. The padded split would otherwise hand the metric ten.

A speedup claim should name which experiment was performed. Strong scaling keeps the global workload fixed and adds workers. The question is how fast one job can finish, the challenge is that communication comes to dominate as local work shrinks, and the metrics are speedup and efficiency. Weak scaling grows the global workload as workers are added. The question is whether larger batches or models can be processed at all, the challenge is that optimization itself changes, and the limit is that it is no longer the same training algorithm. Time-to-quality measures wall-clock time to a fixed validation target. The question is whether scale accelerates useful learning, the challenge is stochastic variance, and the limit is that the choice of target decides the answer.

ResNet-50 trained on ImageNet in one hour, in 2017. That is the reference measurement for strong scaling, and Goyal and colleagues at Facebook AI Research put the claim in one sentence of their abstract: “With these simple techniques, our Caffe2-based system trains ResNet-50 with a minibatch size of 8192 on 256 GPUs in one hour, while matching small minibatch accuracy.” The simple techniques were two. A linear scaling rule — “When the minibatch size is multiplied by k, multiply the learning rate by k.” — and a 5-epoch gradual warmup. On their hardware the implementation “achieves ~90% scaling efficiency when moving from 8 to 256 GPUs”. A tenth of the ideal speedup was already gone at that scale.

The phrase "matching small minibatch accuracy" is checkable, because a different set of institutions checked it. A group at UC Berkeley, CMU and NVIDIA headed a section of its own paper “Reproduce and Extend Facebook's result”. It records the setup independently: “They used data parallelism to process ResNet-50 model on 256 NVIDIA P100 GPUs (equal to 32 NVIDIA DGX-1 stations).” And it puts the two numbers side by side: “The baseline batch size is 256, which achieves 76.40% Top-1 accuracy. The large-batch size is 8192, which achieves 76.26% Top-1 accuracy.” Those figures, not the word "matching", are what a reader can argue with.

Weak scaling asks the question systems metrics cannot answer, and it too has been measured rather than asserted. How many training steps does it take to reach a fixed out-of-sample error as batch size grows? Shallue and colleagues reported the answer in the Journal of Machine Learning Research in 2019. The point where added data parallelism stops buying speedup depends on the workload: they “find extremely large variation between workloads”. The evidence base is stated in the abstract: “Our experimental data is publicly available as a database of 71,638,836 loss measurements taken over the course of training for 168,160 individual models across 35 workloads.” They found no evidence that larger batches degrade generalisation once metaparameters are retuned. OpenAI reached the diminishing-returns conclusion by a different route, introducing the gradient noise scale to predict the largest useful batch size across supervised learning, reinforcement learning and generative models.

Time-to-quality is not advice in this field. It is the definition an industry benchmark works from. The MLPerf Training rules, published by MLCommons, state that “A run is a complete execution of an implementation on a system, training a model from initialization to the quality target.” Then they define the number a submission reports: “A run result is the wallclock time required for a run.” Because that time is noisy, “A benchmark result is the mean of a benchmark-specific number of run results, dropping the highest and lowest results.” The closed division fixes the target per benchmark — in v6.1, 3.3 log perplexity for Llama31_8b, 3.6 for deepseekv3_671b, 0.75 AUC for dlrm_v4_hstu — and vendors report against it. Graphcore discloses its own result in exactly that shape: “For our MLPerf v1.1 results, published in December 2021, we achieved a time to train of 28.3 minutes for ResNet-50 training on ImageNet with 30k images per second throughput and 38 epochs till convergence at 75.9% validation accuracy on an IPU-POD16.” Throughput, epochs and time-to-target are three different claims. That one sentence reports all three.

If workers process disjoint examples and average gradients consistently, the update can match a corresponding global batch. Each of those two conditions is a place where a default can quietly break the equivalence. Disjointness is what the padding sampler gives up when it repeats indices to make the shards divide evenly. Consistent averaging is what uneven masks, per-example weights and variable-length targets give up when the denominators differ across workers while the reduction behaves as though they matched.

The framework may average or sum at different boundaries. Verify the exact semantics rather than compensating twice through the learning rate. A linear scaling rule applied on top of a reduction that already averaged is a second correction to a problem that was solved once.

Begin with a tiny deterministic run before scaling out, and check five things in order.

First, match a single-worker batch: compare gradients and one-step parameter deltas on the same examples. Second, verify sampler coverage — uniqueness, grouping, epoch boundaries and deterministic reshuffling. This is the step with a named trap in it, and the documentation says so outright. The warning on DistributedSampler reads “In distributed mode, calling the set_epoch() method at the beginning of each epoch before creating the DataLoader iterator is necessary to make shuffling work properly across multiple epochs.” Omit that one call and every epoch draws the same shuffle, which no loss curve will tell you. Count the examples the run actually consumed against the size of the dataset, and check whether the difference is the sampler's padding.

Third, verify reduction units: test masks, weights, variable lengths and uneven last batches. Fourth, verify state synchronization across parameters, optimizer buffers, scheduler, scaler and normalization state. Fifth, measure scaling — throughput, efficiency, communication volume, memory and time to quality, reported as separate numbers rather than one.

Every DDP replica commonly stores the full model, gradients, and optimizer state. Microsoft's ZeRO paper puts the premise in one line: “In DP, model parameters are replicated on each device.” Then it turns that into arithmetic. Two bytes per parameter for fp16 parameters, two for fp16 gradients, and K=12 for the fp32 parameter, momentum and variance copies: “Mixed-precision Adam has K = 12. In total, this results in 2Ψ + 2Ψ + KΨ = 16Ψ bytes of memory requirement”. Sixteen bytes per parameter means a 1.5-billion-parameter GPT-2 needs at least 24 GB of model state on every replica. A 7.5-billion-parameter model needs 120 GB under standard DP. Partition the optimizer state 64 ways and that falls to 31.4 GB.

Sharded data parallelism partitions selected states, while tensor, pipeline, or sequence parallelism partition computation differently. PyTorch FSDP, built by a team at Meta AI, states the same limit from a different organisation: DDP-style techniques “require the model to be replicated on every device, which implies that the entire model can be constructed on the target device.” Removing that requirement pays. “With the 175B model, the experiments achieved more than 173 and 186 TFLOPS per GPU with batch size equal to 1 and 2 respectively”, and the paper converts that: “This is equivalent to approximately 55% and 60% of GPU hardware utilization, given that the A100's peak is 312 TFLOPS using the BF16 tensor core.” The run scaled linearly from 128 to 512 GPUs.

Those methods alter communication and failure modes substantially. Introduce them only after the simpler data-parallel reference is correct and profiled. The arithmetic above tells you when you have no choice, and not before.