您好,登錄后才能下訂單哦!
今天小編給大家分享一下NCCL Bootstrap網(wǎng)絡(luò)連接建立的方法是什么的相關(guān)知識(shí)點(diǎn),內(nèi)容詳細(xì),邏輯清晰,相信大部分人都還太了解這方面的知識(shí),所以分享這篇文章給大家參考一下,希望大家閱讀完這篇文章后有所收獲,下面我們一起來(lái)了解一下吧。
上次介紹到rank0的機(jī)器生成了ncclUniqueId,并完成了機(jī)器的bootstrap網(wǎng)絡(luò)和通信網(wǎng)絡(luò)的初始化,這節(jié)接著看下所有節(jié)點(diǎn)間bootstrap的連接是如何建立的。
rank0節(jié)點(diǎn)執(zhí)行ncclGetUniqueId生成ncclUniqueId,通過(guò)mpi將Id廣播到所有節(jié)點(diǎn),然后所有節(jié)點(diǎn)都會(huì)執(zhí)行ncclCommInitRank,這里其他節(jié)點(diǎn)也會(huì)進(jìn)行初始化bootstrap網(wǎng)絡(luò)和通信網(wǎng)絡(luò)的操作,然后會(huì)執(zhí)行到ncclCommInitRankSync。
ncclResult_t ncclCommInitRankSync(ncclComm_t* newcomm, int nranks, ncclUniqueId commId, int myrank, int cudaDev) { ncclResult_t res; CUDACHECK(cudaSetDevice(cudaDev)); NCCLCHECKGOTO(commAlloc(newcomm, nranks, myrank), res, cleanup); NCCLCHECKGOTO(initTransportsRank(*newcomm, &commId), res, cleanup); NCCLCHECKGOTO(devCommSetup(*newcomm), res, cleanup); INFO(NCCL_INIT,"comm %p rank %d nranks %d cudaDev %d busId %x - Init COMPLETE", *newcomm, myrank, nranks, (*newcomm)->cudaDev, (*newcomm)->busId); return ncclSuccess; cleanup: if ((*newcomm) && (*newcomm)->bootstrap) bootstrapAbort((*newcomm)->bootstrap); *newcomm = NULL; return res; }
ncclComm_t是指向ncclComm的指針,ncclComm是一個(gè)大雜燴,包含了通信用到的所有上下文信息,里面的字段等用到的時(shí)候再介紹,然后通過(guò)commAlloc分配newcom,并且完成初始化,比如當(dāng)前是哪個(gè)卡,對(duì)應(yīng)的pcie busid是什么,然后執(zhí)行initTransportsRank。
static ncclResult_t initTransportsRank(struct ncclComm* comm, ncclUniqueId* commId) { // We use 3 AllGathers // 1. { peerInfo, comm } // 2. ConnectTransport[nranks], ConnectValue[nranks] // 3. { nThreads, nrings, compCap, prev[MAXCHANNELS], next[MAXCHANNELS] } int rank = comm->rank; int nranks = comm->nRanks; uint64_t commHash = getHash(commId->internal, NCCL_UNIQUE_ID_BYTES); TRACE(NCCL_INIT, "comm %p, commHash %lx, rank %d nranks %d - BEGIN", comm, commHash, rank, nranks); NCCLCHECK(bootstrapInit(commId, rank, nranks, &comm->bootstrap)); // AllGather1 - begin struct { struct ncclPeerInfo peerInfo; struct ncclComm* comm; } *allGather1Data; NCCLCHECK(ncclCalloc(&allGather1Data, nranks)); allGather1Data[rank].comm = comm; struct ncclPeerInfo* myInfo = &allGather1Data[rank].peerInfo; NCCLCHECK(fillInfo(comm, myInfo, commHash)); NCCLCHECK(bootstrapAllGather(comm->bootstrap, allGather1Data, sizeof(*allGather1Data))); NCCLCHECK(ncclCalloc(&comm->peerInfo, nranks+1)); // Extra rank to represent CollNet root for (int i = 0; i < nranks; i++) { memcpy(comm->peerInfo+i, &allGather1Data[i].peerInfo, sizeof(struct ncclPeerInfo)); if ((i != rank) && (comm->peerInfo[i].hostHash == myInfo->hostHash) && (comm->peerInfo[i].busId == myInfo->busId)) { WARN("Duplicate GPU detected : rank %d and rank %d both on CUDA device %x", rank, i, myInfo->busId); return ncclInvalidUsage; } }
看下bootstrapInit:
ncclResult_t bootstrapInit(ncclUniqueId * id, int rank, int nranks, void** commState) { ncclNetHandle_t* netHandle = (ncclNetHandle_t*) id; bool idFromEnv = getenv("NCCL_COMM_ID") != NULL; struct extState* state; NCCLCHECK(ncclCalloc(&state, 1)); state->rank = rank; state->nranks = nranks; *commState = state; TRACE(NCCL_INIT, "rank %d nranks %d", rank, nranks); struct extInfo info = { 0 }; info.rank = rank; info.nranks = nranks; void *tmpSendComm, *tmpRecvComm; // Pass the remote address to listen via info if (idFromEnv) { memcpy(&info.extHandleListen, netHandle, sizeof(ncclNetHandle_t)); memcpy(&info.extHandleListenRoot, netHandle, sizeof(ncclNetHandle_t)); } // listen will return the local address via info (specify interface type 'findSubnetIf') state->dev = idFromEnv ? findSubnetIf : 0; void* extBstrapListenCommRoot; NCCLCHECK(bootstrapNetListen(state->dev, &info.extHandleListen, &state->extBstrapListenComm)); NCCLCHECK(bootstrapNetListen(state->dev, &info.extHandleListenRoot, &extBstrapListenCommRoot)); // stagger connection times to avoid an overload of the root at very high rank counts if (nranks > 128) { long msec = rank; struct timespec tv; tv.tv_sec = msec / 1000; tv.tv_nsec = 1000000 * (msec % 1000); TRACE(NCCL_INIT, "rank %d delaying connection to root by %ld msec", rank, msec); (void) nanosleep(&tv, NULL); } // send info on my listening socket to root NCCLCHECK(bootstrapNetConnect(state->dev, netHandle, &tmpSendComm)); NCCLCHECK(bootstrapNetSend(tmpSendComm, &info, sizeof(info))); NCCLCHECK(bootstrapNetCloseSend(tmpSendComm)); // get info on my "next" rank in the bootstrap ring from root }
首先看下commState,即ncclComm的bootstrap,類(lèi)型為extState。
struct extState { void* extBstrapListenComm; void* extBstrapRingRecvComm; void* extBstrapRingSendComm; ncclNetHandle_t* peerBstrapHandles; struct unexConn* unexpectedConnections; int rank; int nranks; int dev; };
其中extBstrapRingSendComm是當(dāng)前節(jié)點(diǎn)連接next的socket連接,extBstrapRingRecvComm是當(dāng)前節(jié)點(diǎn)和prev節(jié)點(diǎn)的socket連接,extBstrapListenComm是當(dāng)前節(jié)點(diǎn)的監(jiān)聽(tīng)socket,peerBstrapHandles是所有rank的ip port(對(duì)應(yīng)extBstrapListenComm),dev默認(rèn)為0,表示用第幾個(gè)ip地址。
然后通過(guò)bootstrapNetListen創(chuàng)建extHandleListen和extHandleListenRoot兩個(gè)bootstrap comm,如前文所述,bootstrap comm其實(shí)就是保存了fd,這里創(chuàng)建兩個(gè)comm的原因是extHandleListen是rank之間實(shí)際使用的bootstrap連接,extHandleListenRoot是rank0節(jié)點(diǎn)和其他所有rank進(jìn)行通信使用的連接。
static ncclResult_t bootstrapNetListen(int dev, ncclNetHandle_t* netHandle, void** listenComm)
bootstrapNetListen函數(shù)上節(jié)有介紹過(guò),會(huì)獲取到第dev個(gè)當(dāng)前機(jī)器的ip,然后listen獲取監(jiān)聽(tīng)fd,將ip port寫(xiě)到nethandle,獲取到的bootstrap comm寫(xiě)到listencomm。
然后將rank,nrank,extHandleListen和extHandleListenRoot寫(xiě)到extInfo里。
struct extInfo { int rank; int nranks; ncclNetHandle_t extHandleListenRoot; ncclNetHandle_t extHandleListen; };
netHandle為ncclUniqueId,即rank0的ip port,然后通過(guò)bootstrapNetConnect創(chuàng)建bootstrap send comm,類(lèi)比bootstrapNetListen,bootstrapNetConnect就是建立到netHandle的socket連接,將socket寫(xiě)到sendComm里,這里dev并沒(méi)有用到。
static ncclResult_t bootstrapNetConnect(int dev, ncclNetHandle_t* netHandle, void** sendComm)
然后通過(guò)bootstrapNetSend將extInfo發(fā)送出去,即發(fā)給rank0:
static ncclResult_t bootstrapNetSend(void* sendComm, void* data, int size) { struct bootstrapNetComm* comm = (struct bootstrapNetComm*)sendComm; NCCLCHECK(socketSend(comm->fd, &size, sizeof(int))); NCCLCHECK(socketSend(comm->fd, data, size)); return ncclSuccess; }
其中socketSend就是執(zhí)行send接口發(fā)送數(shù)據(jù)。
然后通過(guò)bootstrapNetCloseSend關(guān)閉fd。
rank0收到數(shù)據(jù)后會(huì)做什么工作呢,回顧一下,rank0的節(jié)執(zhí)行ncclGetUniqueId生成ncclUniqueId,其中在執(zhí)行bootstrapCreateRoot的最后會(huì)啟動(dòng)一個(gè)線程執(zhí)行bootstrapRoot。
static void *bootstrapRoot(void* listenComm) { struct extInfo info; ncclNetHandle_t *rankHandles = NULL; ncclNetHandle_t *rankHandlesRoot = NULL; // for initial rank <-> root information exchange ncclNetHandle_t zero = { 0 }; // for sanity checking void* tmpComm; ncclResult_t res; setFilesLimit(); TRACE(NCCL_INIT, "BEGIN"); /* Receive addresses from all ranks */ int nranks = 0, c = 0; do { NCCLCHECKGOTO(bootstrapNetAccept(listenComm, &tmpComm), res, out); NCCLCHECKGOTO(bootstrapNetRecv(tmpComm, &info, sizeof(info)), res, out); NCCLCHECKGOTO(bootstrapNetCloseRecv(tmpComm), res, out); if (c == 0) { nranks = info.nranks; NCCLCHECKGOTO(ncclCalloc(&rankHandles, nranks), res, out); NCCLCHECKGOTO(ncclCalloc(&rankHandlesRoot, nranks), res, out); } if (nranks != info.nranks) { WARN("Bootstrap Root : mismatch in rank count from procs %d : %d", nranks, info.nranks); goto out; } if (memcmp(&zero, &rankHandlesRoot[info.rank], sizeof(ncclNetHandle_t)) != 0) { WARN("Bootstrap Root : rank %d of %d ranks has already checked in", info.rank, nranks); goto out; } // Save the connection handle for that rank memcpy(rankHandlesRoot+info.rank, info.extHandleListenRoot, sizeof(ncclNetHandle_t)); memcpy(rankHandles+info.rank, info.extHandleListen, sizeof(ncclNetHandle_t)); ++c; TRACE(NCCL_INIT, "Received connect from rank %d total %d/%d", info.rank, c, nranks); } while (c < nranks); TRACE(NCCL_INIT, "COLLECTED ALL %d HANDLES", nranks); // Send the connect handle for the next rank in the AllGather ring for (int r=0; r<nranks; ++r) { int next = (r+1) % nranks; void *tmpSendComm; NCCLCHECKGOTO(bootstrapNetConnect(0, rankHandlesRoot+r, &tmpSendComm), res, out); NCCLCHECKGOTO(bootstrapNetSend(tmpSendComm, rankHandles+next, sizeof(ncclNetHandle_t)), res, out); NCCLCHECKGOTO(bootstrapNetCloseSend(tmpSendComm), res, out); } TRACE(NCCL_INIT, "SENT OUT ALL %d HANDLES", nranks); out: bootstrapNetCloseListen(listenComm); if (rankHandles) free(rankHandles); if (rankHandlesRoot) free(rankHandlesRoot); TRACE(NCCL_INIT, "DONE"); return NULL; }
listenComm是上一個(gè)博文中rank0創(chuàng)建的監(jiān)聽(tīng)fd,bootstrapNetAccept是從listenComm中獲取一個(gè)新連接,使用新連接的fd創(chuàng)建recvcomm。
static ncclResult_t bootstrapNetAccept(void* listenComm, void** recvComm)
然后通過(guò)bootstrapNetRecv讀取tmpComm的數(shù)據(jù),即其他rank發(fā)送來(lái)的extInfo,然后保存其他rank的extHandleListen和extHandleListenRoot,這個(gè)時(shí)候rank0就獲取到其他所有rank的ip和port了。
獲取完所有rank的info之后開(kāi)始建環(huán),將節(jié)點(diǎn)(r+1) % nranks的extHandleListen發(fā)送給節(jié)點(diǎn)r,就是說(shuō)將節(jié)點(diǎn)r的next節(jié)點(diǎn)的nethandle發(fā)送給節(jié)點(diǎn)r。這里可以看出,每個(gè)節(jié)點(diǎn)創(chuàng)建了兩個(gè)listen comm,其中rank0使用extHandleListenRoot進(jìn)行通信,其他節(jié)點(diǎn)之間通過(guò)extHandleListen進(jìn)行通信。
然后再回去接著看bootstrapInit。
ncclResult_t bootstrapInit(ncclUniqueId * id, int rank, int nranks, void** commState) { // get info on my "next" rank in the bootstrap ring from root ncclNetHandle_t extHandleNext; NCCLCHECK(bootstrapNetAccept(extBstrapListenCommRoot, &tmpRecvComm)); NCCLCHECK(bootstrapNetRecv(tmpRecvComm, &extHandleNext, sizeof(extHandleNext))); NCCLCHECK(bootstrapNetCloseRecv(tmpRecvComm)); NCCLCHECK(bootstrapNetCloseListen(extBstrapListenCommRoot)); NCCLCHECK(bootstrapNetConnect(state->dev, &extHandleNext, &state->extBstrapRingSendComm)); // Accept the connect request from the previous rank in the AllGather ring NCCLCHECK(bootstrapNetAccept(state->extBstrapListenComm, &state->extBstrapRingRecvComm)); // AllGather all listen handlers NCCLCHECK(ncclCalloc(&state->peerBstrapHandles, nranks)); memcpy(state->peerBstrapHandles+rank, info.extHandleListen, sizeof(ncclNetHandle_t)); NCCLCHECK(bootstrapAllGather(state, state->peerBstrapHandles, sizeof(ncclNetHandle_t))); TRACE(NCCL_INIT, "rank %d nranks %d - DONE", rank, nranks); return ncclSuccess; }
接著所有rank都會(huì)在extHandleListenRoot上接收新連接創(chuàng)建tmpRecvComm,然后接收到當(dāng)前rank的next的ip,port;然后連接next創(chuàng)建bscomm到state->extBstrapRingSendComm,接收prev的連接創(chuàng)建bscomm到state->extBstrapRingRecvComm,到現(xiàn)在bootstrap網(wǎng)絡(luò)連接就完全建立起來(lái)了,如下圖:
最后gather所有rank的ip port,首先將自己的nethandle放到peerBstrapHandles的對(duì)應(yīng)位置,如下所示。
然后執(zhí)行bootstrapAllGather:
ncclResult_t bootstrapAllGather(void* commState, void* allData, int size) { struct extState* state = (struct extState*)commState; char* data = (char*)allData; int rank = state->rank; int nranks = state->nranks; TRACE(NCCL_INIT, "rank %d nranks %d size %d", rank, nranks, size); /* Simple ring based AllGather * At each step i receive data from (rank-i-1) from left * and send previous step's data from (rank-i) to right */ for (int i=0; i<nranks-1; i++) { size_t rslice = (rank - i - 1 + nranks) % nranks; size_t sslice = (rank - i + nranks) % nranks; // Send slice to the right NCCLCHECK(bootstrapNetSend(state->extBstrapRingSendComm, data+sslice*size, size)); // Recv slice from the left NCCLCHECK(bootstrapNetRecv(state->extBstrapRingRecvComm, data+rslice*size, size)); } TRACE(NCCL_INIT, "rank %d nranks %d size %d - DONE", rank, nranks, size); return ncclSuccess; }
每一次將自己的data發(fā)送給對(duì)應(yīng)的rank,然后接收其他rank發(fā)送過(guò)來(lái)的data,如下圖。
第一步:
第二步:
到這里每個(gè)rank就都有了全局所有rank的ip port。
以上就是“NCCL Bootstrap網(wǎng)絡(luò)連接建立的方法是什么”這篇文章的所有內(nèi)容,感謝各位的閱讀!相信大家閱讀完這篇文章都有很大的收獲,小編每天都會(huì)為大家更新不同的知識(shí),如果還想學(xué)習(xí)更多的知識(shí),請(qǐng)關(guān)注億速云行業(yè)資訊頻道。
免責(zé)聲明:本站發(fā)布的內(nèi)容(圖片、視頻和文字)以原創(chuàng)、轉(zhuǎn)載和分享為主,文章觀點(diǎn)不代表本網(wǎng)站立場(chǎng),如果涉及侵權(quán)請(qǐng)聯(lián)系站長(zhǎng)郵箱:is@yisu.com進(jìn)行舉報(bào),并提供相關(guān)證據(jù),一經(jīng)查實(shí),將立刻刪除涉嫌侵權(quán)內(nèi)容。