mooncake-track
/
Mooncake
0
0
Fork 12
Code Issues Pull Requests 13 Actions Packages Projects Releases Wiki Activity

[CCF Archive] Store object type eviction policy submission #3

Closed
kancel wants to merge 382 commits from kancel:ccf-archive-pr2746 into main
pull from: kancel:ccf-archive-pr2746
merge into: mooncake-track:main
Conversation 0 Commits 382 Files Changed 654 +10 -7
2 changed files with 10 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit 7d26a326e7 - Show all commits

8
mooncake-store/src/master_service.cpp
View File

@ -3476,7 +3476,13 @@ auto MasterService::NotifyPromotionFailure(const UUID& client_id,
void MasterService::EvictionThreadFunc() {
VLOG(1) << "action=eviction_thread_started";
auto last_discard_time = std::chrono::system_clock::now();
// Start with an already-elapsed window so the first loop iteration
// (after kEvictionThreadSleepMs) triggers DiscardExpiredProcessingReplicas.
// Without this, a task admitted shortly after thread startup can survive
// the first reaper cycle and not be cleaned until ~2s later, causing
// promotion-on-hit tests that sleep for 2s to flake.
auto last_discard_time =
std::chrono::system_clock::now() - put_start_release_timeout_sec_;
while (eviction_running_) {
const auto now = std::chrono::system_clock::now();
double used_ratio =

9
mooncake-transfer-engine/src/transfer_metadata.cpp
View File

@ -1128,7 +1128,7 @@ int TransferMetadata::addRpcMetaEntry(const std::string &server_name,
Json::Value rpcMetaJSON;
rpcMetaJSON["ip_or_host_name"] = desc.ip_or_host_name;
rpcMetaJSON["rpc_port"] = static_cast<Json::UInt64>(desc.rpc_port);
rpcMetaJSON["rpc_port"] = static_cast<Json::UInt>(desc.rpc_port);
if (!storage_plugin_->set(rpc_meta_prefix_ + server_name, rpcMetaJSON)) {
LOG(ERROR) << "Failed to set location of " << server_name;
return ERR_METADATA;
@ -1157,19 +1157,16 @@ int TransferMetadata::rePublishRpcMetaEntry(const std::string &server_name) {
if (storage_plugin_->get(full_key, existing)) {
Json::Value desired;
desired["ip_or_host_name"] = local_rpc_meta_.ip_or_host_name;
desired["rpc_port"] =
static_cast<Json::UInt64>(local_rpc_meta_.rpc_port);
desired["rpc_port"] = static_cast<Json::UInt>(local_rpc_meta_.rpc_port);
if (existing == desired) {
return 0;
}
storage_plugin_->remove(full_key);
}
LOG(INFO) << "Re-publishing RPC meta entry for " << server_name;
Json::Value rpcMetaJSON;
rpcMetaJSON["ip_or_host_name"] = local_rpc_meta_.ip_or_host_name;
rpcMetaJSON["rpc_port"] =
static_cast<Json::UInt64>(local_rpc_meta_.rpc_port);
rpcMetaJSON["rpc_port"] = static_cast<Json::UInt>(local_rpc_meta_.rpc_port);
if (!storage_plugin_->set(full_key, rpcMetaJSON)) {
LOG(ERROR) << "Failed to re-publish RPC meta entry for " << server_name;
return ERR_METADATA;