Warmup Guide — Distributed Systems & HPC Collectives

Zero-to-expert primer for Phase 08. Builds on Phase 01 Ch. 9 (interconnect) and Phase 06 (placement). No distributed-training background assumed. By the end you can derive collective costs, reason about parallelism strategies, and design for failure at scale.

Table of Contents

• Chapter 1: Why Distribute, and the Parallelism Taxonomy
• Chapter 2: Collective Operations — the Vocabulary
• Chapter 3: Ring All-Reduce — the Bandwidth-Optimal Algorithm
• Chapter 4: Tree, Hierarchical, and When Each Wins
• Chapter 5: NCCL — the Production Collective Library
• Chapter 6: Topology — NVLink Islands, InfiniBand, and Placement
• Chapter 7: Overlap — Hiding Communication Behind Compute
• Chapter 8: Failure Domains at Scale
• Chapter 9: Distribution on the Serving Side
• Lab Walkthrough Guidance
• Success Criteria
• Interview Q&A
• References

Chapter 1: Why Distribute, and the Parallelism Taxonomy

Zero background: one GPU has finite HBM (80 GB) and finite compute. A 175B-parameter model in FP16 is 350 GB of weights alone — it doesn't fit. Even when a model fits, one GPU's throughput caps training time at "months." So we split the work across many GPUs and make them cooperate. The cooperation is the hard part, and it's all communication.

The four ways to split (you must be able to contrast these cold):

• Data parallel (DP): every GPU holds a full copy of the model, processes a different slice of the batch, and they all-reduce gradients each step to stay in sync. Simple, the workhorse. Communication: one all-reduce of the full gradient per step. Scales batch size, not model size.
• Tensor parallel (TP): split each layer's matrices across GPUs (e.g., each GPU computes part of every attention head), with an all-reduce inside every layer to combine partial results. Lets a model bigger than one GPU run — but the per-layer all-reduce is on the critical path, so TP must live on NVLink (Ch. 6); across slow links it's catastrophic.
• Pipeline parallel (PP): split the model's layers into stages on different GPUs; micro-batches flow through like an assembly line. Communication is just activations between adjacent stages (cheap), but you pay a pipeline bubble (stages idle at fill/drain). Lets very deep models run.
• Expert / sequence / context parallel: (overview) shard along other axes — MoE experts, the sequence dimension for long context. They compose: frontier training is "3D/4D parallelism" — e.g., TP within a node, PP across a few nodes, DP across the rest.

The mental model: DP trades communication for batch scale; TP trades communication for model scale (on fast links only); PP trades latency (bubble) for model depth. Choosing and composing them is a topology problem (Ch. 6).

Chapter 2: Collective Operations — the Vocabulary

A collective is a communication pattern over a group of ranks. The ones that matter, with their communication volume (V = data size per rank, N = ranks):

The headline fact: all-reduce = reduce-scatter + all-gather, and each of those moves (N-1)/N · V, so all-reduce moves ≈ 2V per rank, independent of N — the result you'll prove in Lab 01. This is why data-parallel training scales: adding GPUs doesn't increase per-GPU communication volume (only latency terms grow). All-reduce of gradients is the single most important collective in ML.

Chapter 3: Ring All-Reduce — the Bandwidth-Optimal Algorithm

The algorithm NCCL uses for large messages, and the centerpiece of Lab 01.

Setup: N ranks in a logical ring (each talks to its left and right neighbor). The gradient vector is chopped into N chunks. Two phases:

Phase 1 — reduce-scatter (N-1 steps): in step k, each rank sends one chunk to its right neighbor and receives a chunk from its left, adding the received chunk to its local copy. After N-1 steps, each rank holds the fully reduced sum of one chunk (a different chunk per rank).

Phase 2 — all-gather (N-1 steps): same ring motion, but now ranks forward the completed chunks (no addition) until everyone has all N reduced chunks.

