feat(rccl): Enable hierarchical CE scale-out collectives

projects/rccl/CHANGELOG.md

@@ -12,6 +12,7 @@ Full documentation for RCCL is available at [https://rccl.readthedocs.io](https:
 * RAS diagnostics (`NCCL_RUN_RAS_DIAGNOSTICS`) covering GPU inventory, ROCm runtime versions, ECC counters, XGMI link state, and `NCCL_*` environment consistency (AMDSMI in place of NVML).
 * Communicator init diagnostics (`NCCL_RUN_DIAGNOSTICS`) with an active P2P connectivity check.
 * Multiple GIN proxy progress threads via `NCCL_GIN_PROXY_NTHREADS`.
+* 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; the inter-node rail draws from the internal RMA context pool (`NCCL_NUM_RMA_INT_CTX`, default 4) rather than the user-addressable contexts, so `numRmaCtx` only has to stay above zero. 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
 * **Breaking: `NCCL_GIN_TYPE` values for AMD backends are not compatible with 2.30.7.** NCCL 2.31 inserted EFA GDA at value 5, so rocSHMEM GDA moved 5→6 and Anvil SDMA moved 6→7. The IB proxy remains `2`. Jobs that still set `NCCL_GIN_TYPE=6` now select rocSHMEM GDA, not Anvil SDMA. See `src/gin/README.md`.
@@ -22,6 +23,8 @@ Full documentation for RCCL is available at [https://rccl.readthedocs.io](https:
 ### Resolved issues
 * Restored topo tuning-model init (`ncclTopoTuneModel`) after the 2.31 `ncclTuningInit` switch so multi-node kernels do not launch with `blockDim.x=0`.
 * Grouped multi-rank finalize to match the v2.31 teardown barrier.
+* 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 on the scratch (`RCCL_FORCE_CE`) path as well as the registered-window path, 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 hierarchical Copy Engine collectives dereferencing a NULL RMA proxy context when `NCCL_RMA_DISABLE=1` was combined with the hierarchical opt-ins. `NCCL_RMA_DISABLE` suppresses the window registration that creates those contexts, but the hierarchical availability check did not consult it, so a multi-node communicator with `NCCL_CTA_POLICY=2` and symmetric send and receive windows admitted the path and then crashed at launch. The internal RMA context predicate now defers to `ncclRmaProxyEnabled`, so such a communicator declines the hierarchical path and uses the kernel path instead.
 
 ### Known issues
 * The following NCCL 2.31 features are NVIDIA-specific and are not available in RCCL: Compute Fabric Transport, the GDAKI and EFA GDA GIN backends, PAT combined with NVLS, NVLink-multicast AllGather, TMA-based symmetric kernels, and the CuTeDSL and `nccl4rust` bindings.

projects/rccl/src/ce_coll.cc

@@ -391,6 +391,21 @@ bool ncclCeScratchAvailable(struct ncclComm* comm, ncclFunc_t coll, int /*ncclDe
     TRACE(NCCL_TUNING, "Skipping CE collective: comm is not a single node");
     return false;
   }
+  // The scratch routines address peers by LSA rank, so a team that does not
+  // cover the comm would touch only lsaSize slices, at the wrong indices.
+  // Reachable on a single node via NCCL_LSA_TEAM_SIZE. Prefer the initialized
+  // LSA team; calling ncclTeamLsa before device-runtime init (unit-test mocks,
+  // pre-init comms) can SIGSEGV on a null peerInfo.
+  int lsaRanks = comm->nRanks;
+  if (comm->devrState.bigSize != 0) {
+    lsaRanks = ncclTeamLsa(comm).nRanks;
+  } else if (comm->localRanks > 0) {
+    lsaRanks = comm->localRanks;
+  }
+  if (lsaRanks < comm->nRanks) {
+    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;
@@ -590,10 +605,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;
 
+  // Multicast sync needs a multimem mapping over the LSA team, which a
+  // cross-clique (scale-up) comm does not have. hasLsaMultimem is the same
+  // signal the symmetric kernels and enqueue use, so the selection here, the
+  // op-count sizing below and the [Copy Engine] marker cannot disagree.
+  const bool useMCSync = comm->symkState.hasLsaMultimem;
+
   // Allocate enough slots for all possible ops. PrepUC/MC sync iterate the LSA
   // team (not comm->nRanks); size with nRanks so a stale lsaSize of 0 cannot
   // under-allocate, matching the nRanks-wide ready/complete arrays.
-  size_t batchSize = (comm->nvlsSupport ? NCCL_CE_SYNC_OPS_PER_RANK_MC : NCCL_CE_SYNC_OPS_PER_RANK_UC) * comm->nRanks;
+  size_t batchSize = (useMCSync ? NCCL_CE_SYNC_OPS_PER_RANK_MC : NCCL_CE_SYNC_OPS_PER_RANK_UC) * comm->nRanks;
   size_t opIdx = 0;
   hipStreamBatchMemOpParams* batchParams = nullptr;
 
@@ -602,7 +623,7 @@ ncclResult_t ncclMemOpSync(struct ncclComm* comm, cudaStream_t stream, struct nc
   // Prepare batch memory operations for synchronization
   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);
@@ -688,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);
@@ -697,6 +724,7 @@ void ncclCeFreeBatchOpsParams(struct ncclCeBatchOpsParams* params) {
     free(params->attrIdxs);
     params->attrIdxs = nullptr;
   }
+  params->numAttrs = 0;
 #endif
 }
 
@@ -1224,8 +1252,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;
   }
@@ -1250,6 +1281,10 @@ bool ncclHierCeAvailable(struct ncclComm* comm, ncclFunc_t coll, int /*ncclDevRe
   return true;
 }
 
+bool ncclHierCeDispatch(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 {
@@ -2369,48 +2404,57 @@ 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:
-    // Multi-node CE AllGather is hierarchical (RMA rail + intra-node CE). The
-    // single-node ncclCeAllGather path treats every global rank as an LSA rank
-    // and fails ncclDevrGetLsaRankPtr once nRanks > lsaSize.
-    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. The
+  // single-node ncclCe* paths treat every global rank as an LSA rank and fail
+  // ncclDevrGetLsaRankPtr once nRanks > lsaSize.
+  if (ncclHierCeDispatch(comm)) {
+    switch (args->func) {
+    case ncclFuncAllGather:
       NCCLCHECKGOTO(ncclHierCeAllGather(comm, plan, stream), ret, fail);
-    } else {
-      NCCLCHECKGOTO(ncclCeAllGather(comm, args, stream), ret, fail);
-    }
-    break;
-  case ncclFuncAlltoAll:
-    if (!ncclDevrIsOneLsaTeam(comm)) {
+      break;
+    case ncclFuncAlltoAll:
       NCCLCHECKGOTO(ncclHierCeAlltoAll(comm, plan, stream), ret, fail);
-    } else {
-      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:
-    NCCLCHECKGOTO(ncclCeEnsureAllReduceStaging(comm), ret, fail);
-    if (comm->ceColl.ceARTmpBuf == NULL) {
-      WARN("CE AllReduce invoked without staging buffer");
-      ret = ncclInvalidUsage;
       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:
+      NCCLCHECKGOTO(ncclCeEnsureAllReduceStaging(comm), ret, fail);
+      if (comm->ceColl.ceARTmpBuf == NULL) {
+        WARN("CE AllReduce invoked without staging buffer");
+        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, args),
-                  ret, fail);
-    break;
-  default:
-    ret = ncclInvalidUsage;
+      NCCLCHECKGOTO(ncclCeAllReduce(comm, args->sendBuff, args->recvBuff, args->nElts, args->datatype, args->redOp,
+                                    stream, args->recvWin, args),
+                    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:
@@ -2474,10 +2518,13 @@ ncclResult_t scheduleCeCollTaskToPlan(struct ncclComm* comm, struct ncclKernelPl
   plan->ceCollArgs->sizes = (task->func == ncclFuncAlltoAllv) ? task->sizes : nullptr;
 
   if (comm->rank == 0) {
-    if (!ncclDevrIsOneLsaTeam(comm)) {
-      INFO(NCCL_TUNING, "%s [Hierarchical CE]: %ld Bytes -> RMA proxy + CE", ncclFuncToString(task->func),
+    // 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 {
+      // Matches the useMCSync predicate in ncclMemOpSync.
       const char* nvlsSync = comm->symkState.hasLsaMultimem ? "; CE synchronization with NVLS" : "";
       INFO(NCCL_TUNING, "%s [Copy Engine]: %ld Bytes -> cudaMemcpy%s", ncclFuncToString(task->func),
            task->count * ncclTypeSize(task->datatype), nvlsSync);

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);

projects/rccl/src/enqueue/enqueue.cc

@@ -4303,32 +4303,40 @@ 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, sendWin, recvWin);
+      // 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,
+                                                                 sendWin, recvWin);
+
       // 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. 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 && comm->ceColl.baseUCSymReadyPtr == NULL && ncclIntruQueueEmpty(&comm->ceInitTaskQueue)) {
+          (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));
         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, sendWin, recvWin);
-      // 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,
-                                                                 sendWin, recvWin);
       if (info->coll == ncclFuncAllReduce) {
         const bool ceAllReduceOpSupported =
           (info->op == ncclSum || info->op == ncclProd || info->op == ncclMin || info->op == ncclMax);

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]"
+
 // Largest message eligible for the unregistered forced CE AllReduce path.
 #define NCCL_CE_AR_MAX_MSG_BYTES (256ull * 1024 * 1024)
 
@@ -165,6 +169,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, struct ncclDevrWindow* sendWin, struct ncclDevrWindow* recvWin);
 
+// 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 ncclHierCeDispatch(struct ncclComm* comm);
+
 ncclResult_t ncclCeInit(struct ncclComm* comm);
 
 ncclResult_t ncclCeFinalize(struct ncclComm* comm);