相關(guān)系列

【分布式】NCCL部署與測試 - 01
【分布式】入門級NCCL多機并行實踐 - 02
【分布式】小白看Ring算法 - 03
【分布式】大模型分布式訓(xùn)練入門與實踐 - 04

概述

NCCL（NVIDIA Collective Communications Library）是由NVIDIA開發(fā)的一種用于多GPU間通信的庫。NCCL的RING算法是NCCL庫中的一種通信算法，用于在多個GPU之間進行環(huán)形通信。

RING算法的基本思想是將多個GPU連接成一個環(huán)形結(jié)構(gòu)，每個GPU與相鄰的兩個GPU進行通信。數(shù)據(jù)沿著環(huán)形結(jié)構(gòu)傳遞，直到到達發(fā)送方的位置。這樣的環(huán)形結(jié)構(gòu)可以有效地利用GPU之間的帶寬，提高通信的效率。

RING算法的步驟如下：

Scatter-Reduce

以Scatter-Reduce為例，假設(shè)有4張GPU，RANK_NUM=4。
則需要根據(jù)RANK_NUM把每張CPU劃分為4個chunk。
為什么要這么劃分？

在 NCCL 中，劃分 chunk 的數(shù)量與 GPU 的數(shù)量相關(guān)聯(lián)，這是因為 chunk 的目的是將大的消息劃分為多個小的數(shù)據(jù)塊，以便并行處理和降低通信的延遲。這種劃分通常會基于 GPU 的數(shù)量，以確保每個 GPU 可以處理到一部分數(shù)據(jù)塊，從而提高整體的通信效率。

1. 并行性： 劃分 chunk 可以增加通信的并行性。每個 GPU 處理自己的數(shù)據(jù)塊，不同的 GPU 可以并行地執(zhí)行通信操作，從而提高整體的吞吐量。
2. 減少延遲： 較小的數(shù)據(jù)塊通常可以更快地傳輸，因此通過劃分 chunk，可以減少每個通信操作的延遲。這對于一些對通信延遲敏感的應(yīng)用程序是至關(guān)重要的。
3. 資源分配： NCCL 可能會根據(jù) GPU 的數(shù)量來分配適當?shù)馁Y源，例如內(nèi)存等。通過劃分 chunk，可以更好地管理這些資源。
4. Load Balancing： 均衡負載是分布式系統(tǒng)中的一個關(guān)鍵問題。通過根據(jù) GPU 的數(shù)量劃分 chunk，可以更容易地實現(xiàn)負載均衡，確保每個 GPU 處理的工作量相對均勻。

劃分了chunk以后，我們一次RING的通路將會走通4塊GPU，每次只傳輸一塊chunk的數(shù)據(jù)。這樣需要走很多次通路才能把所有數(shù)據(jù)傳輸完。
假如 ringIx=0，第一次循環(huán)到第三次循環(huán)時：

我們將綠色視為這次循環(huán)需要傳輸?shù)臄?shù)據(jù)。
數(shù)據(jù)ABCD在不同的GPU中流通。
最終達到以下情況，scatter-reduce就完成了：

將圖中藍色部分輸出，就完成了一次ring算法下的Scatter-Reduce。

當然，如果要做All-Reduce，此時不需要繼續(xù)按照原來的規(guī)則計算類，理論上只需要再算一次All-Gather，就能把藍色的塊分發(fā)給其他幾塊卡。All-Reduce的相關(guān)講解網(wǎng)絡(luò)上很多。此處就不講了。

NCCL代碼流程

fillInfo:
這段代碼在init.cc中

static ncclResult_t fillInfo(struct ncclComm* comm, struct ncclPeerInfo* info, uint64_t commHash) {info->rank = comm->rank;CUDACHECK(cudaGetDevice(&info->cudaDev));info->hostHash=getHostHash()+commHash;info->pidHash=getPidHash()+commHash;// Get the device MAJOR:MINOR of /dev/shm so we can use that// information to decide whether we can use SHM for inter-process// communication in a container environmentstruct stat statbuf;SYSCHECK(stat("/dev/shm", &statbuf), "stat");info->shmDev = statbuf.st_dev;info->busId = comm->busId;NCCLCHECK(ncclGpuGdrSupport(&info->gdrSupport));return ncclSuccess;
}

這段代碼的目的是為了獲取和存儲與通信相關(guān)的信息，以便在NCCL通信中使用。其中包括設(shè)備標識、主機哈希、進程ID哈希、共享內(nèi)存設(shè)備標識、總線ID以及對GDR的支持情況等。

