From dbde26e68152a0c5df3e3fb48434c906f6461e9f Mon Sep 17 00:00:00 2001 From: Wenkai Du <43822138+wenkaidu@users.noreply.github.com> Date: Wed, 30 Sep 2020 16:25:36 -0700 Subject: [PATCH] Add Alltoallv RCCL kernel implementation (#269) * Add alltoallv API and implementation * Extend Rome P2P channel limit to multinode and alltoall kernels * topo_expl: fix compilation and sync up with main * gtest: use RCCL alltoallv API * Code review changes [ROCm/rccl commit: b871ea3c0c8690a340818b788321ba8a8942124f] --- projects/rccl/CMakeLists.txt | 2 + projects/rccl/src/channel.cc | 2 + .../rccl/src/collectives/all_to_allv_api.cc | 45 +++++++++ .../rccl/src/collectives/device/all_to_all.h | 2 +- .../src/collectives/device/all_to_allv.cu | 12 +++ .../rccl/src/collectives/device/all_to_allv.h | 95 +++++++++++++++++++ projects/rccl/src/collectives/device/common.h | 5 + projects/rccl/src/collectives/device/gather.h | 2 +- .../rccl/src/collectives/device/scatter.h | 2 +- projects/rccl/src/enqueue.cc | 58 +++++++---- projects/rccl/src/graph/connect.cc | 2 +- projects/rccl/src/graph/paths.cc | 2 +- projects/rccl/src/include/collectives.h | 5 +- projects/rccl/src/include/devcomm.h | 12 ++- projects/rccl/src/include/info.h | 5 + projects/rccl/src/init.cc | 13 ++- projects/rccl/src/misc/argcheck.cc | 6 +- projects/rccl/src/nccl.h.in | 19 ++++ projects/rccl/src/proxy.cc | 20 +++- projects/rccl/test/test_AllToAllv.cpp | 25 +---- projects/rccl/tools/topo_expl/include/utils.h | 1 + projects/rccl/tools/topo_expl/utils.cpp | 25 +++-- 22 files changed, 300 insertions(+), 60 deletions(-) create mode 100755 projects/rccl/src/collectives/all_to_allv_api.cc create mode 100755 projects/rccl/src/collectives/device/all_to_allv.cu create mode 100755 projects/rccl/src/collectives/device/all_to_allv.h diff --git a/projects/rccl/CMakeLists.txt b/projects/rccl/CMakeLists.txt index dfec56d765..450dd48c22 100644 --- a/projects/rccl/CMakeLists.txt +++ b/projects/rccl/CMakeLists.txt @@ -92,6 +92,7 @@ set(CU_SOURCES src/collectives/device/gather.cu src/collectives/device/scatter.cu src/collectives/device/all_to_all.cu + src/collectives/device/all_to_allv.cu src/collectives/device/functions.cu) set(CPP_SOURCES) @@ -123,6 +124,7 @@ set(CC_SOURCES src/collectives/gather_api.cc src/collectives/scatter_api.cc src/collectives/all_to_all_api.cc + src/collectives/all_to_allv_api.cc src/channel.cc src/misc/argcheck.cc src/misc/nvmlwrap_stub.cc diff --git a/projects/rccl/src/channel.cc b/projects/rccl/src/channel.cc index 97d573edfd..8b87a68f68 100644 --- a/projects/rccl/src/channel.cc +++ b/projects/rccl/src/channel.cc @@ -27,12 +27,14 @@ ncclResult_t initChannel(struct ncclComm* comm, int channelid) { // Per-channel operation list. NCCLCHECK(ncclCudaHostCalloc(&channel->collectives, NCCL_MAX_OPS)); + NCCLCHECK(ncclCudaHostCalloc(&channel->collectivesExtra, comm->nRanks*NCCL_MAX_OPS*4)); return ncclSuccess; } ncclResult_t freeChannel(struct ncclChannel* channel, int nRanks) { if (channel->id == -1) return ncclSuccess; // Operation list + NCCLCHECK(ncclCudaHostFree(channel->collectivesExtra)); NCCLCHECK(ncclCudaHostFree(channel->collectives)); // Free Ring index to rank tables diff --git a/projects/rccl/src/collectives/all_to_allv_api.cc b/projects/rccl/src/collectives/all_to_allv_api.cc new file mode 100755 index 0000000000..d3b499bcd7 --- /dev/null +++ b/projects/rccl/src/collectives/all_to_allv_api.cc @@ -0,0 +1,45 @@ +/************************************************************************* + * Copyright (c) 2015-2019, NVIDIA CORPORATION. All rights reserved. + * Modifications Copyright (c) 2019-2020 Advanced Micro Devices, Inc. All rights reserved. + * + * See LICENSE.txt for license information + ************************************************************************/ + +#include "enqueue.h" +#include "collectives.h" + +NCCL_API(ncclResult_t, ncclAllToAllv, const void *sendbuff, const size_t sendcounts[], const size_t sdispls[], + void *recvbuff, const size_t recvcounts[], const size_t rdispls[], + ncclDataType_t datatype, ncclComm_t comm, hipStream_t stream); +ncclResult_t ncclAllToAllv(const void *sendbuff, const size_t sendcounts[], const size_t sdispls[], + void *recvbuff, const size_t recvcounts[], const size_t rdispls[], + ncclDataType_t datatype, ncclComm_t comm, hipStream_t stream) { + if (comm->alltoallDisable) { + int nRanks; + NCCLCHECK(ncclCommCount(comm, &nRanks)); + NCCLCHECK(ncclGroupStart()); + for (int r=0; rdevUserRanks[0]; const int stepSize = comm->buffSizes[NCCL_PROTO_SIMPLE] / (sizeof(T)*NCCL_STEPS); const int chunkSize = stepSize * ALLTOALL_CHUNKSTEPS; - const int peersPerChan = (nChannels >= nranks ? 1 : DIVUP(nranks, nChannels)); + const int peersPerChan = DIVUP(nranks, nChannels); const ssize_t loopSize = (peersPerChan == 1 ? (nChannels/nranks)*(ssize_t)chunkSize : (ssize_t)chunkSize); // Compute pointers diff --git a/projects/rccl/src/collectives/device/all_to_allv.cu b/projects/rccl/src/collectives/device/all_to_allv.cu new file mode 100755 index 0000000000..f78042c939 --- /dev/null +++ b/projects/rccl/src/collectives/device/all_to_allv.cu @@ -0,0 +1,12 @@ +/************************************************************************* + * Copyright (c) 2015-2019, NVIDIA CORPORATION. All rights reserved. + * + * See LICENSE.txt for license information + ************************************************************************/ + +#include "all_to_allv.h" +#include "common.h" +#include "collectives.h" + +IMPL_COLL_FUNC(ncclAllToAllv, copy, FuncSum, i8, int8_t); +IMPL_COLL_KERN(ncclAllToAllv, copy, FuncSum, i8, int8_t, 0); diff --git a/projects/rccl/src/collectives/device/all_to_allv.h b/projects/rccl/src/collectives/device/all_to_allv.h new file mode 100755 index 0000000000..c1bc3cf0d2 --- /dev/null +++ b/projects/rccl/src/collectives/device/all_to_allv.h @@ -0,0 +1,95 @@ +/************************************************************************* + * Copyright (c) 2015-2019, NVIDIA CORPORATION. All rights reserved. + * Modifications Copyright (c) 2019-2020 Advanced Micro Devices, Inc. All rights reserved. + * + * See LICENSE.txt for license information + ************************************************************************/ + +#include "devcomm.h" +#include "primitives.h" +#include "collectives.h" + +template +__attribute__((noinline)) +__device__ void ncclAllToAllvKernel(struct CollectiveArgs* args) { + const int tid = threadIdx.x; + const int nthreads = args->a2av.nThreads; + const int nChannels = args->a2av.nChannels; + struct ncclDevComm* comm = args->comm; + struct ncclChannel* channel = comm->channels+blockIdx.x; + struct ncclRing* ring = &channel->ring; + const ssize_t typesize = args->a2av.count; + const int nranks = comm->nRanks; + const int bid = args->a2av.bid; + const int rank = ring->devUserRanks[0]; + const int stepSize = comm->buffSizes[NCCL_PROTO_SIMPLE] / (sizeof(T)*NCCL_STEPS); + const int chunkSize = stepSize * ALLTOALLV_CHUNKSTEPS; + const int peersPerChan = DIVUP(nranks, nChannels); + const ssize_t loopSize = (peersPerChan == 1 ? (nChannels/nranks)*(ssize_t)chunkSize : (ssize_t)chunkSize); + + // Compute pointers + const T * __restrict__ thisInput = (const T*)args->sendbuff; + T * __restrict__ thisOutput = (T*)args->recvbuff; + + size_t *sendcounts = args->a2av.extra; + size_t *sdispls = args->a2av.extra + nranks; + size_t *recvcounts = args->a2av.extra + nranks*2; + size_t *rdispls = args->a2av.extra + nranks*3; + ssize_t size = sendcounts[rank]*typesize; + for (ssize_t gridOffset = 0; gridOffset < size; gridOffset += loopSize) { + for (int i = 0; i < peersPerChan; i++) { + if ((peersPerChan == 1 && blockIdx.x >= (nChannels/nranks)*nranks) || + (peersPerChan > 1 && blockIdx.x*peersPerChan+i >= nranks)) + continue; + int realChunkSize = min(chunkSize, DIVUP(size-gridOffset, (peersPerChan == 1 ? (nChannels/nranks) : 1))); + ALIGN_SIZE(realChunkSize, nthreads*sizeof(uint64_t)/sizeof(T)); + ssize_t chunkOffset = gridOffset + (peersPerChan == 1 ? (bid/nranks)*realChunkSize : 0); + int nelem = min(realChunkSize, size-chunkOffset); + if ((blockIdx.x*peersPerChan+i)%nranks == 0) { + if (tid < nthreads && thisInput != thisOutput) { + const T* sendbuff = thisInput+chunkOffset+sdispls[rank]*typesize; + T* recvbuff = thisOutput+chunkOffset+rdispls[rank]*typesize; + // local copy + ReduceOrCopyMulti(tid, nthreads, 1, &sendbuff, 1, &recvbuff, nelem); + } + } + } + } + + for (int i = 0; i < peersPerChan; i++) { + if ((peersPerChan == 1 && blockIdx.x >= (nChannels/nranks)*nranks) || + (peersPerChan > 1 && blockIdx.x*peersPerChan+i >= nranks)) + continue; + if ((blockIdx.x*peersPerChan+i)%nranks != 0) { + int nthreadsSplit = nthreads/2; + int peerNone[2] = {-1,-1}; + if (tid < nthreadsSplit ) { + int peerSend = (rank+(blockIdx.x*peersPerChan)+i)%nranks; + ncclPrimitives + prims(tid, nthreadsSplit, peerNone, &peerSend, NULL, stepSize, channel, comm); + size = sendcounts[peerSend]*typesize; + for (ssize_t gridOffset = 0; gridOffset < size; gridOffset += loopSize) { + int realChunkSize = min(chunkSize, DIVUP(size-gridOffset, (peersPerChan == 1 ? (nChannels/nranks) : 1))); + ALIGN_SIZE(realChunkSize, nthreadsSplit*sizeof(uint64_t)/sizeof(T)); + ssize_t chunkOffset = gridOffset + (peersPerChan == 1 ? (bid/nranks)*realChunkSize : 0); + int nelem = min(realChunkSize, size-chunkOffset); + ssize_t send_offset = chunkOffset + sdispls[peerSend]*typesize; + prims.send(thisInput+send_offset, nelem); + } + } else { + int peerRecv = (2*nranks+rank-((blockIdx.x*peersPerChan)%nranks)-(i%nranks))%nranks; + ncclPrimitives + prims(tid-nthreadsSplit, nthreads-nthreadsSplit, &peerRecv, peerNone, NULL, stepSize, channel, comm); + size = recvcounts[peerRecv]*typesize; + for (ssize_t gridOffset = 0; gridOffset < size; gridOffset += loopSize) { + int realChunkSize = min(chunkSize, DIVUP(size-gridOffset, (peersPerChan == 1 ? (nChannels/nranks) : 1))); + ALIGN_SIZE(realChunkSize, (nthreads-nthreadsSplit)*sizeof(uint64_t)/sizeof(T)); + ssize_t chunkOffset = gridOffset + (peersPerChan == 1 ? (bid/nranks)*realChunkSize : 0); + int nelem = min(realChunkSize, size-chunkOffset); + ssize_t recv_offset = chunkOffset + rdispls[peerRecv]*typesize; + prims.recv(thisOutput+recv_offset, nelem); + } + } + } + } +} diff --git a/projects/rccl/src/collectives/device/common.h b/projects/rccl/src/collectives/device/common.h index 76bcb68d5c..f13957dd7e 100644 --- a/projects/rccl/src/collectives/device/common.h +++ b/projects/rccl/src/collectives/device/common.h @@ -99,6 +99,7 @@ static inline __device__ void exitIfAbortBarrier(int abort) { NCCL_COLL_NAME(ncclGather, copy, i8), \ NCCL_COLL_NAME(ncclScatter, copy, i8), \ NCCL_COLL_NAME(ncclAllToAll, copy, i8), \ + NCCL_COLL_NAME(ncclAllToAllv, copy, i8), \ NCCL_COLL_NAME(ncclSendRecv, copy, i8) } // Must be consistent with the ncclFuncSet enum @@ -117,6 +118,7 @@ static const __device__ constexpr ncclKernelFunc_t ncclFuncs[]{ NCCL_COLL_NAME(ncclGather, copy, i8), NCCL_COLL_NAME(ncclScatter, copy, i8), NCCL_COLL_NAME(ncclAllToAll, copy, i8), + NCCL_COLL_NAME(ncclAllToAllv, copy, i8), NCCL_COLL_NAME(ncclSendRecv, copy, i8) #endif }; @@ -174,6 +176,9 @@ void NCCL_CALL_FUNCTIONS(struct ncclColl* const c) noexcept { else if (c->funcIndex == 1802) { ncclAllToAll_copy_i8(&c->args); } + else if (c->funcIndex == 1803) { + ncclAllToAllv_copy_i8(&c->args); + } else ncclSendRecv_copy_i8(&c->args); } diff --git a/projects/rccl/src/collectives/device/gather.h b/projects/rccl/src/collectives/device/gather.h index f263c1db28..c704e39421 100644 --- a/projects/rccl/src/collectives/device/gather.h +++ b/projects/rccl/src/collectives/device/gather.h @@ -24,7 +24,7 @@ __device__ void ncclGatherKernel(struct CollectiveArgs* args) { const int rank = ring->devUserRanks[0]; const int stepSize = comm->buffSizes[NCCL_PROTO_SIMPLE] / (sizeof(T)*NCCL_STEPS); const int chunkSize = stepSize * GATHER_CHUNKSTEPS; - const int peersPerChan = (nChannels >= nranks ? 1 : DIVUP(nranks, nChannels)); + const int peersPerChan = DIVUP(nranks, nChannels); const ssize_t loopSize = (peersPerChan == 1 ? (nChannels/nranks)*(ssize_t)chunkSize : (ssize_t)chunkSize); const int root = args->coll.root; diff --git a/projects/rccl/src/collectives/device/scatter.h b/projects/rccl/src/collectives/device/scatter.h index ccc7a1932f..f08544068a 100644 --- a/projects/rccl/src/collectives/device/scatter.h +++ b/projects/rccl/src/collectives/device/scatter.h @@ -24,7 +24,7 @@ __device__ void ncclScatterKernel(struct CollectiveArgs* args) { const int rank = ring->devUserRanks[0]; const int stepSize = comm->buffSizes[NCCL_PROTO_SIMPLE] / (sizeof(T)*NCCL_STEPS); const int chunkSize = stepSize * SCATTER_CHUNKSTEPS; - const int peersPerChan = (nChannels >= nranks ? 1 : DIVUP(nranks, nChannels)); + const int peersPerChan = DIVUP(nranks, nChannels); const ssize_t loopSize = (peersPerChan == 1 ? (nChannels/nranks)*(ssize_t)chunkSize : (ssize_t)chunkSize); const int root = args->coll.root; diff --git a/projects/rccl/src/enqueue.cc b/projects/rccl/src/enqueue.cc index 1524909055..7f52e089ce 100644 --- a/projects/rccl/src/enqueue.cc +++ b/projects/rccl/src/enqueue.cc @@ -8,6 +8,7 @@ #include "enqueue.h" #include "argcheck.h" #include "coll_net.h" +#include "../graph/topo.h" // Only generate inline kernels for LL #define NCCL_FUNC5(coll, op, dtype) \ @@ -272,7 +273,7 @@ static ncclResult_t getAlgoInfo(struct ncclInfo* info) { } } } - if (info->coll == ncclCollAllToAll || info->coll == ncclCollGather || info->coll == ncclCollScatter) { + if (info->coll == ncclCollAllToAll || info->coll == ncclCollGather || info->coll == ncclCollScatter || info->coll == ncclCollAllToAllv) { info->algorithm = NCCL_ALGO_RING; info->protocol = NCCL_PROTO_SIMPLE; } @@ -284,12 +285,15 @@ static ncclResult_t getAlgoInfo(struct ncclInfo* info) { TRACE(NCCL_COLL, "%ld Bytes -> Algo %d proto %d time %f", info->nBytes, info->algorithm, info->protocol, minTime); int nc = (info->algorithm == NCCL_ALGO_COLLNET) ? comm->nChannels/2 : comm->nChannels; // CollNet uses one channel for up and one channel for down + if (info->comm->topo->type == RCCL_TOPO_4P2H_ROME && (info->coll == ncclCollAllToAll || + info->coll == ncclCollGather || info->coll == ncclCollScatter || info->coll == ncclCollAllToAllv)) + nc = 2; int nt = comm->maxThreads[info->algorithm][info->protocol]; int threadThreshold = comm->threadThresholds[info->algorithm][info->protocol]; while (info->nBytes < nc*nt*threadThreshold) { // do not reduce channels in case of alltoall if (info->algorithm != NCCL_ALGO_COLLNET && info->coll != ncclCollAllToAll && - info->coll != ncclCollGather && info->coll != ncclCollScatter && nc >= 2) nc--; + info->coll != ncclCollGather && info->coll != ncclCollScatter && info->coll != ncclCollAllToAllv && nc >= 2) nc--; #if defined(__HIP_PLATFORM_HCC__) || defined(__HCC__) || defined(__HIPCC__) // do not reduce threads count on VEGA #else @@ -320,6 +324,7 @@ static ncclResult_t getPatternInfo(struct ncclInfo* info) { case ncclCollGather: case ncclCollScatter: case ncclCollAllToAll: + case ncclCollAllToAllv: info->pattern = ncclPatternAll; break; default: WARN("Unknown pattern for collective %d algorithm %d", info->coll, info->algorithm); @@ -337,11 +342,8 @@ static ncclResult_t getLoopInfo(struct ncclInfo* info) { case ncclPatternPipelineTo: case ncclPatternCollTreeUp: case ncclPatternCollTreeDown: - info->nstepsPerLoop = info->nchunksPerLoop = 1; break; case ncclPatternAll: - info->nstepsPerLoop = 1; - info->nchunksPerLoop = info->comm->nRanks; - break; + info->nstepsPerLoop = info->nchunksPerLoop = 1; break; case ncclPatternRing: info->nstepsPerLoop = info->comm->nRanks-1; info->nchunksPerLoop = info->comm->nRanks; break; case ncclPatternRingTwice: @@ -376,10 +378,16 @@ static ncclResult_t computeColl(struct ncclInfo* info /* input */, struct ncclCo NCCLCHECK(getPatternInfo(info)); NCCLCHECK(getLoopInfo(info)); - coll->args.coll.root = info->root; - coll->args.coll.count = info->count; - coll->args.coll.nChannels = info->nChannels; - coll->args.coll.nThreads = info->nThreads; + if (info->coll == ncclCollAllToAllv) { + coll->args.a2av.count = info->count; + coll->args.a2av.nChannels = info->nChannels; + coll->args.a2av.nThreads = info->nThreads; + } else { + coll->args.coll.root = info->root; + coll->args.coll.count = info->count; + coll->args.coll.nChannels = info->nChannels; + coll->args.coll.nThreads = info->nThreads; + } coll->funcIndex = FUNC_INDEX(info->coll, info->op, info->datatype, info->algorithm, info->protocol); @@ -430,7 +438,7 @@ static ncclResult_t computeColl(struct ncclInfo* info /* input */, struct ncclCo if (info->pattern != ncclPatternAll) nLoops = (int)(DIVUP(info->nBytes, (((size_t)(info->nChannels))*info->nchunksPerLoop*chunkEffectiveSize))); else - nLoops = (int)(DIVUP(info->nBytes, (((size_t)((info->nChannels >= info->comm->nRanks ? (info->nChannels/info->comm->nRanks) : 1))))*info->nchunksPerLoop*chunkEffectiveSize)); + nLoops = (int)(DIVUP(info->nBytes, (((size_t)((info->nChannels >= info->comm->nRanks ? (info->nChannels/info->comm->nRanks) : 1))))*info->comm->nRanks*info->nchunksPerLoop*chunkEffectiveSize)); proxyArgs->nsteps = info->nstepsPerLoop * nLoops * chunkSteps; proxyArgs->sliceSteps = sliceSteps; proxyArgs->chunkSteps = chunkSteps; @@ -438,7 +446,7 @@ static ncclResult_t computeColl(struct ncclInfo* info /* input */, struct ncclCo proxyArgs->opCount = info->comm->opCount; proxyArgs->dtype = info->datatype; proxyArgs->redOp = info->op; - TRACE(NCCL_NET,"opCount %lx slicesteps %d spl %d cpl %d ces %d nbytes %zi -> protocol %d nchannels %d nthreads %d, nloops %d nsteps %d comm %p", + if (info->coll != ncclCollAllToAllv) TRACE(NCCL_NET,"opCount %lx slicesteps %d spl %d cpl %d ces %d nbytes %zi -> protocol %d nchannels %d nthreads %d, nloops %d nsteps %d comm %p", coll->args.opCount, proxyArgs->sliceSteps, info->nstepsPerLoop, info->nchunksPerLoop, chunkEffectiveSize, info->nBytes, info->protocol, info->nChannels, info->nThreads, nLoops, proxyArgs->nsteps, info->comm); return ncclSuccess; @@ -492,7 +500,7 @@ ncclResult_t ncclSaveKernel(struct ncclInfo* info) { if (info->coll == ncclCollSendRecv) { info->comm->myParams->gridDim.x = std::max(info->comm->myParams->gridDim.x, channelId+1); NCCLCHECK(ncclProxySaveP2p(info, channel)); - } else if (info->coll == ncclCollAllToAll || info->coll == ncclCollScatter || info->coll == ncclCollGather) { + } else if (info->coll == ncclCollAllToAll || info->coll == ncclCollScatter || info->coll == ncclCollGather || info->coll == ncclCollAllToAllv) { NCCLCHECK(ncclProxySaveA2a(&proxyArgs, info)); } else { NCCLCHECK(ncclProxySaveColl(&proxyArgs, info->pattern, info->root, info->comm->nRanks)); @@ -504,7 +512,15 @@ ncclResult_t ncclSaveKernel(struct ncclInfo* info) { while (LOAD(activePtr) != 0) sched_yield(); memcpy(c, &coll, sizeof(struct ncclColl)); - if (info->coll != ncclCollSendRecv) c->args.coll.bid = bid % coll.args.coll.nChannels; + if (info->coll == ncclCollAllToAllv) { + c->args.a2av.extra = channel->collectivesExtra + info->comm->nRanks*4*opIndex; + memcpy(c->args.a2av.extra, info->sendcounts, sizeof(size_t*)*(info->comm->nRanks)); + memcpy(c->args.a2av.extra+info->comm->nRanks, info->sdispls, sizeof(size_t*)*(info->comm->nRanks)); + memcpy(c->args.a2av.extra+info->comm->nRanks*2, info->recvcounts, sizeof(size_t*)*(info->comm->nRanks)); + memcpy(c->args.a2av.extra+info->comm->nRanks*3, info->rdispls, sizeof(size_t*)*(info->comm->nRanks)); + c->args.a2av.bid = bid % coll.args.coll.nChannels; + } else if (info->coll != ncclCollSendRecv) + c->args.coll.bid = bid % coll.args.coll.nChannels; STORE(&c->active, 1); opIndex = (opIndex+1)%NCCL_MAX_OPS; @@ -569,7 +585,12 @@ ncclResult_t ncclEnqueueCheck(struct ncclInfo* info) { NCCLCHECKGOTO(ncclAsyncColl(info->comm), ret, end); NCCLCHECKGOTO(checkSetStream(info), ret, end); - INFO(NCCL_COLL,"%s: opCount %lx sendbuff %p recvbuff %p count %zi datatype %d op %d root %d comm %p [nranks=%d] stream %p", + if (info->coll == ncclCollAllToAllv) + INFO(NCCL_COLL,"%s: opCount %lx sendbuff %p sendcounts %p sdispls %p recvbuff %p recvcounts %p rdispls %p datatype %d typesize %zi op %d root %d comm %p [nranks=%d] stream %p", + info->opName, info->comm->opCount, info->sendbuff, info->sendcounts, info->sdispls, info->recvbuff, info->recvcounts, info->rdispls, + info->datatype, info->count, info->op, info->root, info->comm, info->comm->nRanks, info->stream); + else + INFO(NCCL_COLL,"%s: opCount %lx sendbuff %p recvbuff %p count %zi datatype %d op %d root %d comm %p [nranks=%d] stream %p", info->opName, info->comm->opCount, info->sendbuff, info->recvbuff, info->count, info->datatype, info->op, info->root, info->comm, info->comm->nRanks, info->stream); @@ -587,7 +608,12 @@ end: NCCLCHECK(ArgsCheck(info)); NCCLCHECK(checkSetStream(info)); - INFO(NCCL_COLL,"%s: opCount %lx sendbuff %p recvbuff %p count %zi datatype %d op %d root %d comm %p [nranks=%d] stream %p", + if (info->coll == ncclCollAllToAllv) + INFO(NCCL_COLL,"%s: opCount %lx sendbuff %p sendcounts %p sdispls %p recvbuff %p recvcounts %p rdispls %p datatype %d typesize %zi op %d root %d comm %p [nranks=%d] stream %p", + info->opName, info->comm->opCount, info->sendbuff, info->sendcounts, info->sdispls, info->recvbuff, info->recvcounts, info->rdispls, + info->datatype, info->count, info->op, info->root, info->comm, info->comm->nRanks, info->stream); + else + INFO(NCCL_COLL,"%s: opCount %lx sendbuff %p recvbuff %p count %zi datatype %d op %d root %d comm %p [nranks=%d] stream %p", info->opName, info->comm->opCount, info->sendbuff, info->recvbuff, info->count, info->datatype, info->op, info->root, info->comm, info->comm->nRanks, info->stream); diff --git a/projects/rccl/src/graph/connect.cc b/projects/rccl/src/graph/connect.cc index 622f8f4dab..76a10c5fc4 100644 --- a/projects/rccl/src/graph/connect.cc +++ b/projects/rccl/src/graph/connect.cc @@ -291,7 +291,7 @@ ncclResult_t ncclTopoPostset(struct ncclComm* comm, int* firstRanks, struct nccl int nc = nChannels*2; if (gcn == 908) nc = std::max(nc, 4); - if (comm->topo->nodes[NET].count == 0 && comm->topo->type == RCCL_TOPO_CR8G) nc = nChannels*3; + if (comm->topo->nodes[NET].count == 0 && comm->topo->type == RCCL_TOPO_CR8G) nc = nChannels*4; if (comm->topo->nodes[NET].count && comm->topo->type == RCCL_TOPO_4P2H_ROME) nc = 4*comm->topo->nodes[NET].count; int end = std::min((int)ncclMaxNchannels(), std::max(nc, ncclMinNchannels())); diff --git a/projects/rccl/src/graph/paths.cc b/projects/rccl/src/graph/paths.cc index 045aba155c..2aeb9fbb20 100644 --- a/projects/rccl/src/graph/paths.cc +++ b/projects/rccl/src/graph/paths.cc @@ -599,7 +599,7 @@ ncclResult_t ncclTopoComputeP2pChannels(struct ncclComm* comm) { } } - if (comm->topo->nodes[NET].count == 0 && comm->topo->type == RCCL_TOPO_4P2H_ROME) { + if (comm->topo->type == RCCL_TOPO_4P2H_ROME) { // Adjust P2P channels on Rome comm->p2pnChannelsPerPeer = 2; comm->p2pnChannels = 2; diff --git a/projects/rccl/src/include/collectives.h b/projects/rccl/src/include/collectives.h index 56c268ba19..ea66063f87 100644 --- a/projects/rccl/src/include/collectives.h +++ b/projects/rccl/src/include/collectives.h @@ -8,7 +8,7 @@ #ifndef NCCL_COLLECTIVES_H_ #define NCCL_COLLECTIVES_H_ -#define FUNC_INDEX_P2P (3+NCCL_NUM_FUNCTIONS*NCCL_NUM_ALGORITHMS*NCCL_NUM_PROTOCOLS*ncclNumTypes*ncclNumOps) +#define FUNC_INDEX_P2P (4+NCCL_NUM_FUNCTIONS*NCCL_NUM_ALGORITHMS*NCCL_NUM_PROTOCOLS*ncclNumTypes*ncclNumOps) #define FUNC_INDEX(coll, redop, dtype, al, pr) ((coll >= NCCL_NUM_FUNCTIONS) \ ? (coll-NCCL_NUM_FUNCTIONS+NCCL_NUM_FUNCTIONS*NCCL_NUM_ALGORITHMS*NCCL_NUM_PROTOCOLS*ncclNumTypes*ncclNumOps) \ : ((((((coll)*ncclNumOps + (redop))*ncclNumTypes) + (dtype))*NCCL_NUM_ALGORITHMS+(al))*NCCL_NUM_PROTOCOLS+(pr))) @@ -61,6 +61,7 @@ DECL_COLL5(ncclGather, copy, i8) \ DECL_COLL5(ncclScatter, copy, i8) \ DECL_COLL5(ncclAllToAll, copy, i8) \ + DECL_COLL5(ncclAllToAllv, copy, i8) \ DECL_COLL5(ncclSendRecv, copy, i8) \ DECL_ALL_COLLS @@ -89,5 +90,7 @@ DECL_ALL_COLLS #define SCATTER_CHUNKSTEPS 4 #define ALLTOALL_SLICESTEPS 4 #define ALLTOALL_CHUNKSTEPS 4 +#define ALLTOALLV_SLICESTEPS 4 +#define ALLTOALLV_CHUNKSTEPS 4 #endif diff --git a/projects/rccl/src/include/devcomm.h b/projects/rccl/src/include/devcomm.h index 5921477cf1..f7b870c5e5 100644 --- a/projects/rccl/src/include/devcomm.h +++ b/projects/rccl/src/include/devcomm.h @@ -23,8 +23,8 @@ #endif #define NCCL_NUM_FUNCTIONS 5 // SendRecv not included for now -typedef enum { ncclCollBroadcast, ncclCollReduce, ncclCollAllGather, ncclCollReduceScatter, ncclCollAllReduce, ncclCollGather, ncclCollScatter, ncclCollAllToAll, ncclCollSendRecv} ncclFunc_t; -extern const char* ncclFuncStr[NCCL_NUM_FUNCTIONS+3]; +typedef enum { ncclCollBroadcast, ncclCollReduce, ncclCollAllGather, ncclCollReduceScatter, ncclCollAllReduce, ncclCollGather, ncclCollScatter, ncclCollAllToAll, ncclCollAllToAllv, ncclCollSendRecv} ncclFunc_t; +extern const char* ncclFuncStr[NCCL_NUM_FUNCTIONS+4]; #define NCCL_NUM_ALGORITHMS 3 // Tree/Ring/CollNet #define NCCL_ALGO_TREE 0 @@ -183,6 +183,13 @@ struct CollectiveArgs { size_t sendCount; size_t recvCount; } p2p; + struct { + uint16_t nThreads; + uint8_t bid; + uint8_t nChannels; + size_t count; + size_t* extra; + } a2av; }; }; struct ncclColl { @@ -215,6 +222,7 @@ struct ncclChannel { // Operation list for aggregation struct ncclColl* collectives; + size_t* collectivesExtra; int collStart; int collCount; int collFifoHead; // Only used by GPU diff --git a/projects/rccl/src/include/info.h b/projects/rccl/src/include/info.h index 11d9bb6db5..fa9402f350 100644 --- a/projects/rccl/src/include/info.h +++ b/projects/rccl/src/include/info.h @@ -40,6 +40,11 @@ struct ncclInfo { // Algorithm details int chunkSteps; int sliceSteps; + // For alltoallv + const size_t *sendcounts; + const size_t *sdispls; + const size_t *recvcounts; + const size_t *rdispls; // Computed later int algorithm; int protocol; diff --git a/projects/rccl/src/init.cc b/projects/rccl/src/init.cc index 371204eeeb..b14d4a1d2b 100644 --- a/projects/rccl/src/init.cc +++ b/projects/rccl/src/init.cc @@ -41,7 +41,7 @@ std::chrono::high_resolution_clock::time_point ncclEpoch; #define NCCL_GROUP_CUDA_STREAM 1 // CGMD: CUDA 9.0,9.1 Need to use an internal CUDA stream #endif -const char* ncclFuncStr[NCCL_NUM_FUNCTIONS+3] = { "Broadcast", "Reduce", "AllGather", "ReduceScatter", "AllReduce", "Gather", "Scatter", "AllToAll" }; +const char* ncclFuncStr[NCCL_NUM_FUNCTIONS+4] = { "Broadcast", "Reduce", "AllGather", "ReduceScatter", "AllReduce", "Gather", "Scatter", "AllToAll", "AllToAllv" }; const char* ncclAlgoStr[NCCL_NUM_ALGORITHMS] = { "Tree", "Ring", "CollNet" }; const char* ncclProtoStr[NCCL_NUM_PROTOCOLS] = { "LL", "LL128", "Simple" }; @@ -847,7 +847,7 @@ static ncclResult_t initTransportsRank(struct ncclComm* comm, ncclUniqueId* comm if (comm->alltoallDisable != alltoallDisable) { comm->alltoallDisable = alltoallDisable; } - INFO(NCCL_INIT, "RCCL AllToAll/Scatter/Gather kernels %s", comm->alltoallDisable ? "disabled" : "enabled"); + INFO(NCCL_INIT, "RCCL AllToAll(v)/Scatter/Gather kernels %s", comm->alltoallDisable ? "disabled" : "enabled"); if (comm->nChannels < nChannelsOrig) { // We started duplicating channels during Preset(), so we need to move the @@ -940,8 +940,11 @@ static ncclResult_t initTransportsRank(struct ncclComm* comm, ncclUniqueId* comm NCCLCHECK(ncclTopoComputeP2pChannels(comm)); if (!alltoallDisable) { - for (int c=0; cnChannels; c++) { - const int peersPerChan = (comm->nChannels >= nranks ? 1 : DIVUP(nranks, comm->nChannels)); + int nc = comm->nChannels; + if (comm->topo->type == RCCL_TOPO_4P2H_ROME) + nc = 2; + for (int c=0; cp2plist.connect; connect->nrecv[c] = 0; connect->nsend[c] = 0; @@ -960,7 +963,7 @@ static ncclResult_t initTransportsRank(struct ncclComm* comm, ncclUniqueId* comm } } - for (int c=0; cnChannels; c++) { + for (int c=0; cchannels+c; struct ncclP2PConnect* connect = &comm->p2plist.connect; #if 0 diff --git a/projects/rccl/src/misc/argcheck.cc b/projects/rccl/src/misc/argcheck.cc index d3531a2b0e..6977c2d6ac 100644 --- a/projects/rccl/src/misc/argcheck.cc +++ b/projects/rccl/src/misc/argcheck.cc @@ -54,7 +54,11 @@ ncclResult_t ArgsCheck(struct ncclInfo* info) { if (info->coll == ncclCollAllGather || info->coll == ncclCollReduceScatter || info->coll == ncclCollGather || info->coll == ncclCollScatter || info->coll == ncclCollAllToAll) info->nBytes *= info->comm->nRanks; // count is per rank - + if (info->coll == ncclCollAllToAllv) { + // Use count to store data type size for alltoallv + info->count = ncclTypeSize(info->datatype); + info->datatype = ncclInt8; + } if (info->op < 0 || info->op >= ncclNumOps) { WARN("%s : invalid reduction operation %d", info->opName, info->op); return ncclInvalidArgument; diff --git a/projects/rccl/src/nccl.h.in b/projects/rccl/src/nccl.h.in index f83cebe0d8..0b6a74cb57 100644 --- a/projects/rccl/src/nccl.h.in +++ b/projects/rccl/src/nccl.h.in @@ -21,6 +21,7 @@ #define RCCL_BFLOAT16 1 #define RCCL_GATHER_SCATTER 1 +#define RCCL_ALLTOALLV 1 #ifdef __cplusplus extern "C" { @@ -313,6 +314,24 @@ ncclResult_t ncclAllToAll(const void* sendbuff, void* recvbuff, size_t count, ncclResult_t pncclAllToAll(const void* sendbuff, void* recvbuff, size_t count, ncclDataType_t datatype, ncclComm_t comm, hipStream_t stream); +/*! @brief All-To-Allv + * + * @details Device (i) sends sendcounts[j] of data from offset sdispls[j] + * to device (j). In the same time, device (i) receives recvcounts[j] of data + * from device (j) to be placed at rdispls[j]. + + * sendcounts, sdispls, recvcounts and rdispls are all measured in the units + * of datatype, not bytes. + * + * In-place operation will happen if sendbuff == recvbuff. + */ +ncclResult_t ncclAllToAllv(const void *sendbuff, const size_t sendcounts[], + const size_t sdispls[], void *recvbuff, const size_t recvcounts[], + const size_t rdispls[], ncclDataType_t datatype, ncclComm_t comm, hipStream_t stream); +ncclResult_t pncclAllToAllv(const void *sendbuff, const size_t sendcounts[], + const size_t sdispls[], void *recvbuff, const size_t recvcounts[], + const size_t rdispls[], ncclDataType_t datatype, ncclComm_t comm, hipStream_t stream); + /* * Group semantics * diff --git a/projects/rccl/src/proxy.cc b/projects/rccl/src/proxy.cc index 636cb4d9a9..6b6cdabb02 100644 --- a/projects/rccl/src/proxy.cc +++ b/projects/rccl/src/proxy.cc @@ -166,7 +166,9 @@ ncclResult_t ncclProxySaveP2p(struct ncclInfo* info, struct ncclChannel* channel } ncclResult_t ncclProxySaveA2a(struct ncclProxyArgs* args, struct ncclInfo* info) { - const int peersPerChan = (info->nChannels >= info->comm->nRanks ? 1 : DIVUP(info->comm->nRanks, info->nChannels)); + const int peersPerChan = DIVUP(info->comm->nRanks, info->nChannels); + const int chunkSize = info->comm->buffSizes[info->protocol]/NCCL_STEPS*info->chunkSteps; + const int loopSize = (info->nChannels >= info->comm->nRanks ? (info->nChannels/info->comm->nRanks) : 1)*info->nchunksPerLoop*chunkSize; for (int p=0; pchannel->id >= (info->nChannels/info->comm->nRanks)*info->comm->nRanks) || (peersPerChan > 1 && args->channel->id*peersPerChan+p >= info->comm->nRanks)) @@ -182,6 +184,22 @@ ncclResult_t ncclProxySaveA2a(struct ncclProxyArgs* args, struct ncclInfo* info) if (info->coll == ncclCollAllToAll || (info->coll == ncclCollGather && info->comm->rank == info->root) || (info->coll == ncclCollScatter && peerRecv == info->root)) NCCLCHECK(SaveProxy(peerRecv, args)); + if (info->coll == ncclCollAllToAllv) { + info->nBytes = info->sendcounts[peerSend]*info->count; + int nLoops = (int)(DIVUP(info->nBytes, loopSize)); + args->nsteps = info->nstepsPerLoop*nLoops*info->chunkSteps; + TRACE(NCCL_NET,"peerSend %d opCount %lx slicesteps %d spl %d cpl %d ces %d nbytes %zi -> protocol %d nchannels %d nthreads %d, nloops %d nsteps %d comm %p", + peerSend, args->opCount, args->sliceSteps, info->nstepsPerLoop, info->nchunksPerLoop, chunkSize, info->nBytes, info->protocol, info->nChannels, info->nThreads, + nLoops, args->nsteps, info->comm); + NCCLCHECK(SaveProxy(peerSend, args)); + info->nBytes = info->recvcounts[peerRecv]*info->count; + nLoops = (int)(DIVUP(info->nBytes, loopSize)); + args->nsteps = info->nstepsPerLoop*nLoops*info->chunkSteps; + TRACE(NCCL_NET,"peerRecv %d opCount %lx slicesteps %d spl %d cpl %d ces %d nbytes %zi -> protocol %d nchannels %d nthreads %d, nloops %d nsteps %d comm %p", + peerRecv, args->opCount, args->sliceSteps, info->nstepsPerLoop, info->nchunksPerLoop, chunkSize, info->nBytes, info->protocol, info->nChannels, info->nThreads, + nLoops, args->nsteps, info->comm); + NCCLCHECK(SaveProxy(peerRecv, args)); + } } return ncclSuccess; } diff --git a/projects/rccl/test/test_AllToAllv.cpp b/projects/rccl/test/test_AllToAllv.cpp index 8223e4e9c8..6f2e317f44 100644 --- a/projects/rccl/test/test_AllToAllv.cpp +++ b/projects/rccl/test/test_AllToAllv.cpp @@ -35,29 +35,10 @@ namespace CorrectnessTests sdispls[i] = rdispls[i] = disp; disp += scount; } - for (int i = 0; i < numDevices; i++) { - if (sendcounts[i] != 0) { - ncclSend( - ((char*)dataset.inputs[r]) + sdispls[i] * DataTypeToBytes(dataType), - sendcounts[i], - dataType, - i, - comms[r], - streams[r]); - } - if (recvcounts[i] != 0) { - ncclRecv( - ((char*)dataset.outputs[r]) + rdispls[i] * DataTypeToBytes(dataType), - recvcounts[i], - dataType, - i, - comms[r], - streams[r]); - } - } + ncclAllToAllv((char*)dataset.inputs[r], sendcounts, sdispls, + (char*)dataset.outputs[r], recvcounts, rdispls, dataType, comms[r], streams[r]); } ncclGroupEnd(); - // Wait for reduction to complete Synchronize(); @@ -89,6 +70,6 @@ namespace CorrectnessTests testing::Values(2,3,4,5,6,7,8), // In-place or not testing::Values(false), - testing::Values("")), + testing::Values("RCCL_ALLTOALL_KERNEL_DISABLE=0", "RCCL_ALLTOALL_KERNEL_DISABLE=1")), CorrectnessTest::PrintToStringParamName()); } // namespace diff --git a/projects/rccl/tools/topo_expl/include/utils.h b/projects/rccl/tools/topo_expl/include/utils.h index 1f9eebdd1d..c60500bdeb 100644 --- a/projects/rccl/tools/topo_expl/include/utils.h +++ b/projects/rccl/tools/topo_expl/include/utils.h @@ -26,6 +26,7 @@ struct allGather3Data_t{ int fullCudaCompCap; int nChannels; int gcn; + int alltoallDisable; struct ncclGraphInfo tree; struct ncclGraphInfo ring; struct ncclGraphInfo collNet; diff --git a/projects/rccl/tools/topo_expl/utils.cpp b/projects/rccl/tools/topo_expl/utils.cpp index 960119ca24..05d84ecf26 100644 --- a/projects/rccl/tools/topo_expl/utils.cpp +++ b/projects/rccl/tools/topo_expl/utils.cpp @@ -30,7 +30,7 @@ #include "model.h" #include "utils.h" -const char* ncclFuncStr[NCCL_NUM_FUNCTIONS+3] = { "Broadcast", "Reduce", "AllGather", "ReduceScatter", "AllReduce", "Gather", "Scatter", "AllToAll" }; +const char* ncclFuncStr[NCCL_NUM_FUNCTIONS+4] = { "Broadcast", "Reduce", "AllGather", "ReduceScatter", "AllReduce", "Gather", "Scatter", "AllToAll", "AllToAllv" }; const char* ncclAlgoStr[NCCL_NUM_ALGORITHMS] = { "Tree", "Ring", "CollNet" }; const char* ncclProtoStr[NCCL_NUM_PROTOCOLS] = { "LL", "LL128", "Simple" }; @@ -220,7 +220,9 @@ ncclResult_t initTransportsRank_1(struct ncclComm* comm, struct allGather1Data_t NCCLCHECK(ncclTopoIdToIndex(comm->topo, GPU, myInfo->busId, &idx)); allGather3Data[rank].cudaCompCap = comm->topo->nodes[GPU].nodes[idx].gpu.cudaCompCap; allGather3Data[rank].gcn = comm->topo->nodes[GPU].nodes[idx].gpu.gcn; - allGather3Data[rank].nChannels = comm->nChannels = std::min(treeGraph.nChannels, ringGraph.nChannels); + allGather3Data[rank].nChannels = comm->nChannels = treeGraph.nChannels = ringGraph.nChannels = + std::min(treeGraph.nChannels, ringGraph.nChannels); + allGather3Data[rank].alltoallDisable = comm->alltoallDisable; allGather3Data[rank].tree.sameChannels = treeGraph.sameChannels; allGather3Data[rank].tree.speedIntra = treeGraph.speedIntra; allGather3Data[rank].tree.speedInter = treeGraph.speedInter; @@ -401,9 +403,11 @@ ncclResult_t initTransportsRank_3(struct ncclComm* comm, struct allGather3Data_t struct ncclTopoRanks** allTopoRanks; NCCLCHECK(ncclCalloc(&allTopoRanks, comm->nRanks)); int gcn = allGather3Data[0].gcn; + int alltoallDisable = 0; for (int i=0; inChannels = std::min(allGather3Data[i].nChannels, comm->nChannels); treeGraph.sameChannels = std::min(allGather3Data[i].tree.sameChannels, treeGraph.sameChannels); @@ -419,6 +423,10 @@ ncclResult_t initTransportsRank_3(struct ncclComm* comm, struct allGather3Data_t collNetGraph.speedInter = std::min(allGather3Data[i].collNet.speedInter, collNetGraph.speedInter); collNetGraph.typeIntra = std::min(allGather3Data[i].collNet.typeIntra, collNetGraph.typeIntra); } + if (comm->alltoallDisable != alltoallDisable) { + comm->alltoallDisable = alltoallDisable; + } + INFO(NCCL_INIT, "RCCL AllToAll(v)/Scatter/Gather kernels %s", comm->alltoallDisable ? "disabled" : "enabled"); if (comm->nChannels < nChannelsOrig) { // We started duplicating channels during Preset(), so we need to move the @@ -432,7 +440,7 @@ ncclResult_t initTransportsRank_3(struct ncclComm* comm, struct allGather3Data_t NCCLCHECK(ncclTopoPostset(comm, nodesFirstRank, allTopoRanks, rings, gcn)); if (comm->nNodes > 1 && ncclParamCollNetEnable() == 1 && - collNetSupport()) { + collNetSupport() && collNetGraph.nChannels) { NCCLCHECK(ncclTopoConnectCollNet(comm, &collNetGraph, rank)); } @@ -511,9 +519,12 @@ ncclResult_t initTransportsRank_3(struct ncclComm* comm, struct allGather3Data_t // Compute nChannels per peer for p2p NCCLCHECK(ncclTopoComputeP2pChannels(comm)); - if (rcclParamAllToAllDisable() == 0) { - for (int c=0; cnChannels; c++) { - const int peersPerChan = (comm->nChannels >= nranks ? 1 : DIVUP(nranks, comm->nChannels)); + if (!alltoallDisable) { + int nc = comm->nChannels; + if (comm->topo->type == RCCL_TOPO_4P2H_ROME) + nc = 2; + for (int c=0; c= nranks ? 1 : DIVUP(nranks, nc)); struct ncclP2PConnect* connect = &comm->p2plist.connect; connect->nrecv[c] = 0; connect->nsend[c] = 0; @@ -532,7 +543,7 @@ ncclResult_t initTransportsRank_3(struct ncclComm* comm, struct allGather3Data_t } } - for (int c=0; cnChannels; c++) { + for (int c=0; cchannels+c; struct ncclP2PConnect* connect = &comm->p2plist.connect; #if 0