具體來(lái)講，本文首先介紹了分布式計(jì)算的基本概念，以及分布式計(jì)算如何用于深度學(xué)習(xí)。然后，列舉了配置處理分布式應(yīng)用的環(huán)境的標(biāo)準(zhǔn)需求(硬件和軟件)。***，為了提供親身實(shí)踐的經(jīng)驗(yàn)，本文從理論角度和實(shí)現(xiàn)的角度演示了一個(gè)用于訓(xùn)練深度學(xué)習(xí)模型的分布式算法(同步隨機(jī)梯度下降，synchronous SGD)。

[[264239]]

何為分布式計(jì)算

分布式計(jì)算指的是一種編寫程序的方式，它利用網(wǎng)絡(luò)中多個(gè)連接的不同組件。通常，大規(guī)模計(jì)算通過(guò)以這種方式布置計(jì)算機(jī)來(lái)實(shí)現(xiàn)，這些計(jì)算機(jī)能夠并行地處理高密度的數(shù)值運(yùn)算。在分布式計(jì)算的術(shù)語(yǔ)中，這些計(jì)算機(jī)通常被稱為節(jié)點(diǎn)(node)，這些節(jié)點(diǎn)的集合就是集群。這些節(jié)點(diǎn)一般是通過(guò)以太網(wǎng)連接的，但是其他的高帶寬網(wǎng)絡(luò)也可以利用分布式架構(gòu)的優(yōu)勢(shì)。

深度學(xué)習(xí)如何從分布式計(jì)算中受益?

作為深度學(xué)習(xí)的主力，神經(jīng)網(wǎng)絡(luò)出現(xiàn)在文獻(xiàn)中已經(jīng)有一段時(shí)間了，但是直到最近，才有人完全利用它的潛力。神經(jīng)網(wǎng)絡(luò)異軍突起的主要原因之一就是巨大的算力，這正是我們?cè)诒疚闹幸獙懙膬?nèi)容。深度學(xué)習(xí)需要基于大量數(shù)據(jù)訓(xùn)練深度神經(jīng)網(wǎng)絡(luò)，它包含大量的參數(shù)。分布式計(jì)算是能夠充分利用現(xiàn)代硬件的***工具。下面是它的核心思想：

精心設(shè)計(jì)的分布式算法可以做到：

• 為了連貫處理，在多個(gè)節(jié)點(diǎn)上「分配」計(jì)算(深度學(xué)習(xí)模型中的前向傳播和反向傳播)和數(shù)據(jù)。
• 為了實(shí)現(xiàn)一致性，它能夠在多個(gè)節(jié)點(diǎn)上建立一種有效的「同步」。

MPI：分布式計(jì)算標(biāo)準(zhǔn)

你還必須習(xí)慣另一個(gè)術(shù)語(yǔ)——消息傳遞接口(MPI)。MPI 幾乎是所有分布式計(jì)算的主力。MPI 是一個(gè)開放標(biāo)準(zhǔn)，它定義了一系列關(guān)于節(jié)點(diǎn)互相通信的規(guī)則，MPI 也是一個(gè)編程模型/API。MPI 不是一款軟件或者工具，它是一種規(guī)范。

1991 年夏天，一批來(lái)自學(xué)術(shù)界和產(chǎn)業(yè)界的組織和個(gè)人聚在一起，最終創(chuàng)建了 MPI 論壇(MPI Forum)。該論壇達(dá)成了一個(gè)共識(shí)，為一個(gè)庫(kù)起草了語(yǔ)法和語(yǔ)義規(guī)范，為不同硬件提供商提出可移植/靈活/優(yōu)化的實(shí)現(xiàn)提供指導(dǎo)。多家硬件提供商都有自己的 MPI 實(shí)現(xiàn)——OpenMPI、MPICH、MVAPICH、Intel MPI 等。

在這份教程中，我們將會(huì)使用 Intel MPI，因?yàn)樗指咝?，而且也針?duì) Intel 平臺(tái)做了優(yōu)化。原始的 Intel MPI 是一個(gè) C 語(yǔ)言庫(kù)，并且級(jí)別非常低。

配置

對(duì)分布式系統(tǒng)而言，合適的配置是非常重要的。如果沒有合適的硬件和網(wǎng)絡(luò)布置，即使你對(duì)它的編程模型有著概念上的理解，也是沒多大用的。下面是需要做的關(guān)鍵布置：

