文具网站建设策划书,润和软件是外包公司吗,郑州做网站公司有多少钱,互联网保险销售行为可回溯管理基于Pytorch的分布式数据并行训练 动机为什么要并行分布数据#xff1f;现有资料的不足 Outline整体框架图带解释的最小demo示例没有multiprocessing开启multiprocessing 分布式训练启动方式混合精度训练#xff08;采用apex#xff09; 参考资料 简介#xff1a; 在PyTorc… 基于Pytorch的分布式数据并行训练 动机为什么要并行分布数据现有资料的不足 Outline整体框架图带解释的最小demo示例没有multiprocessing开启multiprocessing 分布式训练启动方式混合精度训练采用apex 参考资料 简介 在PyTorch中使用DistributedDataParallel进行多GPU分布式模型训练 动机
加速神经网络训练的最简单方法是使用GPU它在神经网络中常见的计算类型矩阵乘法和加法上提供了比CPU更大的加速。随着模型或数据集变得越来越大一个GPU很快就会变得不足。例如像BERT和GPT-2这样的大型语言模型是在数百个GPU上训练的。要执行多GPU训练我们必须有一种方法在不同的GPU之间分割模型和数据并协调训练。
为什么要并行分布数据
很多人喜欢在Pytorch中实现自己的深度学习模型因为它在神经网络框架的控制和易用性之间具有最佳平衡。Pytorch有两种方法可以跨多个GPU拆分模型和数据nn.DataParallel和nn.DistributedDataParallel。
nn.DataParallel更易于使用只需包装模型并运行训练脚本。然而由于它使用一个process来计算模型权重然后在每个批次中将其分配给每个GPU因此网络很快成为瓶颈GPU利用率通常很低。此外nn.DataParallel要求所有GPU都在同一个节点上并且不能与Apex一起用于mixed-precision训练。
因此nn.DataParallel和nn.DistributedDataParallel的主要差异可以总结为以下几点 1. DistributedDataParallel支持模型并行而DataParallel不支持这意味着如果模型太大单卡显存不足时只能使用前者 2. DataParallel是单进程多线程只用于单机情况而DistributedDataParallel是多进程的适用于单机和多机情况真正实现分布式训练 3. DistributedDataParallel的训练更高效因为每个进程都是独立的Python解释器避免GIL问题而且通信成本低其训练速度更快基本上DataParallel已经被弃用 4. 必须要说明的是DistributedDataParallel中每个进程都有独立的优化器执行自己的更新过程但是梯度通过通信传递到每个进程所有执行的内容是相同的。
现有资料的不足
总的来说Pytorch文档是完整且清晰的但是当试图弄清楚如何使用DistributedDataParallel时发现所有的示例和教程都是不可访问、不完整或重载了不相关的功能的组合。
Pytorch提供了一个关于使用AWS进行分布式培训的教程它很好地展示了如何在AWS方面进行设置。然而它的其余部分有点混乱因为出于某种原因它花了很多时间来展示如何计算指标然后再回到展示如何包装模型和启动流程。它也没有描述nn.DistributedDataParallel的作用这使得相关的代码块很难遵循。
关于用Pytorch编写分布式应用程序的教程比第一遍所需的要详细得多而且对于没有Python多处理背景的人来说是无法访问的。它花费了大量时间复制nn.DistributedDataParallel中的功能。然而它没有给出它所做工作的高级概述也没有提供如何使用它的见解(https://pytorch.org/tutorials/intermediate/ddp_tutorial.html)
还有一个Pytorch教程介绍如何开始使用分布式数据并行。这一个展示了如何进行一些设置但没有解释设置的目的然后展示了一些代码以在GPU之间拆分模型并进行一个优化步骤。不幸的是我很确定编写的代码不会运行函数名不匹配而且它也没有告诉你如何运行代码。与前面的教程一样它也没有对分布式培训的工作原理进行高层次的概述。
Pytorch提供的最接近MWE示例的是Imagenet训练示例。不幸的是这个例子还展示了Pytorch几乎所有的其他功能因此很难找出分布式多GPU训练的相关内容。
Apex提供了他们自己版本的Pytorch Imagenet示例。他们的nn.DistributedDataParallel版本是Pytorch的替代品只有在学习如何使用Pytork之后才有帮助。
这个教程很好地描述了引擎盖下发生的事情以及它与nn.DataParallel的区别。但是它没有关于如何使用nn.DataParallel的代码示例。
Outline
本教程真正针对的是那些已经熟悉在Pytorch中训练神经网络模型的人。首先概述整体思想。然后展示了在GPU上使用MNIST进行训练的最小工作示例。我修改了这个例子在多个GPU上进行训练可能跨越多个节点并逐行解释这些变化。重要的是还解释了如何运行代码。作为奖励还演示了如何使用Apex进行简单的混合精度分布式训练。
整体框架图
使用DistributedDataParallel进行Multiprocessing会在多个GPU上复制模型每个GPU都由一个进程控制。进程是在计算机上运行的python的一个实例通过让多个进程并行运行我们可以利用具有多个CPU核心的procressor。如果你愿意可以让每个进程控制多个GPU但这显然比每个进程有一个GPU要慢。也可以让多个工作进程为每个GPU获取数据但为了简单起见将省略这一点。GPU可以全部位于同一节点上也可以分布在多个节点上。一个节点是一台“计算机”包括它的所有CPU和GPU。如果你使用AWS一个节点就是一个EC2实例。每个进程都执行相同的任务每个进程都与所有其他进程通信。只有梯度在进程/GPU之间传递这样网络通信就不会成为瓶颈。 在训练过程中每个进程从磁盘加载自己的mini-batches并将其传递给GPU。每个GPU都有自己的前向通道然后GPU之间的梯度都会减小。每一层的梯度不依赖于前一层因此梯度all-reduce与反向传递同时计算以进一步缓解网络瓶颈。在反向过程结束时每个节点都具有平均梯度确保模型权重保持同步。
所有这些都要求多个进程可能在多个节点上同步并进行通信。Pytorch通过其distributed.init_process_group函数来实现这一点。此函数需要知道在哪里可以找到进程0以便所有进程都可以同步以及预期的进程总数。每个单独的进程还需要知道进程的总数及其在进程中的排名以及使用哪个GPU。将进程的总数称为world size是很常见的。最后每个进程都需要知道要处理哪一部分数据以便批处理不重叠。Pytorch提供了nn.utils.data.DistributedSampler来实现这一点即为各个进程切分数据以保证训练数据不重叠。
更详细的DDP的内部机理见官方的文档介绍DISTRIBUTED DATA PARALLEL
带解释的最小demo示例
为了演示如何做到这一点将创建一个在MNIST上训练的示例然后将其修改为在多个节点的多个GPU上运行最后还允许混合精度训练mixed-precision training。
没有multiprocessing
首先导入所需要的依赖库
import os
import argparse
import torch
import torch.nn as nn
import torch.distributed as dist
import torch.multiprocessing as mp
import torchvision
import torchvision.transforms as transforms
from datetime import datetime
from apex.parallel import DistributedDataParallel as DDP
from apex import amp我们定义了一个非常简单的卷积模型来预测MNIST。
class ConvNet(nn.Module):def __init__(self, num_classes10):super(ConvNet, self).__init__()self.layer1 nn.Sequential(nn.Conv2d(1, 16, kernel_size5, stride1, padding2),nn.BatchNorm2d(16),nn.ReLU(),nn.MaxPool2d(kernel_size2, stride2))self.layer2 nn.Sequential(nn.Conv2d(16, 32, kernel_size5, stride1, padding2),nn.BatchNorm2d(32),nn.ReLU(),nn.MaxPool2d(kernel_size2, stride2))self.fc nn.Linear(7 * 7 * 32, num_classes)def forward(self, x):out self.layer1(x)out self.layer2(out)out out.reshape(out.size(0), -1)out self.fc(out)return out下面是训练过程
def train(gpu, args):torch.manual_seed(0)model ConvNet()torch.cuda.set_device(gpu)model.cuda(gpu)# model nn.DataParallel(model, device_idsdevice_ids)# model model.cuda(devicegpu)batch_size 100# define loss function (criterion) and optimizercriterion nn.CrossEntropyLoss().cuda(gpu)optimizer torch.optim.SGD(model.parameters(), 1e-4)# Data loading codetrain_dataset torchvision.datasets.MNIST(root./data, trainTrue, transformtransforms.ToTensor(),downloadTrue)train_loader torch.utils.data.DataLoader(datasettrain_dataset, batch_sizebatch_size, shuffleTrue,num_workers0, pin_memoryTrue)start datetime.now()total_step len(train_loader)for epoch in range(args.epochs):for i, (images, labels) in enumerate(train_loader):images images.cuda(non_blockingTrue)labels labels.cuda(non_blockingTrue)# Forward passoutputs model(images)loss criterion(outputs, labels)# Backward and optimizeoptimizer.zero_grad()loss.backward()optimizer.step()if (i 1) % 100 0 and gpu 0:print(Epoch [{}/{}], Step [{}/{}], Loss: {:.4f}.format(epoch 1, args.epochs, i 1, total_step,loss.item()))if gpu 0:print(Training complete in: str(datetime.now() - start))main()函数将接受一些参数并运行训练函数。
def main():parser argparse.ArgumentParser()parser.add_argument(-n, --nodes, default1, typeint, metavarN)parser.add_argument(-g, --gpus, default1, typeint, helpnumber of gpus per node)parser.add_argument(-nr, --nr, default0, typeint, helpranking within the nodes)parser.add_argument(--epochs, default2, typeint, metavarN, helpnumber of total epochs to run)args parser.parse_args()train(0, args)最后要确保main()函数被调用。
if __name__ __main__:main()可以通过打开一个终端并键入python src/mnist.py-n 1-g 1-nr 0来运行此代码它将在单个节点上的单个gpu上进行训练。
开启multiprocessing
要使用multiprocessing来做到这一点我们需要一个脚本来为每个GPU启动一个进程。每个进程都需要知道要使用哪个GPU以及它在所有正在运行的进程中的排名。需要在每个节点上运行脚本。
来看看对每个函数的更改。为了便于查找已将新代码隔离开来
def main():parser argparse.ArgumentParser()parser.add_argument(-n, --nodes, default1, typeint, metavarN)parser.add_argument(-g, --gpus, default1, typeint, helpnumber of gpus per node)parser.add_argument(-nr, --nr, default0, typeint, helpranking within the nodes)parser.add_argument(--epochs, default2, typeint, metavarN, helpnumber of total epochs to run)args parser.parse_args()#########################################################args.world_size args.gpus * args.nodesos.environ[MASTER_ADDR] 172.20.109.105os.environ[MASTER_PORT] 8888mp.spawn(train, nprocsargs.gpus, args(args,))########################################################## train(0, args)其中
args.nodes表示节点总数args.gpus表示每个节点的GPU总数每个节点GPU数是一样的args.nr表示当前节点在所有节点中的序号。
根据节点总数和每个节点的GPU数可以计算world_size即要运行的进程总数所有的进程需要知道进程0的IP地址以及端口这样所有进程可以在开始时同步一般情况下称进程0是master进程比如我们会在进程0中打印信息或者保存模型。
PyTorch提供了mp.spawn来在一个节点启动该节点所有进程每个进程运行train(i, args)其中 i 从0到args.gpus - 1。请记住在每个节点上运行main()函数这样总共会有args.nodes*args.gpusargs.world_size进程。
同样我们要修改训练函数
def train(gpu, args):############################################################rank args.nr * args.gpus gpudist.init_process_group(backendnccl, init_methodenv://, world_sizeargs.world_size, rankrank)############################################################torch.manual_seed(0)model ConvNet()torch.cuda.set_device(gpu)model.cuda(gpu)batch_size 100# define loss function (criterion) and optimizercriterion nn.CrossEntropyLoss().cuda(gpu)optimizer torch.optim.SGD(model.parameters(), 1e-4)############################################################# Wrap the modelmodel nn.parallel.DistributedDataParallel(model, device_ids[gpu])############################################################# Data loading codetrain_dataset torchvision.datasets.MNIST(root./data, trainTrue, transformtransforms.ToTensor(),downloadTrue)############################################################train_sampler torch.utils.data.distributed.DistributedSampler(datasettrain_dataset, num_replicasargs.world_size,rankrank)train_loader torch.utils.data.DataLoader(datasettrain_dataset, batch_sizebatch_size, shuffleFalse,num_workers0, pin_memoryTrue, samplertrain_sampler)############################################################start datetime.now()total_step len(train_loader)for epoch in range(args.epochs):for i, (images, labels) in enumerate(train_loader):images images.cuda(non_blockingTrue)labels labels.cuda(non_blockingTrue)# Forward passoutputs model(images)loss criterion(outputs, labels)# Backward and optimizeoptimizer.zero_grad()loss.backward()optimizer.step()if (i 1) % 100 0 and gpu 0:print(Epoch [{}/{}], Step [{}/{}], Loss: {:.4f}.format(epoch 1, args.epochs, i 1, total_step,loss.item()))if gpu 0:print(Training complete in: str(datetime.now() - start))这里首先计算出当前进程序号rank args.nr * args.gpus gpu然后就是通过dist.init_process_group初始化分布式环境其中
backend参数指定通信后端包括mpi、gloo、nccl这里选择nccl它是Nvidia提供的官方多卡通信框架相对比较高效。mpi也是高性能计算常用的通信协议不过需要自己安装MPI实现框架比如OpenMPI。gloo倒是内置通信后端但是不够高效。init_method指的是如何初始化以完成刚开始的进程同步这里我们设置的是env://指的是环境变量初始化方式需要在环境变量中配置4个参数MASTER_PORTMASTER_ADDRWORLD_SIZERANK前面两个参数我们已经配置后面两个参数也可以通过dist.init_process_group函数中world_size和rank参数配置。 其它的初始化方式还包括共享文件系统以及TCP比如采用TCP作为初始化方法init_methodtcp://10.1.1.20:23456其实也是要提供master的IP地址和端口。注意这个调用是阻塞的必须等待所有进程来同步如果任何一个进程出错就会失败。对于模型侧只需要用DistributedDataParallel包装一下原来的model即可将模型复制到GPU上以进行处理在背后它会支持梯度的All-Reduce操作。对于数据侧使用nn.utils.data.DistributedSampler来给各个进程切分数据只需要在dataloader中使用这个sampler就好值得注意的一点是要在训练循环过程的每个epoch开始时调用train_sampler.set_epoch(epoch)主要是为了保证每个epoch的划分是不同的其它的训练代码都保持不变。
最后就可以执行代码了比如我们有4节点每个节点是8张显卡那么需要在4个节点终端上分别执行
python src/mnist-distributed.py -n 4 -g 8 -nr i例如在节点0上执行
python src/mnist-distributed.py -n 4 -g 8 -nr 0换句话说在每个节点上运行这个脚本告诉它在训练开始前启动彼此同步的args.gpus进程。
要注意的是此时的有效batch_size其实是batch_size_per_gpu * world_size对于有BN的模型还可以采用同步BN获取更好的效果
model torch.nn.SyncBatchNorm.convert_sync_batchnorm(model)上述讲述的是分布式训练过程其实同样适用于评估或者测试过程比如我们把数据划分到不同的进程中进行预测这样可以加速预测过程。实现代码和上述过程完全一样不过我们想计算某个指标那就需要从各个进程的统计结果进行All-Reduce因为每个进程仅是计算的部分数据的内容。比如我们要计算分类准确度我们可以统计每个进程的数据总数total和分类正确的数量count然后进行聚合。
这里要提的一点当用dist.init_process_group初始化分布式环境时其实就是建立一个默认的分布式进程组distributed process group这个group同时会初始化Pytorch的torch.distributed包。这样我们可以直接用torch.distributed的API就可以进行分布式基本操作了下面是具体实现
# define tensor on GPU, count and total is the result at each GPU
t torch.tensor([count, total], dtypetorch.float64, devicecuda)
dist.barrier() # synchronizes all processes
dist.all_reduce(t, optorch.distributed.ReduceOp.SUM,) # Reduces the tensor data across all machines in such a way that all get the final result.
t t.tolist()
all_count int(t[0])
all_total int(t[1])
acc all_count / all_total分布式训练启动方式
上述过程中采用PyTorch的torch.multiprocessing包Multiprocessing package - torch.multiprocessing来启动分布式训练目前官方给出的ImageNet训练例子是采用这种方式的detectron2库也是采用这种方式启动https://github.com/facebookresearch/detectron2/blob/main/detectron2/engine/launch.py。
如果使用torch.multiprocessing.spawn启动要注意送入的训练function必须是fn(i,*args) 这种格式其中第一个参数 i 指代的是当前节点的进程编号这个参数其实就充当了local_rank, 所谓的local_rank是指的训练进程在当前节点的序号前面说的 rank 其实是全局的进程序号这个参数很重要因为要根据这个参数来设置每个进程所使用的 device 设备一般情况下直接认为local_rank即为所采用的GPU编号设置如下
torch.cuda.set_device(args.local_rank) # before your code runs# set DDP
model torch.nn.parallel.DistributedDataParallel(model, device_ids[local_rank], output_devicelocal_rank)# 或者
with torch.cuda.device(args.local_rank):# your code to run除了采用mp.spawn还可以采用torch.distributed.launch来启动程序Distributed communication package - torch.distributed这个是更常用的启动方式。比如对于单机多卡训练其启动方式如下
python -m torch.distributed.launch --nproc_per_nodeNUM_GPUS_YOU_HAVEYOUR_TRAINING_SCRIPT.py (--arg1 --arg2 --arg3 and all other arguments of your training script)其中NUM_GPUS_YOU_HAVE是GPU的总量而YOUR_TRAINING_SCRIPT.py是训练的脚本其和上述基本一致不过区别是采用torch.distributed.launch启动会自动设置一些环境变量https://github.com/pytorch/pytorch/blob/master/torch/distributed/run.py#L211比如我们需要的RANK和WORLD_SIZE 就直接可以从环境变量中获取
rank int(os.environ[RANK])
world_size int(os.environ[WORLD_SIZE])对于local_rank的获取有两种方式 1一种是在训练脚本添加一个命令行参数程序启动时会对其自动赋值
import argparse
parser argparse.ArgumentParser()
parser.add_argument(--local_rank, typeint)
args parser.parse_args()local_rank args.local_rank2另外一种方式采用torch.distributed.launch启动加上--use_envTrue此时情况下会设置LOCAL_RANK这个环境变量就可以从环境变量中获取local_rank python -m torch.distributed.launch --nproc_per_nodeNUM_GPUS_YOU_HAVE --use_envTrueYOUR_TRAINING_SCRIPT.py (--arg1 --arg2 --arg3 and all other arguments of your training script)import os
local_rank int(os.environ[LOCAL_RANK]) 对于多机多卡训练比如2个node其启动命令如下所示
# Node 1: (IP: 192.168.1.1, and has a free port: 1234)
python -m torch.distributed.launch --nproc_per_nodeNUM_GPUS_YOU_HAVE --nnodes2 --node_rank0 --master_addr192.168.1.1--master_port1234 YOUR_TRAINING_SCRIPT.py (--arg1 --arg2 --arg3 and all other arguments of your training script)# Node 2
python -m torch.distributed.launch --nproc_per_nodeNUM_GPUS_YOU_HAVE --nnodes2 --node_rank1 --master_addr192.168.1.1--master_port1234 YOUR_TRAINING_SCRIPT.py (--arg1 --arg2 --arg3 and all other arguments of your training script)这里
--nnodes表示传入node数目--node_rank表示传入node的编号world_sizennodes*nproc_per_node。
不过最新版本的PyTorch推出了torchrun来替代torch.distributed.launch。 torchrun和torch.distributed.launch的用法基本一致不过弃用了--use_env命令直接将local_rank设置在环境变量中目前最新版本的torchvision是采用torchrun启动方式具体见 vision/references/classification at main · pytorch/vision。
混合精度训练采用apex
安装Apex
git clone https://github.com/NVIDIA/apex
cd apex
# if pip 23.1 (ref: https://pip.pypa.io/en/stable/news/#v23-1) which supports multiple --config-settings with the same key...
pip install -v --disable-pip-version-check --no-cache-dir --no-build-isolation --config-settings --build-option--cpp_ext --config-settings --build-option--cuda_ext ./
# otherwise
pip install -v --disable-pip-version-check --no-cache-dir --no-build-isolation --global-option--cpp_ext --global-option--cuda_ext ./Apex官方文档Apex (A PyTorch Extension)
混合精度训练浮点FP32和半精度FP16相结合的训练使我们能够使用更大的batch_size并利用NVIDIA Tensor Cores实现更快的计算。AWS p3实例使用带有Tensor内核的NVIDIA Tesla V100 GPU。采用NVIDIA的apex进行混合精度训练非常简单只需要修改部分代码
def train(gpu, args):############################################################rank args.nr * args.gpus gpudist.init_process_group(backendnccl, init_methodenv://, world_sizeargs.world_size, rankrank)############################################################torch.manual_seed(0)model ConvNet()torch.cuda.set_device(gpu)model.cuda(gpu)batch_size 100# define loss function (criterion) and optimizercriterion nn.CrossEntropyLoss().cuda(gpu)optimizer torch.optim.SGD(model.parameters(), 1e-4)############################################################# Wrap the modelmodel, optimizer amp.initialize(model, optimizer, opt_levelO2)model DDP(model)############################################################# Data loading codetrain_dataset torchvision.datasets.MNIST(root./data, trainTrue, transformtransforms.ToTensor(),downloadTrue)############################################################# train_sampler torch.utils.data.distributed.DistributedSampler(datasettrain_dataset, num_replicasargs.world_size,# rankrank)# train_loader torch.utils.data.DataLoader(datasettrain_dataset, batch_sizebatch_size, shuffleFalse,# num_workers0, pin_memoryTrue, samplertrain_sampler)train_loader torch.utils.data.DataLoader(datasettrain_dataset, batch_sizebatch_size, shuffleTrue,num_workers0, pin_memoryTrue)############################################################start datetime.now()total_step len(train_loader)for epoch in range(args.epochs):for i, (images, labels) in enumerate(train_loader):images images.cuda(non_blockingTrue)labels labels.cuda(non_blockingTrue)# Forward passoutputs model(images)loss criterion(outputs, labels)# Backward and optimizeoptimizer.zero_grad()############################################################with amp.scale_loss(loss, optimizer) as scaled_loss:scaled_loss.backward()############################################################optimizer.step()if (i 1) % 100 0 and gpu 0:print(Epoch [{}/{}], Step [{}/{}], Loss: {:.4f}.format(epoch 1, args.epochs, i 1, total_step,loss.item()))if gpu 0:print(Training complete in: str(datetime.now() - start))其实就两处变化:
首先是采用amp.initialize来包装model和optimizer以支持混合精度训练其中opt_level指的是优化级别如果为O0使用all floats或者O3使用half-precision throughout不是真正的混合精度但是可以用来确定模型效果和速度的baseline而O1和O2是混合精度的两种设置可以选择某个进行混合精度训练其详细信息可以在Apex文档中找到。另外一处是在进行根据梯度更新参数前要先通过amp.scale_loss对梯度进行scale以防止梯度下溢underflowing。此外你还可以用apex.parallel.DistributedDataParallel替换nn.DistributedDataParallel。
是的所有这些代码中的第一个字符是大写字母“O”而第二个字符是数字。是的如果用零来代替会得到一条令人费解的错误消息。
apex.parallel.distributedDataParallel是nn.distributedDataParallear的一个替换。不再需要指定GPU因为Apex每个process只允许一个GPU。它还假设脚本在将模型移动到GPU之前调用torch.cuda.set_device(local_rank)。
混合精度训练要求对损失进行缩放以防止梯度下溢。Apex会自动做到这一点。
此脚本的运行方式与分布式训练脚本相同。
python without_multiprocessing.py -n 1 -g 4 -nr 0另外新版本的PyTorch已经内置混合精度训练具体见AUTOMATIC MIXED PRECISION PACKAGE - TORCH.AMP添加链接描述。而且PyTorch官方的分布式实现现在已经比较完善而且性能和效果都不错可以替代的方案是horovod不仅支持PyTorch还支持TensorFlow和MXNet框架实现起来也是比较容易的速度方面也不错。
参考资料
Distributed data parallel training in Pytorch在PyTorch中使用DistributedDataParallel进行多GPU分布式模型训练PyTorch分布式训练简明教程(2022更新版)分布式训练框架Horovod
