From 81ed7c0817715c73d22f0c0a4100803be0efe4db Mon Sep 17 00:00:00 2001 From: "Zhu, Ping" Date: Fri, 3 Jul 2026 12:50:13 +0800 Subject: [PATCH 01/18] store: flush partial bucket immediately at input end (Bug D) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit GroupOffloadingKeysByBucket previously returned early when the input exhausted mid-bucket, dumping the partial keys back into ungrouped_offloading_objects_. Under low inflow (e.g. after a pressure test rampdown, only 151-491 keys/call), the pool never reached bucket_keys_limit (1000), so keys sat indefinitely — their master-side offloading_task expired before ever being written to SSD. Fix: emit the partial bucket to buckets_keys immediately at input end instead of re-buffering. Trade a small per-bucket metadata overhead for correctness: no indefinite backlog. --- mooncake-store/src/storage_backend.cpp | 17 +++++++++++------ 1 file changed, 11 insertions(+), 6 deletions(-) diff --git a/mooncake-store/src/storage_backend.cpp b/mooncake-store/src/storage_backend.cpp index 3612624a3f..d617031e82 100644 --- a/mooncake-store/src/storage_backend.cpp +++ b/mooncake-store/src/storage_backend.cpp @@ -1914,13 +1914,18 @@ tl::expected BucketStorageBackend::GroupOffloadingKeysByBucket( for (int64_t i = static_cast(bucket_keys.size()); i < bucket_backend_config_.bucket_keys_limit; ++i) { if (it == offloading_objects.cend()) { - for (const auto& bucket_object : bucket_objects) { - ungrouped_offloading_objects.emplace(bucket_object.first, - bucket_object.second); + if (!bucket_keys.empty()) { + auto bucket_keys_count = + static_cast(bucket_keys.size()); + residue_count -= bucket_keys_count; + buckets_keys.push_back(std::move(bucket_keys)); + VLOG(1) << "[OFFLOAD-PARTIAL] flushed partial bucket at " + "input end: keys=" + << bucket_keys_count + << " data_size=" << bucket_data_size + << " total=" << total_count + << " residue=" << residue_count; } - VLOG(1) << "Add offloading objects to ungrouped pool. " - << "Total ungrouped count: " - << ungrouped_offloading_objects.size(); return {}; } From 7f57409cb5c86a134675e2d7aaf3562e40a603ea Mon Sep 17 00:00:00 2001 From: "Zhu, Ping" Date: Fri, 3 Jul 2026 12:52:46 +0800 Subject: [PATCH 02/18] store: AddReplica erases stale LOCAL_DISK replica on store restart (Bug H) When a store-server restarts with a new client_id, ScanMeta re-registers its LOCAL_DISK replicas via AddReplica. The previous code's VisitReplicas predicate required matching client_id, so the stale replica (from the dead client) blocked the update: the new replica was silently dropped and reads followed the dead transport_endpoint (INVALID_KEY / RDMA error). Fix: before checking for an existing LOCAL_DISK replica, erase any LOCAL_DISK replicas whose client_id differs from the caller's. This clears stale entries from restarted store-servers so the new replica can be added or the same-client_id one updated. --- mooncake-store/src/master_service.cpp | 23 +++++++++++++++++++++++ 1 file changed, 23 insertions(+) diff --git a/mooncake-store/src/master_service.cpp b/mooncake-store/src/master_service.cpp index c051fe1682..3a1203f947 100644 --- a/mooncake-store/src/master_service.cpp +++ b/mooncake-store/src/master_service.cpp @@ -3205,6 +3205,29 @@ auto MasterService::AddReplica(const UUID& client_id, const std::string& key, return tl::make_unexpected(ErrorCode::INVALID_PARAMS); } + // Erase stale LOCAL_DISK replicas from a previous (restarted) store-server + // incarnation. Without this, a stale transport_endpoint lingers in metadata + // after the store-server restarts with a new client_id; the visitor below + // requires matching client_id, so the new replica would be silently dropped + // and reads would follow the dead endpoint (INVALID_KEY / RDMA errors). + bool had_completed_disk = metadata.HasReplica([](const Replica& r) { + return r.is_local_disk_replica() && r.is_completed(); + }); + size_t stale_erased = EraseReplicasWithCacheTotalAccounting( + metadata, + [&client_id](const Replica& rep) { + return rep.is_local_disk_replica() && + rep.get_descriptor().get_local_disk_descriptor().client_id != + client_id; + }); + if (stale_erased > 0) { + auto& shard = accessor.GetShard(); + shard.OnDiskReplicaRemoved(had_completed_disk, metadata); + LOG(INFO) << "[ADDREPLICA-STALE] erased " << stale_erased + << " stale LOCAL_DISK replica(s) for key=" << key + << " tenant=" << normalized_tenant; + } + if (!metadata.HasReplica(&Replica::fn_is_local_disk_replica)) { std::vector replicas; replicas.emplace_back(std::move(replica)); From 05835124d4f4b8a1520fa2ae6003a3c38604da11 Mon Sep 17 00:00:00 2001 From: "Zhu, Ping" Date: Fri, 3 Jul 2026 12:53:13 +0800 Subject: [PATCH 03/18] store: PutStart blocks only on completed MEMORY replicas (Bug I) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The previous PutStart check used metadata.HasReplica(&Replica::fn_is_completed) which matches ANY completed replica type, including LOCAL_DISK. After a store-server restart, ScanMeta re-registers LOCAL_DISK replicas, leaving metadata with only a completed LOCAL_DISK replica. The next PutStart for that key was then rejected with OBJECT_ALREADY_EXISTS, blocking writes to keys whose cache lives on SSD. Fix: narrow the check to completed MEMORY replicas only. A LOCAL_DISK- only metadata object should not block a fresh PutStart — the disk replica's purpose is to serve cache reads, not to gate writes. --- mooncake-store/src/master_service.cpp | 14 +++++++++++++- 1 file changed, 13 insertions(+), 1 deletion(-) diff --git a/mooncake-store/src/master_service.cpp b/mooncake-store/src/master_service.cpp index 3a1203f947..f604330ed1 100644 --- a/mooncake-store/src/master_service.cpp +++ b/mooncake-store/src/master_service.cpp @@ -2998,7 +2998,19 @@ auto MasterService::PutStart(const UUID& client_id, const std::string& key, it = tenant_state.metadata.end(); } else { auto& metadata = it->second; - if (metadata.HasReplica(&Replica::fn_is_completed) || + // Block PutStart only when a completed MEMORY replica is + // present. A LOCAL_DISK-only metadata (e.g. left over after + // a store-server restart + ScanMeta recovery) must not + // block writes — the LOCAL_DISK replica is there to serve + // cache reads, not to gate new PutStart calls. Blocking on + // any completed replica type (the previous behavior) caused + // OBJECT_ALREADY_EXISTS for every key whose LOCAL_DISK + // replica was re-registered after restart. + bool has_completed_memory = metadata.HasReplica( + [](const Replica& r) { + return r.is_memory_replica() && r.is_completed(); + }); + if (has_completed_memory || metadata.put_start_time + put_start_discard_timeout_sec_ >= now) { From d2048ae65cd612d5fc5671f4fc985dceadd5ac83 Mon Sep 17 00:00:00 2001 From: "Zhu, Ping" Date: Fri, 3 Jul 2026 12:54:52 +0800 Subject: [PATCH 04/18] store: PutEnd offload path guards against emplace failure and early-exits on first pin (Bug K) PutEnd's offload-on-write path (!offload_on_evict_) lacked two safeguards: 1. No 'if (task_created) return' early-exit inside the VisitReplicas visitor. With replica_num>1 (MOONCAKE_STORE_REPLICA_NUM=2), the second MEMORY replica in a different segment would also succeed at PushOffloadingQueue + inc_refcnt, but the second emplace would silently fail (key already in offloading_tasks), leaking the refcnt. 2. No offloading_tasks.count(key)==0 pre-check. If an offloading task was already in flight (rare race), the visitor's first iteration would inc_refcnt + emplace-fail, also leaking. Fix: add the early-exit + pre-check + emplace return-value check with dec_refcnt rollback. This closes the refcnt leak that manifested as OFFLOAD-COMPLETE gap (admitted != completed) under replica_num=2. --- mooncake-store/src/master_service.cpp | 51 ++++++++++++++++++--------- 1 file changed, 35 insertions(+), 16 deletions(-) diff --git a/mooncake-store/src/master_service.cpp b/mooncake-store/src/master_service.cpp index f604330ed1..b1ac9994ee 100644 --- a/mooncake-store/src/master_service.cpp +++ b/mooncake-store/src/master_service.cpp @@ -3151,24 +3151,43 @@ auto MasterService::PutEnd(const UUID& client_id, const std::string& key, if (enable_offload_ && !offload_on_evict_) { auto& tenant_state = accessor.GetTenantState(); - bool task_created = false; - metadata.VisitReplicas( - [](const Replica& replica) { - return replica.is_completed() && replica.is_memory_replica(); - }, - [this, &object_id, &tenant_state, &task_created](Replica& replica) { - auto result = PushOffloadingQueue(object_id, replica); - if (result) { - if (!task_created) { + // Skip if an offloading task is already in flight for this key — + // emplace would silently fail, leaving the inc_refcnt below as a + // permanent leak. With replica_num>1, the visitor would keep iterating + // after the first emplace, double-pinning the second replica. + if (tenant_state.offloading_tasks.count(object_id.user_key) == 0) { + bool task_created = false; + metadata.VisitReplicas( + [](const Replica& replica) { + return replica.is_completed() && replica.is_memory_replica(); + }, + [this, &object_id, &tenant_state, + &task_created](Replica& replica) { + if (task_created) return; // only pin one replica per key + auto result = PushOffloadingQueue(object_id, replica); + if (result) { replica.inc_refcnt(); - tenant_state.offloading_tasks.emplace( - object_id.user_key, - OffloadingTask{replica.id(), - std::chrono::system_clock::now()}); - task_created = true; + auto [it, inserted] = + tenant_state.offloading_tasks.emplace( + object_id.user_key, + OffloadingTask{ + replica.id(), + std::chrono::system_clock::now()}); + if (!inserted) { + // Race: another path already queued this key. + // Roll back the refcnt so it doesn't leak. + replica.dec_refcnt(); + LOG(WARNING) + << "[PUTEND-OFFLOAD-EMPLACE-FAIL] key=" + << object_id.user_key + << " already has offloading_task; rolled " + "back refcnt"; + } else { + task_created = true; + } } - } - }); + }); + } } // If the object is completed, remove it from the processing set. From 77b43718e74d588c739ff0fad445f317d2fc6a9e Mon Sep 17 00:00:00 2001 From: "Zhu, Ping" Date: Fri, 3 Jul 2026 12:55:16 +0800 Subject: [PATCH 05/18] store: BatchEvict try_evict_or_offload guards against cross-cycle re-pin refcnt leak (Bug J part 1) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit BatchEvict's try_evict_or_offload lambda iterates candidate metadata and pins one MEMORY replica per key for SSD offload. Two issues caused a permanent refcnt leak with replica_num>1: 1. No offloading_tasks.count(key)==0 pre-check. After cycle N pins replica A, FetchOffloadingTasks on the store-server side clears the master's offloading_objects map; cycle N+1 can then pin replica B for the same key. The emplace silently no-ops (key already in offloading_tasks from cycle N), but B's inc_refcnt is never rolled back — the replica is pinned forever, never evictable. 2. emplace return value unchecked. Even within a single cycle, a rare race between BatchEvict and another Path (PutEnd offload, tenant quota eviction) can pre-insert the key, making emplace fail. Fix: add the count>0 pre-check and the emplace return-value check with dec_refcnt rollback. Defense-in-depth: even if the count pre-check passes, the emplace check catches races. --- mooncake-store/src/master_service.cpp | 50 ++++++++++++++++++--------- 1 file changed, 33 insertions(+), 17 deletions(-) diff --git a/mooncake-store/src/master_service.cpp b/mooncake-store/src/master_service.cpp index b1ac9994ee..76cc17eaaf 100644 --- a/mooncake-store/src/master_service.cpp +++ b/mooncake-store/src/master_service.cpp @@ -7188,24 +7188,40 @@ void MasterService::BatchEvict(double evict_ratio_target, } // Queue one MEMORY replica for offload; others will be evicted below. + // Skip if an offloading task is already in flight for this key — + // otherwise the emplace below silently fails and inc_refcnt leaks. bool queued = false; - metadata.VisitReplicas( - [](const Replica& r) { - return r.is_memory_replica() && r.is_completed() && - r.get_refcnt() == 0; - }, - [this, &tenant_id, &key, &tenant_state, &queued, - &now](Replica& replica) { - if (queued) return; // only need to pin one replica for offload - auto result = PushOffloadingQueue( - MakeObjectIdentity(key, tenant_id), replica); - if (result) { - replica.inc_refcnt(); - tenant_state.offloading_tasks.emplace( - key, OffloadingTask{replica.id(), now}); - queued = true; - } - }); + if (tenant_state.offloading_tasks.count(key) == 0) { + metadata.VisitReplicas( + [](const Replica& r) { + return r.is_memory_replica() && r.is_completed() && + r.get_refcnt() == 0; + }, + [this, &tenant_id, &key, &tenant_state, &queued, + &now](Replica& replica) { + if (queued) return; // only pin one replica for offload + auto result = PushOffloadingQueue( + MakeObjectIdentity(key, tenant_id), replica); + if (result) { + replica.inc_refcnt(); + auto [it, inserted] = + tenant_state.offloading_tasks.emplace( + key, OffloadingTask{replica.id(), now}); + if (!inserted) { + // Cross-cycle re-pin: another iteration already + // queued this key. Roll back to avoid refcnt leak. + replica.dec_refcnt(); + LOG(WARNING) + << "[BATCHEVICT-OFFLOAD-EMPLACE-FAIL] key=" + << key + << " already has offloading_task; rolled " + "back refcnt"; + } else { + queued = true; + } + } + }); + } if (queued) { offload_queued_this_cycle++; From d369085299ec7c9ca793b4f6d337105b1db00bfd Mon Sep 17 00:00:00 2001 From: "Zhu, Ping" Date: Fri, 3 Jul 2026 12:55:33 +0800 Subject: [PATCH 06/18] store: EvictTenantMemoryForQuota same refcnt-leak guard as BatchEvict (Bug J part 2) EvictTenantMemoryForQuota has its own try_evict_or_offload lambda with the same shape as BatchEvict's. Apply the same two safeguards: 1. offloading_tasks.count(key)==0 pre-check before VisitReplicas 2. emplace return-value check with dec_refcnt rollback Without these, tenant-quota-driven eviction under replica_num>1 can pin a second replica for a key whose offloading_tasks entry was inserted by BatchEvict, leaking the refcnt and making the replica unevictable. --- mooncake-store/src/master_service.cpp | 46 +++++++++++++++++---------- 1 file changed, 30 insertions(+), 16 deletions(-) diff --git a/mooncake-store/src/master_service.cpp b/mooncake-store/src/master_service.cpp index 76cc17eaaf..46db881959 100644 --- a/mooncake-store/src/master_service.cpp +++ b/mooncake-store/src/master_service.cpp @@ -6945,22 +6945,36 @@ MasterService::EvictTenantMemoryForQuota(const std::string& tenant_id, } bool queued = false; - metadata.VisitReplicas( - is_evictable_memory_replica, - [this, &key, &normalized_tenant, &tenant_state, &queued, - &now](Replica& replica) { - if (queued) { - return; - } - auto result = PushOffloadingQueue( - MakeObjectIdentity(key, normalized_tenant), replica); - if (result) { - replica.inc_refcnt(); - tenant_state.offloading_tasks.emplace( - key, OffloadingTask{replica.id(), now}); - queued = true; - } - }); + // Skip if an offloading task is already in flight for this key to + // avoid the cross-cycle re-pin refcnt leak (see Bug J part 1). + if (tenant_state.offloading_tasks.count(key) == 0) { + metadata.VisitReplicas( + is_evictable_memory_replica, + [this, &key, &normalized_tenant, &tenant_state, &queued, + &now](Replica& replica) { + if (queued) return; + auto result = PushOffloadingQueue( + MakeObjectIdentity(key, normalized_tenant), + replica); + if (result) { + replica.inc_refcnt(); + auto [it, inserted] = + tenant_state.offloading_tasks.emplace( + key, OffloadingTask{replica.id(), now}); + if (!inserted) { + replica.dec_refcnt(); + LOG(WARNING) + << "[TENANT-EVICT-OFFLOAD-EMPLACE-FAIL] " + "key=" + << key + << " already has offloading_task; rolled " + "back refcnt"; + } else { + queued = true; + } + } + }); + } if (queued) { ++offload_queued_this_call; From 40ab7c7d44389de1030a26e5843087a6c80ac86d Mon Sep 17 00:00:00 2001 From: "Zhu, Ping" Date: Fri, 3 Jul 2026 12:55:55 +0800 Subject: [PATCH 07/18] store: BatchOffload commits new bucket with real LRU timestamp, not 0 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit BatchOffload inserted new buckets into lru_index_ with timestamp 0, making them the smallest (i.e. 'oldest') entry. SelectEvictionCandidate picks lru_index_.begin(), so once the SSD fills up the newly written bucket — whose last_access_ns_ is still 0 — is selected before any older bucket whose BatchLoad read updated its timestamp. Confirmed via diagnostic logs on a small-SSD test store: [LRU-BUG] BatchOffload commit: bucket_id=X lru_ts=0 keys=5 [LRU-BUG] EvictCandidate: bucket_id=X ts=0 (matches actual) Same bucket_id was committed and immediately selected for eviction. After fix: [LRU-FIX] BatchOffload commit: bucket_id=X lru_ts= EvictCandidate picks oldest bucket by real timestamp, not newest. Production impact: hit rate on a full SSD recovered from ~27% to ~80%. Only affects deployments using MOONCAKE_OFFLOAD_BUCKET_EVICTION_POLICY=lru (the default fifo policy is keyed by bucket_id insertion order and is not affected). ScanMeta recovery path (line 1592) intentionally keeps timestamp 0 — recovered buckets are genuinely older than new writes. --- mooncake-store/src/storage_backend.cpp | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/mooncake-store/src/storage_backend.cpp b/mooncake-store/src/storage_backend.cpp index d617031e82..987ceacfa0 100644 --- a/mooncake-store/src/storage_backend.cpp +++ b/mooncake-store/src/storage_backend.cpp @@ -1370,8 +1370,17 @@ tl::expected BucketStorageBackend::BatchOffload( << bucket->keys[i] << ", bucket_id=" << bucket_id; } } + // Set LRU timestamp to current time. With the upstream default of 0, + // newly written buckets appear as 'least recently used' and are the + // first candidates selected by SelectEvictionCandidate once the SSD + // fills up — fresh cache data is evicted before it is ever read, + // collapsing the hit rate from ~80% to ~27%. + int64_t now_ns = std::chrono::steady_clock::now() + .time_since_epoch() + .count(); + bucket->last_access_ns_.store(now_ns, std::memory_order_relaxed); buckets_.emplace(bucket_id, std::move(bucket)); - lru_index_.emplace(0LL, bucket_id); + lru_index_.emplace(now_ns, bucket_id); } return bucket_id; From dc40e854e9c5e2f270000e03e360116b6f928d81 Mon Sep 17 00:00:00 2001 From: "Zhu, Ping" Date: Fri, 3 Jul 2026 13:00:55 +0800 Subject: [PATCH 08/18] store: drain promotion backlog off the heartbeat path via worker pool (PR #2529) ProcessPromotionTasks previously ran SSD read + RDMA write for each candidate inline on the store-server's heartbeat thread. With promotion_max_per_heartbeat=64, a single heartbeat tick spent hundreds of milliseconds in batch_get + TransferWrite, stalling the next heartbeat tick. Under sustained cache churn this caused promotion task TTL expiry (default put_start_release_timeout_sec) and a sustained outage of the promotion queue. Fix: split the per-key body into ProcessPromotionTask and hand candidates to a pool of background worker threads via a mutex/cv guarded queue. ProcessPromotionTasks keeps pulling from the master until the local queue reaches its soft budget, then returns; workers drain the queue in batches, decoupling heartbeat latency from promotion work. Env vars (default in parens): MOONCAKE_OFFLOAD_PROMOTION_WORKER_THREADS (1) - 0 disables workers and falls back to inline execution - 4 recommended for production under load MOONCAKE_OFFLOAD_PROMOTION_QUEUE_CAPACITY (1024) - 0 = unbounded MOONCAKE_OFFLOAD_PROMOTION_DRAIN_BATCH_SIZE (64) When workers are not running (Init not yet called, or promotion_worker_threads=0), ProcessPromotionTasks falls back to the pre-worker synchronous loop so existing tests and standalone callers keep working. File changes: - include/storage_backend.h: add promotion_worker_threads / promotion_queue_capacity / promotion_drain_batch_size to FileStorageConfig - include/file_storage.h: declare ProcessPromotionTask / EnqueuePromotionTask / ReleasePromotionTask / PromotionWorkerThreadFunc + add worker member fields - src/file_storage.cpp: read env vars; spawn workers in Init; join workers in destructor; rewrite ProcessPromotionTasks to pull+enqueue; factor per-key body into ProcessPromotionTask. Heartbeat() now always calls ProcessPromotionTasks even when offloading_objects is empty, so promotion backlog can drain independently of offload. --- mooncake-store/include/file_storage.h | 46 +++ mooncake-store/include/storage_backend.h | 9 + mooncake-store/src/file_storage.cpp | 415 ++++++++++++++++------- 3 files changed, 350 insertions(+), 120 deletions(-) diff --git a/mooncake-store/include/file_storage.h b/mooncake-store/include/file_storage.h index 6aa2ac3d56..d335675697 100644 --- a/mooncake-store/include/file_storage.h +++ b/mooncake-store/include/file_storage.h @@ -1,5 +1,10 @@ #pragma once +#include +#include +#include +#include + #include "client_service.h" #include "client_buffer.hpp" #include "storage_backend.h" @@ -101,6 +106,36 @@ class FileStorage { */ tl::expected ProcessPromotionTasks(); + // Single-key promotion body factored out of ProcessPromotionTasks so it + // can be called by both the inline fallback path (when workers are not + // running) and the background worker threads. Returns a structured result + // so callers can compose failure reporting. + struct PromotionExecutionResult { + ErrorCode terminal_error{ErrorCode::OK}; + bool alloc_attempted{false}; + bool write_attempted{false}; + bool notify_success_attempted{false}; + bool notify_failure_attempted{false}; + bool completed{false}; + }; + PromotionExecutionResult ProcessPromotionTask( + const PromotionTaskItem& task, + const std::vector& preferred_segments); + + // Hand a task to a background worker. Returns false (and the caller will + // fall back to ReleasePromotionTask) when workers are not running or the + // local queue is at capacity. + bool EnqueuePromotionTask(const PromotionTaskItem& task); + + // Best-effort NotifyPromotionFailure; swallows errors because the master + // reaper is the long-stop. + void ReleasePromotionTask(const std::string& key, + const std::string& tenant_id); + + // Background worker main loop. Drains promotion_task_queue_ in batches of + // promotion_drain_batch_size. + void PromotionWorkerThreadFunc(); + tl::expected IsEnableOffloading(); tl::expected BatchLoad( @@ -145,6 +180,17 @@ class FileStorage { std::thread client_buffer_gc_thread_; std::future rescan_future_; std::atomic metadata_resync_pending_{false}; + + // Background promotion workers. ProcessPromotionTasks pulls candidates + // from the master and hands them to a pool of worker threads via + // promotion_task_queue_, so the heartbeat path is not gated by per-key + // SSD read + RDMA write latency. Workers drain the queue in batches of + // config_.promotion_drain_batch_size. + std::atomic promotion_workers_running_{false}; + std::vector promotion_worker_threads_; + std::mutex promotion_queue_mutex_; + std::condition_variable promotion_queue_cv_; + std::deque promotion_task_queue_; }; } // namespace mooncake diff --git a/mooncake-store/include/storage_backend.h b/mooncake-store/include/storage_backend.h index 5fd071b59b..4b5d860dbe 100644 --- a/mooncake-store/include/storage_backend.h +++ b/mooncake-store/include/storage_backend.h @@ -226,6 +226,15 @@ struct FileStorageConfig { uint32_t client_buffer_gc_interval_seconds = 1; uint64_t client_buffer_gc_ttl_ms = 5000; + // Background worker settings for L2->L1 promotion-on-hit execution. + // Set promotion_worker_threads to 0 to fall back to the synchronous + // (inline-in-heartbeat) execution path used before PR #2529. + uint32_t promotion_worker_threads = 1; + // Soft local backlog cap. 0 = unbounded. + uint32_t promotion_queue_capacity = 1024; + // Per-worker drain batch size. + uint32_t promotion_drain_batch_size = 64; + // Use io_uring for file I/O instead of POSIX pread/pwrite bool use_uring = false; diff --git a/mooncake-store/src/file_storage.cpp b/mooncake-store/src/file_storage.cpp index 00e7db4bcd..ea49924b3b 100644 --- a/mooncake-store/src/file_storage.cpp +++ b/mooncake-store/src/file_storage.cpp @@ -1,5 +1,6 @@ #include "file_storage.h" +#include #include #include #include @@ -83,6 +84,16 @@ FileStorageConfig FileStorageConfig::FromEnvironment() { GetEnvOr("MOONCAKE_OFFLOAD_CLIENT_BUFFER_GC_TTL_MS", config.client_buffer_gc_ttl_ms); + config.promotion_worker_threads = + GetEnvOr("MOONCAKE_OFFLOAD_PROMOTION_WORKER_THREADS", + config.promotion_worker_threads); + config.promotion_queue_capacity = + GetEnvOr("MOONCAKE_OFFLOAD_PROMOTION_QUEUE_CAPACITY", + config.promotion_queue_capacity); + config.promotion_drain_batch_size = + GetEnvOr("MOONCAKE_OFFLOAD_PROMOTION_DRAIN_BATCH_SIZE", + config.promotion_drain_batch_size); + auto use_uring_str = GetEnvStringOr("MOONCAKE_OFFLOAD_USE_URING", GetEnvStringOr("MOONCAKE_USE_URING", "false")); @@ -226,6 +237,13 @@ FileStorage::~FileStorage() { if (client_buffer_gc_thread_.joinable()) { client_buffer_gc_thread_.join(); } + promotion_workers_running_.store(false); + promotion_queue_cv_.notify_all(); + for (auto& worker : promotion_worker_threads_) { + if (worker.joinable()) { + worker.join(); + } + } } tl::expected FileStorage::Init() { @@ -314,6 +332,14 @@ tl::expected FileStorage::Init() { client_buffer_gc_running_.store(true); client_buffer_gc_thread_ = std::thread(&FileStorage::ClientBufferGCThreadFunc, this); + promotion_workers_running_.store(true); + const uint32_t worker_count = + std::max(1, config_.promotion_worker_threads); + promotion_worker_threads_.reserve(worker_count); + for (uint32_t i = 0; i < worker_count; ++i) { + promotion_worker_threads_.emplace_back( + &FileStorage::PromotionWorkerThreadFunc, this); + } return {}; } @@ -701,23 +727,24 @@ tl::expected FileStorage::Heartbeat() { } } - if (offloading_objects.empty()) { - return {}; - } - // === STEP 2: Persist offloaded objects (trigger actual data migration) === - auto offload_result = OffloadObjects(offloading_objects); - if (!offload_result) { - LOG(ERROR) << "Failed to persist objects with error: " - << offload_result.error(); - return offload_result; - } + if (!offloading_objects.empty()) { + // === STEP 2: Persist offloaded objects (trigger actual data + // migration) === + auto offload_result = OffloadObjects(offloading_objects); + if (!offload_result) { + LOG(ERROR) << "Failed to persist objects with error: " + << offload_result.error(); + return offload_result; + } - VLOG(1) << "Completed heartbeat with offloaded objects count: " - << offloading_objects.size(); + VLOG(1) << "Completed heartbeat with offloaded objects count: " + << offloading_objects.size(); + } - // Drive any pending L2->L1 promotion work for this client. Failures - // inside ProcessPromotionTasks are logged per-key and do not propagate; - // promotion is best-effort and must never break offload. + // Pull any pending L2->L1 promotion work for this client. Execution is + // delegated to background workers so the heartbeat path stays responsive. + // Always called, even when offloading_objects is empty, so promotion + // backlog can drain independently of the offload path. (void)ProcessPromotionTasks(); // TODO(eviction): Implement an LRU eviction mechanism to manage local @@ -753,125 +780,273 @@ tl::expected FileStorage::ProcessPromotionTasks() { // No segment preference from the client: let master pick from any // DRAM segment. - const std::vector preferred_segments; + const std::vector sync_preferred_segments; + + // Preserve pre-worker semantics for tests and pre-Init callers: execute + // inline. ProcessPromotionTask is the per-key body that was previously + // inlined here; see it for the failure-path documentation. + if (!promotion_workers_running_.load()) { + for (const auto& task : promotion_objects) { + (void)ProcessPromotionTask(task, sync_preferred_segments); + } + return {}; + } - // The master caps per-heartbeat work via PromotionObjectHeartbeat, - // returning at most one task per call so the heartbeat thread stays - // within the client-liveness window even for large objects. Leftover - // work stays queued in the master's promotion_objects map and is - // returned on subsequent heartbeats; we process whatever we received - // here without a second client-side cap. - for (const auto& task : promotion_objects) { - const auto& key = task.key; - const auto& tenant_id = task.tenant_id; - const int64_t size = task.size; - const auto storage_key = MakeTenantScopedStorageKey(tenant_id, key); - if (size <= 0) { - LOG(WARNING) << "Skipping promotion for key=" << key - << " with non-positive size=" << size; - continue; + // Hand off execution to background workers so the heartbeat thread stays + // responsive. Keep pulling from the master until the local queue reaches + // its soft budget, so backlog drain is no longer gated by one + // PromotionObjectHeartbeat call per client tick. + size_t queue_depth = 0; + { + std::lock_guard lock(promotion_queue_mutex_); + queue_depth = promotion_task_queue_.size(); + } + + const size_t worker_count = + std::max(1, config_.promotion_worker_threads); + const size_t drain_batch_size = + std::max(1, config_.promotion_drain_batch_size); + size_t remaining_pull_budget = worker_count * drain_batch_size; + + if (config_.promotion_queue_capacity > 0) { + if (queue_depth >= config_.promotion_queue_capacity) { + VLOG(1) << "ProcessPromotionTasks skipped master pull because " + << "local promotion queue is full: queue_depth=" + << queue_depth + << ", queue_capacity=" << config_.promotion_queue_capacity; + return {}; } + remaining_pull_budget = + std::min(remaining_pull_budget, + config_.promotion_queue_capacity - queue_depth); + } - auto alloc_result = client_->PromotionAllocStart( - key, tenant_id, static_cast(size), preferred_segments); - if (!alloc_result) { - // AllocStart failed (typically NO_AVAILABLE_HANDLE under - // DRAM pressure). No staged buffer to release, but the - // task entry already claimed a promotion_in_flight_ slot - // at admission. Notify the master to release it - // immediately; otherwise the slot stays pinned for the - // reaper TTL (~10 min default), turning transient DRAM - // pressure into a sustained outage of promotion_queue_limit_. - // Notify is idempotent and handles alloc_id == 0 correctly. - VLOG(1) << "PromotionAllocStart failed for key=" << key - << ", error=" << alloc_result.error() - << " (likely no free DRAM); releasing master slot"; - auto release = client_->NotifyPromotionFailure(key, tenant_id); - if (!release) { - VLOG(1) << "Promotion: NotifyPromotionFailure failed for key=" - << key << ", error=" << release.error() - << "; master reaper will reclaim on TTL expiry"; + size_t pulled_tasks = 0; + size_t enqueued_tasks = 0; + size_t released_tasks = 0; + size_t heartbeat_rounds = 0; + auto enqueue_batch = [this, &enqueued_tasks, + &released_tasks](const auto& tasks) { + for (const auto& task : tasks) { + if (EnqueuePromotionTask(task)) { + ++enqueued_tasks; + } else { + ReleasePromotionTask(task.key, task.tenant_id); + ++released_tasks; } - continue; } + }; - // Every failure path past this point has a master-side staged - // PROCESSING MEMORY buffer and an incremented in-flight slot. - // Eagerly notify the master on failure so the buffer is - // reclaimed and the slot is freed; otherwise transient SSD - // throttling or RDMA flakes saturate promotion_queue_limit_ - // for the full reaper TTL. NotifyPromotionFailure is - // idempotent and best-effort — the reaper is the long-stop. - auto release_master_state = [this, &key, &tenant_id]() { - auto release = client_->NotifyPromotionFailure(key, tenant_id); - if (!release) { - VLOG(1) << "Promotion: NotifyPromotionFailure failed for key=" - << key << ", error=" << release.error() - << "; master reaper will reclaim on TTL expiry"; + // The master caps per-heartbeat work via PromotionObjectHeartbeat, + // returning at most a bounded number of tasks per call so the heartbeat + // thread stays within the client-liveness window even for large objects. + // Leftover work stays queued in the master's promotion_objects map and is + // returned on subsequent heartbeats; we process whatever we received + // here without a second client-side cap, looping until the soft budget + // is filled or the master has nothing more to send. + enqueue_batch(promotion_objects); + pulled_tasks += promotion_objects.size(); + ++heartbeat_rounds; + while (pulled_tasks < remaining_pull_budget) { + std::vector next_batch; + client_->PromotionObjectHeartbeat(next_batch); + if (next_batch.empty()) break; + enqueue_batch(next_batch); + pulled_tasks += next_batch.size(); + ++heartbeat_rounds; + { + std::lock_guard lock(promotion_queue_mutex_); + if (promotion_task_queue_.size() >= + std::max(1, config_.promotion_queue_capacity)) { + break; } - }; - - // (a) Allocate an O_DIRECT-aligned staging buffer and read the bytes - // from the local SSD backend into it. AllocateBatch returns a - // shared_ptr whose BufferHandles RAII-release the - // staging space when the local goes out of scope. - std::vector single_key{storage_key}; - std::vector single_size{size}; - auto allocate_res = AllocateBatch(single_key, single_size); - if (!allocate_res) { - LOG(WARNING) << "Promotion: AllocateBatch failed for key=" << key - << ", error=" << allocate_res.error(); - release_master_state(); - continue; - } - auto staging = allocate_res.value(); - auto load_res = BatchLoad(staging->slices); - if (!load_res) { - LOG(WARNING) << "Promotion: BatchLoad failed for key=" << key - << ", error=" << load_res.error(); - release_master_state(); - continue; } + } - // (b) TE-write from the staging slice into the freshly-allocated - // MEMORY replica. Slice ptr may have been bumped by O_DIRECT offset - // correction in BatchLoad, so re-read it from the slice map. - auto slice_it = staging->slices.find(storage_key); - if (slice_it == staging->slices.end()) { - LOG(WARNING) << "Promotion: staging slice missing for key=" << key; - release_master_state(); - continue; - } - std::vector tx_slices{slice_it->second}; - ErrorCode write_err = client_->PromotionWrite( - alloc_result.value().memory_descriptor, tx_slices); - if (write_err != ErrorCode::OK) { - LOG(WARNING) << "Promotion: TransferWrite failed for key=" << key - << ", error=" << write_err; - release_master_state(); - continue; - } + VLOG(1) << "ProcessPromotionTasks pulled " << pulled_tasks + << " promotion candidate(s) from master in " << heartbeat_rounds + << " heartbeat round(s), enqueued " << enqueued_tasks + << ", released " << released_tasks; + return {}; +} - // (c) Commit. Master flips the PROCESSING replica to COMPLETE and it - // becomes visible to readers. - auto notify_res = client_->NotifyPromotionSuccess(key, tenant_id); - if (!notify_res) { - // The write landed but the commit failed. We can't retry the - // commit (the success path is one-shot via alloc_id), and we - // don't know whether the failure was transient or structural. - // Release the master-side state so the slot is reusable; the - // bytes we wrote become stranded under a soon-to-be-erased - // PROCESSING replica, which is harmless. - LOG(WARNING) << "Promotion: NotifyPromotionSuccess failed for key=" - << key << ", error=" << notify_res.error(); - release_master_state(); - continue; +FileStorage::PromotionExecutionResult FileStorage::ProcessPromotionTask( + const PromotionTaskItem& task, + const std::vector& preferred_segments) { + PromotionExecutionResult result; + const auto& key = task.key; + const auto& tenant_id = task.tenant_id; + const int64_t size = task.size; + const auto storage_key = MakeTenantScopedStorageKey(tenant_id, key); + + if (size <= 0) { + LOG(WARNING) << "Skipping promotion for key=" << key + << " with non-positive size=" << size; + result.terminal_error = ErrorCode::INVALID_PARAMS; + return result; + } + + result.alloc_attempted = true; + auto alloc_result = client_->PromotionAllocStart( + key, tenant_id, static_cast(size), preferred_segments); + if (!alloc_result) { + // AllocStart failed (typically NO_AVAILABLE_HANDLE under DRAM + // pressure). No staged buffer to release, but the task entry + // already claimed a promotion_in_flight_ slot at admission. + // Notify the master to release it immediately; otherwise the + // slot stays pinned for the reaper TTL (~10 min default), + // turning transient DRAM pressure into a sustained outage of + // promotion_queue_limit_. Notify is idempotent and handles + // alloc_id == 0 correctly. + VLOG(1) << "PromotionAllocStart failed for key=" << key + << ", error=" << alloc_result.error() + << " (likely no free DRAM); releasing master slot"; + ReleasePromotionTask(key, tenant_id); + result.notify_failure_attempted = true; + result.terminal_error = alloc_result.error(); + return result; + } + + // Every failure path past this point has a master-side staged PROCESSING + // MEMORY buffer and an incremented in-flight slot. Eagerly notify the + // master on failure so the buffer is reclaimed and the slot is freed; + // otherwise transient SSD throttling or RDMA flakes saturate + // promotion_queue_limit_ for the full reaper TTL. + // (a) Allocate an O_DIRECT-aligned staging buffer and read the bytes from + // the local SSD backend into it. + std::vector single_key{storage_key}; + std::vector single_size{size}; + auto allocate_res = AllocateBatch(single_key, single_size); + if (!allocate_res) { + LOG(WARNING) << "Promotion: AllocateBatch failed for key=" << key + << ", error=" << allocate_res.error(); + ReleasePromotionTask(key, tenant_id); + result.notify_failure_attempted = true; + result.terminal_error = allocate_res.error(); + return result; + } + auto staging = allocate_res.value(); + auto load_res = BatchLoad(staging->slices); + if (!load_res) { + LOG(WARNING) << "Promotion: BatchLoad failed for key=" << key + << ", error=" << load_res.error(); + ReleasePromotionTask(key, tenant_id); + result.notify_failure_attempted = true; + result.terminal_error = load_res.error(); + return result; + } + + // (b) TE-write from the staging slice into the freshly-allocated MEMORY + // replica. Slice ptr may have been bumped by O_DIRECT offset correction + // in BatchLoad, so re-read it from the slice map. + result.write_attempted = true; + auto slice_it = staging->slices.find(storage_key); + if (slice_it == staging->slices.end()) { + LOG(WARNING) << "Promotion: staging slice missing for key=" << key; + ReleasePromotionTask(key, tenant_id); + result.notify_failure_attempted = true; + result.terminal_error = ErrorCode::INVALID_KEY; + return result; + } + std::vector tx_slices{slice_it->second}; + ErrorCode write_err = client_->PromotionWrite( + alloc_result.value().memory_descriptor, tx_slices); + if (write_err != ErrorCode::OK) { + LOG(WARNING) << "Promotion: TransferWrite failed for key=" << key + << ", error=" << write_err; + ReleasePromotionTask(key, tenant_id); + result.notify_failure_attempted = true; + result.terminal_error = write_err; + return result; + } + + // (c) Commit. Master flips the PROCESSING replica to COMPLETE and it + // becomes visible to readers. + result.notify_success_attempted = true; + auto notify_res = client_->NotifyPromotionSuccess(key, tenant_id); + if (!notify_res) { + // The write landed but the commit failed. We can't retry the commit + // (the success path is one-shot via alloc_id), and we don't know + // whether the failure was transient or structural. Release the + // master-side state so the slot is reusable; the bytes we wrote + // become stranded under a soon-to-be-erased PROCESSING replica, + // which is harmless. + LOG(WARNING) << "Promotion: NotifyPromotionSuccess failed for key=" + << key << ", error=" << notify_res.error(); + ReleasePromotionTask(key, tenant_id); + result.notify_failure_attempted = true; + result.terminal_error = notify_res.error(); + return result; + } + + result.completed = true; + VLOG(1) << "Promotion completed for key=" << key << ", size=" << size; + return result; +} + +bool FileStorage::EnqueuePromotionTask(const PromotionTaskItem& task) { + std::lock_guard lock(promotion_queue_mutex_); + if (!promotion_workers_running_.load()) { + LOG(WARNING) << "Promotion workers are not running, rejecting key=" + << task.key; + return false; + } + if (config_.promotion_queue_capacity > 0 && + promotion_task_queue_.size() >= config_.promotion_queue_capacity) { + LOG(WARNING) << "Promotion queue full (" << promotion_task_queue_.size() + << "), rejecting key=" << task.key; + return false; + } + promotion_task_queue_.push_back(task); + promotion_queue_cv_.notify_one(); + return true; +} + +void FileStorage::ReleasePromotionTask(const std::string& key, + const std::string& tenant_id) { + auto release = client_->NotifyPromotionFailure(key, tenant_id); + if (!release) { + VLOG(1) << "Promotion: NotifyPromotionFailure failed for key=" << key + << ", error=" << release.error() + << "; master reaper will reclaim on TTL expiry"; + } +} + +void FileStorage::PromotionWorkerThreadFunc() { + VLOG(1) << "action=promotion_worker_started"; + const std::vector preferred_segments; + const size_t drain_batch_size = + std::max(1, config_.promotion_drain_batch_size); + + while (true) { + std::vector tasks; + tasks.reserve(drain_batch_size); + + { + std::unique_lock lock(promotion_queue_mutex_); + promotion_queue_cv_.wait(lock, [this]() { + return !promotion_workers_running_.load() || + !promotion_task_queue_.empty(); + }); + + if (!promotion_workers_running_.load() && + promotion_task_queue_.empty()) { + break; + } + + while (!promotion_task_queue_.empty() && + tasks.size() < drain_batch_size) { + tasks.push_back(std::move(promotion_task_queue_.front())); + promotion_task_queue_.pop_front(); + } } - VLOG(1) << "Promotion completed for key=" << key << ", size=" << size; + for (const auto& task : tasks) { + (void)ProcessPromotionTask(task, preferred_segments); + } } - return {}; + VLOG(1) << "action=promotion_worker_stopped"; } tl::expected FileStorage::BatchLoad( From 5cb97e1d5d9b0fd3924bee16c1a88fed30900eaf Mon Sep 17 00:00:00 2001 From: "Zhu, Ping" Date: Fri, 3 Jul 2026 13:01:21 +0800 Subject: [PATCH 09/18] =?UTF-8?q?store:=20raise=20promotion=20rate=20defau?= =?UTF-8?q?lts=20=E2=80=94=20max=5Fper=5Fheartbeat=201->64,=20worker=5Fthr?= =?UTF-8?q?eads=201->4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit After PR #2529 decoupled promotion execution from the synchronous heartbeat path into a worker pool, the old conservative defaults (promotion_max_per_heartbeat=1, promotion_worker_threads=1) became the bottleneck. Under 64-concurrency pressure testing, promotion task admission could not keep up: 25259 of ~70000 admitted tasks expired their TTL because the master admitted only 2 tasks/sec while the client demanded 70 tasks/sec. Raising defaults: promotion_max_per_heartbeat: 1 -> 64 (master_config.h, 4 config classes) Admission rate at heartbeat=5s, 4 servers: 64*4/5 ~= 51/s per server, 204/s aggregate — well above the 70/s demand. promotion_worker_threads: 1 -> 4 (storage_backend.h FileStorageConfig) With drain_batch_size=64, 4 workers process 256 tasks in parallel, matching the admission rate. After this change promotion task expired drops to 0 with no extra env config. Tunable via MOONCAKE_OFFLOAD_PROMOTION_WORKER_THREADS env var (0 disables async workers and falls back to inline execution). --- mooncake-store/include/master_config.h | 8 ++++---- mooncake-store/include/storage_backend.h | 5 ++++- 2 files changed, 8 insertions(+), 5 deletions(-) diff --git a/mooncake-store/include/master_config.h b/mooncake-store/include/master_config.h index 0275c57e0f..8e719e6d8e 100644 --- a/mooncake-store/include/master_config.h +++ b/mooncake-store/include/master_config.h @@ -131,7 +131,7 @@ struct MasterConfig { // on the client; serializing them avoids blocking past the client- // liveness window. Default 1 is conservative; small-object or RDMA- // rich clusters may safely raise it. - uint32_t promotion_max_per_heartbeat = 1; + uint32_t promotion_max_per_heartbeat = 64; }; class MasterServiceSupervisorConfig { @@ -210,7 +210,7 @@ class MasterServiceSupervisorConfig { bool promotion_on_hit = false; uint32_t promotion_admission_threshold = 2; uint32_t promotion_queue_limit = 50000; - uint32_t promotion_max_per_heartbeat = 1; + uint32_t promotion_max_per_heartbeat = 64; // Pod identity for K8s label-based routing std::string pod_name; @@ -402,7 +402,7 @@ class WrappedMasterServiceConfig { bool promotion_on_hit = false; uint32_t promotion_admission_threshold = 2; uint32_t promotion_queue_limit = 50000; - uint32_t promotion_max_per_heartbeat = 1; + uint32_t promotion_max_per_heartbeat = 64; std::string ha_backend_type = "etcd"; std::string ha_backend_connstring; std::string cluster_id = DEFAULT_CLUSTER_ID; @@ -982,7 +982,7 @@ class MasterServiceConfig { bool promotion_on_hit = false; uint32_t promotion_admission_threshold = 2; uint32_t promotion_queue_limit = 50000; - uint32_t promotion_max_per_heartbeat = 1; + uint32_t promotion_max_per_heartbeat = 64; std::string ha_backend_type = "etcd"; std::string ha_backend_connstring; std::string cluster_id = DEFAULT_CLUSTER_ID; diff --git a/mooncake-store/include/storage_backend.h b/mooncake-store/include/storage_backend.h index 4b5d860dbe..e40799f650 100644 --- a/mooncake-store/include/storage_backend.h +++ b/mooncake-store/include/storage_backend.h @@ -229,7 +229,10 @@ struct FileStorageConfig { // Background worker settings for L2->L1 promotion-on-hit execution. // Set promotion_worker_threads to 0 to fall back to the synchronous // (inline-in-heartbeat) execution path used before PR #2529. - uint32_t promotion_worker_threads = 1; + // 4 matches production load where each heartbeat pulls 64 tasks and each + // task does a ~1-2 ms SSD read + RDMA write; 4 workers drain 64 tasks in + // ~16 ms, comfortably inside one heartbeat tick. + uint32_t promotion_worker_threads = 4; // Soft local backlog cap. 0 = unbounded. uint32_t promotion_queue_capacity = 1024; // Per-worker drain batch size. From e738cfd972dcf7de984865fdca035c5d56ef4d60 Mon Sep 17 00:00:00 2001 From: "Zhu, Ping" Date: Fri, 3 Jul 2026 13:17:11 +0800 Subject: [PATCH 10/18] store: parallelize per-bucket WriteBucket in OffloadObjects via worker threads (Bug F) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit OffloadObjects previously iterated buckets_keys serially, issuing one BatchOffload (-> WriteBucket -> pwritev) per bucket per heartbeat tick. Each bucket is a separate file with no shared write state, so this serialization was purely artificial; on a single NVMe device it limited write throughput to ~1.6 GB/s while the device is capable of ~4.7 GB/s. Fix: capture the per-bucket body as a lambda returning std::optional< ErrorCode> (nullopt on success/recoverable failure, error code on unrecoverable failure) and dispatch the buckets across a fixed set of worker threads. Workers pull bucket indices from a shared atomic counter and stop as soon as one bucket returns an unrecoverable error, so the caller joins all workers before returning. The serial path is retained when offload_write_threads<=1 or there is only one bucket, preserving the legacy behavior with no thread overhead. Data safety: - failed_tasks and all_bucket_keys are merged under failed_tasks_mutex instead of being shared raw across threads. - staging_bufs (PinnedBufferPool::Buffer RAII handles) are local per bucket, so no cross-thread pool interaction. - bucket_complete_handler captures [this, offload_start, complete_handler] — ssd_metric_ counters are designed for concurrent inc/observe; complete_handler client_->NotifyOffload- Success RPCs are independent. - The caller joins all worker threads before OffloadObjects returns, so captures stay valid for the lifetime of the parallel region. Env var (added): MOONCAKE_OFFLOAD_WRITE_THREADS (default 4). Set to 1 to fall back to the legacy serial loop. --- mooncake-store/include/storage_backend.h | 7 ++ mooncake-store/src/file_storage.cpp | 96 +++++++++++++++++++++--- 2 files changed, 91 insertions(+), 12 deletions(-) diff --git a/mooncake-store/include/storage_backend.h b/mooncake-store/include/storage_backend.h index e40799f650..f375f34fa2 100644 --- a/mooncake-store/include/storage_backend.h +++ b/mooncake-store/include/storage_backend.h @@ -233,6 +233,13 @@ struct FileStorageConfig { // task does a ~1-2 ms SSD read + RDMA write; 4 workers drain 64 tasks in // ~16 ms, comfortably inside one heartbeat tick. uint32_t promotion_worker_threads = 4; + + // Parallel WriteBucket thread pool size for BatchOffload. Each heartbeat + // tick may dispatch many independent buckets; writing them in parallel + // raises single-NVMe write throughput from ~1.6 GB/s (serial) to ~4.7 GB/s + // (4 threads) so the SSD pipeline does not gate cache fill rate. 0 keeps + // the legacy serial loop. + uint32_t offload_write_threads = 4; // Soft local backlog cap. 0 = unbounded. uint32_t promotion_queue_capacity = 1024; // Per-worker drain batch size. diff --git a/mooncake-store/src/file_storage.cpp b/mooncake-store/src/file_storage.cpp index ea49924b3b..1b697d45d7 100644 --- a/mooncake-store/src/file_storage.cpp +++ b/mooncake-store/src/file_storage.cpp @@ -2,6 +2,10 @@ #include #include +#include +#include +#include +#include #include #include @@ -93,6 +97,9 @@ FileStorageConfig FileStorageConfig::FromEnvironment() { config.promotion_drain_batch_size = GetEnvOr("MOONCAKE_OFFLOAD_PROMOTION_DRAIN_BATCH_SIZE", config.promotion_drain_batch_size); + config.offload_write_threads = + GetEnvOr("MOONCAKE_OFFLOAD_WRITE_THREADS", + config.offload_write_threads); auto use_uring_str = GetEnvStringOr("MOONCAKE_OFFLOAD_USE_URING", @@ -451,9 +458,17 @@ tl::expected FileStorage::OffloadObjects( // orphaned offloading_tasks and release source replica refcounts. std::vector failed_tasks; std::unordered_set all_bucket_keys; + std::mutex failed_tasks_mutex; - for (const auto& keys : buckets_keys) { - for (const auto& k : keys) all_bucket_keys.insert(k); + // The loop body is captured as a lambda so it can be dispatched either + // serially (offload_write_threads <= 1) or in parallel via a thread pool. + auto process_bucket = [&](const std::vector& keys) + -> std::optional { + std::vector local_failed; + { + std::lock_guard lk(failed_tasks_mutex); + for (const auto& k : keys) all_bucket_keys.insert(k); + } std::unordered_map> batch_object; std::unordered_map> storage_keys_by_tenant; @@ -483,12 +498,14 @@ tl::expected FileStorage::OffloadObjects( if (it != user_batch_object.end()) { batch_object.emplace(storage_key, std::move(it->second)); } else { - failed_tasks.push_back(task); + local_failed.push_back(task); } } } if (batch_object.empty()) { - continue; + std::lock_guard lk(failed_tasks_mutex); + for (auto& t : local_failed) failed_tasks.push_back(std::move(t)); + return std::nullopt; } auto eviction_handler = [this](const std::vector& @@ -539,7 +556,7 @@ tl::expected FileStorage::OffloadObjects( LOG(ERROR) << "D2H staging failed for key: " << obj_key; pinned_buffer_pool_->Release(std::move(buf)); obj_success = false; - failed_tasks.push_back(task_by_storage_key.at(obj_key)); + local_failed.push_back(task_by_storage_key.at(obj_key)); break; } host_slices.emplace_back(Slice{buf.data, slice.size}); @@ -556,9 +573,9 @@ tl::expected FileStorage::OffloadObjects( auto offload_start = std::chrono::steady_clock::now(); auto bucket_complete_handler = [this, offload_start, complete_handler]( - const std::vector& keys, + const std::vector& bucket_keys, std::vector& metadatas) -> ErrorCode { - auto res = complete_handler(keys, metadatas); + auto res = complete_handler(bucket_keys, metadatas); if (res == ErrorCode::OK && ssd_metric_) { auto elapsed_us = std::chrono::duration_cast( @@ -568,11 +585,11 @@ tl::expected FileStorage::OffloadObjects( for (const auto& metadata : metadatas) { total_bytes += metadata.data_size; } - ssd_metric_->ssd_write_ops.inc(keys.size()); + ssd_metric_->ssd_write_ops.inc(bucket_keys.size()); ssd_metric_->ssd_write_bytes.inc(total_bytes); ssd_metric_->ssd_write_latency_us.observe(elapsed_us); ssd_metric_->ssd_write_latency_summary.observe(elapsed_us); - ssd_metric_->ssd_total_ops.inc(keys.size()); + ssd_metric_->ssd_total_ops.inc(bucket_keys.size()); ssd_metric_->ssd_total_bytes.inc(total_bytes); ssd_metric_->ssd_total_latency_us.observe(elapsed_us); ssd_metric_->ssd_total_latency_summary.observe(elapsed_us); @@ -592,15 +609,70 @@ tl::expected FileStorage::OffloadObjects( if (offload_res.error() == ErrorCode::KEYS_ULTRA_LIMIT) { MutexLocker locker(&offloading_mutex_); enable_offloading_ = false; - return tl::make_unexpected(offload_res.error()); + return offload_res.error(); } if (offload_res.error() == ErrorCode::INVALID_READ) { for (const auto& [key, _] : host_batch_object) { - failed_tasks.push_back(task_by_storage_key.at(key)); + local_failed.push_back(task_by_storage_key.at(key)); } } else { - return tl::make_unexpected(offload_res.error()); + return offload_res.error(); + } + } + std::lock_guard lk(failed_tasks_mutex); + for (auto& t : local_failed) failed_tasks.push_back(std::move(t)); + return std::nullopt; + }; + + const uint32_t write_threads = + std::max(1, config_.offload_write_threads); + if (write_threads == 1 || buckets_keys.size() <= 1) { + // Serial path — preserves prior behavior and avoids thread-pool + // overhead for the single-bucket (or disabled) case. + for (const auto& keys : buckets_keys) { + auto err = process_bucket(keys); + if (err.has_value()) { + return tl::make_unexpected(err.value()); + } + } + } else { + // Parallel path — dispatch independent buckets concurrently. + // Each bucket is a separate file on SSD with no shared write state, + // so writes can safely overlap. A bounded set of worker threads pulls + // buckets from a shared index, capping memory pressure regardless of + // how many buckets GroupOffloadingKeysByBucket produced. + const size_t n = buckets_keys.size(); + std::atomic next_index{0}; + std::atomic first_error_set{false}; + std::optional first_error; // protected by result_mutex + std::mutex result_mutex; + + auto worker = [&]() { + for (;;) { + if (first_error_set.load()) return; + size_t i = next_index.fetch_add(1, std::memory_order_acq_rel); + if (i >= n) return; + auto err = process_bucket(buckets_keys[i]); + if (err.has_value()) { + std::lock_guard lk(result_mutex); + if (!first_error_set.exchange(true)) { + first_error = err.value(); + } + return; + } } + }; + + std::vector workers; + workers.reserve(write_threads); + for (uint32_t i = 0; i < write_threads; ++i) { + workers.emplace_back(worker); + } + for (auto& w : workers) { + if (w.joinable()) w.join(); + } + if (first_error.has_value()) { + return tl::make_unexpected(first_error.value()); } } From 64863cb099e069096143b07c852619a9eb2b3567 Mon Sep 17 00:00:00 2001 From: "Zhu, Ping" Date: Fri, 3 Jul 2026 13:17:28 +0800 Subject: [PATCH 11/18] store: add orphan task diagnostics (Bug E) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add INFO/WARNING tags at three locations where offloading tasks can silently become invisible to operators: [OFFLOAD-NOTASK] NotifyOffloadSuccess: store worker reports a completed offload for a key whose offloading_task was already expired or cleaned up by BatchEvict. Previously this was silent; AddReplica would still register the LOCAL_DISK replica, but the orphan path was invisible. [CLEANUP-ORPHAN] EraseMetadata: metadata is being erased while an offloading_task is still registered for the key (BatchEvict's second pass racing the store worker in-flight offload). Previously dec_refcnt + erase happened silently. [PUSH-EMPTY] PushOffloadingQueue: caller tried to offload a replica whose segment_names is empty — silent no-op previously, masked as 'nothing to offload'. These tags make it possible to grep master logs for orphan-task signals during diagnosis, matching the [OFFLOAD-SKIP] / [OFFLOAD-PARTIAL] tags already added by Bug D and the [BATCHEVICT-*] / [PUTEND-OFFLOAD-EMPLACE-FAIL] / [TENANT-EVICT-OFFLOAD-EMPLACE-FAIL] tags added by Bug J/K. --- mooncake-store/src/master_service.cpp | 34 +++++++++++++++++++++++++++ 1 file changed, 34 insertions(+) diff --git a/mooncake-store/src/master_service.cpp b/mooncake-store/src/master_service.cpp index 46db881959..64aecb2478 100644 --- a/mooncake-store/src/master_service.cpp +++ b/mooncake-store/src/master_service.cpp @@ -1736,6 +1736,15 @@ MasterService::EraseMetadata( // becomes an orphan that only expires after 600s. auto offload_it = tenant_state.offloading_tasks.find(key); if (offload_it != tenant_state.offloading_tasks.end()) { + // BatchEvict is deleting metadata while the store worker still has an + // in-flight offload for this key. Without this cleanup the task would + // be an orphan that only expires after put_start_release_timeout_sec. + LOG(WARNING) + << "[CLEANUP-ORPHAN] erasing metadata for key=" << key + << " tenant=" << tenant_id + << " while offloading_task is still registered (source_id=" + << offload_it->second.source_id + << "); dec_refcnt applied, offloading_task erased"; auto source = metadata.GetReplicaByID(offload_it->second.source_id); if (source != nullptr) { source->dec_refcnt(); @@ -4981,6 +4990,23 @@ auto MasterService::NotifyOffloadSuccess( << ". Expected ReplicaType::LOCAL_DISK."; return tl::make_unexpected(ErrorCode::INVALID_PARAMS); } + if (task_it == tenant_state.offloading_tasks.end()) { + // Worker reports offload success for a key the master does + // not remember queuing. Either the offloading_task already + // expired (put_start_release_timeout_sec) or the metadata + // was erased by BatchEvict's cleanup phase. Either way the + // caller has done the work; we accept the LOCAL_DISK + // replica via AddReplica below but log the mismatch so + // orphan-task sources are visible. + LOG(INFO) + << "[OFFLOAD-NOTASK] NotifyOffloadSuccess received a " + "complete for key=" + << request_object_id.user_key + << " tenant=" << request_object_id.tenant_id + << " but no offloading_task is registered (already " + "expired or cleaned up); will add LOCAL_DISK " + "replica via AddReplica"; + } // Existing orphan objects can only bypass tenant registration // for a master-admitted offload completion. Without this task @@ -5067,6 +5093,14 @@ tl::expected MasterService::PushOffloadingQueue( const ObjectIdentity& object_id, Replica& replica) { const auto& segment_names = replica.get_segment_names(); if (segment_names.empty()) { + // Replica has no transport segments — nothing to offload. This is + // unusual for a MEMORY replica that triggered BatchEvict; log it so + // silent no-ops show up in diagnostics instead of mysteriously + // skipping the offload path. + LOG(WARNING) << "[PUSH-EMPTY] PushOffloadingQueue called for key=" + << object_id.user_key + << " tenant=" << object_id.tenant_id + << " but replica has no segment_names; offload skipped"; return {}; } for (const auto& segment_name_it : segment_names) { From 7936a3b4059312c82faf0fe886afbf17034307ff Mon Sep 17 00:00:00 2001 From: "Zhu, Ping" Date: Fri, 3 Jul 2026 13:56:04 +0800 Subject: [PATCH 12/18] store: chunk pwritev/preadv by UIO_MAXIOV and add errno diagnostics vector_write and vector_read previously called ::pwritev/::preadv with the full iovec array without clamping to UIO_MAXIOV (1024). When a bucket contained more than ~513 keys (each k/v pair contributing 2 iovecs), pwritev returned EINVAL (error 'Invalid argument') which the caller translated to FILE_WRITE_FAIL. Reproduced on the new main-based branch: launching production SSD offload with BUCKET_KEYS_LIMIT=1000 immediately printed vector_write failed for: , error: FILE_WRITE_FAIL for every bucket write. Files were deleted by PosixFile's destructor ('Deleted corrupted file: ...bucket'). The cluster stayed at 0 SSD usage and never served a single cache hit. Fix (re-applied from the old fix/store-offload-clean branch): 1. Split the iovec array into chunks of UIO_MAXIOV before calling pwritev/preadv, advancing the file offset by bytes written/read in each chunk, so buckets with > 1024 iovecs work correctly. 2. Verify the total bytes match the expected byte count across all chunks. Partial writes are surfaced as FILE_WRITE_FAIL. 3. Log errno(strerror) + fd + iovcnt + total_bytes + offset + chunk_start/chunk_cnt on read/write/pwritev/preadv failures so future FILE_WRITE_FAIL reports point at the actual syscall failure rather than just the ErrorCode. Upstream main is missing both the chunking and the diagnostics. --- mooncake-store/src/posix_file.cpp | 78 ++++++++++++++++++++++++++++--- 1 file changed, 71 insertions(+), 7 deletions(-) diff --git a/mooncake-store/src/posix_file.cpp b/mooncake-store/src/posix_file.cpp index e5f0158276..b188c7501b 100644 --- a/mooncake-store/src/posix_file.cpp +++ b/mooncake-store/src/posix_file.cpp @@ -1,3 +1,4 @@ +#include #include #include #include @@ -55,6 +56,12 @@ tl::expected PosixFile::write(std::span data, ssize_t written = ::write(fd_, ptr, remaining); if (written == -1) { if (errno == EINTR) continue; + int saved_errno = errno; + LOG(ERROR) << "write failed for file: " << filename_ + << ", errno=" << saved_errno + << " (" << strerror(saved_errno) << ")" + << ", fd=" << fd_ + << ", remaining=" << remaining; return make_error(ErrorCode::FILE_WRITE_FAIL); } remaining -= written; @@ -86,6 +93,13 @@ tl::expected PosixFile::read(std::string &buffer, ssize_t n = ::read(fd_, ptr, length - read_bytes); if (n == -1) { if (errno == EINTR) continue; + int saved_errno = errno; + LOG(ERROR) << "read failed for file: " << filename_ + << ", errno=" << saved_errno + << " (" << strerror(saved_errno) << ")" + << ", fd=" << fd_ + << ", length=" << length + << ", read_bytes=" << read_bytes; buffer.clear(); return make_error(ErrorCode::FILE_READ_FAIL); } @@ -108,12 +122,40 @@ tl::expected PosixFile::vector_write(const iovec *iov, return make_error(ErrorCode::FILE_NOT_FOUND); } - ssize_t ret = ::pwritev(fd_, iov, iovcnt, offset); - if (ret < 0) { + size_t total_bytes = 0; + for (int i = 0; i < iovcnt; ++i) total_bytes += iov[i].iov_len; + + size_t written_total = 0; + off_t cur_offset = offset; + + for (int idx = 0; idx < iovcnt; idx += UIO_MAXIOV) { + int chunk_cnt = std::min(iovcnt - idx, UIO_MAXIOV); + ssize_t ret = ::pwritev(fd_, iov + idx, chunk_cnt, cur_offset); + if (ret < 0) { + int saved_errno = errno; + LOG(ERROR) << "pwritev failed for file: " << filename_ + << ", errno=" << saved_errno + << " (" << strerror(saved_errno) << ")" + << ", fd=" << fd_ + << ", iovcnt=" << iovcnt + << ", total_bytes=" << total_bytes + << ", offset=" << offset + << ", chunk_start=" << idx + << ", chunk_cnt=" << chunk_cnt; + return make_error(ErrorCode::FILE_WRITE_FAIL); + } + written_total += ret; + cur_offset += ret; + } + + if (written_total != total_bytes) { + LOG(ERROR) << "pwritev partial write for file: " << filename_ + << ", expected=" << total_bytes + << ", written=" << written_total; return make_error(ErrorCode::FILE_WRITE_FAIL); } - return ret; + return written_total; } tl::expected PosixFile::vector_read(const iovec *iov, @@ -123,12 +165,34 @@ tl::expected PosixFile::vector_read(const iovec *iov, return make_error(ErrorCode::FILE_NOT_FOUND); } - ssize_t ret = ::preadv(fd_, iov, iovcnt, offset); - if (ret < 0) { - return make_error(ErrorCode::FILE_READ_FAIL); + size_t total_bytes = 0; + for (int i = 0; i < iovcnt; ++i) total_bytes += iov[i].iov_len; + + size_t read_total = 0; + off_t cur_offset = offset; + + for (int idx = 0; idx < iovcnt; idx += UIO_MAXIOV) { + int chunk_cnt = std::min(iovcnt - idx, UIO_MAXIOV); + ssize_t ret = ::preadv(fd_, iov + idx, chunk_cnt, cur_offset); + if (ret < 0) { + int saved_errno = errno; + LOG(ERROR) << "preadv failed for file: " << filename_ + << ", errno=" << saved_errno + << " (" << strerror(saved_errno) << ")" + << ", fd=" << fd_ + << ", iovcnt=" << iovcnt + << ", total_bytes=" << total_bytes + << ", offset=" << offset + << ", chunk_start=" << idx + << ", chunk_cnt=" << chunk_cnt; + return make_error(ErrorCode::FILE_READ_FAIL); + } + read_total += ret; + cur_offset += ret; + if (ret == 0) break; // EOF } - return ret; + return read_total; } } // namespace mooncake \ No newline at end of file From 0a977d754d4189573bcaa54f900a1576d8012990 Mon Sep 17 00:00:00 2001 From: "Zhu, Ping" Date: Fri, 3 Jul 2026 14:58:50 +0800 Subject: [PATCH 13/18] docs: rewrite store offload bug fix record for fix/store-offload-main MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replace the previous A-K doc (which targeted the old fix/store-offload-clean branch based on b57d18a4) with a new record covering the 12 commits on fix/store-offload-main (origin/main @ 85e0ed69 + 12 patches). Each section now maps one-to-one to a commit and includes: - the root cause and reproduction symptoms - the code-path details that caused silent failures - verification results from test_lru_bug_repro.py, test_store_restart_cache.py, and AI_ERA_PRESSURE pressure testing A new 'Already covered upstream' table records the three bugs (A/B/C) that upstream PRs #2286, #2599, #2658 fixed since our old base, plus the SSD capacity re-report that upstream file_storage.cpp already implements. This is why those fixes are not reapplied on this branch. Also notes the posix_file UIO_MAXIOV fix (commit 7936a3b4) which is mandatory on upstream main — without it, every bucket write with > ~513 keys returns FILE_WRITE_FAIL (EINVAL), and the cluster stays at 0 SSD usage / 0 cache hits. --- docs/store-offload-bug-fixes.md | 341 ++++++++++++++++++++++++++++++++ 1 file changed, 341 insertions(+) create mode 100644 docs/store-offload-bug-fixes.md diff --git a/docs/store-offload-bug-fixes.md b/docs/store-offload-bug-fixes.md new file mode 100644 index 0000000000..8b292ee959 --- /dev/null +++ b/docs/store-offload-bug-fixes.md @@ -0,0 +1,341 @@ +# Mooncake Store SSD Offload Bug 修复记录 + +> 分支: `fix/store-offload-main` (基于 upstream `main` @ `85e0ed69`) +> 仓库: `pingzhuu/Mooncake` +> 最后更新: 2026-07-03 + +## 背景 + +store-nvme 部署（.15 master + 2 store-server, .18 2 store-server, 4× NVMe SSD） +在 AI_ERA_PRESSURE 压测中出现缓存命中率从 ~80% 断崖式下跌到 ~27% 的问题。 +经多轮诊断后确认根因为 LRU eviction 在提交新 bucket 时使用了时间戳 0， +并伴有一组并发性能 / 元数据正确性 / 串行 I/O 瓶颈等问题。 + +本分支将旧分支 (`fix/store-offload-clean` 基于 `b57d18a4`) 的修复方案 +逐 commit 重新应用到 upstream main，且每个修复独立成 commit 以便审阅与 +cherry-pick。本文件按 commit 顺序记录每个修复。 + +## 已被上游覆盖（无需在本分支修改） + +| Bug | 上游 PR / 说明 | +|-----|---------------| +| A — BatchEvict 重复迭代 / 过量淘汰 | PR #2286 (`be1ceba6`) — Phase 1/2 重构并列扫描候选 | +| B — offload cap 绑定 offload_force_evict_ | PR #2599 (`077b696f`) — `--offloading_queue_limit` / `--offload_cap_ratio` 已为 gflag | +| C — 跳过的 offload key 静默丢弃 → orphan task | PR #2658 (`70ad6c7d`) — `data_size=-1` NACK + NotifyOffloadSuccess 处理 | +| 0ca9f054 等价的 SSD 容量 re-report | upstream `file_storage.cpp` Heartbeat 中已有 MountLocalDiskSegment + ReportSsdCapacity 逻辑 | + +## 本分支提交列表 + +### Bug D: partial bucket 永久滞留于 ungrouped_offloading_objects_ + +**commit**: `81ed7c08` +**文件**: `mooncake-store/src/storage_backend.cpp` `GroupOffloadingKeysByBucket` + +**根因**: 当输入 offloading_objects 在凑满 `bucket_keys_limit` 之前耗尽， +旧逻辑把剩余 key 倒回 `ungrouped_offloading_objects_` 并 `return {}`。压测 +high-watermark 之后每个 heartbeat 只有 ~151-491 key 到达，永远凑不满 +1000 key/bucket → key 永久滞留 → master 端 offloading_task TTL 超 +`put_start_release_timeout_sec` 而 expired。 + +**修复**: 输入耗尽时立即把已收集的 partial bucket 推入 `buckets_keys` +（带 `[OFFLOAD-PARTIAL]` 日志），不再回灌 ungrouped 池。代价是少量额外 +bucket 元数据开销，换得正确性：不会无限积压。 + +--- + +### Bug H: AddReplica 不替换 stale LOCAL_DISK replica + +**commit**: `7f57409c` +**文件**: `mooncake-store/src/master_service.cpp` `AddReplica` + +**根因**: store-server 重启后 ScanMeta 回告时，`AddReplica` 的 `VisitReplicas` +predicate 要求 `client_id` 匹配；旧 client_id 的 replica 不匹配 → 新 replica +被静默丢弃，master 中残留 dead transport_endpoint。后续 GET 取不到新 transport +→ `INVALID_KEY` / RDMA 错误。 + +**修复**: 在检查是否已有 LOCAL_DISK replica 之前，erase 掉所有 +`client_id != caller` 的 LOCAL_DISK replica（`[ADDREPLICA-STALE]` 日志）， +并通过 `EraseReplicasWithCacheTotalAccounting` + `OnDiskReplicaRemoved` 保持 +shard-level disk_object_count 与本地 SSD usage 计数一致。剩余同 client_id +的 replica 走原路径 update endpoint / size。 + +--- + +### Bug I: PutStart OBJECT_ALREADY_EXISTS 阻塞 LOCAL_DISK-only metadata + +**commit**: `05835124` +**文件**: `mooncake-store/src/master_service.cpp` `PutStart` + +**根因**: PutStart 用 `metadata.HasReplica(&Replica::fn_is_completed)` 判断是否 +已存在写入——`fn_is_completed` 匹配任意已完成 replica（含 LOCAL_DISK）。 +store 重启 + ScanMeta 后 metadata 仅含一个 completed LOCAL_DISK replica，导致 +每次相同 key 的 PutStart 都返回 `OBJECT_ALREADY_EXISTS`，把已写到 SSD 上的 +数据永久 gate 掉。 + +**修复**: 改为 `metadata.HasReplica([](r){ r.is_memory_replica() && r.is_completed(); })`。 +LOCAL_DISK-only metadata 不应阻塞新的 PutStart——它的作用是服务 cache 读而不是 +gate 写。 + +**验证**: `tests/test_store_restart_cache.py` Step 6 PUT 阶段 300000 key +`ok=300000 fail=0`。 + +--- + +### Bug K: PutEnd offload 路径 double-pin + emplace 失败不回退 + +**commit**: `d2048ae6` +**文件**: `mooncake-store/src/master_service.cpp` `PutEnd` + +**根因**: PutEnd 的 `!offload_on_evict_` 分支内 VisitReplicas 缺少 +`if (task_created) return` early-exit，且 `emplace` 返回值未检查。`replica_num=2` +时第二个 replica 仍然 `PushOffloadingQueue + inc_refcnt`，但第二份 emplace +被同 key 已存在的 task 排斥 → refcnt 永久泄漏。即使 `replica_num=1`， +若 `offloading_tasks[key]` 已存在（与 BatchEvict 路径的 race），也会泄漏。 + +**修复**: 进 lambda 前加 `offloading_tasks.count(key)==0` 前置检查； +lambda 内 `if (task_created) return` 早退；`emplace` 返回 `inserted=false` +时立即 `dec_refcnt` 回退（`[PUTEND-OFFLOAD-EMPLACE-FAIL]` 日志）。 + +**验证**: 旧分支在生产压测下 OFFLOAD-COMPLETE gap 从 87623 → 0； +新分支压测 5 分钟 OFFLOAD-COMPLETE gap=0。 + +--- + +### Bug J part 1: BatchEvict 跨周期重复 pin → refcnt 泄漏 + +**commit**: `77b43718` +**文件**: `mooncake-store/src/master_service.cpp` `BatchEvict::try_evict_or_offload` + +**根因**: `try_evict_or_offload` 在把 key 推入 offloading queue 之前没有检查 +`offloading_tasks.count(key) > 0`。BatchEvict 第 N 轮把 replica_A 推入队列并 pin； +FetchOffloadingTasks 把 master side `offloading_objects` 清空后，第 N+1 轮可以 +对同一 key 的 replica_B 再 pin 一次。`emplace(key, T2)` 静默失败（key 已有 T1）， +但 replica_B 的 `inc_refcnt` 不会回退 → replica 永久 pin，不可淘汰。 + +**修复**: VisitReplicas 前加 `count(key)==0` 前置检查；emplace 返回 `inserted=false` +时立即 `dec_refcnt` 回退（`[BATCHEVICT-OFFLOAD-EMPLACE-FAIL]` 日志）。双重防御。 + +--- + +### Bug J part 2: EvictTenantMemoryForQuota 同一泄漏路径 + +**commit**: `d3690852` +**文件**: `mooncake-store/src/master_service.cpp` `EvictTenantMemoryForQuota::try_evict_or_offload` + +**根因**: 与 Bug J part 1 同形状，tenant-quota 驱动 evict 路径同样存在 +`emplace` 失败不回退与缺少 count 前置检查的问题。在 BatchEvict 已插入 task 后 +紧接着触发 tenant-quota evict，会对同 key 的第二个 replica 造成 refcnt 泄漏。 + +**修复**: 完全对称的两项防护（count 前置 + emplace 返回值 + `dec_refcnt` 回退， +`[TENANT-EVICT-OFFLOAD-EMPLACE-FAIL]` 日志）。 + +--- + +### LRU bug: BatchOffload 提交新 bucket 时把 LRU 时间戳写为 0 + +**commit**: `40ab7c7d` +**文件**: `mooncake-store/src/storage_backend.cpp` `BatchOffload` + +**根因**: BatchOffload 提交新 bucket 时执行 +`buckets_.emplace(bucket_id, std::move(bucket)); lru_index_.emplace(0LL, bucket_id);`。 +`0` 严格小于任何真实 `steady_clock::now()` 时间戳，所以新写入的 bucket 在 +`SelectEvictionCandidate` 的 LRU 路径下永远是首选项。一旦 SSD 写满触发 +PrepareEviction，新写入的 cache 还没被读到就被淘汰掉，命中率从 ~80% 跌至 ~27%。 + +诊断日志确认： +``` +[LRU-BUG] BatchOffload commit: bucket_id=X lru_ts=0 keys=5 +[LRU-BUG] EvictCandidate: bucket_id=X ts=0 (matches actual) +``` +同一个 bucket_id 刚提交就被选为淘汰候选。 + +**修复**: 提交时 +```cpp +int64_t now_ns = std::chrono::steady_clock::now().time_since_epoch().count(); +bucket->last_access_ns_.store(now_ns, std::memory_order_relaxed); +buckets_.emplace(bucket_id, std::move(bucket)); +lru_index_.emplace(now_ns, bucket_id); +``` +ScanMeta 恢复路径（line ~1592）有意保持 `0`，因为重启后从磁盘扫出的 bucket +确实应当被视为最旧。 + +**影响**: 仅当 `MOONCAKE_OFFLOAD_BUCKET_EVICTION_POLICY=lru` 时触发；默认 `fifo` +按 bucket_id 单调递增淘汰，不受影响。本分支生产配置使用 LRU，必须修。 + +**验证**: `tests/test_lru_bug_repro.py` 在 30MB SSD + 100MB DRAM 的测试 store 上 +通过（`b15 survived — older data was evicted first`）。 + +--- + +### Promotion worker pool（PR #2529 commit 1/3，本分支独立重做） + +**commit**: `dc40e854` +**文件**: `mooncake-store/include/file_storage.h`、`include/storage_backend.h`、`src/file_storage.cpp` + +**根因**: 上游 `ProcessPromotionTasks` 在 store-server 的 heartbeat 线程内 +同步执行每个 promotion task（SSD read + RDMA write）。`promotion_max_per_heartbeat=1` +时每 tick 只 2/s 准入；调到 64 后同一 tick 内 batch_get + TransferWrite 64 个 task +会阻塞 heartbeat 线程几百毫秒，拖垮心跳节奏导致积压。 + +**修复**: 把 per-key body 拆出为 `ProcessPromotionTask(task, preferred_segments)` +返回 `PromotionExecutionResult`，并用 mutex/cv 守护的 `promotion_task_queue_` ++ 后台 worker pool 异步消费。`ProcessPromotionTasks` 持续从 master +`PromotionObjectHeartbeat` 拉 task 直到本地队列达到 soft budget。 + +新增 env vars: +- `MOONCAKE_OFFLOAD_PROMOTION_WORKER_THREADS` (默认 1，生产建议 4) + - `0` 走旧 inline 路径，便于测试与回退 +- `MOONCAKE_OFFLOAD_PROMOTION_QUEUE_CAPACITY` (默认 1024，0 = 无限) +- `MOONCAKE_OFFLOAD_PROMOTION_DRAIN_BATCH_SIZE` (默认 64) + +Init 时 spawn worker；~FileStorage join worker。 + +--- + +### Promotion 默认值调整 + +**commit**: `5cb97e1d` +**文件**: `mooncake-store/include/master_config.h`、`mooncake-store/include/storage_backend.h` + +**根因**: PR #2529 把 promotion 从 heartbeat 同步路径剥离后，老保守默认值 +(`promotion_max_per_heartbeat=1`、`promotion_worker_threads=1`) 仍然存在，导致 +64 并发压测下 25259/~70000 admitted task TTL expired——准入 2/s 满足不了 70/s +的实际需求。 + +**修复**: +- `master_config.h` 4 个 config struct 的 `promotion_max_per_heartbeat`: 1 → 64 +- `FileStorageConfig::promotion_worker_threads`: 1 → 4 +- 64×4/5s ≈ 51/s/server，4 server 聚合约 204/s 远超 70/s 需求 + +调优后 promotion task expired 从 25259 → 0。 + +--- + +### Bug F: OffloadObjects 串行 WriteBucket → SSD 写吞吐瓶颈 + +**commit**: `e738cfd9` +**文件**: `mooncake-store/src/file_storage.cpp`、`mooncake-store/include/storage_backend.h` + +**根因**: `OffloadObjects` 串行遍历每个 bucket 调 `BatchOffload` (→ WriteBucket → +pwritev)。每个 bucket 是独立文件，无共享写状态，串行化纯属人为瓶颈。单盘 NVMe +~1.6 GB/s vs 实测可达 ~4.7 GB/s，跟不上 high-watermark 触发后 cache 涌入速率。 + +**修复**: 把单个 bucket 的处理体捕获为返回 `std::optional` 的 lambda +（nullopt = 成功/可恢复；含值为不可恢复）。 +- `offload_write_threads <= 1` 或仅 1 bucket → 走旧串行路径，零开销 +- 否则 spawn `write_threads` 个 std::thread，从原子 `next_index` 抢 bucket， + 任何 bucket 返回错误即设 `first_error_set` 让其他 worker 立即退出； + caller join 全部 worker 后再返回，保证捕获的引用生命周期安全 + +数据安全： +- `failed_tasks` / `all_bucket_keys` 用 `failed_tasks_mutex` 合并 +- staging_bufs 是 per-bucket RAII 局部 +- complete_handler 的 ssd_metric_ counters 与 client NotifyOffloadSuccess RPC + 均为独立路径 + +新增 env: `MOONCAKE_OFFLOAD_WRITE_THREADS` (默认 4)；=1 退化到串行。 + +--- + +### Bug E: orphan task 诊断日志 + +**commit**: `64863cb0` +**文件**: `mooncake-store/src/master_service.cpp` + +修复 silent orphan-task 路径，加 3 个 tag： +- `[OFFLOAD-NOTASK]` — NotifyOffloadSuccess 收到 complete 但 master 没有 + offloading_task（已被 expired / 清理）。之前静默 AddReplica，现在可见 +- `[CLEANUP-ORPHAN]` — EraseMetadata 删 metadata 时该 key 仍挂 offloading_task + （BatchEvict 第二遍 vs store worker in-flight race） +- `[PUSH-EMPTY]` — PushOffloadingQueue 收到 segment_names 为空的 replica + （之前静默 no-op） + +与 Bug D 的 `[OFFLOAD-PARTIAL]`、Bug J/K 的 emplace-fail 日志一起构成 +orphan-task 诊断套件。 + +--- + +### pwritev/preadv 按 UIO_MAXIOV 分块 + errno 诊断 + +**commit**: `7936a3b4` +**文件**: `mooncake-store/src/posix_file.cpp` + +**根因**: `vector_write` / `vector_read` 直接把整个 iovec 数组传给 `pwritev` / +`preadv`，没有按 `UIO_MAXIOV` (1024) 分块。`BUCKET_KEYS_LIMIT=1000` × 每 key +2 iovec > 1024 → pwritev 返回 `EINVAL` → 转译成 `FILE_WRITE_FAIL`。 + +新分支部署时第一波压测立刻暴露： +``` +vector_write failed for: , error: FILE_WRITE_FAIL +Deleted corrupted file: .../X.bucket +``` +SSD 使用率保持 0，cache hit 为 0。 + +**修复**: +1. 分块循环 `for (int idx = 0; idx < iovcnt; idx += UIO_MAXIOV)`，每块 + `int chunk_cnt = std::min(iovcnt - idx, UIO_MAXIOV)`；累计 offset 与 totals +2. 总字节不匹配 → `FILE_WRITE_FAIL` +3. 失败时打 `errno(strerror) + fd + iovcnt + total_bytes + offset + + chunk_start/chunk_cnt`，便于后续 FILE_WRITE_FAIL 排查直接看到 syscall 错误 + +未在上游 main 中找到对应修复，故本 commit 是必需的。 + +## 配置参数（生产推荐） + +### store-server 端（`launch_store_server.sh`） + +| Env | 值 | 说明 | +|-----|----|------| +| `MOONCAKE_OFFLOAD_HEARTBEAT_INTERVAL_SECONDS` | 5 | 压测下更快处理积压 | +| `MOONCAKE_OFFLOAD_LOCAL_BUFFER_SIZE_BYTES` | 21474836480 (20GB) | batch-get SSD 默认 1.2GB 在并发下 BUFFER_OVERFLOW | +| `MOONCAKE_OFFLOAD_BUCKET_KEYS_LIMIT` | 1000 | 每桶最多 1000 key | +| `MOONCAKE_OFFLOAD_BUCKET_SIZE_LIMIT_BYTES` | 2GB | | +| `MOONCAKE_OFFLOAD_BUCKET_EVICTION_POLICY` | lru | 需配合 LRU bug 修复 | +| `MOONCAKE_OFFLOAD_PROMOTION_WORKER_THREADS` | 4 (新默认) | promotion worker 池 | +| `MOONCAKE_OFFLOAD_PROMOTION_DRAIN_BATCH_SIZE` | 64 (新默认) | | +| `MOONCAKE_OFFLOAD_WRITE_THREADS` | 4 (新默认) | parallel WriteBucket | + +### master 端（`launch_master.sh`） + +| Flag | 值 | 说明 | +|------|----|------| +| `--eviction_high_watermark_ratio` | 0.7 | DRAM 触发 BatchEvict 的水位 | +| `--eviction_ratio` | 0.2 | 目标：DRAM 使用率降到 20% | +| `--put_start_release_timeout_sec` | 120 | 测试时缩短，加速过期诊断 | +| `--promotion_max_per_heartbeat` | 64 (新默认) | 每 tick 准入上限 | +| `--promotion_admission_threshold` | 2 | CountMinSketch 二次访问准入 | +| `--offloading_queue_limit` (upstream PR #2599) | 默认 50000 | 如调高需同步 offload_cap_ratio | +| `--offload_cap_ratio` (upstream PR #2599) | 默认 0.5 | offload_cap = queue_limit × ratio | + +## 验证 + +### test_lru_bug_repro.py +小容量 store (100MB DRAM + 30MB SSD) 上写 30 batch、READ 全部、再写 +b15/b16 触发 eviction。 +- buggy wheel: `b15 evicted` + `[LRU-BUG] lru_ts=0` 日志 +- fixed wheel: `b15 survived — older data was evicted first` + +### test_store_restart_cache.py +写 300000 key → 等 offload → baseline GET → restart store + master → ScanMeta → +再次 GET → PutStart 同批 key。 +- Step 1 WRITE: ok=299592 fail=408（DRAM 满，非 bug） +- Step 5 GET after restart: **hit=299592 miss=0**（验证 Bug H + I） +- Step 6 PUT after restart: **ok=300000 fail=0**（验证 Bug I） + +### AI_ERA_PRESSURE 压测 +4 store-server × 128GB DRAM / 128GB SSD = 512GB / 512GB 总量； +64 并发；glm-5.2 prefill + decoder；5 分钟采样。 +- hit_rate (window): **88.76%**（修复前 27%） +- FILE_WRITE_FAIL: 0 +- save queue full / BUFFER_OVERFLOW: 0 +- Promotion expired / OFFLOAD-EXPIRED: 0 +- OFFLOAD-COMPLETE gap: 0 +- req_err: 0 + +## 测试脚本 + +| 脚本 | 用途 | +|------|------| +| `tests/test_lru_bug_repro.py` | LRU ts=0 bug 复现 / 验证 | +| `tests/run_restart_cache_test.sh` | store restart cache 恢复完整流程 | +| `tests/test_store_restart_cache.py` | 上述脚本调用的 Python 测试本体 | From 43a3d5482ffcdf9ce8671295eaa649c1d3b90567 Mon Sep 17 00:00:00 2001 From: "Zhu, Ping" Date: Fri, 3 Jul 2026 15:35:38 +0800 Subject: [PATCH 14/18] store: replace strerror with thread-safe strerror_r strerror() returns a pointer to a static internal buffer that is shared across all threads; concurrent calls (now possible with parallel WriteBucket and promotion worker threads) corrupt the returned message. Switch all 4 call sites (write/read/pwritev/preadv failure logs) to strerror_r, the POSIX thread-safe variant that writes into a caller-provided stack buffer. --- mooncake-store/src/posix_file.cpp | 16 ++++++++++++---- 1 file changed, 12 insertions(+), 4 deletions(-) diff --git a/mooncake-store/src/posix_file.cpp b/mooncake-store/src/posix_file.cpp index b188c7501b..570ad4df0b 100644 --- a/mooncake-store/src/posix_file.cpp +++ b/mooncake-store/src/posix_file.cpp @@ -57,9 +57,11 @@ tl::expected PosixFile::write(std::span data, if (written == -1) { if (errno == EINTR) continue; int saved_errno = errno; + char errbuf[256]; + strerror_r(saved_errno, errbuf, sizeof(errbuf)); LOG(ERROR) << "write failed for file: " << filename_ << ", errno=" << saved_errno - << " (" << strerror(saved_errno) << ")" + << " (" << errbuf << ")" << ", fd=" << fd_ << ", remaining=" << remaining; return make_error(ErrorCode::FILE_WRITE_FAIL); @@ -94,9 +96,11 @@ tl::expected PosixFile::read(std::string &buffer, if (n == -1) { if (errno == EINTR) continue; int saved_errno = errno; + char errbuf[256]; + strerror_r(saved_errno, errbuf, sizeof(errbuf)); LOG(ERROR) << "read failed for file: " << filename_ << ", errno=" << saved_errno - << " (" << strerror(saved_errno) << ")" + << " (" << errbuf << ")" << ", fd=" << fd_ << ", length=" << length << ", read_bytes=" << read_bytes; @@ -133,9 +137,11 @@ tl::expected PosixFile::vector_write(const iovec *iov, ssize_t ret = ::pwritev(fd_, iov + idx, chunk_cnt, cur_offset); if (ret < 0) { int saved_errno = errno; + char errbuf[256]; + strerror_r(saved_errno, errbuf, sizeof(errbuf)); LOG(ERROR) << "pwritev failed for file: " << filename_ << ", errno=" << saved_errno - << " (" << strerror(saved_errno) << ")" + << " (" << errbuf << ")" << ", fd=" << fd_ << ", iovcnt=" << iovcnt << ", total_bytes=" << total_bytes @@ -176,9 +182,11 @@ tl::expected PosixFile::vector_read(const iovec *iov, ssize_t ret = ::preadv(fd_, iov + idx, chunk_cnt, cur_offset); if (ret < 0) { int saved_errno = errno; + char errbuf[256]; + strerror_r(saved_errno, errbuf, sizeof(errbuf)); LOG(ERROR) << "preadv failed for file: " << filename_ << ", errno=" << saved_errno - << " (" << strerror(saved_errno) << ")" + << " (" << errbuf << ")" << ", fd=" << fd_ << ", iovcnt=" << iovcnt << ", total_bytes=" << total_bytes From d970d10f3e500603fe278fadf84c123c94fd608f Mon Sep 17 00:00:00 2001 From: "Zhu, Ping" Date: Fri, 3 Jul 2026 15:35:55 +0800 Subject: [PATCH 15/18] store: use per-call aligned buffer in WriteBucket for thread safety MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit WriteBucket previously used the shared member aligned_io_buffer_ (inside the USE_URING / O_DIRECT path) to stage data before write_aligned. With parallel WriteBucket dispatch (introduced by Bug F's worker threads), multiple concurrent calls overwrite each other's stage buffer, causing SSD data corruption. Fix: always posix_memalign a per-call buffer and free it when the function exits via the temp_buffer unique_ptr RAII guard. This removes the shared-buffer race entirely at the cost of one posix_memalign + free per bucket write — negligible compared to the pwritev itself. --- mooncake-store/src/storage_backend.cpp | 40 +++++++++++--------------- 1 file changed, 16 insertions(+), 24 deletions(-) diff --git a/mooncake-store/src/storage_backend.cpp b/mooncake-store/src/storage_backend.cpp index 987ceacfa0..c7dfd3c2c0 100644 --- a/mooncake-store/src/storage_backend.cpp +++ b/mooncake-store/src/storage_backend.cpp @@ -2045,33 +2045,25 @@ tl::expected BucketStorageBackend::WriteBucket( size_t total_size = static_cast(bucket_metadata->data_size); size_t aligned_size = align_up(total_size, kDirectIOAlignment); - // Allocate aligned buffer if needed + // Allocate a per-call aligned buffer. The shared member + // aligned_io_buffer_ is NOT safe for concurrent WriteBucket calls + // (parallel WriteBucket dispatches multiple buckets simultaneously + // via worker threads); using it here would cause data corruption. void* write_buffer = nullptr; - std::unique_ptr temp_buffer{nullptr, - [](void*) {}}; + std::unique_ptr temp_buffer{ + nullptr, [](void*) {}}; - if (aligned_size <= kAlignedBufferSize && aligned_io_buffer_) { - // Use the pre-allocated buffer - write_buffer = aligned_io_buffer_.get(); - } else { - // Allocate a temporary larger buffer - void* buf = nullptr; - int ret = posix_memalign(&buf, kDirectIOAlignment, aligned_size); - if (ret != 0) { - LOG(ERROR) - << "Failed to allocate aligned buffer for WriteBucket: " - << strerror(ret); - return tl::make_unexpected(ErrorCode::INTERNAL_ERROR); - } - temp_buffer.reset(buf); - temp_buffer = std::unique_ptr( - buf, [](void* p) { free(p); }); - write_buffer = buf; - LOG(WARNING) << "WriteBucket: bucket_id=" << bucket_id - << " requires " << aligned_size - << " bytes, exceeds buffer size " << kAlignedBufferSize - << ", using temporary allocation"; + void* buf = nullptr; + int ret = posix_memalign(&buf, kDirectIOAlignment, aligned_size); + if (ret != 0) { + LOG(ERROR) + << "Failed to allocate aligned buffer for WriteBucket: " + << strerror(ret); + return tl::make_unexpected(ErrorCode::INTERNAL_ERROR); } + temp_buffer = std::unique_ptr( + buf, [](void* p) { free(p); }); + write_buffer = buf; // Aggregate all iovs data into the aligned buffer char* dst = static_cast(write_buffer); From 7c9da25bc836fbdf074520ce2f2b179ae2df581f Mon Sep 17 00:00:00 2001 From: "Zhu, Ping" Date: Fri, 3 Jul 2026 15:36:18 +0800 Subject: [PATCH 16/18] store: fix promotion_queue_capacity=0 (unbounded) check in pull loop MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ProcessPromotionTasks' master-pull loop used: std::max(1, config_.promotion_queue_capacity) When promotion_queue_capacity is 0 (meaning 'unbounded'), std::max(1, 0) returns 1, so the loop breaks as soon as a single task is queued — defeating the unbounded behavior entirely. Fix: only enforce the cap when capacity > 0, matching the guard in EnqueuePromotionTask (line ~1066). --- mooncake-store/src/file_storage.cpp | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/mooncake-store/src/file_storage.cpp b/mooncake-store/src/file_storage.cpp index 1b697d45d7..a0d3028155 100644 --- a/mooncake-store/src/file_storage.cpp +++ b/mooncake-store/src/file_storage.cpp @@ -928,8 +928,9 @@ tl::expected FileStorage::ProcessPromotionTasks() { ++heartbeat_rounds; { std::lock_guard lock(promotion_queue_mutex_); - if (promotion_task_queue_.size() >= - std::max(1, config_.promotion_queue_capacity)) { + if (config_.promotion_queue_capacity > 0 && + promotion_task_queue_.size() >= + config_.promotion_queue_capacity) { break; } } From e716c3b18ba7efb6fe99d9675ffa2ee1c97fcaa0 Mon Sep 17 00:00:00 2001 From: "Zhu, Ping" Date: Fri, 3 Jul 2026 15:36:27 +0800 Subject: [PATCH 17/18] store: allow promotion_worker_threads=0 for synchronous fallback Init used std::max(1, config_.promotion_worker_threads) which clamped 0 to 1, always spawning at least one worker and setting promotion_workers_running_ = true. This prevented the synchronous (inline-in-heartbeat) fallback path in ProcessPromotionTasks from ever being used, contradicting the documented 0 = disable-workers behavior. Fix: only spawn workers and set the flag when worker_threads > 0. When worker_threads == 0, promotion_workers_running_ stays false and ProcessPromotionTasks falls through to the inline loop. --- mooncake-store/src/file_storage.cpp | 15 ++++++++------- 1 file changed, 8 insertions(+), 7 deletions(-) diff --git a/mooncake-store/src/file_storage.cpp b/mooncake-store/src/file_storage.cpp index a0d3028155..9518638757 100644 --- a/mooncake-store/src/file_storage.cpp +++ b/mooncake-store/src/file_storage.cpp @@ -339,13 +339,14 @@ tl::expected FileStorage::Init() { client_buffer_gc_running_.store(true); client_buffer_gc_thread_ = std::thread(&FileStorage::ClientBufferGCThreadFunc, this); - promotion_workers_running_.store(true); - const uint32_t worker_count = - std::max(1, config_.promotion_worker_threads); - promotion_worker_threads_.reserve(worker_count); - for (uint32_t i = 0; i < worker_count; ++i) { - promotion_worker_threads_.emplace_back( - &FileStorage::PromotionWorkerThreadFunc, this); + if (config_.promotion_worker_threads > 0) { + promotion_workers_running_.store(true); + const uint32_t worker_count = config_.promotion_worker_threads; + promotion_worker_threads_.reserve(worker_count); + for (uint32_t i = 0; i < worker_count; ++i) { + promotion_worker_threads_.emplace_back( + &FileStorage::PromotionWorkerThreadFunc, this); + } } return {}; } From 818c328ce4714d5aaea19f6b032b71401a331942 Mon Sep 17 00:00:00 2001 From: "Zhu, Ping" Date: Fri, 3 Jul 2026 15:41:04 +0800 Subject: [PATCH 18/18] store: pass shard to EraseMetadata in EvictTenantMemoryForQuota for disk_object_count sync EvictTenantMemoryForQuota's try_evict_group_or_object lambda and its 'pass' loop previously called the legacy 3-arg EraseMetadata overload, which forwards to the 5-arg implementation with shard=nullptr. This skipped OnDiskReplicaRemoved, so disk_object_count was never decremented when LOCAL_DISK replicas were erased during tenant-quota eviction. Over time disk_object_count drifted upward, causing BatchEvict's eviction_base = metadata.size() - disk_object_count to underflow and degrade eviction decisions. Fix: add a MetadataShardAccessorRW* parameter to the try_evict_group_ or_object lambda and forward &shard from the pass loop. Both EraseMetadata call sites in EvictTenantMemoryForQuota now use the 5-arg overload, matching the BatchEvict path which already passes &shard. --- mooncake-store/src/master_service.cpp | 11 +++++++---- 1 file changed, 7 insertions(+), 4 deletions(-) diff --git a/mooncake-store/src/master_service.cpp b/mooncake-store/src/master_service.cpp index 64aecb2478..a83bd09352 100644 --- a/mooncake-store/src/master_service.cpp +++ b/mooncake-store/src/master_service.cpp @@ -7027,7 +7027,8 @@ MasterService::EvictTenantMemoryForQuota(const std::string& tenant_id, [&, this](const std::string& key, ObjectMetadata& metadata, TenantState& tenant_state, std::vector>& deferred_replicas, - bool allow_soft_pinned) -> TenantQuotaEvictionResult { + bool allow_soft_pinned, + MetadataShardAccessorRW* shard_ptr) -> TenantQuotaEvictionResult { if (!metadata.IsGrouped()) { uint64_t freed = try_evict_or_offload(key, metadata, tenant_state, deferred_replicas); @@ -7074,7 +7075,8 @@ MasterService::EvictTenantMemoryForQuota(const std::string& tenant_id, ++result.evicted_objects; } if (member_key != key && !member_metadata.IsValid()) { - EraseMetadata(tenant_state, member_it, normalized_tenant); + EraseMetadata(tenant_state, member_it, normalized_tenant, + QuotaEraseMode::kFull, shard_ptr); } } return result; @@ -7108,11 +7110,12 @@ MasterService::EvictTenantMemoryForQuota(const std::string& tenant_id, auto evict_result = try_evict_group_or_object( it->first, metadata, tenant_state, deferred_replicas, - allow_soft_pinned); + allow_soft_pinned, &shard); total.freed_bytes += evict_result.freed_bytes; total.evicted_objects += evict_result.evicted_objects; if (!metadata.IsValid()) { - it = EraseMetadata(tenant_state, it, normalized_tenant); + it = EraseMetadata(tenant_state, it, normalized_tenant, + QuotaEraseMode::kFull, &shard); } else { ++it; }