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/NetIbMPITestBase.hpp b/projects/rccl/test/transport/NetIbMPI/NetIbMPITestBase.hpp index d1fbd71ad274..f7653da10d15 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; @@ -1894,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 @@ -1923,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 @@ -1937,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; @@ -1946,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++; } 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_); 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",