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

projects/rccl/test/host/CMakeLists.txt

@@ -422,6 +422,20 @@ function(rccl_define_micro_source_lists)
     fakes/transport_p2p_fakes.cc         # src/transport/p2p.cc shareable buffers
     ../common/main_altrsmi.cpp
     PARENT_SCOPE)
+
+  # rccl-UnitTestsMicroLinux -- src/os/linux*.cc, misc/{socket,shmutils,signals}.cc. Own binary: it runs
+  # the real os/linux.cc that the shared fakes stub, plus real crypt.cc and utils.cc.
+  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)
 endfunction()
 
 # Create one host-only micro-test binary and apply all the boilerplate that
@@ -894,7 +908,7 @@ endif()
 # Mark every hipified misc/ oracle TU any target below compiles as GENERATED.
 # The property is directory-scoped and idempotent, so one call up front covers
 # all targets -- no need to repeat it per binary.
-rccl_mark_oracle_tus_generated(${RCCL_HOST_HIPIFY_DIR} argcheck archinfo utils)
+rccl_mark_oracle_tus_generated(${RCCL_HOST_HIPIFY_DIR} argcheck archinfo crypt utils)
 
 # ---------------------------------------------------------------------------
 # rccl-UnitTestsMicro
@@ -1115,6 +1129,21 @@ rccl_add_micro_binary(rccl-UnitTestsMicroClientSupport
     CLIENT_SUPPORT_CC_PATH="${RCCL_HOST_HIPIFY_DIR}/src/ras/client_support.cc"
   CATEGORIES_YAML "${CMAKE_CURRENT_SOURCE_DIR}/../test_categories_micro.yaml")
 
+# ---------------------------------------------------------------------------
+# rccl-UnitTestsMicroLinux
+# ---------------------------------------------------------------------------
+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-UnitTestsMicroSymKernels
 # ---------------------------------------------------------------------------

projects/rccl/test/host/MICROTEST_README.md

@@ -337,6 +337,12 @@ export colliding non-`static` symbols; otherwise a unit needs its own binary:
   `ncclRmaProxyCreateContext` / `ncclRmaProxyDestroyContext` -- so production
   owns the context and its cleanup. See `test_categories_micro_rma_proxy.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 because it runs the real
+  `os/linux.cc` the shared fakes stub, on real sockets, files and shared memory. See
+  `test_categories_micro_linux.yaml`.
 - **`rccl-UnitTestsMicroDiagnostics`**: `src/diagnostics/p2p.cc` (via
   `DIAG_P2P_CC_PATH`, suite `DiagP2pMicrotest.*`) and `src/diagnostics/ib_write_bw.cc`
   (via `DIAG_IB_WRITE_BW_CC_PATH`, suite `DiagIbWriteBwMicrotest.*`). Its own binary: it fakes the
@@ -857,6 +863,7 @@ can be configured **directly** to build every host binary — `rccl-HostUnitTest
 `rccl-UnitTestsMicro`, `rccl-UnitTestsMicroWarpSpeed`,
 `rccl-UnitTestsMicroInit[-uncached|-faultinj]`, `rccl-UnitTestsMicroEnqueue[-devlinker]`,
 `rccl-UnitTestsMicroGinHost`, `rccl-UnitTestsMicroCe`, `rccl-UnitTestsMicroDiagnostics`,
+`rccl-UnitTestsMicroLinux`,
 `rccl-UnitTestsMicroSymKernels`, `rccl-UnitTestsMicroTaskPrep` and
 `rccl-UnitTestsMicroRmaProxy` — **without configuring/building all of
 librccl**. It compiles just the tests + fakes + the hipified unit-under-test
@@ -899,6 +906,7 @@ cmake --build build -j"$(nproc)"
 ./build/rccl-UnitTestsMicroConnect            # src/graph/connect.cc Preset/Postset tests
 ./build/rccl-UnitTestsMicroDiagnostics        # src/diagnostics/{p2p,ib_write_bw}.cc tests
 ./build/rccl-UnitTestsMicroGinHost            # src/gin/gin_host.cc GIN_PROXY_NTHREADS
+./build/rccl-UnitTestsMicroLinux              # src/os/linux*.cc, misc/{socket,shmutils,signals}.cc
 ./build/rccl-UnitTestsMicroDiagnostics        # src/diagnostics/p2p.cc tests
 ./build/rccl-UnitTestsMicroCe                 # src/ce_coll.cc tests
 ./build/rccl-UnitTestsMicroRmaProxy           # src/rma/rma_proxy.cc tests

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

@@ -0,0 +1,295 @@
+/*************************************************************************
+ * 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);
+}
+
+// Observed: descriptor 0 counts as "not open", like -1, so Close leaves it.
+TEST_F(LinuxIpcsocketMicrotest, Close_SkipsNullAndUnopenedHandles) {
+  EXPECT_EQ(ncclIpcSocketClose(nullptr), ncclInternalError);
+  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