The cost (the result to know cold): each rank sends/receives (N-1) chunks of size V/N in each phase = 2 phases × (N-1) × V/N = 2·(N-1)/N · V ≈ 2V per rank. As N→∞ this approaches exactly 2V, independent of N. Compare the naive "all-gather everything then sum locally": N·V per rank — N times worse. That N-independence is why ring all-reduce made large-scale data-parallel training practical (Baidu's 2017 result that propagated into every framework).

Why bandwidth-optimal: information-theoretically, every rank's contribution must reach every other rank, and each rank must receive the full reduced result — 2V is the lower bound on bytes that must cross each rank's links. Ring hits it. (The cost is N-1 latency hops per phase, so ring is bandwidth-optimal but latency-linear — hence trees for small messages, Ch. 4.)

Chapter 4: Tree, Hierarchical, and When Each Wins

Ring is optimal in bandwidth but its latency grows linearly in N (2(N-1) sequential hops). For small messages (latency-dominated) or many ranks, other structures win:

• Tree all-reduce: reduce up a binary tree to the root, broadcast down. Latency O(log N) instead of O(N) — far better for small messages. Bandwidth is worse (root links are hot), but small messages don't saturate bandwidth anyway. NCCL auto-switches ring↔tree by message size.
• Hierarchical / two-level: reduce within each NVLink island first (fast), then all-reduce across islands over IB (one representative per node), then broadcast back down. This matches the bandwidth hierarchy (Ch. 6) — you do the big-volume reduction on the fast intra-node links and only the small cross-node step on the slow links. Lab 02 quantifies the win.
• Double binary tree (NCCL's default for many cases): two trees using complementary links, recovering bandwidth while keeping log-latency.

The takeaway for a leader: there's no single best collective algorithm — NCCL picks by message size, rank count, and topology, and your job is to give it a good topology (Ch. 6) and large enough messages (bucketing, Ch. 7), not to hand-pick algorithms.

Chapter 5: NCCL — the Production Collective Library

NVIDIA Collective Communications Library — the layer every framework (PyTorch DDP/FSDP, Megatron, DeepSpeed) calls for GPU collectives.

What it does that you won't reimplement: detects the topology (NVLink, NVSwitch, PCIe, IB rails) at init, builds rings/trees over it, picks algorithms by message size, uses GPUDirect RDMA (GPU↔NIC↔GPU without staging through host memory — the Phase 01 PCIe-cliff avoidance), overlaps with compute via CUDA streams, and handles multi-rail IB.

The operational knobs/signals you will use:

• NCCL_DEBUG=INFO — prints the topology and algorithm choices (first stop when collectives are slow).
• NCCL_ALGO, NCCL_PROTO — force ring/tree, LL/Simple protocols (debugging).
• NCCL_IB_*, NCCL_SOCKET_IFNAME, NCCL_P2P_* — interface/transport selection (the cause of half of "training is slow on new hardware" incidents — wrong NIC, P2P disabled, IB not used).
• NCCL_TIMEOUT / the framework's collective timeout — what turns a hang (Ch. 8) into a crash with a stack trace.

The leadership point: NCCL is a black box that's fast when the topology and config are right and silently 10× slow when they aren't. Knowing what it's doing (Ch. 3–4, 6) is what lets you debug "MFU is 50%, why?" instead of filing a ticket and waiting.

Chapter 6: Topology — NVLink Islands, InfiniBand, and Placement

This is where Phase 06 (placement) and this phase (collectives) fuse. The bandwidth hierarchy (Phase 01 Ch. 9), as the substrate for collectives:

within a GPU:    HBM           ~3.35 TB/s
GPU<->GPU (node): NVLink/NVSwitch ~900 GB/s   <- TP lives here
node<->node:     InfiniBand/RoCE ~50 GB/s    <- DP/PP cross this
host:            PCIe           ~64 GB/s

The consequences that decide real throughput:

• Tensor parallelism must stay within an NVLink island. TP does a per-layer all-reduce on the critical path; on NVLink that's ~microseconds, on IB it's ~20× slower and serializes every layer — a 2–5× total slowdown. This is why schedulers (Phase 06) gang-place TP groups on one NVLink-connected node, and why "my 8-GPU TP job is slow" is almost always "it got spread across two nodes."
• Data parallelism can cross nodes — its all-reduce is once per step and overlappable (Ch. 7), so IB bandwidth is tolerable; you still want hierarchical all-reduce (Ch. 4) to minimize cross-node volume.
• Rails and GPUDirect RDMA: multi-NIC nodes have "rails"; a topology-aware collective uses the NIC closest to each GPU (GPUDirect RDMA) to avoid PCIe hops. Misconfiguration here silently routes through host memory (the 50× cliff).

Composing parallelism to the topology is the frontier-training art: TP within node (NVLink), PP across a few nodes (cheap activation passing), DP across the rest (overlappable all-reduce). Lab 02 makes the placement-vs-throughput relationship quantitative.

Chapter 7: Overlap — Hiding Communication Behind Compute

Even optimal collectives cost time; the trick is to not wait for them.

• Gradient bucketing + overlap (DDP's core optimization): instead of one giant all-reduce after the whole backward pass, group gradients into buckets and all-reduce each bucket as soon as it's ready — overlapping communication of early-computed gradients with the still-running backward pass. Done right, communication is almost free (hidden behind compute). Bucket size is a tuning knob (too small = latency overhead per bucket; too large = less overlap).
• Computation/communication overlap generally: collectives run on a separate CUDA stream (Phase 05) so the compute stream proceeds; event edges enforce the dependency. The same stream-scheduling skills from Phase 05, at cluster scale.
• The critical-path mindset: at 60% MFU (model FLOPs utilization), the missing 40% is usually exposed communication (no overlap), pipeline bubbles, or stragglers (Ch. 8) — not raw kernel speed. Profiling distributed training is about finding what's not overlapped.

Chapter 8: Failure Domains at Scale

The chapter that separates "ran a 4-GPU job" from "operated a 10,000-GPU cluster." Synchronous collectives have a brutal property: every rank must participate, so any one failure hangs everyone.

• The hang: a GPU throws an Xid error (Phase 04/10), a NIC flaps, a process OOMs — that rank stops calling the collective, and all other ranks block forever inside NCCL waiting for it. No error, no crash — a hang. At 10k-GPU scale where ~1% of nodes have an issue daily, this happens constantly.
• Detection: collective timeouts (NCCL_TIMEOUT, framework watchdogs) turn hangs into crashes with a culprit rank; health monitoring (Phase 10) catches the failing GPU/NIC; "straggler" detection finds the slow-but-not-dead rank dragging the whole job to its speed.
• Recovery: checkpointing (the only real defense — periodic, async, sharded checkpoints so a failure costs minutes not days), elastic training (frameworks that can drop/re-add ranks and continue), and redundancy/hot spares (replace the failed node, restore from checkpoint, resume). The checkpoint interval is an explicit cost/risk tradeoff you own: more frequent = less lost work per failure, more overhead.
• Blast radius: placement (Ch. 6) determines how much one failure costs — a failure in a TP group stalls that whole group; good placement minimizes the blast radius. Lab 02's extension models this.

The leadership reframe: at scale, failures are not exceptional, they're the steady state. You design the training/serving platform assuming continuous partial failure — checkpoint cadence, fast detection, elastic recovery, and spare capacity are first-class features, not afterthoughts. This is also the sovereign/on-prem story (Phase 07 Ch. 9): air-gapped clusters can't lean on cloud auto-healing.

Chapter 9: Distribution on the Serving Side

Collectives aren't just for training. Large-model serving (Phase 07) is distributed too:

• Tensor-parallel inference: a model too big for one GPU is TP-sharded across an NVLink island; every forward pass does per-layer all-reduces — so serving a 70B+ model has the same NVLink-island requirement as TP training.
• KV-cache sharding: the KV cache (Phase 07) is split across the TP group; attention does its own communication.
• Prefill/decode disaggregation (Phase 07 Ch. 5): separate GPU pools for prefill and decode means shipping the KV cache between them over the interconnect — a new, large communication pattern that topology (Ch. 6) must accommodate.
• Expert parallelism for MoE serving: all-to-all to route tokens to experts — the all-to-all collective at inference time, latency-sensitive.

So the topology and collective skills here apply directly to the JD's serving mandate: a platform serving large models is a distributed system, and its throughput is gated by the same NVLink/IB hierarchy and the same overlap discipline.

Lab Walkthrough Guidance

Order: Lab 01 → Lab 02 (implement the algorithm, then reason about placing it on a topology).

Lab 01 (ring all-reduce over sockets):

1. python solution.py --world 4; confirm correctness, then read the reduce-scatter/all-gather phases against Ch. 3.
2. Verify the byte count: each rank moves ≈2(N-1)/N · V — measure it, don't trust the formula. Compare to the naive all-gather-sum (N·V).
3. Extensions: tree all-reduce (measure latency on small vs large messages — Ch. 4's crossover); bucketing/overlap; then the failure demo — kill a rank, watch the hang, add a timeout (Ch. 8).

Lab 02 (topology simulator):

1. python solution.py; read the bandwidth-hierarchy cost model against Ch. 6.
2. Reproduce: TP within an NVLink island vs TP across nodes — the ratio is the NVLink/IB ratio, quantified. This is the Phase-06-meets-Phase-08 result.
3. Hierarchical all-reduce vs flat across nodes — Ch. 4's win, measured.
4. Extension: pipeline bubble model and failure-domain blast radius.

Success Criteria

• You can contrast DP/TP/PP and state which is NVLink-bound and why (Ch. 1)
• You can derive all-reduce = reduce-scatter + all-gather = 2V/rank and explain N-independence (Ch. 2–3) — and your Lab 01 byte count proves it
• You can say when tree beats ring and what hierarchical all-reduce buys (Ch. 4)
• You can name the NCCL signals/knobs for debugging a slow collective (Ch. 5)
• You can explain why TP across nodes is catastrophic, with the bandwidth ratio (Ch. 6) — Lab 02 quantifies it
• You can explain gradient overlap and diagnose low MFU (Ch. 7)
• You can describe how one failure hangs a job and the detection/recovery stack (Ch. 8)

Interview Q&A

Q1: Derive ring all-reduce's communication cost and explain why it's bandwidth-optimal. A: All-reduce = reduce-scatter + all-gather. Chop the V-byte vector into N chunks. Reduce-scatter: N-1 steps, each rank sends/receives one V/N chunk per step and accumulates → (N-1)·V/N moved, ending with each rank owning one fully- reduced chunk. All-gather: another N-1 steps of V/N chunks → another (N-1)·V/N. Total ≈ 2·(N-1)/N · V ≈ 2V per rank, independent of N. It's bandwidth-optimal because every rank must export its contribution and import the full reduced result — 2V is the information-theoretic minimum bytes across each rank's links, and the ring saturates it. The catch: 2(N-1) sequential hops means latency grows with N, so for small messages a log-latency tree wins — which is why NCCL switches by message size.

Q2: Compare data, tensor, and pipeline parallelism — which is NVLink-bound? A: Data parallel replicates the model, splits the batch, and all-reduces gradients once per step — communication is overlappable with the backward pass and tolerates cross-node IB, so DP scales across nodes. Tensor parallel splits each layer's matrices across GPUs with an all-reduce inside every layer on the critical path — that frequency makes it NVLink-bound: on NVLink it's microseconds, across nodes on IB it's ~20× slower per layer and serializes the whole forward/backward, a 2–5× slowdown. Pipeline parallel splits layers into stages passing only activations between neighbors (cheap), trading a pipeline bubble (idle fill/drain) for depth. Frontier training composes them: TP within a node, PP across a few nodes, DP across the rest.

Q3: Your 512-GPU training run is at 60% MFU. Where do you look? A: The missing 40% is almost never raw kernel speed; it's exposed time. In order: (1) communication not overlapped — is gradient bucketing/overlap on? Profile for all-reduce on the critical path; (2) topology mistakes — is TP accidentally spanning nodes (NCCL_DEBUG=INFO shows the rings; check it's using NVLink/IB not PCIe/sockets), is GPUDirect RDMA active; (3) pipeline bubbles — micro-batch count too low for the PP depth; (4) stragglers — one slow GPU (thermal throttle, ECC retries) dragging every collective to its speed, found by per-rank step-time variance; (5) data loading / checkpoint stalls on the critical path. The discipline is "find what's not overlapped," via the distributed profiler and NCCL logs, not micro-optimizing kernels.

Q4: One GPU fails in a 1024-GPU synchronous job. What happens and how do you make it survivable? A: The failed rank stops entering the collective, so all 1023 others block forever inside NCCL — a silent hang, not a crash. Survivability is a stack: (1) detection — collective timeouts and watchdogs convert the hang into a crash naming the culprit rank; health monitoring (Xid/NIC) flags the failing hardware; straggler detection catches slow-not-dead ranks. (2) recovery — checkpointing is the real defense: frequent, async, sharded checkpoints so a failure costs minutes; elastic frameworks drop/re-add ranks; hot spares replace the node and resume from checkpoint. (3) blast-radius minimization — placement so one failure stalls the smallest group. At 10k-GPU scale ~1% of nodes fail daily, so this isn't exceptional handling — continuous partial failure is the design assumption, and checkpoint cadence is an explicit cost/risk lever I'd own.

Q5: Why does tensor-parallel placement across nodes destroy throughput, with numbers? A: TP all-reduces partial results inside every layer on the critical path. On NVLink (~900 GB/s, ~µs latency) that per-layer collective is negligible; across nodes on InfiniBand (~50 GB/s, ~µs but ~18× less bandwidth and extra hops) each all-reduce is ~20× slower and, because it's serial within each layer's forward and backward, it can't be hidden — so a model with dozens of layers pays that penalty dozens of times per step, often a 2–5× total slowdown. That's why the scheduler (Phase 06) must gang-place a TP group on one NVLink-connected node, and why "8-GPU TP job is slow" almost always means it got spread across two nodes — the single most common distributed-training misconfiguration.

Q6 (leadership): How do you architect a training platform to be efficient AND fault-tolerant at 10k-GPU scale? A: Two intertwined goals. Efficiency: compose parallelism to the topology (TP within NVLink islands, PP across a few nodes, DP across the rest), enforce it through topology-aware gang scheduling (Phase 06), maximize overlap (gradient bucketing, comm on separate streams), and monitor MFU with per-rank step-time variance to catch stragglers. Fault tolerance: assume continuous partial failure — async sharded checkpointing at a cadence tuned to the failure rate (minutes of lost work, not days), collective timeouts + health monitoring for fast detection with a named culprit, elastic training to continue through node loss, and hot-spare capacity for fast replacement. Both rest on observability (Phase 10): without per-rank metrics and NCCL visibility you can neither find the efficiency leaks nor detect the failures. And for sovereign/ on-prem (Phase 07/11), all of this must work without cloud auto-healing — the platform owns the recovery, not the cloud provider.

References

• Baidu, "Bringing HPC Techniques to Deep Learning" (ring all-reduce for DL, 2017) — https://andrew.gibiansky.com/blog/machine-learning/baidu-allreduce/
• Thakur, Rabenseifner, Gropp, "Optimization of Collective Communication Operations in MPICH" (2005) — the algorithm survey
• NVIDIA NCCL documentation + NCCL_DEBUG — https://docs.nvidia.com/deeplearning/nccl/
• Shoeybi et al., "Megatron-LM" (tensor parallelism) — https://arxiv.org/abs/1909.08053
• Narayanan et al., "Efficient Large-Scale Language Model Training on GPU Clusters" (3D parallelism / PTD-P) — https://arxiv.org/abs/2104.04473
• Rajbhandari et al., "ZeRO" (DeepSpeed memory-efficient DP) — https://arxiv.org/abs/1910.02054
• PyTorch DDP design note (gradient bucketing/overlap) — https://pytorch.org/docs/stable/notes/ddp.html
• Meta, "OPT-175B logbook" / "The Llama 3 Herd of Models" — real failure-rate accounts at scale
• Cross-track: Phase 01 WARMUP Ch. 9 (interconnect), Phase 06 WARMUP Ch. 7 (topology-aware placement)