Advanced Distributed Training

import torch
import torch.distributed as dist
from torch.nn.parallel import DistributedDataParallel as DDP
from torch.utils.data.distributed import DistributedSampler

def setup_distributed(rank, world_size):
    """Initialize distributed training"""
    os.environ['MASTER_ADDR'] = 'localhost'
    os.environ['MASTER_PORT'] = '12355'
    dist.init_process_group("nccl", rank=rank, world_size=world_size)

def train_distributed(rank, world_size, model, dataset):
    setup_distributed(rank, world_size)
    
    # Move model to GPU and wrap with DDP
    model = model.to(rank)
    model = DDP(model, device_ids=[rank])
    
    # Create distributed sampler
    sampler = DistributedSampler(dataset, num_replicas=world_size, rank=rank)
    dataloader = DataLoader(dataset, batch_size=32, sampler=sampler)
    
    optimizer = torch.optim.AdamW(model.parameters(), lr=1e-4)
    
    for epoch in range(num_epochs):
        sampler.set_epoch(epoch)  # Important for shuffling
        
        for batch in dataloader:
            inputs, targets = batch
            inputs, targets = inputs.to(rank), targets.to(rank)
            
            optimizer.zero_grad()
            outputs = model(inputs)
            loss = criterion(outputs, targets)
            loss.backward()
            optimizer.step()
    
    dist.destroy_process_group()

# Launch with torch.multiprocessing
if __name__ == "__main__":
    world_size = torch.cuda.device_count()
    torch.multiprocessing.spawn(
        train_distributed,
        args=(world_size, model, dataset),
        nprocs=world_size,
        join=True
    )

import torch.nn as nn
from torch.distributed.pipeline.sync import Pipe

class ModelParallelTransformer(nn.Module):
    def __init__(self, config):
        super().__init__()
        
        # Split layers across devices
        devices = [f'cuda:{i}' for i in range(torch.cuda.device_count())]
        layers_per_device = config.num_layers // len(devices)
        
        self.layers = nn.ModuleList()
        for i, layer in enumerate(self.transformer_layers):
            device_idx = i // layers_per_device
            device = devices[min(device_idx, len(devices)-1)]
            layer.to(device)
            self.layers.append(layer)
    
    def forward(self, x):
        for i, layer in enumerate(self.layers):
            # Move input to layer's device
            x = x.to(layer.weight.device)
            x = layer(x)
        return x

# Pipeline Parallel with automatic balancing
class PipelineParallelModel(nn.Module):
    def __init__(self, layers):
        super().__init__()
        
        # Automatically partition layers across devices
        devices = [f'cuda:{i}' for i in range(torch.cuda.device_count())]
        
        # Use Pipe for automatic pipeline parallelism
        self.pipe_model = Pipe(
            nn.Sequential(*layers),
            balance=[len(layers) // len(devices)] * len(devices),
            devices=devices,
            chunks=8  # Micro-batch size
        )
    
    def forward(self, x):
        return self.pipe_model(x)

# Training with pipeline parallelism
model = PipelineParallelModel(transformer_layers)
optimizer = torch.optim.AdamW(model.parameters())

for batch in dataloader:
    inputs, targets = batch
    
    # Pipeline automatically handles micro-batching
    outputs = model(inputs)
    loss = criterion(outputs, targets)
    loss.backward()
    optimizer.step()
    optimizer.zero_grad()

import deepspeed
from deepspeed.pipe import PipelineModule

class Hybrid3DModel(PipelineModule):
    def __init__(self, layers, topology):
        # topology defines data/model/pipe parallel dimensions
        super().__init__(
            layers=layers,
            topology=topology,
            loss_fn=nn.CrossEntropyLoss()
        )

# DeepSpeed configuration for 3D parallelism
ds_config = {
    "train_batch_size": 512,
    "train_micro_batch_size_per_gpu": 4,
    "gradient_accumulation_steps": 16,
    
    # ZeRO optimizer state partitioning
    "zero_optimization": {
        "stage": 3,  # Partition optimizer states, gradients, and parameters
        "offload_optimizer": {
            "device": "cpu",  # Offload to CPU memory
            "pin_memory": True
        },
        "offload_param": {
            "device": "cpu",
            "pin_memory": True
        }
    },
    
    # Tensor parallelism
    "tensor_parallel": {
        "tp_size": 4  # 4-way tensor parallelism
    },
    
    # Pipeline parallelism  
    "pipeline": {
        "pipe_parallel_size": 8,
        "num_stages": 8
    },
    
    # Mixed precision training
    "fp16": {
        "enabled": True,
        "loss_scale": 0,
        "loss_scale_window": 1000,
        "hysteresis": 2,
        "min_loss_scale": 1
    }
}

# Initialize model with DeepSpeed
model_engine, optimizer, _, _ = deepspeed.initialize(
    model=model,
    config=ds_config
)

# Training loop with automatic mixed precision and gradient accumulation
for batch in dataloader:
    inputs, targets = batch
    
    loss = model_engine(inputs, targets)
    model_engine.backward(loss)
    model_engine.step()