AI/rocm-systems
2
0
Forka 0
Codice Problemi Pull Request Actions Pacchetti Progetti Rilasci Wiki Attività

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

Sfoglia il codice sorgente
This commit is contained in:
Wenkai Du
2020-08-06 23:35:44 +00:00
parent c5d4d9eb76 de49a77074
commit 7e3f841fab
24 ha cambiato i file con 55 aggiunte e 204 eliminazioni
Scarica il file Patch Scarica il file Diff Expand all files Collapse all files
makefiles/version.mk
+1 -1
Vedi File
@@ -1,6 +1,6 @@
##### version
NCCL_MAJOR := 2
NCCL_MINOR := 7
NCCL_PATCH := 6
NCCL_PATCH := 8
NCCL_SUFFIX :=
PKG_REVISION := 1
src/Makefile
+2 -2
Vedi File
@@ -46,7 +46,7 @@ lib : $(INCTARGETS) $(LIBDIR)/$(LIBTARGET) $(PKGDIR)/$(PKGTARGET)
staticlib : $(LIBDIR)/$(STATICLIBTARGET)
$(DEVICELIB): ALWAYS_REBUILD
$(DEVICELIB): ALWAYS_REBUILD $(INCTARGETS)
$(MAKE) -C collectives/device
# Empty target to force rebuild
@@ -107,7 +107,7 @@ $(PKGDIR)/%.pc : %.pc
mkdir -p $(PKGDIR)
install -m 644 $< $@
$(OBJDIR)/%.o : %.cc
$(OBJDIR)/%.o : %.cc $(INCTARGETS)
@printf "Compiling %-35s > %s\n" $< $@
mkdir -p `dirname $@`
$(CXX) -I. -I$(INCDIR) $(CXXFLAGS) -Iinclude -c $< -o $@
src/collectives/device/all_gather.h
+3 -3
Vedi File
@@ -30,7 +30,7 @@ __device__ void ncclAllGatherRingKernel(struct CollectiveArgs* args) {
T * __restrict__ thisOutput = (T*)args->recvbuff;
ncclPrimitives<UNROLL, ALLGATHER_CHUNKSTEPS/ALLGATHER_SLICESTEPS, ALLGATHER_SLICESTEPS, T, 1, 1, 1, FUNC>
prims(tid, nthreads, &ring->prev, &ring->next, thisOutput, stepSize, channel, comm, args->opCount);
prims(tid, nthreads, &ring->prev, &ring->next, thisOutput, stepSize, channel, comm);
for (ssize_t gridOffset = 0; gridOffset < size; gridOffset += loopSize) {
int realChunkSize = min(chunkSize, DIVUP(size-gridOffset,nChannels));
@@ -93,7 +93,7 @@ __device__ void ncclAllGatherRingLLKernel(struct CollectiveArgs* args) {
const ssize_t loopSize = nChannels*chunkSize;
const ssize_t size = args->coll.count;
ncclLLPrimitives<T, FUNC, 1, 1> LLprims(tid, nthreads, &ring->prev, &ring->next, stepLines, channel, comm, args->opCount);
ncclLLPrimitives<T, FUNC, 1, 1> LLprims(tid, nthreads, &ring->prev, &ring->next, stepLines, channel, comm);
// Compute pointers
const T * __restrict__ thisInput = (const T*)args->sendbuff;
@@ -163,7 +163,7 @@ __device__ void ncclAllGatherRingLL128Kernel(struct CollectiveArgs* args) {
const ssize_t loopSize = nChannels*chunkSize;
const ssize_t size = args->coll.count;
ncclLL128Primitives<T, FUNC, 1, 1> LLprims(tid, nthreads, &ring->prev, &ring->next, stepSize, channel, comm, args->opCount);
ncclLL128Primitives<T, FUNC, 1, 1> LLprims(tid, nthreads, &ring->prev, &ring->next, stepSize, channel, comm);
// Compute pointers
const T * __restrict__ thisInput = (const T*)args->sendbuff;
src/collectives/device/all_reduce.h
+14 -14
Vedi File
@@ -35,7 +35,7 @@ __device__ void ncclAllReduceRingKernel(struct CollectiveArgs* args) {
T * __restrict__ thisOutput = (T*)args->recvbuff;
ncclPrimitives<UNROLL, ALLREDUCE_CHUNKSTEPS/ALLREDUCE_SLICESTEPS, ALLREDUCE_SLICESTEPS, T, 1, 1, 1, FUNC>
prims(tid, nthreads, &ring->prev, &ring->next, thisOutput, stepSize, channel, comm, args->opCount);
prims(tid, nthreads, &ring->prev, &ring->next, thisOutput, stepSize, channel, comm);
for (ssize_t gridOffset = 0; gridOffset < size; gridOffset += nranks*loopSize) {
ssize_t realChunkSize = min(chunkSize, DIVUP(size-gridOffset,nranks*nChannels));
@@ -129,7 +129,7 @@ __device__ void ncclAllReduceTreeKernel(struct CollectiveArgs* args) {
do {
struct ncclTree* tree = &channel->treeUp;
// Reduce : max number of recv is 3, max number of send is 1 (binary tree + local)
ncclPrimitives<UNROLL/2, 1, 1, T, NCCL_MAX_TREE_ARITY, 1, 0, FUNC> prims(tid, nthreads, tree->down, &tree->up, NULL, stepSize, channel, comm, args->opCount);
ncclPrimitives<UNROLL/2, 1, 1, T, NCCL_MAX_TREE_ARITY, 1, 0, FUNC> prims(tid, nthreads, tree->down, &tree->up, NULL, stepSize, channel, comm);
for (ssize_t gridOffset = 0; gridOffset < size; gridOffset += loopSize) {
// Up
ssize_t offset = gridOffset + bid*chunkSize;
@@ -147,7 +147,7 @@ __device__ void ncclAllReduceTreeKernel(struct CollectiveArgs* args) {
do {
struct ncclTree* tree = &channel->treeDn;
// Broadcast : max number of recv is 1, max number of send is 3 (binary tree + local)
ncclPrimitives<UNROLL/2, 1, 1, T, 1, NCCL_MAX_TREE_ARITY, 1, FUNC> prims(tid, nthreads, &tree->up, tree->down, thisOutput, stepSize, channel, comm, args->opCount);
ncclPrimitives<UNROLL/2, 1, 1, T, 1, NCCL_MAX_TREE_ARITY, 1, FUNC> prims(tid, nthreads, &tree->up, tree->down, thisOutput, stepSize, channel, comm);
for (ssize_t gridOffset = 0; gridOffset < size; gridOffset += loopSize) {
// Down
ssize_t offset = gridOffset + bid*chunkSize;
@@ -188,7 +188,7 @@ __device__ void ncclAllReduceCollNetKernel(struct CollectiveArgs* args) {
if (blockIdx.x < nChannels) { // first half of the channels do reduce
struct ncclTree* tree = &channel->collTreeUp;
ncclPrimitives<UNROLL, 1, 1, T, 1, 1, 0, FUNC> prims(tid, nthreads, tree->down, &tree->up, NULL, stepSize, channel, comm, args->opCount);
ncclPrimitives<UNROLL, 1, 1, T, 1, 1, 0, FUNC> prims(tid, nthreads, tree->down, &tree->up, NULL, stepSize, channel, comm);
for (ssize_t gridOffset = 0; gridOffset < size; gridOffset += loopSize) {
// Up
ssize_t offset = gridOffset + bid*chunkSize;
@@ -205,7 +205,7 @@ __device__ void ncclAllReduceCollNetKernel(struct CollectiveArgs* args) {
if (blockIdx.x >= nChannels) { // second half of the channels do broadcast
struct ncclTree* tree = &channel->collTreeDn;
ncclPrimitives<UNROLL, 1, 1, T, 1, 1, 0, FUNC> prims(tid, nthreads, &tree->up, tree->down, NULL, stepSize, channel, comm, args->opCount);
ncclPrimitives<UNROLL, 1, 1, T, 1, 1, 0, FUNC> prims(tid, nthreads, &tree->up, tree->down, NULL, stepSize, channel, comm);
for (ssize_t gridOffset = 0; gridOffset < size; gridOffset += loopSize) {
// Down
ssize_t offset = gridOffset + bid*chunkSize;
@@ -238,7 +238,7 @@ __device__ void ncclAllReduceRingLLKernel(struct CollectiveArgs* args) {
const ssize_t loopSize = nChannels*nranks*chunkSize;
const ssize_t size = args->coll.count;
ncclLLPrimitives<T, FUNC, 1, 1> LLprims(tid, nthreads, &ring->prev, &ring->next, stepLines, channel, comm, args->opCount);
ncclLLPrimitives<T, FUNC, 1, 1> LLprims(tid, nthreads, &ring->prev, &ring->next, stepLines, channel, comm);
// Compute pointers
const T * __restrict__ thisInput = (const T*)args->sendbuff;
@@ -321,7 +321,7 @@ __device__ void ncclAllReduceTreeLLKernel(struct CollectiveArgs* args) {
do {
struct ncclTree* tree = &channel->treeUp;
// Reduce : max number of recv is 3, max number of send is 1 (binary tree + local)
ncclLLPrimitives<T, FUNC, NCCL_MAX_TREE_ARITY, 1> LLprims(tid, nthreads, tree->down, &tree->up, stepLines, channel, comm, args->opCount);
ncclLLPrimitives<T, FUNC, NCCL_MAX_TREE_ARITY, 1> LLprims(tid, nthreads, tree->down, &tree->up, stepLines, channel, comm);
for (ssize_t gridOffset = 0; gridOffset < size; gridOffset += loopSize) {
// Up
ssize_t offset = gridOffset + bid*chunkSize;
@@ -339,7 +339,7 @@ __device__ void ncclAllReduceTreeLLKernel(struct CollectiveArgs* args) {
do {
struct ncclTree* tree = &channel->treeDn;
// Broadcast : max number of recv is 1, max number of send is 3 (binary tree + local)
ncclLLPrimitives<T, FUNC, 1, NCCL_MAX_TREE_ARITY> LLprims(tid, nthreads, &tree->up, tree->down, stepLines, channel, comm, args->opCount);
ncclLLPrimitives<T, FUNC, 1, NCCL_MAX_TREE_ARITY> LLprims(tid, nthreads, &tree->up, tree->down, stepLines, channel, comm);
for (ssize_t gridOffset = 0; gridOffset < size; gridOffset += loopSize) {
// Down
ssize_t offset = gridOffset + bid*chunkSize;
@@ -380,7 +380,7 @@ __device__ void ncclAllReduceCollNetLLKernel(struct CollectiveArgs* args) {
if (blockIdx.x < nChannels) { // first half of the channels do reduce
struct ncclTree* tree = &channel->collTreeUp;
ncclLLPrimitives<T, FUNC, 1, 1> LLprims(tid, nthreads, tree->down, &tree->up, stepLines, channel, comm, args->opCount);
ncclLLPrimitives<T, FUNC, 1, 1> LLprims(tid, nthreads, tree->down, &tree->up, stepLines, channel, comm);
for (ssize_t gridOffset = 0; gridOffset < size; gridOffset += loopSize) {
// Up
ssize_t offset = gridOffset + bid*chunkSize;
@@ -397,7 +397,7 @@ __device__ void ncclAllReduceCollNetLLKernel(struct CollectiveArgs* args) {
if (blockIdx.x >= nChannels) { // second half of the channels do broadcast
struct ncclTree* tree = &channel->collTreeDn;
ncclLLPrimitives<T, FUNC, 1, 1> LLprims(tid, nthreads, &tree->up, tree->down, stepLines, channel, comm, args->opCount);
ncclLLPrimitives<T, FUNC, 1, 1> LLprims(tid, nthreads, &tree->up, tree->down, stepLines, channel, comm);
for (ssize_t gridOffset = 0; gridOffset < size; gridOffset += loopSize) {
// Down
ssize_t offset = gridOffset + bid*chunkSize;
@@ -432,7 +432,7 @@ __device__ void ncclAllReduceRingLL128Kernel(struct CollectiveArgs* args) {
const ssize_t loopSize = nChannels*nranks*chunkSize;
const ssize_t size = args->coll.count;
ncclLL128Primitives<T, FUNC, 1, 1> LLprims(tid, nthreads, &ring->prev, &ring->next, stepSize, channel, comm, args->opCount);
ncclLL128Primitives<T, FUNC, 1, 1> LLprims(tid, nthreads, &ring->prev, &ring->next, stepSize, channel, comm);
// Compute pointers
const T * __restrict__ thisInput = (const T*)args->sendbuff;
@@ -517,7 +517,7 @@ __device__ void ncclAllReduceTreeLL128Kernel(struct CollectiveArgs* args) {
if (treeUp->up == -1) {
// ReduceAndBroadcast : max number of recv is 3, max number of send is 3
ncclLL128Primitives<T, FUNC, NCCL_MAX_TREE_ARITY, NCCL_MAX_TREE_ARITY> LLprims(tid, nthreads, treeUp->down, treeDn->down, stepSize, channel, comm, args->opCount);
ncclLL128Primitives<T, FUNC, NCCL_MAX_TREE_ARITY, NCCL_MAX_TREE_ARITY> LLprims(tid, nthreads, treeUp->down, treeDn->down, stepSize, channel, comm);
for (ssize_t gridOffset = 0; gridOffset < size; gridOffset += loopSize) {
ssize_t offset = gridOffset + bid*chunkSize;
int nelem = min(chunkSize, size-offset);
@@ -526,7 +526,7 @@ __device__ void ncclAllReduceTreeLL128Kernel(struct CollectiveArgs* args) {
} else {
if (tid < nthreadsSplit) {
// Reduce : max number of recv is 3, max number of send is 1 (binary tree + local)
ncclLL128Primitives<T, FUNC, NCCL_MAX_TREE_ARITY, 1> LLprims(tid, nthreadsSplit, treeUp->down, &treeUp->up, stepSize, channel, comm, args->opCount);
ncclLL128Primitives<T, FUNC, NCCL_MAX_TREE_ARITY, 1> LLprims(tid, nthreadsSplit, treeUp->down, &treeUp->up, stepSize, channel, comm);
for (ssize_t gridOffset = 0; gridOffset < size; gridOffset += loopSize) {
// Up
ssize_t offset = gridOffset + bid*chunkSize;
@@ -539,7 +539,7 @@ __device__ void ncclAllReduceTreeLL128Kernel(struct CollectiveArgs* args) {
}
} else {
// Broadcast : max number of recv is 1, max number of send is 3 (binary tree + local)
ncclLL128Primitives<T, FUNC, 1, NCCL_MAX_TREE_ARITY> LLprims(tid-nthreadsSplit, nthreads-nthreadsSplit, &treeDn->up, treeDn->down, stepSize, channel, comm, args->opCount);
ncclLL128Primitives<T, FUNC, 1, NCCL_MAX_TREE_ARITY> LLprims(tid-nthreadsSplit, nthreads-nthreadsSplit, &treeDn->up, treeDn->down, stepSize, channel, comm);
for (ssize_t gridOffset = 0; gridOffset < size; gridOffset += loopSize) {
// Down
ssize_t offset = gridOffset + bid*chunkSize;
src/collectives/device/all_to_all.h
+2 -2
Vedi File
@@ -61,7 +61,7 @@ __device__ void ncclAllToAllKernel(struct CollectiveArgs* args) {
if (tid < nthreadsSplit ) {
int peerSend = (rank+(blockIdx.x*peersPerChan)+i)%nranks;
ncclPrimitives<UNROLL, ALLTOALL_CHUNKSTEPS/ALLTOALL_SLICESTEPS, ALLTOALL_SLICESTEPS, T, 2, 1, 0, FUNC>
prims(tid, nthreadsSplit, peerNone, &peerSend, NULL, stepSize, channel, comm, args->opCount);
prims(tid, nthreadsSplit, peerNone, &peerSend, NULL, stepSize, channel, comm);
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*sizeof(uint64_t)/sizeof(T));
@@ -73,7 +73,7 @@ __device__ void ncclAllToAllKernel(struct CollectiveArgs* args) {
} else {
int peerRecv = (2*nranks+rank-((blockIdx.x*peersPerChan)%nranks)-(i%nranks))%nranks;
ncclPrimitives<UNROLL, ALLTOALL_CHUNKSTEPS/ALLTOALL_SLICESTEPS, ALLTOALL_SLICESTEPS, T, 1, 2, 0, FUNC>
prims(tid-nthreadsSplit, nthreads-nthreadsSplit, &peerRecv, peerNone, NULL, stepSize, channel, comm, args->opCount);
prims(tid-nthreadsSplit, nthreads-nthreadsSplit, &peerRecv, peerNone, NULL, stepSize, channel, comm);
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*sizeof(uint64_t)/sizeof(T));
src/collectives/device/broadcast.h
+3 -3
Vedi File
@@ -37,7 +37,7 @@ __device__ void ncclBroadcastRingKernel(struct CollectiveArgs* args) {
T * __restrict__ thisOutput = (T*)args->recvbuff;
ncclPrimitives<UNROLL, BROADCAST_CHUNKSTEPS/BROADCAST_SLICESTEPS, BROADCAST_SLICESTEPS, T, 1, 1, 0, FUNC>
prims(tid, nthreads, &ring->prev, &ring->next, NULL, stepSize, channel, comm, args->opCount);
prims(tid, nthreads, &ring->prev, &ring->next, NULL, stepSize, channel, comm);
for (ssize_t gridOffset = 0; gridOffset < size; gridOffset += loopSize) {
int realChunkSize = min(chunkSize, DIVUP(size-gridOffset,nChannels));
@@ -96,7 +96,7 @@ __device__ void ncclBroadcastRingLLKernel(struct CollectiveArgs* args) {
const int nextRank = ring->devUserRanks[1];
const int root = args->coll.root;
ncclLLPrimitives<T, FUNC, 1, 1> LLprims(tid, nthreads, &ring->prev, &ring->next, stepLines, channel, comm, args->opCount);
ncclLLPrimitives<T, FUNC, 1, 1> LLprims(tid, nthreads, &ring->prev, &ring->next, stepLines, channel, comm);
// Compute pointers
const T * __restrict__ thisInput = (const T*)args->sendbuff;
@@ -151,7 +151,7 @@ __device__ void ncclBroadcastRingLL128Kernel(struct CollectiveArgs* args) {
const int nextRank = ring->devUserRanks[1];
const int root = args->coll.root;
ncclLL128Primitives<T, FUNC, 1, 1> LLprims(tid, nthreads, &ring->prev, &ring->next, stepSize, channel, comm, args->opCount);
ncclLL128Primitives<T, FUNC, 1, 1> LLprims(tid, nthreads, &ring->prev, &ring->next, stepSize, channel, comm);
// Compute pointers
const T * __restrict__ thisInput = (const T*)args->sendbuff;
src/collectives/device/gather.h
+2 -2
Vedi File
@@ -54,7 +54,7 @@ __device__ void ncclGatherKernel(struct CollectiveArgs* args) {
int peerRecv = (2*nranks+rank-((blockIdx.x*peersPerChan)%nranks)-(i%nranks))%nranks;
if (rank == root) {
ncclPrimitives<UNROLL, GATHER_CHUNKSTEPS/GATHER_SLICESTEPS, GATHER_SLICESTEPS, T, 1, 1, 0, FUNC>
prims(tid, nthreads, &peerRecv, &peerSend, NULL, stepSize, channel, comm, args->opCount);
prims(tid, nthreads, &peerRecv, &peerSend, NULL, stepSize, channel, comm);
ssize_t recv_offset = chunkOffset + peerRecv*size;
prims.recv(thisOutput+recv_offset, nelem);
@@ -62,7 +62,7 @@ __device__ void ncclGatherKernel(struct CollectiveArgs* args) {
else {
if (peerSend == root) {
ncclPrimitives<UNROLL, GATHER_CHUNKSTEPS/GATHER_SLICESTEPS, GATHER_SLICESTEPS, T, 1, 1, 0, FUNC>
prims(tid, nthreads, &peerRecv, &peerSend, NULL, stepSize, channel, comm, args->opCount);
prims(tid, nthreads, &peerRecv, &peerSend, NULL, stepSize, channel, comm);
ssize_t send_offset = chunkOffset;
prims.send(thisInput+send_offset, nelem);
src/collectives/device/primitives.h
+2 -22
Vedi File
@@ -111,18 +111,6 @@ class ncclPrimitives {
#endif
}
uint32_t mismatch = 0;
const uint64_t opCount;
inline __device__ void checkMismatch(struct ncclConnInfo* conn) {
if (mismatch) {
// In non-LL, we use _threadfence_system before incrementing opCount, yet we are still waiting for credits here, so there must be a size mismatch
STORE(comm->fatalDevError, ncclDevAssertedMismatch);
} else if (conn && LOAD(conn->opCountRem) > opCount+1) {
mismatch += 1;
}
}
uint32_t spins = 0;
uint32_t abort = 0;
@@ -130,7 +118,6 @@ class ncclPrimitives {
spins++;
if (abort == 0 && spins == SPINS_BEFORE_CHECK_ABORT) {
abort = LOAD(comm->abortFlag);
if (wid == i) checkMismatch(send ? sendConn : recvConn);
spins = 0;
}
return abort;
@@ -138,7 +125,6 @@ class ncclPrimitives {
inline __device__ void waitSend(int nbytes) {
spins = 0;
mismatch = 0;
if (sendConnHeadPtr) {
while (sendConnHeadCache + NCCL_STEPS < sendConnHead + SLICESTEPS) {
sendConnHeadCache = LOAD(sendConnHeadPtr);
@@ -153,7 +139,6 @@ class ncclPrimitives {
inline __device__ void waitRecv() {
spins = 0;
mismatch = 0;
if (recvConnTailPtr) {
#ifdef ENABLE_PROFILING
uint64_t t0 = __rtc64();
@@ -313,8 +298,6 @@ inline __device__ int directSendInc(int i, int directInc, int sliceInc) {
recvConnHeadPtr = LOAD(&recvConn->head);
// Return credits in case we rounded up.
STORE(recvConnHeadPtr, recvConnHead);
// Update opCount in case we skipped some operations
STORE(recvConn->opCountLoc, opCount);
}
}
@@ -340,7 +323,6 @@ inline __device__ int directSendInc(int i, int directInc, int sliceInc) {
sendConnHeadPtr = LOAD(&sendConn->head);
sendConnHeadCache = LOAD(sendConnHeadPtr);
sendConnFifoPtr = LOAD(&sendConn->fifo);
STORE(sendConn->opCountLoc, opCount);
}
if (tid >= nthreads-WARP_SIZE && wid < nsend) {
sendConnTailPtr = LOAD(&sendConn->tail);
@@ -350,7 +332,6 @@ inline __device__ int directSendInc(int i, int directInc, int sliceInc) {
__device__ __forceinline__ void saveRecvSync() {
if (tid >= nthreads-WARP_SIZE && wid < nrecv) {
STORE(&recvConn->step, recvConnHead);
STORE(recvConn->opCountLoc, opCount+1);
__threadfence_system();
}
}
@@ -358,15 +339,14 @@ inline __device__ int directSendInc(int i, int directInc, int sliceInc) {
__device__ __forceinline__ void saveSendSync() {
if (tid < nsend) {
STORE(&sendConn->step, sendConnHead);
STORE(sendConn->opCountLoc, opCount+1);
__threadfence_system();
}
}
public:
__device__ __forceinline__
ncclPrimitives(const int tid, const int nthreads, int* recvPeers, int* sendPeers, T* directBuff, int stepSize, struct ncclChannel* channel, struct ncclDevComm* comm, const uint64_t opCount)
: comm(comm), tid(tid), nthreads(nthreads), wid(tid%WARP_SIZE), stepSize(stepSize), opCount(opCount) {
ncclPrimitives(const int tid, const int nthreads, int* recvPeers, int* sendPeers, T* directBuff, int stepSize, struct ncclChannel* channel, struct ncclDevComm* comm)
: comm(comm), tid(tid), nthreads(nthreads), wid(tid%WARP_SIZE), stepSize(stepSize) {
barriers = channel->barrier;
barrier_next = channel->barrier_next;
// Make sure step is updated before we read it.
src/collectives/device/prims_ll.h
+2 -23
Vedi File
@@ -45,19 +45,6 @@ class ncclLLPrimitives {
#endif
}
uint32_t mismatch = 0;
const uint64_t opCount;
inline __device__ void checkMismatch(struct ncclConnInfo* conn) {
if (mismatch > 20) {
// We have seen that the peer advanced opcount so many times yet we are still waiting for credit of current op, so it is _most likely_ a mismatch
// Note that we are not using _threadfence_system in LL so the error cannot be asserted
STORE(comm->fatalDevError, ncclDevSuspectedMismatch);
} else if (conn && LOAD(conn->opCountRem) > opCount) {
mismatch += 1;
}
}
uint32_t spins = 0;
uint32_t abort = 0;
@@ -65,7 +52,6 @@ class ncclLLPrimitives {
spins++;
if (abort == 0 && spins == SPINS_BEFORE_CHECK_ABORT) {
abort = LOAD(comm->abortFlag);
if (wid == i) checkMismatch(send ? sendConn : recvConn);
spins = 0;
}
return abort;
@@ -73,7 +59,6 @@ class ncclLLPrimitives {
inline __device__ void waitSend(int nbytes) {
spins = 0;
mismatch = 0;
if (sendConnHeadPtr) {
while (sendConnHeadCache + NCCL_STEPS < sendConnHead + 1) {
sendConnHeadCache = LOAD(sendConnHeadPtr);
@@ -110,7 +95,6 @@ class ncclLLPrimitives {
uint32_t flag = recvFlag(i);
uint32_t data1, flag1, data2, flag2;
spins = 0;
mismatch = 0;
#if defined(__HIP_PLATFORM_HCC__) || defined(__HCC__) || defined(__HIPCC__)
using Vec = uint32_t __attribute__((ext_vector_type(4)));
Vec i4;
@@ -209,8 +193,6 @@ class ncclLLPrimitives {
if (tid >= nthreads-WARP_SIZE && wid < nrecv) {
recvConnHeadPtr = LOAD(&recvConn->head);
recvConnHead = LOAD(&recvConn->step);
// Update opCount in case we skipped some operations
STORE(recvConn->opCountLoc, opCount);
}
}
@@ -226,14 +208,12 @@ class ncclLLPrimitives {
sendConnHeadCache = LOAD(sendConnHeadPtr);
sendConnHead = LOAD(&sendConn->step);
sendConnFifoPtr = LOAD(&sendConn->fifo);
STORE(sendConn->opCountLoc, opCount);
}
}
__device__ __forceinline__ void saveRecvSync() {
if (tid >= nthreads-WARP_SIZE && wid < nrecv) {
STORE(&recvConn->step, recvConnHead);
STORE(recvConn->opCountLoc, opCount+1);
__threadfence_block();
}
}
@@ -241,15 +221,14 @@ class ncclLLPrimitives {
__device__ __forceinline__ void saveSendSync() {
if (tid < nsend) {
STORE(&sendConn->step, sendConnHead);
STORE(sendConn->opCountLoc, opCount+1);
__threadfence_block();
}
}
public:
__device__ __forceinline__
ncclLLPrimitives(const int tid, const int nthreads, int* recvPeers, int* sendPeers, int stepLines, struct ncclChannel* channel, struct ncclDevComm* comm, const uint64_t opCount)
: comm(comm), tid(tid), nthreads(nthreads), wid(tid%WARP_SIZE), stepLines(stepLines), opCount(opCount) {
ncclLLPrimitives(const int tid, const int nthreads, int* recvPeers, int* sendPeers, int stepLines, struct ncclChannel* channel, struct ncclDevComm* comm)
: comm(comm), tid(tid), nthreads(nthreads), wid(tid%WARP_SIZE), stepLines(stepLines) {
// Make sure step is updated before we read it.
barrier();
src/collectives/device/prims_ll128.h
+2 -22
Vedi File
@@ -60,19 +60,6 @@ class ncclLL128Primitives {
#endif
}
uint32_t mismatch = 0;
const uint64_t opCount;
inline __device__ void checkMismatch(struct ncclConnInfo* conn) {
if (mismatch > 20) {
// We have seen that the peer advanced opcount so many times yet we are still waiting for credit of current op, so it is _most likely_ a mismatch
// Note that we are not using _threadfence_system in LL so the error cannot be asserted
STORE(comm->fatalDevError, ncclDevSuspectedMismatch);
} else if (conn && LOAD(conn->opCountRem) > opCount) {
mismatch += 1;
}
}
uint32_t spins = 0;
uint32_t abort = 0;
@@ -80,7 +67,6 @@ class ncclLL128Primitives {
spins++;
if (abort == 0 && spins == SPINS_BEFORE_CHECK_ABORT) {
abort = LOAD(comm->abortFlag);
if (wid == i) checkMismatch(send ? sendConn : recvConn);
spins = 0;
}
return abort;
@@ -88,7 +74,6 @@ class ncclLL128Primitives {
inline __device__ void waitSend(int nbytes) {
spins = 0;
mismatch = 0;
if (sendConnHeadPtr) {
while (sendConnHeadCache + NCCL_STEPS < sendConnHead + 1) {
sendConnHeadCache = LOAD(sendConnHeadPtr);
@@ -331,8 +316,6 @@ class ncclLL128Primitives {
if (tid >= nthreads-WARP_SIZE && wid < nrecv) {
recvConnHeadPtr = LOAD(&recvConn->head);
recvConnHead = LOAD(&recvConn->step);
// Update opCount in case we skipped some operations
STORE(recvConn->opCountLoc, opCount);
}
}
@@ -348,7 +331,6 @@ class ncclLL128Primitives {
sendConnHeadCache = LOAD(sendConnHeadPtr);
sendConnHead = LOAD(&sendConn->step);
sendConnFifoPtr = LOAD(&sendConn->fifo);
STORE(sendConn->opCountLoc, opCount);
}
if (tid >= nthreads-WARP_SIZE && wid<nsend) {
if (sendConn->fifo) {
@@ -361,7 +343,6 @@ class ncclLL128Primitives {
__device__ __forceinline__ void saveRecvSync() {
if (tid >= nthreads-WARP_SIZE && wid < nrecv) {
STORE(&recvConn->step, recvConnHead);
STORE(recvConn->opCountLoc, opCount+1);
__threadfence_block();
}
}
@@ -369,15 +350,14 @@ class ncclLL128Primitives {
__device__ __forceinline__ void saveSendSync() {
if (tid < nsend) {
STORE(&sendConn->step, sendConnHead);
STORE(sendConn->opCountLoc, opCount+1);
__threadfence_block();
}
}
public:
__device__ __forceinline__
ncclLL128Primitives(const int tid, const int nthreads, int* recvPeers, int* sendPeers, int stepSize, struct ncclChannel* channel, struct ncclDevComm* comm, const uint64_t opCount)
: comm(comm), tid(tid), nthreads(nthreads), wid(tid%WARP_SIZE), warp(tid/WARP_SIZE), flagThread((tid%8)==7), stepSize(stepSize), opCount(opCount), shmem(ncclShmem+(threadIdx.x/WARP_SIZE)*NCCL_LL128_SHMEM_ELEMS_PER_THREAD*WARP_SIZE+2*wid) {
ncclLL128Primitives(const int tid, const int nthreads, int* recvPeers, int* sendPeers, int stepSize, struct ncclChannel* channel, struct ncclDevComm* comm)
: comm(comm), tid(tid), nthreads(nthreads), wid(tid%WARP_SIZE), warp(tid/WARP_SIZE), flagThread((tid%8)==7), stepSize(stepSize), shmem(ncclShmem+(threadIdx.x/WARP_SIZE)*NCCL_LL128_SHMEM_ELEMS_PER_THREAD*WARP_SIZE+2*wid) {
// for __any_sync
if (NSEND > NRECV)
sync = channel->sync + 2 + tid/WARP_SIZE;
src/collectives/device/reduce.h
+3 -3
Vedi File
@@ -33,7 +33,7 @@ __device__ void ncclReduceRingKernel(struct CollectiveArgs* args) {
T * __restrict__ thisOutput = (T*)args->recvbuff;
ncclPrimitives<UNROLL, REDUCE_CHUNKSTEPS/REDUCE_SLICESTEPS, REDUCE_SLICESTEPS, T, 1, 1, 0, FUNC>
prims(tid, nthreads, &ring->prev, &ring->next, NULL, stepSize, channel, comm, args->opCount);
prims(tid, nthreads, &ring->prev, &ring->next, NULL, stepSize, channel, comm);
for (ssize_t gridOffset = 0; gridOffset < size; gridOffset += loopSize) {
int realChunkSize = min(chunkSize, DIVUP(size-gridOffset,nChannels));
@@ -77,7 +77,7 @@ __device__ void ncclReduceRingLLKernel(struct CollectiveArgs* args) {
const int prevRank = ring->devUserRanks[nranks-1];
const int root = args->coll.root;
ncclLLPrimitives<T, FUNC, 1, 1> LLprims(tid, nthreads, &ring->prev, &ring->next, stepLines, channel, comm, args->opCount);
ncclLLPrimitives<T, FUNC, 1, 1> LLprims(tid, nthreads, &ring->prev, &ring->next, stepLines, channel, comm);
// Compute pointers
const T * __restrict__ thisInput = (const T*)args->sendbuff;
@@ -129,7 +129,7 @@ __device__ void ncclReduceRingLL128Kernel(struct CollectiveArgs* args) {
const int prevRank = ring->devUserRanks[nranks-1];
const int root = args->coll.root;
ncclLL128Primitives<T, FUNC, 1, 1> LLprims(tid, nthreads, &ring->prev, &ring->next, stepSize, channel, comm, args->opCount);
ncclLL128Primitives<T, FUNC, 1, 1> LLprims(tid, nthreads, &ring->prev, &ring->next, stepSize, channel, comm);
// Compute pointers
const T * __restrict__ thisInput = (const T*)args->sendbuff;
src/collectives/device/reduce_scatter.h
+3 -3
Vedi File
@@ -30,7 +30,7 @@ __device__ void ncclReduceScatterRingKernel(struct CollectiveArgs* args) {
T * __restrict__ thisOutput = (T*)args->recvbuff;
ncclPrimitives<UNROLL, REDUCESCATTER_CHUNKSTEPS/REDUCESCATTER_SLICESTEPS, REDUCESCATTER_SLICESTEPS, T, 1, 1, 0, FUNC>
prims(tid, nthreads, &ring->prev, &ring->next, NULL, stepSize, channel, comm, args->opCount);
prims(tid, nthreads, &ring->prev, &ring->next, NULL, stepSize, channel, comm);
for (ssize_t gridOffset = 0; gridOffset < size; gridOffset += loopSize) {
int realChunkSize = min(chunkSize, DIVUP(size-gridOffset,nChannels));
@@ -88,7 +88,7 @@ __device__ void ncclReduceScatterRingLLKernel(struct CollectiveArgs* args) {
const ssize_t loopSize = nChannels*chunkSize;
const ssize_t size = args->coll.count;
ncclLLPrimitives<T, FUNC, 1, 1> LLprims(tid, nthreads, &ring->prev, &ring->next, stepLines, channel, comm, args->opCount);
ncclLLPrimitives<T, FUNC, 1, 1> LLprims(tid, nthreads, &ring->prev, &ring->next, stepLines, channel, comm);
// Compute pointers
const T * __restrict__ thisInput = (const T*)args->sendbuff;
@@ -155,7 +155,7 @@ __device__ void ncclReduceScatterRingLL128Kernel(struct CollectiveArgs* args) {
const ssize_t loopSize = nChannels*chunkSize;
const ssize_t size = args->coll.count;
ncclLL128Primitives<T, FUNC, 1, 1> LLprims(tid, nthreads, &ring->prev, &ring->next, stepSize, channel, comm, args->opCount);
ncclLL128Primitives<T, FUNC, 1, 1> LLprims(tid, nthreads, &ring->prev, &ring->next, stepSize, channel, comm);
// Compute pointers
const T * __restrict__ thisInput = (const T*)args->sendbuff;
src/collectives/device/scatter.h
+2 -2
Vedi File
@@ -54,7 +54,7 @@ __device__ void ncclScatterKernel(struct CollectiveArgs* args) {
int peerRecv = (2*nranks+rank-((blockIdx.x*peersPerChan)%nranks)-(i%nranks))%nranks;
if (rank == root) {
ncclPrimitives<UNROLL, SCATTER_CHUNKSTEPS/SCATTER_SLICESTEPS, SCATTER_SLICESTEPS, T, 1, 1, 0, FUNC>
prims(tid, nthreads, &peerRecv, &peerSend, NULL, stepSize, channel, comm, args->opCount);
prims(tid, nthreads, &peerRecv, &peerSend, NULL, stepSize, channel, comm);
ssize_t send_offset = chunkOffset + peerSend*size;
prims.send(thisInput+send_offset, nelem);
@@ -62,7 +62,7 @@ __device__ void ncclScatterKernel(struct CollectiveArgs* args) {
else {
if (peerRecv == root) {
ncclPrimitives<UNROLL, SCATTER_CHUNKSTEPS/SCATTER_SLICESTEPS, SCATTER_SLICESTEPS, T, 1, 1, 0, FUNC>
prims(tid, nthreads, &peerRecv, &peerSend, NULL, stepSize, channel, comm, args->opCount);
prims(tid, nthreads, &peerRecv, &peerSend, NULL, stepSize, channel, comm);
ssize_t recv_offset = chunkOffset;
prims.recv(thisOutput+recv_offset, nelem);
src/collectives/device/sendrecv.h
+2 -2
Vedi File
@@ -60,7 +60,7 @@ __device__ void ncclSendRecvKernel(struct CollectiveArgs* args) {
int peer = (comm->rank+(int)args->p2p.delta)%comm->nRanks;
ncclPrimitives<UNROLL, 1, 1, T, 2, 1, 1, FUNC>
prims(tid, nthreadsSplit, peerNone, &peer, recvbuff, stepSize*SENDRECV_SLICEFACTOR, channel, comm, args->opCount);
prims(tid, nthreadsSplit, peerNone, &peer, recvbuff, stepSize*SENDRECV_SLICEFACTOR, channel, comm);
if (sendSize == 0) {
prims.send(sendbuff, 0);
@@ -76,7 +76,7 @@ __device__ void ncclSendRecvKernel(struct CollectiveArgs* args) {
int peer = (comm->rank-(int)args->p2p.delta+comm->nRanks)%comm->nRanks;
ncclPrimitives<UNROLL, 1, 1, T, 1, 2, 1, FUNC>
prims(tid-nthreadsSplit, nthreads-nthreadsSplit, &peer, peerNone, recvbuff, stepSize*SENDRECV_SLICEFACTOR, channel, comm, args->opCount);
prims(tid-nthreadsSplit, nthreads-nthreadsSplit, &peer, peerNone, recvbuff, stepSize*SENDRECV_SLICEFACTOR, channel, comm);
if (recvSize == 0) {
prims.recv(recvbuff, 0);
src/graph/topo.cc
+8 -9
Vedi File
@@ -328,21 +328,20 @@ ncclResult_t ncclTopoAddPci(struct ncclXmlNode* xmlPci, struct ncclTopoSystem* s
NCCLCHECK(busIdToInt64(str, &busId));
struct ncclTopoNode* node = NULL;
if (type == GPU) {
struct ncclXmlNode* xmlGpu;
NCCLCHECK(xmlGetSub(xmlPci, "gpu", &xmlGpu));
if (xmlGpu == NULL) return ncclSuccess;
struct ncclXmlNode* xmlGpu = NULL;
NCCLCHECK(xmlGetSub(xmlPci, "gpu", &xmlGpu));
if (xmlGpu != NULL) {
type = GPU;
int index;
NCCLCHECK(xmlGetAttrIndex(xmlGpu, "rank", &index));
if (index == -1) return ncclSuccess;
NCCLCHECK(ncclTopoCreateNode(system, &node, type, busId));
NCCLCHECK(ncclTopoAddGpu(xmlGpu, system, node));
}
if (type == NIC) {
struct ncclXmlNode* xmlNic;
NCCLCHECK(xmlGetSub(xmlPci, "nic", &xmlNic));
if (xmlNic == NULL) return ncclSuccess;
struct ncclXmlNode* xmlNic = NULL;
NCCLCHECK(xmlGetSub(xmlPci, "nic", &xmlNic));
if (xmlNic != NULL) {
type = NIC;
// Ignore sub device ID and merge multi-port NICs into one PCI device.
busId &= 0xfffffffffffffff0;
struct ncclTopoNode* nicNode = NULL;
src/include/comm.h
-5
Vedi File
@@ -41,7 +41,6 @@ struct ncclSendMem {
char pad1[CACHE_LINE_SIZE-sizeof(uint64_t)];
void* ptrExchange;
char pad2[CACHE_LINE_SIZE-sizeof(void*)];
uint64_t opCount;
};
char pad3[MEM_ALIGN];
};
@@ -53,7 +52,6 @@ struct ncclRecvMem {
struct {
uint64_t tail;
char pad1[CACHE_LINE_SIZE-sizeof(uint64_t)];
uint64_t opCount;
char pad2[CACHE_LINE_SIZE-sizeof(uint64_t)];
int sizesFifo[NCCL_STEPS];
};
@@ -113,9 +111,6 @@ struct ncclComm {
// Whether there has been a fatal error in this communicator.
ncclResult_t fatalError;
// Error reported by GPU
volatile ncclDevError_t* fatalDevError;
// Flag to ask NCCL kernels to abort
volatile uint32_t *abortFlag;
src/include/devcomm.h
-9
Vedi File
@@ -97,8 +97,6 @@ struct ncclConnInfo {
char *buffs[NCCL_NUM_PROTOCOLS]; // Local for recv, remote for send
uint64_t *tail; // Local for recv, remote for send
uint64_t *head; // Local for send, remote for recv
uint64_t *opCountLoc; // opCount of local rank
uint64_t *opCountRem; // opCount of remote rank
int direct; // Direct communication
void **ptrExchange; // Pointer exchange for direct communication
@@ -301,12 +299,6 @@ static_assert(sizeof(struct ncclCollTrace) == 8*sizeof(int), "ncclCollTrace must
#define COLLTRACE_NUM_ITEMS 1024
#endif
typedef enum {
ncclDevSuccess,
ncclDevAssertedMismatch,
ncclDevSuspectedMismatch
} ncclDevError_t;
struct ncclDevComm {
int rank;
int nRanks;
@@ -314,7 +306,6 @@ struct ncclDevComm {
// Flag to ask NCCL kernels to abort
volatile uint32_t *abortFlag;
volatile ncclDevError_t *fatalDevError;
// Channels, device side
struct ncclChannel* channels;
src/init.cc
-30
Vedi File
@@ -291,7 +291,6 @@ static ncclResult_t commFree(ncclComm_t comm) {
free(comm->intraCC);
}
CUDACHECK(hipHostFree((void *)comm->abortFlag));
CUDACHECK(hipHostFree((void *)comm->fatalDevError));
// Poison comm to try and catch a double free
commPoison(comm);
@@ -334,10 +333,6 @@ static ncclResult_t commAlloc(ncclComm_t* comret, int ndev, int rank) {
#endif
comm->fatalError = ncclSuccess;
NCCLCHECK(ncclCudaHostCalloc((ncclDevError_t**)&comm->fatalDevError, 1));
comm->hostDevComm.fatalDevError = comm->fatalDevError;
STORE(comm->fatalDevError, ncclDevSuccess);
NCCLCHECK(ncclCudaHostCalloc((uint32_t**)&comm->abortFlag, 1));
comm->hostDevComm.abortFlag = comm->abortFlag;
STORE(comm->abortFlag, 0);
@@ -1149,31 +1144,6 @@ NCCL_API(ncclResult_t, ncclCommGetAsyncError, ncclComm_t comm, ncclResult_t *asy
ncclResult_t ncclCommGetAsyncError(ncclComm_t comm, ncclResult_t *asyncError) {
NCCLCHECK(PtrCheck(comm, "ncclGetAsyncError", "comm"));
NCCLCHECK(PtrCheck(asyncError, "ncclGetAsyncError", "asyncError"));
// Check device reported error
static ncclDevError_t printedDevErr = ncclDevSuccess;
switch(LOAD(comm->fatalDevError)) {
case ncclDevSuccess :
break;
case ncclDevAssertedMismatch :
if (printedDevErr != ncclDevAssertedMismatch) {
WARN("Mismatched collective detected, please check your collective calls at and around rank %d. You can use NCCL_DEBUG=INFO and NCCL_DEBUG_SUBSYS=COLL to see the collective logs", comm->rank);
printedDevErr = ncclDevAssertedMismatch;
}
if (comm->fatalError == ncclSuccess) {
comm->fatalError = ncclInvalidUsage;
}
break;
case ncclDevSuspectedMismatch :
if (printedDevErr != ncclDevSuspectedMismatch) {
WARN("Your program may be hanging, this may be caused by a collective mismatch around rank %d. Please check your collective calls at and around this rank. You can use NCCL_DEBUG=INFO and NCCL_DEBUG_SUBSYS=COLL to see the collective logs", comm->rank);
printedDevErr = ncclDevSuspectedMismatch;
}
break;
default:
WARN("Unknown device error %d", *comm->fatalDevError);
return ncclInternalError;
}
*asyncError = comm->fatalError;
return ncclSuccess;
}
src/transport/coll_net.cc
-10
Vedi File
@@ -140,10 +140,8 @@ ncclResult_t collNetSendConnect(struct ncclConnect* connectInfos, int nranks, in
// Head/Tail/Opcount/Fifos are always on host
send->conn.tail = &resources->devHostRecvMem->tail;
send->conn.opCountRem = &resources->devHostRecvMem->opCount;
send->conn.fifo = resources->devHostRecvMem->sizesFifo;
send->conn.head = &resources->devHostSendMem->head;
send->conn.opCountLoc = &resources->devHostSendMem->opCount;
for (int i=0; i<NCCL_STEPS; i++) send->conn.fifo[i] = -1;
// Get info from recv side
@@ -179,9 +177,7 @@ ncclResult_t collNetRecvConnect(struct ncclConnect* connectInfos, int nranks, in
// Head/Tail/Opcount are always on host
recv->conn.tail = &resources->devHostRecvMem->tail;
recv->conn.opCountLoc = &resources->devHostRecvMem->opCount;
recv->conn.head = &resources->devHostSendMem->head;
recv->conn.opCountRem = &resources->devHostSendMem->opCount;
// Connect to coll comm
collNetHandle_t** handlePtrs = NULL;
@@ -259,9 +255,6 @@ ncclResult_t collNetSendProxy(struct ncclProxyArgs* args) {
}
struct collNetSendResources* resources = (struct collNetSendResources*) (args->connector->transportResources);
if (args->state == ncclProxyOpReady) {
// Update opCount
STORE(&resources->hostRecvMem->opCount, args->opCount);
// Round to next multiple of sliceSteps
resources->step = ROUNDUP(resources->step, args->chunkSteps);
args->head = resources->step;
@@ -366,9 +359,6 @@ ncclResult_t collNetRecvProxy(struct ncclProxyArgs* args) {
}
struct collNetRecvResources* resources = (struct collNetRecvResources*) (args->connector->transportResources);
if (args->state == ncclProxyOpReady) {
// Update opCount
STORE(&resources->hostSendMem->opCount, args->opCount);
// Round to next multiple of sliceSteps
resources->step = ROUNDUP(resources->step, args->chunkSteps);
args->head = resources->step;
src/transport/net.cc
-10
Vedi File
@@ -70,10 +70,8 @@ ncclResult_t netSendSetup(struct ncclTopoSystem* topo, struct ncclTopoGraph* gra
send->conn.direct |= resources->useGdr ? NCCL_DIRECT_NIC : 0;
send->conn.tail = &resources->recvMem->tail;
send->conn.opCountRem = &resources->recvMem->opCount;
send->conn.fifo = resources->recvMem->sizesFifo;
send->conn.head = &resources->sendMem->head;
send->conn.opCountLoc = &resources->sendMem->opCount;
for (int i=0; i<NCCL_STEPS; i++) send->conn.fifo[i] = -1;
int protoLoc[NCCL_NUM_PROTOCOLS];
@@ -126,9 +124,7 @@ ncclResult_t netRecvSetup(struct ncclTopoSystem* topo, struct ncclTopoGraph* gra
recv->conn.direct |= resources->useGdr ? NCCL_DIRECT_NIC : 0;
recv->conn.tail = &resources->recvMem->tail;
recv->conn.opCountLoc = &resources->recvMem->opCount;
recv->conn.head = &resources->sendMem->head;
recv->conn.opCountRem = &resources->sendMem->opCount;
int protoLoc[NCCL_NUM_PROTOCOLS];
for (int p=0; p<NCCL_NUM_PROTOCOLS; p++) {
@@ -238,9 +234,6 @@ ncclResult_t netRecvFree(void* transportResources) {
ncclResult_t netSendProxy(struct ncclProxyArgs* args) {
struct netSendResources* resources = (struct netSendResources*) (args->connector->transportResources);
if (args->state == ncclProxyOpReady) {
// Update opCount
STORE(&resources->recvMem->opCount, args->opCount);
// Round to next multiple of sliceSteps
resources->step = ROUNDUP(resources->step, args->chunkSteps);
args->head = resources->step;
@@ -371,9 +364,6 @@ ncclResult_t netSendProxy(struct ncclProxyArgs* args) {
ncclResult_t netRecvProxy(struct ncclProxyArgs* args) {
struct netRecvResources* resources = (struct netRecvResources*) (args->connector->transportResources);
if (args->state == ncclProxyOpReady) {
// Update opCount
STORE(&resources->sendMem->opCount, args->opCount);
// Round to next multiple of sliceSteps
resources->step = ROUNDUP(resources->step, args->chunkSteps);
args->head = resources->step;
src/transport/p2p.cc
-4
Vedi File
@@ -316,10 +316,8 @@ static ncclResult_t p2pSendConnect(struct ncclConnect* connectInfo, int nranks,
}
}
send->conn.tail = &remDevMem->tail;
send->conn.opCountRem = resources->devRemOpCount;
send->conn.head = &resources->devMem->head;
send->conn.ptrExchange = &resources->devMem->ptrExchange;
send->conn.opCountLoc = resources->devOpCount;
send->conn.next_hdp_reg = resources->next_hdp_reg;
return ncclSuccess;
}
@@ -363,9 +361,7 @@ ncclResult_t p2pRecvConnect(struct ncclConnect* connectInfo, int nranks, int ran
}
}
recv->conn.tail = &resources->devMem->tail;
recv->conn.opCountLoc = resources->devOpCount;
recv->conn.head = &remDevMem->head;
recv->conn.opCountRem = resources->devRemOpCount;
return ncclSuccess;
}
src/transport/shm.cc
-4
Vedi File
@@ -126,10 +126,8 @@ ncclResult_t shmSendConnect(struct ncclConnect* connectInfo, int nranks, int ran
offset += send->comm->buffSizes[p];
}
send->conn.tail = &resources->devRemHostMem->tail;
send->conn.opCountRem = &resources->devRemHostMem->opCount;
send->conn.head = &resources->devHostMem->head;
send->conn.opCountLoc = &resources->devHostMem->opCount;
return ncclSuccess;
}
@@ -145,7 +143,6 @@ ncclResult_t shmRecvConnect(struct ncclConnect* connectInfo, int nranks, int ran
NCCLCHECK(shmOpen(shmName, resources->remShmSize, (void**)&resources->remHostMem, (void**)&resources->devRemHostMem, 0));
NCCLCHECK(shmUnlink(shmName));
recv->conn.head = &resources->devRemHostMem->head;
recv->conn.opCountRem = &resources->devRemHostMem->opCount;
int offset = 0;
for (int p=0; p<NCCL_NUM_PROTOCOLS; p++) {
@@ -153,7 +150,6 @@ ncclResult_t shmRecvConnect(struct ncclConnect* connectInfo, int nranks, int ran
offset += recv->comm->buffSizes[p];
}
recv->conn.tail = &resources->devHostMem->tail;
recv->conn.opCountLoc = &resources->devHostMem->opCount;
return ncclSuccess;
}
test/test_AllReduceAbort.cpp
+2 -9
Vedi File
@@ -40,23 +40,16 @@ namespace CorrectnessTests
hipStream_t stream;
HIPCHECK(hipStreamCreateWithFlags(&stream, hipStreamNonBlocking));
struct ncclChannel* channel = comm->channels;
uint64_t **p_dev_opCount = (uint64_t **)((uint8_t*)(channel->devPeers + channel->ring.next) + offsetof(struct ncclPeer, send.conn.opCountRem));
uint64_t **p_dev_head = (uint64_t **)((uint8_t*)(channel->devPeers + channel->ring.next) + offsetof(struct ncclPeer, send.conn.head));
uint64_t *real_opCount, *fake_opCount, *fake_o;
uint64_t *real_head, *fake_head, *fake_h;
// get original opCount and head
HIPCHECK(hipMemcpy(&real_opCount, p_dev_opCount, sizeof(uint64_t*), hipMemcpyDefault));
// get original head
HIPCHECK(hipMemcpy(&real_head, p_dev_head, sizeof(uint64_t*), hipMemcpyDefault));
// allocate and install fakes
HIPCHECK(hipHostMalloc(&fake_opCount, sizeof(uint64_t*), hipHostMallocMapped));
HIPCHECK(hipMemcpy(p_dev_opCount, &fake_opCount, sizeof(uint64_t*), hipMemcpyDefault));
*fake_opCount = FAKE_OP_COUNT;
HIPCHECK(hipHostMalloc(&fake_head, sizeof(uint64_t*), hipHostMallocMapped));
HIPCHECK(hipMemcpy(p_dev_head, &fake_head, sizeof(uint64_t*), hipMemcpyDefault));
*fake_head = 0;
// read back fakes to confirm
HIPCHECK(hipMemcpy(&fake_o, p_dev_opCount, sizeof(uint64_t*), hipMemcpyDefault));
HIPCHECK(hipMemcpy(&fake_h, p_dev_head, sizeof(uint64_t*), hipMemcpyDefault));
//std::cerr << "[ ] replaced gpu " << gpu << " real_opCount = " << real_opCount << " to fake_opCount = " << fake_o << std::endl;
//std::cerr << "[ ] replaced gpu " << gpu << " real_head = " << real_head << " to fake_head = " << fake_h << std::endl;
@@ -121,7 +114,7 @@ namespace CorrectnessTests
if (idle) pthread_yield();
}
HIPCHECK(hipHostFree(fake_opCount));
HIPCHECK(hipHostFree(fake_head));
HIPCHECK(hipStreamDestroy(stream));
dataset.Release();
}
test/test_BroadcastAbort.cpp
+2 -10
Vedi File
@@ -41,25 +41,17 @@ namespace CorrectnessTests
hipStream_t stream;
HIPCHECK(hipStreamCreateWithFlags(&stream, hipStreamNonBlocking));
struct ncclChannel* channel = comm->channels;
uint64_t **p_dev_opCount = (uint64_t **)((uint8_t*)(channel->devPeers + channel->ring.next) + offsetof(struct ncclPeer, send.conn.opCountRem));
uint64_t **p_dev_head = (uint64_t **)((uint8_t*)(channel->devPeers + channel->ring.next) + offsetof(struct ncclPeer, send.conn.head));
uint64_t *real_opCount, *fake_opCount, *fake_o;
uint64_t *real_head, *fake_head, *fake_h;
// get original opCount and head
HIPCHECK(hipMemcpy(&real_opCount, p_dev_opCount, sizeof(uint64_t*), hipMemcpyDefault));
// get original head
HIPCHECK(hipMemcpy(&real_head, p_dev_head, sizeof(uint64_t*), hipMemcpyDefault));
// allocate and install fakes
HIPCHECK(hipHostMalloc(&fake_opCount, sizeof(uint64_t*), hipHostMallocMapped));
HIPCHECK(hipMemcpy(p_dev_opCount, &fake_opCount, sizeof(uint64_t*), hipMemcpyDefault));
*fake_opCount = FAKE_OP_COUNT;
HIPCHECK(hipHostMalloc(&fake_head, sizeof(uint64_t*), hipHostMallocMapped));
HIPCHECK(hipMemcpy(p_dev_head, &fake_head, sizeof(uint64_t*), hipMemcpyDefault));
*fake_head = 0;
// read back fakes to confirm
HIPCHECK(hipMemcpy(&fake_o, p_dev_opCount, sizeof(uint64_t*), hipMemcpyDefault));
HIPCHECK(hipMemcpy(&fake_h, p_dev_head, sizeof(uint64_t*), hipMemcpyDefault));
//std::cerr << "[ ] replaced gpu " << gpu << " real_opCount = " << real_opCount << " to fake_opCount = " << fake_o << std::endl;
//std::cerr << "[ ] replaced gpu " << gpu << " real_head = " << real_head << " to fake_head = " << fake_h << std::endl;
// Perform a number of iterations and introduce abort
@@ -124,7 +116,7 @@ namespace CorrectnessTests
if (idle) pthread_yield();
}
HIPCHECK(hipHostFree(fake_opCount));
HIPCHECK(hipHostFree(fake_head));
HIPCHECK(hipStreamDestroy(stream));
dataset.Release();
}