NCCL源码解析①：初始化及ncclUniqueId的产生
创建时间：2023-07-25 11:04
NCCL是英伟达开源的GPU通信库，支持集合通信和点对点通信。
作者｜KIDGINBROOK
更新｜潘丽晨
int e = cmd; \if( e != MPI_SUCCESS ) { \printf("Failed: MPI error %s:%d '%d'\n", \__FILE__,__LINE__, e); \exit(EXIT_FAILURE); \} \} while(0)cudaError_t e = cmd; \if( e != cudaSuccess ) { \printf("Failed: Cuda error %s:%d '%s'\n", \__FILE__,__LINE__,cudaGetErrorString(e)); \exit(EXIT_FAILURE); \} \} while(0)ncclResult_t r = cmd; \if (r!= ncclSuccess) { \printf("Failed, NCCL error %s:%d '%s'\n", \__FILE__,__LINE__,ncclGetErrorString(r)); \exit(EXIT_FAILURE); \} \} while(0)static uint64_t getHostHash(const char* string) {// Based on DJB2a, result = result * 33 ^ charuint64_t result = 5381;for (int c = 0; string[c] != '\0'; c++){result = ((result << 5) + result) ^ string[c];}return result;}static void getHostName(char* hostname, int maxlen) {gethostname(hostname, maxlen);for (int i=0; i< maxlen; i++) {if (hostname[i] == '.') {hostname[i] = '\0';return;}}}int main(int argc, char* argv[]){int size = 32*1024*1024;int myRank, nRanks, localRank = 0;//initializing MPIMPICHECK(MPI_Init(&argc, &argv));MPICHECK(MPI_Comm_rank(MPI_COMM_WORLD, &myRank));MPICHECK(MPI_Comm_size(MPI_COMM_WORLD, &nRanks));//calculating localRank which is used in selecting a GPUuint64_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++;}//each process is using two GPUsint nDev = 2;float** sendbuff = (float**)malloc(nDev * sizeof(float*));float** recvbuff = (float**)malloc(nDev * sizeof(float*));cudaStream_t* s = (cudaStream_t*)malloc(sizeof(cudaStream_t)*nDev);//picking GPUs based on localRankfor (int i = 0; i < nDev; ++i) {CUDACHECK(cudaSetDevice(localRank*nDev + i));CUDACHECK(cudaMalloc(sendbuff + i, size * sizeof(float)));CUDACHECK(cudaMalloc(recvbuff + i, size * sizeof(float)));CUDACHECK(cudaMemset(sendbuff[i], 1, size * sizeof(float)));CUDACHECK(cudaMemset(recvbuff[i], 0, size * sizeof(float)));CUDACHECK(cudaStreamCreate(s+i));}ncclUniqueId id;ncclComm_t comms[nDev];//generating NCCL unique ID at one process and broadcasting it to allif (myRank == 0) ncclGetUniqueId(&id);MPICHECK(MPI_Bcast((void *)&id, sizeof(id), MPI_BYTE, 0, MPI_COMM_WORLD));//initializing NCCL, group API is required around ncclCommInitRank as it is//called across multiple GPUs in each thread/processNCCLCHECK(ncclGroupStart());for (int i=0; i<nDev; i++) {CUDACHECK(cudaSetDevice(localRank*nDev + i));NCCLCHECK(ncclCommInitRank(comms+i, nRanks*nDev, id, myRank*nDev + i));}NCCLCHECK(ncclGroupEnd());//calling NCCL communication API. Group API is required when using//multiple devices per thread/processNCCLCHECK(ncclGroupStart());for (int i=0; i<nDev; i++)NCCLCHECK(ncclAllReduce((const void*)sendbuff[i], (void*)recvbuff[i], size, ncclFloat, ncclSum,comms[i], s[i]));NCCLCHECK(ncclGroupEnd());//synchronizing on CUDA stream to complete NCCL communicationfor (int i=0; i<nDev; i++)CUDACHECK(cudaStreamSynchronize(s[i]));//freeing device memoryfor (int i=0; i<nDev; i++) {CUDACHECK(cudaFree(sendbuff[i]));CUDACHECK(cudaFree(recvbuff[i]));}//finalizing NCCLfor (int i=0; i<nDev; i++) {ncclCommDestroy(comms[i]);}//finalizing MPIMPICHECK(MPI_Finalize());printf("[MPI Rank %d] Success \n", myRank);return 0;}
ncclResult_t ncclGetUniqueId(ncclUniqueId* out) {NCCLCHECK(ncclInit());NCCLCHECK(PtrCheck(out, "GetUniqueId", "out"));return bootstrapGetUniqueId(out);}
ncclResult_t initNet() {// Always initialize bootstrap networkNCCLCHECK(bootstrapNetInit());NCCLCHECK(initNetPlugin(&ncclNet, &ncclCollNet));if (ncclNet != NULL) return ncclSuccess;if (initNet(&ncclNetIb) == ncclSuccess) {ncclNet = &ncclNetIb;} else {NCCLCHECK(initNet(&ncclNetSocket));ncclNet = &ncclNetSocket;}return ncclSuccess;}
static int findInterfaces(const char* prefixList, char* names, union socketAddress *addrs, int sock_family, int maxIfNameSize, int maxIfs) {struct netIf userIfs[MAX_IFS];bool searchNot = prefixList && prefixList[0] == '^';if (searchNot) prefixList++;bool searchExact = prefixList && prefixList[0] == '=';if (searchExact) prefixList++;int nUserIfs = parseStringList(prefixList, userIfs, MAX_IFS);int found = 0;struct ifaddrs *interfaces, *interface;getifaddrs(&interfaces);for (interface = interfaces; interface && found < maxIfs; interface = interface->ifa_next) {if (interface->ifa_addr == NULL) continue;int family = interface->ifa_addr->sa_family;if (family != AF_INET && family != AF_INET6)continue;if (sock_family != -1 && family != sock_family)continue;if (family == AF_INET6) {struct sockaddr_in6* sa = (struct sockaddr_in6*)(interface->ifa_addr);if (IN6_IS_ADDR_LOOPBACK(&sa->sin6_addr)) continue;}if (!(matchIfList(interface->ifa_name, -1, userIfs, nUserIfs, searchExact) ^ searchNot)) {continue;}bool duplicate = false;for (int i = 0; i < found; i++) {if (strcmp(interface->ifa_name, names+i*maxIfNameSize) == 0) { duplicate = true; break; }}if (!duplicate) {strncpy(names+found*maxIfNameSize, interface->ifa_name, maxIfNameSize);int salen = (family == AF_INET) ? sizeof(sockaddr_in) : sizeof(sockaddr_in6);memcpy(addrs+found, interface->ifa_addr, salen);found++;}}freeifaddrs(interfaces);return found;}
ncclResult_t ncclIbInit(ncclDebugLogger_t logFunction) {static int shownIbHcaEnv = 0;if(wrap_ibv_symbols() != ncclSuccess) { return ncclInternalError; }if (ncclParamIbDisable()) return ncclInternalError;if (ncclNIbDevs == -1) {pthread_mutex_lock(&ncclIbLock);wrap_ibv_fork_init();if (ncclNIbDevs == -1) {ncclNIbDevs = 0;if (findInterfaces(ncclIbIfName, &ncclIbIfAddr, MAX_IF_NAME_SIZE, 1) != 1) {WARN("NET/IB : No IP interface found.");return ncclInternalError;}// Detect IB cardsint nIbDevs;struct ibv_device** devices;// Check if user defined which IB device:port to usechar* userIbEnv = getenv("NCCL_IB_HCA");if (userIbEnv != NULL && shownIbHcaEnv++ == 0) INFO(NCCL_NET|NCCL_ENV, "NCCL_IB_HCA set to %s", userIbEnv);struct netIf userIfs[MAX_IB_DEVS];bool searchNot = userIbEnv && userIbEnv[0] == '^';if (searchNot) userIbEnv++;bool searchExact = userIbEnv && userIbEnv[0] == '=';if (searchExact) userIbEnv++;int nUserIfs = parseStringList(userIbEnv, userIfs, MAX_IB_DEVS);if (ncclSuccess != wrap_ibv_get_device_list(&devices, &nIbDevs)) return ncclInternalError;for (int d=0; d<nIbDevs && ncclNIbDevs<MAX_IB_DEVS; d++) {struct ibv_context * context;if (ncclSuccess != wrap_ibv_open_device(&context, devices[d]) || context == NULL) {WARN("NET/IB : Unable to open device %s", devices[d]->name);continue;}int nPorts = 0;struct ibv_device_attr devAttr;memset(&devAttr, 0, sizeof(devAttr));if (ncclSuccess != wrap_ibv_query_device(context, &devAttr)) {WARN("NET/IB : Unable to query device %s", devices[d]->name);if (ncclSuccess != wrap_ibv_close_device(context)) { return ncclInternalError; }continue;}for (int port = 1; port <= devAttr.phys_port_cnt; port++) {struct ibv_port_attr portAttr;if (ncclSuccess != wrap_ibv_query_port(context, port, &portAttr)) {WARN("NET/IB : Unable to query port %d", port);continue;}if (portAttr.state != IBV_PORT_ACTIVE) continue;if (portAttr.link_layer != IBV_LINK_LAYER_INFINIBAND&& portAttr.link_layer != IBV_LINK_LAYER_ETHERNET) continue;// check against user specified HCAs/portsif (! (matchIfList(devices[d]->name, port, userIfs, nUserIfs, searchExact) ^ searchNot)) {continue;}TRACE(NCCL_INIT|NCCL_NET,"NET/IB: [%d] %s:%d/%s ", d, devices[d]->name, port,portAttr.link_layer == IBV_LINK_LAYER_INFINIBAND ? "IB" : "RoCE");ncclIbDevs[ncclNIbDevs].device = d;ncclIbDevs[ncclNIbDevs].guid = devAttr.sys_image_guid;ncclIbDevs[ncclNIbDevs].port = port;ncclIbDevs[ncclNIbDevs].link = portAttr.link_layer;ncclIbDevs[ncclNIbDevs].speed = ncclIbSpeed(portAttr.active_speed) * ncclIbWidth(portAttr.active_width);ncclIbDevs[ncclNIbDevs].context = context;strncpy(ncclIbDevs[ncclNIbDevs].devName, devices[d]->name, MAXNAMESIZE);NCCLCHECK(ncclIbGetPciPath(ncclIbDevs[ncclNIbDevs].devName, &ncclIbDevs[ncclNIbDevs].pciPath, &ncclIbDevs[ncclNIbDevs].realPort));ncclIbDevs[ncclNIbDevs].maxQp = devAttr.max_qp;ncclNIbDevs++;nPorts++;pthread_create(&ncclIbAsyncThread, NULL, ncclIbAsyncThreadMain, context);}if (nPorts == 0 && ncclSuccess != wrap_ibv_close_device(context)) { return ncclInternalError; }}if (nIbDevs && (ncclSuccess != wrap_ibv_free_device_list(devices))) { return ncclInternalError; };}if (ncclNIbDevs == 0) {INFO(NCCL_INIT|NCCL_NET, "NET/IB : No device found.");} else {char line[1024];line[0] = '\0';for (int d=0; d<ncclNIbDevs; d++) {snprintf(line+strlen(line), 1023-strlen(line), " [%d]%s:%d/%s", d, ncclIbDevs[d].devName,ncclIbDevs[d].port, ncclIbDevs[d].link == IBV_LINK_LAYER_INFINIBAND ? "IB" : "RoCE");}line[1023] = '\0';char addrline[1024];INFO(NCCL_INIT|NCCL_NET, "NET/IB : Using%s ; OOB %s:%s", line, ncclIbIfName, socketToString(&ncclIbIfAddr.sa, addrline));}pthread_mutex_unlock(&ncclIbLock);}return ncclSuccess;}
ncclResult_t bootstrapCreateRoot(ncclUniqueId* id, bool idFromEnv) {ncclNetHandle_t* netHandle = (ncclNetHandle_t*) id;void* listenComm;NCCLCHECK(bootstrapNetListen(idFromEnv ? dontCareIf : 0, netHandle, &listenComm));pthread_t thread;pthread_create(&thread, NULL, bootstrapRoot, listenComm);return ncclSuccess;}
static ncclResult_t bootstrapNetListen(int dev, ncclNetHandle_t* netHandle, void** listenComm) {union socketAddress* connectAddr = (union socketAddress*) netHandle;static_assert(sizeof(union socketAddress) < NCCL_NET_HANDLE_MAXSIZE, "union socketAddress size is too large");// if dev >= 0, listen based on devif (dev >= 0) {NCCLCHECK(bootstrapNetGetSocketAddr(dev, connectAddr));} else if (dev == findSubnetIf) {...} // Otherwise, handle stores a local addressstruct bootstrapNetComm* comm;NCCLCHECK(bootstrapNetNewComm(&comm));NCCLCHECK(createListenSocket(&comm->fd, connectAddr));*listenComm = comm;return ncclSuccess;}
static ncclResult_t bootstrapNetGetSocketAddr(int dev, union socketAddress* addr) {if (dev >= bootstrapNetIfs) return ncclInternalError;memcpy(addr, bootstrapNetIfAddrs+dev, sizeof(*addr));return ncclSuccess;}
struct bootstrapNetComm {int fd;};
static ncclResult_t createListenSocket(int *fd, union socketAddress *localAddr) {/* IPv4/IPv6 support */int family = localAddr->sa.sa_family;int salen = (family == AF_INET) ? sizeof(sockaddr_in) : sizeof(sockaddr_in6);/* Create socket and bind it to a port */int sockfd = socket(family, SOCK_STREAM, 0);if (sockfd == -1) {WARN("Net : Socket creation failed : %s", strerror(errno));return ncclSystemError;}if (socketToPort(&localAddr->sa)) {// Port is forced by env. Make sure we get the port.int opt = 1;SYSCHECK(setsockopt(sockfd, SOL_SOCKET, SO_REUSEADDR | SO_REUSEPORT, &opt, sizeof(opt)), "setsockopt");SYSCHECK(setsockopt(sockfd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)), "setsockopt");}// localAddr port should be 0 (Any port)SYSCHECK(bind(sockfd, &localAddr->sa, salen), "bind");/* Get the assigned Port */socklen_t size = salen;SYSCHECK(getsockname(sockfd, &localAddr->sa, &size), "getsockname");char line[1024];TRACE(NCCL_INIT|NCCL_NET,"Listening on socket %s", socketToString(&localAddr->sa, line));/* Put the socket in listen mode* NB: The backlog will be silently truncated to the value in /proc/sys/net/core/somaxconn*/SYSCHECK(listen(sockfd, 16384), "listen");*fd = sockfd;return ncclSuccess;}
（本文经授权后由OneFlow发布。原文：https://blog.csdn.net/KIDGIN7439/article/details/126712106?spm=1001.2014.3001.5502）
其他人都在看 点击“阅读原文”，欢迎Star、试用OneFlow新版本
넶浏览量：0
-
RTX PRO 5000 Blackwell：专业桌面算力巅峰，英伟达显卡总代宽恒科技赋能产业 AI 升级
2026 年生成式 AI 与专业创意产业迎来算力升级浪潮，本地 AI 开发、多模态内容生成、工业 3D 设计、影视渲染等场景对桌面端高性能专业显卡需求激增。NVIDIA RTX PRO 5000 Blackwell 作为英伟达最新一代专业桌面 GPU，基于 Blackwell 架构打造，融合 AI 算力、图形渲染与专业稳定性，成为专业人士与中小企业的首选算力设备。宽恒科技作为英伟达显卡核心总代与 NPN Elite 精英级代理，深耕专业显卡领域，依托正品保障、优先供货、原厂技术支持与全栈服务体系，为企业与专业用户提供 RTX PRO 5000 Blackwell 全流程解决方案，赋能本地 AI 开发与专业创意工作流升级，推动产业数字化创新。
넶0 2026-05-22 -
桌面 AI 超级计算机，重构本地大模型开发新范式，宽恒科技赋能个人与中小企业 AI 创新
2026 年生成式 AI 进入 “本地部署” 黄金时代，大模型从云端向桌面端下沉，个人开发者、中小企业对本地高性能 AI 算力需求激增。传统 AI 服务器体积庞大、价格高昂，云端算力存在数据隐私风险与网络延迟问题，难以匹配本地开发需求。NVIDIA DGX Spark 作为全球首款桌面级 AI 超级计算机，基于 Grace Blackwell 架构打造，将超算级算力浓缩至桌面尺寸，支持本地运行千亿参数大模型，彻底打破本地大模型开发的算力瓶颈NVIDIA 英伟达。宽恒科技紧跟 AI 算力下沉趋势，依托英伟达官方合作资源，深耕 DGX Spark 技术服务领域，为个人开发者、中小企业提供产品供应、技术支持与定制化解决方案，赋能本地 AI 创新，推动普惠 AI 发展。
넶0 2026-05-22 -
HTC VIVE Focus Vision 与 VIVE Cosmos 技术解析：XR 技术革新，宽恒科技赋能行业沉浸式应用
2026 年 XR（扩展现实）技术正从消费级娱乐向企业级应用深度渗透，成为空间计算、数字孪生、远程协作、工业培训等领域的核心支撑。HTC VIVE 作为全球 XR 技术领军品牌，凭借多年技术积累与创新能力，推出 VIVE Focus Vision 与 VIVE Cosmos 两款标杆级产品，分别定位高端企业级 XR 一体机与模块化 VR 系统，覆盖不同应用场景，引领 XR 技术发展方向。
넶0 2026-05-22 -
英伟达授权生态全解析：NPN、NVAIE 与 Elite 精英代理，宽恒科技引领产业算力服务升级
2026 年 AI 产业进入规模化落地关键期，英伟达作为全球算力基础设施龙头，其授权体系已成为连接技术、产品与市场的核心纽带。从 NPN 合作伙伴网络到 Elite 精英级别代理，从 NVAIE 认证到 NVIDIA AI Enterprise 软件授权，从数据中心解决方案授权到显卡总代体系，英伟达构建了层级清晰、权责明确、技术赋能的生态体系。宽恒科技深耕英伟达生态多年，凭借技术实力、服务能力与行业资源，成为英伟达授权体系核心参与者，依托全栈授权资质，为企业提供正品保障、原厂技术支持与定制化解决方案，推动英伟达技术在各行业深度应用，助力中国 AI 产业突破算力瓶颈、实现高效升级。
넶0 2026-05-22 -
算力租赁、GPU 集群与 AI 服务器：英伟达生态驱动产业算力升级，宽恒科技赋能企业 AI 转型
在生成式 AI 与大模型爆发的 2026 年，算力已成为数字经济的核心生产力。从千亿参数大模型训练到多模态 AI 推理，从自动驾驶仿真到医疗基因测序，算力需求呈指数级增长，传统算力模式难以匹配产业发展节奏。算力租赁、GPU 集群与 AI 服务器构成的新型算力体系，正成为企业突破算力瓶颈的关键路径，而英伟达凭借完整技术生态主导产业方向，宽恒科技深耕算力服务领域，依托英伟达技术与资源优势，为企业提供全栈算力解决方案，推动 AI 产业高效落地与创新升级。
넶0 2026-05-22 -
RTX PRO 5000、英伟达 pro 5000、pro 5000 blackwell、英伟达显卡总代 —— 宽恒科技赋能专业桌面算力新巅峰
2026 年专业可视化与本地 AI 开发需求爆发，RTX PRO 5000 Blackwell 作为英伟达推出的旗舰级专业显卡，以 Blackwell 架构、超大显存与强劲算力，成为专业设计与本地 AI 开发的核心硬件，宽恒科技作为英伟达显卡总代，依托顶级资质与供应链优势，为用户提供正品保障与全栈服务。
넶2 2026-05-21
