Skip to content
Merged
Show file tree
Hide file tree
Changes from 5 commits
Commits
Show all changes
19 commits
Select commit Hold shift + click to select a range
78918c5
feat(rccl): Enable hierarchical CE scale-out collectives
thomas-huber Aug 7, 2026
69d542a
test(rccl): Cover hierarchical CE scale-out collectives
thomas-huber Aug 7, 2026
a555140
Merge branch 'develop' into users/thomas-huber/AICOMRCCL-630-ce-scaleout
thomas-huber Aug 27, 2026
73ebdc7
Merge remote-tracking branch 'origin/develop' into users/thomas-huber…
thomas-huber Sep 1, 2026
fb7299f
fix(rccl): Address CE scale-out review feedback
thomas-huber Sep 8, 2026
d8d27f2
fix(rccl): Address CE scale-out review round 2
thomas-huber Sep 9, 2026
eb1869e
Merge remote-tracking branch 'origin/develop' into users/thomas-huber…
thomas-huber Sep 9, 2026
d26a711
Merge remote-tracking branch 'origin/develop' into users/thomas-huber…
thomas-huber Sep 15, 2026
59b90eb
fix(rccl): Address CE scale-out review round 3
thomas-huber Sep 15, 2026
7bbec52
Merge remote-tracking branch 'origin/develop' into users/thomas-huber…
thomas-huber Sep 15, 2026
dacfc2c
fix(rccl): address CE scale-out review feedback
thomas-huber Sep 15, 2026
96cc65d
fix(rccl): report hierarchical CE AllGather as CE for -A 1
thomas-huber Sep 16, 2026
9df520c
test(rccl): pin CE eligibility mocks and FreeBatchOps runner filter
thomas-huber Sep 16, 2026
e964ac6
Merge remote-tracking branch 'origin/develop' into users/thomas-huber…
thomas-huber Sep 16, 2026
4c05e51
test(rccl): cover the CE LSA-coverage gate and sharpen CHANGELOG entries
thomas-huber Sep 16, 2026
0f400cf
Merge remote-tracking branch 'origin/develop' into users/thomas-huber…
thomas-huber Sep 17, 2026
1088dde
fix(rccl): restore the CE batch-param resets and cover the RMA-contex…
thomas-huber Sep 17, 2026
5a4b712
fix(rccl): repair two NCCL v2.31.2-1 sync regressions outside CE scal…
thomas-huber Sep 17, 2026
7ca726a
fix(rccl): guard the CE tuning model against an uninitialized sync fr…
thomas-huber Sep 17, 2026
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions projects/rccl/CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Comment thread
thomas-huber marked this conversation as resolved.
Resolved

