深度学习中多机训练概念下的DP与DDP

在进行单机多卡/多机多卡训练时，通常会遇到DP与DDP的概念，为此基于kimi大模型对二者的差异进行梳理。使用DP/DPP的核心是数据并行，也就是根据显卡数量对数据集进行分治，每一个显卡都有一个独立完整的模型和一个局部数据。在多个显卡间进行梯度同步，实现对多卡的训练。DP是对于多卡训练的朴素实现，DDP是对DP的高效升级（通信效率）。调用实现DPP训练的代码，也就是设置全局采样器使数据对多卡环境下实现分治成多个小局部，然后对模型进行warp操作，在模型backward前对多机的梯度进行累积。

DDP模式下对batchnorm的训练有影响，因为DDP模式下每个卡拿到的都是局部数据，故需要将batchnorm替换为syncbatchnorm，在bn层的forward前需要将全局多机所有显卡计算出的均值方差数据进行同步。

这里需要注意的是，DP/DDP可以基于数据划分的模式提升batchsize，达到单机单卡下梯度累积训练的效果，但解决不了显存不够用的情况（单机下batch为1无法训练的模型，DDP模式下也无法训练）。

以下章节信息均来自kimi对互联网知识的搜集与提炼。

1、基本差异

分布式训练中，DP（Data Parallelism）和DDP（Distributed Data Parallelism）是两种常见的数据并行策略，它们的主要区别如下：

实现方式

• DP：基于单进程多线程实现，适用于单机多卡场景。
• DDP：基于多进程实现，每个GPU对应一个独立的进程，适用于单机多卡和多机多卡场景。

参数更新方式

• DP：梯度汇总到主卡（GPU0），然后在主卡上进行参数更新，再将更新后的参数广播到其他GPU。
• DDP：各进程在梯度计算完成后，通过all-reduce操作汇总梯度，然后每个进程独立更新参数。

通信效率

• DP：通信集中在主GPU，随着GPU数量的增加，通信成本呈线性上升，可能导致通信瓶颈。
• DDP：采用高效的通信模式（如ring-all-reduce，只实现相邻卡通信），减少了通信瓶颈，提高了通信效率。

适用场景

• DP：主要适用于单机多卡场景。
• DDP：适用于单机多卡和多机多卡场景，具有更好的扩展性。

负载均衡

• DP：主GPU的通信和计算压力较大，存在负载不均衡的问题。
• DDP：各GPU独立工作，负载均衡，避免了单点瓶颈。

性能表现

• DP：受GIL影响，性能受限，且随着GPU数量增加，性能提升有限。
• DDP：性能更优，可扩展性更好，适用于大规模分布式训练。

2、DDP训练的基本使用案例

在多机训练中利用DDP（Distributed Data Parallel）可以有效地提高训练效率和缩短训练时间。以下是使用DDP进行多机训练的步骤和要点：

1. 环境准备

• 确保每台机器上都安装了PyTorch，同时具备相同的文件系统，确保训练代码、模型、数据的路径都是一致的（如果不是同一个文件系统，数据路径不一致，请酌情修改 datasets对应的路径）。
• 确保机器之间可以通过网络相互通信，通常需要配置好SSH免密登录。

2. 代码实现

• 初始化进程组：使用torch.distributed.init_process_group来初始化进程组，指定通信后端（如NCCL）和初始化方法（如env://）。
• 设置分布式采样器：使用torch.utils.data.distributed.DistributedSampler来确保每个进程处理不同的数据子集。
• 封装模型：使用torch.nn.parallel.DistributedDataParallel来封装模型，指定设备ID。
• 训练循环：在训练循环中，每个进程独立进行前向传播、反向传播和参数更新，通过All-Reduce操作同步梯度。

3. 启动训练

• 使用torchrun命令来启动多机多卡训练，指定每个节点的进程数、节点数、节点排名和 rendezvous 端点。
  torchrun 支持与 torch.distributed.launch 相同的参数，除了 已弃用的 --use-env

示例代码

import torch
import torch.distributed as dist
import torch.nn as nn
import torch.optim as optim
from torch.nn.parallel import DistributedDataParallel as DDP

