[源碼解析] 深度學習分散式訓練框架 Horovod (1) --- 基礎知識

[源碼解析] 深度學習分散式訓練框架 Horovod --- (1) 基礎知識

0x00 摘要

Horovod 是Uber於2017年發布的一個易於使用的高性能的分散式訓練框架，在業界得到了廣泛應用。

本系列將通過源碼分析來帶領大家了解 Horovod。系列大約有15 ～ 18 篇，本文是系列第一篇，介紹相關背景知識。

0x01 分散式並行訓練

我們首先要介紹下分散式並行訓練。

1.1 分散式並行訓練的必要

傳統的模型訓練中，迭代計算只能利用當前進程所在主機上的所有硬體資源，可是單機擴展性始終有限。而目前的機器學習有如下特點：

樣本數量大。目前訓練數據越來越多，在大型互聯網場景下，每天的樣本量可以達到百億級別。
特徵維度多。因為巨大樣本量導致機器學習模型參數越來越多，特徵維度可以達到千億或者萬億級別。
訓練性能要求高。雖然樣本量和模型參數巨大，但是業務需要我們在短期內訓練出一個優秀的模型來驗證。
模型實時上線。對於推薦資訊類應用，往往要求根據用戶最新行為及時調整模型進行預測。

因此，單機面對海量數據和巨大模型時是無能為力的，有必要把數據或者模型分割成為多分，在多個機器上藉助不同主機上的硬體資源進行訓練加速。

1.2 分散式訓練

本文所說的訓練，指的是利用訓練數據通過計算梯度下降的方式迭代地去優化神經網路參數，並最終輸出網路模型的過程。在單次模型訓練迭代中，會有如下操作：

首先利用數據對模型進行前向的計算。所謂的前向計算，就是將模型上一層的輸出作為下一層的輸入，並計算下一層的輸出，從輸入層一直算到輸出層為止。
其次會根據目標函數，我們將反向計算模型中每個參數的導數，並且結合學習率來更新模型的參數。

而並行梯度下降的基本思想便是：多個處理器分別利用自己的數據來計算梯度，最後通過聚合或其他方式來實現並行計算梯度下降以加速模型訓練過程。比如兩個處理器分別處理一半數據計算梯度 g_1, g_2，然後把兩個梯度結果進行聚合更新，這樣就實現了並行梯度下降。

1.3 訓練並行機制

1.3.1 三種機制

由於使用小批量演算法，可以把寬度（∝W）和深度（∝D）的前向傳播和反向傳播分發到並行的處理器上，這樣深度訓練的並行機制主要有三種：

第一個是模型並行機制（按照網路結構分區）。
- 通常是針對一個節點無法存下整個模型的情況下，去對圖進行拆分。
- 將模型參數進行分散式存儲。電腦上每個計算可以建模為一個有向無環圖（DAG），頂點是計算指令，邊是數據依賴（數據流）。 "基於圖去拆分" 會根據每一層中的神經元（即四維張量中的C、H或W維）來把一張大的圖拆分成很多部分，每個部分都會在很多設備上去計算。
- 或者可以這麼理解：深度學習的計算主要是矩陣運算，有時候矩陣非常大無法放到顯示記憶體中，就只能把超大矩陣拆分了放到不同卡上計算。
- 模型較後部分的計算必須等前面計算完成，因此不同節點間的計算實際是串列的。但每個部分計算互不妨礙，更像是流水線結構。
第二個是數據並行機制（按照輸入樣本分區）。
- 更多場景下我們模型規模不大，在一張 GPU 可以容納，但是訓練數據量會比較大，這時候就採用數據並行機制。
- 具體就是在多節點上並行分割數據和訓練。
第三種不常用的並行機制是流水線機制（按層分區）。
- 在深度學習中，流水線可以是指重疊的計算，即在一層和下一層之間（當數據準備就緒時）連續計算；或者根據深度劃分DNN，將層分配給特定處理器。
- 流水線可以看作是數據並行的一種形式，因為元素（樣本）是通過網路並行處理的，但也可以看作是模型並行，因為流水線的長度是由DNN結構決定的。

具體可見下圖：

1.3.2 如何使用

數據的並行往往意味著計算性能的可擴展，而模型的並行往往意味著記憶體使用的可擴展。

