Supercache io pattern

mooncake-store/benchmarks/CMakeLists.txt

@@ -38,6 +38,16 @@ target_link_libraries(
   stress_cluster_bench PRIVATE mooncake_store transfer_engine asio_shared
                                gflags::gflags glog::glog pthread)
 
+# CFM client benchmark. Simulates vLLM inference requests as KV-cache block
+# accesses and reports both client performance and IO Pattern observability.
+add_executable(cfm_client_bench cfm_client_bench.cpp)
+target_link_libraries(
+  cfm_client_bench PRIVATE mooncake_store transfer_engine asio_shared
+                          gflags::gflags glog::glog pthread)
+if(STORE_USE_ETCD)
+  target_link_libraries(cfm_client_bench PRIVATE ${ETCD_WRAPPER_LIB})
+endif()
+
 # Benchmark for vLLM Store Connector path
 # Triggers: batch_put_from_multi_buffers / batchIsExist /
 # batch_get_into_multi_buffers (and setup/register_buffer/tearDownAll).

mooncake-store/include/admission_ops.h

@@ -0,0 +1,3 @@
+#pragma once
+
+#include "io_pattern/ops.h"

mooncake-store/include/cfm_client.h

@@ -0,0 +1,3 @@
+#pragma once
+
+#include "io_pattern/client.h"

mooncake-store/include/cfm_client_impl.h

@@ -0,0 +1,2 @@
+#pragma once
+#include "io_pattern/cfm_client_impl.h"

mooncake-store/include/collector_impl.h

@@ -0,0 +1,2 @@
+#pragma once
+#include "io_pattern/collector_impl.h"

mooncake-store/include/degrading_policy_engine.h

@@ -0,0 +1,2 @@
+#pragma once
+#include "io_pattern/degrading_policy_engine.h"

mooncake-store/include/eviction_ops.h

@@ -0,0 +1,3 @@
+#pragma once
+
+#include "io_pattern/ops.h"

mooncake-store/include/feedback.h

@@ -0,0 +1,2 @@
+#pragma once
+#include "io_pattern/feedback.h"

mooncake-store/include/io_pattern.h

@@ -0,0 +1,3 @@
+#pragma once
+
+#include "io_pattern/io_pattern.h"

mooncake-store/include/io_pattern/analyzer.h

@@ -0,0 +1,19 @@
+#pragma once
+
+#include "io_pattern/types.h"
+
+namespace mooncake::io_pattern {
+
+// Converts an immutable snapshot into workload and per-object features.
+class IoPatternAnalyzer {
+   public:
+    virtual ~IoPatternAnalyzer() = default;
+
+    virtual PatternResult Analyze(const IoPatternSnapshot& snapshot) const = 0;
+    virtual WorkloadType DetectWorkloadType(
+        const IoPatternSnapshot& snapshot) const = 0;
+    virtual float CalculateConfidence(
+        const ObjectRef& object, const IoPatternSnapshot& snapshot) const = 0;
+};
+
+}  // namespace mooncake::io_pattern

mooncake-store/include/io_pattern/cfm_channel.h

@@ -0,0 +1,26 @@
+#pragma once
+
+#include <optional>
+#include <utility>
+
+#include "reporter.h"
+#include "types.h"
+#include "../types.h"
+
+namespace mooncake::io_pattern {
+
+// Transport-neutral CFM reporting channel. In the embedded architecture a
+// SubMaster owns CFM for the keys in its slots, so a channel delivers reports
+// (snapshots and metric batches) and explicit prefetch plans to the owning
+// SubMaster endpoint. There is no policy polling or delivery acknowledgement
+// loop: the receiver merges reports into its local runtime and policy
+// commands execute on the SubMaster that owns the reported keys.
+class CfmChannel {
+   public:
+    virtual ~CfmChannel() = default;
+    virtual bool SendSnapshot(const IoPatternSnapshot& snapshot) = 0;
+    virtual bool SendMetricBatch(const MetricBatch& batch) = 0;
+    virtual ErrorCode ExecutePrefetch(const PrefetchPlan& plan) = 0;
+};
+
+}  // namespace mooncake::io_pattern

