[meta] lru cache support inner last access time field

Author: DWAE86

kv_cache_manager/common/cache/advanced_cache.h

@@ -415,12 +415,22 @@ class Cache {
             void(const std::string_view &key, ObjectPtr obj, size_t charge, const CacheItemHelper *helper)> &callback,
         const ApplyToAllEntriesOptions &opts) = 0;
 
+    // Callback invoked (while holding the shard mutex) whenever the LRU tail
+    // of a shard changes. Parameters: (shard_id, tail_value_ptr).
+    // tail_value_ptr is nullptr when the shard becomes empty.
+    using TailChangeCallback = std::function<void(uint32_t shard_id, ObjectPtr tail_value)>;
+
+    // Register a callback that is invoked whenever the LRU tail of any shard
+    // changes. The callback is called inside the shard lock, so it must be
+    // very lightweight (e.g. a single atomic store).
+    // Default implementation is a no-op (not supported).
+    virtual void SetTailChangeCallback(TailChangeCallback /*callback*/) {}
+
     // Returns up to `count` oldest (least recently used) keys from the
     // specified shard. Returns the number of keys actually collected.
     // Default implementation returns 0 (not supported).
-    virtual size_t GetOldestKeysInShard(uint32_t /*shard_id*/,
-                                        size_t /*count*/,
-                                        std::vector<std::string> & /*out_keys*/) {
+    virtual size_t
+    GetOldestKeysInShard(uint32_t /*shard_id*/, size_t /*count*/, std::vector<std::string> & /*out_keys*/) {
         return 0;
     }
 
@@ -431,9 +441,9 @@ class Cache {
     // Default implementation is a no-op (not supported).
     virtual void ApplyToSingleShard(
         uint32_t /*shard_id*/,
-        const std::function<
-            void(const std::string_view &key, ObjectPtr obj, size_t charge, const CacheItemHelper *helper)>
-            & /*callback*/) {}
+        const std::function<void(
+            const std::string_view &key, ObjectPtr obj, size_t charge, const CacheItemHelper *helper)> & /*callback*/) {
+    }
 
     // Insert a mapping from key->object only if the key does not already exist.
     // Returns EC_OK on successful insertion, EC_EXIST if the key is already
@@ -680,8 +690,8 @@ class CacheWrapper : public Cache {
     void ApplyToSingleShard(
         uint32_t shard_id,
         const std::function<
-            void(const std::string_view &key, ObjectPtr value, size_t charge, const CacheItemHelper *helper)>
-            &callback) override {
+            void(const std::string_view &key, ObjectPtr value, size_t charge, const CacheItemHelper *helper)> &callback)
+        override {
         target_->ApplyToSingleShard(shard_id, callback);
     }
 

kv_cache_manager/common/cache/lru_cache.cc

@@ -87,7 +87,7 @@ void LRUHandleTable::Resize() {
 
     uint32_t old_length = uint32_t{1} << length_bits_;
     int new_length_bits = length_bits_ + 1;
-    std::unique_ptr<LRUHandle *[]> new_list{new LRUHandle *[size_t{1} << new_length_bits] {}};
+    std::unique_ptr<LRUHandle *[]> new_list { new LRUHandle *[size_t{1} << new_length_bits] {} };
     [[maybe_unused]] uint32_t count = 0;
     for (uint32_t i = 0; i < old_length; i++) {
         LRUHandle *h = list_[i];
@@ -225,6 +225,7 @@ double LRUCacheShard::GetLowPriPoolRatio() {
 void LRUCacheShard::LRU_Remove(LRUHandle *e) {
     assert(e->next != nullptr);
     assert(e->prev != nullptr);
+    bool was_tail = (lru_.next == e);
     if (lru_low_pri_ == e) {
         lru_low_pri_ = e->prev;
     }
@@ -244,11 +245,15 @@ void LRUCacheShard::LRU_Remove(LRUHandle *e) {
         assert(low_pri_pool_usage_ >= e->total_charge);
         low_pri_pool_usage_ -= e->total_charge;
     }
+    if (was_tail) {
+        NotifyTailChange();
+    }
 }
 
 void LRUCacheShard::LRU_Insert(LRUHandle *e) {
     assert(e->next == nullptr);
     assert(e->prev == nullptr);
+    bool was_empty = (lru_.next == &lru_);
     if (high_pri_pool_ratio_ > 0 && (e->IsHighPri() || e->HasHit())) {
         // Inset "e" to head of LRU list.
         e->next = &lru_;
@@ -285,6 +290,10 @@ void LRUCacheShard::LRU_Insert(LRUHandle *e) {
         lru_bottom_pri_ = e;
     }
     lru_usage_ += e->total_charge;
+    if (was_empty) {
+        // List went from 0 to 1 entry: the newly inserted node is the tail.
+        NotifyTailChange();
+    }
 }
 
 void LRUCacheShard::MaintainPoolSize() {
@@ -358,8 +367,8 @@ void LRUCacheShard::SetStrictCapacityLimit(bool strict_capacity_limit) {
     strict_capacity_limit_ = strict_capacity_limit;
 }
 
-ErrorCode LRUCacheShard::DoInsertItemUnsafe(LRUHandle *e, LRUHandle **handle,
-                                      autovector<LRUHandle *> *last_reference_list) {
+ErrorCode
+LRUCacheShard::DoInsertItemUnsafe(LRUHandle *e, LRUHandle **handle, autovector<LRUHandle *> *last_reference_list) {
     // Free the space following strict LRU policy until enough space
     // is freed or the lru list is empty.
     EvictFromLRU(e->total_charge, last_reference_list);
@@ -679,6 +688,28 @@ size_t LRUCacheShard::GetOldestKeys(size_t count, std::vector<std::string> &out_
     return collected;
 }
 
+void LRUCacheShard::SetTailChangeCallback(uint32_t shard_id, const Cache::TailChangeCallback &callback) {
+    std::lock_guard<std::mutex> l(mutex_);
+    shard_id_ = shard_id;
+    tail_change_callback_ = callback;
+    // Notify immediately with current tail state so the caller gets the
+    // initial value.
+    NotifyTailChange();
+}
+
+void LRUCacheShard::NotifyTailChange() {
+    if (!tail_change_callback_) {
+        return;
+    }
+    LRUHandle *tail = lru_.next;
+    if (tail == &lru_) {
+        // LRU list is empty.
+        tail_change_callback_(shard_id_, nullptr);
+    } else {
+        tail_change_callback_(shard_id_, tail->value);
+    }
+}
+
 void LRUCacheShard::AppendPrintableOptions(std::string &str) const {
     const int kBufferSize = 200;
     char buffer[kBufferSize];
@@ -739,16 +770,21 @@ size_t LRUCache::TEST_GetLRUSize() {
 
 double LRUCache::GetHighPriPoolRatio() { return GetShard(0).GetHighPriPoolRatio(); }
 
-size_t LRUCache::GetOldestKeysInShard(uint32_t shard_id,
-                                       size_t count,
-                                       std::vector<std::string> &out_keys) {
+size_t LRUCache::GetOldestKeysInShard(uint32_t shard_id, size_t count, std::vector<std::string> &out_keys) {
     uint32_t num_shards = GetNumShards();
     if (shard_id >= num_shards || count == 0) {
         return 0;
     }
     return GetShard(shard_id).GetOldestKeys(count, out_keys);
 }
 
+void LRUCache::SetTailChangeCallback(TailChangeCallback callback) {
+    uint32_t num_shards = GetNumShards();
+    for (uint32_t i = 0; i < num_shards; ++i) {
+        GetShard(i).SetTailChangeCallback(i, callback);
+    }
+}
+
 } // namespace lru_cache
 
 std::shared_ptr<Cache> LRUCacheOptions::MakeSharedCache() const {

kv_cache_manager/common/cache/lru_cache.h

@@ -367,6 +367,10 @@ class ALIGN_AS(CACHE_LINE_SIZE) LRUCacheShard final : public CacheShardBase {
     // Returns the number of keys actually collected.
     size_t GetOldestKeys(size_t count, std::vector<std::string> &out_keys);
 
+    // Set the shard id and tail-change callback for this shard.
+    // The callback is invoked inside the shard mutex whenever the LRU tail changes.
+    void SetTailChangeCallback(uint32_t shard_id, const Cache::TailChangeCallback &callback);
+
 private:
     friend class LRUCache;
     // Insert an item into the hash table and, if handle is null, insert into
@@ -468,6 +472,14 @@ class ALIGN_AS(CACHE_LINE_SIZE) LRUCacheShard final : public CacheShardBase {
 
     // A reference to Cache::eviction_callback_
     const Cache::EvictionCallback &eviction_callback_;
+
+    // Tail-change notification support.
+    uint32_t shard_id_{0};
+    Cache::TailChangeCallback tail_change_callback_;
+
+    // Must be called while holding mutex_. Invokes tail_change_callback_ with
+    // the current tail value (nullptr if the LRU list is empty).
+    void NotifyTailChange();
 };
 
 class LRUCache
@@ -494,9 +506,10 @@ class LRUCache
     double GetHighPriPoolRatio();
 
     // Returns up to `count` oldest keys from the specified shard.
-    size_t GetOldestKeysInShard(uint32_t shard_id,
-                                size_t count,
-                                std::vector<std::string> &out_keys) override;
+    size_t GetOldestKeysInShard(uint32_t shard_id, size_t count, std::vector<std::string> &out_keys) override;
+
+    // Register a callback invoked whenever the LRU tail of any shard changes.
+    void SetTailChangeCallback(TailChangeCallback callback) override;
 };
 
 } // namespace lru_cache

kv_cache_manager/meta/meta_cached_backend.cc

@@ -305,15 +305,6 @@ std::vector<ErrorCode> MetaCachedBackend::Upsert(const KeyTypeVec &keys, const F
     return local_backend_->Upsert(keys, field_maps, persistent_results);
 }
 
-std::vector<ErrorCode> MetaCachedBackend::IncrFields(const KeyTypeVec &keys,
-                                                     const std::map<std::string, int64_t> &field_amounts) noexcept {
-    if (recover_state_.load(std::memory_order_acquire) == RecoverState::kRecover) {
-        EnsureKeyInLocal(keys);
-    }
-    std::vector<ErrorCode> persistent_results = persistent_backend_->IncrFields(keys, field_amounts);
-    return local_backend_->IncrFields(keys, field_amounts, persistent_results);
-}
-
 std::vector<ErrorCode> MetaCachedBackend::Delete(const KeyTypeVec &keys) noexcept {
     std::vector<ErrorCode> persistent_results = persistent_backend_->Delete(keys);
 
@@ -451,16 +442,7 @@ ErrorCode MetaCachedBackend::SampleReclaimKeys(const int64_t count, std::vector<
 }
 
 ErrorCode MetaCachedBackend::PutMetaData(const FieldMap &field_maps) noexcept {
-    ErrorCode persistent_ec = persistent_backend_->PutMetaData(field_maps);
-    if (persistent_ec != EC_OK) {
-        return persistent_ec;
-    }
-    ErrorCode cache_ec = local_backend_->PutMetaData(field_maps);
-    if (cache_ec != EC_OK) {
-        KVCM_LOG_ERROR("meta cached backend: cache PutMetaData failed ec[%d], persistent succeeded", cache_ec);
-        return cache_ec;
-    }
-    return EC_OK;
+    return persistent_backend_->PutMetaData(field_maps);
 }
 
 ErrorCode MetaCachedBackend::GetMetaData(FieldMap &field_maps) noexcept {

kv_cache_manager/meta/meta_cached_backend.h

@@ -38,8 +38,6 @@ class MetaCachedBackend : public MetaStorageBackend {
     std::vector<ErrorCode> Put(const KeyTypeVec &keys, const FieldMapVec &field_maps) noexcept override;
     std::vector<ErrorCode> UpdateFields(const KeyTypeVec &keys, const FieldMapVec &field_maps) noexcept override;
     std::vector<ErrorCode> Upsert(const KeyTypeVec &keys, const FieldMapVec &field_maps) noexcept override;
-    std::vector<ErrorCode> IncrFields(const KeyTypeVec &keys,
-                                      const std::map<std::string, int64_t> &field_amounts) noexcept override;
     std::vector<ErrorCode> Delete(const KeyTypeVec &keys) noexcept override;
 
     // read