test(rccl): add host microtests for the OS and socket layer

projects/rccl/test/host/CMakeLists.txt

@@ -55,6 +55,10 @@ rccl_assert_source_line("${CMAKE_CURRENT_LIST_DIR}/fakes/tuning_fakes.cc"
   "int64_t g_paramMaxNchannels = -2;" "the MaxNchannels fake default")
 rccl_assert_source_line("${CMAKE_CURRENT_LIST_DIR}/fakes/wrap_fakes.cc"
   "static int64_t DefaultParamForceCe() { return 1; }" "the ForceCe fake default")
+rccl_assert_source_line("${_rccl_src_root}/init.cc"
+  "NCCL_PARAM(SetCpuStackSize, \"SET_CPU_STACK_SIZE\", 1)" "linux-test.cc: kSetCpuStackSizeDefault")
+rccl_assert_source_line("${CMAKE_CURRENT_LIST_DIR}/linux-test.cc"
+  "constexpr int kSetCpuStackSizeDefault = 1;" "the SetCpuStackSize fake default")
 # gdr-probe-test.cc pins the GDR probe's registration flags to production's; these
 # keep its kRegAccessFlags and both reg.cc copies spelling the same set.
 set(_rccl_reg_access_flags
@@ -311,6 +315,20 @@ function(rccl_define_micro_source_lists)
     ../common/main_altrsmi.cpp
     PARENT_SCOPE)
 
