The Communication Substrate

"I only ever talk to my right-hand neighbour. Somehow, by the end of the round, I am holding everyone's gradient. Nobody has explained to me how this is not gossip."

A Worker On A Ring

The previous section, Section 4.1, argued that communication, not compute, is what bounds distributed training once you add enough machines. That argument leaves a concrete question unanswered: what exactly is "communication", mechanically, at the level a program controls? The answer is smaller than the word suggests. At bottom there is only one operation, one worker handing a block of bytes to another worker across a link. Everything else, including the whole vocabulary of collectives this chapter develops, is a pattern of those single handoffs orchestrated so that many workers reach a shared result. This section builds up from that one operation to the abstraction that organizes it and the hardware that carries it.

1. Point-to-Point: Send and Recv, the Atom of Communication Beginner

The most basic act of communication is a point-to-point transfer: one rank calls send to push a buffer toward a named destination, and the destination rank calls recv to accept it into a buffer of its own. The two calls are a matched pair; a send with no matching recv simply waits, and a recv with no matching send waits too. This pairing is what makes point-to-point both flexible and dangerous. It is flexible because you can express any communication pattern at all by choosing who sends to whom and in what order. It is dangerous because if every rank tries to send before anyone receives, and the buffers are large enough that the system cannot quietly absorb them, every rank blocks waiting for a partner who is also blocked, and the job deadlocks. The discipline of ordering sends and recvs so this cannot happen is the price of working at this level.

Point-to-point comes in two flavors that matter for performance. A blocking send does not return until the buffer is safe to reuse; a non-blocking send returns immediately and hands you a handle you later wait on, which lets a worker post a transfer and keep computing while the bytes move. That overlap, computing during a transfer instead of stalling for it, is the seed of one of the most important optimizations in the whole chapter, the overlap of gradient communication with the backward pass that Section 4.10 develops in full. For now the point is narrower: send and recv are the only two verbs the hardware truly needs, and the cost of one such transfer of $n$ bytes is exactly the alpha-beta cost $\alpha + n\beta$ from Section 3.8.

2. Collective Operations: One Call, Every Rank Participates Beginner

Writing distributed training as raw send and recv would be both tedious and error-prone, so the field standardized a small set of named patterns, the collective operations, in which a single call is issued by every participating rank and the library executes the underlying schedule of point-to-point messages for you. The defining feature of a collective is that all ranks call it together: where point-to-point names a single source and a single destination, a collective names the whole group and produces a result defined over the whole group. An all-reduce, the collective that sums one vector per rank and leaves the sum on every rank, is the one that synchronizes gradients in data-parallel training and the subject of Section 4.3; it is the same operation you performed by hand in Section 1.1 when eight workers combined their partial gradients.

The value of the collective is not only convenience. Because the library knows the full pattern in advance, it can pick a schedule tuned to the hardware (a ring when bandwidth dominates, a tree when latency dominates) and overlap the hops optimally, which a hand-rolled sequence of sends almost never achieves. The trade is expressiveness for performance and safety: you give up the freedom to specify an arbitrary message pattern, and in return you get a deadlock-free, topology-aware implementation. Almost all of distributed deep learning is expressed in this collective vocabulary, which is why the rest of this chapter is organized as one section per collective. Table 4.2.1 places point-to-point beside the collectives it composes into.

3. The Process Group: A World of Ranks Beginner

For any of this to work, the participants must be named, and that naming is the job of the process group, also called a communicator. When a distributed job starts, the framework forms a group of all the worker processes and assigns each one an integer rank from $0$ to $K-1$; the total count $K$ is the world size, and the default group containing everyone is the world. A send addresses its destination by rank; a collective acts over a group, so it sums or gathers exactly the ranks in that group and no others. The process group is the small piece of shared bookkeeping that turns a pile of independent processes into a coordinated team that can speak the collective vocabulary.

Groups can be carved into subgroups, and this is not a detail but a central tool. Three-dimensional parallelism splits a model along data, tensor, and pipeline dimensions at once, and each dimension is its own process group: a tensor-parallel all-reduce runs only over the few ranks that share a layer, while a data-parallel all-reduce runs over the ranks holding replicas, and the two groups overlap on the same physical machines. Forming the right subgroups so that frequent, latency-sensitive collectives stay inside one machine while rare, bandwidth-heavy ones cross the network is exactly the topology-aware placement that Section 4.9 is devoted to. The process-group abstraction is what makes that placement expressible: you decide which ranks share a group, and the library confines each collective to it.

