相關系列

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

概述

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

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

RING算法的步驟如下：

Scatter-Reduce

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

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

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

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

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

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

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

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;
}

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

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

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

來到scatter-reduce的實現部分：

		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協議的通信模式：
一共有三種，分別是NCCL_PROTO_SIMPLE、NCCL_PROTO_LL和NCCL_PROTO_LL128。

NCCL_PROTO_SIMPLE：

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

NCCL_PROTO_LL：

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

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

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 是一個偏移量變量，用于指定數據在通信緩沖區中的位置。
nelem 表示每次操作的元素個數，取 realChunkSize 和 size-chunkOffset 的較小值。
rankDest 是目標GPU的排名。

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

第2到第k-1步：
將數據在環形路徑上經過各個GPU節點，依次進行Reduce操作，并將結果復制到下一個GPU。
通過循環，依次計算目標GPU的排名 rankDest 和在通信緩沖區中的偏移量 offset。
調用 prims.recvReduceSend 函數，接收數據并執行Reduce操作，然后將結果發送到下一個GPU。

第k-1步：
將最后一個GPU的數據進行Reduce操作，得到最終的結果。
計算目標GPU的排名 rankDest 和在通信緩沖區中的偏移量 offset。
調用 prims.recvReduceCopy 函數，接收數據并執行Reduce操作，然后將結果復制到指定的位置，最終產生最終的ReduceScatter結果。

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

加入Barria，在本地（intra-node）執行一個屏障操作，確保同一節點內的所有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的流程。

相關系列

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