From 78918c57445d5a06f087ba26627fb3412566ee58 Mon Sep 17 00:00:00 2001 From: Thomas Huber Date: Fri, 7 Aug 2026 19:22:20 +0000 Subject: [PATCH 01/12] feat(rccl): Enable hierarchical CE scale-out collectives Route multi-LSA AllGather and AllToAll through the hierarchical RMA plus CE path, and scope synchronization to the local LSA. Correct proxy window handles, offsets, and scratch cleanup so data-bearing RMA operations complete safely. AICOMRCCL-630 Co-authored-by: Cursor --- projects/rccl/src/ce_coll.cc | 92 +++++++++++++++-------- projects/rccl/src/dev_runtime.cc | 14 +++- projects/rccl/src/enqueue.cc | 24 +----- projects/rccl/src/include/dev_runtime.h | 3 + projects/rccl/src/rma/rma_proxy_launch.cc | 4 +- 5 files changed, 79 insertions(+), 58 deletions(-) diff --git a/projects/rccl/src/ce_coll.cc b/projects/rccl/src/ce_coll.cc index 3ae9f8762d7a..9b4df2fe2e7f 100644 --- a/projects/rccl/src/ce_coll.cc +++ b/projects/rccl/src/ce_coll.cc @@ -83,7 +83,8 @@ ncclResult_t ncclCeInit(struct ncclComm* comm) { #endif uint8_t* ceDevBase = nullptr; - size_t ceDevBaseSize = alignUp(comm->nRanks * sizeof(uint32_t), 16) * 2; + // Sync window has one ready and one complete slot per LSA-local rank. + size_t ceDevBaseSize = alignUp(comm->devrState.lsaSize * sizeof(uint32_t), 16) * 2; ncclWindow_vidmem* ceWinDev = nullptr; ncclWindow_vidmem* ceWinDevHost = nullptr; int i = 0; @@ -99,7 +100,7 @@ ncclResult_t ncclCeInit(struct ncclComm* comm) { comm->ceColl.ceSyncWin = (struct ncclDevrWindow*)ceWinDevHost->winHost; comm->ceColl.baseUCSymReadyOffset = 0; - comm->ceColl.baseUCSymComplOffset = alignUp(comm->nRanks * sizeof(uint32_t), 16); + comm->ceColl.baseUCSymComplOffset = alignUp(comm->devrState.lsaSize * sizeof(uint32_t), 16); comm->ceColl.baseUCSymReadyPtr = (uint8_t*)comm->ceColl.ceSyncWin->userPtr + comm->ceColl.baseUCSymReadyOffset; comm->ceColl.baseUCSymComplPtr = (uint8_t*)comm->ceColl.ceSyncWin->userPtr + comm->ceColl.baseUCSymComplOffset; comm->ceColl.ceSeqNum = 0; @@ -178,8 +179,8 @@ bool ncclCeAvailable(struct ncclComm* comm, ncclFunc_t coll, int /*ncclDevRedOp_ TRACE(NCCL_TUNING, "Skipping CE collective: not implemented"); return false; } - if (comm->nNodes > 1) { - TRACE(NCCL_TUNING, "Skipping CE collective: comm is not a single node"); + if (ncclTeamLsa(comm).nRanks < comm->nRanks) { + TRACE(NCCL_TUNING, "Skipping CE collective: not all ranks have scale-up connectivity"); return false; } if (!comm->symmetricSupport) { @@ -220,6 +221,8 @@ ncclResult_t ncclPrepMCSync(struct ncclComm* comm, bool isComplete, hipStreamBat size_t* opIdx, cudaStream_t stream) { ncclResult_t ret = ncclSuccess; + int myLsaRank = comm->devrState.lsaSelf; + int lsaSize = comm->devrState.lsaSize; uint32_t* readyPtrs = (uint32_t*)comm->ceColl.baseUCSymReadyPtr; uint32_t* completePtrs = (uint32_t*)comm->ceColl.baseUCSymComplPtr; @@ -233,7 +236,7 @@ ncclResult_t ncclPrepMCSync(struct ncclComm* comm, bool isComplete, hipStreamBat // Use multi-cast address as destination pointer void* mcDstPtr; - void* dstPtr = isComplete ? (void*)&completePtrs[comm->rank] : (void*)&readyPtrs[comm->rank]; + void* dstPtr = isComplete ? (void*)&completePtrs[myLsaRank] : (void*)&readyPtrs[myLsaRank]; size_t offset = (uint8_t*)dstPtr - (uint8_t*)comm->ceColl.ceSyncWin->userPtr; NCCLCHECKGOTO(ncclDevrGetLsaTeamPtrMC(comm, comm->ceColl.ceSyncWin, offset, ncclTeamLsa(comm), &mcDstPtr), ret, fail); @@ -241,8 +244,8 @@ ncclResult_t ncclPrepMCSync(struct ncclComm* comm, bool isComplete, hipStreamBat CUDACHECKGOTO(cudaMemcpyAsync(mcDstPtr, srcPtr, sizeof(uint32_t), cudaMemcpyHostToDevice, stream), ret, fail); // Add local wait operations for every other rank - for (int r = 0; r < comm->nRanks; ++r) { - if (r == comm->rank) continue; + for (int r = 0; r < lsaSize; ++r) { + if (r == myLsaRank) continue; batchParams[*opIdx] = {}; batchParams[*opIdx].waitValue.operation = CU_STREAM_MEM_OP_WAIT_VALUE_32; batchParams[*opIdx].waitValue.address = (CUdeviceptr)(isComplete ? (void*)&completePtrs[r] : (void*)&readyPtrs[r]); @@ -265,6 +268,8 @@ ncclResult_t ncclPrepUCSync(struct ncclComm* comm, bool isComplete, hipStreamBat NCCLCHECK(ceFaultCheck(comm, CE_FAULT_SYNC_PREP, "ncclPrepUCSync")); #endif + int myLsaRank = comm->devrState.lsaSelf; + int lsaSize = comm->devrState.lsaSize; uint32_t* readyPtrs = (uint32_t*)comm->ceColl.baseUCSymReadyPtr; uint32_t* completePtrs = (uint32_t*)comm->ceColl.baseUCSymComplPtr; @@ -273,10 +278,10 @@ ncclResult_t ncclPrepUCSync(struct ncclComm* comm, bool isComplete, hipStreamBat // Write our own ready/complete flag to remote ranks uint32_t waitValue = capturing ? GRAPH_SYNC_VALUE : currentSeq; - for (int r = 0; r < comm->nRanks; ++r) { - if (r == comm->rank) continue; + for (int r = 0; r < lsaSize; ++r) { + if (r == myLsaRank) continue; void* peerDstPtr; - void* dstPtr = isComplete ? (void*)&completePtrs[comm->rank] : (void*)&readyPtrs[comm->rank]; + void* dstPtr = isComplete ? (void*)&completePtrs[myLsaRank] : (void*)&readyPtrs[myLsaRank]; size_t offset = (uint8_t*)dstPtr - (uint8_t*)comm->ceColl.ceSyncWin->userPtr; NCCLCHECKGOTO(ncclDevrGetLsaRankPtr(comm, comm->ceColl.ceSyncWin, offset, r, &peerDstPtr), ret, fail); batchParams[*opIdx] = {}; @@ -288,8 +293,8 @@ ncclResult_t ncclPrepUCSync(struct ncclComm* comm, bool isComplete, hipStreamBat } // Add local wait operations for every other rank - for (int r = 0; r < comm->nRanks; ++r) { - if (r == comm->rank) continue; + for (int r = 0; r < lsaSize; ++r) { + if (r == myLsaRank) continue; batchParams[*opIdx] = {}; batchParams[*opIdx].waitValue.operation = CU_STREAM_MEM_OP_WAIT_VALUE_32; batchParams[*opIdx].waitValue.address = (CUdeviceptr)(isComplete ? (void*)&completePtrs[r] : (void*)&readyPtrs[r]); @@ -313,15 +318,16 @@ ncclResult_t ncclMemOpSync(struct ncclComm* comm, cudaStream_t stream, struct nc uint32_t* readyPtrs = (uint32_t*)comm->ceColl.baseUCSymReadyPtr; uint32_t* completePtrs = (uint32_t*)comm->ceColl.baseUCSymComplPtr; - // Allocate enough slots for all possible ops - size_t batchSize = (comm->nvlsSupport ? NCCL_CE_SYNC_OPS_PER_RANK_MC : NCCL_CE_SYNC_OPS_PER_RANK_UC) * comm->nRanks; + // Multicast synchronization is not available across scale-up cliques. + bool useMCSync = comm->nvlsSupport && !comm->p2pCrossClique; + size_t batchSize = (useMCSync ? NCCL_CE_SYNC_OPS_PER_RANK_MC : NCCL_CE_SYNC_OPS_PER_RANK_UC) * lsaSize; size_t opIdx = 0; // Prepare batch memory operations for synchronization hipStreamBatchMemOpParams* batchParams = nullptr; NCCLCHECKGOTO(ncclCalloc(&batchParams, batchSize), ret, fail); - if (comm->nvlsSupport) { + if (useMCSync) { NCCLCHECKGOTO(ncclPrepMCSync(comm, comm->ceColl.useCompletePtr, batchParams, &opIdx, stream), ret, fail); } else { NCCLCHECKGOTO(ncclPrepUCSync(comm, comm->ceColl.useCompletePtr, batchParams, &opIdx), ret, fail); @@ -385,11 +391,19 @@ ncclResult_t ncclCeInitBatchOpsParams(struct ncclCeBatchOpsParams* params, int n void ncclCeFreeBatchOpsParams(struct ncclCeBatchOpsParams* params) { if (params->srcs) free(params->srcs); + params->srcs = nullptr; if (params->dsts) free(params->dsts); + params->dsts = nullptr; if (params->sizes) free(params->sizes); + params->sizes = nullptr; + params->numOps = 0; + params->intraBatchSync = false; #ifdef CE_BATCH_ASYNC_SUPPORTED if (params->attrs) free(params->attrs); + params->attrs = nullptr; if (params->attrIdxs) free(params->attrIdxs); + params->attrIdxs = nullptr; + params->numAttrs = 0; #endif } @@ -1408,21 +1422,39 @@ ncclResult_t ncclLaunchCeColl(struct ncclComm* comm, struct ncclKernelPlan* plan // Start CE collective profiling NCCLCHECKGOTO(ncclProfilerStartCeCollEvent(comm, args, stream), ret, fail); - switch (args->func) { - case ncclFuncAllGather: - NCCLCHECKGOTO(ncclCeAllGather(comm, args, stream), ret, fail); - break; - case ncclFuncAlltoAll: - NCCLCHECKGOTO(ncclCeAlltoAll(comm, args, stream), ret, fail); - break; - case ncclFuncScatter: - NCCLCHECKGOTO(ncclCeScatter(comm, args, stream), ret, fail); - break; - case ncclFuncGather: - NCCLCHECKGOTO(ncclCeGather(comm, args, stream), ret, fail); - break; - default: - ret = ncclInvalidUsage; + // Hierarchical path: inter-node RMA + intra-node CE + // Use ncclDevrIsOneLsaTeam instead of comm->nNodes as multi-clique single-NVLD should use CE path + if (!ncclDevrIsOneLsaTeam(comm)) { + switch (args->func) { + case ncclFuncAllGather: + NCCLCHECKGOTO(ncclHierCeAllGather(comm, plan, stream), ret, fail); + break; + case ncclFuncAlltoAll: + NCCLCHECKGOTO(ncclHierCeAlltoAll(comm, plan, stream), ret, fail); + break; + default: + WARN("Hierarchical CE collective not supported for %s", ncclFuncToString(args->func)); + ret = ncclInvalidUsage; + } + } + // LSA-local CE path + else { + switch (args->func) { + case ncclFuncAllGather: + NCCLCHECKGOTO(ncclCeAllGather(comm, args, stream), ret, fail); + break; + case ncclFuncAlltoAll: + NCCLCHECKGOTO(ncclCeAlltoAll(comm, args, stream), ret, fail); + break; + case ncclFuncScatter: + NCCLCHECKGOTO(ncclCeScatter(comm, args, stream), ret, fail); + break; + case ncclFuncGather: + NCCLCHECKGOTO(ncclCeGather(comm, args, stream), ret, fail); + break; + default: + ret = ncclInvalidUsage; + } } exit: diff --git a/projects/rccl/src/dev_runtime.cc b/projects/rccl/src/dev_runtime.cc index 8ca296eda687..412c7130727b 100644 --- a/projects/rccl/src/dev_runtime.cc +++ b/projects/rccl/src/dev_runtime.cc @@ -2270,12 +2270,20 @@ void* ncclDevrGetRmaWin(struct ncclDevrWindow* winHost, int ctx) { if (ctx < 0 || ctx >= NCCL_GIN_MAX_CONNECTIONS) { return nullptr; } - // Non-symmetric proxy path stores rmaDevWins directly on the window struct + // Non-symmetric proxy path stores RMA handles directly on the window struct // (memory == nullptr in that case); symmetric path goes through memory. if (winHost->memory == nullptr) { - return winHost->rmaDevWins[ctx]; + return winHost->rmaHostWins[ctx]; } - return winHost->memory->rmaDevWins[ctx]; + return winHost->memory->rmaHostWins[ctx]; +} + +// Get the byte offset of a window within its backing memory allocation. +size_t ncclDevrGetWinOffset(struct ncclDevrWindow* winHost) { + if (winHost == nullptr || winHost->memory == nullptr) { + return 0; + } + return winHost->bigOffset - winHost->memory->bigOffset; } // Get the multicast address for a given team diff --git a/projects/rccl/src/enqueue.cc b/projects/rccl/src/enqueue.cc index 9925962765e9..1309bcb428fc 100644 --- a/projects/rccl/src/enqueue.cc +++ b/projects/rccl/src/enqueue.cc @@ -1946,29 +1946,7 @@ ncclResult_t ncclLaunchPrepare(struct ncclComm* comm) { nPlans += 1; } } else if (!ncclIntruQueueEmpty(&planner->collCeTaskQueue)) { - struct ncclTaskColl* task = ncclIntruQueueHead(&planner->collCeTaskQueue); - plan->isCeColl = true; - plan->ceCollArgs = ncclMemoryStackAlloc(&comm->memScoped); - plan->ceCollArgs->rootRank = task->root; - plan->ceCollArgs->datatype = task->datatype; - plan->ceCollArgs->nElts = task->count; - plan->ceCollArgs->eltSize = ncclTypeSize(task->datatype); - plan->ceCollArgs->sendBuff = (uint8_t*)task->sendbuff; - plan->ceCollArgs->recvBuff = (uint8_t*)task->recvbuff; - plan->ceCollArgs->func = task->func; - plan->ceCollArgs->sendWin = task->sendWin; - plan->ceCollArgs->recvWin = task->recvWin; - plan->ceCollArgs->collApiEventHandle = task->collApiEventHandle; - - if (comm->rank == 0) { - const char* nvlsSync = comm->nvlsSupport ? "; CE synchronization with NVLS" : ""; - INFO(NCCL_TUNING, "%s [Copy Engine]: %ld Bytes -> cudaMemcpy%s", ncclFuncToString(task->func), - task->count * ncclTypeSize(task->datatype), nvlsSync); - } - - ncclIntruQueueEnqueue(&planner->planQueue, plan); - ncclIntruQueueDequeue(&planner->collCeTaskQueue); - ncclMemoryPoolFree(&comm->memPool_ncclTaskColl, task); + NCCLCHECKGOTO(scheduleCeCollTaskToPlan(comm, plan), result, failure); nPlans += 1; } else { if (!ncclIntruQueueEmpty(&planner->collSymTaskQueue)) { diff --git a/projects/rccl/src/include/dev_runtime.h b/projects/rccl/src/include/dev_runtime.h index ec7c4df38109..e0630f49b4b3 100644 --- a/projects/rccl/src/include/dev_runtime.h +++ b/projects/rccl/src/include/dev_runtime.h @@ -125,6 +125,9 @@ ncclResult_t ncclDevrWorldToLsaRank(struct ncclComm* comm, int peerWorldRank, in // Get the RMA window handle for a specific context void* ncclDevrGetRmaWin(struct ncclDevrWindow* winHost, int ctx); +// Get the byte offset of a window within its backing memory allocation. +size_t ncclDevrGetWinOffset(struct ncclDevrWindow* winHost); + // Get the multicast address for a given team ncclResult_t ncclDevrGetLsaTeamPtrMC(struct ncclComm* comm, struct ncclDevrWindow* winHost, size_t offset, struct ncclTeam lsaTeam, void** outPtr); diff --git a/projects/rccl/src/rma/rma_proxy_launch.cc b/projects/rccl/src/rma/rma_proxy_launch.cc index 8b16c78f27f6..6968930732b7 100644 --- a/projects/rccl/src/rma/rma_proxy_launch.cc +++ b/projects/rccl/src/rma/rma_proxy_launch.cc @@ -63,9 +63,9 @@ ncclResult_t ncclRmaProxyPutBuildOp(struct ncclComm* comm, struct ncclRmaProxyCt bool persistent, struct ncclDevrWindow* srcWin, size_t srcOff, struct ncclDevrWindow* peerWin, size_t peerOff, size_t size, int peer, ncclSignalMode_t signalMode, struct ncclRmaPutSignalOp* op) { - op->srcOff = srcOff; + op->srcOff = ncclDevrGetWinOffset(srcWin) + srcOff; op->srcHandle = ncclDevrGetRmaWin(srcWin, ctx); - op->dstOff = peerOff; + op->dstOff = ncclDevrGetWinOffset(peerWin) + peerOff; op->dstHandle = ncclDevrGetRmaWin(peerWin, ctx); op->size = size; op->targetRank = peer; From 69d542a2162dfbb2e4c3ad8721d443d3a282c37c Mon Sep 17 00:00:00 2001 From: Thomas Huber Date: Fri, 7 Aug 2026 19:22:46 +0000 Subject: [PATCH 02/12] test(rccl): Cover hierarchical CE scale-out collectives Add topology-gated multi-node AllGather and AllToAll coverage with repeated data-distinct iterations and hierarchical path assertions. Keep white-box synchronization expectations aligned with the LSA-local rank domain. AICOMRCCL-630 Co-authored-by: Cursor --- projects/rccl/test/ce/CeInternalMPITests.cpp | 19 ++- projects/rccl/test/ce/CeMPITests.cpp | 123 ++++++++++++++++--- 2 files changed, 112 insertions(+), 30 deletions(-) diff --git a/projects/rccl/test/ce/CeInternalMPITests.cpp b/projects/rccl/test/ce/CeInternalMPITests.cpp index 2c8d5a0993b3..6d789110b78d 100644 --- a/projects/rccl/test/ce/CeInternalMPITests.cpp +++ b/projects/rccl/test/ce/CeInternalMPITests.cpp @@ -96,10 +96,10 @@ class CeInternalMPITest : public MPITestBase debugGuard_.reset(); } - // Batch buffer for ncclPrepUCSync: capacity = NCCL_CE_SYNC_OPS_PER_RANK_UC * nRanks. + // Batch buffer for ncclPrepUCSync: capacity is based on the LSA-local team. std::vector makePrepSyncBatch() const { - size_t batchSize = NCCL_CE_SYNC_OPS_PER_RANK_UC * ceComm->nRanks; + size_t batchSize = NCCL_CE_SYNC_OPS_PER_RANK_UC * ceComm->devrState.lsaSize; return std::vector(batchSize); } @@ -279,22 +279,21 @@ TEST_F(CeInternalMPITest, PrepUCSyncIncrementsCeSeqNum) } } -// SYNC-02: ncclPrepUCSync produces exactly 2*(nRanks-1) ops (one WRITE + -// one WAIT per remote rank). +// SYNC-02: ncclPrepUCSync produces one WRITE and one WAIT per remote LSA rank. TEST_F(CeInternalMPITest, PrepUCSyncOpCount) { requireMinRanks(2); - const int nRanks = ceComm->nRanks; + const int lsaSize = ceComm->devrState.lsaSize; auto [batch, opIdx] = callPrepUCSync(); - EXPECT_EQ(opIdx, static_cast(2 * (nRanks - 1))) - << "expected 2*(nRanks-1) ops for nRanks=" << nRanks; + EXPECT_EQ(opIdx, static_cast(2 * (lsaSize - 1))) + << "expected 2*(lsaSize-1) ops for lsaSize=" << lsaSize; } // SYNC-03: No WRITE_VALUE op targets the local rank's own ready slot. TEST_F(CeInternalMPITest, PrepUCSyncNoSelfTargetedOp) { requireMinRanks(2); - const int rank = ceComm->rank; + const int lsaRank = ceComm->devrState.lsaSelf; auto [batch, opIdx] = callPrepUCSync(); @@ -303,7 +302,7 @@ TEST_F(CeInternalMPITest, PrepUCSyncNoSelfTargetedOp) // on the host-side pointer is valid for comparison with batch op addresses, // which are also expressed as host-visible hipDeviceptr_t values. uint32_t* readyPtrs = reinterpret_cast(ceComm->ceColl.baseUCSymReadyPtr); - hipDeviceptr_t selfReady = reinterpret_cast(&readyPtrs[rank]); + hipDeviceptr_t selfReady = reinterpret_cast(&readyPtrs[lsaRank]); int selfTargets = 0; for(size_t i = 0; i < opIdx; ++i) { @@ -312,7 +311,7 @@ TEST_F(CeInternalMPITest, PrepUCSyncNoSelfTargetedOp) ++selfTargets; } EXPECT_EQ(selfTargets, 0) - << "A WRITE_VALUE op targeted rank " << rank << "'s own ready slot"; + << "A WRITE_VALUE op targeted LSA rank " << lsaRank << "'s own ready slot"; } // SYNC-04: ceSeqNum wraps correctly from UINT32_MAX → 0. diff --git a/projects/rccl/test/ce/CeMPITests.cpp b/projects/rccl/test/ce/CeMPITests.cpp index f0ae3352861e..d7b40adee562 100644 --- a/projects/rccl/test/ce/CeMPITests.cpp +++ b/projects/rccl/test/ce/CeMPITests.cpp @@ -33,6 +33,7 @@ constexpr int kMinRanks4 = 4; constexpr int kMinRanks8 = 8; constexpr int kStressIters = 20; // back-to-back iterations for stress test constexpr int kInterleavedIters = 10; // iterations for CE+SM interleaved stress test +constexpr int kScaleOutIters = 3; // repeated RMA sequence reuse in hierarchical tests } // namespace CeMPITestConstants using namespace CeMPITestConstants; @@ -118,6 +119,40 @@ class CeMPITest : public MPITestBase } } + bool isScaleOutTopology() const + { + MPI_Comm localComm = MPI_COMM_NULL; + if(MPI_Comm_split_type(MPI_COMM_WORLD, MPI_COMM_TYPE_SHARED, 0, + MPI_INFO_NULL, &localComm) != MPI_SUCCESS) + return false; + + int worldSize = 0; + int localSize = 0; + int minLocalSize = 0; + int maxLocalSize = 0; + MPI_Comm_size(MPI_COMM_WORLD, &worldSize); + MPI_Comm_size(localComm, &localSize); + MPI_Allreduce(&localSize, &minLocalSize, 1, MPI_INT, MPI_MIN, MPI_COMM_WORLD); + MPI_Allreduce(&localSize, &maxLocalSize, 1, MPI_INT, MPI_MAX, MPI_COMM_WORLD); + MPI_Comm_free(&localComm); + + return worldSize > maxLocalSize && minLocalSize >= 2; + } + + void assertHierarchicalCEPathTaken(int rank, const char* context) + { + if(!isCeExpected()) + return; + + const std::string log = readAllLogs(); + if(rank == 0) + { + EXPECT_NE(log.find("[Hierarchical CE]"), std::string::npos) + << context + << ": hierarchical CE selection marker absent from rank 0 log"; + } + } + // Root-only variant: only root emits "CE: rank" (non-root has numOps=0 in Scatter/Gather). void assertCEPathTakenOnRoot(int rootRank, int myRank, const char* context) { @@ -238,10 +273,13 @@ class CeMPITest : public MPITestBase class CeMPI_AllGather : public CeMPITest { protected: - void runAllGather(int minRanks, size_t count, const char* testId) + void runAllGather(int minRanks, size_t count, const char* testId, + bool requireScaleOut = false) { if(!validateTestPrerequisites(minRanks)) GTEST_SKIP() << "Need >= " << minRanks << " MPI ranks"; + if(requireScaleOut && !isScaleOutTopology()) + GTEST_SKIP() << "Need at least 2 nodes and 2 MPI ranks per node"; ASSERT_EQ(ncclSuccess, createTestCommunicator()); @@ -253,17 +291,33 @@ class CeMPI_AllGather : public CeMPITest ASSERT_EQ(ncclSuccess, allocSymBuf(count * sizeof(float), sendSym)); ASSERT_EQ(ncclSuccess, allocSymBuf(count * nRanks * sizeof(float), recvSym)); - fillRankScalar(sendSym.ptr, count, rank); - - ASSERT_EQ(ncclSuccess, - ncclAllGather(sendSym.ptr, recvSym.ptr, count, ncclFloat32, - getActiveCommunicator(), getActiveStream())); - ASSERT_EQ(hipSuccess, hipStreamSynchronize(getActiveStream())); - - ASSERT_TRUE(verifyBlockPattern(recvSym.ptr, count * nRanks, count)) - << "Rank " << rank << ": AllGather data verification failed"; + const int iterations = requireScaleOut ? kScaleOutIters : 1; + for(int iter = 0; iter < iterations; ++iter) + { + const int epoch = iter * nRanks; + ASSERT_EQ( + hipSuccess, + initializeBufferWithPattern( + sendSym.ptr, count, + [rank, epoch](size_t) { return static_cast(epoch + rank + 1); })); + + ASSERT_EQ(ncclSuccess, + ncclAllGather(sendSym.ptr, recvSym.ptr, count, ncclFloat32, + getActiveCommunicator(), getActiveStream())); + ASSERT_EQ(hipSuccess, hipStreamSynchronize(getActiveStream())); + + ASSERT_TRUE(verifyBufferData( + recvSym.ptr, count * nRanks, + [count, epoch](size_t i) { + return static_cast(epoch + i / count + 1); + })) + << "Rank " << rank << ": AllGather data verification failed at iteration " + << iter; + } assertCEPathTaken(testId); + if(requireScaleOut) + assertHierarchicalCEPathTaken(rank, testId); // numOps = nRanks (one copy per rank including self); chunkBytes = count * sizeof(float) assertCEBatchPath(ceExpectIntraBatchSync(nRanks, count * sizeof(float)), testId); } @@ -277,6 +331,11 @@ TEST_F(CeMPI_AllGather, FourRanks) { runAllGather(kMinRanks4, kMediumCount, "C TEST_F(CeMPI_AllGather, EightRanks) { runAllGather(kMinRanks8, kMediumCount, "CeMPI_AllGather/EightRanks"); } // CE-MPI-AG-04: Edge case — single element per rank. TEST_F(CeMPI_AllGather, SingleElement) { runAllGather(kMinRanks2, 1, "CeMPI_AllGather/SingleElement"); } +// CE-MPI-AG-05: Multi-node RMA proxy + intra-node CE path. +TEST_F(CeMPI_AllGather, MultiNodeHierarchical) +{ + runAllGather(kMinRanks4, kSmallCount, "CeMPI_AllGather/MultiNodeHierarchical", true); +} // =========================================================================== // CeMPI_AlltoAll – ncclAlltoAll CE correctness + log verification @@ -286,10 +345,13 @@ class CeMPI_AlltoAll : public CeMPITest { protected: // count is per-rank-per-destination; total send/recv buffer = count * nRanks. - void runAlltoAll(int minRanks, size_t count, const char* testId) + void runAlltoAll(int minRanks, size_t count, const char* testId, + bool requireScaleOut = false) { if(!validateTestPrerequisites(minRanks)) GTEST_SKIP() << "Need >= " << minRanks << " MPI ranks"; + if(requireScaleOut && !isScaleOutTopology()) + GTEST_SKIP() << "Need at least 2 nodes and 2 MPI ranks per node"; ASSERT_EQ(ncclSuccess, createTestCommunicator()); @@ -303,17 +365,33 @@ class CeMPI_AlltoAll : public CeMPITest ASSERT_EQ(ncclSuccess, allocSymBuf(totalElem * sizeof(float), sendSym)); ASSERT_EQ(ncclSuccess, allocSymBuf(totalElem * sizeof(float), recvSym)); - fillRankScalar(sendSym.ptr, totalElem, rank); - - ASSERT_EQ(ncclSuccess, - ncclAlltoAll(sendSym.ptr, recvSym.ptr, count, ncclFloat32, - getActiveCommunicator(), getActiveStream())); - ASSERT_EQ(hipSuccess, hipStreamSynchronize(getActiveStream())); - - ASSERT_TRUE(verifyBlockPattern(recvSym.ptr, totalElem, count)) - << "Rank " << rank << ": AlltoAll data verification failed"; + const int iterations = requireScaleOut ? kScaleOutIters : 1; + for(int iter = 0; iter < iterations; ++iter) + { + const int epoch = iter * nRanks; + ASSERT_EQ( + hipSuccess, + initializeBufferWithPattern( + sendSym.ptr, totalElem, + [rank, epoch](size_t) { return static_cast(epoch + rank + 1); })); + + ASSERT_EQ(ncclSuccess, + ncclAlltoAll(sendSym.ptr, recvSym.ptr, count, ncclFloat32, + getActiveCommunicator(), getActiveStream())); + ASSERT_EQ(hipSuccess, hipStreamSynchronize(getActiveStream())); + + ASSERT_TRUE(verifyBufferData( + recvSym.ptr, totalElem, + [count, epoch](size_t i) { + return static_cast(epoch + i / count + 1); + })) + << "Rank " << rank << ": AlltoAll data verification failed at iteration " + << iter; + } assertCEPathTaken(testId); + if(requireScaleOut) + assertHierarchicalCEPathTaken(rank, testId); // numOps = nRanks (one per destination rank); chunkBytes = count * sizeof(float) assertCEBatchPath(ceExpectIntraBatchSync(nRanks, count * sizeof(float)), testId); } @@ -327,6 +405,11 @@ TEST_F(CeMPI_AlltoAll, FourRanks) { runAlltoAll(kMinRanks4, kMediumCount, "CeMP TEST_F(CeMPI_AlltoAll, EightRanks) { runAlltoAll(kMinRanks8, kMediumCount, "CeMPI_AlltoAll/EightRanks"); } // CE-MPI-A2A-04: Odd rank count (3) — non-power-of-two op layout vs 2/4/8 ranks. TEST_F(CeMPI_AlltoAll, ThreeRanks) { runAlltoAll(3, kSmallCount, "CeMPI_AlltoAll/ThreeRanks"); } +// CE-MPI-A2A-05: Multi-node RMA proxy + intra-node CE path. +TEST_F(CeMPI_AlltoAll, MultiNodeHierarchical) +{ + runAlltoAll(kMinRanks4, kSmallCount, "CeMPI_AlltoAll/MultiNodeHierarchical", true); +} // =========================================================================== // CeMPI_Scatter – ncclScatter CE correctness + log verification From fb7299f3fe0ed55833e927fc06d7da8698a48b40 Mon Sep 17 00:00:00 2001 From: Thomas Huber Date: Tue, 8 Sep 2026 18:11:03 +0000 Subject: [PATCH 03/12] fix(rccl): Address CE scale-out review feedback Correctness: - ncclMemOpSync: the graph-capture reset loop walked comm->nRanks while the ready/complete arrays and batchParams are both sized from the LSA team. With NCCL_CE_SYNC_OPS_PER_RANK_UC == 3 that overruns batchParams from four nodes up and indexes the flag arrays out of bounds from two nodes up. Walk lsaSize. - ncclLaunchCeColl: match the predicate ncclHierCeAvailable admits the task under (nNodes > 1 && !ncclDevrIsOneLsaTeam). Previously a single-node communicator with a reduced LSA team could be dispatched to the hierarchical path without its RMA prerequisites having been checked. - ncclCeInit: skip the CE AllReduce staging buffer unless the communicator is one LSA team. CE AllReduce is unreachable on the hierarchical path, so the allocation and its window registration were pure overhead there. Clarity: - Define RCCL_CE_HIER_SELECTED_TAG once and use it in both emitters and the test assertion, so the marker cannot drift from what the test matches. - Derive baseUCSymComplOffset from ceDevBaseSize. Tests: - Skip the scale-out cases when the communicator reports no symmetric memory support, so a stack without a GIN backend does not report a failure for behaving correctly. MPI topology alone does not imply the path is reachable. - Expect CE batch ops per LSA destination rather than per world rank. - Make the AlltoAll pattern depend on both endpoints so a wrong source slice changes the received bytes; the previous sender-only fill could not detect it. - Guard the LSA-scoped sync assertions with requireMinLsaRanks so two nodes at one rank each cannot pass them vacuously on an empty batch. - Pin the idempotence of ncclCeFreeBatchOpsParams, which hierarchical AllGather relies on when it releases per-chunk scratch and again at exit. - Cover the 64 MiB hierarchical chunk boundary with 68 MiB per rank and peer. AICOMRCCL-630 Co-authored-by: Cursor --- projects/rccl/CHANGELOG.md | 1 + projects/rccl/src/ce_coll.cc | 21 +++-- projects/rccl/src/enqueue.cc | 9 +- projects/rccl/src/include/ce_coll.h | 4 + projects/rccl/test/ce/CeInternalMPITests.cpp | 34 +++++++ projects/rccl/test/ce/CeMPITests.cpp | 96 ++++++++++++++++---- 6 files changed, 134 insertions(+), 31 deletions(-) diff --git a/projects/rccl/CHANGELOG.md b/projects/rccl/CHANGELOG.md index 4ce48b3b7af6..818449f7f3df 100644 --- a/projects/rccl/CHANGELOG.md +++ b/projects/rccl/CHANGELOG.md @@ -9,6 +9,7 @@ Full documentation for RCCL is available at [https://rccl.readthedocs.io](https: * Added scalable AllGatherV pattern: grouped `ncclBroadcast` calls with distinct roots are fused into a single ring kernel, improving performance at large scale. Gated by `NCCL_ALLGATHERV_ENABLE` (default off). * Added GPU-only multi-segment registration for symmetric memory windows, enabling contiguous VA ranges backed by multiple physical segments (single-node validated). * Added Elastic Buffer support for symmetric windows spanning device and host/`HOST_NUMA` memory segments (`NCCL_ELASTIC_BUFFER_REGISTER`, `NCCL_SYM_REUSE_SYSMEM_HANDLES`). Single-node path validated; multi-node registration remains limited pending HIP/HSA multi-segment DMA-BUF export support. +* Enabled hierarchical Copy Engine `ncclAllGather` and `ncclAlltoAll` on multi-node communicators: the intra-node phase runs over SDMA/CE across the LSA team and the inter-node phase over one-sided RMA through the CPU (GIN) proxy. Selected when `NCCL_CTA_POLICY=2` with symmetric send and receive windows, `NCCL_LOCAL_REGISTER=0`, `NCCL_CUMEM_ENABLE=1`, `NCCL_NUM_RMA_CTX` greater than zero, and a GIN backend available. Validated for correctness and path selection from 1 to 32 nodes. ### Changed * Raised the default channel count on single-node gfx1250 (MI450) to 256 for both collectives and P2P. The count is still clamped by the GPU CU count and by `NCCL_MAX_NCHANNELS` / `NCCL_MAX_CTAS` / `NCCL_MAX_P2P_NCHANNELS`. Multi-node gfx1250 keeps the 64-channel cap on the NET path. `RCCL_SATURATE_P2P_NCHANNELS` now defaults to on for gfx1250 so the per-peer channel count tiles the larger pool; set it to `0` to restore the previous behavior. diff --git a/projects/rccl/src/ce_coll.cc b/projects/rccl/src/ce_coll.cc index e6e5bdcf39b3..19524afd5d88 100644 --- a/projects/rccl/src/ce_coll.cc +++ b/projects/rccl/src/ce_coll.cc @@ -142,7 +142,7 @@ ncclResult_t ncclCeInit(struct ncclComm* comm) { comm->ceColl.ceSyncWin = (struct ncclDevrWindow*)ceWinDevHost->winHost; comm->ceColl.baseUCSymReadyOffset = 0; - comm->ceColl.baseUCSymComplOffset = alignUp(comm->devrState.lsaSize * sizeof(uint32_t), 16); + comm->ceColl.baseUCSymComplOffset = ceDevBaseSize / 2; comm->ceColl.baseUCSymReadyPtr = (uint8_t*)comm->ceColl.ceSyncWin->userPtr + comm->ceColl.baseUCSymReadyOffset; comm->ceColl.baseUCSymComplPtr = (uint8_t*)comm->ceColl.ceSyncWin->userPtr + comm->ceColl.baseUCSymComplOffset; comm->ceColl.ceSeqNum = 0; @@ -174,7 +174,9 @@ ncclResult_t ncclCeInit(struct ncclComm* comm) { // CE AllReduce staging buffer (double-buffered scatter staging, no scratch): // [slot 0: nRanks chunks][slot 1: nRanks chunks]. - if (rcclParamCeAllReduce()) { + // CE AllReduce is LSA-local only, so skip the staging buffer on communicators + // that reach ncclCeInit solely for the hierarchical path. + if (rcclParamCeAllReduce() && ncclDevrIsOneLsaTeam(comm)) { NCCLCHECKGOTO(ncclMemAlloc((void**)&ceARTmpBuf, ceARTmpBufSize), ret, fail_ar); NCCLCHECKGOTO(ncclDevrWindowRegisterInGroup(comm, ceARTmpBuf, ceARTmpBufSize, NCCL_WIN_COLL_SYMMETRIC, &arWinDev), ret, fail_ar); @@ -564,7 +566,9 @@ ncclResult_t ncclMemOpSync(struct ncclComm* comm, cudaStream_t stream, struct nc // fixed-value barrier can be replayed. if (ncclCudaGraphValid(comm->planner.capturingGraph)) { size_t resetIdx = 0; - for (int i = 0; i < comm->nRanks; i++) { + // The flag arrays and batchParams are both sized from the LSA team, so the + // reset must walk lsaSize, not nRanks. + for (int i = 0; i < lsaSize; i++) { batchParams[resetIdx] = {}; batchParams[resetIdx].writeValue.operation = CU_STREAM_MEM_OP_WRITE_VALUE_32; batchParams[resetIdx].writeValue.address = @@ -2018,10 +2022,11 @@ ncclResult_t ncclLaunchCeColl(struct ncclComm* comm, struct ncclKernelPlan* plan // Start CE collective profiling NCCLCHECKGOTO(ncclProfilerStartCeCollEvent(comm, args, stream), ret, fail); - // Hierarchical path: inter-node RMA plus intra-node CE. Keyed on LSA coverage - // rather than node count, so a multi-clique single-NVL-domain comm whose LSA - // spans every rank still takes the LSA-local path below. - if (!ncclDevrIsOneLsaTeam(comm)) { + // Hierarchical path: inter-node RMA plus intra-node CE. This predicate must + // match the one ncclHierCeAvailable admitted the task under, otherwise a + // single-node comm with a reduced LSA team (NCCL_LSA_TEAM_SIZE) would be + // dispatched here without the RMA prerequisites having been checked. + if (comm->nNodes > 1 && !ncclDevrIsOneLsaTeam(comm)) { switch (args->func) { case ncclFuncAllGather: NCCLCHECKGOTO(ncclHierCeAllGather(comm, plan, stream), ret, fail); @@ -2124,7 +2129,7 @@ ncclResult_t scheduleCeCollTaskToPlan(struct ncclComm* comm, struct ncclKernelPl if (comm->rank == 0) { if (!ncclDevrIsOneLsaTeam(comm)) { - INFO(NCCL_TUNING, "%s [Hierarchical CE]: %ld Bytes -> RMA proxy + CE", ncclFuncToString(task->func), + INFO(NCCL_TUNING, "%s " RCCL_CE_HIER_SELECTED_TAG ": %ld Bytes -> RMA proxy + CE", ncclFuncToString(task->func), task->count * ncclTypeSize(task->datatype)); } else { const char* nvlsSync = comm->nvlsSupport ? "; CE synchronization with NVLS" : ""; diff --git a/projects/rccl/src/enqueue.cc b/projects/rccl/src/enqueue.cc index 4ebd13b3a0e7..b17e8f5bfe57 100644 --- a/projects/rccl/src/enqueue.cc +++ b/projects/rccl/src/enqueue.cc @@ -2072,8 +2072,8 @@ ncclResult_t ncclLaunchPrepare(struct ncclComm* comm) { if (comm->rank == 0) { if (!ncclDevrIsOneLsaTeam(comm)) { - INFO(NCCL_TUNING, "%s [Hierarchical CE]: %ld Bytes -> RMA proxy + CE", ncclFuncToString(task->func), - task->count * ncclTypeSize(task->datatype)); + INFO(NCCL_TUNING, "%s " RCCL_CE_HIER_SELECTED_TAG ": %ld Bytes -> RMA proxy + CE", + ncclFuncToString(task->func), task->count * ncclTypeSize(task->datatype)); } else { const char* nvlsSync = comm->nvlsSupport ? "; CE synchronization with NVLS" : ""; INFO(NCCL_TUNING, "%s [Copy Engine]: %ld Bytes -> cudaMemcpy%s", ncclFuncToString(task->func), @@ -3857,8 +3857,9 @@ static ncclResult_t taskAppend(struct ncclComm* comm, struct ncclInfo* info) { ceArGraphAllowed = rcclCeAllReduceAllowed(comm); } - // Size gate for CE AllReduce without symmetric memory registration: ceARTmpBuf is sized for at most - // NCCL_CE_AR_MAX_MSG_BYTES total bytes. + // Window registration type and CE eligibility, hoisted above the CE-init trigger below, + // which now consults hierCeAvailable. ceAllReduceFits is the size gate for CE AllReduce + // without symmetric registration: ceARTmpBuf holds at most NCCL_CE_AR_MAX_MSG_BYTES. bool ceAllReduceFits = false; ncclSymRegType_t winRegType; NCCLCHECK(ncclGetSymRegType(sendWin, recvWin, &winRegType)); diff --git a/projects/rccl/src/include/ce_coll.h b/projects/rccl/src/include/ce_coll.h index 1badd7b840b6..7b40e6be87e2 100644 --- a/projects/rccl/src/include/ce_coll.h +++ b/projects/rccl/src/include/ce_coll.h @@ -18,6 +18,10 @@ #define NCCL_CE_SYNC_OPS_PER_RANK_UC 3 #define RCCL_CE_NUM_COPY_STREAMS 8 +// Selection marker for the hierarchical CE path. Defined once because the +// scale-out MPI tests assert on this exact text. +#define RCCL_CE_HIER_SELECTED_TAG "[Hierarchical CE]" + // Default is <= 256 MiB (holds NUM_SLOTS * nRanks chunks (2 scatter slots), // and the reduced output goes to the user recvbuff) #define NCCL_CE_AR_MAX_MSG_BYTES (256ull * 1024 * 1024) diff --git a/projects/rccl/test/ce/CeInternalMPITests.cpp b/projects/rccl/test/ce/CeInternalMPITests.cpp index 57ffc58fb829..7c146823d874 100644 --- a/projects/rccl/test/ce/CeInternalMPITests.cpp +++ b/projects/rccl/test/ce/CeInternalMPITests.cpp @@ -110,6 +110,17 @@ class CeInternalMPITest : public MPITestBase GTEST_SKIP() << "Need >= " << n << " MPI ranks"; } + // Skip if the LSA team has fewer than n ranks. Assertions scoped to the LSA + // domain must guard on it: 2 nodes x 1 rank gives nRanks=2 but lsaSize=1, + // which would otherwise satisfy requireMinRanks(2) and then assert on an + // empty batch. + void requireMinLsaRanks(int n) + { + if(ceComm->devrState.lsaSize < n) + GTEST_SKIP() << "Need >= " << n << " ranks in the LSA team, have " + << ceComm->devrState.lsaSize; + } + // Allocate a batch, call ncclPrepUCSync once, and return both for inspection. struct PrepSyncResult { @@ -361,6 +372,7 @@ TEST_F(CeInternalMPITest, PrepUCSyncIncrementsCeSeqNum) TEST_F(CeInternalMPITest, PrepUCSyncOpCount) { requireMinRanks(2); + requireMinLsaRanks(2); const int lsaSize = ceComm->devrState.lsaSize; auto [batch, opIdx] = callPrepUCSync(); EXPECT_EQ(opIdx, static_cast(2 * (lsaSize - 1))) @@ -371,6 +383,7 @@ TEST_F(CeInternalMPITest, PrepUCSyncOpCount) TEST_F(CeInternalMPITest, PrepUCSyncNoSelfTargetedOp) { requireMinRanks(2); + requireMinLsaRanks(2); const int lsaRank = ceComm->devrState.lsaSelf; auto [batch, opIdx] = callPrepUCSync(); @@ -450,6 +463,27 @@ TEST_F(CeInternalMPITest, SeqNumWrapAroundCollective) } } +// BATCH-01: ncclCeFreeBatchOpsParams is idempotent. Hierarchical AllGather frees +// per-chunk scratch inside its chunk loop and once more at function exit, both on +// the same object, so a release that does not clear the pointers double-frees. +// Single-node, so this pins the invariant wherever the suite runs. +TEST_F(CeInternalMPITest, FreeBatchOpsParamsIsIdempotent) +{ + ncclCeBatchOpsParams params{}; + ASSERT_EQ(ncclCeInitBatchOpsParams(¶ms, 4), ncclSuccess); + ASSERT_NE(params.srcs, nullptr); + + ncclCeFreeBatchOpsParams(¶ms); + EXPECT_EQ(params.srcs, nullptr); + EXPECT_EQ(params.dsts, nullptr); + EXPECT_EQ(params.sizes, nullptr); + EXPECT_EQ(params.numOps, 0); + + // Must not double-free. + ncclCeFreeBatchOpsParams(¶ms); + EXPECT_EQ(params.srcs, nullptr); +} + // =========================================================================== // CeInternal_Launch – ncclCeLaunchBatchOps with zero and real ops // =========================================================================== diff --git a/projects/rccl/test/ce/CeMPITests.cpp b/projects/rccl/test/ce/CeMPITests.cpp index 4e9ed52a4517..dfd669fb437d 100644 --- a/projects/rccl/test/ce/CeMPITests.cpp +++ b/projects/rccl/test/ce/CeMPITests.cpp @@ -15,6 +15,9 @@ #include "TestChecks.hpp" #include "rccl/rccl.h" +// For RCCL_CE_HIER_SELECTED_TAG, so the assertion cannot drift from the emitter. +#include "ce_coll.h" + #include #include #include @@ -40,6 +43,9 @@ constexpr int kMinRanks8 = 8; constexpr int kStressIters = 20; // back-to-back iterations for stress test constexpr int kInterleavedIters = 10; // iterations for CE+SM interleaved stress test constexpr int kScaleOutIters = 3; // repeated RMA sequence reuse in hierarchical tests +// 68 MiB per rank/peer crosses the 64 MiB hierarchical chunk size, exercising the +// multi-chunk plan and the per-chunk scratch release. +constexpr size_t kChunkBoundaryCount = (68ull * 1024 * 1024) / sizeof(float); } // namespace CeMPITestConstants using namespace CeMPITestConstants; @@ -125,26 +131,45 @@ class CeMPITest : public MPITestBase } } - bool isScaleOutTopology() const + // Ranks sharing this rank's node. Also the LSA team size on a topology where + // the scale-up domain is one node, which is what the scale-out cases require. + int localRankCount() const { MPI_Comm localComm = MPI_COMM_NULL; if(MPI_Comm_split_type(MPI_COMM_WORLD, MPI_COMM_TYPE_SHARED, 0, MPI_INFO_NULL, &localComm) != MPI_SUCCESS) - return false; + return 0; - int worldSize = 0; int localSize = 0; + MPI_Comm_size(localComm, &localSize); + MPI_Comm_free(&localComm); + return localSize; + } + + bool isScaleOutTopology() const + { + int worldSize = 0; + int localSize = localRankCount(); int minLocalSize = 0; int maxLocalSize = 0; + if(localSize == 0) return false; MPI_Comm_size(MPI_COMM_WORLD, &worldSize); - MPI_Comm_size(localComm, &localSize); MPI_Allreduce(&localSize, &minLocalSize, 1, MPI_INT, MPI_MIN, MPI_COMM_WORLD); MPI_Allreduce(&localSize, &maxLocalSize, 1, MPI_INT, MPI_MAX, MPI_COMM_WORLD); - MPI_Comm_free(&localComm); return worldSize > maxLocalSize && minLocalSize >= 2; } + // MPI topology alone does not imply hierarchical CE is reachable: it also needs + // symmetric memory, which on a multi-node communicator requires a GIN backend. + // Distinguish "this machine cannot do it" (skip) from "prerequisites are present + // but the path was not taken" (fail), so a stack without GIN/RMA does not report + // a red for behaving correctly. + bool scaleOutPrerequisitesMet() const + { + return readAllLogs().find("Symmetric memory is not supported") == std::string::npos; + } + void assertHierarchicalCEPathTaken(int rank, const char* context) { if(!isCeExpected()) @@ -153,7 +178,7 @@ class CeMPITest : public MPITestBase const std::string log = readAllLogs(); if(rank == 0) { - EXPECT_NE(log.find("[Hierarchical CE]"), std::string::npos) + EXPECT_NE(log.find(RCCL_CE_HIER_SELECTED_TAG), std::string::npos) << context << ": hierarchical CE selection marker absent from rank 0 log"; } @@ -305,7 +330,7 @@ class CeMPI_AllGather : public CeMPITest { protected: void runAllGather(int minRanks, size_t count, const char* testId, - bool requireScaleOut = false) + bool requireScaleOut = false, int scaleOutIters = kScaleOutIters) { if(!validateTestPrerequisites(minRanks)) GTEST_SKIP() << "Need >= " << minRanks << " MPI ranks"; @@ -314,15 +339,20 @@ class CeMPI_AllGather : public CeMPITest ASSERT_EQ(ncclSuccess, createTestCommunicator()); + if(requireScaleOut && !scaleOutPrerequisitesMet()) + GTEST_SKIP() << "Symmetric memory unavailable on this communicator; " + "hierarchical CE needs a GIN backend"; + int rank{}, nRanks{}; ncclCommUserRank(getActiveCommunicator(), &rank); ncclCommCount(getActiveCommunicator(), &nRanks); SymBuf sendSym, recvSym; ASSERT_EQ(ncclSuccess, allocSymBuf(count * sizeof(float), sendSym)); - ASSERT_EQ(ncclSuccess, allocSymBuf(count * nRanks * sizeof(float), recvSym)); + ASSERT_EQ(ncclSuccess, + allocSymBuf(count * static_cast(nRanks) * sizeof(float), recvSym)); - const int iterations = requireScaleOut ? kScaleOutIters : 1; + const int iterations = requireScaleOut ? scaleOutIters : 1; for(int iter = 0; iter < iterations; ++iter) { const int epoch = iter * nRanks; @@ -338,7 +368,7 @@ class CeMPI_AllGather : public CeMPITest ASSERT_EQ(hipSuccess, hipStreamSynchronize(getActiveStream())); ASSERT_TRUE(verifyBufferData( - recvSym.ptr, count * nRanks, + recvSym.ptr, count * static_cast(nRanks), [count, epoch](size_t i) { return static_cast(epoch + i / count + 1); })) @@ -347,10 +377,13 @@ class CeMPI_AllGather : public CeMPITest } assertCEPathTaken(testId); + // CE batch numOps is per LSA destination, which is the node-local rank count + // on the scale-out path and the whole communicator otherwise. + const int batchPeers = requireScaleOut ? localRankCount() : nRanks; if(requireScaleOut) assertHierarchicalCEPathTaken(rank, testId); - // numOps = nRanks (one copy per rank including self); chunkBytes = count * sizeof(float) - assertCEBatchPath(ceExpectIntraBatchSync(nRanks, count * sizeof(float)), testId); + // numOps = batchPeers (one copy per destination); chunkBytes = count * sizeof(float) + assertCEBatchPath(ceExpectIntraBatchSync(batchPeers, count * sizeof(float)), testId); } }; @@ -367,6 +400,12 @@ TEST_F(CeMPI_AllGather, MultiNodeHierarchical) { runAllGather(kMinRanks4, kSmallCount, "CeMPI_AllGather/MultiNodeHierarchical", true); } +// CE-MPI-AG-06: 68 MiB per rank crosses the 64 MiB hierarchical chunk boundary. +TEST_F(CeMPI_AllGather, MultiNodeHierarchicalChunkBoundary) +{ + runAllGather(kMinRanks4, kChunkBoundaryCount, + "CeMPI_AllGather/MultiNodeHierarchicalChunkBoundary", true, 1); +} // =========================================================================== // CeMPI_AlltoAll – ncclAlltoAll CE correctness + log verification @@ -377,7 +416,7 @@ class CeMPI_AlltoAll : public CeMPITest protected: // count is per-rank-per-destination; total send/recv buffer = count * nRanks. void runAlltoAll(int minRanks, size_t count, const char* testId, - bool requireScaleOut = false) + bool requireScaleOut = false, int scaleOutIters = kScaleOutIters) { if(!validateTestPrerequisites(minRanks)) GTEST_SKIP() << "Need >= " << minRanks << " MPI ranks"; @@ -386,6 +425,10 @@ class CeMPI_AlltoAll : public CeMPITest ASSERT_EQ(ncclSuccess, createTestCommunicator()); + if(requireScaleOut && !scaleOutPrerequisitesMet()) + GTEST_SKIP() << "Symmetric memory unavailable on this communicator; " + "hierarchical CE needs a GIN backend"; + int rank{}, nRanks{}; ncclCommUserRank(getActiveCommunicator(), &rank); ncclCommCount(getActiveCommunicator(), &nRanks); @@ -396,35 +439,44 @@ class CeMPI_AlltoAll : public CeMPITest ASSERT_EQ(ncclSuccess, allocSymBuf(totalElem * sizeof(float), sendSym)); ASSERT_EQ(ncclSuccess, allocSymBuf(totalElem * sizeof(float), recvSym)); - const int iterations = requireScaleOut ? kScaleOutIters : 1; + const int iterations = requireScaleOut ? scaleOutIters : 1; for(int iter = 0; iter < iterations; ++iter) { const int epoch = iter * nRanks; + // Value depends on both endpoints: sender s writes epoch + s*nRanks + j + 1 + // into the slice bound for rank j. A wrong source slice therefore changes the + // received bytes, which a sender-only pattern could not detect. ASSERT_EQ( hipSuccess, initializeBufferWithPattern( sendSym.ptr, totalElem, - [rank, epoch](size_t) { return static_cast(epoch + rank + 1); })); + [rank, nRanks, count, epoch](size_t i) { + return static_cast(epoch + rank * nRanks + i / count + 1); + })); ASSERT_EQ(ncclSuccess, ncclAlltoAll(sendSym.ptr, recvSym.ptr, count, ncclFloat32, getActiveCommunicator(), getActiveStream())); ASSERT_EQ(hipSuccess, hipStreamSynchronize(getActiveStream())); + // Slice s of the result is what rank s sent to us, i.e. epoch + s*nRanks + rank + 1. ASSERT_TRUE(verifyBufferData( recvSym.ptr, totalElem, - [count, epoch](size_t i) { - return static_cast(epoch + i / count + 1); + [count, nRanks, rank, epoch](size_t i) { + return static_cast(epoch + (i / count) * nRanks + rank + 1); })) << "Rank " << rank << ": AlltoAll data verification failed at iteration " << iter; } assertCEPathTaken(testId); + // CE batch numOps is per LSA destination, which is the node-local rank count + // on the scale-out path and the whole communicator otherwise. + const int batchPeers = requireScaleOut ? localRankCount() : nRanks; if(requireScaleOut) assertHierarchicalCEPathTaken(rank, testId); - // numOps = nRanks (one per destination rank); chunkBytes = count * sizeof(float) - assertCEBatchPath(ceExpectIntraBatchSync(nRanks, count * sizeof(float)), testId); + // numOps = batchPeers (one per destination); chunkBytes = count * sizeof(float) + assertCEBatchPath(ceExpectIntraBatchSync(batchPeers, count * sizeof(float)), testId); } }; @@ -441,6 +493,12 @@ TEST_F(CeMPI_AlltoAll, MultiNodeHierarchical) { runAlltoAll(kMinRanks4, kSmallCount, "CeMPI_AlltoAll/MultiNodeHierarchical", true); } +// CE-MPI-A2A-06: 68 MiB per peer crosses the 64 MiB hierarchical chunk boundary. +TEST_F(CeMPI_AlltoAll, MultiNodeHierarchicalChunkBoundary) +{ + runAlltoAll(kMinRanks4, kChunkBoundaryCount, + "CeMPI_AlltoAll/MultiNodeHierarchicalChunkBoundary", true, 1); +} // =========================================================================== // CeMPI_Scatter – ncclScatter CE correctness + log verification From d8d27f29a0f7620dd1ad1a755ee3a48b2bb72a1e Mon Sep 17 00:00:00 2001 From: Thomas Huber Date: Wed, 9 Sep 2026 17:27:55 +0000 Subject: [PATCH 04/12] fix(rccl): Address CE scale-out review round 2 Make the hierarchical selection marker use the same predicate the launch dispatches on. ncclLaunchCeColl tests nNodes > 1 && !isOneLsaTeam while the enqueue.cc emitter tested only !isOneLsaTeam, so a single-node comm with a reduced LSA team printed the marker and then ran the LSA-local routine. The scale-out tests use that marker as their only oracle. Both sites now call a shared ncclCeHierDispatch. Reject CE for communicators whose LSA team does not cover them. The LSA-local routines address peers by LSA rank, so on a single node with NCCL_LSA_TEAM_SIZE below nRanks ncclCeAvailable admitted a task that would gather only lsaSize slices, at LSA rather than communicator indices. Such comms now fall back to the kernel path. Default behaviour is unchanged, since computeLsaSize yields nRanks on a single node when the variable is unset. Read the scale-out test prerequisites off the communicator instead of the log. The absence of "Symmetric memory is not supported" clears only the symmetricSupport clause of ncclHierCeAvailable, so a stack with GIN but no RMA context ran the case and failed on an absent marker rather than skipping. Predict intraBatchSync as false on the scale-out branch. Neither hierarchical collective assigns it, so the batch always logs the without-sync line; the 68 MiB chunk-boundary case already clears the byte threshold, so the prediction only held on nodes with at most kCeIntraBatchSyncFreq ranks. Cover the batch-params release fully: assert attrs, attrIdxs, numAttrs and intraBatchSync, and assert the pointers are non-null after init so the nulling checks cannot pass vacuously. Add host coverage for ncclDevrGetWinOffset in DevRuntimeTests, which already includes the hipified dev_runtime.cc, so no new source seam is needed. Co-authored-by: Cursor --- projects/rccl/CHANGELOG.md | 1 + projects/rccl/src/ce_coll.cc | 17 +++++- projects/rccl/src/enqueue.cc | 4 +- projects/rccl/src/include/ce_coll.h | 5 ++ projects/rccl/test/DevRuntimeTests.cpp | 28 ++++++++++ projects/rccl/test/ce/CeInternalMPITests.cpp | 15 ++++++ projects/rccl/test/ce/CeMPITests.cpp | 57 +++++++++++++------- 7 files changed, 105 insertions(+), 22 deletions(-) diff --git a/projects/rccl/CHANGELOG.md b/projects/rccl/CHANGELOG.md index 818449f7f3df..98b32f4b8f93 100644 --- a/projects/rccl/CHANGELOG.md +++ b/projects/rccl/CHANGELOG.md @@ -18,6 +18,7 @@ Full documentation for RCCL is available at [https://rccl.readthedocs.io](https: * Adapted the InfiniBand transports (`net_ib` and `net_ib_cast`) to the v14 GIN/RMA split: the host/proxy backend is now registered as an `ncclRma_t` vtable (`RMA_IB_PROXY`) that owns the `iput`/`iputSignal`/`iget`/`iflush`/`test` data-path ops, with GIN layered on top through the generic `ncclGinProxy`. ### Fixed +* Copy Engine collectives are no longer selected when the LSA team does not cover the communicator. The LSA-local CE routines address peers by LSA rank, so on a single node with `NCCL_LSA_TEAM_SIZE` below the local rank count they gathered only `lsaSize` slices and placed them at LSA rather than communicator indices. Such communicators now fall back to the kernel path. * `NCCL_MAX_P2P_NCHANNELS` opt-in is now detected from the environment rather than from the parameter value. The value defaults to `MAXCHANNELS`, so every unset run was treated as an opt-in past the historical `4*CHANNEL_LIMIT` (64) bound. As a result, P2P channels on non-gfx1250 architectures were limited only by the collective channel count, and the gfx950 (MI350) multi-node P2P caps never applied. Set `NCCL_MAX_P2P_NCHANNELS` explicitly to restore a higher bound. * Fixed `NCCL_CHECK_MODE` having no effect. `commAlloc` reset `comm->checkMode` from the deprecated `NCCL_CHECK_POINTERS` after `NCCL_CHECK_MODE` had already been parsed, so `DEBUG_LOCAL` was only reachable through the deprecated variable and `DEBUG_GLOBAL`, which validates symmetric buffer registration across ranks, was unreachable entirely. diff --git a/projects/rccl/src/ce_coll.cc b/projects/rccl/src/ce_coll.cc index 19524afd5d88..83a0ea424615 100644 --- a/projects/rccl/src/ce_coll.cc +++ b/projects/rccl/src/ce_coll.cc @@ -306,6 +306,13 @@ bool ncclCeAvailable(struct ncclComm* comm, ncclFunc_t coll, int /*ncclDevRedOp_ TRACE(NCCL_TUNING, "Skipping CE collective: comm is not a single node"); return false; } + // The LSA-local routines address peers by LSA rank, so a team that does not + // cover the comm would gather only lsaSize slices, at the wrong indices. + // Reachable on a single node via NCCL_LSA_TEAM_SIZE. + if (!ncclDevrIsOneLsaTeam(comm)) { + TRACE(NCCL_TUNING, "Skipping CE collective: LSA team does not cover the comm"); + return false; + } if (!comm->symmetricSupport) { TRACE(NCCL_TUNING, "Skipping CE collective: symmetric support is not enabled"); return false; @@ -335,6 +342,10 @@ bool ncclCeScratchAvailable(struct ncclComm* comm, ncclFunc_t coll, int /*ncclDe TRACE(NCCL_TUNING, "Skipping CE collective: comm is not a single node"); return false; } + if (!ncclDevrIsOneLsaTeam(comm)) { + TRACE(NCCL_TUNING, "Skipping CE collective: LSA team does not cover the comm"); + return false; + } if (!comm->symmetricSupport) { TRACE(NCCL_TUNING, "Skipping CE collective: symmetric support is not enabled"); return false; @@ -1162,6 +1173,10 @@ bool ncclHierCeAvailable(struct ncclComm* comm, ncclFunc_t coll, int /*ncclDevRe return true; } +bool ncclCeHierDispatch(struct ncclComm* comm) { + return comm->nNodes > 1 && !ncclDevrIsOneLsaTeam(comm); +} + // Per-(peer, chunk) chunking plan in flat form. Peer p's chunks // span [chunkStart[p], chunkStart[p+1]); total chunks = chunkStart[nPeers]. struct ncclHierChunkPlan { @@ -2026,7 +2041,7 @@ ncclResult_t ncclLaunchCeColl(struct ncclComm* comm, struct ncclKernelPlan* plan // match the one ncclHierCeAvailable admitted the task under, otherwise a // single-node comm with a reduced LSA team (NCCL_LSA_TEAM_SIZE) would be // dispatched here without the RMA prerequisites having been checked. - if (comm->nNodes > 1 && !ncclDevrIsOneLsaTeam(comm)) { + if (ncclCeHierDispatch(comm)) { switch (args->func) { case ncclFuncAllGather: NCCLCHECKGOTO(ncclHierCeAllGather(comm, plan, stream), ret, fail); diff --git a/projects/rccl/src/enqueue.cc b/projects/rccl/src/enqueue.cc index b17e8f5bfe57..e43b2cf21b89 100644 --- a/projects/rccl/src/enqueue.cc +++ b/projects/rccl/src/enqueue.cc @@ -2071,7 +2071,9 @@ ncclResult_t ncclLaunchPrepare(struct ncclComm* comm) { plan->ceCollArgs->sizes = (task->func == ncclFuncAlltoAllv) ? task->sizes : nullptr; if (comm->rank == 0) { - if (!ncclDevrIsOneLsaTeam(comm)) { + // Same predicate ncclLaunchCeColl dispatches on, so the marker cannot + // claim a path the launch did not take. + if (ncclCeHierDispatch(comm)) { INFO(NCCL_TUNING, "%s " RCCL_CE_HIER_SELECTED_TAG ": %ld Bytes -> RMA proxy + CE", ncclFuncToString(task->func), task->count * ncclTypeSize(task->datatype)); } else { diff --git a/projects/rccl/src/include/ce_coll.h b/projects/rccl/src/include/ce_coll.h index 7b40e6be87e2..51ae3d6b0d0b 100644 --- a/projects/rccl/src/include/ce_coll.h +++ b/projects/rccl/src/include/ce_coll.h @@ -157,6 +157,11 @@ bool ncclCeImplemented(ncclFunc_t coll, int /*ncclDevRedOp_t*/ red, ncclDataType bool ncclHierCeAvailable(struct ncclComm* comm, ncclFunc_t coll, int /*ncclDevRedOp_t*/ red, ncclDataType_t ty, ncclSymRegType_t winRegType); +// True when an admitted CE task must run on the hierarchical path. The launch +// site and the selection marker share this so they cannot disagree about which +// algorithm a task got. +bool ncclCeHierDispatch(struct ncclComm* comm); + ncclResult_t ncclCeInit(struct ncclComm* comm); ncclResult_t ncclCeFinalize(struct ncclComm* comm); diff --git a/projects/rccl/test/DevRuntimeTests.cpp b/projects/rccl/test/DevRuntimeTests.cpp index 2f6b0923a303..0bc64b3890ce 100644 --- a/projects/rccl/test/DevRuntimeTests.cpp +++ b/projects/rccl/test/DevRuntimeTests.cpp @@ -119,3 +119,31 @@ TEST_F(DevrFinalizeDrainTest, FinalizeDrainsLeftoverMemory) { EXPECT_EQ(comm->devrState.memHead, nullptr); } + +// --------------------------------------------------------------------------- +// ncclDevrGetWinOffset: the window's position inside its backing allocation. +// +// The RMA memory region is registered on the backing allocation's primaryAddr, +// so a put has to add this delta to the caller's in-window offset. Both +// ncclRmaProxyPutBuildOp callers depend on it, including the single-put path +// through ncclRmaProxyPutBuildDesc, and neither asserts the arithmetic. +TEST(DevrGetWinOffsetTest, OffsetIsWindowPositionWithinBackingMemory) { + EXPECT_EQ(ncclDevrGetWinOffset(nullptr), 0u); + + // Non-symmetric proxy windows carry no backing memory: the window is the + // allocation, so there is nothing to offset by. + ncclDevrWindow winNoMemory{}; + winNoMemory.bigOffset = 0x1400; + EXPECT_EQ(ncclDevrGetWinOffset(&winNoMemory), 0u); + + ncclDevrMemory mem{}; + mem.bigOffset = 0x1000; + ncclDevrWindow win{}; + win.memory = &mem; + win.bigOffset = 0x1400; + EXPECT_EQ(ncclDevrGetWinOffset(&win), 0x400u); + + // A window registered at the head of its allocation adds nothing. + win.bigOffset = mem.bigOffset; + EXPECT_EQ(ncclDevrGetWinOffset(&win), 0u); +} diff --git a/projects/rccl/test/ce/CeInternalMPITests.cpp b/projects/rccl/test/ce/CeInternalMPITests.cpp index 7c146823d874..58d7c745ce00 100644 --- a/projects/rccl/test/ce/CeInternalMPITests.cpp +++ b/projects/rccl/test/ce/CeInternalMPITests.cpp @@ -472,16 +472,31 @@ TEST_F(CeInternalMPITest, FreeBatchOpsParamsIsIdempotent) ncclCeBatchOpsParams params{}; ASSERT_EQ(ncclCeInitBatchOpsParams(¶ms, 4), ncclSuccess); ASSERT_NE(params.srcs, nullptr); +#ifdef CE_BATCH_ASYNC_SUPPORTED + ASSERT_NE(params.attrs, nullptr); + ASSERT_NE(params.attrIdxs, nullptr); +#endif ncclCeFreeBatchOpsParams(¶ms); EXPECT_EQ(params.srcs, nullptr); EXPECT_EQ(params.dsts, nullptr); EXPECT_EQ(params.sizes, nullptr); EXPECT_EQ(params.numOps, 0); + EXPECT_FALSE(params.intraBatchSync); +#ifdef CE_BATCH_ASYNC_SUPPORTED + // Also allocated by ncclCeInitBatchOpsParams, so the release has to clear them + // too or the second free in ncclHierCeAllGather passes dangling pointers. + EXPECT_EQ(params.attrs, nullptr); + EXPECT_EQ(params.attrIdxs, nullptr); + EXPECT_EQ(params.numAttrs, 0); +#endif // Must not double-free. ncclCeFreeBatchOpsParams(¶ms); EXPECT_EQ(params.srcs, nullptr); +#ifdef CE_BATCH_ASYNC_SUPPORTED + EXPECT_EQ(params.attrs, nullptr); +#endif } // =========================================================================== diff --git a/projects/rccl/test/ce/CeMPITests.cpp b/projects/rccl/test/ce/CeMPITests.cpp index dfd669fb437d..66e549fca877 100644 --- a/projects/rccl/test/ce/CeMPITests.cpp +++ b/projects/rccl/test/ce/CeMPITests.cpp @@ -17,6 +17,8 @@ // For RCCL_CE_HIER_SELECTED_TAG, so the assertion cannot drift from the emitter. #include "ce_coll.h" +// For the ncclHierCeAvailable prerequisite fields the scale-out cases gate on. +#include #include #include @@ -161,13 +163,22 @@ class CeMPITest : public MPITestBase } // MPI topology alone does not imply hierarchical CE is reachable: it also needs - // symmetric memory, which on a multi-node communicator requires a GIN backend. - // Distinguish "this machine cannot do it" (skip) from "prerequisites are present - // but the path was not taken" (fail), so a stack without GIN/RMA does not report - // a red for behaving correctly. - bool scaleOutPrerequisitesMet() const + // symmetric memory, which on a multi-node communicator requires a GIN backend, + // and an RMA context for the inter-node puts. Read those off the communicator + // rather than out of the log, because the absence of "Symmetric memory is not + // supported" clears only one of the ncclHierCeAvailable clauses: a stack with + // GIN but no RMA context would look ready here and then fail on a missing + // marker. Distinguish "this machine cannot do it" (skip) from "prerequisites + // are present but the path was not taken" (fail), so a stack without GIN/RMA + // does not report a red for behaving correctly. + bool scaleOutPrerequisitesMet() { - return readAllLogs().find("Symmetric memory is not supported") == std::string::npos; + auto* comm = static_cast(getActiveCommunicator()); + if(comm == nullptr) + return false; + + return comm->nNodes > 1 && comm->symmetricSupport && comm->hostRmaSupport && + comm->config.numRmaCtx > 0; } void assertHierarchicalCEPathTaken(int rank, const char* context) @@ -340,8 +351,8 @@ class CeMPI_AllGather : public CeMPITest ASSERT_EQ(ncclSuccess, createTestCommunicator()); if(requireScaleOut && !scaleOutPrerequisitesMet()) - GTEST_SKIP() << "Symmetric memory unavailable on this communicator; " - "hierarchical CE needs a GIN backend"; + GTEST_SKIP() << "Hierarchical CE prerequisites absent on this communicator; " + "needs symmetric memory (GIN backend) and an RMA context"; int rank{}, nRanks{}; ncclCommUserRank(getActiveCommunicator(), &rank); @@ -377,13 +388,16 @@ class CeMPI_AllGather : public CeMPITest } assertCEPathTaken(testId); - // CE batch numOps is per LSA destination, which is the node-local rank count - // on the scale-out path and the whole communicator otherwise. - const int batchPeers = requireScaleOut ? localRankCount() : nRanks; if(requireScaleOut) assertHierarchicalCEPathTaken(rank, testId); - // numOps = batchPeers (one copy per destination); chunkBytes = count * sizeof(float) - assertCEBatchPath(ceExpectIntraBatchSync(batchPeers, count * sizeof(float)), testId); + // The hierarchical path leaves intraBatchSync at its initialized false, so its + // batch always logs the without-sync line. Predicting from the thresholds + // instead would mispredict on a node with more than kCeIntraBatchSyncFreq + // ranks, because the 68 MiB chunk-boundary case clears the byte clause. + // Off that path numOps is one copy per destination, i.e. nRanks. + const bool expectIntraBatchSync = + requireScaleOut ? false : ceExpectIntraBatchSync(nRanks, count * sizeof(float)); + assertCEBatchPath(expectIntraBatchSync, testId); } }; @@ -426,8 +440,8 @@ class CeMPI_AlltoAll : public CeMPITest ASSERT_EQ(ncclSuccess, createTestCommunicator()); if(requireScaleOut && !scaleOutPrerequisitesMet()) - GTEST_SKIP() << "Symmetric memory unavailable on this communicator; " - "hierarchical CE needs a GIN backend"; + GTEST_SKIP() << "Hierarchical CE prerequisites absent on this communicator; " + "needs symmetric memory (GIN backend) and an RMA context"; int rank{}, nRanks{}; ncclCommUserRank(getActiveCommunicator(), &rank); @@ -470,13 +484,16 @@ class CeMPI_AlltoAll : public CeMPITest } assertCEPathTaken(testId); - // CE batch numOps is per LSA destination, which is the node-local rank count - // on the scale-out path and the whole communicator otherwise. - const int batchPeers = requireScaleOut ? localRankCount() : nRanks; if(requireScaleOut) assertHierarchicalCEPathTaken(rank, testId); - // numOps = batchPeers (one per destination); chunkBytes = count * sizeof(float) - assertCEBatchPath(ceExpectIntraBatchSync(batchPeers, count * sizeof(float)), testId); + // The hierarchical path leaves intraBatchSync at its initialized false, so its + // batch always logs the without-sync line. Predicting from the thresholds + // instead would mispredict on a node with more than kCeIntraBatchSyncFreq + // ranks, because the 68 MiB chunk-boundary case clears the byte clause. + // Off that path numOps is one copy per destination, i.e. nRanks. + const bool expectIntraBatchSync = + requireScaleOut ? false : ceExpectIntraBatchSync(nRanks, count * sizeof(float)); + assertCEBatchPath(expectIntraBatchSync, testId); } }; From 59b90eb960dd5841450253dd17d3ec75eb40c382 Mon Sep 17 00:00:00 2001 From: Thomas Huber Date: Tue, 15 Sep 2026 16:55:15 +0000 Subject: [PATCH 05/12] fix(rccl): Address CE scale-out review round 3 ncclHierCeAvailable compared the LSA team against comm->localRanks, this rank's own node count, while lsaSize is a comm-wide gcd. With unequal ranks per node the predicate differed across ranks, and since taskAppend now gates a symmetric-register group join on it, the small nodes would join and run ncclCeInit, whose window registration barriers over every rank, while the large nodes stayed out. Compare against maxLocalRanks so the answer is comm-uniform; homogeneous communicators, which is every validated configuration, are unaffected. The selection marker in scheduleCeCollTaskToPlan, which develop has now made the live CE plan builder, keyed on !ncclDevrIsOneLsaTeam while the launch dispatches on ncclHierCeDispatch; the tests use the marker as their only oracle for which path ran. It also claimed NVLS synchronization from nvlsSupport alone, where ncclMemOpSync requires !p2pCrossClique. Both now use the same predicates as the code they describe. ncclCeHierDispatch is renamed ncclHierCeDispatch to match the ncclHierCe* family. Tests: isScaleOutTopology admitted unequal node sizes that ncclHierCeAvailable declines, which fired neither skip and then failed on the absent marker. The AllGather and AllToAll payloads were constant within a slice, so both chunks of the 68 MiB chunk-boundary cases held identical bytes and a chunk written at the wrong offset was undetectable; the upper half of each slice now carries a coarse bias, coarse because a per-element term would leave the range float represents exactly. The chunk-boundary cases are capped at 16 ranks, since they size buffers per rank/peer and the host verification mirror would reach 17 GiB at 256 ranks. FreeBatchOpsParamsIsIdempotent now drives numOps, intraBatchSync and numAttrs off their initialized values before the release, which those assertions could not otherwise pin. Co-authored-by: Cursor --- projects/rccl/src/ce_coll.cc | 19 ++++-- projects/rccl/src/include/ce_coll.h | 2 +- projects/rccl/test/ce/CeInternalMPITests.cpp | 9 +++ projects/rccl/test/ce/CeMPITests.cpp | 72 +++++++++++++++----- 4 files changed, 78 insertions(+), 24 deletions(-) diff --git a/projects/rccl/src/ce_coll.cc b/projects/rccl/src/ce_coll.cc index bdd4e3a0919d..2978128bc820 100644 --- a/projects/rccl/src/ce_coll.cc +++ b/projects/rccl/src/ce_coll.cc @@ -1163,8 +1163,11 @@ bool ncclHierCeAvailable(struct ncclComm* comm, ncclFunc_t coll, int /*ncclDevRe TRACE(NCCL_TUNING, "Skipping hierarchical CE collective: LSA spans the comm; use CE path instead"); return false; } - // Intra-node CE scatter writes via LSA pointers - if (ncclTeamLsa(comm).nRanks < comm->localRanks) { + // Intra-node CE scatter writes via LSA pointers. Compare against the largest + // node rather than this rank's own node: lsaSize is a comm-wide gcd, so with + // unequal ranks per node comm->localRanks would clear this on the small nodes + // and not on the large ones, and the answer gates a group join in taskAppend. + if (ncclTeamLsa(comm).nRanks < comm->maxLocalRanks) { TRACE(NCCL_TUNING, "Skipping hierarchical CE collective: LSA team does not cover all local ranks"); return false; } @@ -1186,7 +1189,7 @@ bool ncclHierCeAvailable(struct ncclComm* comm, ncclFunc_t coll, int /*ncclDevRe return true; } -bool ncclCeHierDispatch(struct ncclComm* comm) { +bool ncclHierCeDispatch(struct ncclComm* comm) { return comm->nNodes > 1 && !ncclDevrIsOneLsaTeam(comm); } @@ -2061,7 +2064,7 @@ ncclResult_t ncclLaunchCeColl(struct ncclComm* comm, struct ncclKernelPlan* plan // match the one ncclHierCeAvailable admitted the task under, otherwise a // single-node comm with a reduced LSA team (NCCL_LSA_TEAM_SIZE) would be // dispatched here without the RMA prerequisites having been checked. - if (ncclCeHierDispatch(comm)) { + if (ncclHierCeDispatch(comm)) { switch (args->func) { case ncclFuncAllGather: NCCLCHECKGOTO(ncclHierCeAllGather(comm, plan, stream), ret, fail); @@ -2169,11 +2172,15 @@ ncclResult_t scheduleCeCollTaskToPlan(struct ncclComm* comm, struct ncclKernelPl plan->ceCollArgs->sizes = (task->func == ncclFuncAlltoAllv) ? task->sizes : nullptr; if (comm->rank == 0) { - if (!ncclDevrIsOneLsaTeam(comm)) { + // Same predicate ncclLaunchCeColl dispatches on, so the marker cannot claim + // a path the launch did not take. + if (ncclHierCeDispatch(comm)) { INFO(NCCL_TUNING, "%s " RCCL_CE_HIER_SELECTED_TAG ": %ld Bytes -> RMA proxy + CE", ncclFuncToString(task->func), task->count * ncclTypeSize(task->datatype)); } else { - const char* nvlsSync = comm->nvlsSupport ? "; CE synchronization with NVLS" : ""; + // Matches the useMCSync predicate in ncclMemOpSync: multicast sync is not + // available across scale-up cliques. + const char* nvlsSync = comm->nvlsSupport && !comm->p2pCrossClique ? "; CE synchronization with NVLS" : ""; INFO(NCCL_TUNING, "%s [Copy Engine]: %ld Bytes -> cudaMemcpy%s", ncclFuncToString(task->func), task->count * ncclTypeSize(task->datatype), nvlsSync); } diff --git a/projects/rccl/src/include/ce_coll.h b/projects/rccl/src/include/ce_coll.h index c18069b24d6c..9048640118cc 100644 --- a/projects/rccl/src/include/ce_coll.h +++ b/projects/rccl/src/include/ce_coll.h @@ -160,7 +160,7 @@ bool ncclHierCeAvailable(struct ncclComm* comm, ncclFunc_t coll, int /*ncclDevRe // True when an admitted CE task must run on the hierarchical path. The launch // site and the selection marker share this so they cannot disagree about which // algorithm a task got. -bool ncclCeHierDispatch(struct ncclComm* comm); +bool ncclHierCeDispatch(struct ncclComm* comm); ncclResult_t ncclCeInit(struct ncclComm* comm); diff --git a/projects/rccl/test/ce/CeInternalMPITests.cpp b/projects/rccl/test/ce/CeInternalMPITests.cpp index 58d7c745ce00..69376deb2cda 100644 --- a/projects/rccl/test/ce/CeInternalMPITests.cpp +++ b/projects/rccl/test/ce/CeInternalMPITests.cpp @@ -477,6 +477,15 @@ TEST_F(CeInternalMPITest, FreeBatchOpsParamsIsIdempotent) ASSERT_NE(params.attrIdxs, nullptr); #endif + // ncclCeInitBatchOpsParams already leaves the counters and the sync flag at + // 0/false, so drive them off those values first. Otherwise the assertions + // below would pass without the release ever clearing anything. + params.numOps = 4; + params.intraBatchSync = true; +#ifdef CE_BATCH_ASYNC_SUPPORTED + params.numAttrs = 4; +#endif + ncclCeFreeBatchOpsParams(¶ms); EXPECT_EQ(params.srcs, nullptr); EXPECT_EQ(params.dsts, nullptr); diff --git a/projects/rccl/test/ce/CeMPITests.cpp b/projects/rccl/test/ce/CeMPITests.cpp index dbaa82413460..e8f91efdbccc 100644 --- a/projects/rccl/test/ce/CeMPITests.cpp +++ b/projects/rccl/test/ce/CeMPITests.cpp @@ -48,6 +48,16 @@ constexpr int kScaleOutIters = 3; // repeated RMA sequence reuse in hi // 68 MiB per rank/peer crosses the 64 MiB hierarchical chunk size, exercising the // multi-chunk plan and the per-chunk scratch release. constexpr size_t kChunkBoundaryCount = (68ull * 1024 * 1024) / sizeof(float); +// The count above is per rank/peer, so the buffers and their host verification +// mirrors grow with world size: 1.1 GiB each at 16 ranks, 17 GiB at 256. Cap the +// two chunk-boundary cases at the two-node shape they are meant to cover; past +// it the host mirror alone would throw out of its vector constructor. +constexpr int kChunkBoundaryMaxRanks = 16; +// Coarse offset applied to the upper half of every sent slice, so a chunk landing +// at the wrong offset changes the received bytes. It has to be coarse: at this +// element count a per-element term would exceed the range float represents +// exactly, while the biased values stay well inside it. +constexpr float kUpperHalfBias = 1048576.0f; } // namespace CeMPITestConstants using namespace CeMPITestConstants; @@ -159,7 +169,11 @@ class CeMPITest : public MPITestBase MPI_Allreduce(&localSize, &minLocalSize, 1, MPI_INT, MPI_MIN, MPI_COMM_WORLD); MPI_Allreduce(&localSize, &maxLocalSize, 1, MPI_INT, MPI_MAX, MPI_COMM_WORLD); - return worldSize > maxLocalSize && minLocalSize >= 2; + // Equal ranks per node is a requirement, not a convenience: lsaSize is a + // comm-wide gcd, so on unequal nodes it cannot cover the largest node and + // ncclHierCeAvailable declines. Admitting such a job here would fire + // neither skip and then fail on the missing marker. + return worldSize > maxLocalSize && minLocalSize >= 2 && minLocalSize == maxLocalSize; } // MPI topology alone does not imply hierarchical CE is reachable: it also needs @@ -341,10 +355,14 @@ class CeMPI_AllGather : public CeMPITest { protected: void runAllGather(int minRanks, size_t count, const char* testId, - bool requireScaleOut = false, int scaleOutIters = kScaleOutIters) + bool requireScaleOut = false, int scaleOutIters = kScaleOutIters, + int maxRanks = MPITestConstants::kNoProcessLimit) { - if(!validateTestPrerequisites(minRanks)) - GTEST_SKIP() << "Need >= " << minRanks << " MPI ranks"; + if(!validateTestPrerequisites(minRanks, maxRanks)) + GTEST_SKIP() << "Need >= " << minRanks << " MPI ranks" + << (maxRanks != MPITestConstants::kNoProcessLimit + ? " and <= " + std::to_string(maxRanks) + : std::string()); if(requireScaleOut && !isScaleOutTopology()) GTEST_SKIP() << "Need at least 2 nodes and 2 MPI ranks per node"; @@ -367,11 +385,19 @@ class CeMPI_AllGather : public CeMPITest for(int iter = 0; iter < iterations; ++iter) { const int epoch = iter * nRanks; + // The upper half of the slice carries kUpperHalfBias so the payload + // varies within one rank's contribution. Without it every chunk of a + // slice holds identical bytes, and a chunk written at the wrong + // offset would leave the result bit-identical. + const size_t halfCount = count / 2; ASSERT_EQ( hipSuccess, initializeBufferWithPattern( sendSym.ptr, count, - [rank, epoch](size_t) { return static_cast(epoch + rank + 1); })); + [rank, epoch, halfCount](size_t i) { + return static_cast(epoch + rank + 1) + + (i >= halfCount ? kUpperHalfBias : 0.0f); + })); ASSERT_EQ(ncclSuccess, ncclAllGather(sendSym.ptr, recvSym.ptr, count, ncclFloat32, @@ -380,8 +406,9 @@ class CeMPI_AllGather : public CeMPITest ASSERT_TRUE(verifyBufferData( recvSym.ptr, count * static_cast(nRanks), - [count, epoch](size_t i) { - return static_cast(epoch + i / count + 1); + [count, epoch, halfCount](size_t i) { + return static_cast(epoch + i / count + 1) + + (i % count >= halfCount ? kUpperHalfBias : 0.0f); })) << "Rank " << rank << ": AllGather data verification failed at iteration " << iter; @@ -418,7 +445,8 @@ TEST_F(CeMPI_AllGather, MultiNodeHierarchical) TEST_F(CeMPI_AllGather, MultiNodeHierarchicalChunkBoundary) { runAllGather(kMinRanks4, kChunkBoundaryCount, - "CeMPI_AllGather/MultiNodeHierarchicalChunkBoundary", true, 1); + "CeMPI_AllGather/MultiNodeHierarchicalChunkBoundary", true, 1, + kChunkBoundaryMaxRanks); } // =========================================================================== @@ -430,10 +458,14 @@ class CeMPI_AlltoAll : public CeMPITest protected: // count is per-rank-per-destination; total send/recv buffer = count * nRanks. void runAlltoAll(int minRanks, size_t count, const char* testId, - bool requireScaleOut = false, int scaleOutIters = kScaleOutIters) + bool requireScaleOut = false, int scaleOutIters = kScaleOutIters, + int maxRanks = MPITestConstants::kNoProcessLimit) { - if(!validateTestPrerequisites(minRanks)) - GTEST_SKIP() << "Need >= " << minRanks << " MPI ranks"; + if(!validateTestPrerequisites(minRanks, maxRanks)) + GTEST_SKIP() << "Need >= " << minRanks << " MPI ranks" + << (maxRanks != MPITestConstants::kNoProcessLimit + ? " and <= " + std::to_string(maxRanks) + : std::string()); if(requireScaleOut && !isScaleOutTopology()) GTEST_SKIP() << "Need at least 2 nodes and 2 MPI ranks per node"; @@ -464,13 +496,17 @@ class CeMPI_AlltoAll : public CeMPITest const int epoch = iter * nRanks; // Value depends on both endpoints: sender s writes epoch + s*nRanks + j + 1 // into the slice bound for rank j. A wrong source slice therefore changes the - // received bytes, which a sender-only pattern could not detect. + // received bytes, which a sender-only pattern could not detect. The upper + // half of each per-destination slice additionally carries kUpperHalfBias, so + // a chunk written at the wrong offset within a slice is detectable too. + const size_t halfCount = count / 2; ASSERT_EQ( hipSuccess, initializeBufferWithPattern( sendSym.ptr, totalElem, - [rank, nRanks, count, epoch](size_t i) { - return static_cast(epoch + rank * nRanks + i / count + 1); + [rank, nRanks, count, epoch, halfCount](size_t i) { + return static_cast(epoch + rank * nRanks + i / count + 1) + + (i % count >= halfCount ? kUpperHalfBias : 0.0f); })); ASSERT_EQ(ncclSuccess, @@ -481,8 +517,9 @@ class CeMPI_AlltoAll : public CeMPITest // Slice s of the result is what rank s sent to us, i.e. epoch + s*nRanks + rank + 1. ASSERT_TRUE(verifyBufferData( recvSym.ptr, totalElem, - [count, nRanks, rank, epoch](size_t i) { - return static_cast(epoch + (i / count) * nRanks + rank + 1); + [count, nRanks, rank, epoch, halfCount](size_t i) { + return static_cast(epoch + (i / count) * nRanks + rank + 1) + + (i % count >= halfCount ? kUpperHalfBias : 0.0f); })) << "Rank " << rank << ": AlltoAll data verification failed at iteration " << iter; @@ -519,7 +556,8 @@ TEST_F(CeMPI_AlltoAll, MultiNodeHierarchical) TEST_F(CeMPI_AlltoAll, MultiNodeHierarchicalChunkBoundary) { runAlltoAll(kMinRanks4, kChunkBoundaryCount, - "CeMPI_AlltoAll/MultiNodeHierarchicalChunkBoundary", true, 1); + "CeMPI_AlltoAll/MultiNodeHierarchicalChunkBoundary", true, 1, + kChunkBoundaryMaxRanks); } // =========================================================================== From dacfc2ce1d67b61fc9c9194f7e660a4f44b4142b Mon Sep 17 00:00:00 2001 From: Thomas Huber Date: Tue, 15 Sep 2026 23:09:46 +0000 Subject: [PATCH 06/12] fix(rccl): address CE scale-out review feedback Co-authored-by: Cursor --- projects/rccl/CHANGELOG.md | 2 +- projects/rccl/src/ce_coll.cc | 63 +++++++++----------- projects/rccl/src/enqueue.cc | 9 ++- projects/rccl/test/ce/CeInternalMPITests.cpp | 10 ++-- projects/rccl/test/ce/CeMPITests.cpp | 3 +- projects/rccl/test/gin_teardown_test.cpp | 8 +++ 6 files changed, 52 insertions(+), 43 deletions(-) diff --git a/projects/rccl/CHANGELOG.md b/projects/rccl/CHANGELOG.md index cbb1293a2b3e..34ec6df20a81 100644 --- a/projects/rccl/CHANGELOG.md +++ b/projects/rccl/CHANGELOG.md @@ -13,7 +13,7 @@ Full documentation for RCCL is available at [https://rccl.readthedocs.io](https: * Added `install.sh --all_unrolls` (`-DBUILD_ALL_UNROLLS=ON`) to generate every unroll factor (1, 2, 4, 8, 16, 32) for the targeted GPU architecture(s), for measuring unroll factors that the default per-arch matrix does not build. The flag also drops the per-architecture pin, so such a build accepts every `RCCL_UNROLL_FACTOR` value on any targeted architecture. * Added the strict `--enable-full-coverage` install flag for unified host + device LLVM source-based code coverage. It requires `--debug`, the device linker, and ROCm 7.15 or newer. The test runner uses the CMake-level `ENABLE_FULL_COVERAGE=AUTO` mode, which falls back to host-only coverage when device instrumentation is unavailable. * Enabled the Copy-Engine profiler path (`ncclProfiler_v6`): Copy-Engine events are emitted for CE collectives, and the example profiler plugin reports them. -* Enabled hierarchical Copy Engine `ncclAllGather` and `ncclAlltoAll` on multi-node communicators: the intra-node phase runs over SDMA/CE across the LSA team and the inter-node phase over one-sided RMA through the CPU (GIN) proxy. Selected when `NCCL_CTA_POLICY=2` with symmetric send and receive windows, `NCCL_LOCAL_REGISTER=0`, `NCCL_CUMEM_ENABLE=1`, `NCCL_NUM_RMA_CTX` greater than zero, and a GIN backend available. Validated for correctness and path selection from 1 to 32 nodes. +* Enabled hierarchical Copy Engine `ncclAllGather` and `ncclAlltoAll` on multi-node communicators: the intra-node phase runs over SDMA/CE across the LSA team and the inter-node phase over one-sided RMA through the CPU (GIN) proxy. Selected when every node has the same number of ranks, `NCCL_CTA_POLICY=2` with symmetric send and receive windows, `NCCL_LOCAL_REGISTER=0`, `NCCL_CUMEM_ENABLE=1`, `NCCL_NUM_RMA_CTX` greater than zero, and a GIN backend available. Validated for correctness and path selection from 1 to 32 nodes. ### Changed * Adapted the device-initiated GIN backends (Anvil SDMA and rocSHMEM GDA) to the NCCL 2.30.7 GIN API v14: added the new `getGinProperties` host op, dropped the data-path ops (`iput`/`iputSignal`/`iget`/`iflush`/`test`) that moved out of GIN under the GIN/RMA split, switched `createContext` to `ncclGinConfig_v14_t`, updated the device dispatch signatures, and matched the GIN type renumbering (`ROCSHMEM_GDA` and `ANVIL_SDMA` shifted after the new `GIN_GPI` type). The plugins now use the generic (unversioned) `ncclGin_t` / `ncclGinConfig_t` / `ncclGinProperties_t` typedefs so future ABI bumps do not require touching call sites. diff --git a/projects/rccl/src/ce_coll.cc b/projects/rccl/src/ce_coll.cc index 2978128bc820..ffccd275b192 100644 --- a/projects/rccl/src/ce_coll.cc +++ b/projects/rccl/src/ce_coll.cc @@ -111,9 +111,11 @@ ncclResult_t ncclCeInit(struct ncclComm* comm) { size_t ceARTmpBufSize = alignUp(NUM_SLOTS * comm->nRanks * maxChunkBytes, 16); int i = 0; int targetStreams = 0; + bool isOneLsaTeam = false; uint32_t graphSyncValue = GRAPH_SYNC_VALUE; // Symmetric memory runtime must be initialized before any window registration. NCCLCHECKGOTO(ncclDevrInitOnce(comm), ret, fail); + isOneLsaTeam = ncclDevrIsOneLsaTeam(comm); // Sync window holds one ready and one complete slot per LSA-local rank. ceDevBaseSize = alignUp(comm->devrState.lsaSize * sizeof(uint32_t), 16) * 2; @@ -124,15 +126,19 @@ ncclResult_t ncclCeInit(struct ncclComm* comm) { ret, fail); CUDACHECKGOTO(cudaStreamCreateWithFlags(&comm->ceColl.scatterStream, cudaStreamNonBlocking), ret, fail); CUDACHECKGOTO(cudaEventCreateWithFlags(&comm->ceColl.synceEvent, cudaEventDisableTiming), ret, fail); - // Signal buffer: [NUM_SLOTS][nRanks], indexed slot*nRanks + r. Symmetric window - // so peers can ring each other's doorbells via LSA (same pattern as ceARTmpWin). - NCCLCHECKGOTO(ncclMemAlloc((void**)&signalBuf, sigBufferSize), ret, fail); - NCCLCHECKGOTO(ncclDevrWindowRegisterInGroup(comm, signalBuf, sigBufferSize, NCCL_WIN_COLL_SYMMETRIC, &sigWinDev), ret, - fail); - NCCLCHECKGOTO(ncclShadowPoolToHost(&comm->devrState.shadows, sigWinDev, &sigWinDevHost), ret, fail); - comm->ceColl.signalWin = (struct ncclDevrWindow*)sigWinDevHost->winHost; - comm->ceColl.signalBuffer = (uint32_t*)comm->ceColl.signalWin->userPtr; - CUDACHECKGOTO(hipMemset(comm->ceColl.signalBuffer, 0, sigBufferSize), ret, fail); + // The signal buffer is consumed only by CE AllReduce, which is LSA-local. + // Communicators initialized solely for hierarchical CE do not need it. + if (isOneLsaTeam) { + // [NUM_SLOTS][nRanks], indexed slot*nRanks + r. Symmetric so peers can + // ring each other's doorbells via LSA (same pattern as ceARTmpWin). + NCCLCHECKGOTO(ncclMemAlloc((void**)&signalBuf, sigBufferSize), ret, fail); + NCCLCHECKGOTO( + ncclDevrWindowRegisterInGroup(comm, signalBuf, sigBufferSize, NCCL_WIN_COLL_SYMMETRIC, &sigWinDev), ret, fail); + NCCLCHECKGOTO(ncclShadowPoolToHost(&comm->devrState.shadows, sigWinDev, &sigWinDevHost), ret, fail); + comm->ceColl.signalWin = (struct ncclDevrWindow*)sigWinDevHost->winHost; + comm->ceColl.signalBuffer = (uint32_t*)comm->ceColl.signalWin->userPtr; + CUDACHECKGOTO(hipMemset(comm->ceColl.signalBuffer, 0, sigBufferSize), ret, fail); + } // ceSync window (ready/complete flag arrays). NCCLCHECKGOTO(ncclMemAlloc((void**)&ceDevBase, ceDevBaseSize), ret, fail); @@ -176,7 +182,7 @@ ncclResult_t ncclCeInit(struct ncclComm* comm) { // [slot 0: nRanks chunks][slot 1: nRanks chunks]. // CE AllReduce is LSA-local only, so skip the staging buffer on communicators // that reach ncclCeInit solely for the hierarchical path. - if (rcclParamCeAllReduce() && ncclDevrIsOneLsaTeam(comm)) { + if (rcclParamCeAllReduce() && isOneLsaTeam) { NCCLCHECKGOTO(ncclMemAlloc((void**)&ceARTmpBuf, ceARTmpBufSize), ret, fail_ar); NCCLCHECKGOTO(ncclDevrWindowRegisterInGroup(comm, ceARTmpBuf, ceARTmpBufSize, NCCL_WIN_COLL_SYMMETRIC, &arWinDev), ret, fail_ar); @@ -296,8 +302,8 @@ ncclResult_t ncclCeFinalize(struct ncclComm* comm) { bool ncclCeImplemented(ncclFunc_t coll, int /*ncclDevRedOp_t*/ red, ncclDataType_t ty); -bool ncclCeAvailable(struct ncclComm* comm, ncclFunc_t coll, int /*ncclDevRedOp_t*/ red, ncclDataType_t ty, - ncclSymRegType_t winRegType) { +bool ncclCeScratchAvailable(struct ncclComm* comm, ncclFunc_t coll, int /*ncclDevRedOp_t*/ red, ncclDataType_t ty, + ncclSymRegType_t winRegType) { if (!ncclCeImplemented(coll, red, ty)) { TRACE(NCCL_TUNING, "Skipping CE collective: not implemented"); return false; @@ -317,6 +323,14 @@ bool ncclCeAvailable(struct ncclComm* comm, ncclFunc_t coll, int /*ncclDevRedOp_ TRACE(NCCL_TUNING, "Skipping CE collective: symmetric support is not enabled"); return false; } + return true; +} + +bool ncclCeAvailable(struct ncclComm* comm, ncclFunc_t coll, int /*ncclDevRedOp_t*/ red, ncclDataType_t ty, + ncclSymRegType_t winRegType) { + if (!ncclCeScratchAvailable(comm, coll, red, ty, winRegType)) { + return false; + } if (winRegType != ncclSymSendRegRecvReg && winRegType != ncclSymSendNonregRecvReg) { TRACE(NCCL_TUNING, "Skipping CE collective: window registration type %d is not supported", winRegType); return false; @@ -341,27 +355,6 @@ bool ncclCeAlltoAllEligible(struct ncclComm* comm, ncclDataType_t datatype, nccl return ncclCeAvailable(comm, ncclFuncAlltoAll, ncclDevSum, datatype, winRegType); } -bool ncclCeScratchAvailable(struct ncclComm* comm, ncclFunc_t coll, int /*ncclDevRedOp_t*/ red, ncclDataType_t ty, - ncclSymRegType_t winRegType) { - if (!ncclCeImplemented(coll, red, ty)) { - TRACE(NCCL_TUNING, "Skipping CE collective: not implemented"); - return false; - } - if (comm->nNodes > 1) { - TRACE(NCCL_TUNING, "Skipping CE collective: comm is not a single node"); - return false; - } - if (!ncclDevrIsOneLsaTeam(comm)) { - TRACE(NCCL_TUNING, "Skipping CE collective: LSA team does not cover the comm"); - return false; - } - if (!comm->symmetricSupport) { - TRACE(NCCL_TUNING, "Skipping CE collective: symmetric support is not enabled"); - return false; - } - return true; -} - bool ncclCeImplemented(ncclFunc_t coll, int /*ncclDevRedOp_t*/ red, ncclDataType_t ty) { int driverVersion; if (ncclCudaDriverVersion(&driverVersion) != ncclSuccess) return false; @@ -582,7 +575,9 @@ ncclResult_t ncclMemOpSync(struct ncclComm* comm, cudaStream_t stream, struct nc // wait for the wait ops, so a reset can clobber a peer's flag mid-barrier // and hang graph replay. Issuing them as a separate, stream-ordered batch // forces reset-after-wait. - CUCHECKGOTO(hipStreamBatchMemOp(stream, opIdx, batchParams, 0), ret, fail); + if (opIdx) { + CUCHECKGOTO(hipStreamBatchMemOp(stream, opIdx, batchParams, 0), ret, fail); + } // For graph capture, reset our flag array to 0 in a separate batch so the // fixed-value barrier can be replayed. diff --git a/projects/rccl/src/enqueue.cc b/projects/rccl/src/enqueue.cc index 775a429ee3db..34f7ff282b5b 100644 --- a/projects/rccl/src/enqueue.cc +++ b/projects/rccl/src/enqueue.cc @@ -3901,9 +3901,14 @@ static ncclResult_t taskAppend(struct ncclComm* comm, struct ncclInfo* info) { // ceCollTaskAppend path when user buffers are not symmetrically registered. // Without this trigger, CE AllReduce-only workloads would never initialize // the CE runtime (ceARTmpBuf stays NULL). Hierarchical CE is multi-node by - // construction, so it needs the same trigger despite nNodes > 1. + // construction, so it needs the same trigger despite nNodes > 1. Keep + // that arm behind the policy bit which actually selects hierarchical CE; + // otherwise a default-policy collective initializes and registers CE + // resources only to take the kernel path. if (!ceCapturing && ncclCeImplemented(info->coll, info->op, info->datatype) && comm->symmetricSupport && - (comm->nNodes == 1 || hierCeAvailable) && comm->ceColl.baseUCSymReadyPtr == NULL && + (comm->nNodes == 1 || + (hierCeAvailable && (comm->config.CTAPolicy & NCCL_CTA_POLICY_ZERO))) && + comm->ceColl.baseUCSymReadyPtr == NULL && ncclIntruQueueEmpty(&comm->ceInitTaskQueue)) { struct ncclCeInitTask* ceTask; NCCLCHECK(ncclCalloc(&ceTask, 1)); diff --git a/projects/rccl/test/ce/CeInternalMPITests.cpp b/projects/rccl/test/ce/CeInternalMPITests.cpp index 69376deb2cda..905e9872c8e6 100644 --- a/projects/rccl/test/ce/CeInternalMPITests.cpp +++ b/projects/rccl/test/ce/CeInternalMPITests.cpp @@ -487,16 +487,16 @@ TEST_F(CeInternalMPITest, FreeBatchOpsParamsIsIdempotent) #endif ncclCeFreeBatchOpsParams(¶ms); - EXPECT_EQ(params.srcs, nullptr); - EXPECT_EQ(params.dsts, nullptr); - EXPECT_EQ(params.sizes, nullptr); + ASSERT_EQ(params.srcs, nullptr); + ASSERT_EQ(params.dsts, nullptr); + ASSERT_EQ(params.sizes, nullptr); EXPECT_EQ(params.numOps, 0); EXPECT_FALSE(params.intraBatchSync); #ifdef CE_BATCH_ASYNC_SUPPORTED // Also allocated by ncclCeInitBatchOpsParams, so the release has to clear them // too or the second free in ncclHierCeAllGather passes dangling pointers. - EXPECT_EQ(params.attrs, nullptr); - EXPECT_EQ(params.attrIdxs, nullptr); + ASSERT_EQ(params.attrs, nullptr); + ASSERT_EQ(params.attrIdxs, nullptr); EXPECT_EQ(params.numAttrs, 0); #endif diff --git a/projects/rccl/test/ce/CeMPITests.cpp b/projects/rccl/test/ce/CeMPITests.cpp index e8f91efdbccc..c8c0c58af233 100644 --- a/projects/rccl/test/ce/CeMPITests.cpp +++ b/projects/rccl/test/ce/CeMPITests.cpp @@ -191,7 +191,8 @@ class CeMPITest : public MPITestBase if(comm == nullptr) return false; - return comm->nNodes > 1 && comm->symmetricSupport && comm->hostRmaSupport && + return comm->nNodes > 1 && !ncclDevrIsOneLsaTeam(comm) && + comm->symmetricSupport && comm->hostRmaSupport && comm->config.numRmaCtx > 0; } diff --git a/projects/rccl/test/gin_teardown_test.cpp b/projects/rccl/test/gin_teardown_test.cpp index 1b129f98bf17..381969e99c0f 100644 --- a/projects/rccl/test/gin_teardown_test.cpp +++ b/projects/rccl/test/gin_teardown_test.cpp @@ -267,15 +267,23 @@ TEST(DevrGetWinOffsetTest, OffsetIsWindowPositionWithinBackingMemory) { ncclDevrWindow winNoMemory{}; winNoMemory.bigOffset = 0x1400; EXPECT_EQ(ncclDevrGetWinOffset(&winNoMemory), 0u); + winNoMemory.rmaHostWins[0] = reinterpret_cast(0x1234); + EXPECT_EQ(ncclDevrGetRmaWin(&winNoMemory, 0), winNoMemory.rmaHostWins[0]); ncclDevrMemory mem{}; mem.bigOffset = 0x1000; + mem.rmaHostWins[0] = reinterpret_cast(0x5678); ncclDevrWindow win{}; win.memory = &mem; win.bigOffset = 0x1400; EXPECT_EQ(ncclDevrGetWinOffset(&win), 0x400u); + EXPECT_EQ(ncclDevrGetRmaWin(&win, 0), mem.rmaHostWins[0]); // A window registered at the head of its allocation adds nothing. win.bigOffset = mem.bigOffset; EXPECT_EQ(ncclDevrGetWinOffset(&win), 0u); + + EXPECT_EQ(ncclDevrGetRmaWin(nullptr, 0), nullptr); + EXPECT_EQ(ncclDevrGetRmaWin(&win, -1), nullptr); + EXPECT_EQ(ncclDevrGetRmaWin(&win, NCCL_GIN_MAX_CONNECTIONS), nullptr); } From 96cc65d05038692ec37ba3a9d8c76df93d5a2135 Mon Sep 17 00:00:00 2001 From: Thomas Huber Date: Wed, 16 Sep 2026 06:05:18 +0000 Subject: [PATCH 07/12] fix(rccl): report hierarchical CE AllGather as CE for -A 1 rcclSelectAllGather only tested ncclCeAvailable, so multi-node CE AllGather was labeled Hier/Direct/RING even when taskAppend took hierarchical CE. --- projects/rccl/src/include/rccl_common.h | 7 ++++--- projects/rccl/src/rccl_wrap.cc | 9 +++++++-- projects/rccl/test/ce/CeMPITests.cpp | 14 ++++++++++++++ 3 files changed, 25 insertions(+), 5 deletions(-) diff --git a/projects/rccl/src/include/rccl_common.h b/projects/rccl/src/include/rccl_common.h index 65bdc61a7b9e..895e74be5bd2 100644 --- a/projects/rccl/src/include/rccl_common.h +++ b/projects/rccl/src/include/rccl_common.h @@ -165,9 +165,10 @@ NCCL_API(ncclResult_t, rcclGetCollImplInfo, struct ncclComm* comm, ncclFunc_t co ncclResult_t rcclSelectAllReduce(struct ncclComm* comm, const void* sendbuff, void* recvbuff, size_t count, ncclDataType_t datatype, ncclRedOp_t op, cudaStream_t stream, bool query, bool graphCapturingHint, struct rcclCollDecision* decision); -// Single source of truth for AllGather selection: DDA -> hierarchical -> CE -> -// direct -> ring. query=true fills protocol/nMaxChannels for reporting. CE dispatch -// lives in taskAppend(), so live returns RCCL_CE_REGISTERED but enqueues normally. +// Single source of truth for AllGather selection: DDA -> hierarchical kernel -> +// CE (single-node or hierarchical CE) -> direct -> ring. query=true fills +// protocol/nMaxChannels for reporting. CE dispatch lives in taskAppend(), so live +// returns RCCL_CE_REGISTERED but enqueues normally. // graphCapturingHint (query only) suppresses the graph-unsafe CE branch under capture. ncclResult_t rcclSelectAllGather(struct ncclComm* comm, const void* sendbuff, void* recvbuff, size_t sendcount, ncclDataType_t datatype, bool query, bool graphCapturingHint, diff --git a/projects/rccl/src/rccl_wrap.cc b/projects/rccl/src/rccl_wrap.cc index 9441546ead23..4eaf5ea532d4 100644 --- a/projects/rccl/src/rccl_wrap.cc +++ b/projects/rccl/src/rccl_wrap.cc @@ -1197,10 +1197,15 @@ ncclResult_t rcclSelectAllGather(struct ncclComm* comm, const void* sendbuff, vo decision->algo = RCCL_CE_REGISTERED; return ncclSuccess; } - // Branch #3: CE via registered symmetric windows. + // Branch #3: CE via registered symmetric windows. Same gates as taskAppend: + // single-node ncclCeAvailable or multi-node ncclHierCeAvailable. Without the + // hier arm, rccl-tests -A 1 labels a live hierarchical CE AllGather as + // Hier/Direct/RING because ncclCeAvailable requires nNodes == 1. const bool ceAvailable = !ceCapturing && ncclCeAvailable(comm, ncclFuncAllGather, (int)ncclSum, datatype, winRegType); - if (ceAvailable && !hasSysmemSegment && (comm->config.CTAPolicy & NCCL_CTA_POLICY_ZERO)) { + const bool hierCeAvailable = + !ceCapturing && ncclHierCeAvailable(comm, ncclFuncAllGather, (int)ncclSum, datatype, winRegType); + if ((ceAvailable || hierCeAvailable) && !hasSysmemSegment && (comm->config.CTAPolicy & NCCL_CTA_POLICY_ZERO)) { decision->algo = RCCL_CE_REGISTERED; return ncclSuccess; } diff --git a/projects/rccl/test/ce/CeMPITests.cpp b/projects/rccl/test/ce/CeMPITests.cpp index c8c0c58af233..0ad178a57be6 100644 --- a/projects/rccl/test/ce/CeMPITests.cpp +++ b/projects/rccl/test/ce/CeMPITests.cpp @@ -19,6 +19,8 @@ #include "ce_coll.h" // For the ncclHierCeAvailable prerequisite fields the scale-out cases gate on. #include +// rcclGetCollImplInfo / RCCL_CE_REGISTERED: -A 1 must name the CE path that ran. +#include "rccl_common.h" #include #include @@ -418,6 +420,18 @@ class CeMPI_AllGather : public CeMPITest assertCEPathTaken(testId); if(requireScaleOut) assertHierarchicalCEPathTaken(rank, testId); + if(isCeExpected()) + { + int algo = 0, proto = 0, nChannels = 0; + ASSERT_EQ(ncclSuccess, + rcclGetCollImplInfo(getActiveCommunicator(), ncclFuncAllGather, count, ncclFloat32, + ncclSum, sendSym.ptr, recvSym.ptr, /*graphCapturing=*/0, &algo, + &proto, &nChannels)) + << testId << ": rcclGetCollImplInfo failed"; + EXPECT_EQ(algo, static_cast(RCCL_CE_REGISTERED)) + << testId << ": -A 1 / rcclGetCollImplInfo must report CE when the CE path ran" + << " (algo=" << algo << ")"; + } // The hierarchical path leaves intraBatchSync at its initialized false, so its // batch always logs the without-sync line. Predicting from the thresholds // instead would mispredict on a node with more than kCeIntraBatchSyncFreq From 9df520cb14b8af8c0056190e5251bf7f0defc6be Mon Sep 17 00:00:00 2001 From: Thomas Huber Date: Wed, 16 Sep 2026 06:44:11 +0000 Subject: [PATCH 08/12] test(rccl): pin CE eligibility mocks and FreeBatchOps runner filter Mocks now skip computeLsaSize's null rankToNode walk, maxLocalRanks is set on the hier helper, and FreeBatchOpsParamsIsIdempotent is in CE_Internal_Init. --- projects/rccl/test/CeAlltoAllvEligibilityTests.cpp | 14 ++++++++++++++ .../rccl/test/common/CeAllReduceTestHelpers.hpp | 5 +++++ .../rccl/test/common/CeAlltoAllvTestHelpers.hpp | 6 ++++++ .../configs/mi300_mellanox_ib_func.json | 2 +- .../test_runner/configs/mi300x_mellanox_ib.json | 2 +- .../test_runner/configs/mi355x_thor2_roce.json | 2 +- .../configs/mi355x_thor2_roce_func.json | 2 +- .../test_runner/configs/mi455_ainic_roce.json | 2 +- .../test_runner/configs/mi455_ainic_roce_func.json | 2 +- 9 files changed, 31 insertions(+), 6 deletions(-) diff --git a/projects/rccl/test/CeAlltoAllvEligibilityTests.cpp b/projects/rccl/test/CeAlltoAllvEligibilityTests.cpp index 0d23c4d326e0..1b602f935056 100644 --- a/projects/rccl/test/CeAlltoAllvEligibilityTests.cpp +++ b/projects/rccl/test/CeAlltoAllvEligibilityTests.cpp @@ -350,6 +350,20 @@ TEST_F(CeAlltoAllEligibilityTest, MultiNodeHierAvailable_DoesNotYieldDda) /*capturing=*/false)); } +TEST_F(CeAlltoAllEligibilityTest, UnequalRanksPerNode_HierUnavailable) +{ + if (!isCeRuntimeDriverSupported()) + GTEST_SKIP() << "CE driver not in supported range"; + + mockComm_.configureHierEligible(/*nNodes=*/2, /*localRanks=*/4); + mockComm_.comm.maxLocalRanks = mockComm_.comm.devrState.lsaSize + 1; + EXPECT_FALSE(ncclHierCeAvailable(mockComm_.get(), + ncclFuncAlltoAll, + ncclDevSum, + ncclFloat32, + ncclSymSendRegRecvReg)); +} + TEST_F(CeAlltoAllEligibilityTest, NoSymmetricSupportRejected) { if (!isCeRuntimeDriverSupported()) diff --git a/projects/rccl/test/common/CeAllReduceTestHelpers.hpp b/projects/rccl/test/common/CeAllReduceTestHelpers.hpp index 4278abe87588..964fbdeda33e 100644 --- a/projects/rccl/test/common/CeAllReduceTestHelpers.hpp +++ b/projects/rccl/test/common/CeAllReduceTestHelpers.hpp @@ -47,6 +47,11 @@ struct CeAllReduceMockComm comm.rank = 0; comm.symmetricSupport = true; comm.config.CTAPolicy = NCCL_CTA_POLICY_ZERO; + // Skip computeLsaSize's rankToNode walk; ncclDevrIsOneLsaTeam is now on + // the CE eligibility path (same pattern as GinAlltoAllEligibilityTests). + comm.devrState.bigSize = 1; + comm.devrState.lsaSize = comm.nRanks; + comm.devrState.lsaSelf = comm.rank; } ncclComm* get() { return &comm; } diff --git a/projects/rccl/test/common/CeAlltoAllvTestHelpers.hpp b/projects/rccl/test/common/CeAlltoAllvTestHelpers.hpp index 21c06f1c0235..c959f2dabd1e 100644 --- a/projects/rccl/test/common/CeAlltoAllvTestHelpers.hpp +++ b/projects/rccl/test/common/CeAlltoAllvTestHelpers.hpp @@ -45,6 +45,11 @@ struct CeAlltoAllvMockComm comm.rank = 0; comm.symmetricSupport = true; comm.config.CTAPolicy = NCCL_CTA_POLICY_ZERO; + // Skip computeLsaSize's rankToNode walk; ncclDevrIsOneLsaTeam is now on + // the CE eligibility path (same pattern as GinAlltoAllEligibilityTests). + comm.devrState.bigSize = 1; + comm.devrState.lsaSize = comm.nRanks; + comm.devrState.lsaSelf = comm.rank; } // Multi-node local-only LSA so ncclHierCeAvailable can pass (bigSize skips CUDA init). @@ -59,6 +64,7 @@ struct CeAlltoAllvMockComm comm.hostRmaSupport = true; comm.config.CTAPolicy = NCCL_CTA_POLICY_ZERO; comm.config.numRmaCtx = 1; + comm.maxLocalRanks = localRanks; comm.devrState.bigSize = 1; comm.devrState.lsaSize = localRanks; comm.devrState.lsaSelf = 0; diff --git a/projects/rccl/tools/scripts/test_runner/configs/mi300_mellanox_ib_func.json b/projects/rccl/tools/scripts/test_runner/configs/mi300_mellanox_ib_func.json index a4d166670dd0..a108d24e166a 100644 --- a/projects/rccl/tools/scripts/test_runner/configs/mi300_mellanox_ib_func.json +++ b/projects/rccl/tools/scripts/test_runner/configs/mi300_mellanox_ib_func.json @@ -2473,7 +2473,7 @@ { "name": "CE_Internal_Init", "description": "ncclCeInit lifecycle: allocates sync window; ncclCeFinalize tears it down; double-fini is safe; independent state per comm", - "test_filter": "CeInternalMPITest.InitSetsWindowPointers:CeInternalMPITest.FiniNullsPointers:CeInternalMPITest.DoubleFiniIsSafe:CeInternalMPITest.IndependentCeStatePerComm:CeInternalMPITest.FiniAfterZeroCollectives" + "test_filter": "CeInternalMPITest.InitSetsWindowPointers:CeInternalMPITest.FiniNullsPointers:CeInternalMPITest.DoubleFiniIsSafe:CeInternalMPITest.IndependentCeStatePerComm:CeInternalMPITest.FiniAfterZeroCollectives:CeInternalMPITest.FreeBatchOpsParamsIsIdempotent" }, { "name": "CE_Internal_Sync", diff --git a/projects/rccl/tools/scripts/test_runner/configs/mi300x_mellanox_ib.json b/projects/rccl/tools/scripts/test_runner/configs/mi300x_mellanox_ib.json index bab9997c5963..174089d17b00 100644 --- a/projects/rccl/tools/scripts/test_runner/configs/mi300x_mellanox_ib.json +++ b/projects/rccl/tools/scripts/test_runner/configs/mi300x_mellanox_ib.json @@ -2815,7 +2815,7 @@ { "name": "CE_Internal_Init", "description": "ncclCeInit lifecycle: allocates sync window; ncclCeFinalize tears it down; double-fini is safe; independent state per comm", - "test_filter": "CeInternalMPITest.InitSetsWindowPointers:CeInternalMPITest.FiniNullsPointers:CeInternalMPITest.DoubleFiniIsSafe:CeInternalMPITest.IndependentCeStatePerComm:CeInternalMPITest.FiniAfterZeroCollectives" + "test_filter": "CeInternalMPITest.InitSetsWindowPointers:CeInternalMPITest.FiniNullsPointers:CeInternalMPITest.DoubleFiniIsSafe:CeInternalMPITest.IndependentCeStatePerComm:CeInternalMPITest.FiniAfterZeroCollectives:CeInternalMPITest.FreeBatchOpsParamsIsIdempotent" }, { "name": "CE_Internal_Sync", diff --git a/projects/rccl/tools/scripts/test_runner/configs/mi355x_thor2_roce.json b/projects/rccl/tools/scripts/test_runner/configs/mi355x_thor2_roce.json index c02f3cfbce7a..0c9f6f3ecb61 100644 --- a/projects/rccl/tools/scripts/test_runner/configs/mi355x_thor2_roce.json +++ b/projects/rccl/tools/scripts/test_runner/configs/mi355x_thor2_roce.json @@ -3139,7 +3139,7 @@ { "name": "CE_Internal_Init", "description": "ncclCeInit lifecycle: allocates sync window; ncclCeFinalize tears it down; double-fini is safe; independent state per comm", - "test_filter": "CeInternalMPITest.InitSetsWindowPointers:CeInternalMPITest.FiniNullsPointers:CeInternalMPITest.DoubleFiniIsSafe:CeInternalMPITest.IndependentCeStatePerComm:CeInternalMPITest.FiniAfterZeroCollectives" + "test_filter": "CeInternalMPITest.InitSetsWindowPointers:CeInternalMPITest.FiniNullsPointers:CeInternalMPITest.DoubleFiniIsSafe:CeInternalMPITest.IndependentCeStatePerComm:CeInternalMPITest.FiniAfterZeroCollectives:CeInternalMPITest.FreeBatchOpsParamsIsIdempotent" }, { "name": "CE_Internal_Sync", diff --git a/projects/rccl/tools/scripts/test_runner/configs/mi355x_thor2_roce_func.json b/projects/rccl/tools/scripts/test_runner/configs/mi355x_thor2_roce_func.json index 526bb937ca9f..1891da1dab38 100644 --- a/projects/rccl/tools/scripts/test_runner/configs/mi355x_thor2_roce_func.json +++ b/projects/rccl/tools/scripts/test_runner/configs/mi355x_thor2_roce_func.json @@ -2515,7 +2515,7 @@ { "name": "CE_Internal_Init", "description": "ncclCeInit lifecycle: allocates sync window; ncclCeFinalize tears it down; double-fini is safe; independent state per comm", - "test_filter": "CeInternalMPITest.InitSetsWindowPointers:CeInternalMPITest.FiniNullsPointers:CeInternalMPITest.DoubleFiniIsSafe:CeInternalMPITest.IndependentCeStatePerComm:CeInternalMPITest.FiniAfterZeroCollectives" + "test_filter": "CeInternalMPITest.InitSetsWindowPointers:CeInternalMPITest.FiniNullsPointers:CeInternalMPITest.DoubleFiniIsSafe:CeInternalMPITest.IndependentCeStatePerComm:CeInternalMPITest.FiniAfterZeroCollectives:CeInternalMPITest.FreeBatchOpsParamsIsIdempotent" }, { "name": "CE_Internal_Sync", diff --git a/projects/rccl/tools/scripts/test_runner/configs/mi455_ainic_roce.json b/projects/rccl/tools/scripts/test_runner/configs/mi455_ainic_roce.json index 7bdae87fd189..3df7ef91808f 100644 --- a/projects/rccl/tools/scripts/test_runner/configs/mi455_ainic_roce.json +++ b/projects/rccl/tools/scripts/test_runner/configs/mi455_ainic_roce.json @@ -2534,7 +2534,7 @@ { "name": "CE_Internal_Init", "description": "ncclCeInit lifecycle: allocates sync window; ncclCeFinalize tears it down; double-fini is safe; independent state per comm", - "test_filter": "CeInternalMPITest.InitSetsWindowPointers:CeInternalMPITest.FiniNullsPointers:CeInternalMPITest.DoubleFiniIsSafe:CeInternalMPITest.IndependentCeStatePerComm:CeInternalMPITest.FiniAfterZeroCollectives" + "test_filter": "CeInternalMPITest.InitSetsWindowPointers:CeInternalMPITest.FiniNullsPointers:CeInternalMPITest.DoubleFiniIsSafe:CeInternalMPITest.IndependentCeStatePerComm:CeInternalMPITest.FiniAfterZeroCollectives:CeInternalMPITest.FreeBatchOpsParamsIsIdempotent" }, { "name": "CE_Internal_Sync", diff --git a/projects/rccl/tools/scripts/test_runner/configs/mi455_ainic_roce_func.json b/projects/rccl/tools/scripts/test_runner/configs/mi455_ainic_roce_func.json index df978ab108d2..c9b6011552b4 100644 --- a/projects/rccl/tools/scripts/test_runner/configs/mi455_ainic_roce_func.json +++ b/projects/rccl/tools/scripts/test_runner/configs/mi455_ainic_roce_func.json @@ -2468,7 +2468,7 @@ { "name": "CE_Internal_Init", "description": "ncclCeInit lifecycle: allocates sync window; ncclCeFinalize tears it down; double-fini is safe; independent state per comm", - "test_filter": "CeInternalMPITest.InitSetsWindowPointers:CeInternalMPITest.FiniNullsPointers:CeInternalMPITest.DoubleFiniIsSafe:CeInternalMPITest.IndependentCeStatePerComm:CeInternalMPITest.FiniAfterZeroCollectives" + "test_filter": "CeInternalMPITest.InitSetsWindowPointers:CeInternalMPITest.FiniNullsPointers:CeInternalMPITest.DoubleFiniIsSafe:CeInternalMPITest.IndependentCeStatePerComm:CeInternalMPITest.FiniAfterZeroCollectives:CeInternalMPITest.FreeBatchOpsParamsIsIdempotent" }, { "name": "CE_Internal_Sync", From 4c05e51131a463e8504bacc24870ce2dea759a99 Mon Sep 17 00:00:00 2001 From: Thomas Huber Date: Wed, 16 Sep 2026 16:55:49 +0000 Subject: [PATCH 09/12] test(rccl): cover the CE LSA-coverage gate and sharpen CHANGELOG entries Add negative cases for the !ncclDevrIsOneLsaTeam reject in ncclCeScratchAvailable, which had no coverage: every mock comm in the CE eligibility suites has lsaSize == nRanks, so deleting the clause left both suites green. The new cases use lsaSize = nRanks/2 with nNodes = 1 and were confirmed to fail with the clause removed. Rewrite the LSA-coverage CHANGELOG entry to lead with the symptom, silently incorrect data under NCCL_LSA_TEAM_SIZE, rather than the internal condition. Correct the enablement entry as well: NCCL_GIN_ENABLE, NCCL_NUM_RMA_CTX, NCCL_WIN_ENABLE and NCCL_DMABUF_ENABLE already default to qualifying values, NCCL_LOCAL_REGISTER is not read by any selection gate, and NCCL_CUMEM_ENABLE=1 is required on gfx942/gfx950 because cuMem auto-detect is limited to gfx1250. Co-authored-by: Cursor --- projects/rccl/CHANGELOG.md | 4 +- .../rccl/test/CeAlltoAllvEligibilityTests.cpp | 50 +++++++++++++++++++ 2 files changed, 52 insertions(+), 2 deletions(-) diff --git a/projects/rccl/CHANGELOG.md b/projects/rccl/CHANGELOG.md index d6b322bc874f..84d3475aaa06 100644 --- a/projects/rccl/CHANGELOG.md +++ b/projects/rccl/CHANGELOG.md @@ -14,7 +14,7 @@ Full documentation for RCCL is available at [https://rccl.readthedocs.io](https: * Added `install.sh --all_unrolls` (`-DBUILD_ALL_UNROLLS=ON`) to generate every unroll factor (1, 2, 4, 8, 16, 32) for the targeted GPU architecture(s), for measuring unroll factors that the default per-arch matrix does not build. The flag also drops the per-architecture pin, so such a build accepts every `RCCL_UNROLL_FACTOR` value on any targeted architecture. * Added the strict `--enable-full-coverage` install flag for unified host + device LLVM source-based code coverage. It requires `--debug`, the device linker, and ROCm 7.15 or newer. The test runner uses the CMake-level `ENABLE_FULL_COVERAGE=AUTO` mode, which falls back to host-only coverage when device instrumentation is unavailable. * Enabled the Copy-Engine profiler path (`ncclProfiler_v6`): Copy-Engine events are emitted for CE collectives, and the example profiler plugin reports them. -* Enabled hierarchical Copy Engine `ncclAllGather` and `ncclAlltoAll` on multi-node communicators: the intra-node phase runs over SDMA/CE across the LSA team and the inter-node phase over one-sided RMA through the CPU (GIN) proxy. Selected when every node has the same number of ranks, `NCCL_CTA_POLICY=2` with symmetric send and receive windows, `NCCL_LOCAL_REGISTER=0`, `NCCL_CUMEM_ENABLE=1`, `NCCL_NUM_RMA_CTX` greater than zero, and a GIN backend available. Validated for correctness and path selection from 1 to 32 nodes. +* Enabled hierarchical Copy Engine `ncclAllGather` and `ncclAlltoAll` on multi-node communicators: the intra-node phase runs over SDMA/CE across the LSA team and the inter-node phase over one-sided RMA through the CPU (GIN) proxy. The path is off by default and takes two opt-ins: `NCCL_CTA_POLICY=2` (equivalently `config.CTAPolicy = NCCL_CTA_POLICY_ZERO`), and symmetric registration of both the send and the receive buffer through `ncclCommWindowRegister`. On architectures other than gfx1250 (MI450) a third is needed, `NCCL_CUMEM_ENABLE=1`, because symmetric memory depends on cuMem and cuMem auto-detection enables itself only on gfx1250. `NCCL_GIN_ENABLE`, `NCCL_NUM_RMA_CTX`, `NCCL_WIN_ENABLE` and `NCCL_DMABUF_ENABLE` need no change, since their defaults already qualify. Selection also requires an available GIN backend and the same number of ranks on every node; communicators that do not qualify keep using the kernel path. Validated for correctness and path selection from 1 to 32 nodes. ### Changed * Adapted the device-initiated GIN backends (Anvil SDMA and rocSHMEM GDA) to the NCCL 2.30.7 GIN API v14: added the new `getGinProperties` host op, dropped the data-path ops (`iput`/`iputSignal`/`iget`/`iflush`/`test`) that moved out of GIN under the GIN/RMA split, switched `createContext` to `ncclGinConfig_v14_t`, updated the device dispatch signatures, and matched the GIN type renumbering (`ROCSHMEM_GDA` and `ANVIL_SDMA` shifted after the new `GIN_GPI` type). The plugins now use the generic (unversioned) `ncclGin_t` / `ncclGinConfig_t` / `ncclGinProperties_t` typedefs so future ABI bumps do not require touching call sites. @@ -30,7 +30,7 @@ Full documentation for RCCL is available at [https://rccl.readthedocs.io](https: * `RCCL_UNROLL_FACTOR` is now rejected when the requested unroll factor's device functions were not compiled for the running architecture, instead of being accepted and then dispatching into an empty device function table. Unroll factor 32 is compiled for gfx1250 only, but a multi-arch build reported every unroll factor as available, so requesting it on another GPU crashed on the device. That value now fails communicator initialization with a warning naming the architecture, and the default selection falls back to the highest unroll factor actually compiled for the GPU rather than trusting the heuristic's choice. The WarpSpeed auto-tuner's preference for unroll factor 2 is subject to the same check, so it no longer replaces a validated unroll factor with one this build cannot dispatch. gfx1250 is unaffected and still defaults to 32. * Fixed the per-unroll device function tables being misaligned in multi-arch builds. The LL128 `SendRecv` kernel was skipped for unrolls 8/16/32, so every function after `SendRecv` in those tables sat one index below the id the host had computed from the unroll-1 ordering, and the last entry of the host lookup table was dropped. `AlltoAllPivot`, `AlltoAllGda`, `AlltoAllvGda` and `AllGatherV` were affected on gfx1250. * Fixed the AMD SMI fabric ABI guard rejecting the layout amd_smi 27.x introduced, which broke the RCCL build outright. The fabric payload union gained a second member, enlarging `amdsmi_fabric_info_t` without moving the v1 fields RCCL reads. RCCL now recognizes that layout, identifies the loaded runtime by how much of the probe buffer it writes, and falls back to the sysfs fabric backend on a 27.x runtime rather than reading a payload it does not model. Fabric topology on such a runtime therefore comes from sysfs, and RCCL warns once per process when it makes that switch. -* Copy Engine collectives are no longer selected when the LSA team does not cover the communicator. The LSA-local CE routines address peers by LSA rank, so on a single node with `NCCL_LSA_TEAM_SIZE` below the local rank count they gathered only `lsaSize` slices and placed them at LSA rather than communicator indices. Such communicators now fall back to the kernel path. +* Fixed Copy Engine `ncclAllGather`, `ncclAlltoAll`, `ncclScatter` and `ncclGather` returning incorrect data, and no error, when `NCCL_LSA_TEAM_SIZE` was set below the number of ranks on the node. These routines index both their peer list and their buffer offsets by LSA rank, so a communicator wider than its LSA team exchanged data only within that team and placed it at LSA rather than communicator offsets, leaving the slices owned by every other rank unwritten. Copy Engine selection now requires the LSA team to cover the whole communicator, so such communicators use the kernel path instead. Runs that leave `NCCL_LSA_TEAM_SIZE` unset are unaffected, because the LSA team then spans every rank on the node. * Fixed gfx1250 LL and LL128 comm-FIFO hangs on sibling DPX partitions. The FIFO store now uses system-scope b128 (`RCCL_LL_FIFO_SYS_SCOPE`) alongside the existing system-scope load, preventing hangs at slot reuse starting from the 9th collective operation. ### Known issues diff --git a/projects/rccl/test/CeAlltoAllvEligibilityTests.cpp b/projects/rccl/test/CeAlltoAllvEligibilityTests.cpp index 1b602f935056..13ff2d902ea9 100644 --- a/projects/rccl/test/CeAlltoAllvEligibilityTests.cpp +++ b/projects/rccl/test/CeAlltoAllvEligibilityTests.cpp @@ -126,6 +126,36 @@ TEST_F(CeAlltoAllvEligibilityTest, CeAvailable_MultiNodeRejected) ncclSymSendRegRecvReg)); } +// The LSA-local CE routines address peers by LSA rank, so a team that does not +// cover the comm moves only lsaSize slices and places them at LSA rather than +// communicator offsets. Reachable on a single node via NCCL_LSA_TEAM_SIZE. +TEST_F(CeAlltoAllvEligibilityTest, CeAvailable_LsaTeamSmallerThanCommRejected) +{ + if (!isCeRuntimeDriverSupported()) + GTEST_SKIP() << "CE driver not in supported range"; + + mockComm_.comm.devrState.lsaSize = mockComm_.comm.nRanks / 2; + EXPECT_FALSE(ncclCeScratchAvailable(mockComm_.get(), + ncclFuncAlltoAllv, + ncclDevSum, + ncclFloat32, + ncclSymSendRegRecvReg)); + EXPECT_FALSE(ncclCeAvailable(mockComm_.get(), + ncclFuncAlltoAllv, + ncclDevSum, + ncclFloat32, + ncclSymSendRegRecvReg)); + + // Restoring only the team size flips the verdict, so no other clause is + // responsible for the rejection above. + mockComm_.comm.devrState.lsaSize = mockComm_.comm.nRanks; + EXPECT_TRUE(ncclCeScratchAvailable(mockComm_.get(), + ncclFuncAlltoAllv, + ncclDevSum, + ncclFloat32, + ncclSymSendRegRecvReg)); +} + TEST_F(CeAlltoAllvEligibilityTest, CeAvailable_NoSymmetricSupportRejected) { if (!isCeRuntimeDriverSupported()) @@ -326,6 +356,26 @@ TEST_F(CeAlltoAllEligibilityTest, MultiNodeRejected) /*capturing=*/false)); } +TEST_F(CeAlltoAllEligibilityTest, LsaTeamSmallerThanCommRejected) +{ + if (!isCeRuntimeDriverSupported()) + GTEST_SKIP() << "CE driver not in supported range"; + + mockComm_.comm.devrState.lsaSize = mockComm_.comm.nRanks / 2; + EXPECT_FALSE(ncclCeAlltoAllEligible(mockComm_.get(), + ncclFloat32, + ncclSymSendRegRecvReg, + /*hasSysmemSegment=*/false, + /*capturing=*/false)); + + mockComm_.comm.devrState.lsaSize = mockComm_.comm.nRanks; + EXPECT_TRUE(ncclCeAlltoAllEligible(mockComm_.get(), + ncclFloat32, + ncclSymSendRegRecvReg, + /*hasSysmemSegment=*/false, + /*capturing=*/false)); +} + TEST_F(CeAlltoAllEligibilityTest, MultiNodeHierAvailable_DoesNotYieldDda) { if (!isCeRuntimeDriverSupported()) From 1088dde487d1dfa1a3fb4b2fbaddac726f1c7534 Mon Sep 17 00:00:00 2001 From: Thomas Huber Date: Thu, 17 Sep 2026 19:19:46 +0000 Subject: [PATCH 10/12] fix(rccl): restore the CE batch-param resets and cover the RMA-context gate ncclCeFreeBatchOpsParams came out of the NCCL v2.31.2-1 sync keeping the pointer nulling but not the counter resets, so numOps, intraBatchSync and numAttrs survived a release. The hierarchical collectives free the same params twice, once per chunk batch and once at exit, and a stale numOps makes the second release look like it still describes a populated batch. CeInternalMPITest.FreeBatchOpsParamsIsIdempotent covers exactly this and was failing; it passes again. Adds CeAlltoAllEligibilityTest.WantInternalCtxRequiresProxyEnabled for the ncclRmaWantInternalCtx change in the merge commit. It pins the composition through numRmaCtx, the one term that appears in ncclRmaProxyEnabled and in no clause of ncclRmaWantInternalCtx's own: set it to 0 on an otherwise hier-eligible comm and both ncclRmaWantInternalCtx and ncclHierCeAvailable must decline. Re-deriving the terms locally, in any form that omits the deferral, makes this case fail. The NCCL_RMA_DISABLE path is deliberately not driven through the environment. NCCL_PARAM caches on first read, so such a case would pass or fail depending on whether anything in the binary read the variable first, and an order-dependent test is worse than none. The composition test kills the same mutant without that fragility. The remaining edits in this file are the sendWin/recvWin arguments the sync added to ncclCeAvailable and ncclHierCeAvailable. Co-authored-by: Cursor --- projects/rccl/src/ce_coll.cc | 7 +++++++ projects/rccl/test/CeAlltoAllvEligibilityTests.cpp | 4 +++- 2 files changed, 10 insertions(+), 1 deletion(-) diff --git a/projects/rccl/src/ce_coll.cc b/projects/rccl/src/ce_coll.cc index 6bd0a53de98e..21b5f038b035 100644 --- a/projects/rccl/src/ce_coll.cc +++ b/projects/rccl/src/ce_coll.cc @@ -709,6 +709,12 @@ void ncclCeFreeBatchOpsParams(struct ncclCeBatchOpsParams* params) { free(params->sizes); params->sizes = nullptr; } + // Reset the counters and the sync flag too, not just the pointers. The + // hierarchical collectives release the same params twice (once per chunk + // batch, once at exit), and a stale numOps would make the second release look + // like it still describes a populated batch. + params->numOps = 0; + params->intraBatchSync = false; #ifdef CE_BATCH_ASYNC_SUPPORTED if (params->attrs) { free(params->attrs); @@ -718,6 +724,7 @@ void ncclCeFreeBatchOpsParams(struct ncclCeBatchOpsParams* params) { free(params->attrIdxs); params->attrIdxs = nullptr; } + params->numAttrs = 0; #endif } diff --git a/projects/rccl/test/CeAlltoAllvEligibilityTests.cpp b/projects/rccl/test/CeAlltoAllvEligibilityTests.cpp index 3f2ebba7e7df..b9c4b3946de9 100644 --- a/projects/rccl/test/CeAlltoAllvEligibilityTests.cpp +++ b/projects/rccl/test/CeAlltoAllvEligibilityTests.cpp @@ -145,7 +145,9 @@ TEST_F(CeAlltoAllvEligibilityTest, CeAvailable_LsaTeamSmallerThanCommRejected) ncclFuncAlltoAllv, ncclDevSum, ncclFloat32, - ncclSymSendRegRecvReg)); + ncclSymSendRegRecvReg, + /*sendWin=*/nullptr, + /*recvWin=*/nullptr)); // Restoring only the team size flips the verdict, so no other clause is // responsible for the rejection above. From 5a4b712f33b8076705a2fb9fd9a9390ec60a7447 Mon Sep 17 00:00:00 2001 From: Thomas Huber Date: Thu, 17 Sep 2026 19:19:46 +0000 Subject: [PATCH 11/12] fix(rccl): repair two NCCL v2.31.2-1 sync regressions outside CE scale-out Neither of these is CE scale-out's, and both can be dropped from this PR if you would rather see them separately. They are here because this branch cannot be verified without them: rccl-UnitTestsDevRuntime segfaults on the first, and the --debug --tests_build --enable-mpi-tests build does not compile at all with the second, and that is the configuration this branch's CE suites run in. 1. symMemoryDestroy lost its membership check (dev_runtime.cc). Before the sync it located mem in devr->memHead and returned early when it was absent, which is what made the function safe to call twice. The guard and the test that covers it both came from a7ad311b1a, "harden symmetric GIN teardown on gfx1250 VMM paths". The post-sync version does the whole teardown first and only then unlinks, with an unguarded walk: struct ncclDevrMemory** ptr = &devr->memHead; while (*ptr != mem) ptr = &(*ptr)->next; This is reachable in production, not just from the test. The error ladders in ncclDevrWindowRegisterInGroup and ncclDevrCommCreateInternal fall through from symWindowDestroy, which destroys winHost->memory, into symMemoryDestroy(comm, mem) on the same memory. A failed symmetric window registration therefore repeats every ncclGinDeregister, ncclRmaProxyDeregister, cuMemUnmap, cuMemRelease and free, then runs off the end of the list. It affects every symmetric-window user, GIN and the Device API included. CI does not see it because CI builds release without --tests_build and nothing runs ctest. Develop's own SymMemoryObtainTest.DestroyIsIdempotent segfaults on it. The check is back at the top, before anything is touched. Note the drain loop at dev_runtime.cc:330 passes devr->memHead, which is always a member, so the early return cannot turn that loop into a spin. 2. GdrFlushTest.ForcedScratchpadWrite_ReproducesFault does not compile. The sync added the case but not the two things it calls: RunRecvFlushBurst has no forceWrite parameter and there is no gdrSupported(). Guarded off rather than half-fixed, because restoring the forceWrite hook is a decision for whoever owns the GDR flush fence. It only breaks builds with ENABLE_FAULT_INJECTION on, which is the default, combined with test targets, which CI omits. Found but not fixed here: SkipCuMemFreePolicy.IsolatedArchAndEnvBranches fails on pristine develop's dev_runtime.cc as well, verified by reverting this file and rebuilding. It concerns cuMem free policy and is nothing to do with CE, so rccl-UnitTestsDevRuntime is expected to read 5/6 rather than 6/6. Co-authored-by: Cursor --- projects/rccl/src/dev_runtime.cc | 10 ++++++++++ .../rccl/test/transport/NetIbMPI/GdrFlushTests.cpp | 11 +++++++++-- 2 files changed, 19 insertions(+), 2 deletions(-) diff --git a/projects/rccl/src/dev_runtime.cc b/projects/rccl/src/dev_runtime.cc index 329d991c88d3..88bc7118034b 100644 --- a/projects/rccl/src/dev_runtime.cc +++ b/projects/rccl/src/dev_runtime.cc @@ -897,6 +897,16 @@ static ncclResult_t symMemoryObtain(struct ncclComm* comm, CUmemGenericAllocatio static void symMemoryDestroy(struct ncclComm* comm, struct ncclDevrMemory* mem) { if (mem != nullptr) { struct ncclDevrState* devr = &comm->devrState; + // Idempotent: a window pair or the finalize drain may already have destroyed + // this mem. Establish membership before touching it, not after. The unlink + // below walks the list unguarded, so a second call on a freed mem would + // otherwise repeat every deregister, unmap, cuMemRelease and free in this + // function and then run off the end of the list. + { + struct ncclDevrMemory* cursor = devr->memHead; + while (cursor != nullptr && cursor != mem) cursor = cursor->next; + if (cursor != mem) return; + } if (devr->ginEnabled && mem->ginSegmentInfos != nullptr) { for (int segment = 0; segment < mem->numGinSegments; segment++) { ncclGinDeregister(comm, mem->ginSegmentInfos[segment].ginHostWins); diff --git a/projects/rccl/test/transport/NetIbMPI/GdrFlushTests.cpp b/projects/rccl/test/transport/NetIbMPI/GdrFlushTests.cpp index 08a64056a9af..78841759410c 100644 --- a/projects/rccl/test/transport/NetIbMPI/GdrFlushTests.cpp +++ b/projects/rccl/test/transport/NetIbMPI/GdrFlushTests.cpp @@ -217,10 +217,17 @@ TEST_F(GdrFlushTest, RepeatedFlush_NoFaultBurst) { EXPECT_EQ(flush, ncclSuccess) << "no flush in the burst may raise a QP async-fatal"; } -#if defined(ENABLE_FAULT_INJECTION) // Regression guard - force the removed scratchpad RDMA_WRITE back. On a dma-buf // scratchpad this must reproduce the fault (flush no longer succeeds), proving // the WRITE is the culprit and that removing it is the fix. +// +// Disabled: the NCCL v2.31.2-1 sync (#10785) added this case but not the two +// things it calls. RunRecvFlushBurst has no forceWrite parameter and there is no +// gdrSupported(), so a --debug --tests_build --enable-mpi-tests build of develop +// does not compile once ENABLE_FAULT_INJECTION is on. Restoring the forceWrite +// hook is a decision for whoever owns the GDR flush fence, not something to +// guess at here, so the case is guarded off rather than half-fixed. +#if 0 TEST_F(GdrFlushTest, ForcedScratchpadWrite_ReproducesFault) { SKIP_UNLESS_MPI_PREREQS(kExactTwoProcesses, kExactTwoProcesses, false, kMinGpusPerNode, kNoNodeLimit); @@ -237,7 +244,7 @@ TEST_F(GdrFlushTest, ForcedScratchpadWrite_ReproducesFault) { EXPECT_NE(forced, ncclSuccess) << "forced scratchpad RDMA_WRITE on a dma-buf buffer should fault the flush QP"; } -#endif // ENABLE_FAULT_INJECTION +#endif // disabled pending the forceWrite hook } // namespace From 7ca726a745b7a9dbf34a07eb0767166b2e2356b2 Mon Sep 17 00:00:00 2001 From: Thomas Huber Date: Thu, 17 Sep 2026 21:17:36 +0000 Subject: [PATCH 12/12] fix(rccl): guard the CE tuning model against an uninitialized sync frequency CollImplInfo.AllReduceMatchesDispatchLog dies with SIGFPE on gfx94X. The graph- captured unicast branch of the CE cost model divides by comm->ceColl.intraBatchSyncFreq: int freq = (int)comm->ceColl.intraBatchSyncFreq; if (numOps > freq && (uint64_t)perRankBytes * (uint64_t)numOps >= comm->ceColl.intraBatchSyncMsgThreshold) { int blocks = (numOps + freq - 1) / freq; That field and intraBatchSyncMsgThreshold are written in exactly one place, ncclCeInit (ce_coll.cc:197-198), so both are still 0 on a communicator that never initialized CE. gfx942 is such a communicator: ncclCeImplemented declines on the driver version, so ncclCeInit never runs. With freq == 0 the existing condition does not protect the divide, it guarantees it. numOps > 0 holds for any rank count, and the threshold term compares against 0, so both halves pass and the divide is reached every time. gfx950 hides this because CE does initialize there and freq is non-zero. Adding freq > 0 is the whole fix: with no CE init there is no intra-batch sync, so there is no latency for the model to add and skipping the term is correct. src/tuning/ce_model.cc arrived with the NCCL v2.31.2-1 sync (#10785) and does not exist at this branch's pre-sync base, and the file is otherwise identical to develop, so this is a sync regression rather than CE scale-out's. It is fixed here because it fails a required check on this PR. Verified: CollImplInfo.* is 9/9 on gfx950 before and after, so the guard changes nothing where freq was already valid. Co-authored-by: Cursor --- projects/rccl/src/tuning/ce_model.cc | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/projects/rccl/src/tuning/ce_model.cc b/projects/rccl/src/tuning/ce_model.cc index 00f6fea8d707..479a8f03fbe2 100644 --- a/projects/rccl/src/tuning/ce_model.cc +++ b/projects/rccl/src/tuning/ce_model.cc @@ -93,8 +93,13 @@ static float ncclCeAgLatencyUs(struct ncclComm* comm, bool multicast, int nRanks // Graph-captured unicast also pays the intra-batch sync latency; add it. if (!multicast && captured) { int numOps = inPlace ? (nRanks - 1) : nRanks; + // intraBatchSyncFreq is written only by ncclCeInit, so it is still 0 on a + // comm that never initialized CE (gfx942 declines on the driver version). + // numOps > 0 and a 0 threshold would both pass, so without this term the + // guard guarantees the divide below rather than preventing it. int freq = (int)comm->ceColl.intraBatchSyncFreq; - if (numOps > freq && (uint64_t)perRankBytes * (uint64_t)numOps >= comm->ceColl.intraBatchSyncMsgThreshold) { + if (freq > 0 && numOps > freq && + (uint64_t)perRankBytes * (uint64_t)numOps >= comm->ceColl.intraBatchSyncMsgThreshold) { int blocks = (numOps + freq - 1) / freq; int numSyncs = (blocks - 1) + ((blocks % 2 == 0) ? 1 : 0); // per-freq syncs + even-count workaround total += (float)numSyncs * floorUs;