# Launch with: torchrun --nproc_per_node=4 thisfile.py
import torch, torch.distributed as dist

dist.init_process_group("nccl")          # form the world; assign ranks
rank = dist.get_rank()                    # this process's id, 0 .. K-1
world = dist.get_world_size()             # K, the world size

# Point-to-point: rank 0 sends one tensor to its right neighbour, which recvs it.
buf = torch.full((3,), float(rank), device="cuda")
if rank == 0:
    dist.send(buf, dst=1)                 # blocking send to rank 1
elif rank == 1:
    dist.recv(buf, src=0)                 # blocking recv from rank 0

# A subgroup over just the even ranks; a collective on it touches only those.
evens = dist.new_group(ranks=list(range(0, world, 2)))
if rank % 2 == 0:
    dist.all_reduce(buf, group=evens)     # confined to the even-rank world

4. Building a Collective From Bare Send and Recv Intermediate

The cleanest way to believe that a collective is just choreographed point-to-point is to build one. The simulation below arranges $K$ ranks on a logical ring where each rank only ever talks to its right neighbor, then implements an all-reduce-by-sum using nothing but those one-hop sends. A first set of $K-1$ hops walks a running sum around the ring until one rank holds the total; a second set of $K-1$ hops walks that finished total back around so every rank ends with it. We count every message and check the result against the single-machine sum, so the claim "no approximation, just regrouping" from Section 1.1 is verified rather than asserted.

import numpy as np

K = 4
rng = np.random.default_rng(0)
local = [rng.integers(1, 10, size=3) for _ in range(K)]   # one vector per rank

# A naive ring all-reduce written ONLY with point-to-point hops. Phase A walks a
# running sum around the ring; phase B walks the finished sum back so every rank
# holds it. msgs counts each point-to-point send (no collective is used).
buf = [v.copy() for v in local]
msgs = 0
acc = buf[0].copy()                       # rank 0 starts the running sum
for step in range(1, K):                  # K-1 send/recv hops: gather the sum
    acc = acc + buf[step]                 # recv from left, add local share
    msgs += 1
total = acc                               # rank K-1 now holds the full sum
result = [None] * K
result[K - 1] = total.copy()
for step in range(K - 1):                 # K-1 hops: broadcast the sum back
    result[step] = total.copy()
    msgs += 1

reference = np.sum(local, axis=0)         # what a single machine would compute
print("per-rank input vectors:")
for r in range(K):
    print(f"  rank {r}: {local[r]}")
print()
print("point-to-point messages sent :", msgs, f" (= 2*(K-1) for K={K})")
print("reference single-machine sum :", reference)
print("all ranks hold the sum?      :",
      all(np.array_equal(result[r], reference) for r in range(K)))
print("max abs error vs reference   :",
      max(int(np.max(np.abs(result[r] - reference))) for r in range(K)))

per-rank input vectors:
  rank 0: [8 6 5]
  rank 1: [3 3 1]
  rank 2: [1 1 2]
  rank 3: [8 6 9]

point-to-point messages sent : 6  (= 2*(K-1) for K=4)
reference single-machine sum : [20 16 17]
all ranks hold the sum?      : True
max abs error vs reference   : 0

Six messages, no shared memory, every rank blind to the others' data except through one neighbor, and the result is the exact sum on every rank. This is the entire idea of a collective in miniature. The cost of this hand-built version is $2(K-1)$ messages, which is precisely the message count the alpha-beta model in Section 3.8 assigned to ring all-reduce; the real ring algorithm improves on the naive version above only by slicing each vector so the per-message payload is $n/K$ rather than $n$, which is the refinement Section 4.4 develops. What does not change, ever, is that the bytes cross a physical wire, and which wire decides how long each of those six hops takes.

5. The Interconnect Hierarchy, Thinly Intermediate

The send in Code 4.2.2 was free because it happened in one process; on real hardware each hop crosses a link with a finite $\alpha$ and $\beta$, and those links form a hierarchy. The crucial fact, the only one this chapter needs you to internalize about hardware, is that the tiers differ by orders of magnitude. Inside a single machine, accelerators talk over NVLink, a dedicated high-bandwidth fabric reaching hundreds of gigabytes per second between GPUs, or over PCIe, the general-purpose bus shared with the rest of the system at tens of gigabytes per second. Between machines, bytes leave the box and cross a network: commodity Ethernet, or InfiniBand, a low-latency network built for high-performance computing and now standard in AI clusters. The span from NVLink down to Ethernet is roughly two orders of magnitude in bandwidth and a similar gap in latency, and Figure 4.2.1 draws the tiers to scale.

