Add topology-safe live decoder reconfiguration

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

@@ -17,6 +17,7 @@
 #include <fstream>
 #include <iostream>
 #include <iterator>
+#include <mutex>
 #include <stdexcept>
 #include <thread>
 #include <vector>
@@ -87,7 +88,7 @@ DecodingServer::make_transport(DecoderDispatch dispatch,
 
 DecodingServer::DecodingServer(const std::string &config_yaml) {
   // Parse the YAML once: SessionRegistry validates the decoder entries
-  // (including the uniform-dispatch rule — MVP limitation: heterogeneous
+  // (including the uniform-dispatch rule -- MVP limitation: heterogeneous
   // deployments require per-session transceiver binding, deferred to a
   // follow-up once CpuRoce/DeviceGraphTransceiverAdapter are available) and
   // required_dispatch() then drives transceiver creation.
@@ -196,6 +197,7 @@ DecodingServer::DecodingServer(std::vector<std::unique_ptr<ITransceiver>> owned,
 }
 
 void *DecodingServer::graph_resources_for(uint64_t decoder_id) const {
+  std::shared_lock lifecycle_guard(lifecycle_mutex_);
   const auto &sessions = registry_.sessions();
   const auto iter = sessions.find(decoder_id);
   if (iter == sessions.end() || !iter->second->graph_resources)
@@ -213,7 +215,7 @@ DecodingServer::~DecodingServer() {
 }
 
 // ---------------------------------------------------------------------------
-// init — load sessions and register RPC handlers
+// init -- load sessions and register RPC handlers
 // ---------------------------------------------------------------------------
 
 void DecodingServer::init(const std::string &config_yaml) {
@@ -235,12 +237,22 @@ void DecodingServer::install_direct_dispatch() {
 }
 
 void DecodingServer::register_handlers() {
-  // enqueue_syndromes — fire-and-forget at the RPC level; the transport
+  // enqueue_syndromes -- fire-and-forget at the RPC level; the transport
   // layer ACKs delivery (ACCEPTED), and a queue-full drop is reported both
   // here and at the next get_corrections.
   dispatcher_.register_handler(
       kEnqueueSyndromesFunctionId,
       [this](RxFrame frame, ResponseWriter &writer) {
+        if (applying_config_.load(std::memory_order_acquire)) {
+          writer.write_error(RpcStatus::NOT_READY);
+          return;
+        }
+        std::shared_lock lifecycle_guard(lifecycle_mutex_, std::try_to_lock);
+        if (!lifecycle_guard.owns_lock() ||
+            applying_config_.load(std::memory_order_acquire)) {
+          writer.write_error(RpcStatus::NOT_READY);
+          return;
+        }
         if (frame.buf.size() <
             sizeof(RPCHeader) + sizeof(EnqueueRequestPayload)) {
           writer.write_error(RpcStatus::BAD_REQUEST);
@@ -268,9 +280,19 @@ void DecodingServer::register_handlers() {
         }
       });
 
-  // get_corrections — response sent by the worker thread.
+  // get_corrections -- response sent by the worker thread.
   dispatcher_.register_handler(
       kGetCorrectionsFunctionId, [this](RxFrame frame, ResponseWriter &writer) {
+        if (applying_config_.load(std::memory_order_acquire)) {
+          writer.write_error(RpcStatus::NOT_READY);
+          return;
+        }
+        std::shared_lock lifecycle_guard(lifecycle_mutex_, std::try_to_lock);
+        if (!lifecycle_guard.owns_lock() ||
+            applying_config_.load(std::memory_order_acquire)) {
+          writer.write_error(RpcStatus::NOT_READY);
+          return;
+        }
         if (frame.buf.size() <
             sizeof(RPCHeader) + sizeof(GetCorrectionsRequestPayload)) {
           writer.write_error(RpcStatus::BAD_REQUEST);
@@ -296,9 +318,19 @@ void DecodingServer::register_handlers() {
           writer.write_error(RpcStatus::BUSY);
       });
 
-  // reset_decoder — response sent by the worker thread.
+  // reset_decoder -- response sent by the worker thread.
   dispatcher_.register_handler(
       kResetDecoderFunctionId, [this](RxFrame frame, ResponseWriter &writer) {
+        if (applying_config_.load(std::memory_order_acquire)) {
+          writer.write_error(RpcStatus::NOT_READY);
+          return;
+        }
+        std::shared_lock lifecycle_guard(lifecycle_mutex_, std::try_to_lock);
+        if (!lifecycle_guard.owns_lock() ||
+            applying_config_.load(std::memory_order_acquire)) {
+          writer.write_error(RpcStatus::NOT_READY);
+          return;
+        }
         if (frame.buf.size() <
             sizeof(RPCHeader) + sizeof(ResetRequestPayload)) {
           writer.write_error(RpcStatus::BAD_REQUEST);
@@ -352,7 +384,7 @@ void DecodingServer::run() {
   CUDA_QEC_INFO("DecodingServer: starting {} receiver thread(s)",
                 unique_transports.size());
 
-  // All threads share dispatcher_ — routing is by function_id, not transport.
+  // All threads share dispatcher_ -- routing is by function_id, not transport.
   std::vector<std::thread> recv_threads;
   recv_threads.reserve(unique_transports.size());
   for (ITransceiver *t : unique_transports) {
@@ -374,6 +406,7 @@ void DecodingServer::run() {
 }
 
 void DecodingServer::print_session_stats() const {
+  std::shared_lock lifecycle_guard(lifecycle_mutex_);
   for (const auto &[id, session] : registry_.sessions()) {
     std::cout << "QEC_DECODING_SERVER_DECODER_STATS id=" << id
               << " decodes=" << session->decode_count.load()
@@ -384,6 +417,36 @@ void DecodingServer::print_session_stats() const {
   }
 }
 
+ConfigApplyResult DecodingServer::apply_config(
+    const cudaq::qec::decoding::config::multi_decoder_config &config,
+    const std::string &source_name) {
+  bool expected = false;
+  if (!applying_config_.compare_exchange_strong(
+          expected, true, std::memory_order_acq_rel, std::memory_order_acquire))
+    return {ConfigApplyState::busy, "another config apply is in progress"};
+  struct ResetApplyFlag {
+    std::atomic<bool> &flag;
+    ~ResetApplyFlag() { flag.store(false, std::memory_order_release); }
+  } reset_flag{applying_config_};
+
+  // Once applying_config_ is visible, new handlers fail fast. Wait only for
+  // handlers already between admission and queue insertion, then drain and
+  // replace their sessions under exclusive ownership.
+  std::unique_lock lifecycle_guard(lifecycle_mutex_);
+  DeviceGraphLifecycle dg_lifecycle;
+  dg_lifecycle.stop = [this](uint64_t) {
+    if (owned_transports_.size() != 1)
+      return false;
+    return owned_transports_.front()->stop_device_scheduler();
+  };
+  dg_lifecycle.launch = [this](uint64_t, void *graph_resources) {
+    if (!graph_resources || owned_transports_.size() != 1)
+      return false;
+    return owned_transports_.front()->launch_device_scheduler(graph_resources);
+  };
+  return registry_.apply_config(config, source_name, dg_lifecycle);
+}
+
 void DecodingServer::stop() {
   shutdown_.store(true, std::memory_order_release);
   // Unblock any receive loop parked in recv().

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

@@ -15,6 +15,7 @@
 
 #include <atomic>
 #include <memory>
+#include <shared_mutex>
 #include <string>
 #include <unordered_map>
 #include <vector>
@@ -25,7 +26,7 @@ namespace cudaq::qec::decoding_server {
 /// cuda_device_id (-1 when unpinned). An unpinned decoder defaults to device 0.
 int resolve_decode_device(int decoder_pin);
 
-/// Maps function_id → non-owning ITransceiver pointer.
+/// Maps function_id -> non-owning ITransceiver pointer.
 /// Ownership lives in DecodingServer::owned_transports_.
 using TransportMap = std::unordered_map<uint32_t, ITransceiver *>;
 
@@ -84,6 +85,14 @@ class DecodingServer {
   /// (test/diagnostic evidence; callers gate on QEC_DECODING_SERVER_STATS).
   void print_session_stats() const;
 
+  /// Apply a decoder configuration without rebinding process-owned transports
+  /// or per-decoder rings. Requests fail fast with NOT_READY while the apply
+  /// owns the lifecycle lock. See SessionRegistry::apply_config for the
+  /// topology and device_graph restrictions.
+  ConfigApplyResult
+  apply_config(const cudaq::qec::decoding::config::multi_decoder_config &config,
+               const std::string &source_name = "<live-config>");
+
 private:
   void init(const std::string &config_yaml);
   void register_handlers();
@@ -113,8 +122,10 @@ class DecodingServer {
   // registry_ must be declared BEFORE owned_transports_.
   SessionRegistry registry_;
   RpcDispatcher dispatcher_;
+  mutable std::shared_mutex lifecycle_mutex_;
+  std::atomic<bool> applying_config_{false};
   std::atomic<bool> shutdown_{false};
-  /// Maps function_id → transceiver; used to deduplicate receiver threads.
+  /// Maps function_id -> transceiver; used to deduplicate receiver threads.
   /// Routing within the server is by function_id, not by decoder_id.
   TransportMap function_transport_;
   std::vector<std::unique_ptr<ITransceiver>> owned_transports_;

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

@@ -277,8 +277,13 @@ void DeviceGraphRingConsumer::shutdown() {
     if (device_pinned)
       cudaSetDevice(prev_device);
   }
-  if (shutdown_host_)
+  if (shutdown_host_) {
     __atomic_store_n(shutdown_host_, 1, __ATOMIC_RELEASE);
+    // The device scheduler polls this mapped host-memory flag. In particular
+    // on weakly ordered ARM hosts, publish the store before waiting for the
+    // self-relaunch chain to drain in the destructor.
+    __sync_synchronize();
+  }
 }
 
 std::uint64_t DeviceGraphRingConsumer::dispatched() const {

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

@@ -241,6 +241,13 @@ DeviceGraphTransceiver::DeviceGraphTransceiver(const DeviceGraphConfig &config)
 // ---------------------------------------------------------------------------
 
 void DeviceGraphTransceiver::launch_scheduler(void *raw_graph_resources) {
+  if (stopped_.load(std::memory_order_acquire))
+    throw std::runtime_error(
+        "DeviceGraphTransceiver::launch_scheduler: transport is stopped");
+  if (consumer_)
+    throw std::runtime_error(
+        "DeviceGraphTransceiver::launch_scheduler: scheduler already active");
+
   // All scheduler wiring (pinned function table + populate shims + dispatch
   // graph create/launch) lives in DeviceGraphRingConsumer; this transceiver
   // contributes only its provider's ring context and geometry.
@@ -251,26 +258,40 @@ void DeviceGraphTransceiver::launch_scheduler(void *raw_graph_resources) {
   ring.tx_data = tx_ring_data_;
   ring.rx_stride_sz = page_size_;
   ring.tx_stride_sz = page_size_;
-  consumer_ = std::make_unique<DeviceGraphRingConsumer>(
+  auto next_consumer = std::make_unique<DeviceGraphRingConsumer>(
       ring, num_pages_, page_size_, gpu_id_, raw_graph_resources);
 
-  // Start the provider's I/O loop (Hololink RX/TX kernels + monitor thread,
-  // owned by the provider) now that the scheduler is polling the rings.
-  if (cudaq_bridge_launch(bridge_) != CUDAQ_OK) {
-    consumer_->shutdown();
-    throw std::runtime_error(
-        "DeviceGraphTransceiver::launch_scheduler: provider launch() failed");
+  // Start the provider's I/O loop on the first scheduler bind. Live decoder
+  // reloads preserve that provider and only replace the scheduler consumer.
+  if (!provider_launched_) {
+    if (cudaq_bridge_launch(bridge_) != CUDAQ_OK) {
+      next_consumer->shutdown();
+      throw std::runtime_error(
+          "DeviceGraphTransceiver::launch_scheduler: provider launch() "
+          "failed");
+    }
+    provider_launched_ = true;
+
+    // Publish the endpoint only once. A decoder reload must not imply that the
+    // stable QP, rkey, or ring allocation was rebound.
+    std::cout << "QEC_DECODING_SERVER_ENDPOINT " << endpoint_info_ << "\n";
+    std::cout.flush();
   }
+  consumer_ = std::move(next_consumer);
 
   CUDA_QEC_INFO("DeviceGraphTransceiver: GPU scheduler launched ({})",
                 endpoint_info_);
+}
 
-  // Publish the provider's endpoint description VERBATIM so the
-  // orchestration layer can scrape whatever rendezvous tokens its wire
-  // needs (qp=/rkey=/buffer_addr= for RDMA playback, port= for sockets).
-  // This class does not know or care which tokens are present.
-  std::cout << "QEC_DECODING_SERVER_ENDPOINT " << endpoint_info_ << "\n";
-  std::cout.flush();
+bool DeviceGraphTransceiver::stop_device_scheduler() {
+  if (!consumer_)
+    return true;
+  consumer_->shutdown();
+  consumer_.reset();
+  CUDA_QEC_INFO(
+      "DeviceGraphTransceiver: GPU scheduler stopped; provider preserved ({})",
+      endpoint_info_);
+  return true;
 }
 
 // ---------------------------------------------------------------------------
@@ -302,9 +323,7 @@ void DeviceGraphTransceiver::shutdown() {
   if (stopped_.exchange(true, std::memory_order_acq_rel))
     return; // already stopped
 
-  // Signal the GPU scheduler's self-relaunch loop to stop.
-  if (consumer_)
-    consumer_->shutdown();
+  stop_device_scheduler();
 
   // Stop the Hololink RX/TX kernels and join the provider's monitor thread.
   if (bridge_)
@@ -314,14 +333,10 @@ void DeviceGraphTransceiver::shutdown() {
 DeviceGraphTransceiver::~DeviceGraphTransceiver() {
   // Ensure clean shutdown even if the caller omitted shutdown().
   if (!stopped_.exchange(true, std::memory_order_acq_rel)) {
-    if (consumer_)
-      consumer_->shutdown();
+    stop_device_scheduler();
     if (bridge_)
       cudaq_bridge_disconnect(bridge_);
   }
-  // Drain + destroy the scheduler BEFORE the provider (it polls the
-  // provider's ring memory).
-  consumer_.reset();
   if (bridge_)
     cudaq_bridge_destroy(bridge_);
 }

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

@@ -99,8 +99,8 @@ class DeviceGraphTransceiver final : public ITransceiver {
   ~DeviceGraphTransceiver() override;
 
   /// Wire the DOCA ring buffers to the CUDAQ device-graph scheduler and launch
-  /// the GPU dispatch loop.  Must be called exactly once after the transceiver
-  /// is created and before `run()`.
+  /// the GPU dispatch loop. May be called again after stop_device_scheduler()
+  /// to bind a replacement decoder graph to the same provider rings.
   ///
   /// \p raw_graph_resources is the `void *` returned by
   /// `decoder::capture_decode_graph()`; it is cast internally to
@@ -113,6 +113,10 @@ class DeviceGraphTransceiver final : public ITransceiver {
     return true;
   }
 
+  /// Stop and destroy the scheduler without disconnecting the provider or
+  /// releasing its rings. Idempotent.
+  bool stop_device_scheduler() override;
+
   /// Block until shutdown() is called.  The GPU scheduler handles RX/TX;
   /// this method only satisfies the ITransceiver contract for DecodingServer.
   RxFrame recv() override;
@@ -149,6 +153,7 @@ class DeviceGraphTransceiver final : public ITransceiver {
   // launch_scheduler; owns all scheduler-side CUDA state).
   std::unique_ptr<DeviceGraphRingConsumer> consumer_;
 
+  bool provider_launched_{false};
   std::atomic<bool> stopped_{false};
 };
 

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

@@ -112,6 +112,10 @@ struct ITransceiver {
     return false;
   }
 
+  /// Stop and destroy only the device scheduler while preserving the
+  /// transport provider and its ring allocation.
+  virtual bool stop_device_scheduler() { return false; }
+
   virtual ~ITransceiver() = default;
 };
 

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

@@ -7,6 +7,7 @@
  ******************************************************************************/
 
 #include "RpcDispatcher.h"
+#include "SessionRegistry.h"
 
 #include "cudaq/qec/logger.h"
 
@@ -69,6 +70,8 @@ void RpcDispatcher::dispatch(RxFrame frame, ITransceiver &transport) {
 
   try {
     it->second(std::move(frame), writer);
+  } catch (const SessionNotReady &) {
+    writer.write_error(RpcStatus::NOT_READY);
   } catch (const std::out_of_range &) {
     // SessionRegistry::get() throws std::out_of_range for unknown decoder_id.
     writer.write_error(RpcStatus::INVALID_DECODER);