class SimpleNN(nn.Module):
    def __init__(self):
        super(SimpleNN, self).__init__()
        self.fc = nn.Linear(28 * 28, 10)

    def forward(self, x):
        x = x.view(-1, 28 * 28)
        x = self.fc(x)
        return x

def main():
    # 初始化进程组
    dist.init_process_group("nccl", init_method='env://')
    rank = dist.get_rank()
    world_size = dist.get_world_size()
    
    # 设置设备
    device = torch.device("cuda", rank)
    torch.cuda.set_device(device)
    
    # 加载数据集和分布式采样器
    transform = transforms.ToTensor()
    train_dataset = datasets.MNIST(root='./', train=True, download=True, transform=transform)
    train_sampler = torch.utils.data.distributed.DistributedSampler(train_dataset)
    train_loader = DataLoader(train_dataset, batch_size=32, sampler=train_sampler)
    
    # 构建模型并封装为DDP
    model = SimpleNN().to(device)
    ddp_model = DDP(model, device_ids=[rank])
    
    # 定义损失函数和优化器
    criterion = nn.CrossEntropyLoss()
    optimizer = optim.SGD(ddp_model.parameters(), lr=0.01 * world_size)
    
    # 训练循环
    for epoch in range(5):
        train_sampler.set_epoch(epoch)
        model.train()
        for batch_idx, (data, target) in enumerate(train_loader):
            data, target = data.to(device), target.to(device)
            optimizer.zero_grad()
            output = ddp_model(data)
            loss = criterion(output, target)
            loss.backward()
            optimizer.step()
    
    # 保存模型（仅在主进程中）
    if rank == 0:
        torch.save(model.state_dict(), 'ddp_model.pth')
    
    # 清理进程组
    dist.destroy_process_group()

if __name__ == '__main__':
    main()

启动命令

在每台机器上运行以下命令，假设使用两台机器，每台机器有8个GPU：

# 节点1
torchrun --nproc_per_node 8  --nnodes 2 --node_rank 0 --rdzv_endpoint 10.192.2.1:62111  ddp_example.py

# 节点2
torchrun --nproc_per_node 8  --nnodes 2 --node_rank 1 --rdzv_endpoint 10.192.2.1:62111  ddp_example.py

在具体训练前，可以设置CUDA_VISIBLE_DEVICES信息，限制参与训练的显卡量。具体可以参考：https://blog.csdn.net/ababab12345/article/details/134940973

注意事项

• 确保每台机器的环境变量（如MASTER_ADDR和MASTER_PORT）配置正确。
• 在多机环境下，网络连接可能会成为瓶颈，需要确保网络配置良好。
• 使用DistributedSampler来确保数据在不同进程中的分布是均匀的，并且在每个epoch开始时调用set_epoch来更新随机种子。
• 在保存模型时，通常只在主进程中保存，以避免重复保存。

3、DDP训练的优势

在单个服务器上使用DDP（Distributed Data Parallel）和DP（Data Parallel）训练模型的效率存在显著差异，以下是两者的对比：

实现方式

• DP：基于单进程多线程实现，适用于单机多GPU场景。
• DDP：基于多进程实现，每个GPU对应一个独立的进程，适用于单机多GPU和多机多卡场景。

通信效率

• DP：通信集中在主GPU，随着GPU数量的增加，通信成本呈线性上升，可能导致通信瓶颈。
• DDP：采用高效的通信模式（如ring-all-reduce），减少了通信瓶颈，提高了通信效率。

负载均衡

• DP：主GPU的通信和计算压力较大，存在负载不均衡的问题。
• DDP：各GPU独立工作，负载均衡，避免了单点瓶颈。

内存占用

• DP：所有GPU都维护完整的模型副本，显存占用较高。
• DDP：每个进程有独立的模型副本，显存占用相对较低。

容错性

• DP：如果主GPU出现问题，整个训练可能会中断。
• DDP：有较好的容错机制，单个GPU出问题不会导致整个训练中断。

性能表现

• DP：受GIL影响，性能受限，且随着GPU数量增加，性能提升有限。
• DDP：性能更优，可扩展性更好，适用于大规模分布式训练。

综上所述，在单机多卡环境下，DDP相较于DP在通信效率、负载均衡、内存占用和容错性等方面具有明显优势，因此在单机多卡训练中，DDP通常比DP更高效。

4、DPP训练命令中的相关参数