需要注意的是：數據並行和模型並行也並不衝突，兩者可以同時存在，而流水線機制也可以和模型並行一起混用。比如，DistBelief分散式深度學習系統結合了三種並行策略。訓練在同時複製的多個模型上訓練，每個模型副本在不同的樣本上訓練（數據並行），每個副本上，依據同一層的神經元（模型並行性）和不同層（流水線）上劃分任務，進行分布訓練。

另外也需要根據具體問題具體分析，比如現代卷積神經網路主要由兩種層構成，他們具有不一樣的屬性和性能。

卷積層，佔據了90% ~ 95% 的計算量，5% 的參數，但是對結果具有很大的表達能力。
全連接層，佔據了 5% ~ 10% 的計算量， 95% 的參數，但是對於結果具有相對較小的表達的能力。

綜上：卷積層計算量大，所需參數係數 W 少，全連接層計算量小，所需參數係數 W 多。因此對於卷積層適合使用數據並行，對於全連接層適合使用模型並行。

1.4 數據並行訓練

我們本系列主要討論數據並行訓練（其中的一種架構）。

數據並行訓練只是一種邏輯架構。我們從沐神的書裡面摘錄：

假設機器上有\(k\)個GPU。給定要訓練的模型，每個GPU將獨立地維護一組完整的模型參數，儘管GPU上的參數值是相同且同步的。例如，下圖演示了在\(k=2\)時使用數據並行的訓練。

一般來說，訓練過程如下：

在訓練的任何迭代中，給定一個隨機的小批量，我們將該小批量中的樣本分成\(k\)個部分，並將它們均勻地分在多個GPU上。

每個GPU根據分配給它的小批量子集計算模型參數的損失和梯度。

將\(k\)個GPU中每個GPU的局部梯度聚合以獲得當前的小批量隨機梯度。

聚合梯度被重新分配到每個GPU。

每個GPU使用這個小批量隨機梯度來更新它維護的完整的模型參數集。

0x02 通訊 & 架構

前面提到並行梯度下降的例子：兩個處理器分別處理一般數據計算梯度 g_1, g_2，然後把兩個梯度結果進行聚合，最後再把最新參數發給各個分布計算單元，這種訓練演算法叫模型一致性方法（consistent model methods）。這就涉及到了通訊問題，即如何做聚合。

2.1 方法和架構

一般有兩種通訊方法：Share memory 和 Message passing。

Share memory 就是所有處理器共享同一塊記憶體，這樣通訊很容易，但是同一個節點內的處理器之間才可以共享記憶體，不同節點處理器之間無法共享記憶體。

Message passing 就是不同節點之間用消息（比如基於 TCP/IP 或者 RDMA）進行傳遞/通訊，這樣容易擴展，可以進行大規模訓練。

因此我們知道，Message passing 才是解決方案，於是帶來了問題：如何協調這些節點之間的通訊。

有兩種架構：

Client-Server 架構: 一個 server 節點協調其他節點工作，其他節點是用來執行計算任務的 worker。
Peer-to-Peer 架構：每個節點都有鄰居，鄰居之間可以互相通訊。

2.2 非同步 vs 同步

非同步 vs 同步是通訊的另外一個側面。

在數據並行訓練之中，各個計算設備分別根據各自獲得的batch，前向計算獲得損失，進而反向傳播計算梯度。計算好梯度後，就涉及到一個梯度同步的問題：每個計算設備都有根據自己的數據計算的梯度，如何在不同GPU之間維護模型的不同副本之間的一致性。如果不同的模型以某種方式最終獲得不同的權重，則權重更新將變得不一致，並且模型訓練將有所不同。

怎麼做這個同步就是設計分散式機器學習系統的一個核心問題。

分散式訓練的梯度同步策略可分為非同步（asynchronous）梯度更新和同步（synchronous）梯度更新機制。

