Add hardware pinning for QEC decoders

libs/qec/include/cudaq/qec/decoder.h

@@ -13,6 +13,7 @@
 #include "cuda-qx/core/tensor.h"
 #include "sparse_binary_matrix.h"
 #include "cudaq/qec/detector_error_model.h"
+#include "cudaq/qec/device_affinity.h"
 #include <algorithm>
 #include <functional>
 #include <future>
@@ -245,6 +246,30 @@ class decoder
   std::size_t get_block_size() { return block_size; }
   std::size_t get_syndrome_size() { return syndrome_size; }
 
+  /// @brief Store hardware affinity parameters read from the constructor params
+  /// map. Called by decoder::get() after the plugin constructor returns.
+  /// Not intended for direct use by plugin authors.
+  void set_hardware_params(const cudaqx::heterogeneous_map &params);
+
+  /// @brief Target NUMA node for this decoder (-1 = no binding).
+  int numa_node_id() const { return numa_node_id_; }
+
+  /// @brief Target CUDA device for this decoder (-1 = inherit caller's device).
+  int cuda_device_id() const { return cuda_device_id_; }
+
+  /// @brief Persistently bind the CALLING thread to this decoder's NUMA node
+  /// (and CUDA device). Call once from the thread that will own decode()/
+  /// enqueue for this decoder (e.g. a realtime worker). No restore.
+  /// @return the node bound to, or -1 if nothing was applied. Warns if a node
+  /// was requested (>= 0) but could not be honored.
+  int bind_current_thread();
+
+  /// @brief Run decode(syndrome) on a fresh worker thread bound (via
+  /// bind_current_thread) to this decoder's CUDA device / NUMA node, and return
+  /// the result. Convenience for a one-off pinned decode; for sustained
+  /// throughput drive decode on a long-lived bound thread instead.
+  decoder_result decode_on_pinned_thread(const std::vector<float_t> &syndrome);
+
   // -- Begin realtime decoding API --
 
   // Note: all of the current realtime decoding API is designed to be used with
@@ -357,8 +382,28 @@ class decoder
   /// @brief The decoder's D matrix in sparse format
   std::vector<std::vector<uint32_t>> D_sparse;
 
+  /// Target CUDA device for this decoder. -1 = inherit caller's device (no-op).
+  int cuda_device_id_ = -1;
+  /// Target NUMA node for this decoder. -1 = no binding.
+  int numa_node_id_ = -1;
+  /// Set once bind_current_thread() has pinned the owning thread, so the
+  /// synchronous decode_batch() guard can skip its redundant set/restore.
+  bool bound_persistently_ = false;
+  /// NUMA memory-policy mode for this decoder (default soft PREFERRED).
+  cudaq::qec::mempolicy_mode mempolicy_ = cudaq::qec::mempolicy_mode::preferred;
+  /// Explicit CPU cores for this decoder's owning thread (empty = use node
+  /// cpuset).
+  std::vector<int> cpu_affinity_;
+  /// Set once the current-device mismatch warning has fired, so it prints at
+  /// most once per decoder instance.
+  bool device_mismatch_warned_ = false;
+
 private:
   decode_result_type result_type_ = decode_result_type::decode_to_errs;
+
+  /// Warn (once) if this decoder has an explicit cuda_device_id and the
+  /// calling thread's current CUDA device does not match it.
+  void warn_if_device_mismatch();
 };
 
 /// @brief Convert a single soft probability to a hard 0/1 decision.

libs/qec/include/cudaq/qec/decoder_pool.h