mooncake-store/include/io_pattern/cfm_client_impl.h

@@ -0,0 +1,28 @@
+#pragma once
+
+#include <memory>
+#include <utility>
+
+#include "cfm_channel.h"
+#include "client.h"
+
+namespace mooncake::io_pattern {
+
+// Reporting client used by inference-side connectors and integration tests.
+// Network behavior is delegated to the injected channel; the client is what a
+// vLLM/SGLang connector sees when it reports IO Pattern observations for keys
+// owned by the addressed SubMaster.
+class CfmClientImpl final : public CfmClient {
+   public:
+    explicit CfmClientImpl(std::shared_ptr<CfmChannel> channel)
+        : channel_(std::move(channel)) {}
+
+    ErrorCode ReportSnapshot(const IoPatternSnapshot& snapshot) override;
+    ErrorCode ReportMetricBatch(const MetricBatch& batch) override;
+    ErrorCode ExecutePrefetch(const PrefetchPlan& plan) override;
+
+   private:
+    std::shared_ptr<CfmChannel> channel_;
+};
+
+}  // namespace mooncake::io_pattern

mooncake-store/include/io_pattern/cfm_ingress.h

@@ -0,0 +1,28 @@
+#pragma once
+
+#include <memory>
+#include <string_view>
+
+#include "cfm_protocol.h"
+#include "runtime.h"
+
+namespace mooncake::io_pattern {
+
+// Server-side counterpart of CfmRpcChannel. Bind Handle() as an
+// InProcessCfmRpcTransport::SendHandler or adapt it to a network RPC server.
+class CfmIngress final {
+   public:
+    explicit CfmIngress(std::shared_ptr<IoPatternRuntime> runtime,
+                        std::shared_ptr<CfmBinaryCodec> codec =
+                            std::make_shared<CfmBinaryCodec>())
+        : runtime_(std::move(runtime)), codec_(std::move(codec)) {}
+
+    bool Handle(std::string_view method, std::string_view payload,
+                std::string_view source_id = {});
+
+   private:
+    std::shared_ptr<IoPatternRuntime> runtime_;
+    std::shared_ptr<CfmBinaryCodec> codec_;
+};
+
+}  // namespace mooncake::io_pattern

mooncake-store/include/io_pattern/cfm_ownership_client.h

