horovod和pytorch多机多卡分布式并行训练代码配置和训练启动

pytorch 和 horovod 多机多卡并行训练总结

1 pytorch 中的多GPU训练

只需要安装pytorch GPU版本即可，使用其内部DistributedDataParallel 方法即可实现，方便简单。
从终端torchrun启动，初始化使用环境变量，并行实际上是给每个GPU启动一个进程
先看整体改动架构，只列出改动部分，适合单机多卡，多机多卡

# 1.导入库
# 分布式数据并行
from torch.utils.data.distributed import DistributedSampler
# 分布式并行
import torch.distributed as dist
from torch.nn.parallel import DistributedDataParallel as DDP
...

# 2.初始化
# 使用nccl初始化，windows不支持nccl，可以用gloo
'''
backend 是通讯方式，最好都用nccl，nccl作用可以百度一下，
init_method初始化方式，这里选使用环境变量初始化，其他初始化方式可以自己查询
world_size这里要给出总进程的数量，即用了几个GPU就会开几个进程，无论多机还是单机，如用了2个GPU则为2
rank进程唯一编号，如2个进程则每个进程都有唯一编号，一个为0，另一个为1，无论多机还是单机
'''
dist.init_process_group(backend='nccl', 
                        init_method='env://', 
                        world_size=2, 
                        rank=int(os.environ['RANK']))
# 3.分布式数据
'''
# DistributedSampler中，num_replicas和rank 参数如果省略则是以下形式，可以到源代码里面看一下
DistributedSampler(Data, num_replicas=dist.get_world_size(), rank=dist.get_rank())
dist.get_world_size() 总进程数跟上面说的world_size是一个意思
dist.get_rank()进程唯一编号, 跟上面的rank是一个意思
# 疑问：那为什么在初始化的时候不能用dist.get_world_size() 和 dist.get_rank()给world_size和rank
赋值，因为这两个方法必须初始化完成后才能使用，不然我系统怎么知道我用了几个进程。
Data_loader 中的shuffle不能用True, 使用默认值即可
'''
train_sampler = DistributedSampler(Data)
Data_loader = DataLoader(dataset=Data, sampler=train_sampler, batch_size=batch_size)

# 4.模型加载 , gpuid为每个locl_rank对应的gpuid
'''
local_rank 为当前机器（节点）使用GPU个数的唯一标识，如有两台机器、每台机器2个GPU并行，总共4个GPU，
就有4个进程，对于机器0，local_rank为0,1，对于机器1，local_rank还是0,1，0和1就代表每台机器要使用的
GPU编号。比如我0机器上有3个GPU，但是我想用gpuid为2和3的GPU，但是local_rank不会知道你使用哪个gpuid,
他只知道我要用两个GPU，标识为0和1，方法就是把local_rank 为0和1跟gpuid为2和3映射起来
'''
device = torch.device(f"cuda:{gpuid}")
...

# 5.分布式模型构建
'''
这里似乎没有什么可以讲的
'''
model = torch.nn.parallel.DistributedDataParallel(model, 
                                                  device_ids=[gpuid], 
                                                  output_device=gpuid,
                                                  find_unused_parameters=True)
optimizer = optim...

# 5.训练
for epoch ...
	...

终端启动：

# 1.单机多卡形态, 自动导入环境变量
'''
在终端执行后，自动传入环境变量
os.environ['RANK'] = nproc_per_node * nnodes  # RANK上面已经解释过
os.environ['LOCAL_RANK'] = nproc_per_node     # LOCAL_RANK 上面已经解释过
os.environ['MASTER_ADDR'] = ...               # 主节点ip, 单节点不用管，自动传入默认ip
os.environ['MASTER_PORT'] = ...               # 主节点端口, 单节点不用管，自动传入默认端口
nproc_per_node 每个节点（机器）启动的进程数
nnodes 使用多少个机器
node_rank 每台机器的唯一标识，跟rank是两个概念，一定注意
master_addr 主机器的ip地址
master_port 主机器的端口，随便给一个没有占用的
'''
# 1.单机多卡形态, 自动导入环境变量
torchrun --nproc_per_node=2 --nnodes=1 train.py  # 单节点不用传入master ip 和master port

# 2.多机多卡，如两个机器，每个机器上都要有环境、运行脚本、数据，因为每台机器都会执行脚本,注意node_rank的标识
# 机器1运行脚本
torchrun --nnodes=2 --nproc_per_node=1 --node_rank=0 --master_addr=xx.xx.xx.xx --master_port=12345 train.py
# 机器2运行脚本
torchrun --nnodes=2 --nproc_per_node=1 --node_rank=1 --master_addr=xx.xx.xx.xx --master_port=12345 train.py

这里强调一下几个比较重要的参数；我们拿两台机器，每台机器四个显卡来举例说明
–node_rank 这个是在运行torchrun时候指定，指多机的时候每个机器都有一个唯一标识，两台机器就是0,1,一台机器上指定0，另一台机器指定1
world_size 总进程数，也即使用的gpu数，这里也就是2*4=8
rank 指进程的唯一标识，这里有8个进程，也就是0,1,2,…7,8
local_rank 每个机器上GPU的标识，这里机器0上为0,1,2,3。机器1上还为0,1,2,3

