Home › Neural network training & inference › Multi-GPU parallelism
algorithm

Ring all-reduce

Problem Data-parallel training on many GPUs must add up large gradient buffers from every worker each step. Without a good scheme, communication time grows with worker count and becomes the bottleneck.

Sum a buffer across N workers via reduce-scatter then all-gather around a ring; each link carries 2(N−1)/N of the buffer, which is bandwidth-optimal.

time ≈ 2(N−1)·α + 2(N−1)/N · S/B; per-worker bytes sent 2(N−1)/N · S (bandwidth-optimal, independent of N)
collectiveallreducedata-parallelncclreduce-scatterall-gather
Overview

Data-parallel training must sum every gradient across N GPUs each step. The ring splits the buffer into N chunks. In the reduce-scatter phase, each worker passes a chunk to its neighbor and adds the chunk it receives; after N−1 steps each worker owns one fully summed chunk. All-gather then circulates the finished chunks. Every link is busy every step.

Key facts

Use when

  • Large messages: gradient all-reduce, ZeRO reduce-scatter and all-gather.
  • Any bulk-synchronous cluster job that needs a global sum: MPI_Allreduce in HPC codes (Thakur et al.), and Horovod-style data-parallel training in other frameworks.

Avoid when

  • Small messages at large N: use double binary trees.
  • Only shards are needed: reduce-scatter alone is half the traffic.

Deep dive

How it works

In: a buffer of S bytes on each of N ranks, arranged in a ring · Out: the elementwise sum, identical on every rank · State: none between calls (chunk offsets are derived from the rank) · Cost: 2(N−1) steps, 2(N−1)/N · S bytes sent per rank, independent of N for large N

Ranks 0..N−1 form a ring, and the buffer is split into N chunks.

reduce-scatter, step s = 0..N−2:
   rank r sends chunk (r − s) mod N to r+1, receives chunk (r − s − 1) mod N from r−1, adds it
   → after N−1 steps rank r holds the complete sum of chunk (r + 1) mod N
all-gather, step s = 0..N−2:
   rank r sends chunk (r + 1 − s) mod N, receives chunk (r − s) mod N, overwrites
   → every rank has every summed chunk

N=4, chunk c travels  r=c → c+1 → c+2 → c+3  accumulating, then keeps going to broadcast

Cost model (α = per-message latency, B = link bandwidth, S = bytes):

  • Ring: 2(N−1)α + 2(N−1)/N · S/B.
  • Double binary tree: about 2 log₂N · α + 2 · S/B. Bandwidth-optimal up to a constant, with logarithmic latency. NCCL has used it since 2.4.
  • Rabenseifner (recursive-halving reduce-scatter + recursive-doubling all-gather): 2 log₂N · α + 2(N−1)/N · S/B for power-of-two N.

Code

A runnable ring all-reduce over threads and channels, with the same index arithmetic NCCL's ring kernel uses.

use std::ops::Range;
use std::sync::mpsc::{channel, Receiver, Sender};
use std::thread;

fn chunk(c: usize, n: usize, len: usize) -> Range<usize> { c * len / n..(c + 1) * len / n }

/// Called on every rank: `tx` goes to rank+1, `rx` comes from rank−1.
fn ring_allreduce(rank: usize, n: usize, data: &mut [f32], tx: &Sender<Vec<f32>>, rx: &Receiver<Vec<f32>>) {
    let len = data.len();
    for s in 0..n - 1 {                                   // reduce-scatter
        let send = (rank + n - s) % n;
        let recv = (rank + 2 * n - s - 1) % n;
        tx.send(data[chunk(send, n, len)].to_vec()).unwrap();
        let incoming = rx.recv().unwrap();
        for (d, x) in data[chunk(recv, n, len)].iter_mut().zip(incoming) { *d += x; }
    }
    for s in 0..n - 1 {                                   // all-gather
        let send = (rank + 1 + n - s) % n;
        let recv = (rank + n - s) % n;
        tx.send(data[chunk(send, n, len)].to_vec()).unwrap();
        let incoming = rx.recv().unwrap();
        data[chunk(recv, n, len)].copy_from_slice(&incoming);
    }
}

