[optimizer] report fixed-target MRC metrics

Co-authored-by: putaopi <72243274+putaopi7@users.noreply.github.com>

Author: Tyndalllll

kv_cache_manager/optimizer/manager/BUILD

@@ -62,6 +62,7 @@ cc_library(
         "//kv_cache_manager/optimizer/liteHit:hit_curve",
         "//kv_cache_manager/optimizer/liteHit:lite_hit",
         "//kv_cache_manager/optimizer/liteHit:request_preprocess",
+        "//kv_cache_manager/optimizer/metrics:mrc_window",
     ],
 )
 

kv_cache_manager/optimizer/manager/online_runtime/online_optimizer_manager.cc

@@ -62,41 +62,6 @@ int64_t SaturatingMultiplyToInt64(uint64_t lhs, uint64_t rhs) {
     return static_cast<int64_t>(lhs * rhs);
 }
 
-void AccumulateMrc(const RequestFact &fact, InstanceState &state) {
-    for (const auto &segment : fact.hit_curve) {
-        if (segment.run_length == 0) {
-            continue;
-        }
-        const uint64_t end_required_blocks = segment.start_required_blocks + segment.run_length - 1;
-        if (end_required_blocks + 1 >= state.mrc_interval_hit_count_deltas.size()) {
-            state.mrc_interval_hit_count_deltas.resize(static_cast<std::size_t>(end_required_blocks + 2), 0);
-        }
-        ++state.mrc_interval_hit_count_deltas[segment.start_required_blocks];
-        --state.mrc_interval_hit_count_deltas[end_required_blocks + 1];
-        state.mrc_interval_total_hits += segment.run_length;
-    }
-}
-
-uint64_t ComputeMrcRequiredBlocks(const InstanceState &state) {
-    if (state.mrc_interval_total_hits == 0) {
-        return 0;
-    }
-
-    // ceil(total * 0.95), written without multiplication to avoid overflow.
-    const uint64_t target_hits = state.mrc_interval_total_hits - state.mrc_interval_total_hits / 20;
-    uint64_t accumulated_hits = 0;
-    int64_t hits_at_required_blocks = 0;
-    for (std::size_t required_blocks = 0; required_blocks < state.mrc_interval_hit_count_deltas.size();
-         ++required_blocks) {
-        hits_at_required_blocks += state.mrc_interval_hit_count_deltas[required_blocks];
-        accumulated_hits += static_cast<uint64_t>(hits_at_required_blocks);
-        if (accumulated_hits >= target_hits) {
-            return required_blocks;
-        }
-    }
-    return 0;
-}
-
 } // namespace
 
 int64_t OnlineOptimizerManager::ComputeSizeForGroup(const std::vector<LocationSpecInfo> &specs,
@@ -529,7 +494,6 @@ ErrorCode OnlineOptimizerManager::TraceQuery(const std::string &instance_id,
         }
 
         const RequestFact fact = state->lite_hit->ProcessRequest(normalized.block_keys, replay_timestamp_ns);
-        AccumulateMrc(fact, *state);
         result.input_token_len = ClampToInt64(normalized.input_token_len);
 
         const uint64_t block_size = static_cast<uint64_t>(state->instance_info->block_size());
@@ -554,6 +518,7 @@ ErrorCode OnlineOptimizerManager::TraceQuery(const std::string &instance_id,
 
         if (state->instance_group->enable_theoretical_max_cache()) {
             const uint64_t max_hits = HitCurveProjector::ProjectInfinite(fact);
+            state->mrc_window.Record(fact);
             result.max_hit_count = ClampToInt64(max_hits);
             result.max_hit_rate =
                 normalized.input_token_len == 0 ? 0.0 : static_cast<double>(max_hits * block_size) / token_denominator;
@@ -737,21 +702,21 @@ ErrorCode OnlineOptimizerManager::ListInstances(const std::string &instance_grou
 ErrorCode OnlineOptimizerManager::TakeMrcMetrics(std::vector<MrcMetricInfo> &metrics) {
     std::shared_lock lock(instances_mutex_);
     metrics.clear();
-    metrics.reserve(instances_.size());
 
     for (const auto &[id, state] : instances_) {
         std::lock_guard<std::mutex> guard(state->mutex);
-        if (state->linear_step != 0 || !state->lite_hit) {
+        if (state->linear_step != 0 || !state->lite_hit || !state->instance_group->enable_theoretical_max_cache()) {
             continue;
         }
 
-        MrcMetricInfo metric;
-        metric.instance_id = id;
-        metric.capacity_bytes =
-            SaturatingMultiplyToInt64(ComputeMrcRequiredBlocks(*state), static_cast<uint64_t>(state->size_full));
-        metrics.push_back(std::move(metric));
-        state->mrc_interval_hit_count_deltas.clear();
-        state->mrc_interval_total_hits = 0;
+        for (const auto &point : state->mrc_window.Take()) {
+            MrcMetricInfo metric;
+            metric.instance_id = id;
+            metric.target_basis_points = point.target_basis_points;
+            metric.capacity_bytes =
+                SaturatingMultiplyToInt64(point.required_blocks, static_cast<uint64_t>(state->size_full));
+            metrics.push_back(std::move(metric));
+        }
     }
     return EC_OK;
 }
@@ -795,8 +760,7 @@ ErrorCode OnlineOptimizerManager::ResetStats(const std::string &instance_id) {
     state->total_input_tokens = 0;
     std::fill(state->total_hits_per_capacity.begin(), state->total_hits_per_capacity.end(), 0);
     state->total_max_hits = 0;
-    state->mrc_interval_hit_count_deltas.clear();
-    state->mrc_interval_total_hits = 0;
+    state->mrc_window.Reset();
     KVCM_LOG_INFO("ResetStats OK: instance[%s]", instance_id.c_str());
     return EC_OK;
 }

kv_cache_manager/optimizer/manager/online_runtime/online_optimizer_manager.h

@@ -14,6 +14,7 @@
 #include "kv_cache_manager/optimizer/config/optimizer_instance_info.h"
 #include "kv_cache_manager/optimizer/index/online/cache_indexer.h"
 #include "kv_cache_manager/optimizer/liteHit/lite_hit.h"
+#include "kv_cache_manager/optimizer/metrics/mrc_window.h"
 
 namespace kv_cache_manager {
 
@@ -45,11 +46,7 @@ struct InstanceState {
     std::vector<int64_t> total_hits_per_capacity;
     int64_t total_max_hits = 0;
 
-    // Range-add difference array for the current reporting interval's
-    // distribution of minimum LRU capacities required by theoretically
-    // hittable blocks.
-    std::vector<int64_t> mrc_interval_hit_count_deltas;
-    uint64_t mrc_interval_total_hits = 0;
+    MrcWindow mrc_window;
 };
 
 struct TraceQueryResult {
@@ -84,6 +81,7 @@ struct HitAgeBucketRatio {
 
 struct MrcMetricInfo {
     std::string instance_id;
+    uint32_t target_basis_points = 0;
     int64_t capacity_bytes = 0;
 };
 

kv_cache_manager/optimizer/metrics/BUILD

@@ -1,5 +1,14 @@
 package(default_visibility = ["//visibility:public"])
 
+cc_library(
+    name = "mrc_window",
+    srcs = ["mrc_window.cc"],
+    hdrs = ["mrc_window.h"],
+    deps = [
+        "//kv_cache_manager/optimizer/liteHit:hit_curve",
+    ],
+)
+
 cc_library(
     name = "optimizer_metrics_collector",
     srcs = ["optimizer_metrics_collector.cc"],

kv_cache_manager/optimizer/metrics/mrc_window.cc

@@ -0,0 +1,92 @@
+#include "kv_cache_manager/optimizer/metrics/mrc_window.h"
+
+#include <array>
+#include <iterator>
+#include <limits>
+
+#include "kv_cache_manager/optimizer/liteHit/hit_curve.h"
+
+namespace kv_cache_manager {
+
+namespace {
+
+constexpr uint32_t kBasisPointScale = 10000;
+constexpr std::array<uint32_t, 6> kTargetBasisPoints = {6000, 8000, 9000, 9500, 9900, 9950};
+
+uint64_t SaturatingAdd(uint64_t lhs, uint64_t rhs) {
+    return rhs > std::numeric_limits<uint64_t>::max() - lhs ? std::numeric_limits<uint64_t>::max() : lhs + rhs;
+}
+
+uint64_t SaturatingMultiply(uint64_t lhs, uint64_t rhs) {
+    if (lhs != 0 && rhs > std::numeric_limits<uint64_t>::max() / lhs) {
+        return std::numeric_limits<uint64_t>::max();
+    }
+    return lhs * rhs;
+}
+
+} // namespace
+
+void MrcWindow::Record(const RequestFact &fact) {
+    uint64_t request_hits = 0;
+    for (const auto &segment : fact.hit_curve) {
+        if (segment.run_length == 0) {
+            continue;
+        }
+
+        const uint64_t end_required_blocks = segment.start_required_blocks + segment.run_length - 1;
+        ++hit_count_deltas_[segment.start_required_blocks];
+        --hit_count_deltas_[end_required_blocks + 1];
+        request_hits = SaturatingAdd(request_hits, segment.run_length);
+    }
+
+    total_hits_ = SaturatingAdd(total_hits_, request_hits);
+}
+
+std::vector<MrcWindowPoint> MrcWindow::Take() {
+    std::vector<MrcWindowPoint> curve;
+    curve.reserve(kTargetBasisPoints.size());
+    for (uint32_t target_basis_points : kTargetBasisPoints) {
+        curve.push_back({target_basis_points, ComputeRequiredBlocks(target_basis_points)});
+    }
+    Reset();
+    return curve;
+}
+
+void MrcWindow::Reset() {
+    hit_count_deltas_.clear();
+    total_hits_ = 0;
+}
+
+uint64_t MrcWindow::ComputeRequiredBlocks(uint32_t target_basis_points) const {
+    if (total_hits_ == 0) {
+        return 0;
+    }
+
+    // ceil(total * target / 10000), split into quotient/remainder to avoid overflow.
+    const uint64_t quotient = total_hits_ / kBasisPointScale;
+    const uint64_t remainder = total_hits_ % kBasisPointScale;
+    const uint64_t target_hits =
+        quotient * target_basis_points + (remainder * target_basis_points + kBasisPointScale - 1) / kBasisPointScale;
+    uint64_t accumulated_hits = 0;
+    int64_t hits_at_required_blocks = 0;
+
+    for (auto it = hit_count_deltas_.begin(); it != hit_count_deltas_.end(); ++it) {
+        hits_at_required_blocks += it->second;
+        const auto next = std::next(it);
+        if (hits_at_required_blocks <= 0 || next == hit_count_deltas_.end()) {
+            continue;
+        }
+
+        const uint64_t span = next->first - it->first;
+        const uint64_t span_hits = SaturatingMultiply(static_cast<uint64_t>(hits_at_required_blocks), span);
+        if (span_hits >= target_hits - accumulated_hits) {
+            const uint64_t remaining_hits = target_hits - accumulated_hits;
+            const uint64_t offset = (remaining_hits - 1) / static_cast<uint64_t>(hits_at_required_blocks);
+            return it->first + offset;
+        }
+        accumulated_hits = SaturatingAdd(accumulated_hits, span_hits);
+    }
+    return 0;
+}
+
+} // namespace kv_cache_manager

kv_cache_manager/optimizer/metrics/mrc_window.h

@@ -0,0 +1,32 @@
+#pragma once
+
+#include <cstdint>
+#include <map>
+#include <vector>
+
+namespace kv_cache_manager {
+
+struct RequestFact;
+
+struct MrcWindowPoint {
+    uint32_t target_basis_points = 0;
+    uint64_t required_blocks = 0;
+};
+
+// Accumulates one reporting window of full-attention theoretical hits.
+// Synchronization is provided by the owning InstanceState mutex.
+class MrcWindow {
+public:
+    void Record(const RequestFact &fact);
+    std::vector<MrcWindowPoint> Take();
+    void Reset();
+
+private:
+    uint64_t ComputeRequiredBlocks(uint32_t target_basis_points) const;
+
+    // Sparse difference points of required capacity -> theoretical hit count.
+    std::map<uint64_t, int64_t> hit_count_deltas_;
+    uint64_t total_hits_ = 0;
+};
+
+} // namespace kv_cache_manager

kv_cache_manager/optimizer/metrics/optimizer_metrics_reporter.cc

@@ -15,6 +15,22 @@
 
 namespace kv_cache_manager {
 
+namespace {
+
+std::string FormatTargetHitRatePercent(uint32_t target_basis_points) {
+    const uint32_t whole = target_basis_points / 100;
+    const uint32_t fraction = target_basis_points % 100;
+    if (fraction == 0) {
+        return std::to_string(whole);
+    }
+    if (fraction % 10 == 0) {
+        return std::to_string(whole) + "." + std::to_string(fraction / 10);
+    }
+    return std::to_string(whole) + "." + (fraction < 10 ? "0" : "") + std::to_string(fraction);
+}
+
+} // namespace
+
 #define DECLARE_METRICS(group, name) std::unique_ptr<kmonitor::MutableMetric> group##_##name##_metrics;
 
 struct OptimizerMetricsReporter::KmonContext {
@@ -221,7 +237,6 @@ bool OptimizerMetricsReporter::InitMetrics() {
     REGISTER_GAUGE_METRIC(query, capacity_efficiency);
 
     REGISTER_GAUGE_METRIC(trace, query_hit_age_bucket_ratio);
-
     kmon_ctx_->mrc_metrics.reset(reporter->RegisterMetric("mrc", kmonitor::GAUGE, kmonitor::FATAL));
     if (!kmon_ctx_->mrc_metrics) {
         KVCM_LOG_ERROR("failed to register metric:[mrc]");
@@ -332,7 +347,8 @@ void OptimizerMetricsReporter::ReportInterval() {
     }
 
     for (const auto &metric : mrc_metrics) {
-        MetricsTags prom_tags = {{"instance_id", metric.instance_id}};
+        MetricsTags prom_tags = {{"instance_id", metric.instance_id},
+                                 {"target_hit_rate_percent", FormatTargetHitRatePercent(metric.target_basis_points)}};
         Gauge mrc = metrics_registry_->GetGauge("mrc", prom_tags);
         mrc = static_cast<double>(metric.capacity_bytes);
     }
@@ -379,7 +395,8 @@ void OptimizerMetricsReporter::ReportInterval() {
     }
 
     for (const auto &metric : mrc_metrics) {
-        MetricsTags base_tags = {{"instance_id", metric.instance_id}};
+        MetricsTags base_tags = {{"instance_id", metric.instance_id},
+                                 {"target_hit_rate_percent", FormatTargetHitRatePercent(metric.target_basis_points)}};
         kmonitor::MetricsTags tags = kmon_ctx_->GetKmonitorTags(base_tags);
         kmon_ctx_->mrc_metrics->Report(&tags, static_cast<double>(metric.capacity_bytes));
     }

kv_cache_manager/optimizer/test/BUILD

@@ -273,6 +273,16 @@ cc_test(
     ],
 )
 
+cc_test(
+    name = "MrcWindowTest",
+    srcs = ["mrc_window_test.cc"],
+    deps = [
+        "//kv_cache_manager/common:unittest",
+        "//kv_cache_manager/optimizer/liteHit:hit_curve",
+        "//kv_cache_manager/optimizer/metrics:mrc_window",
+    ],
+)
+
 cc_test(
     name = "OnlineOptimizerServerConfigTest",
     srcs = [