[QEC] Decoding server: guard tx-slot writes, drop dead code from #769

libs/qec/lib/realtime/decoding-server-cqr/CMakeLists.txt

@@ -74,8 +74,9 @@ else()
 endif()
 
 # ---------------------------------------------------------------------------
-# Core decoding server: transport abstraction, session registry, per-decoder
-# worker threads, RPC dispatcher.  The wire format (function IDs, payload
+# Core decoding server: transport abstraction, session registry, and the
+# per-decoder sessions whose handle_* entry points serve requests inline on
+# the calling CUDAQ dispatcher thread.  The wire format (function IDs, payload
 # structs, RPCHeader/RPCResponse) comes from cudaq/qec/realtime/
 # decoder_rpc_wire_format.h, which includes the cudaq realtime headers -- so
 # the server (and everything below) is only built when a realtime-enabled
@@ -124,8 +125,8 @@ target_link_libraries(cudaq-qec-decoding-server
     cudaq-qec-decoders
     cudaq-qec-realtime-decoding
   PRIVATE
-    # DecodingSession worker threads pin themselves to their decoder's
-    # cuda_device_id (cudaSetDevice).
+    # DecodingSession pins the calling dispatcher thread to its decoder's
+    # cuda_device_id (pin_decode_device_cached -> cudaSetDevice).
     CUDA::cudart
 )
 

libs/qec/lib/realtime/decoding-server-cqr/DecodingServer.cpp

@@ -13,7 +13,6 @@
 #include "cudaq/qec/realtime/decoding_config.h"
 
 #include <fstream>
-#include <iostream>
 #include <iterator>
 #include <stdexcept>
 
@@ -145,17 +144,6 @@ void DecodingServer::run() {
   stop_cv_.wait(lk, [this] { return shutdown_; });
 }
 
