diff --git a/docs/sphinx/examples_rst/qec/realtime_relay_bp.rst b/docs/sphinx/examples_rst/qec/realtime_relay_bp.rst index f680c0786..a9a8e1f59 100644 --- a/docs/sphinx/examples_rst/qec/realtime_relay_bp.rst +++ b/docs/sphinx/examples_rst/qec/realtime_relay_bp.rst @@ -8,10 +8,9 @@ Relay BP Decoding with CUDA-Q Realtime This guide explains how to build, test, and run the nv-qldpc-decoder Relay BP decoder using CUDA-Q's realtime dispatch system. The decoder is driven by a -**self-relaunching device-graph scheduler** and can operate in three +**self-relaunching device-graph scheduler** and can operate in two configurations: -- **CI unit test** -- standalone executable, no FPGA or network hardware needed - **Emulated end-to-end test** -- software FPGA emulator replaces real hardware - **FPGA end-to-end test** -- real FPGA connected via ConnectX RDMA/RoCE @@ -70,10 +69,6 @@ Hardware - GPU - ConnectX NIC - FPGA - * - CI unit test - - Any CUDA-capable GPU - - Not required - - Not required * - Emulated E2E - CUDA GPU with GPUDirect RDMA - Required (loopback cable) @@ -108,7 +103,7 @@ built from this (cudaqx) repository: needed at **build** time. It contains the ``enqueue_syndromes`` / ``get_corrections`` / ``reset_decoder`` ``DEVICE_CALL`` handlers (the device functions the scheduler dispatches). It is linked ``WHOLE_ARCHIVE`` and - device-linked into the bridge and the CI test, and is pointed at via the + device-linked into the bridge, and is pointed at via the ``-DCUDAQ_QEC_REALTIME_CUDEVICE_PROPRIETARY_ARCHIVE=`` CMake variable. Both artifacts come from the same closed-source decoder package; build the ``cudaq-qec-realtime-cudevice-proprietary`` target from the proprietary @@ -185,11 +180,6 @@ provides the ``GpuRoceTransceiver`` library for RDMA transport. fire-and-forget decode launch, and tail self-relaunch on top of the device-side graph dispatch). -.. note:: - - ``holoscan-sensor-bridge`` is only needed for the emulated and FPGA - end-to-end tests. The CI unit test requires only ``libcudaq-realtime``. - Repository Layout ----------------- @@ -200,8 +190,6 @@ Key files within ``cudaqx``: libs/qec/ unittests/ realtime/ - qec_graph_decode_test/ - test_realtime_qldpc_graph_decoding.cpp # CI unit test qec_roce_decode_test/ data/ config_nv_qldpc_relay.yml # Relay BP decoder config @@ -222,37 +210,7 @@ The FPGA emulator is in the ``cuda-quantum`` repository: Building -------- -CI unit test only (no HSB tools) -^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ - -If you only need to run the CI unit test, you can build without -``holoscan-sensor-bridge``: - -.. code-block:: bash - - # 1. Build libcudaq-realtime - git clone https://github.com/NVIDIA/cuda-quantum.git cudaq-realtime-src - cd cudaq-realtime-src - git checkout releases/v0.15.1 - cd realtime && mkdir -p build && cd build - cmake -G Ninja -DCMAKE_INSTALL_PREFIX=/tmp/cudaq-realtime .. - ninja && ninja install - cd ../../.. - - # 2. Build cudaqx with the nv-qldpc-decoder test. - # CUDAQ_QEC_REALTIME_CUDEVICE_PROPRIETARY_ARCHIVE points at the static - # archive with the DEVICE_CALL handlers; it is linked WHOLE_ARCHIVE into - # the test (see "Obtaining the proprietary components"). - cmake -S cudaqx -B cudaqx/build \ - -DCMAKE_BUILD_TYPE=Release \ - -DCUDAQ_DIR=/path/to/cudaq-install/lib/cmake/cudaq/ \ - -DCUDAQ_REALTIME_ROOT=/tmp/cudaq-realtime \ - -DCUDAQ_QEC_REALTIME_CUDEVICE_PROPRIETARY_ARCHIVE=/path/to/libcudaq-qec-realtime-cudevice-proprietary.a \ - -DCUDAQX_ENABLE_LIBS="qec" \ - -DCUDAQX_INCLUDE_TESTS=ON - cmake --build cudaqx/build --target test_realtime_qldpc_graph_decoding - -Full build (CI test + HSB bridge/playback tools) +Build the HSB bridge and playback tools ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ To also build the bridge and playback tools for emulated or FPGA testing: @@ -322,7 +280,6 @@ To also build the bridge and playback tools for emulated or FPGA testing: -DHOLOSCAN_SENSOR_BRIDGE_SOURCE_DIR=/path/to/holoscan-sensor-bridge \ -DHOLOSCAN_SENSOR_BRIDGE_BUILD_DIR=/path/to/holoscan-sensor-bridge/build cmake --build cudaqx/build --target \ - test_realtime_qldpc_graph_decoding \ gpu_roce_qldpc_graph_decoder_bridge \ hsb_fpga_syndrome_playback @@ -339,58 +296,16 @@ The orchestration script can build everything automatically: --cuda-quantum-dir /path/to/cuda-quantum \ --no-run -CI Unit Test ------------- - -The CI unit test (``test_realtime_qldpc_graph_decoding``) exercises the full -device-graph scheduler decode path without any network hardware. It: - -1. Loads the Relay BP config and syndrome data from YAML/text files -2. Creates the decoder via the ``decoder::get("nv-qldpc-decoder", ...)`` plugin API -3. Constructs a ``qec_realtime_session``, which captures the decoder's - device-launchable cooperative decode graph and starts the device-graph - scheduler on a pinned-mapped ring (3 ``DEVICE_CALL`` entries: - ``enqueue_syndromes`` / ``get_corrections`` / ``reset_decoder``) -4. Drives the per-round protocol with ``rpc_producer``: for each shot it sends - one ``enqueue_syndromes`` per round, then a ``get_corrections``; the - scheduler fires the decode when a window completes and tail self-relaunches -5. Verifies each shot's correction against the fixture, then a final - ``reset_decoder`` + ``get_corrections`` confirms reset - -Running -^^^^^^^ - -.. code-block:: bash - - cd cudaqx/build - - # The nv-qldpc-decoder plugin must be in /lib/decoder-plugins/ - # before running -- see "Obtaining the proprietary components" above. - - ./libs/qec/unittests/test_realtime_qldpc_graph_decoding - -Expected output: - -.. code-block:: text - - [==========] Running 1 test from 1 test suite. - [----------] 1 test from GraphDecodeTest - [ RUN ] GraphDecodeTest.DecodesAllSyndromes - ... - [ OK ] GraphDecodeTest.DecodesAllSyndromes (XXX ms) - [==========] 1 test from 1 test suite ran. - [ PASSED ] 1 test. - Surface Code Test (Relay BP) ---------------------------- The ``surface_code-1-local`` app example runs a surface code memory experiment with the nv-qldpc-decoder configured for Relay BP. It simulates a surface code -with ``stim`` and generates syndromes on the fly, so -- unlike the fixed-fixture -CI unit test -- it can run an arbitrary number of shots. +with ``stim`` and generates syndromes on the fly, so it can run an arbitrary +number of shots. Build the app example (it links the same plugin + proprietary archive as the -CI test): +bridge): .. code-block:: bash diff --git a/libs/qec/lib/realtime/CMakeLists.txt b/libs/qec/lib/realtime/CMakeLists.txt index e5479f01f..8073299a1 100644 --- a/libs/qec/lib/realtime/CMakeLists.txt +++ b/libs/qec/lib/realtime/CMakeLists.txt @@ -82,23 +82,11 @@ if(CMAKE_CUDA_COMPILER) get_filename_component(_CUDAQ_REALTIME_LIB_DIR "${CUDAQ_REALTIME_LIBRARY}" DIRECTORY) endif() - # The host-side dispatch loop (cudaq_host_dispatcher_loop), used by - # qec_realtime_session, lives in libcudaq-realtime-host-dispatch -- separate - # from libcudaq-realtime. Locate it so cudaq-qec-realtime-decoding can link - # it (mirrors the cudaq-realtime-pipeline target below). - cudaq_qec_find_realtime_library(CUDAQ_REALTIME_HOST_DISPATCH_LIBRARY - NAMES cudaq-realtime-host-dispatch - PATH_SUFFIXES lib - ) - if(CUDAQ_REALTIME_HOST_DISPATCH_LIBRARY) - message(STATUS "Found cuda-quantum realtime host dispatch library at ${CUDAQ_REALTIME_HOST_DISPATCH_LIBRARY}") - endif() - # The device-graph dispatch API (cudaq_create_dispatch_graph_regular, # cudaq_launch_dispatch_graph, cudaq_destroy_dispatch_graph) lives in # libcudaq-realtime-dispatch, separate from libcudaq-realtime. It is - # relocatable device code, and qec_realtime_session resolves it at runtime - # via dlsym(RTLD_DEFAULT, ...) from the host executable's absorbed copy -- + # relocatable device code, resolved at runtime via dlsym(RTLD_DEFAULT, ...) + # from the host executable's absorbed copy -- # it must NOT be linked into cudaq-qec-realtime-decoding.so: a second # device-linked copy of the dispatch kernel would live in a different CUDA # module than the decoder's DEVICE_CALL handlers (captured from the @@ -206,8 +194,6 @@ endif() add_library(cudaq-qec-realtime-decoding SHARED realtime_decoding.cpp config.cpp - qec_realtime_session.cpp - rpc_producer.cpp ) target_compile_options(cudaq-qec-realtime-decoding @@ -240,7 +226,6 @@ target_link_libraries(cudaq-qec-realtime-decoding LLVMSupport CUDA::cudart $<$:${CUDAQ_REALTIME_LIBRARY}> - $<$:${CUDAQ_REALTIME_HOST_DISPATCH_LIBRARY}> ) set_target_properties(cudaq-qec-realtime-decoding PROPERTIES diff --git a/libs/qec/lib/realtime/decoding-server-cqr/DeviceGraphRingConsumer.cpp b/libs/qec/lib/realtime/decoding-server-cqr/DeviceGraphRingConsumer.cpp index eaaadfe60..e75549191 100644 --- a/libs/qec/lib/realtime/decoding-server-cqr/DeviceGraphRingConsumer.cpp +++ b/libs/qec/lib/realtime/decoding-server-cqr/DeviceGraphRingConsumer.cpp @@ -147,8 +147,8 @@ DeviceGraphRingConsumer::DeviceGraphRingConsumer(const cudaq_ringbuffer_t &ring, // Resolve the dispatch graph API via dlsym; cudaq-realtime-dispatch is // linked into the process (not this library) to keep the CUDA module in - // one copy. Signatures must match create/launch/destroy_dispatch_graph_ - // fn_t in qec_realtime_session.cpp/.h exactly. + // one copy. Signatures must match cudaq-realtime's + // create/launch/destroy_dispatch_graph exports exactly. using create_fn_t = cudaError_t (*)( volatile std::uint64_t *, volatile std::uint64_t *, std::uint8_t *, std::uint8_t *, std::size_t, std::size_t, cudaq_function_entry_t *, diff --git a/libs/qec/lib/realtime/qec_realtime_session.cpp b/libs/qec/lib/realtime/qec_realtime_session.cpp deleted file mode 100644 index e033bec09..000000000 --- a/libs/qec/lib/realtime/qec_realtime_session.cpp +++ /dev/null @@ -1,1012 +0,0 @@ -/******************************************************************************* - * Copyright (c) 2024 - 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. * - ******************************************************************************/ - -#ifdef CUDAQ_REALTIME_ROOT - -#include "qec_realtime_session.h" - -#include "../hardware_guards.h" - -#include "cudaq/qec/logger.h" -#include "cudaq/qec/realtime/decoder_rpc_wire_format.h" -#include "cudaq/qec/realtime/graph_resources.h" -#include "cudaq/realtime/daemon/dispatcher/dispatch_kernel_launch.h" - -#include -#include -#include -#include -#include -#include -#include -#include - -namespace cudaq::qec::realtime { - -namespace { - -using Decoders = std::vector>; - -//============================================================================== -// DEVICE-mode helpers -//============================================================================== - -// Resolves a host-side C-ABI shim `cudaqx_qec_realtime_dispatch_populate_*` -// at runtime via dlsym(RTLD_DEFAULT, ...). These shims are defined in -// libcudaq-qec-realtime-cudevice.a and only enter the process when the final -// executable absorbs that static archive (typically via the -// `qec_realtime_app_link_options()` CMake helper). Resolving by name rather -// than by direct symbol reference keeps libcudaq-qec-realtime-decoding.so free -// of unresolved C-ABI symbols, so it can be safely dlopen'd from consumers that -// do NOT link the cudevice archive (notably the Python extension). Any such -// consumer that tries to actually USE the device dispatch path lands here, does -// not find the symbol, and surfaces a clean runtime_error with actionable -// linker guidance. -using populate_device_entry_fn = void (*)(void *); -populate_device_entry_fn resolve_populate_shim(const char *symbol_name) { - void *sym = ::dlsym(RTLD_DEFAULT, symbol_name); - return reinterpret_cast(sym); -} - -// The device-graph dispatch API (cudaq_create_dispatch_graph_regular / -// cudaq_launch_dispatch_graph / cudaq_destroy_dispatch_graph) lives in -// libcudaq-realtime-dispatch.a -- relocatable device code the host executable -// absorbs and device-links (see qec_realtime_app_link_options()). It must be -// resolved from the SAME image as the populate shims above: the dispatch -// kernel invokes the decoder's __device__ function pointers captured by those -// shims, and a device function pointer is only valid inside the CUDA module -// (device-link unit) that produced it. If this .so carried its own copy of -// the archive, its dispatch kernel would live in a different CUDA module than -// the executable's decoder handlers and the DEVICE_CALL would trap with -// cudaErrorIllegalInstruction. Resolving via dlsym(RTLD_DEFAULT, ...) binds -// to the executable's copy (exported via --export-dynamic + -// CUDAQ_REALTIME_DISPATCH_API default visibility), keeping kernel and handlers -// in one module -- and keeps this .so free of undefined C-ABI symbols for -// consumers that never touch the device path. -using create_dispatch_graph_fn_t = cudaError_t (*)( - volatile std::uint64_t *, volatile std::uint64_t *, std::uint8_t *, - std::uint8_t *, std::size_t, std::size_t, cudaq_function_entry_t *, - std::size_t, void *, volatile int *, std::uint64_t *, std::size_t, - std::uint32_t, std::uint32_t, cudaGraphExec_t, cudaStream_t, - cudaq_dispatch_graph_context **); -using launch_dispatch_graph_fn_t = - cudaError_t (*)(cudaq_dispatch_graph_context *, cudaStream_t); - -template -FnT resolve_dispatch_graph_api(const char *symbol_name) { - void *sym = ::dlsym(RTLD_DEFAULT, symbol_name); - if (!sym) - throw std::runtime_error( - std::string("qec_realtime_session::initialize: ") + symbol_name + - " not found via dlsym(RTLD_DEFAULT, ...). The host executable must " - "absorb libcudaq-realtime-dispatch.a and link with --export-dynamic " - "(see qec_realtime_app_link_options())."); - return reinterpret_cast(sym); -} - -// Pinned mapped flags + pinned mapped data, with the device pointer obtained -// via UVA so the GPU dispatcher can read the same backing. -bool allocate_pinned_mapped(std::size_t bytes, void **host_out, - void **device_out) { - void *h = nullptr; - if (cudaHostAlloc(&h, bytes, cudaHostAllocMapped) != cudaSuccess) - return false; - void *d = nullptr; - if (cudaHostGetDevicePointer(&d, h, 0) != cudaSuccess) { - cudaFreeHost(h); - return false; - } - std::memset(h, 0, bytes); - *host_out = h; - *device_out = d; - return true; -} - -//============================================================================== -// HOST-mode helpers (two-ring CUDAQ_DISPATCH_HOST_CALL handlers) -//============================================================================== - -namespace rpc = cudaq::qec::decoding::rpc; - -// The HOST_CALL handlers below are plain C function pointers with no -// user-context argument, so the active decoder table must be reachable from a -// process-global. It is published/cleared by a HOST-mode initialize()/ -// finalize() and read concurrently by the host dispatcher thread, hence the -// atomic. Only one HOST-mode session may be live at a time (enforced in -// initialize()). -std::atomic g_active_decoders{nullptr}; - -// Throws std::out_of_range for an unknown decoder_id so the handlers below can -// map it to RpcStatus::INVALID_DECODER, mirroring SessionRegistry::get() / -// RpcDispatcher on the network-transport path. -cudaq::qec::decoder *get_decoder_or_throw(std::int64_t decoder_id) { - Decoders *decoders = g_active_decoders.load(std::memory_order_acquire); - if (!decoders || decoder_id < 0 || - static_cast(decoder_id) >= decoders->size() || - !(*decoders)[static_cast(decoder_id)]) - throw std::out_of_range("invalid decoder_id " + std::to_string(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; - // 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::pin_decode_device(*dec); -} - -// 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 -// caller (cudaq_host_dispatcher_loop::handle_host_call) publishes tx_flags -// AFTER the handler returns; the handler only needs to write the response body -// + header and release-store the magic. -void write_response(void *tx_slot, const void *rx_slot, rpc::RpcStatus status, - std::uint32_t result_len = 0) { - const auto *request = - static_cast(rx_slot); - auto *response = static_cast(tx_slot); - response->status = static_cast(status); - response->result_len = result_len; - response->request_id = request->request_id; - response->ptp_timestamp = request->ptp_timestamp; - __atomic_store_n(&response->magic, cudaq::realtime::RPC_MAGIC_RESPONSE, - __ATOMIC_RELEASE); -} - -std::uint8_t *response_body(void *tx_slot) { - return static_cast(tx_slot) + - sizeof(cudaq::realtime::RPCResponse); -} - -void enqueue_syndromes_host(const void *rx_slot, void *tx_slot, - std::size_t slot_size) { - try { - const auto *header = - static_cast(rx_slot); - if (header->arg_len < sizeof(rpc::EnqueueRequestPayload)) { - write_response(tx_slot, rx_slot, rpc::RpcStatus::BAD_REQUEST); - return; - } - const auto *body = reinterpret_cast( - static_cast(rx_slot) + - sizeof(cudaq::realtime::RPCHeader)); - if (body->num_syndromes < 0 || body->syndrome_mapping_id != 0) { - write_response(tx_slot, rx_slot, rpc::RpcStatus::BAD_REQUEST); - return; - } - const auto num_syndromes = static_cast(body->num_syndromes); - const std::size_t expected_arg_len = sizeof(rpc::EnqueueRequestPayload) + - rpc::bit_packed_bytes(num_syndromes); - if (header->arg_len != expected_arg_len || - sizeof(cudaq::realtime::RPCHeader) + expected_arg_len > slot_size) { - write_response(tx_slot, rx_slot, rpc::RpcStatus::BAD_REQUEST); - return; - } - - 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 - // overflow the decoder's accumulation buffer and be silently dropped by - // enqueue_syndrome (which returns false) while we ACK success. - if (num_syndromes > decoder->get_num_msyn_per_decode()) { - write_response(tx_slot, rx_slot, rpc::RpcStatus::BAD_REQUEST); - return; - } - const std::uint8_t *bits = reinterpret_cast(body + 1); - std::vector syndromes(num_syndromes, 0); - for (std::uint64_t bit = 0; bit < num_syndromes; ++bit) - syndromes[bit] = (bits[bit >> 3] >> (bit & 7)) & 0x1u; - - // enqueue_syndrome's bool return means "a decode was triggered", not - // success, so it is intentionally not treated as an error here; the - // oversize guard above is what rejects malformed lengths. - (void)decoder->enqueue_syndrome(syndromes.data(), syndromes.size()); - write_response(tx_slot, rx_slot, rpc::RpcStatus::OK); - } catch (const std::out_of_range &) { - write_response(tx_slot, rx_slot, rpc::RpcStatus::INVALID_DECODER); - } catch (...) { - write_response(tx_slot, rx_slot, rpc::RpcStatus::INTERNAL_ERROR); - } -} - -void get_corrections_host(const void *rx_slot, void *tx_slot, - std::size_t slot_size) { - try { - const auto *header = - static_cast(rx_slot); - if (header->arg_len != sizeof(rpc::GetCorrectionsRequestPayload)) { - write_response(tx_slot, rx_slot, rpc::RpcStatus::BAD_REQUEST); - return; - } - const auto *body = - reinterpret_cast( - static_cast(rx_slot) + - sizeof(cudaq::realtime::RPCHeader)); - if (body->return_size < 0) { - write_response(tx_slot, rx_slot, rpc::RpcStatus::BAD_REQUEST); - return; - } - - auto *decoder = get_decoder_or_throw(body->decoder_id); - const auto return_size = static_cast(body->return_size); - if (return_size > decoder->get_num_observables()) { - write_response(tx_slot, rx_slot, rpc::RpcStatus::BAD_REQUEST); - return; - } - // result_len = ceil(R/8) exactly per decoder_server_runtime.md (no pad). - const std::size_t result_len = rpc::bit_packed_bytes(return_size); - if (sizeof(cudaq::realtime::RPCResponse) + result_len > slot_size) { - write_response(tx_slot, rx_slot, rpc::RpcStatus::BAD_REQUEST); - return; - } - - std::uint8_t *out = response_body(tx_slot); - std::memset(out, 0, result_len); - const std::uint8_t *corrections = decoder->get_obs_corrections(); - for (std::uint64_t i = 0; i < return_size; ++i) { - if (corrections[i] & 0x1u) - out[i >> 3] |= static_cast(1u << (i & 7)); - } - if (body->reset != 0) - decoder->clear_corrections(); - write_response(tx_slot, rx_slot, rpc::RpcStatus::OK, - static_cast(result_len)); - } catch (const std::out_of_range &) { - write_response(tx_slot, rx_slot, rpc::RpcStatus::INVALID_DECODER); - } catch (...) { - write_response(tx_slot, rx_slot, rpc::RpcStatus::INTERNAL_ERROR); - } -} - -void reset_decoder_host(const void *rx_slot, void *tx_slot, std::size_t) { - try { - const auto *header = - static_cast(rx_slot); - if (header->arg_len != sizeof(rpc::ResetRequestPayload)) { - write_response(tx_slot, rx_slot, rpc::RpcStatus::BAD_REQUEST); - return; - } - const auto *body = reinterpret_cast( - static_cast(rx_slot) + - sizeof(cudaq::realtime::RPCHeader)); - auto *decoder = get_decoder_or_throw(body->decoder_id); - apply_decoder_cuda_device(decoder); - decoder->reset_decoder(); - write_response(tx_slot, rx_slot, rpc::RpcStatus::OK); - } catch (const std::out_of_range &) { - write_response(tx_slot, rx_slot, rpc::RpcStatus::INVALID_DECODER); - } catch (...) { - write_response(tx_slot, rx_slot, rpc::RpcStatus::INTERNAL_ERROR); - } -} - -} // namespace - -//============================================================================== -// ctor / dtor -//============================================================================== - -qec_realtime_session::qec_realtime_session( - std::vector> &decoders, - cudaq_dispatch_launch_fn_t device_launch_fn) - : decoders_(decoders), device_launch_fn_(device_launch_fn) {} - -qec_realtime_session::~qec_realtime_session() { - // Best-effort teardown. finalize() is null-safe at every step (each resource - // has its own guard), so calling it from a never-fully-initialized or - // already-finalized session is a no-op beyond the trace message. - finalize(); -} - -//============================================================================== -// classify_mode() -//============================================================================== - -void qec_realtime_session::classify_mode() { - bool any_graph = false; - bool any_host = false; - std::size_t non_null = 0; - for (auto &decoder : decoders_) { - if (!decoder) - continue; - ++non_null; - if (decoder->supports_graph_dispatch()) - any_graph = true; - else - any_host = true; - } - - if (non_null == 0) - throw std::runtime_error( - "qec_realtime_session::initialize: no (non-null) decoders to serve"); - - if (any_graph && any_host) - throw std::runtime_error( - "qec_realtime_session::initialize: mixed decoder set -- some decoders " - "support graph dispatch (DEVICE mode) and some do not (HOST mode). A " - "single session must be homogeneous: the host loop resolves a slot to " - "a function table entry by function_id alone, so a GRAPH_LAUNCH and a " - "HOST_CALL enqueue sharing kEnqueueSyndromesFunctionId would collide. " - "Use one decoder per session (or a homogeneous decoder set)."); - - device_mode_ = any_graph; - - if (device_mode_ && !device_launch_fn_) - throw std::runtime_error( - "qec_realtime_session::initialize: DEVICE mode requires a non-null " - "device_launch_fn (typically &cudaq_launch_dispatch_kernel_regular " - "from libcudaq-realtime-dispatch), but the session was constructed " - "without one."); -} - -//============================================================================== -// initialize() -//============================================================================== - -void qec_realtime_session::initialize() { - if (initialized_) - return; - - classify_mode(); - - // Reset the monotonic producer cursor so it starts in lockstep with the - // strict-FIFO consumer (both begin at slot 0). - producer_cursor_ = 0; - - if (device_mode_) { - // The device-graph scheduler relies on device-side graph launch, which - // requires compute capability 9.0+ (Hopper). Below sm_90 the dispatch - // kernel's TRIGGER_GRAPH interception is compiled out and the enqueue RPC - // would surface the raw sentinel (0x12A6E5) as a non-zero status -- fail - // fast with an actionable message instead. - { - int device = 0; - cudaDeviceProp prop{}; - if (cudaGetDevice(&device) != cudaSuccess || - cudaGetDeviceProperties(&prop, device) != cudaSuccess) - throw std::runtime_error( - "qec_realtime_session::initialize: unable to query the active " - "CUDA device for the DEVICE-mode compute-capability check"); - if (prop.major < 9) - throw std::runtime_error( - "qec_realtime_session::initialize: DEVICE mode (the device-graph " - "scheduler) requires device-side graph launch, i.e. compute " - "capability 9.0+ (Hopper); found " + - std::to_string(prop.major) + "." + std::to_string(prop.minor) + - ". Use a CPU decoder (HOST mode) or a Hopper+ GPU."); - } - - // Be tolerant of being called before any CUDA setup. The pinned - // allocations below require cudaDeviceMapHost on the active device. - cudaError_t flags_err = cudaSetDeviceFlags(cudaDeviceMapHost); - if (flags_err != cudaSuccess && flags_err != cudaErrorSetOnActiveProcess) - throw std::runtime_error( - std::string("qec_realtime_session::initialize: " - "cudaSetDeviceFlags(cudaDeviceMapHost) failed: ") + - cudaGetErrorString(flags_err)); - } else { - // HOST mode: claim the process-global decoder table up front. With a - // single global, a second HOST-mode initialize() would hijack decoder-id - // resolution for the first session's host loop. - Decoders *expected = nullptr; - if (!g_active_decoders.compare_exchange_strong(expected, &decoders_, - std::memory_order_acq_rel)) - throw std::runtime_error( - "qec_realtime_session::initialize: another HOST_CALL session is " - "already active; concurrent HOST-mode sessions are not supported"); - } - - // Everything below acquires resources, so it must be transactional. - // finalize() is null-safe at every step, so we can roll a half-built session - // back from any throw. - try { - if (device_mode_) - capture_decoder_graphs(); - allocate_ring_buffer(); - populate_function_table(); - if (device_mode_) - start_device_loop(); // self-relaunching device-graph scheduler - else - start_host_loop(); - initialized_ = true; - } catch (...) { - CUDA_QEC_WARN("qec_realtime_session::initialize: rolling back partial " - "initialization after exception"); - finalize(); - throw; - } - - if (device_mode_) - CUDA_QEC_INFO("qec_realtime_session: initialized DEVICE mode " - "(num_decoders_with_graph={}, num_slots={}, slot_size={})", - num_decoders_with_graph_, num_slots_, slot_size_); - else - CUDA_QEC_INFO("qec_realtime_session: initialized HOST mode " - "(num_slots={}, slot_size={})", - num_slots_, slot_size_); -} - -//============================================================================== -// finalize() -//============================================================================== - -void qec_realtime_session::finalize() { - const bool was_initialized = initialized_; - if (was_initialized) - initialized_ = false; - // Note: we intentionally do NOT early-return on !was_initialized; the - // initialize() rollback path depends on running through the cleanup below. - - stop_loops(); - - if (device_mode_) { - // After stop_loops() the host monitor thread is joined and the persistent - // device kernel has been signalled to exit, but in-flight worker-stream - // graph launches submitted before the join can still be running. - // cudaDeviceSynchronize() drains all outstanding work, so the subsequent - // release_decode_graph() can't free buffers a still-running enqueue graph - // dereferences. finalize() is cold-path so the sync cost is irrelevant. - cudaDeviceSynchronize(); - - for (std::size_t i = 0; i < captured_graphs_.size(); ++i) { - if (captured_graphs_[i] && i < decoders_.size() && decoders_[i]) - decoders_[i]->release_decode_graph(captured_graphs_[i]); - } - } - captured_graphs_.clear(); - num_decoders_with_graph_ = 0; - - // Release the HOST-mode global if this session owns it (no-op otherwise). - Decoders *self = &decoders_; - g_active_decoders.compare_exchange_strong(self, nullptr, - std::memory_order_acq_rel); - - // Free the function table with the allocator that matches the mode it was - // built with. - if (function_table_host_) { - if (device_mode_) - cudaFreeHost(function_table_host_); - else - std::free(function_table_host_); - function_table_host_ = nullptr; - function_table_dev_ = nullptr; - } - function_table_count_ = 0; - get_corrections_fn_id_ = 0; - reset_decoder_fn_id_ = 0; - - if (device_stats_dev_) { - cudaFree(device_stats_dev_); - device_stats_dev_ = nullptr; - } - - // Free the ring. DEVICE mode allocated pinned-mapped backings; HOST mode - // allocated plain host memory with _dev aliasing _host (free _host only). - auto free_ring_u64 = [&](volatile std::uint64_t *&host, - volatile std::uint64_t *&dev) { - if (host) { - if (device_mode_) - cudaFreeHost(const_cast(host)); - else - std::free(const_cast(host)); - } - host = nullptr; - dev = nullptr; - }; - auto free_ring_u8 = [&](std::uint8_t *&host, std::uint8_t *&dev) { - if (host) { - if (device_mode_) - cudaFreeHost(host); - else - std::free(host); - } - host = nullptr; - dev = nullptr; - }; - free_ring_u64(tx_flags_host_, tx_flags_dev_); - free_ring_u64(rx_flags_host_, rx_flags_dev_); - free_ring_u8(tx_data_host_, tx_data_dev_); - free_ring_u8(rx_data_host_, rx_data_dev_); - - if (shutdown_flag_host_) { - cudaFreeHost(shutdown_flag_host_); - shutdown_flag_host_ = nullptr; - shutdown_flag_dev_ = nullptr; - } - - std::memset(&ringbuffer_, 0, sizeof(ringbuffer_)); - std::memset(&host_table_, 0, sizeof(host_table_)); - std::memset(&host_config_, 0, sizeof(host_config_)); - host_engine_ = nullptr; - - if (was_initialized) - CUDA_QEC_INFO("qec_realtime_session: finalized"); -} - -//============================================================================== -// capture_decoder_graphs() [DEVICE mode] -//============================================================================== - -void qec_realtime_session::capture_decoder_graphs() { - captured_graphs_.assign(decoders_.size(), nullptr); - num_decoders_with_graph_ = 0; - - // kMaxDispatchedDecoders sizes the device-side g_decoder_state_table[]. - if (decoders_.size() > cudaq::qec::decoding::rpc::kMaxDispatchedDecoders) - throw std::runtime_error( - "qec_realtime_session::initialize: requested " + - std::to_string(decoders_.size()) + - " decoders but the realtime dispatch supports at most " + - std::to_string(cudaq::qec::decoding::rpc::kMaxDispatchedDecoders) + - " (kMaxDispatchedDecoders)."); - - for (std::size_t i = 0; i < decoders_.size(); ++i) { - auto *dec = decoders_[i].get(); - if (!dec) - continue; - if (!dec->supports_graph_dispatch()) - throw std::runtime_error( - "qec_realtime_session::initialize: decoder " + std::to_string(i) + - " does not support graph dispatch in DEVICE mode."); - - // reserved_sms = 0 is intentional for the inproc_rpc desktop / CI path. - void *raw = - cudaq::qec::detail_affinity::capture_graph_pinned(*dec, /*sms=*/0); - if (!raw) - throw std::runtime_error("qec_realtime_session::initialize: decoder " + - std::to_string(i) + - " returned null from capture_decode_graph()"); - captured_graphs_[i] = raw; - - auto *gres = static_cast(raw); - if (!gres->graph_exec || !gres->function_id) - throw std::runtime_error( - "qec_realtime_session::initialize: decoder " + std::to_string(i) + - " produced incomplete graph_resources (graph_exec / function_id)"); - - // All N enqueue_syndromes graphs share a single canonical function_id; the - // host monitor disambiguates per-decoder via routing_key == decoder_id. - if (gres->function_id != - cudaq::qec::decoding::rpc::kEnqueueSyndromesFunctionId) - throw std::runtime_error( - "qec_realtime_session::initialize: decoder " + std::to_string(i) + - " published a non-canonical enqueue function_id"); - - ++num_decoders_with_graph_; - } - - if (num_decoders_with_graph_ == 0) - throw std::runtime_error( - "qec_realtime_session::initialize: no decoders to capture graphs for"); -} - -//============================================================================== -// allocate_ring_buffer() [branches on device_mode_] -//============================================================================== - -void qec_realtime_session::allocate_ring_buffer() { - namespace rpc = cudaq::qec::decoding::rpc; - using cudaq::realtime::RPCHeader; - using cudaq::realtime::RPCResponse; - - // Slot size: largest body across the RPC trio, over all served decoders. - std::size_t max_measurements = 0; - std::size_t max_observables = 0; - for (std::size_t i = 0; i < decoders_.size(); ++i) { - auto *dec = decoders_[i].get(); - if (!dec) - continue; - // DEVICE mode only sizes for decoders that captured a graph. - if (device_mode_ && !captured_graphs_[i]) - continue; - max_measurements = - std::max(max_measurements, dec->get_num_msyn_per_decode()); - max_observables = - std::max(max_observables, dec->get_num_observables()); - } - - const std::size_t enqueue_req = sizeof(RPCHeader) + - sizeof(rpc::EnqueueRequestPayload) + - rpc::bit_packed_bytes(max_measurements); - const std::size_t get_req = - sizeof(RPCHeader) + sizeof(rpc::GetCorrectionsRequestPayload); - const std::size_t reset_req = - sizeof(RPCHeader) + sizeof(rpc::ResetRequestPayload); - const std::size_t enqueue_resp = sizeof(RPCResponse); - const std::size_t get_resp = - sizeof(RPCResponse) + rpc::bit_packed_bytes(max_observables); - const std::size_t reset_resp = sizeof(RPCResponse); - - slot_size_ = std::max({enqueue_req, get_req, reset_req, enqueue_resp, - get_resp, reset_resp, std::size_t{64}}); - - // Round the slot stride up to an alignment boundary so every slot -- and thus - // every RPCHeader/RPCResponse placed at slot offset i*slot_size_ -- starts - // aligned. write_response() (and the producer) perform a 4-byte __atomic RMW - // on the header/response `magic` field at slot offset 0; with an unaligned - // stride (e.g. slot_size_ == 75 at distance 5) that atomic lands on an - // unaligned address, which the CPU splits into a bus-locked, cache-line- - // crossing access -- fatal SIGBUS on hosts with split-lock detection enabled - // (common on cloud / CI runners, tolerated silently elsewhere). A distance-3 - // config happens to floor at the aligned 64-byte minimum and so never hit - // this. DEVICE mode additionally wants a 256-byte stride for deterministic - // GPU-visible slot addressing; HOST mode only needs the atomics aligned. - { - constexpr std::size_t kDeviceSlotAlignment = 256; - constexpr std::size_t kHostSlotAlignment = 16; - const std::size_t alignment = - device_mode_ ? kDeviceSlotAlignment : kHostSlotAlignment; - slot_size_ = (slot_size_ + (alignment - 1)) & ~(alignment - 1); - } - - if (device_mode_) { - auto alloc_u64 = [&](volatile std::uint64_t *&host, - volatile std::uint64_t *&dev, const char *what) { - void *h = nullptr; - void *d = nullptr; - if (!allocate_pinned_mapped(num_slots_ * sizeof(std::uint64_t), &h, &d)) - throw std::runtime_error( - std::string( - "qec_realtime_session::initialize: failed to allocate ") + - what); - host = static_cast(h); - dev = static_cast(d); - }; - auto alloc_u8 = [&](std::uint8_t *&host, std::uint8_t *&dev, - const char *what) { - void *h = nullptr; - void *d = nullptr; - if (!allocate_pinned_mapped(num_slots_ * slot_size_, &h, &d)) - throw std::runtime_error( - std::string( - "qec_realtime_session::initialize: failed to allocate ") + - what); - host = static_cast(h); - dev = static_cast(d); - }; - alloc_u64(rx_flags_host_, rx_flags_dev_, "rx_flags"); - alloc_u64(tx_flags_host_, tx_flags_dev_, "tx_flags"); - alloc_u8(rx_data_host_, rx_data_dev_, "RX ring data"); - alloc_u8(tx_data_host_, tx_data_dev_, "TX ring data"); - - { - void *h = nullptr; - void *d = nullptr; - if (!allocate_pinned_mapped(sizeof(int), &h, &d)) - throw std::runtime_error("qec_realtime_session::initialize: failed to " - "allocate shutdown flag"); - shutdown_flag_host_ = static_cast(h); - *shutdown_flag_host_ = 0; - shutdown_flag_dev_ = static_cast(d); - } - } else { - // HOST mode: plain host memory; the device-visible pointers alias the host - // backings (no GPU required at runtime). The host loop reads only the - // *_host views; the producer's address-as-flag publish uses rx_data_dev() - // (== rx_data_host_ here), which the host loop dereferences as host memory. - auto alloc_u64 = [&](volatile std::uint64_t *&host, - volatile std::uint64_t *&dev, const char *what) { - void *p = std::calloc(num_slots_, sizeof(std::uint64_t)); - if (!p) - throw std::runtime_error( - std::string( - "qec_realtime_session::initialize: failed to allocate ") + - what); - host = static_cast(p); - dev = host; - }; - auto alloc_u8 = [&](std::uint8_t *&host, std::uint8_t *&dev, - const char *what) { - void *p = std::calloc(num_slots_, slot_size_); - if (!p) - throw std::runtime_error( - std::string( - "qec_realtime_session::initialize: failed to allocate ") + - what); - host = static_cast(p); - dev = host; - }; - alloc_u64(rx_flags_host_, rx_flags_dev_, "rx_flags"); - alloc_u64(tx_flags_host_, tx_flags_dev_, "tx_flags"); - alloc_u8(rx_data_host_, rx_data_dev_, "RX ring data"); - alloc_u8(tx_data_host_, tx_data_dev_, "TX ring data"); - } - - std::memset(&ringbuffer_, 0, sizeof(ringbuffer_)); - ringbuffer_.rx_flags = rx_flags_dev_; - ringbuffer_.tx_flags = tx_flags_dev_; - ringbuffer_.rx_data = rx_data_dev_; - ringbuffer_.tx_data = tx_data_dev_; - ringbuffer_.rx_stride_sz = slot_size_; - ringbuffer_.tx_stride_sz = slot_size_; - ringbuffer_.rx_flags_host = rx_flags_host_; - ringbuffer_.tx_flags_host = tx_flags_host_; - ringbuffer_.rx_data_host = rx_data_host_; - ringbuffer_.tx_data_host = tx_data_host_; -} - -//============================================================================== -// populate_function_table() [branches on device_mode_] -//============================================================================== - -void qec_realtime_session::populate_function_table() { - namespace rpc = cudaq::qec::decoding::rpc; - - if (!device_mode_) { - // HOST mode: 3 HOST_CALL entries (enqueue, get_corrections, reset). Plain - // host allocation -- host_fn pointers are host code addresses; _dev aliases - // _host. decoder_id routing happens inside each handler via the payload. - function_table_count_ = 3; - void *p = - std::calloc(function_table_count_, sizeof(cudaq_function_entry_t)); - if (!p) - throw std::runtime_error("qec_realtime_session::initialize: failed to " - "allocate function table"); - function_table_host_ = static_cast(p); - function_table_dev_ = function_table_host_; - - function_table_host_[0].handler.host_fn = enqueue_syndromes_host; - function_table_host_[0].function_id = rpc::kEnqueueSyndromesFunctionId; - function_table_host_[0].dispatch_mode = CUDAQ_DISPATCH_HOST_CALL; - - function_table_host_[1].handler.host_fn = get_corrections_host; - function_table_host_[1].function_id = rpc::kGetCorrectionsFunctionId; - function_table_host_[1].dispatch_mode = CUDAQ_DISPATCH_HOST_CALL; - - function_table_host_[2].handler.host_fn = reset_decoder_host; - function_table_host_[2].function_id = rpc::kResetDecoderFunctionId; - function_table_host_[2].dispatch_mode = CUDAQ_DISPATCH_HOST_CALL; - - get_corrections_fn_id_ = rpc::kGetCorrectionsFunctionId; - reset_decoder_fn_id_ = rpc::kResetDecoderFunctionId; - return; - } - - // DEVICE mode: 3 DEVICE_CALL entries (enqueue_syndromes accumulate, - // get_corrections, reset_decoder), all serviced by the self-relaunching - // device-graph scheduler. enqueue_syndromes is an accumulate handler (NOT a - // GRAPH_LAUNCH): it appends the round's syndromes into the registered - // GpuDecoderState and returns CUDAQ_DISPATCH_STATUS_TRIGGER_GRAPH when a full - // window has accumulated, which tells the scheduler to fire the per-decoder - // device-launchable decode graph fire-and-forget. All three handlers live - // in the cudevice archive and are resolved by name via dlsym so a .so that - // did not link that archive can still load. Pinned-mapped so the scheduler - // kernel reads the same backing. - function_table_count_ = 3; - - void *h = nullptr; - void *d = nullptr; - if (!allocate_pinned_mapped( - function_table_count_ * sizeof(cudaq_function_entry_t), &h, &d)) - throw std::runtime_error( - "qec_realtime_session::initialize: failed to allocate function table"); - function_table_host_ = static_cast(h); - function_table_dev_ = static_cast(d); - - // Resolve a DEVICE_CALL populate shim by name, invoke it on the target entry, - // stamp the function_id/routing_key, and validate it produced a real - // DEVICE_CALL handler. routing_key is unused by the scheduler (single - // decoder) but kept 0 for forward compatibility with per-decoder routing. - auto populate_device_call = [&](std::size_t slot, const char *symbol, - std::uint32_t function_id) { - auto shim = resolve_populate_shim(symbol); - if (!shim) - throw std::runtime_error( - std::string("qec_realtime_session::initialize: ") + symbol + - " not found via dlsym(RTLD_DEFAULT, ...). The final binary must " - "link libcudaq-qec-realtime-cudevice.a (or the static parts of " - "decoder_rpc_dispatch.cu via qec_realtime_app_link_options())."); - shim(&function_table_host_[slot]); - function_table_host_[slot].function_id = function_id; - function_table_host_[slot].routing_key = 0; - if (function_table_host_[slot].dispatch_mode != - CUDAQ_DISPATCH_DEVICE_CALL || - !function_table_host_[slot].handler.device_fn_ptr) - throw std::runtime_error( - std::string("qec_realtime_session::initialize: ") + symbol + - " did not produce a valid DEVICE_CALL entry (plugin bug)"); - }; - - // [0] enqueue_syndromes accumulate. - populate_device_call( - 0, "cudaqx_qec_realtime_dispatch_populate_enqueue_syndromes_device_entry", - rpc::kEnqueueSyndromesFunctionId); - // [1] get_corrections. - get_corrections_fn_id_ = rpc::kGetCorrectionsFunctionId; - populate_device_call( - 1, "cudaqx_qec_realtime_dispatch_populate_get_corrections_device_entry", - get_corrections_fn_id_); - // [2] reset_decoder. - reset_decoder_fn_id_ = rpc::kResetDecoderFunctionId; - populate_device_call( - 2, "cudaqx_qec_realtime_dispatch_populate_reset_decoder_device_entry", - reset_decoder_fn_id_); -} - -//============================================================================== -// start_device_loop() [DEVICE mode] -//============================================================================== - -void qec_realtime_session::start_device_loop() { - // The scheduler fires a single per-decoder decode graph fire-and-forget when - // the enqueue accumulate handler signals a full window. The public host - // dispatcher exposes one triggered graph per scheduler, so DEVICE mode is - // scoped to a single graph-dispatch decoder (matching the prior single- - // mailbox-bank limitation). A multi-decoder scheduler would need a - // function-id/routing-key -> triggered-graph map inside the dispatch kernel. - if (num_decoders_with_graph_ != 1) - throw std::runtime_error( - "qec_realtime_session::initialize: the device-graph scheduler supports " - "exactly one graph-dispatch decoder per session (got " + - std::to_string(num_decoders_with_graph_) + ")."); - - // The lone captured decoder's device-launchable decode graph. - cudaGraphExec_t decode_graph_exec = nullptr; - for (std::size_t i = 0; i < captured_graphs_.size(); ++i) { - if (!captured_graphs_[i]) - continue; - auto *gres = static_cast( - captured_graphs_[i]); - decode_graph_exec = gres->graph_exec; - break; - } - if (!decode_graph_exec) - throw std::runtime_error( - "qec_realtime_session::initialize: no captured decode graph for the " - "scheduler to trigger"); - - if (cudaMalloc(&device_stats_dev_, sizeof(std::uint64_t)) != cudaSuccess || - cudaMemset(device_stats_dev_, 0, sizeof(std::uint64_t)) != cudaSuccess) - throw std::runtime_error( - "qec_realtime_session::initialize: device_stats_dev allocation failed"); - - if (cudaStreamCreate(&scheduler_stream_) != cudaSuccess) - throw std::runtime_error( - "qec_realtime_session::initialize: cudaStreamCreate for the scheduler " - "stream failed"); - - // Bind the graph dispatch API from the host executable's device-link unit - // (NOT from any copy this .so might carry) so the dispatch kernel and the - // decoder's DEVICE_CALL handlers share one CUDA module -- see - // resolve_dispatch_graph_api(). The destroy fn is stashed for stop_loops(). - auto create_fn = resolve_dispatch_graph_api( - "cudaq_create_dispatch_graph_regular"); - auto launch_fn = resolve_dispatch_graph_api( - "cudaq_launch_dispatch_graph"); - destroy_dispatch_graph_fn_ = - resolve_dispatch_graph_api( - "cudaq_destroy_dispatch_graph"); - - // The scheduler kernel itself is lightweight (poll + parse + DEVICE_CALL + - // fire-and-forget); the heavy cooperative decode lives in the triggered - // graph, so a single-block scheduler is sufficient. - cudaError_t err = - create_fn(rx_flags_dev_, tx_flags_dev_, rx_data_dev_, tx_data_dev_, - slot_size_, slot_size_, function_table_dev_, - static_cast(function_table_count_), - /*graph_io_ctx=*/nullptr, shutdown_flag_dev_, device_stats_dev_, - num_slots_, /*num_blocks=*/1, /*threads_per_block=*/64, - decode_graph_exec, scheduler_stream_, &scheduler_ctx_); - if (err != cudaSuccess) - throw std::runtime_error( - std::string("qec_realtime_session::initialize: " - "cudaq_create_dispatch_graph_regular failed: ") + - cudaGetErrorString(err)); - - err = launch_fn(scheduler_ctx_, scheduler_stream_); - if (err != cudaSuccess) - throw std::runtime_error( - std::string("qec_realtime_session::initialize: " - "cudaq_launch_dispatch_graph failed: ") + - cudaGetErrorString(err)); -} - -//============================================================================== -// start_host_loop() [HOST mode only] -//============================================================================== - -void qec_realtime_session::start_host_loop() { - // HOST mode only: inline HOST_CALL handlers, no graph worker pool. DEVICE - // mode uses the self-relaunching device-graph scheduler launched in - // start_device_loop() (see initialize()'s dispatch), so this is reached only - // when device_mode_ is false. Every table entry is HOST_CALL, so a - // GRAPH_LAUNCH engine would build no workers -- we drive the ring loop with a - // NULL engine (the loop runs the HOST_CALL handlers inline). - std::memset(&host_config_, 0, sizeof(host_config_)); - host_config_.num_slots = static_cast(num_slots_); - host_config_.slot_size = static_cast(slot_size_); - host_config_.dispatch_path = CUDAQ_DISPATCH_PATH_HOST; - host_config_.dispatch_mode = CUDAQ_DISPATCH_HOST_CALL; - host_config_.skip_tx_markers = 1; - // Strict-FIFO: the host loop is the sole consumer and its current_slot - // already advances monotonically; with shared-ring scanning OFF it simply - // waits at current_slot for the next frame. The rpc_producer advances its - // slot monotonically too (see producer_cursor()), so producer and consumer - // walk the ring in lockstep -- no scan needed. - host_config_.shared_ring_mode = 0; - host_table_.entries = function_table_host_; - host_table_.count = static_cast(function_table_count_); - host_engine_ = nullptr; - shutdown_flag_ = 0; - - host_loop_thread_ = std::thread([this]() { - cudaq_host_ring_dispatch_loop(&ringbuffer_, &host_table_, &host_config_, - /*engine=*/nullptr, &shutdown_flag_, - &host_stats_counter_); - }); -} - -//============================================================================== -// stop_loops() -//============================================================================== - -void qec_realtime_session::stop_loops() { - // Signal shutdown to whichever flag the active host loop polls (and, in - // DEVICE mode, the persistent device kernel which shares the pinned flag). - if (device_mode_) { - if (shutdown_flag_host_) { - __atomic_store_n(shutdown_flag_host_, 1, __ATOMIC_RELEASE); - __sync_synchronize(); - } - } else { - __atomic_store_n(&shutdown_flag_, 1, __ATOMIC_RELEASE); - __sync_synchronize(); - } - - if (host_loop_thread_.joinable()) - host_loop_thread_.join(); - - // Drain and destroy the device-graph scheduler (DEVICE mode). The shutdown - // flag was set above; the scheduler observes it, stops tail-self-relaunching, - // and the in-flight self-relaunch chain drains. cudaStreamSynchronize waits - // for that to finish before we destroy the graph context. - if (scheduler_ctx_) { - if (scheduler_stream_) - cudaStreamSynchronize(scheduler_stream_); - if (destroy_dispatch_graph_fn_) - destroy_dispatch_graph_fn_(scheduler_ctx_); - scheduler_ctx_ = nullptr; - } - if (scheduler_stream_) { - cudaStreamDestroy(scheduler_stream_); - scheduler_stream_ = nullptr; - } - - // Legacy DEVICE_LOOP teardown (unused by the scheduler; both are null in - // scheduler mode, so these are harmless no-ops). - if (device_dispatcher_) { - cudaq_dispatcher_stop(device_dispatcher_); - cudaq_dispatcher_destroy(device_dispatcher_); - device_dispatcher_ = nullptr; - } - if (device_manager_) { - cudaq_dispatch_manager_destroy(device_manager_); - device_manager_ = nullptr; - } - - // The engine owns the worker streams, idle mask, inflight-slot tags, and - // GraphIOContext array; destroy it after the driving loop has stopped. - if (host_engine_) { - cudaq_graph_launch_engine_destroy(host_engine_); - host_engine_ = nullptr; - } -} - -} // namespace cudaq::qec::realtime - -#endif // CUDAQ_REALTIME_ROOT diff --git a/libs/qec/lib/realtime/qec_realtime_session.h b/libs/qec/lib/realtime/qec_realtime_session.h deleted file mode 100644 index fef32a6bc..000000000 --- a/libs/qec/lib/realtime/qec_realtime_session.h +++ /dev/null @@ -1,263 +0,0 @@ -/****************************************************************-*- C++ -*-**** - * Copyright (c) 2024 - 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 - -#ifdef CUDAQ_REALTIME_ROOT - -// Defines CUDA_VERSION, which gates the graph-based dispatch API (the -// self-relaunching scheduler: cudaq_dispatch_graph_context, -// cudaq_create_dispatch_graph_regular, ...) inside cudaq_realtime.h. This TU -// is compiled by the host C++ compiler (no __CUDACC__), so without this the -// graph API would be preprocessed out and the scheduler members below would -// not name a type. -#include - -#include "cudaq/qec/decoder.h" -#include "cudaq/realtime/daemon/dispatcher/cudaq_realtime.h" -#include "cudaq/realtime/daemon/dispatcher/dispatch_kernel_launch.h" -#include "cudaq/realtime/daemon/dispatcher/graph_launch_engine.h" - -#include -#include -#include -#include - -namespace cudaq::qec::realtime { - -/// @brief Per-process realtime decoding session, dual-mode. -/// -/// A session is **homogeneous**: at `initialize()` it inspects the decoders and -/// runs one of two dispatch modes, chosen by -/// `decoder::supports_graph_dispatch()`: -/// -/// - DEVICE mode (every decoder supports graph dispatch -- e.g. the Relay BP -/// GPU decoder). A single self-relaunching device-graph scheduler (see -/// `start_device_loop()`) services all three RPCs as `DEVICE_CALL` -/// handlers: `enqueue_syndromes` accumulates the round's syndromes into the -/// decoder's registered GpuDecoderState and returns -/// `CUDAQ_DISPATCH_STATUS_TRIGGER_GRAPH` when a full window is ready, which -/// makes the scheduler fire the decoder's device-launchable cooperative -/// decode graph fire-and-forget; `get_corrections` / `reset_decoder` run -/// inline on device. The scheduler tail-self-relaunches via -/// `cudaGetCurrentGraphExec()` so the 120 fire-and-forget launches per -/// parent-graph execution budget resets each relaunch. Scoped to one -/// graph-dispatch decoder per session. Requires a non-null -/// `device_launch_fn`. -/// -/// - HOST mode (no decoder supports graph dispatch -- e.g. PyMatching, a CPU -/// decoder). All three RPCs (enqueue_syndromes, get_corrections, -/// reset_decoder) are `CUDAQ_DISPATCH_HOST_CALL` handlers invoked inline by -/// the CPU HOST_LOOP using the two-ring callback ABI -/// `cudaq_host_rpc_fn_t(const void *rx, void *tx, size_t)`. The ring is -/// plain host memory (no GPU required at runtime); the device-visible -/// pointers alias the host backings so `rpc_producer` is mode-agnostic. -/// -/// A *mixed* session (some decoders support graph dispatch and some do not) is -/// rejected: the host loop resolves a slot to a function table entry by -/// `function_id` alone, so a `GRAPH_LAUNCH` enqueue and a `HOST_CALL` enqueue -/// sharing `kEnqueueSyndromesFunctionId` would collide. One decoder per -/// session is the supported (and tested) configuration; homogeneous multi-CPU -/// sessions also work. See the throw in `classify_mode()`. -/// -/// Ring layout (both modes): producer writes RPCHeader + payload into the RX -/// backing, the dispatcher writes RPCResponse + result into the TX backing. -/// RX and TX are separate physical allocations. -/// -/// Constructed with a reference to a vector of realized decoder instances -- -/// the same vector held by `realtime_decoding.cpp::g_decoders` in the -/// production path, or a one-element vector in a unit test. The session keeps -/// a non-owning reference so it can call `release_decode_graph()` (DEVICE mode) -/// on each captured graph at finalize time. -/// -/// The class is marked `default`-visible so its constructor / destructor / -/// `initialize` / `finalize` symbols cross the `cudaq-qec-realtime-decoding` -/// shared-library boundary (the library is built with `-fvisibility=hidden`). -class __attribute__((visibility("default"))) qec_realtime_session { -public: - /// @brief Construct a session over the given realized decoders. - /// @param decoders Reference must outlive this session. The session calls - /// `supports_graph_dispatch()` on each non-null entry at - /// `initialize()` time to choose its dispatch mode, and (in - /// DEVICE mode) `capture_decode_graph()` / - /// `release_decode_graph()`. - /// @param device_launch_fn Function pointer passed to - /// `cudaq_dispatcher_set_launch_fn` in DEVICE mode. Typically - /// `&cudaq_launch_dispatch_kernel_regular` from libcudaq- - /// realtime-dispatch. Passed in (rather than referenced - /// directly) so this shared library stays free of references - /// to symbols that live only in static archives linked by the - /// final executable. May be null for a HOST-mode (CPU - /// decoder) session; `initialize()` throws if a DEVICE-mode - /// decoder set is given without it. - explicit qec_realtime_session( - std::vector> &decoders, - cudaq_dispatch_launch_fn_t device_launch_fn = nullptr); - - ~qec_realtime_session(); - - qec_realtime_session(const qec_realtime_session &) = delete; - qec_realtime_session &operator=(const qec_realtime_session &) = delete; - qec_realtime_session(qec_realtime_session &&) = delete; - qec_realtime_session &operator=(qec_realtime_session &&) = delete; - - /// @brief Bring up the ring + dispatcher(s) for the selected mode. - /// Idempotent: a second call is a no-op. Throws `std::runtime_error` on any - /// failure (mixed decoder set, DEVICE mode without device_launch_fn, decoder - /// lacks graph dispatch in DEVICE mode, CUDA allocation failure, plugin - /// failed to populate device entries, a second concurrent HOST-mode session, - /// ...). - void initialize(); - - /// @brief Tear down dispatcher(s), release captured graphs (DEVICE mode), - /// free ring. Idempotent. Safe to call from a destructor. In DEVICE mode - /// must be called BEFORE the decoders vector is cleared. - void finalize(); - - /// @brief True if `initialize()` has completed and `finalize()` has not. - bool initialized() const { return initialized_; } - - /// @brief True if this session runs in DEVICE (GPU graph-dispatch) mode. - /// Only meaningful after `initialize()`. - bool device_mode() const { return device_mode_; } - - // ---- Accessors used by rpc_producer.cpp (and by tests). ---------------- - // In HOST mode the `_dev` pointers alias the `_host` backings (host memory), - // so the producer's address-as-flag publish works unchanged in both modes. - - volatile std::uint64_t *rx_flags_host() const { return rx_flags_host_; } - volatile std::uint64_t *tx_flags_host() const { return tx_flags_host_; } - std::uint8_t *rx_data_host() const { return rx_data_host_; } - std::uint8_t *rx_data_dev() const { return rx_data_dev_; } - std::uint8_t *tx_data_host() const { return tx_data_host_; } - std::uint8_t *tx_data_dev() const { return tx_data_dev_; } - - std::size_t num_slots() const { return num_slots_; } - std::size_t slot_size() const { return slot_size_; } - - /// @brief Monotonic producer cursor (rpc_producer ring discipline). - /// The single serialized producer advances this each RPC so it walks the - /// ring in lockstep with the strict-FIFO consumer (device-graph scheduler or - /// host loop, both with shared-ring scanning OFF) instead of reusing slot 0. - /// Reset to 0 by initialize(). Single-producer today; a future multi- - /// producer design would make the advance an atomic fetch-add. - std::size_t producer_cursor() const { return producer_cursor_; } - void set_producer_cursor(std::size_t slot) { producer_cursor_ = slot; } - - /// @brief (DEVICE mode) Number of decoders that captured a CUDA graph. - std::size_t num_decoders_with_graph() const { - return num_decoders_with_graph_; - } - -private: - // Inspect decoders_ and set device_mode_. Throws on an empty set, a mixed - // (graph + non-graph) set, or a DEVICE-mode set without device_launch_fn_. - void classify_mode(); - - // ---- DEVICE-mode internals ---- - void capture_decoder_graphs(); - void start_device_loop(); - - // ---- shared internals (branch on device_mode_) ---- - // allocate_ring_buffer() computes slot_size_, allocates rx/tx flags + data - // (pinned-mapped in DEVICE mode, host memory in HOST mode), and fully - // populates ringbuffer_. - void allocate_ring_buffer(); - // populate_function_table() builds the shared function table: N GRAPH_LAUNCH - // + 2 DEVICE_CALL entries in DEVICE mode; 3 HOST_CALL entries in HOST mode. - void populate_function_table(); - // start_host_loop() launches the CPU HOST_LOOP thread. In DEVICE mode it - // wires the per-decoder graph workers + GraphIOContext mailbox; in HOST mode - // it runs the inline HOST_CALL handlers (no worker pool). - void start_host_loop(); - - // Signal shutdown, join host thread, stop device dispatcher (if any), free - // worker streams + per-worker storage. - void stop_loops(); - - // ---- References / external state ---- - std::vector> &decoders_; - cudaq_dispatch_launch_fn_t device_launch_fn_ = nullptr; - - // ---- Lifetime / mode ---- - bool initialized_ = false; - bool device_mode_ = false; - - // ---- Ring buffer (raw pointers; _dev aliases _host in HOST mode) ---- - static constexpr std::size_t kDefaultNumSlots = 8; - std::size_t num_slots_ = kDefaultNumSlots; - std::size_t slot_size_ = 0; - // Monotonic producer ring cursor (see producer_cursor()). Reset in - // initialize(). - std::size_t producer_cursor_ = 0; - volatile std::uint64_t *rx_flags_host_ = nullptr; - volatile std::uint64_t *rx_flags_dev_ = nullptr; - volatile std::uint64_t *tx_flags_host_ = nullptr; - volatile std::uint64_t *tx_flags_dev_ = nullptr; - std::uint8_t *rx_data_host_ = nullptr; - std::uint8_t *rx_data_dev_ = nullptr; - std::uint8_t *tx_data_host_ = nullptr; - std::uint8_t *tx_data_dev_ = nullptr; - cudaq_ringbuffer_t ringbuffer_{}; - - // ---- Function table ---- - // DEVICE mode: pinned-mapped (host + device same UVA). HOST mode: plain host - // allocation (host_fn pointers are host code addresses); _dev aliases _host. - std::size_t function_table_count_ = 0; - cudaq_function_entry_t *function_table_host_ = nullptr; - cudaq_function_entry_t *function_table_dev_ = nullptr; - std::uint32_t get_corrections_fn_id_ = 0; - std::uint32_t reset_decoder_fn_id_ = 0; - - // ---- DEVICE-mode scheduler wiring ---- - // A single self-relaunching device-graph scheduler replaces the legacy - // HOST_LOOP(graph-worker) + DEVICE_LOOP(get/reset) pair. The scheduler - // runs all three RPCs as DEVICE_CALL handlers and fires the - // per-decoder device-launchable decode graph fire-and-forget when the - // enqueue accumulate handler signals a full window - // (CUDAQ_DISPATCH_STATUS_TRIGGER_GRAPH). device_manager_/device_dispatcher_ - // are retained (unused in scheduler mode) to keep the legacy DEVICE_LOOP - // teardown in stop_loops() a harmless no-op. - cudaq_dispatch_manager_t *device_manager_ = nullptr; - cudaq_dispatcher_t *device_dispatcher_ = nullptr; - std::uint64_t *device_stats_dev_ = nullptr; - cudaq_dispatch_graph_context *scheduler_ctx_ = nullptr; - cudaStream_t scheduler_stream_ = nullptr; - // cudaq_destroy_dispatch_graph, resolved by start_device_loop() via - // dlsym(RTLD_DEFAULT, ...) from the host executable's absorbed - // libcudaq-realtime-dispatch.a (same image as the create/launch fns), so - // stop_loops() tears down the graph context with the copy that created it. - using destroy_dispatch_graph_fn_t = - cudaError_t (*)(cudaq_dispatch_graph_context *); - destroy_dispatch_graph_fn_t destroy_dispatch_graph_fn_ = nullptr; - // Pinned-mapped shutdown flag polled by the scheduler graph (DEVICE mode). - int *shutdown_flag_host_ = nullptr; - int *shutdown_flag_dev_ = nullptr; - - // ---- HOST_LOOP wiring (both modes) ---- - // The GRAPH_LAUNCH engine (DEVICE mode) owns the worker streams, idle mask, - // inflight-slot tags, and per-worker GraphIOContext array. It is built from - // the GRAPH_LAUNCH entries of `function_table_host_` and is NULL in HOST mode - // (all-HOST_CALL table -> no graph workers). - cudaq_function_table_t host_table_{}; - cudaq_dispatcher_config_t host_config_{}; - cudaq_graph_launch_engine_t *host_engine_ = nullptr; - std::thread host_loop_thread_; - std::uint64_t host_stats_counter_ = 0; - // Plain (non-pinned) shutdown flag for HOST mode (no device kernel shares - // it). - int shutdown_flag_ = 0; - - // ---- Graph state (DEVICE mode only) ---- - std::vector captured_graphs_; - std::size_t num_decoders_with_graph_ = 0; -}; - -} // namespace cudaq::qec::realtime - -#endif // CUDAQ_REALTIME_ROOT diff --git a/libs/qec/lib/realtime/realtime_decoding.cpp b/libs/qec/lib/realtime/realtime_decoding.cpp index b22dc70f6..adf86e993 100644 --- a/libs/qec/lib/realtime/realtime_decoding.cpp +++ b/libs/qec/lib/realtime/realtime_decoding.cpp @@ -14,23 +14,11 @@ #include "cudaq/qec/realtime/decoding_config.h" #include #include -#include -#include -#include #include #include #include #include -#ifdef CUDAQ_REALTIME_ROOT -#include "qec_realtime_session.h" -#include "rpc_producer.h" -#else -namespace cudaq::qec::realtime { -class qec_realtime_session {}; -} // namespace cudaq::qec::realtime -#endif - // Optional syndrome capture callback for --save_syndrome feature namespace { using SyndromeCaptureCallback = void (*)(const uint8_t *, size_t); @@ -38,103 +26,6 @@ SyndromeCaptureCallback g_syndrome_capture_callback = nullptr; } // namespace std::vector> g_decoders; -std::unique_ptr g_realtime_session; - -namespace { - -#ifdef CUDAQ_REALTIME_ROOT -inline cudaq_dispatch_launch_fn_t resolve_launch_dispatch_kernel_regular() { - return reinterpret_cast( - ::dlsym(RTLD_DEFAULT, "cudaq_launch_dispatch_kernel_regular")); -} -#endif - -bool realtime_mode_inproc_rpc_requested() { - const char *env = std::getenv("CUDAQ_QEC_REALTIME_MODE"); - if (!env || env[0] == '\0') - return false; - return std::strcmp(env, "inproc_rpc") == 0; -} - -bool any_decoder_supports_graph_dispatch() { - for (const auto &dec : g_decoders) { - if (dec && dec->supports_graph_dispatch()) - return true; - } - return false; -} - -} // namespace - -#ifdef CUDAQ_REALTIME_ROOT -namespace { - -void maybe_init_realtime_session() { - if (!realtime_mode_inproc_rpc_requested()) { - CUDA_QEC_INFO("CUDAQ_QEC_REALTIME_MODE not set to inproc_rpc; using " - "legacy direct-call decoding path."); - return; - } - - // Pick DEVICE vs HOST dispatch the same way qec_realtime_session does at - // initialize(): any graph-capable decoder => DEVICE mode (per-round - // GRAPH_LAUNCH enqueue + DEVICE_CALL get/reset, driven by the device dispatch - // kernel); otherwise HOST mode -- CPU decoders such as pymatching run all - // three RPCs inline on the CPU host loop. A mixed (graph + non-graph) set is - // rejected by qec_realtime_session::initialize() below. - const bool device_mode = any_decoder_supports_graph_dispatch(); - - cudaq_dispatch_launch_fn_t launch_fn = nullptr; - if (device_mode) { - // DEVICE mode needs the dispatch-kernel launch helper from - // libcudaq-realtime-dispatch.a (absorbed into the final executable). HOST - // mode uses no device launch helper. - launch_fn = resolve_launch_dispatch_kernel_regular(); - if (!launch_fn) - throw std::runtime_error( - "CUDAQ_QEC_REALTIME_MODE=inproc_rpc requested with a graph-capable " - "decoder but cudaq_launch_dispatch_kernel_regular could not be " - "resolved via dlsym(RTLD_DEFAULT, ...). The host executable must " - "absorb libcudaq-realtime-dispatch.a and link with " - "--export-dynamic."); - } else { - CUDA_QEC_INFO("CUDAQ_QEC_REALTIME_MODE=inproc_rpc with CPU (non-graph) " - "decoder(s); using HOST dispatch mode (no device kernel / no " - "device shared-ring setup)."); - } - - try { - g_realtime_session = - std::make_unique(g_decoders, - launch_fn); - g_realtime_session->initialize(); - } catch (const std::exception &e) { - const std::string what = e.what(); - g_realtime_session.reset(); - throw std::runtime_error("CUDAQ_QEC_REALTIME_MODE=inproc_rpc requested but " - "qec_realtime_session::initialize() threw: " + - what); - } -} - -void maybe_finalize_realtime_session() { - if (g_realtime_session) { - try { - g_realtime_session->finalize(); - } catch (const std::exception &e) { - CUDA_QEC_WARN("qec_realtime_session::finalize threw: {}", e.what()); - } - g_realtime_session.reset(); - } -} - -} // namespace -#else -namespace { -void maybe_init_realtime_session() {} -void maybe_finalize_realtime_session() {} -} // namespace -#endif // Helper to pack syndrome bits into bytes (8 bits per byte, MSB first for // readability) @@ -287,10 +178,6 @@ std::unique_ptr create_realtime_decoder( return decoder; } -cudaq::qec::realtime::qec_realtime_session *get_realtime_session() { - return g_realtime_session.get(); -} - int configure_decoders( cudaq::qec::decoding::config::multi_decoder_config &config) { CUDA_QEC_INFO("Initializing decoders..."); @@ -323,45 +210,6 @@ int configure_decoders( return 3; } -#ifdef CUDAQ_REALTIME_ROOT - // inproc_rpc DEVICE sessions allocate pinned, device-mapped ring buffers - // (cudaHostAlloc(cudaHostAllocMapped) + cudaHostGetDevicePointer). - // cudaSetDeviceFlags(cudaDeviceMapHost) only takes effect BEFORE the device's - // CUDA context is created, and the per-decoder dry-run below - // (new_decoder->decode(...)) can create that context for GPU decoders -- so - // set the flag here, before any decoder is realized, rather than (only) later - // in qec_realtime_session::initialize(). Best-effort: if a context already - // exists this returns cudaErrorSetOnActiveProcess, which is harmless (mapped - // host allocation still works via UVA regardless of this device-wide flag), - // and HOST-mode CPU sessions do not use mapped memory at all. - if (realtime_mode_inproc_rpc_requested()) { - // The device-mapped ring buffers guarded by cudaDeviceMapHost are used only - // by the DEVICE-mode graph scheduler, which needs a usable GPU. CPU - // decoders run in HOST mode with plain host memory and never touch the - // device, so probe for a GPU first and skip the flag entirely when none is - // present. This keeps CPU-only / GPU-less machines from executing the - // device-flag call at all -- previously it ran unconditionally and logged a - // spurious "CUDA driver version is insufficient" warning. (If a graph - // decoder is later selected without a usable device, - // qec_realtime_session::initialize() still fails with a clear DEVICE-mode - // error.) - int device_count = 0; - cudaError_t count_err = cudaGetDeviceCount(&device_count); - if (count_err == cudaSuccess && device_count > 0) { - cudaError_t flags_err = cudaSetDeviceFlags(cudaDeviceMapHost); - if (flags_err != cudaSuccess && flags_err != cudaErrorSetOnActiveProcess) - CUDA_QEC_WARN( - "cudaSetDeviceFlags(cudaDeviceMapHost) returned '{}' before " - "decoder init; continuing (mapped alloc works via UVA).", - cudaGetErrorString(flags_err)); - } else { - // Reset the sticky runtime error so a later benign cudaGetLastError() - // isn't surprised by the no-device / insufficient-driver probe result. - cudaGetLastError(); - } - } -#endif - // Create the decoders based on the decoder configs. try { g_decoders.clear(); @@ -374,13 +222,11 @@ int configure_decoders( return 4; } - maybe_init_realtime_session(); return 0; } void finalize_decoders() { CUDA_QEC_INFO("Finalizing the realtime decoding library."); - maybe_finalize_realtime_session(); g_decoders.clear(); } @@ -432,22 +278,6 @@ void enqueue_syndromes(std::size_t decoder_id, uint8_t *syndromes, } }; -#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); - } catch ( - const cudaq::qec::decoding::rpc_producer::dispatcher_unresponsive_error - &) { - maybe_finalize_realtime_session(); - throw; - } - return; - } -#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 @@ -505,22 +335,6 @@ void get_corrections(std::size_t decoder_id, uint8_t *corrections, correction_length, num_observables)); } -#ifdef CUDAQ_REALTIME_ROOT - if (g_realtime_session) { - try { - cudaq::qec::decoding::rpc_producer::get_corrections( - *g_realtime_session, decoder_id, corrections, correction_length, - reset ? 1u : 0u); - } catch ( - const cudaq::qec::decoding::rpc_producer::dispatcher_unresponsive_error - &) { - maybe_finalize_realtime_session(); - throw; - } - return; - } -#endif - // clear_corrections may touch device memory in some plugins. cudaq::qec::detail_affinity::pin_decode_device(*decoder); auto ret = decoder->get_obs_corrections(); @@ -543,21 +357,6 @@ void reset_decoder(std::size_t decoder_id) { fmt::format("Decoder {} not found", decoder_id)); } -#ifdef CUDAQ_REALTIME_ROOT - if (g_realtime_session) { - try { - cudaq::qec::decoding::rpc_producer::reset_decoder(*g_realtime_session, - decoder_id); - } catch ( - const cudaq::qec::decoding::rpc_producer::dispatcher_unresponsive_error - &) { - maybe_finalize_realtime_session(); - throw; - } - return; - } -#endif - cudaq::qec::detail_affinity::pin_decode_device(*decoder); decoder->reset_decoder(); } diff --git a/libs/qec/lib/realtime/realtime_decoding.h b/libs/qec/lib/realtime/realtime_decoding.h index d4e6558f7..f65fdf13f 100644 --- a/libs/qec/lib/realtime/realtime_decoding.h +++ b/libs/qec/lib/realtime/realtime_decoding.h @@ -15,19 +15,8 @@ // Note: none of these are intended to be user-facing functions. -namespace cudaq::qec::realtime { -class qec_realtime_session; -} // namespace cudaq::qec::realtime - namespace cudaq::qec::decoding::host { -/// @brief Accessor for the per-process realtime session. Returns nullptr -/// unless CUDAQ_QEC_REALTIME_MODE=inproc_rpc has initialized the shared-ring -/// dispatch session. -__attribute__((visibility("default"))) -cudaq::qec::realtime::qec_realtime_session * -get_realtime_session(); - __attribute__((visibility("default"))) void enqueue_syndromes(std::size_t decoder_id, uint8_t *syndromes, std::uint64_t syndrome_length, std::uint64_t tag); diff --git a/libs/qec/lib/realtime/rpc_producer.cpp b/libs/qec/lib/realtime/rpc_producer.cpp deleted file mode 100644 index 87fb97ef1..000000000 --- a/libs/qec/lib/realtime/rpc_producer.cpp +++ /dev/null @@ -1,449 +0,0 @@ -/******************************************************************************* - * Copyright (c) 2024 - 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. * - ******************************************************************************/ - -#ifdef CUDAQ_REALTIME_ROOT - -#include "rpc_producer.h" - -#include "qec_realtime_session.h" -#include "cudaq/qec/realtime/decoder_rpc_wire_format.h" -#include "cudaq/realtime/daemon/dispatcher/cudaq_realtime.h" -#include "cudaq/realtime/daemon/dispatcher/dispatch_kernel_launch.h" - -#include -#include -#include -#include -#include -#include -#include // for usleep -#include - -namespace cudaq::qec::decoding::rpc_producer { - -namespace { - -// Process-wide monotonic counter for `RPCHeader::request_id`. The wire -// protocol echoes request_id in `RPCResponse::request_id`; today we don't -// match-by-id (we wait on the magic of the slot we wrote to), but the -// handlers still validate it's nonzero, and DEVICE_LOOP stats keys off it. -std::atomic g_request_id_counter{1}; - -std::uint32_t next_request_id() { - return g_request_id_counter.fetch_add(1, std::memory_order_relaxed); -} - -// Enforces the single-producer contract documented in rpc_producer.h. The -// producer path is single-producer by design -- the sole caller is the single- -// threaded QEC decode loop. acquire_slot() selects a free slot without an -// atomic reservation, so two concurrent producers could pick the same slot and -// corrupt each other's RPC. Rather than pay for full multi-producer support -// (CAS slot-claim / per-producer arena), we DETECT a contract violation and -// fail loudly. This is a real throw, NOT assert(): release builds compile -// assert() out, so an assert would enforce nothing in production. -std::atomic g_producer_active{false}; - -struct single_producer_guard { - single_producer_guard() { - bool expected = false; - if (!g_producer_active.compare_exchange_strong(expected, true, - std::memory_order_acquire)) - throw std::runtime_error( - "rpc_producer: concurrent producer detected. This RPC path is " - "single-producer (the single-threaded QEC decode loop); serialize " - "calls or add multi-producer support (CAS slot-claim / per-producer " - "arena)."); - } - ~single_producer_guard() { - g_producer_active.store(false, std::memory_order_release); - } - single_producer_guard(const single_producer_guard &) = delete; - single_producer_guard &operator=(const single_producer_guard &) = delete; -}; - -// Bounded spin for the NEXT slot in monotonic ring order. Returns -// UINT32_MAX on timeout. -// -// Ring discipline: the producer walks slots monotonically -// (session.producer_cursor(), advanced mod num_slots each acquire) rather -// than picking the lowest free slot. This keeps it in lockstep with the -// strict-FIFO consumer (the device-graph scheduler / host loop, both with -// shared-ring scanning OFF), which waits at exactly its own monotonically- -// advancing cursor. The cursor is reset to 0 by -// qec_realtime_session::initialize(), matching the consumer's reset, so both -// start at slot 0. Back-pressure is preserved: we wait until the chosen slot -// is free (rx_flags[s]==0 AND tx_flags[s]==0 -- request consumed by the -// dispatcher and response consumed by release_slot()). -// -// THREAD-SAFETY ASSUMPTION (single producer): -// We read-modify-write producer_cursor() non-atomically, which is correct -// under the single-producer invariant documented in rpc_producer.h (the QEC -// main loop is the only producer, single-threaded by construction). A future -// multi-producer design would make the cursor advance an atomic fetch-add -// (the natural multi-producer ring head) and add a CAS claim of the slot; -// this is deferred to a follow-up MR. -std::uint32_t acquire_slot(cudaq::qec::realtime::qec_realtime_session &session, - int timeout_ms) { - volatile std::uint64_t *rx = session.rx_flags_host(); - volatile std::uint64_t *tx = session.tx_flags_host(); - const std::size_t n = session.num_slots(); - if (rx == nullptr || tx == nullptr || n == 0) - return UINT32_MAX; - const std::uint32_t s = - static_cast(session.producer_cursor() % n); - for (int waited = 0; waited < timeout_ms; ++waited) { - if (rx[s] == 0 && tx[s] == 0) { - session.set_producer_cursor((s + 1u) % n); - return s; - } - // 1 ms granularity matches the test's spin cadence. The shared ring - // is host-pinned + UVA-mapped, so the producer's view of rx/tx flags - // is coherent with the GPU consumer's writes after a __sync_- - // synchronize on the consumer side; usleep here keeps the busy-wait - // off the critical path. - usleep(1000); - } - return UINT32_MAX; -} - -// Write an RPC request into `slot` and publish it by writing the device- -// visible slot address into rx_flags[slot]. Mirrors WriteAndSignal in the -// test; the slot is pre-acquired so the caller owns it from acquire_slot() -// through release_slot(). -void write_and_signal(cudaq::qec::realtime::qec_realtime_session &session, - std::uint32_t slot, std::uint32_t function_id, - std::uint32_t request_id, const void *payload, - std::size_t payload_len) { - // Two-ring wire format: requests go into the RX backing, responses - // come back via the TX backing. We zero just the RX slot here; the - // TX slot is cleared lazily on release_slot() so a stale response - // body from a previous round doesn't confound the next reader. - std::uint8_t *rx_slot_host = - session.rx_data_host() + slot * session.slot_size(); - std::memset(rx_slot_host, 0, session.slot_size()); - auto *header = reinterpret_cast(rx_slot_host); - header->magic = cudaq::realtime::RPC_MAGIC_REQUEST; - header->function_id = function_id; - header->arg_len = static_cast(payload_len); - header->request_id = request_id; - header->ptp_timestamp = 0; - std::memcpy(rx_slot_host + sizeof(cudaq::realtime::RPCHeader), payload, - payload_len); - __sync_synchronize(); - // Address-as-flag publish: the DEVICE-visible RX slot address is what - // the dispatcher polls for on rx_flags. Same UVA on host and device - // because the RX data backing is pinned+mapped. - session.rx_flags_host()[slot] = reinterpret_cast( - session.rx_data_dev() + slot * session.slot_size()); -} - -// Bounded spin for response publication. Returns false on timeout. The -// response is considered complete only after the writer has produced an -// RPCResponse header and published the matching tx_flags entry. -bool wait_for_response(cudaq::qec::realtime::qec_realtime_session &session, - std::uint32_t slot, int timeout_ms) { - // Two-ring wire format: the response lives in the TX slot. The - // dispatcher's writer (the captured graph for GRAPH_LAUNCH; the - // DEVICE_LOOP kernel for DEVICE_CALL) writes RPCResponse and then signals - // tx_flags[slot]. Wait for both before reading or releasing the slot. - std::uint8_t *tx_slot_host = - session.tx_data_host() + slot * session.slot_size(); - auto *resp = - reinterpret_cast(tx_slot_host); - for (int waited = 0; waited < timeout_ms; ++waited) { - __sync_synchronize(); - if (resp->magic == cudaq::realtime::RPC_MAGIC_RESPONSE && - session.tx_flags_host()[slot] != 0) - return true; - // 200us granularity matches the test. Shorter than acquire_slot's - // sleep because get_corrections / reset round-trips are sub-ms on - // typical GPUs and a 1ms cadence would dominate the round-trip - // budget for small per-shot payloads. - usleep(200); - } - return false; -} - -// Release a slot after the caller has finished consuming the response. -// Two-ring wire format: clears the TX backing (so the next reader of -// this slot won't see a stale `RPC_MAGIC_RESPONSE`). The RX backing -// for this slot was already overwritten by the dispatcher (the captured -// graph) when it parsed the request; we zero it again defensively in -// write_and_signal() before the next request, so we don't need to wipe -// it here. Clears tx_flags[slot] to unblock acquire_slot for the next -// caller. rx_flags is cleared by the dispatcher (see -// host_dispatcher.cu::finish_slot_and_advance), not by the producer. -void release_slot(cudaq::qec::realtime::qec_realtime_session &session, - std::uint32_t slot) { - std::uint8_t *tx_slot_host = - session.tx_data_host() + slot * session.slot_size(); - std::memset(tx_slot_host, 0, session.slot_size()); - __sync_synchronize(); - session.tx_flags_host()[slot] = 0; -} - -// Common pre-flight: session must be initialized + ring must be live. -// Pulled into a helper so each public function's pre-flight error is -// uniform (and so the runtime error includes the function name). -void require_initialized(cudaq::qec::realtime::qec_realtime_session &session, - const char *fn) { - if (!session.initialized()) { - std::ostringstream os; - os << "cudaq::qec::decoding::rpc_producer::" << fn - << ": session is not initialized(). Call qec_realtime_session::" - "initialize() before sending RPCs."; - throw std::runtime_error(os.str()); - } - if (session.rx_flags_host() == nullptr || - session.tx_flags_host() == nullptr || session.rx_data_host() == nullptr || - session.rx_data_dev() == nullptr || session.tx_data_host() == nullptr || - session.tx_data_dev() == nullptr) { - std::ostringstream os; - os << "cudaq::qec::decoding::rpc_producer::" << fn - << ": session ring buffer pointers are null even though " - "initialized()==true. This indicates a teardown race or a " - "double-finalize bug."; - throw std::runtime_error(os.str()); - } -} - -} // namespace - -void enqueue_syndromes(cudaq::qec::realtime::qec_realtime_session &session, - std::size_t decoder_id, const std::uint8_t *syndromes, - std::uint64_t num_syndromes, std::uint64_t tag) { - single_producer_guard producer_guard; - require_initialized(session, "enqueue_syndromes"); - - if (syndromes == nullptr && num_syndromes > 0) - throw std::runtime_error( - "rpc_producer::enqueue_syndromes: syndromes == nullptr but " - "num_syndromes > 0"); - - // Build the wire payload per decoder_server_runtime.md#enqueue_syndromes: - // 32-byte EnqueueRequestPayload (decoder_id, counter, - // syndrome_mapping_id, num_syndromes; all INT64) - // + ceil(num_syndromes/8) bit-packed syndrome bytes (LSB-first), no pad. - const std::size_t bp_bytes = - cudaq::qec::decoding::rpc::bit_packed_bytes(num_syndromes); - const std::size_t body_bytes = - sizeof(cudaq::qec::decoding::rpc::EnqueueRequestPayload) + bp_bytes; - std::vector payload(body_bytes, 0); - auto *p = - reinterpret_cast( - payload.data()); - p->decoder_id = static_cast(decoder_id); - p->counter = static_cast(tag); - p->syndrome_mapping_id = 0; - p->num_syndromes = static_cast(num_syndromes); - if (num_syndromes > 0) { - std::uint8_t *bits = - payload.data() + - sizeof(cudaq::qec::decoding::rpc::EnqueueRequestPayload); - for (std::uint64_t i = 0; i < num_syndromes; ++i) { - // Source format: one bit per byte (low bit significant), matching - // the existing plugin / test contract. Bit i lands at bit (i mod - // 8) of bits[i / 8] -- LSB-first, per the spec. - if (syndromes[i] & 0x1u) - bits[i / 8] |= - static_cast(1u << static_cast(i % 8)); - } - } - - std::uint32_t slot = acquire_slot(session, kAcquireSlotTimeoutMs); - if (slot == UINT32_MAX) - throw dispatcher_unresponsive_error( - "rpc_producer::enqueue_syndromes: timed out acquiring a free slot"); - - // request_id is for correlation. The full-width application tag travels as - // payload arg1 (`counter`); request_id is a 32-bit window so use the low 32 - // bits of tag. When `tag = (shot << 16) | round` (the production - // convention), this yields a unique-per-shot request_id within a 65k-shot - // window, more than enough for in-flight correlation in DEVICE_LOOP stats / - // debugging. - const std::uint32_t request_id = static_cast(tag); - write_and_signal(session, slot, - cudaq::qec::decoding::rpc::kEnqueueSyndromesFunctionId, - request_id, payload.data(), payload.size()); - if (!wait_for_response(session, slot, kResponseTimeoutMs)) { - release_slot(session, slot); - std::ostringstream os; - os << "rpc_producer::enqueue_syndromes: timed out waiting for ACK " - "(decoder_id=" - << decoder_id << ", tag=0x" << std::hex << tag << ")"; - throw dispatcher_unresponsive_error(os.str()); - } - - std::uint8_t *tx_slot_host = - session.tx_data_host() + slot * session.slot_size(); - const auto *resp = - reinterpret_cast(tx_slot_host); - - if (resp->status != 0) { - const std::int32_t status = resp->status; - release_slot(session, slot); - std::ostringstream os; - os << "rpc_producer::enqueue_syndromes: non-zero status (" << status - << ") for decoder_id=" << decoder_id << ", tag=0x" << std::hex << tag; - throw std::runtime_error(os.str()); - } - // Per decoder_server_runtime.md the dispatcher always emits a 24-byte - // RPCResponse, even for fire-and-forget calls; the body is empty - // (result_len == 0). Drop the ACK and release the slot. - if (resp->result_len != 0) { - const std::uint32_t got = resp->result_len; - release_slot(session, slot); - std::ostringstream os; - os << "rpc_producer::enqueue_syndromes: unexpected non-empty ACK " - "(decoder_id=" - << decoder_id << "), expected result_len=0, got " << got; - throw std::runtime_error(os.str()); - } - - release_slot(session, slot); -} - -void get_corrections(cudaq::qec::realtime::qec_realtime_session &session, - std::size_t decoder_id, std::uint8_t *corrections, - std::uint64_t correction_length, std::uint64_t reset) { - single_producer_guard producer_guard; - require_initialized(session, "get_corrections"); - - if (corrections == nullptr && correction_length > 0) - throw std::runtime_error( - "rpc_producer::get_corrections: corrections == nullptr but " - "correction_length > 0"); - - // Build the wire payload per decoder_server_runtime.md#get_corrections: - // 17 bytes total: decoder_id (INT64) + return_size (INT64, the OUT - // std::vector length) + reset (UINT8, trailing bool, no pad). The - // struct is laid out exactly this way. - cudaq::qec::decoding::rpc::GetCorrectionsRequestPayload payload{}; - payload.decoder_id = static_cast(decoder_id); - payload.return_size = static_cast(correction_length); - payload.reset = reset ? std::uint8_t{1} : std::uint8_t{0}; - - std::uint32_t slot = acquire_slot(session, kAcquireSlotTimeoutMs); - if (slot == UINT32_MAX) - throw dispatcher_unresponsive_error( - "rpc_producer::get_corrections: timed out acquiring a free slot"); - - const std::uint32_t request_id = next_request_id(); - write_and_signal(session, slot, - cudaq::qec::decoding::rpc::kGetCorrectionsFunctionId, - request_id, &payload, sizeof(payload)); - - if (!wait_for_response(session, slot, kResponseTimeoutMs)) { - release_slot(session, slot); - std::ostringstream os; - os << "rpc_producer::get_corrections: timed out waiting for response " - "(decoder_id=" - << decoder_id << ")"; - throw dispatcher_unresponsive_error(os.str()); - } - - std::uint8_t *tx_slot_host = - session.tx_data_host() + slot * session.slot_size(); - const auto *resp = - reinterpret_cast(tx_slot_host); - - if (resp->status != 0) { - const std::int32_t status = resp->status; - release_slot(session, slot); - std::ostringstream os; - os << "rpc_producer::get_corrections: non-zero status (" << status - << ") for decoder_id=" << decoder_id; - throw std::runtime_error(os.str()); - } - // Per spec result_len = ceil(R/8) exactly (no trailing pad). - const std::size_t expected_bp = - cudaq::qec::decoding::rpc::bit_packed_bytes(correction_length); - if (resp->result_len != static_cast(expected_bp)) { - const std::uint32_t got = resp->result_len; - release_slot(session, slot); - std::ostringstream os; - os << "rpc_producer::get_corrections: result_len mismatch (decoder_id=" - << decoder_id << "), expected " << expected_bp - << " (ceil(R/8) for R=" << correction_length << "), got " << got; - throw std::runtime_error(os.str()); - } - - if (correction_length > 0) { - // Unpack the bit-packed result (LSB-first) into the caller's - // byte-per-bit output buffer to preserve the API surface used by - // realtime_decoding.cpp / the ABI seen by test code. Bit i of the - // correction vector lives at bit (i mod 8) of bits[i/8]. - const std::uint8_t *bits = - tx_slot_host + sizeof(cudaq::realtime::RPCResponse); - for (std::uint64_t i = 0; i < correction_length; ++i) { - corrections[i] = static_cast( - (bits[i / 8] >> static_cast(i % 8)) & 0x1u); - } - } - - release_slot(session, slot); -} - -void reset_decoder(cudaq::qec::realtime::qec_realtime_session &session, - std::size_t decoder_id) { - single_producer_guard producer_guard; - require_initialized(session, "reset_decoder"); - - cudaq::qec::decoding::rpc::ResetRequestPayload payload{}; - payload.decoder_id = static_cast(decoder_id); - - std::uint32_t slot = acquire_slot(session, kAcquireSlotTimeoutMs); - if (slot == UINT32_MAX) - throw dispatcher_unresponsive_error( - "rpc_producer::reset_decoder: timed out acquiring a free slot"); - - const std::uint32_t request_id = next_request_id(); - write_and_signal(session, slot, - cudaq::qec::decoding::rpc::kResetDecoderFunctionId, - request_id, &payload, sizeof(payload)); - - if (!wait_for_response(session, slot, kResponseTimeoutMs)) { - release_slot(session, slot); - std::ostringstream os; - os << "rpc_producer::reset_decoder: timed out waiting for response " - "(decoder_id=" - << decoder_id << ")"; - throw dispatcher_unresponsive_error(os.str()); - } - - std::uint8_t *tx_slot_host = - session.tx_data_host() + slot * session.slot_size(); - const auto *resp = - reinterpret_cast(tx_slot_host); - - if (resp->status != 0) { - const std::int32_t status = resp->status; - release_slot(session, slot); - std::ostringstream os; - os << "rpc_producer::reset_decoder: non-zero status (" << status - << ") for decoder_id=" << decoder_id; - throw std::runtime_error(os.str()); - } - // Per spec the dispatcher always emits an empty 24-byte RPCResponse for - // fire-and-forget reset; result_len must be 0. - if (resp->result_len != 0) { - const std::uint32_t got = resp->result_len; - release_slot(session, slot); - std::ostringstream os; - os << "rpc_producer::reset_decoder: unexpected non-empty ACK " - "(decoder_id=" - << decoder_id << "), expected result_len=0, got " << got; - throw std::runtime_error(os.str()); - } - - release_slot(session, slot); -} - -} // namespace cudaq::qec::decoding::rpc_producer - -#endif // CUDAQ_REALTIME_ROOT diff --git a/libs/qec/lib/realtime/rpc_producer.h b/libs/qec/lib/realtime/rpc_producer.h deleted file mode 100644 index ce2280743..000000000 --- a/libs/qec/lib/realtime/rpc_producer.h +++ /dev/null @@ -1,184 +0,0 @@ -/****************************************************************-*- C++ -*-**** - * Copyright (c) 2024 - 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::realtime { -class qec_realtime_session; -} // namespace cudaq::qec::realtime - -namespace cudaq::qec::decoding::rpc_producer { - -/// @brief Thrown when an RPC round-trip cannot make progress because the -/// dispatcher is unresponsive -- either no free slot appears within -/// `kAcquireSlotTimeoutMs`, or no `RPCResponse` appears within -/// `kResponseTimeoutMs`. Distinct from `std::runtime_error` so the host-side -/// caller (realtime_decoding.cpp) can recognize it as fatal-to-the-session -/// (the device kernel / host monitor has stopped servicing the ring) and tear -/// the session down rather than retry into a slow ghost-slot leak. All other -/// producer errors (bad args, non-zero RPC status, malformed result_len) -/// stay plain `std::runtime_error`. -struct dispatcher_unresponsive_error : std::runtime_error { - using std::runtime_error::runtime_error; -}; - -//============================================================================== -// Host-side RPC producer for the inproc_rpc realtime decoding path. -// -// These three functions are the host-process counterparts to the three RPC -// handlers wired up by qec_realtime_session + decoder_rpc_dispatch.cu, all -// conformant with proposals/decoder_server_runtime.md: -// - enqueue_syndromes GRAPH_LAUNCH; function_id == -// fnv1a("enqueue_syndromes") -// == kEnqueueSyndromesFunctionId; one captured graph -// per decoder; host monitor sub-routes by -// (function_id, routing_key=decoder_id). -// - get_corrections DEVICE_CALL; function_id == fnv1a("get_corrections") -// == kGetCorrectionsFunctionId. -// - reset_decoder DEVICE_CALL; function_id == fnv1a("reset_decoder") -// == kResetDecoderFunctionId. -// -// All three RPCs are request/response on the wire: the dispatcher always -// emits a 24-byte RPCResponse (status=0, result_len=0 for the two fire- -// and-forget calls; bit-packed correction bytes + 0..7 pad for get_- -// corrections). Producers wait for the ACK and drop the response body -// for enqueue/reset. -// -// Each function: -// 1. Resolves the per-process session via the `session` argument (no global -// lookup) so test code can drive a private session without touching -// `g_realtime_session`. -// 2. Acquires a free RX slot from the ring buffer. -// 3. Writes RPCHeader + the matching wire-format payload from -// decoder_rpc_wire_format.h. Payload scalars are INT64; bools are -// UINT8+pad; bit-packed arrays are LSB-first; whole payload is padded to -// an 8-byte multiple. -// 4. Publishes the slot by writing the DEVICE-visible RX slot address -// into rx_flags[slot] (the "address-as-flag" convention of the shared -// ring; both the HOST_LOOP and DEVICE_LOOP dispatchers expect this). -// 5. Spins on RPCResponse::magic, checks status, and (for -// get_corrections) reads back the bit-packed result. -// 6. Releases the slot (clears rx_flags + tx_flags + the slot's first -// bytes so the response magic doesn't get misread on slot reuse). -// -// On any error (no slot within timeout_ms, response timeout, non-zero -// RPC status, malformed result_len), the corresponding function throws -// `std::runtime_error` with a contextual message. The host-side caller in -// the production path (realtime_decoding.cpp::enqueue_syndromes / -// get_corrections / reset) is also a free function in -// `cudaq::qec::decoding::host`, so it gets to choose how to surface the -// error -- e.g. by logging then rethrowing. -// -// THREAD-SAFETY: -// All three functions are SINGLE-PRODUCER -- this is a hard contract, not -// merely an assumption. The production caller in realtime_decoding.cpp:: -// enqueue_syndromes / get_corrections / reset_decoder is invoked from the QEC -// main loop, which is single-threaded by construction. The contract is now -// ENFORCED at runtime: each function holds a single_producer_guard that throws -// if a second producer is active concurrently (an always-on check -- a real -// throw, not assert(), so it stays active in release builds). Full multi- -// producer support (CAS slot-claim or a per-producer arena) remains a -// deliberate follow-up. -// -// acquire_slot() in rpc_producer.cpp scans for a free slot (rx_flags == -// tx_flags == 0) but does NOT atomically claim it on return. That's -// fine while there's a single producer (the chosen slot is written -// before the next call to acquire_slot()), but it would race if a -// second producer thread were ever introduced. Multi-producer support -// is a deliberate follow-up: it requires either (a) a CAS-based atomic -// claim on the rx_flags slot ("0 -> in-progress") inside acquire_slot, -// or (b) a per-producer arena of slots in the ring. See -// `acquire_slot()` for the inline assumption comment. -// -// As a separate constraint, the *handlers* themselves are not -// re-entrant per-decoder (the plugin's GpuDecoderState is shared -// across in-flight enqueue rounds for that decoder_id), so the -// wire-level contract is "one in-flight RPC per (decoder_id, -// function_id) on the wire at a time" regardless of producer count. -//============================================================================== - -/// @brief Send a per-round enqueue RPC for `decoder_id` carrying -/// `num_syndromes` syndrome bits, bit-packed LSB-first into the wire payload. -/// -/// The session holds N GRAPH_LAUNCH entries that all share the canonical -/// `kEnqueueSyndromesFunctionId`; the host monitor disambiguates them by -/// `routing_key = decoder_id` (see proposals/cudaq_realtime_host_api.bs# -/// host-path-graph-routing-key). The dispatcher always emits a 24-byte -/// `RPCResponse` with `status=0, result_len=0` (no body); this producer -/// waits for the ACK and drops it. -/// -/// @param session Realtime session. Must be `initialized()`. -/// @param decoder_id Routing key. Written into payload arg0 AND -/// matched against the function table's -/// `routing_key` field. -/// @param syndromes Pointer to `syndrome_length` raw syndrome bytes, -/// one bit-per-byte at the source (each input byte -/// contributes one bit to the wire-side bit-packed -/// array). -/// @param num_syndromes Number of syndrome BITS for this round -/// (== `syndrome_length`). Written into payload -/// arg3. -/// @param tag Application-level breadcrumb written into -/// `RPCHeader::request_id` (low 32 bits) and payload -/// arg1 (`counter`, full 64 bits). This first pass -/// always emits payload arg2 (`syndrome_mapping_id`) as -/// 0 for contiguous identity mapping. Production callers -/// typically pack `(shot << 16) | round` into `tag`. -__attribute__((visibility("default"))) void -enqueue_syndromes(cudaq::qec::realtime::qec_realtime_session &session, - std::size_t decoder_id, const std::uint8_t *syndromes, - std::uint64_t num_syndromes, std::uint64_t tag); - -/// @brief Fetch `correction_length` correction bytes for `decoder_id` and -/// optionally reset the device-side accumulated correction buffer. -/// -/// Handler is the shared DEVICE_CALL `get_corrections_ui64`; routing happens -/// in `decoder_rpc_dispatch.cu` based on `decoder_id` in the payload. -/// -/// @param session Realtime session. Must be `initialized()`. -/// @param decoder_id Decoder index for state lookup. -/// @param corrections Output buffer (caller-owned, at least -/// `correction_length` bytes). -/// @param correction_length # of correction bytes to read. Must match the -/// decoder's declared num_observables. -/// @param reset 1 to zero the device-side correction buffer -/// after the read, 0 to leave it accumulated. -__attribute__((visibility("default"))) void -get_corrections(cudaq::qec::realtime::qec_realtime_session &session, - std::size_t decoder_id, std::uint8_t *corrections, - std::uint64_t correction_length, std::uint64_t reset); - -/// @brief Send a reset RPC, clearing per-decoder device-side state -/// (correction buffer + the plugin's BP context). -/// -/// Handler is the shared DEVICE_CALL `reset_decoder_ui64`. -/// -/// @param session Realtime session. Must be `initialized()`. -/// @param decoder_id Decoder index for state lookup. -__attribute__((visibility("default"))) void -reset_decoder(cudaq::qec::realtime::qec_realtime_session &session, - std::size_t decoder_id); - -//============================================================================== -// Spin/timeout knobs (visible so tests can shorten timeouts for negative -// fixtures without touching the production defaults). -//============================================================================== - -/// @brief Max time (ms) `AcquireSlot` will spin waiting for a free slot -/// before throwing. -constexpr int kAcquireSlotTimeoutMs = 5000; - -/// @brief Max time (ms) `WaitForResponse` will spin waiting for -/// `RPCResponse::magic` before throwing. -constexpr int kResponseTimeoutMs = 5000; - -} // namespace cudaq::qec::decoding::rpc_producer diff --git a/libs/qec/unittests/CMakeLists.txt b/libs/qec/unittests/CMakeLists.txt index 0140a9bcc..2c9aba7cd 100644 --- a/libs/qec/unittests/CMakeLists.txt +++ b/libs/qec/unittests/CMakeLists.txt @@ -528,87 +528,6 @@ if(CUDAQ_REALTIME_ROOT AND CMAKE_CUDA_COMPILER) ) # test_realtime_predecoder_w_pymatching is now in unittests/realtime/ - # ------------------------------------------------------------------ - # Realtime QLDPC graph decode CI test (CPU-launched CUDA graph + relay BP) - # ------------------------------------------------------------------ - - # cudaq-qec-realtime-decoding transitively pulls in cudaq-qec, which - # references CUDA-Q quantum runtime symbols. Satisfy them here. - find_library(_CUDAQ_LIBRARY NAMES cudaq - PATHS ${CUDAQ_INSTALL_PREFIX} $ENV{CUDAQ_INSTALL_PREFIX} - ${CUDAQ_DIR}/.. $ENV{CUDAQ_DIR}/.. - PATH_SUFFIXES lib) - find_library(_NVQIR_LIBRARY NAMES nvqir - PATHS ${CUDAQ_INSTALL_PREFIX} $ENV{CUDAQ_INSTALL_PREFIX} - ${CUDAQ_DIR}/.. $ENV{CUDAQ_DIR}/.. - PATH_SUFFIXES lib) - - add_executable(test_realtime_qldpc_graph_decoding - ${CMAKE_CURRENT_SOURCE_DIR}/realtime/qec_graph_decode_test/test_realtime_qldpc_graph_decoding.cpp - ${CMAKE_CURRENT_SOURCE_DIR}/realtime/qec_graph_decode_test/qldpc_config_loader.cpp - ) - - set_target_properties(test_realtime_qldpc_graph_decoding PROPERTIES - CUDA_SEPARABLE_COMPILATION ON - CUDA_RESOLVE_DEVICE_SYMBOLS ON - CUDA_STANDARD 17 - LINKER_LANGUAGE CUDA - ) - - target_include_directories(test_realtime_qldpc_graph_decoding PRIVATE - ${CUDAToolkit_INCLUDE_DIRS} - ${CUDAQ_REALTIME_INCLUDE_DIR} - ${CMAKE_CURRENT_SOURCE_DIR}/../include - ${CMAKE_SOURCE_DIR}/libs/core/include - ${CMAKE_CURRENT_SOURCE_DIR}/../lib/realtime - ) - - target_compile_definitions(test_realtime_qldpc_graph_decoding PRIVATE - TEST_DATA_DIR="${CMAKE_CURRENT_SOURCE_DIR}/realtime/qec_roce_decode_test/data" - CUDAQ_REALTIME_ROOT - ) - - target_link_libraries(test_realtime_qldpc_graph_decoding PRIVATE - GTest::gtest_main - CUDA::cudart - ${CUDAQ_REALTIME_LIBRARY} - ${CUDAQ_REALTIME_DISPATCH_LIBRARY} - $<$:${CUDAQ_REALTIME_HOST_DISPATCH_LIBRARY}> - cudaq-qec-realtime-decoding - $<$:$> - $<$:${_CUDAQ_LIBRARY}> - $<$:${_NVQIR_LIBRARY}> - ) - - target_link_options(test_realtime_qldpc_graph_decoding PRIVATE - "LINKER:--allow-shlib-undefined" - "LINKER:--disable-new-dtags" - "LINKER:--export-dynamic" - ) - - # Resolve CUDAQ SDK lib dir for RPATH (libnvqir, libcudaq) - set(_cudaq_sdk_lib_dir "") - if(CUDAQ_INSTALL_PREFIX) - set(_cudaq_sdk_lib_dir "${CUDAQ_INSTALL_PREFIX}/lib") - elseif(CUDAQ_DIR) - get_filename_component(_cudaq_sdk_lib_dir "${CUDAQ_DIR}/../../.." ABSOLUTE) - set(_cudaq_sdk_lib_dir "${_cudaq_sdk_lib_dir}/lib") - endif() - - set_target_properties(test_realtime_qldpc_graph_decoding PROPERTIES - BUILD_RPATH "${CUDAQ_REALTIME_LIB_DIR};${CMAKE_BINARY_DIR}/lib;${CMAKE_BINARY_DIR}/lib/decoder-plugins;${_cudaq_sdk_lib_dir}" - INSTALL_RPATH "${CUDAQ_REALTIME_LIB_DIR};${CMAKE_INSTALL_PREFIX}/${CMAKE_INSTALL_LIBDIR};${CMAKE_BINARY_DIR}/lib/decoder-plugins;${_cudaq_sdk_lib_dir}" - ) - - add_dependencies(CUDAQXQECUnitTests test_realtime_qldpc_graph_decoding) - - if((QEC_EXTERNAL_DECODERS OR DEFINED ENV{QEC_EXTERNAL_DECODERS}) - AND TARGET cudaq-qec-realtime-cudevice-proprietary) - add_test(NAME test_realtime_qldpc_graph_decoding - COMMAND test_realtime_qldpc_graph_decoding) - set_tests_properties(test_realtime_qldpc_graph_decoding PROPERTIES - LABELS "qec") - endif() else() message(WARNING "cuda-quantum realtime dependency not found. " diff --git a/libs/qec/unittests/decoders/pymatching/CMakeLists.txt b/libs/qec/unittests/decoders/pymatching/CMakeLists.txt index cc8d5a0f8..03c9c7a8c 100644 --- a/libs/qec/unittests/decoders/pymatching/CMakeLists.txt +++ b/libs/qec/unittests/decoders/pymatching/CMakeLists.txt @@ -29,28 +29,6 @@ target_link_libraries(test_pymatching PRIVATE GTest::gtest_main cudaq-qec-decode add_dependencies(CUDAQXQECUnitTests test_pymatching) gtest_discover_tests(test_pymatching) -if(CUDAQ_REALTIME_ROOT AND CUDAQ_REALTIME_INCLUDE_DIR) - add_executable(test_pymatching_realtime test_pymatching_realtime.cpp) - target_include_directories(test_pymatching_realtime PRIVATE - ${CUDAQ_REALTIME_INCLUDE_DIR} - ${CUDAToolkit_INCLUDE_DIRS} - ${CUDAQX_QEC_SOURCE_DIR}/lib/realtime - ) - target_compile_definitions(test_pymatching_realtime PRIVATE - CUDAQ_REALTIME_ROOT) - target_link_libraries(test_pymatching_realtime PRIVATE - GTest::gtest_main - cudaq-qec-decoders - cudaq-qec-realtime-decoding - ) - set_target_properties(test_pymatching_realtime PROPERTIES - BUILD_RPATH "${CMAKE_BINARY_DIR}/lib;${CMAKE_BINARY_DIR}/lib/decoder-plugins" - ) - add_dependencies(test_pymatching_realtime cudaq-qec-pymatching) - add_dependencies(CUDAQXQECUnitTests test_pymatching_realtime) - gtest_discover_tests(test_pymatching_realtime) -endif() - if(CUDAQ_REALTIME_INCLUDE_DIR AND TARGET cudaq-qec-realtime-decoding-server-cqr) add_executable(test_pymatching_device_call_realtime) set_target_properties(test_pymatching_device_call_realtime PROPERTIES @@ -108,9 +86,7 @@ if(CUDAQ_REALTIME_INCLUDE_DIR AND TARGET cudaq-qec-realtime-decoding-server-cqr) add_dependencies(CUDAQXQECUnitTests test_pymatching_device_call_realtime) # CUDAQ_DEVICE_CALL_CHANNEL=host_dispatch routes the device_call over the # cudaq-realtime host-dispatch ring to the decoding server, which decodes on - # the CPU host. No CUDAQ_QEC_REALTIME_MODE=inproc_rpc: that second, - # server-internal decode ring is redundant now that the device_call already - # crosses cudaq-realtime here -- the server decodes inline instead. + # the CPU host. gtest_discover_tests(test_pymatching_device_call_realtime WORKING_DIRECTORY ${CMAKE_CURRENT_BINARY_DIR} PROPERTIES ENVIRONMENT "CUDAQ_DEVICE_CALL_CHANNEL=host_dispatch") diff --git a/libs/qec/unittests/decoders/pymatching/test_pymatching_realtime.cpp b/libs/qec/unittests/decoders/pymatching/test_pymatching_realtime.cpp deleted file mode 100644 index 150de67fb..000000000 --- a/libs/qec/unittests/decoders/pymatching/test_pymatching_realtime.cpp +++ /dev/null @@ -1,209 +0,0 @@ -/******************************************************************************* - * Copyright (c) 2024 - 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. * - ******************************************************************************/ - -#include "qec_realtime_session.h" -#include "realtime_decoding.h" -#include "rpc_producer.h" -#include "cudaq/qec/decoder.h" -#include "cudaq/qec/realtime/decoding_config.h" - -#include - -#include -#include -#include -#include -#include - -namespace { - -using DecoderVec = std::vector>; - -DecoderVec make_pymatching_decoders(const std::vector &h_vec, - std::size_t syndrome_size, - std::size_t block_size) { - cudaqx::tensor h; - h.copy(h_vec.data(), {syndrome_size, block_size}); - - DecoderVec decoders; - auto decoder = - cudaq::qec::decoder::get("pymatching", h, cudaqx::heterogeneous_map{}); - decoder->set_decoder_id(0); - std::vector> d_sparse(syndrome_size); - for (std::size_t row = 0; row < syndrome_size; ++row) - d_sparse[row].push_back(static_cast(row)); - decoder->set_D_sparse(d_sparse); - std::vector> o_sparse(block_size); - for (std::size_t row = 0; row < block_size; ++row) - o_sparse[row].push_back(static_cast(row)); - decoder->set_O_sparse(o_sparse); - decoders.push_back(std::move(decoder)); - return decoders; -} - -void expect_corrections(cudaq::qec::realtime::qec_realtime_session &session, - const std::vector &syndrome, - std::span expected, - std::uint64_t counter, bool reset_on_read = true) { - cudaq::qec::decoding::rpc_producer::enqueue_syndromes( - session, /*decoder_id=*/0, syndrome.data(), syndrome.size(), - /*tag=*/counter); - - std::vector corrections(expected.size(), 0xCC); - cudaq::qec::decoding::rpc_producer::get_corrections( - session, /*decoder_id=*/0, corrections.data(), corrections.size(), - reset_on_read ? 1 : 0); - EXPECT_EQ(corrections, - std::vector(expected.begin(), expected.end())); -} - -std::vector -read_corrections(cudaq::qec::realtime::qec_realtime_session &session, - std::size_t num_corrections) { - std::vector corrections(num_corrections, 0xCC); - cudaq::qec::decoding::rpc_producer::get_corrections( - session, /*decoder_id=*/0, corrections.data(), corrections.size(), - /*reset=*/0); - return corrections; -} - -void run_case(const std::vector &h_vec, std::size_t syndrome_size, - std::size_t block_size, - const std::vector, - std::vector>> &cases) { - auto decoders = make_pymatching_decoders(h_vec, syndrome_size, block_size); - cudaq::qec::realtime::qec_realtime_session session(decoders); - session.initialize(); - - std::uint64_t counter = 1; - for (const auto &[syndrome, expected] : cases) { - expect_corrections(session, syndrome, expected, counter++); - EXPECT_EQ(read_corrections(session, block_size), - std::vector(block_size, 0)); - } - - session.finalize(); -} - -} // namespace - -TEST(PyMatchingRealtime, CheckRegularEdges) { - run_case(/*H=*/{1, 0, 1, 1, 0, 1}, /*syndrome_size=*/3, /*block_size=*/2, - {{{1, 1, 0}, {1, 0}}, {{0, 1, 1}, {0, 1}}, {{1, 0, 1}, {1, 1}}}); -} - -TEST(PyMatchingRealtime, CheckBoundaryEdges) { - run_case(/*H=*/{1, 0, 0, 0, 1, 0, 0, 0, 1}, - /*syndrome_size=*/3, /*block_size=*/3, - {{{1, 0, 0}, {1, 0, 0}}, - {{0, 1, 0}, {0, 1, 0}}, - {{0, 0, 1}, {0, 0, 1}}, - {{1, 0, 0}, {1, 0, 0}}}); -} - -TEST(PyMatchingRealtime, PreservesCallerColumnOrderUnderNonCanonicalOrdering) { - run_case(/*H=*/{1, 0, 0, 0, 0, 1, 0, 0, 0, 0, 0, 1, 0, 0, 1, 0}, - /*syndrome_size=*/4, /*block_size=*/4, - {{{0, 0, 0, 1}, {0, 0, 1, 0}}, - {{0, 0, 1, 0}, {0, 0, 0, 1}}, - {{1, 0, 0, 0}, {1, 0, 0, 0}}}); -} - -TEST(PyMatchingRealtime, ResetDecoderClearsCorrections) { - auto decoders = - make_pymatching_decoders(/*H=*/{1, 0, 1, 1, 0, 1}, /*syndrome_size=*/3, - /*block_size=*/2); - cudaq::qec::realtime::qec_realtime_session session(decoders); - session.initialize(); - - expect_corrections(session, {1, 0, 1}, std::vector{1, 1}, - /*counter=*/1, /*reset_on_read=*/false); - EXPECT_EQ(read_corrections(session, 2), (std::vector{1, 1})); - cudaq::qec::decoding::rpc_producer::reset_decoder(session, /*decoder_id=*/0); - EXPECT_EQ(read_corrections(session, 2), (std::vector{0, 0})); - - session.finalize(); -} - -TEST(PyMatchingRealtime, RejectsOversizedSyndromeRequest) { - // Single decoder per session (the supported configuration). The ring slot - // has headroom beyond the decoder's per-decode window -- it is also sized for - // the response payload and a 64-byte floor -- so an oversized enqueue can - // still fit the slot, pass the slot-size check, and reach the new per-decoder - // length guard. Without that guard enqueue_syndrome would overflow the - // decoder's accumulation buffer, silently drop the data (it returns false), - // and the handler would still ACK success. - auto decoders = make_pymatching_decoders(/*H=*/{1, 0, 0, 0, 1, 0, 0, 0, 1}, - /*syndrome_size=*/3, - /*block_size=*/3); - cudaq::qec::realtime::qec_realtime_session session(decoders); - session.initialize(); - - const std::uint64_t capacity = decoders[0]->get_num_msyn_per_decode(); - const std::uint64_t oversized = capacity + 1; - std::vector oversized_syndrome(oversized, 0); - EXPECT_THROW(cudaq::qec::decoding::rpc_producer::enqueue_syndromes( - session, /*decoder_id=*/0, oversized_syndrome.data(), - oversized_syndrome.size(), /*tag=*/1), - std::runtime_error); - - // A correctly-sized request to the same decoder is still accepted (confirms - // the guard rejected only the oversized one and released the slot). - std::vector ok_syndrome(capacity, 0); - EXPECT_NO_THROW(cudaq::qec::decoding::rpc_producer::enqueue_syndromes( - session, /*decoder_id=*/0, ok_syndrome.data(), ok_syndrome.size(), - /*tag=*/2)); - - session.finalize(); -} - -TEST(PyMatchingRealtime, ConfiguresViaRealtimeDecoderConfig) { - namespace config = cudaq::qec::decoding::config; - - config::decoder_config decoder_config; - decoder_config.id = 0; - decoder_config.type = "pymatching"; - decoder_config.block_size = 3; - decoder_config.syndrome_size = 3; - decoder_config.H_sparse = {0, -1, 1, -1, 2, -1}; - decoder_config.O_sparse = {0, -1, 1, -1, 2, -1}; - decoder_config.D_sparse = {0, -1, 1, -1, 2, -1}; - - cudaqx::heterogeneous_map pymatching_args; - pymatching_args.insert("error_rate_vec", std::vector{0.1, 0.1, 0.1}); - pymatching_args.insert("merge_strategy", "smallest_weight"); - decoder_config.decoder_custom_args = pymatching_args; - - config::multi_decoder_config multi_config; - multi_config.decoders.push_back(decoder_config); - - const auto yaml_str = multi_config.to_yaml_str(200); - auto multi_config_from_yaml = - config::multi_decoder_config::from_yaml_str(yaml_str); - EXPECT_EQ(multi_config_from_yaml, multi_config); - - EXPECT_EQ(config::configure_decoders(multi_config), 0); - - std::vector syndrome{0, 1, 0}; - EXPECT_NO_THROW(cudaq::qec::decoding::host::enqueue_syndromes( - /*decoder_id=*/0, syndrome.data(), syndrome.size(), /*tag=*/1)); - - std::vector corrections(3, 0xCC); - EXPECT_NO_THROW(cudaq::qec::decoding::host::get_corrections( - /*decoder_id=*/0, corrections.data(), corrections.size(), - /*reset=*/true)); - EXPECT_EQ(corrections, (std::vector{0, 1, 0})); - - corrections.assign(3, 0xCC); - EXPECT_NO_THROW(cudaq::qec::decoding::host::get_corrections( - /*decoder_id=*/0, corrections.data(), corrections.size(), - /*reset=*/false)); - EXPECT_EQ(corrections, (std::vector{0, 0, 0})); - - config::finalize_decoders(); -} diff --git a/libs/qec/unittests/realtime/app_examples/CMakeLists.txt b/libs/qec/unittests/realtime/app_examples/CMakeLists.txt index 591fb7f35..85c5198e4 100644 --- a/libs/qec/unittests/realtime/app_examples/CMakeLists.txt +++ b/libs/qec/unittests/realtime/app_examples/CMakeLists.txt @@ -15,8 +15,8 @@ # build/libs/qec/unittests/realtime/app_examples/qir-*.ll #------------------------------------------------------------------------------# -# Resolve cudaq-realtime dispatch for app examples that opt into -# CUDAQ_QEC_REALTIME_MODE=inproc_rpc. +# Resolve cudaq-realtime dispatch (DEVICE-mode graph launch) for app examples +# that link it. set(_app_cudaq_realtime_prefixes "") if(CUDAQ_REALTIME_ROOT) list(APPEND _app_cudaq_realtime_prefixes "${CUDAQ_REALTIME_ROOT}") @@ -83,29 +83,6 @@ add_test( WORKING_DIRECTORY ${CMAKE_CURRENT_BINARY_DIR} ) -if(_APP_CUDAQ_REALTIME_DISPATCH_LIB - AND TARGET cudaq-qec-realtime-cudevice-proprietary) - add_test( - NAME app_examples.surface_code-1-local-test-distance-3-inproc-rpc - COMMAND - bash "${CMAKE_CURRENT_SOURCE_DIR}/surface_code-1-test.sh" - ${CMAKE_CURRENT_BINARY_DIR}/surface_code-1-local - ${CMAKE_CURRENT_BINARY_DIR}/surface_code-1-local - 3 60 30 NULL 12 - ${CMAKE_BINARY_DIR}/lib - nv-qldpc-decoder - WORKING_DIRECTORY ${CMAKE_CURRENT_BINARY_DIR} - ) - set_tests_properties( - app_examples.surface_code-1-local-test-distance-3-inproc-rpc - PROPERTIES ENVIRONMENT - "CUDAQ_QEC_REALTIME_MODE=inproc_rpc" - # The wrapper exits 77 on GPUs below compute capability 9.0 (the - # device-graph scheduler needs device-side graph launch); report that - # as SKIPPED rather than FAILED. - SKIP_RETURN_CODE 77) -endif() - # Test with sliding_window decoder add_test( NAME app_examples.surface_code-1-local-test-distance-3-sliding-window @@ -791,9 +768,8 @@ if(TARGET decoding_server endif() # Surface-code-4 YAML tests: exercise both d=3 and d=5 for each decoder entry. -# All entries run through the realtime inproc RPC mode; relay-BP is registered -# only when the private cudevice/archive and external decoder plugin are -# configured. +# Regular entries run through the direct-call decode path; the cqr variants +# exercise the external decoding server over UDP. find_package(Python COMPONENTS Interpreter QUIET) set(_SC4_TRY_TRT_TEST FALSE) if(Python_Interpreter_FOUND) @@ -806,6 +782,19 @@ if(Python_Interpreter_FOUND) set(_SC4_TRY_TRT_TEST TRUE) endif() endif() +set(_SC4_HAVE_NV_QLDPC FALSE) +if(TARGET cudaq-qec-nv-qldpc-decoder + OR QEC_EXTERNAL_DECODERS + OR DEFINED ENV{QEC_EXTERNAL_DECODERS}) + set(_SC4_HAVE_NV_QLDPC TRUE) +endif() +# Same lesson as the trt gate below: name the gate that actually failed rather +# than letting the nv-qldpc cases evaporate silently. +if(NOT _SC4_HAVE_NV_QLDPC) + message(WARNING "surface_code-4-yaml: nv-qldpc-decoder tests NOT registered " + "(no cudaq-qec-nv-qldpc-decoder target and QEC_EXTERNAL_DECODERS unset); " + "point QEC_EXTERNAL_DECODERS at the plugin to enable them.") +endif() # A silent skip here once masked a dead test check for a whole release cycle: # the trt surface-code tests register only when BOTH the trt plugin target and # python-onnx are available, so name whichever gate actually failed. @@ -825,9 +814,9 @@ endif() function(add_surface_code_4_yaml_test test_suffix distance num_rounds decoder_type num_shots) - set(_sc4_env "CUDAQ_QEC_REALTIME_MODE=inproc_rpc") + set(_sc4_env "") if(Python_Interpreter_FOUND) - list(APPEND _sc4_env "PYTHON=${Python_EXECUTABLE}") + set(_sc4_env "PYTHON=${Python_EXECUTABLE}") endif() add_test( @@ -886,16 +875,6 @@ if(TARGET cudaq-qec-trt-decoder AND _SC4_TRY_TRT_TEST) --generate-identity-onnx) endif() -# nv-qldpc relay-BP requires both the external nv-qldpc decoder plugin and the -# proprietary cudevice archive. Match the other realtime graph/inproc tests: -# register only when those dependencies are configured; do not runtime-skip. -if(_APP_CUDAQ_REALTIME_DISPATCH_LIB - AND TARGET cudaq-qec-realtime-cudevice-proprietary - AND (QEC_EXTERNAL_DECODERS OR DEFINED ENV{QEC_EXTERNAL_DECODERS})) - add_surface_code_4_yaml_decoder_cases(relay-bp nv-qldpc-decoder - --use-relay-bp) -endif() - # Relaxed round rule (T=6 not a multiple of d=5) and multi-logical patches, each # through the full generate + reload + decode driver -- so no hand-written # generate/reload duplication, and both get the driver's residual-LER check. @@ -951,48 +930,6 @@ if(TARGET surface_code-4-yaml-cqr) endif() -# The multi-type cases must PROVE the patches traveled through one realtime -# session: without realtime support maybe_init_realtime_session() is a no-op -# and enqueue/get/reset silently take the legacy direct-call path, so the -# tests would pass on YAML routing alone. Assert HOST-mode initialization -# whenever this build has realtime support. Gate on the compile definition -# actually applied to cudaq-qec-realtime-decoding (set when -# CUDAQ_REALTIME_INCLUDE_DIR is discovered), NOT the optional -# CUDAQ_REALTIME_ROOT search hint -- the headers can be found through -# CUDAQ_INSTALL_DIR while the hint stays empty. -get_target_property(_sc4_rt_defs cudaq-qec-realtime-decoding - COMPILE_DEFINITIONS) -if(_sc4_rt_defs MATCHES "CUDAQ_REALTIME_ROOT") - set(_sc4_hostmode_env "REQUIRE_HOST_MODE=1;CUDAQ_LOG_LEVEL=info") - set_property(TEST app_examples.surface_code-4-yaml-list-parse-d3 APPEND - PROPERTY ENVIRONMENT "${_sc4_hostmode_env}") - if(TARGET cudaq-qec-trt-decoder AND _SC4_TRY_TRT_TEST) - set_property(TEST app_examples.surface_code-4-yaml-trio-trt-d3 APPEND - PROPERTY ENVIRONMENT "${_sc4_hostmode_env}") - set_property(TEST app_examples.surface_code-4-yaml-trio-trt-d5 APPEND - PROPERTY ENVIRONMENT "${_sc4_hostmode_env}") - endif() -endif() - -# Three-way heterogeneous trio incl. relay-BP. Requires a HOST-dispatch -# (non-graph) nv-qldpc plugin via QEC_EXTERNAL_DECODERS, built against the -# CURRENT cudaq-qec headers (a stale plugin fails to load): the inproc session -# must be dispatch-homogeneous, and pymatching/trt are host-mode. A -# graph-flavor plugin (built with CUDAQ_REALTIME_ROOT) would be rejected as a -# mixed set. 1000 shots make working-vs-broken unambiguous for the BP entry -# under the predeclared 2% correctness bound. REQUIRE_HOST_MODE + -# CUDAQ_LOG_LEVEL=info make the driver assert the realtime session actually -# initialized (no vacuous pass through the legacy direct-call stub). -if((QEC_EXTERNAL_DECODERS OR DEFINED ENV{QEC_EXTERNAL_DECODERS}) - AND TARGET cudaq-qec-trt-decoder AND _SC4_TRY_TRT_TEST) - add_surface_code_4_yaml_test(trio-relay-d3 3 3 - "pymatching,trt_decoder,nv-qldpc-decoder" 1000 - --generate-identity-onnx - --num_logical 3 --use-relay-bp) - set_property(TEST app_examples.surface_code-4-yaml-trio-relay-d3 APPEND - PROPERTY ENVIRONMENT "REQUIRE_HOST_MODE=1;CUDAQ_LOG_LEVEL=info") -endif() - # --------------------------------------------------------------------------- # # Negative / boundary tests for geometry, replay, and multi-logical reload. # PASS_REGULAR_EXPRESSION alone ignores exit status, so _sc4_reject checks both. @@ -1000,8 +937,7 @@ endif() set(_sc4_exe ${CMAKE_CURRENT_BINARY_DIR}/surface_code-4-yaml) set(_sc4_neg_dir ${CMAKE_CURRENT_BINARY_DIR}/negative) file(MAKE_DIRECTORY ${_sc4_neg_dir}) -set(_sc4_neg_env "CUDAQ_QEC_REALTIME_MODE=inproc_rpc" - "CUDAQ_DEFAULT_SIMULATOR=stim") +set(_sc4_neg_env "CUDAQ_DEFAULT_SIMULATOR=stim") # A rejection case must satisfy BOTH conditions: a nonzero exit AND the expected # diagnostic. CTest's PASS_REGULAR_EXPRESSION ignores the exit code, so a wrapper @@ -1227,9 +1163,7 @@ fi set_tests_properties(app_examples.surface_code-4-yaml-dual-parse-gen-d${distance} PROPERTIES ENVIRONMENT "${_sc4_neg_env}") endfunction() -if(TARGET cudaq-qec-nv-qldpc-decoder - OR QEC_EXTERNAL_DECODERS - OR DEFINED ENV{QEC_EXTERNAL_DECODERS}) +if(_SC4_HAVE_NV_QLDPC) _sc4_dual_parse_test(3 3) _sc4_dual_parse_test(5 5) endif() diff --git a/libs/qec/unittests/realtime/app_examples/surface_code-1-test.sh b/libs/qec/unittests/realtime/app_examples/surface_code-1-test.sh index 49b9ef888..635d24475 100644 --- a/libs/qec/unittests/realtime/app_examples/surface_code-1-test.sh +++ b/libs/qec/unittests/realtime/app_examples/surface_code-1-test.sh @@ -43,22 +43,6 @@ SW_WINDOW_SIZE=${10:-5} SW_STEP_SIZE=${11:-1} EXTRA_CLI_ARGS=${EXTRA_CLI_ARGS:-} -# The inproc_rpc realtime path is served by the device-graph scheduler, which -# uses device-side graph launch -- compute capability 9.0+ (Hopper) only; the -# dispatch kernel's TRIGGER_GRAPH interception is compiled out below sm_90 and -# the enqueue RPC surfaces the raw sentinel as a non-zero status. Skip on -# older GPUs (e.g. A100/sm_80 CI runners), matching the skip in -# test_realtime_qldpc_graph_decoding. Exit code 77 pairs with the test's -# SKIP_RETURN_CODE property. -if [[ "${CUDAQ_QEC_REALTIME_MODE:-}" == "inproc_rpc" ]]; then - compute_cap=$(nvidia-smi --query-gpu=compute_cap --format=csv,noheader 2>/dev/null | head -1) - if [[ -n "$compute_cap" && "${compute_cap%%.*}" -lt 9 ]]; then - echo "SKIP: CUDAQ_QEC_REALTIME_MODE=inproc_rpc requires device-side graph" \ - "launch (compute capability 9.0+); found ${compute_cap}" - exit 77 - fi -fi - export CUDAQ_DEFAULT_SIMULATOR=stim NUM_SHOTS=1000 diff --git a/libs/qec/unittests/realtime/app_examples/surface_code-4-yaml-test.sh b/libs/qec/unittests/realtime/app_examples/surface_code-4-yaml-test.sh index 9755cdeea..aeb223f10 100755 --- a/libs/qec/unittests/realtime/app_examples/surface_code-4-yaml-test.sh +++ b/libs/qec/unittests/realtime/app_examples/surface_code-4-yaml-test.sh @@ -143,11 +143,6 @@ for ((i = 0; i < ${#EXTRA_APP_ARGS[@]}; i++)); do done export CUDAQ_DEFAULT_SIMULATOR=stim -if [[ -n "${QEC_DECODING_SERVER:-}" ]]; then - export CUDAQ_QEC_REALTIME_MODE=external_server -else - export CUDAQ_QEC_REALTIME_MODE=${CUDAQ_QEC_REALTIME_MODE:-inproc_rpc} -fi P_SPAM=0.01 @@ -241,7 +236,6 @@ echo " num_rounds = $NUM_ROUNDS" echo " decoder_type = $DECODER_TYPE" echo " num_shots = $NUM_SHOTS" echo " num_logical = $NUM_LOGICAL" -echo " realtime mode = $CUDAQ_QEC_REALTIME_MODE" if [[ -n "$ONNX_PATH" ]]; then echo " onnx_path = $ONNX_PATH" fi @@ -630,16 +624,6 @@ if [[ -n "$SERVER_PORT" ]]; then echo "Server evidence: dispatches=$server_dispatches, max_concurrent=$server_max_concurrent" fi -# REQUIRE_HOST_MODE (relay trio ctest): assert the realtime session actually -# initialized in HOST dispatch mode, so the test cannot pass vacuously through -# the legacy direct-call path. Needs CUDAQ_LOG_LEVEL=info. -if [[ -n "${REQUIRE_HOST_MODE:-}" ]]; then - if ! grep -q "using HOST dispatch mode" "$REALTIME_LOG"; then - echo "FAIL: 'using HOST dispatch mode' not found (realtime session did not initialize; is CUDAQ_LOG_LEVEL=info set?)" - return_code=1 - fi -fi - echo "" if [[ "$return_code" -eq 0 ]]; then echo "PASS: surface_code-4-yaml ($DECODER_TYPE, d=$DISTANCE) realtime decode succeeded" diff --git a/libs/qec/unittests/realtime/qec_graph_decode_test/qldpc_config_loader.cpp b/libs/qec/unittests/realtime/qec_graph_decode_test/qldpc_config_loader.cpp deleted file mode 100644 index 921f812ed..000000000 --- a/libs/qec/unittests/realtime/qec_graph_decode_test/qldpc_config_loader.cpp +++ /dev/null @@ -1,95 +0,0 @@ -/****************************************************************-*- C++ -*-**** - * Copyright (c) 2025-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. * - ******************************************************************************/ - -#include "qldpc_config_loader.h" - -#include "cudaq/qec/decoder.h" -#include "cudaq/qec/realtime/decoding_config.h" -#include "cudaq/qec/realtime/sparse_to_csr.h" - -#include -#include -#include -#include -#include - -namespace test_realtime_qldpc { - -namespace { - -std::string read_file(const std::string &path) { - std::ifstream f(path); - if (!f.is_open()) - throw std::runtime_error( - "test_realtime_qldpc_config_loader: failed to open YAML: " + path); - return std::string((std::istreambuf_iterator(f)), - std::istreambuf_iterator()); -} - -} // namespace - -LoadedDecoder load_decoder_from_yaml(const std::string &yaml_path) { - using namespace cudaq::qec; - - auto mdc = decoding::config::multi_decoder_config::from_yaml_str( - read_file(yaml_path)); - if (mdc.decoders.size() != 1u) - throw std::runtime_error( - "test_realtime_qldpc_config_loader: expected exactly one decoder in " - "YAML, found " + - std::to_string(mdc.decoders.size()) + ": " + yaml_path); - - auto &dec = mdc.decoders[0]; - - // H_sparse -> dense tensor (the plugin reads it through the standard - // decoder::get() entry point). - std::vector h_row_ptr, h_col_idx; - std::size_t h_rows = - realtime::sparse_vec_to_csr(dec.H_sparse, h_row_ptr, h_col_idx); - if (h_rows != dec.syndrome_size) - throw std::runtime_error( - "test_realtime_qldpc_config_loader: H_sparse row count " + - std::to_string(h_rows) + " does not match dec.syndrome_size " + - std::to_string(dec.syndrome_size)); - - const std::size_t bs = dec.block_size; - const std::size_t ss = dec.syndrome_size; - cudaqx::tensor H_tensor({ss, bs}); - for (std::size_t r = 0; r < ss; ++r) - for (std::uint32_t j = h_row_ptr[r]; j < h_row_ptr[r + 1]; ++j) - H_tensor.at({r, static_cast(h_col_idx[j])}) = 1; - - auto params = dec.decoder_custom_args_to_heterogeneous_map(); - auto plugin = decoder::get("nv-qldpc-decoder", H_tensor, params); - if (!plugin) - throw std::runtime_error( - "test_realtime_qldpc_config_loader: decoder::get(\"nv-qldpc-decoder\"," - " ...) returned nullptr; is the plugin built and discoverable?"); - plugin->set_D_sparse(dec.D_sparse); - plugin->set_O_sparse(dec.O_sparse); - - LoadedDecoder out{}; - out.decoder = std::move(plugin); - - // num_measurements is the highest column index referenced by D_sparse, +1. - // (D_sparse columns are measurement-bit indices; CSR sparse_vec_to_csr - // gives us them via `d_ci`.) - std::vector d_rp, d_ci; - realtime::sparse_vec_to_csr(dec.D_sparse, d_rp, d_ci); - out.num_measurements = 0; - for (auto c : d_ci) - out.num_measurements = - std::max(out.num_measurements, static_cast(c + 1)); - - std::vector o_rp, o_ci; - out.num_observables = realtime::sparse_vec_to_csr(dec.O_sparse, o_rp, o_ci); - - return out; -} - -} // namespace test_realtime_qldpc diff --git a/libs/qec/unittests/realtime/qec_graph_decode_test/qldpc_config_loader.h b/libs/qec/unittests/realtime/qec_graph_decode_test/qldpc_config_loader.h deleted file mode 100644 index 781741526..000000000 --- a/libs/qec/unittests/realtime/qec_graph_decode_test/qldpc_config_loader.h +++ /dev/null @@ -1,45 +0,0 @@ -/****************************************************************-*- C++ -*-**** - * Copyright (c) 2025-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 - -// Tiny shim that does the YAML -> decoder construction in a .cpp TU so the -// .cu test file doesn't have to include cudaq/qec/realtime/decoding_config.h. -// That header pulls in C++20-only `bool operator==(...) const = default;` -// patterns that nvcc 13 chokes on when forced to C++20 (libstdc++ 13 ICE). -// -// The test's .cu file calls `load_decoder_from_yaml` to get back exactly the -// state it needs to bring up the dispatchers and replay syndromes (decoder, -// H_tensor's shape via num_measurements / num_observables, decoder_id is -// always 0 for this single-decoder test). - -#include -#include -#include - -namespace cudaq::qec { -class decoder; -} - -namespace test_realtime_qldpc { - -struct LoadedDecoder { - std::unique_ptr decoder; - std::size_t num_measurements = 0; ///< total per-shot, derived from D_sparse - std::size_t num_observables = 0; ///< rows of O_sparse -}; - -/// Read the YAML file at `yaml_path`, parse it via -/// `cudaq::qec::decoding::config::multi_decoder_config::from_yaml_str`, build -/// the H_tensor, instantiate the nv-qldpc-decoder plugin, and feed it the -/// D_sparse + O_sparse from the YAML. Returns the constructed decoder plus -/// derived shape information. Aborts via std::runtime_error on YAML/decoder -/// errors so the gtest body can ASSERT_NO_THROW around the call. -LoadedDecoder load_decoder_from_yaml(const std::string &yaml_path); - -} // namespace test_realtime_qldpc diff --git a/libs/qec/unittests/realtime/qec_graph_decode_test/test_realtime_qldpc_graph_decoding.cpp b/libs/qec/unittests/realtime/qec_graph_decode_test/test_realtime_qldpc_graph_decoding.cpp deleted file mode 100644 index 9a4690cd5..000000000 --- a/libs/qec/unittests/realtime/qec_graph_decode_test/test_realtime_qldpc_graph_decoding.cpp +++ /dev/null @@ -1,490 +0,0 @@ -/****************************************************************-*- C++ -*-**** - * Copyright (c) 2025-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_realtime_qldpc_graph_decoding.cpp -/// @brief CI test for the per-round CUDA-graph relay BP decode trio, -/// exercising the full shared-ring two-dispatcher path -/// (CUDAQ_DISPATCH_PATH_HOST for per-round enqueue + CUDAQ_DISPATCH_PATH_DEVICE -/// for get_corrections and reset_decoder). Post-Step-8 of the realtime-session -/// refactor, this test consumes `qec_realtime_session` (for ring/dispatcher -/// setup) and `rpc_producer` (for host-side enqueue / get_corrections / reset -/// RPC calls), which are the exact same components surface_code-1-local uses in -/// its CUDAQ_QEC_REALTIME_MODE=inproc_rpc path -- so this test is now a -/// wire-format-and-orchestration-faithful contract test of the production -/// stack rather than a parallel re-implementation. -/// -/// Flow: -/// 1. Loads the relay BP config YAML and the syndrome fixture -/// (post-ROUND_START markers). -/// 2. Creates the nv-qldpc decoder via the .cpp shim around decoder::get() -/// and stores it in a single-element vector owned by the fixture (the -/// session takes a reference to that vector). -/// 3. Constructs a `qec_realtime_session` over that vector + the -/// libcudaq-realtime-dispatch.a launch fn (passed in from the exe -/// because the .a is hidden-visibility and the .so can't reference it -/// directly). `session.initialize()` then: -/// - calls capture_decode_graph() per decoder (each captured graph -/// publishes the canonical kEnqueueSyndromesFunctionId; the -/// session sub-routes them by routing_key = decoder_id per -/// proposals/cudaq_realtime_host_api.bs#host-path-graph-routing-key), -/// - allocates the TWO-RING data backing (rx_data != tx_data) + -/// flags per proposals/decoder_server_runtime.md, -/// - builds the N+2 function table (N GRAPH_LAUNCH sharing -/// function_id + distinct routing_key, 2 DEVICE_CALL), -/// - starts HOST_LOOP (CPU thread) + DEVICE_LOOP (persistent GPU -/// cooperative kernel). -/// 4. For each shot: replays per-round measurements via N -/// rpc_producer::enqueue_syndromes calls (each returns an empty -/// 24-byte RPCResponse, per the spec's Always-Emitted RPCResponse -/// rule -- no did_decode flag on the wire), then one -/// rpc_producer::get_corrections (asserts first byte matches -/// SyndromeEntry::expected_correction). -/// 5. After the loop: one rpc_producer::reset_decoder, then one more -/// rpc_producer::get_corrections, asserting the corrections buffer is -/// now zero. - -#include - -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include - -#include "cudaq/qec/decoder.h" - -#include "cudaq/realtime/daemon/dispatcher/cudaq_realtime.h" -#include "cudaq/realtime/daemon/dispatcher/dispatch_kernel_launch.h" - -#include "qec_realtime_session.h" -#include "rpc_producer.h" -#include "cudaq/qec/realtime/decoder_rpc_wire_format.h" - -// YAML -> decoder construction lives in a .cpp shim so this .cu file doesn't -// need to include decoding_config.h, which uses C++20 `= default` operator== -// patterns that nvcc 13 ICEs on under -std=c++20 + libstdc++ 13. -#include "qldpc_config_loader.h" - -#ifndef TEST_DATA_DIR -#define TEST_DATA_DIR "." -#endif - -#define CUDA_CHECK(call) \ - do { \ - cudaError_t err = call; \ - ASSERT_EQ(err, cudaSuccess) << "CUDA error: " << cudaGetErrorString(err); \ - } while (0) - -using namespace cudaq::qec; -using namespace cudaq::realtime; - -//============================================================================== -// Syndrome file loader (unchanged from pre-Step-8 -- this only touches the -// fixture format, not the dispatcher contract). -//============================================================================== - -struct SyndromeEntry { - // Flat per-shot bits (all rounds concatenated). Kept for backward - // compatibility with the pre-migration test body. Total size equals - // num_measurements when the shot is complete. - std::vector measurements; - // Authoritative per-round slices, populated from ROUND_START markers. - // Each inner vector is one round's worth of measurement bytes - // (measurements_per_round bytes for nv-qldpc). num_rounds = inner size. - std::vector> per_round_measurements; - uint8_t expected_correction; -}; - -static std::vector load_syndromes(const std::string &path, - std::size_t num_measurements) { - std::ifstream file(path); - if (!file.is_open()) - return {}; - - std::vector entries; - std::string line; - bool in_corrections = false; - std::size_t correction_idx = 0; - // Per-round slicing state, valid only while not in CORRECTIONS_START block. - // saw_round_start_in_current_shot lets us assert that every SHOT_START - // block (post-migration) contains at least one ROUND_START before either - // the next SHOT_START or CORRECTIONS_START -- per the recorder-round-markers - // todo "fail-fast (assertion) if any shot lacks ROUND_START markers". - bool saw_round_start_in_current_shot = false; - - auto seal_current_shot = [&]() { - if (entries.empty()) - return; - if (!saw_round_start_in_current_shot) { - throw std::runtime_error( - "test_realtime_qldpc_graph_decoding: load_syndromes: shot " + - std::to_string(entries.size() - 1) + - " is missing ROUND_START markers. Re-record the fixture (see " - "re-record-relay-fixture todo) -- pre-marker fixtures cannot be " - "consumed by this loader because per-round slicing is mandatory " - "for the migrated dispatcher-contract test."); - } - }; - - while (std::getline(file, line)) { - if (line.empty()) - continue; - if (line.rfind("NUM_DATA", 0) == 0 || line.rfind("NUM_LOGICAL", 0) == 0) - continue; - if (line.rfind("CORRECTIONS_START", 0) == 0) { - seal_current_shot(); - in_corrections = true; - correction_idx = 0; - continue; - } - if (line.rfind("CORRECTIONS_END", 0) == 0) - break; - - if (line.rfind("SHOT_START", 0) == 0) { - seal_current_shot(); - entries.emplace_back(); - entries.back().measurements.reserve(num_measurements); - entries.back().expected_correction = 0; - saw_round_start_in_current_shot = false; - continue; - } - - if (line.rfind("ROUND_START", 0) == 0) { - if (entries.empty()) { - throw std::runtime_error( - "test_realtime_qldpc_graph_decoding: load_syndromes: " - "ROUND_START before any SHOT_START in fixture: " + - path); - } - entries.back().per_round_measurements.emplace_back(); - saw_round_start_in_current_shot = true; - continue; - } - - if (in_corrections) { - if (correction_idx < entries.size()) - entries[correction_idx].expected_correction = - static_cast(std::stoi(line)); - correction_idx++; - } else if (!entries.empty()) { - uint8_t bit = static_cast(std::stoi(line)); - entries.back().measurements.push_back(bit); - if (saw_round_start_in_current_shot) - entries.back().per_round_measurements.back().push_back(bit); - } - } - if (!in_corrections) - seal_current_shot(); - return entries; -} - -//============================================================================== -// GTest fixture -//============================================================================== - -class GraphDecodeTest : public ::testing::Test { -protected: - // The decoder vector that backs the session. Single-element today; the - // session is constructed with a reference to this so the test (and the - // session) share ownership semantics with the production path's - // `g_decoders` global. - std::vector> decoders_; - - // The realtime session abstracts: shared ring buffer, function table, - // HOST_LOOP CPU dispatcher (for per-round GRAPH_LAUNCH enqueue), and - // DEVICE_LOOP persistent GPU dispatcher (for DEVICE_CALL get_corrections - // and reset_decoder). Lifetime managed by SetUp/TearDown so each test - // method gets a clean dispatcher pair. - std::unique_ptr session_; - - std::vector syndromes_; - std::size_t num_measurements_ = 0; - std::size_t num_observables_ = 0; - std::size_t num_rounds_ = 0; - std::size_t measurements_per_round_ = 0; - // decoder_id chosen by the plugin at capture_decode_graph() time; lives - // in the [0, kNvQldpcMaxDecoders) range. Cached so we can echo it into - // every RPC payload (enqueue / get_corrections / reset). Session - // initialize() walks decoders_ from index 0; this test creates exactly - // one decoder, so decoder_id_ is 0. - std::size_t decoder_id_ = 0; - - void SetUp() override { - int device_count = 0; - cudaGetDeviceCount(&device_count); - if (device_count == 0) - GTEST_SKIP() << "No CUDA devices available"; - - // The self-relaunching device-graph scheduler this test drives uses - // device-side graph launch, which requires compute capability 9.0+ - // (Hopper). Skip on older GPUs (e.g. A100/sm_80 CI runners), matching the - // cuda-quantum dispatch-kernel tests. - int device = 0; - cudaGetDevice(&device); - cudaDeviceProp prop{}; - cudaGetDeviceProperties(&prop, device); - if (prop.major < 9) - GTEST_SKIP() << "Graph device launch requires compute capability 9.0+, " - "found " - << prop.major << "." << prop.minor; - - cudaError_t flags_err = cudaSetDeviceFlags(cudaDeviceMapHost); - ASSERT_TRUE(flags_err == cudaSuccess || - flags_err == cudaErrorSetOnActiveProcess); - - // ---- Load config + build decoder via the .cpp shim ---- - test_realtime_qldpc::LoadedDecoder loaded; - try { - loaded = test_realtime_qldpc::load_decoder_from_yaml( - std::string(TEST_DATA_DIR) + "/config_nv_qldpc_relay.yml"); - } catch (const std::exception &e) { - FAIL() << "load_decoder_from_yaml threw: " << e.what(); - } - num_measurements_ = loaded.num_measurements; - num_observables_ = loaded.num_observables; - ASSERT_NE(loaded.decoder, nullptr); - printf("Config: num_measurements=%zu, num_observables=%zu\n", - num_measurements_, num_observables_); - - decoders_.clear(); - decoders_.push_back(std::move(loaded.decoder)); - - // ---- Load syndromes (post-ROUND_START fixture) ---- - syndromes_ = load_syndromes(std::string(TEST_DATA_DIR) + - "/syndromes_nv_qldpc_relay.txt", - num_measurements_); - printf("Loaded %zu test syndromes\n", syndromes_.size()); - ASSERT_GT(syndromes_.size(), 0u); - - // Per-round shape: take it from the first shot's slicing; assert - // remaining shots match (otherwise the fixture is internally - // inconsistent and the per-round dispatch contract has no meaning). - ASSERT_GT(syndromes_[0].per_round_measurements.size(), 0u); - num_rounds_ = syndromes_[0].per_round_measurements.size(); - measurements_per_round_ = syndromes_[0].per_round_measurements[0].size(); - ASSERT_GT(measurements_per_round_, 0u); - for (std::size_t i = 0; i < syndromes_.size(); ++i) { - ASSERT_EQ(syndromes_[i].per_round_measurements.size(), num_rounds_) - << "Shot " << i << " has " - << syndromes_[i].per_round_measurements.size() << " rounds; expected " - << num_rounds_; - for (std::size_t r = 0; r < num_rounds_; ++r) - ASSERT_EQ(syndromes_[i].per_round_measurements[r].size(), - measurements_per_round_) - << "Shot " << i << " round " << r << " has wrong measurement count"; - } - // Sanity-check that the per-round slicing covers exactly the same - // measurements the decoder's D matrix expects. - ASSERT_EQ(num_rounds_ * measurements_per_round_, num_measurements_); - printf("Per-shot shape: %zu rounds x %zu measurements/round = %zu total\n", - num_rounds_, measurements_per_round_, num_measurements_); - - // ---- Initialize the realtime session ---- - // - // The session needs the dispatch-kernel launch function pointer; that - // symbol lives in libcudaq-realtime-dispatch.a (a static archive with - // hidden visibility), which this exe -- and ONLY this exe -- links. - // The session's .so cannot reference it directly, so we hand it in as - // a constructor parameter. See qec_realtime_session.h for the - // rationale (cudaq_dispatch_launch_fn_t docstring). - session_ = std::make_unique( - decoders_, &cudaq_launch_dispatch_kernel_regular); - try { - session_->initialize(); - } catch (const std::exception &e) { - FAIL() << "qec_realtime_session::initialize threw: " << e.what(); - } - - // Under decoder_server_runtime.md every decoder shares the canonical - // enqueue_syndromes function_id; per-decoder routing is by routing_key. - // rpc_producer::enqueue_syndromes writes that canonical fid into the - // RPCHeader and `decoder_id_` into payload arg0 (which doubles as the - // routing_key the host monitor matches against the function table). - printf("Session initialized: enqueue_fn_id=0x%08X (canonical, " - "decoder_id=%zu), %zu slots\n", - cudaq::qec::decoding::rpc::kEnqueueSyndromesFunctionId, decoder_id_, - session_->num_slots()); - } - - void TearDown() override { - // Session finalize is idempotent + safe to call after a partially- - // initialized SetUp (some ASSERTs above can return before initialize() - // succeeds; the unique_ptr might still be null in that case). - if (session_) { - session_->finalize(); - session_.reset(); - } - - // Drop the decoder vector AFTER the session releases its captured - // graphs (session.finalize() above). Order matters because the - // session holds non-owning pointers into decoders_ for graph release. - decoders_.clear(); - } -}; - -//============================================================================== -// Test: Graph decode of all test syndromes via HOST_LOOP dispatch -// -// Post-Step-8 the test body is reduced to: build a payload, hand it to -// rpc_producer, validate the returned bytes against the fixture. All slot -// leasing / RPCHeader assembly / spin-on-magic logic now lives in -// rpc_producer.cpp and is shared with surface_code-1-local. -//============================================================================== - -TEST_F(GraphDecodeTest, DecodesAllSyndromes) { - ASSERT_NE(session_, nullptr); - auto &session = *session_; - using namespace cudaq::qec::decoding; - - int enqueue_ok = 0; - int correction_matched = 0; - int correction_mismatched = 0; - - using clock_t = std::chrono::high_resolution_clock; - std::vector shot_durations_us; - shot_durations_us.reserve(syndromes_.size()); - - // Re-usable correction output buffer (one shot's worth of observables). - std::vector corrections(num_observables_, 0); - - for (std::size_t shot = 0; shot < syndromes_.size(); ++shot) { - auto t_start = clock_t::now(); - - // ---------------------------------------------------------------------- - // (a) Per-round enqueue. Under decoder_server_runtime.md the - // dispatcher always emits an empty (result_len == 0) RPCResponse - // for enqueue_syndromes -- there is no did_decode flag on the - // wire anymore. Whether a window closed and a decode latched - // is observable post-suite through get_corrections(reset=1). - // ---------------------------------------------------------------------- - for (std::size_t round = 0; round < num_rounds_; ++round) { - const auto &round_bytes = syndromes_[shot].per_round_measurements[round]; - ASSERT_EQ(round_bytes.size(), measurements_per_round_); - - // `tag` carries (shot, round) for diagnostic / ordering checks. - // rpc_producer::enqueue_syndromes writes `tag` into BOTH: - // - payload arg1 (full 64 bits, per decoder_server_runtime.md - // #enqueue_syndromes), and - // - RPCHeader::request_id (low 32 bits, for the realtime layer's - // id-echo correlation). - // Either path is sufficient to correlate the response with the - // request. - const std::uint64_t tag = (static_cast(shot) << 16) | - static_cast(round); - - try { - rpc_producer::enqueue_syndromes( - session, decoder_id_, round_bytes.data(), - static_cast(measurements_per_round_), tag); - } catch (const std::exception &e) { - FAIL() << "rpc_producer::enqueue_syndromes threw at shot " << shot - << " round " << round << ": " << e.what(); - } - enqueue_ok++; - } - - // ---------------------------------------------------------------------- - // (b) get_corrections. reset=1 zeros corrections after fetch so the - // next shot starts from cleared state on the device side. The - // plugin's BP iteration warm-start is governed by reset_decoder - // (issued at end-of-suite below), not by this reset flag. - // ---------------------------------------------------------------------- - std::fill(corrections.begin(), corrections.end(), - static_cast(0xCC)); // poison - try { - rpc_producer::get_corrections( - session, decoder_id_, corrections.data(), - static_cast(num_observables_), - /*reset=*/1); - } catch (const std::exception &e) { - FAIL() << "rpc_producer::get_corrections threw at shot " << shot << ": " - << e.what(); - } - const std::uint8_t got = corrections[0]; - const std::uint8_t expected = syndromes_[shot].expected_correction; - const bool matches = (got == expected); - if (matches) - correction_matched++; - else - correction_mismatched++; - EXPECT_EQ(got, expected) - << "Correction byte 0 mismatch at shot " << shot - << " (got=" << static_cast(got) - << ", expected=" << static_cast(expected) << ")"; - - auto t_end = clock_t::now(); - double duration_us = - std::chrono::duration(t_end - t_start).count(); - shot_durations_us.push_back(duration_us); - - printf("Shot %zu: enqueue_rounds=%zu, corr=[", shot, num_rounds_); - for (std::size_t i = 0; i < num_observables_ && i < 8; i++) { - printf("%u", corrections[i]); - if (i + 1 < num_observables_ && i + 1 < 8) - printf(","); - } - printf("] expected=%u %s (%.1f us shot-total)\n", - static_cast(expected), matches ? "OK" : "MISMATCH", - duration_us); - } - - // -------------------------------------------------------------------------- - // (c) End-of-suite reset_decoder + get_corrections to verify the reset - // path actually zeros device-side state. Without this, a regression - // where reset is a no-op would still pass the per-shot checks - // (because we already pass reset=1 to get_corrections; that uses a - // different code path inside the plugin). - // -------------------------------------------------------------------------- - try { - rpc_producer::reset_decoder(*session_, decoder_id_); - } catch (const std::exception &e) { - FAIL() << "rpc_producer::reset_decoder threw at end-of-suite: " << e.what(); - } - - std::fill(corrections.begin(), corrections.end(), - static_cast(0xCC)); - try { - rpc_producer::get_corrections(*session_, decoder_id_, corrections.data(), - static_cast(num_observables_), - /*reset=*/0); - } catch (const std::exception &e) { - FAIL() << "rpc_producer::get_corrections post-reset threw: " << e.what(); - } - for (std::size_t i = 0; i < num_observables_; ++i) - EXPECT_EQ(corrections[i], 0) - << "reset_decoder did not zero corrections[" << i << "]"; - - printf("\nCompleted: enqueue OK = %d (expected %zu)\n", enqueue_ok, - syndromes_.size() * num_rounds_); - printf("Correction comparison vs fixture: %d matched, %d mismatched " - "(over %zu shots)\n", - correction_matched, correction_mismatched, syndromes_.size()); - - if (shot_durations_us.size() > 1) { - auto begin = shot_durations_us.begin() + 1; - auto end = shot_durations_us.end(); - std::size_t n = std::distance(begin, end); - double sum = std::accumulate(begin, end, 0.0); - double avg = sum / n; - double min_val = *std::min_element(begin, end); - double max_val = *std::max_element(begin, end); - printf("\n[GraphDecodeTiming] shots=%zu (excluding warmup shot 0)\n", n); - printf("[GraphDecodeTiming] min=%.1f us avg=%.1f us max=%.1f us\n", - min_val, avg, max_val); - } - - EXPECT_EQ(enqueue_ok, static_cast(syndromes_.size() * num_rounds_)); - EXPECT_EQ(correction_matched, static_cast(syndromes_.size())); - EXPECT_EQ(correction_mismatched, 0); -} diff --git a/libs/qec/unittests/utils/gpu_roce_qldpc_graph_decoder_bridge.cpp b/libs/qec/unittests/utils/gpu_roce_qldpc_graph_decoder_bridge.cpp index a56404b8d..3630d33db 100644 --- a/libs/qec/unittests/utils/gpu_roce_qldpc_graph_decoder_bridge.cpp +++ b/libs/qec/unittests/utils/gpu_roce_qldpc_graph_decoder_bridge.cpp @@ -12,10 +12,8 @@ /// self-relaunching device-graph scheduler. /// /// This bridge wires the per-round decode-server protocol onto a real -/// (or emulated) FPGA over RoCE. Unlike the inproc_rpc path -/// (qec_realtime_session, which allocates its own pinned ring), the bridge -/// runs the SAME device-graph scheduler directly on the GpuRoceTransceiver DOCA -/// ring: +/// (or emulated) FPGA over RoCE. The bridge runs the device-graph scheduler +/// directly on the GpuRoceTransceiver DOCA ring: /// /// FPGA --RDMA--> GpuRoceTransceiver RX kernel --writes rx_flags--> scheduler /// graph scheduler graph --DEVICE_CALL append/get/reset; fires decode on a @@ -83,9 +81,9 @@ std::string read_file(const std::string &path) { } // Resolve a proprietary DEVICE_CALL populate shim by name and stamp the entry. -// Same dlsym(RTLD_DEFAULT) contract as qec_realtime_session: the symbols are -// exported from this executable because it absorbs the cudevice proprietary -// archive (WHOLE_ARCHIVE) and links with --export-dynamic. +// dlsym(RTLD_DEFAULT) contract: the symbols are exported from this executable +// because it absorbs the cudevice proprietary archive (WHOLE_ARCHIVE) and links +// with --export-dynamic. using populate_fn = void (*)(void *); bool populate_device_call(cudaq_function_entry_t &entry, const char *symbol, std::uint32_t function_id) {