分布式训练

深度学习

分为计划、想法和测试三部分。

计划部分：用于进度共享

想法部分：我的一些自问自答，都是关于具体应用的时候一些需要注意的问题，没有回答的原因是因为没有来得及去调研。

测试部分：用来记录一些实现的时候需要注意的东西。

计划

1.公网情况下的多机多卡------------ 100%
2.ZeroTier局域网数据互通
3.局域网下的多机多卡
4.模型使用PointNet++进行多机多卡训练、并记录性能占用
5.性能瓶颈与优化

想法

需要注意的问题

多机多卡情况的梯度合并

Ring-Reduce： 这个方法对于带宽要求确实低、但是害怕算力提供端不稳定、 某个提供点突然下线怎么办？ 目前有没有解决方案？ 需要再找找。

不同显存与算力机器的Batch Size设置

这个应该根据当前机器的性能去动态的指定、 需要考虑一下， 不同的BS下 是否会对精度产生影响？ 梯度聚合的时候需不需要额外操作？

动态增加算力提供端

现在的ddp是否支持动态增加机器？如何实现？
torchrun支持动态算力提供、可以设置worker范围。

测试环境：

代码测试部分

import argparse
from tqdm import tqdm
import torch
import torchvision
import torch.nn as nn
import torch.nn.functional as F
import torch.distributed as dist
from torch.nn.parallel import DistributedDataParallel as DDP
import random
import numpy as np
import os
def init_seeds(seed=0, cuda_deterministic=True):
    random.seed(seed)
    np.random.seed(seed)
    torch.manual_seed(seed)
# 模型无需修改
class ToyModel(nn.Module):
    def __init__(self):
        super(ToyModel, self).__init__()
        self.conv1 = nn.Conv2d(3, 6, 5)
        self.pool = nn.MaxPool2d(2, 2)
        self.conv2 = nn.Conv2d(6, 16, 5)
        self.fc1 = nn.Linear(16 * 5 * 5, 120)
        self.fc2 = nn.Linear(120, 84)
        self.fc3 = nn.Linear(84, 10)
    def forward(self, x):
        x = self.pool(F.relu(self.conv1(x)))
        x = self.pool(F.relu(self.conv2(x)))
        x = x.view(-1, 16 * 5 * 5)
        x = F.relu(self.fc1(x))
        x = F.relu(self.fc2(x))
        x = self.fc3(x)
        return x
def get_dataset():
    transform = torchvision.transforms.Compose([
        torchvision.transforms.ToTensor(),
        torchvision.transforms.Normalize((0.5, 0.5, 0.5), (0.5, 0.5, 0.5))
    ])
    my_trainset = torchvision.datasets.CIFAR10(root='./data', train=True,
        download=True, transform=transform)
    # dataLoader需要使用DistributedSampler去分发数据。
    train_sampler = torch.utils.data.distributed.DistributedSampler(my_trainset)
    trainloader = torch.utils.data.DataLoader(my_trainset,
        batch_size=16, num_workers=2, sampler=train_sampler)
    return trainloader
# 在torchrun时已经指定参数、 此时无序传参
dist.init_process_group()
rank = torch.distributed.get_rank()
init_seeds(1 + rank)
#torchrun的新改动，替代原先的LOCAL_RANK获取方式
local_rank = int(os.environ["LOCAL_RANK"])
torch.cuda.set_device(local_rank)
# 准备就绪、开发分割数据。
trainloader = get_dataset()
# 分发模型
model = ToyModel().to(local_rank)
# load模型数据要在构造DDP模型之前，且只需要在master上加载就行了。
ckpt_path = None
if dist.get_rank() == 0 and ckpt_path is not None:
    model.load_state_dict(torch.load(ckpt_path))
# DDP: 构造DDP model
model = DDP(model, device_ids=[local_rank], output_device=local_rank)
# DDP: 要在构造DDP model之后，才能用model初始化optimizer。
optimizer = torch.optim.SGD(model.parameters(), lr=0.001)
# loss分发
loss_func = nn.CrossEntropyLoss().to(local_rank)
model.train()
iterator = tqdm(range(100))
for epoch in iterator:
    # DDP：设置sampler的epoch，
    # DistributedSampler需要这个来指定shuffle方式，
    # 通过维持各个进程之间的相同随机数种子使不同进程能获得同样的shuffle效果。
    trainloader.sampler.set_epoch(epoch)
    # 后面这部分，则与原来完全一致了。
    for data, label in trainloader:
        data, label = data.to(local_rank), label.to(local_rank)
        optimizer.zero_grad()
        prediction = model(data)
        loss = loss_func(prediction, label)
        loss.backward()
        iterator.desc = "loss = %0.3f" % loss
        optimizer.step()
    # DDP:
    # 1. save模型的时候，和DP模式一样，有一个需要注意的点：保存的是model.module而不是model。
    #    因为model其实是DDP model，参数是被`model=DDP(model)`包起来的。
    # 2. 只需要在进程0上保存一次就行了，避免多次保存重复的东西。
    if dist.get_rank() == 0:
        torch.save(model.module.state_dict(), "%d.ckpt" % epoch)
# 多机环境下 $PORT需要一致。 此处可以自行实现分发程序、指定端口号。
CUDA_VISIBLE_DEVICES="2,3" torchrun --nnodes=2 --nproc-per-node=2 --max-restarts=3 --rdzv-id=777 --rdzv-backend=c10d --rdzv-endpoint=localhost:8777 main.py
CUDA_VISIBLE_DEVICES="0,1" torchrun --nnodes=2 --nproc-per-node=2 --max-restarts=3 --rdzv-id=777 --rdzv-backend=c10d --rdzv-endpoint=36.138.4.117:8777 main.py

CUDA_VISIBLE_DEVICES="2,3" python -m torch.distributed.launch --nnodes 2 --node_rank 1 --nproc_per_node 2 --master_addr 36.138.4.117 --master_port 8777 main.py