[event] add optimizer event stream publisher #289
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,44 @@ | ||
| #include "kv_cache_manager/event/event_publishers_config.h" | ||
|
|
||
| namespace kv_cache_manager { | ||
|
|
||
| bool LogEventPublisherConfig::FromRapidValue(const rapidjson::Value &rapid_value) { | ||
| KVCM_JSON_GET_DEFAULT_MACRO(rapid_value, "enable", enable_, true); | ||
| KVCM_JSON_GET_DEFAULT_MACRO(rapid_value, "queue_size", queue_size_, std::size_t(10000)); | ||
| return queue_size_ > 0; | ||
| } | ||
|
|
||
| void LogEventPublisherConfig::ToRapidWriter(rapidjson::Writer<rapidjson::StringBuffer> &writer) const noexcept { | ||
| Put(writer, "enable", enable_); | ||
| Put(writer, "queue_size", queue_size_); | ||
| } | ||
|
|
||
| bool OptimizerEventPublisherConfig::FromRapidValue(const rapidjson::Value &rapid_value) { | ||
| KVCM_JSON_GET_DEFAULT_MACRO(rapid_value, "enable", enable_, false); | ||
| KVCM_JSON_GET_DEFAULT_MACRO(rapid_value, "queue_size", queue_size_, std::size_t(100000)); | ||
| KVCM_JSON_GET_DEFAULT_MACRO(rapid_value, "max_subscribers", max_subscribers_, std::size_t(4)); | ||
| KVCM_JSON_GET_DEFAULT_MACRO(rapid_value, "subscriber_queue_size", subscriber_queue_size_, std::size_t(10000)); | ||
| return queue_size_ > 0 && max_subscribers_ > 0 && subscriber_queue_size_ > 0; | ||
| } | ||
|
|
||
| void OptimizerEventPublisherConfig::ToRapidWriter(rapidjson::Writer<rapidjson::StringBuffer> &writer) const noexcept { | ||
| Put(writer, "enable", enable_); | ||
| Put(writer, "queue_size", queue_size_); | ||
| Put(writer, "max_subscribers", max_subscribers_); | ||
| Put(writer, "subscriber_queue_size", subscriber_queue_size_); | ||
| } | ||
|
|
||
| bool EventPublishersConfig::FromRapidValue(const rapidjson::Value &rapid_value) { | ||
| log_ = LogEventPublisherConfig{}; | ||
| optimizer_ = OptimizerEventPublisherConfig{}; | ||
| KVCM_JSON_GET_MACRO(rapid_value, "log", log_); | ||
| KVCM_JSON_GET_MACRO(rapid_value, "optimizer", optimizer_); | ||
| return true; | ||
| } | ||
|
|
||
| void EventPublishersConfig::ToRapidWriter(rapidjson::Writer<rapidjson::StringBuffer> &writer) const noexcept { | ||
| Put(writer, "log", log_); | ||
| Put(writer, "optimizer", optimizer_); | ||
| } | ||
|
|
||
| } // namespace kv_cache_manager |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,54 @@ | ||
| #pragma once | ||
|
|
||
| #include <cstddef> | ||
|
|
||
| #include "kv_cache_manager/common/jsonizable.h" | ||
|
|
||
| namespace kv_cache_manager { | ||
|
|
||
| class LogEventPublisherConfig : public Jsonizable { | ||
| public: | ||
| bool FromRapidValue(const rapidjson::Value &rapid_value) override; | ||
| void ToRapidWriter(rapidjson::Writer<rapidjson::StringBuffer> &writer) const noexcept override; | ||
|
|
||
| bool enable() const { return enable_; } | ||
| std::size_t queue_size() const { return queue_size_; } | ||
|
|
||
| private: | ||
| bool enable_ = true; | ||
| std::size_t queue_size_ = 10000; | ||
| }; | ||
|
|
||
| class OptimizerEventPublisherConfig : public Jsonizable { | ||
| public: | ||
| bool FromRapidValue(const rapidjson::Value &rapid_value) override; | ||
| void ToRapidWriter(rapidjson::Writer<rapidjson::StringBuffer> &writer) const noexcept override; | ||
|
|
||
| bool enable() const { return enable_; } | ||
| std::size_t queue_size() const { return queue_size_; } | ||
| std::size_t max_subscribers() const { return max_subscribers_; } | ||
| std::size_t subscriber_queue_size() const { return subscriber_queue_size_; } | ||
|
|
||
| private: | ||
| bool enable_ = false; | ||
| std::size_t queue_size_ = 100000; | ||
| std::size_t max_subscribers_ = 4; | ||
| std::size_t subscriber_queue_size_ = 10000; | ||
| }; | ||
|
|
||
| class EventPublishersConfig : public Jsonizable { | ||
| public: | ||
| bool FromRapidValue(const rapidjson::Value &rapid_value) override; | ||
| void ToRapidWriter(rapidjson::Writer<rapidjson::StringBuffer> &writer) const noexcept override; | ||
|
|
||
| bool enable_log_event_publisher() const { return log_.enable(); } | ||
| const LogEventPublisherConfig &log_event_publisher_config() const { return log_; } | ||
| bool enable_optimizer_event_publisher() const { return optimizer_.enable(); } | ||
| const OptimizerEventPublisherConfig &optimizer_event_publisher_config() const { return optimizer_; } | ||
|
|
||
| private: | ||
| LogEventPublisherConfig log_; | ||
| OptimizerEventPublisherConfig optimizer_; | ||
| }; | ||
|
|
||
| } // namespace kv_cache_manager |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,123 @@ | ||
| #include "kv_cache_manager/event/optimizer_event_publisher.h" | ||
|
|
||
| #include <utility> | ||
|
|
||
| #include "kv_cache_manager/common/logger.h" | ||
| #include "kv_cache_manager/event/base_event.h" | ||
| #include "kv_cache_manager/event/spec_events/optimizer_event.h" | ||
| #include "kv_cache_manager/protocol/protobuf/optimizer_service.pb.h" | ||
|
|
||
| namespace kv_cache_manager { | ||
|
|
||
| OptimizerEventPublisher::OptimizerEventPublisher(std::shared_ptr<EventSink> sink, | ||
| const OptimizerEventPublisherConfig &config) | ||
| : sink_(std::move(sink)), config_(config) {} | ||
|
|
||
| OptimizerEventPublisher::~OptimizerEventPublisher() { | ||
| if (running_) { | ||
| Stop(); | ||
| } | ||
| } | ||
|
|
||
| bool OptimizerEventPublisher::Init(const std::string & /*config*/) { | ||
| if (!sink_) { | ||
| KVCM_LOG_ERROR("OptimizerEventPublisher: no sink provided"); | ||
| return false; | ||
| } | ||
| InitBasicQueue(config_.queue_size()); | ||
| running_ = true; | ||
| worker_ = std::thread(&OptimizerEventPublisher::WorkerThread, this); | ||
| KVCM_LOG_INFO("OptimizerEventPublisher: initialized, queue_size=%zu", config_.queue_size()); | ||
| return true; | ||
| } | ||
|
|
||
| bool OptimizerEventPublisher::Publish(const std::shared_ptr<BaseEvent> &event) { | ||
| if (!event || !running_) { | ||
| return false; | ||
| } | ||
| // Runs on a serving thread: enqueue and return, nothing else. A full queue | ||
| // drops the event (counted by the base class) without surfacing an expected | ||
| // best-effort drop as a publish failure and triggering a warning per request. | ||
| BasicEnqueue(event); | ||
| return true; | ||
| } | ||
|
|
||
| bool OptimizerEventPublisher::Stop() { | ||
| if (!running_) { | ||
| return true; | ||
| } | ||
| running_ = false; | ||
| ClearBasicQueue(); | ||
| // Required, not belt-and-braces: BasicWait() blocks in | ||
| // condition_variable::wait(), whose predicate is only re-evaluated when it | ||
| // is notified. Clearing running_ alone leaves the worker asleep and join() | ||
| // below would never return. | ||
| if (basic_queue_) { | ||
| basic_queue_->queue_cv.notify_all(); | ||
| } | ||
| if (worker_.joinable()) { | ||
| worker_.join(); | ||
| } | ||
| if (sink_) { | ||
| sink_->Stop(); | ||
| } | ||
| KVCM_LOG_INFO("OptimizerEventPublisher: stopped, forwarded=%zu skipped=%zu queue_dropped=%zu", | ||
| forwarded_.load(), | ||
| skipped_.load(), | ||
| BasicDroppedCount()); | ||
| return true; | ||
| } | ||
|
|
||
| void OptimizerEventPublisher::WorkerThread() { | ||
| // A single worker keeps conversion and delivery off serving threads | ||
| // without adding synchronization between multiple consumers of the queue. | ||
| while (running_) { | ||
| BasicWait(); | ||
|
|
||
| std::shared_ptr<BaseEvent> event; | ||
| while (BasicDequeue(event)) { | ||
| proto::optimizer::TraceQueryRequest request; | ||
| if (!Convert(event, &request)) { | ||
| skipped_.fetch_add(1); | ||
| continue; | ||
| } | ||
| if (sink_->Send(request)) { | ||
| forwarded_.fetch_add(1); | ||
| } | ||
| } | ||
| } | ||
| } | ||
|
|
||
| bool OptimizerEventPublisher::Convert(const std::shared_ptr<BaseEvent> &event, | ||
| proto::optimizer::TraceQueryRequest *out) { | ||
| // Every publisher registered with EventManager sees every event, so write | ||
| // and reclaim events arrive here too. Only cache reads can be replayed. | ||
| const auto *get_event = dynamic_cast<const CacheGetEvent *>(event.get()); | ||
| if (get_event == nullptr) { | ||
| return false; | ||
| } | ||
|
|
||
| out->set_trace_id(get_event->trace_id()); | ||
| // CacheGetEvent's source is the instance the request was served for. | ||
| out->set_instance_id(get_event->event_source()); | ||
| for (const auto key : get_event->get_keys()) { | ||
| out->add_block_keys(key); | ||
| } | ||
|
Comment on lines
+103
to
+105
There was a problem hiding this comment.
When KVCM serves Useful? React with 👍 / 👎.
Collaborator
Author
There was a problem hiding this comment. 按 KVCM 职责边界不在这里过滤。KVCM 只透明采集和转发已发生的 Query，不理解或重解释 QT_BATCH_GET 等具体查询语义；如不同查询类型需要不同回放模型，应由协议语义和 Optimizer 消费实现统一处理。 |
||
| // The event carries microseconds; the replay works in nanoseconds. | ||
| out->set_timestamp_ns(get_event->event_trigger_time_us() * 1000); | ||
|
There was a problem hiding this comment.
For a slow metadata or backend lookup, this forwards a completion-time timestamp rather than the time the cache access occurred: each producing Useful? React with 👍 / 👎. |
||
|
|
||
| // Only the token count is needed, never the ids. An empty tokens vector | ||
| // means the caller passed pre-computed keys instead, and then the exact | ||
| // input length is simply not known here: 0 says "unknown" and the | ||
| // consumer infers it from block count times block size. That inference | ||
| // loses the trailing partial block, which biases the hit rate upwards - | ||
| // the opposite direction from dropped events, so the two do not cancel. | ||
| out->set_input_token_len(static_cast<std::int64_t>(get_event->get_tokens().size())); | ||
|
|
||
| // An empty block_keys list is legitimate, not junk: a prompt shorter than | ||
| // one block has no complete block. Its input_token_len still belongs in | ||
| // the hit-rate denominator, so the event is forwarded as-is. | ||
| return true; | ||
| } | ||
|
|
||
| } // namespace kv_cache_manager | ||
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This target introduces a direct
event → protocolmodule dependency, but neither the root dependency diagram nordocs/design/module_architecture.mdwas updated; the latter still depictseventas depending only oncommon. Update both architecture diagrams and the event module description so future dependency checks account for this new edge.AGENTS.md reference: AGENTS.md:L38-L38
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
已修复。根目录 AGENTS.md 与 docs/design/module_architecture.md 的依赖图均补充了 event → protocol 边，并更新 event/protocol 模块说明，记录 OptimizerEventPublisher、TraceQueryRequest、SubscriptionEventSink 与 gRPC 事件流的关系。