[manager/meta] support V6D and subscriber reporting for multi-engine instances

Author: lucky-zzz

kv_cache_manager/manager/cache_manager.cc

@@ -3,6 +3,7 @@
 #include <algorithm>
 #include <array>
 #include <cassert>
+#include <charconv>
 #include <chrono>
 #include <cinttypes>
 #include <limits>
@@ -89,6 +90,39 @@ namespace kv_cache_manager {
     } while (0)
 
 namespace {
+struct ReporterIdentityView {
+    std::string_view base_host;
+    std::optional<uint64_t> engine_rank;
+};
+
+bool ParseReporterIdentity(std::string_view host_ip_port, ReporterIdentityView &out) {
+    out = {};
+    if (host_ip_port.empty()) {
+        return false;
+    }
+    const size_t separator = host_ip_port.find('@');
+    if (separator == std::string_view::npos) {
+        out.base_host = host_ip_port;
+        return true;
+    }
+    if (separator == 0 || separator + 1 >= host_ip_port.size() ||
+        host_ip_port.find('@', separator + 1) != std::string_view::npos) {
+        return false;
+    }
+    const std::string_view rank_text = host_ip_port.substr(separator + 1);
+    if (rank_text.size() > 1 && rank_text.front() == '0') {
+        return false;
+    }
+    uint64_t rank = 0;
+    const auto [end, ec] = std::from_chars(rank_text.data(), rank_text.data() + rank_text.size(), rank);
+    if (ec != std::errc{} || end != rank_text.data() + rank_text.size()) {
+        return false;
+    }
+    out.base_host = host_ip_port.substr(0, separator);
+    out.engine_rank = rank;
+    return true;
+}
+
 CacheManager::KeyVector GenKeyVector(const CacheManager::TokenIdsVector &tokens, int64_t block_size) {
     std::vector<int64_t> block_keys;
     size_t total_blocks = tokens.size() / block_size;
@@ -2668,7 +2702,9 @@ ErrorCode CacheManager::ReportEvent(RequestContext *request_context,
     const std::string &host_ip_port = request->host_ip_port();
     auto *response_status = response->mutable_header()->mutable_status();
 
-    if (instance_id.empty() || !SnapshotUriUtils::IsValidLocationIdComponent(host_ip_port)) {
+    ReporterIdentityView reporter_identity;
+    if (instance_id.empty() || !SnapshotUriUtils::IsValidLocationIdComponent(host_ip_port) ||
+        !ParseReporterIdentity(host_ip_port, reporter_identity)) {
         KVCM_LOG_WARN("trace_id [%s] | ReportEvent: invalid instance_id or host_ip_port", trace_id.c_str());
         response_status->set_code(proto::meta::INVALID_ARGUMENT);
         response_status->set_message("invalid instance_id or host_ip_port");
@@ -4596,6 +4632,7 @@ CacheManager::GetHostCacheStateCheckLocDataExistFunc(const std::string &instance
     struct EventVisibilitySnapshot {
         std::shared_ptr<EventReportBackend> backend;
         EventReportBackend::QueryVisibilitySnapshot reporters;
+        std::map<std::string, std::vector<std::string>, std::less<>> logical_hosts_by_reporter;
     };
     struct EventVisibilitySnapshots {
         std::once_flag initialize_once;
@@ -4627,6 +4664,42 @@ CacheManager::GetHostCacheStateCheckLocDataExistFunc(const std::string &instance
             snapshot.backend->GetQueryVisibilitySnapshot(instance_id, snapshot.reporters);
             event_snapshots->by_storage_type.emplace(storage_type, std::move(snapshot));
         }
+
+        std::map<std::string, std::vector<std::string>, std::less<>> ranked_hosts_by_base;
+        const auto l1p5_it = event_snapshots->by_storage_type.find(
+            DataStorageType::DATA_STORAGE_TYPE_EVENT_REPORT_L1P5);
+        if (l1p5_it != event_snapshots->by_storage_type.end()) {
+            for (const auto &[reporter, state] : l1p5_it->second.reporters) {
+                (void)state;
+                ReporterIdentityView identity;
+                if (ParseReporterIdentity(reporter, identity) && identity.engine_rank.has_value()) {
+                    ranked_hosts_by_base[std::string(identity.base_host)].push_back(reporter);
+                }
+            }
+        }
+        for (auto &[base, ranked_hosts] : ranked_hosts_by_base) {
+            (void)base;
+            std::sort(ranked_hosts.begin(), ranked_hosts.end());
+            ranked_hosts.erase(std::unique(ranked_hosts.begin(), ranked_hosts.end()), ranked_hosts.end());
+        }
+        for (auto &[storage_type, snapshot] : event_snapshots->by_storage_type) {
+            for (const auto &[reporter, state] : snapshot.reporters) {
+                (void)state;
+                auto &logical_hosts = snapshot.logical_hosts_by_reporter[reporter];
+                ReporterIdentityView identity;
+                const bool parsed = ParseReporterIdentity(reporter, identity);
+                if (parsed && storage_type == DataStorageType::DATA_STORAGE_TYPE_EVENT_REPORT_L2 &&
+                    !identity.engine_rank.has_value()) {
+                    const auto ranked_it = ranked_hosts_by_base.find(identity.base_host);
+                    if (ranked_it != ranked_hosts_by_base.end()) {
+                        logical_hosts = ranked_it->second;
+                    }
+                }
+                if (logical_hosts.empty()) {
+                    logical_hosts.push_back(reporter);
+                }
+            }
+        }
     };
 
     return [fallback = std::move(fallback),
@@ -4661,6 +4734,10 @@ CacheManager::GetHostCacheStateCheckLocDataExistFunc(const std::string &instance
         out_info.has_reporter_identity = true;
         out_info.reporter_medium = reporter_medium;
         out_info.reporter_host = reporter_host;
+        const auto logical_hosts_it = snapshot_it->second.logical_hosts_by_reporter.find(reporter_host);
+        if (logical_hosts_it != snapshot_it->second.logical_hosts_by_reporter.end()) {
+            out_info.logical_hosts = &logical_hosts_it->second;
+        }
         return true;
     };
 }

kv_cache_manager/manager/meta_searcher.cc

@@ -481,14 +481,26 @@ void VisitHostSpecsForOneKey(const LocationRange &locations,
             // applying the reporter generation fence. Reuse that result.
             const bool specs_already_validated =
                 request_check_location != nullptr && is_event_report && location_info.has_reporter_identity;
+            const auto visit_logical_hosts = [&](std::string_view spec_name) {
+                if (location_info.logical_hosts && !location_info.logical_hosts->empty()) {
+                    for (const auto &logical_host : *location_info.logical_hosts) {
+                        visitor(std::string_view(logical_host),
+                                spec_name,
+                                is_vineyard,
+                                location_info.reporter_host);
+                    }
+                    return;
+                }
+                visitor(location_info.reporter_host, spec_name, is_vineyard, location_info.reporter_host);
+            };
             if (!visit_spec_names) {
                 if (specs_already_validated) {
-                    visitor(location_info.reporter_host, std::string_view{}, is_vineyard);
+                    visit_logical_hosts(std::string_view{});
                     continue;
                 }
                 for (const auto &spec : loc->location_specs()) {
                     if (StandardUri(spec.uri()).Valid()) {
-                        visitor(location_info.reporter_host, std::string_view{}, is_vineyard);
+                        visit_logical_hosts(std::string_view{});
                         break;
                     }
                 }
@@ -498,7 +510,7 @@ void VisitHostSpecsForOneKey(const LocationRange &locations,
                 if (!specs_already_validated && !StandardUri(spec.uri()).Valid()) {
                     continue;
                 }
-                visitor(location_info.reporter_host, std::string_view(spec.name()), is_vineyard);
+                visit_logical_hosts(std::string_view(spec.name()));
             }
             continue;
         }
@@ -510,7 +522,7 @@ void VisitHostSpecsForOneKey(const LocationRange &locations,
             }
             const std::string host = uri.GetHostPort();
             if (!host.empty()) {
-                visitor(std::string_view(host), std::string_view(spec.name()), false);
+                visitor(std::string_view(host), std::string_view(spec.name()), false, std::string_view(host));
             }
         }
     }
@@ -527,7 +539,9 @@ void BuildHostsForOneKey(const LocationRange &locations,
                             request_check_location,
                             medium_set,
                             false,
-                            [&hosts](std::string_view host, std::string_view, bool) { hosts.emplace_back(host); });
+                            [&hosts](std::string_view host, std::string_view, bool, std::string_view) {
+                                hosts.emplace_back(host);
+                            });
     std::sort(hosts.begin(), hosts.end());
     hosts.erase(std::unique(hosts.begin(), hosts.end()), hosts.end());
 }
@@ -544,7 +558,8 @@ void BuildCandidatePresenceForOneKey(const LocationRange &locations,
                             request_check_location,
                             medium_set,
                             false,
-                            [&candidate_hosts, presence_words](std::string_view host, std::string_view, bool) {
+                            [&candidate_hosts, presence_words](
+                                std::string_view host, std::string_view, bool, std::string_view) {
                                 const auto it =
                                     std::lower_bound(candidate_hosts.begin(),
                                                      candidate_hosts.end(),
@@ -601,11 +616,17 @@ void BuildHostSpecNamesForOneKey(const LocationRange &locations,
         request_check_location,
         medium_set,
         true,
-        [&host_specs, &vineyard_host_specs](std::string_view host, std::string_view spec_name, bool is_vineyard) {
+        [&host_specs, &vineyard_host_specs](std::string_view host,
+                                           std::string_view spec_name,
+                                           bool is_vineyard,
+                                           std::string_view physical_reporter) {
             auto &local_names = host_specs[std::string(host)];
             local_names.emplace(spec_name);
             if (is_vineyard) {
-                vineyard_host_specs[std::string(host)].emplace(spec_name);
+                // Keep one P2P candidate per physical EventReport reporter.
+                // A shared L2 reporter may project onto several logical ranks,
+                // but must not become several Vineyard peers.
+                vineyard_host_specs[std::string(physical_reporter)].emplace(spec_name);
             }
         });
 }
@@ -1114,7 +1135,7 @@ ErrorCode PrefixMatchWithMambaByHostWithoutP2P(MetaIndexer *meta_indexer,
                 medium_set,
                 true,
                 [&candidate_hosts, &required_spec_names, &seen_words, &host_present, spec_word_count](
-                    std::string_view host, std::string_view spec_name, bool) {
+                    std::string_view host, std::string_view spec_name, bool, std::string_view) {
                     const auto host_it = std::lower_bound(candidate_hosts.begin(),
                                                           candidate_hosts.end(),
                                                           host,

kv_cache_manager/manager/meta_searcher.h

@@ -65,6 +65,11 @@ class MetaSearcher {
         // never retained in output.
         std::string_view reporter_medium;
         std::string_view reporter_host;
+        // Optional request-owned projection of one physical reporter onto one
+        // or more schedulable engine identities. Shared L2 reporters use this
+        // to contribute their specs to every active ranked L1P5 engine without
+        // duplicating the stored CacheLocation or its physical Vineyard URI.
+        const std::vector<std::string> *logical_hosts = nullptr;
     };
     using CheckHostCacheLocationFunc =
         std::function<bool(const CacheLocation &location, HostCacheLocationInfo &out_info)>;