[store]feat: add conductor#2833
Conversation
Chase-Rong
requested review from
Ann-1024,
ShangmingCai,
luketong777,
stmatengss and
ykwd
as code owners
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces the mooncake-conductor service, which coordinates and indexes KV cache metadata for Mooncake and vLLM instances. It includes a prefix indexer (PrefixCacheTable), a ZMQ client for event subscription, and an HTTP server for service registration and querying. The review highlights several critical issues, including a concurrency bug in GetContextData that can cause silent cache misses, a data race in Start() during concurrent service registration, an ineffective timeout loop in the HTTP server shutdown, and missing error checks during ZMQ client reconnection. Additionally, the message decoder overly restricts parent block hashes, and service subscriptions lack checks for a stopped manager state.
Important
The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.
| auto [it, inserted] = | ||
| context_map_.try_emplace(model_context, new_context_data); | ||
| if (inserted) { | ||
| return new_context_data; | ||
| } | ||
| } | ||
|
|
||
| context_count_.fetch_add(1); | ||
| VLOG(1) << "in getContextData modelcontext model=" | ||
| << model_context.model_name | ||
| << " instance=" << model_context.instance_id; | ||
| return new_context_data; |
There was a problem hiding this comment.
There is a critical concurrency bug in GetContextData. If the insertion fails because another thread has already inserted the context (inserted is false), the function still returns new_context_data (which is orphaned and not in the map) instead of the existing context data (it->second). Furthermore, context_count_ is incorrectly incremented only on duplicate/failed insertions instead of successful ones. This will lead to lost updates and silent cache misses.
| auto [it, inserted] = | |
| context_map_.try_emplace(model_context, new_context_data); | |
| if (inserted) { | |
| return new_context_data; | |
| } | |
| } | |
| context_count_.fetch_add(1); | |
| VLOG(1) << "in getContextData modelcontext model=" | |
| << model_context.model_name | |
| << " instance=" << model_context.instance_id; | |
| return new_context_data; | |
| auto [it, inserted] = | |
| context_map_.try_emplace(model_context, new_context_data); | |
| if (inserted) { | |
| context_count_.fetch_add(1); | |
| VLOG(1) << "in getContextData modelcontext model=" | |
| << model_context.model_name | |
| << " instance=" << model_context.instance_id; | |
| return new_context_data; | |
| } | |
| return it->second; | |
| } |
| std::atomic<int> failure_count{0}; | ||
| std::vector<std::thread> workers; | ||
| workers.reserve(services_.size()); | ||
| for (const auto& svc : services_) { | ||
| workers.emplace_back([this, svc, &failure_count] { |
There was a problem hiding this comment.
The Start() method iterates over services_ without holding any lock. Since the HTTP server is started before Start() is called, a concurrent /register request can modify services_ via services_.push_back(svc) under mu_, leading to a critical data race and undefined behavior. Copy services_ under a shared lock before iterating.
std::vector<common::ServiceConfig> services_copy;
{
std::shared_lock lock(mu_);
services_copy = services_;
}
std::atomic<int> failure_count{0};
std::vector<std::thread> workers;
workers.reserve(services_copy.size());
for (const auto& svc : services_copy) {
workers.emplace_back([this, svc, &failure_count] {| std::thread shutdown_thread([this] { | ||
| LOG(INFO) << "Shutting down HTTP server"; | ||
| http_server_->stop(); | ||
| }); | ||
| // 5-second timeout for forced shutdown | ||
| const auto deadline = | ||
| std::chrono::steady_clock::now() + std::chrono::seconds(5); | ||
| // stop() blocks until connections drain; poll for completion. | ||
| while (std::chrono::steady_clock::now() < deadline && | ||
| shutdown_thread.joinable()) { | ||
| // join with timeout emulation: try a short join via native | ||
| // handle is non-portable; instead detach after deadline. | ||
| break; | ||
| } | ||
| shutdown_thread.join(); |
There was a problem hiding this comment.
The HTTP server shutdown loop contains an immediate break; on the very first iteration. This makes the 5-second timeout completely ineffective, and the subsequent shutdown_thread.join() call will block the calling thread indefinitely if the server fails to stop or if connections do not drain. Consider using an atomic flag to poll the thread status and detaching the thread if the timeout is exceeded.
std::atomic<bool> done{false};
std::thread shutdown_thread([this, &done] {
LOG(INFO) << "Shutting down HTTP server";
http_server_->stop();
done.store(true);
});
const auto deadline =
std::chrono::steady_clock::now() + std::chrono::seconds(5);
while (std::chrono::steady_clock::now() < deadline && !done.load()) {
std::this_thread::sleep_for(std::chrono::milliseconds(100));
}
if (done.load()) {
shutdown_thread.join();
} else {
LOG(WARNING) << "HTTP server shutdown timed out, detaching thread";
shutdown_thread.detach();
}| if (auto err = Connect(); !err.empty()) { | ||
| LOG(ERROR) << "Reconnect failed service=" << config_.cache_pool_key | ||
| << " error=" << err; | ||
| } |
There was a problem hiding this comment.
If Connect() fails during reconnection, the error is logged but the function still proceeds to request a replay. Since the sockets are not connected or are null, calling RequestReplay will fail or cause undefined behavior. Return early from HandleReconnect if the connection attempt fails.
| if (auto err = Connect(); !err.empty()) { | |
| LOG(ERROR) << "Reconnect failed service=" << config_.cache_pool_key | |
| << " error=" << err; | |
| } | |
| if (auto err = Connect(); !err.empty()) { | |
| LOG(ERROR) << "Reconnect failed service=" << config_.cache_pool_key | |
| << " error=" << err; | |
| return; | |
| } |
| } else if (f[2].type == object_type::POSITIVE_INTEGER && | ||
| f[2].via.u64 > 0xFFFFFFFFULL) { | ||
| // Require a full 8-byte integer marker: minimal-width encoders | ||
| // only emit this width for values above 2**32. | ||
| event.parent_block_hash = f[2].via.u64; |
There was a problem hiding this comment.
The parser rejects any parent block hash that is less than or equal to 0xFFFFFFFFULL (2^32) with an expected integer error. Since parent block hashes can naturally be small integers (especially in test environments or with specific hash generation schemes), this check is overly restrictive and causes valid messages to be rejected. Msgpack-cxx automatically decodes integers of any size into via.u64, so checking f[2].type == object_type::POSITIVE_INTEGER is sufficient.
| } else if (f[2].type == object_type::POSITIVE_INTEGER && | |
| f[2].via.u64 > 0xFFFFFFFFULL) { | |
| // Require a full 8-byte integer marker: minimal-width encoders | |
| // only emit this width for values above 2**32. | |
| event.parent_block_hash = f[2].via.u64; | |
| } else if (f[2].type == object_type::POSITIVE_INTEGER) { | |
| event.parent_block_hash = f[2].via.u64; |
| std::pair<bool, std::string> EventManager::SubscribeToService( | ||
| const common::ServiceConfig& svc) { | ||
| // Use (instance_id, tenant_id) as composite key to support | ||
| // multi-tenant replicas |
There was a problem hiding this comment.
The SubscribeToService method does not check if the manager has been stopped. If a /register request is processed concurrently with or after Stop(), a new ZMQClient will be started and added to subscribers_, leaking background threads and resources. Check stopped_ and reject the subscription if the manager is stopped.
std::pair<bool, std::string> EventManager::SubscribeToService(
const common::ServiceConfig& svc) {
if (stopped_) {
return {false, "manager is stopped"};
}
// Use (instance_id, tenant_id) as composite key to support
// multi-tenant replicas| LOG(WARNING) << "Config file does not exist, exiting. path=" | ||
| << config_path; |
There was a problem hiding this comment.
The log message says exiting when the config file does not exist, but the code actually returns an empty list and continues execution normally. Update the log message to accurately reflect that execution is continuing.
| LOG(WARNING) << "Config file does not exist, exiting. path=" | |
| << config_path; | |
| LOG(WARNING) << "Config file does not exist, continuing with empty service list. path=" | |
| << config_path; |
github-actions
Bot
added
documentation
1 participant
Description
Add the initial C++ Mooncake Conductor service. Reference RFC：#977
This change introduces
mooncake-conductor, including:BlockStored/BlockRemovedevents.WITH_CONDUCTOR.Module
mooncake-transfer-engine)mooncake-store)mooncake-ep)mooncake-pg)mooncake-integration)mooncake-p2p-store)mooncake-wheel)mooncake-common)mooncake-rl)Type of Change
How Has This Been Tested?
Added C++ unit tests for conductor config/util helpers, hash golden vectors, JSON uint64
handling, msgpack decoders, prefix indexer behavior, event manager HTTP/service flows,
ZMQ client behavior, and concurrency coverage.
Checklist
AI Assistance Disclosure
Codex was used to draft this PR description from the local HEAD commit. The human
submitter is responsible for reviewing and defending all changed lines.