Cloud Computing for AI Systems from First PrinciplesPost 11 of 19 · 19 of 19 written
- How to use this guide
- 01The core mental model
- 02The physical layer
- 03Virtualization: how one machine becomes many
- 04Compute for AI: GPUs and accelerators
- 05Back-of-envelope math for AI workloads
- 06Storage and data
- 07Networking
- 08Identity, security and compliance
- 09Orchestration: running fleets, not machines
- 10Infrastructure as code and delivery
- 11Distributed training in the cloud
- 12Inference and serving
- 13Reliability and observability
- 14Cloud economics
- 15The provider landscape
- 16Fluency toolkit: the questions to ask
- 17Hands-on learning path
- 18Glossary
- 19Resources worth your time
11. Distributed training in the cloud
Distributed training splits one training job across many GPUs. The design question is always which tensors to split and over which link they travel, because communication, not computation, sets the ceiling.
Why split at all
There are only two reasons:
- Too slow on one GPU. You want more throughput, so you process more data in parallel.
- Too big for one GPU. The model, its gradients and optimizer state (Part 5: about 16 bytes per parameter) do not fit, so you shard them.
Every strategy below answers one or both.
The collective operations
GPUs in a training job coordinate through a handful of group operations called collectives. NVIDIA’s NCCL library implements them (RCCL on AMD).
- All-reduce: every GPU contributes a tensor, and every GPU ends up with the sum. Used to average gradients.
- All-gather: every GPU holds one shard, and every GPU ends up with the whole tensor.
- Reduce-scatter: sum across GPUs, but each GPU keeps only its own shard of the result. An all-reduce is a reduce-scatter followed by an all-gather.
- All-to-all: each GPU sends a different piece to every other GPU. Used to route tokens to experts in mixture-of-experts models.
The cost of a ring all-reduce is worth memorizing. Each GPU sends and receives about 2 × (N − 1)/N × the tensor size, which is close to twice the size no matter how many GPUs there are.
Worked example: the gradients of an 8B model in BF16 are 16 GB, so each GPU moves about 32 GB per step. Over NVLink at 900 GB/s that takes about 36 ms. Between nodes at 50 GB/s it takes about 640 ms. This is why frameworks overlap communication with computation, and why topology decides speed.
Parallelism strategies
| Strategy | What is split | What travels | Link it needs | Use when |
|---|---|---|---|---|
| Data parallel (DDP) | The batch; every GPU holds the full model | Gradient all-reduce once per step | Any decent network | Model and optimizer state fit on one GPU |
| Fully sharded data parallel (FSDP, ZeRO-3) | The batch, plus weights, gradients and optimizer state | All-gather weights per layer; reduce-scatter gradients | Good network | State is too big for one GPU; the default for most fine-tuning and mid-size training |
| Tensor parallel (TP) | Individual weight matrices | Partial results inside every layer | NVLink only | Very wide layers; also used to serve big models |
| Pipeline parallel (PP) | Groups of layers, as stages on different GPUs | Activations between stages | Across nodes is fine | Very large models; idle “bubbles” reduced with micro-batches |
| Context / sequence parallel | The sequence-length dimension | Attention data between GPUs | Fast links | Very long contexts |
| Expert parallel (EP) | The experts of a mixture-of-experts model | All-to-all token routing | Fast links | MoE models |
Large runs combine them, often called 3D or 4D parallelism. The governing rule: put the chattiest parallelism on the fastest link. Tensor parallelism stays inside a server’s NVLink domain, pipeline stages span a few nodes, and data or fully sharded parallelism spans everything else.
The software you will meet
- PyTorch native: DDP, FSDP2, the
torchrunlauncher, and Distributed Checkpoint. TorchTitan is PyTorch’s reference for large-scale training. - DeepSpeed: ZeRO stages 1–3, with optional offloading of state to CPU memory or NVMe.
- Megatron-LM / Megatron Core and NeMo: NVIDIA’s stack for tensor, pipeline and expert parallelism at very large scale.
- Fine-tuning tools: Hugging Face Accelerate and TRL (SFT, DPO, GRPO), Axolotl, torchtune, and Unsloth for fast single-GPU fine-tunes.
- JAX: the main path on TPUs, with MaxText as a reference implementation.
A two-node launch looks like torchrun --nnodes=2 --nproc-per-node=8 --rdzv-endpoint=<head-node>:29500 train.py: 16 processes, one per GPU, that find each other through the rendezvous address.
Measuring efficiency
- MFU (Part 5) is the headline number: the share of peak FLOPs you actually achieve.
- Step time breakdown: how much is compute, how much is communication that is not hidden behind compute, how much is waiting for data, and how much is checkpointing.
- Tools: the PyTorch profiler, NVIDIA Nsight Systems, and DCGM metrics. Note that
nvidia-smi’s “GPU utilization” only means a kernel was running; it can read 100% while the chip does little useful math.
Common fixes, roughly in order of payoff: larger micro-batches or gradient accumulation, BF16 or FP8, FlashAttention, torch.compile, activation checkpointing to fit larger batches, overlapping communication, tuning NCCL, and fixing the data loader.
Failures at scale
With thousands of GPUs, something breaks every few hours: GPU hardware errors, memory errors, NVLink or network-card faults, host crashes, silent data corruption (wrong numbers with no error), and stragglers. Because training is synchronized, one slow GPU slows all of them.
The math is simple. If a single GPU averages one failure every 5 years (about 44,000 hours), a 1,000-GPU job sees one failure about every 44 hours, and a 16,000-GPU job one about every 3 hours. Meta reported 419 unexpected interruptions over 54 days when training Llama 3 on about 16,000 H100s, which matches this order of magnitude.
The defenses:
- Frequent asynchronous checkpoints, so a failure costs minutes of work, not hours.
- Automatic restart: the scheduler requeues the job and it resumes from the last checkpoint.
- Health checks before and during jobs (burn-in tests, NCCL bandwidth tests) to remove bad nodes early.
- Spare nodes kept ready, and elastic training that can continue on fewer nodes.
Track goodput: the share of wall-clock time that produced real training progress, after subtracting restarts, lost work and checkpoint pauses.
Spot capacity and reservations
- Spot or preemptible instances cost 50–90% less but can be reclaimed with short notice (about 2 minutes on AWS, 30 seconds on Google Cloud). They suit single-node jobs and loosely coupled work with frequent checkpoints and automatic restarts. They suit large synchronized multi-node jobs badly, because one reclaimed node stalls everyone.
- Reservations guarantee capacity for planned runs: AWS Capacity Blocks for ML, Google’s Dynamic Workload Scheduler, or term contracts with GPU clouds.
Post-training and reinforcement learning
Much practical LLM work today is post-training: supervised fine-tuning (SFT), preference optimization (DPO), and reinforcement learning with verifiable rewards (GRPO, PPO). RL is mainly a systems problem. A loop generates rollouts with a fast inference engine, scores them (often by running code in sandboxes), and updates the model on separate training GPUs. Frameworks such as veRL, OpenRLHF and TRL coordinate this, often on Ray, and the rollout inference usually dominates the cost.
Check yourself
- Why must tensor parallelism stay inside the NVLink domain?
- With DDP, how many bytes does each GPU send per step to all-reduce the gradients of a 3B model in BF16?
- A 512-GPU job’s throughput drops 30% but nothing has crashed. What is your first hypothesis, and how do you test it?
- When does spot capacity make sense for training, and when is it a false saving?