@@ -0,0 +1,83 @@
+#pragma once
+
+#include <atomic>
+#include <chrono>
+#include <functional>
+#include <memory>
+#include <mutex>
+#include <optional>
+#include <string>
+#include <unordered_map>
+#include <vector>
+
+#include "client.h"
+#include "cfm_channel.h"
+#include "cfm_protocol.h"
+#include "cfm_service.h"
+#include "reporter.h"
+#include "rpc_transport.h"
+#include "types.h"
+
+namespace mooncake::io_pattern {
+
+// Resolves the SubMaster that owns a reported object. Production callers feed
+// this from the CVM key->slot->submaster mapping (cvm::KeySlot over the etcd
+// /cvm/<ns> master registry) inside the Store client / connector layer;
+// returns the SubMaster's regular coro_rpc endpoint ("host:port"), the same
+// endpoint every other Mooncake RPC of that SubMaster uses.
+using SubmasterEndpointResolver =
+    std::function<std::optional<std::string>(const TenantId&, const std::string&)>;
+
+// Ownership-addressed CFM reporting client.
+//
+// CFM is a component of every SubMaster; there is no standalone CFM Master
+// and no auth token. A caller (inference connector / Store client) observes
+// keys that may live on many SubMasters, so reports are bucketed by the CVM
+// ownership resolver: observations whose key belongs to the same SubMaster are
+// aggregated into one metric batch and delivered to that SubMaster's embedded
+// CFM over its ordinary coro_rpc endpoint. Metrics whose owner cannot be
+// resolved are dropped and counted so callers can degrade instead of guessing.
+class CfmOwnershipClient final : public CfmClient {
+   public:
+    // `resolver` maps a reported object to the owning SubMaster endpoint. It
+    // must be kept up to date with the CVM route (slot rebalance is rare).
+    // `timeout` applies to every RPC.
+    explicit CfmOwnershipClient(SubmasterEndpointResolver resolver,
+                                std::chrono::milliseconds timeout =
+                                    std::chrono::milliseconds(500));
+
+    // Sends the snapshot, grouped by the owning SubMaster of each key.
+    ErrorCode ReportSnapshot(const IoPatternSnapshot& snapshot) override;
+
+    // Sends the metric batch, grouped by the owning SubMaster of each
+    // inference/access object. Storage observations are deliberately not
+    // routed by default: the SubMaster that owns the underlying storage
+    // already reports its own watermark to its local runtime. Benchmark /
+    // simulation callers can enable forward_storage to have the storage
+    // metrics ride along with each owner-addressed report so a remote run can
+    // drive the report-driven eviction dimension.
+    ErrorCode ReportMetricBatch(const MetricBatch& batch) override;
+
+    void set_forward_storage(bool forward) { forward_storage_ = forward; }
+
+    // Sends an explicit prefetch plan to the SubMaster that owns the first
+    // candidate; that SubMaster executes it through its local storage-safe
+    // handlers.
+    ErrorCode ExecutePrefetch(const PrefetchPlan& plan) override;
+
+    // Observations dropped because no owner could be resolved.
+    uint64_t dropped_observations() const;
+
+   private:
+    std::shared_ptr<CfmChannel> ChannelFor(const std::string& endpoint);
+    std::string OwnerEndpoint(const ObjectRef& object) const;
+
+    SubmasterEndpointResolver resolver_;
+    std::chrono::milliseconds timeout_;
+    std::unordered_map<std::string, std::shared_ptr<CfmChannel>> channels_;
+    mutable std::mutex channels_mutex_;
+    std::atomic<uint64_t> dropped_observations_{0};
+    bool forward_storage_{false};
+};
+
+}  // namespace mooncake::io_pattern

mooncake-store/include/io_pattern/cfm_protocol.h

@@ -0,0 +1,22 @@
+#pragma once
+
+#include "rpc_transport.h"
+
+namespace mooncake::io_pattern {
+
+// Versioned binary wire codec for the CFM RPC methods. It deliberately owns
+// every serialization detail so transports only deal in method/payload bytes.
+class CfmBinaryCodec final : public CfmRpcCodec {
+   public:
+    std::string EncodeSnapshot(const IoPatternSnapshot& snapshot) const override;
+    std::string EncodePrefetch(const PrefetchPlan& plan) const override;
+    std::string EncodeMetricBatch(const MetricBatch& batch) const override;
+    std::optional<PolicyCommand> DecodePolicy(const std::string& payload) const override;
+
+    std::optional<IoPatternSnapshot> DecodeSnapshot(
+        const std::string& payload) const;
+    std::optional<MetricBatch> DecodeMetricBatch(const std::string& payload) const;
+    std::string EncodePolicy(const PolicyCommand& command) const;
+};
+
+}  // namespace mooncake::io_pattern

mooncake-store/include/io_pattern/cfm_service.h

