net-ib: fix fault injection API issues from review

- ncclIbCastFaultSetQpDelay/SetQpError: use comm->base.nqps for bounds
  check instead of NCCL_IB_MAX_QPS
- ncclIbCastFaultClear: atomically reset fatalErrorCount to 0
- move net_ib_fault_inject.h into transport/net_ib_cast/; drop local
  #define NCCL_IB_MAX_QPS 128, include net_ib_cast_inspect.h and add
  static_assert so a size mismatch becomes a compile error
- fault hook in IbCastMultiSend: use IbCastStatsFatalError (renamed from
  ncclIbStatsFatalError in asanniko's split)
- FaultInjCastQpErrorClearRecovers: ASSERT_EQ on SetQpError return value
  (was silently ignored); drain recvReq before DeregisterMemory
- FaultInjCastSingleQpErrorIsFatal: EXPECT_EQ on ncclIbCastSetTokens and
  ncclIbCastFaultClear return values (Copilot review)

Co-Authored-By: Claude Sonnet 4 <noreply@anthropic.com>

Author: i-kosarev

projects/rccl/src/CMakeLists.txt

@@ -217,7 +217,7 @@ set(SRC_FILES
   include/plugin/net/net_v11.h
   include/plugin/profiler/net_ib_v1.h
   include/plugin/profiler/net_ib.h
-  include/net_ib_fault_inject.h
+  transport/net_ib_cast/net_ib_fault_inject.h
   include/plugin/profiler/net_socket_v1.h
   include/plugin/profiler/net_socket.h
   include/plugin/profiler/profiler_v1.h

…s/rccl/src/include/net_ib_fault_inject.h …nsport/net_ib_cast/net_ib_fault_inject.hprojects/rccl/src/include/net_ib_fault_inject.h renamed to projects/rccl/src/transport/net_ib_cast/net_ib_fault_inject.h projects/rccl/src/include/net_ib_fault_inject.h renamed to projects/rccl/src/transport/net_ib_cast/net_ib_fault_inject.h

@@ -22,13 +22,11 @@ extern "C" {
 /*
  * Test-only per-QP fault injection API for the net-ib CAST transport.
  *
- * Implemented in src/transport/net_ib_cast.cc (CAST multi-QP path).
- *
- * NCCL_IB_MAX_QPS is defined in net_ib_cast_inspect.h; guard against
- * double-definition when both headers are included together.
+ * Implemented in src/transport/net_ib_cast/p2p.cc (CAST multi-QP path).
  */
-#ifndef NCCL_IB_MAX_QPS
-#define NCCL_IB_MAX_QPS 128
+#include "net_ib_cast_inspect.h"
+#ifdef __cplusplus
+static_assert(NCCL_IB_MAX_QPS == 128, "fault injection arrays sized for 128 QPs; update if NCCL_IB_MAX_QPS changes");
 #endif
 
 /* ── CAST path (net_ib_cast.cc) ───────────────────────────────────────── */
@@ -39,7 +37,7 @@ extern "C" {
 ncclResult_t ncclIbCastFaultSetQpDelay(void* sendComm, int qpIdx, uint32_t delayUs);
 
 /* Arm error injection on a specific QP index.
- * When armed, the hook calls ncclIbStatsFatalError and then returns
+ * When armed, the hook calls IbCastStatsFatalError and then returns
  * ncclSystemError instead of calling wrap_ibv_post_send.
  * Set inject=false to disarm. */
 ncclResult_t ncclIbCastFaultSetQpError(void* sendComm, int qpIdx, bool inject);

projects/rccl/src/transport/net_ib_cast/p2p.cc

@@ -298,7 +298,7 @@ ncclResult_t IbCastMultiSend(struct ncclIbSendComm* comm, int slot, int nqps, in
       const uint32_t faultDelay = comm->base.faultQpDelayUs[qpIndex];
       if (faultDelay) usleep(faultDelay);
       if (comm->base.faultQpError[qpIndex]) {
-        ncclIbStatsFatalError(&comm->base.stats);
+        IbCastStatsFatalError(&comm->base.stats);
         return ncclSystemError;
       }
     }
@@ -1094,15 +1094,15 @@ ncclResult_t IbCastTest(void* request, int* done, int* sizes) {
 ncclResult_t ncclIbCastFaultSetQpDelay(void* sendComm, int qpIdx, uint32_t delayUs) {
   if (!sendComm) return ncclInvalidArgument;
   struct ncclIbSendComm* comm = (struct ncclIbSendComm*)sendComm;
-  if (qpIdx < 0 || qpIdx >= NCCL_IB_MAX_QPS) return ncclInvalidArgument;
+  if (qpIdx < 0 || qpIdx >= comm->base.nqps) return ncclInvalidArgument;
   comm->base.faultQpDelayUs[qpIdx] = delayUs;
   return ncclSuccess;
 }
 
 ncclResult_t ncclIbCastFaultSetQpError(void* sendComm, int qpIdx, bool inject) {
   if (!sendComm) return ncclInvalidArgument;
   struct ncclIbSendComm* comm = (struct ncclIbSendComm*)sendComm;
-  if (qpIdx < 0 || qpIdx >= NCCL_IB_MAX_QPS) return ncclInvalidArgument;
+  if (qpIdx < 0 || qpIdx >= comm->base.nqps) return ncclInvalidArgument;
   comm->base.faultQpError[qpIdx] = inject;
   return ncclSuccess;
 }
@@ -1112,6 +1112,7 @@ ncclResult_t ncclIbCastFaultClear(void* sendComm) {
   struct ncclIbSendComm* comm = (struct ncclIbSendComm*)sendComm;
   memset(comm->base.faultQpDelayUs, 0, sizeof(comm->base.faultQpDelayUs));
   memset(comm->base.faultQpError, 0, sizeof(comm->base.faultQpError));
+  __atomic_store_n(&comm->base.stats.fatalErrorCount, 0, __ATOMIC_RELEASE);
   return ncclSuccess;
 }
 

projects/rccl/test/CMakeLists.txt

@@ -94,6 +94,7 @@ if(BUILD_TESTS)
     ${PROJECT_BINARY_DIR}/hipify/gensrc # for rccl_bfloat16.h
     ${PROJECT_BINARY_DIR}/hipify/src # for graph/topo.h
     ${PROJECT_BINARY_DIR}/hipify/src/include/plugin # for recorder tests, nccl_tuner.h
+    ${PROJECT_BINARY_DIR}/hipify/src/transport/net_ib_cast # for net_ib_fault_inject.h
     ${ROCM_PATH}/include
     ${ROCM_PATH}
   )

projects/rccl/test/transport/NetIbMPI/FaultInjectTests.cpp

@@ -462,7 +462,7 @@ TEST_F(NetIbMPITest, FaultInjCastSingleQpErrorIsFatal) {
         // Concentrate all tokens on QP 0: tokens[0]=1, tokens[1..n-1]=0.
         std::vector<int> tokens(actualNqps, 0);
         tokens[0] = 1;
-        ncclIbCastSetTokens(sendComm, tokens.data(), actualNqps);
+        EXPECT_EQ(ncclIbCastSetTokens(sendComm, tokens.data(), actualNqps), ncclSuccess);
         r1.setErrRet = static_cast<int>(ncclIbCastFaultSetQpError(sendComm, targetQp, /*inject=*/true));
     }
     MPI_Barrier(MPI_COMM_WORLD);
@@ -602,22 +602,27 @@ TEST_F(NetIbMPITest, FaultInjCastQpErrorClearRecovers) {
 
     if (rank == 1) {
         for (int q = 0; q < actualNqps; ++q)
-            ncclIbCastFaultSetQpError(sendComm1, q, /*inject=*/true);
+            ASSERT_EQ(ncclIbCastFaultSetQpError(sendComm1, q, /*inject=*/true), ncclSuccess);
     }
     MPI_Barrier(MPI_COMM_WORLD);
 
     // Trigger the fault (rank 0 posts recv, rank 1 posts send that will fail).
+    // rank 1 forwards its fault result to rank 0 so we can assert the fault
+    // was actually observed before testing recovery in phase 2.
+    static constexpr int kPhase1MpiTag = 9882;
+    FaultInjectResult p1 = {};
+    void* recvReq1 = nullptr;
+    bool  recvDone1 = false;
     if (rank == 0) {
         void*  bufs[1]    = {buf1};
         size_t sizes[1]   = {kMsgSize};
         int    tags[1]    = {501};
         void*  handles[1] = {mhandle1};
-        void* recvReq = nullptr;
-        ASSERT_EQ(PostRecv(recvComm1, 1, bufs, sizes, tags, handles, &recvReq), ncclSuccess);
+        ASSERT_EQ(PostRecv(recvComm1, 1, bufs, sizes, tags, handles, &recvReq1), ncclSuccess);
         for (int poll = 0; poll < 100; poll++) {
             int done = 0, sz = 0;
-            if (TestRequest(recvReq, &done, &sz) != ncclSuccess) break;
-            if (done) break;
+            if (TestRequest(recvReq1, &done, &sz) != ncclSuccess) break;
+            if (done) { recvDone1 = true; break; }
             usleep(kPollIntervalUs);
         }
     } else {
@@ -628,21 +633,47 @@ TEST_F(NetIbMPITest, FaultInjCastQpErrorClearRecovers) {
             if (sendRet != ncclSuccess || sendReq != nullptr) break;
             usleep(kPollIntervalUs);
         }
+        int fatalCount = 0;
         if (sendRet == ncclSuccess && sendReq != nullptr) {
             for (int poll = 0; poll < 200; poll++) {
                 int done = 0, sz = 0;
-                int fc = 0;
                 TestRequest(sendReq, &done, &sz);
-                ncclIbCastFaultGetFatalCount(sendComm1, &fc);
-                if (done || fc > 0) break;
+                ncclIbCastFaultGetFatalCount(sendComm1, &fatalCount);
+                if (done || fatalCount > 0) break;
                 usleep(kPollIntervalUs);
             }
+        } else {
+            ncclIbCastFaultGetFatalCount(sendComm1, &fatalCount);
         }
-        ncclIbCastFaultClear(sendComm1);
+        p1.sendRet    = static_cast<int>(sendRet);
+        p1.fatalCount = fatalCount;
+        EXPECT_EQ(ncclIbCastFaultClear(sendComm1), ncclSuccess);
+        MPI_Send(&p1, sizeof(p1), MPI_BYTE, 0, kPhase1MpiTag, MPI_COMM_WORLD);
+    }
+
+    if (rank == 0) {
+        MPI_Recv(&p1, sizeof(p1), MPI_BYTE, 1, kPhase1MpiTag, MPI_COMM_WORLD,
+                 MPI_STATUS_IGNORE);
+        bool isendFailed = (p1.sendRet != static_cast<int>(ncclSuccess));
+        EXPECT_TRUE(isendFailed || p1.fatalCount > 0)
+            << "Phase 1: fault injection did not trigger — isend returned "
+            << p1.sendRet << ", fatalCount=" << p1.fatalCount
+            << "; recovery test is meaningless without a confirmed fault";
     }
 
     MPI_Barrier(MPI_COMM_WORLD);
 
+    // Drain any outstanding recv request before closing the comm.
+    // The recv may not complete if the sender faulted — that is expected here.
+    if (rank == 0 && !recvDone1 && recvReq1 != nullptr) {
+        for (int poll = 0; poll < 500; poll++) {
+            int done = 0, sz = 0;
+            if (TestRequest(recvReq1, &done, &sz) != ncclSuccess) break;
+            if (done) break;
+            usleep(kPollIntervalUs);
+        }
+    }
+
     ASSERT_EQ(DeregisterMemory(comm1, mhandle1), ncclSuccess);
     if (rank == 0) {
         ASSERT_EQ(CloseRecvComm(recvComm1), ncclSuccess);