2. horovod

horovod安装教程，查看官网流程。horovod只能用在linux系统上
https://horovod.readthedocs.io/en/stable/install_include.html

horovod 启动相比torchrun启动要简单一点，无论单机还是多机都只需要在master机器上启动即可

#1. 初始化horovod， 放在执行代码最前面
hvd.init()

#2. gpuid获取, 跟pytorch里一样
gpuid = hvd.locl_rank()

#3. 分布式数据并行
# 分布式数据集,默认是按照rank分的，多机之间是同步，快的机器会等慢的机器， 这里跟pytorch是一样的，因为本身用的就是torch中的api
train_sampler = DistributedSampler(DataA, num_replicas=hvd.size(), rank=hvd.rank())
DataA_loader = DataLoader(dataset=DataA, sampler=train_sampler, batch_size=batch_size, )

#4. 优化器, 这里和torch就不一样了，这里假设定义了三个优化器，并且三个优化器有部分权重参数是共享的
...
optimizer_1 = optim.Adam(...)
optimizer_2 = optim.Adam(...)
optimizer_3 = optim.Adam(...)

# horovod----
# 广播参数 把rank 0 to all process
hvd.broadcast_parameters(model.state_dict(), root_rank=0)
hvd.broadcast_optimizer_state(optimizer_1, root_rank=0)
hvd.broadcast_optimizer_state(optimizer_2, root_rank=0)
hvd.broadcast_optimizer_state(optimizer_3, root_rank=0)
#  horovod----

# horovod----
'''
Add Horovod Distributed Optimizer，这里要注意，由于我们定义的三个优化器，部分参数是共享的，
因此三个优化器中参数的名字有部分是一样的，而分布式优化器中要求每个优化器中name不能有重复，否则会报错，
多优化器出现的问题大多是因此导致的。多优化器问题可参考这个issues。如果优化器之间有参数name是一样的，
一定要手动去改一下参数的name,这不会导致什么异常，只要参数名称格式和优化器中的参数长度一致，而且必须一致就行
如下改动例子：
name_list1 = [('a'   name, param) for name, param in model.named_parameters() if re.match('encoder', name) or re.match('decoder_A', name)]
name_list2 = [('b'   name, param) for name, param in model.named_parameters() if re.match('encoder', name) or re.match('decoder_B', name)]
name_list3 = [(name, param) for name, param in model.named_parameters() if re.match('decoder_C', name)]
具体情况请跟自己模型架构去改，如这里优化器1和2有部分参数是共享的，因此将相同部分的name分别在前面   'a' 和 'b'，这样就可以了
https://github.com/horovod/horovod/issues/1417#issuecomment-588367250
'''
optimizer_1 = hvd.DistributedOptimizer(optimizer_1, named_parameters=name_list1)
optimizer_2 = hvd.DistributedOptimizer(optimizer_2, named_parameters=name_list2)
optimizer_3 = hvd.DistributedOptimizer(optimizer_3, named_parameters=name_list3)
# horovod----
'''
这for循环里horovod完成all-reduce部分是在loss.backword()完成梯度计算之后，optimizer.step()去广播参数，
因此每次step()之后必须要对其他优化器进行参数更新，因此多优化器场景必须要加同步synchronize(),而且是对其他所有优化器来同步，
所以对多优化器来说horovod更加耗时间
'''
for epoch in ...
	...
	model ...
	optimizer_1.zero_grad()
	loss1.backward()
	optimizer_1.step()
	optimizer_2.synchronize()
	optimizer_3.synchronize()
	
	model ...
	...
	optimizer_2.zero_grad()
	loss2.backward()
	optimizer_2.step()
	optimizer_1.synchronize()
	optimizer_3.synchronize()
	
	model ...
	...
	optimizer_1.zero_grad()
	loss3.backward()
	optimizer_3.step()
	optimizer_1.synchronize()
	optimizer_2.synchronize()
	

多优化器问题参考issue
https://github.com/horovod/horovod/issues/1417#issuecomment-588367250
终端启动：
horovodrun -np 2 -H hostname1:1,hostname2:1 python train.py

np：后面是进程总数
hostname1是机器1的ip或者名，:1，表示该机型启一个进程；
hostname2是机器2的ip或者名，:1，表示该机型启一个进程；

相比较单纯的pytorch中的并行，horovod需要改动的部分更大，需要注意的地方更多

3. 速度问题

无论是pytorch并行还是horovod并行，有时后你会发现没单机速度快，这是由于在做all-reduce的时候所花费的时间已经是整个eopch时间的主要，有可能是网路、底层通信局限导致了数据同步花了更多时间。如果你当前的代码更多花费的时间在GPU计算、预处理、后处理上，而不是all-reduce做数据同步上，那毫无疑问并行速度上更有优势。此外并行的优势我想更多在于，当显存是整个模型训练的瓶颈时，并行可以享受更大的显存，能够训练更大的batchsize，对于很多大模型，更大的batchsize是模型训练的必要因素，所以我个人觉得小模型并不需要并行

这篇好文章是转载于：学新通技术网