AI/rocm-systems
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'nccl/master' into develop

Browse Source
This commit is contained in:
BertanDogancay
2024-04-23 13:33:19 -07:00
parent 220066197a 6dd51f15bf
commit e1a835910e
76 changed files with 5157 additions and 2296 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/init.cc
+142 -120
View File
@@ -329,6 +329,10 @@ static ncclResult_t commFree(ncclComm_t comm) {
* resource cleanup in commFree(). */
if (comm->proxyState && comm->proxyRefCountOld == 0 && comm->proxyState->thread) {
pthread_join(comm->proxyState->thread, nullptr);
if (comm->proxyState->threadUDS) {
// UDS support
pthread_join(comm->proxyState->threadUDS, nullptr);;
}
}
delete[] comm->userRedOps;
@@ -417,17 +421,7 @@ static ncclResult_t commFree(ncclComm_t comm) {
free(comm->topParentRanks);
free(comm->topParentLocalRanks);
while (!ncclIntruQueueEmpty(&comm->regRecordQueue)) {
struct ncclRegRecord* rec = ncclIntruQueueDequeue(&comm->regRecordQueue);
NCCLCHECK(ncclNvlsDeregBuffer(&rec->mcHandle, rec->regAddr, rec->dev, rec->regSize));
free(rec->addrs);
free(rec);
}
while (!ncclIntruQueueEmpty(&comm->regRequestQueue)) {
struct ncclRegRequest* req = ncclIntruQueueDequeue(&comm->regRequestQueue);
free(req);
}
NCCLCHECK(ncclRegCleanup(comm));
commPoison(comm); // poison comm before free to avoid comm reuse.
free(comm);
@@ -472,7 +466,7 @@ ncclResult_t ncclCommEnsureReady(ncclComm_t comm) {
/* comm must be ready, or error will be reported */
ncclResult_t ret = ncclSuccess;
if (*comm->abortFlag) {
if (__atomic_load_n(comm->abortFlag, __ATOMIC_RELAXED)) {
ncclGroupJobAbort(comm->groupJob);
} else {
NCCLCHECK(ncclCommGetAsyncError(comm, &ret));
@@ -555,7 +549,6 @@ static ncclResult_t commAlloc(struct ncclComm* comm, struct ncclComm* parent, in
comm->groupNext = reinterpret_cast<struct ncclComm*>(0x1);
comm->preconnectNext = reinterpret_cast<struct ncclComm*>(0x1);
comm->channelSize = ncclParamAggChannelSize();
static_assert(MAXCHANNELS <= sizeof(*comm->connectSend)*8, "comm->connectSend must have enough bits for all channels");
static_assert(MAXCHANNELS <= sizeof(*comm->connectRecv)*8, "comm->connectRecv must have enough bits for all channels");
@@ -588,9 +581,9 @@ static ncclResult_t commAlloc(struct ncclComm* comm, struct ncclComm* parent, in
comm->topParentRanks[i] = i;
}
ncclIntruQueueConstruct(&comm->regRequestQueue);
ncclIntruQueueConstruct(&comm->regRecordQueue);
ncclIntruQueueMpscConstruct(&comm->callbackQueue);
comm->regCache.pageSize = sysconf(_SC_PAGESIZE);
return ncclSuccess;
}
@@ -606,6 +599,8 @@ static ncclResult_t devCommSetup(ncclComm_t comm) {
comm->devComm = &devCommAndChans->comm;
tmpCommAndChans.comm.rank = comm->rank;
tmpCommAndChans.comm.nRanks = nRanks;
tmpCommAndChans.comm.node = comm->node;
tmpCommAndChans.comm.nNodes = comm->nNodes;
tmpCommAndChans.comm.abortFlag = comm->abortFlag;
for (int p=0; p < NCCL_NUM_PROTOCOLS; p++) {
tmpCommAndChans.comm.buffSizes[p] = comm->buffSizes[p];
@@ -638,6 +633,12 @@ static ncclResult_t devCommSetup(ncclComm_t comm) {
comm->workFifoSent = 0;
comm->workFifoAckdMin = 0;
if (comm->collNetDenseToUserRank != nullptr) {
NCCLCHECKGOTO(ncclCudaCallocAsync(&tmpCommAndChans.comm.collNetDenseToUserRank, nRanks, comm->sharedRes->deviceStream.cudaStream), ret, fail);
ncclCommPushCudaFree(comm, tmpCommAndChans.comm.collNetDenseToUserRank);
NCCLCHECKGOTO(ncclCudaMemcpyAsync(tmpCommAndChans.comm.collNetDenseToUserRank, comm->collNetDenseToUserRank, nRanks, comm->sharedRes->deviceStream.cudaStream), ret, fail);
}
for (int c=0; c < MAXCHANNELS; c++) {
tmpCommAndChans.channels[c].peers = comm->channels[c].devPeers;
tmpCommAndChans.channels[c].ring = comm->channels[c].ring;
@@ -731,6 +732,26 @@ static ncclResult_t fillInfo(struct ncclComm* comm, struct ncclPeerInfo* info, u
info->comm = comm;
info->cudaCompCap = comm->minCompCap = comm->maxCompCap = comm->compCap;
#if !defined(__HIP_PLATFORM_HCC__) && !defined(__HCC__) && !defined(__HIPCC__)
// MNNVL support
{
// MNNVL: Request the fabric UUID and partition info
char busId[NVML_DEVICE_PCI_BUS_ID_BUFFER_SIZE];
nvmlDevice_t nvmlDev;
NCCLCHECK(int64ToBusId(info->busId, busId));
NCCLCHECK(ncclNvmlDeviceGetHandleByPciBusId(busId, &nvmlDev));
info->fabricInfo.state = NVML_GPU_FABRIC_STATE_NOT_SUPPORTED;
(void) ncclNvmlDeviceGetGpuFabricInfoV(nvmlDev, &info->fabricInfo);
if (info->fabricInfo.state != NVML_GPU_FABRIC_STATE_NOT_SUPPORTED) {
INFO(NCCL_INIT, "MNNVL busId 0x%lx fabric UUID %lx.%lx cliqueId 0x%x state %d healthMask 0x%x",
info->busId,
((long *)&info->fabricInfo.clusterUuid)[0], ((long *)&info->fabricInfo.clusterUuid)[1],
info->fabricInfo.cliqueId, info->fabricInfo.state, info->fabricInfo.healthMask);
}
}
#endif
return ncclSuccess;
}
@@ -774,8 +795,9 @@ static ncclResult_t computeBuffSizes(struct ncclComm* comm) {
comm->buffSizes[p] = envs[p] != -2 ? envs[p] : defaults[p];
}
if (comm->nNodes > 1) comm->p2pChunkSize = ncclParamP2pNetChunkSize();
else if (ncclTopoPathAllNVLink(comm->topo)) comm->p2pChunkSize = ncclParamP2pNvlChunkSize();
// MNNVL support
if (!comm->MNNVL && comm->nNodes > 1) comm->p2pChunkSize = ncclParamP2pNetChunkSize();
else if (comm->MNNVL || ncclTopoPathAllNVLink(comm->topo)) comm->p2pChunkSize = ncclParamP2pNvlChunkSize();
else comm->p2pChunkSize = ncclParamP2pPciChunkSize();
// Make sure P2P chunksize is not larger than coll chunksize.
@@ -805,6 +827,8 @@ static ncclResult_t collNetTrySetup(ncclComm_t comm, ncclComm_t parent, struct n
int highestTypes[NCCL_MAX_LOCAL_RANKS] = { TRANSPORT_P2P };
// Find all head ranks
int nHeads = collNetGraph->nChannels;
int nHeadsUnique = 0;
int headsUnique[NCCL_MAX_LOCAL_RANKS];
int highestTransportType0, highestTransportType1;
char line[1024];
bool share;
@@ -816,13 +840,20 @@ static ncclResult_t collNetTrySetup(ncclComm_t comm, ncclComm_t parent, struct n
struct collnetShareInfo* infos = NULL;
NCCLCHECKGOTO(ncclCalloc(&heads, nHeads), ret, fail);
// Head GPU index is always 0
for (int c = 0; c < nHeads; c++) {
heads[c] = collNetGraph->intra[c * comm->localRanks + 0];
{ uint64_t mask = 0;
// Head GPU index is always 0
for (int c = 0; c < nHeads; c++) {
heads[c] = collNetGraph->intra[c * comm->localRanks + 0];
assert(comm->rankToNode[heads[c]] == comm->node);
uint64_t mask0 = mask;
mask |= 1ull<<comm->rankToLocalRank[heads[c]];
if (mask != mask0) headsUnique[nHeadsUnique++] = heads[c];
}
}
comm->collNetHeads = heads;
comm->collNetHeadsNum = nHeads;
comm->collNetHeadsUniqueNum = nHeadsUnique;
if (parent && parent->collNetSupport && parent->config.splitShare && parent->nNodes == comm->nNodes) {
NCCLCHECKGOTO(ncclCalloc(&infos, comm->nRanks), ret, fail);
/* check whether child can share collnet resources of parent. Since parent builds each collnet communicator
@@ -883,6 +914,26 @@ static ncclResult_t collNetTrySetup(ncclComm_t comm, ncclComm_t parent, struct n
NCCLCHECK(ncclCalloc(&comm->collNetSharedRes, 1));
comm->collNetChannels = comm->collNetSharedRes->nChannels = comm->nChannels;
comm->collNetSharedRes->buffSize = comm->buffSizes[NCCL_PROTO_SIMPLE];
comm->collNetDenseToUserRank = ncclMemoryStackAlloc<int>(&comm->memPermanent, comm->nRanks);
comm->collNetUserToDenseRank = ncclMemoryStackAlloc<int>(&comm->memPermanent, comm->nRanks);
{ // initialize collNetUserToDenseRank[rank]
uint64_t nonHeadMask = (1ull<<comm->localRanks)-1;
comm->collNetUserToDenseRank[rank] = -1;
for (int h=0; h < nHeadsUnique; h++) {
nonHeadMask ^= 1ull<<comm->rankToLocalRank[headsUnique[h]];
if (headsUnique[h] == rank) { comm->collNetUserToDenseRank[rank] = h; break; }
}
if (comm->collNetUserToDenseRank[rank] == -1) {
comm->collNetUserToDenseRank[rank] = __builtin_popcountll(nonHeadMask & ((1ull<<comm->localRank)-1));
}
comm->collNetUserToDenseRank[rank] += comm->node*comm->localRanks;
}
NCCLCHECK(bootstrapAllGather(comm->bootstrap, comm->collNetUserToDenseRank, sizeof(int)));
for (int r=0; r < comm->nRanks; r++) {
comm->collNetDenseToUserRank[comm->collNetUserToDenseRank[r]] = r;
}
for (int c = 0; c < comm->collNetChannels; c++) {
struct ncclChannel* channel = comm->channels + c;
NCCLCHECKGOTO(initCollnetChannel(comm, c, parent, false), ret, fail);
@@ -1000,6 +1051,9 @@ fail:
goto exit;
}
// MNNVL: Flag to indicate whether to enable Multi-Node NVLink
NCCL_PARAM(MNNVL, "MNNVL", -2);
static ncclResult_t initTransportsRank(struct ncclComm* comm, struct ncclComm* parent = NULL) {
// We use 2 AllGathers
// 1. { peerInfo, comm, compCap}
@@ -1007,6 +1061,7 @@ static ncclResult_t initTransportsRank(struct ncclComm* comm, struct ncclComm* p
ncclResult_t ret = ncclSuccess;
int rank = comm->rank;
int nranks = comm->nRanks;
int nNodes = 1;
cpu_set_t affinitySave;
struct ncclTopoGraph ringGraph;
struct ncclTopoGraph treeGraph;
@@ -1054,6 +1109,7 @@ static ncclResult_t initTransportsRank(struct ncclComm* comm, struct ncclComm* p
NCCLCHECKGOTO(bootstrapAllGather(comm->bootstrap, comm->peerInfo, sizeof(struct ncclPeerInfo)), ret, fail);
for (int i = 0; i < nranks; i++) {
if (comm->peerInfo[i].hostHash != comm->peerInfo[rank].hostHash) nNodes++;
if ((i != rank) && (comm->peerInfo[i].hostHash == comm->peerInfo[rank].hostHash) && (comm->peerInfo[i].busId == comm->peerInfo[rank].busId)) {
WARN("Duplicate GPU detected : rank %d and rank %d both on CUDA device %lx", rank, i, comm->peerInfo[rank].busId);
ret = ncclInvalidUsage;
@@ -1063,6 +1119,56 @@ static ncclResult_t initTransportsRank(struct ncclComm* comm, struct ncclComm* p
// AllGather1 - end
#if CUDART_VERSION >= 11030
#include <cuda.h>
#include "cudawrap.h"
// MNNVL support
if (nNodes > 1) {
int cliqueSize = 0;
comm->MNNVL = 0;
// Determine the size of the MNNVL domain/clique
for (int i = 0; i < nranks; i++) {
nvmlGpuFabricInfoV_t *fabricInfo1 = &comm->peerInfo[rank].fabricInfo;
nvmlGpuFabricInfoV_t *fabricInfo2 = &comm->peerInfo[i].fabricInfo;
// Check that the Fabric state is fully initialized
if (fabricInfo2->state != NVML_GPU_FABRIC_STATE_COMPLETED) continue;
// Check that the cluster UUID and cliqueId match in each rank
// A zero UUID means we don't have MNNVL fabric info - disable MNNVL
if ((((long *)&fabricInfo2->clusterUuid)[0]|((long *)fabricInfo2->clusterUuid)[1]) == 0) continue;
if ((memcmp(fabricInfo1->clusterUuid, fabricInfo2->clusterUuid, NVML_GPU_FABRIC_UUID_LEN) == 0) &&
(fabricInfo1->cliqueId == fabricInfo2->cliqueId)) {
cliqueSize++;
}
}
// Determine whether this is a MNNVL system
comm->MNNVL = ncclParamMNNVL() < 0 ? cliqueSize == comm->nRanks : ncclParamMNNVL();
// MNNVL requires cuMem to be enabled
if (!ncclCuMemEnable()) comm->MNNVL = 0;
if (comm->MNNVL) {
// MNNVL also requires FABRIC handle support
int cudaDev;
int flag = 0;
CUdevice currentDev;
CUDACHECK(cudaGetDevice(&cudaDev));
CUCHECK(cuDeviceGet(&currentDev, cudaDev));
// Ignore error if CU_DEVICE_ATTRIBUTE_HANDLE_TYPE_FABRIC_SUPPORTED is not supported
(void) CUPFN(cuDeviceGetAttribute(&flag, CU_DEVICE_ATTRIBUTE_HANDLE_TYPE_FABRIC_SUPPORTED, currentDev));;
if (!flag)
comm->MNNVL = 0;
else
// Force the handle type to be FABRIC for MNNVL
ncclCuMemHandleType = CU_MEM_HANDLE_TYPE_FABRIC;
}
if (ncclParamMNNVL() == 1 && !comm->MNNVL) {
WARN("MNNVL is not supported on this system");
ret = ncclSystemError;
goto fail;
}
}
#endif
do {
// Compute intra-process ranks
int intraProcRank0 = -1, intraProcRank = -1, intraProcRanks = 0;
@@ -1347,6 +1453,9 @@ static ncclResult_t initTransportsRank(struct ncclComm* comm, struct ncclComm* p
goto fail;
}
INFO(NCCL_INIT, "comm %p rank %d nRanks %d nNodes %d localRanks %d localRank %d MNNVL %d",
comm, rank, comm->nRanks, comm->nNodes, comm->localRanks, comm->localRank, comm->MNNVL);
nChannelsOrig = comm->nChannels;
NCCLCHECKGOTO(ncclCalloc(&allTopoRanks, comm->nRanks), ret, fail);
int nc;
@@ -1439,7 +1548,12 @@ static ncclResult_t initTransportsRank(struct ncclComm* comm, struct ncclComm* p
comm->topParentLocalRanks = topParentLocalRanks;
// Launch proxy service thread, after this, the proxy calls can be used.
NCCLCHECKGOTO(ncclProxyCreate(comm), ret, fail);
if (parent && parent->config.splitShare) {
comm->proxyState = parent->sharedRes->proxyState;
ncclAtomicRefCountIncrement(&parent->sharedRes->proxyState->refCount);
} else {
NCCLCHECKGOTO(ncclProxyCreate(comm), ret, fail);
}
// Connect with prev/next for each ring
for (int c=0; c<comm->nChannels; c++) {
@@ -1476,8 +1590,8 @@ static ncclResult_t initTransportsRank(struct ncclComm* comm, struct ncclComm* p
// Setup NVLS
NCCLCHECKGOTO(ncclNvlsSetup(comm, parent), ret, fail);
// And NVLS trees if needed
if (comm->nvlsSupport && comm->localRanks > 1) {
for (int c=0; c<comm->nvlsChannels; c++) {
if (comm->nvlsSupport && comm->nNodes > 1) {
for (int c=0; c<comm->nChannels; c++) {
struct ncclChannel* channel = comm->channels+c;
NCCLCHECKGOTO(ncclTransportP2pConnect(comm, c, NCCL_MAX_NVLS_TREE_ARITY, channel->nvls.treeDown, 1, &channel->nvls.treeUp, 0), ret, fail);
NCCLCHECKGOTO(ncclTransportP2pConnect(comm, c, 1, &channel->nvls.treeUp, NCCL_MAX_NVLS_TREE_ARITY, channel->nvls.treeDown, 0), ret, fail);
@@ -1496,7 +1610,7 @@ static ncclResult_t initTransportsRank(struct ncclComm* comm, struct ncclComm* p
// Compute time models for algorithm and protocol combinations
NCCLCHECKGOTO(ncclTopoTuneModel(comm, comm->minCompCap, comm->maxCompCap, graphs), ret, fail);
INFO(NCCL_INIT, "%d coll channels, %d nvls channels, %d p2p channels, %d p2p channels per peer", comm->nChannels, comm->nvlsChannels, comm->p2pnChannels, comm->p2pnChannelsPerPeer);
INFO(NCCL_INIT, "%d coll channels, %d collnet channels, %d nvls channels, %d p2p channels, %d p2p channels per peer", comm->nChannels, comm->collNetChannels, comm->nvlsChannels, comm->p2pnChannels, comm->p2pnChannelsPerPeer);
do { // Setup p2p structures in comm->tasks
struct ncclTasks* tasks = &comm->tasks;
@@ -2281,7 +2395,7 @@ static ncclResult_t commReclaim(ncclComm_t comm) {
NCCLCHECKGOTO(ncclCommGetAsyncError(comm, &state), ret, fail);
TRACE(NCCL_INIT, "commReclaim: reclaim comm %p rank %d state %d", comm, comm->rank, state);
if (state == ncclSuccess && *comm->abortFlag == 0 && comm->finalizeCalled == false) {
if (state == ncclSuccess && __atomic_load_n(comm->abortFlag, __ATOMIC_RELAXED) == 0 && comm->finalizeCalled == false) {
/* user does not call ncclCommFinalize and this is a normal comm destroy. ncclCommDestroy
* should be nonblocking until last call of ncclCommDestroy. */
NCCLCHECKGOTO(commFinalize(comm, false), ret, fail);
@@ -2406,9 +2520,9 @@ ncclResult_t ncclCommAbort(ncclComm_t comm) {
// Ask anything that might still be running on the device to quit
childAbortFlag = __atomic_load_n(&comm->childAbortFlag, __ATOMIC_ACQUIRE);
if (childAbortFlag != NULL) {
*childAbortFlag = 1;
__atomic_store_n(childAbortFlag, 1, __ATOMIC_RELAXED);
}
*comm->abortFlag = 1;
__atomic_store_n(comm->abortFlag, 1, __ATOMIC_RELAXED);
/* init thread must be joined before we destroy the comm,
* and we should ignore the init error here. */
ncclCommEnsureReady(comm);
@@ -2556,98 +2670,6 @@ ncclResult_t ncclCommUserRank(const ncclComm_t comm, int* rank) {
return ncclSuccess;
}
NCCL_PARAM(LocalRegister, "LOCAL_REGISTER", 1);
NCCL_API(ncclResult_t, ncclCommRegister, const ncclComm_t comm, void* buff, size_t size, void** handle);
ncclResult_t ncclCommRegister(const ncclComm_t comm, void* buff, size_t size, void** handle) {
NVTX3_FUNC_RANGE_IN(nccl_domain);
ncclResult_t ret = ncclSuccess;
#if CUDART_VERSION >= 12010
size_t granularity;
if (ncclParamLocalRegister()) {
if (comm == NCCL_COMM_NULL || buff == NULL || handle == NULL || size == 0) {
WARN("Invalid arguments comm %p, buff %p, size %ld, handle %p", comm, buff, size, handle);
ret = ncclInvalidArgument;
} else if (comm->nvlsSupport) {
CUmulticastObjectProp prop = comm->nvlsResources->properties;
prop.size = size;
CUCHECK(cuMulticastGetGranularity(&granularity, &prop, CU_MULTICAST_GRANULARITY_RECOMMENDED));
if ((uintptr_t)buff % comm->nvlsResources->ucGran == 0 && size % granularity == 0) {
/* we can direct register what user provide */
struct ncclRegRequest* req;
NCCLCHECK(ncclCalloc(&req, 1));
req->buff = (uintptr_t)buff;
req->size = size;
ncclIntruQueueEnqueue(&comm->regRequestQueue, req);
*handle = (void*)req;
} else {
void* base;
size_t baseSize;
/* Since we don't provide actually allocated buffer size for users by ncclMemAlloc,
* therefore, we need to get the full range of the buffer by cuMemGetAddressRange to
* register buffers. */
CUCHECK(cuMemGetAddressRange((CUdeviceptr*)&base, &baseSize, (CUdeviceptr)buff));
if ((uintptr_t)base % comm->nvlsResources->ucGran == 0 && baseSize % granularity == 0) {
struct ncclRegRequest* req;
NCCLCHECK(ncclCalloc(&req, 1));
req->buff = (uintptr_t)base;
req->size = baseSize;
ncclIntruQueueEnqueue(&comm->regRequestQueue, req);
*handle = (void*)req;
} else {
WARN("register fails, buffer %p (aligned %s, granularity %ld) and size %ld (aligned %s, granularity %ld) for registration", buff, (uintptr_t)buff % comm->nvlsResources->ucGran == 0 ? "TRUE" : "FALSE", comm->nvlsResources->ucGran, size, size % granularity == 0 ? "TRUE" : "FALSE", granularity);
ret = ncclInvalidArgument;
}
}
}
}
#endif
return ret;
}
NCCL_API(ncclResult_t, ncclCommDeregister, const ncclComm_t comm, void* handle);
ncclResult_t ncclCommDeregister(const ncclComm_t comm, void* handle) {
ncclResult_t ret = ncclSuccess;
#if CUDART_VERSION >= 12010
struct ncclRegRequest* dreq = (struct ncclRegRequest*)handle;
if (ncclParamLocalRegister()) {
if (comm == NCCL_COMM_NULL || handle == NULL) {
WARN("Invalid arguments comm %p, handle %p", comm, handle);
ret = ncclInvalidArgument;
} else {
struct ncclRegRecord* rec;
/* first release register record */
rec = ncclIntruQueueHead(&comm->regRecordQueue);
while (rec) {
if (rec->buff == dreq->buff && rec->size == dreq->size) {
NCCLCHECK(ncclNvlsDeregBuffer(&rec->mcHandle, rec->regAddr, rec->dev, rec->regSize));
ncclIntruQueueDelete(&comm->regRecordQueue, rec);
free(rec->addrs);
free(rec);
break;
}
rec = rec->next;
}
/* then free register request */
if (ncclIntruQueueDelete(&comm->regRequestQueue, dreq) == false) {
WARN("Invalid handle %p", handle);
ret = ncclInvalidArgument;
}
}
}
#endif
return ret;
}
NCCL_API(ncclResult_t, ncclMemAlloc, void **ptr, size_t size);
ncclResult_t ncclMemAlloc(void **ptr, size_t size) {
NVTX3_FUNC_RANGE_IN(nccl_domain);
@@ -2759,4 +2781,4 @@ exit:
return ret;
fail:
goto exit;
}
}