Skip to content

Commit f905d96

Browse files
charptycharpty
committed
[manager] optimize and harden ReportEvent hot path
Flatten ReportEvent aggregation, reduce URI/copy and local-lock overhead, and preserve validation, ordering, and capacity semantics. Harden EventReportBackend shutdown synchronization and expand correctness, sanitizer, and performance coverage and documentation.
1 parent ca3bfcf commit f905d96

21 files changed

Lines changed: 1330 additions & 215 deletions

‎docs/api/report_event.md‎

Lines changed: 11 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -619,22 +619,22 @@ HTTP 接口为 `POST /api/getCacheLocation`：
619619
`ST_EVENT_REPORT_L1P5`、`ST_EVENT_REPORT_L2` 等 backend，适合验证两种 EventReport storage 的
620620
隔离状态。
621621

622-
`location_spec_names` 不只是返回结果的投影条件，也是 backend/peer 选择前的候选条件：
622+
`location_spec_names` 不只是返回结果的投影条件，也是 backend/peer 选择前按 query key 生效的候选条件：
623623

624624
- 为空时，location 中任意合法 spec 都可使该 location 成为候选；
625-
- 非空时，location 至少包含一个请求的 spec name 才能成为候选；
625+
- 非空时，数组长度必须等于 query key 数量，且每个 name 都不能为空；第 i 个 name 只过滤第 i 个 key；
626+
- 第 i 个 key 的 location 必须包含对应的 spec name 才能成为候选，selector 也从该 spec URI 提取 peer；
626627
- 同一 EventReport location 由 `storage_type + medium + host_ip_port` 标识，其中各 spec 必须属于
627-
同一个 reporter endpoint；location 命中过滤后，selector 从第一个合法 spec URI 提取 peer；
628-
- 多个 peer 的 prefix/coverage 相同时按 endpoint 字典序选择，保证调用方按 spec 分批查询时
629-
各批次不会因为容器遍历顺序选择不同 peer；
630-
- peer 选择完成后，响应仍只保留 `location_spec_names` 指定的 specs。
628+
同一个 reporter endpoint；
629+
- 多个 peer 的 prefix/coverage 相同时按 endpoint 字典序选择，避免容器遍历顺序引起选择抖动；
630+
- peer 选择完成后，第 i 个 key 的响应仍只保留其对应 name 指定的 spec。
631631

632632
因此 spec name 是 reporter 与查询方之间的稳定协议字段，不能用 object size 代替：不同 cache
633-
group 即使 byte size 相同，也必须使用不同且稳定的 spec name。调用方如果每个 key 需要的 spec
634-
不同，应先按 spec name 分组发起查询，再按原 object key 合并结果；`location_spec_names` 是一次
635-
请求级过滤条件，不是 per-key 数组。确定性 tie-break 只消除无序遍历造成的抖动；若各组候选
636-
peer 集合不同，分组请求无法保证得到全局最优公共 peer。该能力需要后续扩展 per-key spec filter
637-
或等价的联合选择接口。
633+
group 即使 byte size 相同，也必须使用不同且稳定的 spec name。调用方必须让
634+
`location_spec_names` 与 `block_keys`（或由 token 生成的 query keys）同序对齐。同一个 block key
635+
可以在不同位置重复并请求不同 spec，用于 mixed-attention/Mamba groups。长度不匹配或包含空
636+
name 会返回 `INVALID_ARGUMENT`。确定性 tie-break 只消除无序遍历造成的抖动；各 key 经过
637+
spec 过滤后的候选 peer 集合仍可能不同。
638638

639639
### 11.4 GetHostCacheState
640640

‎docs/design/report_event_performance.md‎

Lines changed: 254 additions & 0 deletions
Large diffs are not rendered by default.

