diff --git a/projects/rccl/test/host/CMakeLists.txt b/projects/rccl/test/host/CMakeLists.txt index 9d365aca5a01..b923cf3320d7 100644 --- a/projects/rccl/test/host/CMakeLists.txt +++ b/projects/rccl/test/host/CMakeLists.txt @@ -190,6 +190,18 @@ function(rccl_define_micro_source_lists) ../common/ProcessIsolatedTestRunner.cpp PARENT_SCOPE) + # rccl-UnitTestsMicroBootstrap -- src/bootstrap.cc. Own binary: bootstrap_micro_fakes.cc replaces misc/socket.cc + # wholesale, and other binaries link bootstrap_stubs.cc for this unit's symbols. + set(TEST_MICRO_BOOTSTRAP_SOURCE_FILES + bootstrap-test.cc + fakes/bootstrap_micro_fakes.cc + fakes/env_fakes.cc + fakes/hip_fakes.cc + fakes/os_fakes.cc + ${RCCL_HOST_HIPIFY_DIR}/src/misc/utils.cc + ../common/main_altrsmi.cpp + PARENT_SCOPE) + # rccl-UnitTestsMicroWarpSpeed -- scheduler UUTs under ENABLE_WARP_SPEED. set(TEST_MICRO_SCHEDULER_SOURCE_FILES scheduler-test.cc @@ -1187,6 +1199,16 @@ endif() # all targets -- no need to repeat it per binary. rccl_mark_oracle_tus_generated(${RCCL_HOST_HIPIFY_DIR} argcheck archinfo utils kernel_config) +# --------------------------------------------------------------------------- +# rccl-UnitTestsMicroBootstrap +# --------------------------------------------------------------------------- +rccl_add_micro_binary(rccl-UnitTestsMicroBootstrap + SOURCES ${TEST_MICRO_BOOTSTRAP_SOURCE_FILES} + EXTRA_DEFS + ${_MICRO_MODE_DEFS} + BOOTSTRAP_CC_PATH="${RCCL_HOST_HIPIFY_DIR}/src/bootstrap.cc" + CATEGORIES_YAML "${CMAKE_CURRENT_SOURCE_DIR}/../test_categories_micro_bootstrap.yaml") + # --------------------------------------------------------------------------- # rccl-UnitTestsMicro # --------------------------------------------------------------------------- diff --git a/projects/rccl/test/host/MICROTEST_README.md b/projects/rccl/test/host/MICROTEST_README.md index 5f234b4fa491..8773df521cb6 100644 --- a/projects/rccl/test/host/MICROTEST_README.md +++ b/projects/rccl/test/host/MICROTEST_README.md @@ -227,6 +227,11 @@ export colliding non-`static` symbols; otherwise a unit needs its own binary: tables, prime strides, Walecki, Walecki+greedy) and the load-balanced cut selection in `findRingCutIndices`. `greedyRingGen`'s out-of-memory arm is driven through the heap half of `fakes/libc_seam.h`. +- **`rccl-UnitTestsMicroBootstrap`**: `bootstrap.cc` (via `BOOTSTRAP_CC_PATH`, from + `bootstrap-test.cc`); suites `Bootstrap*Microtest.*`. Ranks run as threads over the + in-memory socket layer and net plugin in `fakes/bootstrap_micro_fakes.cc`. Its own binary + because those seams replace `misc/socket.cc` wholesale, and other binaries link + `bootstrap_stubs.cc` for this unit's symbols. See `test_categories_micro_bootstrap.yaml`. - **`rccl-UnitTestsMicroEnqueue`** — `enqueue.cc` (via `ENQUEUE_CC_PATH`); suite `EnqueueMicrotest.*`. All tests live in `enqueue-test.cc`, grouped by unit under test; several fixtures are reused by later groups, so the order within the file @@ -587,6 +592,7 @@ symbol. | `src/algorithms/dda/*.cc` (targets that do not compile the real file) | `fakes/dda_fakes.cc` | | `src/allocator.cc` | `fakes/allocator_fakes.cc` | | `src/bootstrap.cc` | `fakes/bootstrap_stubs.cc` | +| externals `src/bootstrap.cc` reaches (`rccl-UnitTestsMicroBootstrap`, which compiles it) | `fakes/bootstrap_micro_fakes.cc` | | `src/ce_coll.cc` (targets that do not compile the real file) | `fakes/ce_fakes.cc` | | `src/collectives.cc` | `fakes/collectives_fakes.cc` | | `src/dev_runtime.cc` (targets that do not compile the real file) | `fakes/dev_runtime_fakes.cc` | @@ -1005,6 +1011,7 @@ cd projects/rccl/test/host cmake -B build -DCMAKE_BUILD_TYPE=Release \ -DRCCL_BUILD_DIR=/path/to/projects/rccl/build/release cmake --build build -j"$(nproc)" +./build/rccl-UnitTestsMicroBootstrap # src/bootstrap.cc tests ./build/rccl-UnitTestsMicro # p2p tests, ldd shows no HIP/ROCm/HSA/RCCL ./build/rccl-UnitTestsMicroWarpSpeed # src/scheduler/*.cc tests ./build/rccl-UnitTestsMicroCrypt # misc/crypt.cc tests diff --git a/projects/rccl/test/host/bootstrap-test.cc b/projects/rccl/test/host/bootstrap-test.cc new file mode 100644 index 000000000000..5bd1b4bb4a81 --- /dev/null +++ b/projects/rccl/test/host/bootstrap-test.cc @@ -0,0 +1,1460 @@ +/************************************************************************* + * Copyright (c) 2026 Advanced Micro Devices, Inc. All rights reserved. + * + * See LICENSE.txt for license information + ************************************************************************/ + +// Host-only microtests for src/bootstrap.cc. Ranks run as threads over the in-memory socket layer and net +// plugin in fakes/bootstrap_micro_fakes.cc, so the root rendezvous, both rings and the collectives run for real. + +#include + +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "nccl.h" +#include "comm.h" +#include "debug.h" +#include "net.h" +#include "proxy.h" +#include "ras.h" +#include "socket.h" +#include "utils.h" + +#include "fakes/bootstrap_micro_fakes.h" +#include "fakes/env_fakes.h" +#include "fakes/param_redirect.h" + +#include BOOTSTRAP_CC_PATH + +namespace { + +using namespace bootstrap_micro; + +struct Rank { + std::unique_ptr comm = std::make_unique(); + uint32_t abort = 0; + std::vector topParentRanks; +}; + +class BootstrapMicrotest : public ::testing::Test { + protected: + std::vector> ranks_; + std::vector> parents_; // the comm a split or grow starts from + ncclBootstrapHandle handle_{}; + std::chrono::milliseconds watchdog_ = g_waitLimit; + uint32_t abort_ = 0; // for sockets the test thread drives itself + Rank self_; // the comm sendToRoot reads its abort flag from + std::thread::id main_ = std::this_thread::get_id(); + + void SetUp() override { + ResetSeams(); + bootstrapNetInitDone = 0; + ASSERT_EQ(ncclSuccess, bootstrapNetInit()); + } + void TearDown() override { + CloseAll(); + ShutdownSockets(); + ResetSeams(); + } + // Closes every comm still holding a bootstrap state, parents included. + void CloseAll() { + for (auto& r : ranks_) parents_.push_back(std::move(r)); + ranks_.clear(); + for (auto& r : parents_) { + if (r->comm->bootstrap == nullptr) continue; + __atomic_store_n(&r->abort, 1u, __ATOMIC_RELEASE); // a failed test may leave unmatched connections + (void)bootstrapClose(r->comm->bootstrap); + r->comm->bootstrap = nullptr; + } + parents_.clear(); + } + // For a test that leaves a root waiting and then reuses the fake network. + void RestartSockets() { + ShutdownSockets(); + std::lock_guard lock(g_sockMu); + g_sockShutdown = false; + } + + // A connected pair over a fresh listener; `a` is the connecting side. + void Pair(struct ncclSocket* a, struct ncclSocket* b) { + struct ncclSocket l; + ASSERT_EQ(ncclSuccess, ncclSocketInit(&l, &bootstrapNetIfAddr, 7, ncclSocketTypeBootstrap, &abort_)); + ASSERT_EQ(ncclSuccess, ncclSocketListen(&l)); + ASSERT_EQ(ncclSuccess, ncclSocketInit(a, &l.addr, 7, ncclSocketTypeBootstrap)); + ASSERT_EQ(ncclSuccess, ncclSocketConnect(a)); + ASSERT_EQ(ncclSuccess, ncclSocketInit(b)); + ASSERT_EQ(ncclSuccess, ncclSocketAccept(b, &l)); + ASSERT_EQ(ncclSuccess, ncclSocketClose(&l)); + } + // Plays `rank` against the root behind handle_: listens for the answer, then checks in. + void CheckIn(int rank, int nranks, struct ncclSocket* listen, int nroots = 1, int iroot = 0, int offset = 0) { + struct extInfo info = {}; + info.rank = rank; + info.nranks = nranks; + info.nroots = nroots; + info.iroot = iroot; + info.offset = offset; + info.connectInfo.fwd.handle[0] = char(1 + rank); + info.connectInfo.revHandle[0] = char(101 + rank); + ASSERT_EQ(ncclSuccess, ncclSocketInit(listen, &bootstrapNetIfAddr, handle_.magic, ncclSocketTypeBootstrap, &abort_)); + ASSERT_EQ(ncclSuccess, ncclSocketListen(listen)); + info.listenRootAddress = listen->addr; + self_.comm->abortFlag = &self_.abort; + ASSERT_EQ(ncclSuccess, sendToRoot(&handle_, self_.comm.get(), &info)); + } + struct ringConnectInfo Answer(struct ncclSocket* listen) { + struct ringConnectInfo out = {}; + struct ncclSocket sock; + EXPECT_EQ(ncclSuccess, bootstrapAccept(&sock, listen, &abort_)); + EXPECT_EQ(ncclSuccess, socketRecv(&sock, &out, sizeof(out))); + (void)ncclSocketClose(&sock); + (void)ncclSocketClose(listen); + return out; + } + void WaitRoots() { + std::unique_lock lock(g_sockMu); + auto deadline = Clock::now() + g_waitLimit; + while (g_liveRoots > 0 && Clock::now() < deadline) g_sockCv.wait_for(lock, std::chrono::milliseconds(1)); + EXPECT_EQ(0, g_liveRoots); + } + size_t PendingOn(const struct ncclSocket& listen) { + std::lock_guard lock(g_sockMu); + return g_listeners[listen.socketDescriptor - kListenFd].pending.size(); + } + int OpenListeners() { + std::lock_guard lock(g_sockMu); + return std::count_if(g_openFds.begin(), g_openFds.end(), [](int fd) { return fd >= kListenFd; }); + } + + ncclComm* Comm(int r) { return ranks_[r]->comm.get(); } + void* State(int r) { return Comm(r)->bootstrap; } + void MakeRanks(int n) { + for (int r = 0; r < n; r++) { + auto rank = std::make_unique(); + rank->comm->rank = r; + rank->comm->nRanks = n; + rank->comm->cudaDev = r; + rank->comm->nvmlDev = 10 + r; + rank->comm->abortFlag = &rank->abort; + rank->comm->ncclNet = &g_fakeNet; + ranks_.push_back(std::move(rank)); + } + } + // One thread per rank; past the wait limit every abort flag is raised so busy loops unwind too. + std::vector OnRanks(const std::function& fn) { + std::vector res(ranks_.size(), ncclSuccess); + std::atomic running(ranks_.size()); + std::vector threads; + for (int r = 0; r < (int)ranks_.size(); r++) + threads.emplace_back([&, r] { + res[r] = fn(r); + running--; + }); + auto deadline = Clock::now() + watchdog_; + while (running > 0 && Clock::now() < deadline) std::this_thread::sleep_for(std::chrono::milliseconds(1)); + if (running > 0) + for (auto& rank : ranks_) __atomic_store_n(&rank->abort, 1u, __ATOMIC_RELEASE); + for (auto& t : threads) t.join(); + return res; + } + static bool AllOk(const std::vector& res) { + return std::all_of(res.begin(), res.end(), [](ncclResult_t r) { return r == ncclSuccess; }); + } + ncclResult_t InitRanks(int n) { + MakeRanks(n); + ncclResult_t res = bootstrapGetUniqueId(&handle_, nullptr); + if (res != ncclSuccess) return res; + auto all = OnRanks([&](int r) { return bootstrapInit(1, &handle_, Comm(r), nullptr); }); + for (ncclResult_t r : all) + if (r != ncclSuccess) return r; + WaitRoots(); // so a fault hook installed next is never reached from the root thread + return ncclSuccess; + } + std::vector CloseRanks() { + auto res = OnRanks([&](int r) { return bootstrapClose(State(r)); }); + for (int r = 0; r < (int)ranks_.size(); r++) + if (res[r] == ncclSuccess) Comm(r)->bootstrap = nullptr; + return res; + } + // Every rank contributes `size` bytes of (rank, slice) and must end with every rank's slice. + // Works across several comms at once: each rank gathers within its own. + void ExpectAllGather(int size) { + std::vector> data(ranks_.size()); + for (int r = 0; r < (int)ranks_.size(); r++) { + data[r].resize((size_t)Comm(r)->nRanks * size); + std::fill_n(data[r].begin() + (size_t)Comm(r)->rank * size, size, char('A' + Comm(r)->rank)); + } + EXPECT_TRUE(AllOk(OnRanks([&](int r) { return bootstrapAllGather(State(r), data[r].data(), size); }))); + for (int r = 0; r < (int)ranks_.size(); r++) + for (int s = 0; s < Comm(r)->nRanks; s++) + EXPECT_EQ(std::string(size, char('A' + s)), std::string(&data[r][(size_t)s * size], size)) + << "rank " << r << " slice " << s; + } + // Splits ranks_ (which become parents_) into one child comm per group; group members are parent ranks in + // child-rank order. + std::vector Split(const std::vector>& groups, bool shareResources) { + parents_ = std::move(ranks_); + ranks_.clear(); + std::vector*> parentRanks; + std::vector parentOf; + for (auto& p : parents_) { + p->comm->shareResources = shareResources; + p->topParentRanks.resize(parents_.size()); + for (int i = 0; i < (int)parents_.size(); i++) p->topParentRanks[i] = 100 + i; + p->comm->topParentRanks = p->topParentRanks.data(); + } + std::vector> members = groups; + for (auto& g : members) { + for (int i = 0; i < (int)g.size(); i++) { + auto child = std::make_unique(); + child->comm->rank = i; + child->comm->nRanks = g.size(); + child->comm->abortFlag = &child->abort; + child->comm->ncclNet = &g_fakeNet; + child->topParentRanks.resize(g.size()); + child->comm->topParentRanks = child->topParentRanks.data(); + parentRanks.push_back(&g); + parentOf.push_back(parents_[g[i]]->comm.get()); + ranks_.push_back(std::move(child)); + } + } + return OnRanks([&](int r) { + return bootstrapSplit(0xC0FFEE, Comm(r), parentOf[r], 0, 0, parentRanks[r]->data()); + }); + } +}; + +TEST_F(BootstrapMicrotest, RootMath_EveryRankPastTheOffsetBelongsToExactlyOneRoot) { + for (int nRanks = 1; nRanks <= 12; nRanks++) + for (int nRoots = 1; nRoots <= std::min(4, nRanks); nRoots++) + for (int offset = 0; offset < std::min(3, nRanks - nRoots + 1); offset++) { + SCOPED_TRACE(testing::Message() << nRanks << " ranks " << nRoots << " roots offset " << offset); + int next = offset; + for (int root = 0; root < nRoots; root++) { + int first = firstRankFromRoot(root, nRanks, nRoots, offset); + EXPECT_EQ(next, first); + EXPECT_EQ(nRankFromRoot(root, nRanks, nRoots, offset), nRankFromRoot(root + nRoots, nRanks, nRoots, offset)); + for (int k = 0; k < nRankFromRoot(root, nRanks, nRoots, offset); k++, next++) { + EXPECT_EQ(root, rootIdFromRank(next, nRanks, nRoots, offset)); + EXPECT_EQ(k, localIdFromRoot(next, root, nRanks, nRoots, offset)); + EXPECT_EQ(k == 0, isFirstFromRoot(next, root, nRanks, nRoots, offset)); + } + } + EXPECT_EQ(nRanks, next); + if (offset > 0) EXPECT_EQ(-1, rootIdFromRank(offset - 1, nRanks, nRoots, offset)); + } + EXPECT_EQ(-1, rootIdFromRank(3, 8, 0, 0)); + EXPECT_EQ(0, firstRankFromRoot(-1, 8, 2, 0)); + EXPECT_EQ(0, nRankFromRoot(-1, 8, 2, 0)); + EXPECT_EQ(5, localIdFromRoot(5, -1, 8, 2, 0)); +} + +TEST_F(BootstrapMicrotest, BidirEnabled_SocketKnobIsOnUnlessDisabledOrNetTakesOver) { + EXPECT_FALSE(bootstrapBidirEnabled(2, 0)); + EXPECT_TRUE(bootstrapBidirEnabled(3, 0)); + g_params["BOOTSTRAP_BIDIR_ALLGATHER"] = 0; + EXPECT_FALSE(bootstrapBidirEnabled(3, 0)); + g_params["BOOTSTRAP_BIDIR_ALLGATHER"] = 1; + g_params["OOB_NET_ENABLE"] = 1; + EXPECT_FALSE(bootstrapBidirEnabled(3, 0)); + EXPECT_FALSE(bootstrapBidirEnabled(3, 2)); + EXPECT_FALSE(bootstrapBidirEnabled(3, -1)); +} + +TEST_F(BootstrapMicrotest, BidirEnabled_NetKnobIsATristateGatedByTheThreshold) { + EXPECT_FALSE(bootstrapBidirEnabled(200, 1)); // net OOB off by default + g_params["OOB_NET_ENABLE"] = 1; + EXPECT_FALSE(bootstrapBidirEnabled(200, 1)); // bidir net off by default + g_params["BOOTSTRAP_BIDIR_NET"] = 1; + EXPECT_TRUE(bootstrapBidirEnabled(3, 1)); + g_params["BOOTSTRAP_BIDIR_NET"] = -1; + g_params["BOOTSTRAP_BIDIR_THRESHOLD"] = 8; + EXPECT_FALSE(bootstrapBidirEnabled(7, 1)); + EXPECT_TRUE(bootstrapBidirEnabled(8, 1)); + g_params["BOOTSTRAP_BIDIR_THRESHOLD"] = 0; + EXPECT_FALSE(bootstrapBidirEnabled(8, 1)); +} + +TEST_F(BootstrapMicrotest, NetEnabled_AutoModeUsesTheBidirThreshold) { + EXPECT_FALSE(bootstrapNetEnabledEffective(1000)); + g_params["OOB_NET_ENABLE"] = 2; + EXPECT_TRUE(bootstrapNetEnabledEffective(1)); + g_params["OOB_NET_ENABLE"] = -1; + g_params["BOOTSTRAP_BIDIR_THRESHOLD"] = 4; + EXPECT_FALSE(bootstrapNetEnabledEffective(3)); + EXPECT_TRUE(bootstrapNetEnabledEffective(4)); + g_params["BOOTSTRAP_BIDIR_THRESHOLD"] = 0; + EXPECT_FALSE(bootstrapNetEnabledEffective(4)); +} + +TEST_F(BootstrapMicrotest, CheckAbort_ReadsTheFlagOnlyOnceEveryTenThousandCalls) { + uint32_t flag = 1; + int cntr = 1; + EXPECT_EQ(ncclSuccess, checkAbort(&flag, &cntr)); + EXPECT_EQ(2, cntr); + cntr = BOOTSTRAP_N_CHECK_ABORT - 1; + EXPECT_EQ(ncclSuccess, checkAbort(&flag, &cntr)); + EXPECT_EQ(0, cntr); + EXPECT_EQ(ncclInternalError, checkAbort(&flag, &cntr)); + EXPECT_EQ(0, cntr); + EXPECT_EQ(ncclSuccess, checkAbort(nullptr, &cntr)); + flag = 0; + cntr = 0; + EXPECT_EQ(ncclSuccess, checkAbort(&flag, &cntr)); +} + +TEST_F(BootstrapMicrotest, SocketRecv_SizePrefixedMessageFitsOrIsRejectedAsTruncated) { + struct ncclSocket a, b; + Pair(&a, &b); + char out[8] = {}; + ASSERT_EQ(ncclSuccess, socketSend(&a, (void*)"abc", 3)); + EXPECT_EQ(ncclSuccess, socketRecv(&b, out, sizeof(out))); + EXPECT_STREQ("abc", out); + ASSERT_EQ(ncclSuccess, socketSend(&a, nullptr, 0)); // header only + EXPECT_EQ(ncclSuccess, socketRecv(&b, nullptr, 0)); + ASSERT_EQ(ncclSuccess, socketSend(&a, (void*)"too long", 8)); + EXPECT_EQ(ncclInternalError, socketRecv(&b, out, 4)); + EXPECT_TRUE(Warned("Message truncated : received 8 bytes instead of 4")); + (void)ncclSocketClose(&a); + (void)ncclSocketClose(&b); +} + +TEST_F(BootstrapMicrotest, SocketSendRecv_ShortMessageReceivedTooLongRejected) { + struct ncclSocket a, b; + Pair(&a, &b); + char fromA[4] = {}, fromB[4] = {}; + ASSERT_EQ(ncclSuccess, socketSend(&b, (void*)"xy", 2)); // b's half of the exchange, sent ahead + EXPECT_EQ(ncclSuccess, socketSendRecv(&a, (void*)"pq", 2, &a, fromB, sizeof(fromB))); + EXPECT_STREQ("xy", fromB); + EXPECT_EQ(ncclSuccess, socketRecv(&b, fromA, sizeof(fromA))); + EXPECT_STREQ("pq", fromA); + ASSERT_EQ(ncclSuccess, socketSend(&b, (void*)"long", 4)); + EXPECT_EQ(ncclInternalError, socketSendRecv(&a, (void*)"pq", 2, &a, fromB, 2)); + EXPECT_TRUE(Warned("Message truncated : received 4 bytes instead of 2")); + (void)ncclSocketClose(&a); + (void)ncclSocketClose(&b); +} + +TEST_F(BootstrapMicrotest, SocketDoubleSendRecv_EitherDirectionTooLongIsRejected) { + struct ncclSocket a, b; + Pair(&a, &b); + char r1[4] = {}, r2[4] = {}; + // Both "peers" are b, so its two size words and both payloads are queued up front. + int three = 3, one = 1; + ASSERT_EQ(ncclSuccess, ncclSocketSend(&b, &three, sizeof(int))); + ASSERT_EQ(ncclSuccess, ncclSocketSend(&b, &one, sizeof(int))); + ASSERT_EQ(ncclSuccess, ncclSocketSend(&b, (void*)"abcd", 4)); + struct ncclSocketOp ops[4] = {{NCCL_SOCKET_SEND, &a, (void*)"s1", 2, 0}, + {NCCL_SOCKET_RECV, &a, r1, 4, 0}, + {NCCL_SOCKET_SEND, &a, (void*)"s2", 2, 0}, + {NCCL_SOCKET_RECV, &a, r2, 4, 0}}; + EXPECT_EQ(ncclSuccess, socketDoubleSendRecv(ops)); + EXPECT_EQ(3, ops[1].size); + EXPECT_EQ(1, ops[3].size); + EXPECT_STREQ("abc", r1); + EXPECT_STREQ("d", r2); + int big = 9; + ASSERT_EQ(ncclSuccess, ncclSocketSend(&b, &one, sizeof(int))); + ASSERT_EQ(ncclSuccess, ncclSocketSend(&b, &big, sizeof(int))); + ops[1].size = ops[3].size = 4; + EXPECT_EQ(ncclInternalError, socketDoubleSendRecv(ops)); + EXPECT_TRUE(Warned("Message truncated : received 1,9 bytes instead of 4,4")); + (void)ncclSocketClose(&a); + (void)ncclSocketClose(&b); +} + +// Loops one comm back to itself so the ring helpers can be driven on a single thread. +struct NetLoop { + void *listen = nullptr, *send = nullptr, *recv = nullptr; + NetLoop() { + char handle[NCCL_NET_HANDLE_MAXSIZE]; + EXPECT_EQ(ncclSuccess, g_fakeNet.listen(nullptr, 0, handle, &listen)); + EXPECT_EQ(ncclSuccess, g_fakeNet.connect(nullptr, 0, handle, &send, nullptr)); + EXPECT_EQ(ncclSuccess, g_fakeNet.accept(listen, &recv, nullptr)); + } +}; + +TEST_F(BootstrapMicrotest, NetMultiOp_RejectsEachMalformedOpBeforePostingAny) { + NetLoop loop; + char buf[4]; + int posts = 0; + g_netFault = [&](const char* op) { + if (std::strcmp(op, "isend") == 0 || std::strcmp(op, "irecv") == 0) posts++; + return ncclSuccess; + }; + EXPECT_EQ(ncclInvalidArgument, netMultiOp(&g_fakeNet, nullptr, 1, nullptr)); + EXPECT_TRUE(Warned("netMultiOp: invalid arguments")); + struct ncclNetOp ok = {NCCL_NET_OP_SEND, loop.send, buf, 4, nullptr, 0, nullptr, 0}; + EXPECT_EQ(ncclInvalidArgument, netMultiOp(&g_fakeNet, &ok, 0, nullptr)); + struct ncclNetOp bad[4] = {ok, ok, ok, ok}; + bad[0].op = 7; + bad[1].comm = nullptr; + bad[2].size = -1; + bad[3].data = nullptr; + const char* warns[4] = {"invalid op 7 at index 1", "invalid comm at index 1", "invalid size -1 at index 1", + "NULL data with size 4 at index 1"}; + for (int i = 0; i < 4; i++) { + struct ncclNetOp ops[2] = {ok, bad[i]}; + EXPECT_EQ(ncclInvalidArgument, netMultiOp(&g_fakeNet, ops, 2, nullptr)) << warns[i]; + EXPECT_TRUE(Warned(warns[i])) << warns[i]; + } + EXPECT_EQ(0, posts); +} + +TEST_F(BootstrapMicrotest, NetMultiOp_PollsUntilDeferredPostsAndCompletionsLand) { + void *listen, *send, *recv; + char handle[NCCL_NET_HANDLE_MAXSIZE]; + ASSERT_EQ(ncclSuccess, g_fakeNet.listen(nullptr, 0, handle, &listen)); + ASSERT_EQ(ncclSuccess, g_fakeNet.connect(nullptr, 0, handle, &send, nullptr)); + ASSERT_EQ(ncclSuccess, g_fakeNet.accept(listen, &recv, nullptr)); + char out[3] = {}; + struct ncclNetOp ops[2] = {{NCCL_NET_OP_RECV, recv, out, 2, nullptr, 5, nullptr, 0}, + {NCCL_NET_OP_SEND, send, (void*)"hi", 2, nullptr, 5, nullptr, 0}}; + g_netDeferIrecv = 2; + g_netDeferIsend = 2; + g_netDeferTest = 3; + uint32_t flag = 0; + EXPECT_EQ(ncclSuccess, netMultiOp(&g_fakeNet, ops, 2, &flag)); + EXPECT_STREQ("hi", out); + EXPECT_EQ(1, ops[0].done); + EXPECT_EQ(nullptr, ops[0].req); + EXPECT_EQ(0, g_netDeferTest.load()); + struct ncclNetOp empty[2] = {{NCCL_NET_OP_RECV, recv, nullptr, 0, nullptr, 6, nullptr, 0}, + {NCCL_NET_OP_SEND, send, nullptr, 0, nullptr, 6, nullptr, 0}}; + EXPECT_EQ(ncclSuccess, netMultiOp(&g_fakeNet, empty, 2, &flag)); // no payload needs no buffer +} + +TEST_F(BootstrapMicrotest, NetMultiOp_AbortStopsTheWaitAndPluginErrorsPropagate) { + void *listen, *send, *recv; + char handle[NCCL_NET_HANDLE_MAXSIZE], out[2]; + ASSERT_EQ(ncclSuccess, g_fakeNet.listen(nullptr, 0, handle, &listen)); + ASSERT_EQ(ncclSuccess, g_fakeNet.connect(nullptr, 0, handle, &send, nullptr)); + ASSERT_EQ(ncclSuccess, g_fakeNet.accept(listen, &recv, nullptr)); + struct ncclNetOp recvOp = {NCCL_NET_OP_RECV, recv, out, 2, nullptr, 1, nullptr, 0}; + uint32_t flag = 1; // nothing was sent, so only the abort ends this + EXPECT_EQ(ncclInternalError, netMultiOp(&g_fakeNet, &recvOp, 1, &flag)); + g_netFault = [](const char* op) { return std::strcmp(op, "isend") == 0 ? ncclSystemError : ncclSuccess; }; + struct ncclNetOp sendOp = {NCCL_NET_OP_SEND, send, out, 2, nullptr, 1, nullptr, 0}; + EXPECT_EQ(ncclSystemError, netMultiOp(&g_fakeNet, &sendOp, 1, nullptr)); +} + +TEST_F(BootstrapMicrotest, NetIsendIrecv_ADoneOpIsNotPostedAgain) { + int done = 1; + void* req = nullptr; + EXPECT_EQ(ncclSuccess, netIsend(&g_fakeNet, nullptr, nullptr, 0, nullptr, 0, &req, &done)); + EXPECT_EQ(ncclSuccess, netIrecv(&g_fakeNet, nullptr, nullptr, 0, nullptr, 0, &req, &done)); + EXPECT_EQ(nullptr, req); +} + +TEST_F(BootstrapMicrotest, NetRingAllGather_RegistrationOrTransferFailureStillDeregisters) { + NetLoop loop; + char data[8] = "abcdefg"; + for (const char* op : {"regMr", "isend"}) { + SCOPED_TRACE(op); + std::atomic calls(0); + g_netFault = [&](const char* o) { + return std::strcmp(o, op) == 0 && ++calls == 2 ? ncclSystemError : ncclSuccess; + }; + g_netRegs = g_netDeregs = 0; + EXPECT_EQ(ncclSystemError, netRingAllGather(&g_fakeNet, loop.send, loop.recv, 0, 4, data, 2, nullptr)); + EXPECT_EQ(g_netRegs.load(), g_netDeregs.load()); + } +} + +TEST_F(BootstrapMicrotest, NetBiDirRingAllGather_EachFailurePointStillDeregistersWhatItRegistered) { + NetLoop fwd, rev; + char data[8] = "abcdefg"; + struct Fault { + const char* op; + int nth, regs; + } faults[] = {{"regMr", 1, 0}, {"regMr", 2, 1}, {"regMr", 3, 2}, {"regMr", 4, 3}, {"isend", 1, 4}, {"isend", 3, 4}}; + for (const auto& f : faults) { + SCOPED_TRACE(testing::Message() << f.op << " #" << f.nth); + std::atomic calls(0); + g_netFault = [&](const char* o) { + return std::strcmp(o, f.op) == 0 && ++calls == f.nth ? ncclSystemError : ncclSuccess; + }; + g_netRegs = g_netDeregs = 0; + EXPECT_EQ(ncclSystemError, + netBiDirRingAllGather(&g_fakeNet, fwd.send, fwd.recv, rev.send, rev.recv, 0, 4, data, 2, nullptr)); + EXPECT_EQ(f.regs, g_netRegs.load()); + EXPECT_EQ(f.regs, g_netDeregs.load()); + } +} + +TEST_F(BootstrapMicrotest, Accept_AbortRaisedBeforehandFailsWithoutAccepting) { + struct ncclSocket l, sock; + ASSERT_EQ(ncclSuccess, ncclSocketInit(&l, &bootstrapNetIfAddr, 7, ncclSocketTypeBootstrap, &abort_)); + ASSERT_EQ(ncclSuccess, ncclSocketListen(&l)); + uint32_t flag = 1; + EXPECT_EQ(ncclInternalError, bootstrapAccept(&sock, &l, &flag)); + (void)ncclSocketClose(&l); +} + +// The contract bootstrapAccept was written for: a rejected peer comes back as BadHandshake and is skipped. +TEST_F(BootstrapMicrotest, Accept_BadHandshakePeerIsClosedAndTheNextOneAccepted) { + g_rejectedState = ncclSocketStateBadHandshake; + struct ncclSocket l, stray, good, sock; + ASSERT_EQ(ncclSuccess, ncclSocketInit(&l, &bootstrapNetIfAddr, 7, ncclSocketTypeBootstrap, &abort_)); + ASSERT_EQ(ncclSuccess, ncclSocketListen(&l)); + ASSERT_EQ(ncclSuccess, ncclSocketInit(&stray, &l.addr, 8, ncclSocketTypeBootstrap)); + ASSERT_EQ(ncclSuccess, ncclSocketConnect(&stray)); + ASSERT_EQ(ncclSuccess, ncclSocketInit(&good, &l.addr, 7, ncclSocketTypeBootstrap)); + ASSERT_EQ(ncclSuccess, ncclSocketConnect(&good)); + ASSERT_EQ(ncclSuccess, socketSend(&good, (void*)"ok", 2)); + + EXPECT_EQ(ncclSuccess, bootstrapAccept(&sock, &l, &abort_)); + + char out[3] = {}; + EXPECT_EQ(ncclSuccess, socketRecv(&sock, out, 2)); + EXPECT_STREQ("ok", out); + for (auto* s : {&l, &stray, &good, &sock}) (void)ncclSocketClose(s); +} + +// bootstrapAccept retries only on BadHandshake, so it returns a rejected peer unconnected; a fix flips this pin. +TEST_F(BootstrapMicrotest, Accept_KnownBug_RejectedPeerReportedAsAccepting_IsReturnedUnconnected) { + struct ncclSocket l, stray, good, sock; + ASSERT_EQ(ncclSuccess, ncclSocketInit(&l, &bootstrapNetIfAddr, 7, ncclSocketTypeBootstrap, &abort_)); + ASSERT_EQ(ncclSuccess, ncclSocketListen(&l)); + ASSERT_EQ(ncclSuccess, ncclSocketInit(&stray, &l.addr, 8, ncclSocketTypeBootstrap)); + ASSERT_EQ(ncclSuccess, ncclSocketConnect(&stray)); + ASSERT_EQ(ncclSuccess, ncclSocketInit(&good, &l.addr, 7, ncclSocketTypeBootstrap)); + ASSERT_EQ(ncclSuccess, ncclSocketConnect(&good)); + + EXPECT_EQ(ncclSuccess, bootstrapAccept(&sock, &l, &abort_)); + + EXPECT_EQ(ncclSocketStateAccepting, sock.state); + char out[2]; + EXPECT_NE(ncclSuccess, socketRecv(&sock, out, 2)); + for (auto* s : {&l, &stray, &good, &sock}) (void)ncclSocketClose(s); +} + +TEST_F(BootstrapMicrotest, NetInit_CommIdPicksTheMatchingSubnetOrFails) { + bootstrapNetInitDone = 0; + SetMicroEnv("NCCL_COMM_ID", "not-an-address"); + EXPECT_EQ(ncclInvalidArgument, bootstrapNetInit()); + EXPECT_TRUE(Warned("Invalid NCCL_COMM_ID")); + SetMicroEnv("NCCL_COMM_ID", "10.1.2.3:4242"); + g_ifCount = 0; + EXPECT_EQ(ncclSystemError, bootstrapNetInit()); + EXPECT_TRUE(Warned("No usable listening interface found")); + EXPECT_EQ(4242, PortOf(g_subnetRemote)); + g_ifCount = 1; + EXPECT_EQ(ncclSuccess, bootstrapNetInit()); + EXPECT_EQ(1, bootstrapNetInitDone); + EXPECT_STREQ("fake0", bootstrapNetIfName); +} + +TEST_F(BootstrapMicrotest, NetInit_NoInterfaceFailsAndSuccessIsCachedForever) { + bootstrapNetInitDone = 0; + g_ifCount = 0; + EXPECT_EQ(ncclInvalidUsage, bootstrapNetInit()); + EXPECT_TRUE(Warned("Bootstrap : no socket interface found")); + EXPECT_EQ(0, bootstrapNetInitDone); + g_ifCount = 1; + ASSERT_EQ(ncclSuccess, bootstrapNetInit()); + g_ifCount = 0; + EXPECT_EQ(ncclSuccess, bootstrapNetInit()); +} + +TEST_F(BootstrapMicrotest, GetUniqueId_CommIdFromEnvIsUsedAsIsWithoutARoot) { + SetMicroEnv("NCCL_COMM_ID", "10.1.2.3:4242"); + ncclBootstrapHandle h; + std::memset(&h, 0xff, sizeof(h)); + EXPECT_EQ(ncclSuccess, bootstrapGetUniqueId(&h, nullptr)); + EXPECT_EQ((uint64_t)NCCL_MAGIC, h.magic); + EXPECT_EQ(4242, PortOf(h.addr)); + EXPECT_EQ(0, h.nRanks); + EXPECT_EQ(0, OpenListeners()); +} + +TEST_F(BootstrapMicrotest, GetUniqueId_CommIdFromEnvRejectsGrowAndBadAddresses) { + SetMicroEnv("NCCL_COMM_ID", "10.1.2.3:4242"); + ncclBootstrapHandle h; + EXPECT_EQ(ncclInvalidUsage, bootstrapGetUniqueId(&h, self_.comm.get())); + EXPECT_TRUE(Warned("should not be called when NCCL_COMM_ID is set")); + SetMicroEnv("NCCL_COMM_ID", "nonsense"); + EXPECT_EQ(ncclInvalidArgument, bootstrapGetUniqueId(&h, nullptr)); + EXPECT_TRUE(Warned("Invalid NCCL_COMM_ID")); +} + +TEST_F(BootstrapMicrotest, GetUniqueId_GrowDerivesTheNextChildMagicAndStartsARoot) { + self_.comm->magic = 0x1234; + self_.comm->childCount = 2; + self_.comm->nRanks = 6; + ncclBootstrapHandle h; + ASSERT_EQ(ncclSuccess, bootstrapGetUniqueId(&h, self_.comm.get())); + EXPECT_EQ(hashCombine(0x1234, 3), h.magic); + EXPECT_EQ(6, h.nRanks); + EXPECT_NE(0, PortOf(h.addr)); + EXPECT_EQ(1, OpenListeners()); +} + +TEST_F(BootstrapMicrotest, CreateRoot_EachSocketStepFailureIsReturnedAndNoThreadStarts) { + for (const char* step : {"init", "listen", "getaddr"}) { + SCOPED_TRACE(step); + SetSockFault([step](const char* op, struct ncclSocket*) { + return std::strcmp(op, step) == 0 ? ncclSystemError : ncclSuccess; + }); + ncclBootstrapHandle h = {}; + EXPECT_EQ(ncclSystemError, bootstrapCreateRoot(&h, false)); + EXPECT_EQ(0, g_liveRoots); + } +} + +// The fail path after a successful listen frees the socket without closing it. +// Pins AICOMRCCL-3002; a fix flips this pin. +TEST_F(BootstrapMicrotest, CreateRoot_KnownBug_GetAddrFailureLeaksTheListeningSocket) { + SetSockFault([](const char* op, struct ncclSocket*) { + return std::strcmp(op, "getaddr") == 0 ? ncclSystemError : ncclSuccess; + }); + ncclBootstrapHandle h = {}; + ASSERT_EQ(ncclSystemError, bootstrapCreateRoot(&h, false)); + EXPECT_EQ(1, OpenListeners()); +} + +TEST_F(BootstrapMicrotest, Root_EveryArrivalOrderGivesEachRankItsNextAndItsPrevsReverseHandle) { + std::vector order = {0, 1, 2, 3}; + do { + SCOPED_TRACE(testing::Message() << order[0] << order[1] << order[2] << order[3]); + ASSERT_EQ(ncclSuccess, bootstrapGetUniqueId(&handle_, nullptr)); + struct ncclSocket listen[4]; + for (int r : order) CheckIn(r, 4, &listen[r]); + for (int r = 0; r < 4; r++) { + struct ringConnectInfo got = Answer(&listen[r]); + EXPECT_EQ(1 + (r + 1) % 4, got.fwd.handle[0]) << "rank " << r; + EXPECT_EQ(101 + (r + 3) % 4, got.revHandle[0]) << "rank " << r; + } + WaitRoots(); + } while (std::next_permutation(order.begin(), order.end())); +} + +TEST_F(BootstrapMicrotest, Root_InconsistentRankInfoMakesTheRootGiveUp) { + struct Case { + const char* warn; + int rank, nranks, nroots; + } cases[] = {{"mismatch in info from procs, nranks 3 vs 4", 1, 4, 1}, + {"localId 5 is out of range", 5, 3, 1}, + {"rank 0 of 3 ranks has already checked in", 0, 3, 1}}; + for (const auto& c : cases) { + SCOPED_TRACE(c.warn); + ASSERT_EQ(ncclSuccess, bootstrapGetUniqueId(&handle_, nullptr)); + struct ncclSocket first, second; + CheckIn(0, 3, &first); + CheckIn(c.rank, c.nranks, &second, c.nroots); + WaitRoots(); + EXPECT_TRUE(Warned(c.warn)); + (void)ncclSocketClose(&first); + (void)ncclSocketClose(&second); + } +} + +TEST_F(BootstrapMicrotest, Root_RankGoneBeforeItsAnswer_MakesTheRootGiveUp) { + ASSERT_EQ(ncclSuccess, bootstrapGetUniqueId(&handle_, nullptr)); + struct ncclSocket listen[2]; + CheckIn(0, 2, &listen[0]); + (void)ncclSocketClose(&listen[0]); + CheckIn(1, 2, &listen[1]); // completes the ring: the root now answers both, starting with rank 0 + WaitRoots(); + EXPECT_EQ(0u, PendingOn(listen[1])); // gave up rather than answering rank 1 anyway + (void)ncclSocketClose(&listen[1]); +} + +// The Accept_KnownBug_ above makes one stray connection end the root; a fix flips this pin. +TEST_F(BootstrapMicrotest, Root_KnownBug_StrayPeerWithTheWrongMagicMakesTheRootExit) { + ASSERT_EQ(ncclSuccess, bootstrapGetUniqueId(&handle_, nullptr)); + struct ncclSocket stray; + ASSERT_EQ(ncclSuccess, ncclSocketInit(&stray, &handle_.addr, handle_.magic + 1, ncclSocketTypeBootstrap)); + ASSERT_EQ(ncclSuccess, ncclSocketConnect(&stray)); + WaitRoots(); + struct extInfo info = {}; + EXPECT_EQ(ncclRemoteError, sendToRoot(&handle_, self_.comm.get(), &info)); + (void)ncclSocketClose(&stray); +} + +TEST_F(BootstrapMicrotest, Init_FourRanks_AllGathersEveryRanksAddresses) { + ASSERT_EQ(ncclSuccess, InitRanks(4)); + + ASSERT_EQ(4u, g_proxyInits.size()); + const auto& addrs = g_proxyInits[Comm(0)].addrs; + const auto& uds = g_proxyInits[Comm(0)].uds; + EXPECT_EQ(4u, std::set(uds.begin(), uds.end()).size()); + auto* p2p = static_cast(State(0))->peerP2pAddresses; + for (int r = 0; r < 4; r++) { + EXPECT_EQ(0, std::memcmp(addrs.data(), g_proxyInits[Comm(r)].addrs.data(), 4 * sizeof(ncclSocketAddress))); + EXPECT_EQ(uds, g_proxyInits[Comm(r)].uds); + EXPECT_EQ(0, std::memcmp(&addrs[r], &g_proxyInits[Comm(r)].own, sizeof(ncclSocketAddress))) << r; + auto* state = static_cast(State(r)); + EXPECT_EQ(0, std::memcmp(&p2p[r], &state->listen.peerSocket.addr, sizeof(ncclSocketAddress))) << r; + } + ASSERT_EQ(4u, g_rasAdded.size()); + EXPECT_EQ(13, g_rasAdded[3].nvmlDev); + EXPECT_TRUE(AllOk(CloseRanks())); +} + +TEST_F(BootstrapMicrotest, Init_WithNeitherHandlesNorParent_FailsBeforeTouchingTheNetwork) { + MakeRanks(1); + EXPECT_EQ(ncclSystemError, bootstrapInit(0, nullptr, Comm(0), nullptr)); + EXPECT_TRUE(Warned("bootstrapInit: handles and parent are NULL")); + EXPECT_EQ(0, OpenListeners()); +} + +TEST_F(BootstrapMicrotest, Init_SingleRankEachSocketFailureIsReturned) { + struct Fault { + const char* op; + int nth; + } faults[] = {{"listen", 1}, {"listen", 2}, {"listen", 3}, {"listen", 4}, {"connect", 1}, {"send", 1}, + {"accept", 1}, {"recv", 2}, {"connect", 2}, {"accept", 2}, {"ready", 1}, {"send", 2}}; + for (const auto& f : faults) { + SCOPED_TRACE(testing::Message() << f.op << " #" << f.nth); + ranks_.clear(); + MakeRanks(1); + ASSERT_EQ(ncclSuccess, bootstrapGetUniqueId(&handle_, nullptr)); + std::map calls; + // The root thread's own socket calls are left alone. + SetSockFault([&](const char* op, struct ncclSocket*) { + if (std::this_thread::get_id() != main_) return ncclSuccess; + return ++calls[op] == f.nth && std::strcmp(op, f.op) == 0 ? ncclSystemError : ncclSuccess; + }); + EXPECT_NE(ncclSuccess, bootstrapInit(1, &handle_, Comm(0), nullptr)); + SetSockFault(nullptr); + __atomic_store_n(&ranks_[0]->abort, 1u, __ATOMIC_RELEASE); + EXPECT_EQ(ncclSuccess, bootstrapClose(State(0))); + Comm(0)->bootstrap = nullptr; + RestartSockets(); + } +} + +TEST_F(BootstrapMicrotest, Init_StaggeredRanksStillMeetAtTheRoot) { + g_params["UID_STAGGER_THRESHOLD"] = 1; + g_params["UID_STAGGER_RATE"] = 100; // 100 msgs/s: local rank k waits k * 10 ms + auto start = Clock::now(); + ASSERT_EQ(ncclSuccess, InitRanks(3)); + EXPECT_GE(Clock::now() - start, std::chrono::milliseconds(20)); + ExpectAllGather(2); +} + +TEST_F(BootstrapMicrotest, Init_RasOffOrFailingIsSkippedWithoutFailingInit) { + g_params["RAS_ENABLE"] = 0; + ASSERT_EQ(ncclSuccess, InitRanks(2)); + EXPECT_EQ(0, g_rasCommInits); + EXPECT_TRUE(g_rasAdded.empty()); + EXPECT_TRUE(AllOk(CloseRanks())); + + ranks_.clear(); + g_params.erase("RAS_ENABLE"); + g_rasCommInitResult = ncclSystemError; + ASSERT_EQ(ncclSuccess, InitRanks(2)); + EXPECT_EQ(2, g_rasCommInits); + EXPECT_TRUE(g_rasAdded.empty()); + EXPECT_TRUE(AllOk(CloseRanks())); + + ranks_.clear(); + g_rasCommInitResult = ncclSuccess; + g_rasAddRanksResult = ncclSystemError; + ASSERT_EQ(ncclSuccess, InitRanks(2)); + EXPECT_TRUE(g_rasAdded.empty()); +} + +// Any failure past the proxy listen frees proxySocket without closing it. Pins AICOMRCCL-3002; a fix flips this pin. +TEST_F(BootstrapMicrotest, Init_KnownBug_ProxyInitFailureLeaksTheProxyListenSocket) { + g_proxyInitResult = ncclSystemError; + EXPECT_EQ(ncclSystemError, InitRanks(2)); + EXPECT_TRUE(AllOk(CloseRanks())); + WaitRoots(); + EXPECT_EQ(2, OpenListeners()); +} + +TEST_F(BootstrapMicrotest, Init_AbortWhileARankIsMissing_UnblocksTheOthers) { + MakeRanks(3); + ASSERT_EQ(ncclSuccess, bootstrapGetUniqueId(&handle_, nullptr)); + watchdog_ = std::chrono::milliseconds(50); + auto res = OnRanks([&](int r) { return r == 2 ? ncclSuccess : bootstrapInit(1, &handle_, Comm(r), nullptr); }); + EXPECT_EQ(ncclInternalError, res[0]); + EXPECT_EQ(ncclInternalError, res[1]); +} + +TEST_F(BootstrapMicrotest, Init_SeveralRootsShareTheRendezvousAndStillCloseOneRing) { + for (int nRoots : {2, 3}) + for (int n : {nRoots, 5, 7}) { + SCOPED_TRACE(testing::Message() << nRoots << " roots " << n << " ranks"); + ranks_.clear(); + MakeRanks(n); + std::vector ids(nRoots); + for (auto& id : ids) ASSERT_EQ(ncclSuccess, bootstrapGetUniqueId((ncclBootstrapHandle*)&id, nullptr)); + ASSERT_TRUE(AllOk(OnRanks([&](int r) { return bootstrapInit(nRoots, ids.data(), Comm(r), nullptr); }))); + ExpectAllGather(5); + EXPECT_TRUE(AllOk(CloseRanks())); + WaitRoots(); + } +} + +TEST_F(BootstrapMicrotest, AllGather_BidirAndUnidirRingsAgreeForEveryRankCount) { + for (int bidir = 0; bidir <= 1; bidir++) { + g_params["BOOTSTRAP_BIDIR_ALLGATHER"] = bidir; + for (int n = 1; n <= 6; n++) { + SCOPED_TRACE(testing::Message() << "bidir " << bidir << " nranks " << n); + ranks_.clear(); + ASSERT_EQ(ncclSuccess, InitRanks(n)); + g_multiOps = 0; + ExpectAllGather(3); + EXPECT_EQ(bidir && n >= 3, g_multiOps > 0); + EXPECT_TRUE(AllOk(CloseRanks())); + } + } +} + +TEST_F(BootstrapMicrotest, InitOverNet_UnidirRingAllGathersAndCloseReleasesEveryComm) { + g_params["OOB_NET_ENABLE"] = 1; + for (int n = 1; n <= 4; n++) { + SCOPED_TRACE(n); + ranks_.clear(); + g_netCloses = 0; + ASSERT_EQ(ncclSuccess, InitRanks(n)); + g_netRegs = g_netDeregs = 0; + ExpectAllGather(4); + EXPECT_EQ(2 * n, g_netRegs.load()); // the buffer, once per ring comm + EXPECT_EQ(g_netRegs.load(), g_netDeregs.load()); + EXPECT_TRUE(AllOk(CloseRanks())); + EXPECT_EQ(3 * n, g_netCloses.load()); // send, recv, listen + WaitRoots(); + } +} + +TEST_F(BootstrapMicrotest, InitOverNet_BidirRingUsesThePiggybackedReverseHandles) { + g_params["OOB_NET_ENABLE"] = 1; + g_params["BOOTSTRAP_BIDIR_NET"] = 1; + for (int n = 3; n <= 6; n++) { + SCOPED_TRACE(n); + ranks_.clear(); + g_netCloses = 0; + ASSERT_EQ(ncclSuccess, InitRanks(n)); + for (int r = 0; r < n; r++) { + auto* state = static_cast(State(r)); + EXPECT_NE(nullptr, state->ring.net.revSendComm); + EXPECT_NE(nullptr, state->ring.net.revRecvComm); + } + g_netRegs = g_netDeregs = 0; + ExpectAllGather(3); + EXPECT_EQ(4 * n, g_netRegs.load()); // forward and reverse pairs + EXPECT_EQ(g_netRegs.load(), g_netDeregs.load()); + EXPECT_TRUE(AllOk(CloseRanks())); + EXPECT_EQ(6 * n, g_netCloses.load()); + WaitRoots(); + } +} + +TEST_F(BootstrapMicrotest, InitOverNet_SlowConnectsAndPostsAreRetriedUntilTheyLand) { + g_params["OOB_NET_ENABLE"] = 1; + g_params["BOOTSTRAP_BIDIR_NET"] = 1; + g_netDeferConnect = 40; + g_netDeferIsend = 40; + g_netDeferIrecv = 40; + ASSERT_EQ(ncclSuccess, InitRanks(4)); + ExpectAllGather(2); + EXPECT_EQ(0, g_netDeferConnect.load()); +} + +TEST_F(BootstrapMicrotest, InitOverNet_ListenFailuresAreReturned) { + g_params["OOB_NET_ENABLE"] = 1; + g_params["BOOTSTRAP_BIDIR_NET"] = 1; + for (int nth : {1, 2}) { + SCOPED_TRACE(nth); + ranks_.clear(); + MakeRanks(3); + std::atomic listens(0); + g_netFault = [&](const char* op) { + return std::strcmp(op, "listen") == 0 && ++listens == nth ? ncclSystemError : ncclSuccess; + }; + ASSERT_EQ(ncclSuccess, bootstrapGetUniqueId(&handle_, nullptr)); + EXPECT_EQ(ncclSystemError, bootstrapInit(1, &handle_, Comm(0), nullptr)); + g_netFault = nullptr; + EXPECT_EQ(ncclSuccess, bootstrapClose(State(0))); + Comm(0)->bootstrap = nullptr; + RestartSockets(); + } +} + +// The reverse handles cannot cross roots, so several roots fall back to the forward net ring. +TEST_F(BootstrapMicrotest, InitOverNet_SeveralRootsKeepTheUnidirectionalRing) { + g_params["OOB_NET_ENABLE"] = 1; + g_params["BOOTSTRAP_BIDIR_NET"] = 1; + MakeRanks(5); + ncclUniqueId ids[2]; + for (auto& id : ids) ASSERT_EQ(ncclSuccess, bootstrapGetUniqueId((ncclBootstrapHandle*)&id, nullptr)); + ASSERT_TRUE(AllOk(OnRanks([&](int r) { return bootstrapInit(2, ids, Comm(r), nullptr); }))); + for (int r = 0; r < 5; r++) EXPECT_EQ(nullptr, static_cast(State(r))->ring.net.revSendComm); + ExpectAllGather(2); + EXPECT_TRUE(AllOk(CloseRanks())); +} + +TEST_F(BootstrapMicrotest, BidirRingSetup_HandleExchangeFailureDeregistersAndIsReturned) { + g_params["OOB_NET_ENABLE"] = 1; + g_params["BOOTSTRAP_BIDIR_NET"] = 1; + NetLoop loop; + struct bootstrapState state = {}; + state.nranks = 3; + state.net = &g_fakeNet; + state.ring.net.sendComm = loop.send; + state.ring.net.recvComm = loop.recv; + state.abortFlag = &abort_; + for (int nth : {1, 2}) { + SCOPED_TRACE(nth); + std::atomic regs(0); + g_netFault = [&](const char* o) { + return std::strcmp(o, "regMr") == 0 && ++regs == nth ? ncclSystemError : ncclSuccess; + }; + g_netRegs = g_netDeregs = 0; + EXPECT_EQ(ncclSystemError, bootstrapBidirRingSetup(self_.comm.get(), &state, nullptr)); + EXPECT_EQ(g_netRegs.load(), g_netDeregs.load()); + } + EXPECT_NE(nullptr, state.listen.net.revComm); // the in-place listen, done once +} + +TEST_F(BootstrapMicrotest, BidirRingSetup_PiggybackedHandleSkipsTheExchange) { + g_params["OOB_NET_ENABLE"] = 1; + g_params["BOOTSTRAP_BIDIR_NET"] = 1; + struct bootstrapState state = {}; + state.nranks = 3; + state.net = &g_fakeNet; + state.abortFlag = &abort_; + ASSERT_EQ(ncclSuccess, g_fakeNet.listen(nullptr, 0, state.listen.net.revHandle, &state.listen.net.revComm)); + char prevRevHandle[NCCL_NET_HANDLE_MAXSIZE]; + std::memcpy(prevRevHandle, state.listen.net.revHandle, sizeof(prevRevHandle)); // loops back to itself + + EXPECT_EQ(ncclSuccess, bootstrapBidirRingSetup(self_.comm.get(), &state, prevRevHandle)); + + EXPECT_NE(nullptr, state.ring.net.revSendComm); + EXPECT_NE(nullptr, state.ring.net.revRecvComm); + EXPECT_EQ(0, g_netRegs.load()); + EXPECT_EQ(1u, g_netListens.size()); +} + +TEST_F(BootstrapMicrotest, BidirRingSetupStart_NeedsAPiggybackedHandleToStartConnecting) { + g_params["OOB_NET_ENABLE"] = 1; + g_params["BOOTSTRAP_BIDIR_NET"] = 1; + struct bootstrapState state = {}; + state.nranks = 3; + state.net = &g_fakeNet; + char out[NCCL_NET_HANDLE_MAXSIZE], zeros[NCCL_NET_HANDLE_MAXSIZE] = {}; + bool started = true; + EXPECT_EQ(ncclSuccess, bootstrapBidirRingSetupStart(self_.comm.get(), &state, nullptr, out, &started)); + EXPECT_FALSE(started); + EXPECT_NE(nullptr, state.listen.net.revComm); + EXPECT_EQ(ncclSuccess, bootstrapBidirRingSetupStart(self_.comm.get(), &state, zeros, out, &started)); + EXPECT_FALSE(started); + EXPECT_EQ(1u, g_netListens.size()); + state.nranks = 2; + EXPECT_EQ(ncclSuccess, bootstrapBidirRingSetupStart(self_.comm.get(), &state, zeros, out, &started)); + EXPECT_EQ(ncclSuccess, bootstrapBidirRingSetupFinish(self_.comm.get(), &state, zeros)); + EXPECT_EQ(ncclSuccess, bootstrapBidirRingSetup(self_.comm.get(), &state, zeros)); +} + +// devOOB is a function-local static, so each NCCL_OOB_NET_IFNAME case runs in a fresh process. +class BootstrapOobDeviceMicrotest : public BootstrapMicrotest { + protected: + void ExpectDevice(const char* ifname, int expectDev, ncclResult_t expectRes = ncclSuccess, + const char* warn = nullptr) { + const std::string style = ::testing::GTEST_FLAG(death_test_style); + ::testing::GTEST_FLAG(death_test_style) = "threadsafe"; + EXPECT_EXIT( + { + SetMicroEnv("NCCL_OOB_NET_IFNAME", ifname); + self_.comm->ncclNet = &g_fakeNet; + int dev = -1; + bool ok = netGetDevice(0, self_.comm.get(), &dev) == expectRes && (warn == nullptr || Warned(warn)); + if (ok && expectRes == ncclSuccess) { + SetMicroEnv("NCCL_OOB_NET_IFNAME", "nothing-matches"); // the first answer is cached + int again = -1; + ok = dev == expectDev && netGetDevice(0, self_.comm.get(), &again) == ncclSuccess && again == expectDev; + } + std::exit(ok ? 0 : 1); + }, + ::testing::ExitedWithCode(0), ""); + ::testing::GTEST_FLAG(death_test_style) = style; + } + void SetUp() override { + BootstrapMicrotest::SetUp(); + g_netDevs = {"mlx5_10", "mlx5_1", "ib0"}; + } +}; + +TEST_F(BootstrapOobDeviceMicrotest, Unset_PicksDeviceZero) { ExpectDevice("", 0); } +TEST_F(BootstrapOobDeviceMicrotest, Prefix_PicksTheFirstDeviceItPrefixes) { ExpectDevice("mlx5_1", 0); } +TEST_F(BootstrapOobDeviceMicrotest, Exact_SkipsLongerNames) { ExpectDevice("=mlx5_1", 1); } +TEST_F(BootstrapOobDeviceMicrotest, Excluded_PicksTheFirstOtherDevice) { ExpectDevice("^mlx5_10", 1); } +TEST_F(BootstrapOobDeviceMicrotest, ExcludedExact_PicksTheFirstOtherDevice) { ExpectDevice("^=mlx5_10", 1); } +TEST_F(BootstrapOobDeviceMicrotest, NoMatch_IsAnInvalidArgument) { + ExpectDevice("=mlx5", -1, ncclInvalidArgument, "no device found matching exactly mlx5"); +} +TEST_F(BootstrapOobDeviceMicrotest, EverythingExcluded_IsAnInvalidArgument) { + ExpectDevice("^mlx5_1,ib", -1, ncclInvalidArgument, "no device found after excluding mlx5_1,ib"); +} +TEST_F(BootstrapOobDeviceMicrotest, DeviceQueryFailure_IsReturned) { + g_netFault = [](const char* op) { return std::strcmp(op, "devices") == 0 ? ncclSystemError : ncclSuccess; }; + ExpectDevice("ib", -1, ncclSystemError); +} +TEST_F(BootstrapOobDeviceMicrotest, PropertiesFailureOnlyAffectsTheLog) { + g_netFault = [](const char* op) { return std::strcmp(op, "getProperties") == 0 ? ncclSystemError : ncclSuccess; }; + ExpectDevice("", 0); +} + +TEST_F(BootstrapMicrotest, Split_OneChildPerColorEachWithItsOwnRing) { + ASSERT_EQ(ncclSuccess, InitRanks(5)); + ASSERT_TRUE(AllOk(Split({{4, 1, 3}, {0, 2}}, /*shareResources=*/false))); + ExpectAllGather(3); + EXPECT_EQ(5u, g_proxyInits.size() - 5); // the parents' five, then one per child + EXPECT_TRUE(AllOk(CloseRanks())); +} + +TEST_F(BootstrapMicrotest, Split_SharedResourcesMapsTopParentRanksAndSkipsTheProxy) { + ASSERT_EQ(ncclSuccess, InitRanks(4)); + size_t proxies = g_proxyInits.size(); + ASSERT_TRUE(AllOk(Split({{3, 1, 0, 2}}, /*shareResources=*/true))); + ExpectAllGather(2); + EXPECT_EQ(proxies, g_proxyInits.size()); + EXPECT_EQ(std::vector({103, 101, 100, 102}), ranks_[0]->topParentRanks); + EXPECT_TRUE(AllOk(CloseRanks())); +} + +TEST_F(BootstrapMicrotest, Split_NetModeExchangesReverseHandlesOverTheForwardRing) { + g_params["OOB_NET_ENABLE"] = 1; + g_params["BOOTSTRAP_BIDIR_NET"] = 1; + ASSERT_EQ(ncclSuccess, InitRanks(4)); + g_netDeferConnect = 20; + ASSERT_TRUE(AllOk(Split({{1, 2, 3, 0}}, false))); + for (int r = 0; r < 4; r++) EXPECT_NE(nullptr, static_cast(State(r))->ring.net.revSendComm); + ExpectAllGather(4); + EXPECT_TRUE(AllOk(CloseRanks())); +} + +TEST_F(BootstrapMicrotest, Split_RasInitFailureIsOnlyLogged) { + ASSERT_EQ(ncclSuccess, InitRanks(2)); + g_rasCommInitResult = ncclSystemError; + EXPECT_TRUE(AllOk(Split({{1, 0}}, false))); + ExpectAllGather(1); +} + +// The same proxySocket leak on the split path. Pins AICOMRCCL-3002; a fix flips this pin. +TEST_F(BootstrapMicrotest, Split_KnownBug_ProxyInitFailureLeaksTheProxyListenSocket) { + ASSERT_EQ(ncclSuccess, InitRanks(2)); + g_proxyInitResult = ncclSystemError; + auto res = Split({{1, 0}}, false); + EXPECT_EQ(ncclSystemError, res[0]); + EXPECT_EQ(ncclSystemError, res[1]); + EXPECT_TRUE(AllOk(CloseRanks())); + for (auto& p : parents_) { + EXPECT_EQ(ncclSuccess, bootstrapClose(p->comm->bootstrap)); + p->comm->bootstrap = nullptr; + } + WaitRoots(); + EXPECT_EQ(2, OpenListeners()); +} + +TEST_F(BootstrapMicrotest, Split_EachSetupFailureIsReturned) { + struct Fault { + bool net, plugin; + const char* op; + int nth; + } faults[] = {{false, false, "listen", 1}, {false, false, "listen", 2}, {false, false, "listen", 3}, + {false, false, "connect", 1}, {false, false, "connect", 2}, {false, false, "accept", 1}, + {true, false, "listen", 2}, {true, true, "listen", 1}, {true, true, "connect", 1}}; + for (const auto& f : faults) { + SCOPED_TRACE(testing::Message() << f.net << f.plugin << " " << f.op << " #" << f.nth); + g_params["OOB_NET_ENABLE"] = f.net; + ASSERT_EQ(ncclSuccess, InitRanks(1)); + std::map calls; + auto fail = [&](const char* op) { return ++calls[op] == f.nth && std::strcmp(op, f.op) == 0; }; + if (f.plugin) + g_netFault = [&](const char* op) { return fail(op) ? ncclSystemError : ncclSuccess; }; + else + SetSockFault([&](const char* op, struct ncclSocket*) { return fail(op) ? ncclSystemError : ncclSuccess; }); + EXPECT_EQ(ncclSystemError, Split({{0}}, false)[0]); + SetSockFault(nullptr); + g_netFault = nullptr; + CloseAll(); // Split moved the parent into parents_, which the next iteration's Split overwrites + WaitRoots(); + EXPECT_EQ(0, OpenListeners()); + } +} + +TEST_F(BootstrapMicrotest, BcastGrowHandle_CoordinatorReachesOnlyTheBoundaryRanks) { + ncclBootstrapHandle h = {}; + EXPECT_EQ(ncclInvalidArgument, bcastGrowHandle(nullptr, self_.comm.get(), true)); + EXPECT_EQ(ncclInvalidArgument, bcastGrowHandle(&h, nullptr, false)); + EXPECT_TRUE(Warned("parent comm and handle must be provided")); + self_.comm->nRanks = 1; + EXPECT_EQ(ncclSuccess, bcastGrowHandle(&h, self_.comm.get(), false)); // would block if it received + + ASSERT_EQ(ncclSuccess, InitRanks(4)); + ncclBootstrapHandle got[4] = {}; + got[1].magic = 0xAB; + got[1].nRanks = 4; + EXPECT_TRUE(AllOk(OnRanks([&](int r) { + if (r == 2) return ncclSuccess; + return bcastGrowHandle(&got[r], Comm(r), /*isRoot=*/r == 1); + }))); + EXPECT_EQ(0xABu, got[0].magic); + EXPECT_EQ(0xABu, got[3].magic); + EXPECT_EQ(4, got[3].nRanks); + for (int r : {1, 2}) // neither the coordinator nor an inner rank was sent anything + EXPECT_EQ(0u, PendingOn(static_cast(State(r))->listen.peerSocket)) << r; + EXPECT_TRUE(AllOk(CloseRanks())); +} + +TEST_F(BootstrapMicrotest, Grow_ExistingAndNewRanksCloseOneRing) { + ASSERT_EQ(ncclSuccess, InitRanks(3)); + parents_ = std::move(ranks_); + ranks_.clear(); + ncclComm* coordinator = parents_[1]->comm.get(); + ncclBootstrapHandle grow; + ASSERT_EQ(ncclSuccess, bootstrapGetUniqueId(&grow, coordinator)); + ncclBootstrapHandle boundary[3] = {{}, grow, {}}; + std::swap(ranks_, parents_); // OnRanks runs over ranks_ + ASSERT_TRUE(AllOk(OnRanks([&](int r) { return bcastGrowHandle(&boundary[r], Comm(r), /*isRoot=*/r == 1); }))); + std::swap(ranks_, parents_); + for (auto& p : parents_) p->comm->childCount++; // ncclCommGrow bumps it before init + + MakeRanks(5); + auto res = OnRanks([&](int r) { + Comm(r)->isGrow = true; + if (r >= 3) return bootstrapInit(1, &grow, Comm(r), nullptr); + bool boundaryRank = r == 0 || r == 2; + return bootstrapInit(boundaryRank ? 1 : 0, boundaryRank ? &boundary[r] : nullptr, Comm(r), parents_[r]->comm.get()); + }); + ASSERT_TRUE(AllOk(res)); + EXPECT_EQ(Comm(0)->magic, Comm(4)->magic); + EXPECT_EQ(Comm(1)->magic, grow.magic); + ExpectAllGather(3); + EXPECT_TRUE(AllOk(CloseRanks())); +} + +// With OOB net + bidir net a grow's reverse ring never connects, so init hangs until abort. +// Pins AICOMRCCL-3004; a fix flips this pin. +TEST_F(BootstrapMicrotest, Grow_KnownBug_NetBidirReverseRingNeverConnectsSoInitHangs) { + g_params["OOB_NET_ENABLE"] = 1; + for (int bidir : {0, 1}) + for (int existing : {2, 3}) { + SCOPED_TRACE(testing::Message() << "bidir " << bidir << " existing " << existing); + g_params["BOOTSTRAP_BIDIR_NET"] = bidir; + for (auto& r : ranks_) parents_.push_back(std::move(r)); + ranks_.clear(); + ASSERT_EQ(ncclSuccess, InitRanks(existing)); + std::vector parent; + for (auto& r : ranks_) parent.push_back(r.get()); + for (auto& r : ranks_) parents_.push_back(std::move(r)); + ranks_.clear(); + ncclBootstrapHandle grow; + ASSERT_EQ(ncclSuccess, bootstrapGetUniqueId(&grow, parent[0]->comm.get())); + for (auto* p : parent) p->comm->childCount++; + MakeRanks(existing + 2); + watchdog_ = bidir ? std::chrono::milliseconds(300) : g_waitLimit; + auto res = OnRanks([&](int r) { + Comm(r)->isGrow = true; + if (r >= existing) return bootstrapInit(1, &grow, Comm(r), nullptr); + bool boundary = r == 0 || r == existing - 1; // only these receive the handle + return bootstrapInit(boundary ? 1 : 0, boundary ? &grow : nullptr, Comm(r), parent[r]->comm.get()); + }); + for (int r = 0; r < existing + 2; r++) EXPECT_EQ(bidir ? ncclInternalError : ncclSuccess, res[r]) << r; + WaitRoots(); + } +} + +// Plain sends complete without the receiver, so one thread can fix the order connections arrive in. +TEST_F(BootstrapMicrotest, SendRecv_OutOfOrderConnectionsAreParkedUntilAsked) { + ASSERT_EQ(ncclSuccess, InitRanks(3)); + char got[3] = {}; + for (const char* tag : {"1", "2", "3"}) ASSERT_EQ(ncclSuccess, bootstrapSend(State(0), 1, tag[0], (void*)tag, 2)); + EXPECT_EQ(ncclSuccess, bootstrapRecv(State(1), 0, '3', got, 2)); // parks 1 and 2 + EXPECT_STREQ("3", got); + EXPECT_EQ(ncclSuccess, bootstrapRecv(State(1), 0, '2', got, 2)); // from the middle of the parked list + EXPECT_STREQ("2", got); + EXPECT_EQ(ncclSuccess, bootstrapRecv(State(1), 0, '1', got, 2)); + EXPECT_STREQ("1", got); + + ASSERT_EQ(ncclSuccess, bootstrapSend(State(2), 1, 7, (void*)"2", 2)); + ASSERT_EQ(ncclSuccess, bootstrapSend(State(0), 1, 7, (void*)"0", 2)); + EXPECT_EQ(ncclSuccess, bootstrapRecv(State(1), 0, 7, got, 2)); // same tag, other peer first: parked + EXPECT_STREQ("0", got); + EXPECT_EQ(ncclSuccess, bootstrapRecv(State(1), 2, 7, got, 2)); + EXPECT_STREQ("2", got); + + ASSERT_EQ(ncclSuccess, bootstrapSend(State(0), 1, 8, (void*)"a", 2)); + ASSERT_EQ(ncclSuccess, bootstrapSend(State(2), 1, 9, (void*)"b", 2)); + EXPECT_EQ(ncclSuccess, bootstrapRecv(State(1), -1, 9, got, 2)); // any peer + EXPECT_STREQ("b", got); + EXPECT_EQ(ncclSuccess, bootstrapRecv(State(1), -1, 8, got, 2)); // any peer, from the parked list + EXPECT_STREQ("a", got); + EXPECT_TRUE(AllOk(CloseRanks())); +} + +TEST_F(BootstrapMicrotest, SendRecv_TransferFailuresAreReturned) { + ASSERT_EQ(ncclSuccess, InitRanks(1)); + ASSERT_EQ(ncclSuccess, bootstrapSend(State(0), 0, 2, (void*)"long", 4)); + char c[2]; + EXPECT_EQ(ncclInternalError, bootstrapRecv(State(0), 0, 2, c, 2)); + EXPECT_TRUE(Warned("Message truncated : received 4 bytes instead of 2")); + for (int nth : {1, 3}) { // the (rank, tag) ack, then the payload + int sends = 0; + SetSockFault([&](const char* op, struct ncclSocket*) { + return std::strcmp(op, "send") == 0 && ++sends == nth ? ncclSystemError : ncclSuccess; + }); + EXPECT_EQ(ncclSystemError, bootstrapSend(State(0), 0, 1, (void*)"x", 1)) << nth; + } +} + +// The failing close frees the parked list but leaves it linked, so a second close frees it again. +// Pins AICOMRCCL-3003; a fix flips this pin. +TEST_F(BootstrapMicrotest, Close_KnownBug_ParkedConnectionLeavesTheFreedListLinked) { + ASSERT_EQ(ncclSuccess, InitRanks(2)); + auto res = OnRanks([&](int r) -> ncclResult_t { + if (r == 0) { + NCCLCHECK(bootstrapSend(State(0), 1, 5, (void*)"x", 1)); + return bootstrapSend(State(0), 1, 6, (void*)"y", 1); + } + char c; + return bootstrapRecv(State(1), 0, 6, &c, 1); + }); + ASSERT_TRUE(AllOk(res)); + auto* state = static_cast(State(1)); + ASSERT_NE(nullptr, state->unexpectedConnections); + + EXPECT_EQ(ncclInternalError, bootstrapClose(state)); + + EXPECT_TRUE(Warned("Unexpected connections are not empty")); + EXPECT_NE(nullptr, state->unexpectedConnections); // already freed + state->unexpectedConnections = nullptr; +} + +TEST_F(BootstrapMicrotest, Close_WhileAbortingDropsParkedConnectionsQuietly) { + ASSERT_EQ(ncclSuccess, InitRanks(2)); + ASSERT_TRUE(AllOk(OnRanks([&](int r) -> ncclResult_t { + char c; + if (r == 0) { + NCCLCHECK(bootstrapSend(State(0), 1, 5, (void*)"x", 1)); + return bootstrapSend(State(0), 1, 6, (void*)"y", 1); + } + return bootstrapRecv(State(1), 0, 6, &c, 1); + }))); + ranks_[1]->abort = 1; + EXPECT_TRUE(AllOk(CloseRanks())); + EXPECT_FALSE(Warned("Unexpected connections")); +} + +TEST_F(BootstrapMicrotest, Barrier_EveryRankLeavesAfterAllArrive) { + ASSERT_EQ(ncclSuccess, InitRanks(5)); + // The last rank arrives late; nobody may leave before it does. + std::atomic lateArrived(false); + auto late = [&] { + std::this_thread::sleep_for(std::chrono::milliseconds(50)); + lateArrived = true; + }; + EXPECT_TRUE(AllOk(OnRanks([&](int r) { + if (r == 4) late(); + ncclResult_t res = bootstrapBarrier(State(r), r, 5, 0x77); + EXPECT_TRUE(lateArrived) << "rank " << r << " left early"; + return res; + }))); + // Intra-node barrier over ranks 1 and 3 only, in local-rank order. + int local[2] = {1, 3}; + lateArrived = false; + EXPECT_TRUE(AllOk(OnRanks([&](int r) { + if (r != 1 && r != 3) return ncclSuccess; + if (r == 3) late(); + ncclResult_t res = bootstrapIntraNodeBarrier(State(r), local, r == 1 ? 0 : 1, 2, 0x78); + EXPECT_TRUE(lateArrived) << "rank " << r << " left early"; + return res; + }))); + EXPECT_EQ(ncclSuccess, bootstrapBarrier(State(0), 0, 1, 0x79)); + EXPECT_TRUE(AllOk(CloseRanks())); +} + +TEST_F(BootstrapMicrotest, Broadcast_RootPayloadReachesEveryRank) { + ASSERT_EQ(ncclSuccess, InitRanks(4)); + char buf[4][4] = {}; + std::strcpy(buf[2], "abc"); + std::atomic connects(0); + SetSockFault([&](const char* op, struct ncclSocket*) { + if (std::strcmp(op, "connect") == 0) connects++; + return ncclSuccess; + }); + EXPECT_TRUE(AllOk(OnRanks([&](int r) { return bootstrapBroadcast(State(r), r, 4, 2, buf[r], 4); }))); + SetSockFault(nullptr); + for (int r = 0; r < 4; r++) EXPECT_STREQ("abc", buf[r]) << r; + EXPECT_EQ(3, connects.load()); // not to itself + // Intra-node broadcast over ranks {0, 2, 3}, rooted at local rank 1 (= rank 2). + int local[3] = {0, 2, 3}; + char nbuf[4][4] = {}; + std::strcpy(nbuf[2], "xyz"); + EXPECT_TRUE(AllOk(OnRanks([&](int r) { + if (r == 1) return ncclSuccess; + return bootstrapIntraNodeBroadcast(State(r), local, r == 0 ? 0 : r - 1, 3, 1, nbuf[r], 4); + }))); + EXPECT_STREQ("xyz", nbuf[0]); + EXPECT_STREQ("xyz", nbuf[3]); + EXPECT_STREQ("", nbuf[1]); + EXPECT_EQ(ncclSuccess, bootstrapBroadcast(State(0), 0, 1, 0, buf[0], 4)); + EXPECT_TRUE(AllOk(CloseRanks())); +} + +TEST_F(BootstrapMicrotest, IntraNodeAllGather_GathersOverALocalSubsetInBothRingModes) { + ASSERT_EQ(ncclSuccess, InitRanks(4)); + int local[3] = {3, 0, 2}; // local rank i is rank local[i] + for (int bidir = 0; bidir <= 1; bidir++) { + SCOPED_TRACE(bidir); + g_params["BOOTSTRAP_BIDIR_ALLGATHER"] = bidir; + char data[4][3] = {}; + for (int i = 0; i < 3; i++) data[local[i]][i] = char('a' + i); + EXPECT_TRUE(AllOk(OnRanks([&](int r) { + int i = std::find(local, local + 3, r) - local; + return i == 3 ? ncclSuccess : bootstrapIntraNodeAllGather(State(r), local, i, 3, data[r], 1); + }))); + for (int r : local) EXPECT_EQ("abc", std::string(data[r], 3)) << r; + } + char one = 'q'; + EXPECT_EQ(ncclSuccess, bootstrapIntraNodeAllGather(State(0), local, 0, 1, &one, 1)); + EXPECT_TRUE(AllOk(CloseRanks())); +} + +TEST_F(BootstrapMicrotest, IntraNodeAllGather_EncryptedConnectsOnAHelperThread) { + ASSERT_EQ(ncclSuccess, InitRanks(3)); + g_encrypted = true; + int local[3] = {0, 1, 2}; + char data[3][3] = {}; + for (int r = 0; r < 3; r++) data[r][r] = char('a' + r); + // Thread ids get reused, so mark the rank threads themselves. + static thread_local bool onRankThread = false; + std::atomic connects(0), connectsOnRankThreads(0); + SetSockFault([&](const char* op, struct ncclSocket*) { + if (std::strcmp(op, "connect") == 0) { + connects++; + if (onRankThread) connectsOnRankThreads++; + } + return ncclSuccess; + }); + EXPECT_TRUE(AllOk(OnRanks([&](int r) { + onRankThread = true; + return bootstrapIntraNodeAllGather(State(r), local, r, 3, data[r], 1); + }))); + SetSockFault(nullptr); + for (int r = 0; r < 3; r++) EXPECT_EQ("abc", std::string(data[r], 3)) << r; + EXPECT_EQ(3, connects.load()); + EXPECT_EQ(0, connectsOnRankThreads.load()); + EXPECT_TRUE(AllOk(CloseRanks())); +} + +TEST_F(BootstrapMicrotest, IntraNodeAllGather_ConnectFailureIsReturnedPlainOrEncrypted) { + ASSERT_EQ(ncclSuccess, InitRanks(2)); + int local[2] = {0, 1}; + char data[2][2]; + SetSockFault([](const char* op, struct ncclSocket*) { + return std::strcmp(op, "connect") == 0 ? ncclSystemError : ncclSuccess; + }); + for (bool encrypted : {false, true}) { + SCOPED_TRACE(encrypted); + g_encrypted = encrypted; + watchdog_ = std::chrono::milliseconds(50); // encrypted, the accept side waits for a connect that never comes + auto res = OnRanks([&](int r) { return bootstrapIntraNodeAllGather(State(r), local, r, 2, data[r], 1); }); + EXPECT_EQ(ncclSystemError, res[0]); + EXPECT_EQ(ncclSystemError, res[1]); + for (auto& rank : ranks_) rank->abort = 0; + } +} + +TEST_F(BootstrapMicrotest, Send_EncryptedSendsToOnePeerAndTagArriveInCallOrder) { + ASSERT_EQ(ncclSuccess, InitRanks(2)); + g_encrypted = true; + char got[3][2] = {}; + const char* sent[3] = {"1", "2", "3"}; + EXPECT_TRUE(AllOk(OnRanks([&](int r) -> ncclResult_t { + for (int i = 0; i < 3; i++) + NCCLCHECK(r == 0 ? bootstrapSend(State(0), 1, 9, (void*)sent[i], 2) : bootstrapRecv(State(1), 0, 9, got[i], 2)); + return r == 0 ? bootstrapSend(State(0), 1, 10, nullptr, 0) : bootstrapRecv(State(1), 0, 10, nullptr, 0); + }))); + for (int i = 0; i < 3; i++) EXPECT_STREQ(sent[i], got[i]); + EXPECT_TRUE(AllOk(CloseRanks())); +} + +TEST_F(BootstrapMicrotest, Send_EncryptedOnceCloseBegan_IsRejected) { + ASSERT_EQ(ncclSuccess, InitRanks(1)); + g_encrypted = true; + ASSERT_EQ(ncclSuccess, bootstrapAsyncSendDrain(static_cast(State(0)))); + EXPECT_EQ(ncclInternalError, bootstrapSend(State(0), 0, 1, (void*)"x", 1)); + EXPECT_TRUE(Warned("bootstrapSend: bootstrap state is closing")); +} + +class BootstrapAsyncSendMicrotest : public BootstrapMicrotest { + protected: + std::mutex gateMu_; + std::condition_variable gateCv_; + bool open_ = false; + std::atomic connects_{0}; + struct bootstrapState* state_ = nullptr; + + // Rank 0 queues two encrypted sends to (1, 4); the first one's connect is held, then fails. + void FailFirstOfTwoSends() { + ASSERT_EQ(ncclSuccess, InitRanks(2)); + g_encrypted = true; + state_ = static_cast(State(0)); + SetSockFault([this](const char* op, struct ncclSocket*) { + if (std::strcmp(op, "connect") != 0 || ++connects_ != 1) return ncclSuccess; + std::unique_lock lock(gateMu_); + EXPECT_TRUE(gateCv_.wait_for(lock, g_waitLimit, [this] { return open_; })); + return ncclRemoteError; + }); + ASSERT_EQ(ncclSuccess, bootstrapSend(state_, 1, 4, (void*)"a", 1)); + ASSERT_EQ(ncclSuccess, bootstrapSend(state_, 1, 4, (void*)"b", 1)); + { + std::lock_guard lock(gateMu_); + open_ = true; + } + gateCv_.notify_all(); + std::unique_lock lock(state_->asyncSendLock); + ASSERT_TRUE(state_->asyncSendCond.wait_for(lock, g_waitLimit, + [this] { return ncclIntruQueueEmpty(&state_->asyncSendQueue); })); + } +}; + +TEST_F(BootstrapAsyncSendMicrotest, FailureAbortsDropsLaterSendsAndSurfacesOnTheNextCalls) { + FailFirstOfTwoSends(); + + EXPECT_EQ(1, connects_.load()); // "b" was dropped, not sent after the failed "a" + EXPECT_EQ(1u, __atomic_load_n(&ranks_[0]->abort, __ATOMIC_ACQUIRE)); + char c; + EXPECT_EQ(ncclRemoteError, bootstrapRecv(state_, 1, 1, &c, 1)); + EXPECT_TRUE(Warned("bootstrapRecv: an earlier asynchronous bootstrap send failed")); + EXPECT_EQ(ncclRemoteError, bootstrapSend(state_, 1, 1, &c, 1)); + EXPECT_TRUE(Warned("bootstrapSend: an earlier asynchronous send to peer 1 failed")); + EXPECT_EQ(ncclSuccess, bootstrapClose(state_)); // already reported, and aborting + Comm(0)->bootstrap = nullptr; +} + +TEST_F(BootstrapAsyncSendMicrotest, FailureNotYetReportedIsReturnedByCloseOnce) { + FailFirstOfTwoSends(); + + EXPECT_EQ(ncclRemoteError, bootstrapClose(state_)); + EXPECT_TRUE(Warned("bootstrapClose: an asynchronous bootstrap send failed")); + EXPECT_EQ(ncclSuccess, bootstrapClose(state_)); + Comm(0)->bootstrap = nullptr; +} +} // namespace diff --git a/projects/rccl/test/host/fakes/bootstrap_micro_fakes.cc b/projects/rccl/test/host/fakes/bootstrap_micro_fakes.cc new file mode 100644 index 000000000000..cf86f5e710f9 --- /dev/null +++ b/projects/rccl/test/host/fakes/bootstrap_micro_fakes.cc @@ -0,0 +1,571 @@ +/************************************************************************* + * Copyright (c) 2026 Advanced Micro Devices, Inc. All rights reserved. + * + * See LICENSE.txt for license information + ************************************************************************/ + +// See bootstrap_micro_fakes.h. + +#include "bootstrap_micro_fakes.h" + +#include + +#include + +#include +#include +#include +#include +#include +#include + +#include "debug.h" +#include "env_fakes.h" +#include "proxy.h" + +using namespace bootstrap_micro; + +// nccl_fakes.cc's ncclDebugLog only prints, and it cannot be told apart per rank thread; these tests assert +// on the WARNs the ranks logged, so this binary records them. +uint32_t ncclDebugLevelMask = NCCL_DEBUG_LEVEL_MASK_UNINITIALIZED; // log everything +uint64_t ncclDebugMask = ~uint64_t(0); +thread_local int ncclDebugNoWarn = 0; + +namespace bootstrap_micro { +std::mutex g_logMu; +std::vector g_warns; + +bool Warned(const char* needle) { + std::lock_guard lock(g_logMu); + for (const auto& w : g_warns) + if (w.find(needle) != std::string::npos) return true; + return false; +} +} // namespace bootstrap_micro + +void ncclDebugLog(ncclDebugLogLevel level, unsigned long, const char*, int, const char* fmt, ...) { + if (level != NCCL_LOG_WARN) return; + char buf[512]; + va_list ap; + va_start(ap, fmt); + std::vsnprintf(buf, sizeof(buf), fmt, ap); + va_end(ap); + std::lock_guard lock(g_logMu); + g_warns.push_back(buf); +} + +std::map g_params; +std::function g_loadParam = [](const char* env, int64_t deftVal) { + auto it = g_params.find(env); + return it == g_params.end() ? deftVal : it->second; +}; +int64_t ncclParamRasEnable() { return g_loadParam("RAS_ENABLE", 1); } + +// src/misc/signals.cc, which has no shared owner: the real one would install process-wide handlers. +void RegisterSignalHandlers() {} + +bool g_encrypted = false; +ncclResult_t ncclGetCryptConnectionMode(bool* encrypted) { + *encrypted = g_encrypted; + return ncclSuccess; +} + +namespace bootstrap_micro { +std::mutex g_seamMu; +std::map g_proxyInits; +ncclResult_t g_proxyInitResult = ncclSuccess; +int g_rasCommInits = 0; +ncclResult_t g_rasCommInitResult = ncclSuccess; +std::vector g_rasAdded; +ncclResult_t g_rasAddRanksResult = ncclSuccess; +int g_ifCount = 1; +union ncclSocketAddress g_subnetRemote; +} // namespace bootstrap_micro + +// On success the proxy owns the listen socket, as in production. +ncclResult_t ncclProxyInit(struct ncclComm* comm, struct ncclSocket* sock, union ncclSocketAddress* peerAddresses, + uint64_t* peerUDS) { + { + std::lock_guard lock(g_seamMu); + g_proxyInits[comm] = {std::vector(peerAddresses, peerAddresses + comm->nRanks), + std::vector(peerUDS, peerUDS + comm->nRanks), sock->addr}; + if (g_proxyInitResult != ncclSuccess) return g_proxyInitResult; + } + (void)ncclSocketClose(sock); + free(sock); + return ncclSuccess; +} +ncclResult_t ncclRasCommInit(struct ncclComm*, struct rasRankInit*) { + std::lock_guard lock(g_seamMu); + g_rasCommInits++; + return g_rasCommInitResult; +} +// On success the RAS thread owns (and frees) the array. +ncclResult_t ncclRasAddRanks(struct rasRankInit* ranks, int nranks) { + std::lock_guard lock(g_seamMu); + if (g_rasAddRanksResult != ncclSuccess) return g_rasAddRanksResult; + g_rasAdded.assign(ranks, ranks + nranks); + free(ranks); + return ncclSuccess; +} + +namespace bootstrap_micro { + + +std::mutex g_sockMu; +std::condition_variable g_sockCv; +std::deque g_endpoints; +std::deque g_listeners; +std::map g_ports; +std::set g_openFds; +uint16_t g_nextPort = 1000; +bool g_sockShutdown = false; +int g_liveRoots = 0; +std::chrono::milliseconds g_waitLimit(5000); +ncclSocketState g_rejectedState = ncclSocketStateAccepting; +int g_readyPolls = 0; +std::atomic g_multiOps{0}; +std::shared_ptr g_sockFault; // swapped atomically: a root thread may be calling it + +void SetSockFault(SockFaultFn fn) { + std::atomic_store(&g_sockFault, fn ? std::make_shared(std::move(fn)) : nullptr); +} +ncclResult_t SockFault(const char* op, struct ncclSocket* sock) { + auto fault = std::atomic_load(&g_sockFault); + return fault ? (*fault)(op, sock) : ncclSuccess; +} +uint16_t PortOf(const union ncclSocketAddress& addr) { return ntohs(addr.sin.sin_port); } +FakeEndpoint& Endpoint(const struct ncclSocket* sock) { return g_endpoints[sock->socketDescriptor - kEndpointFd]; } + +ncclResult_t WaitFor(std::unique_lock& lock, volatile uint32_t* abortFlag, + const std::function& ready) { + auto deadline = Clock::now() + g_waitLimit; + while (!ready()) { + if (abortFlag && __atomic_load_n(abortFlag, __ATOMIC_ACQUIRE)) return ncclInternalError; + if (g_sockShutdown) return ncclSystemError; + if (Clock::now() > deadline) { + ADD_FAILURE() << "fake socket wait timed out"; + return ncclSystemError; + } + g_sockCv.wait_for(lock, std::chrono::milliseconds(1)); + } + return ncclSuccess; +} + +// Takes the next pending connection, rejecting a wrong-magic peer the way socketFinalizeAccept does. +bool PopPending(struct ncclSocket* sock, FakeListener& l, bool retry) { + while (!l.pending.empty()) { + int e = l.pending.front(); + l.pending.pop_front(); + if (g_endpoints[e].magic != l.magic) { + g_endpoints[e].open = false; + g_sockCv.notify_all(); + if (retry) continue; + sock->state = g_rejectedState; + return true; + } + sock->socketDescriptor = kEndpointFd + e; + sock->state = ncclSocketStateReady; + g_openFds.insert(sock->socketDescriptor); + return true; + } + return false; +} +} // namespace bootstrap_micro + +uint64_t ncclSocketDefaultMagic(void) { return 0x5eed; } + +ncclResult_t ncclSocketInit(struct ncclSocket* sock, const union ncclSocketAddress* addr, uint64_t magic, + enum ncclSocketType type, volatile uint32_t* abortFlag, int asyncFlag, int customRetry) { + if (sock == nullptr) return ncclSuccess; + std::memset(sock, 0, sizeof(*sock)); + if (addr) sock->addr = *addr; + sock->magic = magic; + sock->type = type; + sock->abortFlag = abortFlag; + sock->asyncFlag = asyncFlag; + sock->customRetry = customRetry; + sock->state = ncclSocketStateInitialized; + sock->socketDescriptor = sock->acceptSocketDescriptor = NCCL_INVALID_SOCKET; + return SockFault("init", sock); +} + +void ncclSocketMove(struct ncclSocket* dst, struct ncclSocket* src) { + *dst = *src; + src->socketDescriptor = src->acceptSocketDescriptor = NCCL_INVALID_SOCKET; + src->state = ncclSocketStateNone; +} + +ncclResult_t ncclSocketListen(struct ncclSocket* sock) { + if (ncclResult_t res = SockFault("listen", sock)) return res; + std::lock_guard lock(g_sockMu); + uint16_t port = PortOf(sock->addr); + if (port == 0) port = g_nextPort++; + if (g_ports.count(port)) return ncclSystemError; + sock->addr.sin.sin_family = AF_INET; + sock->addr.sin.sin_port = htons(port); + g_listeners.emplace_back(); + FakeListener& l = g_listeners.back(); + l.port = port; + l.magic = sock->magic; + l.root = sock->type == ncclSocketTypeBootstrap && sock->abortFlag == nullptr; // only bootstrapCreateRoot's + g_ports[port] = g_listeners.size() - 1; + sock->socketDescriptor = kListenFd + g_listeners.size() - 1; + sock->state = ncclSocketStateReady; + g_openFds.insert(sock->socketDescriptor); + return ncclSuccess; +} + +ncclResult_t ncclSocketGetAddr(struct ncclSocket* sock, union ncclSocketAddress* addr) { + if (ncclResult_t res = SockFault("getaddr", sock)) return res; + *addr = sock->addr; + return ncclSuccess; +} + +ncclResult_t ncclSocketConnect(struct ncclSocket* sock) { + if (ncclResult_t res = SockFault("connect", sock)) return res; + std::lock_guard lock(g_sockMu); + auto it = g_ports.find(PortOf(sock->addr)); + if (it == g_ports.end() || !g_listeners[it->second].open) return ncclRemoteError; + int a = g_endpoints.size(); + g_endpoints.emplace_back(); + g_endpoints.emplace_back(); + g_endpoints[a].peer = a + 1; + g_endpoints[a + 1].peer = a; + g_endpoints[a + 1].magic = sock->magic; + g_listeners[it->second].pending.push_back(a + 1); + sock->socketDescriptor = kEndpointFd + a; + sock->state = sock->asyncFlag ? ncclSocketStateConnecting : ncclSocketStateReady; + g_openFds.insert(sock->socketDescriptor); + g_sockCv.notify_all(); + return ncclSuccess; +} + +ncclResult_t ncclSocketAccept(struct ncclSocket* sock, struct ncclSocket* listenSock, bool retry) { + if (ncclResult_t res = SockFault("accept", sock)) return res; + if (listenSock->state != ncclSocketStateReady) return ncclInternalError; + if (sock->acceptSocketDescriptor == NCCL_INVALID_SOCKET) { // clones the listener, as socket.cc does + *sock = *listenSock; + sock->acceptSocketDescriptor = listenSock->socketDescriptor; + sock->socketDescriptor = NCCL_INVALID_SOCKET; + sock->state = ncclSocketStateAccepting; + } + std::unique_lock lock(g_sockMu); + FakeListener& l = g_listeners[listenSock->socketDescriptor - kListenFd]; + if (sock->asyncFlag) { + PopPending(sock, l, retry); + return ncclSuccess; + } + ncclResult_t res = WaitFor(lock, sock->abortFlag, [&] { return !l.pending.empty() || !l.open; }); + if (res != ncclSuccess) return res; + if (!PopPending(sock, l, retry)) return ncclSystemError; + return ncclSuccess; +} + +ncclResult_t ncclSocketReady(struct ncclSocket* sock, int* running) { + if (ncclResult_t res = SockFault("ready", sock)) return res; + std::lock_guard lock(g_sockMu); + *running = 0; + if (g_readyPolls > 0) { + g_readyPolls--; + return ncclSuccess; + } + if (sock->state == ncclSocketStateConnecting) sock->state = ncclSocketStateReady; + if (sock->state == ncclSocketStateAccepting) + PopPending(sock, g_listeners[sock->acceptSocketDescriptor - kListenFd], /*retry=*/true); + *running = sock->state == ncclSocketStateReady; + return ncclSuccess; +} + +ncclResult_t ncclSocketSend(struct ncclSocket* sock, void* ptr, int size) { + if (ncclResult_t res = SockFault("send", sock)) return res; + std::lock_guard lock(g_sockMu); + if (sock->state != ncclSocketStateReady) return ncclInternalError; + FakeEndpoint& peer = g_endpoints[Endpoint(sock).peer]; + if (!peer.open) return ncclRemoteError; + peer.in.insert(peer.in.end(), (char*)ptr, (char*)ptr + size); + g_sockCv.notify_all(); + return ncclSuccess; +} + +ncclResult_t ncclSocketRecv(struct ncclSocket* sock, void* ptr, int size) { + if (ncclResult_t res = SockFault("recv", sock)) return res; + std::unique_lock lock(g_sockMu); + if (sock->state != ncclSocketStateReady) return ncclInternalError; + FakeEndpoint& me = Endpoint(sock); + ncclResult_t res = WaitFor(lock, sock->abortFlag, + [&] { return (int)me.in.size() >= size || !g_endpoints[me.peer].open; }); + if (res != ncclSuccess) return res; + if ((int)me.in.size() < size) return ncclRemoteError; // peer closed short + std::copy(me.in.begin(), me.in.begin() + size, (char*)ptr); + me.in.erase(me.in.begin(), me.in.begin() + size); + return ncclSuccess; +} + +ncclResult_t ncclSocketSendRecv(struct ncclSocket* sendSock, void* sendPtr, int sendSize, struct ncclSocket* recvSock, + void* recvPtr, int recvSize) { + if (ncclResult_t res = ncclSocketSend(sendSock, sendPtr, sendSize)) return res; + return ncclSocketRecv(recvSock, recvPtr, recvSize); +} + +// Sends never block here, so issuing them all before any receive cannot deadlock. +ncclResult_t ncclSocketMultiOp(struct ncclSocketOp* ops, int numOps) { + g_multiOps++; + for (int pass = NCCL_SOCKET_SEND; pass <= NCCL_SOCKET_RECV; pass++) { + for (int i = 0; i < numOps; i++) { + if (ops[i].op != pass || ops[i].size == 0) continue; + ncclResult_t res = pass == NCCL_SOCKET_SEND ? ncclSocketSend(ops[i].sock, ops[i].ptr, ops[i].size) + : ncclSocketRecv(ops[i].sock, ops[i].ptr, ops[i].size); + if (res != ncclSuccess) return res; + } + } + return ncclSuccess; +} + +ncclResult_t ncclSocketClose(struct ncclSocket* sock, bool wait) { + if (sock == nullptr) return ncclSuccess; + if (ncclResult_t res = SockFault("close", sock)) return res; + std::unique_lock lock(g_sockMu); + int fd = sock->socketDescriptor; + if (sock->state > ncclSocketStateNone && fd >= kListenFd) { + FakeListener& l = g_listeners[fd - kListenFd]; + l.open = false; + g_ports.erase(l.port); + for (int e : l.pending) g_endpoints[e].open = false; + if (l.root) g_liveRoots--; + } else if (sock->state > ncclSocketStateNone && fd >= kEndpointFd) { + FakeEndpoint& me = Endpoint(sock); + if (wait) (void)WaitFor(lock, sock->abortFlag, [&] { return !g_endpoints[me.peer].open; }); + me.open = false; + } + g_openFds.erase(fd); + sock->socketDescriptor = NCCL_INVALID_SOCKET; + sock->state = ncclSocketStateClosed; + g_sockCv.notify_all(); + return ncclSuccess; +} + +const char* ncclSocketToString(const union ncclSocketAddress* addr, char* buf, size_t bufLen, const int) { + std::snprintf(buf, bufLen, "fake:%u", PortOf(*addr)); + return buf; +} + +ncclResult_t ncclSocketGetAddrFromString(union ncclSocketAddress* ua, const char* ipPortPair) { + const char* colon = std::strrchr(ipPortPair, ':'); + if (colon == nullptr) return ncclInvalidArgument; + std::memset(ua, 0, sizeof(*ua)); + ua->sin.sin_family = AF_INET; + ua->sin.sin_port = htons(std::atoi(colon + 1)); + return ncclSuccess; +} + +ncclResult_t ncclFindInterfaces(char* ifNames, union ncclSocketAddress* ifAddrs, int ifNameMaxSize, int, int* nIfs) { + *nIfs = g_ifCount; + if (g_ifCount > 0) { + std::snprintf(ifNames, ifNameMaxSize, "fake0"); + std::memset(ifAddrs, 0, sizeof(*ifAddrs)); + ifAddrs->sin.sin_family = AF_INET; + } + return ncclSuccess; +} +ncclResult_t ncclFindInterfaceMatchSubnet(char* ifName, union ncclSocketAddress* localAddr, + union ncclSocketAddress* remoteAddr, int ifNameMaxSize, int* found) { + g_subnetRemote = *remoteAddr; + return ncclFindInterfaces(ifName, localAddr, ifNameMaxSize, 1, found); +} + +// The root's thread is the only one named, and its listener the only one closed by it. +void ncclSetThreadName(std::thread&, const char* fmt, ...) { + std::lock_guard lock(g_sockMu); + if (std::strcmp(fmt, "NCCL BootstrapR") == 0) g_liveRoots++; +} + +namespace bootstrap_micro { +struct FakeNetReq { + bool send; + FakeNetConn* conn; + void* data; + size_t size; + int tag; +}; +std::mutex g_netMu; +std::deque g_netListens; +std::vector> g_netConns; +std::vector g_netDevs = {"mlx5_0"}; +std::atomic g_netDeferConnect{0}, g_netDeferIsend{0}, g_netDeferIrecv{0}, g_netDeferTest{0}; +std::atomic g_netRegs{0}, g_netDeregs{0}, g_netCloses{0}; +std::function g_netFault; +std::set g_netListenDevs; + +ncclResult_t NetFault(const char* op) { return g_netFault ? g_netFault(op) : ncclSuccess; } +bool Defer(std::atomic& n) { + int v = n.load(); + while (v > 0) + if (n.compare_exchange_weak(v, v - 1)) return true; + return false; +} + +ncclResult_t FakeNetDevices(int* ndev) { + *ndev = g_netDevs.size(); + return NetFault("devices"); +} +ncclResult_t FakeNetGetProperties(int dev, ncclNetProperties_t* props) { + std::memset(props, 0, sizeof(*props)); + props->name = const_cast(g_netDevs[dev].c_str()); + props->port = 1; + return NetFault("getProperties"); +} +ncclResult_t FakeNetListen(void*, int dev, void* handle, void** listenComm) { + if (ncclResult_t res = NetFault("listen")) return res; + std::lock_guard lock(g_netMu); + g_netListens.emplace_back(); + g_netListens.back().dev = dev; + g_netListenDevs.insert(dev); + std::memset(handle, 0, NCCL_NET_HANDLE_MAXSIZE); + uint64_t id = g_netListens.size(); + std::memcpy(handle, &id, sizeof(id)); + *listenComm = &g_netListens.back(); + return ncclSuccess; +} +ncclResult_t FakeNetConnect(void*, int, void* handle, void** sendComm, ncclNetDeviceHandle_t**) { + if (ncclResult_t res = NetFault("connect")) return res; + *sendComm = nullptr; + if (Defer(g_netDeferConnect)) return ncclSuccess; + uint64_t id; + std::memcpy(&id, handle, sizeof(id)); + std::lock_guard lock(g_netMu); + if (id == 0 || id > g_netListens.size()) return ncclSystemError; + g_netConns.push_back(std::make_unique()); + g_netListens[id - 1].pending.push_back(g_netConns.back().get()); + *sendComm = g_netConns.back().get(); + return ncclSuccess; +} +ncclResult_t FakeNetAccept(void* listenComm, void** recvComm, ncclNetDeviceHandle_t**) { + if (ncclResult_t res = NetFault("accept")) return res; + std::lock_guard lock(g_netMu); + auto* l = static_cast(listenComm); + *recvComm = nullptr; + if (!l->pending.empty()) { + *recvComm = l->pending.front(); + l->pending.pop_front(); + } + return ncclSuccess; +} +ncclResult_t FakeNetRegMr(void*, void*, size_t, int, void** mhandle) { + if (ncclResult_t res = NetFault("regMr")) return res; + g_netRegs++; + *mhandle = &g_netRegs; + return ncclSuccess; +} +ncclResult_t FakeNetDeregMr(void*, void*) { + g_netDeregs++; + return NetFault("deregMr"); +} +ncclResult_t FakeNetIsend(void* sendComm, void* data, size_t size, int tag, void*, void*, void** request) { + if (ncclResult_t res = NetFault("isend")) return res; + *request = nullptr; + if (Defer(g_netDeferIsend)) return ncclSuccess; + std::lock_guard lock(g_netMu); + static_cast(sendComm)->msgs.emplace_back(tag, std::string((char*)data, size)); + *request = new FakeNetReq{true, nullptr, nullptr, 0, tag}; + return ncclSuccess; +} +ncclResult_t FakeNetIrecv(void* recvComm, int, void** data, size_t* sizes, int* tags, void**, void**, void** request) { + if (ncclResult_t res = NetFault("irecv")) return res; + *request = nullptr; + if (Defer(g_netDeferIrecv)) return ncclSuccess; + *request = new FakeNetReq{false, static_cast(recvComm), data[0], sizes[0], tags[0]}; + return ncclSuccess; +} +ncclResult_t FakeNetTest(void* request, int* done, int*) { + if (ncclResult_t res = NetFault("test")) return res; + auto* req = static_cast(request); + *done = 0; + if (Defer(g_netDeferTest)) return ncclSuccess; + std::lock_guard lock(g_netMu); + if (!req->send) { + auto& msgs = req->conn->msgs; + auto it = std::find_if(msgs.begin(), msgs.end(), [&](const auto& m) { return m.first == req->tag; }); + if (it == msgs.end()) return ncclSuccess; + std::memcpy(req->data, it->second.data(), std::min(req->size, it->second.size())); + msgs.erase(it); + } + *done = 1; + delete req; + return ncclSuccess; +} +ncclResult_t FakeNetClose(void*) { + g_netCloses++; + return NetFault("close"); +} + +ncclNet_t MakeFakeNet() { + ncclNet_t net{}; + net.name = "fake"; + net.devices = FakeNetDevices; + net.getProperties = FakeNetGetProperties; + net.listen = FakeNetListen; + net.connect = FakeNetConnect; + net.accept = FakeNetAccept; + net.regMr = FakeNetRegMr; + net.deregMr = FakeNetDeregMr; + net.isend = FakeNetIsend; + net.irecv = FakeNetIrecv; + net.test = FakeNetTest; + net.closeSend = FakeNetClose; + net.closeRecv = FakeNetClose; + net.closeListen = FakeNetClose; + return net; +} +ncclNet_t g_fakeNet = MakeFakeNet(); + +void ResetSeams() { + { + std::lock_guard lock(g_logMu); + g_warns.clear(); + } + g_params.clear(); + g_encrypted = false; + g_ifCount = 1; + g_proxyInits.clear(); + g_proxyInitResult = ncclSuccess; + g_rasCommInits = 0; + g_rasCommInitResult = ncclSuccess; + g_rasAdded.clear(); + g_rasAddRanksResult = ncclSuccess; + { + std::lock_guard lock(g_sockMu); + g_endpoints.clear(); + g_listeners.clear(); + g_ports.clear(); + g_openFds.clear(); + g_sockShutdown = false; + g_liveRoots = 0; + } + g_waitLimit = std::chrono::milliseconds(5000); + g_rejectedState = ncclSocketStateAccepting; + g_readyPolls = 0; + g_multiOps = 0; + SetSockFault(nullptr); + { + std::lock_guard lock(g_netMu); + g_netListens.clear(); + g_netConns.clear(); + g_netListenDevs.clear(); + } + g_netDevs = {"mlx5_0"}; + g_netDeferConnect = g_netDeferIsend = g_netDeferIrecv = g_netDeferTest = 0; + g_netRegs = g_netDeregs = g_netCloses = 0; + g_netFault = nullptr; + ResetEnvFakes(); +} + +void ShutdownSockets() { + std::unique_lock lock(g_sockMu); + g_sockShutdown = true; + g_sockCv.notify_all(); + auto deadline = Clock::now() + g_waitLimit; + while (g_liveRoots > 0 && Clock::now() < deadline) g_sockCv.wait_for(lock, std::chrono::milliseconds(1)); + EXPECT_EQ(0, g_liveRoots) << "a bootstrap root thread did not finish"; +} +} // namespace bootstrap_micro diff --git a/projects/rccl/test/host/fakes/bootstrap_micro_fakes.h b/projects/rccl/test/host/fakes/bootstrap_micro_fakes.h new file mode 100644 index 000000000000..09c8160d6a5c --- /dev/null +++ b/projects/rccl/test/host/fakes/bootstrap_micro_fakes.h @@ -0,0 +1,114 @@ +/************************************************************************* + * Copyright (c) 2026 Advanced Micro Devices, Inc. All rights reserved. + * + * See LICENSE.txt for license information + ************************************************************************/ + +// Externals src/bootstrap.cc reaches, for rccl-UnitTestsMicroBootstrap, which compiles it. Distinct from +// bootstrap_stubs.cc, which fakes bootstrap.cc itself. Ranks run as threads over the in-memory socket layer +// and net plugin here; misc/utils.cc is the real one, and ncclGetEnv and linux.cc come from env_fakes.cc and +// os_fakes.cc. + +#ifndef RCCL_TEST_HOST_BOOTSTRAP_MICRO_FAKES_H_ +#define RCCL_TEST_HOST_BOOTSTRAP_MICRO_FAKES_H_ + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "comm.h" +#include "net.h" +#include "ras.h" +#include "socket.h" + +// Params are read on every call (via param_redirect.h) so a test can flip one; unset names keep their default. +extern std::map g_params; +extern std::function g_loadParam; +extern bool g_encrypted; // what ncclGetCryptConnectionMode reports + +namespace bootstrap_micro { + +extern std::mutex g_logMu; +extern std::vector g_warns; +bool Warned(const char* needle); + +struct ProxyInitCall { + std::vector addrs; + std::vector uds; + ncclSocketAddress own; // the proxy listen socket's +}; +extern std::map g_proxyInits; +extern ncclResult_t g_proxyInitResult; +extern int g_rasCommInits; +extern ncclResult_t g_rasCommInitResult; +extern std::vector g_rasAdded; +extern ncclResult_t g_rasAddRanksResult; +extern int g_ifCount; +extern union ncclSocketAddress g_subnetRemote; + +// In-memory socket layer: a listener is a port with a queue of pending connections, a connection is +// two endpoints whose inbound byte queues the other side appends to. Blocking calls honour the +// socket's abort flag and fail the test instead of hanging. +using Clock = std::chrono::steady_clock; +constexpr int kEndpointFd = 10000; +constexpr int kListenFd = 20000; +struct FakeEndpoint { + std::deque in; + int peer = -1; + bool open = true; + uint64_t magic = 0; // the connector's, checked by accept +}; +struct FakeListener { + uint16_t port = 0; + uint64_t magic = 0; + bool open = true; + bool root = false; + std::deque pending; +}; +extern std::mutex g_sockMu; +extern std::condition_variable g_sockCv; +extern std::deque g_listeners; +extern std::set g_openFds; +extern bool g_sockShutdown; +extern int g_liveRoots; +extern std::chrono::milliseconds g_waitLimit; +// Mirrors socket.cc: ncclSocketAccept(retry=false) hands back a rejected peer as Accepting, not BadHandshake. +extern ncclSocketState g_rejectedState; +extern int g_readyPolls; // ncclSocketReady says "not yet" this many times first +extern std::atomic g_multiOps; // only the bidirectional socket ring uses ncclSocketMultiOp +using SockFaultFn = std::function; +void SetSockFault(SockFaultFn fn); +uint16_t PortOf(const union ncclSocketAddress& addr); + +// In-memory net plugin. listen() writes a listener id into the handle; connect() queues a connection +// on it that accept() picks up. isend copies into the connection at once; irecv matches by tag. +struct FakeNetConn { + std::deque> msgs; +}; +struct FakeNetListener { + int dev = 0; + std::deque pending; +}; +extern std::deque g_netListens; +extern std::vector g_netDevs; +extern std::atomic g_netDeferConnect, g_netDeferIsend, g_netDeferIrecv, g_netDeferTest; +extern std::atomic g_netRegs, g_netDeregs, g_netCloses; +extern std::function g_netFault; +extern ncclNet_t g_fakeNet; + +void ResetSeams(); +// Lets every detached root thread unwind before the next test reuses the fake network. +void ShutdownSockets(); + +} // namespace bootstrap_micro + +#endif // RCCL_TEST_HOST_BOOTSTRAP_MICRO_FAKES_H_ diff --git a/projects/rccl/test/host/fakes/os_fakes.cc b/projects/rccl/test/host/fakes/os_fakes.cc index 206f35fb8b50..eb16af38a432 100644 --- a/projects/rccl/test/host/fakes/os_fakes.cc +++ b/projects/rccl/test/host/fakes/os_fakes.cc @@ -26,6 +26,8 @@ void ncclOsAlignedFree(void* ptr) { free(ptr); } ncclResult_t ncclOsInitialize() { return ncclSuccess; } uint64_t ncclOsGetPid() { return 4321; } +ncclResult_t ncclOsSetFilesLimit() { return ncclSuccess; } +char* ncclOsStrSep(char** stringp, const char* delim) { return strsep(stringp, delim); } size_t ncclOsGetPageSize() { return 4096; } size_t ncclOsGetCommMempoolMaxSize() { return 0; } // src/os/linux.cc: 0 = no cap diff --git a/projects/rccl/test/host/run_host_tests.sh b/projects/rccl/test/host/run_host_tests.sh index a4d739927ecc..87e80445e215 100755 --- a/projects/rccl/test/host/run_host_tests.sh +++ b/projects/rccl/test/host/run_host_tests.sh @@ -150,6 +150,7 @@ do_host_tests() { # rccl-HostUnitTests keeps host_tests.xml for backward compatibility. local -a binaries=( "rccl-HostUnitTests:$XML_FILE" + "rccl-UnitTestsMicroBootstrap:$SCRIPT_DIR/host_tests_micro_bootstrap.xml" "rccl-UnitTestsMicro:$SCRIPT_DIR/host_tests_micro.xml" "rccl-UnitTestsMicroWarpSpeed:$SCRIPT_DIR/host_tests_micro_warpspeed.xml" "rccl-UnitTestsMicroCrypt:$SCRIPT_DIR/host_tests_micro_crypt.xml" diff --git a/projects/rccl/test/test_categories_micro_bootstrap.yaml b/projects/rccl/test/test_categories_micro_bootstrap.yaml new file mode 100644 index 000000000000..a7dd344b8c1b --- /dev/null +++ b/projects/rccl/test/test_categories_micro_bootstrap.yaml @@ -0,0 +1,46 @@ +# rccl-UnitTestsMicroBootstrap — shared/ctest schema. +# Host-only micro-tests for src/bootstrap.cc (no GPU, no librccl). +# RCCL does not support Windows: exclude_windows matches all tests so generated suites are empty on Windows. + +test_categories: + quick: µ_bootstrap_quick + description: "Smoke — host-only bootstrap micro-tests, about two seconds." + test_patterns: + - "Bootstrap*" + exclude_windows: + - "*" + labels: + - "rccl" + - "quick" + - "smoke" + - "precheck" + - "microtest" + standard: + <<: *micro_bootstrap_quick + labels: + - "rccl" + - "standard" + - "regression" + - "microtest" + comprehensive: + <<: *micro_bootstrap_quick + labels: + - "rccl" + - "comprehensive" + - "comprehensive_regression" + - "microtest" + full: + <<: *micro_bootstrap_quick + labels: + - "rccl" + - "full" + - "full_regression" + - "microtest" + +execution_settings: + timeout_multiplier: 1 + category_timeouts: + quick: 600 + standard: 600 + comprehensive: 600 + full: 600 diff --git a/projects/rccl/tools/scripts/test_runner/configs/ci-precheckin.json b/projects/rccl/tools/scripts/test_runner/configs/ci-precheckin.json index a256d289843e..ef9cc70b7044 100644 --- a/projects/rccl/tools/scripts/test_runner/configs/ci-precheckin.json +++ b/projects/rccl/tools/scripts/test_runner/configs/ci-precheckin.json @@ -24,6 +24,17 @@ "generator": "Unix Makefiles" }, "test_configurations": { + "unit_tests_micro_bootstrap": { + "is_gtest": true, + "binary": "host/rccl-UnitTestsMicroBootstrap", + "num_ranks": 1, + "num_nodes": 1, + "num_gpus": 0, + "timeout": 60, + "tests": [ + {"name": "BootstrapMicrotests", "description": "Host-only bootstrap microtests (no GPU, no HIP/RCCL runtime)", "test_filter": "*"} + ] + }, "unit_tests": { "is_gtest": true, "binary": "rccl-UnitTests", @@ -487,6 +498,12 @@ } }, "test_suites": [ + { + "name": "RCCL Microtests - bootstrap", + "description": "rccl-UnitTestsMicroBootstrap (host-only bootstrap microtests, no GPU/HIP/RCCL runtime)", + "config": "unit_tests_micro_bootstrap", + "enabled": true + }, { "name": "RCCL Unit Tests - All Collectives", "description": "Standard RCCL unit tests for all collective operations", diff --git a/projects/rccl/tools/scripts/test_runner/lib/test_executor.py b/projects/rccl/tools/scripts/test_runner/lib/test_executor.py index e3a4ab071357..0e0777a1ec25 100644 --- a/projects/rccl/tools/scripts/test_runner/lib/test_executor.py +++ b/projects/rccl/tools/scripts/test_runner/lib/test_executor.py @@ -3327,7 +3327,8 @@ def generate_coverage_report(self): "rccl-UnitTestsMicroGinHost", "rccl-UnitTestsMicroRomeModels", "rccl-UnitTestsMicroDda", - "rccl-UnitTestsMicroTopo"]: + "rccl-UnitTestsMicroTopo", + "rccl-UnitTestsMicroBootstrap"]: binary_path = os.path.join(host_test_dir, binary) if os.path.isfile(binary_path): object_files.extend(["--object", binary_path])