From 96d1f2d09471af244bd87bd0ef361328e9b6e6fb Mon Sep 17 00:00:00 2001 From: Ilya Kosarev Date: Wed, 16 Sep 2026 13:29:45 +0000 Subject: [PATCH 1/4] net-ib: add worker-safe IB-CAST scheduler inspection helpers Token ledgers and the round-robin cursor are per send communicator, so a worker can arm and read its own connection without disturbing its siblings -- which is what lets the CAST scheduler tests run threaded at all. WorkerCastSetTokens/WorkerCastGetSchedState/WorkerCastLiveNqps wrap the inspection API in the ThreadResult convention the rest of the threaded harness uses, and ThreadedCastAgreedNqps establishes the one fact the whole family needs before it can size anything: the connection's live queue-pair count, agreed across ranks. That count cannot be taken from NCCL_IB_QPS_PER_CONNECTION. The variable states a request; on a merged device the plugin creates that many per member, so a split-data threshold derived from the environment sits on the wrong side of the boundary and the token-delta assertions test nothing. A worker cannot broadcast the real count either, so the probe brings up its own throwaway connection on the main thread, reads the count from the scheduler once it is warm, and agrees it before any worker starts. --- .../transport/NetIbMPI/NetIbMPITestBase.hpp | 347 ++++++++++++++++++ 1 file changed, 347 insertions(+) diff --git a/projects/rccl/test/transport/NetIbMPI/NetIbMPITestBase.hpp b/projects/rccl/test/transport/NetIbMPI/NetIbMPITestBase.hpp index d1fbd71ad274..639293ecb890 100644 --- a/projects/rccl/test/transport/NetIbMPI/NetIbMPITestBase.hpp +++ b/projects/rccl/test/transport/NetIbMPI/NetIbMPITestBase.hpp @@ -1454,6 +1454,353 @@ class NetIbMPITest : public MPITestBase { return result; } + // ── Worker-safe IB-CAST scheduler inspection ───────────────────── + // Token and cursor state is per send communicator, so a worker can arm and + // read its own connection without disturbing the others. + + ThreadResult WorkerCastSetTokens(void* sendComm, const std::vector& tokens) { + ThreadResult result; + if (ncclIbCastSetTokens(sendComm, tokens.data(), (int)tokens.size()) != ncclSuccess) { + result.ok = false; + result.msg = "ncclIbCastSetTokens failed"; + } + return result; + } + + ThreadResult WorkerCastGetSchedState(void* sendComm, struct ncclIbCastSchedState* out) { + ThreadResult result; + memset(out, 0, sizeof(*out)); + if (ncclIbCastGetSchedState(sendComm, out) != ncclSuccess) { + result.ok = false; + result.msg = "ncclIbCastGetSchedState failed"; + } + return result; + } + + // QPs the connection actually uses. NCCL_IB_QPS_PER_CONNECTION states only a + // request: on a merged device the plugin creates that many per member, so a + // count taken from the environment would not describe this connection, and a + // split threshold built from it would sit on the wrong side of the boundary. + // Valid once the scheduler is warm, which the first successful send does. + ThreadResult WorkerCastLiveNqps(void* sendComm, int* nqps) { + struct ncclIbCastSchedState state; + ThreadResult result = WorkerCastGetSchedState(sendComm, &state); + if (!result.ok) return result; + if (state.nqps <= 0) { + result.ok = false; + result.msg = "scheduler reports nqps=" + std::to_string(state.nqps); + return result; + } + *nqps = state.nqps; + return result; + } + + // Every CAST helper below measures and arms the sender's scheduler, because that is + // the only side the scheduler exists on. None of the signatures carry that, so the + // convention is named once here rather than left as a bare literal at each use: a + // future CAST body that sends from rank 0 would otherwise silently size its + // transfers against the peer's connection. + static constexpr int kCastSenderRank = 1; + static constexpr int kCastReceiverRank = 0; + + // The connection's live QP count, agreed across ranks: on a merged device the plugin + // creates the requested QPs per member, so an environment-derived split threshold is + // wrong, and a worker cannot broadcast the real one. + // 0 usable, 1 single queue pair (caller skips), -1 failure (reported here). + int ThreadedCastAgreedNqps(int dev, int* nqps) { + // The count describes the plugin's device configuration, not any one connection, + // so it cannot change between tests in the same process. Probing it once per CAST + // body per worker count rebuilt an entire connection -- listen, accept, connect, + // register, warm-up transfer, teardown -- for an answer that was already known. + // Cache it instead. Both ranks run the same test sequence, so they populate and + // hit this cache in lockstep and the collectives below stay matched; a failed + // probe is deliberately not cached, so a later caller re-probes rather than + // inheriting the failure. + static std::map agreedNqpsByDev; + const auto cached = agreedNqpsByDev.find(dev); + if (cached != agreedNqpsByDev.end()) { + *nqps = cached->second; + return (*nqps == 1) ? 1 : 0; + } + + void* listenComm = nullptr; + void* sendComm = nullptr; + void* recvComm = nullptr; + + // The probe brings up its own connection rather than calling + // SetupCastConnection, which asserts fatally and sends the handle only after + // its assertions: a listen that fails on rank 0 leaves rank 1 waiting in + // MPI_Recv forever, which is a hang instead of the failure this helper + // promises. Here the handle carries a status word, so both ranks agree before + // either one waits on the other. + const int rank = MPIEnvironment::world_rank; + const int peer = 1 - rank; + ncclNetHandle_t handle; + memset(&handle, 0, sizeof(handle)); + int localOk = 1; + + // ncclNetHandle_t is a char array, so the handshake copies rather than assigns. + struct ProbeHandshake { + int ok; + ncclNetHandle_t handle; + }; + + if (rank == kCastReceiverRank) { + if (CreateListenComm(dev, &handle, &listenComm) != ncclSuccess || !listenComm) + localOk = 0; + ProbeHandshake msg{}; + msg.ok = localOk; + memcpy(msg.handle, handle, sizeof(handle)); + MPI_Send(&msg, sizeof(msg), MPI_BYTE, peer, 0, MPI_COMM_WORLD); + if (localOk) { + for (int i = 0; i < kMaxRetryAttempts && recvComm == nullptr; i++) { + if (AcceptConnection(listenComm, &recvComm) != ncclSuccess) break; + if (!recvComm) usleep(kPollIntervalUs); + } + if (!recvComm) localOk = 0; + } + } else { + ProbeHandshake msg{}; + MPI_Recv(&msg, sizeof(msg), MPI_BYTE, peer, 0, MPI_COMM_WORLD, MPI_STATUS_IGNORE); + localOk = msg.ok; + if (localOk) { + memcpy(handle, msg.handle, sizeof(handle)); + for (int i = 0; i < kMaxRetryAttempts && sendComm == nullptr; i++) { + if (ConnectToRemote(dev, &handle, &sendComm) != ncclSuccess) break; + if (!sendComm) usleep(kPollIntervalUs); + } + if (!sendComm) localOk = 0; + } + } + + int bothUp = 0; + if (MPI_Allreduce(&localOk, &bothUp, 1, MPI_INT, MPI_MIN, MPI_COMM_WORLD) != MPI_SUCCESS) + bothUp = 0; + if (!bothUp) { + ADD_FAILURE() << "could not establish the probe connection used to agree the " + "connection's live QP count"; + TeardownConnection(recvComm, listenComm, sendComm, nullptr); + return -1; + } + + void* comm = (rank == kCastReceiverRank) ? recvComm : sendComm; + // A guarded allocation rather than a std::vector: if the warm-up below times out + // the buffer has to outlive this function, and a vector's storage cannot be + // retained -- clearing it frees exactly the memory that must not be freed. + static constexpr size_t kProbeBytes = 128; + void* probe = malloc(kProbeBytes); + // Agreed before either rank acts on it, for the same reason the registration + // below is: a rank that returned here on its own would walk into + // TeardownConnection's barrier while its peer was still inside the registration + // MPI_Allreduce, and the two would wait on each other instead of reporting + // anything. free(nullptr) is a no-op, so the failing rank can join the same exit. + int allocated = (probe != nullptr) ? 1 : 0; + int bothAllocated = 0; + if (MPI_Allreduce(&allocated, &bothAllocated, 1, MPI_INT, MPI_MIN, MPI_COMM_WORLD) + != MPI_SUCCESS) { + bothAllocated = 0; + } + if (!bothAllocated) { + ADD_FAILURE() << "allocating the probe buffer failed on at least one rank, so the " + "connection's live QP count could not be agreed"; + free(probe); + TeardownConnection(recvComm, listenComm, sendComm, nullptr); + return -1; + } + auto probeGuard = makeHostBufferAutoGuard(probe); + memset(probe, 0, kProbeBytes); + void* mhandle = nullptr; + const int registered = RegisterMemory(comm, probe, kProbeBytes, + NCCL_PTR_HOST, &mhandle) == ncclSuccess; + // Agreed before either side moves: a one-sided failure would otherwise send + // the failing rank into the teardown barrier while its peer waits inside + // GetActualNqps for traffic that is never coming, and the test would hang + // instead of reporting anything. TeardownConnection accepts a null handle. + int bothOk = 0; + if (MPI_Allreduce(®istered, &bothOk, 1, MPI_INT, MPI_MIN, MPI_COMM_WORLD) + != MPI_SUCCESS) { + bothOk = 0; + } + if (!bothOk) { + ADD_FAILURE() << "registering the probe buffer failed on at least one rank, so the " + "connection's live QP count could not be agreed"; + TeardownConnection(recvComm, listenComm, sendComm, mhandle); + return -1; + } + // Not GetActualNqps: its CastDoSendRecv asserts fatally and reports nothing about + // whether a timed-out request is still posted. A fatal assertion there returns from + // that helper alone, so execution would arrive back here and tear down the + // registration and free `probe` underneath a request the NIC can still write into -- + // the lifetime hazard the worker paths in this file exist to avoid. This warm-up + // reports instead, and the ranks agree on it before either one tears anything down. + bool outstanding = false; + int warmOk = 1; + { + void* req = nullptr; + void* bufs[1] = {probe}; + size_t sizes[1] = {kProbeBytes}; + int tags[1] = {1}; + void* handles[1] = {mhandle}; + if (rank == kCastReceiverRank) { + if (PostRecv(recvComm, 1, bufs, sizes, tags, handles, &req) != ncclSuccess + || req == nullptr) { + warmOk = 0; + } + } else { + for (int attempt = 0; attempt < kMaxRetryAttempts && req == nullptr; attempt++) { + if (PostSend(sendComm, probe, kProbeBytes, 1, mhandle, &req) + != ncclSuccess) { + break; + } + if (req == nullptr) usleep(kPollIntervalUs); + } + if (req == nullptr) warmOk = 0; + } + if (warmOk) { + int sz = 0; + if (WaitForCompletion(req, &sz, kConnectTimeoutMs) != ncclSuccess) { + warmOk = 0; + outstanding = true; // the wait timed out; the request is still posted + } + } + } + + int warmBothOk = 0; + if (MPI_Allreduce(&warmOk, &warmBothOk, 1, MPI_INT, MPI_MIN, MPI_COMM_WORLD) + != MPI_SUCCESS) { + warmBothOk = 0; + } + if (!warmBothOk) { + ADD_FAILURE() << "the probe warm-up transfer did not complete on at least one " + "rank, so the connection's live QP count could not be agreed"; + if (outstanding) { + // Deregistering now would hand the device freed memory. Leaking one 128-byte + // probe and its registration on a failing run is the cheaper mistake, and it + // is the same trade the worker retain paths make. + ADD_FAILURE() << "the probe buffer and its registration are retained, since " + "the warm-up request may still reference them"; + probeGuard.release(); + TeardownConnection(recvComm, listenComm, sendComm, nullptr); + } else { + TeardownConnection(recvComm, listenComm, sendComm, mhandle); + } + return -1; + } + + // Both sides completed, so the scheduler is warm and nothing references the probe. + // Read from the sender: the scheduler only exists on that side, which is why the + // count is broadcast from the same rank rather than computed independently. + if (rank == kCastSenderRank) { + struct ncclIbCastSchedState state = {}; + if (ncclIbCastGetSchedState(sendComm, &state) != ncclSuccess) { + ADD_FAILURE() << "ncclIbCastGetSchedState failed after the warm-up transfer"; + *nqps = 0; + } else { + *nqps = state.nqps; + } + } + MPI_Bcast(nqps, 1, MPI_INT, kCastSenderRank, MPI_COMM_WORLD); + TeardownConnection(recvComm, listenComm, sendComm, mhandle); + if (*nqps > 0) agreedNqpsByDev[dev] = *nqps; + // One queue pair is a legitimate configuration -- CastSingleQPBypassesWrr + // covers it -- but not one these branches can make claims about: the scheduler + // returns before split selection and token accounting when nqps is 1, so a + // token delta of 1 never appears and the split path is never taken. Reported + // separately from a failure so the caller can skip rather than fail. + if (*nqps == 1) return 1; + if (*nqps > 0) return 0; + ADD_FAILURE() << "the scheduler reported " << *nqps + << " queue pairs after a successful warm-up transfer"; + return -1; + } + + // Warm the scheduler up (it initializes on the first send) and arm equal weights. + // nqps is the live count ThreadedCastAgreedNqps agreed on the main thread, so both + // ranks size their transfers the same way and merged devices work; the sender + // confirms its own connection reports the same count. + // totTokens is the ledger to arm. Callers whose audit asserts the ledger ends empty + // pass the number of sub-threshold sends they make, so the two cannot drift into a + // false "the sends did not take the WRR path"; the serial bodies couple them the + // same way with their own kTotTokens. + ThreadResult WorkerCastPrepareTokens(int rank, ConnectionPair& pair, void* buffer, + void* mhandle, int nqps, int tag, int seed, + int totTokens = 100, bool* outstanding = nullptr) { + ThreadResult result = + WorkerSendRecvPattern(rank, pair, buffer, 64, tag, mhandle, seed, + kDefaultTimeoutMs, outstanding); + if (!result.ok || rank != kCastSenderRank) return result; + + // The expectations are built from the agreed live count, so this checks that + // this worker's own connection reports the same thing; a mismatch means the + // expectations belong to a different connection than the one carrying data. + // This is also what enforces the kCastSenderRank convention at run time: the + // agreed count comes from the sender's connection, and a body that sent from + // the other rank would arm a scheduler whose nqps this check does not match. + int liveNqps = 0; + result = WorkerCastLiveNqps(pair.sendComm, &liveNqps); + if (!result.ok) return result; + if (liveNqps != nqps) { + result.ok = false; + result.msg = "this worker's connection reports nqps=" + std::to_string(liveNqps) + + " but the run agreed on " + std::to_string(nqps); + return result; + } + // Scope of what arming the tokens here buys, because it is narrower than it + // looks. ncclIbCastSetTokens fills the token ledger and sets qpTxSchedInit, but + // it does not touch qpTxSched[].weight -- scheduler.cc writes those only from + // IbCastQpSchedUpdateTx, which fires off the RTT timer. The threaded CAST suites + // pin RCCL_IB_QP_SCHED_UPDATE_INTERVAL to 10 s precisely so that timer cannot + // rewrite the ledger mid-assertion, so in those runs it never fires at all and + // every weight stays 0. With zero weights the split branch in p2p.cc:209-216 + // computes a zero-length chunk for every QP but the last, which then takes the + // whole payload: the data still arrives and still verifies, and the split path + // is still the one selected (that is what the 0-token deltas prove), but nothing + // is actually striped across QPs. + // + // So these branches cover split-vs-WRR path *selection* and data integrity, not + // striping, and not the RTT/weight path the serial bodies reach via their + // "initQpTokens were rewritten" checks. Closing that needs the scheduler weights + // (or per-QP tx counters) exposed through net_ib_cast_inspect.h plus a + // short-interval threaded suite -- a transport-side change, out of scope for a + // test-only branch. Tracked as follow-up; do not read the split coverage below + // as proof that concurrent senders stripe correctly. + return WorkerCastSetTokens(pair.sendComm, EqualTokens(liveNqps, totTokens)); + } + + // Worker-safe CAST transfer with a token-consumption expectation. Only the + // sender owns scheduler state, so it checks the delta while the receiver + // verifies the payload. Pass a negative delta to skip the token check. + ThreadResult WorkerCastTransferExpectTokens(int rank, ConnectionPair& pair, void* buffer, + size_t size, int tag, void* mhandle, int seed, + int expectedTokenDelta, + int timeoutMs = kLargeTransferTimeoutMs, + bool* outstanding = nullptr) { + struct ncclIbCastSchedState before = {}; + ThreadResult result; + if (rank == kCastSenderRank) { + result = WorkerCastGetSchedState(pair.sendComm, &before); + if (!result.ok) return result; + } + + result = WorkerSendRecvPattern(rank, pair, buffer, size, tag, mhandle, seed, timeoutMs, + outstanding); + if (!result.ok) return result; + + if (rank == kCastSenderRank && expectedTokenDelta >= 0) { + struct ncclIbCastSchedState after = {}; + result = WorkerCastGetSchedState(pair.sendComm, &after); + if (!result.ok) return result; + const int delta = before.activeTotTokens - after.activeTotTokens; + if (delta != expectedTokenDelta) { + result.ok = false; + result.msg = "expected a WRR token delta of " + + std::to_string(expectedTokenDelta) + " at size " + + std::to_string(size) + ", observed " + std::to_string(delta); + } + } + return result; + } + ncclResult_t InitNetIbCtx(void** ctxOut) { ncclNetCommConfig_t commConfig = {}; commConfig.trafficClass = NCCL_NET_TRAFFIC_CLASS_UNDEF; From 73af111236ff1a1ebcbd0307aacc984ae714ba90 Mon Sep 17 00:00:00 2001 From: Ilya Kosarev Date: Wed, 16 Sep 2026 13:30:04 +0000 Subject: [PATCH 2/4] net-ib: let a threaded size sweep register per step, not only once RunThreadedSizeSweep registered one buffer covering its largest step and reused that handle for every step. That is the right shape for some tests and the wrong one for the fused-device sweeps this branch adds: on a merged device every registration fans out across both members' protection domains and MR caches, which is the thing under test, and registering sub-ranges of one block keeps handing back the same cache entry once a covering registration is live -- so the churn never happens. SweepRegistration::PerSize allocates and registers each step fresh and releases it at the end of that step; Once keeps the previous behaviour and stays the default. A threaded branch has to match whichever shape its serial body uses or it quietly covers less, which is why this is a parameter rather than a change of default. The timeout cleanup picks up whichever holder owns the step -- the fresh one under PerSize, the sweep-wide one under Once -- so a request still in flight cannot have its memory released underneath it either way. --- .../transport/NetIbMPI/NetIbMPITestBase.hpp | 56 +++++++++++++++---- 1 file changed, 46 insertions(+), 10 deletions(-) diff --git a/projects/rccl/test/transport/NetIbMPI/NetIbMPITestBase.hpp b/projects/rccl/test/transport/NetIbMPI/NetIbMPITestBase.hpp index 639293ecb890..f7653da10d15 100644 --- a/projects/rccl/test/transport/NetIbMPI/NetIbMPITestBase.hpp +++ b/projects/rccl/test/transport/NetIbMPI/NetIbMPITestBase.hpp @@ -2241,14 +2241,30 @@ class NetIbMPITest : public MPITestBase { AssertNoRdmaLeaks(before, CaptureRdmaResources(), label); } + void RunThreadedBody(int dev, int nThreads, const char* label, + std::function body) { + RunThreadedBody(ThreadDevPolicy::Fixed(dev), nThreads, label, std::move(body)); + } + + // How a threaded size sweep gets its memory. On a fused device the PerSize shape is + // the point of the test, since every registration fans out across both members' + // protection domains and caches. The allocation churns with it, as the serial body + // does: registering sub-ranges of one block would keep handing back the same cache + // entry once a covering registration were live. The threaded branch has to match + // whichever its serial body does, or it quietly covers less. + enum class SweepRegistration { Once, PerSize }; + // Threaded size sweep: every worker walks the steps that fit its share of the - // registration budget, on its own connection and with a per-worker payload seed, so - // a transfer delivered on the wrong connection fails verification. Registers one - // buffer covering the largest step it will run and reuses that handle. Wraps the run + // registration budget, on its own connection and with a per-worker payload seed, so a + // transfer delivered on the wrong connection fails verification. Memory comes from + // `registration`: Once registers a buffer covering the largest step it will run and + // reuses that handle, PerSize allocates and registers each step fresh. Wraps the run // in an RDMA resource leak check. + void RunThreadedSizeSweep(ThreadDevPolicy policy, int nThreads, const std::vector& sizes, int repeats, - const char* label) { + const char* label, + SweepRegistration registration = SweepRegistration::Once) { const int rank = MPIEnvironment::world_rank; // Capped by worker count, the way MemoryRegistrationStorm holds its own @@ -2270,11 +2286,15 @@ class NetIbMPITest : public MPITestBase { RunThreadedBody( policy, nThreads, label, [&](int threadIdx, ConnectionPair& pair) -> ThreadResult { - WorkerHostBuffer h = WorkerSetupHostBuffer(rank, pair, maxSize); - if (!h.result.ok) return h.result; - void* buffer = h.buffer; - void* mhandle = h.mhandle; - ThreadResult result; + const bool perSize = (registration == SweepRegistration::PerSize); + + // Once: one allocation and one registration covering the largest + // step, reused by all of them. + WorkerHostBuffer whole; + if (!perSize) { + whole = WorkerSetupHostBuffer(rank, pair, maxSize); + if (!whole.result.ok) return whole.result; + } // One pattern for this worker across the whole sweep. Varying it per // step aliased modulo 256 -- WorkerSeed(0, 8) and WorkerSeed(8, 0) are @@ -2284,8 +2304,19 @@ class NetIbMPITest : public MPITestBase { // and the payload check. The tag and the expected size identify the // step; the payload identifies the worker. const int workerPattern = WorkerSeed(threadIdx, 0); + ThreadResult result; int tag = 0; for (size_t size : steps) { + // PerSize: allocated and registered fresh for this step and + // released at the end of it, so the next step starts over. + WorkerHostBuffer step; + if (perSize) { + step = WorkerSetupHostBuffer(rank, pair, size); + if (!step.result.ok) return step.result; + } + void* buffer = perSize ? step.buffer : whole.buffer; + void* mhandle = perSize ? step.mhandle : whole.mhandle; + for (int repeat = 0; repeat < repeats; repeat++) { const int timeout = (size > 1024 * 1024) ? kLargeTransferTimeoutMs : kDefaultTimeoutMs; @@ -2293,7 +2324,12 @@ class NetIbMPITest : public MPITestBase { result = WorkerSendRecvPattern(rank, pair, buffer, size, tag, mhandle, workerPattern, timeout, &outstanding); if (!result.ok) { - return outstanding ? WorkerRetainHostBuffer(result, h) : result; + // Whichever holder owns this step's registration is the one to + // retain -- the same choice buffer/mhandle make above. Splitting + // the old single `h` into these two left this site naming a + // variable that no longer exists. + WorkerHostBuffer& held = perSize ? step : whole; + return outstanding ? WorkerRetainHostBuffer(result, held) : result; } tag++; } From 94d60590b2b8728e594f5ea4ba245f9e0bab7bc2 Mon Sep 17 00:00:00 2001 From: Ilya Kosarev Date: Wed, 16 Sep 2026 13:30:34 +0000 Subject: [PATCH 3/4] net-ib: run the IB-CAST scheduler and fused-device tests at several worker counts Threaded bodies for the CAST scheduler tests (token distribution, the WRR/split boundary, the size ladder, zero-size, the large transfer, the multi-round stress) and for the fused-device tests (registration cycling, repeated flush, sequential transfers, the vNIC size sweep). Each worker drives its own connection with a per-worker payload pattern, so a transfer delivered on the wrong connection fails verification rather than passing quietly. Two things the bodies have to get right that the serial versions do not: The payload pattern is per worker and constant across the phase, not folded from the size or the message index. makeBytePattern keeps only the seed modulo 256 and the worker stride is small, so a per-message seed collides across workers -- one worker's message 0 and another's message 95 produce the same byte -- and with every worker sending the same size the size check cannot separate them either. Freshness per message still holds because the receiver clears its buffer first. Token-delta assertions need the RTT-driven ledger update suspended: with several workers the wall-clock gap between the before and after snapshots is wide enough for the timer to rewrite the ledger underneath them, so those bodies require a long update interval and skip otherwise. Every transfer captures the outstanding signal and retains its buffer and registration on a timeout, since a timed-out request is still live and the guards would otherwise free memory the NIC can write into. NicFusionTests needs three shapes of that: the shared cross-worker allocation in RegDeregCycling_VNic is dismissed from inside the worker, FlushRepeated_VNic retains its per-worker GPU buffer and registration directly, and SequentialTransfers_VNic uses the plain WorkerHostBuffer path. --- .../test/transport/NetIbMPI/CastTests.cpp | 376 +++++++++++++++++- .../transport/NetIbMPI/NicFusionTests.cpp | 282 ++++++++++++- 2 files changed, 647 insertions(+), 11 deletions(-) diff --git a/projects/rccl/test/transport/NetIbMPI/CastTests.cpp b/projects/rccl/test/transport/NetIbMPI/CastTests.cpp index 04a871ec9707..e14267e7de97 100644 --- a/projects/rccl/test/transport/NetIbMPI/CastTests.cpp +++ b/projects/rccl/test/transport/NetIbMPI/CastTests.cpp @@ -8,6 +8,11 @@ #include "NetIbCastInspect.hpp" #include +// ThreadedCastAgreedNqps returns this when the connection uses one queue pair: real, +// but the scheduler then returns before split selection and token accounting, so the +// threaded branches skip rather than assert. +static constexpr int kThreadedNqpsSingleQp = 1; + #ifdef MPI_TESTS_ENABLED // ============================================================================= @@ -590,6 +595,65 @@ TEST_F(NetIbMPITest, CastSplitDataThresholdBoundary) { net_ = &netIbCast; AssertInitAndGetDevices(nullptr); + // Parameterized by MPIEnvironment::nThreads. Scheduler state is per send + // communicator, so every worker arms and audits its own tokens while N + // schedulers run the boundary decision concurrently. + if (MPIEnvironment::nThreads > 1) { + // Token-delta assertions need the RTT-driven update suspended: with + // several workers the wall-clock gap between the before and after + // snapshots is wide enough for the timer to rewrite the ledger. + CAST_REQUIRE_UPDATE_INTERVAL_OR_SKIP(10000000); + // The live count, agreed across ranks: on a merged device the plugin creates + // the requested QPs per member, so an environment-derived threshold is wrong. + int nqps = 0; + // The helper reports failures itself; both ranks exit together. One queue pair + // is not a failure but bypasses split selection and token accounting. + const int nqpsStatus = ThreadedCastAgreedNqps(/*dev=*/0, &nqps); + if (nqpsStatus == kThreadedNqpsSingleQp) + GTEST_SKIP() << "the connection uses a single queue pair, which bypasses WRR and " + "split selection; CastSingleQPBypassesWrr covers that case"; + if (nqpsStatus != 0) return; + const size_t splitDataMin = GetSplitDataMin(); + if (splitDataMin == 0) + GTEST_SKIP() << "RCCL_IB_QP_SCHED_SPLIT_DATA_MIN=0: no WRR/split boundary exists"; + const size_t threshold = splitDataMin * static_cast(nqps); + // WorkerCastPrepareTokens warms up with 64 bytes, which a small threshold would + // otherwise put past the allocation. + const size_t bufSize = std::max(64, threshold); + + RunThreadedBody( + 0, MPIEnvironment::nThreads, "threaded CastSplitDataThresholdBoundary", + [&](int threadIdx, ConnectionPair& pair) -> ThreadResult { + WorkerHostBuffer host = WorkerSetupHostBuffer(rank, pair, bufSize); + if (!host.result.ok) return host.result; + void* buffer = host.buffer; + void* mhandle = host.mhandle; + ThreadResult result; + + bool outstanding = false; + const int seed = WorkerSeed(threadIdx, 1599); + result = WorkerCastPrepareTokens(rank, pair, buffer, mhandle, nqps, 1599, seed, + /*totTokens=*/100, &outstanding); + if (!result.ok) + return outstanding ? WorkerRetainHostBuffer(result, host) : result; + + // At the threshold the split path must leave WRR tokens alone. + result = WorkerCastTransferExpectTokens(rank, pair, buffer, threshold, 1600, + mhandle, seed + 1, /*tokens=*/0, + kLargeTransferTimeoutMs, &outstanding); + if (!result.ok) + return outstanding ? WorkerRetainHostBuffer(result, host) : result; + + // One byte below it, WRR must consume exactly one token. + result = WorkerCastTransferExpectTokens(rank, pair, buffer, threshold - 1, 1601, + mhandle, seed + 2, /*tokens=*/1, + kLargeTransferTimeoutMs, &outstanding); + if (!result.ok && outstanding) return WorkerRetainHostBuffer(result, host); + return result; + }); + return; + } + void* listenComm = nullptr; void* sendComm = nullptr; void* recvComm = nullptr; @@ -1029,6 +1093,83 @@ TEST_F(NetIbMPITest, CastSendRecvMultipleSizes) { net_ = &netIbCast; AssertInitAndGetDevices(nullptr); + // Parameterized by MPIEnvironment::nThreads: N schedulers sweep both sides of + // the WRR/split boundary at once, each auditing its own token ledger. + if (MPIEnvironment::nThreads > 1) { + // Token-delta assertions need the RTT-driven update suspended: with + // several workers the wall-clock gap between the before and after + // snapshots is wide enough for the timer to rewrite the ledger. + CAST_REQUIRE_UPDATE_INTERVAL_OR_SKIP(10000000); + // The live count, agreed across ranks, not the environment's request: on a + // merged device the plugin creates that many QPs per member, so a threshold + // derived from the environment sits on the wrong side of the split boundary. + int nqps = 0; + // The helper reports a failure itself; both ranks take this exit together. A + // single queue pair is not a failure but is not testable here either: the + // scheduler returns before split selection and token accounting, so the token + // deltas these branches expect never appear. + const int nqpsStatus = ThreadedCastAgreedNqps(/*dev=*/0, &nqps); + if (nqpsStatus == kThreadedNqpsSingleQp) + GTEST_SKIP() << "the connection uses a single queue pair, which bypasses WRR and " + "split selection; CastSingleQPBypassesWrr covers that case"; + if (nqpsStatus != 0) return; + const size_t splitDataMin = GetSplitDataMin(); + if (splitDataMin == 0) + GTEST_SKIP() << "RCCL_IB_QP_SCHED_SPLIT_DATA_MIN=0: no WRR/split boundary exists"; + const size_t threshold = splitDataMin * static_cast(nqps); + // Floored at the 64 bytes the token warm-up transfers, for the same reason. + const size_t bufSize = std::max(64, threshold * 2); + + std::vector wrrSizes; + for (size_t size : std::initializer_list{512, 4096, splitDataMin, threshold - 1}) + if (size > 0 && size < threshold) wrrSizes.push_back(size); + // Sorted first: std::unique only collapses *adjacent* equals, and this list is + // built in a fixed order, not a sorted one. splitDataMin can equal 512 or 4096 + // without being next to it, so the duplicate survived and that size ran twice. + std::sort(wrrSizes.begin(), wrrSizes.end()); + wrrSizes.erase(std::unique(wrrSizes.begin(), wrrSizes.end()), wrrSizes.end()); + const std::vector splitSizes = {threshold, threshold * 2}; + + RunThreadedBody( + 0, MPIEnvironment::nThreads, "threaded CastSendRecvMultipleSizes", + [&](int threadIdx, ConnectionPair& pair) -> ThreadResult { + WorkerHostBuffer host = WorkerSetupHostBuffer(rank, pair, bufSize); + if (!host.result.ok) return host.result; + void* buffer = host.buffer; + void* mhandle = host.mhandle; + ThreadResult result; + + bool outstanding = false; + const int seedBase = WorkerSeed(threadIdx, 1999); + result = WorkerCastPrepareTokens(rank, pair, buffer, mhandle, nqps, 1999, + seedBase, /*totTokens=*/100, &outstanding); + if (!result.ok) + return outstanding ? WorkerRetainHostBuffer(result, host) : result; + + int tag = 2000; + for (size_t size : wrrSizes) { + const int stepTag = tag++; + result = WorkerCastTransferExpectTokens(rank, pair, buffer, size, stepTag, + mhandle, seedBase + stepTag, + /*tokens=*/1, kLargeTransferTimeoutMs, + &outstanding); + if (!result.ok) + return outstanding ? WorkerRetainHostBuffer(result, host) : result; + } + for (size_t size : splitSizes) { + const int stepTag = tag++; + result = WorkerCastTransferExpectTokens(rank, pair, buffer, size, stepTag, + mhandle, seedBase + stepTag, + /*tokens=*/0, kLargeTransferTimeoutMs, + &outstanding); + if (!result.ok) + return outstanding ? WorkerRetainHostBuffer(result, host) : result; + } + return result; + }); + return; + } + void* listenComm = nullptr; void* sendComm = nullptr; void* recvComm = nullptr; @@ -1069,7 +1210,9 @@ TEST_F(NetIbMPITest, CastSendRecvMultipleSizes) { if (sz > 0 && sz < kThreshold && sz <= kBufSz) kWrrSizes.push_back(sz); } - // Deduplicate (kSplitDataMin or 512 might equal kThreshold-1). + // Deduplicate (kSplitDataMin or 512 might equal kThreshold-1). Sorted first because + // std::unique only collapses adjacent equals and this list is built in a fixed order. + std::sort(kWrrSizes.begin(), kWrrSizes.end()); kWrrSizes.erase(std::unique(kWrrSizes.begin(), kWrrSizes.end()), kWrrSizes.end()); // Sizes at/above threshold → split path (0 tokens consumed). const std::vector kSplitSizes = {kThreshold, kThreshold * 2}; @@ -1140,6 +1283,51 @@ TEST_F(NetIbMPITest, CastLargeTransfer) { net_ = &netIbCast; AssertInitAndGetDevices(nullptr); + // Parameterized by MPIEnvironment::nThreads: concurrent 16 MB split-path + // transfers, each worker confirming its own scheduler stayed untouched. + if (MPIEnvironment::nThreads > 1) { + // Token-delta assertions need the RTT-driven update suspended: with + // several workers the wall-clock gap between the before and after + // snapshots is wide enough for the timer to rewrite the ledger. + CAST_REQUIRE_UPDATE_INTERVAL_OR_SKIP(10000000); + // The live count, agreed across ranks, not the environment's request: on a + // merged device the plugin creates that many QPs per member, so a threshold + // derived from the environment sits on the wrong side of the split boundary. + int nqps = 0; + // The helper reports a failure itself; both ranks take this exit together. A + // single queue pair is not a failure but is not testable here either: the + // scheduler returns before split selection and token accounting, so the token + // deltas these branches expect never appear. + const int nqpsStatus = ThreadedCastAgreedNqps(/*dev=*/0, &nqps); + if (nqpsStatus == kThreadedNqpsSingleQp) + GTEST_SKIP() << "the connection uses a single queue pair, which bypasses WRR and " + "split selection; CastSingleQPBypassesWrr covers that case"; + if (nqpsStatus != 0) return; + RunThreadedBody( + 0, MPIEnvironment::nThreads, "threaded CastLargeTransfer", + [&](int threadIdx, ConnectionPair& pair) -> ThreadResult { + WorkerHostBuffer host = WorkerSetupHostBuffer(rank, pair, kLargeBufferSize); + if (!host.result.ok) return host.result; + void* buffer = host.buffer; + void* mhandle = host.mhandle; + ThreadResult result; + + bool outstanding = false; + const int seed = WorkerSeed(threadIdx, 2099); + result = WorkerCastPrepareTokens(rank, pair, buffer, mhandle, nqps, 2099, seed, + /*totTokens=*/100, &outstanding); + if (!result.ok) + return outstanding ? WorkerRetainHostBuffer(result, host) : result; + + result = WorkerCastTransferExpectTokens(rank, pair, buffer, kLargeBufferSize, + 2100, mhandle, seed + 1, /*tokens=*/0, + kLargeTransferTimeoutMs, &outstanding); + if (!result.ok && outstanding) return WorkerRetainHostBuffer(result, host); + return result; + }); + return; + } + void* listenComm = nullptr; void* sendComm = nullptr; void* recvComm = nullptr; @@ -1206,6 +1394,52 @@ TEST_F(NetIbMPITest, CastSendRecvZeroSize) { net_ = &netIbCast; AssertInitAndGetDevices(nullptr); + // Parameterized by MPIEnvironment::nThreads: a zero-byte send must still take + // the WRR path on every worker's own scheduler. + if (MPIEnvironment::nThreads > 1) { + // Token-delta assertions need the RTT-driven update suspended: with + // several workers the wall-clock gap between the before and after + // snapshots is wide enough for the timer to rewrite the ledger. + CAST_REQUIRE_UPDATE_INTERVAL_OR_SKIP(10000000); + // The live count, agreed across ranks, not the environment's request: on a + // merged device the plugin creates that many QPs per member, so a threshold + // derived from the environment sits on the wrong side of the split boundary. + int nqps = 0; + // The helper reports a failure itself; both ranks take this exit together. A + // single queue pair is not a failure but is not testable here either: the + // scheduler returns before split selection and token accounting, so the token + // deltas these branches expect never appear. + const int nqpsStatus = ThreadedCastAgreedNqps(/*dev=*/0, &nqps); + if (nqpsStatus == kThreadedNqpsSingleQp) + GTEST_SKIP() << "the connection uses a single queue pair, which bypasses WRR and " + "split selection; CastSingleQPBypassesWrr covers that case"; + if (nqpsStatus != 0) return; + RunThreadedBody( + 0, MPIEnvironment::nThreads, "threaded CastSendRecvZeroSize", + [&](int threadIdx, ConnectionPair& pair) -> ThreadResult { + const size_t regSize = 64; + WorkerHostBuffer host = WorkerSetupHostBuffer(rank, pair, regSize); + if (!host.result.ok) return host.result; + void* buffer = host.buffer; + void* mhandle = host.mhandle; + ThreadResult result; + + bool outstanding = false; + const int seed = WorkerSeed(threadIdx, 2199); + result = WorkerCastPrepareTokens(rank, pair, buffer, mhandle, nqps, 2199, seed, + /*totTokens=*/100, &outstanding); + if (!result.ok) + return outstanding ? WorkerRetainHostBuffer(result, host) : result; + + result = WorkerCastTransferExpectTokens(rank, pair, buffer, 0, 2200, mhandle, + seed + 1, /*tokens=*/1, + kLargeTransferTimeoutMs, &outstanding); + if (!result.ok && outstanding) return WorkerRetainHostBuffer(result, host); + return result; + }); + return; + } + void* listenComm = nullptr; void* sendComm = nullptr; void* recvComm = nullptr; @@ -1280,6 +1514,146 @@ TEST_F(NetIbMPITest, CastStressMultiRoundTwoConns) { net_ = &netIbCast; AssertInitAndGetDevices(nullptr); + // Parameterized by MPIEnvironment::nThreads. The single-threaded body already + // drives many connections, but strictly one after another; here every worker + // owns a connection so the schedulers advance simultaneously. Each worker + // audits its own ledger: tokens are per send communicator. + if (MPIEnvironment::nThreads > 1) { + if (GetSplitDataMin() == 0) + GTEST_SKIP() << "RCCL_IB_QP_SCHED_SPLIT_DATA_MIN=0: no WRR/split boundary exists"; + // As the other four threaded branches do: a short update interval refills + // tokens mid-phase, and the audit below then reads a ledger the sends did + // not produce. + // + // The cost is that it suspends IbCastQpSchedUpdateTx outright, so the RTT + // assertions the serial body makes below -- initQpTokens no longer holding the + // seeded 70/30, i.e. "the RTT timer fired" -- have no counterpart here, and the + // scheduler weights stay 0 for the whole run. That path has no threaded coverage + // today; restoring it needs a separate short-interval threaded suite, which + // cannot reuse these token audits. See WorkerCastPrepareTokens for the full + // scope note. + CAST_REQUIRE_UPDATE_INTERVAL_OR_SKIP(10000000); + + // The live count, agreed across ranks, not the environment's request: on a + // merged device the plugin creates that many QPs per member, so a threshold + // derived from the environment sits on the wrong side of the split boundary. + int nqps = 0; + // The helper reports a failure itself; both ranks take this exit together. A + // single queue pair is not a failure but is not testable here either: the + // scheduler returns before split selection and token accounting, so the token + // deltas these branches expect never appear. + const int nqpsStatus = ThreadedCastAgreedNqps(/*dev=*/0, &nqps); + if (nqpsStatus == kThreadedNqpsSingleQp) + GTEST_SKIP() << "the connection uses a single queue pair, which bypasses WRR and " + "split selection; CastSingleQPBypassesWrr covers that case"; + if (nqpsStatus != 0) return; + const size_t splitDataMin = GetSplitDataMin(); + // Split is taken when size / liveNqps >= splitDataMin, so the boundary scales + // with the live count; a ramp built from splitDataMin alone never crosses it. + const size_t threshold = splitDataMin * static_cast(nqps); + if (threshold < 2) + GTEST_SKIP() << "live split threshold is " << threshold + << " bytes: there is no WRR band below it to stress"; + // One byte under the threshold and not floored: a 64-byte floor would push it + // past the threshold when the threshold is small. Only the buffer is floored. + const size_t msgSize = threshold - 1; + const std::vector rampSizes = {threshold / 4, threshold / 2, + threshold, threshold * 2}; + const size_t bufSize = std::max(64, std::max(msgSize, threshold * 2)); + static constexpr int kThreadedMsgs = 100; + static constexpr int kThreadedRampRounds = 5; + + RunThreadedBody( + 0, MPIEnvironment::nThreads, "threaded CastStressMultiRoundTwoConns", + [&](int threadIdx, ConnectionPair& pair) -> ThreadResult { + WorkerHostBuffer host = WorkerSetupHostBuffer(rank, pair, bufSize); + if (!host.result.ok) return host.result; + void* buffer = host.buffer; + void* mhandle = host.mhandle; + ThreadResult result; + + bool outstanding = false; + const int tagBase = 10000 + threadIdx * 200; + const int seedBase = WorkerSeed(threadIdx, 3999); + result = WorkerCastPrepareTokens(rank, pair, buffer, mhandle, nqps, tagBase, + seedBase, /*totTokens=*/kThreadedMsgs, + &outstanding); + if (!result.ok) + return outstanding ? WorkerRetainHostBuffer(result, host) : result; + + // Phase 1: small messages, all on the WRR path. One pattern for the + // whole phase rather than seedBase + i: makeBytePattern keeps the seed + // modulo 256, and the worker stride is 161 there, so a per-message seed + // collides across workers -- worker t at message 0 and worker t+1 at + // message 95 produce the same byte. Every worker sends the same msgSize, + // so the size check cannot separate the pair either, and a payload + // delivered on the wrong connection would verify clean. Freshness per + // message survives because the receiver is cleared before each transfer. + for (int i = 0; i < kThreadedMsgs; i++) { + result = WorkerSendRecvPattern(rank, pair, buffer, msgSize, tagBase + 1 + i, + mhandle, seedBase, kDefaultTimeoutMs, + &outstanding); + if (!result.ok) + return outstanding ? WorkerRetainHostBuffer(result, host) : result; + } + + if (rank == 1) { + struct ncclIbCastSchedState state = {}; + result = WorkerCastGetSchedState(pair.sendComm, &state); + if (!result.ok) return result; + if (!state.schedInit) { + result.ok = false; + result.msg = "scheduler never initialized after the WRR phase"; + return result; + } + int sum = 0; + for (int qp = 0; qp < state.nqps; qp++) sum += state.activeQpTokens[qp]; + if (sum != state.activeTotTokens) { + result.ok = false; + result.msg = "per-QP tokens do not sum to activeTotTokens"; + return result; + } + if (state.activeTotTokens < 0 + || state.activeTotTokens > state.initTotTokens) { + result.ok = false; + result.msg = "activeTotTokens left the [0, initTotTokens] range"; + return result; + } + // Exact, because everything above holds whether or not WRR ran: + // schedInit is set unconditionally, and the sum and the range hold + // both at 0 and at an untouched full ledger. This phase arms + // exactly kThreadedMsgs tokens against exactly that many + // sub-threshold sends and the refill only fires on the next one, so + // a WRR path that ran leaves none and one that did not leaves all. + if (state.activeTotTokens != 0) { + result.ok = false; + result.msg = "the WRR phase left " + + std::to_string(state.activeTotTokens) + " of " + + std::to_string(state.initTotTokens) + + " tokens unspent, so the sends did not take the " + "WRR path"; + return result; + } + } + + // Phase 2: ramp across the WRR/split boundary. + int tag = tagBase + 1 + kThreadedMsgs; + for (int round = 0; round < kThreadedRampRounds; round++) { + for (size_t size : rampSizes) { + if (size == 0 || size > bufSize) continue; + const int stepTag = tag++; + result = WorkerSendRecvPattern(rank, pair, buffer, size, stepTag, mhandle, + seedBase + stepTag, + kLargeTransferTimeoutMs, &outstanding); + if (!result.ok) + return outstanding ? WorkerRetainHostBuffer(result, host) : result; + } + } + return result; + }); + return; + } + constexpr int kNConns = 100; std::vector listenComms(kNConns, nullptr); std::vector sendComms(kNConns, nullptr); diff --git a/projects/rccl/test/transport/NetIbMPI/NicFusionTests.cpp b/projects/rccl/test/transport/NetIbMPI/NicFusionTests.cpp index 4cf7cdcd7bab..c97593c9b366 100644 --- a/projects/rccl/test/transport/NetIbMPI/NicFusionTests.cpp +++ b/projects/rccl/test/transport/NetIbMPI/NicFusionTests.cpp @@ -615,6 +615,77 @@ TEST_F(NetIbMPITest, RegDeregCycling_VNic) { << "Failed to create fused vNIC from devices 0 and 1"; ASSERT_GE(vdev, 0); + // Parameterized by MPIEnvironment::nThreads. Every regMr on a fused device + // fans out across both members' protection domains and MR caches, so + // concurrent cycling on one address doubles the contended surface. The vNIC + // itself is created once on the main thread: the merged-device table is + // process-global. + const int nThreads = MPIEnvironment::nThreads; + if (nThreads > 1) { + // One region, cycled by every worker so they contend on a single MR cache + // entry, but sliced for the traffic that follows: the serial body moves + // data through the address it just recycled, and a shared payload buffer + // would have the workers overwriting each other's verification. + const size_t slotSize = kSmallBufferSize; + const size_t sharedSize = slotSize * nThreads; + auto shared = makeHostBufferAutoGuard(malloc(sharedSize)); + ASSERT_NE(shared.get(), nullptr); + // Workers cannot dismiss the shared guard themselves: several can time out at + // once and AutoGuard::release() writes a plain bool, so concurrent calls would + // be a data race. They raise this instead, and the guard is dismissed once on + // this thread after RunThreadedBody has joined them all. + std::atomic retainShared{false}; + + RunThreadedBody( + ThreadDevPolicy::Fixed(vdev), nThreads, "threaded RegDeregCycling_VNic", + [&](int threadIdx, ConnectionPair& pair) -> ThreadResult { + ThreadResult result; + void* workerComm = (rank == 0) ? pair.recvComm : pair.sendComm; + for (int i = 0; i < 50; i++) { + void* mhandle = nullptr; + result = WorkerRegister(workerComm, shared.get(), sharedSize, + NCCL_PTR_HOST, &mhandle); + if (!result.ok) return result; + if (DeregisterMemory(workerComm, mhandle) != ncclSuccess) { + result.ok = false; + result.msg = "deregMr failed during vNIC cache cycling"; + return result; + } + } + + void* mhandle = nullptr; + result = WorkerRegister(workerComm, shared.get(), sharedSize, NCCL_PTR_HOST, + &mhandle); + if (!result.ok) return result; + NetMHandleWorkerGuard mhandleGuard(mhandle, + NetMHandleWorkerDeleter(net_, workerComm)); + + void* slot = static_cast(shared.get()) + threadIdx * slotSize; + bool outstanding = false; + result = WorkerSendRecvPattern(rank, pair, slot, slotSize, 530, mhandle, + WorkerSeed(threadIdx, 0), kDefaultTimeoutMs, + &outstanding); + if (!result.ok && outstanding) { + // shared is one allocation split into per-worker slots and outlives + // every worker thread, so a timed-out request here means a live NIC + // write into memory the whole test -- not just this worker -- is + // about to free. Dismissing it leaks the region rather than freeing + // memory another rank's hardware can still touch. + result.msg += "; the shared buffer and this worker's registration are " + "retained, since the request may still reference them"; + // This worker's own registration is its own to dismiss; the shared + // allocation is flagged for the main thread. + mhandleGuard.release(); + retainShared.store(true, std::memory_order_release); + } + return result; + }); + // After the join, so this is the only thread touching the guard. Leaks the one + // region rather than freeing memory a timed-out request may still be written into. + if (retainShared.load(std::memory_order_acquire)) shared.release(); + return; + } + ConnectionPair pair; NetConnectionGuard connGuard(net_); SetupConnectionWithGuard(vdev, pair, connGuard); @@ -694,6 +765,25 @@ TEST_F(NetIbMPITest, LargeTransfer_VNic) { << "Failed to create fused vNIC from devices 0 and 1"; ASSERT_GE(vdev, 0); + // Parameterized by MPIEnvironment::nThreads: several striped transfers run at + // once over the fused device. The per-worker size shrinks with the worker + // count: every registration on a fused device is pinned once per member, so + // 64 MB per worker exhausts registration resources at high worker counts + // while adding nothing to the striping coverage. + if (MPIEnvironment::nThreads > 1) { + const size_t threadedSize = + std::max(4 * 1024 * 1024, + (64 * 1024 * 1024) / MPIEnvironment::nThreads); + RunThreadedBody( + ThreadDevPolicy::Fixed(vdev), MPIEnvironment::nThreads, + "threaded LargeTransfer_VNic", + [&](int threadIdx, ConnectionPair& pair) -> ThreadResult { + return WorkerHostTransfer(rank, pair, threadedSize, 540, + WorkerSeed(threadIdx, 5400), kLargeTransferTimeoutMs); + }); + return; + } + ConnectionPair pair; NetConnectionGuard connGuard(net_); SetupConnectionWithGuard(vdev, pair, connGuard); @@ -761,18 +851,31 @@ TEST_F(NetIbMPITest, MixedSizes_VNic) { << "Failed to create fused vNIC from devices 0 and 1"; ASSERT_GE(vdev, 0); - ConnectionPair pair; - NetConnectionGuard connGuard(net_); - SetupConnectionWithGuard(vdev, pair, connGuard); - // Sizes: 1B, 3MB, 3B, 5MB, 7B, 7MB, 64B, 16MB, 1B, 11MB, 4MB, 1B. // Tiny sizes may use only one QP, large ones stripe across both. // Odd MB sizes (3, 5, 7, 11) produce uneven QP splits. - std::vector testSizes = { + // Shared by both halves: a size added for one of them belongs to the other too. + const std::vector testSizes = { 1, 3*1024*1024, 3, 5*1024*1024, 7, 7*1024*1024, 64, 16*1024*1024, 1, 11*1024*1024, 4*1024*1024, 1 }; + // Parameterized by MPIEnvironment::nThreads: the uneven-split ladder runs on + // every worker's own fused connection simultaneously. + if (MPIEnvironment::nThreads > 1) { + // Per-size allocation and registration, as the serial body does: on a fused + // device every regMr fans out across both members, and that churn is what + // this test is about. + RunThreadedSizeSweep(ThreadDevPolicy::Fixed(vdev), MPIEnvironment::nThreads, testSizes, + /*repeats=*/1, "threaded MixedSizes_VNic", + SweepRegistration::PerSize); + return; + } + + ConnectionPair pair; + NetConnectionGuard connGuard(net_); + SetupConnectionWithGuard(vdev, pair, connGuard); + for (size_t idx = 0; idx < testSizes.size(); idx++) { size_t size = testSizes[idx]; const int tag = 550; @@ -838,16 +941,26 @@ TEST_F(NetIbMPITest, UnalignedSizeTransfer_VNic) { << "Failed to create fused vNIC from devices 0 and 1"; ASSERT_GE(vdev, 0); - ConnectionPair pair; - NetConnectionGuard connGuard(net_); - SetupConnectionWithGuard(vdev, pair, connGuard); - // Sizes around 128-byte QP striping alignment boundaries. // ncclIbMultiSend computes chunkSize = DIVUP(DIVUP(size, nqps), 128) * 128. // These sizes produce uneven QP splits where one QP gets more data than the other. // 127: all on QP 0, QP 1 posts zero-sge. 129: 128B on QP 0, 1B on QP 1. // 255: 128B each, QP 1 gets 127B. 257: 256B on QP 0, 1B remainder on QP 1. - std::vector testSizes = {127, 129, 255, 257, 511, 513}; + // Shared by both halves: a size added for one of them belongs to the other too. + const std::vector testSizes = {127, 129, 255, 257, 511, 513}; + + // Parameterized by MPIEnvironment::nThreads: concurrent workers hit the + // 128-byte striping boundary on both members of the fused device at once. + if (MPIEnvironment::nThreads > 1) { + RunThreadedSizeSweep(ThreadDevPolicy::Fixed(vdev), MPIEnvironment::nThreads, testSizes, + /*repeats=*/2, "threaded UnalignedSizeTransfer_VNic", + SweepRegistration::PerSize); + return; + } + + ConnectionPair pair; + NetConnectionGuard connGuard(net_); + SetupConnectionWithGuard(vdev, pair, connGuard); for (size_t idx = 0; idx < testSizes.size(); idx++) { size_t size = testSizes[idx]; @@ -1057,6 +1170,125 @@ TEST_F(NetIbMPITest, FlushRepeated_VNic) { ASSERT_GE(vdev, 0); const int rank = MPIEnvironment::world_rank; + + // Parameterized by MPIEnvironment::nThreads: concurrent GPU receives and + // flushes on one fused device. Workers inherit this rank's HIP device from + // the harness, since the current device is thread-local. + if (MPIEnvironment::nThreads > 1) { + static constexpr int kThreadedIters = 20; + RunThreadedBody( + ThreadDevPolicy::Fixed(vdev), MPIEnvironment::nThreads, + "threaded FlushRepeated_VNic", + [&](int threadIdx, ConnectionPair& pair) -> ThreadResult { + ThreadResult result; + const size_t size = kSmallBufferSize; + void* gpuBuffer = nullptr; + if (hipMalloc(&gpuBuffer, size) != hipSuccess) { + result.ok = false; + result.msg = "hipMalloc failed"; + return result; + } + auto gpuGuard = makeDeviceBufferAutoGuard(gpuBuffer); + + void* comm = (rank == 0) ? pair.recvComm : pair.sendComm; + void* mhandle = nullptr; + result = WorkerRegister(comm, gpuBuffer, size, NCCL_PTR_CUDA, &mhandle); + if (!result.ok) return result; + NetMHandleWorkerGuard mhandleGuard(mhandle, NetMHandleWorkerDeleter(net_, comm)); + + // One pattern per worker: a per-iteration seed aliases modulo 256. + const int seed = WorkerSeed(threadIdx, 6000); + for (int iter = 0; iter < kThreadedIters; iter++) { + if (rank == 1 + && initializeBufferWithPattern(gpuBuffer, size, + makeBytePattern(seed)) + != hipSuccess) { + result.ok = false; + result.msg = "GPU buffer initialization failed"; + return result; + } + + // Cleared before every receive, not just the first: the expected + // pattern is the same on all iterations, so after iteration 0 a + // flush that did nothing would leave the previous payload in place + // and verify clean. Zeroing first makes each iteration prove that + // this transfer's data became visible. + if (rank == 0 && hipMemset(gpuBuffer, 0, size) != hipSuccess) { + result.ok = false; + result.msg = "clearing the GPU receive buffer failed at iteration " + + std::to_string(iter); + return result; + } + + // The size is checked too, so a short receive cannot verify on a + // tail this clearing happened to leave at the expected value. + int received = 0; + bool outstanding = false; + result = WorkerSendRecvRaw(rank, pair, gpuBuffer, size, 600, mhandle, + kDefaultTimeoutMs, &received, + /*busyPoll=*/false, &outstanding); + if (!result.ok) { + // Per-worker gpuGuard/mhandleGuard, unlike the shared allocation + // above: a timed-out request here is this worker's own, so only + // its own guards need to survive it. + if (outstanding) { + result.msg += "; the GPU buffer and its registration are " + "retained, since the request may still " + "reference them"; + mhandleGuard.release(); + gpuGuard.release(); + } + return result; + } + + if (rank != 0) continue; + if (received != static_cast(size)) { + result.ok = false; + result.msg = "iteration " + std::to_string(iter) + " received " + + std::to_string(received) + " of " + + std::to_string(size) + " bytes"; + return result; + } + + void* flushBuffers[1] = {gpuBuffer}; + int flushSizes[1] = {static_cast(size)}; + void* flushHandles[1] = {mhandle}; + void* flushRequest = nullptr; + // A null request means the flush was a no-op; an error + // return means the flush itself failed. + if (FlushRecv(pair.recvComm, 1, flushBuffers, flushSizes, flushHandles, + &flushRequest) + != ncclSuccess) { + result.ok = false; + result.msg = "FlushRecv failed at iteration " + std::to_string(iter); + return result; + } + if (flushRequest != nullptr) { + result = WorkerWait(flushRequest, nullptr); + if (!result.ok) { + // A failed wait means the timeout was reached, so this flush + // request -- unlike the ordinary send/recv above -- has no + // separate outstanding flag: not completing within the + // timeout is what the failure already says. + result.msg += "; the GPU buffer and its registration are " + "retained, since the flush may still " + "reference them"; + mhandleGuard.release(); + gpuGuard.release(); + return result; + } + } + if (!verifyBufferData(gpuBuffer, size, makeBytePattern(seed))) { + result.ok = false; + result.msg = "GPU data verification failed after flush"; + return result; + } + } + return result; + }); + return; + } + ConnectionPair pair; NetConnectionGuard connGuard(net_); SetupConnectionWithGuard(vdev, pair, connGuard); @@ -1104,6 +1336,7 @@ TEST_F(NetIbMPITest, FlushRepeated_VNic) { ncclResult_t flushResult = FlushRecv(pair.recvComm, 1, flushBuffers, flushSizes, flushHandles, &flushRequest); + EXPECT_EQ(flushResult, ncclSuccess) << "Iter " << iter << ": FlushRecv failed"; if (flushResult == ncclSuccess && flushRequest != nullptr) { ASSERT_EQ(WaitForCompletion(flushRequest, nullptr), ncclSuccess); } @@ -1141,6 +1374,35 @@ TEST_F(NetIbMPITest, SequentialTransfers_VNic) { const int rank = MPIEnvironment::world_rank; + // Parameterized by MPIEnvironment::nThreads: every worker reuses one + // registration across its iterations, so the fused device serves many + // long-lived MRs at once. + if (MPIEnvironment::nThreads > 1) { + static constexpr int kThreadedIters = 100; + RunThreadedBody( + ThreadDevPolicy::Fixed(vdev), MPIEnvironment::nThreads, + "threaded SequentialTransfers_VNic", + [&](int threadIdx, ConnectionPair& pair) -> ThreadResult { + const size_t size = kSmallBufferSize; + WorkerHostBuffer host = WorkerSetupHostBuffer(rank, pair, size); + if (!host.result.ok) return host.result; + + // Worker-constant, as above. + ThreadResult result; + const int seed = WorkerSeed(threadIdx, 7000); + bool outstanding = false; + for (int iter = 0; iter < kThreadedIters; iter++) { + result = WorkerSendRecvPattern(rank, pair, host.buffer, size, 700, + host.mhandle, seed, kDefaultTimeoutMs, + &outstanding); + if (!result.ok) + return outstanding ? WorkerRetainHostBuffer(result, host) : result; + } + return result; + }); + return; + } + // Single connection through the vNIC, reused across all 100 iterations. ConnectionPair pair; NetConnectionGuard connGuard(net_); From 2d8b53a3814be979d06d45c7c4ff4cb6e2cdb6c4 Mon Sep 17 00:00:00 2001 From: Ilya Kosarev Date: Wed, 16 Sep 2026 13:30:51 +0000 Subject: [PATCH 4/4] rccl(ainic): register the threaded IB-CAST and fused-device suites Adds the new threaded suites to every config that carries their serial siblings, and marks a_cast_multithread (A18) as smoke on the AINIC config -- it is a Component suite, so the name-glob the gate selects on already covers it and the flag makes that explicit. --- .../configs/mi300_mellanox_ib_func.json | 28 ++++++ .../configs/mi300x_mellanox_ib.json | 28 ++++++ .../configs/mi355x_ainic_tw_roce.json | 17 ++++ .../configs/mi355x_thor2_roce.json | 28 ++++++ .../configs/mi355x_thor2_roce_func.json | 28 ++++++ .../test_runner/configs/mi455_ainic_roce.json | 28 ++++++ .../configs/mi455_ainic_roce_func.json | 28 ++++++ .../test_runner/configs/net_ib_transport.json | 96 +++++++++++++++++++ 8 files changed, 281 insertions(+) diff --git a/projects/rccl/tools/scripts/test_runner/configs/mi300_mellanox_ib_func.json b/projects/rccl/tools/scripts/test_runner/configs/mi300_mellanox_ib_func.json index a4d166670dd0..1b84424a0185 100644 --- a/projects/rccl/tools/scripts/test_runner/configs/mi300_mellanox_ib_func.json +++ b/projects/rccl/tools/scripts/test_runner/configs/mi300_mellanox_ib_func.json @@ -227,6 +227,28 @@ } ] }, + "cast_multithread": { + "extends": "cast_base", + "timeout": 600, + "command_args": "--net_ib_nthreads=4", + "env_variables": { + "NCCL_IB_QPS_PER_CONNECTION": "2", + "NCCL_IB_SPLIT_DATA_ON_QPS": "1", + "RCCL_IB_QP_SCHED_UPDATE_INTERVAL": "10000000" + }, + "tests": [ + { + "name": "Cast_MultiThread_StressMultiRound", + "description": "Concurrent IB-CAST connections, each auditing its own WRR token ledger", + "test_filter": "NetIbMPITest.CastStressMultiRoundTwoConns" + }, + { + "name": "Cast_MultiThread_SendRecvMultipleSizes", + "description": "Concurrent schedulers sweep both sides of the WRR/split boundary", + "test_filter": "NetIbMPITest.CastSendRecvMultipleSizes" + } + ] + }, "net_ib_gdr_flush": { "extends": "net_ib_base", "timeout": 120, @@ -2794,6 +2816,12 @@ "config": "net_ib_multithread_independent", "enabled": true }, + { + "name": "NET IB - CAST Multithread (4)", + "description": "4 worker threads/rank driving independent IB-CAST connections and WRR schedulers", + "config": "cast_multithread", + "enabled": true + }, { "name": "GDR Flush - Whitebox ncclIbIflush (Single-Node, 2-Rank)", "description": "Whitebox coverage of the NET/IB GDR flush: dma-buf and peermem RO=0 scratchpad fence, feature-disabled fallback, and repeated-flush burst", diff --git a/projects/rccl/tools/scripts/test_runner/configs/mi300x_mellanox_ib.json b/projects/rccl/tools/scripts/test_runner/configs/mi300x_mellanox_ib.json index bab9997c5963..2f7585ae98bf 100644 --- a/projects/rccl/tools/scripts/test_runner/configs/mi300x_mellanox_ib.json +++ b/projects/rccl/tools/scripts/test_runner/configs/mi300x_mellanox_ib.json @@ -478,6 +478,28 @@ } ] }, + "cast_multithread": { + "extends": "cast_base", + "timeout": 600, + "command_args": "--net_ib_nthreads=4", + "env_variables": { + "NCCL_IB_QPS_PER_CONNECTION": "2", + "NCCL_IB_SPLIT_DATA_ON_QPS": "1", + "RCCL_IB_QP_SCHED_UPDATE_INTERVAL": "10000000" + }, + "tests": [ + { + "name": "Cast_MultiThread_StressMultiRound", + "description": "Concurrent IB-CAST connections, each auditing its own WRR token ledger", + "test_filter": "NetIbMPITest.CastStressMultiRoundTwoConns" + }, + { + "name": "Cast_MultiThread_SendRecvMultipleSizes", + "description": "Concurrent schedulers sweep both sides of the WRR/split boundary", + "test_filter": "NetIbMPITest.CastSendRecvMultipleSizes" + } + ] + }, "net_ib_initialization": { "extends": "net_ib_base", "timeout": 60, @@ -4204,6 +4226,12 @@ "config": "net_ib_multithread_independent", "enabled": true }, + { + "name": "NET IB - CAST Multithread (4)", + "description": "4 worker threads/rank driving independent IB-CAST connections and WRR schedulers", + "config": "cast_multithread", + "enabled": true + }, { "name": "GDR Flush - Whitebox ncclIbIflush (Single-Node, 2-Rank)", "description": "Whitebox coverage of the NET/IB GDR flush: dma-buf and peermem RO=0 scratchpad fence, feature-disabled fallback, and repeated-flush burst", diff --git a/projects/rccl/tools/scripts/test_runner/configs/mi355x_ainic_tw_roce.json b/projects/rccl/tools/scripts/test_runner/configs/mi355x_ainic_tw_roce.json index 6869c36e2419..b330932f3c17 100644 --- a/projects/rccl/tools/scripts/test_runner/configs/mi355x_ainic_tw_roce.json +++ b/projects/rccl/tools/scripts/test_runner/configs/mi355x_ainic_tw_roce.json @@ -199,6 +199,22 @@ } ] }, + "a_cast_multithread": { + "extends": "cast_base", + "timeout": 600, + "command_args": "--net_ib_nthreads=4", + "description": "A18: four worker threads per rank on independent IB-CAST connections, so several WRR schedulers advance at once.", + "env_variables": { + "RCCL_IB_QP_SCHED_UPDATE_INTERVAL": "10000000" + }, + "tests": [ + { + "name": "A18_Cast_MultiThread_StressMultiRound", + "description": "Concurrent CAST connections, each auditing its own WRR token ledger.", + "test_filter": "NetIbMPITest.CastStressMultiRoundTwoConns" + } + ] + }, "gin_proxy": { "extends": "gtest_base", @@ -489,6 +505,7 @@ { "config": "gin_ainic_backend", "enabled": true }, { "config": "bootstrap_socket_poll", "enabled": true, "smoke": true }, { "config": "a_netib_multithread", "enabled": true, "smoke": true }, + { "config": "a_cast_multithread", "enabled": true, "smoke": true }, { "config": "ib_baseline", "enabled": true, "smoke": true }, { "config": "cast_alltoall_full", "enabled": true, "smoke": true }, diff --git a/projects/rccl/tools/scripts/test_runner/configs/mi355x_thor2_roce.json b/projects/rccl/tools/scripts/test_runner/configs/mi355x_thor2_roce.json index c02f3cfbce7a..cfef0f679023 100644 --- a/projects/rccl/tools/scripts/test_runner/configs/mi355x_thor2_roce.json +++ b/projects/rccl/tools/scripts/test_runner/configs/mi355x_thor2_roce.json @@ -246,6 +246,28 @@ } ] }, + "cast_multithread": { + "extends": "cast_base", + "timeout": 600, + "command_args": "--net_ib_nthreads=4", + "env_variables": { + "NCCL_IB_QPS_PER_CONNECTION": "2", + "NCCL_IB_SPLIT_DATA_ON_QPS": "1", + "RCCL_IB_QP_SCHED_UPDATE_INTERVAL": "10000000" + }, + "tests": [ + { + "name": "Cast_MultiThread_StressMultiRound", + "description": "Concurrent IB-CAST connections, each auditing its own WRR token ledger", + "test_filter": "NetIbMPITest.CastStressMultiRoundTwoConns" + }, + { + "name": "Cast_MultiThread_SendRecvMultipleSizes", + "description": "Concurrent schedulers sweep both sides of the WRR/split boundary", + "test_filter": "NetIbMPITest.CastSendRecvMultipleSizes" + } + ] + }, "net_ib_gdr_flush": { "extends": "net_ib_base", "timeout": 120, @@ -4490,6 +4512,12 @@ "config": "net_ib_multithread_independent", "enabled": true }, + { + "name": "NET IB - CAST Multithread (4)", + "description": "4 worker threads/rank driving independent IB-CAST connections and WRR schedulers", + "config": "cast_multithread", + "enabled": true + }, { "name": "GDR Flush - Whitebox ncclIbIflush (Single-Node, 2-Rank)", "description": "Whitebox coverage of the NET/IB GDR flush: dma-buf and peermem RO=0 scratchpad fence, feature-disabled fallback, and repeated-flush burst", diff --git a/projects/rccl/tools/scripts/test_runner/configs/mi355x_thor2_roce_func.json b/projects/rccl/tools/scripts/test_runner/configs/mi355x_thor2_roce_func.json index 526bb937ca9f..3d10a992af38 100644 --- a/projects/rccl/tools/scripts/test_runner/configs/mi355x_thor2_roce_func.json +++ b/projects/rccl/tools/scripts/test_runner/configs/mi355x_thor2_roce_func.json @@ -233,6 +233,28 @@ } ] }, + "cast_multithread": { + "extends": "cast_base", + "timeout": 600, + "command_args": "--net_ib_nthreads=4", + "env_variables": { + "NCCL_IB_QPS_PER_CONNECTION": "2", + "NCCL_IB_SPLIT_DATA_ON_QPS": "1", + "RCCL_IB_QP_SCHED_UPDATE_INTERVAL": "10000000" + }, + "tests": [ + { + "name": "Cast_MultiThread_StressMultiRound", + "description": "Concurrent IB-CAST connections, each auditing its own WRR token ledger", + "test_filter": "NetIbMPITest.CastStressMultiRoundTwoConns" + }, + { + "name": "Cast_MultiThread_SendRecvMultipleSizes", + "description": "Concurrent schedulers sweep both sides of the WRR/split boundary", + "test_filter": "NetIbMPITest.CastSendRecvMultipleSizes" + } + ] + }, "net_ib_gdr_flush": { "extends": "net_ib_base", "timeout": 120, @@ -2836,6 +2858,12 @@ "config": "net_ib_multithread_independent", "enabled": true }, + { + "name": "NET IB - CAST Multithread (4)", + "description": "4 worker threads/rank driving independent IB-CAST connections and WRR schedulers", + "config": "cast_multithread", + "enabled": true + }, { "name": "GDR Flush - Whitebox ncclIbIflush (Single-Node, 2-Rank)", "description": "Whitebox coverage of the NET/IB GDR flush: dma-buf and peermem RO=0 scratchpad fence, feature-disabled fallback, and repeated-flush burst", diff --git a/projects/rccl/tools/scripts/test_runner/configs/mi455_ainic_roce.json b/projects/rccl/tools/scripts/test_runner/configs/mi455_ainic_roce.json index 7bdae87fd189..bee239332b26 100644 --- a/projects/rccl/tools/scripts/test_runner/configs/mi455_ainic_roce.json +++ b/projects/rccl/tools/scripts/test_runner/configs/mi455_ainic_roce.json @@ -242,6 +242,28 @@ } ] }, + "cast_multithread": { + "extends": "cast_base", + "timeout": 600, + "command_args": "--net_ib_nthreads=4", + "env_variables": { + "NCCL_IB_QPS_PER_CONNECTION": "2", + "NCCL_IB_SPLIT_DATA_ON_QPS": "1", + "RCCL_IB_QP_SCHED_UPDATE_INTERVAL": "10000000" + }, + "tests": [ + { + "name": "Cast_MultiThread_StressMultiRound", + "description": "Concurrent IB-CAST connections, each auditing its own WRR token ledger", + "test_filter": "NetIbMPITest.CastStressMultiRoundTwoConns" + }, + { + "name": "Cast_MultiThread_SendRecvMultipleSizes", + "description": "Concurrent schedulers sweep both sides of the WRR/split boundary", + "test_filter": "NetIbMPITest.CastSendRecvMultipleSizes" + } + ] + }, "net_ib_gdr_flush": { "extends": "net_ib_base", "timeout": 120, @@ -3648,6 +3670,12 @@ "config": "net_ib_multithread_independent", "enabled": true }, + { + "name": "NET IB - CAST Multithread (4)", + "description": "4 worker threads/rank driving independent IB-CAST connections and WRR schedulers", + "config": "cast_multithread", + "enabled": true + }, { "name": "GDR Flush - Whitebox ncclIbIflush (Single-Node, 2-Rank)", "description": "Whitebox coverage of the NET/IB GDR flush: dma-buf and peermem RO=0 scratchpad fence, feature-disabled fallback, and repeated-flush burst", diff --git a/projects/rccl/tools/scripts/test_runner/configs/mi455_ainic_roce_func.json b/projects/rccl/tools/scripts/test_runner/configs/mi455_ainic_roce_func.json index df978ab108d2..aabf7bf7053a 100644 --- a/projects/rccl/tools/scripts/test_runner/configs/mi455_ainic_roce_func.json +++ b/projects/rccl/tools/scripts/test_runner/configs/mi455_ainic_roce_func.json @@ -227,6 +227,28 @@ } ] }, + "cast_multithread": { + "extends": "cast_base", + "timeout": 600, + "command_args": "--net_ib_nthreads=4", + "env_variables": { + "NCCL_IB_QPS_PER_CONNECTION": "2", + "NCCL_IB_SPLIT_DATA_ON_QPS": "1", + "RCCL_IB_QP_SCHED_UPDATE_INTERVAL": "10000000" + }, + "tests": [ + { + "name": "Cast_MultiThread_StressMultiRound", + "description": "Concurrent IB-CAST connections, each auditing its own WRR token ledger", + "test_filter": "NetIbMPITest.CastStressMultiRoundTwoConns" + }, + { + "name": "Cast_MultiThread_SendRecvMultipleSizes", + "description": "Concurrent schedulers sweep both sides of the WRR/split boundary", + "test_filter": "NetIbMPITest.CastSendRecvMultipleSizes" + } + ] + }, "net_ib_gdr_flush": { "extends": "net_ib_base", "timeout": 120, @@ -2789,6 +2811,12 @@ "config": "net_ib_multithread_independent", "enabled": true }, + { + "name": "NET IB - CAST Multithread (4)", + "description": "4 worker threads/rank driving independent IB-CAST connections and WRR schedulers", + "config": "cast_multithread", + "enabled": true + }, { "name": "GDR Flush - Whitebox ncclIbIflush (Single-Node, 2-Rank)", "description": "Whitebox coverage of the NET/IB GDR flush: dma-buf and peermem RO=0 scratchpad fence, feature-disabled fallback, and repeated-flush burst", diff --git a/projects/rccl/tools/scripts/test_runner/configs/net_ib_transport.json b/projects/rccl/tools/scripts/test_runner/configs/net_ib_transport.json index 259f8abcbef4..4fff0df71838 100644 --- a/projects/rccl/tools/scripts/test_runner/configs/net_ib_transport.json +++ b/projects/rccl/tools/scripts/test_runner/configs/net_ib_transport.json @@ -1068,6 +1068,36 @@ "name": "NetIB_MultiThread_GpuMemoryTransfer", "description": "Concurrent GPU-buffer registrations drive the peer-memory path into the shared MR cache", "test_filter": "NetIbMPITest.GpuMemoryTransferStress" + }, + { + "name": "NetIB_MultiThread_VNicRegDeregCycling", + "description": "Concurrent register/deregister cycling on a fused device, fanning out across both members' protection domains", + "test_filter": "NetIbMPITest.RegDeregCycling_VNic" + }, + { + "name": "NetIB_MultiThread_VNicSequentialTransfers", + "description": "Concurrent long-lived registrations on one fused device", + "test_filter": "NetIbMPITest.SequentialTransfers_VNic" + }, + { + "name": "NetIB_MultiThread_VNicMixedSizes", + "description": "Concurrent uneven QP splits across a fused device", + "test_filter": "NetIbMPITest.MixedSizes_VNic" + }, + { + "name": "NetIB_MultiThread_VNicLargeTransfer", + "description": "Concurrent striped transfers over a fused device, 64 MB divided by the worker count so the registrations stay within the device limits", + "test_filter": "NetIbMPITest.LargeTransfer_VNic" + }, + { + "name": "NetIB_MultiThread_VNicUnalignedSizes", + "description": "Concurrent transfers at the 128-byte striping boundary of a fused device", + "test_filter": "NetIbMPITest.UnalignedSizeTransfer_VNic" + }, + { + "name": "NetIB_MultiThread_VNicFlushRepeated", + "description": "Concurrent GPU receives and flushes on one fused device", + "test_filter": "NetIbMPITest.FlushRepeated_VNic" } ] }, @@ -1119,6 +1149,54 @@ "extends": "net_ib_multithread_multiqp_nosplit", "command_args": "--net_ib_nthreads=4" }, + "cast_multithread": { + "extends": "cast_base", + "timeout": 300, + "env_variables": { + "NCCL_IB_QPS_PER_CONNECTION": "2", + "NCCL_IB_SPLIT_DATA_ON_QPS": "1", + "RCCL_IB_QP_SCHED_UPDATE_INTERVAL": "10000000" + }, + "tests": [ + { + "name": "Cast_MultiThread_StressMultiRound", + "description": "Every worker drives its own CAST connection, so the WRR schedulers advance simultaneously instead of one after another", + "test_filter": "NetIbMPITest.CastStressMultiRoundTwoConns" + }, + { + "name": "Cast_MultiThread_SendRecvMultipleSizes", + "description": "Concurrent schedulers sweep both sides of the WRR/split boundary, each auditing its own token ledger", + "test_filter": "NetIbMPITest.CastSendRecvMultipleSizes" + }, + { + "name": "Cast_MultiThread_LargeTransfer", + "description": "Concurrent 16 MB split-path transfers, each confirming its scheduler consumed no tokens", + "test_filter": "NetIbMPITest.CastLargeTransfer" + }, + { + "name": "Cast_MultiThread_SendRecvZeroSize", + "description": "Zero-byte sends still take the WRR path on every worker's scheduler", + "test_filter": "NetIbMPITest.CastSendRecvZeroSize" + }, + { + "name": "Cast_MultiThread_SplitDataThresholdBoundary", + "description": "Concurrent schedulers decide the split boundary at the same time, each auditing its own tokens", + "test_filter": "NetIbMPITest.CastSplitDataThresholdBoundary" + } + ] + }, + "cast_multithread_2": { + "extends": "cast_multithread", + "command_args": "--net_ib_nthreads=2" + }, + "cast_multithread_4": { + "extends": "cast_multithread", + "command_args": "--net_ib_nthreads=4" + }, + "cast_multithread_16": { + "extends": "cast_multithread", + "command_args": "--net_ib_nthreads=16" + }, "cast_wrr_split_long_interval": { "extends": "cast_base", "timeout": 120, @@ -1459,6 +1537,24 @@ "config": "net_ib_multithread_multiqp_nosplit_4", "enabled": true }, + { + "name": "NET IB - CAST Multithread (2)", + "description": "2 worker threads/rank driving independent IB-CAST connections and WRR schedulers", + "config": "cast_multithread_2", + "enabled": true + }, + { + "name": "NET IB - CAST Multithread (4)", + "description": "4 worker threads/rank driving independent IB-CAST connections and WRR schedulers", + "config": "cast_multithread_4", + "enabled": true + }, + { + "name": "NET IB - CAST Multithread (16)", + "description": "16 worker threads/rank driving independent IB-CAST connections and WRR schedulers", + "config": "cast_multithread_16", + "enabled": true + }, { "name": "NET IB - CAST WRR+Split, long update interval", "description": "Scheduler toggle tests that need RCCL_IB_QP_SCHED_UPDATE_INTERVAL >= 10 s; they skip in every other config",