fix(rccl): make CE AlltoAllv intra-batch sync count rank-invariant (AICOMRCCL-2474)

projects/rccl/CHANGELOG.md

@@ -45,6 +45,7 @@ Full documentation for RCCL is available at [https://rccl.readthedocs.io](https:
 * Fixed tuner plugins receiving uninitialized cost-model constants.
 * Fixed GIN proxy descriptor shared-memory sizing and alignment.
 * Fixed virtual address space exhaustion when symmetric windows backed by the same physical allocation are registered repeatedly.
+* Fixed Copy Engine `ncclAlltoAllv` hanging on intra-node communicators of more than 8 ranks with uneven send sizes. Each rank decided from its own send sizes whether to add intra-batch barriers, so ranks could issue different barrier counts. The decision now uses the full send-size matrix, which every rank shares.
 * Fixed `NCCL_P2P_DISABLE` and `NCCL_P2P_LEVEL` being ignored on Intel, ARM and Zhaoxin hosts, where the CPU-based default P2P level overwrote the user setting. On such hosts `NCCL_P2P_DISABLE=1` now falls back to SHM as documented, and `NCCL_P2P_LEVEL` is honored.
 
 ### Known issues

projects/rccl/src/ce_coll.cc

@@ -1150,17 +1150,38 @@ ncclResult_t ncclCeAlltoAllv(struct ncclComm* comm, struct ncclCeCollArgs* args,
   uint8_t* myRecvBuff = (uint8_t*)args->recvBuff;
   void* peerRecvBuff;
   size_t offset, winOff;
-  size_t totalBytes = 0;
+  const int freq = (int)comm->ceColl.intraBatchSyncFreq;
+  bool roundSync = false;
   struct ncclCeBatchOpsParams batchOpsParams = {};
   NCCLCHECKGOTO(ncclCeInitBatchOpsParams(&batchOpsParams, comm->nRanks), ret, fail);
 
   NCCLCHECKGOTO(ncclAlltoAllvValidateSizeMatrix(args->sizes, comm->nRanks), ret, fail);
 
+  // ncclMemOpSync is collective: decide intra-batch sync from the full size matrix, not this rank's nonzero peers.
+  for (int src = 0; freq > 0 && src < comm->nRanks && !roundSync; src++) {
+    size_t* srcSizes = ncclAlltoAllvSendSizes(args->sizes, src, comm->nRanks);
+    int ops = 0;
+    size_t bytes = 0;
+    for (int dst = 0; dst < comm->nRanks; dst++) {
+      if (srcSizes[dst] != 0) {
+        ops++;
+        bytes += srcSizes[dst];
+      }
+    }
+    roundSync = ops > freq && bytes >= comm->ceColl.intraBatchSyncMsgThreshold;
+  }
+
   // Ensure all ranks are ready before starting transfers
   NCCLCHECKGOTO(ncclMemOpSync(comm, stream, args), ret, fail);
 
   // Copy data to other ranks: send variable-sized chunk for each destination rank
   for (int r = 0; r < comm->nRanks; r++) {
+    // Sync per round of freq destination slots, so every rank issues the same number of barriers.
+    if (roundSync && r > 0 && r % freq == 0) {
+      NCCLCHECKGOTO(ncclCeLaunchBatchOps(comm, &batchOpsParams, stream, args), ret, fail);
+      batchOpsParams.numOps = 0;
+      NCCLCHECKGOTO(ncclMemOpSync(comm, stream, args), ret, fail);
+    }
     int dstRank = (comm->rank + r) % comm->nRanks;
     const size_t chunkBytes = sendSizes[dstRank];
     if (chunkBytes == 0) {
@@ -1169,7 +1190,6 @@ ncclResult_t ncclCeAlltoAllv(struct ncclComm* comm, struct ncclCeCollArgs* args,
 
     uint8_t* srcPtr = mySendBuff + sendDispls[dstRank];
     uint8_t* dstPtr = myRecvBuff + recvDispls[comm->rank];
-    totalBytes += chunkBytes;
 
     if (dstRank == comm->rank) {
       // Local copy for own data
@@ -1197,13 +1217,14 @@ ncclResult_t ncclCeAlltoAllv(struct ncclComm* comm, struct ncclCeCollArgs* args,
     }
   }
 
-  // Check if we need to perform intra-batch synchronization
-  batchOpsParams.intraBatchSync =
-    (batchOpsParams.numOps > comm->ceColl.intraBatchSyncFreq && totalBytes >= comm->ceColl.intraBatchSyncMsgThreshold);
-
-  // Launch the batch operations
+  // Launch the last round; intraBatchSync stays false so ncclCeLaunchBatchOps adds no rank-local barriers
   NCCLCHECKGOTO(ncclCeLaunchBatchOps(comm, &batchOpsParams, stream, args), ret, fail);
 
+  // Graph replay needs an even barrier count so useCompletePtr returns to its pre-launch state.
+  if (roundSync && ncclCudaGraphValid(comm->planner.capturingGraph) && ((comm->nRanks - 1) / freq) % 2 != 0) {
+    NCCLCHECKGOTO(ncclMemOpSync(comm, stream, args), ret, fail);
+  }
+
   // Ensure all transfers are complete across all ranks
   NCCLCHECKGOTO(ncclMemOpSync(comm, stream, args), ret, fail);
 

projects/rccl/test/host/CMakeLists.txt

@@ -938,11 +938,25 @@ endif()
 # ---------------------------------------------------------------------------
 # rccl-UnitTestsMicroCe
 # ---------------------------------------------------------------------------
+# In-build, CE_BATCH_ASYNC_SUPPORTED arrives with rccl's own compile definitions when CE_ENABLED.
+set(_MICRO_CE_BATCH_DEF "")
+if(NOT BUILD_TESTS)
+  include(CheckCXXSymbolExists)
+  set(CMAKE_TRY_COMPILE_TARGET_TYPE STATIC_LIBRARY)
+  set(CMAKE_REQUIRED_DEFINITIONS -D__HIP_PLATFORM_AMD__)
+  check_cxx_symbol_exists(hipMemcpyBatchAsync "hip/hip_runtime_api.h" _MICRO_HIP_MEMCPY_BATCH_ASYNC)
+  unset(CMAKE_REQUIRED_DEFINITIONS)
+  set(CMAKE_TRY_COMPILE_TARGET_TYPE EXECUTABLE)
+  if(_MICRO_HIP_MEMCPY_BATCH_ASYNC)
+    set(_MICRO_CE_BATCH_DEF CE_BATCH_ASYNC_SUPPORTED)
+  endif()
+endif()
 rccl_add_micro_binary(rccl-UnitTestsMicroCe
   SOURCES ${TEST_MICRO_CE_SOURCE_FILES}
   EXTRA_DEFS
     ${_MICRO_MODE_DEFS}
     ${_MICRO_AMDSMI_DEF}
+    ${_MICRO_CE_BATCH_DEF}
     RCCL_STUBS_OMIT_ncclCeFinalize       # ce_coll.cc defines it
     CE_COLL_CC_PATH="${RCCL_HOST_HIPIFY_DIR}/src/ce_coll.cc"
   CATEGORIES_YAML "${CMAKE_CURRENT_SOURCE_DIR}/../test_categories_micro.yaml")

projects/rccl/test/host/MICROTEST_README.md

@@ -201,8 +201,12 @@ export colliding non-`static` symbols; otherwise a unit needs its own binary:
   because `enqueue.cc` defines them itself. See
   `test_categories_micro_enqueue.yaml`.
 - **`rccl-UnitTestsMicroCe`**: `ce_coll.cc` (via `CE_COLL_CC_PATH`); suite
-  `CeCollMicrotest.*`. Its own binary because every other target links
-  `fakes/ce_fakes.cc`, which fakes the symbols this file defines.
+  `CeCollMicrotest.*` and `LaunchPaths/CeAlltoAllvSyncMicrotest.*`, both in
+  `ce-coll-test.cc`. Its own binary because every other target links
+  `fakes/ce_fakes.cc`, which fakes the symbols this file defines, and the
+  `rccl-UnitTestsMicroInit` cleanup-order oracle needs the fake `ncclCeFinalize`
+  in `fakes/nccl_stubs.cc`, which this target drops via
+  `RCCL_STUBS_OMIT_ncclCeFinalize`.
 - **`rccl-UnitTestsMicroSymKernels`** — the REAL `src/sym_kernels.cc` (via
   `SYM_KERNELS_CC_PATH`, from `sym-kernels-test.cc`), compiled together with the
   GENERATED `sym_kernels_host.cc` it calls into; suites `SymKernelMicrotest.*`,
@@ -394,7 +398,7 @@ symbol.
 |---|---|
 | `src/algorithms/dda/*.cc` | `fakes/dda_fakes.cc` |
 | `src/bootstrap.cc` | `fakes/bootstrap_stubs.cc` |
-| `src/ce_coll.cc` | `fakes/ce_fakes.cc` |
+| `src/ce_coll.cc` (targets that do not compile the real file) | `fakes/ce_fakes.cc` |
 | `src/collectives.cc` | `fakes/collectives_fakes.cc` |
 | `src/dev_runtime.cc` (targets that do not compile the real file) | `fakes/dev_runtime_fakes.cc` |
 | `src/diagnostics.cc` (`ncclDiagChildRun*` external-tool runners) | `fakes/diagnostics_fakes.cc` |
@@ -781,7 +785,7 @@ above (`./install.sh -t`, wired via `add_subdirectory(host)`), the same file
 can be configured **directly** to build every host binary — `rccl-HostUnitTests`,
 `rccl-UnitTestsMicro`, `rccl-UnitTestsMicroWarpSpeed`,
 `rccl-UnitTestsMicroInit[-uncached|-faultinj]`, `rccl-UnitTestsMicroEnqueue[-devlinker]`,
-`rccl-UnitTestsMicroGinHost, `rccl-UnitTestsMicroDiagnostics`,
+`rccl-UnitTestsMicroGinHost`, `rccl-UnitTestsMicroCe`, `rccl-UnitTestsMicroDiagnostics`,
 `rccl-UnitTestsMicroSymKernels` and `rccl-UnitTestsMicroTaskPrep` — **without configuring/building all of
 librccl**. It compiles just the tests + fakes + the hipified unit-under-test
 sources.
@@ -823,6 +827,7 @@ cmake --build build -j"$(nproc)"
 ./build/rccl-UnitTestsMicroDiagnostics        # src/diagnostics/{p2p,ib_write_bw}.cc tests
 ./build/rccl-UnitTestsMicroGinHost            # src/gin/gin_host.cc GIN_PROXY_NTHREADS
 ./build/rccl-UnitTestsMicroDiagnostics        # src/diagnostics/p2p.cc tests
+./build/rccl-UnitTestsMicroCe                 # src/ce_coll.cc tests
 ./build/rccl-HostUnitTests
 ```