feat(umbp): Redis/RESP master metadata store backend

.gitmodules

@@ -9,3 +9,6 @@
 	path = 3rdparty/spdk
 	url = https://github.com/spdk/spdk.git
 	shallow = true
+[submodule "3rdparty/redis-plus-plus"]
+	path = 3rdparty/redis-plus-plus
+	url = https://github.com/sewenew/redis-plus-plus.git

include/mori/metrics/prometheus_metrics_server.hpp

@@ -114,6 +114,12 @@ class MetricsServer {
   void observe(std::string_view name, std::string_view help, const std::vector<double>& bounds,
                double value);
 
+  // Record one histogram observation into a LABELED series (e.g.
+  // {op=...,backend=...}). Same first-write-wins bucket layout as the unlabeled
+  // observe(); reuses the labeled-histogram storage that observeAggregated uses.
+  void observe(std::string_view name, std::string_view help, const Labels& labels,
+               const std::vector<double>& bounds, double value);
+
   // Merge a pre-aggregated histogram batch into a labeled series.
   // `bucket_counts` is CUMULATIVE (bucket_counts[i] = #obs with value <=
   // bounds[i]) and is added per-bucket into the stored series; `count` and

setup.py

@@ -381,6 +381,7 @@ def build_extension(self, ext: Extension) -> None:
         build_tests = os.environ.get("BUILD_TESTS", "OFF")
         build_umbp = "ON" if build_umbp_enabled else "OFF"
         build_umbp_spdk = "ON" if build_umbp_spdk_enabled else "OFF"
+        use_redis_backend = os.environ.get("USE_REDIS_BACKEND", "OFF")
         build_xla_ffi_ops = os.environ.get("BUILD_XLA_FFI_OPS", "OFF")
         with_mpi = (
             "ON"
@@ -411,6 +412,7 @@ def build_extension(self, ext: Extension) -> None:
             f"-DBUILD_BENCHMARK={build_benchmark}",
             f"-DBUILD_TESTS={build_tests}",
             f"-DBUILD_UMBP={build_umbp}",
+            f"-DUSE_REDIS_BACKEND={use_redis_backend}",
             f"-DUSE_SPDK={build_umbp_spdk}",
             f"-DWITH_MPI={with_mpi}",
             "-DBUILD_TORCH_BOOTSTRAP=OFF",

src/application/transport/rdma/providers/ibverbs/ibverbs.cpp

@@ -170,25 +170,32 @@ RdmaEndpoint IBVerbsDeviceContext::CreateRdmaEndpoint(const RdmaEndpointConfig&
   // TODO: we need to add more options in config, include min cqe num for ib_create_cq
   endpoint.ibvHandle.compCh = config.withCompChannel ? ibv_create_comp_channel(context) : nullptr;
   if (config.withCompChannel && !endpoint.ibvHandle.compCh) {
-    MORI_APP_ERROR("ibv_create_comp_channel failed: errno={} ({}); dev={}", errno, strerror(errno),
+    // Capture errno immediately: any intervening libc call before we read it
+    // (e.g. isatty() inside the logging sink when stderr is not a tty, which
+    // sets ENOTTY) would otherwise overwrite the real failure code.
+    const int err = errno;
+    MORI_APP_ERROR("ibv_create_comp_channel failed: errno={} ({}); dev={}", err, strerror(err),
                    GetRdmaDevice()->Name());
-    throw std::runtime_error("ibv_create_comp_channel failed: " + std::string(strerror(errno)));
+    throw std::runtime_error("ibv_create_comp_channel failed: " + std::string(strerror(err)));
   }
 
   endpoint.ibvHandle.cq =
       ibv_create_cq(context, config.maxCqeNum, NULL, endpoint.ibvHandle.compCh, 0);
   if (!endpoint.ibvHandle.cq) {
+    // Capture errno before the lock/log path can overwrite it (see the note at
+    // the comp-channel site).
+    const int err = errno;
     size_t cqPoolSize = 0;
     {
       std::lock_guard<std::mutex> lock(poolMu);
       cqPoolSize = cqPool.size();
     }
     MORI_APP_ERROR(
-        "ibv_create_cq failed: errno={} ({}); dev={} max_cqe={} dev_max_cqe={} cqs_in_pool={}",
-        errno, strerror(errno), GetRdmaDevice()->Name(), config.maxCqeNum,
-        deviceAttr->orig_attr.max_cqe, cqPoolSize);
+        "ibv_create_cq failed: errno={} ({}); dev={} max_cqe={} dev_max_cqe={} cqs_in_pool={}", err,
+        strerror(err), GetRdmaDevice()->Name(), config.maxCqeNum, deviceAttr->orig_attr.max_cqe,
+        cqPoolSize);
     if (endpoint.ibvHandle.compCh) ibv_destroy_comp_channel(endpoint.ibvHandle.compCh);
-    throw std::runtime_error("ibv_create_cq failed: " + std::string(strerror(errno)));
+    throw std::runtime_error("ibv_create_cq failed: " + std::string(strerror(err)));
   }
 
   // TODO: should also manage the lifecycle of completion channel && srq
@@ -216,6 +223,12 @@ RdmaEndpoint IBVerbsDeviceContext::CreateRdmaEndpoint(const RdmaEndpointConfig&
                              .qp_type = IBV_QPT_RC};
   endpoint.ibvHandle.qp = ibv_create_qp(pd, &qpAttr);
   if (!endpoint.ibvHandle.qp) {
+    // Capture errno immediately. Otherwise the lock/log path below overwrites
+    // it before we read it: spdlog's console sink calls isatty(stderr), which
+    // sets errno=ENOTTY when stderr is not a terminal, masking the real error
+    // (commonly EINVAL when the requested QP capacity — max_send_wr x per-WQE
+    // size from sge+inline — exceeds the device's per-QP work-queue budget).
+    const int err = errno;
     size_t qpPoolSize = 0;
     {
       std::lock_guard<std::mutex> lock(poolMu);
@@ -224,13 +237,13 @@ RdmaEndpoint IBVerbsDeviceContext::CreateRdmaEndpoint(const RdmaEndpointConfig&
     MORI_APP_ERROR(
         "ibv_create_qp failed: errno={} ({}); dev={} port={} max_send_wr={} max_recv_wr={} "
         "max_send_sge={} max_cqe={} dev_caps(max_qp_wr={} max_qp={} max_cqe={}) qps_in_pool={}",
-        errno, strerror(errno), GetRdmaDevice()->Name(), config.portId, qpAttr.cap.max_send_wr,
+        err, strerror(err), GetRdmaDevice()->Name(), config.portId, qpAttr.cap.max_send_wr,
         qpAttr.cap.max_recv_wr, qpAttr.cap.max_send_sge, config.maxCqeNum,
         deviceAttr->orig_attr.max_qp_wr, deviceAttr->orig_attr.max_qp,
         deviceAttr->orig_attr.max_cqe, qpPoolSize);
     ibv_destroy_cq(endpoint.ibvHandle.cq);
     if (endpoint.ibvHandle.compCh) ibv_destroy_comp_channel(endpoint.ibvHandle.compCh);
-    throw std::runtime_error("ibv_create_qp failed: " + std::string(strerror(errno)));
+    throw std::runtime_error("ibv_create_qp failed: " + std::string(strerror(err)));
   }
   endpoint.handle.qpn = endpoint.ibvHandle.qp->qp_num;
 

src/metrics/prometheus_metrics_server.cpp

@@ -179,6 +179,25 @@ void MetricsServer::observe(std::string_view name, std::string_view help,
 static std::string FormatLabels(
     const mori::metrics::MetricsServer::Labels& labels);  // forward decl
 
+void MetricsServer::observe(std::string_view name, std::string_view help, const Labels& labels,
+                            const std::vector<double>& bounds, double value) {
+  auto label_str = FormatLabels(labels);
+  std::lock_guard<std::mutex> lk(mutex_);
+  auto& family = labeled_histograms_[std::string(name)];
+  family.help = help;
+  auto& s = family.series[label_str];
+  if (s.bounds.empty()) {
+    s.bounds = bounds;
+    s.bucket_counts.assign(bounds.size(), 0u);
+  }
+  // Increment all cumulative buckets whose upper bound >= value.
+  for (std::size_t i = 0; i < s.bounds.size(); ++i) {
+    if (value <= s.bounds[i]) s.bucket_counts[i]++;
+  }
+  s.count++;
+  s.sum += value;
+}
+
 void MetricsServer::observeAggregated(std::string_view name, std::string_view help,
                                       const Labels& labels, const std::vector<double>& bounds,
                                       const std::vector<uint64_t>& bucket_counts, uint64_t count,

src/umbp/CMakeLists.txt

@@ -309,6 +309,7 @@ add_library(
   ${UMBP_PEER_PROTO_SRCS}
   ${UMBP_PEER_GRPC_SRCS}
   distributed/master/in_memory_master_metadata_store.cpp
+  distributed/master/master_metadata_store_factory.cpp
   distributed/master/master_server.cpp
   distributed/master/master_client.cpp
   distributed/master/rpc_latency_timer.cpp
@@ -347,6 +348,53 @@ target_link_libraries(
 
 target_compile_features(umbp_common PUBLIC cxx_std_17)
 
+# ---------- Optional Redis / RESP master metadata backend --------------------
+# Makes the master stateless by storing all metadata in an external RESP store
+# (Redis / Dragonfly / Valkey). OFF by default so existing builds are
+# unaffected; enable with -DUSE_REDIS_BACKEND=ON. The RESP client seam is built
+# on hiredis (RespClient isolates the client library from the store).
+option(USE_REDIS_BACKEND
+       "Build the RESP-compatible Redis master metadata backend" OFF)
+if(USE_REDIS_BACKEND)
+  find_path(HIREDIS_INCLUDE_DIR hiredis/hiredis.h REQUIRED)
+  find_library(HIREDIS_LIB NAMES hiredis REQUIRED)
+  message(
+    STATUS
+      "UMBP Redis backend ENABLED (hiredis lib: ${HIREDIS_LIB}, include: ${HIREDIS_INCLUDE_DIR})"
+  )
+
+  # redis-plus-plus (3rdparty submodule) — the cluster/sentinel/pooling-capable
+  # client the Redis backend standardizes on. Built from source so any build
+  # image works without an extra system package; only the client library needs
+  # a submodule (hiredis stays the system lib). Static + PIC (umbp_common is
+  # linked into the shared pybind module), C++17, tests off. The store code does
+  # not depend on it yet (that migration is a later step); building it here
+  # de-risks the dependency (toolchain, -Werror, link) before the seam rebase.
+  set(_REDISPP_DIR "${CMAKE_SOURCE_DIR}/3rdparty/redis-plus-plus")
+  if(NOT EXISTS "${_REDISPP_DIR}/CMakeLists.txt")
+    message(
+      FATAL_ERROR
+        "USE_REDIS_BACKEND=ON but 3rdparty/redis-plus-plus is empty; run "
+        "'git submodule update --init --recursive 3rdparty/redis-plus-plus'")
+  endif()
+  set(REDIS_PLUS_PLUS_BUILD_TEST OFF CACHE BOOL "" FORCE)
+  set(REDIS_PLUS_PLUS_BUILD_SHARED OFF CACHE BOOL "" FORCE)
+  set(REDIS_PLUS_PLUS_BUILD_STATIC ON CACHE BOOL "" FORCE)
+  set(REDIS_PLUS_PLUS_CXX_STANDARD 17 CACHE STRING "" FORCE)
+  add_subdirectory("${_REDISPP_DIR}"
+                   "${CMAKE_BINARY_DIR}/3rdparty/redis-plus-plus")
+
+  target_sources(
+    umbp_common PRIVATE distributed/master/redis/resp_client.cpp
+                        distributed/master/redis/resp_cluster_client.cpp
+                        distributed/master/redis_master_metadata_store.cpp)
+  target_include_directories(umbp_common PRIVATE ${HIREDIS_INCLUDE_DIR})
+  # redis++_static carries its own (and hiredis') include dirs; link it before
+  # hiredis so the static archive's hiredis symbols resolve.
+  target_link_libraries(umbp_common PUBLIC redis++_static ${HIREDIS_LIB})
+  target_compile_definitions(umbp_common PUBLIC USE_REDIS_BACKEND)
+endif()
+
 if(MORI_UMBP_TESTING)
   target_compile_definitions(umbp_common PUBLIC MORI_UMBP_TESTING)
 endif()

src/umbp/distributed/master/eviction_manager.cpp

@@ -71,7 +71,13 @@ void EvictionManager::EvictionLoop() {
                    [this] { return !running_.load(std::memory_order_relaxed); });
     }
     if (!running_.load(std::memory_order_relaxed)) break;
-    RunOnce();
+    // A metadata-store hiccup (e.g. RESP transport error) must not kill the
+    // eviction thread; log and retry on the next tick.
+    try {
+      RunOnce();
+    } catch (const std::exception& e) {
+      MORI_UMBP_ERROR("[EvictionManager] metadata-store error: {}", e.what());
+    }
   }
 }