[人工智能]从tensorflow集群到horovod分布式计算框架

本文参考：https://juejin.cn/post/6844903827787743239

一、分布式训练介绍

当数据较多或者模型较大时，为提高机器学习模型训练效率，一般采用多GPU的分布式训练。
按照并行方式，分布式训练一般分为数据并行和模型并行两种，

两种方式——模型并行和数据并行

• 模型并行：分布式系统中的不同GPU负责网络模型的不同部分。例如，神经网络模型的不同网络层被分配到不同的GPU，或者同一层内部的不同参数被分配到不同GPU；
• 数据并行：不同的GPU有同一个模型的多个副本，每个GPU分配到不同的数据，然后将所有GPU的计算结果按照某种方式合并。
  
  

注意，上述中的不用GPU可以是同一台机上的多个GPU，也可以是不用机上的GPU。

数据并行的优势

因为模型并行各个部分存在一定的依赖，规模伸缩性差（意思是不能随意增加GPU的数量），在实际训练中用的不多。

数据并行，则各部分独立，规模伸缩性好，实际训练中更为常用，提速效果也更好。

模型参数更新方式——同步更新，异步更新

数据并行会涉及到各个GPU之间同步模型参数，一般分为同步更新和异步更新。

• 同步更新：等到所有GPU的梯度计算完成，再统一计算新权值，然后所有GPU同步新值后，才进行下一轮计算。
• 异步更新：每个GPU梯度计算完后，无需等待其他GPU的梯度计算（有时可以设置需要等待的梯度个数），可立即更新整体权值，然后同步此权值，即可进行下一轮计算。

同步更新有等待时长，异步更新基本没有等待，但异步更新涉及到梯度过时等更复杂问题。

在实际应用中，单机多卡的同步式数据并行是最常用的，在论文中最常见的训练方式是单机八卡。数据再多时，一般就需要多机多卡了。

数据并行流程

「数据并行」方法在分布式训练上包含在多节点上并行分割数据和训练。在同步情况下，不同批次数据的梯度将在不同节点上分别进行计算，但在节点之间进行互相平均，以对每个节点中的模型副本应用一致化更新。

具体算法可以简化为下面几步：

1. 运行训练脚本的多个副本，对于每个每个副本：

   a）读取数据块 ；
   b）将其输入模型；
   c）计算模型更新（梯度）

2. 计算这些副本梯度的均值

3. 更新模型

4. 重复 1a 步骤

无论是单机多卡，还是多机多卡，均是分布式训练，在horovod出现之前，使用tensorflow，一般只有官方推荐的集群训练方式。可是tensorflow的集群训练，用起来并不轻松。

二、tensorflow集群缺点

0 tensorflow集群简介

标准分布式 TensorFlow 包使用parameter server,ps参数服务器的方法来平均梯度。在这种方法之下，每个进程都有一到两个角色：

1. 工作线程: 工作线程处理训练数据，计算梯度，并把它们传递到参数服务器上进行平均。
2. 参数服务器: 可以按照不同比例的参数服务器和工作线程进行配置，每个参数服务器都有着不同的配置数据。

数据传导方式为：

1 概念多，学习曲线陡峭

tensorflow的集群采用的是parameter server（ps）架构，因此引入了比较多复杂概念，罗列如下：

server
client
master
cluster
parameter server
worker
job
task
replica_device_setter
master service
worker service
clone

涉及到函数：

tf.train.Server
tf.train.Supervisor
tf.train.SessionManager
tf.train.ClusterSpec
tf.train.replica_device_setter
tf.train.MonitoredTrainingSession
tf.train.MonitoredSession
tf.train.SingularMonitoredSession
tf.train.Scaffold
tf.train.SessionCreator
tf.train.ChiefSessionCreator
tf.train.WorkerSessionCreator

大致是，在client中创建server实例，session与server一一对应，server内含master service和worker service两个服务，

• master service负责与外界通讯，比如sess.run一般都是告诉server的master service要开始工作了，server的master service通知同一个server的worker service去干活
• worker service调动GPU运算，完成后，返回结果给master service，做权值更新，如果是多机多卡的分布式，parameter server与master service之间做梯度传递和权值同步。