• 通常需要由一系列通過(guò)通用網(wǎng)絡(luò)互聯(lián)形成集群的節(jié)點(diǎn)。推薦使用高端服務(wù)器作為節(jié)點(diǎn)，以及高帶寬的網(wǎng)絡(luò)，例如 InfiniBand。
• 集群中的所有節(jié)點(diǎn)都需要具有完全相同用戶名的 Linux 系統(tǒng)。
• 節(jié)點(diǎn)之間必須擁有無(wú)密碼 SSH 連接，這對(duì)無(wú)縫連接至關(guān)重要。
• 必須安裝一種 MPI 實(shí)現(xiàn)。本文只聚焦于 Intel MPI。
• 需要一個(gè)共同的文件系統(tǒng)，它對(duì)所有的節(jié)點(diǎn)都是可見的，而且分布式應(yīng)用必須駐留在上面。網(wǎng)絡(luò)文件系統(tǒng)(NFS，Network Filesystem)是實(shí)現(xiàn)此目的一種方式。

并行策略的類型

并行深度學(xué)習(xí)模型有兩種流行的方式：

• 模型并行
• 數(shù)據(jù)并行

1. 模型并行

模型并行指的是一個(gè)模型從邏輯上被分成了幾個(gè)部分(例如，一些層在一部分，其他層在另一部分)，然后把它們部署在不同的硬件/設(shè)備上。

盡管從執(zhí)行時(shí)間上來(lái)看，將模型的不同部分部署在不同設(shè)備上確實(shí)有好處，但是它通常是出于避免內(nèi)存限制才使用。具有特別多參數(shù)的模型會(huì)受益于這種并行策略，因?yàn)檫@類模型需要很高的內(nèi)存占用，很難適應(yīng)到單個(gè)系統(tǒng)。

2. 數(shù)據(jù)并行

另一方面，數(shù)據(jù)并行指的是，通過(guò)位于不同硬件/設(shè)備上的同一個(gè)網(wǎng)絡(luò)的多個(gè)副本來(lái)處理數(shù)據(jù)的不同批(batch)。不同于模型并行，每個(gè)副本可能是整個(gè)網(wǎng)絡(luò)，而不僅僅是一部分。

正如你可能猜到的，這種策略隨著數(shù)據(jù)的增長(zhǎng)可以很好地?cái)U(kuò)展。但是，由于整個(gè)網(wǎng)絡(luò)必須部署在一個(gè)設(shè)備上，因此可能無(wú)法幫助到具有高內(nèi)存占用的模型。下圖應(yīng)該可以說(shuō)清楚這個(gè)問(wèn)題。

模型并行 VS 數(shù)據(jù)并行

實(shí)際上，在大組織里，為了執(zhí)行生產(chǎn)質(zhì)量的深度學(xué)習(xí)訓(xùn)練算法，數(shù)據(jù)并行更加流行也更加常用。所以，本教程主要介紹數(shù)據(jù)并行。

torch.distributed API

PyTorch提供了一個(gè)非常優(yōu)雅并且易于使用的 API，作為用 C 語(yǔ)言寫的底層 MPI 庫(kù)的接口。PyTorch 需要從源碼編譯，并且必須與安裝在系統(tǒng)中的 Intel MPI 進(jìn)行鏈接。我們現(xiàn)在就看一下 torch.distributed 的基本用法，以及如何執(zhí)行它。

# filename 'ptdist.py' 
import torch 
import torch.distributed as dist 
 
def main(rank, world): 
    if rank == 0: 
        x = torch.tensor([1., -1.]) # Tensor of interest 
        dist.send(x, dst=1) 
        print('Rank-0 has sent the following tensor to Rank-1') 
        print(x) 
    else: 
        z = torch.tensor([0., 0.]) # A holder for recieving the tensor 
        dist.recv(z, src=0) 
        print('Rank-1 has recieved the following tensor from Rank-0') 
        print(z) 
 
if __name__ == '__main__': 
    dist.init_process_group(backend='mpi') 
    main(dist.get_rank(), dist.get_world_size()) 

點(diǎn)對(duì)點(diǎn)通信

用 mpiexec 執(zhí)行上面的代碼，能夠得到一個(gè)分布式進(jìn)程調(diào)度器，基于任何標(biāo)準(zhǔn) MPI 實(shí)現(xiàn)都可以，結(jié)果如下：

