PyTorch分布式overview

PyTorch分布式overview

• Distributed Data-Parallel Training (DDP)：是一个广泛 采用的单平台（单机），多数据训练范式。使用DDP, 模型is replicated on every process 每一个模型复制品将会被fead with a different set of input data samples. DDP照顾好了梯度通信来保持模型替代品同步，同时，overlaps it with the gradient computations to speed up training.
• RPC-Based Distributed Training (RPC)
• Collective Communication (c10d)

从简单到复杂，从原型到生产，普遍的研发轨迹(trajectory)是：

1.使用单设备(single-device)训练，也就是使用单个GPU训练，如果数据和模型能够适应一个GPU, 则训练速度不是一个问题。

所以后面是数据和模型无法适应一个GPU, 例如batch_size设置的比较大。

2.使用单个机器，多个GPU数据并行(DataParallel)来利用一个机器上的多个GPU来加速训练with minimal code changes.

3.使用单个机器，多个GPU分布式数据并行(DistributedDataParallel), 如果你would like to further speed up training and are willing to write a little more code to set it up.

4.使用多个机器的分布式数据训练和启动脚本, 如果这个应用需要在多个机器上跑。

5.使用torch.distributed.elastic(弹性的），来启动分步式训练 if errors(例如，out-of-memory) are expected or if resources can join and leave dynamically during training.

torch.nn.DataParallel

DataParallel包允许了单个机器多个GPU并行 with the lowest coding hurdle(障碍）. 它仅需要对应用代码做一行改变。尽管DataParallel很容易使用，它通常没有提供最好的性能, 因为它在每一个forward pass的过程中，均需要replicates the model. 它的单进程多线程机制自然地遭受哦了GIL不一致问题。为了获得更好的性能， 考虑使用分布式数据并行 。

torch.nn.parallell.DistributedDataParallel.

和DataParallel相比，DistributedDataParallel还需要一个设置的步骤，即调用该init_process_group. DDP使用的是读进程机制，并且在模型复制品之间不存在GIL冲突。并且，模型是在DDP构建的过程中广播的，而不是在每一个前向传播的过程中传播，这也帮助了加速训练。

分布式数据并行训练(DistributedDataParallel)

PyTorch提供了好几个进行数据并行训练的选项。

Pytorch分布式的数据并行

分布式数据并行DistributedDataParallel (DDP) 实现了在模块级别的数据并行，可以运行在多个机器上。使用DDP的应用应该产生(spawn)多个进程，并且为每一个进程创建一个单个的DDP实例。

更具体地，DDP为model.parameters()中每一个参数注册了一个自动求导hook, 并且这个hook will fire 当相应的梯度在后向传播的过程中被计算的时候。然后DDP使用那个信号触发跨进程的梯度同步。

使用DDP的推荐的方式是为每一个模型复制品产生一个进程，其中每一个模型复制品可以分布在多个设备（多张GPU)上。DDP进程可以被放在相同的机器上或者是跨机器，但是GPU设备不能够被多个进程所共享。

即一个进程可以使用多个GPU，每一个GPU只能被一个进程使用。

本教程从一个基本的DDP用例开始，然后展示了更多的高级用例，包括checkpointing model以及将DDP和模型并行结合起来。

这个教程中的代码运行在一个8-GPU的服务器上，但是它可以很容易地被泛化到其他的环境。

接触DDP, 从这个例子开始

参考DDP notes

下面的例子使用了一个torch.nn.Linear作为本地模型，并使用DDP将它包装。

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

def example(rank, world_size):
	# create default process group
  dist.init_process_group("gloo", rank = rank, world_size = world_size)
  # create local model
  model = nn.Linear(10,10).to(rank) #rank是GPU的编号
  # construct DDP model
  ddp_model = DDP(model, device_ids = [rank]) # 从这里可以看出rank 是GPU的编号
  # define loss function and optimizer
  loss_fn = nn.MSELoss()
  optimizer = optim.SGD(ddp_model.parameters(), lr = 0.001)

  # forward pass
  outputs = ddp_model(torch.randn(20,10).to(rank))
  labels = torch.randn(20,10).to(rank)
  # backward pass
  loss_fn(outputs, labels).backward()
  # update parameters
  optimizer.step()
  
def main():
  world__siize = 2
  mp.spawn(example,args = (world_size,)), nprocs=world_size, join=True) # world_size就是进程的个数。猜测：同时由于一个GPU不可以被多个进程共享，所以这里一般就是GPU的个数。
  # 根据官方文档：这个函数的目的是：Spawns nprocs processes that ruun example with args.
  
if __name__=='__main__':
   # Environment variables which need to be
    # set when using c10d's default "env"
    # initialization mode.
    os.environ["MASTER_ADDR"] = "localhost"
    os.environ["MASTER_PORT"] = "29500"
    main()

分布式数据并行(DistributedDataParallel)基本用例(Basic Use Case)

参考ddp tutorial

为了创建一个DDP模块 ，你必须首先设定进程组properly. 更多的细节可以被发现在Writing Distributed Applications with PyTorch.

Setup

包含在PyTorch中的分布式的包， torch.distributed 使得研究者和实践者能够容易在进程和集群之间并行他们的计算。为了这么做，它借助了消息传递语义，允许每个进程将数据传递给其他的任何一个进程。和多进程包(torch.multiprocessing)相反，进程可以使用不同的通讯后端，不仅限于在同一台机器上执行。

为了开始，我们需要同时运行多个进程的能力 。如果你有一台机器集群，你要check with your local sysadmin or use the favorite coordination tool (e.g., pdsh, clustershell, or others).

