Extend NetIbMPI multithread validation

projects/rccl/.claude/skills/feature-unit-testing/multithread-validation.md

@@ -51,6 +51,41 @@ turns a multithread configuration into a single-threaded run:
 - reduce the value across all ranks and fail before the suite starts unless
   every rank resolved the same one.
 
+## Choose which tests get a threaded variant
+
+Pick by the state concurrency would actually stress, not by how easy the body is
+to convert. In descending order of value:
+
+- **Registration paths.** The MR cache and protection domain are per physical
+  device behind one mutex, so registration storms, refcount churn on a single
+  shared address, and fused-device cycling (where one registration fans out
+  across both members) are the only way to race insert, lookup and eviction.
+- **Fault injection whose state is per communicator.** One worker breaking its
+  own connection while the others keep transferring is an assertion no serial
+  test can make. Concurrent link failures are also the only way to load a
+  process-global recovery thread.
+- **Per-communicator scheduler state**, such as WRR tokens and cursor: every
+  worker arms and audits its own ledger, so N schedulers advance at once while
+  sharing nothing.
+- **Data paths under pressure** — size ladders, multi-QP splits, request-pool
+  churn, GPU buffers. There is no software race to find in per-comm state; the
+  value is real queue contention plus proof that completion routing never
+  crosses communicators.
+
+Give every worker a distinct payload seed. A transfer delivered on the wrong
+connection then fails verification instead of passing silently, which is the one
+cheap defence against the whole model being wrong.
+
+Leave alone anything that asserts on process-global state: device and merged-NIC
+tables, counters snapshotted process-wide, and timing claims about a shared
+background thread. Resource-leak checks still work, but only on the main thread
+around the entire run — an in-loop snapshot sees other workers' live resources
+and reports a leak that is not there.
+
+Spread workers across devices when the suite has several NICs. Pinning every
+worker to device 0 leaves the other devices' protection domains and caches
+untouched.
+
 ## Harness structure
 
 - The main thread creates every context and connection, exchanges handles over

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

@@ -350,6 +350,9 @@ TEST_F(NetIbMPITest, SimpleSendRecv) {
     AssertNoRdmaLeaks(before, CaptureRdmaResources(), "threaded SimpleSendRecv");
 }
 
+// Parameterized by MPIEnvironment::nThreads. Concurrent workers each sweep the
+// same size ladder on an independent connection, so registrations of many
+// different sizes hit the shared per-device MR cache at once.
 TEST_F(NetIbMPITest, SendRecvMultipleSizes) {
     ASSERT_TRUE(validateTestPrerequisites(kExactTwoProcesses, kExactTwoProcesses,
                                          false, kMinGpusPerNode, kNoNodeLimit))
@@ -359,13 +362,34 @@ TEST_F(NetIbMPITest, SendRecvMultipleSizes) {
     AssertInitAndGetDevices(&ndev);
 
     const int rank = MPIEnvironment::world_rank;
-    ConnectionPair pair;
-    NetConnectionGuard connGuard(net_);
-    SetupConnectionWithGuard(0, pair, connGuard);
+    const int nThreads = MPIEnvironment::nThreads;
 
     // Test various sizes
     std::vector<size_t> testSizes = {1, 64, 256, 1024, 4096, 16384, 65536};
 
+    if (nThreads > 1) {
+        const RdmaResourceCounts before = CaptureRdmaResources();
+        RunMultiThreadedIndependent(
+            ThreadDevPolicy::Spread(), nThreads,
+            [&](int threadIdx, ConnectionPair& pair) -> ThreadResult {
+                ThreadResult result;
+                for (size_t size : testSizes) {
+                    const int seed = kMultiSizeSeedOffset + threadIdx * 10000
+                                     + static_cast<int>(size);
+                    result = WorkerHostTransfer(rank, pair, size, 100, seed);
+                    if (!result.ok) return result;
+                }
+                return result;
+            });
+        MPI_Barrier(MPI_COMM_WORLD);
+        AssertNoRdmaLeaks(before, CaptureRdmaResources(), "threaded SendRecvMultipleSizes");
+        return;
+    }
+
+    ConnectionPair pair;
+    NetConnectionGuard connGuard(net_);
+    SetupConnectionWithGuard(0, pair, connGuard);
+
     for (size_t size : testSizes) {
         const int tag = 100;
         const int seed = 2000 + static_cast<int>(size);  // Unique seed per size
@@ -736,6 +760,9 @@ TEST_F(NetIbMPITest, MultipleSequentialTransfers) {
     AssertNoRdmaLeaks(before, CaptureRdmaResources(), "threaded MultipleSequentialTransfers");
 }
 
+// Parameterized by MPIEnvironment::nThreads. Concurrent 16 MB transfers put
+// several large registrations in the per-device MR cache simultaneously and
+// keep the NIC saturated while every worker verifies its own payload.
 TEST_F(NetIbMPITest, LargeTransfer) {
     ASSERT_TRUE(validateTestPrerequisites(kExactTwoProcesses, kExactTwoProcesses,
                                          false, kMinGpusPerNode, kNoNodeLimit))
@@ -745,6 +772,22 @@ TEST_F(NetIbMPITest, LargeTransfer) {
     AssertInitAndGetDevices(&ndev);
 
     const int rank = MPIEnvironment::world_rank;
+    const int nThreads = MPIEnvironment::nThreads;
+
+    if (nThreads > 1) {
+        const RdmaResourceCounts before = CaptureRdmaResources();
+        RunMultiThreadedIndependent(
+            ThreadDevPolicy::Spread(), nThreads,
+            [&](int threadIdx, ConnectionPair& pair) -> ThreadResult {
+                return WorkerHostTransfer(rank, pair, kLargeBufferSize, 400,
+                                          kBaseSeedOffset + threadIdx,
+                                          kLargeTransferTimeout);
+            });
+        MPI_Barrier(MPI_COMM_WORLD);
+        AssertNoRdmaLeaks(before, CaptureRdmaResources(), "threaded LargeTransfer");
+        return;
+    }
+
     ConnectionPair pair;
     NetConnectionGuard connGuard(net_);
     SetupConnectionWithGuard(0, pair, connGuard);