cluster@miriad2a:~/nfs$ mpiexec -n 2 -ppn 1 -hosts miriad2a,miriad2b python ptdist.py 
Rank-0 has sent the following tensor to Rank-1 
tensor([ 1., -1.]) 
Rank-1 has recieved the following tensor from Rank-0 
tensor([ 1., -1.]) 

• ***行要被執(zhí)行的是 dist.init_process_group(backend)，它基本上設(shè)置了參與節(jié)點(diǎn)之間的內(nèi)部通信通道。它使用了一個(gè)參數(shù)來(lái)指定使用哪個(gè)后端(backend)。因?yàn)槲覀兺耆褂? MPI，所以在我們的例子中 backend='mpi'。也有其他的后端(例如 TCP、Gloo、NCCL)。
• 需要檢索的兩個(gè)參數(shù)——world size 和 rank。World 指的是在特定 mpiexec 調(diào)用環(huán)境中所有節(jié)點(diǎn)的集合(參見 mpiexec 中的 -hosts flag)。rank 是由 MPI 運(yùn)行時(shí)為每一個(gè)進(jìn)程分配的唯一整數(shù)。它從 0 開始。它們?cè)?-hosts 中被指定的順序用于分配數(shù)值。所以，在這個(gè)例子中，節(jié)點(diǎn)「miriad2a」上的進(jìn)程會(huì)被賦值 Rank 0，節(jié)點(diǎn)「miriad2b」上的進(jìn)程會(huì)被賦值為 Rank 1.
• x 是 Rank 0 打算發(fā)送到 Rank 1 的張量，通過(guò) dist.send(x, dst=1) 完成。
• z 是 Rank 1 在接收到張量之前就創(chuàng)建的東西。我們需要一個(gè)早就創(chuàng)建好的同維度的張量作為接收傳送來(lái)的張量的占位符。z 的值最終會(huì)被 x 替代。
• 與 dist.send(..) 類似，負(fù)責(zé)接收的對(duì)應(yīng)函數(shù)是 dist.recv(z, src=0)，它將張量接收到 z。

通信集體

我們?cè)谏弦徊糠挚吹降氖且粋€(gè)「點(diǎn)對(duì)點(diǎn)」通信的例子，在給定的環(huán)境中，rank(s) 將數(shù)據(jù)發(fā)送到特定的 rank(s)。盡管這種通信是有用的，因?yàn)樗鼘?duì)通信提供了細(xì)粒度的控制，但是還有其他被經(jīng)常使用的標(biāo)準(zhǔn)通信模式，叫作集體(collectives)。下面介紹了 Synchronous SGD 算法中我們感興趣的一個(gè)集體——all-reduce 集體。

1. ALL-REDUCE 集體

All-reduce 是一種同步通信方式，所有的 ranks 都被執(zhí)行了一個(gè) reduction 運(yùn)算，并且得到的結(jié)果對(duì)所有的 ranks 都是可見的。下圖介紹了這個(gè)思想(將求和作為 reduction 運(yùn)算)。

all-reduce 集體

def main(rank, world): 
    if rank == 0: 
        x = torch.tensor([1.]) 
    elif rank == 1: 
        x = torch.tensor([2.]) 
    elif rank == 2: 
        x = torch.tensor([-3.]) 
 
    dist.all_reduce(x, op=dist.reduce_op.SUM) 
    print('Rank {} has {}'.format(rank, x)) 
 
if __name__ == '__main__': 
    dist.init_process_group(backend='mpi') 
    main(dist.get_rank(), dist.get_world_size()) 

PyTorch 中 all-reduce 集體的基本用法

在 world of 3 環(huán)境中啟動(dòng)時(shí)，結(jié)果如下：

cluster@miriad2a:~/nfs$ mpiexec -n 3 -ppn 1 -hosts miriad2a,miriad2b,miriad2c python ptdist.py 
Rank 1 has tensor([0.]) 
Rank 0 has tensor([0.]) 
Rank 2 has tensor([0.]) 

• if rank == … elif 是我們?cè)诜植际接?jì)算中多次遇到的模式。在這個(gè)例子中，它被用來(lái)在不同的 rank 上創(chuàng)建張量。
• 它們一起執(zhí)行了 all-reduce(可以看見，dist.all_reduce(..) 在 if … elif block 邏輯塊的外部)，求和 (dist.reduce_op.SUM) 作為 reduction 運(yùn)算。
• 將來(lái)自每個(gè) rank 的 x 求和，再把得到的求和結(jié)果放置在每個(gè) rank 的 x 內(nèi)。