为了这个目的，我们要使用一个单个的机器并使用下列模板产生多个进程。

import os
import torch
import torch.distributed as dist
import torch.multiprocessing as mp

def run(rank, size):
    """Distributed function to be implemented later"""
    pass

def init_process(rank, size, fn, backend='gloo'):
    """Initialize the distributed environment"""
    os.environ['MASTER_ADDR'] = '127.0.0.1'
    os.environ['MASTER_PORT'] = '29501'
    dist.init_process_group(backend, rank=rank, world_size=size)
    fn(rank, size)

if __name__ == '__main__':
    size = 2
    processes = []
    mp.set_start_method('spawn')
    for rank in range(size):
        p = mp.Process(target=init_process, args=(rank, size, run))
        p.start()
        processes.append(p)
    for p in processes:
        p.join()

以上的脚本生成了两个进程，每个进程都会设定分布式环境，初始化进程组(dist.init_process_group), 并且最后执行了给定的run函数。

让我们来看一下init_process函数。它确保了每一个进程都能够coordinate through a master, using the same ip address and port. 注意到，我们使用了gloo backend but other backends are available.

我们会go over the magic happening in dist.init_process_group at the end of this tutorial, 但是它本质上允许了进程彼此通信by sharing their locations.

数据从一个进程转移到另一个进程称为点到点通讯。这些事通过send和recv函数来取得的。

Ok, 写到这里，基本就已经了解了如何使用分布式数据并行了，了解了分布式数据并行和数据并行的区别了。

分布式数据并行是多进程的，数据并行是多线程的。

单机的模型并行最好的实践

模型并行

模型并行在分布式训练技巧中广泛应用。之前的博客已经解释了如何使用DataParallel 来再多个GPU上训练一个网络。这个特征在所有GPU上重复了相同的模型，其中每一个消耗了不同划分的输入数据。

尽管DataParallel可以很大程度上加速训练过程，it does not work for some uses cases where the model is too large to fit into a single GPU.

本博客显示了如何使用模型并行解决这个问题。模型并行将一个单个模型划分到不同的GPU上，而不是在每一个GPU上复制整个模型（例如，一个模型m包含10层，当使用DataParallel的时候，每一个GPU都会有一个这10层的复制品，然而当使用模型并行的时候，每一个GPU可以装5层）。

模型并行的高视角的想法是：将一个模型不同子网络放在不同的设备上，并且相应地实现forwad方法，在不同的GPU之间移动中间输出。由于仅有模型的一部分运行在任意单个的GPU上，一个设备集合可以集体承担一个更大的模型。在这一篇博文中，我们不会尝试构造大模型然后将他们防止到有限的GPU上。Instead, 这篇博文关注展示模型并行的思路。取决于读者将这些思路应用到真实世界的应用中。

import torch
import torch.nn as nn
import torch.optim as optim


class ToyModel(nn.Module):
    def __init__(self):
        super(ToyModel, self).__init__()
        self.net1 = torch.nn.Linear(10, 10).to('cuda:0')
        self.relu = torch.nn.ReLU()
        self.net2 = torch.nn.Linear(10, 5).to('cuda:1')

    def forward(self, x):
        x = self.relu(self.net1(x.to('cuda:0')))
        return self.net2(x.to('cuda:1'))

可以看出，模型并行和数据并行是两码事。

DataParallel和DistributedDataParallel的区别

在我们进入之前，让我们阐明一下为什么尽管增加了复杂度，你会考虑使用DistributedDataParallel over DataParallel.

• 首先，**DataParallel是单进程的，多线程的，并且仅运行在一个机器上, while DistributedDataParallel 是一个多进程的，**既可以进行单机器训练，也可以进行多机器训练。DataParallel通常比DistributedDataParallel要慢even on a single machine due to GIL contention (不一致）across threads, per-iteration replicated model, and additional overhead introduced by scattering inputs and gathering outputs.（所以，如果数据量没有超过一张GPU的显存，则直接在一张GPU上训练即可。使用多张GPU分散训练可能还会减慢速度。）
• 回顾一下之前的教程，如果你的模型太大以至于难以放在单个GPU上，你必须使用model parallel来将其分散在多个GPU上。DistributedDataParallel works with model parallel. DataParallel无法和模型并行一起使用。当DDP和模型并行结合起来之后，每一个DDP进程均会使用模型并行，并且所有的进程会集体使用数据并行。
• 如果你的模型需要分布在多个机器上，或者你的用例不适合数据并行范式，请参考RPC API.

学习经历总结

今天上午来了之后，遇到了多卡并行的bug, 就想着学习一下这方面的知识，最后找到了官方的教程，然后一边看一边翻译，一边做笔记，一边理解，最后到了下午两点半的时候，对GPU多卡并行基本知道了一个大概的思路和框架，没有那么畏惧了。

看完这些东西，我对分布式数据并行有了一个基本的了解。分布式学习包括数据并行、模型并行、分布式数据并行、多个机器的分布式数据并行、弹性分布式数据并行。

数据并行是单机、单进程、多线程(多GPU)的。

模型并行是单机，将多个模型分布在不同的GPU上。每一个GPU上只存储模型的一部分。

分布式数据并行是多进程的，每一个GPU仅可以被一个进程功能占用。速度更快。

多个机器的分布式数据并行是在多个机器上。

弹性分布式数据并行是当cuda out of memory等类似的现象出现之后再采用数据并行。

还有，再看到分布式数据并行的代码中有init之类的代码，不会被吓到，知道init代码的功能是分布式数据并行的配置代码。