fn main() {
    let (n, len) = (5, 1003);                             // uneven chunks on purpose
    let (txs, rxs): (Vec<_>, Vec<_>) = (0..n).map(|_| channel::<Vec<f32>>()).unzip();
    let mut bufs: Vec<Vec<f32>> = (0..n).map(|r| (0..len).map(|i| (r * len + i) as f32).collect()).collect();
    let expect: Vec<f32> = (0..len).map(|i| (0..n).map(|r| (r * len + i) as f32).sum()).collect();

    thread::scope(|sc| {
        for (rank, (buf, rx)) in bufs.iter_mut().zip(rxs).enumerate() {
            let tx = txs[(rank + 1) % n].clone();         // send to the right neighbour
            sc.spawn(move || ring_allreduce(rank, n, buf, &tx, &rx));
        }
    });
    for b in &bufs { assert_eq!(b, &expect); }            // every rank holds the identical sum
    println!("ring all-reduce ok: {n} ranks x {len} floats");
}

In PyTorch you never write this. dist.all_reduce(t) calls NCCL, and DDP/FSDP call it per bucket from autograd hooks.

Complexity and constants

  • Example: 1 GB of bf16 gradients (a 500M-param model) on 8 GPUs at an NCCL busbw of about 400 GB/s takes 1 GB × 1.75 / 400 GB/s ≈ 4.4 ms.
  • The same buffer across 64 nodes at about 45 GB/s per GPU takes about 45 ms, which must hide under the backward pass. Hence bucketing and overlap.
  • Per-hop latency is of order 2-10 µs (NVLink a few µs, IB/RoCE more). A ring over 1024 GPUs pays more than 10 ms of pure latency per all-reduce, which is why NCCL switches to trees for small messages and large N.
  • NVLink SHARP (NVLS) on NVSwitch and InfiniBand SHARP do the reduction in the switch. Each GPU then sends S once instead of about 2S.

Efficiency

  • Cost: 2(N−1)α + 2(N−1)/N · S/B (α latency per hop, B link bandwidth). Example from above: 1 GB of bf16 gradients on 8 GPUs at about 400 GB/s bus bandwidth takes about 4.4 ms; across 64 nodes at about 45 GB/s per GPU about 45 ms. Per-hop latency is order of 2-10 µs, so a ring over 1024 GPUs pays over 10 ms in latency alone. Bound: link bandwidth for large buffers, hop latency for small ones.
  • Levers:
    • bucketed all-reduce overlapped with backward (DDP default bucket 25 MB) — hides most of the 45 ms behind compute (Li et al., VLDB 2020) — data-parallel training
    • double binary tree for small messages — about 2·log₂N·α latency instead of 2(N−1)·α (NCCL 2.4) — small buffers, large N
    • hierarchical: reduce-scatter in the node, all-reduce across nodes on 1/8 of the data — cross-node bytes ÷ GPUs per node — multi-GPU nodes
    • in-network reduction (NVLS, InfiniBand SHARP) — each GPU sends S once instead of about 2S — switch support
    • compress or reduce-scatter only (PowerSGD, ZeRO) — fewer bytes on the wire — scarce cross-node bandwidth
  • Naive → standard → state of the art:
    • gather to one rank, sum, broadcast · root link carries (N−1)·S
    • ring all-reduce · 2(N−1)/N · S per rank
    • topology-aware multi-channel rings or trees with in-switch reduction · bandwidth-optimal bytes, log-depth latency

Implementation notes

  • Overlap: start each bucket's all-reduce as soon as its last gradient is produced (backward runs in reverse layer order). DDP's default bucket of 25 MB balances launch count against overlap.
  • Use several rings (NCCL "channels"), each on a different NVLink, SM and NIC, to saturate all links.
  • Hierarchical: reduce-scatter inside the node over NVLink, all-reduce across nodes over IB on 1/8 of the data, then all-gather inside the node. NCCL builds this topology-aware.
  • Reduce in fp32, or at least bf16 with fp32 accumulation, at large N. bf16 sums over 1000+ ranks lose bits.
  • The result is identical on all ranks (each chunk is summed in exactly one place, then copied), so replicas don't drift.