-void DecodingServer::print_session_stats() const {
-  for (const auto &[id, session] : registry_.sessions()) {
-    std::cout << "QEC_DECODING_SERVER_DECODER_STATS id=" << id
-              << " decodes=" << session->decode_count.load()
-              << " enqueues=" << session->enqueue_count.load()
-              << " corrections=" << session->get_corrections_count.load()
-              << " resets=" << session->reset_count.load()
-              << " errors=" << session->error_count.load() << std::endl;
-  }
-}
-
 void DecodingServer::stop() {
   {
     std::lock_guard<std::mutex> lk(stop_mutex_);

libs/qec/lib/realtime/decoding-server-cqr/DecodingServer.h

@@ -59,10 +59,6 @@ class DecodingServer {
   /// Thread-safe; releases run() and shuts the transports down.
   void stop();
 
-  /// Print one QEC_DECODING_SERVER_DECODER_STATS line per session to stdout
-  /// (test/diagnostic evidence; callers gate on QEC_DECODING_SERVER_STATS).
-  void print_session_stats() const;
-
 private:
   /// Create a transceiver for \p dispatch.  Throws for host dispatch (served
   /// by the CQR plugin) and when the device-graph component is not linked.

libs/qec/lib/realtime/decoding-server-cqr/DecodingSession.cpp

@@ -25,7 +25,6 @@ namespace cudaq::qec::decoding_server {
 using cudaq::qec::decoding::rpc::bit_packed_bytes;
 using cudaq::qec::decoding::rpc::RpcStatus;
 using cudaq::realtime::RPCHeader;
-using cudaq::realtime::RPCResponse;
 
 // Busy high-water mark across all sessions (bumped while a session executes
 // a request inline on a dispatcher thread).
@@ -371,10 +370,10 @@ void DecodingSession::handle_get_corrections(const void *rx_slot, void *tx_slot,
     // The corrections pack straight into the tx slot's payload area; the
     // header (and the magic, last) follow in commit().  result-too-large is
     // detected by the core against the slot capacity — no truncation.
+    const std::size_t out_capacity = writer.payload_capacity();
     status = get_corrections_core(req.return_size, req.reset,
-                                  static_cast<uint8_t *>(tx_slot) +
-                                      sizeof(RPCResponse),
-                                  slot_size - sizeof(RPCResponse), out_len);
+                                  writer.payload(out_capacity), out_capacity,
+                                  out_len);
   } catch (const std::exception &e) {
     cudaq::qec::error("DecodingSession::handle_get_corrections: {}", e.what());
     ++error_count;

libs/qec/lib/realtime/decoding-server-cqr/RpcSlot.h

@@ -176,6 +176,15 @@ inline bool peek_decoder_id(const void *rx_slot, std::size_t slot_size,
   return true;
 }
 
+/// Callers gate on `slot_size >= sizeof(RPCHeader)` (dispatch_rpc) and then
+/// write an RPCResponse into a slot of that same size, so the response header
+/// must fit wherever a request header did.  Both are 24 bytes today; assert
+/// the relation rather than leave it implicit -- these structs live in the
+/// cudaq realtime headers and can change independently of this repo.
+static_assert(sizeof(RPCResponse) <= sizeof(RPCHeader),
+              "an RPCResponse must fit in any slot large enough to have "
+              "carried an RPCHeader");
+
 /// Write a header-only RPCResponse (no result payload) into \p tx_slot.
 /// The magic is release-stored LAST so the CUDAQ runtime sees a complete
 /// response before observing the magic word.
@@ -200,9 +209,17 @@ class ResultWriter {
   ResultWriter(void *tx_slot, std::size_t slot_size) noexcept
       : tx_(tx_slot), capacity_(slot_size) {}
 
+  /// Bytes available for the result after the response header, 0 if the slot
+  /// cannot even hold the header.  Order matters: the capacity check must
+  /// precede any subtraction, or a short slot underflows to a huge size_t.
+  std::size_t payload_capacity() const noexcept {
+    if (!tx_ || capacity_ < sizeof(RPCResponse))
+      return 0;
+    return capacity_ - sizeof(RPCResponse);
+  }
+
   uint8_t *payload(std::size_t result_len) noexcept {
-    if (!tx_ || result_len > capacity_ - sizeof(RPCResponse) ||
-        capacity_ < sizeof(RPCResponse))
+    if (!tx_ || result_len > payload_capacity())
       return nullptr;
     return static_cast<uint8_t *>(tx_) + sizeof(RPCResponse);
   }

libs/qec/unittests/test_decoding_server_core.cpp

@@ -178,6 +178,32 @@ TEST(RpcSlotParse, PeeksTheDecoderIdFromEveryRequestKind) {
   EXPECT_FALSE(slot::peek_decoder_id(bare.data(), bare.size(), id));
 }
 
+TEST(RpcSlotWrite, ResultWriterRefusesASlotTooSmallForTheResponseHeader) {
+  // The capacity check must precede the subtraction: a slot shorter than the
+  // response header must report zero room and hand back nullptr, not the huge
+  // capacity an unguarded `slot_size - sizeof(RPCResponse)` underflows to.
+  // get_corrections_core turns that nullptr into INTERNAL_ERROR; a wrapped
+  // capacity would instead let it write past the slot.
+  std::vector<uint8_t> tx(sizeof(RPCResponse) + 8, 0);
+  slot::ResultWriter tight(tx.data(), sizeof(RPCResponse));
+  EXPECT_EQ(tight.payload_capacity(), 0u);
+  EXPECT_NE(tight.payload(0), nullptr);
+  EXPECT_EQ(tight.payload(1), nullptr);
+
+  slot::ResultWriter undersized(tx.data(), sizeof(RPCResponse) - 1);
+  EXPECT_EQ(undersized.payload_capacity(), 0u);
+  EXPECT_EQ(undersized.payload(1), nullptr);
+
+  slot::ResultWriter none(nullptr, 4096);
+  EXPECT_EQ(none.payload_capacity(), 0u);
+  EXPECT_EQ(none.payload(0), nullptr);
+
+  slot::ResultWriter roomy(tx.data(), sizeof(RPCResponse) + 8);
+  EXPECT_EQ(roomy.payload_capacity(), 8u);
+  EXPECT_EQ(roomy.payload(8), tx.data() + sizeof(RPCResponse));
+  EXPECT_EQ(roomy.payload(9), nullptr);
+}
+
 // ---------------------------------------------------------------------------
 // Device resolution + pin probes
 // ---------------------------------------------------------------------------