轉(zhuǎn)向深度學(xué)習(xí)

假設(shè)讀者熟知標(biāo)準(zhǔn)的隨機(jī)梯度下降算法(SGD)，該算法常用于訓(xùn)練深度學(xué)習(xí)模型。我們現(xiàn)在看到的是 SGD 的一個(gè)變體——同步 SGD(synchronous SGD)，它利用 all-reduce collective 來(lái)進(jìn)行擴(kuò)展。我們先從標(biāo)準(zhǔn) SGD 的數(shù)學(xué)公式開始吧。

其中 D 是一個(gè)樣本集合(mini-batch)，θ 是所有參數(shù)的集合，λ 是學(xué)習(xí)率，Loss(X, y) 是某個(gè)損失函數(shù)在 D 中所有樣本上的均值。

同步 SGD 所依賴的核心技巧是將更新規(guī)則中的求和在更小的 (mini)batch 子集上進(jìn)行分割。D 被分割成 R 個(gè)子集 D₁, D₂, . .(推薦每個(gè)子集具有相同數(shù)量的樣本)，所以將標(biāo)準(zhǔn)的 SGD 更新公式中的求和進(jìn)行分割，得到：

現(xiàn)在，因?yàn)樘荻人阕釉谇蠛退阕由鲜欠植际降?，所以我們得到?/p>

我們從中得到了什么?

看一下上面方程中單獨(dú)的梯度項(xiàng)(方括號(hào)里面)。它們現(xiàn)在可以被獨(dú)立地計(jì)算，然后加起來(lái)得到原始的梯度，而且沒有任何損失/近似。這就是數(shù)據(jù)并行。下面是整個(gè)過(guò)程：

• 將整個(gè)數(shù)據(jù)集分成 R 個(gè)等大的數(shù)據(jù)塊(子集)。這里的字母 R 代表的是 replica(副本)。
• 使用 MPI 啟動(dòng) R 個(gè)進(jìn)程/rank，將每個(gè)進(jìn)程綁定到一個(gè)數(shù)據(jù)塊上。
• 讓每個(gè) rank 使用大小為 B 的 mini-batch(dᵣ)(dᵣ來(lái)自該 rank 分配到的數(shù)據(jù)塊 D_r)計(jì)算梯度，即 rank r 計(jì)算。
• 將所有 rank 的梯度進(jìn)行求和，然后將得到的梯度對(duì)每個(gè) rank 可見，再進(jìn)行進(jìn)一步處理。

***一點(diǎn)就是 all-reduce 算法。所以，每次在所有 rank 使用大小為 B 的 mini-batch(dᵣ)計(jì)算完梯度以后，都必須執(zhí)行 all-reduce。需要注意的一點(diǎn)是，將全部 R 個(gè) rank(使用大小為 B 的 mini-batch 計(jì)算出)的梯度相加之后會(huì)得到一個(gè)有效的批大?。?/p>

下面是實(shí)現(xiàn)的關(guān)鍵部分(沒有展示樣板代碼)：

model = LeNet() 
# first synchronization of initial weights 
sync_initial_weights(model, rank, world_size) 
 
optimoptimizer = optim.SGD(model.parameters(), lr=1e-3, momentum=0.85) 
 
model.train() 
for epoch in range(1, epochs + 1): 
    for data, target in train_loader: 
        optimizer.zero_grad() 
        output = model(data) 
        loss = F.nll_loss(output, target) 
        loss.backward() 
 
        # The all-reduce on gradients 
        sync_gradients(model, rank, world_size) 
 
        optimizer.step() 
 
def sync_initial_weights(model, rank, world_size): 
    for param in model.parameters(): 
        if rank == 0: 
            # Rank 0 is sending it's own weight 
            # to all it's siblings (1 to world_size) 
            for sibling in range(1, world_size): 
                dist.send(param.data, dst=sibling) 
        else: 
            # Siblings must recieve the parameters 
            dist.recv(param.data, src=0) 
 
 
def sync_gradients(model, rank, world_size): 
    for param in model.parameters(): 
        dist.all_reduce(param.grad.data, op=dist.reduce_op.SUM) 

