是時候放棄tensorflow集羣投入horovod的懷抱

當數據較多或者模型較大時，爲提升機器學習模型訓練效率，通常採用多GPU的分佈式訓練。

按照並行方式，分佈式訓練通常分爲數據並行和模型並行兩種， 模型並行：分佈式系統中的不一樣GPU負責網絡模型的不一樣部分。例如，神經網絡模型的不一樣網絡層被分配到不一樣的GPU，或者同一層內部的不一樣參數被分配到不一樣GPU；

數據並行：不一樣的GPU有同一個模型的多個副本，每一個GPU分配到不一樣的數據，而後將全部GPU的計算結果按照某種方式合併。

注意，上述中的不用GPU能夠是同一臺機上的多個GPU，也能夠是不用機上的GPU。

注：圖中的Machine其實就是GPU，固然也能夠指CPU，但深度學習不多采用CPU訓練

固然也有數據並行和模型並行的混合模式。

注：圖中的Machine其實就是GPU，固然也能夠包含CPU，但深度學習不多采用CPU訓練

由於模型並行各個部分存在必定的依賴，規模伸縮性差（意思是不能隨意增長GPU的數量），在實際訓練中用的很少。而數據並行，則各部分獨立，規模伸縮性好，實際訓練中更爲經常使用，提速效果也更好。

數據並行會涉及到各個GPU之間同步模型參數，通常分爲同步更新和異步更新。同步更新要等到全部GPU的梯度計算完成，再統一計算新權值，而後全部GPU同步新值後，才進行下一輪計算。異步更新，每一個GPU梯度計算完後，無需等待其餘GPU的梯度計算（有時能夠設置須要等待的梯度個數），可當即更新總體權值，而後同步此權值，便可進行下一輪計算。同步更新有等待，異步更新基本沒有等待，但異步更新涉及到梯度過期等更復雜問題。

在實際應用中，單機多卡的同步式數據並行是最經常使用的，在論文中最多見的訓練方式是單機八卡。數據再多時，通常就須要多機多卡了。

不管是單機多卡，仍是多機多卡，均是分佈式訓練，在horovod出現以前，使用tensorflow，通常只有官方推薦的集羣訓練方式。

但是tensorflow的集羣訓練，用起來並不輕鬆。

tensorflow集羣的缺點

1. 概念多，學習曲線陡峭

tensorflow的集羣採用的是parameter server架構，所以引入了比較多複雜概念，羅列以下

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
複製代碼

我反覆研究過屢次，仍是沒有完全弄清楚，server，client，master，master service，worker service，clone，session之間的關係。 大體是，在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之間作梯度傳遞和權值同步。 stackoverflow.com/questions/3…

2. 修改的代碼量大

若是想把單機單卡的模型，移植到多機多卡，涉及的代碼量是以天記的，慢的話甚至須要一週。

3. 須要多臺機子跑不一樣的腳本

tensorflow集羣是採用parameter server架構的，要想跑多機多卡的集羣，每一個機子都要啓動一個client，即跑一個腳本，來啓動訓練，100個機子，人就要崩潰了。

4. ps和worker的比例很差選取

tensorflow集羣要將服務器分爲ps和worker兩種job類型，ps設置多少性能最近並無肯定的計算公式。

5. 性能損失較大

tensorflow的集羣性能並很差，當超過必定規模時，性能甚至會掉到理想性能的一半如下。

Horovod

因爲tensorflow集羣太不友好，業內也一直在嘗試新的集羣方案。 2017年Facebook發佈了《Accurate, large minibatch SGD: Training ImageNet in 1 hour 》驗證了大數據並行的高效性，同年百度發表了《Bringing HPC techniques to deep learning 》，驗證了全新的梯度同步和權值更新算法的可行性。受這兩篇論文的啓發，Uber開發了Horovod集羣方案。

約定以下： 網絡帶寬記爲：B（單位Mb/s）， 模型總參數數據量記爲：D（單位Mb）， 總服務器數量記爲：n， 參數服務器數量記爲：n_p（其中有n= n_p+ n_w）， worker服務器數量記爲：n_w（其中有n= n_p+ n_w） 單服務器計算一次耗時記爲：T_0

梯度同步和權值更新算法

1） parameter server架構

tensorflow的集羣架構是parameter server架構，數據的傳導模型以下圖。

則能夠計算出，parameter server架構的集羣方案，總耗時：

能夠看出T與總節點數n基本成線性關係，但不一樣的參數服務器和woker服務器分配方案，總性能也將不一樣。 假設，e表示worker服務器佔比，即e=n_w/n，則能夠計算出最優的e值爲：

能夠看出，最優worker服務器佔比與模型大小、網絡帶寬、單機運行市場都有關係，並非一個一眼能最優值得超參數。

2）horovod的ring-allreduce算法