同步指的是所有的設備都是採用相同的模型參數來訓練，等待所有設備的mini-batch訓練完成後，收集它們的梯度然後取均值，然後執行模型的一次參數更新。
- 同步訓練相當於通過聚合很多設備上的mini-batch形成一個很大的batch來訓練模型，Facebook就是這樣做的，但是他們發現當batch大小增加時，同時線性增加學習速率會取得不錯的效果。
- 同步訓練看起來很不錯，但是實際上需要各個設備的計算能力要均衡，而且要求集群的通訊也要均衡。
- 因為每一輪結束時算得快的節點都需等待算得慢的節點算完，再進行下一輪迭代。類似於木桶效應，一個拖油瓶會嚴重拖慢訓練進度，所以同步訓練方式相對來說訓練速度會慢一些。這個拖油瓶一般就叫做 straggler。
非同步訓練中，各個設備完成一個mini-batch訓練之後，不需要等待其它節點，直接去更新模型的參數，這樣總體會訓練速度會快很多。
- 非同步訓練的一個很嚴重的問題是梯度失效問題（stale gradients），剛開始所有設備採用相同的參數來訓練，但是非同步情況下，某個設備完成一步訓練後，可能發現模型參數其實已經被其它設備更新過了，此時這個梯度就過期了，因為現在的模型參數和訓練前採用的參數是不一樣的。由於梯度失效問題，非同步訓練雖然速度快，但是可能陷入次優解（sub-optimal training performance）。

具體如下圖所示：

這兩種更新方式各有優缺點：

非同步更新可能會更快速地完成整個梯度計算。
同步更新可以更快地進行一個收斂。

選擇哪種方式取決於實際的應用場景。

0x03 具體架構

接下來，我們看看幾種具體架構實現，先給出一個總體說明：

名稱	通訊	架構	並行性
MapReduce	消息傳遞	client-server	批同步
Parameter Server	消息傳遞	client-server	非同步
Decentralized	消息傳遞	P2P	同步或非同步

3.1 MapReduce

MapReduce是Client-Server架構。以 Spark 為例看看是如何進行並行化：

Spark Driver 就是 Server，Spark Executor 就是 Worker 節點，每一個梯度下降過程包含一個廣播、map和一個 reduce 操作。
Server 定義了 map操作（就是具體的訓練），也可以把資訊廣播到worker節點。
Worker 會執行 map 操作進行訓練，在此過程中，數據被分給 worker 進行計算。
計算結束後，worker把計算結果傳回 driver 處理，這個叫做reduce。
在 reduce 過程中，Server 節點對 worker 傳來的計算結果進行聚合之後，把聚合結果廣播到各個worker節點，進行下一次迭代。

3.2 參數伺服器 (PS)

Parameter server 也是一種client-server架構。和MapReduce不同在於 Parameter server 可以是非同步的，MapReduce只有等所有map都完成了才能做reduce操作。

在參數伺服器架構中，計算設備被劃分為參數伺服器（PS）和worker。

參數伺服器（server）。是中心化的組件，主要是負責模型參數的存儲，平均梯度和交換更新。參數伺服器可以按照不同比例的參數伺服器和工作執行緒進行配置，每個參數伺服器都有著不同的配置數據。
工作節點（worker）。每個工作節點會負責它領域內的數據分片所對應模型參數的更新計算（比如前向和反向傳播這類計算密集的運算），同時它們又會向參數伺服器去傳遞它所計算的梯度，由參數伺服器來匯總所有的梯度，再進一步回饋到所有節點。

具體步驟如下：

所有的參數都存儲在參數伺服器中，而工作節點（worker）是萬年打工仔。
工作節點們只負責計算梯度，待所有計算設備完成梯度計算之後，把計算好的梯度發送給參數伺服器，這樣參數伺服器收到梯度之後，執行一定的計算（梯度平均等）之後，就更新其維護的參數，做到了在節點之間對梯度進行平均，利用平均梯度對模型進行更新。
然後參數伺服器再把更新好的新參數返回給所有的工作節點，以對每個節點中的模型副本應用一致化更新。
打工仔們會再進行下一輪的前後向計算。

邏輯如下：

     +----------------------------------------------+
     |  Parameter Server                            |
     |                                              |
     |                                              |
     |   Compute : New P = P + Sum(Delta P ...)     |
     |                                              |
     |                                              |
     |   Parameter 1, Parameter 2, Parameter 3 ...  |
     |                                              |
     |                                              |
     +--+----+----------+--+----------------+--+----+
        ^    |          ^  |                ^  |
        |    |          |  |                |  |
Delta P |    |   Delta P|  |         Delta P|  |
  +-----+    |          |  |                |  +------+
  |    +-----+          |  |                |         |
  |    | New P          |  | New P          +------+  |
  |    |                |  |                       |  |  New P
  |    v                |  |                       |  |
  |                     |  v                       |  v