### 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.
Expand Down
142 changes: 96 additions & 46 deletions projects/rccl/src/ce_coll.cc
Original file line number Diff line number Diff line change
Expand Up @@ -104,7 +104,8 @@ ncclResult_t ncclCeInit(struct ncclComm* comm) {
ncclWindow_vidmem* sigWinDevHost = nullptr;
ncclWindow_vidmem* arWinDev = nullptr;
ncclWindow_vidmem* arWinDevHost = nullptr;
size_t ceDevBaseSize = alignUp(comm->nRanks * sizeof(uint32_t), 16) * 2;
// Sized from the LSA team once ncclDevrInitOnce has populated devrState.
size_t ceDevBaseSize = 0;
Comment thread
garzaran marked this conversation as resolved.
Resolved
size_t sigBufferSize = NUM_SLOTS * comm->nRanks * sizeof(uint32_t);
size_t maxChunkBytes = ncclCeAllReduceMaxChunkBytes(comm->nRanks);
size_t ceARTmpBufSize = alignUp(NUM_SLOTS * comm->nRanks * maxChunkBytes, 16);
Expand All @@ -114,6 +115,9 @@ ncclResult_t ncclCeInit(struct ncclComm* comm) {
// Symmetric memory runtime must be initialized before any window registration.
NCCLCHECKGOTO(ncclDevrInitOnce(comm), ret, fail);

// Sync window holds one ready and one complete slot per LSA-local rank.
ceDevBaseSize = alignUp(comm->devrState.lsaSize * sizeof(uint32_t), 16) * 2;
Comment thread
thomas-huber marked this conversation as resolved.
Resolved

// Local-only control words (no peer access -> no window registration needed).
CUDACHECKGOTO(hipExtMallocWithFlags((void**)&comm->ceColl.d_barrierSync, 2 * sizeof(uint32_t),
hipDeviceMallocUncached),
Expand All @@ -138,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->nRanks * 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;
Expand Down Expand Up @@ -170,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)) {
Comment thread
thomas-huber marked this conversation as resolved.
Resolved
NCCLCHECKGOTO(ncclMemAlloc((void**)&ceARTmpBuf, ceARTmpBufSize), ret, fail_ar);
NCCLCHECKGOTO(ncclDevrWindowRegisterInGroup(comm, ceARTmpBuf, ceARTmpBufSize, NCCL_WIN_COLL_SYMMETRIC, &arWinDev),
ret, fail_ar);
Expand Down Expand Up @@ -365,6 +371,7 @@ ncclResult_t ncclPrepMCSync(struct ncclComm* comm, bool isComplete, hipStreamBat
size_t* opIdx, cudaStream_t stream) {
ncclResult_t ret = ncclSuccess;

// Sync slots are addressed through LSA pointers, so index by LSA rank.
int myLsaRank = comm->devrState.lsaSelf;
int lsaSize = comm->devrState.lsaSize;
uint32_t* readyPtrs = (uint32_t*)comm->ceColl.baseUCSymReadyPtr;
Expand Down Expand Up @@ -418,12 +425,15 @@ static ncclResult_t ncclPrepUCSyncCapture(struct ncclComm* comm, bool isComplete
ncclResult_t ret = ncclSuccess;
hipStreamBatchMemOpParams* writeParams = nullptr;
size_t writeIdx = 0;
void* dstPtr = isComplete ? (void*)&completePtrs[comm->rank] : (void*)&readyPtrs[comm->rank];
// Sync slots are addressed through LSA pointers, so index by LSA rank.
Comment thread
thomas-huber marked this conversation as resolved.
Resolved
int myLsaRank = comm->devrState.lsaSelf;
int lsaSize = comm->devrState.lsaSize;
void* dstPtr = isComplete ? (void*)&completePtrs[myLsaRank] : (void*)&readyPtrs[myLsaRank];
size_t offset = (uint8_t*)dstPtr - (uint8_t*)comm->ceColl.ceSyncWin->userPtr;

NCCLCHECKGOTO(ncclCalloc(&writeParams, comm->nRanks), ret, fail);
for (int r = 0; r < comm->nRanks; ++r) {
if (r == comm->rank) continue;
NCCLCHECKGOTO(ncclCalloc(&writeParams, lsaSize), ret, fail);
for (int r = 0; r < lsaSize; ++r) {
if (r == myLsaRank) continue;
void* peerDstPtr;
NCCLCHECKGOTO(ncclDevrGetLsaRankPtr(comm, comm->ceColl.ceSyncWin, offset, r, &peerDstPtr), ret, fail);
writeParams[writeIdx] = {};
Expand All @@ -449,11 +459,14 @@ static ncclResult_t ncclPrepUCSyncNonCapture(struct ncclComm* comm, bool isCompl
uint32_t* completePtrs, uint32_t waitValue,
hipStreamBatchMemOpParams* batchParams, size_t* opIdx) {
ncclResult_t ret = ncclSuccess;
void* dstPtr = isComplete ? (void*)&completePtrs[comm->rank] : (void*)&readyPtrs[comm->rank];
// Sync slots are addressed through LSA pointers, so index by LSA rank.
int myLsaRank = comm->devrState.lsaSelf;
int lsaSize = comm->devrState.lsaSize;
void* dstPtr = isComplete ? (void*)&completePtrs[myLsaRank] : (void*)&readyPtrs[myLsaRank];
size_t offset = (uint8_t*)dstPtr - (uint8_t*)comm->ceColl.ceSyncWin->userPtr;

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;
NCCLCHECKGOTO(ncclDevrGetLsaRankPtr(comm, comm->ceColl.ceSyncWin, offset, r, &peerDstPtr), ret, fail);
batchParams[*opIdx] = {};
Expand All @@ -478,6 +491,9 @@ ncclResult_t ncclPrepUCSync(struct ncclComm* comm, bool isComplete, hipStreamBat
NCCLCHECK(ceFaultCheck(comm, CE_FAULT_SYNC_PREP, "ncclPrepUCSync"));
#endif

// Sync slots are addressed through LSA pointers, so index by LSA rank.
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;

Expand All @@ -493,8 +509,8 @@ ncclResult_t ncclPrepUCSync(struct ncclComm* comm, bool isComplete, hipStreamBat
}

// Waits always go in batchParams (submitted by ncclMemOpSync; reset follows if capturing).
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 =
Expand All @@ -514,20 +530,25 @@ ncclResult_t ncclMemOpSync(struct ncclComm* comm, cudaStream_t stream, struct nc
ncclResult_t ret = ncclSuccess;
void* ceSyncHandle = NULL;
int lsaSize = comm->devrState.lsaSize;
bool useMCSync = false;
size_t batchSize = 0;
size_t opIdx = 0;
hipStreamBatchMemOpParams* batchParams = nullptr;

// Get pointers to the ready and complete synchronization arrays
uint32_t* readyPtrs = (uint32_t*)comm->ceColl.baseUCSymReadyPtr;
uint32_t* completePtrs = (uint32_t*)comm->ceColl.baseUCSymComplPtr;

// Multicast synchronization is not available across scale-up cliques.
useMCSync = comm->nvlsSupport && !comm->p2pCrossClique;

// 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;
size_t opIdx = 0;
batchSize = (useMCSync ? NCCL_CE_SYNC_OPS_PER_RANK_MC : NCCL_CE_SYNC_OPS_PER_RANK_UC) * lsaSize;
Comment thread
thomas-huber marked this conversation as resolved.
Resolved

// 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, stream), ret, fail);
Expand All @@ -545,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)) {
Comment thread
thomas-huber marked this conversation as resolved.
Resolved
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 =
Expand Down Expand Up @@ -597,13 +620,22 @@ ncclResult_t ncclCeInitBatchOpsParams(struct ncclCeBatchOpsParams* params, int n
goto exit;
}

// Safe to call more than once on the same params.
Comment thread
thomas-huber marked this conversation as resolved.
Resolved
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
}

Expand Down Expand Up @@ -1990,38 +2022,56 @@ 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 ncclFuncAlltoAllv:
NCCLCHECKGOTO(ncclCeAlltoAllv(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;
case ncclFuncAllReduce:
// CE init runs (in ncclCommGroupRegisterSymmetric) before doLaunches, so
// ceARTmpBuf is guaranteed non-NULL by the time we get here.
if (comm->ceColl.ceARTmpBuf == NULL) {
WARN("CE AllReduce invoked before CE init; this should not happen");
ret = ncclInvalidUsage;
// 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);
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;
}
} 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 ncclFuncAlltoAllv:
NCCLCHECKGOTO(ncclCeAlltoAllv(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;
case ncclFuncAllReduce:
// CE init runs (in ncclCommGroupRegisterSymmetric) before doLaunches, so
// ceARTmpBuf is guaranteed non-NULL by the time we get here.
if (comm->ceColl.ceARTmpBuf == NULL) {
WARN("CE AllReduce invoked before CE init; this should not happen");
ret = ncclInvalidUsage;
break;
}
// Pass args->recvWin so ncclCeAllReduce can take the fast path
// (AG written directly into user recvbuff, no final D2D copy).
NCCLCHECKGOTO(ncclCeAllReduce(comm, args->sendBuff, args->recvBuff, args->nElts, args->datatype, args->redOp,
stream, args->recvWin),
ret, fail);
break;
default:
ret = ncclInvalidUsage;
NCCLCHECKGOTO(ncclCeAllReduce(comm, args->sendBuff, args->recvBuff, args->nElts, args->datatype, args->redOp,
stream, args->recvWin),
ret, fail);
break;
default:
ret = ncclInvalidUsage;
}
}
// DDA path: results were staged in scratch (args->recvBuff). Copy them back to
// the user's recv buffer. Copy-back semantics are collective-specific:
Expand Down Expand Up @@ -2079,7 +2129,7 @@ ncclResult_t scheduleCeCollTaskToPlan(struct ncclComm* comm, struct ncclKernelPl

if (comm->rank == 0) {
if (!ncclDevrIsOneLsaTeam(comm)) {
Comment thread
thomas-huber marked this conversation as resolved.
Resolved
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" : "";
Expand Down
15 changes: 12 additions & 3 deletions projects/rccl/src/dev_runtime.cc
Original file line number Diff line number Diff line change
Expand Up @@ -2278,12 +2278,21 @@ 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.
// The host proxy consumes the handles produced by ncclRmaProxyRegister.
if (winHost->memory == nullptr) {
return winHost->rmaDevWins[ctx];
return winHost->rmaHostWins[ctx];
Comment thread
thomas-huber marked this conversation as resolved.
Resolved
}
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) {
Comment thread
thomas-huber marked this conversation as resolved.
Resolved
Comment thread
thomas-huber marked this conversation as resolved.
Resolved
if (winHost == nullptr || winHost->memory == nullptr) {
return 0;
}
return winHost->bigOffset - winHost->memory->bigOffset;
}

// Get the multicast address for a given team
Expand Down
40 changes: 24 additions & 16 deletions projects/rccl/src/enqueue.cc
Original file line number Diff line number Diff line change
Expand Up @@ -2071,9 +2071,14 @@ ncclResult_t ncclLaunchPrepare(struct ncclComm* comm) {
plan->ceCollArgs->sizes = (task->func == ncclFuncAlltoAllv) ? task->sizes : nullptr;

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);
if (!ncclDevrIsOneLsaTeam(comm)) {
Comment thread
thomas-huber marked this conversation as resolved.
Resolved
Comment thread
thomas-huber marked this conversation as resolved.
Resolved
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" : "";
Comment thread
thomas-huber marked this conversation as resolved.
Resolved
INFO(NCCL_TUNING, "%s [Copy Engine]: %ld Bytes -> cudaMemcpy%s", ncclFuncToString(task->func),
task->count * ncclTypeSize(task->datatype), nvlsSync);
}
}

ncclIntruQueueEnqueue(&planner->planQueue, plan);
Expand Down Expand Up @@ -3852,31 +3857,34 @@ static ncclResult_t taskAppend(struct ncclComm* comm, struct ncclInfo* info) {
ceArGraphAllowed = rcclCeAllReduceAllowed(comm);
}

// 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));
bool ceAvailable = !ceCapturing && ncclCeAvailable(comm, info->coll, info->op, info->datatype, winRegType);
// Hierarchical CE (sync feature) rides the same symmetric-window ceCollTaskAppend
// path; keep it graph-capture-safe by gating on !ceCapturing like ceAvailable.
bool hierCeAvailable =
!ceCapturing && ncclHierCeAvailable(comm, info->coll, info->op, info->datatype, winRegType);

// Trigger CE initialization on the first CE-capable collective.
// This covers collectives whose user buffers ARE registered (AllGather,
// AlltoAll, Scatter, Gather) as well as AllReduce, which may bypass the
// 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).
// the CE runtime (ceARTmpBuf stays NULL). Hierarchical CE is multi-node by
// construction, so it needs the same trigger despite nNodes > 1.
if (!ceCapturing && ncclCeImplemented(info->coll, info->op, info->datatype) && comm->symmetricSupport &&
comm->nNodes == 1 && comm->ceColl.baseUCSymReadyPtr == NULL && ncclIntruQueueEmpty(&comm->ceInitTaskQueue)) {
(comm->nNodes == 1 || hierCeAvailable) && comm->ceColl.baseUCSymReadyPtr == NULL &&
Comment thread
thomas-huber marked this conversation as resolved.
Resolved
Comment thread
thomas-huber marked this conversation as resolved.
Resolved
Comment thread
thomas-huber marked this conversation as resolved.
Resolved
ncclIntruQueueEmpty(&comm->ceInitTaskQueue)) {
struct ncclCeInitTask* ceTask;
NCCLCHECK(ncclCalloc(&ceTask, 1));
ceTask->comm = comm;
ncclIntruQueueEnqueue(&comm->ceInitTaskQueue, ceTask);
ncclGroupCommJoin(comm, ncclGroupTaskTypeSymRegister);
}

// Size gate for CE AllReduce without symmetric memory registration: ceARTmpBuf is sized for at most
// NCCL_CE_AR_MAX_MSG_BYTES total bytes.
bool ceAllReduceFits = false;
ncclSymRegType_t winRegType;
NCCLCHECK(ncclGetSymRegType(sendWin, recvWin, &winRegType));
bool ceAvailable = !ceCapturing && ncclCeAvailable(comm, info->coll, info->op, info->datatype, winRegType);
// Hierarchical CE (sync feature) rides the same symmetric-window ceCollTaskAppend
// path; keep it graph-capture-safe by gating on !ceCapturing like ceAvailable.
bool hierCeAvailable =
!ceCapturing && ncclHierCeAvailable(comm, info->coll, info->op, info->datatype, winRegType);
if (info->coll == ncclFuncAllReduce) {
const bool ceAllReduceOpSupported =
(info->op == ncclSum || info->op == ncclProd || info->op == ncclMin || info->op == ncclMax);
Expand Down
Loading
Loading