Distributed Launch Mechanics

Before any distributed training can happen, you need to launch processes, assign ranks, and establish communication. This chapter covers the mechanics — from single-node to multi-node elastic launch with fault tolerance.

---

1. The One-GPU-One-Process Model

Distributed training uses one process per GPU. Each process:

• Owns one GPU (torch.cuda.set_device(local_rank))
• Has a unique identity (rank) in the global communicator
• Knows the total number of processes (world_size)

Variable	Meaning	Example (2 nodes × 4 GPUs)
RANK	Global process ID (0 to world_size-1)	0, 1, 2, 3, 4, 5, 6, 7
LOCAL_RANK	GPU ID within this node (0 to nproc-1)	0, 1, 2, 3 on each node
WORLD_SIZE	Total number of processes	8
MASTER_ADDR	IP of rank 0’s node	10.0.0.1
MASTER_PORT	Port for initial rendezvous	29500

Process Initialization

import os
import torch
import torch.distributed as dist

def setup():
    rank = int(os.environ["RANK"])
    local_rank = int(os.environ["LOCAL_RANK"])
    world_size = int(os.environ["WORLD_SIZE"])
    
    # Bind this process to its GPU
    torch.cuda.set_device(local_rank)
    
    # Initialize the default process group
    dist.init_process_group(
        backend="nccl",      # GPU communication via NCCL
        rank=rank,
        world_size=world_size,
    )
    return rank, local_rank, world_size

init_process_group blocks until all world_size processes reach this line. If any process fails to start, all others hang indefinitely. This is the most common cause of “training stuck at startup.”

---

2. torchrun (Single Node)

# Launch 8 processes on 8 GPUs, single node
torchrun --nproc_per_node=8 train.py --batch-size 32

# torchrun automatically sets:
# RANK=0..7, LOCAL_RANK=0..7, WORLD_SIZE=8, MASTER_ADDR=localhost, MASTER_PORT=29500

What torchrun Does

1. Detects available GPUs
2. Forks N child processes (one per GPU)
3. Sets environment variables for each process
4. Monitors processes for failure
5. Restarts failed processes (elastic mode)

---

3. torchrun (Multi-Node)

# Node 0 (master):
torchrun \
  --nproc_per_node=8 \
  --nnodes=4 \
  --node_rank=0 \
  --master_addr=10.0.0.1 \
  --master_port=29500 \
  train.py

# Node 1:
torchrun \
  --nproc_per_node=8 \
  --nnodes=4 \
  --node_rank=1 \
  --master_addr=10.0.0.1 \
  --master_port=29500 \
  train.py

# Nodes 2, 3: same pattern with node_rank=2, 3

All nodes must run the same command (except node_rank). They rendezvous at master_addr:master_port to discover each other.

With SLURM

#!/bin/bash
#SBATCH --nodes=4
#SBATCH --ntasks-per-node=8
#SBATCH --gpus-per-node=8

# SLURM sets SLURM_PROCID, SLURM_LOCALID, SLURM_NTASKS
# Use srun to launch on all nodes simultaneously
srun torchrun \
  --nproc_per_node=8 \
  --nnodes=$SLURM_NNODES \
  --node_rank=$SLURM_NODEID \
  --master_addr=$(scontrol show hostname $SLURM_NODELIST | head -n1) \
  --master_port=29500 \
  train.py

---

4. Elastic Launch

Elastic training allows the job to continue when nodes fail or are added:

torchrun \
  --nproc_per_node=8 \
  --nnodes=2:8 \         # min 2 nodes, max 8 nodes
  --rdzv_backend=c10d \  # rendezvous backend
  --rdzv_endpoint=10.0.0.1:29500 \
  train.py

How Elastic Works

1. Rendezvous: Processes join a rendezvous (like a waiting room)
2. Barrier: When enough processes join (≥ min_nodes × nproc), training starts
3. Monitoring: torchrun watches for process failures
4. Recovery: On failure, surviving processes re-rendezvous with fewer nodes
5. Scale-up: New nodes can join at the next re-rendezvous

Elastic Requirements

Your training script must handle:

• Checkpoint/resume: Save state frequently, reload after restart
• Variable world_size: DP degree changes when nodes join/leave
• Re-initialization: Recreate process groups after re-rendezvous

def train():
    setup()  # init_process_group
    
    # Load checkpoint if resuming
    start_step = load_checkpoint_if_exists()
    
    for step in range(start_step, total_steps):
        train_step()
        
        if step % save_interval == 0:
            save_checkpoint(step)  # survive node failures

---

5. Backend Selection

Backend	Transport	Best for	Notes
nccl	GPU↔GPU (NVLink, IB)	All GPU training	De facto standard
gloo	CPU, TCP	CPU training, fallback	Also used for non-tensor operations
mpi	MPI library	HPC environments	Requires MPI installation

# Common pattern: NCCL for GPU ops, Gloo for CPU coordination
dist.init_process_group(backend="nccl")
# PyTorch automatically uses Gloo for CPU-based operations

---

6. Debugging Launch Issues

Symptom	Cause	Fix
Hangs at init_process_group	One process didn’t start	Check all nodes launched, verify MASTER_ADDR reachable
“Connection refused”	Master not ready	Launch master node first, or use elastic rendezvous
Wrong GPU assignment	LOCAL_RANK mismatch	Verify CUDA_VISIBLE_DEVICES matches torchrun
NCCL timeout	Network issue	Set NCCL_DEBUG=INFO, check firewall, verify IB
Different world_size on nodes	Mismatched --nnodes	Ensure all nodes use same configuration

# Debug environment
python -c "import os; [print(f'{k}={v}') for k,v in os.environ.items() if 'RANK' in k or 'WORLD' in k or 'MASTER' in k]"

# Verify NCCL can communicate
NCCL_DEBUG=INFO torchrun --nproc_per_node=8 -c "
import torch.distributed as dist
dist.init_process_group('nccl')
print(f'Rank {dist.get_rank()} of {dist.get_world_size()} ready')
"

For NCCL debugging and distributed troubleshooting, see Debugging Distributed Training.

---

Self-Check

• Explain RANK vs LOCAL_RANK and when to use each
• Write a torchrun command for 4-node, 8-GPU-per-node training
• Explain the rendezvous mechanism for multi-node startup
• Set up elastic launch with min/max node counts
• Debug a training job that hangs at init_process_group
• Convert a SLURM job script to use torchrun

---

References

• PyTorch Distributed Overview
• torchrun Documentation
• Elastic Training (TorchElastic)