百度2017年發表的《Bringing HPC techniques to deep learning 》中，採用了全新的梯度同步和權值同步算法，叫作ring-allreduce。此種算法各個節點之間只與相鄰的兩個節點通訊，並不須要參數服務器。所以，全部節點都參與計算也參與存儲。 一次權重更新，主要包含兩個過程， 1）累計梯度 將全部梯度分爲n個片斷，每次只與相鄰節點傳遞1個片斷的梯度，n-1次後，每一片斷的梯度都完成了全部節點這一片斷梯度的累計，但不用片斷的累計值分佈在不一樣節點上。以下圖的第二、第3步； 2）將累計後的梯度分發到全部節點 將第一步累計的梯度再次經過n-1次的相互交換後，全部節點的梯度完成同步。以下圖的第四、第5步。再平均後，更新權重，就完成了全部節點權重的更新。

能夠計算出ring-allreduce算法的總耗時爲：

能夠看出，總耗時基本與總節點數n成線性關係（n較大時，1/n基本爲0）

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

概念

horovod的數據傳遞是基於MPI，所以其涉及的概念也是MPI中的概念。以4個服務器，每一個服務器4個GPU爲例，

• size 進程數量，也即全部GPU數量，爲16
• rank 進程的惟一ID，0-15
• local rank 每個server中的進程的本地惟一ID，0-3
• allreduce 累加全部數據，並同步到全部節點的操做，以下圖

• allgather 收集全部數據，並同步到全部節點的操做，完成後每一個節點都包含全部節點的數據，而且這些數據單獨存在。以下圖。

• broadcast 將數據（須要由根節點確認）從一個節點傳播到其餘全部節點的操做

大概就這麼多概念，簡單清晰。

將單機單卡改成多機多卡

將一個只支持單機單卡的訓練腳本修改成支持多機多卡的訓練腳本，以tensorflow爲例，只須要作以下改動：

import tensorflow as tf
import horovod.tensorflow as hvd


# Initialize Horovod
hvd.init()

# Pin GPU to be used to process local rank (one GPU per process)
config = tf.ConfigProto()
config.gpu_options.visible_device_list = str(hvd.local_rank())

# Build model...
loss = ...
opt = tf.train.AdagradOptimizer(0.01 * hvd.size())

# Add Horovod Distributed Optimizer
opt = hvd.DistributedOptimizer(opt)

# Add hook to broadcast variables from rank 0 to all other processes during
# initialization.
hooks = [hvd.BroadcastGlobalVariablesHook(0)]

# Make training operation
train_op = opt.minimize(loss)

# Save checkpoints only on worker 0 to prevent other workers from corrupting them.
checkpoint_dir = '/tmp/train_logs' if hvd.rank() == 0 else None

# The MonitoredTrainingSession takes care of session initialization,
# restoring from a checkpoint, saving to a checkpoint, and closing when done
# or an error occurs.
with tf.train.MonitoredTrainingSession(checkpoint_dir=checkpoint_dir,
                                       config=config,
                                       hooks=hooks) as mon_sess:
  while not mon_sess.should_stop():
    # Perform synchronous training.
    mon_sess.run(train_op)
複製代碼

能夠看出，改動不大，只需添加10行左右的代碼，主要分爲6步：

1）初始化horovod

hvd.init()
複製代碼

2）一個GPU與一個進程綁定

config = tf.ConfigProto()
config.gpu_options.visible_device_list = str(hvd.local_rank())
複製代碼

3）根據總GPU數量放大學習率

opt = tf.train.AdagradOptimizer(0.01 * hvd.size())
複製代碼

由於BatchSize會根據GPU數量放大，因此學習率也應該放大

4）使用hvd.DistributedOptimizer封裝原有的optimizer

opt = hvd.DistributedOptimizer(opt)
複製代碼

分佈式訓練涉及到梯度同步，每個GPU的梯度計算仍然由原有的optimizer 計算，只是梯度同步由hvd.DistributedOptimizer負責。

5）廣播初始變量值到全部進程

hooks = [hvd.BroadcastGlobalVariablesHook(0)]
複製代碼

主要爲了確保全部進程變量初始值相同

6）只在worker 0上保存checkpoint

checkpoint_dir = '/tmp/train_logs' if hvd.rank() == 0 else None
複製代碼

防止checkpoint保存錯亂

horovod只是須要改動必要改動的，不涉及parameter server架構的device設置等，繁瑣的操做。

起訓練

在單機4卡的機上起訓練，只需執行如下命令:

horovodrun -np 4 -H localhost:4 python train.py
複製代碼

在4機，每機4卡的機子上起訓練，只需在一個機子上執行如下命令便可：

horovodrun -np 16 -H server1:4,server2:4,server3:4,server4:4 python train.py
複製代碼

注意不管是單機多卡，仍是多機多卡，都只需在一個機子上執行一次命令便可，其餘機horovod會用MPI啓動進程和傳遞數據。

性能對比

horovod隨着規模增大，性能損失遠小於tensorflow，基本是線性增長的。

結論

經過tensorflow集羣的人，會深入體會到horovod有多好用，感謝百度、Facebook和Uber讓深度學習更美好。

不過，也要注意到，horovod的分佈式貌似只支持同步更新式的數據並行，模型並行和異步更新式的數據並行，我沒有嘗試過，根據ring-allreduce算法可知，應該是不支持的。