一、NCCL介绍

1.1NCCL简介

1.2分布式训练策略

1.2.1数据并行

1.2.2模型并行

1.2.3流水并行

1.3通信原语

1.3.1Reduce

从多个sender接收数据，最终combine到一个节点上

从多个sender接收数据，最终combine到每一个结点上，前面提到的数据并行中反向传播就是利用AllReduce将各个设备中训练后的梯度统一后广播到各个设备中

1.3.2Ring-base collective

• 单机4卡通过同一个PCIe switch挂载在一颗CPU上，
• 单机8卡通过两个CPU下不同的PCIe switch挂载

1.3.3NCCL中的通信原语实现

1.4专有名词解释

1.4.1PCIe（高速串行总线）

1.4.2NVlink（NVIDIA的高速互连技术）

1.4.3Infiniband（无限带宽）

1.4.4MPI（并行计算Message Passing Interface）

二、NCCL源码阅读-初始化部分

2.1官方文档中的example代码

• 上述代码中，显示进行了一些MPI相关的操作，主要是为了在NCCL连接建立之前，在设备之间传输一些额外的信息，比如ncclUniquedId。然后是CUDA相关的一些操作：分配内存、创建stream等，其中涉及到NCCL的代码包括：ncclGetUniqueId、ncclCommInitRank和ncclAllReduce。 本文先针对ncclGetUniqueId进行探究

int main(int argc, char* argv[])
{
  int size = 32*1024*1024;
  int myRank, nRanks, localRank = 0;
  //initializing MPI
  MPICHECK(MPI_Init(&argc, &argv));
  MPICHECK(MPI_Comm_rank(MPI_COMM_WORLD, &myRank));
  MPICHECK(MPI_Comm_size(MPI_COMM_WORLD, &nRanks));
  //calculating localRank based on hostname which is used in selecting a GPU
  uint64_t hostHashs[nRanks];
  char hostname[1024];
  getHostName(hostname, 1024);
  hostHashs[myRank] = getHostHash(hostname);
  MPICHECK(MPI_Allgather(MPI_IN_PLACE, 0, MPI_DATATYPE_NULL, hostHashs, sizeof(uint64_t), MPI_BYTE, MPI_COMM_WORLD));
  for (int p=0; p<nRanks; p++) {
     if (p == myRank) break;
     if (hostHashs[p] == hostHashs[myRank]) localRank++;
  }
  ncclUniqueId id;
  ncclComm_t comm;
  float *sendbuff, *recvbuff;
  cudaStream_t s;
  //get NCCL unique ID at rank 0 and broadcast it to all others
  if (myRank == 0) ncclGetUniqueId(&id);
  MPICHECK(MPI_Bcast((void *)&id, sizeof(id), MPI_BYTE, 0, MPI_COMM_WORLD));
  //picking a GPU based on localRank, allocate device buffers
  CUDACHECK(cudaSetDevice(localRank));
  CUDACHECK(cudaMalloc(&sendbuff, size * sizeof(float)));
  CUDACHECK(cudaMalloc(&recvbuff, size * sizeof(float)));
  CUDACHECK(cudaStreamCreate(&s));
  //initializing NCCL
  NCCLCHECK(ncclCommInitRank(&comm, nRanks, id, myRank));
  //communicating using NCCL
  NCCLCHECK(ncclAllReduce((const void*)sendbuff, (void*)recvbuff, size, ncclFloat, ncclSum,
        comm, s));
  //completing NCCL operation by synchronizing on the CUDA stream
  CUDACHECK(cudaStreamSynchronize(s));
  //free device buffers
  CUDACHECK(cudaFree(sendbuff));
  CUDACHECK(cudaFree(recvbuff));
  //finalizing NCCL
  ncclCommDestroy(comm);
  //finalizing MPI
  MPICHECK(MPI_Finalize());
  printf("[MPI Rank %d] Success \n", myRank);
  return 0;
}

2.2ncclGetUniqueId[init.cc]

• 在doc的example代码中，如果是rank0，则会调用ncclGetUniqueId来获取id，并通过mpi来广播给其他rank

//get NCCL unique ID at rank 0 and broadcast it to all others
  if (myRank == 0) ncclGetUniqueId(&id);
  MPICHECK(MPI_Bcast((void *)&id, sizeof(id), MPI_BYTE, 0, MPI_COMM_WORLD));

• ncclGetUniqueId产生uniqueId

NCCL_API(ncclResult_t, ncclGetUniqueId, ncclUniqueId* out);
ncclResult_t ncclGetUniqueId(ncclUniqueId* out) {
  NCCLCHECK(ncclInit());
  NCCLCHECK(PtrCheck(out, "GetUniqueId", "out"));
  ncclResult_t res = bootstrapGetUniqueId((struct ncclBootstrapHandle*)out);
  TRACE_CALL("ncclGetUniqueId(0x%llx)", (unsigned long long)hashUniqueId(*out));
  return res;
}

2.3ncclInit

• nccl初始化时会调用ncclInit，为了防止重复初始化，加互斥量锁。然后会调用initEnv进行环境的初始化

pthread_mutex_t initLock = PTHREAD_MUTEX_INITIALIZER;
static bool initialized = false;