@@ -0,0 +1,67 @@
+#pragma once
+
+#include <chrono>
+#include <cstdint>
+#include <memory>
+#include <optional>
+#include <string>
+#include <string_view>
+#include <utility>
+
+#include "cfm_ingress.h"
+#include "observability.h"
+#include "types.h"
+
+namespace mooncake::io_pattern {
+
+// Embedded CFM endpoint of one SubMaster. There is no standalone CFM Master:
+// every SubMaster runs the full IO Pattern pipeline locally and accepts
+// ownership-addressed metric reports over its regular coro_rpc port (the same
+// endpoint the rest of Mooncake uses). Received batches are merged straight
+// into the local runtime so collection, analysis and policy execution all stay
+// on the SubMaster that owns the reported keys.
+//
+// Reports are addressed by key ownership, so a receiving SubMaster only ever
+// merges observations for keys that belong to its own slots. A separate node
+// identity, delivery queues and a poll/ack loop are therefore unnecessary: the
+// reporting client fans each batch out to the owning SubMaster(s) using the
+// CVM key->slot->submaster mapping.
+class CfmService final {
+   public:
+    explicit CfmService(std::shared_ptr<IoPatternRuntime> runtime);
+
+    // RPC entry point. Supported methods:
+    //   report_metric_batch / report_snapshot  -> merge into the local runtime
+    //   execute_prefetch / execute_policy      -> execute through the local
+    //                                             runtime's storage handlers
+    // `source_id` identifies the reporting process and is used to normalize
+    // remote StorageMetric watermarks.
+    bool Send(std::string_view method, std::string_view payload,
+              std::string_view source_id = {});
+
+    IoPatternSnapshot Snapshot() const;
+    IoPatternObservabilitySnapshot Observability(
+        double window_seconds = 0.0) const;
+
+   private:
+    std::shared_ptr<IoPatternRuntime> runtime_;
+    std::shared_ptr<CfmBinaryCodec> codec_;
+    CfmIngress ingress_;
+};
+
+// coro_rpc-facing adapter. Keeping the RPC signature here lets both the
+// SubMaster server and integration tests register the exact production
+// endpoint.
+class CfmRpcService final {
+   public:
+    explicit CfmRpcService(std::shared_ptr<CfmService> service)
+        : service_(std::move(service)) {}
+
+    bool Send(const std::string& method, const std::string& payload,
+              const std::string& source_id = {});
+
+   private:
+    std::shared_ptr<CfmService> service_;
+};
+
+}  // namespace mooncake::io_pattern

mooncake-store/include/io_pattern/client.h

@@ -0,0 +1,22 @@
+#pragma once
+
+#include "../types.h"
+#include "reporter.h"
+#include "types.h"
+
+namespace mooncake::io_pattern {
+
+// Adapter seam between an inference node and the Cache Flow Manager of the
+// SubMaster(s) that own the reported keys. Reports are sent over the same
+// coro_rpc endpoint every other Mooncake API uses; there is no separate CFM
+// Master and no credential to obtain.
+class CfmClient {
+   public:
+    virtual ~CfmClient() = default;
+
+    virtual ErrorCode ReportSnapshot(const IoPatternSnapshot& snapshot) = 0;
+    virtual ErrorCode ReportMetricBatch(const MetricBatch& batch) = 0;
+    virtual ErrorCode ExecutePrefetch(const PrefetchPlan& plan) = 0;
+};
+
+}  // namespace mooncake::io_pattern

mooncake-store/include/io_pattern/collector.h

@@ -0,0 +1,20 @@
+#pragma once
+
+#include "io_pattern/types.h"
+
+namespace mooncake::io_pattern {
+
+// Collects non-blocking, already-aggregated observations from data paths.
+class IoPatternCollector {
+   public:
+    virtual ~IoPatternCollector() = default;
+
+    // Implementations must not block the caller on RPC or storage I/O.
+    virtual void ReportInferenceMetrics(const InferenceMetrics& metrics) = 0;
+    virtual void RecordAccess(const std::string& key,
+                              const AccessRecord& record) = 0;
+    virtual void RecordStorageMetric(const StorageMetric& metric) = 0;
+    virtual IoPatternSnapshot GetSnapshot() const = 0;
+};
+
+}  // namespace mooncake::io_pattern