@@ -0,0 +1,60 @@
+/*******************************************************************************
+ * 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
+
+#include "cudaq/qec/decoder.h"
+#include <future>
+#include <memory>
+#include <string>
+#include <unordered_map>
+#include <vector>
+
+namespace cudaq::qec {
+
+/// @brief One decoder's placement + construction inputs for a decoder_pool.
+struct pool_decoder_spec {
+  int id;                            ///< Caller-chosen routing id.
+  std::string name;                  ///< Registered decoder name.
+  cudaqx::tensor<uint8_t> H;         ///< Parity-check matrix.
+  cudaqx::heterogeneous_map options; ///< get() options (cuda_device_id, ...).
+};
+
+/// @brief Runs a set of decoders concurrently, each on its own worker thread
+/// bound (bind_current_thread) to that decoder's CUDA device / NUMA node. The
+/// decoder is constructed on its worker thread so its resources land on-node.
+class decoder_pool {
+public:
+  explicit decoder_pool(std::vector<pool_decoder_spec> specs);
+  ~decoder_pool();
+  decoder_pool(const decoder_pool &) = delete;
+  decoder_pool &operator=(const decoder_pool &) = delete;
+
+  /// @brief Decode every id's syndromes on that id's pinned worker, with all
+  /// decoders running concurrently; blocks until they finish and returns the
+  /// results grouped by id. Each id maps to that decoder's syndromes (the
+  /// per-decoder batched decode happens inside its worker). Throws if an id has
+  /// no matching decoder, or rethrows a worker's decode exception.
+  std::unordered_map<int, std::vector<decoder_result>> decode_all(
+      const std::unordered_map<int, std::vector<std::vector<float_t>>> &work);
+
+  /// @brief Non-blocking streaming submit: enqueue `syndromes` on `id`'s pinned
+  /// worker and return immediately with a future for that chunk's results.
+  /// Submit chunks over time and consume the futures as they resolve to stream
+  /// per id, concurrently; a size-1 chunk streams a single syndrome. Throws if
+  /// `id` has no matching decoder; the future rethrows a worker decode
+  /// exception.
+  std::future<std::vector<decoder_result>>
+  submit(int id, std::vector<std::vector<float_t>> syndromes);
+
+private:
+  struct worker;
+  std::vector<std::unique_ptr<worker>> workers_;
+  std::unordered_map<int, worker *> by_id_;
+};
+
+} // namespace cudaq::qec

libs/qec/include/cudaq/qec/device_affinity.h

@@ -0,0 +1,74 @@
+/*******************************************************************************
+ * 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
+
+#include "cuda-qx/core/heterogeneous_map.h"
+#include <stdexcept>
+#include <string>
+#include <vector>
+
+namespace cudaq::qec {
+
+namespace detail {
+/// Returns -1 if key is absent; throws if stored value is negative.
+/// Handles both int (YAML path) and std::size_t (Python kwargs path) storage.
+inline int read_pin_key(const cudaqx::heterogeneous_map &params,
+                        const std::string &key) {
+  if (!params.contains(key))
+    return -1;
+  int value = params.get<int>(key);
+  if (value < 0)
+    throw std::runtime_error(key + " must be >= 0 (got " +
+                             std::to_string(value) + ")");
+  return value;
+}
+} // namespace detail
+
+/// @brief GPU device id for a decoder. -1 = inherit current device (no-op).
+inline int read_cuda_device_id(const cudaqx::heterogeneous_map &params) {
+  return detail::read_pin_key(params, "cuda_device_id");
+}
+
+/// @brief NUMA node id for a decoder. -1 = no binding.
+inline int read_numa_node_id(const cudaqx::heterogeneous_map &params) {
+  return detail::read_pin_key(params, "numa_node_id");
+}
+
+/// @brief Public NUMA memory-policy selector. preferred = soft; bind = strict.
+enum class mempolicy_mode { preferred, bind };
+
+/// @brief Read "mempolicy": "bind"->bind, "preferred"/absent->preferred, else
+/// throw.
+inline mempolicy_mode read_mempolicy(const cudaqx::heterogeneous_map &params) {
+  if (!params.contains("mempolicy"))
+    return mempolicy_mode::preferred;
+  const std::string v = params.get<std::string>("mempolicy");
+  if (v == "bind")
+    return mempolicy_mode::bind;
+  if (v == "preferred")
+    return mempolicy_mode::preferred;
+  throw std::runtime_error(
+      "mempolicy must be \"preferred\" or \"bind\" (got \"" + v + "\")");
+}
+
+/// @brief Read "cpu_affinity": a list of CPU core ids. Absent -> empty (no
+/// override).
+inline std::vector<int>
+read_cpu_affinity(const cudaqx::heterogeneous_map &params) {
+  if (!params.contains("cpu_affinity"))
+    return {};
+  return params.get<std::vector<int>>("cpu_affinity");
+}
+
+/// @brief NUMA node local to a CUDA device (via PCIe locality), or -1 if the
+/// device id is negative or the topology can't be resolved. Defined in
+/// decoder.cpp.
+int numa_node_for_cuda_device(int cuda_device_id);
+
+} // namespace cudaq::qec

libs/qec/include/cudaq/qec/realtime/decoding_config.h

@@ -179,6 +179,8 @@ struct decoder_config {
   std::vector<std::int64_t> H_sparse;
   std::vector<std::int64_t> O_sparse;
   std::vector<std::int64_t> D_sparse;
+  std::optional<int> cuda_device_id;
+  std::optional<int> numa_node_id;
   std::variant<single_error_lut_config, multi_error_lut_config,
                nv_qldpc_decoder_config, sliding_window_config,
                trt_decoder_config, pymatching_config>

libs/qec/lib/CMakeLists.txt

@@ -41,6 +41,7 @@ endif()
 set(QEC_SOURCES
   code.cpp
   decoder.cpp
+  decoder_pool.cpp
   detector_error_model.cpp
   experiments.cpp
   logger.cpp