static ncclResult_t ncclInit() {
  if (__atomic_load_n(&initialized, __ATOMIC_ACQUIRE)) return ncclSuccess;
  pthread_mutex_lock(&initLock);
  if (!initialized) {
    initEnv();
    initGdrCopy();
    // Always initialize bootstrap network
    NCCLCHECK(bootstrapNetInit());
    // 网络插件
    NCCLCHECK(ncclNetPluginInit());
    initNvtxRegisteredEnums();
    __atomic_store_n(&initialized, true, __ATOMIC_RELEASE);
  }
  pthread_mutex_unlock(&initLock);
  return ncclSuccess;
}

2.4initEnv

• 初始化环境变量，通过解析nccl.conf文件，并使用setenv系统调用设置环境变量

void initEnv() {
  char confFilePath[1024];
  const char * userDir = userHomeDir();
  if (userDir) {
    sprintf(confFilePath, "%s/.nccl.conf", userDir);
    setEnvFile(confFilePath);
  }
  sprintf(confFilePath, "/etc/nccl.conf");
  setEnvFile(confFilePath);
}

2.5initGdrCopy

• GDR是指GPU Direct RDMA，两个显卡通过网卡直接访问内存。ncclGdrInit加载GDR动态库

// GDRCOPY support
gdr_t ncclGdrCopy = NULL;
ncclResult_t initGdrCopy() {
  if (ncclParamGdrCopyEnable() == 1) {
    ncclGdrCopy = ncclGdrInit();
  }
  return ncclSuccess;
}

2.6bootstrapNetInit

• nccl所需要两个网络，一个是bootstrap网络，一个是数据通信网络。
  • bootstrap网络主要用于初始化时交换一些简单的信息，如果每个机器的ip端口，由于数据量较小，而且只在初始化阶段执行一次，因此bootstrap使用的是tcp
  • 而通信网络是用于实际数据的传输，因此优先使用RDMA（支持GDR的话会优先GDR）
• 通过C库函数getenv来获取环境变量的值“NCCL_COMM_ID”，作为NCCL库中用于通信的标识符，NCCL_COMM_ID的格式为“nccl_socket:ip:port”，其中ip指定要使用的IP地址，port指定要使用的端口号。后续这个ID会作为该实例的uniqueId。

/* Init functions */
static char bootstrapNetIfName[MAX_IF_NAME_SIZE+1]; 
static union ncclSocketAddress bootstrapNetIfAddr; 
static int bootstrapNetInitDone = 0;
pthread_mutex_t bootstrapNetLock = PTHREAD_MUTEX_INITIALIZER;

ncclResult_t bootstrapNetInit() {
  if (bootstrapNetInitDone == 0) {
    pthread_mutex_lock(&bootstrapNetLock);
    if (bootstrapNetInitDone == 0) {
    
      char* env = getenv("NCCL_COMM_ID");
      if (env) {
        union ncclSocketAddress remoteAddr;
        if (ncclSocketGetAddrFromString(&remoteAddr, env) != ncclSuccess) {
          WARN("Invalid NCCL_COMM_ID, please use format: <ipv4>:<port> or [<ipv6>]:<port> or <hostname>:<port>");
          return ncclInvalidArgument;
        }
        if (ncclFindInterfaceMatchSubnet(bootstrapNetIfName, &bootstrapNetIfAddr, &remoteAddr, MAX_IF_NAME_SIZE, 1) <= 0) {
          WARN("NET/Socket : No usable listening interface found");
          return ncclSystemError;
        }
      } else {
        int nIfs = ncclFindInterfaces(bootstrapNetIfName, &bootstrapNetIfAddr, MAX_IF_NAME_SIZE, 1);
        if (nIfs <= 0) {
          WARN("Bootstrap : no socket interface found");
          return ncclInternalError;
        }
      }
      char line[SOCKET_NAME_MAXLEN+MAX_IF_NAME_SIZE+2];
      sprintf(line, " %s:", bootstrapNetIfName);
      ncclSocketToString(&bootstrapNetIfAddr, line+strlen(line));
      INFO(NCCL_INIT, "Bootstrap : Using%s", line);
      bootstrapNetInitDone = 1;
    }
    pthread_mutex_unlock(&bootstrapNetLock);
  }
  return ncclSuccess;
}

2.7bootstrapGetUniqueId

• 在bootstrap网络建立初期获取uniqueId的函数，这个uniqueId包含一个magic number用于通信时校验，还包含一个ncclSocketAddress结构体，其中包含一个地址族和地址数据，这个ncclSocketAddress是属于root gpu的，作为一个单机实例的标识id，其实就是当前机器的ip和port.

ncclResult_t bootstrapGetUniqueId(struct ncclBootstrapHandle* handle) {
  memset(handle, 0, sizeof(ncclBootstrapHandle));
  // magic在通信中起到验证作用
  NCCLCHECK(getRandomData(&handle->magic, sizeof(handle->magic)));

  char* env = getenv("NCCL_COMM_ID");
  if (env) {
    INFO(NCCL_ENV, "NCCL_COMM_ID set by environment to %s", env);
    if (ncclSocketGetAddrFromString(&handle->addr, env) != ncclSuccess) {
      WARN("Invalid NCCL_COMM_ID, please use format: <ipv4>:<port> or [<ipv6>]:<port> or <hostname>:<port>");
      return ncclInvalidArgument;
    }
  } else {
    memcpy(&handle->addr, &bootstrapNetIfAddr, sizeof(union ncclSocketAddress));
    NCCLCHECK(bootstrapCreateRoot(handle, false));
  }

  return ncclSuccess;
}