【分布式】小白看Ring算法

概述

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 可以处理到一部分数据块，从而提高整体的通信效率。

并行性： 划分 chunk 可以增加通信的并行性。每个 GPU 处理自己的数据块，不同的 GPU 可以并行地执行通信操作，从而提高整体的吞吐量。
减少延迟： 较小的数据块通常可以更快地传输，因此通过划分 chunk，可以减少每个通信操作的延迟。这对于一些对通信延迟敏感的应用程序是至关重要的。
资源分配： NCCL 可能会根据 GPU 的数量来分配适当的资源，例如内存等。通过划分 chunk，可以更好地管理这些资源。
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, nChannelsminChunkSizeLL128)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));