One more idea is needed to make inter-node transfers fast, and it is RDMA, remote direct memory access. Ordinarily, moving a buffer from one machine to another means copying it through the operating-system kernel and the network stack at both ends, burning CPU cycles and adding latency on every hop. RDMA lets a network adapter read from one machine's memory and write directly into another machine's memory without involving either CPU or kernel on the data path, a zero-copy transport. This is what lets InfiniBand (and RDMA-over-Ethernet variants) hit the low latencies in Figure 4.2.1, and it is why an all-reduce across nodes can approach the alpha-beta cost rather than drowning in copy overhead. For our purposes RDMA is a single fact: it is the transport that makes the inter-node $\alpha$ and $\beta$ small enough that the cost model of Section 3.8 is the right tool, and the named libraries that drive it appear in Section 4.8.

6. Costing a Transfer Across the Tiers Intermediate

The reason to keep the hardware tour thin is that two numbers per tier, an $\alpha$ and a $\beta$, are all the cost model needs. The program below takes a single 256 MiB gradient buffer, the kind of payload a data-parallel step exchanges, and applies the alpha-beta time $T(n) = \alpha + n\beta$ from Section 3.8 to each tier in Figure 4.2.1. It is the same model, fed the four tiers' constants, and it makes the orders-of-magnitude claim concrete: the identical transfer that costs under a millisecond on NVLink costs nearly a tenth of a second on commodity Ethernet.

n = 256 * 1024**2     # 256 MiB payload, in bytes
tiers = [
    # name,                  alpha (s),  bandwidth (bytes/s)
    ("NVLink (intra-node)",   1.0e-6,  300e9),
    ("PCIe   (intra-node)",   2.0e-6,   25e9),
    ("InfiniBand NDR (inter)",3.0e-6,   50e9),
    ("Ethernet 25G (inter)",  30.0e-6,  3.1e9),
]
print(f"alpha-beta transfer time of a {n//1024//1024} MiB buffer across tiers:")
print(f"{'tier':<24}{'alpha(us)':>10}{'BW(GB/s)':>10}{'time(ms)':>10}{'vs NVLink':>11}")
base = None
for name, alpha, bw in tiers:
    t = alpha + n * (1.0 / bw)            # T(n) = alpha + n*beta, beta = 1/bandwidth
    base = base or t                      # NVLink is the reference (fastest)
    print(f"{name:<24}{alpha*1e6:>10.1f}{bw/1e9:>10.1f}{t*1e3:>10.2f}{t/base:>10.1f}x")

alpha-beta transfer time of a 256 MiB buffer across tiers:
tier                     alpha(us)  BW(GB/s)  time(ms)  vs NVLink
NVLink (intra-node)            1.0     300.0      0.90       1.0x
PCIe   (intra-node)            2.0      25.0     10.74      12.0x
InfiniBand NDR (inter)         3.0      50.0      5.37       6.0x
Ethernet 25G (inter)          30.0       3.1     86.62      96.7x

The 97-fold spread is the whole argument for caring about topology. A collective that keeps its traffic on NVLink inside a machine pays the top row's cost; the same collective forced across Ethernet pays the bottom row's, and at large message sizes the difference is set by bandwidth, exactly the bandwidth-bound regime that Section 3.8 named. This is why a real cluster does not treat all ranks as interchangeable: it places the ranks that must exchange the most bytes on the fastest links and lets the rare, smaller transfers cross the network. Turning that instinct into a placement strategy is the job of Section 4.9, and every collective in the sections between here and there is priced with the per-tier constants you just used.

7. Why the Substrate Decides Everything Above It Advanced

You now hold the substrate beneath the entire chapter: a single point-to-point atom, a process group that names the ranks a collective acts over, and an interconnect hierarchy whose tiers differ by orders of magnitude and whose inter-node transfers are made fast by RDMA. Every named collective in the sections ahead is a schedule of the atom over a group, and its cost is that schedule's message count and bytes evaluated against the tier its ranks happen to sit on. The all-reduce of Section 4.3, the all-gather and reduce-scatter of Section 4.5, and the all-to-all of Section 4.6 are three such schedules; what makes one fast or slow in a given system is not the schedule alone but the schedule meeting the substrate.

With the atom, the group, and the tiers in hand, we are ready to name the first and most important collective in distributed training. The next section, Section 4.3, takes the all-reduce you built by hand in Code 4.2.2 and develops it as the operation that synchronizes gradients in data-parallel SGD, complete with the cost model that decides how it scales.