• 全部 R 個(gè) rank 都使用隨機(jī)權(quán)重創(chuàng)建自己的模型副本。
• 單個(gè)具有隨機(jī)權(quán)重的副本可能導(dǎo)致在初始的時(shí)候不同步。推薦在所有的副本上同步初始權(quán)重，sync_initial_weights(..) 就是在做這件事。讓任何一個(gè) rank 將自己的權(quán)重發(fā)送到它的兄弟 rank，兄弟 rank 必須接收這些權(quán)重并用來(lái)初始化它們自身。
• 從每個(gè) rank 對(duì)應(yīng)的數(shù)據(jù)部分取出一個(gè) mini-batch(大小為 B)，計(jì)算前向和反向傳遞(梯度)。作為配置的一部分，這里需要重點(diǎn)注意的一點(diǎn)是：所有的進(jìn)程/rank 應(yīng)該讓自己那部分?jǐn)?shù)據(jù)可見(通常是在自己的硬盤上或者在共享文件系統(tǒng)中)。
• 把求和作為 reduction 運(yùn)算，對(duì)每一個(gè)副本上的梯度執(zhí)行 all-reduce 集體。sync_gradients(..) 會(huì)完成梯度同步。
• 梯度同步之后，每個(gè)副本能夠在自己的權(quán)重上獨(dú)立地執(zhí)行標(biāo)準(zhǔn)的 SGD 更新。optimizer.step() 正常運(yùn)行。

現(xiàn)在問(wèn)題來(lái)了：我們?nèi)绾未_保獨(dú)立的更新保持同步?

我們看一下更新方程的***更新：

上面的第 2 點(diǎn)和第 4 點(diǎn)保證每個(gè)初始權(quán)重和梯度都是同步的。顯然，它們的線性組合也是同步的(λ 是常數(shù))。以后的所有更新都是類似的邏輯，因此也是同步的。

性能對(duì)比

所有分布式算法的***瓶頸就是同步。只有當(dāng)同步時(shí)間顯著小于計(jì)算時(shí)間的時(shí)候，分布式算法才是有益的。讓我們?cè)跇?biāo)準(zhǔn) SGD 和同步 SGD 之間做一個(gè)簡(jiǎn)單的對(duì)比，來(lái)看一下什么時(shí)候后者是比較好的。

定義：我們假設(shè)整個(gè)數(shù)據(jù)集的規(guī)模為 N。網(wǎng)絡(luò)處理大小為 B 的 mini-batch 需要花費(fèi)時(shí)間 Tcomp。在分布式情況下，all-reduce 同步花費(fèi)的時(shí)間為 Tsync。

對(duì)于非分布式(標(biāo)準(zhǔn))SGD，每個(gè) epoch 花費(fèi)的時(shí)間為：

對(duì)于同步 SGD，每個(gè) epoch 花費(fèi)的時(shí)間為：

因此，對(duì)于分布式環(huán)境，為了與非分布式環(huán)境相比有顯著的優(yōu)勢(shì)，我們需要滿足：

我們可以調(diào)整影響上述不等式的三個(gè)因子，從分布式算法中得到更多的好處。

• 通過(guò)以高帶寬的快速網(wǎng)絡(luò)連接節(jié)點(diǎn)，來(lái)減小 Tsync。
• 通過(guò)增加批大小 B，來(lái)增加 Tcomp。
• 通過(guò)連接更多的節(jié)點(diǎn)和擁有更多的副本來(lái)增加 R。

本文清晰地介紹了深度學(xué)習(xí)環(huán)境中的分布式計(jì)算的核心思想。盡管同步 SGD 很流行，但是也有其他被頻繁使用的分布式算法(如異步 SGD 及其變體)。然而，更重要的是能夠以并行的方式來(lái)思考深度學(xué)習(xí)方法。請(qǐng)注意，不是所有的算法都可以開箱即用地并行化，有的需要做一些近似處理，這破壞了原算法給出的理論保證。能否高效處理這些近似，取決于算法的設(shè)計(jì)者和實(shí)現(xiàn)者。

原文地址：

https://medium.com/intel-student-ambassadors/distributed-training-of-deep-learning-models-with-pytorch-1123fa538848

【本文是51CTO專欄機(jī)構(gòu)“機(jī)器之心”的原創(chuàng)譯文，微信公眾號(hào)“機(jī)器之心( id: almosthuman2014)”】 

戳這里，看該作者更多好文