diff --git a/src/channel.cc b/src/channel.cc index e9cfa6664f..5018a39ae5 100644 --- a/src/channel.cc +++ b/src/channel.cc @@ -18,11 +18,11 @@ ncclResult_t initChannel(struct ncclComm* comm, int channelid) { channel->id = channelid; // Ring index to user rank table. - NCCLCHECK(ncclCudaCalloc(&channel->ring.devUserRanks, comm->nRanks)); + NCCLCHECK(ncclCudaCalloc(&channel->ring.devUserRanks, comm->nRanks, comm->sideStream)); NCCLCHECK(ncclCalloc(&channel->ring.userRanks, comm->nRanks)); // Communication structures with peers. - NCCLCHECK(ncclCudaCalloc(&channel->devPeers, comm->nRanks+1)); // The extra one rank is for collnet root (i.e. network) + NCCLCHECK(ncclCudaCalloc(&channel->devPeers, comm->nRanks+1, comm->sideStream)); // The extra one rank is for collnet root (i.e. network) NCCLCHECK(ncclCalloc(&channel->peers, comm->nRanks+1)); for (size_t i=0; inRanks+1; ++i) { for (int b=0; b -static ncclResult_t ncclCudaCallocDebug(const char *filefunc, int line, T** ptr, size_t nelem, bool isFineGrain = false) { - - // Need async stream for P2P pre-connect + CUDA Graph - static bool streamCreated = false; - static hipStream_t stream; - if (rcclParamEnableHipGraph() && !streamCreated) - { - // Create stream only once to avoid performance penalty - CUDACHECK(hipStreamCreateWithFlags(&stream, hipStreamNonBlocking)); - streamCreated = true; - } +static ncclResult_t ncclCudaCallocDebug(const char *filefunc, int line, T** ptr, size_t nelem, hipStream_t sideStream = nullptr, bool isFineGrain = false) { if (isFineGrain) CUDACHECK(hipExtMallocWithFlags((void**)ptr, nelem*sizeof(T), hipDeviceMallocFinegrained)); @@ -96,10 +86,14 @@ static ncclResult_t ncclCudaCallocDebug(const char *filefunc, int line, T** ptr, CUDACHECK(hipMalloc(ptr, nelem*sizeof(T))); if (rcclParamEnableHipGraph()) { + hipStream_t stream = sideStream; + if (stream == nullptr) + CUDACHECK(hipStreamCreateWithFlags(&stream, hipStreamNonBlocking)); + CUDACHECK(hipMemsetAsync(*ptr, 0, nelem*sizeof(T), stream)); CUDACHECK(hipStreamSynchronize(stream)); - // NOTE: Currently the re-used stream is not destroyed - //CUDACHECK(hipStreamDestroy(stream)); + if (sideStream == nullptr) + CUDACHECK(hipStreamDestroy(stream)); } else { CUDACHECK(hipMemset(*ptr, 0, nelem*sizeof(T))); CUDACHECK(hipStreamSynchronize(NULL)); diff --git a/src/include/comm.h b/src/include/comm.h index c7f04d44a7..41ac83922f 100644 --- a/src/include/comm.h +++ b/src/include/comm.h @@ -220,6 +220,7 @@ struct ncclComm { int p2pRecvCount; // [RCCL] + hipStream_t sideStream; //CliqueManager* cliqueManager; // CliqueManager handles pointer collection / distribution for clique-based kernels //int rootPid; // Process ID of root // [/RCCL] diff --git a/src/init.cc b/src/init.cc index 31cdf595b2..1f921d41f2 100644 --- a/src/init.cc +++ b/src/init.cc @@ -303,6 +303,8 @@ static ncclResult_t commFree(ncclComm_t comm) { CUDACHECK(hipStreamDestroy(comm->groupStream)); } + CUDACHECK(hipStreamDestroy(comm->sideStream)); + // Last rank frees shared resources between threads int isLast; NCCLCHECK(ncclCpuBarrierIn(comm, &isLast)); @@ -365,6 +367,9 @@ static ncclResult_t commAlloc(ncclComm_t* comret, int ndev, int rank, int virtua NCCLCHECK(getBusId(comm->cudaDev, &comm->busId)); TRACE(NCCL_INIT,"comm %p rank %d nranks %d cudaDev %d busId %lx", comm, rank, ndev, comm->cudaDev, comm->busId); + // RCCL: create persistent stream for calloc + CUDACHECK(hipStreamCreateWithFlags(&comm->sideStream, hipStreamNonBlocking)); + comm->doneEvent = doneEvent; comm->intDoneEvent = intDoneEvent; comm->checkPointers = ncclParamCheckPointers() == 1 ? true : false; @@ -385,7 +390,7 @@ static ncclResult_t commAlloc(ncclComm_t* comret, int ndev, int rank, int virtua comm->argsptrs[0] = &comm->devComm; #ifdef ENABLE_PROFILING - NCCLCHECK(ncclCudaCalloc(&comm->hostDevComm.devProf, MAXCHANNELS*PROFILE_NUM_LAUNCHES)); + NCCLCHECK(ncclCudaCalloc(&comm->hostDevComm.devProf, MAXCHANNELS*PROFILE_NUM_LAUNCHES, comm->sideStream)); #endif #ifdef ENABLE_COLLTRACE @@ -445,7 +450,7 @@ static ncclResult_t commAlloc(ncclComm_t* comret, int ndev, int rank, int virtua static ncclResult_t devCommSetup(ncclComm_t comm) { ncclDevCommAndChannels *devCommAndChans; - NCCLCHECK(ncclCudaCalloc(&devCommAndChans, 1)); + NCCLCHECK(ncclCudaCalloc(&devCommAndChans, 1, comm->sideStream)); comm->devComm = &devCommAndChans->comm; comm->hostDevComm.channels = devCommAndChans->channels; diff --git a/src/transport.cc b/src/transport.cc index 6b279cbdbb..da8fe8d2da 100644 --- a/src/transport.cc +++ b/src/transport.cc @@ -81,9 +81,6 @@ void dumpData(struct ncclConnect* data, int ndata) { } ncclResult_t ncclTransportP2pSetup(struct ncclComm* comm, struct ncclTopoGraph* graph, int connIndex, int* highestTransportType/*=NULL*/) { - // Stream used during transport setup; need for P2P pre-connect + CUDA Graph - hipStream_t transportSetupStream; - CUDACHECK(hipStreamCreateWithFlags(&transportSetupStream, hipStreamNonBlocking)); int highestType = TRANSPORT_P2P; // track highest transport type struct ncclConnect data[2*MAXCHANNELS]; @@ -137,7 +134,7 @@ ncclResult_t ncclTransportP2pSetup(struct ncclComm* comm, struct ncclTopoGraph* struct ncclConnector* conn = comm->channels[c].peers[sendPeer].send + connIndex; NCCLCHECK(conn->transportComm->connect(comm, sendData++, 1, comm->rank, conn)); conn->connected = 1; - CUDACHECK(hipMemcpyAsync(comm->channels[c].devPeers[sendPeer].send+connIndex, conn, sizeof(struct ncclConnector), hipMemcpyHostToDevice, transportSetupStream)); + CUDACHECK(hipMemcpyAsync(comm->channels[c].devPeers[sendPeer].send+connIndex, conn, sizeof(struct ncclConnector), hipMemcpyHostToDevice, comm->sideStream)); } } TIME_STOP(3); @@ -147,14 +144,13 @@ ncclResult_t ncclTransportP2pSetup(struct ncclComm* comm, struct ncclTopoGraph* struct ncclConnector* conn = comm->channels[c].peers[recvPeer].recv + connIndex; NCCLCHECK(conn->transportComm->connect(comm, recvData++, 1, comm->rank, conn)); conn->connected = 1; - CUDACHECK(hipMemcpyAsync(comm->channels[c].devPeers[recvPeer].recv+connIndex, conn, sizeof(struct ncclConnector), hipMemcpyHostToDevice, transportSetupStream)); + CUDACHECK(hipMemcpyAsync(comm->channels[c].devPeers[recvPeer].recv+connIndex, conn, sizeof(struct ncclConnector), hipMemcpyHostToDevice, comm->sideStream)); } } TIME_STOP(4); comm->connectRecv[recvPeer+comm->nRanks*connIndex] = comm->connectSend[sendPeer+comm->nRanks*connIndex] = 0; } - CUDACHECK(hipStreamSynchronize(transportSetupStream)); - CUDACHECK(hipStreamDestroy(transportSetupStream)); + CUDACHECK(hipStreamSynchronize(comm->sideStream)); if (highestTransportType != NULL) *highestTransportType = highestType; TIME_PRINT("P2P Setup/Connect"); return ncclSuccess; diff --git a/src/transport/coll_net.cc b/src/transport/coll_net.cc index e7bec6e70b..cb0ae2a482 100644 --- a/src/transport/coll_net.cc +++ b/src/transport/coll_net.cc @@ -351,7 +351,7 @@ static ncclResult_t sharedBuffersInit(struct ncclComm* comm, int cuda, char** gp *size = state->size; if (cuda && state->cudaBuff == NULL) { - NCCLCHECK(ncclCudaCalloc(&state->cudaBuff, *size, cuda)); + NCCLCHECK(ncclCudaCalloc(&state->cudaBuff, *size, comm->sideStream, cuda)); } if (!cuda && state->hostBuff == NULL) { NCCLCHECK(ncclCudaHostCalloc(&state->hostBuff, *size)); diff --git a/src/transport/net.cc b/src/transport/net.cc index 10572d3af4..56d5748bff 100644 --- a/src/transport/net.cc +++ b/src/transport/net.cc @@ -396,7 +396,7 @@ static ncclResult_t sharedBuffersInit(struct ncclComm* comm, int cuda, int local if (size) *size = state->size; if (cuda && state->cudaBuff == NULL) { - NCCLCHECK(ncclCudaCalloc(&state->cudaBuff, state->size, cuda)); + NCCLCHECK(ncclCudaCalloc(&state->cudaBuff, state->size, comm->sideStream, cuda)); if (sameProcess == 0) { CUDACHECK(hipIpcGetMemHandle(&state->ipc, state->cudaBuff)); } @@ -575,7 +575,7 @@ static ncclResult_t sendProxyConnect(struct ncclProxyConnection* connection, str if (!map->sameProcess) { ALIGN_SIZE(map->mems[NCCL_NET_MAP_DEVMEM].size, CUDA_IPC_MIN); } - NCCLCHECK(ncclCudaCalloc(&map->mems[NCCL_NET_MAP_DEVMEM].gpuPtr, map->mems[NCCL_NET_MAP_DEVMEM].size, resources->useGdr)); + NCCLCHECK(ncclCudaCalloc(&map->mems[NCCL_NET_MAP_DEVMEM].gpuPtr, map->mems[NCCL_NET_MAP_DEVMEM].size, comm->sideStream, resources->useGdr)); map->mems[NCCL_NET_MAP_DEVMEM].cpuPtr = map->mems[NCCL_NET_MAP_DEVMEM].gpuPtr; } if (!map->sameProcess) { @@ -689,7 +689,7 @@ static ncclResult_t recvProxyConnect(struct ncclProxyConnection* connection, str if (map->mems[NCCL_NET_MAP_DEVMEM].size) { if (resources->shared == 0) { - NCCLCHECK(ncclCudaCalloc(&map->mems[NCCL_NET_MAP_DEVMEM].gpuPtr, map->mems[NCCL_NET_MAP_DEVMEM].size, resources->useGdr)); + NCCLCHECK(ncclCudaCalloc(&map->mems[NCCL_NET_MAP_DEVMEM].gpuPtr, map->mems[NCCL_NET_MAP_DEVMEM].size, comm->sideStream, resources->useGdr)); map->mems[NCCL_NET_MAP_DEVMEM].cpuPtr = map->mems[NCCL_NET_MAP_DEVMEM].gpuPtr; } } diff --git a/src/transport/p2p.cc b/src/transport/p2p.cc index 5101eae3bc..481b86319c 100644 --- a/src/transport/p2p.cc +++ b/src/transport/p2p.cc @@ -115,7 +115,7 @@ ncclResult_t p2pCanConnect(int* ret, struct ncclTopoSystem* topo, struct ncclTop // Check that legacy IPC support is available (WSL WAR) char *dummy; hipIpcMemHandle_t ipc; - NCCLCHECK(ncclCudaCalloc(&dummy, CUDA_IPC_MIN)); + CUDACHECK(hipMalloc(&dummy, CUDA_IPC_MIN)); if (hipIpcGetMemHandle(&ipc, dummy) != hipSuccess) { INFO(NCCL_INIT|NCCL_P2P,"Legacy IPC not supported"); *ret = 0; @@ -347,7 +347,7 @@ static ncclResult_t p2pProxySetup(struct ncclProxyConnection* connection, struct int size = *((int*)reqBuff); if (respSize != sizeof(struct ncclP2pBuff)) return ncclInternalError; struct ncclP2pBuff* p2pBuff = (struct ncclP2pBuff*)respBuff; - NCCLCHECK(ncclCudaCalloc((char**)&p2pBuff->directPtr, size, true)); + NCCLCHECK(ncclCudaCalloc((char**)&p2pBuff->directPtr, size, comm->sideStream, true)); connection->transportResources = p2pBuff->directPtr; hipError_t res = hipIpcGetMemHandle(&p2pBuff->devIpc, p2pBuff->directPtr); if (res != hipSuccess) { diff --git a/tools/ib-test/ib_test.cpp b/tools/ib-test/ib_test.cpp index 7a5e09f7a5..640b4f4328 100755 --- a/tools/ib-test/ib_test.cpp +++ b/tools/ib-test/ib_test.cpp @@ -245,7 +245,7 @@ private: printf("GDR Read %s\n", use_gdr_read ? "enabled" : "disabled"); if (use_gdr_read) { - NCCLCHECK(ncclCudaCalloc(&sendDevBuffer, sendBuffSize, 1)); + NCCLCHECK(ncclCudaCalloc(&sendDevBuffer, sendBuffSize, nullptr, 1)); printf("Allocated sendDevBuffer %p of %d bytes, sliceSteps %d\n", sendDevBuffer, sendBuffSize, sliceSteps); } @@ -397,7 +397,7 @@ private: printf("GDR Write %s\n", use_gdr_write ? "enabled" : "disabled"); if (use_gdr_write) { - NCCLCHECK(ncclCudaCalloc(&recvDevBuffer, recvBuffSize, 1)); + NCCLCHECK(ncclCudaCalloc(&recvDevBuffer, recvBuffSize, nullptr, 1)); printf("Allocated recvDevBuffer %p of %d bytes, sliceSteps %d\n", recvDevBuffer, recvBuffSize, sliceSteps); }