在initTransportsRank中，搜索完信息并作第一次AllGather, 收集所有通信節(jié)點的信息。
然后再為通信組分配額外的內(nèi)存，以存儲每個通信節(jié)點的信息（包括一個額外的用于表示CollNet root的位置）。
遍歷節(jié)點和復(fù)制信息時，需要檢查是否存在相同主機哈希和總線ID的重復(fù)GPU。如果是，發(fā)出警告并返回ncclInvalidUsage錯誤。

后面的一系列過程就是計算路徑，然后這里涉及一些搜索算法，通常會將BFS搜索到的路徑都存在一個位置，選擇更優(yōu)的路徑。
搜索時也會根據(jù)實際情況判斷選擇ring算法或者tree算法。
搜索內(nèi)容可能是無窮的，因此NCCL設(shè)置了一個超時時間，超過該時間則終端搜索。
完成路徑的計算后，再做一次AllGather。

來到scatter-reduce的實現(xiàn)部分：

		size_t realChunkSize;if (Proto::Id == NCCL_PROTO_SIMPLE) {realChunkSize = min(chunkSize, divUp(size-gridOffset, nChannels));realChunkSize = roundUp(realChunkSize, (nthreads-WARP_SIZE)*sizeof(uint64_t)/sizeof(T));}else if (Proto::Id == NCCL_PROTO_LL)realChunkSize = size-gridOffset < loopSize ? args->coll.lastChunkSize : chunkSize;else if (Proto::Id == NCCL_PROTO_LL128)realChunkSize = min(divUp(size-gridOffset, nChannels*minChunkSizeLL128)*minChunkSizeLL128, chunkSize);realChunkSize = int(realChunkSize);ssize_t chunkOffset = gridOffset + bid*int(realChunkSize);

這里涉及了NCCL協(xié)議的通信模式：
一共有三種，分別是NCCL_PROTO_SIMPLE、NCCL_PROTO_LL和NCCL_PROTO_LL128。

NCCL_PROTO_SIMPLE：

描述： 使用簡單的通信協(xié)議。
差異點： 計算realChunkSize時，采用了一些特殊的邏輯，其中min(chunkSize, divUp(size-gridOffset, nChannels))用于確定實際的塊大小，并通過roundUp調(diào)整為合適的大小。這可能涉及到性能和資源的考慮，以及對通信模式的調(diào)整。

NCCL_PROTO_LL：

描述： 使用連續(xù)鏈表（Linked List，LL）的通信協(xié)議。
差異點： 在計算realChunkSize時，首先檢查size-gridOffset < loopSize條件，如果為真，則使用args->coll.lastChunkSize，否則使用默認的chunkSize。這可能與LL協(xié)議的特性有關(guān)，具體考慮了循環(huán)的情況。
NCCL_PROTO_LL128：

描述： 使用連續(xù)鏈表的通信協(xié)議，每次傳輸128字節(jié)。
差異點： 計算realChunkSize時，采用了min(divUp(size-gridOffset, nChannels*minChunkSizeLL128)*minChunkSizeLL128, chunkSize)的邏輯。這考慮了128字節(jié)的限制，以及對通信塊大小的一些限制。
總體來說，這三種協(xié)議模式的區(qū)別主要體現(xiàn)在計算realChunkSize的邏輯上，這可能受到性能、資源利用、通信模式等方面的不同考慮。具體選擇哪種協(xié)議模式通常取決于系統(tǒng)的特性和應(yīng)用場景的需求。

Protocol Mode	Description	Calculation of realChunkSize
NCCL_PROTO_SIMPLE	Uses a simple communication protocol.	realChunkSize = roundUp(min(chunkSize, divUp(size-gridOffset, nChannels)), (nthreads-WARP_SIZE)*sizeof(uint64_t)/sizeof(T))
NCCL_PROTO_LL	Uses a linked list (LL) communication protocol.	realChunkSize = size-gridOffset < loopSize ? args->coll.lastChunkSize : chunkSize
NCCL_PROTO_LL128	Uses a linked list (LL) communication protocol, with each transfer involving 128 bytes.	realChunkSize = min(divUp(size-gridOffset, nChannels*minChunkSizeLL128)*minChunkSizeLL128, chunkSize)

最后是正式計算部分：

 /////////////// begin ReduceScatter steps ///////////////ssize_t offset;int nelem = min(realChunkSize, size-chunkOffset);int rankDest;// step 0: push data to next GPUrankDest = ringRanks[nranks-1];offset = chunkOffset + rankDest * size;prims.send(offset, nelem);// k-2 steps: reduce and copy to next GPUfor (int j=2; j<nranks; ++j) {rankDest = ringRanks[nranks-j];offset = chunkOffset + rankDest * size;prims.recvReduceSend(offset, nelem);}// step k-1: reduce this buffer and data, which will produce the final resultrankDest = ringRanks[0];offset = chunkOffset + rankDest * size;prims.recvReduceCopy(offset, chunkOffset, nelem, /*postOp=*/true);