+-+-----------+   +-----+--+---+             +-----+--+---+
| Worker      |   | Worker     |             | Worker     |
|             |   |            |             |            |
|             |   |            |   ......    |            |
|       Model |   |     Model  |             |     Model  |
+------+------+   +------+-----+             +----+-------+
       ^                 ^                        ^
       |                 |                        |
       |                 |                        |
  +----+----+       +----+-----+               +--+-----+
  | Data 1  |       |  Data 2  |               | Data 3 |
  +---------+       +----------+               +--------+

手機如下：

參數伺服器既可以用在數據並行上，也可以被用到模型並行訓練上。比如可以將模型切分為多個部分，存儲在不同的PS Server節點上，並提供方便的訪問服務，這是參數伺服器的本質。

3.3 Decentralized Network

Decentralized Network 就是去中心化網路，其特點如下：

去中心化網路沒有一個中心節點，屬於 Peer-to-Peer 架構。
採用 message passing 進行通訊，且節點只和鄰居通訊。
並行方式可以採用非同步或者同步。
去中心化網路的收斂情況取決於網路連接情況：
- 連接越緊密，收斂性越快，當強連接時候，模型可以很快收斂；
- 如果不是強連接，它可能不收斂；

0x04 All Reduce

因為本系列是 Horovod，所以我們要先說說參數伺服器的劣勢，下一個系列我們再說參數伺服器優勢。

4.1 參數伺服器劣勢

儘管參數伺服器可以提升表現，但仍然面臨幾個問題：

確定工作者與參數伺服器的正確比例：如果使用一個參數伺服器，它可能會成為網路或計算瓶頸。如果使用多個參數伺服器，則通訊模式變為「All-to-All」，這可能使網路飽和。
處理程式複雜性：參數伺服器的概念較多，這通常導致陡峭的學習曲線和大量的程式碼重構，壓縮了實際建模的時間。
硬體成本 : 參數伺服器的引入也增加了系統的硬體成本。

人們發現，MPI_AllReduce 語義也可以很好地滿足數據並行訓練這一需要。

需要注意的是：AllReduce 既可以是去中心化，也可以是主從式的。

4.2 並行任務通訊分類

並行任務的通訊一般可以分為 Point-to-point communication 和 Collective communication。

P2P 這種模式只有一個sender和一個receiver，實現起來比較簡單，比如NV GPU Direct P2P技術服務於單機多卡的單機卡間數據通訊。
Collective communication包含多個sender和多個receiver，一般的通訊原語包括 broadcast，gather,all-gather，scatter，reduce，all-reduce，reduce-scatter，all-to-all等。

4.3 MPI_AllReduce

AllReduce（對 m 個獨立參數進行規約，並將規約結果返回給所有進程）其實是最顯然和直接的分散式機器學習抽象，因為大部分演算法的結構都是分布數據。在每個子集上面算出一些局部統計量，然後整合出全局統計量，並且再分配給各個節點去進行下一輪的迭代，這樣一個過程就是AllReduce。

可以把每個 Worker 看作是 MPI 概念中的一個進程，比如可以用 4 個 Worker 組成了一個組，該組由 4 個進程組成。我們在這四個進程中對梯度進行一次 MPI_AllReduce。
根據 MPI_AllReduce 的語義，所有參與計算的進程都有結果，所以梯度就完成了分發。只要在初始化的時候，我們可以保證每個 Worker 的參數是一致的，那在後續的迭代計算中，參數會一直保持一致，因為梯度資訊是一致的。
AllReduce 跟 MapReduce 有類似，但後者採用的是面向通用任務處理的多階段執行任務的方式，而AllReduce則讓一個程式在必要的時候佔領一台機器，並且在所有迭代的時候一直跑到底，來防止重新分配資源的開銷，這更加適合於機器學習的任務處理。

所以，MPI_AllReduce 的語義可以很好地解決深度學習中梯度同步的問題。但是到底能不能使用它，還是要看下層的實現對這一場景是否足夠友好。

0x05 ring-allreduce

百度提出使用新演算法來平均梯度，取消 Reducer，並讓這些梯度在所有節點之間交流，這被稱為 ring-allreduce，他們使用 TensorFlow 也實現了這種演算法（//github.com/baidu-research/tensorflow-allreduce）。

