[Store] Reuse an in-process master for tests and evaluation

.github/workflows/ci.yml

@@ -87,6 +87,13 @@ jobs:
         bash scripts/test_installation.sh
       shell: bash
 
+    - name: Run in-process master e2e tests (no mooncake_master)
+      run: |
+        source test_env/bin/activate
+        bash scripts/run_standalone_store_e2e.sh
+        deactivate
+      shell: bash
+
     - name: Run tensor snapshot regressions
       env:
         MASTER_SERVER: "127.0.0.1:50061"

mooncake-integration/store/store_py.cpp

@@ -2407,7 +2407,8 @@ PYBIND11_MODULE(store, m) {
                const std::string &ssd_offload_path = "",
                const std::string &tenant_id = "default",
                bool enable_client_http_server = false,
-               int client_http_port = DEFAULT_CLIENT_HTTP_PORT) {
+               int client_http_port = DEFAULT_CLIENT_HTTP_PORT,
+               bool enable_embedded_master = false) {
                 auto real_client = self.init_real_client();
                 std::shared_ptr<mooncake::TransferEngine> transfer_engine =
                     nullptr;
@@ -2418,12 +2419,15 @@ PYBIND11_MODULE(store, m) {
                 int ret;
                 {
                     py::gil_scoped_release release;
-                    ret = real_client->setup_real(
+                    auto result = real_client->setup_internal(
                         local_hostname, metadata_server, global_segment_size,
                         local_buffer_size, protocol, rdma_devices,
-                        master_server_addr, transfer_engine, "",
-                        enable_ssd_offload, ssd_offload_path, tenant_id,
-                        enable_client_http_server, client_http_port);
+                        master_server_addr, transfer_engine, "", 50052,
+                        enable_ssd_offload, true, ssd_offload_path, tenant_id,
+                        enable_client_http_server, client_http_port,
+                        enable_embedded_master);
+                    ret = result.has_value() ? 0
+                                             : static_cast<int>(result.error());
                 }
                 return ret;
             },
@@ -2434,7 +2438,8 @@ PYBIND11_MODULE(store, m) {
             py::arg("enable_ssd_offload") = false,
             py::arg("ssd_offload_path") = "", py::arg("tenant_id") = "default",
             py::arg("enable_client_http_server") = false,
-            py::arg("client_http_port") = DEFAULT_CLIENT_HTTP_PORT)
+            py::arg("client_http_port") = DEFAULT_CLIENT_HTTP_PORT,
+            py::arg("enable_embedded_master") = false)
         .def(
             "setup",
             [](MooncakeStorePyWrapper &self, const py::dict &config_dict) {
@@ -2474,7 +2479,10 @@ PYBIND11_MODULE(store, m) {
             "  tenant_id: Tenant identifier (default 'default').\n"
             "  enable_client_http_server: Enable client HTTP endpoints "
             "(default false).\n"
-            "  client_http_port: Client HTTP metrics port (default 9300).")
+            "  client_http_port: Client HTTP metrics port (default 9300).\n"
+            "  enable_embedded_master: Start an in-process master so no "
+            "external mooncake_master is required (default false). This is "
+            "test/eval plumbing, not a documented deployment mode.")
         .def(
             "setup_dummy",
             [](MooncakeStorePyWrapper &self, size_t mem_pool_size,

mooncake-store/include/embedded_master.h

@@ -0,0 +1,59 @@
+#pragma once
+
+#include <memory>
+#include <string>
+
+#include <ylt/coro_rpc/coro_rpc_server.hpp>
+
+#include "http_metadata_server.h"
+#include "master_admin_service.h"
+#include "master_config.h"
+#include "rpc_service.h"
+
+namespace mooncake {
+
+// In-process master used by tests and by opt-in enable_embedded_master.
+// When enabled, a store client starts this master in the same process so no
+// external mooncake_master is required. This is not a user-facing deployment
+// mode. All listeners bind to the IPv4 loopback address.
+class EmbeddedMaster {
+   public:
+    EmbeddedMaster() = default;
+    ~EmbeddedMaster() { Stop(); }
+
+    EmbeddedMaster(const EmbeddedMaster&) = delete;
+    EmbeddedMaster& operator=(const EmbeddedMaster&) = delete;
+    EmbeddedMaster(EmbeddedMaster&&) = delete;
+    EmbeddedMaster& operator=(EmbeddedMaster&&) = delete;
+
+    bool Start(InProcMasterConfig config);
+    void Stop();
+
+    int rpc_port() const { return rpc_port_; }
+    int http_metrics_port() const { return http_metrics_port_; }
+    int http_metadata_port() const { return http_metadata_port_; }
+    std::string master_address() const {
+        return std::string("127.0.0.1:") + std::to_string(rpc_port_);
+    }
+    std::string metadata_url() const {
+        if (http_metadata_port_ <= 0) return {};
+        return std::string("http://127.0.0.1:") +
+               std::to_string(http_metadata_port_) + "/metadata";
+    }
+    std::string http_metrics_base() const {
+        return std::string("http://127.0.0.1:") +
+               std::to_string(http_metrics_port_);
+    }
+    std::shared_ptr<WrappedMasterService> service() const { return wrapped_; }
+
+   private:
+    std::unique_ptr<coro_rpc::coro_rpc_server> server_;
+    std::shared_ptr<WrappedMasterService> wrapped_;
+    std::unique_ptr<MasterAdminServer> admin_server_;
+    std::unique_ptr<HttpMetadataServer> meta_server_;
+    int rpc_port_ = 0;
+    int http_metrics_port_ = 0;
+    int http_metadata_port_ = 0;
+};
+
+}  // namespace mooncake

mooncake-store/include/real_client.h

@@ -629,7 +629,8 @@ class RealClient : public PyClient {
         const std::string &ssd_offload_path = "",
         const std::string &tenant_id = "default",
         bool enable_client_http_server = false,
-        int client_http_port = DEFAULT_CLIENT_HTTP_PORT);
+        int client_http_port = DEFAULT_CLIENT_HTTP_PORT,
+        bool enable_embedded_master = false);
 
     // Overload that accepts a configuration dictionary
     tl::expected<void, ErrorCode> setup_internal(const ConfigDict &config);
@@ -1151,6 +1152,11 @@ class RealClient : public PyClient {
         size_t local_buffer_size);
 
    private:
+    tl::expected<std::string, ErrorCode> StartEmbeddedMaster(
+        const std::string &master_server_addr, bool enable_ssd_offload);
+
+    std::unique_ptr<class EmbeddedMaster> embedded_master_;
+
     struct SessionRangeReadRequest {
         std::string object_key;
         size_t result_index;

mooncake-store/include/types.h

@@ -222,6 +222,8 @@ constexpr const char* CONFIG_KEY_TENANT_ID = "tenant_id";
 constexpr const char* CONFIG_KEY_ENABLE_CLIENT_HTTP_SERVER =
     "enable_client_http_server";
 constexpr const char* CONFIG_KEY_CLIENT_HTTP_PORT = "client_http_port";
+constexpr const char* CONFIG_KEY_ENABLE_EMBEDDED_MASTER =
+    "enable_embedded_master";
 
 // Store client configuration defaults
 static constexpr size_t DEFAULT_GLOBAL_SEGMENT_SIZE = 1024 * 1024 * 16;  // 16MB

mooncake-store/src/CMakeLists.txt

@@ -52,6 +52,7 @@ set(MOONCAKE_STORE_MASTER_SOURCES
     master_admin_service.cpp
     offset_allocator.cpp
     http_metadata_server.cpp
+    embedded_master.cpp
     serialize/serializer.cpp
     storage/distributed/shard_allocator.cpp
     storage/distributed/immutable_bucket_allocator.cpp

mooncake-store/src/embedded_master.cpp

@@ -0,0 +1,213 @@
+#include "embedded_master.h"
+
+#include <algorithm>
+#include <chrono>
+#include <cstdlib>
+#include <exception>
+#include <string_view>
+#include <thread>
+#include <vector>
+
+#include <glog/logging.h>
+#include <ylt/coro_rpc/coro_rpc_server.hpp>
+
+#include "common/network.h"
+#include "http_metadata_server.h"
+#include "master_admin_service.h"
+#include "types.h"
+
+namespace mooncake {
+
+bool EmbeddedMaster::Start(InProcMasterConfig config) {
+    if (server_) {
+        LOG(ERROR) << "Embedded master is already running on port "
+                   << rpc_port_;
+        return false;
+    }
+    try {
+        // Allocate all needed ports atomically to avoid collisions from
+        // rapid sequential getFreeTcpPort() calls (TOCTOU).
+        int needed = (!config.rpc_port.has_value()) +
+                     (!config.http_metrics_port.has_value()) +
+                     (!config.http_metadata_port.has_value());
+        std::vector<int> available_ports;
+        if (needed > 0) {
+            std::vector<int> reserved_ports;
+            if (config.rpc_port.has_value()) {
+                reserved_ports.push_back(config.rpc_port.value());
+            }
+            if (config.http_metrics_port.has_value()) {
+                reserved_ports.push_back(config.http_metrics_port.value());
+            }
+            if (config.http_metadata_port.has_value()) {
+                reserved_ports.push_back(config.http_metadata_port.value());
+            }
+            auto free_ports = getFreeTcpPorts(
+                needed + static_cast<int>(reserved_ports.size()));
+            available_ports.reserve(needed);
+            for (int port : free_ports) {
+                if (std::find(reserved_ports.begin(), reserved_ports.end(),
+                              port) == reserved_ports.end()) {
+                    available_ports.push_back(port);
+                    if (static_cast<int>(available_ports.size()) == needed) {
+                        break;
+                    }
+                }
+            }
+            if (static_cast<int>(available_ports.size()) < needed) {
+                LOG(ERROR) << "Failed to reserve " << needed
+                           << " free TCP ports for embedded master";
+                return false;
+            }
+        }
+        int idx = 0;
+        rpc_port_ = config.rpc_port.has_value() ? config.rpc_port.value()
+                                                : available_ports[idx++];
+        http_metrics_port_ = config.http_metrics_port.has_value()
+                                 ? config.http_metrics_port.value()
+                                 : available_ports[idx++];
+        http_metadata_port_ = config.http_metadata_port.has_value()
+                                  ? config.http_metadata_port.value()
+                                  : available_ports[idx++];
+
+        if (http_metadata_port_ > 0) {
+            meta_server_ = std::make_unique<HttpMetadataServer>(
+                static_cast<uint16_t>(http_metadata_port_), "127.0.0.1");
+            if (!meta_server_->start()) {
+                LOG(ERROR) << "Failed to start embedded HTTP metadata server "
+                              "on port "
+                           << http_metadata_port_;
+                Stop();
+                return false;
+            }
+        }
+
+        server_ = std::make_unique<coro_rpc::coro_rpc_server>(
+            /*thread_num=*/4, /*port=*/rpc_port_, /*address=*/"127.0.0.1",
+            std::chrono::seconds(0), /*tcp_no_delay=*/true);
+        const char* value = std::getenv("MC_RPC_PROTOCOL");
+        if (value && std::string_view(value) == "rdma") {
+#ifdef YLT_ENABLE_IBV
+            server_->init_ibv();
+#else
+            LOG(WARNING)
+                << "RDMA RPC is disabled at compile time; using TCP RPC";
+#endif
+        }
+
+        uint64_t default_kv_lease_ttl = DEFAULT_DEFAULT_KV_LEASE_TTL;
+        if (config.default_kv_lease_ttl.has_value()) {
+            default_kv_lease_ttl = config.default_kv_lease_ttl.value();
+        } else if (const char* ttl_env = std::getenv("DEFAULT_KV_LEASE_TTL")) {
+            char* endptr = nullptr;
+            unsigned long parsed = std::strtoul(ttl_env, &endptr, 10);
+            if (endptr != ttl_env && endptr && *endptr == '\0') {
+                default_kv_lease_ttl = static_cast<uint64_t>(parsed);
+            }
+        }
+
+        WrappedMasterServiceConfig wms_cfg;
+        wms_cfg.default_kv_lease_ttl = default_kv_lease_ttl;
+        wms_cfg.default_kv_soft_pin_ttl = DEFAULT_KV_SOFT_PIN_TTL_MS;
+        wms_cfg.allow_evict_soft_pinned_objects = true;
+        wms_cfg.enable_metric_reporting = false;
+        wms_cfg.enable_offload = config.enable_offload.has_value()
+                                     ? config.enable_offload.value()
+                                     : false;
+        wms_cfg.eviction_ratio = DEFAULT_EVICTION_RATIO;
+        wms_cfg.eviction_high_watermark_ratio =
+            config.eviction_high_watermark_ratio.has_value()
+                ? config.eviction_high_watermark_ratio.value()
+                : DEFAULT_EVICTION_HIGH_WATERMARK_RATIO;
+        wms_cfg.view_version = 0;
+        wms_cfg.enable_ha = false;
+        wms_cfg.http_port = static_cast<uint16_t>(http_metrics_port_);
+        wms_cfg.cluster_id = DEFAULT_CLUSTER_ID;
+        wms_cfg.root_fs_dir = config.root_fs_dir.has_value()
+                                  ? config.root_fs_dir.value()
+                                  : DEFAULT_ROOT_FS_DIR;
+        wms_cfg.memory_allocator = BufferAllocatorType::OFFSET;
+        if (config.enable_disk_eviction.has_value()) {
+            wms_cfg.enable_disk_eviction = config.enable_disk_eviction.value();
+        }
+        if (config.quota_bytes.has_value()) {
+            wms_cfg.quota_bytes = config.quota_bytes.value();
+        }
+
+        wms_cfg.enable_cxl =
+            config.enable_cxl.has_value() ? config.enable_cxl.value() : false;
+        if (config.cxl_path.has_value()) {
+            wms_cfg.cxl_path = config.cxl_path.value();
+        } else if (const char* cxl_path_env = std::getenv("MC_CXL_DEV_PATH")) {
+            wms_cfg.cxl_path = cxl_path_env;
+        }
+
+        if (config.cxl_size.has_value()) {
+            wms_cfg.cxl_size = config.cxl_size.value();
+        } else if (const char* cxl_size_env = std::getenv("MC_CXL_DEV_SIZE")) {
+            char* endptr = nullptr;
+            unsigned long long val = std::strtoull(cxl_size_env, &endptr, 10);
+            if (endptr != cxl_size_env && *endptr == '\0') {
+                wms_cfg.cxl_size = static_cast<size_t>(val);
+            }
+        }
+
+        wrapped_ = std::make_shared<WrappedMasterService>(wms_cfg);
+        admin_server_ = std::make_unique<MasterAdminServer>(
+            static_cast<uint16_t>(http_metrics_port_),
+            /*enable_metric_reporting=*/false, /*http_host=*/"127.0.0.1");
+        if (!admin_server_->Start()) {
+            LOG(ERROR) << "Failed to start embedded master admin server on "
+                          "port "
+                       << http_metrics_port_;
+            Stop();
+            return false;
+        }
+        admin_server_->SetRuntimeState(ha::MasterRuntimeState::kServing);
+        admin_server_->SetServiceDelegate(wrapped_);
+        admin_server_->SetServiceAvailable(true);
+        RegisterRpcService(*server_, *wrapped_);
+
+        auto ec = server_->async_start();
+        if (ec.hasResult()) {
+            LOG(ERROR) << "Failed to start embedded master RPC server on port "
+                       << rpc_port_;
+            Stop();
+            return false;
+        }
+        std::this_thread::sleep_for(std::chrono::milliseconds(200));
+        LOG(INFO) << "Embedded master started on " << master_address()
+                  << ", metrics_port=" << http_metrics_port_
+                  << ", metadata_port=" << http_metadata_port_;
+        return true;
+    } catch (const std::exception& e) {
+        LOG(ERROR) << "Failed to start embedded master: " << e.what();
+        Stop();
+        return false;
+    } catch (...) {
+        LOG(ERROR) << "Failed to start embedded master";
+        Stop();
+        return false;
+    }
+}
+
+void EmbeddedMaster::Stop() {
+    if (admin_server_) {
+        admin_server_->Stop();
+        admin_server_.reset();
+    }
+    if (server_) {
+        server_->stop();
+        server_.reset();
+    }
+    wrapped_.reset();
+    if (meta_server_) {
+        meta_server_->stop();
+        meta_server_.reset();
+    }
+    rpc_port_ = 0;
+    http_metrics_port_ = 0;
+    http_metadata_port_ = 0;
+}
+
+}  // namespace mooncake

mooncake-store/src/http_metadata_server.cpp

@@ -12,7 +12,7 @@ namespace mooncake {
 HttpMetadataServer::HttpMetadataServer(uint16_t port, const std::string& host)
     : port_(port),
       host_(host),
-      server_(std::make_unique<coro_http::coro_http_server>(4, port)),
+      server_(std::make_unique<coro_http::coro_http_server>(4, port, host_)),
       running_(false) {
     init_server();
 }