2 修改代码量大

3 需要多台机子一起运转

tensorflow集群是采用parameter server架构的，要想跑多机多卡的集群，每个机子都要启动一个client，即跑一个脚本，来启动训练，100个机子，人就要崩溃了。

4 ps和worker的比例不好选取

tensorflow集群要将服务器分为ps和worker两种job类型，ps设置多少性能最近并没有确定的计算公式。

如果使用一个ps，它可能会成为网络或计算瓶颈。 如果使用多个ps，则通信模式变为“All-to-All”，这可能使网络互连饱和。

5 性能损失较大

tensorflow的集群性能并不好，当超过一定规模时，性能甚至会掉到理想性能的一半以下。例如，在使用 128 个 GPU 进行训练时，我们因为低效率损失了一半的计算资源。

三、Horvod

由于tensorflow集群太不友好，业内也一直在尝试新的集群方案。

2017年Facebook发布了《Accurate, large minibatch SGD: Training ImageNet in 1 hour 》验证了大数据并行的高效性，同年百度发表了《Bringing HPC techniques to deep learning 》，验证了全新的梯度同步和权值更新算法的可行性。受这两篇论文的启发，Uber开发了Horovod集群方案。

1 parameter server 架构

约定如下：
网络带宽记为：$C（单位Mb/s）$，
模型总参数数据量记为：$D（单位Mb）$，
总服务器数量记为：$n$，
参数服务器数量记为：$n_p（其中有n=n_p+n_w）$，
worker服务器数量记为：$n_w（其中有n=n_p+n_w）$
单服务器计算一次耗时记为：$T_0$

tensorflow的集群架构是parameter server架构，数据的传导模型如下图。

则可以计算出，parameter server架构的集群方案，总耗时：
$$T=\frac{T_0+2\frac{D}{C}\frac{n_w}{n_p}}{n_w}$$

可以看出T与总节点数n基本成线性关系，但不同的参数服务器和woker服务器分配方案，总性能也将不同。

假设，$\xi$表示worker服务器占比，即$\xi =n_w/n$，则可以计算出最优的e值为：
$${\xi }^{?}=\frac{1}{1+\sqrt{\frac{2D/C}{T_0}}}$$
可以看出，最优worker服务器占比与模型大小、网络带宽、单机运行时长都有关系，并不是一个一眼能最优值得超参数。

2 百度的ring-allreduce算法

百度2017年发表的《Bringing HPC techniques to deep learning 》中，采用了全新的梯度同步和权值同步算法，叫做ring-allreduce。此种算法各个节点之间只与相邻的两个节点通信，并不需要参数服务器。因此，所有节点都参与计算也参与存储。
一次权重更新，主要包含两个过程：

• 1）累计梯度
  将所有梯度分为n个片段，每次只与相邻节点传递1个片段的梯度，n-1次后，每一片段的梯度都完成了所有节点这一片段梯度的累计，但不用片段的累计值分布在不同节点上。如下图的第2、第3步；

• 2）将累计后的梯度分发到所有节点
  将第一步累计的梯度再次通过n-1次的相互交换后，所有节点的梯度完成同步。如下图的第4、第5步。再平均后，更新权重，就完成了所有节点权重的更新。

可以计算出ring-allreduce算法的总耗时为：
$$T=\frac{T_0+4\frac{D}{C}(1?1/n)}{n}$$

可以看出，总耗时基本与总节点数n成线性关系（n较大时，1/n基本为0）

Horovod的梯度同步和权值同步就采用了ring-allreduce算法。

3 Horovod

horovod的数据传递是基于MPI，因此其涉及的概念也是MPI中的概念。以4个服务器，每个服务器4个GPU为例，

• size 进程数量，也即所有GPU数量，为16
• rank 进程的唯一ID，0-15
• local rank 每一个server中的进程的本地唯一ID，0-3
• allreduce 累加所有数据，并同步到所有节点的操作，如下图