Pitfalls

  • Mismatched collective order across ranks (one rank skips a layer, or unused parameters in DDP): a hang with no error. Set NCCL_DEBUG=INFO and TORCH_NCCL_ASYNC_ERROR_HANDLING, and use find_unused_parameters only when needed.
  • Summation order differs from a single-GPU sum, so results are not bitwise identical across different N.
  • PCIe-only nodes, or a ring that crosses CPU sockets: bandwidth drops 5-10x. Check nvidia-smi topo -m.
  • Too-small buckets are latency-bound. Too-large buckets start late and lose the overlap.
  • Gradient accumulation: skip the all-reduce on the non-final micro-batches (no_sync()), or traffic multiplies.

Variants and alternatives

Double binary treelog-latency, near-optimal bandwidth. Wins for small and medium messages at scale.
Reduce-scatter / all-gather alonethe building blocks of ZeRO/FSDP and sequence parallelism.
In-network reduction (SHARP, NVLS)halves the traffic.
Parameter serverasynchronous and heterogeneous clusters. Rare for large-model training.
Compressed all-reduce (PowerSGD, 1-bit Adam)when cross-node bandwidth is scarce.

In production

NCCL (ring, tree, NVLS, CollNet), RCCL (AMD), Gloo (CPU), Horovod (popularized ring all-reduce for DL), PyTorch DDP and FSDP, JAX/XLA psum, and MPI implementations (MPICH, Open MPI) in scientific computing.

Implementations

namelangrepoinstallverdictlicencehealthas_ofnotes
NCCLC++ / CUDANVIDIA/ncclpip install nvidia-nccl-cu12————ring, tree, NVLS; used by PyTorch and JAX on NVIDIA GPUs
RCCLC++ / HIPROCm/rccln/a (ships with ROCm)————NCCL-compatible collectives for AMD GPUs
Open MPICopen-mpi/ompibrew install open-mpi————MPI_Allreduce with ring and recursive-doubling algorithms
MPICHCpmodels/mpichbrew install mpich————the library benchmarked in Thakur et al.
GlooC++pytorch/gloon/a (library, built with CMake)————CPU collectives, PyTorch's CPU backend

Sources

  • Patarasuk and Yuan, Bandwidth Optimal All-reduce Algorithms for Clusters of Workstations, JPDC, 2009. paper
  • Thakur, Rabenseifner, Gropp, Optimization of Collective Communication Operations in MPICH, IJHPCA, 2005. paper
  • Sergeev and Del Balso, Horovod: fast and easy distributed deep learning in TensorFlow, 2018. paper · code
  • Li et al., PyTorch Distributed: Experiences on Accelerating Data Parallel Training, VLDB 2020. paper
  • NVIDIA, NCCL documentation: Performance (algbw vs busbw). doc · code

Used by

Neural network trainingA team has a model, a dataset and a cluster, and needs the model to reach its best accuracy without diverging, wasting GPU hours or losing a long run to a crash. This is the loop that does the learning.Tensor parallelismA single layer of a very large neural network is too big or too slow for one GPU. Engineers need to divide the layer's work across several GPUs on a fast link.ZeRO and FSDP shardingTraining a model whose weights, gradients and optimizer state do not fit in one GPU's memory, without rewriting the model to split its layers across devices, is the problem for data-parallel training.

Systems (9):

ai acceleratortraining fabricgigawatt ai campusscale up fabricllm pretrainingelastic trainingsurrogate modelingmpi clustermulti datacenter training

Alternatives and related

in network aggregation · plannedReduce gradients or data inside switches (SHARP, SwitchML) to cut allreduce traffic
Pipeline parallelismA neural network is too large to fit on one GPU, and the machines are linked by slow connections. Engineers need to spread consecutive layers across devices without leaving most of them idle.SGD with momentumWhen training a model with noisy minibatch gradients, plain gradient descent is slow and zig-zags. Engineers training vision models want faster, steadier progress without paying for heavy optimizer memory.GPU cloudTeams training or serving neural nets find GPU time is the biggest line on the bill and prices vary widely between providers. This guide answers where to rent GPUs and what trade-offs each kind of provider brings.LLM pretrainingHow is a large language model trained from scratch in practice? This guide covers deciding data mix, scale, schedules and parallel layout so a months-long run on thousands of GPUs stays stable and efficient, and whether to pretrain at all.