From 7fe9529b906a496f0a1e73c3af777e6ae1016a0a Mon Sep 17 00:00:00 2001 From: kvmto Date: Thu, 9 Jul 2026 17:25:46 +0200 Subject: [PATCH 01/10] feat(qec): add cuda_device_id placement knob for GPU decoders Decoders can now be pinned to a specific CUDA device at construction via a cuda_device_id parameter, settable through C++/Python kwargs and the YAML realtime config (top-level decoder field, next to type/ transport). Model: one thread owns one decoder. decoder::get() validates the id (negative or >= device count throws), persistently pins the constructing thread with cudaSetDevice (no restore), strips the key before the plugin constructor, and stores it (get_cuda_device_id()). Plugin constructors therefore allocate on the right device with zero plugin changes -- this covers trt_decoder and the closed-source nv-qldpc-decoder transparently. decode_async() is the one exception: its fresh std::async worker pins itself for the call's duration via a lib-private RAII CudaDeviceGuard (libs/qec/lib/hardware_guards.h). The realtime host dispatcher (one thread serving all decoders) applies each decoder's device with set-if-different before enqueue and before DEVICE-mode graph capture; no-op for unpinned decoders. Tests: 7 C++ unit tests incl. 2-GPU placement and async-worker pinning, YAML round-trip + prepare_decoder_params coverage, Python kwargs tests. NUMA/mempolicy/cpu_affinity and thread-binding APIs are deferred to a follow-up PR per review feedback on #634. Signed-off-by: kvmto --- .../examples_rst/qec/realtime_decoding.rst | 9 + libs/qec/include/cudaq/qec/decoder.h | 10 + .../cudaq/qec/realtime/decoding_config.h | 5 + libs/qec/lib/CMakeLists.txt | 4 + libs/qec/lib/decoder.cpp | 52 ++++- libs/qec/lib/hardware_guards.h | 57 +++++ libs/qec/lib/realtime/config.cpp | 1 + .../qec/lib/realtime/qec_realtime_session.cpp | 23 ++ libs/qec/lib/realtime/realtime_decoding.cpp | 4 + .../python/bindings/py_decoding_config.cpp | 1 + libs/qec/python/tests/test_decoder.py | 26 +++ libs/qec/unittests/CMakeLists.txt | 2 +- libs/qec/unittests/test_decoders.cpp | 213 ++++++++++++++++++ libs/qec/unittests/test_decoders_yaml.cpp | 33 +++ 14 files changed, 436 insertions(+), 4 deletions(-) create mode 100644 libs/qec/lib/hardware_guards.h diff --git a/docs/sphinx/examples_rst/qec/realtime_decoding.rst b/docs/sphinx/examples_rst/qec/realtime_decoding.rst index 814c9c8f2..2f5cff664 100644 --- a/docs/sphinx/examples_rst/qec/realtime_decoding.rst +++ b/docs/sphinx/examples_rst/qec/realtime_decoding.rst @@ -155,6 +155,7 @@ arguments: decoders: - id: 0 type: pymatching + cuda_device_id: 0 # optional: pin this decoder to a CUDA device block_size: 3 syndrome_size: 3 H_sparse: [ 0, -1, 1, -1, 2, -1 ] @@ -164,6 +165,14 @@ arguments: error_rate_vec: [ 0.1, 0.1, 0.1 ] merge_strategy: smallest_weight +``cuda_device_id`` pins a GPU-accelerated decoder (e.g. ``nv-qldpc-decoder`` +or ``trt_decoder``) to a specific CUDA device. The same knob is available as +a construction parameter in C++ and Python +(``qec.get_decoder("trt_decoder", H, cuda_device_id=1)``). The thread that +creates a decoder is pinned to that device and is expected to drive its +decode calls; create each pinned decoder on its own thread to place several +decoders on different GPUs. + Here is how to create and save a decoder configuration: .. tab:: Python diff --git a/libs/qec/include/cudaq/qec/decoder.h b/libs/qec/include/cudaq/qec/decoder.h index 0a2f4c0c0..7c1d2e351 100644 --- a/libs/qec/include/cudaq/qec/decoder.h +++ b/libs/qec/include/cudaq/qec/decoder.h @@ -260,6 +260,12 @@ class decoder /// depends on D_sparse, so you must have called set_D_sparse() first. uint32_t get_num_msyn_per_decode() const; + /// @brief The CUDA device this decoder was pinned to at construction via + /// the "cuda_device_id" parameter, or -1 when no pin was requested. + /// Construction pins the constructing thread persistently (the thread that + /// creates a decoder is the thread expected to drive its decode calls). + int get_cuda_device_id() const { return cuda_device_id_; } + /// @brief Set the observable matrix. void set_O_sparse(const std::vector> &O_sparse); @@ -357,6 +363,10 @@ class decoder /// @brief The decoder's D matrix in sparse format std::vector> D_sparse; + /// @brief CUDA device id consumed from the construction parameters by + /// decoder::get(); -1 = unpinned. See get_cuda_device_id(). + int cuda_device_id_ = -1; + private: decode_result_type result_type_ = decode_result_type::decode_to_errs; }; diff --git a/libs/qec/include/cudaq/qec/realtime/decoding_config.h b/libs/qec/include/cudaq/qec/realtime/decoding_config.h index 7d4a3f2d2..3608ec43e 100644 --- a/libs/qec/include/cudaq/qec/realtime/decoding_config.h +++ b/libs/qec/include/cudaq/qec/realtime/decoding_config.h @@ -183,6 +183,11 @@ struct decoder_config { /// Defaults to cpu_roce. Set to gpu_roce for decoders where syndrome bits /// are DMA'd directly to GPU VRAM (e.g. nv_qldpc_decoder with RelayBP). DecoderTransport transport = DecoderTransport::cpu_roce; + /// CUDA device this decoder is pinned to at construction (see the + /// "cuda_device_id" decoder parameter). Placement knob common to any + /// GPU-accelerated decoder, hence at this level rather than inside the + /// per-decoder custom args. Unset = unpinned. + std::optional cuda_device_id; uint64_t block_size = 0; uint64_t syndrome_size = 0; std::vector H_sparse; diff --git a/libs/qec/lib/CMakeLists.txt b/libs/qec/lib/CMakeLists.txt index f9a52e1d6..0a8fad91e 100644 --- a/libs/qec/lib/CMakeLists.txt +++ b/libs/qec/lib/CMakeLists.txt @@ -150,6 +150,10 @@ target_link_libraries(${DECODERS_LIBRARY_NAME} PUBLIC $ fmt::fmt-header-only + PRIVATE + # decoder::get() consumes the "cuda_device_id" construction parameter + # (validation via cudaGetDeviceCount + persistent cudaSetDevice pin). + CUDA::cudart ) target_link_libraries(${LIBRARY_NAME} diff --git a/libs/qec/lib/decoder.cpp b/libs/qec/lib/decoder.cpp index 796427a54..eef1def36 100644 --- a/libs/qec/lib/decoder.cpp +++ b/libs/qec/lib/decoder.cpp @@ -8,10 +8,12 @@ #include "cudaq/qec/decoder.h" #include "cuda-qx/core/library_utils.h" +#include "hardware_guards.h" #include "cudaq/qec/logger.h" #include "cudaq/qec/plugin_loader.h" #include "cudaq/qec/version.h" #include +#include #include #include #include @@ -123,8 +125,33 @@ std::string decoder::get_version() const { std::future decoder::decode_async(const std::vector &syndrome) { - return std::async(std::launch::async, - [this, syndrome] { return this->decode(syndrome); }); + // Captured by value: the worker must not dereference decoder members to + // find its device. The std::async thread is brand-new and unpinned, so it + // guards itself for the duration of the call (the one exception to the + // one-thread-owns-one-decoder persistent pin). + const int cuda_id = cuda_device_id_; + return std::async(std::launch::async, [this, syndrome, cuda_id] { + cudaq::qec::detail_affinity::CudaDeviceGuard dev(cuda_id); + return this->decode(syndrome); + }); +} + +/// Reads "cuda_device_id" from the construction parameters. Absent -> -1. +/// Negative or >= cudaGetDeviceCount() -> std::runtime_error (fail fast: +/// never silently decode on the wrong GPU). +static int read_cuda_device_id(const cudaqx::heterogeneous_map ¶ms) { + if (!params.contains("cuda_device_id")) + return -1; + const int value = params.get("cuda_device_id"); + if (value < 0) + throw std::runtime_error("cuda_device_id must be >= 0 (got " + + std::to_string(value) + ")"); + int count = 0; + if (cudaGetDeviceCount(&count) != cudaSuccess || value >= count) + throw std::runtime_error("cuda_device_id " + std::to_string(value) + + " is out of range: " + std::to_string(count) + + " CUDA device(s) visible"); + return value; } std::unique_ptr @@ -138,7 +165,26 @@ decoder::get(const std::string &name, const decoder_init &init, "invalid decoder requested: " + name + ". Run with CUDAQ_LOG_LEVEL=info (environment variable) to see " "additional plugin diagnostics at startup."); - return iter->second(init, param_map); + const int cuda_device_id = read_cuda_device_id(param_map); + if (cuda_device_id < 0) + return iter->second(init, param_map); + // Pin the constructing thread persistently (no restore): one thread owns + // one decoder, so every later allocation and kernel launch on this thread + // -- including lazy allocations inside a plugin's decode() -- lands on the + // requested device with no per-call machinery. + cudaError_t err = cudaSetDevice(cuda_device_id); + if (err != cudaSuccess) + throw std::runtime_error("cudaSetDevice(" + std::to_string(cuda_device_id) + + ") failed: " + cudaGetErrorString(err)); + // The key is consumed here; strip it so plugins that strictly validate + // their parameter keys do not reject it. + cudaqx::heterogeneous_map plugin_params; + for (const auto &kv : param_map) + if (kv.first != "cuda_device_id") + plugin_params.insert(kv.first, kv.second); + auto d = iter->second(init, plugin_params); + d->cuda_device_id_ = cuda_device_id; + return d; } namespace details { diff --git a/libs/qec/lib/hardware_guards.h b/libs/qec/lib/hardware_guards.h new file mode 100644 index 000000000..3f01196fb --- /dev/null +++ b/libs/qec/lib/hardware_guards.h @@ -0,0 +1,57 @@ +/******************************************************************************* + * Copyright (c) 2022 - 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 + +#include +#include +#include + +namespace cudaq::qec::detail_affinity { + +/// RAII: set the calling thread's CUDA device, restore the previous device on +/// scope exit. No-op for target < 0. Lib-private and header-only so decoder +/// plugins built as separate .so files can reuse it (PR2 extends this header +/// with NUMA guards; the nv-qldpc follow-up mirrors its use). +/// +/// This guard is for threads that do NOT follow the one-thread-owns-one- +/// decoder persistent pin (e.g. the fresh worker spawned by decode_async). +class CudaDeviceGuard { +public: + explicit CudaDeviceGuard(int target) { + if (target < 0) + return; + int count = 0; + if (cudaGetDeviceCount(&count) != cudaSuccess || target >= count) + throw std::runtime_error("cuda_device_id " + std::to_string(target) + + " is out of range: " + std::to_string(count) + + " CUDA device(s) visible"); + // If the current device is unreadable, skip restoration rather than + // restore to a guessed device; the set below still applies. + if (cudaGetDevice(&prev_) != cudaSuccess) + prev_ = -1; + cudaError_t err = cudaSetDevice(target); + if (err != cudaSuccess) + throw std::runtime_error("CudaDeviceGuard: cudaSetDevice(" + + std::to_string(target) + + ") failed: " + cudaGetErrorString(err)); + restore_ = (prev_ >= 0 && prev_ != target); + } + ~CudaDeviceGuard() { + if (restore_) + (void)cudaSetDevice(prev_); + } + CudaDeviceGuard(const CudaDeviceGuard &) = delete; + CudaDeviceGuard &operator=(const CudaDeviceGuard &) = delete; + +private: + int prev_ = -1; + bool restore_ = false; +}; + +} // namespace cudaq::qec::detail_affinity diff --git a/libs/qec/lib/realtime/config.cpp b/libs/qec/lib/realtime/config.cpp index a568e6673..9b100319f 100644 --- a/libs/qec/lib/realtime/config.cpp +++ b/libs/qec/lib/realtime/config.cpp @@ -717,6 +717,7 @@ struct MappingTraits { io.mapRequired("type", config.type); io.mapOptional("transport", config.transport, cudaq::qec::decoding::config::DecoderTransport::cpu_roce); + io.mapOptional("cuda_device_id", config.cuda_device_id); io.mapRequired("block_size", config.block_size); io.mapRequired("syndrome_size", config.syndrome_size); io.mapRequired("H_sparse", config.H_sparse); diff --git a/libs/qec/lib/realtime/qec_realtime_session.cpp b/libs/qec/lib/realtime/qec_realtime_session.cpp index dbe35c2f5..a0bdfe6c0 100644 --- a/libs/qec/lib/realtime/qec_realtime_session.cpp +++ b/libs/qec/lib/realtime/qec_realtime_session.cpp @@ -19,6 +19,7 @@ #include #include #include +#include #include #include #include @@ -126,6 +127,26 @@ cudaq::qec::decoder *get_decoder_or_throw(std::int64_t decoder_id) { return (*decoders)[static_cast(decoder_id)].get(); } +// Point the calling thread at the decoder's pinned CUDA device before work +// that allocates or launches on it. Set-and-leave (no restore): one +// dispatcher thread serves all decoders, so the thread simply converges to +// the device of the decoder it is currently serving; cudaSetDevice on an +// already-current device is a cheap no-op. +static void apply_decoder_cuda_device(cudaq::qec::decoder *dec) { + if (!dec) + return; + const int id = dec->get_cuda_device_id(); + if (id < 0) + return; + int cur = -1; + if (cudaGetDevice(&cur) == cudaSuccess && cur == id) + return; + cudaError_t err = cudaSetDevice(id); + if (err != cudaSuccess) + CUDA_QEC_WARN("apply_decoder_cuda_device: cudaSetDevice({}) failed: {}", id, + cudaGetErrorString(err)); +} + // Two-ring response writer: the request stays in `rx_slot` (read-only); the // response is written into the distinct `tx_slot`. The preserved header fields // (request_id, ptp_timestamp) must be echoed explicitly from rx to tx. The @@ -178,6 +199,7 @@ void enqueue_syndromes_host(const void *rx_slot, void *tx_slot, } auto *decoder = get_decoder_or_throw(body->decoder_id); + apply_decoder_cuda_device(decoder); // Reject requests larger than this decoder's per-decode window. The slot // is sized for the largest decoder in the session, so an oversized request // for a smaller decoder can still fit the slot; without this guard it would @@ -537,6 +559,7 @@ void qec_realtime_session::capture_decoder_graphs() { "qec_realtime_session::initialize: decoder " + std::to_string(i) + " does not support graph dispatch in DEVICE mode."); + apply_decoder_cuda_device(dec); // reserved_sms = 0 is intentional for the inproc_rpc desktop / CI path. void *raw = dec->capture_decode_graph(/*reserved_sms=*/0); if (!raw) diff --git a/libs/qec/lib/realtime/realtime_decoding.cpp b/libs/qec/lib/realtime/realtime_decoding.cpp index 23c0ad4c0..d423712b3 100644 --- a/libs/qec/lib/realtime/realtime_decoding.cpp +++ b/libs/qec/lib/realtime/realtime_decoding.cpp @@ -156,6 +156,10 @@ namespace cudaq::qec::decoding::host { cudaqx::heterogeneous_map prepare_decoder_params( const cudaq::qec::decoding::config::decoder_config &decoder_config) { auto params = decoder_config.decoder_custom_args_to_heterogeneous_map(); + // Placement knob: surfaced for every decoder type (deliberately before the + // trt-only early return below); consumed by decoder::get() at construction. + if (decoder_config.cuda_device_id.has_value()) + params.insert("cuda_device_id", decoder_config.cuda_device_id.value()); if (decoder_config.type != "trt_decoder") return params; diff --git a/libs/qec/python/bindings/py_decoding_config.cpp b/libs/qec/python/bindings/py_decoding_config.cpp index 283c90af1..98a4f9aa8 100644 --- a/libs/qec/python/bindings/py_decoding_config.cpp +++ b/libs/qec/python/bindings/py_decoding_config.cpp @@ -279,6 +279,7 @@ void bindDecodingConfig(nb::module_ &mod) { .def(nb::init<>()) .def_rw("id", &decoder_config::id) .def_rw("type", &decoder_config::type) + .def_rw("cuda_device_id", &decoder_config::cuda_device_id) .def_rw("block_size", &decoder_config::block_size) .def_rw("syndrome_size", &decoder_config::syndrome_size) .def_rw("H_sparse", &decoder_config::H_sparse) diff --git a/libs/qec/python/tests/test_decoder.py b/libs/qec/python/tests/test_decoder.py index 3201de1c6..8f00900a8 100644 --- a/libs/qec/python/tests/test_decoder.py +++ b/libs/qec/python/tests/test_decoder.py @@ -1002,6 +1002,32 @@ def test_get_decoder_rejects_unknown_decoder_for_stim_dem_text(): qec.get_decoder("__no_such_decoder__", "error(0.1) D0 L0\n") +def test_decoder_cuda_device_id_invalid_raises(): + H = create_test_matrix() + # A negative id is rejected before reaching the C++ guard: the kwargs + # marshalling layer stores Python ints as size_t, so nanobind refuses the + # negative value with a bare RuntimeError ("std::bad_cast"). + with pytest.raises(RuntimeError): + qec.get_decoder("single_error_lut", H, cuda_device_id=-2) + # An out-of-range id flows through kwargs to decoder::get(), which raises + # a runtime_error naming the offending parameter. + with pytest.raises(RuntimeError, match="cuda_device_id"): + qec.get_decoder("single_error_lut", H, cuda_device_id=1 << 20) + + +def test_decoder_cuda_device_id_valid(): + H = create_test_matrix() + try: + d = qec.get_decoder("single_error_lut", H, cuda_device_id=0) + except RuntimeError as e: + if "out of range" in str(e): + pytest.skip("no CUDA device visible") + raise + syndrome = create_test_syndrome() + result = d.decode(syndrome) + assert len(result.result) == H.shape[1] + + def test_get_decoder_user_O_wins_over_dem_derived(): dem_text = ("error(0.1) D0 L0\n" "error(0.1) D1 L0\n" diff --git a/libs/qec/unittests/CMakeLists.txt b/libs/qec/unittests/CMakeLists.txt index 344d05449..211ba17e2 100644 --- a/libs/qec/unittests/CMakeLists.txt +++ b/libs/qec/unittests/CMakeLists.txt @@ -35,7 +35,7 @@ find_package(CUDAToolkit REQUIRED) add_compile_options(-Wno-attributes) add_executable(test_decoders test_decoders.cpp decoders/sample_decoder.cpp) -target_link_libraries(test_decoders PRIVATE GTest::gtest_main cudaq-qec-decoders libstim) +target_link_libraries(test_decoders PRIVATE GTest::gtest_main cudaq-qec-decoders libstim CUDA::cudart) add_dependencies(CUDAQXQECUnitTests test_decoders) gtest_discover_tests(test_decoders) diff --git a/libs/qec/unittests/test_decoders.cpp b/libs/qec/unittests/test_decoders.cpp index 0f2f4c8d5..e254979a4 100644 --- a/libs/qec/unittests/test_decoders.cpp +++ b/libs/qec/unittests/test_decoders.cpp @@ -10,12 +10,15 @@ #include "cudaq/qec/decoder.h" #include "cudaq/qec/detector_error_model.h" #include "cudaq/qec/pcm_utils.h" +#include #include #include +#include #include #include #include #include +#include namespace { class ScopedEnv { @@ -1190,3 +1193,213 @@ TEST(SlidingWindowDecoder, BaseStreamingCopiesFirstRoundDetectors) { << "First-round detector copy runs, but the sliding window is not full " "yet so no final correction is committed."; } + +namespace { + +int cuda_device_count() { + int count = 0; + if (cudaGetDeviceCount(&count) != cudaSuccess) + return 0; + return count; +} + +/// Restores the caller's CUDA device on scope exit so the persistent pin +/// made by one test does not leak into the next (gtest shares the process). +class ScopedDeviceRestore { +public: + ScopedDeviceRestore() { + if (cudaGetDevice(&prev_) != cudaSuccess) + prev_ = -1; + } + ~ScopedDeviceRestore() { + if (prev_ >= 0) + (void)cudaSetDevice(prev_); + } + +private: + int prev_ = -1; +}; + +/// A decoder that rejects any construction parameter it does not know, +/// proving decoder::get() strips cuda_device_id before the plugin ctor. +class strict_keys_decoder : public cudaq::qec::decoder { +public: + strict_keys_decoder(const cudaq::qec::sparse_binary_matrix &H, + const cudaqx::heterogeneous_map ¶ms) + : decoder(H) { + auto invalid = + cudaq::qec::validate_config_parameters(params, {"decode_to_obs"}); + if (!invalid.empty()) + throw std::runtime_error("strict_keys_decoder: unexpected key " + + invalid.front()); + } + cudaq::qec::decoder_result + decode(const std::vector &syndrome) override { + cudaq::qec::decoder_result r; + r.converged = true; + r.result = std::vector(block_size, 0.0); + return r; + } + CUDAQ_EXTENSION_CUSTOM_CREATOR_FUNCTION( + strict_keys_decoder, static std::unique_ptr create( + const cudaq::qec::decoder_init &init, + const cudaqx::heterogeneous_map ¶ms) { + return cudaq::qec::make_pcm_decoder(init, params); + }) +}; +CUDAQ_EXT_PT_REGISTER_TYPE(strict_keys_decoder) + +cudaq::qec::sparse_binary_matrix make_test_H() { + cudaqx::tensor H({std::size_t{4}, std::size_t{10}}); + return cudaq::qec::sparse_binary_matrix(H); +} + +/// Records the CUDA device current on the thread that runs decode(), so a +/// test can observe which device an async worker thread actually used. +class device_recording_decoder : public cudaq::qec::decoder { +public: + std::atomic last_decode_device{-2}; + device_recording_decoder(const cudaq::qec::sparse_binary_matrix &H, + const cudaqx::heterogeneous_map &) + : decoder(H) {} + cudaq::qec::decoder_result + decode(const std::vector &) override { + int dev = -1; + if (cudaGetDevice(&dev) != cudaSuccess) + dev = -1; + last_decode_device.store(dev); + cudaq::qec::decoder_result r; + r.converged = true; + r.result = std::vector(block_size, 0.0); + return r; + } + CUDAQ_EXTENSION_CUSTOM_CREATOR_FUNCTION( + device_recording_decoder, + static std::unique_ptr create( + const cudaq::qec::decoder_init &init, + const cudaqx::heterogeneous_map ¶ms) { + return cudaq::qec::make_pcm_decoder(init, + params); + }) +}; +CUDAQ_EXT_PT_REGISTER_TYPE(device_recording_decoder) + +} // namespace + +TEST(DecoderCudaDeviceId, AbsentKeyIsNoOp) { + auto d = cudaq::qec::decoder::get("sample_decoder", make_test_H()); + EXPECT_EQ(d->get_cuda_device_id(), -1); + std::vector syndrome(4); + auto r = d->decode(syndrome); + EXPECT_EQ(r.result.size(), 10); +} + +TEST(DecoderCudaDeviceId, NegativeIdThrows) { + cudaqx::heterogeneous_map params; + params.insert("cuda_device_id", -2); + try { + auto d = cudaq::qec::decoder::get("sample_decoder", make_test_H(), params); + FAIL() << "expected std::runtime_error"; + } catch (const std::runtime_error &e) { + EXPECT_NE(std::string(e.what()).find("cuda_device_id"), std::string::npos); + } +} + +TEST(DecoderCudaDeviceId, OutOfRangeIdThrows) { + cudaqx::heterogeneous_map params; + params.insert("cuda_device_id", 1 << 20); + try { + auto d = cudaq::qec::decoder::get("sample_decoder", make_test_H(), params); + FAIL() << "expected std::runtime_error"; + } catch (const std::runtime_error &e) { + EXPECT_NE(std::string(e.what()).find("cuda_device_id"), std::string::npos); + } +} + +TEST(DecoderCudaDeviceId, PersistentPinAtConstruction) { + if (cuda_device_count() < 2) + GTEST_SKIP() << "needs >= 2 CUDA devices"; + ScopedDeviceRestore restore; + cudaqx::heterogeneous_map params; + params.insert("cuda_device_id", 1); + auto d = cudaq::qec::decoder::get("sample_decoder", make_test_H(), params); + EXPECT_EQ(d->get_cuda_device_id(), 1); + // The pin is persistent: the constructing thread is still on device 1 + // after decoder::get() returns. + int cur = -1; + ASSERT_EQ(cudaGetDevice(&cur), cudaSuccess); + EXPECT_EQ(cur, 1); + // Decode entry points need no per-call guard on this thread. + std::vector syndrome(4); + auto r = d->decode(syndrome); + EXPECT_EQ(r.result.size(), 10); + ASSERT_EQ(cudaGetDevice(&cur), cudaSuccess); + EXPECT_EQ(cur, 1); +} + +TEST(DecoderCudaDeviceId, KeyStrippedFromPluginParams) { + if (cuda_device_count() < 1) + GTEST_SKIP() << "needs >= 1 CUDA device"; + ScopedDeviceRestore restore; + // Sanity: strict_keys_decoder does reject unknown keys. + cudaqx::heterogeneous_map bogus; + bogus.insert("bogus_key", 1); + EXPECT_THROW( + cudaq::qec::decoder::get("strict_keys_decoder", make_test_H(), bogus), + std::runtime_error); + // cuda_device_id must be consumed by the base and never reach the plugin, + // while permitted keys (decode_to_obs) must survive the rebuild-and-strip. + cudaqx::heterogeneous_map params; + params.insert("cuda_device_id", 0); + params.insert("decode_to_obs", true); + EXPECT_NO_THROW( + cudaq::qec::decoder::get("strict_keys_decoder", make_test_H(), params)); +} + +TEST(DecoderCudaDeviceId, AsyncWorkerPinsItself) { + if (cuda_device_count() < 2) + GTEST_SKIP() << "needs >= 2 CUDA devices"; + ScopedDeviceRestore restore; + cudaqx::heterogeneous_map params; + params.insert("cuda_device_id", 1); + auto d = cudaq::qec::decoder::get("device_recording_decoder", make_test_H(), + params); + auto *rec = dynamic_cast(d.get()); + ASSERT_NE(rec, nullptr); + // decode_async spawns a brand-new std::async thread whose current device + // defaults to 0, NOT the owning thread's device 1. The worker must pin + // itself to the decoder's device for the duration of the call. + std::vector syndrome(4); + auto r = d->decode_async(syndrome).get(); + EXPECT_EQ(r.result.size(), 10); + EXPECT_EQ(rec->last_decode_device.load(), 1); + // The owning (calling) thread's device is untouched by the async call. + int cur = -1; + ASSERT_EQ(cudaGetDevice(&cur), cudaSuccess); + EXPECT_EQ(cur, 1); +} + +TEST(DecoderCudaDeviceId, TwoThreadsTwoDevices) { + if (cuda_device_count() < 2) + GTEST_SKIP() << "needs >= 2 CUDA devices"; + auto worker = [](int id, int &observed_device, bool &decode_ok) { + cudaqx::heterogeneous_map params; + params.insert("cuda_device_id", id); + auto d = cudaq::qec::decoder::get("sample_decoder", make_test_H(), params); + std::vector syndrome(4); + auto r = d->decode(syndrome); + decode_ok = (r.result.size() == 10); + int cur = -1; + observed_device = (cudaGetDevice(&cur) == cudaSuccess) ? cur : -1; + }; + int dev0 = -1, dev1 = -1; + bool ok0 = false, ok1 = false; + std::thread t0(worker, 0, std::ref(dev0), std::ref(ok0)); + std::thread t1(worker, 1, std::ref(dev1), std::ref(ok1)); + t0.join(); + t1.join(); + EXPECT_TRUE(ok0); + EXPECT_TRUE(ok1); + EXPECT_EQ(dev0, 0); + EXPECT_EQ(dev1, 1); +} diff --git a/libs/qec/unittests/test_decoders_yaml.cpp b/libs/qec/unittests/test_decoders_yaml.cpp index dc8be3e41..1b3b0178a 100644 --- a/libs/qec/unittests/test_decoders_yaml.cpp +++ b/libs/qec/unittests/test_decoders_yaml.cpp @@ -720,3 +720,36 @@ TEST(DecoderConfigTest, SimulationHostPointerWrappersForwardToHostRuntime) { EXPECT_EQ(corrections, (std::vector{0})); finalize_decoders(); } + +TEST(DecoderYAMLTest, CudaDeviceIdRoundTrip) { + cudaq::qec::decoding::config::multi_decoder_config multi_config; + auto config = create_test_empty_decoder_config(0); + config.cuda_device_id = 2; + multi_config.decoders.push_back(config); + test_decoder_yaml_roundtrip(multi_config); +} + +TEST(DecoderYAMLTest, PrepareDecoderParamsSurfacesCudaDeviceId) { + // Non-trt type: the insert must happen before prepare_decoder_params()'s + // trt-only early return, so the knob reaches every decoder type. + auto config = create_test_empty_decoder_config(0); + config.cuda_device_id = 3; + auto params = cudaq::qec::decoding::host::prepare_decoder_params(config); + ASSERT_TRUE(params.contains("cuda_device_id")); + EXPECT_EQ(params.get("cuda_device_id"), 3); + + // Absent -> key absent (decoder::get() treats absence as unpinned). + auto config2 = create_test_empty_decoder_config(1); + auto params2 = cudaq::qec::decoding::host::prepare_decoder_params(config2); + EXPECT_FALSE(params2.contains("cuda_device_id")); + + // trt type: still surfaced on the trt branch. + auto config3 = create_test_empty_decoder_config(2); + config3.type = "trt_decoder"; + config3.decoder_custom_args = + cudaq::qec::decoding::config::trt_decoder_config{}; + config3.cuda_device_id = 1; + auto params3 = cudaq::qec::decoding::host::prepare_decoder_params(config3); + ASSERT_TRUE(params3.contains("cuda_device_id")); + EXPECT_EQ(params3.get("cuda_device_id"), 1); +} From bad363098178ed0ae6a6524231ddf1454fad31a9 Mon Sep 17 00:00:00 2001 From: Melody Ren Date: Fri, 10 Jul 2026 11:50:59 -0700 Subject: [PATCH 02/10] Select each decoder's pinned CUDA device on every dispatch path cuda_device_id pinned only the constructing thread, so two dispatch paths decoded on whatever device was current: - decoding-server DecodingSession workers start unpinned; every session decoded on the default device regardless of its pin. The worker now pins itself once at worker_loop entry. - the direct (no realtime session) path decodes on the caller thread, which configure_decoders leaves on the LAST decoder's device. It now selects the decoder's device before each decode. Both paths share a new fail-fast helper (hardware_guards.h, set-if-different, throws). apply_decoder_cuda_device also uses it now: a cudaSetDevice failure previously warned and continued on the wrong device; it now surfaces as a dispatch error response, and graph capture aborts during initialization. Signed-off-by: Melody Ren --- libs/qec/lib/hardware_guards.h | 16 ++++++++++++++++ .../realtime/decoding-server-cqr/CMakeLists.txt | 4 ++++ .../decoding-server-cqr/DecodingSession.cpp | 10 ++++++++++ libs/qec/lib/realtime/qec_realtime_session.cpp | 17 +++++++---------- libs/qec/lib/realtime/realtime_decoding.cpp | 9 +++++++++ 5 files changed, 46 insertions(+), 10 deletions(-) diff --git a/libs/qec/lib/hardware_guards.h b/libs/qec/lib/hardware_guards.h index 3f01196fb..e4178ccce 100644 --- a/libs/qec/lib/hardware_guards.h +++ b/libs/qec/lib/hardware_guards.h @@ -14,6 +14,22 @@ namespace cudaq::qec::detail_affinity { +/// Point the calling thread at \p target before work that allocates or +/// launches on it (no restore; set-if-different). No-op for target < 0. +/// Throws on failure: never silently decode on the wrong GPU. +inline void set_cuda_device_for_decode(int target) { + if (target < 0) + return; + int current = -1; + if (cudaGetDevice(¤t) == cudaSuccess && current == target) + return; + cudaError_t err = cudaSetDevice(target); + if (err != cudaSuccess) + throw std::runtime_error("set_cuda_device_for_decode: cudaSetDevice(" + + std::to_string(target) + + ") failed: " + cudaGetErrorString(err)); +} + /// RAII: set the calling thread's CUDA device, restore the previous device on /// scope exit. No-op for target < 0. Lib-private and header-only so decoder /// plugins built as separate .so files can reuse it (PR2 extends this header diff --git a/libs/qec/lib/realtime/decoding-server-cqr/CMakeLists.txt b/libs/qec/lib/realtime/decoding-server-cqr/CMakeLists.txt index 18ddb6ce0..a6adcdabe 100644 --- a/libs/qec/lib/realtime/decoding-server-cqr/CMakeLists.txt +++ b/libs/qec/lib/realtime/decoding-server-cqr/CMakeLists.txt @@ -149,6 +149,10 @@ target_link_libraries(cudaq-qec-decoding-server # violates the realtime-server dependency-closure contract. cudaq-qec-decoders cudaq-qec-realtime-decoding + PRIVATE + # DecodingSession worker threads pin themselves to their decoder's + # cuda_device_id (cudaSetDevice). + CUDA::cudart ) if(CUDAQ_GPU_ROCE_AVAILABLE) diff --git a/libs/qec/lib/realtime/decoding-server-cqr/DecodingSession.cpp b/libs/qec/lib/realtime/decoding-server-cqr/DecodingSession.cpp index 9e2e7b09c..dc81410b2 100644 --- a/libs/qec/lib/realtime/decoding-server-cqr/DecodingSession.cpp +++ b/libs/qec/lib/realtime/decoding-server-cqr/DecodingSession.cpp @@ -12,6 +12,7 @@ #include #include +#include #include #include @@ -299,6 +300,15 @@ void DecodingSession::on_reset(const WorkItem &item) { } void DecodingSession::worker_loop() { + // One worker owns one decoder: pin this thread to the decoder's requested + // CUDA device once, so every decode (including lazy plugin allocations) + // lands there. Construction pinned only the registry thread. + if (const int cuda_id = dec->get_cuda_device_id(); cuda_id >= 0) { + const cudaError_t err = cudaSetDevice(cuda_id); + if (err != cudaSuccess) + cudaq::qec::error("DecodingSession worker: cudaSetDevice({}) failed: {}", + cuda_id, cudaGetErrorString(err)); + } while (true) { WorkItem item; { diff --git a/libs/qec/lib/realtime/qec_realtime_session.cpp b/libs/qec/lib/realtime/qec_realtime_session.cpp index 620b26b45..f01db70c3 100644 --- a/libs/qec/lib/realtime/qec_realtime_session.cpp +++ b/libs/qec/lib/realtime/qec_realtime_session.cpp @@ -10,6 +10,8 @@ #include "qec_realtime_session.h" +#include "../hardware_guards.h" + #include "cudaq/qec/logger.h" #include "cudaq/qec/realtime/decoder_rpc_ids.h" #include "cudaq/qec/realtime/graph_resources.h" @@ -135,16 +137,11 @@ cudaq::qec::decoder *get_decoder_or_throw(std::int64_t decoder_id) { static void apply_decoder_cuda_device(cudaq::qec::decoder *dec) { if (!dec) return; - const int id = dec->get_cuda_device_id(); - if (id < 0) - return; - int cur = -1; - if (cudaGetDevice(&cur) == cudaSuccess && cur == id) - return; - cudaError_t err = cudaSetDevice(id); - if (err != cudaSuccess) - CUDA_QEC_WARN("apply_decoder_cuda_device: cudaSetDevice({}) failed: {}", id, - cudaGetErrorString(err)); + // Throws on failure (fail fast): host dispatch surfaces it as an error + // response and graph initialization aborts, rather than continuing on + // whichever device happened to be current. + cudaq::qec::detail_affinity::set_cuda_device_for_decode( + dec->get_cuda_device_id()); } // Two-ring response writer: the request stays in `rx_slot` (read-only); the diff --git a/libs/qec/lib/realtime/realtime_decoding.cpp b/libs/qec/lib/realtime/realtime_decoding.cpp index 30519ab45..12e9fe197 100644 --- a/libs/qec/lib/realtime/realtime_decoding.cpp +++ b/libs/qec/lib/realtime/realtime_decoding.cpp @@ -7,6 +7,7 @@ ******************************************************************************/ #include "realtime_decoding.h" +#include "../hardware_guards.h" #include "cudaq/qec/decoder.h" #include "cudaq/qec/logger.h" #include "cudaq/qec/pcm_utils.h" @@ -422,6 +423,14 @@ void enqueue_syndromes(std::size_t decoder_id, uint8_t *syndromes, } #endif + // Direct-call path: this caller thread runs the decode, but + // configure_decoders() constructed every decoder sequentially on one thread, + // leaving the LAST decoder's device current. Point the thread at this + // decoder's pinned device before decoding (set-if-different; throws on + // failure). + cudaq::qec::detail_affinity::set_cuda_device_for_decode( + decoder->get_cuda_device_id()); + std::vector syndrome_u8(syndrome_length); bool did_decode = false; for (std::size_t i = 0; i < syndrome_length; i++) { From 41ea84de7cf9aa0cf9e54d381c4aeedfcbcfaad9 Mon Sep 17 00:00:00 2001 From: Melody Ren Date: Fri, 10 Jul 2026 21:31:11 -0700 Subject: [PATCH 03/10] Fail server startup when a decoder pin cannot be honored Two paths could previously run with a decoder's cuda_device_id silently violated -- the worst failure mode, because decoding can still succeed on the wrong GPU with nothing but an unread log line as evidence: - DecodingSession workers logged a cudaSetDevice failure and kept serving. The pin now runs on the worker thread behind a promise/future handshake in start_worker(): the worker either starts pinned or the exception is rethrown on the registry thread and server startup aborts, the same channel as a decoder that fails to construct. - gpu_roce chose its GPU from HOLOLINK_GPU_ID (default 0) with no knowledge of the decoder's pin, splitting graph capture (pinned device) from ring buffers and device-side graph launch (env device). Both knobs name the same topology fact -- the FPGA-affine GPU -- so they are now reconciled at transport creation: agreement or a single set knob resolves the device, a conflict throws, and an unset environment defers to the decoder's pin. The reconciliation is a plain function compiled in every configuration and unit-tested; the gpu_roce call site remains behind CUDAQ_GPU_ROCE_AVAILABLE and needs hardware validation. Signed-off-by: Melody Ren --- .../decoding-server-cqr/DecodingServer.cpp | 34 ++++++++++++++++-- .../decoding-server-cqr/DecodingServer.h | 10 +++++- .../decoding-server-cqr/DecodingSession.cpp | 36 +++++++++++++------ .../decoding-server-cqr/DecodingSession.h | 5 ++- .../GpuRoceTransceiver.cpp | 4 ++- .../decoding-server-cqr/GpuRoceTransceiver.h | 4 +++ .../unittests/test_decoding_server_core.cpp | 26 ++++++++++++++ 7 files changed, 103 insertions(+), 16 deletions(-) diff --git a/libs/qec/lib/realtime/decoding-server-cqr/DecodingServer.cpp b/libs/qec/lib/realtime/decoding-server-cqr/DecodingServer.cpp index b66f7b242..739f00832 100644 --- a/libs/qec/lib/realtime/decoding-server-cqr/DecodingServer.cpp +++ b/libs/qec/lib/realtime/decoding-server-cqr/DecodingServer.cpp @@ -28,12 +28,35 @@ using cudaq::qec::decoding::config::DecoderTransport; // Constructors // --------------------------------------------------------------------------- +/// gpu_roce runs the whole pipeline -- rings, dispatch scheduler, device-side +/// graph fire -- on ONE GPU: the one the FPGA/NIC is affine to +/// (HOLOLINK_GPU_ID). A decoder pinned elsewhere would split graph capture +/// and graph launch across devices, which CUDA graphs cannot do. Both knobs +/// name the same topology fact, so they must agree. +int reconcile_gpu_roce_device(std::optional env_gpu_id, int decoder_pin) { + if (env_gpu_id && decoder_pin >= 0 && *env_gpu_id != decoder_pin) + throw std::runtime_error( + "gpu_roce device conflict: HOLOLINK_GPU_ID=" + + std::to_string(*env_gpu_id) + " but the decoder is pinned to " + + std::to_string(decoder_pin) + + " (cuda_device_id). The FPGA-affine GPU and the decoder pin must be " + "the same device."); + if (env_gpu_id) + return *env_gpu_id; + return decoder_pin >= 0 ? decoder_pin : 0; +} + std::unique_ptr -DecodingServer::make_transport(DecoderTransport transport_type) { +DecodingServer::make_transport(DecoderTransport transport_type, + int pinned_cuda_device) { switch (transport_type) { case DecoderTransport::gpu_roce: #ifdef CUDAQ_GPU_ROCE_AVAILABLE - return std::make_unique(GpuRoceConfig::from_env()); + { + auto cfg = GpuRoceConfig::from_env(); + cfg.gpu_id = reconcile_gpu_roce_device(cfg.gpu_id_env, pinned_cuda_device); + return std::make_unique(cfg); + } #else throw std::runtime_error( "gpu_roce transport requested but CUDAQ_GPU_ROCE_AVAILABLE is not set. " @@ -65,7 +88,12 @@ DecodingServer::DecodingServer(const std::string &config_yaml) { registry_.load_from_config(config, config_yaml); register_handlers(); - auto t = make_transport(registry_.required_transport()); + const auto &boot_sessions = registry_.sessions(); + const int pinned_cuda_device = + boot_sessions.size() == 1 + ? boot_sessions.begin()->second->dec->get_cuda_device_id() + : -1; + auto t = make_transport(registry_.required_transport(), pinned_cuda_device); ITransceiver *raw = t.get(); owned_transports_.push_back(std::move(t)); function_transport_[kEnqueueSyndromesFunctionId] = raw; diff --git a/libs/qec/lib/realtime/decoding-server-cqr/DecodingServer.h b/libs/qec/lib/realtime/decoding-server-cqr/DecodingServer.h index c56ee69b2..fbc235bd7 100644 --- a/libs/qec/lib/realtime/decoding-server-cqr/DecodingServer.h +++ b/libs/qec/lib/realtime/decoding-server-cqr/DecodingServer.h @@ -15,12 +15,19 @@ #include #include +#include #include #include #include namespace cudaq::qec::decoding_server { +/// Resolve the single GPU a gpu_roce pipeline runs on from the two knobs that +/// can name it: HOLOLINK_GPU_ID (FPGA/NIC affinity; nullopt when unset) and +/// the decoder's cuda_device_id (-1 when unpinned). Throws when both are set +/// and disagree; unset env defers to the pin; neither set -> 0. +int reconcile_gpu_roce_device(std::optional env_gpu_id, int decoder_pin); + /// Maps function_id → non-owning ITransceiver pointer. /// Ownership lives in DecodingServer::owned_transports_. using TransportMap = std::unordered_map; @@ -74,7 +81,8 @@ class DecodingServer { /// until CpuRoceTransceiverAdapter / GpuRoceTransceiverAdapter are /// available via CUDAQ_REALTIME. static std::unique_ptr - make_transport(cudaq::qec::decoding::config::DecoderTransport transport_type); + make_transport(cudaq::qec::decoding::config::DecoderTransport transport_type, + int pinned_cuda_device); // Destruction order matters: the GPU RoCE scheduler (inside // owned_transports_) holds a cudaGraphExec_t captured from a session's diff --git a/libs/qec/lib/realtime/decoding-server-cqr/DecodingSession.cpp b/libs/qec/lib/realtime/decoding-server-cqr/DecodingSession.cpp index dc81410b2..76f7223f1 100644 --- a/libs/qec/lib/realtime/decoding-server-cqr/DecodingSession.cpp +++ b/libs/qec/lib/realtime/decoding-server-cqr/DecodingSession.cpp @@ -8,11 +8,13 @@ #include "DecodingSession.h" #include "RpcWireFormat.h" +#include "../../hardware_guards.h" #include "cudaq/qec/logger.h" #include #include #include +#include #include #include @@ -62,7 +64,30 @@ DecodingSession::create(std::unique_ptr decoder, } void DecodingSession::start_worker() { - worker = std::thread([this] { worker_loop(); }); + // The pin must happen ON the worker thread (CUDA device selection is + // thread-local), but a failure is a startup error that belongs to the + // caller: hand it back through a promise so load_from_config aborts the + // server instead of a worker silently decoding on the wrong device. + std::promise pinned; + auto pin_result = pinned.get_future(); + worker = std::thread([this, &pinned] { + try { + cudaq::qec::detail_affinity::set_cuda_device_for_decode( + dec->get_cuda_device_id()); + pinned.set_value(); + } catch (...) { + pinned.set_exception(std::current_exception()); + return; // never serve work from a mispinned thread + } + worker_loop(); + }); + try { + pin_result.get(); + } catch (...) { + if (worker.joinable()) + worker.join(); + throw; + } } bool DecodingSession::try_enqueue(WorkItem item) { @@ -300,15 +325,6 @@ void DecodingSession::on_reset(const WorkItem &item) { } void DecodingSession::worker_loop() { - // One worker owns one decoder: pin this thread to the decoder's requested - // CUDA device once, so every decode (including lazy plugin allocations) - // lands there. Construction pinned only the registry thread. - if (const int cuda_id = dec->get_cuda_device_id(); cuda_id >= 0) { - const cudaError_t err = cudaSetDevice(cuda_id); - if (err != cudaSuccess) - cudaq::qec::error("DecodingSession worker: cudaSetDevice({}) failed: {}", - cuda_id, cudaGetErrorString(err)); - } while (true) { WorkItem item; { diff --git a/libs/qec/lib/realtime/decoding-server-cqr/DecodingSession.h b/libs/qec/lib/realtime/decoding-server-cqr/DecodingSession.h index 226210191..3a4468ef6 100644 --- a/libs/qec/lib/realtime/decoding-server-cqr/DecodingSession.h +++ b/libs/qec/lib/realtime/decoding-server-cqr/DecodingSession.h @@ -106,7 +106,10 @@ struct DecodingSession { create(std::unique_ptr decoder, SyndromeMappingTable mapping_table); - /// Start the FIFO worker thread. Must be called after create(). + /// Start the FIFO worker thread. Must be called after create(). The + /// worker pins itself to the decoder's cuda_device_id before serving work; + /// a pin failure throws HERE (one worker owns one decoder -- a worker on + /// the wrong device must never serve). void start_worker(); /// Signal shutdown and join the worker (drains any queued items first). diff --git a/libs/qec/lib/realtime/decoding-server-cqr/GpuRoceTransceiver.cpp b/libs/qec/lib/realtime/decoding-server-cqr/GpuRoceTransceiver.cpp index ce7a54ce6..b1ee1f184 100644 --- a/libs/qec/lib/realtime/decoding-server-cqr/GpuRoceTransceiver.cpp +++ b/libs/qec/lib/realtime/decoding-server-cqr/GpuRoceTransceiver.cpp @@ -117,7 +117,9 @@ GpuRoceConfig GpuRoceConfig::from_env() { c.device_name = env_str("HOLOLINK_DEVICE"); c.peer_ip = env_str("HOLOLINK_PEER_IP"); c.remote_qp = env_u32("HOLOLINK_REMOTE_QP", 0); - c.gpu_id = env_int("HOLOLINK_GPU_ID", 0); + if (std::getenv("HOLOLINK_GPU_ID")) + c.gpu_id_env = env_int("HOLOLINK_GPU_ID", 0); + c.gpu_id = c.gpu_id_env.value_or(0); c.frame_size = env_size("HOLOLINK_FRAME_SIZE", 384); c.page_size = env_size("HOLOLINK_PAGE_SIZE", 0); // 0 → derived below c.num_pages = env_size("HOLOLINK_NUM_PAGES", 64); diff --git a/libs/qec/lib/realtime/decoding-server-cqr/GpuRoceTransceiver.h b/libs/qec/lib/realtime/decoding-server-cqr/GpuRoceTransceiver.h index b1c490f02..4456edcc6 100644 --- a/libs/qec/lib/realtime/decoding-server-cqr/GpuRoceTransceiver.h +++ b/libs/qec/lib/realtime/decoding-server-cqr/GpuRoceTransceiver.h @@ -15,6 +15,7 @@ #include #include #include +#include #include #include @@ -38,6 +39,9 @@ struct GpuRoceConfig { std::string device_name; ///< HOLOLINK_DEVICE (IB netdev, e.g. "mlx5_0") uint32_t remote_qp{0}; ///< HOLOLINK_REMOTE_QP (FPGA/emulator QP number) int gpu_id{0}; ///< HOLOLINK_GPU_ID + /// Set iff HOLOLINK_GPU_ID was present in the environment (the FPGA/NIC + /// affinity is a topology fact; absence defers to the decoder's pin). + std::optional gpu_id_env; size_t frame_size{384}; ///< HOLOLINK_FRAME_SIZE (max RPC frame bytes) size_t page_size{0}; ///< HOLOLINK_PAGE_SIZE (0 → derived from frame_size) size_t num_pages{64}; ///< HOLOLINK_NUM_PAGES (ring depth) diff --git a/libs/qec/unittests/test_decoding_server_core.cpp b/libs/qec/unittests/test_decoding_server_core.cpp index 348a5436d..e17d3b7f5 100644 --- a/libs/qec/unittests/test_decoding_server_core.cpp +++ b/libs/qec/unittests/test_decoding_server_core.cpp @@ -6,6 +6,7 @@ * the terms of the Apache License 2.0 which accompanies this distribution. * *******************************************************************************/ +#include "DecodingServer.h" #include "DecodingSession.h" #include "RoundAccumulator.h" #include "RpcDispatcher.h" @@ -256,4 +257,29 @@ TEST(RpcDispatcherTest, ConvertsHandlerExceptionsToErrorResponses) { expect_status(transport, RpcStatus::INTERNAL_ERROR); } +TEST(GpuRoceDeviceReconcile, BothUnsetDefaultsToZero) { + EXPECT_EQ( + cudaq::qec::decoding_server::reconcile_gpu_roce_device(std::nullopt, -1), + 0); +} + +TEST(GpuRoceDeviceReconcile, EnvOnlyWins) { + EXPECT_EQ(cudaq::qec::decoding_server::reconcile_gpu_roce_device(2, -1), 2); +} + +TEST(GpuRoceDeviceReconcile, PinOnlyWins) { + EXPECT_EQ( + cudaq::qec::decoding_server::reconcile_gpu_roce_device(std::nullopt, 3), + 3); +} + +TEST(GpuRoceDeviceReconcile, AgreementPasses) { + EXPECT_EQ(cudaq::qec::decoding_server::reconcile_gpu_roce_device(1, 1), 1); +} + +TEST(GpuRoceDeviceReconcile, ConflictThrows) { + EXPECT_THROW(cudaq::qec::decoding_server::reconcile_gpu_roce_device(0, 2), + std::runtime_error); +} + } // namespace From 310fdfff462e4a5d4c4f5356ce47bd7ad6dc2e58 Mon Sep 17 00:00:00 2001 From: Melody Ren Date: Fri, 10 Jul 2026 21:45:59 -0700 Subject: [PATCH 04/10] CI-runnable coverage for the pin handshake and its helper - set_cuda_device_for_decode: -1 is a no-op and an impossible device id throws -- both runnable on GPU-less CI (cudaSetDevice past the device count fails there too), covering the failure transport the worker handshake rides on, which cannot be induced end-to-end after construction-time validation. - DecodingSession handshake smoke: a decoder pinned to device 0 starts its worker through the promise/future handshake and serves a queued item (skips below 1 GPU). Signed-off-by: Melody Ren --- libs/qec/unittests/CMakeLists.txt | 3 +- .../unittests/test_decoding_server_core.cpp | 49 +++++++++++++++++++ 2 files changed, 51 insertions(+), 1 deletion(-) diff --git a/libs/qec/unittests/CMakeLists.txt b/libs/qec/unittests/CMakeLists.txt index 5c0a2ab05..359258a51 100644 --- a/libs/qec/unittests/CMakeLists.txt +++ b/libs/qec/unittests/CMakeLists.txt @@ -56,7 +56,8 @@ add_executable(test_decoding_server_core test_decoding_server_core.cpp) target_link_libraries(test_decoding_server_core PRIVATE GTest::gtest_main cudaq-qec-decoding-server - cudaq::cudaq) + cudaq::cudaq + CUDA::cudart) add_dependencies(CUDAQXQECUnitTests test_decoding_server_core) gtest_discover_tests(test_decoding_server_core) diff --git a/libs/qec/unittests/test_decoding_server_core.cpp b/libs/qec/unittests/test_decoding_server_core.cpp index e17d3b7f5..d12197d66 100644 --- a/libs/qec/unittests/test_decoding_server_core.cpp +++ b/libs/qec/unittests/test_decoding_server_core.cpp @@ -11,15 +11,19 @@ #include "RoundAccumulator.h" #include "RpcDispatcher.h" #include "RpcWireFormat.h" +#include "../lib/hardware_guards.h" #include "cudaq/qec/decoder.h" #include "cudaq/qec/sparse_binary_matrix.h" #include +#include #include +#include #include #include +#include #include #include @@ -282,4 +286,49 @@ TEST(GpuRoceDeviceReconcile, ConflictThrows) { std::runtime_error); } +TEST(SetCudaDeviceForDecode, UnpinnedIsNoOp) { + // -1 = unpinned: must never touch the device or throw, even on a machine + // with no CUDA devices at all. + EXPECT_NO_THROW(cudaq::qec::detail_affinity::set_cuda_device_for_decode(-1)); +} + +TEST(SetCudaDeviceForDecode, ImpossibleDeviceThrows) { + // The handshake's failure transport rides on this throw; an id beyond the + // device count fails cudaSetDevice on any machine, including GPU-less CI. + int count = 0; + if (cudaGetDeviceCount(&count) != cudaSuccess) + count = 0; + EXPECT_THROW( + cudaq::qec::detail_affinity::set_cuda_device_for_decode(count + 7), + std::runtime_error); +} + +TEST(DecodingSessionPinHandshake, PinnedWorkerStartsAndServes) { + // start_worker() must resolve the pin handshake (throwing on failure per + // its contract) and leave a live worker serving items. + int count = 0; + if (cudaGetDeviceCount(&count) != cudaSuccess || count < 1) + GTEST_SKIP() << "needs >= 1 CUDA device"; + + cudaqx::heterogeneous_map params; + params.insert("cuda_device_id", 0); + auto dec = cudaq::qec::decoder::get( + "single_error_lut", + cudaq::qec::sparse_binary_matrix::from_csr(1, 1, {0, 1}, {0}), params); + dec->set_O_sparse(std::vector>{{0}}); + dec->set_D_sparse(std::vector>{{0, 1}}); + + SyndromeMappingTable table; + table[0] = {{}}; + auto session = DecodingSession::create(std::move(dec), std::move(table)); + ASSERT_NO_THROW(session->start_worker()); + + CaptureTransceiver transport; + ASSERT_TRUE(session->try_enqueue(make_reset(transport))); + for (int i = 0; i < 200 && session->reset_count.load() == 0; ++i) + std::this_thread::sleep_for(std::chrono::milliseconds(5)); + EXPECT_EQ(session->reset_count.load(), 1u) + << "pinned worker did not serve the queued item"; +} + } // namespace From 27327e9e4821dfc24a8e8d003d1a56cee1dba016 Mon Sep 17 00:00:00 2001 From: Melody Ren Date: Fri, 10 Jul 2026 23:42:29 -0700 Subject: [PATCH 05/10] Address independent review findings on the pinning stack - Add the handshake failure-injection test: cuda_device_id_ is protected, so a test decoder can carry an impossible id past the construction-time range check, and start_worker() must throw and join the worker. This is the test that fails if the handshake ever reverts to log-and-continue. - Select the decoder's device on the get_corrections and reset paths (host dispatcher and direct call), not just enqueue: a plugin's clear_corrections or reset_decoder may touch device memory, and interleaved pinned decoders left the wrong device current. - On the direct path, pin before the syndrome-capture callback so a pin failure cannot record a round in a --save_syndrome trace that was never decoded. Session-forwarded requests still capture before forwarding and still skip the pin. - Reject a negative HOLOLINK_GPU_ID in the gpu_roce reconciler instead of resolving to a garbage device. - Join already-started workers before the transports are destroyed when a constructor that installs transports first fails mid load_from_config; drop a dead include. Signed-off-by: Melody Ren --- .../decoding-server-cqr/DecodingServer.cpp | 26 +++++++++++-- .../decoding-server-cqr/DecodingSession.cpp | 1 - .../qec/lib/realtime/qec_realtime_session.cpp | 4 +- libs/qec/lib/realtime/realtime_decoding.cpp | 24 ++++++++---- .../unittests/test_decoding_server_core.cpp | 38 +++++++++++++++++++ 5 files changed, 81 insertions(+), 12 deletions(-) diff --git a/libs/qec/lib/realtime/decoding-server-cqr/DecodingServer.cpp b/libs/qec/lib/realtime/decoding-server-cqr/DecodingServer.cpp index 739f00832..f0df198f4 100644 --- a/libs/qec/lib/realtime/decoding-server-cqr/DecodingServer.cpp +++ b/libs/qec/lib/realtime/decoding-server-cqr/DecodingServer.cpp @@ -34,6 +34,9 @@ using cudaq::qec::decoding::config::DecoderTransport; /// and graph launch across devices, which CUDA graphs cannot do. Both knobs /// name the same topology fact, so they must agree. int reconcile_gpu_roce_device(std::optional env_gpu_id, int decoder_pin) { + if (env_gpu_id && *env_gpu_id < 0) + throw std::runtime_error("HOLOLINK_GPU_ID must be >= 0 (got " + + std::to_string(*env_gpu_id) + ")"); if (env_gpu_id && decoder_pin >= 0 && *env_gpu_id != decoder_pin) throw std::runtime_error( "gpu_roce device conflict: HOLOLINK_GPU_ID=" + @@ -130,7 +133,12 @@ DecodingServer::DecodingServer(std::unique_ptr transport, function_transport_[kEnqueueSyndromesFunctionId] = raw; function_transport_[kGetCorrectionsFunctionId] = raw; function_transport_[kResetDecoderFunctionId] = raw; - init(config_yaml); + try { + init(config_yaml); + } catch (...) { + registry_.stop_workers(); + throw; + } } DecodingServer::DecodingServer( @@ -141,7 +149,14 @@ DecodingServer::DecodingServer( function_transport_[kEnqueueSyndromesFunctionId] = raw; function_transport_[kGetCorrectionsFunctionId] = raw; function_transport_[kResetDecoderFunctionId] = raw; - registry_.load_from_config(config, "configure_decoders()"); + try { + registry_.load_from_config(config, "configure_decoders()"); + } catch (...) { + // Members destroy in reverse order (transports before registry); join any + // already-started workers while the transports still exist. + registry_.stop_workers(); + throw; + } register_handlers(); } @@ -150,7 +165,12 @@ DecodingServer::DecodingServer(std::vector> owned, const std::string &config_yaml) : owned_transports_(std::move(owned)), function_transport_(std::move(function_transport)) { - init(config_yaml); + try { + init(config_yaml); + } catch (...) { + registry_.stop_workers(); + throw; + } } DecodingServer::~DecodingServer() { diff --git a/libs/qec/lib/realtime/decoding-server-cqr/DecodingSession.cpp b/libs/qec/lib/realtime/decoding-server-cqr/DecodingSession.cpp index 76f7223f1..1ce711dda 100644 --- a/libs/qec/lib/realtime/decoding-server-cqr/DecodingSession.cpp +++ b/libs/qec/lib/realtime/decoding-server-cqr/DecodingSession.cpp @@ -13,7 +13,6 @@ #include #include -#include #include #include #include diff --git a/libs/qec/lib/realtime/qec_realtime_session.cpp b/libs/qec/lib/realtime/qec_realtime_session.cpp index f01db70c3..e949bba82 100644 --- a/libs/qec/lib/realtime/qec_realtime_session.cpp +++ b/libs/qec/lib/realtime/qec_realtime_session.cpp @@ -279,7 +279,9 @@ void reset_decoder_host(const void *rx_slot, void *tx_slot, std::size_t) { const auto *body = reinterpret_cast( static_cast(rx_slot) + sizeof(cudaq::realtime::RPCHeader)); - get_decoder_or_throw(body->decoder_id)->reset_decoder(); + auto *decoder = get_decoder_or_throw(body->decoder_id); + apply_decoder_cuda_device(decoder); + decoder->reset_decoder(); write_response(tx_slot, rx_slot, 0); } catch (...) { write_response(tx_slot, rx_slot, -2); diff --git a/libs/qec/lib/realtime/realtime_decoding.cpp b/libs/qec/lib/realtime/realtime_decoding.cpp index 12e9fe197..22ba4b1e4 100644 --- a/libs/qec/lib/realtime/realtime_decoding.cpp +++ b/libs/qec/lib/realtime/realtime_decoding.cpp @@ -401,15 +401,18 @@ void enqueue_syndromes(std::size_t decoder_id, uint8_t *syndromes, syndrome_length, max_syndromes)); } - // Invoke syndrome capture callback if registered (for --save_syndrome - // feature) - if (g_syndrome_capture_callback) { - auto packed_syndrome = pack_syndrome_bits(syndromes, syndrome_length); - g_syndrome_capture_callback(packed_syndrome.data(), packed_syndrome.size()); - } + const auto capture_syndromes = [&] { + // --save_syndrome feature: record what is actually submitted for decode. + if (g_syndrome_capture_callback) { + auto packed_syndrome = pack_syndrome_bits(syndromes, syndrome_length); + g_syndrome_capture_callback(packed_syndrome.data(), + packed_syndrome.size()); + } + }; #ifdef CUDAQ_REALTIME_ROOT if (g_realtime_session) { + capture_syndromes(); try { cudaq::qec::decoding::rpc_producer::enqueue_syndromes( *g_realtime_session, decoder_id, syndromes, syndrome_length, tag); @@ -427,9 +430,11 @@ void enqueue_syndromes(std::size_t decoder_id, uint8_t *syndromes, // configure_decoders() constructed every decoder sequentially on one thread, // leaving the LAST decoder's device current. Point the thread at this // decoder's pinned device before decoding (set-if-different; throws on - // failure). + // failure) -- and before the capture callback, so a pin failure cannot + // record a round that was never decoded. cudaq::qec::detail_affinity::set_cuda_device_for_decode( decoder->get_cuda_device_id()); + capture_syndromes(); std::vector syndrome_u8(syndrome_length); bool did_decode = false; @@ -495,6 +500,9 @@ void get_corrections(std::size_t decoder_id, uint8_t *corrections, } #endif + // clear_corrections may touch device memory in some plugins. + cudaq::qec::detail_affinity::set_cuda_device_for_decode( + decoder->get_cuda_device_id()); auto ret = decoder->get_obs_corrections(); for (std::size_t i = 0; i < correction_length; ++i) { corrections[i] = ret[i]; @@ -530,6 +538,8 @@ void reset_decoder(std::size_t decoder_id) { } #endif + cudaq::qec::detail_affinity::set_cuda_device_for_decode( + decoder->get_cuda_device_id()); decoder->reset_decoder(); } diff --git a/libs/qec/unittests/test_decoding_server_core.cpp b/libs/qec/unittests/test_decoding_server_core.cpp index d12197d66..76115afdb 100644 --- a/libs/qec/unittests/test_decoding_server_core.cpp +++ b/libs/qec/unittests/test_decoding_server_core.cpp @@ -303,6 +303,44 @@ TEST(SetCudaDeviceForDecode, ImpossibleDeviceThrows) { std::runtime_error); } +/// cuda_device_id_ is protected: setting an impossible id directly bypasses +/// decoder::get()'s construction-time range check, the only front door -- +/// which is exactly what makes the handshake's failure path injectable here. +class MispinnedDecoder final : public cudaq::qec::decoder { +public: + MispinnedDecoder() + : decoder(cudaq::qec::sparse_binary_matrix::from_csr( + /*num_rows=*/1, /*num_cols=*/1, /*row_ptrs=*/{0, 1}, + /*col_indices=*/{0})) { + set_O_sparse(std::vector>{{0}}); + set_D_sparse(std::vector>{{0, 1}}); + cuda_device_id_ = 1 << 20; + } + cudaq::qec::decoder_result + decode(const std::vector &) override { + return {}; + } +}; + +TEST(DecodingSessionPinHandshake, UnhonorablePinFailsStartWorker) { + // The contract under test: a worker that cannot pin must never serve, and + // the failure must surface on the caller (server-startup) thread. This is + // the test that fails if start_worker ever reverts to log-and-continue. + SyndromeMappingTable table; + table[0] = {{}}; + auto session = DecodingSession::create(std::make_unique(), + std::move(table)); + EXPECT_THROW(session->start_worker(), std::runtime_error); + // The failed worker was joined inside start_worker; nothing is left to + // serve and destruction must not hang. + EXPECT_FALSE(session->worker.joinable()); +} + +TEST(GpuRoceDeviceReconcile, NegativeEnvThrows) { + EXPECT_THROW(cudaq::qec::decoding_server::reconcile_gpu_roce_device(-1, -1), + std::runtime_error); +} + TEST(DecodingSessionPinHandshake, PinnedWorkerStartsAndServes) { // start_worker() must resolve the pin handshake (throwing on failure per // its contract) and leave a live worker serving items. From b9c3ded946c1e9a30e7c8eef4ce39e63685a2dab Mon Sep 17 00:00:00 2001 From: Melody Ren Date: Mon, 13 Jul 2026 14:30:42 -0700 Subject: [PATCH 06/10] Fix clang-format: remove extra blank line in hardware_guards.h Signed-off-by: Melody Ren --- libs/qec/lib/hardware_guards.h | 1 - 1 file changed, 1 deletion(-) diff --git a/libs/qec/lib/hardware_guards.h b/libs/qec/lib/hardware_guards.h index 9c878d754..e4178ccce 100644 --- a/libs/qec/lib/hardware_guards.h +++ b/libs/qec/lib/hardware_guards.h @@ -30,7 +30,6 @@ inline void set_cuda_device_for_decode(int target) { ") failed: " + cudaGetErrorString(err)); } - /// RAII: set the calling thread's CUDA device, restore the previous device on /// scope exit. No-op for target < 0. Lib-private and header-only so decoder /// plugins built as separate .so files can reuse it (PR2 extends this header From 8c266d3dc8ece6ab7d4a3751fbf659e6bb5f63ef Mon Sep 17 00:00:00 2001 From: Melody Ren Date: Mon, 13 Jul 2026 15:21:31 -0700 Subject: [PATCH 07/10] Restore the calling thread's CUDA device when decoder construction fails Signed-off-by: Melody Ren --- libs/qec/lib/decoder.cpp | 51 +++++++++++++++++++++++++++++++++------- 1 file changed, 43 insertions(+), 8 deletions(-) diff --git a/libs/qec/lib/decoder.cpp b/libs/qec/lib/decoder.cpp index eef1def36..3f0947e27 100644 --- a/libs/qec/lib/decoder.cpp +++ b/libs/qec/lib/decoder.cpp @@ -154,6 +154,47 @@ static int read_cuda_device_id(const cudaqx::heterogeneous_map ¶ms) { return value; } +/// Selects the construction device and restores the previous device unless +/// commit() is called. Makes failed plugin construction transactional: if the +/// plugin ctor throws, the calling thread is left on its original device +/// instead of leaking the pin to whichever device was selected for the attempt. +class ConstructionDevicePin { +public: + explicit ConstructionDevicePin(int target) : target_(target) { + cudaError_t err = cudaGetDevice(&previous_); + if (err != cudaSuccess) + throw std::runtime_error( + "cuda_device_id " + std::to_string(target_) + + " could not be selected because cudaGetDevice() failed: " + + cudaGetErrorString(err)); + err = cudaSetDevice(target_); + if (err != cudaSuccess) + throw std::runtime_error("cudaSetDevice(" + std::to_string(target_) + + ") failed: " + cudaGetErrorString(err)); + selected_ = true; + } + + ~ConstructionDevicePin() { + if (selected_ && !committed_ && previous_ != target_) + (void)cudaSetDevice(previous_); + } + + ConstructionDevicePin(const ConstructionDevicePin &) = delete; + ConstructionDevicePin &operator=(const ConstructionDevicePin &) = delete; + + // Keep the pin: one thread owns one decoder, so leaving the constructing + // thread on the target device lets later allocations and kernel launches -- + // including lazy ones inside decode() -- land there with no per-call + // machinery. + void commit() { committed_ = true; } + +private: + int target_ = -1; + int previous_ = -1; + bool selected_ = false; + bool committed_ = false; +}; + std::unique_ptr decoder::get(const std::string &name, const decoder_init &init, const cudaqx::heterogeneous_map ¶m_map) { @@ -168,14 +209,7 @@ decoder::get(const std::string &name, const decoder_init &init, const int cuda_device_id = read_cuda_device_id(param_map); if (cuda_device_id < 0) return iter->second(init, param_map); - // Pin the constructing thread persistently (no restore): one thread owns - // one decoder, so every later allocation and kernel launch on this thread - // -- including lazy allocations inside a plugin's decode() -- lands on the - // requested device with no per-call machinery. - cudaError_t err = cudaSetDevice(cuda_device_id); - if (err != cudaSuccess) - throw std::runtime_error("cudaSetDevice(" + std::to_string(cuda_device_id) + - ") failed: " + cudaGetErrorString(err)); + ConstructionDevicePin device_pin(cuda_device_id); // The key is consumed here; strip it so plugins that strictly validate // their parameter keys do not reject it. cudaqx::heterogeneous_map plugin_params; @@ -184,6 +218,7 @@ decoder::get(const std::string &name, const decoder_init &init, plugin_params.insert(kv.first, kv.second); auto d = iter->second(init, plugin_params); d->cuda_device_id_ = cuda_device_id; + device_pin.commit(); return d; } From 98d19bd04a7ef212d395ac2f6ef6ff9d69231fa6 Mon Sep 17 00:00:00 2001 From: Melody Ren Date: Mon, 13 Jul 2026 16:06:16 -0700 Subject: [PATCH 08/10] Source the gpu_roce transport device from cuda_device_id instead of the HOLOLINK_GPU_ID env var Signed-off-by: Melody Ren --- .../decoding-server-cqr/DecodingServer.cpp | 24 ++++----------- .../decoding-server-cqr/DecodingServer.h | 9 ++---- .../GpuRoceTransceiver.cpp | 6 ++-- .../decoding-server-cqr/GpuRoceTransceiver.h | 14 ++++----- .../unittests/test_decoding_server_core.cpp | 30 +++---------------- 5 files changed, 22 insertions(+), 61 deletions(-) diff --git a/libs/qec/lib/realtime/decoding-server-cqr/DecodingServer.cpp b/libs/qec/lib/realtime/decoding-server-cqr/DecodingServer.cpp index f0df198f4..14e4b1e79 100644 --- a/libs/qec/lib/realtime/decoding-server-cqr/DecodingServer.cpp +++ b/libs/qec/lib/realtime/decoding-server-cqr/DecodingServer.cpp @@ -29,23 +29,11 @@ using cudaq::qec::decoding::config::DecoderTransport; // --------------------------------------------------------------------------- /// gpu_roce runs the whole pipeline -- rings, dispatch scheduler, device-side -/// graph fire -- on ONE GPU: the one the FPGA/NIC is affine to -/// (HOLOLINK_GPU_ID). A decoder pinned elsewhere would split graph capture -/// and graph launch across devices, which CUDA graphs cannot do. Both knobs -/// name the same topology fact, so they must agree. -int reconcile_gpu_roce_device(std::optional env_gpu_id, int decoder_pin) { - if (env_gpu_id && *env_gpu_id < 0) - throw std::runtime_error("HOLOLINK_GPU_ID must be >= 0 (got " + - std::to_string(*env_gpu_id) + ")"); - if (env_gpu_id && decoder_pin >= 0 && *env_gpu_id != decoder_pin) - throw std::runtime_error( - "gpu_roce device conflict: HOLOLINK_GPU_ID=" + - std::to_string(*env_gpu_id) + " but the decoder is pinned to " + - std::to_string(decoder_pin) + - " (cuda_device_id). The FPGA-affine GPU and the decoder pin must be " - "the same device."); - if (env_gpu_id) - return *env_gpu_id; +/// graph fire -- on ONE GPU: the one the FPGA/NIC is affine to. The decoder +/// must be pinned to that same device via cuda_device_id, because CUDA graphs +/// cannot split capture and launch across devices. An unpinned decoder defaults +/// to device 0. +int reconcile_gpu_roce_device(int decoder_pin) { return decoder_pin >= 0 ? decoder_pin : 0; } @@ -57,7 +45,7 @@ DecodingServer::make_transport(DecoderTransport transport_type, #ifdef CUDAQ_GPU_ROCE_AVAILABLE { auto cfg = GpuRoceConfig::from_env(); - cfg.gpu_id = reconcile_gpu_roce_device(cfg.gpu_id_env, pinned_cuda_device); + cfg.gpu_id = reconcile_gpu_roce_device(pinned_cuda_device); return std::make_unique(cfg); } #else diff --git a/libs/qec/lib/realtime/decoding-server-cqr/DecodingServer.h b/libs/qec/lib/realtime/decoding-server-cqr/DecodingServer.h index fbc235bd7..b9f59ec91 100644 --- a/libs/qec/lib/realtime/decoding-server-cqr/DecodingServer.h +++ b/libs/qec/lib/realtime/decoding-server-cqr/DecodingServer.h @@ -15,18 +15,15 @@ #include #include -#include #include #include #include namespace cudaq::qec::decoding_server { -/// Resolve the single GPU a gpu_roce pipeline runs on from the two knobs that -/// can name it: HOLOLINK_GPU_ID (FPGA/NIC affinity; nullopt when unset) and -/// the decoder's cuda_device_id (-1 when unpinned). Throws when both are set -/// and disagree; unset env defers to the pin; neither set -> 0. -int reconcile_gpu_roce_device(std::optional env_gpu_id, int decoder_pin); +/// Resolve the single GPU a gpu_roce pipeline runs on from the decoder's +/// cuda_device_id (-1 when unpinned). An unpinned decoder defaults to device 0. +int reconcile_gpu_roce_device(int decoder_pin); /// Maps function_id → non-owning ITransceiver pointer. /// Ownership lives in DecodingServer::owned_transports_. diff --git a/libs/qec/lib/realtime/decoding-server-cqr/GpuRoceTransceiver.cpp b/libs/qec/lib/realtime/decoding-server-cqr/GpuRoceTransceiver.cpp index b1ee1f184..3255cee80 100644 --- a/libs/qec/lib/realtime/decoding-server-cqr/GpuRoceTransceiver.cpp +++ b/libs/qec/lib/realtime/decoding-server-cqr/GpuRoceTransceiver.cpp @@ -117,9 +117,9 @@ GpuRoceConfig GpuRoceConfig::from_env() { c.device_name = env_str("HOLOLINK_DEVICE"); c.peer_ip = env_str("HOLOLINK_PEER_IP"); c.remote_qp = env_u32("HOLOLINK_REMOTE_QP", 0); - if (std::getenv("HOLOLINK_GPU_ID")) - c.gpu_id_env = env_int("HOLOLINK_GPU_ID", 0); - c.gpu_id = c.gpu_id_env.value_or(0); + // gpu_id is not read from the environment: the device is the decoder's + // cuda_device_id, resolved by reconcile_gpu_roce_device() at transport + // creation. c.frame_size = env_size("HOLOLINK_FRAME_SIZE", 384); c.page_size = env_size("HOLOLINK_PAGE_SIZE", 0); // 0 → derived below c.num_pages = env_size("HOLOLINK_NUM_PAGES", 64); diff --git a/libs/qec/lib/realtime/decoding-server-cqr/GpuRoceTransceiver.h b/libs/qec/lib/realtime/decoding-server-cqr/GpuRoceTransceiver.h index 4456edcc6..4a189f9a1 100644 --- a/libs/qec/lib/realtime/decoding-server-cqr/GpuRoceTransceiver.h +++ b/libs/qec/lib/realtime/decoding-server-cqr/GpuRoceTransceiver.h @@ -15,7 +15,6 @@ #include #include #include -#include #include #include @@ -32,16 +31,15 @@ struct cudaq_dispatch_graph_context; namespace cudaq::qec::decoding_server { -/// Runtime configuration for GpuRoceTransceiver. All fields are read from -/// environment variables so that the server can be reconfigured without a -/// rebuild. +/// Runtime configuration for GpuRoceTransceiver. Transport fields are read +/// from environment variables so that the server can be reconfigured without a +/// rebuild; gpu_id is the exception -- it is the decoder's cuda_device_id, +/// filled in at transport creation. struct GpuRoceConfig { std::string device_name; ///< HOLOLINK_DEVICE (IB netdev, e.g. "mlx5_0") uint32_t remote_qp{0}; ///< HOLOLINK_REMOTE_QP (FPGA/emulator QP number) - int gpu_id{0}; ///< HOLOLINK_GPU_ID - /// Set iff HOLOLINK_GPU_ID was present in the environment (the FPGA/NIC - /// affinity is a topology fact; absence defers to the decoder's pin). - std::optional gpu_id_env; + int gpu_id{0}; ///< FPGA-affine GPU; set from the decoder's + ///< cuda_device_id by reconcile_gpu_roce_device() size_t frame_size{384}; ///< HOLOLINK_FRAME_SIZE (max RPC frame bytes) size_t page_size{0}; ///< HOLOLINK_PAGE_SIZE (0 → derived from frame_size) size_t num_pages{64}; ///< HOLOLINK_NUM_PAGES (ring depth) diff --git a/libs/qec/unittests/test_decoding_server_core.cpp b/libs/qec/unittests/test_decoding_server_core.cpp index 76115afdb..7d9af4a8f 100644 --- a/libs/qec/unittests/test_decoding_server_core.cpp +++ b/libs/qec/unittests/test_decoding_server_core.cpp @@ -261,29 +261,12 @@ TEST(RpcDispatcherTest, ConvertsHandlerExceptionsToErrorResponses) { expect_status(transport, RpcStatus::INTERNAL_ERROR); } -TEST(GpuRoceDeviceReconcile, BothUnsetDefaultsToZero) { - EXPECT_EQ( - cudaq::qec::decoding_server::reconcile_gpu_roce_device(std::nullopt, -1), - 0); +TEST(GpuRoceDeviceReconcile, UnpinnedDefaultsToZero) { + EXPECT_EQ(cudaq::qec::decoding_server::reconcile_gpu_roce_device(-1), 0); } -TEST(GpuRoceDeviceReconcile, EnvOnlyWins) { - EXPECT_EQ(cudaq::qec::decoding_server::reconcile_gpu_roce_device(2, -1), 2); -} - -TEST(GpuRoceDeviceReconcile, PinOnlyWins) { - EXPECT_EQ( - cudaq::qec::decoding_server::reconcile_gpu_roce_device(std::nullopt, 3), - 3); -} - -TEST(GpuRoceDeviceReconcile, AgreementPasses) { - EXPECT_EQ(cudaq::qec::decoding_server::reconcile_gpu_roce_device(1, 1), 1); -} - -TEST(GpuRoceDeviceReconcile, ConflictThrows) { - EXPECT_THROW(cudaq::qec::decoding_server::reconcile_gpu_roce_device(0, 2), - std::runtime_error); +TEST(GpuRoceDeviceReconcile, PinSelectsDevice) { + EXPECT_EQ(cudaq::qec::decoding_server::reconcile_gpu_roce_device(3), 3); } TEST(SetCudaDeviceForDecode, UnpinnedIsNoOp) { @@ -336,11 +319,6 @@ TEST(DecodingSessionPinHandshake, UnhonorablePinFailsStartWorker) { EXPECT_FALSE(session->worker.joinable()); } -TEST(GpuRoceDeviceReconcile, NegativeEnvThrows) { - EXPECT_THROW(cudaq::qec::decoding_server::reconcile_gpu_roce_device(-1, -1), - std::runtime_error); -} - TEST(DecodingSessionPinHandshake, PinnedWorkerStartsAndServes) { // start_worker() must resolve the pin handshake (throwing on failure per // its contract) and leave a live worker serving items. From dacd989be0b5b356fc9df592ca4d00b019b503ee Mon Sep 17 00:00:00 2001 From: Melody Ren Date: Tue, 14 Jul 2026 12:48:34 -0700 Subject: [PATCH 09/10] Rename reconcile_gpu_roce_device to resolve_decode_device Signed-off-by: Melody Ren --- .../realtime/decoding-server-cqr/DecodingServer.cpp | 13 +++++++------ .../realtime/decoding-server-cqr/DecodingServer.h | 4 ++-- .../realtime/decoding-server-cqr/GpuRoceFactory.cpp | 4 ++-- .../decoding-server-cqr/GpuRoceTransceiver.cpp | 2 +- .../decoding-server-cqr/GpuRoceTransceiver.h | 2 +- libs/qec/unittests/test_decoding_server_core.cpp | 8 ++++---- 6 files changed, 17 insertions(+), 16 deletions(-) diff --git a/libs/qec/lib/realtime/decoding-server-cqr/DecodingServer.cpp b/libs/qec/lib/realtime/decoding-server-cqr/DecodingServer.cpp index beb65667f..37aa9327f 100644 --- a/libs/qec/lib/realtime/decoding-server-cqr/DecodingServer.cpp +++ b/libs/qec/lib/realtime/decoding-server-cqr/DecodingServer.cpp @@ -37,12 +37,13 @@ using cudaq::qec::decoding::config::DecoderTransport; // Constructors // --------------------------------------------------------------------------- -/// gpu_roce runs the whole pipeline -- rings, dispatch scheduler, device-side -/// graph fire -- on ONE GPU: the one the FPGA/NIC is affine to. The decoder -/// must be pinned to that same device via cuda_device_id, because CUDA graphs -/// cannot split capture and launch across devices. An unpinned decoder defaults -/// to device 0. -int reconcile_gpu_roce_device(int decoder_pin) { +/// Resolve the CUDA device a decode pipeline runs on from the decoder's +/// cuda_device_id pin; an unpinned decoder (-1) defaults to device 0. The +/// gpu_roce path relies on this to place its rings, dispatch scheduler, and +/// device-side graph fire on the one GPU the FPGA/NIC is affine to -- CUDA +/// graphs cannot split capture and launch across devices, so the decoder must +/// be pinned to that device. +int resolve_decode_device(int decoder_pin) { return decoder_pin >= 0 ? decoder_pin : 0; } diff --git a/libs/qec/lib/realtime/decoding-server-cqr/DecodingServer.h b/libs/qec/lib/realtime/decoding-server-cqr/DecodingServer.h index b9f59ec91..4902e256a 100644 --- a/libs/qec/lib/realtime/decoding-server-cqr/DecodingServer.h +++ b/libs/qec/lib/realtime/decoding-server-cqr/DecodingServer.h @@ -21,9 +21,9 @@ namespace cudaq::qec::decoding_server { -/// Resolve the single GPU a gpu_roce pipeline runs on from the decoder's +/// Resolve the CUDA device a decode pipeline runs on from the decoder's /// cuda_device_id (-1 when unpinned). An unpinned decoder defaults to device 0. -int reconcile_gpu_roce_device(int decoder_pin); +int resolve_decode_device(int decoder_pin); /// Maps function_id → non-owning ITransceiver pointer. /// Ownership lives in DecodingServer::owned_transports_. diff --git a/libs/qec/lib/realtime/decoding-server-cqr/GpuRoceFactory.cpp b/libs/qec/lib/realtime/decoding-server-cqr/GpuRoceFactory.cpp index 27e74494c..651290121 100644 --- a/libs/qec/lib/realtime/decoding-server-cqr/GpuRoceFactory.cpp +++ b/libs/qec/lib/realtime/decoding-server-cqr/GpuRoceFactory.cpp @@ -13,7 +13,7 @@ // component WHOLE_ARCHIVE: the sole reference to this symbol is weak, which // does not pull archive members on its own. -#include "DecodingServer.h" // reconcile_gpu_roce_device (core symbol) +#include "DecodingServer.h" // resolve_decode_device (core symbol) #include "GpuRoceTransceiver.h" extern "C" cudaq::qec::decoding_server::ITransceiver * @@ -22,6 +22,6 @@ cudaqx_qec_make_gpu_roce_transceiver(int pinned_cuda_device) { // The gpu_roce device is the decoder's cuda_device_id pin; resolve it here, // inside the component, where GpuRoceConfig is visible. auto cfg = GpuRoceConfig::from_env(); - cfg.gpu_id = reconcile_gpu_roce_device(pinned_cuda_device); + cfg.gpu_id = resolve_decode_device(pinned_cuda_device); return new GpuRoceTransceiver(cfg); } diff --git a/libs/qec/lib/realtime/decoding-server-cqr/GpuRoceTransceiver.cpp b/libs/qec/lib/realtime/decoding-server-cqr/GpuRoceTransceiver.cpp index 3255cee80..42c26a398 100644 --- a/libs/qec/lib/realtime/decoding-server-cqr/GpuRoceTransceiver.cpp +++ b/libs/qec/lib/realtime/decoding-server-cqr/GpuRoceTransceiver.cpp @@ -118,7 +118,7 @@ GpuRoceConfig GpuRoceConfig::from_env() { c.peer_ip = env_str("HOLOLINK_PEER_IP"); c.remote_qp = env_u32("HOLOLINK_REMOTE_QP", 0); // gpu_id is not read from the environment: the device is the decoder's - // cuda_device_id, resolved by reconcile_gpu_roce_device() at transport + // cuda_device_id, resolved by resolve_decode_device() at transport // creation. c.frame_size = env_size("HOLOLINK_FRAME_SIZE", 384); c.page_size = env_size("HOLOLINK_PAGE_SIZE", 0); // 0 → derived below diff --git a/libs/qec/lib/realtime/decoding-server-cqr/GpuRoceTransceiver.h b/libs/qec/lib/realtime/decoding-server-cqr/GpuRoceTransceiver.h index 17b2078f6..89200f81a 100644 --- a/libs/qec/lib/realtime/decoding-server-cqr/GpuRoceTransceiver.h +++ b/libs/qec/lib/realtime/decoding-server-cqr/GpuRoceTransceiver.h @@ -39,7 +39,7 @@ struct GpuRoceConfig { std::string device_name; ///< HOLOLINK_DEVICE (IB netdev, e.g. "mlx5_0") uint32_t remote_qp{0}; ///< HOLOLINK_REMOTE_QP (FPGA/emulator QP number) int gpu_id{0}; ///< FPGA-affine GPU; set from the decoder's - ///< cuda_device_id by reconcile_gpu_roce_device() + ///< cuda_device_id by resolve_decode_device() size_t frame_size{384}; ///< HOLOLINK_FRAME_SIZE (max RPC frame bytes) size_t page_size{0}; ///< HOLOLINK_PAGE_SIZE (0 → derived from frame_size) size_t num_pages{64}; ///< HOLOLINK_NUM_PAGES (ring depth) diff --git a/libs/qec/unittests/test_decoding_server_core.cpp b/libs/qec/unittests/test_decoding_server_core.cpp index 7d9af4a8f..9fc9c5514 100644 --- a/libs/qec/unittests/test_decoding_server_core.cpp +++ b/libs/qec/unittests/test_decoding_server_core.cpp @@ -261,12 +261,12 @@ TEST(RpcDispatcherTest, ConvertsHandlerExceptionsToErrorResponses) { expect_status(transport, RpcStatus::INTERNAL_ERROR); } -TEST(GpuRoceDeviceReconcile, UnpinnedDefaultsToZero) { - EXPECT_EQ(cudaq::qec::decoding_server::reconcile_gpu_roce_device(-1), 0); +TEST(ResolveDecodeDevice, UnpinnedDefaultsToZero) { + EXPECT_EQ(cudaq::qec::decoding_server::resolve_decode_device(-1), 0); } -TEST(GpuRoceDeviceReconcile, PinSelectsDevice) { - EXPECT_EQ(cudaq::qec::decoding_server::reconcile_gpu_roce_device(3), 3); +TEST(ResolveDecodeDevice, PinSelectsDevice) { + EXPECT_EQ(cudaq::qec::decoding_server::resolve_decode_device(3), 3); } TEST(SetCudaDeviceForDecode, UnpinnedIsNoOp) { From a75365766b30f9d9a25274a995446dae943e211b Mon Sep 17 00:00:00 2001 From: Melody Ren Date: Tue, 14 Jul 2026 14:28:11 -0700 Subject: [PATCH 10/10] Drop the dead HOLOLINK_GPU_ID env from the HSB decoding-server harness Signed-off-by: Melody Ren --- libs/qec/unittests/utils/hsb_fpga_decoding_server_test.sh | 1 - 1 file changed, 1 deletion(-) diff --git a/libs/qec/unittests/utils/hsb_fpga_decoding_server_test.sh b/libs/qec/unittests/utils/hsb_fpga_decoding_server_test.sh index 91d6b9ba9..47cd35759 100755 --- a/libs/qec/unittests/utils/hsb_fpga_decoding_server_test.sh +++ b/libs/qec/unittests/utils/hsb_fpga_decoding_server_test.sh @@ -1009,7 +1009,6 @@ start_server() { HOLOLINK_REMOTE_QP="$((remote_qp))" \ HOLOLINK_FRAME_SIZE="$PAGE_SIZE" \ HOLOLINK_NUM_PAGES="$GPU_ROCE_NUM_PAGES" \ - HOLOLINK_GPU_ID="$GPU_ID" \ "$SERVER_BIN" \ --config="$CONFIG_FILE" \ --transport=gpu_roce \