‎integration_test/meta_service/http_interface_test.py‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -243,7 +243,7 @@ def test_event_report_requested_spec_filters_before_peer_selection(self):
243243
"query_type": "QT_BATCH_GET",
244244
"block_keys": block_keys,
245245
"block_mask": {"offset": 0},
246-
"location_spec_names": ["linear_1"],
246+
"location_spec_names": ["linear_1"] * len(block_keys),
247247
"backend_selectors": [{
248248
"backend_type": "ST_EVENT_REPORT_L2",
249249
"strategy": strategy,
@@ -264,7 +264,7 @@ def test_event_report_requested_spec_filters_before_peer_selection(self):
264264
"query_type": "QT_BATCH_GET",
265265
"block_keys": block_keys,
266266
"block_mask": {"offset": 0},
267-
"location_spec_names": ["unknown_spec"],
267+
"location_spec_names": ["unknown_spec"] * len(block_keys),
268268
"backend_selectors": [{
269269
"backend_type": "ST_EVENT_REPORT_L2",
270270
"strategy": "LSS_V6D_PREFIX",

‎kv_cache_manager/common/standard_uri.cc‎

Lines changed: 84 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
#include "kv_cache_manager/common/standard_uri.h"
22

3-
#include "kv_cache_manager/common/string_util.h"
3+
#include <sstream>
44

55
namespace kv_cache_manager {
66

@@ -28,67 +28,77 @@ bool StandardUri::Parse(const std::string &uri) {
2828
protocol_ = uri.substr(0, pos_protocol_end);
2929

3030
size_t authority_start = pos_protocol_end + 3; // skip ://
31+
// Locate the end of authority before interpreting '@' or ':'. Delimiter
32+
// characters in the path/query belong to their values, not user-info or
33+
// host/port (for example callback URLs and email addresses).
34+
size_t pos_path_start = uri.find('/', authority_start);
35+
size_t pos_query_start = uri.find('?', authority_start);
36+
size_t host_end = std::min((pos_path_start != std::string::npos ? pos_path_start : uri.size()),
37+
(pos_query_start != std::string::npos ? pos_query_start : uri.size()));
3138
size_t host_start = authority_start;
3239
size_t pos_at = uri.find('@', authority_start);
33-
if (pos_at != std::string::npos) {
40+
if (pos_at != std::string::npos && pos_at < host_end) {
3441
user_info_ = uri.substr(authority_start, pos_at - authority_start);
3542
host_start = pos_at + 1; // hostname 开始位置
3643
}
3744

38-
// 找 hostname 结束的位置（可能有 port）
39-
size_t pos_path_start = uri.find('/', authority_start);
40-
size_t pos_query_start = uri.find('?', authority_start);
41-
size_t host_end = std::min((pos_path_start != std::string::npos ? pos_path_start : uri.size()),
42-
(pos_query_start != std::string::npos ? pos_query_start : uri.size())
43-
44-
);
45-
// 分离 hostname 和 port
46-
std::string host_port = uri.substr(host_start, host_end - host_start);
47-
size_t colon_pos = host_port.find(':');
48-
if (colon_pos == std::string::npos) {
49-
hostname_ = host_port;
45+
// 分离 hostname 和 port。直接在输入中定位，避免为每个 URI 先复制
46+
// 一份 host:port 临时字符串。
47+
size_t colon_pos = uri.find(':', host_start);
48+
if (colon_pos == std::string::npos || colon_pos >= host_end) {
49+
hostname_ = uri.substr(host_start, host_end - host_start);
5050
} else {
51-
hostname_ = host_port.substr(0, colon_pos);
52-
std::string port_str = host_port.substr(colon_pos + 1);
51+
hostname_ = uri.substr(host_start, colon_pos - host_start);
5352
int64_t tmp_port = 0;
54-
if (!StringUtil::StrToInt64(port_str.c_str(), tmp_port)) {
53+
const char *port_begin = uri.data() + colon_pos + 1;
54+
const char *port_end = uri.data() + host_end;
55+
const auto [parsed_end, parse_ec] = std::from_chars(port_begin, port_end, tmp_port);
56+
if (port_begin == port_end || *port_begin == '-' || parse_ec != std::errc{} || parsed_end != port_end) {
57+
// Parse() is also used through the direct string constructor,
58+
// whose caller observes validity rather than the return value.
59+
// Do not leave a partially parsed object looking valid.
60+
protocol_.clear();
61+
user_info_.clear();
62+
hostname_.clear();
63+
port_ = 0;
64+
path_.clear();
65+
params_.clear();
5566
return false;
5667
} else {
5768
port_ = tmp_port;
5869
}
5970
}
6071

6172
// 提取 path 和 query
62-
if (pos_path_start != std::string::npos && pos_path_start < uri.size()) {
73+
if (pos_path_start != std::string::npos &&
74+
(pos_query_start == std::string::npos || pos_path_start < pos_query_start)) {
6375
if (pos_query_start != std::string::npos && pos_path_start < pos_query_start) {
6476
path_ = uri.substr(pos_path_start, pos_query_start - pos_path_start);
65-
std::string query_str = uri.substr(pos_query_start + 1);
66-
ParseParams(query_str);
77+
ParseParams(std::string_view(uri).substr(pos_query_start + 1));
6778
} else {
6879
path_ = uri.substr(pos_path_start);
6980
}
7081
} else if (pos_query_start != std::string::npos && pos_query_start < uri.size()) {
71-
std::string query_str = uri.substr(pos_query_start + 1);
72-
ParseParams(query_str);
82+
ParseParams(std::string_view(uri).substr(pos_query_start + 1));
7383
}
7484
return true;
7585
}
7686

77-
bool StandardUri::ParseParams(const std::string &uri_params) {
78-
auto start = 0;
87+
bool StandardUri::ParseParams(std::string_view uri_params) {
88+
size_t start = 0;
7989
while (start < uri_params.size()) {
8090
auto end = uri_params.find('&', start);
8191
if (end == std::string::npos) {
8292
end = uri_params.size();
8393
}
8494
auto eq_pos = uri_params.find('=', start);
8595
if (eq_pos != std::string::npos && eq_pos < end) {
86-
std::string key = uri_params.substr(start, eq_pos - start);
87-
std::string value = uri_params.substr(eq_pos + 1, end - eq_pos - 1);
96+
std::string key(uri_params.substr(start, eq_pos - start));
97+
std::string value(uri_params.substr(eq_pos + 1, end - eq_pos - 1));
8898
params_[key] = value;
8999
} else {
90100
// key但无value，value空字符串
91-
std::string key = uri_params.substr(start, end - start);
101+
std::string key(uri_params.substr(start, end - start));
92102
params_[key] = "";
93103
}
94104
start = end + 1;
@@ -125,6 +135,53 @@ std::string StandardUri::ToUriString() const {
125135
return ss.str();
126136
}
127137

138+
std::string StandardUri::ToUriStringWithExtraParam(const std::string &key, const std::string &value) const {
139+
if (!Valid() || key.empty() || HasParam(key)) {
140+
return "";
141+
}
142+
143+
size_t estimated_size =
144+
protocol_.size() + user_info_.size() + hostname_.size() + path_.size() + key.size() + value.size() + 8;
145+
for (const auto &[param_key, param_value] : params_) {
146+
estimated_size += param_key.size() + param_value.size() + 2;
147+
}
148+
std::string result;
149+
result.reserve(estimated_size);
150+
result.append(protocol_).append("://");
151+
if (!user_info_.empty()) {
152+
result.append(user_info_).push_back('@');
153+
}
154+
result.append(hostname_);
155+
if (port_ > 0) {
156+
result.push_back(':');
157+
result.append(std::to_string(port_));
158+
}
159+
result.append(path_);
160+
result.push_back('?');
161+
162+
bool first = true;
163+
bool extra_written = false;
164+
auto append_param = [&result, &first](const std::string &param_key, const std::string &param_value) {
165+
if (!first) {
166+
result.push_back('&');
167+
}
168+
result.append(param_key).push_back('=');
169+
result.append(param_value);
170+
first = false;
171+
};
172+
for (const auto &[param_key, param_value] : params_) {
173+
if (!extra_written && key < param_key) {
174+
append_param(key, value);
175+
extra_written = true;
176+
}
177+
append_param(param_key, param_value);
178+
}
179+
if (!extra_written) {
180+
append_param(key, value);
181+
}
182+
return result;
183+
}
184+
128185
StandardUri StandardUri::FromUri(const std::string &source) {
129186
StandardUri result;
130187
if (!result.Parse(source)) {

‎kv_cache_manager/common/standard_uri.h‎

Lines changed: 9 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@
33
#include <charconv>
44
#include <map>
55
#include <string>
6+
#include <string_view>
67

78
namespace kv_cache_manager {
89

@@ -14,6 +15,10 @@ class StandardUri {
1415
public:
1516
bool Parse(const std::string &Uri);
1617
std::string ToUriString() const;
18+
// Serialize the URI as if one new query parameter had been inserted,
19+
// without cloning/mutating the parameter map. The output keeps the same
20+
// sorted canonical form as SetParam() followed by ToUriString().
21+
std::string ToUriStringWithExtraParam(const std::string &key, const std::string &value) const;
1722

1823
bool Valid() const { return !protocol_.empty(); }
1924
const std::string &GetProtocol() const { return protocol_; }
@@ -39,10 +44,11 @@ class StandardUri {
3944
std::string GetParam(const std::string &key) const;
4045
template <typename T>
4146
void GetParamAs(const std::string &key, T &t) const {
42-
std::string val = GetParam(key);
43-
if (val.empty()) {
47+
const auto it = params_.find(key);
48+
if (it == params_.end() || it->second.empty()) {
4449
return;
4550
}
51+
const std::string &val = it->second;
4652
T result;
4753
auto [ptr, ec] = std::from_chars(val.data(), val.data() + val.size(), result);
4854
if (ec == std::errc{} && ptr == val.data() + val.size()) {
@@ -62,7 +68,7 @@ class StandardUri {
6268
static std::string ToUri(const StandardUri &source);
6369

6470
private:
65-
bool ParseParams(const std::string &Uri_params);
71+
bool ParseParams(std::string_view Uri_params);
6672

6773
private:
6874
std::string protocol_;

‎kv_cache_manager/common/test/standard_uri_test.cc‎

Lines changed: 50 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -369,6 +369,42 @@ TEST_F(StandardUriTest, TestFileUri) {
369369
}
370370
}
371371

372+
TEST_F(StandardUriTest, TestSerializeWithExtraParamMatchesCanonicalMutation) {
373+
for (const std::string &raw_uri : {
374+
"event_report://host:8080/mem",
375+
"event_report://host:8080/mem?block=7&phase=add",
376+
"event_report://user@host:8080/mem?z=last&a=first",
377+
"file://host/path?empty=&flag",
378+
}) {
379+
StandardUri uri = StandardUri::FromUri(raw_uri);
380+
ASSERT_TRUE(uri.Valid());
381+
StandardUri expected = uri;
382+
expected.SetParam("s_version", "0123456789abcdef0123456789abcdef");
383+
EXPECT_EQ(expected.ToUriString(),
384+
uri.ToUriStringWithExtraParam("s_version", "0123456789abcdef0123456789abcdef"));
385+
}
386+
387+
StandardUri uri = StandardUri::FromUri("event_report://host:8080/mem?s_version=existing");
388+
ASSERT_TRUE(uri.Valid());
389+
EXPECT_TRUE(uri.ToUriStringWithExtraParam("s_version", "replacement").empty());
390+
EXPECT_TRUE(uri.ToUriStringWithExtraParam("", "value").empty());
391+
EXPECT_TRUE(StandardUri().ToUriStringWithExtraParam("key", "value").empty());
392+
}
393+
394+
TEST_F(StandardUriTest, TestQueryDelimitersDoNotChangeAuthorityOrPath) {
395+
const std::string raw_uri = "event_report://cache-host?callback=http://peer/path&owner=user@example.com";
396+
StandardUri uri(raw_uri);
397+
ASSERT_TRUE(uri.Valid());
398+
EXPECT_TRUE(uri.GetUserInfo().empty());
399+
EXPECT_EQ("cache-host", uri.GetHostName());
400+
EXPECT_TRUE(uri.GetPath().empty());
401+
EXPECT_EQ("http://peer/path", uri.GetParam("callback"));
402+
EXPECT_EQ("user@example.com", uri.GetParam("owner"));
403+
EXPECT_EQ(raw_uri, uri.ToUriString());
404+
EXPECT_EQ("event_report://cache-host?callback=http://peer/path&owner=user@example.com&s_version=token",
405+
uri.ToUriStringWithExtraParam("s_version", "token"));
406+
}
407+
372408
TEST_F(StandardUriTest, TestInvalidPort) {
373409
{
374410
std::string redis_uri_str = "redis://user:pw@127.0.0.1";
@@ -378,10 +414,21 @@ TEST_F(StandardUriTest, TestInvalidPort) {
378414
ASSERT_EQ("127.0.0.1", redis_uri.GetHostName());
379415
ASSERT_EQ(0, redis_uri.GetPort()); // default 0
380416
}
381-
{
382-
std::string redis_uri_str = "redis://user:pw@127.0.0.1:abcd/";
383-
StandardUri redis_uri = StandardUri::FromUri(redis_uri_str);
417+
for (const std::string &invalid_uri : {
418+
"redis://user:pw@127.0.0.1:abcd/",
419+
"redis://user:pw@127.0.0.1:-1/",
420+
"redis://user:pw@127.0.0.1:-0/",
421+
"redis://user:pw@127.0.0.1:+6379/",
422+
"redis://user:pw@127.0.0.1: 6379/",
423+
"redis://user:pw@127.0.0.1:/",
424+
}) {
425+
StandardUri redis_uri = StandardUri::FromUri(invalid_uri);
384426
ASSERT_FALSE(redis_uri.Valid());
427+
428+
StandardUri directly_constructed(invalid_uri);
429+
EXPECT_FALSE(directly_constructed.Valid());
430+
EXPECT_TRUE(directly_constructed.ToUriString().empty());
431+
EXPECT_TRUE(directly_constructed.ToUriStringWithExtraParam("key", "value").empty());
385432
}
386433
}
387434
} // namespace kv_cache_manager

‎kv_cache_manager/data_storage/event_report_backend.cc‎

Lines changed: 26 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -166,23 +166,41 @@ ErrorCode EventReportBackend::Close() {
166166
if (liveness_checker_thread_.joinable()) {
167167
liveness_checker_thread_.join();
168168
}
169-
std::lock_guard<std::mutex> fences_guard(lifecycle_fences_mutex_);
170-
std::vector<std::unique_lock<std::shared_mutex>> fence_locks;
171-
fence_locks.reserve(lifecycle_fences_.size());
172-
for (const auto &entry : lifecycle_fences_) {
173-
const auto &fence = entry.second;
174-
if (fence) {
175-
fence_locks.emplace_back(fence->mutex);
169+
std::vector<std::shared_ptr<LifecycleFence>> fence_refs;
170+
{
171+
std::lock_guard<std::mutex> fences_guard(lifecycle_fences_mutex_);
172+
fence_refs.reserve(lifecycle_fences_.size());
173+
for (const auto &entry : lifecycle_fences_) {
174+
if (entry.second) {
175+
fence_refs.push_back(entry.second);
176+
}
176177
}
177178
}
179+
180+
// Never wait for a lifecycle fence while holding lifecycle_fences_mutex_.
181+
// Cleanup deliberately takes lifecycle -> metadata, while a metadata RMW
182+
// may already hold metadata when it briefly looks up and try-locks its
183+
// lifecycle fence. Close holding the table mutex while waiting for the
184+
// cleanup lease would complete a three-lock cycle. The strong references
185+
// also keep each shared_mutex alive until its unique_lock is released.
186+
std::vector<std::unique_lock<std::shared_mutex>> fence_locks;
187+
fence_locks.reserve(fence_refs.size());
188+
for (const auto &fence : fence_refs) {
189+
fence_locks.emplace_back(fence->mutex);
190+
}
178191
{
179192
std::unique_lock<std::shared_mutex> lock(nodes_mutex_);
180193
instance_nodes_.clear();
181194
node_generation_.clear();
182195
snapshot_versions_.clear();
183196
snapshot_token_owners_.clear();
184197
}
185-
lifecycle_fences_.clear();
198+
{
199+
std::lock_guard<std::mutex> fences_guard(lifecycle_fences_mutex_);
200+
lifecycle_fences_.clear();
201+
}
202+
fence_locks.clear();
203+
fence_refs.clear();
186204
snapshot_state_cv_.notify_all();
187205
{
188206
std::lock_guard<std::mutex> lock(cleanup_cb_mutex_);

‎kv_cache_manager/data_storage/snapshot_uri_utils.h‎

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -197,8 +197,7 @@ class SnapshotUriUtils {
197197
if (!uri.Valid() || !IsValidSnapshotVersionToken(version) || HasEventReportInternalUriMetadata(uri)) {
198198
return false;
199199
}
200-
uri.SetParam(kSnapshotVersionParam, version);
201-
out_uri = uri.ToUriString();
200+
out_uri = uri.ToUriStringWithExtraParam(kSnapshotVersionParam, version);
202201
return !out_uri.empty();
203202
}
204203

0 commit comments

Comments
 (0)