5.1 特點

Ring-Allreduce特點如下：

Ring Allreduce 演算法使用定義良好的成對消息傳遞步驟序列在一組進程之間同步狀態（在這種情況下為張量）。
Ring-Allreduce 的命名中 Ring 意味著設備之間的拓撲結構為一個邏輯環形，每個設備都應該有一個左鄰和一個右鄰居，且本設備只會向它右鄰居發送數據，並且從它的左鄰居接受數據。
Ring-Allreduce 的命名中的 Allreduce 則代表著沒有中心節點，架構中的每個節點都是梯度的匯總計算節點。
此種演算法各個節點之間只與相鄰的兩個節點通訊，並不需要參數伺服器。因此，所有節點都參與計算也參與存儲，也避免產生中心化的通訊瓶頸。
相比PS架構，Ring-Allreduce 架構是頻寬優化的，因為集群中每個節點的頻寬都被充分利用。
- 在 ring-allreduce 演算法中，每個 N 節點與其他兩個節點進行 2 * (N-1) 次通訊。在這個通訊過程中，一個節點發送並接收數據緩衝區傳來的塊。在第一個 N - 1 迭代中，接收的值被添加到節點緩衝區中的值。在第二個 N - 1 迭代中，接收的值代替節點緩衝區中保存的值。百度的文章證明了這種演算法是頻寬上最優的，這意味著如果緩衝區足夠大，它將最大化地利用可用的網路。
在深度學習訓練過程中，計算梯度採用BP演算法，其特點是後面層的梯度先被計算，而前面層的梯度慢於後面層，Ring-allreduce架構可以充分利用這個特點，在前面層梯度計算的同時進行後面層梯度的傳遞，從而進一步減少訓練時間。
Ring架構下的同步演算法將參數在通訊環中依次傳遞，往往需要多步才能完成一次參數同步。在大規模訓練時會引入很大的通訊開銷，並且對小尺寸張量（tensor）不夠友好。對於小尺寸張量，可以採用批量操作（batch）的方法來減小通訊開銷。

綜上所述，Ring-based AllReduce 架構的網路通訊量如果處理適當，不會隨著機器增加而增加，而僅僅和模型 & 網路頻寬有關，這針對參數伺服器是個巨大的提升。

5.2 策略

Ring-based AllReduce 策略包括 Scatter-Reduce 和 AllGather 兩個階段。

首先是scatter-reduce，scatter-reduce 會逐步交換彼此的梯度並融合，最後每個 GPU 都會包含完整融合梯度的一部分，是最終結果的一個塊。

假設環中有 N 個 worker，每個 worker 有長度相同的數組，需要將 worker 的數組進行求和。在 Scatter-Reduce 階段，每個 worker 會將數組分成 N 份數據塊，然後 worker 之間進行 N 次數據交換。在第 k 次數據交換時，第 i 個 worker 會將自己的 (i - k) % N 份數據塊發送給下一個 worker。接收到上一個 worker 的數據塊後，worker 會將其與自己對應的數據塊求和。
然後是allgather。GPU 會逐步交換彼此不完整的融合梯度，最後所有 GPU 都會得到完整的最終融合梯度。

在執行完 Scatter-Reduce 後，每個 worker 的數組裡都有某個數據塊是最終求和的結果，現在需要將各數據塊的最後求和結果發送到每個 worker 上。和 Scatter-Reduce 一樣，也需要 N 次循環。在第 k 次循環時，第 i 個 worker 會將其第 (i+1-k)%N 個數據塊發送給下一個 worker 。接收到前一個 worker 的數據塊後，worker 會用接收的數據快覆蓋自己對應的數據塊。進行 N 次循環後，每個 worker 就擁有了數組各數據塊的最終求和結果了。

以下部分來自 //andrew.gibiansky.com/blog/machine-learning/baidu-allreduce/，這是我能找到最優秀的解讀。

5.2.1 結構

環形結構如下，每個 GPU 應該有一個左鄰居和一個右鄰居；它只會向其右側鄰居發送數據，並從其左側鄰居接收數據。：

5.2.2 Scatter-Reduce

scatter-reduce：會逐步交換彼此的梯度並融合，最後每個 GPU 都會包含完整融合梯度的一部分。