ssize_t offset; int nelem = min(realChunkSize, size-chunkOffset); int rankDest;：

offset 是一個偏移量變量，用于指定數(shù)據(jù)在通信緩沖區(qū)中的位置。
nelem 表示每次操作的元素個數(shù)，取 realChunkSize 和 size-chunkOffset 的較小值。
rankDest 是目標GPU的排名。

第一步：將數(shù)據(jù)推送到下一個GPU。
計算目標GPU的排名 rankDest 和在通信緩沖區(qū)中的偏移量 offset。
調(diào)用 prims.send 函數(shù)，將數(shù)據(jù)從當前GPU發(fā)送到目標GPU。
// k-2 steps: reduce and copy to next GPU：

第2到第k-1步：
將數(shù)據(jù)在環(huán)形路徑上經(jīng)過各個GPU節(jié)點，依次進行Reduce操作，并將結(jié)果復(fù)制到下一個GPU。
通過循環(huán)，依次計算目標GPU的排名 rankDest 和在通信緩沖區(qū)中的偏移量 offset。
調(diào)用 prims.recvReduceSend 函數(shù)，接收數(shù)據(jù)并執(zhí)行Reduce操作，然后將結(jié)果發(fā)送到下一個GPU。

第k-1步：
將最后一個GPU的數(shù)據(jù)進行Reduce操作，得到最終的結(jié)果。
計算目標GPU的排名 rankDest 和在通信緩沖區(qū)中的偏移量 offset。
調(diào)用 prims.recvReduceCopy 函數(shù)，接收數(shù)據(jù)并執(zhí)行Reduce操作，然后將結(jié)果復(fù)制到指定的位置，最終產(chǎn)生最終的ReduceScatter結(jié)果。

在實際運行中，我們在host端的代碼只是規(guī)定計算流，當這些定義好的原子操作加入到stream中去以后，就由固定的流來分配實際運行的情況了。

加入Barria，在本地（intra-node）執(zhí)行一個屏障操作，確保同一節(jié)點內(nèi)的所有GPU都達到了同步點。

 // Compute time models for algorithm and protocol combinationsNCCLCHECK(ncclTopoTuneModel(comm, minCompCap, maxCompCap, &treeGraph, &ringGraph, &collNetGraph));// Compute nChannels per peer for p2pNCCLCHECK(ncclTopoComputeP2pChannels(comm));if (ncclParamNvbPreconnect()) {// Connect p2p when using NVB pathint nvbNpeers;int* nvbPeers;NCCLCHECK(ncclTopoGetNvbGpus(comm->topo, comm->rank, &nvbNpeers, &nvbPeers));for (int r=0; r<nvbNpeers; r++) {int peer = nvbPeers[r];int delta = (comm->nRanks + (comm->rank-peer)) % comm->nRanks;for (int c=0; c<comm->p2pnChannelsPerPeer; c++) {int channelId = (delta+comm->p2pChannels[c]) % comm->p2pnChannels;if (comm->channels[channelId].peers[peer].recv[0].connected == 0) { // P2P uses only 1 connectorcomm->connectRecv[peer] |= (1<<channelId);}}delta = (comm->nRanks - (comm->rank-peer)) % comm->nRanks;for (int c=0; c<comm->p2pnChannelsPerPeer; c++) {int channelId = (delta+comm->p2pChannels[c]) % comm->p2pnChannels;if (comm->channels[channelId].peers[peer].send[0].connected == 0) { // P2P uses only 1 connectorcomm->connectSend[peer] |= (1<<channelId);}}}NCCLCHECK(ncclTransportP2pSetup(comm, NULL, 0));free(nvbPeers);}NCCLCHECK(ncclCommSetIntraProc(comm, intraProcRank, intraProcRanks, intraProcRank0Comm));/* Local intra-node barrier */NCCLCHECK(bootstrapBarrier(comm->bootstrap, comm->intraNodeGlobalRanks, intraNodeRank, intraNodeRanks, (int)intraNodeRank0pidHash));if (comm->nNodes) NCCLCHECK(ncclProxyCreate(comm));

以上就是整個scatter-reduce的流程。

相關(guān)系列

【分布式】NCCL部署與測試 - 01
【分布式】入門級NCCL多機并行實踐 - 02
【分布式】小白看Ring算法 - 03
【分布式】大模型分布式訓(xùn)練入門與實踐 - 04