io pattern实现

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,48 @@
+#pragma once
+
+#include <optional>
+#include <utility>
+
+#include "types.h"
+#include "../types.h"
+
+namespace mooncake::io_pattern {
+
+struct CfmPollResult {
+    enum class Status { kCommand, kEmpty, kError };
+
+    static CfmPollResult Command(PolicyCommand command,
+                                 uint64_t delivery_id = 0) {
+        return {.status = Status::kCommand,
+                .command = std::move(command),
+                .delivery_id = delivery_id};
+    }
+    static CfmPollResult Empty() { return {.status = Status::kEmpty}; }
+    static CfmPollResult Error() { return {.status = Status::kError}; }
+
+    Status status{Status::kEmpty};
+    std::optional<PolicyCommand> command;
+    uint64_t delivery_id{0};
+};
+
+// Transport-neutral CFM RPC channel. Implementations own serialization,
+// retries and connection lifecycle.
+class CfmChannel {
+   public:
+    virtual ~CfmChannel() = default;
+    virtual bool SendSnapshot(const IoPatternSnapshot& snapshot) = 0;
+    virtual CfmPollResult PollPolicyResult() = 0;
+    std::optional<PolicyCommand> PollPolicy() {
+        auto result = PollPolicyResult();
+        if (result.status != CfmPollResult::Status::kCommand ||
+            !result.command) {
+            return std::nullopt;
+        }
+        if (!AcknowledgePolicy(result.delivery_id, true)) return std::nullopt;
+        return std::move(result.command);
+    }
+    virtual bool AcknowledgePolicy(uint64_t, bool) { return true; }
+    virtual ErrorCode ExecutePrefetch(const PrefetchPlan& plan) = 0;
+};
+
+}  // namespace mooncake::io_pattern

mooncake-store/include/io_pattern/cfm_client_impl.h

@@ -0,0 +1,34 @@
+#pragma once
+
+#include <memory>
+#include <functional>
+
+#include "cfm_channel.h"
+#include "client.h"
+
+namespace mooncake::io_pattern {
+
+// Production CFM client orchestration. Network behavior is delegated to the
+// injected channel so this class remains independent of RPC libraries.
+class CfmClientImpl final : public CfmClient {
+   public:
+    using PolicyCommandHandler = std::function<ErrorCode(const PolicyCommand&)>;
+
+    explicit CfmClientImpl(std::shared_ptr<CfmChannel> channel,
+                           PolicyCommandHandler policy_handler = {})
+        : channel_(std::move(channel)),
+          policy_handler_(std::move(policy_handler)) {}
+
+    ErrorCode ReportSnapshot(const IoPatternSnapshot& snapshot) override;
+    ErrorCode ReceivePolicy(const PolicyCommand& command) override;
+    ErrorCode ExecutePrefetch(const PrefetchPlan& plan) override;
+
+    std::optional<PolicyCommand> PollPolicy();
+    ErrorCode PollAndDispatchPolicy();
+
+   private:
+    std::shared_ptr<CfmChannel> channel_;
+    PolicyCommandHandler policy_handler_;
+};
+
+}  // 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_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 authenticated 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,91 @@
+#pragma once
+
+#include <cstddef>
+#include <cstdint>
+#include <condition_variable>
+#include <deque>
+#include <memory>
+#include <mutex>
+#include <optional>
+#include <string>
+#include <string_view>
+#include <thread>
+#include <unordered_map>
+#include <utility>
+
+#include "cfm_ingress.h"
+
+namespace mooncake::io_pattern {
+
+// Authenticated server-side CFM endpoint. The RPC layer delegates to this
+// class, keeping authentication, bounded policy queues and runtime dispatch
+// independent of the concrete network transport.
+class CfmService final {
+   public:
+    CfmService(std::shared_ptr<IoPatternRuntime> runtime,
+               std::string auth_token, size_t policy_queue_capacity = 4096,
+               std::string producer_auth_token = {});
+    ~CfmService();
+
+    bool Authenticate(std::string_view token) const;
+    bool AuthenticateNode(std::string_view token) const;
+    bool AuthenticateProducer(std::string_view token) const;
+    bool Send(std::string_view node_id, std::string_view method,
+              std::string_view payload, std::string_view token);
+    std::optional<std::pair<uint64_t, std::string>> PollPolicy(
+        std::string_view node_id, std::string_view token);
+    bool AcknowledgePolicy(std::string_view node_id, uint64_t delivery_id,
+                           bool success, std::string_view token);
+    bool EnqueuePolicy(std::string node_id, std::string payload,
+                       std::string_view token);
+
+   private:
+    bool EnqueueValidated(std::string node_id, std::string payload);
+    void SchedulePolicyProduction(std::string node_id, MetricBatch batch);
+    void PolicyProducerWorker();
+    void ProducePolicies(std::string_view node_id, const MetricBatch& batch);
+
+    std::shared_ptr<IoPatternRuntime> runtime_;
+    std::shared_ptr<CfmBinaryCodec> codec_;
+    CfmIngress ingress_;
+    const std::string auth_token_;
+    const std::string producer_auth_token_;
+    const size_t policy_queue_capacity_;
+    std::mutex mutex_;
+    std::unordered_map<std::string,
+                       std::deque<std::pair<uint64_t, std::string>>>
+        policy_queues_;
+    uint64_t next_delivery_id_{1};
+    size_t total_queued_policies_{0};
+    std::mutex producer_mutex_;
+    std::condition_variable producer_cv_;
+    std::deque<std::pair<std::string, MetricBatch>> pending_metric_batches_;
+    bool producer_stopping_{false};
+    std::thread producer_worker_;
+};
+
+// coro_rpc-facing adapter. Keeping RPC signatures here lets both the Master
+// server and integration tests register the exact production endpoints.
+class CfmRpcService final {
+   public:
+    explicit CfmRpcService(std::shared_ptr<CfmService> service)
+        : service_(std::move(service)) {}
+
+    bool Authenticate(const std::string& auth_token);
+    bool Send(const std::string& node_id, const std::string& method,
+              const std::string& payload, const std::string& auth_token);
+    // The boolean explicitly distinguishes a rejected request from an
+    // authenticated queue that currently has no policy.
+    std::pair<bool, std::optional<std::pair<uint64_t, std::string>>> Receive(
+        const std::string& method, const std::string& node_id,
+        const std::string& auth_token);
+    bool Acknowledge(const std::string& node_id, uint64_t delivery_id,
+                     bool success, const std::string& auth_token);
+    bool EnqueuePolicy(const std::string& node_id, const std::string& payload,
+                       const std::string& auth_token);
+
+   private:
+    std::shared_ptr<CfmService> service_;
+};
+
+}  // namespace mooncake::io_pattern

