相關(guān)系列

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

概述

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

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

RING算法的步驟如下：

Scatter-Reduce

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

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

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

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

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

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

當(dāng)然，如果要做All-Reduce，此時(shí)不需要繼續(xù)按照原來的規(guī)則計(jì)算類，理論上只需要再算一次All-Gather，就能把藍(lán)色的塊分發(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;
}

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

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

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

來到scatter-reduce的實(shí)現(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：

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

NCCL_PROTO_LL：

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

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

最后是正式計(jì)算部分：

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

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

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

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

在實(shí)際運(yùn)行中，我們?cè)趆ost端的代碼只是規(guī)定計(jì)算流，當(dāng)這些定義好的原子操作加入到stream中去以后，就由固定的流來分配實(shí)際運(yùn)行的情況了。

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

 // 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));

以上就是整個(gè)scatter-reduce的流程。

相關(guān)系列

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