每日基础课

一句话定义： 数据并行（DDP, Distributed Data Parallel）让每张显卡都持有一份完整的模型副本，各自喂入不同的数据子集算梯度，最后通过跨卡通信把梯度求平均，让所有卡同步更新为同一份新参数。

生活类比： 想象 4 个厨师拿着完全一样的炒菜秘方（模型副本）。老板给了他们 4 箱不同产地的新鲜食材（分块数据）。4 个厨师各自开火炒菜、尝味道、记录调味建议（算损失与梯度）。炒完后，4 人围在一起把调味建议取平均值（All-Reduce 梯度同步），同时在各自的秘方本上改写相同的调味量（更新参数）。第二天，大家继续用更新后的相同秘方炒下一批菜。

核心数学公式： 全局平均梯度计算： $$g = \frac{1}{K} \sum_{k=1}^{K} \nabla_\theta \mathcal{L}(x^{(k)}, y^{(k)}; \theta)$$ 参数同步更新： $$\theta \leftarrow \theta - \eta \cdot g$$

符号解释： - $K$：参与训练的 GPU 总卡数（进程数 / World Size）。 - $\theta$：模型的全部权重参数（每张卡上数值完全相同）。 - $(x^{(k)}, y^{(k)})$：分配给第 $k$ 张显卡的专属数据批次（通过采样器切分，互不重复）。 - $\nabla_\theta \mathcal{L}(\dots)$：第 $k$ 张卡在自己的数据上算出的局部梯度。 - $g$：全卡汇总平均后的全局梯度。 - $\eta$：学习率（Learning Rate）。

它解决什么问题？没有它会怎样？ - 解决的问题：单卡算力慢、显存小，无法用大 Batch 训练，导致训练周期以月计算。 - 没有它会怎样： 1. 说话人日志（如 CAM++ / ECAPA-TDNN）和对比学习非常依赖大 Batch Size 来提供丰富的负样本，单卡 Batch 太小会导致模型表征能力严重退化。 2. 音视频多模态模型（如 AV-Hubert）单条视频+音频数据体量极大，单卡只能塞进极少样本，没有数据并行根本无法在合理时间内完成预训练。

在音视频与说话人日志中的实际位置： 1. 数据切分层：使用 `DistributedSampler` 确保每张卡分配到不同说话人的音频片段或音视频对，保证各卡数据无重复。 2. 训练包装层：将特征提取器（如 Conformer、ResNet）和分类头包进 `torch.nn.parallel.DistributedDataParallel`。 3. 反向传播层：各卡前向传播计算损失后，PyTorch 在反向传播算梯度的同时，后台自动通过 Ring-AllReduce 通信拉齐梯度，完全不需要手动写通信代码。

学习建议：学习基础概念和通信拓扑直觉推荐看《Dive into Deep Learning》对应章节（先看直觉和公式，再运行章节中的代码）；掌握真实的多卡训练工程写法推荐看《PyTorch Tutorials》（对照官方示例掌握张量与训练写法）。

思路： 1. 编写单进程训练任务函数 `demo_basic`。 2. 使用 `dist.init_process_group` 初始化进程组（后端选 `gloo` 支持 CPU/GPU）。 3. 构建小型 Linear 分类模型，并用 `nn.parallel.DistributedDataParallel` 包装。 4. 使用 `DistributedSampler` 分发数据，展示 `sampler.set_epoch(epoch)`。 5. 使用 `torch.multiprocessing.spawn` 启动 2 个独立进程模拟 2 张卡。

完整可运行代码：

import os
import torch
import torch.distributed as dist
import torch.multiprocessing as mp
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import DataLoader, Dataset, DistributedSampler


# 1. 模拟说话人嵌入特征数据集 (64维特征, 2分类)
class SyntheticSpeakerDataset(Dataset):

    def __init__(self, size=128):
        self.data = torch.randn(size, 64)
        self.labels = torch.randint(0, 2, (size,))

    def __len__(self):
        return len(self.data)

    def __getitem__(self, idx):
        return self.data[idx], self.labels[idx]


# 2. 简易说话人分类模型
class SpeakerClassifier(nn.Module):

    def __init__(self, input_dim=64, num_classes=2):
        super().__init__()
        self.fc = nn.Linear(input_dim, num_classes)

    def forward(self, x):
        return self.fc(x)


# 3. 单个进程的训练主体
def train_worker(rank: int, world_size: int):
    # 步骤 A: 设置主节点地址与初始化进程组
    os.environ["MASTER_ADDR"] = "localhost"
    os.environ["MASTER_PORT"] = "12355"
    # CPU 环境使用 gloo 后端；若有多张 GPU 则改为 nccl
    dist.init_process_group("gloo", rank=rank, world_size=world_size)

    # 步骤 B: 实例化模型并用 DDP 包装
    model = SpeakerClassifier(input_dim=64, num_classes=2)
    ddp_model = nn.parallel.DistributedDataParallel(model)

    loss_fn = nn.CrossEntropyLoss()
    optimizer = optim.SGD(ddp_model.parameters(), lr=0.05)

    # 步骤 C: 数据集与分布式采样器
    dataset = SyntheticSpeakerDataset(size=128)
    sampler = DistributedSampler(
        dataset, num_replicas=world_size, rank=rank, shuffle=True
    )
    dataloader = DataLoader(dataset, batch_size=16, sampler=sampler)

    # 步骤 D: 训练循环
    for epoch in range(3):
        # 极重要：保证各 Epoch 打乱顺序不同且各卡不重叠
        sampler.set_epoch(epoch)
        total_loss = 0.0

        for feats, labels in dataloader:
            optimizer.zero_grad()
            outputs = ddp_model(feats)
            loss = loss_fn(outputs, labels)
            loss.backward()  # DDP 在此步自动触发 All-Reduce 梯度同步
            optimizer.step()
            total_loss += loss.item()

        avg_loss = total_loss / len(dataloader)
        if rank == 0:  # 仅在 Rank 0 打印日志，避免重复刷屏
            print(f"[Epoch {epoch + 1}] Rank {rank} - Avg Loss: {avg_loss:.4f}")

    # 步骤 E: 验证两卡参数是否严格一致
    weight_data = ddp_model.module.fc.weight.data.clone()
    dist.all_reduce(weight_data, op=dist.ReduceOp.SUM)
    # 若严格一致，Sum 后的平均值应等于单卡值
    if rank == 0:
        synced_weight = weight_data / world_size
        diff = (
            torch.sum(torch.abs(synced_weight - ddp_model.module.fc.weight.data))
            .item()
        )
        print(f"\n[验证] 卡间权重差异绝对值求和: {diff:.6f} (0即为完全同步)")

    # 步骤 F: 销毁进程组
    dist.destroy_process_group()


if __name__ == "__main__":
    WORLD_SIZE = 2
    print(f"启动 {WORLD_SIZE} 个进程模拟 DDP 训练...")
    mp.spawn(train_worker, args=(WORLD_SIZE,), nprocs=WORLD_SIZE, join=True)
    print("DDP 训练与同步测试完成！")