+  # rccl-UnitTestsMicroLinux -- src/os/linux*.cc, misc/{socket,shmutils,signals}.cc. Own binary:
+  # fakes/os_fakes.cc, which most targets link, defines ncclOs* symbols that os/linux.cc defines.
+  set(TEST_MICRO_LINUX_SOURCE_FILES
+    linux-test.cc
+    linux-ipcsocket-test.cc
+    linux-socket-pair-test.cc
+    socket-test.cc
+    shmutils-test.cc
+    signals-test.cc
+    ${RCCL_HOST_HIPIFY_DIR}/src/misc/crypt.cc
+    ${RCCL_HOST_HIPIFY_DIR}/src/misc/utils.cc
+    ../common/main_altrsmi.cpp
+    PARENT_SCOPE)
+
   # rccl-UnitTestsMicroRmaProxy -- REAL src/rma/rma_proxy.cc.
   set(TEST_MICRO_RMA_PROXY_SOURCE_FILES
     rma-proxy-test.cc
@@ -1098,6 +1116,22 @@ foreach(_variant "" "-uncached" "-faultinj")
   endif()
 endforeach()
 
+# ---------------------------------------------------------------------------
+# rccl-UnitTestsMicroLinux
+# ---------------------------------------------------------------------------
+rccl_mark_oracle_tus_generated(${RCCL_HOST_HIPIFY_DIR} crypt)
+rccl_add_micro_binary(rccl-UnitTestsMicroLinux
+  SOURCES ${TEST_MICRO_LINUX_SOURCE_FILES}
+  EXTRA_DEFS
+    ${_MICRO_MODE_DEFS}
+    LINUX_CC_PATH="${RCCL_HOST_HIPIFY_DIR}/src/os/linux.cc"
+    LINUX_IPCSOCKET_CC_PATH="${RCCL_HOST_HIPIFY_DIR}/src/os/linux_ipcsocket.cc"
+    LINUX_SOCKET_PAIR_CC_PATH="${RCCL_HOST_HIPIFY_DIR}/src/os/linux_socket_pair.cc"
+    SOCKET_CC_PATH="${RCCL_HOST_HIPIFY_DIR}/src/misc/socket.cc"
+    SHMUTILS_CC_PATH="${RCCL_HOST_HIPIFY_DIR}/src/misc/shmutils.cc"
+    SIGNALS_CC_PATH="${RCCL_HOST_HIPIFY_DIR}/src/misc/signals.cc"
+  CATEGORIES_YAML "${CMAKE_CURRENT_SOURCE_DIR}/../test_categories_micro_linux.yaml")
+
 # ---------------------------------------------------------------------------
 # rccl-UnitTestsMicroConnect -- ncclTopoPreset / ncclTopoPostset unstubbed.
 # ---------------------------------------------------------------------------

projects/rccl/test/host/MICROTEST_README.md

@@ -292,6 +292,12 @@ export colliding non-`static` symbols; otherwise a unit needs its own binary:
   so a fake tests this binary's own error-propagation logic just as well
   without pulling in that file's dependency floor. See
   `test_categories_micro_scheduler.yaml`.
+- **`rccl-UnitTestsMicroLinux`**: `os/linux.cc` (`LINUX_CC_PATH`, from `linux-test.cc`),
+  plus `os/linux_ipcsocket.cc`, `os/linux_socket_pair.cc`, `misc/socket.cc`,
+  `misc/shmutils.cc` and `misc/signals.cc`, one test TU each; suites `Linux*`,
+  `Socket*`, `Shmutils*`, `Signals*`. Its own binary: `fakes/os_fakes.cc`, which most
+  targets link, defines `ncclOsInitialize`, `ncclOsTopoGetStrFromSys` and other `ncclOs*`
+  symbols `os/linux.cc` defines. See `test_categories_micro_linux.yaml`.
 
 - **`rccl-UnitTestsMicroTaskPrep`** — `src/enqueue/task_prep/task_prep.cc` (via
   `TASK_PREP_CC_PATH`, suite `TaskPrepMicrotest.*`),
@@ -917,6 +923,7 @@ cmake --build build -j"$(nproc)"
 ./build/rccl-UnitTestsMicroEnqueue-devlinker  # same, RCCL_DEVICE_LINKER arm
 ./build/rccl-UnitTestsMicroSymKernels         # sym_kernels.cc tests
 ./build/rccl-UnitTestsMicroTaskPrep           # src/enqueue/task_prep/ + task_sched/ tests
+./build/rccl-UnitTestsMicroLinux              # src/os/linux*.cc, misc/{socket,shmutils,signals}.cc
 ./build/rccl-UnitTestsMicroConnect            # src/graph/connect.cc Preset/Postset tests
 ./build/rccl-UnitTestsMicroDiagnostics        # src/diagnostics/{p2p,ib_write_bw}.cc tests
 ./build/rccl-UnitTestsMicroGraph              # src/graph/{xml,topo,paths,search,rome_models}.cc tests

projects/rccl/test/host/linux-ipcsocket-test.cc

@@ -0,0 +1,298 @@
+/*************************************************************************
+ * Copyright (c) 2026 Advanced Micro Devices, Inc. All rights reserved.
+ *
+ * See LICENSE.txt for license information
+ ************************************************************************/
+
+// Host-only microtests for src/os/linux_ipcsocket.cc on real Unix datagram sockets, abstract and
+// path-bound. socket(), sendmsg() and recvmsg() are seamed only to fail or to forge a reply.
+
+#include <gtest/gtest.h>
+
+#include <cerrno>
+#include <cstring>
+#include <fcntl.h>
+#include <functional>
+#include <string>
+#include <sys/socket.h>
+#include <sys/stat.h>
+#include <sys/un.h>
+#include <unistd.h>
+
+#include "ipcsocket.h"
+#include "os.h"
+#include "param.h"
+#include "utils.h"
+
+#include "linux-test-common.h"
+
+namespace {
+std::function<int(int, int, int)> g_ipcSocket = ::socket;
+std::function<ssize_t(int, msghdr*, int)> g_ipcRecvmsg = ::recvmsg;
+std::function<ssize_t(int, const msghdr*, int)> g_ipcSendmsg = ::sendmsg;
+}  // namespace
+
+#define socket(...) g_ipcSocket(__VA_ARGS__)
+#define recvmsg(...) g_ipcRecvmsg(__VA_ARGS__)
+#define sendmsg(...) g_ipcSendmsg(__VA_ARGS__)
+#include LINUX_IPCSOCKET_CC_PATH
+#undef socket
+#undef recvmsg
+#undef sendmsg
+
+namespace {
+
+bool FdOpen(int fd) { return fcntl(fd, F_GETFD) != -1; }
+
+class LinuxIpcsocketMicrotest : public ::testing::Test {
+ protected:
+  void SetUp() override {
+    g_linuxLog.clear();
+    hash_ = ((uint64_t)getpid() << 20) ^ ++counter_;  // unique across concurrent runs
+  }
+  void TearDown() override {
+    for (auto* s : {&a_, &b_}) ncclIpcSocketClose(s);
+    g_ipcSocket = ::socket;
+    g_ipcRecvmsg = ::recvmsg;
+    g_ipcSendmsg = ::sendmsg;
+  }
+  std::string Name(int rank) {
+    char name[NCCL_IPC_SOCKNAME_LEN];
+    snprintf(name, sizeof(name), "/tmp/nccl-socket-%d-%lx", rank, (unsigned long)hash_);
+    return name;
+  }
+  // a_ is rank 0 and b_ rank 1, both on hash_.
+  void Open(volatile uint32_t* abortFlag = nullptr) {
+    ASSERT_EQ(ncclIpcSocketInit(&a_, 0, hash_, abortFlag), ncclSuccess);
+    ASSERT_EQ(ncclIpcSocketInit(&b_, 1, hash_, abortFlag), ncclSuccess);
+  }
+  // Passes a pipe's write end from a_ to b_ and checks the received copy writes into the pipe.
+  void PassDescriptor() {
+    int p[2];
+    ASSERT_EQ(pipe(p), 0);
+    ASSERT_EQ(ncclIpcSocketSendFd(&a_, p[1], 1, hash_), ncclSuccess);
+    ncclIpcFd got = -1;
+    ASSERT_EQ(ncclIpcSocketRecvFd(&b_, &got), ncclSuccess);
+    EXPECT_NE(got, p[1]);
+    EXPECT_EQ(write(got, "z", 1), 1);
+    char c = 0;
+    EXPECT_EQ(read(p[0], &c, 1), 1);
+    EXPECT_EQ(c, 'z');
+    EXPECT_EQ(ncclIpcFdClose(got), 0);
+    EXPECT_FALSE(FdOpen(got));
+    close(p[0]);
+    close(p[1]);
+  }
+  static uint64_t counter_;
+  uint64_t hash_ = 0;
+  ncclIpcSocket a_ = {-1, "", nullptr}, b_ = {-1, "", nullptr};
+};
+uint64_t LinuxIpcsocketMicrotest::counter_ = 0;
+
+TEST_F(LinuxIpcsocketMicrotest, Init_NullHandleIsAnInternalError) {
+  EXPECT_EQ(ncclIpcSocketInit(nullptr, 0, hash_, nullptr), ncclInternalError);
+}
+
+TEST_F(LinuxIpcsocketMicrotest, Init_AbstractSocketLeavesNoFile) {
+  ASSERT_EQ(ncclIpcSocketInit(&a_, 3, hash_, nullptr), ncclSuccess);
+  EXPECT_TRUE(FdOpen(a_.fd));
+  EXPECT_EQ(a_.socketName, Name(3));
+  EXPECT_EQ(a_.abortFlag, nullptr);
+  EXPECT_NE(access(a_.socketName, F_OK), 0);
+  EXPECT_EQ(fcntl(a_.fd, F_GETFL) & O_NONBLOCK, 0);
+}
+
+TEST_F(LinuxIpcsocketMicrotest, Init_PathSocketReplacesAStaleFileAndCloseRemovesIt) {
+  LinuxScopedEnv env("NCCL_IPC_USE_ABSTRACT_SOCKET", "0");
+  std::string name = Name(0);
+  FILE* stale = fopen(name.c_str(), "w");
+  ASSERT_NE(stale, nullptr);
+  fclose(stale);
+  ASSERT_EQ(ncclIpcSocketInit(&a_, 0, hash_, nullptr), ncclSuccess);
+  struct stat st;
+  ASSERT_EQ(stat(name.c_str(), &st), 0);
+  EXPECT_TRUE(S_ISSOCK(st.st_mode));
+  int fd = a_.fd;
+  ASSERT_EQ(ncclIpcSocketClose(&a_), ncclSuccess);
+  EXPECT_FALSE(FdOpen(fd));
+  EXPECT_NE(access(name.c_str(), F_OK), 0);
+  a_.fd = -1;
+}
+
+TEST_F(LinuxIpcsocketMicrotest, Init_AbortFlagMakesTheSocketNonBlocking) {
+  volatile uint32_t abortFlag = 0;
+  ASSERT_EQ(ncclIpcSocketInit(&a_, 0, hash_, &abortFlag), ncclSuccess);
+  EXPECT_EQ(a_.abortFlag, &abortFlag);
+  EXPECT_NE(fcntl(a_.fd, F_GETFL) & O_NONBLOCK, 0);
+}
+
+TEST_F(LinuxIpcsocketMicrotest, Init_NameInUseIsASystemError) {
+  ASSERT_EQ(ncclIpcSocketInit(&a_, 0, hash_, nullptr), ncclSuccess);
+  EXPECT_EQ(ncclIpcSocketInit(&b_, 0, hash_, nullptr), ncclSystemError);
+  EXPECT_EQ(b_.fd, -1);
+  EXPECT_STREQ(b_.socketName, "");
+  EXPECT_TRUE(LinuxLogged(("WARN UDS: Binding to socket " + Name(0) + " failed : Address already in use").c_str()));
+}
+
+TEST_F(LinuxIpcsocketMicrotest, Init_SocketFailureIsASystemError) {
+  g_ipcSocket = [](int family, int type, int) {
+    EXPECT_EQ(family, AF_UNIX);
+    EXPECT_EQ(type, SOCK_DGRAM);
+    errno = EMFILE;
+    return -1;
+  };
+  EXPECT_EQ(ncclIpcSocketInit(&a_, 0, hash_, nullptr), ncclSystemError);
+  EXPECT_EQ(a_.fd, -1);
+  EXPECT_TRUE(LinuxLogged("WARN UDS: Socket creation error : Too many open files"));
+}
+
+TEST_F(LinuxIpcsocketMicrotest, GetFd) {
+  ncclIpcFd fd = -1;
+  EXPECT_EQ(ncclIpcSocketGetFd(nullptr, &fd), ncclInvalidArgument);
+  EXPECT_TRUE(LinuxLogged("WARN ncclIpcSocketGetFd: pass NULL socket"));
+  Open();
+  EXPECT_EQ(ncclIpcSocketGetFd(&a_, nullptr), ncclSuccess);
+  ASSERT_EQ(ncclIpcSocketGetFd(&a_, &fd), ncclSuccess);
+  EXPECT_EQ(fd, a_.fd);
+}
+
+TEST_F(LinuxIpcsocketMicrotest, Close_RejectsANullSocket) {
+  EXPECT_EQ(ncclIpcSocketClose(nullptr), ncclInternalError);
+}
+
+// Descriptor 0 counts as "not open", like -1, so Close leaves it; a fix flips this pin.
+TEST_F(LinuxIpcsocketMicrotest, Close_KnownBug_TreatsDescriptorZeroAsUnopened) {
+  ncclIpcSocket zero = {0, "", nullptr};
+  EXPECT_EQ(ncclIpcSocketClose(&zero), ncclSuccess);
+  EXPECT_TRUE(FdOpen(0));
+}
+
+TEST_F(LinuxIpcsocketMicrotest, SendFd_PassesADescriptorOverAnAbstractSocket) {
+  Open();
+  PassDescriptor();
+}
+
+TEST_F(LinuxIpcsocketMicrotest, SendFd_PassesADescriptorOverAPathSocket) {
+  LinuxScopedEnv env("NCCL_IPC_USE_ABSTRACT_SOCKET", "0");
+  Open();
+  EXPECT_EQ(access(b_.socketName, F_OK), 0);
+  PassDescriptor();
+  for (auto* s : {&a_, &b_}) {
+    ASSERT_EQ(ncclIpcSocketClose(s), ncclSuccess);  // while the param still says path sockets: unlinks
+    EXPECT_NE(access(s->socketName, F_OK), 0);
+    s->fd = -1;
+  }
+}
+
+TEST_F(LinuxIpcsocketMicrotest, SendMsg_CarriesAHeaderWithOrWithoutADescriptor) {
+  Open();
+  char hdr[6] = "hello";
+  ASSERT_EQ(ncclIpcSocketSendMsg(&a_, hdr, sizeof(hdr), -1, 1, hash_), ncclSuccess);
+  char got[6] = {};
+  ASSERT_EQ(ncclIpcSocketRecvMsg(&b_, got, sizeof(got), nullptr), ncclSuccess);
+  EXPECT_STREQ(got, "hello");
+
+  ASSERT_EQ(ncclIpcSocketSendMsg(&a_, hdr, sizeof(hdr), STDOUT_FILENO, 1, hash_), ncclSuccess);
+  std::memset(got, 0, sizeof(got));
+  ncclIpcFd fd = -1;
+  ASSERT_EQ(ncclIpcSocketRecvMsg(&b_, got, sizeof(got), &fd), ncclSuccess);
+  EXPECT_STREQ(got, "hello");
+  EXPECT_TRUE(FdOpen(fd));
+  close(fd);
+}
+
+TEST_F(LinuxIpcsocketMicrotest, RecvFd_MessageWithoutADescriptorIsASystemError) {
+  Open();
+  ASSERT_EQ(ncclIpcSocketSendMsg(&b_, nullptr, 0, -1, 0, hash_), ncclSuccess);
+  ncclIpcFd fd = -1;
+  EXPECT_EQ(ncclIpcSocketRecvFd(&a_, &fd), ncclSystemError);
+  EXPECT_TRUE(LinuxLogged(("WARN UDS: Receiving data over socket " + Name(0) + " failed").c_str()));
+}
+
+TEST_F(LinuxIpcsocketMicrotest, RecvFd_ForeignControlMessageIsASystemError) {
+  Open();
+  g_ipcRecvmsg = [](int, msghdr* msg, int) -> ssize_t {
+    cmsghdr* c = CMSG_FIRSTHDR(msg);
+    c->cmsg_len = CMSG_LEN(sizeof(int));
+    c->cmsg_level = SOL_SOCKET;
+    c->cmsg_type = SCM_CREDENTIALS;
+    return 1;
+  };
+  ncclIpcFd fd = -1;
+  EXPECT_EQ(ncclIpcSocketRecvFd(&a_, &fd), ncclSystemError);
+  EXPECT_EQ(g_linuxLog, "WARN UDS: Receiving data over socket failed\n");
+  g_ipcRecvmsg = [](int, msghdr* msg, int) -> ssize_t {
+    cmsghdr* c = CMSG_FIRSTHDR(msg);
+    c->cmsg_len = CMSG_LEN(sizeof(int));
+    c->cmsg_level = SOL_IP;
+    c->cmsg_type = SCM_RIGHTS;
+    return 1;
+  };
+  EXPECT_EQ(ncclIpcSocketRecvFd(&a_, &fd), ncclSystemError);
+}
+
+TEST_F(LinuxIpcsocketMicrotest, RecvMsg_RetriesTransientErrorsAndFailsOnOthers) {
+  Open();
+  const int transient[] = {EINTR, EAGAIN, EWOULDBLOCK};
+  int failures = 0;
+  g_ipcRecvmsg = [&](int fd, msghdr* msg, int flags) -> ssize_t {
+    if (failures < 3) {
+      errno = transient[failures++];
+      return -1;
+    }
+    return ::recvmsg(fd, msg, flags);
+  };
+  char hdr[2] = "x", got[2] = {};
+  ASSERT_EQ(ncclIpcSocketSendMsg(&b_, hdr, sizeof(hdr), -1, 0, hash_), ncclSuccess);
+  ASSERT_EQ(ncclIpcSocketRecvMsg(&a_, got, sizeof(got), nullptr), ncclSuccess);
+  EXPECT_EQ(failures, 3);
+  EXPECT_STREQ(got, "x");
+
+  g_ipcRecvmsg = [](int, msghdr*, int) -> ssize_t {
+    errno = ENOTSOCK;
+    return -1;
+  };
+  EXPECT_EQ(ncclIpcSocketRecvMsg(&a_, got, sizeof(got), nullptr), ncclSystemError);
+  EXPECT_TRUE(LinuxLogged("WARN UDS: Receiving data over socket failed : 88"));
+}
+
+TEST_F(LinuxIpcsocketMicrotest, RecvMsg_AbortStopsWaiting) {
+  volatile uint32_t abortFlag = 1;
+  Open(&abortFlag);
+  ncclIpcFd fd = -1;
+  EXPECT_EQ(ncclIpcSocketRecvFd(&a_, &fd), ncclInternalError);  // non-blocking: EAGAIN, then the flag
+  EXPECT_TRUE(g_linuxLog.empty());
+}
+
+TEST_F(LinuxIpcsocketMicrotest, SendMsg_NoPeerIsASystemError) {
+  Open();
+  EXPECT_EQ(ncclIpcSocketSendFd(&a_, STDOUT_FILENO, 7, hash_), ncclSystemError);
+  EXPECT_TRUE(LinuxLogged(("WARN UDS: Sending data over socket " + Name(7) + " failed").c_str()));
+}
+
+TEST_F(LinuxIpcsocketMicrotest, SendMsg_RetriesTransientErrorsUntilAborted) {
+  volatile uint32_t abortFlag = 0;
+  Open(&abortFlag);
+  int attempts = 0;
+  g_ipcSendmsg = [&](int fd, const msghdr* msg, int flags) -> ssize_t {
+    if (++attempts < 3) {
+      errno = attempts == 1 ? EINTR : EAGAIN;
+      return -1;
+    }
+    return ::sendmsg(fd, msg, flags);
+  };
+  char hdr[2] = "y", got[2] = {};
+  ASSERT_EQ(ncclIpcSocketSendMsg(&a_, hdr, sizeof(hdr), -1, 1, hash_), ncclSuccess);
+  EXPECT_EQ(attempts, 3);
+  ASSERT_EQ(ncclIpcSocketRecvMsg(&b_, got, sizeof(got), nullptr), ncclSuccess);
+  EXPECT_STREQ(got, "y");
+
+  abortFlag = 1;
+  g_ipcSendmsg = [](int, const msghdr*, int) -> ssize_t {
+    errno = EWOULDBLOCK;
+    return -1;
+  };
+  EXPECT_EQ(ncclIpcSocketSendMsg(&a_, hdr, sizeof(hdr), -1, 1, hash_), ncclInternalError);
+}
+
+}  // namespace