fix(rccl): Fix DDA IPC hang

projects/rccl/src/algorithms/dda/ipc/ipc_init.cu

@@ -8,6 +8,7 @@
 
 #include "alloc.h"
 #include "archinfo.h"
+#include "bootstrap.h"
 #include "checks.h"
 #include "comm.h"
 #include "debug.h"
@@ -29,10 +30,35 @@
     } \
   } while (0)
 
-ncclResult_t ncclDdaIpcCommInit(ncclComm* comm) {
-  if (comm == nullptr) {
-    return ncclSuccess;
+namespace {
+
+constexpr int kDdaIpcFiniEnteredTag = 0xDDA0;
+constexpr int kDdaIpcFiniClosedTag = 0xDDA1;
+
+bool ddaIpcCommEligible(const ncclComm* comm) {
+  if (comm == nullptr || !rcclParamDdaEnable()) {
+    return false;
   }
+  const bool ddaArchSupported =
+    comm->archName != nullptr && (IsArchMatch(comm->archName, "gfx942") || IsArchMatch(comm->archName, "gfx950"));
+  return comm->nRanks == kDdaNranks && comm->nNodes == 1 && comm->bootstrap != nullptr && !comm->directMode &&
+         !comm->MNNVL && ddaArchSupported;
+}
+
+void rememberFirstError(ncclResult_t candidate, ncclResult_t* result) {
+  if (*result == ncclSuccess && candidate != ncclSuccess) {
+    *result = candidate;
+  }
+}
+
+bool ddaIpcCanCollectivelyTeardown(const ncclComm* comm) {
+  return ddaIpcCommEligible(comm) && comm->abortFlag != nullptr && !comm->revokedFlag &&
+         COMPILER_ATOMIC_LOAD(comm->abortFlag, std::memory_order_acquire) == 0;
+}
+
+} // namespace
+
+ncclResult_t ncclDdaIpcCommInit(ncclComm* comm) {
   // Skip DDA if:
   // - nRanks is not exactly kDdaNranks (currently hardcoded to 8)
   // - multi-node runs
@@ -44,10 +70,7 @@
   //   scratch/barrier here is not necessary. On gfx12xx (RDNA4) the
   //   uncached-memory IPC export fails (hipIpcGetMemHandle -> hipErrorInvalidValue),
   //   which aborts comm init entirely. Gate init to match dispatch.
-  const bool ddaArchSupported =
-    comm->archName != nullptr && (IsArchMatch(comm->archName, "gfx942") || IsArchMatch(comm->archName, "gfx950"));
-  if (comm->nRanks != kDdaNranks || comm->nNodes != 1 || comm->bootstrap == nullptr || comm->directMode ||
-      comm->MNNVL || !ddaArchSupported) {
+  if (!ddaIpcCommEligible(comm)) {
     return ncclSuccess;
   }
 
@@ -200,8 +223,43 @@
   if (comm == nullptr) {
     return ncclSuccess;
   }
+
+  // IPC importers must release both the barrier and scratch mappings before any
+  // exporting rank frees its local allocation. Keep abort/revoke cleanup local:
+  // those paths cannot require every rank to enter another collective.
+  ncclResult_t result = ncclSuccess;
+  bool teardownCollectiveReady = false;
+  if (ddaIpcCanCollectivelyTeardown(comm)) {
+    ncclResult_t barrierResult =
+      bootstrapBarrier(comm->bootstrap, comm->rank, comm->nRanks, kDdaIpcFiniEnteredTag);
+    if (barrierResult == ncclSuccess) {
+      teardownCollectiveReady = true;
+    } else {
+      WARN("ncclDdaIpcCommFini: entry barrier failed");
+      rememberFirstError(barrierResult, &result);
+    }
+  }
+
+  auto* barrierState = static_cast<DdaIpcBarrierState*>(comm->ddaIpcBarrierState);
+  if (barrierState != nullptr && barrierState->resources != nullptr &&
+      barrierState->resources->ipcMemHandler != nullptr) {
+    rememberFirstError(barrierState->resources->ipcMemHandler->closePeerMappings(), &result);
+  }
+  if (comm->ddaIpcMemHandler != nullptr) {
+    rememberFirstError(comm->ddaIpcMemHandler->closePeerMappings(), &result);
+  }
+
+  if (teardownCollectiveReady) {
+    ncclResult_t barrierResult =
+      bootstrapBarrier(comm->bootstrap, comm->rank, comm->nRanks, kDdaIpcFiniClosedTag);
+    if (barrierResult != ncclSuccess) {
+      WARN("ncclDdaIpcCommFini: peer-close barrier failed");
+      rememberFirstError(barrierResult, &result);
+    }
+  }
+
   if (comm->ddaIpcBarrierState != nullptr) {
-    delete static_cast<DdaIpcBarrierState*>(comm->ddaIpcBarrierState);
+    delete barrierState;
     comm->ddaIpcBarrierState = nullptr;
   }
   CUDACHECKIGNORE(cudaFree(comm->ddaPeerPtrsDev));
@@ -215,5 +273,5 @@
   CUDACHECKIGNORE(cudaFree(comm->ddaScratch));
   comm->ddaScratch = nullptr;
   comm->ddaScratchBytes = 0;
-  return ncclSuccess;
+  return result;
 }

projects/rccl/src/algorithms/dda/ipc/ipc_mem_handler.cc

@@ -18,15 +18,27 @@ ncclIpcMemHandler::ncclIpcMemHandler(void* bootstrap, int rank, int nranks)
     exchanged_(false) {}
 
 ncclIpcMemHandler::~ncclIpcMemHandler() {
-  if (!exchanged_) {
-    return;
-  }
+  (void)closePeerMappings();
+}
+
+ncclResult_t ncclIpcMemHandler::closePeerMappings() {
+  // Close every mapping that was opened, including a partial exchange that
+  // returned before exchanged_ could be set.
+  ncclResult_t result = ncclSuccess;
   for (int i = 0; i < nranks_; ++i) {
-    if (i == rank_) {
+    if (i == rank_ || memPtrs_[static_cast<size_t>(i)] == nullptr) {
       continue;
     }
-    CUDACHECKIGNORE(cudaIpcCloseMemHandle(memPtrs_[static_cast<size_t>(i)]));
+    cudaError_t err = cudaIpcCloseMemHandle(memPtrs_[static_cast<size_t>(i)]);
+    if (err != cudaSuccess) {
+      WARN("ncclIpcMemHandler::closePeerMappings: cudaIpcCloseMemHandle(rank %d) failed (%s)", i,
+           cudaGetErrorString(err));
+      result = ncclUnhandledCudaError;
+    }
+    memPtrs_[static_cast<size_t>(i)] = nullptr;
   }
+  exchanged_ = false;
+  return result;
 }
 
 ncclResult_t ncclIpcMemHandler::addSelfDeviceMemPtr(void* deviceMemPtr) {

projects/rccl/src/algorithms/dda/ipc/ipc_mem_handler.h

@@ -38,6 +38,7 @@ class ncclIpcMemHandler {
 
   ncclResult_t addSelfDeviceMemPtr(void* deviceMemPtr);
   ncclResult_t exchangeMemPtrs();
+  ncclResult_t closePeerMappings();
   ncclResult_t getPeerDeviceMemPtr(int peerRank, void** outPeerPtr) const;
 
 private:

projects/rccl/test/host/CMakeLists.txt

@@ -388,6 +388,19 @@ function(rccl_define_micro_source_lists)
     fakes/transport_p2p_fakes.cc         # src/transport/p2p.cc shareable buffers
     ../common/main_altrsmi.cpp
     PARENT_SCOPE)
+
+  # src/algorithms/dda/ipc/{ipc_init.cu,ipc_mem_handler.cc,ipc_gpu_barrier.cu},
+  # included by the test TU so teardown ordering can be tested with host fakes.
+  set(TEST_MICRO_DDA_IPC_SOURCE_FILES
+    dda-ipc-lifecycle-test.cc
+    fakes/bootstrap_stubs.cc
+    fakes/hip_fakes.cc
+    fakes/nccl_fakes.cc
+    fakes/utils_fakes.cc
+    ${RCCL_HOST_HIPIFY_DIR}/src/algorithms/dda/device/device_buffer.cc
+    ${RCCL_HOST_HIPIFY_DIR}/src/misc/archinfo.cc
+    ../common/main_altrsmi.cpp
+    PARENT_SCOPE)
 endfunction()
 
 # Create one host-only micro-test binary and apply all the boilerplate that
@@ -1121,6 +1134,21 @@ rccl_add_micro_binary(rccl-UnitTestsMicroDiagnostics
     DIAG_IB_WRITE_BW_CC_PATH="${RCCL_HOST_HIPIFY_DIR}/src/diagnostics/ib_write_bw.cc"
   CATEGORIES_YAML "${CMAKE_CURRENT_SOURCE_DIR}/../test_categories_micro_diagnostics.yaml")
 
+# ---------------------------------------------------------------------------
+# rccl-UnitTestsMicroDdaIpc -- host-only DDA IPC init/teardown lifecycle
+# ---------------------------------------------------------------------------
+set_source_files_properties(
+  ${RCCL_HOST_HIPIFY_DIR}/src/algorithms/dda/device/device_buffer.cc
+  PROPERTIES GENERATED TRUE)
+rccl_add_micro_binary(rccl-UnitTestsMicroDdaIpc
+  SOURCES ${TEST_MICRO_DDA_IPC_SOURCE_FILES}
+  EXTRA_DEFS
+    ${_MICRO_MODE_DEFS}
+    DDA_IPC_INIT_CU_PATH="${RCCL_HOST_HIPIFY_DIR}/src/algorithms/dda/ipc/ipc_init.cu.cpp"
+    DDA_IPC_MEM_HANDLER_CC_PATH="${RCCL_HOST_HIPIFY_DIR}/src/algorithms/dda/ipc/ipc_mem_handler.cc"
+    DDA_IPC_GPU_BARRIER_CU_PATH="${RCCL_HOST_HIPIFY_DIR}/src/algorithms/dda/ipc/ipc_gpu_barrier.cu.cpp"
+  CATEGORIES_YAML "${CMAKE_CURRENT_SOURCE_DIR}/../test_categories_micro_dda_ipc.yaml")
+
 # =====================================================================
 # Standalone-only tail: coverage + no CTest registration.
 # =====================================================================