学习目标
- 了解模型并行与数据并行的区别.
- 了解分布式训练与并行训练的关系.
- 掌握在单机多GPU上进行模型并行训练的解决方案.
相关知识
- 并行/分布训练及其两者的关系：
* 在机器学习领域（深度学习），并行/分布方式一般主要应用在模型的训练阶段以加速模型的训练效率。因此，利用计算机系统的多线程或多进程来提升
模型训练效率的方式都可以称作并行训练。其中，利用多进程训练的方式又可以叫做并行分布式训练，简称分布式训练（因为单台计算机多进程间的通信等同于多>台计算机间的通信）。由此可见，分布式训练是并行训练的一种特殊形式。
- 数据并行训练：
* 数据并行是一般指训练数据的每个批次数据被分割成n等份，分别送给同一模型，此时模型被复制了n份以接受不同数据，之后每个模型都会计算对应数>据的梯度，然后所有的梯度求均值用以更新每个模型的参数，进而进行下个批次数据的并行（因为我们常用的批次SGD优化方法，就是求解该批次数据的平均梯度来
更新参数）。
- 模型并行训练:
* 模型并行是指模型网络结构被分割成n个部分，每一部分都会在处理完一条数据后立即处理下一条（如果模型不被分割成独立的各个部分，模型中的每一
部分必须等待该条数据全部处理后，才能开始下一条数据处理）。
-
本案例着重讲解单机多GPU的模型并行方案，解决大型模型无法在单GPU上整体加载的问题。
“6.md” 486L, 18286C 33,1 Top -
本案例着重讲解单机多GPU的模型并行方案，解决大型模型无法在单GPU上整体加载的问题。
单机多GPU的模型并行
- 第一步: 查看硬件配置并以一个简单示例理解模型分配
- 第二步: 将大型模型ResNet50结构分配到多个GPU上
- 第三步: 对比模型多GPU并行和单GPU的耗时
- 第四步: 使用流水线技术加速多GPU训练
- 第五步: 寻找流水线参数以进一步加速多GPU训练
第一步: 查看硬件配置并以一个简单示例理解模型分配
- 查看硬件配置
import subprocess
# 打印nvidia显卡信息，包括cuda版本，显卡数量，当前使用情况等等
print(subprocess.check_output("nvidia-smi", universal_newlines=True))
- 输出效果:
# 这里我们可以看到:
# GPU Driver和CUDA的版本信息
# 两台GTX1080Ti的GPU运行情况
+-----------------------------------------------------------------------------+
| NVIDIA-SMI 430.50 Driver Version: 430.50 CUDA Version: 10.1 |
32,0-1 6%
+-----------------------------------------------------------------------------+
| NVIDIA-SMI 430.50 Driver Version: 430.50 CUDA Version: 10.1 |
|-------------------------------+----------------------+----------------------+
| GPU Name Persistence-M| Bus-Id Disp.A | Volatile Uncorr. ECC |
| Fan Temp Perf Pwr:Usage/Cap| Memory-Usage | GPU-Util Compute M. |
|===============================+======================+======================|
| 0 GeForce GTX 1080Ti Off | 00000000:03:00.0 Off | N/A |
| 20% 38C P0 54W / 250W | 0MiB / 11178MiB | 0% Default |
+-------------------------------+----------------------+----------------------+
| 1 GeForce GTX 1080Ti Off | 00000000:04:00.0 Off | N/A |
| 26% 45C P0 53W / 250W | 0MiB / 11178MiB | 3% Default |
+-------------------------------+----------------------+----------------------+
+-----------------------------------------------------------------------------+
| Processes: GPU Memory |
| GPU PID Type Process name Usage |
|=============================================================================|
| No running processes found |
+-----------------------------------------------------------------------------+
- 定义只有两个线性层的玩具模型:
# 导入构建模型的必备工具包
import torch
import torch.nn as nn
import torch.optim as optim
class ToyModel(nn.Module):
"""定义一个玩具模型类"""
def __init__(self):
super(ToyModel, self).__init__()
# 实例化第一个线性层(参数)，放在'0'号GPU上
self.net1 = torch.nn.Linear(10, 10).to('cuda:0')
68,1 14%
# 实例化第一个线性层(参数)，放在'0'号GPU上
self.net1 = torch.nn.Linear(10, 10).to('cuda:0')
# 实例化ReLU层，无参数计算层不需要任何分配
# 不占任何存储空间，只是一条计算指令
self.relu = torch.nn.ReLU()
# 实例化第二个线性层(参数)，放在'1'号GPU上
self.net2 = torch.nn.Linear(10, 5).to('cuda:1')
def forward(self, x):
# 输入x需要与第一个线性层参数相乘，因此需要发送到'0'号GPU上
# 接着在'0'号GPU上被ReLU函数激活
x = self.relu(self.net1(x.to('cuda:0')))
# 为了继续和第二个线性层参数相乘，因此需要发送到'1'号GPU上
# 最后在'1'号GPU上返回计算结果
return self.net2(x.to('cuda:1'))
- 定义玩具模型的训练配置:
# 实例化模型
model = ToyModel()
# 选择损失函数
loss_fn = nn.MSELoss()
# 选择优化方法
optimizer = optim.SGD(model.parameters(), lr=0.001)
# 梯度初始化为0
optimizer.zero_grad()
# 使用随机张量输入模型获得输出
outputs = model(torch.randn(20, 10))
# 因为模型的结果是在'1'号GPU上返回
# 所以也要将真实标签分配给'1'号GPU
labels = torch.randn(20, 5).to('cuda:1')
# 计算损失
104,9 22%
# 计算损失
loss_fn(outputs, labels).backward()
# 更新权重
optimizer.step()
第二步: 将大型模型ResNet50结构分配到多个GPU上
# 导入ResNet的主结构，和ResNet50的组成单元Bottleneck
from torchvision.models.resnet import ResNet, Bottleneck
# 原生ResNet50输出类别为1000
num_classes = 1000
class ModelParallelResNet50(ResNet):
"""在两台GPU上分配的并行ResNet50模型"""
def __init__(self, *args, **kwargs):
# 从ResNet主结构中初始化特定参数使其成为ResNet50
# 第一个初始化参数Bottleneck是ResNet50的特定块单元
# 第二个初始化参数[3, 4, 6, 3]是指ResNet50四个块单元对应的层数
# [3, 4, 6, 3]对于ResNet50是固定的，如果ResNet101，则对应[3, 4, 23, 3]
super(ModelParallelResNet50, self).__init__(
Bottleneck, [3, 4, 6, 3], num_classes=num_classes, *args, **kwargs)
# 重写ResNet50结构，使其分配在两台GPU上
# 内部的计算层和顺序都是固定的
# 前两个块单元(layer1, layer2)在'0'号GPU上
140,0-1 31%
# 内部的计算层和顺序都是固定的
# 前两个块单元(layer1, layer2)在'0'号GPU上
self.seq1 = nn.Sequential(
self.conv1,
self.bn1,
self.relu,
self.maxpool,
self.layer1,
self.layer2
).to('cuda:0')
# 后两个块单元(layer3, layer4)在'1'号GPU上
self.seq2 = nn.Sequential(
self.layer3,
self.layer4,
self.avgpool,
).to('cuda:1')
self.fc.to('cuda:1')
def forward(self, x):
# seq1处理后，将结果发送到'1'号GPU上
x = self.seq2(self.seq1(x).to('cuda:1'))
return self.fc(x.view(x.size(0), -1))
- 定义ResNet50模型训练配置:
# 定义模型训练的相关配置
num_batches = 3
batch_size = 120
image_w = 128
image_h = 128
176,9 39%
image_h = 128
def train(model):
"""模型训练函数"""
model.train(True)
# 定义损失函数
loss_fn = nn.MSELoss()
# 定义优化方法
optimizer = optim.SGD(model.parameters(), lr=0.001)
# 生成一个[batch, 1]形状的张量，里面的每个值都是[0, 1000)值域内的随机数
# 这个张量将用于之后生成真实标签
one_hot_indices = torch.LongTensor(batch_size) \
.random_(0, num_classes) \
.view(batch_size, 1)
# 开始batch循环
for _ in range(num_batches):
# 随机初始化指定尺寸的输入
inputs = torch.randn(batch_size, 3, image_w, image_h)
# 初始化一个[batch_size, num_classes]大小的零张量
# 使用scatter_方法向这个张量中填充数值
# 第一个参数为1，代表每次按照纵轴方向填充
# 第二个参数为one_hot_indices，代表每一列填充的位置索引
# 第三个参数为1，填充的值为1
labels = torch.zeros(batch_size, num_classes) \
.scatter_(1, one_hot_indices, 1)
# 梯度归零
optimizer.zero_grad()
# 首先还是将输入发送到'0'号GPU上
# 再调用模型得到输出
outputs = model(inputs.to('cuda:0'))
# 为了计算损失，需要把真实标签发送到输出结果的设备上
labels = labels.to(outputs.device)
# 在指定设备上计算损失
loss_fn(outputs, labels).backward()
212,1 47%
# 在指定设备上计算损失
loss_fn(outputs, labels).backward()
# 根据梯度更新参数
optimizer.step()
第三步: 对比模型多GPU并行和单GPU的耗时
- 绘制模型双GPU并行和单GPU的耗时图
# 导入matplotlib用于绘图
import matplotlib.pyplot as plt
# 设置绘图风格
plt.switch_backend('Agg')
import numpy as np
# 导入timeit，这是专门用于并行计算统计模型耗时的工具包
import timeit
# 设定timeit的重复参数，为了凸显训练的时间的差异，将重复10次
num_repeat = 10
# 设定timeit的目标函数(将计算该函数的耗时)
stmt = "train(model)"
# 设定timeit的启动语句，即计算耗时开始前运行的语句
# 启动语句为实例化并行的ResNet50模型
setup = "model = ModelParallelResNet50()"
# 连续计算10次并行的ResNet50模型的耗时
# stmt为执行的目标函数字符串形式
# setup为执行前的启动语句
# number为目标函数执行的次数，number=1表示目标函数只执行一次就计算耗时
248,9 55%
# setup为执行前的启动语句
# number为目标函数执行的次数，number=1表示目标函数只执行一次就计算耗时
# repeat为计算耗时的次数，number=1，repeat=10表示目标函数执行一次并计算该次耗时；
# 反复进行10次，得到10个结果
# globals=globals()表示代码能在当前的全局名称空间中执行，使用所有变量
mp_run_times = timeit.repeat(
stmt, setup, number=1, repeat=num_repeat, globals=globals())
# 计算10次结果的平均值和标准差
mp_mean, mp_std = np.mean(mp_run_times), np.std(mp_run_times)
# 启动语句为实例化单GPU的ResNet50模型
setup = "import torchvision.models as models;" + \
"model = models.resnet50(num_classes=num_classes).to('cuda:0')"
# 计算单GPU的ResNet50模型耗时
rn_run_times = timeit.repeat(
stmt, setup, number=1, repeat=num_repeat, globals=globals())
# 计算10次结果的平均值和标准差
rn_mean, rn_std = np.mean(rn_run_times), np.std(rn_run_times)
def plot(means, stds, labels, fig_name):
"""绘图函数"""
# 创建子图画布
fig, ax = plt.subplots()
# 在画布上绘制柱状图, 设置相关配置
ax.bar(np.arange(len(means)), means, yerr=stds,
align='center', alpha=0.5, ecolor='red', capsize=10, width=0.6)
# 设置纵轴说明
ax.set_ylabel('ResNet50 Execution Time (Second)')
# 设置横轴刻度
ax.set_xticks(np.arange(len(means)))
# 设置横轴刻度标签
ax.set_xticklabels(labels)
# 设置y轴网格线
ax.yaxis.grid(True)
# 设置布局
284,1 63%
* 模型训练的流水线技术:
* 流水线技术旨在使分布在不同GPU上的模型能够在同一时间都在处理对应工作，以此提升训练效率。流水线技术的原理是通过将数据划分为N份(N>1)，每
份数据称作一个数据堆。当第一个GPU处理完第一个数据堆后，将数据发送给第二个GPU，之后第一个GPU不会像之前一样等待第二个GPU处理完成，而是马上处理第>二个数据堆，此时间点上，两个GPU都在运行处理对应的工作，直到将所有数据堆处理完成。
* 以上是标准的流水线过程，必须开启与GPU等数量的线程来控制这些异步行为。而在实际工程中，为了避免代码的复杂度过高，往往不去使用异步的处理
机制，这是因为当我们把批次数据切分为足够小的数据堆时，单个GPU处理它们的速度已经非常快，其他GPU的等待时间可以忽略。也就是说，第二个GPU在处理第一
个数据堆时，不需要使用其他线程让第一个GPU异步处理数据，而只是等待其完成后，再继续处理第二个数据堆。接下来，我们将按照这种方式实现流水线并对比效
果。
---
* 使用流水线技术加速多GPU训练的实现:
```python
class PipelineParallelResNet50(ModelParallelResNet50):
"""带有流水线技术的并行模型ResNet50"""
def __init__(self, split_size=20, *args, **kwargs):
# 继承ModelParallelResNet50的初始化函数
# 加入了新的初始化参数split_size，代表每个批次数据划分的大小
# 如: batch_size=120, split_size=20说明将120条数据划分成6份，
# 每份20条作为流水线处理的条数
super(PipelineParallelResNet50, self).__init__(*args, **kwargs)
self.split_size = split_size
def forward(self, x):
"""重写流水线的forward函数"""
# 将输入的批次数据按照split_size划分，并使用迭代器封装
splits = iter(x.split(self.split_size, dim=0))
# 使用next方法取出迭代器中的第一份数据(第一个数据堆)
s_next = next(splits)
# 将数据在'0'号GPU上处理后发送给'1'号GPU
s_prev = self.seq1(s_next).to('cuda:1')
# 创建一个存储最终处理结果的列表
ret = []
# 循环遍历迭代器中的所有数据堆
353,1 77%
# 循环遍历迭代器中的所有数据堆
for s_next in splits:
# 在'1'号GPU上处理'0'号GPU上发来的数据
s_prev = self.seq2(s_prev)
# 将结果view成指定维度输入到全连接层
# 最后装进结果列表
ret.append(self.fc(s_prev.view(s_prev.size(0), -1)))
# 继续将数据在'0'号GPU上处理后发送给'1'号GPU
s_prev = self.seq1(s_next).to('cuda:1')
# 当最后一个数据堆循环遍历完成后，只是发送给'1'号GPU并没有处理
# 所以这里要在'1'号GPU上处理完成
s_prev = self.seq2(s_prev)
# 同样将结果view成指定维度输入到全连接层
# 最后装进结果列表
ret.append(self.fc(s_prev.view(s_prev.size(0), -1)))
# 返回结果的张量形式
return torch.cat(ret)
# 启动语句为实例化带有流水线的多GPU并行ResNet50模型
setup = "model = PipelineParallelResNet50()"
# 使用timeit进行耗时计算，参数与上述使用时相同
pp_run_times = timeit.repeat(
stmt, setup, number=1, repeat=num_repeat, globals=globals())
# 计算均值和标准差
pp_mean, pp_std = np.mean(pp_run_times), np.std(pp_run_times)
# 绘制耗时对比图
plot([mp_mean, rn_mean, pp_mean],
[mp_std, rn_std, pp_std],
['Model Parallel', 'Single GPU', 'Pipelining Model Parallel'],
'mp_vs_rn_vs_pp.png')
384,0-1 85%
> * 输出效果:
> * 分析:
* 从图中可知，带有流水线技术的模型训练耗时(运行时间)最短，相对比单GPU运行已经有了明显改善。但是我们发现，流水线技术引进了一个新的参数split_size，它代表数据堆的大小，也直接影响了模型训练的耗时，我们可以使用两个极端的例子来解释这种影响，当split_size与batch_size大小相同时，即等效>是没有使用流水线的情况，耗时大于单GPU。而当split_size=1时，计算时间和等待时间虽然足够小，但是GPU之间的数据传输时间将被放大，导致训练耗时变长，>下面我们将从实验中寻找最佳的split_size。
---
* 第五步: 寻找流水线参数以进一步加速多GPU训练
```python
# 创建存储均值和标准差的列表
means = []
stds = []
# 设置一组split_size的采样点
split_sizes = [1, 3, 5, 8, 10, 12, 20, 40, 60]
# 遍历采样点
for split_size in split_sizes:
# 启动语句为实例化带有流水线的多GPU并行ResNet50模型
setup = "model = PipelineParallelResNet50(split_size=%d)" % split_size
# 使用timeit计算各个采样点的耗时
pp_run_times = timeit.repeat(
stmt, setup, number=1, repeat=num_repeat, globals=globals())
# 保存均值和标准差
means.append(np.mean(pp_run_times))
stds.append(np.std(pp_run_times))
420,1 92%
drwxr-xr-x 5 root root 4096 Sep 5 13:47 mkdocs_solutions
drwxr-xr-x 3 root root 4096 Feb 25 2021 mkdocs_student
drwxr-xr-x 4 root root 4096 Oct 25 10:50 mkdocs_SZ
drwxr-xr-x 8 root root 4096 Jan 13 2020 mkdocs_torch
drwxr-xr-x 5 root root 4096 Feb 26 2021 mkdocs_transformer
drwxr-xr-x 4 root root 4096 Feb 26 2021 mkdocs_translearning
drwxr-xr-x 3 root root 4096 Jul 21 2020 NLP
-rw------- 1 root root 99 Feb 24 2021 nohup.out
-rw-r--r-- 1 root root 11109 Jan 10 2021 pylintrc
-rw-r--r-- 1 root root 1008 Jan 10 2021 weixin.py
[root@iZ8vbdyg5rlsodjyzp1fdyZ zhoumingzhen]# cd mkdocs_EaseLibrary/
[root@iZ8vbdyg5rlsodjyzp1fdyZ mkdocs_EaseLibrary]# ll
total 16596
drwxr-xr-x 4 root root 4096 May 3 2021 docs
-rw-r--r-- 1 root root 1726 Aug 29 2020 mkdocs.yml
-rw------- 1 root root 16976148 Jul 11 17:00 nohup.out
drwxr-xr-x 11 root root 4096 Jun 3 2020 site
-rw-r--r-- 1 root root 38 May 12 2020 start.sh
[root@iZ8vbdyg5rlsodjyzp1fdyZ mkdocs_EaseLibrary]# cd docs/
[root@iZ8vbdyg5rlsodjyzp1fdyZ docs]# ll
total 180
-rw-r--r-- 1 root root 35344 May 3 2021 1.md
-rw-r--r-- 1 root root 13466 Mar 7 2020 2.md
-rw-r--r-- 1 root root 25158 Mar 7 2020 3.md
-rw-r--r-- 1 root root 32912 Sep 2 2020 4.md
-rw-r--r-- 1 root root 32407 Sep 2 2020 5.md
-rw-r--r-- 1 root root 18286 Jul 4 2020 6.md
-rw-r--r-- 1 root root 3232 Aug 29 2020 7.md
drwxr-xr-x 2 root root 4096 Mar 23 2020 img
-rw-r--r-- 1 root root 41 May 12 2020 README.md
[root@iZ8vbdyg5rlsodjyzp1fdyZ docs]# vim 1.md
[root@iZ8vbdyg5rlsodjyzp1fdyZ docs]# vim 2.md
[root@iZ8vbdyg5rlsodjyzp1fdyZ docs]# vim 3.md
[root@iZ8vbdyg5rlsodjyzp1fdyZ docs]# vim 4.md
[root@iZ8vbdyg5rlsodjyzp1fdyZ docs]# vim 5.md
[root@iZ8vbdyg5rlsodjyzp1fdyZ docs]# vim 6.md
[root@iZ8vbdyg5rlsodjyzp1fdyZ docs]# vim 7.md
[root@iZ8vbdyg5rlsodjyzp1fdyZ docs]# vim 6.md
[root@iZ8vbdyg5rlsodjyzp1fdyZ docs]# cat 6.md
<!-- Single-Machine Model Parallel Best Practices -->
### 学习目标
* 了解模型并行与数据并行的区别.
* 了解分布式训练与并行训练的关系.
* 掌握在单机多GPU上进行模型并行训练的解决方案.
---
<center></center>
---
### 相关知识
* 并行/分布训练及其两者的关系：
* 在机器学习领域（深度学习），并行/分布方式一般主要应用在模型的训练阶段以加速模型的训练效率。因此，利用计算机系统的多线程或多进程来提升模型训练效率的方式都可以称作并行训练。其中，利用多进程训练的方式又可以叫做并行分布式训练，简称分布式训练（因为单台计算机多进程间的通信等同于多台计算机间的通信）。由此可见，分布式训练是并行训练的一种特殊形式。
---
* 数据并行训练：
* 数据并行是一般指训练数据的每个批次数据被分割成n等份，分别送给同一模型，此时模型被复制了n份以接受不同数据，之后每个模型都会计算对应数据的梯度，然后所有的梯度求均值用以更新每个模型的参数，进而进行下个批次数据的并行（因为我们常用的批次SGD优化方法，就是求解该批次数据的平均梯度来更新参数）。
---
* 模型并行训练:
* 模型并行是指模型网络结构被分割成n个部分，每一部分都会在处理完一条数据后立即处理下一条（如果模型不被分割成独立的各个部分，模型中的每一部分必须等待该条数据全部处理后，才能开始下一条数据处理）。
---
* 本案例着重讲解单机多GPU的模型并行方案，解决大型模型无法在单GPU上整体加载的问题。
---
### 单机多GPU的模型并行
* 第一步: 查看硬件配置并以一个简单示例理解模型分配
* 第二步: 将大型模型ResNet50结构分配到多个GPU上
* 第三步: 对比模型多GPU并行和单GPU的耗时
* 第四步: 使用流水线技术加速多GPU训练
* 第五步: 寻找流水线参数以进一步加速多GPU训练
---
#### 第一步: 查看硬件配置并以一个简单示例理解模型分配
* 查看硬件配置
```python
import subprocess
# 打印nvidia显卡信息，包括cuda版本，显卡数量，当前使用情况等等
print(subprocess.check_output("nvidia-smi", universal_newlines=True))
- 输出效果:
# 这里我们可以看到:
# GPU Driver和CUDA的版本信息
# 两台GTX1080Ti的GPU运行情况
+-----------------------------------------------------------------------------+
| NVIDIA-SMI 430.50 Driver Version: 430.50 CUDA Version: 10.1 |
|-------------------------------+----------------------+----------------------+
| GPU Name Persistence-M| Bus-Id Disp.A | Volatile Uncorr. ECC |
| Fan Temp Perf Pwr:Usage/Cap| Memory-Usage | GPU-Util Compute M. |
|===============================+======================+======================|
| 0 GeForce GTX 1080Ti Off | 00000000:03:00.0 Off | N/A |
| 20% 38C P0 54W / 250W | 0MiB / 11178MiB | 0% Default |
+-------------------------------+----------------------+----------------------+
| 1 GeForce GTX 1080Ti Off | 00000000:04:00.0 Off | N/A |
| 26% 45C P0 53W / 250W | 0MiB / 11178MiB | 3% Default |
+-------------------------------+----------------------+----------------------+
+-----------------------------------------------------------------------------+
| Processes: GPU Memory |
| GPU PID Type Process name Usage |
|=============================================================================|
| No running processes found |
+-----------------------------------------------------------------------------+
- 定义只有两个线性层的玩具模型:
# 导入构建模型的必备工具包
import torch
import torch.nn as nn
import torch.optim as optim
class ToyModel(nn.Module):
"""定义一个玩具模型类"""
def __init__(self):
super(ToyModel, self).__init__()
# 实例化第一个线性层(参数)，放在'0'号GPU上
self.net1 = torch.nn.Linear(10, 10).to('cuda:0')
# 实例化ReLU层，无参数计算层不需要任何分配
# 不占任何存储空间，只是一条计算指令
self.relu = torch.nn.ReLU()
# 实例化第二个线性层(参数)，放在'1'号GPU上
self.net2 = torch.nn.Linear(10, 5).to('cuda:1')
def forward(self, x):
# 输入x需要与第一个线性层参数相乘，因此需要发送到'0'号GPU上
# 接着在'0'号GPU上被ReLU函数激活
x = self.relu(self.net1(x.to('cuda:0')))
# 为了继续和第二个线性层参数相乘，因此需要发送到'1'号GPU上
# 最后在'1'号GPU上返回计算结果
return self.net2(x.to('cuda:1'))
- 定义玩具模型的训练配置:
# 实例化模型
model = ToyModel()
# 选择损失函数
loss_fn = nn.MSELoss()
# 选择优化方法
optimizer = optim.SGD(model.parameters(), lr=0.001)
# 梯度初始化为0
optimizer.zero_grad()
# 使用随机张量输入模型获得输出
outputs = model(torch.randn(20, 10))
# 因为模型的结果是在'1'号GPU上返回
# 所以也要将真实标签分配给'1'号GPU
labels = torch.randn(20, 5).to('cuda:1')
# 计算损失
loss_fn(outputs, labels).backward()
# 更新权重
optimizer.step()
第二步: 将大型模型ResNet50结构分配到多个GPU上
# 导入ResNet的主结构，和ResNet50的组成单元Bottleneck
from torchvision.models.resnet import ResNet, Bottleneck
# 原生ResNet50输出类别为1000
num_classes = 1000
class ModelParallelResNet50(ResNet):
"""在两台GPU上分配的并行ResNet50模型"""
def __init__(self, *args, **kwargs):
# 从ResNet主结构中初始化特定参数使其成为ResNet50
# 第一个初始化参数Bottleneck是ResNet50的特定块单元
# 第二个初始化参数[3, 4, 6, 3]是指ResNet50四个块单元对应的层数
# [3, 4, 6, 3]对于ResNet50是固定的，如果ResNet101，则对应[3, 4, 23, 3]
super(ModelParallelResNet50, self).__init__(
Bottleneck, [3, 4, 6, 3], num_classes=num_classes, *args, **kwargs)
# 重写ResNet50结构，使其分配在两台GPU上
# 内部的计算层和顺序都是固定的
# 前两个块单元(layer1, layer2)在'0'号GPU上
self.seq1 = nn.Sequential(
self.conv1,
self.bn1,
self.relu,
self.maxpool,
self.layer1,
self.layer2
).to('cuda:0')
# 后两个块单元(layer3, layer4)在'1'号GPU上
self.seq2 = nn.Sequential(
self.layer3,
self.layer4,
self.avgpool,
).to('cuda:1')
self.fc.to('cuda:1')
def forward(self, x):
# seq1处理后，将结果发送到'1'号GPU上
x = self.seq2(self.seq1(x).to('cuda:1'))
return self.fc(x.view(x.size(0), -1))
- 定义ResNet50模型训练配置:
# 定义模型训练的相关配置
num_batches = 3
batch_size = 120
image_w = 128
image_h = 128
def train(model):
"""模型训练函数"""
model.train(True)
# 定义损失函数
loss_fn = nn.MSELoss()
# 定义优化方法
optimizer = optim.SGD(model.parameters(), lr=0.001)
# 生成一个[batch, 1]形状的张量，里面的每个值都是[0, 1000)值域内的随机数
# 这个张量将用于之后生成真实标签
one_hot_indices = torch.LongTensor(batch_size) \
.random_(0, num_classes) \
.view(batch_size, 1)
# 开始batch循环
for _ in range(num_batches):
# 随机初始化指定尺寸的输入
inputs = torch.randn(batch_size, 3, image_w, image_h)
# 初始化一个[batch_size, num_classes]大小的零张量
# 使用scatter_方法向这个张量中填充数值
# 第一个参数为1，代表每次按照纵轴方向填充
# 第二个参数为one_hot_indices，代表每一列填充的位置索引
# 第三个参数为1，填充的值为1
labels = torch.zeros(batch_size, num_classes) \
.scatter_(1, one_hot_indices, 1)
# 梯度归零
optimizer.zero_grad()
# 首先还是将输入发送到'0'号GPU上
# 再调用模型得到输出
outputs = model(inputs.to('cuda:0'))
# 为了计算损失，需要把真实标签发送到输出结果的设备上
labels = labels.to(outputs.device)
# 在指定设备上计算损失
loss_fn(outputs, labels).backward()
# 根据梯度更新参数
optimizer.step()
第三步: 对比模型多GPU并行和单GPU的耗时
- 绘制模型双GPU并行和单GPU的耗时图
# 导入matplotlib用于绘图
import matplotlib.pyplot as plt
# 设置绘图风格
plt.switch_backend('Agg')
import numpy as np
# 导入timeit，这是专门用于并行计算统计模型耗时的工具包
import timeit
# 设定timeit的重复参数，为了凸显训练的时间的差异，将重复10次
num_repeat = 10
# 设定timeit的目标函数(将计算该函数的耗时)
stmt = "train(model)"
# 设定timeit的启动语句，即计算耗时开始前运行的语句
# 启动语句为实例化并行的ResNet50模型
setup = "model = ModelParallelResNet50()"
# 连续计算10次并行的ResNet50模型的耗时
# stmt为执行的目标函数字符串形式
# setup为执行前的启动语句
# number为目标函数执行的次数，number=1表示目标函数只执行一次就计算耗时
# repeat为计算耗时的次数，number=1，repeat=10表示目标函数执行一次并计算该次耗时；
# 反复进行10次，得到10个结果
# globals=globals()表示代码能在当前的全局名称空间中执行，使用所有变量
mp_run_times = timeit.repeat(
stmt, setup, number=1, repeat=num_repeat, globals=globals())
# 计算10次结果的平均值和标准差
mp_mean, mp_std = np.mean(mp_run_times), np.std(mp_run_times)
# 启动语句为实例化单GPU的ResNet50模型
setup = "import torchvision.models as models;" + \
"model = models.resnet50(num_classes=num_classes).to('cuda:0')"
# 计算单GPU的ResNet50模型耗时
rn_run_times = timeit.repeat(
stmt, setup, number=1, repeat=num_repeat, globals=globals())
# 计算10次结果的平均值和标准差
rn_mean, rn_std = np.mean(rn_run_times), np.std(rn_run_times)
def plot(means, stds, labels, fig_name):
"""绘图函数"""
# 创建子图画布
fig, ax = plt.subplots()
# 在画布上绘制柱状图, 设置相关配置
ax.bar(np.arange(len(means)), means, yerr=stds,
align='center', alpha=0.5, ecolor='red', capsize=10, width=0.6)
# 设置纵轴说明
ax.set_ylabel('ResNet50 Execution Time (Second)')
# 设置横轴刻度
ax.set_xticks(np.arange(len(means)))
# 设置横轴刻度标签
ax.set_xticklabels(labels)
# 设置y轴网格线
ax.yaxis.grid(True)
# 设置布局
plt.tight_layout()
# 保存图片
plt.savefig(fig_name)
# 关闭图片
plt.close(fig)
# 向函数中传入对应参数
plot([mp_mean, rn_mean],
[mp_std, rn_std],
['Model Parallel', 'Single GPU'],
'mp_vs_rn.png')
- 输出效果:
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-5esaecnq-1639285726178)(./img/mp_vs_rn.png)]
- 分析:
* 由图可知，单GPU的运行时间小于模型分配在两台GPU上的运行时间，这是因为: 在当前状态下，两台GPU上的模型同一时间仅有一台GPU工作，并他们还要花费时间在相互的数据传输上。为了改善这种状况，我们将使用模型训练的流水线技术，下面将详细讲解。
第四步: 使用流水线技术加速多GPU训练
- 模型训练的流水线技术:
- 流水线技术旨在使分布在不同GPU上的模型能够在同一时间都在处理对应工作，以此提升训练效率。流水线技术的原理是通过将数据划分为N份(N>1)，每份数据称作一个数据堆。当第一个GPU处理完第一个数据堆后，将数据发送给第二个GPU，之后第一个GPU不会像之前一样等待第二个GPU处理完成，而是马上处理第二个数据堆，此时间点上，两个GPU都在运行处理对应的工作，直到将所有数据堆处理完成。
- 以上是标准的流水线过程，必须开启与GPU等数量的线程来控制这些异步行为。而在实际工程中，为了避免代码的复杂度过高，往往不去使用异步的处理机制，这是因为当我们把批次数据切分为足够小的数据堆时，单个GPU处理它们的速度已经非常快，其他GPU的等待时间可以忽略。也就是说，第二个GPU在处理第一个数据堆时，不需要使用其他线程让第一个GPU异步处理数据，而只是等待其完成后，再继续处理第二个数据堆。接下来，我们将按照这种方式实现流水线并对比效果。
- 使用流水线技术加速多GPU训练的实现:
class PipelineParallelResNet50(ModelParallelResNet50):
"""带有流水线技术的并行模型ResNet50"""
def __init__(self, split_size=20, *args, **kwargs):
# 继承ModelParallelResNet50的初始化函数
# 加入了新的初始化参数split_size，代表每个批次数据划分的大小
# 如: batch_size=120, split_size=20说明将120条数据划分成6份，
# 每份20条作为流水线处理的条数
super(PipelineParallelResNet50, self).__init__(*args, **kwargs)
self.split_size = split_size
def forward(self, x):
"""重写流水线的forward函数"""
# 将输入的批次数据按照split_size划分，并使用迭代器封装
splits = iter(x.split(self.split_size, dim=0))
# 使用next方法取出迭代器中的第一份数据(第一个数据堆)
s_next = next(splits)
# 将数据在'0'号GPU上处理后发送给'1'号GPU
s_prev = self.seq1(s_next).to('cuda:1')
# 创建一个存储最终处理结果的列表
ret = []
# 循环遍历迭代器中的所有数据堆
for s_next in splits:
# 在'1'号GPU上处理'0'号GPU上发来的数据
s_prev = self.seq2(s_prev)
# 将结果view成指定维度输入到全连接层
# 最后装进结果列表
ret.append(self.fc(s_prev.view(s_prev.size(0), -1)))
# 继续将数据在'0'号GPU上处理后发送给'1'号GPU
s_prev = self.seq1(s_next).to('cuda:1')
# 当最后一个数据堆循环遍历完成后，只是发送给'1'号GPU并没有处理
# 所以这里要在'1'号GPU上处理完成
s_prev = self.seq2(s_prev)
# 同样将结果view成指定维度输入到全连接层
# 最后装进结果列表
ret.append(self.fc(s_prev.view(s_prev.size(0), -1)))
# 返回结果的张量形式
return torch.cat(ret)
# 启动语句为实例化带有流水线的多GPU并行ResNet50模型
setup = "model = PipelineParallelResNet50()"
# 使用timeit进行耗时计算，参数与上述使用时相同
pp_run_times = timeit.repeat(
stmt, setup, number=1, repeat=num_repeat, globals=globals())
# 计算均值和标准差
pp_mean, pp_std = np.mean(pp_run_times), np.std(pp_run_times)
# 绘制耗时对比图
plot([mp_mean, rn_mean, pp_mean],
[mp_std, rn_std, pp_std],
['Model Parallel', 'Single GPU', 'Pipelining Model Parallel'],
'mp_vs_rn_vs_pp.png')
- 输出效果:
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-dnHlSkXk-1639285726180)(./img/mp_vs_rn_vs_pp.png)]
- 分析:
* 从图中可知，带有流水线技术的模型训练耗时(运行时间)最短，相对比单GPU运行已经有了明显改善。但是我们发现，流水线技术引进了一个新的参数split_size，它代表数据堆的大小，也直接影响了模型训练的耗时，我们可以使用两个极端的例子来解释这种影响，当split_size与batch_size大小相同时，即等效是没有使用流水线的情况，耗时大于单GPU。而当split_size=1时，计算时间和等待时间虽然足够小，但是GPU之间的数据传输时间将被放大，导致训练耗时变长，下面我们将从实验中寻找最佳的split_size。
- 第五步: 寻找流水线参数以进一步加速多GPU训练
# 创建存储均值和标准差的列表
means = []
stds = []
# 设置一组split_size的采样点
split_sizes = [1, 3, 5, 8, 10, 12, 20, 40, 60]
# 遍历采样点
for split_size in split_sizes:
# 启动语句为实例化带有流水线的多GPU并行ResNet50模型
setup = "model = PipelineParallelResNet50(split_size=%d)" % split_size
# 使用timeit计算各个采样点的耗时
pp_run_times = timeit.repeat(
stmt, setup, number=1, repeat=num_repeat, globals=globals())
# 保存均值和标准差
means.append(np.mean(pp_run_times))
stds.append(np.std(pp_run_times))
# 创建画布
fig, ax = plt.subplots()
# 绘制均值曲线
ax.plot(split_sizes, means)
# 绘制均值点的上下浮动范围(标准差)
ax.errorbar(split_sizes, means, yerr=stds, ecolor='red', fmt='ro')
# 设置横纵坐标名称
ax.set_ylabel('ResNet50 Execution Time (Second)')
ax.set_xlabel('Pipeline Split Size')
# 设置刻度
ax.set_xticks(split_sizes)
# 设置网格显示
ax.yaxis.grid(True)
# 设置布局
plt.tight_layout()
# 保存图片
plt.savefig("split_size_tradeoff.png")
# 关闭画布
plt.close(fig)
- 输出效果:
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-cLdxhMLe-1639285726182)(./img/split_size_tradeoff.png)]
- 分析:
* 从图中可以看出，最佳的split_size是12，此时耗时最短。如果继续减小split_size的值，硬件间的数据传输时间将显著增加。所以，在使用模型并行的流水线技术时，一般应该先通过采样点找到合适的split_size值作为参数，再进行模型并行训练。