為簡單起見，我們假設目標是按元素對單個大型浮點數數組的所有元素求和；系統中有 N 個 GPU，每個 GPU 都有一個相同大小的數組，在 allreduce 的最後環節，每個 GPU 都應該有一個相同大小的數組，其中包含原始數組中數字的總和。

5.2.2.1 分塊

首先，GPU 將陣列劃分為 N 個較小的塊（其中 N 是環中的 GPU 數量）。

接下來，GPU 將進行 N-1 次 scatter-reduce 迭代。

在每次迭代中，GPU 會將其一個塊發送到其右鄰居，並將從其左鄰居接收一個塊並累積到該塊中。每個 GPU 發送和接收的數據塊每次迭代都不同。第 n 個 GPU 通過發送塊 n 和接收塊 n – 1 開始，然後逐步向後進行，每次迭代發送它在前一次迭代中接收到的塊。

5.2.2.2 第一次迭代

在第一次迭代中，上圖中的五個 GPU 將發送和接收以下塊：

GPU	發送	收到
0	塊 0	塊 4
1	塊 1	塊 0
2	塊 2	塊 1
3	塊 3	塊 2
4	塊 4	塊 3

scatter-reduce 的第一次迭代中的數據傳輸如下：

第一次發送和接收完成後，每個 GPU 都會有一個塊，該塊由兩個不同 GPU 上相同塊的總和組成。例如，第二個 GPU 上的第一個塊將是該塊中來自第二個 GPU 和第一個 GPU 的值的總和。

5.2.2.3 全部迭代

在後續迭代中，該過程繼續直到最後。最終每個 GPU 將有一個塊，這個塊包含所有 GPU 中該塊中所有值的總和。

下面系列圖展示了所有數據傳輸和中間結果，從第一次迭代開始，一直持續到scatter-reduce完成。

第一次迭代

第二次迭代

第三次迭代

第四次迭代

所有 scatter-reduce 傳輸後的最終狀態

5.2.3 Allgather

在 scatter-reduce 步驟完成後，在每個 GPU 的數組中都有某一些值（每個 GPU 有一個塊）是最終值，其中包括來自所有 GPU 的貢獻。為了完成 allreduce，GPU 必須接下來交換這些塊，以便所有 GPU 都具有最終所需的值。

ring allgather 與 scatter-reduce 進行相同的處理（發送和接收的 N-1 次迭代），但是他們這次不是累積 GPU 接收的值，而只是簡單地覆蓋塊。第 n 個 GPU 開始發送第 n+1 個塊並接收第 n 個塊，然後在以後的迭代中始終發送它剛剛接收到的塊。

5.2.3.1 第一次迭代

例如，在我們的 5-GPU 設置的第一次迭代中，GPU 將發送和接收以下塊：

圖形處理器	發送	收到
0	塊 1	塊 0
1	塊 2	塊 1
2	塊 3	塊 2
3	塊 4	塊 3
4	塊 0	塊 4

allgather 的第一次迭代中的數據傳輸如下。

第一次迭代完成後，每個 GPU 都會有最終數組的兩個塊。在接下來的迭代中，該過程繼續一直到最後，最終每個 GPU 將擁有整個數組的完全累加值。

5.2.3.2 全部迭代

下面系列圖展示了所有數據傳輸和中間結果，從第一次迭代開始，一直持續到全部收集完成。

Allgather 數據傳輸（迭代 1）

Allgather 數據傳輸（迭代 2）如下：

Allgather 數據傳輸（迭代 3）

Allgather 數據傳輸（迭代 4）

所有全部轉移後的最終狀態。

5.2.4 Horovod 架構圖

工作原理也可以藉助Horovod的發布帖子來看看。

5.2.5 百度思路

或者我們從百度的源碼中也可以直接看到思路，現在摘錄給大家。

具體程式碼參見 //github.com/baidu-research/tensorflow-allreduce/commit/66d5b855e90b0949e9fa5cca5599fd729a70e874#diff-3d530d590e551619acd776cfe7eaff06R517

tensorflow/contrib/mpi_collectives/ring.h