• allgather 收集所有数据，并同步到所有节点的操作，完成后每个节点都包含所有节点的数据，并且这些数据单独存在。如下图。
  
• broadcast 将数据（需要由根节点确认）从一个节点传播到其他所有节点的操作

大概就这么多概念，简单清晰。

四、实践

1 SageMaker pipe mode下使用 Horovod 实现多 GPU 分布式训练

https://aws.amazon.com/cn/blogs/china/multi-gpu-and-distributed-training-using-horovod-in-amazon-sagemaker-pipe-mode/

pipe mode 是在训练实例和S3存储桶之间直接创建一个输入管道，允许进程直接访问对象，消除了file mode：训练之前将S3的数据复制到训练实例的工作中。在pipe mode，训练数据将作为FIFO流的形式进行交付，TensorFlow扩展的dataset类极大降低了访问流数据集的难度。

要对给定Amazon S3 URI中的数据集以管道模式加以访问，请在创建Amazon SageMaker Estimator时将输入模式设置为 Pipe ，具体参见以下代码：

from sagemaker.tensorflow import TensorFlow

tf_estimator = TensorFlow(entry_point='train.py',
                          role='SageMakerRole',
                          train_instance_type='ml.p3.2xlarge',
                          train_instance_count=2,
                          framework_version='2.1.0',
                          py_version='py3',
                          input_mode='Pipe')

当您配合Horovod使用管道模式执行单机多卡或者多机多卡的分布式训练时，有一点需要特别注意。下图所示为这类场景的基本架构。

管道模式将数据从Amazon S3流式的传送到训练实例当中的Unix命名管道/FIFOs当中。一个FIFO文件仅支持一对写入/读取程序，且每轮训练周期内我们只能为一条通道创建一个FIFO文件。通常，人们会为训练数据集定义一条通道，并为验证或测试数据集定义另一条单独的通道，而后将这些输入通道作为Amazon SageMaker Estimator中 fit() 函数的参数传递至训练作业。

from sagemaker.session import s3_input

input_channel = {'train': s3_input('s3://your-bucket-name/train-dataset/')}

tf_estimator.fit(inputs=input_channel)     

这种方式对于Horovod多卡训练就会出错。因为Horovod在多gpu中，每个GPU都有对应的进程，启动各个进程来争夺这个FIFO流，导致多个进程无法同时访问这些FIFO。而且由于同一时间内只有单一工作进程能够访问FIFO，且在完成训练作业之前不会释放句柄，就导致所有其他工作进程无法从该FIFO中读取数据，最终令训练作业陷入死锁式的无限循环。

因此，需要对数据进行先分片，分片数量和GPU的数量对应，如果您拥有4000个TensorFlow记录文件，且使用一台带有4 GPU的ml.p3.8xlarge实例进行模型训练，则可以为互不重复的1000个TensorFLow记录文件设定不同的前缀，如以下代码所示：

s3://your-bucket-name/train/0/
s3://your-bucket-name/train/1/
s3://your-bucket-name/train/2/
s3://your-bucket-name/train/3/

使用SharedByS3Key 作为Amazon S3数据类型分配方式进行的数据集分片方法，并不完全适用于Horovod。 这是因为在使用SharedByS3Key 时，分片只会以实例为单位、而非以工作进程为单位进行，且实例中的工作进程与GPU的数量保持一致。同样的，各个实例仍然只拥有一条输入通道。因此，大家需要将数据集的分片数量，设定为与Horovod集群内GPU数相同。

接下来，我们需要为Amazon SageMaker训练定义四条输入通道，具体参见以下代码：

from sagemaker.session import s3_input

shuffle_config = sagemaker.session.ShuffleConfig(234)

train_s3_uri_prefix = 's3://your-bucket-name/train'
input_channels = {}

for idx in range(4):
    train_s3_uri = f'{train_s3_uri_prefix}/train/{idx}/'
    train_s3_input = s3_input(train_s3_uri, shuffle_config=shuffle_config)
    input_channels[f'train_{idx}'] = train_s3_input