Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
lucky-zzz wants to merge 1 commit into
base: main
Choose a base branch
from
Open

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
79 changes: 78 additions & 1 deletion kv_cache_manager/manager/cache_manager.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
#include <algorithm>
#include <array>
#include <cassert>
#include <charconv>
#include <chrono>
#include <cinttypes>
#include <limits>
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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");
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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),
Expand Down Expand Up @@ -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;
};
}
Expand Down
39 changes: 30 additions & 9 deletions kv_cache_manager/manager/meta_searcher.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
}
}
Expand All @@ -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;
}
Expand All @@ -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));
}
}
}
Expand All @@ -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());
}
Expand All @@ -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(),
Expand Down Expand Up @@ -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);
}
});
}
Expand Down Expand Up @@ -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,
Expand Down
5 changes: 5 additions & 0 deletions kv_cache_manager/manager/meta_searcher.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)>;
Expand Down
Loading
Loading