在使用DDP（Distributed Data Parallel）进行分布式训练时，torchrun命令中的参数配置对于训练的启动和运行至关重要。以下是对这些参数的详细解释：

1. --nproc_per_node

• 含义：指定每个节点上启动的进程数。通常等于该节点上的GPU数量。
• 示例：--nproc_per_node=8 表示每个节点启动8个进程，适用于单机8卡的情况。

2. --nnodes

• 含义：指定参与训练的总节点数。
• 示例：--nnodes=2 表示使用2个节点进行分布式训练。

3. --node_rank

• 含义：指定当前节点的排名，通常从0开始编号。
• 示例：--node_rank=0 表示当前节点是主节点，--node_rank=1 表示第二个节点。

4. --master_addr

• 含义：指定主节点的IP地址，用于初始化进程组。
• 示例：--master_addr=192.168.1.1 表示主节点的IP地址为192.168.1.1。

5. --master_port

• 含义：指定主节点的端口号，用于初始化进程组。
• 示例：--master_port=12355 表示主节点的端口号为12355。

6. --use_env

• 含义：使用环境变量来传递rank和world_size等信息，而不是通过命令行参数。
• 示例：--use_env 可以简化命令行参数的配置，尤其在复杂的分布式环境中。

示例命令

以下是一个完整的启动命令示例，适用于单机8卡的训练场景：

torchrun --nproc_per_node=8 --nnodes=1 --node_rank=0 --master_addr=localhost --master_port=12355 train.py

参数总结

• --nproc_per_node：每个节点的进程数（通常等于GPU数量）。
• --nnodes：总节点数。
• --node_rank：当前节点的排名。
• --master_addr：主节点的IP地址。
• --master_port：主节点的端口号。
• --use_env：使用环境变量传递配置信息。

这些参数共同决定了分布式训练的启动和运行方式，确保每个进程能够正确地初始化和通信。

此外，还有一个rank参数，它是DDP核心概念之一，它用于标识每个进程在整个分布式训练环境中的唯一身份。以下是关于rank参数及其相关参数的详细解释：

rank参数

• 含义：rank是一个全局唯一标识符，用于区分分布式训练中的每个进程。它是一个非负整数，通常从0开始编号。
• 作用：
  • 在初始化进程组时，rank用于指定当前进程的身份。
  • 在数据采样器（DistributedSampler）中，rank用于确定每个进程应处理的数据子集。
  • 在日志记录和模型保存时，通常只有rank=0的进程执行这些操作，以避免重复。

相关参数

• world_size：

  • 含义：表示分布式训练中总的进程数（或GPU数）。
  • 作用：在初始化进程组和数据采样器时，world_size用于指定总的进程数量。
  • 示例：如果使用4个GPU进行训练，则world_size=4。

• local_rank：

  • 含义：表示当前进程在本地节点上的排名。
  • 作用：用于指定当前进程在本地机器上使用的GPU设备。
  • 示例：如果一个节点上有4个GPU，local_rank可以是0、1、2或3。

初始化进程组

在初始化进程组时，rank和world_size是必需的参数。以下是一个示例代码：

import torch.distributed as dist

dist.init_process_group(
    backend='nccl',  # 使用NCCL后端，适用于GPU训练
    init_method='tcp://127.0.0.1:23456',  # 指定通信初始化方法
    world_size=4,  # 总的进程数
    rank=0  # 当前进程的rank
)

数据采样器

在使用DistributedSampler时，rank和world_size用于确保每个进程处理不同的数据子集：

from torch.utils.data import DistributedSampler

train_sampler = DistributedSampler(
    train_dataset,  # 数据集
    num_replicas=dist.get_world_size(),  # 总的进程数
    rank=dist.get_rank()  # 当前进程的rank
)

训练循环中的使用

在训练循环中，通常只有rank=0的进程执行日志记录和模型保存操作，以避免重复：

if dist.get_rank() == 0:
    # 执行日志记录和模型保存操作
    print("Logging and saving model...")

总结

• rank：全局唯一标识符，用于区分每个进程。
• world_size：总的进程数，用于初始化进程组和数据采样器。
• local_rank：本地节点上的排名，用于指定当前进程使用的GPU设备。

这些参数共同确保了分布式训练的正确性和效率。