diff --git a/.github/pre-commit/spelling_allowlist.txt b/.github/pre-commit/spelling_allowlist.txt index 26e781ce877..abd44017d94 100644 --- a/.github/pre-commit/spelling_allowlist.txt +++ b/.github/pre-commit/spelling_allowlist.txt @@ -228,6 +228,7 @@ YAML ZZ Zener accessor +accessors acknowledgement adaptor adaptors diff --git a/python/README.md.in b/python/README.md.in index b5427f9f80d..2b544ce952c 100644 --- a/python/README.md.in +++ b/python/README.md.in @@ -53,7 +53,7 @@ able to use it without any further installation steps. If you want to perform multi-GPU simulations, additional components must be installed. We recommend using [Conda](https://docs.conda.io/en/latest/) to do so. If you are not already using Conda, you can install a minimal version -following [miniconda instructions here](https://docs.anaconda.com/miniconda/). +following [Miniconda instructions here](https://docs.anaconda.com/miniconda/). The following commands will create and activate a complete environment for CUDA-Q with all its dependencies: @@ -73,7 +73,7 @@ source $CONDA_PREFIX/lib/python3.11/site-packages/distributed_interfaces/activat [//]: # (End conda install) -**Warning (conda-forge)**: Installing `cudaq` from the `conda-forge` channel on recent versions of Python can lead to segmentation faults. +**Warning (`conda-forge`)**: Installing `cudaq` from the `conda-forge` channel on recent versions of Python can lead to segmentation faults. If you see such errors, please switch to one of the Python versions 3.11.10 or 3.12.7 as per the issue [#2999](https://github.com/NVIDIA/cuda-quantum/issues/2999) or [#3104](https://github.com/NVIDIA/cuda-quantum/issues/3104). diff --git a/realtime/include/cudaq/realtime/testing/server_process.h b/realtime/include/cudaq/realtime/testing/server_process.h new file mode 100644 index 00000000000..8bc89e71201 --- /dev/null +++ b/realtime/include/cudaq/realtime/testing/server_process.h @@ -0,0 +1,326 @@ +/****************************************************************-*- C++ -*-**** + * Copyright (c) 2026 NVIDIA Corporation & Affiliates. * + * All rights reserved. * + * * + * This source code and the accompanying materials are made available under * + * the terms of the Apache License 2.0 which accompanies this distribution. * + ******************************************************************************/ + +#pragma once + +/// @file server_process.h +/// @brief Header-only fork/exec harness for two-process tests: spawn a server +/// binary, wait for its readiness line, read the `key=value` endpoint +/// description out of it, then shut it down and collect the summary +/// lines it prints on the way out. +/// +/// The readiness handshake, rather than a sleep, is what makes a two-process +/// test deterministic: the server publishes its endpoint (an ephemeral port, a +/// rendezvous port, an RDMA QP -- whatever the transport uses) on `stdout`, +/// and the client only dials once that line has been read. Both `stdout` and +/// `stderr` are folded into one pipe so a server that dies during bring-up +/// leaves its diagnostic in `output()` instead of the test seeing a bare +/// timeout. +/// +/// Every complete line read from the child is also echoed to the test's own +/// `stdout` behind a `[server]` prefix, so `ctest -V` shows the server's side +/// of a run that passed rather than only of one that failed. Nothing reads +/// the pipe except the waits for a specific line, so `stop()` drains what is +/// left before closing it -- without that, the lines a server prints while +/// shutting down would never be seen. +/// +/// The READY line is parsed as free-form whitespace-separated `key=value` +/// tokens into `fields()`, so one harness serves any transport: the `udp` +/// provider publishes `transport=udp port=N`, cpu_roce publishes +/// `transport=cpu_roce port=N roce_ip=A` in rendezvous mode and +/// `qp=`/`rkey=`/`buffer_addr=` in hsb_fpga mode. Nothing here knows which. +/// +/// Usage: +/// \code +/// ServerProcess server; +/// ASSERT_TRUE(server.start({binary, "--transport=udp", "--port=0"}, +/// "CUDAQ_REALTIME_SERVER_READY")) +/// << server.output(); +/// ... talk to server.port() ... +/// const std::string line = server.stopAndReadLine("..._PROCESSED", 5s); +/// \endcode + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace cudaq::realtime::testing { + +class ServerProcess { +public: + ServerProcess() = default; + ServerProcess(const ServerProcess &) = delete; + ServerProcess &operator=(const ServerProcess &) = delete; + + ~ServerProcess() { + if (pid_ > 0) { + ::kill(pid_, SIGKILL); + int status = 0; + ::waitpid(pid_, &status, 0); + pid_ = -1; + } + closeFd(); + } + + /// Spawn `argv[0]` with `argv` and read its output until a line beginning + /// with `ready_prefix` appears. Returns false on spawn failure, on timeout, + /// or if the child exits first (a rejected command line) -- in every case + /// `output()` holds everything the child printed. + bool start(const std::vector &argv, + const std::string &ready_prefix, + std::chrono::milliseconds timeout = std::chrono::seconds(15)) { + if (argv.empty()) + return false; + + int out_pipe[2] = {-1, -1}; + if (::pipe(out_pipe) != 0) + return false; + + pid_ = ::fork(); + if (pid_ < 0) { + ::close(out_pipe[0]); + ::close(out_pipe[1]); + return false; + } + if (pid_ == 0) { + // Fold stderr into the same pipe: a bring-up failure prints there, and + // the parent needs it to explain the missing READY line. + ::dup2(out_pipe[1], STDOUT_FILENO); + ::dup2(out_pipe[1], STDERR_FILENO); + ::close(out_pipe[0]); + ::close(out_pipe[1]); + std::vector args = argv; + std::vector raw; + raw.reserve(args.size() + 1); + for (auto &a : args) + raw.push_back(a.data()); + raw.push_back(nullptr); + ::execv(args[0].c_str(), raw.data()); + std::perror(("execv " + args[0]).c_str()); + _exit(127); + } + + ::close(out_pipe[1]); + out_fd_ = out_pipe[0]; + // Non-blocking reads let the drain loop empty the pipe without an extra + // poll() per byte and without ever blocking mid-line. + ::fcntl(out_fd_, F_SETFL, ::fcntl(out_fd_, F_GETFL, 0) | O_NONBLOCK); + + std::string ready_line; + if (!readLineWithPrefix(ready_prefix, timeout, ready_line)) + return false; + parseFields(ready_line); + return true; + } + + /// Ask the server to shut down, then return the first line beginning with + /// `prefix` that it prints on the way out (empty string if none appears). + /// Servers whose counters are only final after their dispatch loop exits + /// report them here rather than while running. + std::string stopAndReadLine(const std::string &prefix, + std::chrono::milliseconds timeout) { + if (pid_ <= 0) + return {}; + ::kill(pid_, SIGTERM); + std::string line; + const bool found = readLineWithPrefix(prefix, timeout, line); + reapBounded(std::chrono::milliseconds(1000)); + drain(); // anything printed after the line we waited for + closeFd(); + return found ? line : std::string{}; + } + + /// SIGTERM, then SIGKILL if the child has not exited within a second. A + /// server that ignores SIGTERM must not hang the test until the `ctest` + /// timeout. + void stop() { + if (pid_ > 0) { + ::kill(pid_, SIGTERM); + reapBounded(std::chrono::milliseconds(1000)); + } + // After the reap: the child has closed its end, so this collects + // everything it printed on the way out rather than racing it. + drain(); + closeFd(); + } + + /// Reap a child that exited on its own and return its exit code (-1 when it + /// died by signal, never ran, or had to be killed). Used for the cases + /// where the server is expected to reject its command line. + int exitCode(std::chrono::milliseconds grace = std::chrono::seconds(5)) { + if (pid_ <= 0) + return -1; + drain(); + const int code = reapBounded(grace); + closeFd(); + return code; + } + + /// `key=value` tokens parsed out of the READY line. + const std::map &fields() const { return fields_; } + + /// Convenience accessor for the near-universal `port=` field (0 when the + /// transport does not publish one). + std::uint16_t port() const { + const auto it = fields_.find("port"); + if (it == fields_.end()) + return 0; + try { + return static_cast(std::stoul(it->second)); + } catch (const std::exception &) { + return 0; + } + } + + /// Everything read from the child so far, for failure messages. + const std::string &output() const { return output_; } + +private: + bool readLineWithPrefix(const std::string &prefix, + std::chrono::milliseconds timeout, + std::string &line_out) { + if (out_fd_ < 0) + return false; + // Deadline rather than a per-iteration tick count: charging a fixed cost + // per poll() would time out a chatty server long before `timeout`. + const auto deadline = std::chrono::steady_clock::now() + timeout; + for (;;) { + const auto now = std::chrono::steady_clock::now(); + if (now >= deadline) + return false; + const auto remaining = + std::chrono::duration_cast(deadline - now) + .count(); + pollfd pfd{out_fd_, POLLIN, 0}; + const int ready = + ::poll(&pfd, 1, static_cast(remaining < 100 ? remaining : 100)); + if (ready < 0) { + if (errno == EINTR) + continue; + return false; + } + if (ready == 0) + continue; + + char c = 0; + ssize_t n = 0; + while ((n = ::read(out_fd_, &c, 1)) == 1) { + output_.push_back(c); + if (c != '\n') { + partial_.push_back(c); + continue; + } + const bool match = partial_.rfind(prefix, 0) == 0; + echoLine(partial_); + line_out = partial_; + partial_.clear(); + if (match) + return true; + } + if (n == 0) + return false; // EOF: the child exited without printing the prefix + } + } + + // Read whatever is already buffered without waiting, so output() is complete + // for a child that has already exited. Split on newlines as we go so the + // tail of the run is echoed a line at a time like the rest. + void drain() { + if (out_fd_ < 0) + return; + char buffer[512]; + ssize_t n = 0; + while ((n = ::read(out_fd_, buffer, sizeof(buffer))) > 0) { + output_.append(buffer, static_cast(n)); + for (ssize_t i = 0; i < n; ++i) { + if (buffer[i] != '\n') { + partial_.push_back(buffer[i]); + continue; + } + echoLine(partial_); + partial_.clear(); + } + } + // A child killed mid-line still deserves to have that line shown. + echoLine(partial_); + partial_.clear(); + } + + // Echo one line of the child's output to the test's own stdout. printf + // rather than std::cout so it shares gtest's stream and buffering, which is + // what keeps the interleaving with the test log in order. + void echoLine(const std::string &line) const { + if (line.empty()) + return; + std::printf("[server] %s\n", line.c_str()); + std::fflush(stdout); + } + + // Wait up to `grace` for a voluntary exit, then SIGKILL and wait. + int reapBounded(std::chrono::milliseconds grace) { + if (pid_ <= 0) + return -1; + const auto deadline = std::chrono::steady_clock::now() + grace; + int status = 0; + for (;;) { + const pid_t done = ::waitpid(pid_, &status, WNOHANG); + if (done == pid_) + break; + if (done < 0) { + pid_ = -1; + return -1; + } + if (std::chrono::steady_clock::now() >= deadline) { + ::kill(pid_, SIGKILL); + ::waitpid(pid_, &status, 0); + pid_ = -1; + return -1; + } + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + } + pid_ = -1; + return WIFEXITED(status) ? WEXITSTATUS(status) : -1; + } + + void parseFields(const std::string &ready_line) { + std::istringstream tokens(ready_line); + std::string token; + while (tokens >> token) { + const auto eq = token.find('='); + if (eq == std::string::npos || eq == 0) + continue; + fields_[token.substr(0, eq)] = token.substr(eq + 1); + } + } + + void closeFd() { + if (out_fd_ >= 0) { + ::close(out_fd_); + out_fd_ = -1; + } + } + + pid_t pid_ = -1; + int out_fd_ = -1; + std::string output_; // everything read + std::string partial_; // bytes of the line currently being assembled + std::map fields_; +}; + +} // namespace cudaq::realtime::testing diff --git a/realtime/include/cudaq/realtime/testing/test_utils.h b/realtime/include/cudaq/realtime/testing/test_utils.h new file mode 100644 index 00000000000..5556359d4ef --- /dev/null +++ b/realtime/include/cudaq/realtime/testing/test_utils.h @@ -0,0 +1,74 @@ +/****************************************************************-*- C++ -*-**** + * Copyright (c) 2026 NVIDIA Corporation & Affiliates. * + * All rights reserved. * + * * + * This source code and the accompanying materials are made available under * + * the terms of the Apache License 2.0 which accompanies this distribution. * + ******************************************************************************/ + +#pragma once + +/// @file test_utils.h +/// @brief Accessors for the address-as-flag ring protocol, for tests and tools +/// that drive a ring buffer by hand instead of through a dispatcher. +/// + +#include +#include +#include +#include + +namespace cudaq::realtime::testing { + +/// @brief Read slot `slot`'s flag: 0 when free, otherwise the slot address. +inline std::uint64_t load_flag(std::uint64_t flags_addr, unsigned slot) { + const auto *flags = reinterpret_cast(flags_addr); + return __atomic_load_n(&flags[slot], __ATOMIC_ACQUIRE); +} + +/// @brief Publish (`value` = slot address) or release (`value` = 0) a slot. +/// The release ordering is what makes the slot's payload visible to the +/// consumer that observes the flag. +inline void store_flag(std::uint64_t flags_addr, unsigned slot, + std::uint64_t value) { + auto *flags = reinterpret_cast(flags_addr); + __atomic_store_n(&flags[slot], value, __ATOMIC_RELEASE); +} + +/// @brief First byte of slot `slot`, given the ring's slot stride. +inline std::uint8_t *slot_data(std::uint64_t data_addr, unsigned slot, + std::size_t stride) { + return reinterpret_cast(data_addr) + slot * stride; +} + +/// @brief Poll until slot `slot` is published, or the timeout expires. +/// @return true if the flag became non-zero. +inline bool wait_for_flag( + std::uint64_t flags_addr, unsigned slot, + std::chrono::milliseconds timeout = std::chrono::milliseconds(5000)) { + const auto deadline = std::chrono::steady_clock::now() + timeout; + while (std::chrono::steady_clock::now() < deadline) { + if (load_flag(flags_addr, slot) != 0) + return true; + std::this_thread::sleep_for(std::chrono::microseconds(100)); + } + return false; +} + +/// @brief Poll until slot `slot` is released, or the timeout expires. This is +/// how a producer observes back-pressure: the slot is reusable once the +/// consumer has cleared it. +/// @return true if the flag became zero. +inline bool wait_for_flag_clear( + std::uint64_t flags_addr, unsigned slot, + std::chrono::milliseconds timeout = std::chrono::milliseconds(5000)) { + const auto deadline = std::chrono::steady_clock::now() + timeout; + while (std::chrono::steady_clock::now() < deadline) { + if (load_flag(flags_addr, slot) == 0) + return true; + std::this_thread::sleep_for(std::chrono::microseconds(100)); + } + return false; +} + +} // namespace cudaq::realtime::testing diff --git a/realtime/scripts/deps_common.sh b/realtime/scripts/deps_common.sh index 7efb1415af9..bf07f4075f2 100644 --- a/realtime/scripts/deps_common.sh +++ b/realtime/scripts/deps_common.sh @@ -8,6 +8,25 @@ # the terms of the Apache License 2.0 which accompanies this distribution. # # ============================================================================ # +# Retry a command, clearing package-manager metadata between attempts. The CUDA +# yum repo CDN intermittently serves a stale repomd.xml that points at rotated +# repodata files, producing 404s; clearing metadata forces a fresh fetch. +function retry { + local n=0 max=5 delay=15 + until "$@"; do + n=$((n+1)) + if [ "$n" -ge "$max" ]; then + echo "Command failed after $max attempts: $*" >&2 + return 1 + fi + echo "Attempt $n/$max failed; clearing repo metadata and retrying in ${delay}s..." >&2 + if [ -x "$(command -v dnf)" ]; then dnf clean all || true + elif [ -x "$(command -v apt-get)" ]; then apt-get clean || true; fi + sleep "$delay" + done +} + + # Version pins and helpers shared by the CUDA-Q Realtime dependency scripts, # i.e., install_dev_prerequisites.sh (standard apt path) and install_devdeps.sh # (containers that already ship Mellanox OFED). @@ -62,7 +81,8 @@ cudaq_realtime_cuda_native_arch() { # Register the DOCA host apt repository for this architecture and distro. cudaq_realtime_add_doca_repo() { if [ ! -x "$(command -v curl)" ] || [ ! -x "$(command -v gpg)" ]; then - apt-get update && apt-get install -y --no-install-recommends curl gnupg + retry apt-get update + retry apt-get install -y --no-install-recommends curl gnupg fi echo "Installing DOCA version $CUDAQ_REALTIME_DOCA_VERSION..." @@ -75,7 +95,7 @@ cudaq_realtime_add_doca_repo() { echo "Using DOCA_REPO_LINK=${DOCA_URL}" curl https://linux.mellanox.com/public/repo/doca/GPG-KEY-Mellanox.pub | gpg --dearmor > /etc/apt/trusted.gpg.d/GPG-KEY-Mellanox.pub echo "deb [signed-by=/etc/apt/trusted.gpg.d/GPG-KEY-Mellanox.pub] $DOCA_URL ./" > /etc/apt/sources.list.d/doca.list - apt-get update + retry apt-get update } # Install the Holoscan SDK matching the CUDA toolkit in use. Set @@ -84,8 +104,8 @@ cudaq_realtime_add_doca_repo() { # resolving the Holoscan dependency chain. cudaq_realtime_install_holoscan() { _cudaq_realtime_holoscan_cuda_major=$(cudaq_realtime_cuda_major) || return 1 - apt-get update - if apt-get install -y --no-install-recommends \ + retry apt-get update + if retry apt-get install -y --no-install-recommends \ holoscan-cuda-$_cudaq_realtime_holoscan_cuda_major; then return 0 fi @@ -94,7 +114,7 @@ cudaq_realtime_install_holoscan() { fi _cudaq_realtime_holoscan_tmp=$(mktemp -d) (cd "$_cudaq_realtime_holoscan_tmp" && - apt-get download holoscan holoscan-cuda-$_cudaq_realtime_holoscan_cuda_major && + retry apt-get download holoscan holoscan-cuda-$_cudaq_realtime_holoscan_cuda_major && dpkg --force-depends -i holoscan*.deb) _cudaq_realtime_holoscan_status=$? rm -rf "$_cudaq_realtime_holoscan_tmp" diff --git a/realtime/scripts/install_dev_prerequisites.sh b/realtime/scripts/install_dev_prerequisites.sh index 6d441e35cd9..5b75ea2d38b 100755 --- a/realtime/scripts/install_dev_prerequisites.sh +++ b/realtime/scripts/install_dev_prerequisites.sh @@ -28,6 +28,7 @@ set -e . "$(dirname "$0")/deps_common.sh" +retry apt-get update if [ -x "$(command -v apt-get)" ]; then # Fail early if the CUDA toolkit is missing. @@ -35,15 +36,15 @@ if [ -x "$(command -v apt-get)" ]; then # [Build tools] # Needed to build HSB from source below. - apt-get update && apt-get install -y --no-install-recommends git ninja-build pkg-config + retry apt-get install -y --no-install-recommends git ninja-build pkg-config # [libibverbs] echo "Installing libibverbs..." - apt-get update && apt-get install -y --no-install-recommends libibverbs-dev + retry apt-get install -y --no-install-recommends libibverbs-dev # [DOCA Host] cudaq_realtime_add_doca_repo - DEBIAN_FRONTEND=noninteractive apt-get -y install doca-all libdoca-sdk-gpunetio-dev + DEBIAN_FRONTEND=noninteractive retry apt-get -y install doca-all libdoca-sdk-gpunetio-dev # [Holoscan SDK] cudaq_realtime_install_holoscan diff --git a/realtime/scripts/install_devdeps.sh b/realtime/scripts/install_devdeps.sh index 371f19bb71d..e0d2e96862f 100755 --- a/realtime/scripts/install_devdeps.sh +++ b/realtime/scripts/install_devdeps.sh @@ -35,17 +35,18 @@ fi # Fail early if the CUDA toolkit is missing; the exact version is needed below. CUDA_FULL_VERSION=$(cudaq_realtime_cuda_version) -apt-get update && apt-get install -y --no-install-recommends \ +retry apt-get update +retry apt-get install -y --no-install-recommends \ git ninja-build curl pkg-config # [DOCA Host] # Only the GPUNetIO dev package, not doca-all. cudaq_realtime_add_doca_repo -apt-get -y install --no-install-recommends libdoca-sdk-gpunetio-dev +retry apt-get -y install --no-install-recommends libdoca-sdk-gpunetio-dev # hololink_core links CUDA::nvrtc -- must match the exact toolkit version CUDA_VER_DASH=$(echo $CUDA_FULL_VERSION | sed 's/\./-/') -apt-get install -y cuda-nvrtc-dev-$CUDA_VER_DASH 2>/dev/null || true +retry apt-get install -y cuda-nvrtc-dev-$CUDA_VER_DASH 2>/dev/null || true # [Holoscan SDK] export CUDAQ_REALTIME_HOLOSCAN_FORCE_DEPS=1 diff --git a/realtime/unittests/CMakeLists.txt b/realtime/unittests/CMakeLists.txt index b3669995621..70ef396bc43 100644 --- a/realtime/unittests/CMakeLists.txt +++ b/realtime/unittests/CMakeLists.txt @@ -115,6 +115,13 @@ if (CUDAQ_REALTIME_ENABLE_HSB_TOOLS) add_subdirectory(bridge_interface/gpu_roce) endif() +# Transport-agnostic two-process test server. Links cudaq-realtime, which +# only exists when CUDA was found, and must be configured BEFORE the transport +# test directories that exec it. +if (TARGET cudaq-realtime) + add_subdirectory(test_server) +endif() + # UDP bridge provider tests: self-gated on the always-built udp provider # target, so they run in CI without any HSB/RDMA setup. add_subdirectory(bridge_interface/udp) diff --git a/realtime/unittests/bridge_interface/udp/CMakeLists.txt b/realtime/unittests/bridge_interface/udp/CMakeLists.txt index 844eef8bb86..9597e0d508b 100644 --- a/realtime/unittests/bridge_interface/udp/CMakeLists.txt +++ b/realtime/unittests/bridge_interface/udp/CMakeLists.txt @@ -9,6 +9,11 @@ # UDP bridge provider tests # ============================================================================== +# Everything here loads or drives the udp provider .so. +if (NOT TARGET cudaq-realtime-bridge-udp) + return() +endif() + # Functional: load the built provider by absolute path and bring a plain # (non-pinned) udp bridge up through the v2 queries. Links only the bridge # loader (cudaq-realtime), NOT the CUDA runtime, so a regressed dynamic cudart @@ -25,3 +30,32 @@ target_link_libraries(test_udp_bridge_provider PRIVATE ) cudaq_gtest_discover_tests(test_udp_bridge_provider DISCOVERY_TIMEOUT 120) + +# End-to-end: spawn the transport-agnostic test server on a udp bridge and +# drive it from a caller transceiver in this process. A separate binary from +# test_udp_bridge_provider on purpose -- that one links no transport and no CUDA +# runtime, which is what makes it catch a regressed cudart dependency in the +# provider. This one needs the transceiver to speak the wire. +if (TARGET cudaq-realtime-test-server AND TARGET cudaq-realtime-udp-transport) + add_executable(test_udp_two_process test_udp_two_process.cpp) + + target_include_directories(test_udp_two_process PRIVATE + ${CUDAQ_REALTIME_INCLUDE_DIR} + ) + + target_compile_definitions(test_udp_two_process PRIVATE + CUDAQ_REALTIME_TEST_SERVER_PATH="$" + ) + + target_link_libraries(test_udp_two_process PRIVATE + GTest::gtest_main + cudaq-realtime-udp-transport + Threads::Threads + ) + + # The server is exec'd, not linked. + add_dependencies(test_udp_two_process cudaq-realtime-test-server) + + cudaq_gtest_discover_tests(test_udp_two_process DISCOVERY_TIMEOUT 120) + message(STATUS " - test_udp_two_process (UDP dispatch, two processes)") +endif() diff --git a/realtime/unittests/bridge_interface/udp/test_udp_two_process.cpp b/realtime/unittests/bridge_interface/udp/test_udp_two_process.cpp new file mode 100644 index 00000000000..ddda5835fb0 --- /dev/null +++ b/realtime/unittests/bridge_interface/udp/test_udp_two_process.cpp @@ -0,0 +1,295 @@ +/******************************************************************************* + * Copyright (c) 2026 NVIDIA Corporation & Affiliates. * + * All rights reserved. * + * * + * This source code and the accompanying materials are made available under * + * the terms of the Apache License 2.0 which accompanies this distribution. * + ******************************************************************************/ + +/// @file test_udp_two_process.cpp +/// @brief End-to-end two-process test of the UDP transport: a real dispatcher +/// in a child process serving rpc_increment, driven over loopback UDP. +/// +/// This is the first test that exercises the whole service path at once -- +/// provider, ring buffer, host dispatch loop, RPC framing, HOST_CALL handler -- +/// against a caller in a separate process. The in-process tests either stop at +/// the provider (test_udp_bridge_provider), at the wire (test_udp_transceiver), +/// or at the dispatch loop (test_host_dispatcher). +/// +/// The caller is a second UDP transceiver rather than a raw socket, for two +/// reasons: it ships one full slot stride per datagram (getting that wrong is +/// how a hand-rolled client silently loses every request -- see the "Wire +/// behavior" note in udp_wrapper.h), and it gives the same ring contract the +/// production caller uses. Malformed frames are still expressible, because the +/// test writes the request bytes into the ring slot itself. +/// +/// Both rings are strict FIFO, so the fixture tracks a TX and an RX cursor +/// instead of naming slots: a request occupies the next TX slot, a response +/// arrives in the next RX slot. The two cursors move independently, which is +/// exactly what a dropped request does -- it consumes a TX slot and produces no +/// response -- and it keeps the test bodies free of slot arithmetic. +/// +/// CONSEQUENCE FOR THE NEGATIVE CASES: an undispatchable request must be the +/// LAST one a test sends. The transceiver's TX pump consumes slots in strict +/// cursor order and parks on the first slot whose flag is clear (udp_wrapper.h +/// assumes every slot gets a response, which is true of the device_call caller +/// but not of the dispatcher), while the dispatcher publishes a TX slot only +/// for requests it actually answers. A drop therefore leaves a permanent gap +/// that stalls every later response on the wire. The dispatcher itself is +/// unaffected -- it keeps consuming and counting -- so what these tests can +/// verify is that a refused frame produces no response and is not counted as +/// dispatched, measured against a round trip that is known to work. +/// +/// Parameterized on the dispatch shape, which crosses the process boundary as a +/// command-line token. Only "ring" is instantiated today; the client half of +/// the wire is byte-identical under "unified" (that shape changes only how the +/// server services its own rings), so adding it is one value in the +/// instantiation list once the provider exposes a CPU data-plane. + +#include "cudaq/realtime/cpu_transport/udp_wrapper.h" +#include "cudaq/realtime/daemon/dispatcher/dispatch_kernel_launch.h" +#include "cudaq/realtime/testing/server_process.h" +#include "cudaq/realtime/testing/test_utils.h" +#include +#include +#include +#include +#include +#include +#include +#include + +#ifndef CUDAQ_REALTIME_TEST_SERVER_PATH +#error "CUDAQ_REALTIME_TEST_SERVER_PATH must be defined (path to the server)" +#endif + +using cudaq::realtime::fnv1a_hash; +using cudaq::realtime::RPC_MAGIC_REQUEST; +using cudaq::realtime::RPC_MAGIC_RESPONSE; +using cudaq::realtime::RPCHeader; +using cudaq::realtime::RPCResponse; +using cudaq::realtime::testing::ServerProcess; +using cudaq::realtime::testing::slot_data; +using cudaq::realtime::testing::store_flag; +using cudaq::realtime::testing::wait_for_flag; + +namespace { + +constexpr const char *kReadyPrefix = "CUDAQ_REALTIME_SERVER_READY"; +constexpr const char *kProcessedPrefix = "CUDAQ_REALTIME_SERVER_PROCESSED"; +constexpr std::uint32_t kIncrementId = fnv1a_hash("rpc_increment"); + +// Small enough to keep the burst test quick, large enough that the burst wraps +// the ring several times. +constexpr unsigned kNumSlots = 4; +constexpr std::size_t kSlotSize = 256; + +// A round trip over loopback is sub-millisecond; these are backstops, not +// expected waits. +constexpr auto kResponseTimeout = std::chrono::milliseconds(5000); +constexpr auto kNoResponseWindow = std::chrono::milliseconds(500); + +class UdpTwoProcess : public ::testing::TestWithParam { +protected: + void SetUp() override { + const std::vector argv = { + CUDAQ_REALTIME_TEST_SERVER_PATH, "--transport=udp", + std::string("--dispatch=") + GetParam(), "--port=0", + "--num-slots=" + std::to_string(kNumSlots), + "--slot-size=" + std::to_string(kSlotSize), + // Backstop in case a failing test leaves the child unreaped. + "--timeout=120"}; + ASSERT_TRUE(server.start(argv, kReadyPrefix)) + << "server did not become ready; output:\n" + << server.output(); + + // The shape the server actually brought up, not the one we asked for. + EXPECT_EQ(std::string(GetParam()), field("dispatch")) << server.output(); + ASSERT_EQ("udp", field("transport")) << server.output(); + + // Geometry comes from the handshake: both ends must agree on the slot + // stride or every datagram is dropped as oversized. + slots = static_cast(std::stoul(field("slots"))); + slotSize = static_cast(std::stoul(field("slot_size"))); + ASSERT_EQ(kNumSlots, slots); + ASSERT_EQ(kSlotSize, slotSize); + ASSERT_NE(0, server.port()) << server.output(); + + caller = cpu_udp_create_transceiver(slotSize, slots); + ASSERT_NE(nullptr, caller); + ASSERT_EQ(1, cpu_udp_connect(caller, "127.0.0.1", server.port())); + ASSERT_EQ(1, cpu_udp_start(caller)); + + txFlags = cpu_udp_get_tx_ring_flag_addr(caller); + txData = cpu_udp_get_tx_ring_data_addr(caller); + rxFlags = cpu_udp_get_rx_ring_flag_addr(caller); + rxData = cpu_udp_get_rx_ring_data_addr(caller); + } + + void TearDown() override { + // Caller first: no datagrams in flight while the server tears its rings + // down. + if (caller) + cpu_udp_destroy_transceiver(caller); + server.stop(); + } + + std::string field(const std::string &key) const { + const auto it = server.fields().find(key); + return it == server.fields().end() ? std::string{} : it->second; + } + + // Publish one request from the next TX slot: a 24-byte RPCHeader followed by + // `args` verbatim. The framing defaults to a well-formed rpc_increment + // call; the negative cases override `magic` or `function_id` to post a frame + // the dispatcher has to refuse. + void postIncrement(std::uint32_t request_id, std::string_view args, + std::uint32_t magic = RPC_MAGIC_REQUEST, + std::uint32_t function_id = kIncrementId) { + ASSERT_LE(sizeof(RPCHeader) + args.size(), slotSize); + std::uint8_t *tx = slot_data(txData, txCursor, slotSize); + std::memset(tx, 0, slotSize); + + RPCHeader header{}; + header.magic = magic; + header.function_id = function_id; + header.arg_len = static_cast(args.size()); + header.request_id = request_id; + header.ptp_timestamp = 0; + std::memcpy(tx, &header, sizeof(header)); + if (!args.empty()) + std::memcpy(tx + sizeof(header), args.data(), args.size()); + + store_flag(txFlags, txCursor, reinterpret_cast(tx)); + txCursor = (txCursor + 1) % slots; + } + + // Wait for the next response, verify it is `sent` with every byte + // incremented, and recycle the slot so the RX pump's back-pressure releases. + void expectIncrement(std::uint32_t request_id, std::string_view sent) { + const unsigned slot = rxCursor; + ASSERT_TRUE(waitForResponse(slot, kResponseTimeout)) + << "no response in rx slot " << slot << " for request " << request_id + << "; server output:\n" + << server.output(); + + const std::uint8_t *rx = slot_data(rxData, slot, slotSize); + RPCResponse response{}; + std::memcpy(&response, rx, sizeof(response)); + EXPECT_EQ(RPC_MAGIC_RESPONSE, response.magic); + EXPECT_EQ(0, response.status); + EXPECT_EQ(static_cast(sent.size()), response.result_len); + EXPECT_EQ(request_id, response.request_id); + + std::string expected(sent.size(), '\0'); + for (std::size_t i = 0; i < sent.size(); ++i) + expected[i] = static_cast(sent[i] + 1); + const auto *payload = reinterpret_cast(rx + sizeof(response)); + EXPECT_EQ(expected, std::string(payload, sent.size())); + + store_flag(rxFlags, slot, 0); + rxCursor = (rxCursor + 1) % slots; + } + + // The dispatcher consumes an undispatchable slot without producing a + // response, so the next RX slot must stay empty. + void expectNoResponse() { + EXPECT_FALSE(waitForResponse(rxCursor, kNoResponseWindow)) + << "expected no response in rx slot " << rxCursor + << "; server output:\n" + << server.output(); + } + + // Shut the server down and check the count it reports on the way out. Only + // successful HOST_CALL invocations are counted, so an exact match is also + // evidence that nothing else was dispatched. + void expectProcessedCount(unsigned long long expected) { + const std::string line = + server.stopAndReadLine(kProcessedPrefix, std::chrono::seconds(10)); + ASSERT_FALSE(line.empty()) << "no processed-count line; server output:\n" + << server.output(); + unsigned long long count = 0; + ASSERT_EQ(1, + std::sscanf(line.c_str(), + "CUDAQ_REALTIME_SERVER_PROCESSED count=%llu", &count)) + << "unparsable line: " << line; + EXPECT_EQ(expected, count); + } + + bool waitForResponse(unsigned slot, std::chrono::milliseconds timeout) const { + return wait_for_flag(rxFlags, slot, timeout); + } + + ServerProcess server; + cpu_udp_transceiver_t caller = nullptr; + unsigned slots = 0; + std::size_t slotSize = 0; + std::uint64_t txFlags = 0, txData = 0, rxFlags = 0, rxData = 0; + unsigned txCursor = 0, rxCursor = 0; +}; + +TEST_P(UdpTwoProcess, IncrementsStringPayload) { + const std::string payload = "hello dispatcher"; + postIncrement(/*request_id=*/1, payload); + expectIncrement(/*request_id=*/1, payload); +} + +TEST_P(UdpTwoProcess, IncrementsBurstAcrossRingWrap) { + // Three times around both rings: every slot is reused, so a slot that failed + // to recycle would stall the run rather than pass. + const unsigned requests = 3 * slots; + for (unsigned i = 0; i < requests; ++i) { + const std::string payload = "burst-" + std::to_string(i); + postIncrement(/*request_id=*/i + 1, payload); + expectIncrement(/*request_id=*/i + 1, payload); + } +} + +TEST_P(UdpTwoProcess, DropsBadMagic) { + // Round trip first, so the refusal below is measured against a path proven + // to work rather than against a server that might never have come up. + const std::string payload = "before bad magic"; + postIncrement(/*request_id=*/1, payload); + expectIncrement(/*request_id=*/1, payload); + + postIncrement(/*request_id=*/2, "unframed", /*magic=*/0xdeadbeefu); + expectNoResponse(); + + // Bad framing retires the slot without reaching a handler, so only the + // well-formed request counts. + expectProcessedCount(1); +} + +TEST_P(UdpTwoProcess, DropsUnknownFunctionId) { + const std::string payload = "before unknown id"; + postIncrement(/*request_id=*/1, payload); + expectIncrement(/*request_id=*/1, payload); + + // Correct framing, no such entry in the function table. + postIncrement(/*request_id=*/2, "unroutable", RPC_MAGIC_REQUEST, + /*function_id=*/fnv1a_hash("no_such_function")); + expectNoResponse(); + + expectProcessedCount(1); +} + +TEST_P(UdpTwoProcess, ReportsProcessedCount) { + constexpr unsigned kRequests = 5; + for (unsigned i = 0; i < kRequests; ++i) { + const std::string payload = "counted-" + std::to_string(i); + postIncrement(/*request_id=*/i + 1, payload); + expectIncrement(/*request_id=*/i + 1, payload); + } + + // The dispatcher's counter is only final once its loop has exited, which is + // why the server reports it during shutdown rather than on request. + expectProcessedCount(kRequests); +} + +INSTANTIATE_TEST_SUITE_P( + Shapes, UdpTwoProcess, ::testing::Values("ring"), + [](const ::testing::TestParamInfo &info) { + return std::string(info.param); + }); + +} // namespace diff --git a/realtime/unittests/cpu_transport/test_udp_transceiver.cpp b/realtime/unittests/cpu_transport/test_udp_transceiver.cpp index 3e286429a04..835cf66da30 100644 --- a/realtime/unittests/cpu_transport/test_udp_transceiver.cpp +++ b/realtime/unittests/cpu_transport/test_udp_transceiver.cpp @@ -11,56 +11,29 @@ // sockets: no CUDA, no ibverbs, so these tests run anywhere. #include "cudaq/realtime/cpu_transport/udp_wrapper.h" -#include - +#include "cudaq/realtime/testing/test_utils.h" #include #include #include +#include #include -#include + +using cudaq::realtime::testing::load_flag; +using cudaq::realtime::testing::slot_data; +using cudaq::realtime::testing::store_flag; +using cudaq::realtime::testing::wait_for_flag; +using cudaq::realtime::testing::wait_for_flag_clear; namespace { constexpr std::size_t kPageSize = 256; constexpr unsigned kNumPages = 4; -std::uint64_t loadFlag(std::uint64_t flagsAddr, unsigned slot) { - const auto *flags = reinterpret_cast(flagsAddr); - return __atomic_load_n(&flags[slot], __ATOMIC_ACQUIRE); -} - -void storeFlag(std::uint64_t flagsAddr, unsigned slot, std::uint64_t value) { - auto *flags = reinterpret_cast(flagsAddr); - __atomic_store_n(&flags[slot], value, __ATOMIC_RELEASE); -} - +// Every ring in this file has the same stride, so default it; the oversize test +// passes its own to build a datagram the receiver has to reject. std::uint8_t *slotData(std::uint64_t dataAddr, unsigned slot, - std::size_t pageSize = kPageSize) { - return reinterpret_cast(dataAddr) + slot * pageSize; -} - -bool waitForFlag( - std::uint64_t flagsAddr, unsigned slot, - std::chrono::milliseconds timeout = std::chrono::milliseconds(5000)) { - const auto deadline = std::chrono::steady_clock::now() + timeout; - while (std::chrono::steady_clock::now() < deadline) { - if (loadFlag(flagsAddr, slot) != 0) - return true; - std::this_thread::sleep_for(std::chrono::microseconds(100)); - } - return false; -} - -bool waitForFlagClear( - std::uint64_t flagsAddr, unsigned slot, - std::chrono::milliseconds timeout = std::chrono::milliseconds(5000)) { - const auto deadline = std::chrono::steady_clock::now() + timeout; - while (std::chrono::steady_clock::now() < deadline) { - if (loadFlag(flagsAddr, slot) == 0) - return true; - std::this_thread::sleep_for(std::chrono::microseconds(100)); - } - return false; + std::size_t stride = kPageSize) { + return slot_data(dataAddr, slot, stride); } // A bound (service) and connected (caller) transceiver pair over loopback, @@ -102,7 +75,7 @@ class UdpTransceiverPairTest : public ::testing::Test { std::uint8_t *tx = slotData(callerTxData, slot); std::memset(tx, 0, kPageSize); std::memcpy(tx, payload.data(), payload.size()); - storeFlag(callerTxFlags, slot, reinterpret_cast(tx)); + store_flag(callerTxFlags, slot, reinterpret_cast(tx)); } // Consume the service's RX slot: return its payload and recycle the slot @@ -110,7 +83,7 @@ class UdpTransceiverPairTest : public ::testing::Test { std::string consumeAtService(unsigned slot) { std::string payload( reinterpret_cast(slotData(serviceRxData, slot))); - storeFlag(serviceRxFlags, slot, 0); + store_flag(serviceRxFlags, slot, 0); return payload; } @@ -173,11 +146,11 @@ TEST(UdpTransceiverLifecycle, DeliversAcrossAnyInterfaceBind) { std::uint8_t *tx = slotData(txData, 0); std::memset(tx, 0, kPageSize); std::memcpy(tx, "any-if", 6); - storeFlag(txFlags, 0, reinterpret_cast(tx)); + store_flag(txFlags, 0, reinterpret_cast(tx)); const std::uint64_t rxFlags = cpu_udp_get_rx_ring_flag_addr(service); const std::uint64_t rxData = cpu_udp_get_rx_ring_data_addr(service); - ASSERT_TRUE(waitForFlag(rxFlags, 0)); + ASSERT_TRUE(wait_for_flag(rxFlags, 0)); EXPECT_EQ(0, std::memcmp(slotData(rxData, 0), "any-if", 6)); cpu_udp_destroy_transceiver(caller); @@ -203,18 +176,18 @@ TEST(UdpTransceiverLifecycle, StartRequiresSocketAndCloseIsIdempotent) { TEST_F(UdpTransceiverPairTest, DeliversPublishedSlotToServiceRxRing) { publishFromCaller(0, "request-0"); - ASSERT_TRUE(waitForFlag(serviceRxFlags, 0)); + ASSERT_TRUE(wait_for_flag(serviceRxFlags, 0)); // The RX flag carries the slot's data address, same contract as RoCE. EXPECT_EQ(reinterpret_cast(slotData(serviceRxData, 0)), - loadFlag(serviceRxFlags, 0)); + load_flag(serviceRxFlags, 0)); EXPECT_EQ("request-0", consumeAtService(0)); // The caller's TX pump recycles the published slot. - EXPECT_TRUE(waitForFlagClear(callerTxFlags, 0)); + EXPECT_TRUE(wait_for_flag_clear(callerTxFlags, 0)); } TEST_F(UdpTransceiverPairTest, RoundTripsResponseToCallerRxRing) { publishFromCaller(0, "ping"); - ASSERT_TRUE(waitForFlag(serviceRxFlags, 0)); + ASSERT_TRUE(wait_for_flag(serviceRxFlags, 0)); EXPECT_EQ("ping", consumeAtService(0)); // Service answers through its own TX ring; responses go to the source of @@ -222,11 +195,11 @@ TEST_F(UdpTransceiverPairTest, RoundTripsResponseToCallerRxRing) { std::uint8_t *tx = slotData(serviceTxData, 0); std::memset(tx, 0, kPageSize); std::memcpy(tx, "pong", 4); - storeFlag(serviceTxFlags, 0, reinterpret_cast(tx)); + store_flag(serviceTxFlags, 0, reinterpret_cast(tx)); - ASSERT_TRUE(waitForFlag(callerRxFlags, 0)); + ASSERT_TRUE(wait_for_flag(callerRxFlags, 0)); EXPECT_EQ(0, std::memcmp(slotData(callerRxData, 0), "pong", 4)); - storeFlag(callerRxFlags, 0, 0); + store_flag(callerRxFlags, 0, 0); } TEST_F(UdpTransceiverPairTest, FillsRxSlotsInStrictRingOrder) { @@ -234,7 +207,7 @@ TEST_F(UdpTransceiverPairTest, FillsRxSlotsInStrictRingOrder) { const unsigned slot = i % kNumPages; const std::string payload = "msg-" + std::to_string(i); publishFromCaller(slot, payload); - ASSERT_TRUE(waitForFlag(serviceRxFlags, slot)) << "message " << i; + ASSERT_TRUE(wait_for_flag(serviceRxFlags, slot)) << "message " << i; EXPECT_EQ(payload, consumeAtService(slot)) << "message " << i; } } @@ -248,9 +221,9 @@ TEST_F(UdpTransceiverPairTest, ShipsWrappingPublishBurstInFifoOrder) { // Advance the caller's TX cursor to the last slot. for (unsigned slot = 0; slot < kNumPages - 1; ++slot) { publishFromCaller(slot, "warmup-" + std::to_string(slot)); - ASSERT_TRUE(waitForFlag(serviceRxFlags, slot)); + ASSERT_TRUE(wait_for_flag(serviceRxFlags, slot)); consumeAtService(slot); - ASSERT_TRUE(waitForFlagClear(callerTxFlags, slot)); + ASSERT_TRUE(wait_for_flag_clear(callerTxFlags, slot)); } // Publish the wrapped slot 0 FIRST while the TX cursor still waits on slot @@ -261,9 +234,9 @@ TEST_F(UdpTransceiverPairTest, ShipsWrappingPublishBurstInFifoOrder) { // The service's RX ring assigns slots in arrival order, so the payload // arriving first lands in the in-order RX slot kNumPages-1. - ASSERT_TRUE(waitForFlag(serviceRxFlags, kNumPages - 1)); + ASSERT_TRUE(wait_for_flag(serviceRxFlags, kNumPages - 1)); EXPECT_EQ("first", consumeAtService(kNumPages - 1)); - ASSERT_TRUE(waitForFlag(serviceRxFlags, 0)); + ASSERT_TRUE(wait_for_flag(serviceRxFlags, 0)); EXPECT_EQ("second", consumeAtService(0)); } @@ -282,12 +255,13 @@ TEST_F(UdpTransceiverPairTest, DropsDatagramsLargerThanOwnStride) { const std::uint64_t txData = cpu_udp_get_tx_ring_data_addr(bigCaller); std::uint8_t *tx = slotData(txData, 0, 2 * kPageSize); std::memset(tx, 0xAB, 2 * kPageSize); - storeFlag(txFlags, 0, reinterpret_cast(tx)); + store_flag(txFlags, 0, reinterpret_cast(tx)); // The oversized datagram was shipped (TX flag recycled) ... - ASSERT_TRUE(waitForFlagClear(txFlags, 0)); + ASSERT_TRUE(wait_for_flag_clear(txFlags, 0)); // ... but never lands in the service's RX ring. - EXPECT_FALSE(waitForFlag(serviceRxFlags, 0, std::chrono::milliseconds(250))); + EXPECT_FALSE( + wait_for_flag(serviceRxFlags, 0, std::chrono::milliseconds(250))); cpu_udp_destroy_transceiver(bigCaller); } diff --git a/realtime/unittests/test_server/CMakeLists.txt b/realtime/unittests/test_server/CMakeLists.txt new file mode 100644 index 00000000000..0bf80365206 --- /dev/null +++ b/realtime/unittests/test_server/CMakeLists.txt @@ -0,0 +1,52 @@ +# ============================================================================ # +# Copyright (c) 2026 NVIDIA Corporation & Affiliates. # +# All rights reserved. # +# # +# This source code and the accompanying materials are made available under # +# the terms of the Apache License 2.0 which accompanies this distribution. # +# ============================================================================ # + +# Transport-agnostic two-process test server +# ============================================================================== +# Service half of the realtime two-process tests: a HOST/HOST_CALL dispatcher +# serving rpc_increment over whichever bridge provider --transport= names. Not +# tied to any one transport, hence its own directory rather than living under +# bridge_interface//. + +# The increment handler + function-table initialiser is shared with +# hsb_bridge_cpu, where it is compiled BELOW that directory's ibverbs return() +# guard. Referencing the source directly keeps this target independent of +# libibverbs being present. +add_executable(cudaq-realtime-test-server + cudaq_realtime_test_server.cpp + ${CMAKE_CURRENT_SOURCE_DIR}/../cpu_transport/init_rpc_increment_function_table_host.cpp +) + +# CUDA headers only (no CUDA link): cudaq_realtime.h transitively includes +# for the cudaGraphExec_t typedef in the function-entry union. +target_include_directories(cudaq-realtime-test-server PRIVATE + ${CUDAQ_REALTIME_INCLUDE_DIR} + ${CUDAToolkit_INCLUDE_DIRS} +) + +# Where to look for provider .so's named by --transport=, so the tests +# resolve them out of the build tree without an install. Both shipped +# providers set LIBRARY_OUTPUT_DIRECTORY to this directory. +target_compile_definitions(cudaq-realtime-test-server PRIVATE + CUDAQ_REALTIME_BRIDGE_PROVIDER_DIR="${CMAKE_BINARY_DIR}/lib" +) + +target_link_libraries(cudaq-realtime-test-server PRIVATE + cudaq-realtime + Threads::Threads +) + +# Providers are dlopen'd, not linked, so express the ordering explicitly for +# whichever ones this configure builds. +foreach(provider cudaq-realtime-bridge-udp cudaq-realtime-bridge-cpu-roce) + if (TARGET ${provider}) + add_dependencies(cudaq-realtime-test-server ${provider}) + endif() +endforeach() + +message(STATUS " - cudaq-realtime-test-server (two-process dispatch server)") diff --git a/realtime/unittests/test_server/cudaq_realtime_test_server.cpp b/realtime/unittests/test_server/cudaq_realtime_test_server.cpp new file mode 100644 index 00000000000..4ed3546f4d5 --- /dev/null +++ b/realtime/unittests/test_server/cudaq_realtime_test_server.cpp @@ -0,0 +1,387 @@ +/******************************************************************************* + * Copyright (c) 2026 NVIDIA Corporation & Affiliates. * + * All rights reserved. * + * * + * This source code and the accompanying materials are made available under * + * the terms of the Apache License 2.0 which accompanies this distribution. * + ******************************************************************************/ + +/// @file cudaq_realtime_test_server.cpp +/// @brief Service half of the realtime two-process tests: a dispatcher serving +/// the `rpc_increment` HOST_CALL handler over any bridge provider. +/// +/// Transport-agnostic by construction. Every transport is reached through the +/// same bridge vtable (bridge_interface.h), and each provider owns its own +/// bring-up -- udp binds a socket in create(), cpu_roce does the full TCP +/// rendezvous plus QP/rkey swap across create()/connect() -- so there is no +/// per-transport code here. `--transport=` names the provider library and +/// every unrecognized argument is forwarded to it verbatim, which is how +/// `--port=`, `--device=`, `--local-ip=`, `--qp_config=`, `--peer-ip=`, +/// `--remote-qp=`, `--num-slots=` and `--slot-size=` reach the provider that +/// understands them. Providers ignore arguments they do not recognize, by +/// contract, so forwarding the whole command line is safe. +/// +/// ORDERING RULE: the READY line is printed BEFORE cudaq_bridge_connect(). +/// A rendezvous transport blocks in connect() until the caller dials in, while +/// the caller waits for READY before dialing -- announcing after connecting +/// would deadlock the pair. Providers are built for this order: endpoint info +/// is valid as soon as create() returns. +/// +/// Two dispatch shapes are wired through the three functions below, so a shape +/// is one command-line token rather than a second binary: +/// --dispatch=ring dispatcher polls the provider's ring buffer while the +/// provider's own pump threads move bytes to the wire. +/// --dispatch=unified dispatcher drives the transport itself through the +/// provider's rx_poll/tx_publish hooks and the provider +/// starts no threads. Fails cleanly with UNSUPPORTED +/// against a provider whose get_cpu_dataplane is NULL. +/// +/// Handshake, both parsed by cudaq::realtime::testing::ServerProcess: +/// CUDAQ_REALTIME_SERVER_READY dispatch= +/// slots= slot_size= (one line; wrapped here) +/// CUDAQ_REALTIME_SERVER_PROCESSED count= (printed after shutdown) +/// +/// The processed count is only final once the dispatch loop has exited, so it +/// is printed on the way out rather than served on request. + +#include "cudaq/realtime/daemon/bridge/bridge_interface.h" +#include "cudaq/realtime/daemon/dispatcher/cudaq_realtime.h" +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +// Provided by init_rpc_increment_function_table_host.cpp. +extern "C" void +setup_rpc_increment_function_table_host(cudaq_function_entry_t *h_entries); + +namespace { + +enum class Shape { Ring, Unified }; + +const char *shape_name(Shape shape) { + return shape == Shape::Unified ? "unified" : "ring"; +} + +struct ServerConfig { + std::string transport = "udp"; + Shape shape = Shape::Ring; + int timeout_sec = 60; +}; + +std::atomic g_shutdown{0}; +void on_signal(int) { g_shutdown.store(1, std::memory_order_release); } + +bool starts_with(const std::string &s, const char *prefix) { + const std::size_t n = std::strlen(prefix); + return s.size() >= n && std::memcmp(s.data(), prefix, n) == 0; +} + +void print_usage(const char *program) { + std::cout + << "Usage: " << program << " [options] [provider options]\n\n" + << "Serves the rpc_increment HOST_CALL handler over any bridge " + "provider.\n\n" + << "Options:\n" + << " --transport=NAME provider to load: a bare name resolved to\n" + << " libcudaq-realtime-bridge-.so (udp,\n" + << " cpu_roce, ...) or a path to a provider .so\n" + << " [default udp]\n" + << " --dispatch=SHAPE ring | unified [default ring]\n" + << " --timeout=N run timeout in seconds [default 60]\n\n" + << "All other options are forwarded to the provider, e.g. --port=N,\n" + << "--num-slots=N, --slot-size=N, --device=NAME, --local-ip=ADDR,\n" + << "--qp_config=MODE, --peer-ip=ADDR, --remote-qp=N.\n"; +} + +// Ring geometry is queried from the provider rather than parsed here, so the +// dispatcher and the transport cannot disagree about slot count or stride. +bool parse_args(int argc, char **argv, ServerConfig &cfg, bool &help) { + for (int i = 1; i < argc; ++i) { + const std::string a = argv[i] ? argv[i] : ""; + if (a == "--help" || a == "-h") { + help = true; + return false; + } + try { + if (starts_with(a, "--transport=")) + cfg.transport = a.substr(12); + else if (starts_with(a, "--dispatch=")) { + const std::string shape = a.substr(11); + if (shape == "ring") + cfg.shape = Shape::Ring; + else if (shape == "unified") + cfg.shape = Shape::Unified; + else { + std::cerr << "ERROR: unknown --dispatch=" << shape + << " (expected ring or unified)" << std::endl; + return false; + } + } else if (starts_with(a, "--timeout=")) + cfg.timeout_sec = std::stoi(a.substr(10)); + // Everything else belongs to the provider; forwarded untouched. + } catch (const std::exception &) { + std::cerr << "ERROR: bad numeric value in '" << a << "'" << std::endl; + return false; + } + } + return true; +} + +// Resolve a --transport value to the provider library to dlopen: anything with +// a '/' is a caller-supplied path, a bare name maps to the shipped soname. +// Provider names are snake_case (udp, cpu_roce) while the sonames are +// hyphenated, so '_' maps to '-'; the literal spelling is probed too so an +// out-of-tree provider whose soname really contains an underscore resolves. +// The build directory is probed first so the tests run without an install. +std::string resolve_provider_lib(const std::string &transport) { + if (transport.find('/') != std::string::npos) + return transport; + std::string hyphenated = transport; + std::replace(hyphenated.begin(), hyphenated.end(), '_', '-'); + const std::string soname = "libcudaq-realtime-bridge-" + hyphenated + ".so"; +#ifdef CUDAQ_REALTIME_BRIDGE_PROVIDER_DIR + const std::string literal = "libcudaq-realtime-bridge-" + transport + ".so"; + for (const auto &name : {soname, literal}) { + const std::string candidate = + std::string(CUDAQ_REALTIME_BRIDGE_PROVIDER_DIR) + "/" + name; + if (std::ifstream(candidate).good()) + return candidate; + } +#endif + return soname; // fall back to the dynamic loader's search path +} + +//===----------------------------------------------------------------------===// +// The shape seam: the only three shape-aware functions in this file. +//===----------------------------------------------------------------------===// + +// Transport context the dispatcher is wired to. The unified data-plane is +// held by pointer inside the dispatcher, so this must outlive it (declared +// before the dispatcher in main, destroyed after). +struct ShapeState { + cudaq_ringbuffer_t ring{}; + cudaq_cpu_dataplane_t dataplane{}; +}; + +// (1) Read at cudaq_dispatcher_create time. +void configure_shape(Shape shape, cudaq_dispatcher_config_t &dcfg) { + if (shape == Shape::Unified) { + // The unified loop keeps the per-slot TX markers (its publish hook reads + // them to tell a running graph from a finished one), so skip_tx_markers + // stays 0; the dispatcher overrides it regardless. + dcfg.kernel_type = CUDAQ_KERNEL_UNIFIED; + return; + } + dcfg.kernel_type = CUDAQ_KERNEL_REGULAR; + // The provider's TX pump owns the wire and treats any non-zero flag as a + // slot address, so the in-flight sentinel must not be written. + dcfg.skip_tx_markers = 1; +} + +// (2) Called after create, before start. +cudaq_status_t wire_shape(Shape shape, cudaq_realtime_bridge_handle_t bridge, + cudaq_dispatcher_t *dispatcher, ShapeState &state) { + if (shape == Shape::Unified) { + const cudaq_status_t status = + cudaq_bridge_get_cpu_dataplane(bridge, &state.dataplane); + if (status != CUDAQ_OK) + return status; + return cudaq_dispatcher_set_cpu_dataplane(dispatcher, &state.dataplane); + } + const cudaq_status_t status = + cudaq_bridge_get_transport_context(bridge, RING_BUFFER, &state.ring); + if (status != CUDAQ_OK) + return status; + return cudaq_dispatcher_set_ringbuffer(dispatcher, &state.ring); +} + +// (3) Unified drives the transport from the dispatcher's own thread; starting +// the provider's pump threads as well would race its hooks for the rings. +bool needs_bridge_launch(Shape shape) { return shape != Shape::Unified; } + +//===----------------------------------------------------------------------===// + +// Reverse-order teardown, so any early `return 1` releases the transport too. +// The dispatcher goes first: its loop reads the provider's rings (and under +// the unified shape calls straight into the provider), so it must be stopped +// before the bridge is disconnected. Every call is idempotent, which lets the +// normal shutdown path stop the dispatcher, read its final count, and then +// leave the rest to this destructor. +struct ServerResources { + cudaq_realtime_bridge_handle_t bridge = nullptr; + cudaq_dispatch_manager_t *manager = nullptr; + cudaq_dispatcher_t *dispatcher = nullptr; + + ~ServerResources() { + if (dispatcher) { + cudaq_dispatcher_stop(dispatcher); + cudaq_dispatcher_destroy(dispatcher); + } + if (manager) + cudaq_dispatch_manager_destroy(manager); + if (bridge) { + cudaq_bridge_disconnect(bridge); + cudaq_bridge_destroy(bridge); + } + } +}; + +} // namespace + +int main(int argc, char **argv) { + ServerConfig cfg; + bool help = false; + if (!parse_args(argc, argv, cfg, help)) { + if (help) { + print_usage(argv[0]); + return 0; + } + return 1; + } + + std::signal(SIGINT, on_signal); + std::signal(SIGTERM, on_signal); + + // Declared before the resources so it outlives the dispatcher that points + // into it (see ShapeState). + ShapeState state; + ServerResources res; + + // [1] Load the provider and let it parse our whole command line. + const std::string library = resolve_provider_lib(cfg.transport); + if (cudaq_bridge_create_from_library(&res.bridge, library.c_str(), argc, + argv) != CUDAQ_OK) { + std::cerr << "ERROR: cannot create a bridge from '" << library + << "' (--transport=" << cfg.transport << ")" << std::endl; + return 1; + } + + // [2] Take the dispatcher's ring geometry and the endpoint to publish from + // the provider. Both are interface-version-2 queries; a provider + // without them cannot be served or advertised. + std::uint32_t num_slots = 0, slot_size = 0; + if (cudaq_bridge_get_ring_geometry(res.bridge, &num_slots, &slot_size) != + CUDAQ_OK || + num_slots == 0 || slot_size == 0) { + std::cerr << "ERROR: provider '" << library + << "' did not report a usable ring geometry" << std::endl; + return 1; + } + char endpoint[512] = {0}; + if (cudaq_bridge_get_endpoint_info(res.bridge, endpoint, sizeof(endpoint)) != + CUDAQ_OK) { + std::cerr << "ERROR: provider '" << library + << "' does not report endpoint info; the caller would have " + "nothing to dial" + << std::endl; + return 1; + } + + // [3] Dispatcher: HOST path, HOST_CALL mode, geometry from the transport. + if (cudaq_dispatch_manager_create(&res.manager) != CUDAQ_OK) { + std::cerr << "ERROR: cudaq_dispatch_manager_create failed" << std::endl; + return 1; + } + cudaq_dispatcher_config_t dcfg{}; + dcfg.dispatch_path = CUDAQ_DISPATCH_PATH_HOST; + dcfg.dispatch_mode = CUDAQ_DISPATCH_HOST_CALL; + dcfg.num_slots = num_slots; + dcfg.slot_size = slot_size; + configure_shape(cfg.shape, dcfg); + if (cudaq_dispatcher_create(res.manager, &dcfg, &res.dispatcher) != + CUDAQ_OK) { + std::cerr << "ERROR: cudaq_dispatcher_create failed" << std::endl; + return 1; + } + + // [4] Wire the transport to the dispatcher (shape-dependent). + const cudaq_status_t wired = + wire_shape(cfg.shape, res.bridge, res.dispatcher, state); + if (wired != CUDAQ_OK) { + std::cerr << "ERROR: cannot wire --dispatch=" << shape_name(cfg.shape) + << " to provider '" << library << "': " + << (wired == CUDAQ_ERR_UNSUPPORTED + ? "the provider does not support this shape" + : "wiring failed") + << std::endl; + return 1; + } + + // [5] Function table: the single host-side increment handler. + cudaq_function_entry_t entries[1]; + setup_rpc_increment_function_table_host(entries); + cudaq_function_table_t table{}; + table.entries = entries; + table.count = 1; + if (cudaq_dispatcher_set_function_table(res.dispatcher, &table) != CUDAQ_OK) { + std::cerr << "ERROR: cudaq_dispatcher_set_function_table failed" + << std::endl; + return 1; + } + + // [6] Control variables, then start polling. A HOST_CALL-only table needs + // no graph engine and therefore no GPU. + volatile int shutdown_flag = 0; + std::uint64_t stats = 0; + if (cudaq_dispatcher_set_control(res.dispatcher, &shutdown_flag, &stats) != + CUDAQ_OK) { + std::cerr << "ERROR: cudaq_dispatcher_set_control failed" << std::endl; + return 1; + } + if (cudaq_dispatcher_start(res.dispatcher) != CUDAQ_OK) { + std::cerr << "ERROR: cudaq_dispatcher_start failed" << std::endl; + return 1; + } + + // [7] Announce BEFORE connecting: see the ORDERING RULE at the top. The + // endpoint string is the provider's own key=value description, passed + // through so the harness needs no per-transport parsing. std::endl + // flushes, which matters because stdout is a pipe here. + std::cout << "CUDAQ_REALTIME_SERVER_READY " << endpoint + << " dispatch=" << shape_name(cfg.shape) << " slots=" << num_slots + << " slot_size=" << slot_size << std::endl; + + // [8] Connect: a no-op for connectionless transports, a blocking wait for + // the caller on a rendezvous transport. + if (cudaq_bridge_connect(res.bridge) != CUDAQ_OK) { + std::cerr << "ERROR: cudaq_bridge_connect failed" << std::endl; + return 1; + } + + // [9] Start the provider's pump threads last, once the dispatcher is + // already polling. + if (needs_bridge_launch(cfg.shape) && + cudaq_bridge_launch(res.bridge) != CUDAQ_OK) { + std::cerr << "ERROR: cudaq_bridge_launch failed" << std::endl; + return 1; + } + + // [10] Serve until SIGTERM/SIGINT or the timeout backstop. + const auto started = std::chrono::steady_clock::now(); + while (g_shutdown.load(std::memory_order_acquire) == 0) { + const auto elapsed = std::chrono::duration_cast( + std::chrono::steady_clock::now() - started) + .count(); + if (elapsed >= cfg.timeout_sec) { + std::cerr << "timeout reached (" << cfg.timeout_sec << "s)" << std::endl; + break; + } + std::this_thread::sleep_for(std::chrono::milliseconds(50)); + } + + // [11] Stop the dispatch loop, then report the count it accumulated (only + // final once the loop has exited). ServerResources unwinds the rest. + cudaq_dispatcher_stop(res.dispatcher); + std::uint64_t processed = 0; + cudaq_dispatcher_get_processed(res.dispatcher, &processed); + std::cout << "CUDAQ_REALTIME_SERVER_PROCESSED count=" << processed + << std::endl; + return 0; +}