mooncake-store/include/io_pattern/client.h

@@ -0,0 +1,18 @@
+#pragma once
+
+#include "../types.h"
+#include "io_pattern/types.h"
+
+namespace mooncake::io_pattern {
+
+// Adapter seam between an inference node and the remote Cache Flow Manager.
+class CfmClient {
+   public:
+    virtual ~CfmClient() = default;
+
+    virtual ErrorCode ReportSnapshot(const IoPatternSnapshot& snapshot) = 0;
+    virtual ErrorCode ReceivePolicy(const PolicyCommand& command) = 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

mooncake-store/include/io_pattern/collector_impl.h

@@ -0,0 +1,80 @@
+#pragma once
+
+#include <deque>
+#include <functional>
+#include <memory>
+#include <mutex>
+#include <unordered_map>
+#include <utility>
+
+#include "collector.h"
+#include "reporter.h"
+
+namespace mooncake::io_pattern {
+
+// Production collector that aggregates observations by tenant and object.
+// It owns only metrics state; reporting to CFM is intentionally external.
+class IoPatternCollectorImpl final : public IoPatternCollector {
+   public:
+    struct Config {
+        size_t max_keys_per_tenant{0};
+        size_t max_total_keys{0};
+        uint64_t access_window_ns{60'000'000'000ULL};
+        uint64_t access_bucket_ns{1'000'000'000ULL};
+        size_t max_access_buckets_per_key{64};
+        std::function<uint64_t()> now_ns;
+    };
+
+    explicit IoPatternCollectorImpl(Config config = {},
+                                    std::shared_ptr<IoPatternReporter> reporter =
+                                        nullptr)
+        : config_(config), reporter_(std::move(reporter)) {}
+    void ReportInferenceMetrics(const InferenceMetrics& metrics) override;
+    void RecordAccess(const std::string& key,
+                      const AccessRecord& record) override;
+    void RecordStorageMetric(const StorageMetric& metric) override;
+    // Ingests a CFM snapshot without replaying it through the asynchronous
+    // reporter. The sender is already the reporting side of that pipeline.
+    void MergeSnapshot(const IoPatternSnapshot& snapshot);
+    IoPatternSnapshot GetSnapshot() const override;
+    uint64_t dropped() const;
+    bool degraded() const;
+    bool FlushReports();
+
+   private:
+    struct StorageMetricKey {
+        std::string source_id;
+        CacheTier tier{CacheTier::kL2Segment};
+        bool operator==(const StorageMetricKey&) const = default;
+    };
+    struct StorageMetricKeyHash {
+        size_t operator()(const StorageMetricKey& key) const noexcept {
+            return std::hash<std::string>{}(key.source_id) ^
+                   (static_cast<size_t>(key.tier) << 1);
+        }
+    };
+    struct AccessWindowBucket {
+        uint64_t observed_at_ns{0};
+        uint64_t access_count{0};
+        uint64_t write_count{0};
+        uint64_t overwrite_count{0};
+        uint32_t max_write_batch_size{0};
+    };
+
+    void ApplyAccessWindow(const ObjectRef& object, uint64_t now_ns,
+                           KeyMetrics& metrics) const;
+
+    mutable std::mutex mutex_;
+    Config config_;
+    std::shared_ptr<IoPatternReporter> reporter_;
+    uint64_t dropped_{0};
+    bool degraded_{false};
+    std::unordered_map<ObjectRef, KeyMetrics, ObjectRefHash> key_metrics_;
+    std::unordered_map<ObjectRef, std::deque<AccessWindowBucket>, ObjectRefHash>
+        access_windows_;
+    std::unordered_map<TenantId, size_t, TenantIdHash> tenant_key_counts_;
+    std::unordered_map<StorageMetricKey, StorageMetric, StorageMetricKeyHash>
+        storage_metrics_;
+};
+
+}  // namespace mooncake::io_pattern