/* Perform a ring allreduce on the data. Allocate the necessary output tensor and
 * store it in the output parameter.
 *
 * Assumes that all MPI processes are doing an allreduce of the same tensor,
 * with the same dimensions.
 *
 * A ring allreduce is a bandwidth-optimal way to do an allreduce. To do the allreduce,
 * the nodes involved are arranged in a ring:
 *
 *                   .--0--.
 *                  /       \
 *                 3         1
 *                  \       /
 *                   *--2--*
 *
 *  Each node always sends to the next clockwise node in the ring, and receives
 *  from the previous one.
 *
 *  The allreduce is done in two parts: a scatter-reduce and an allgather. In
 *  the scatter reduce, a reduction is done, so that each node ends up with a
 *  chunk of the final output tensor which has contributions from all other
 *  nodes.  In the allgather, those chunks are distributed among all the nodes,
 *  so that all nodes have the entire output tensor.
 *
 *  Both of these operations are done by dividing the input tensor into N
 *  evenly sized chunks (where N is the number of nodes in the ring).
 *
 *  The scatter-reduce is done in N-1 steps. In the ith step, node j will send
 *  the (j - i)th chunk and receive the (j - i - 1)th chunk, adding it in to
 *  its existing data for that chunk. For example, in the first iteration with
 *  the ring depicted above, you will have the following transfers:
 *
 *      Segment 0:  Node 0 --> Node 1
 *      Segment 1:  Node 1 --> Node 2
 *      Segment 2:  Node 2 --> Node 3
 *      Segment 3:  Node 3 --> Node 0
 *
 *  In the second iteration, you'll have the following transfers:
 *
 *      Segment 0:  Node 1 --> Node 2
 *      Segment 1:  Node 2 --> Node 3
 *      Segment 2:  Node 3 --> Node 0
 *      Segment 3:  Node 0 --> Node 1
 *
 *  After this iteration, Node 2 has 3 of the four contributions to Segment 0.
 *  The last iteration has the following transfers:
 *
 *      Segment 0:  Node 2 --> Node 3
 *      Segment 1:  Node 3 --> Node 0
 *      Segment 2:  Node 0 --> Node 1
 *      Segment 3:  Node 1 --> Node 2
 *
 *  After this iteration, Node 3 has the fully accumulated Segment 0; Node 0
 *  has the fully accumulated Segment 1; and so on. The scatter-reduce is complete.
 *
 *  Next, the allgather distributes these fully accumululated chunks across all nodes.
 *  Communication proceeds in the same ring, once again in N-1 steps. At the ith step,
 *  node j will send chunk (j - i + 1) and receive chunk (j - i). For example, at the
 *  first iteration, the following transfers will occur:
 *
 *      Segment 0:  Node 3 --> Node 0
 *      Segment 1:  Node 0 --> Node 1
 *      Segment 2:  Node 1 --> Node 2
 *      Segment 3:  Node 2 --> Node 3
 *
 * After the first iteration, Node 0 will have a fully accumulated Segment 0
 * (from Node 3) and Segment 1. In the next iteration, Node 0 will send its
 * just-received Segment 0 onward to Node 1, and receive Segment 3 from Node 3.
 * After this has continued for N - 1 iterations, all nodes will have a the fully
 * accumulated tensor.
 *
 * Each node will do (N-1) sends for the scatter-reduce and (N-1) sends for the allgather.
 * Each send will contain K / N bytes, if there are K bytes in the original tensor on every node.
 * Thus, each node sends and receives 2K(N - 1)/N bytes of data, and the performance of the allreduce
 * (assuming no latency in connections) is constrained by the slowest interconnect between the nodes.
 *
 */

5.3 區別

在中等規模模型情況下，all-reduce 更適合。當規模巨大時候則應該使用參數伺服器。

參數伺服器適合的是高緯稀疏模型訓練，它利用的是維度稀疏的特點，每次 pull or push 只更新有效的值。但是深度學習模型是典型的dense場景，embedding做的就是把稀疏變成稠密。所以這種 pull or push 的不太適合。而網路通訊上更優化的 all-reduce 適合中等規模的深度學習。

又比如由於推薦搜索領域模型的 Embedding 層規模龐大以及訓練數據樣本長度不固定等原因，導致容易出現顯示記憶體不足和卡間同步時間耗費等問題，所以 all-reduce 架構很少被用於搜索推薦領域。

至此，背景知識已經介紹完畢，下一篇我們開始介紹 Horovod 的使用。