branch-4.1: [improvement](filecache) limit file cache LRU replay queues #64381 (#64387)

Cherry-picked from #64381

Co-authored-by: zhengyu <zhangzhengyu@selectdb.com>

Author: github-actions[bot]

be/src/common/config.cpp

@@ -1231,6 +1231,7 @@ DEFINE_mInt64(file_cache_remove_block_qps_limit, "1000");
 DEFINE_mInt64(file_cache_background_gc_interval_ms, "100");
 DEFINE_mInt64(file_cache_background_block_lru_update_interval_ms, "5000");
 DEFINE_mInt64(file_cache_background_block_lru_update_qps_limit, "1000");
+DEFINE_mInt64(file_cache_background_block_lru_update_queue_max_size, "500000");
 DEFINE_mBool(enable_file_cache_async_touch_on_get_or_set, "false");
 DEFINE_mBool(enable_reader_dryrun_when_download_file_cache, "true");
 DEFINE_mInt64(file_cache_background_monitor_interval_ms, "5000");
@@ -1242,7 +1243,8 @@ DEFINE_mInt64(file_cache_background_lru_dump_interval_ms, "60000");
 // dump queue only if the queue update specific times through several dump intervals
 DEFINE_mInt64(file_cache_background_lru_dump_update_cnt_threshold, "1000");
 DEFINE_mInt64(file_cache_background_lru_dump_tail_record_num, "5000000");
-DEFINE_mInt64(file_cache_background_lru_log_replay_interval_ms, "1000");
+DEFINE_mInt64(file_cache_background_lru_log_queue_max_size, "500000");
+DEFINE_mInt64(file_cache_background_lru_log_replay_interval_ms, "1");
 DEFINE_mBool(enable_evaluate_shadow_queue_diff, "false");
 
 DEFINE_mBool(file_cache_enable_only_warm_up_idx, "false");

be/src/common/config.h

@@ -1268,6 +1268,7 @@ DECLARE_mInt64(file_cache_remove_block_qps_limit);
 DECLARE_mInt64(file_cache_background_gc_interval_ms);
 DECLARE_mInt64(file_cache_background_block_lru_update_interval_ms);
 DECLARE_mInt64(file_cache_background_block_lru_update_qps_limit);
+DECLARE_mInt64(file_cache_background_block_lru_update_queue_max_size);
 DECLARE_mBool(enable_file_cache_async_touch_on_get_or_set);
 DECLARE_mBool(enable_reader_dryrun_when_download_file_cache);
 DECLARE_mInt64(file_cache_background_monitor_interval_ms);
@@ -1282,6 +1283,7 @@ DECLARE_mInt64(file_cache_background_lru_dump_interval_ms);
 // dump queue only if the queue update specific times through several dump intervals
 DECLARE_mInt64(file_cache_background_lru_dump_update_cnt_threshold);
 DECLARE_mInt64(file_cache_background_lru_dump_tail_record_num);
+DECLARE_mInt64(file_cache_background_lru_log_queue_max_size);
 DECLARE_mInt64(file_cache_background_lru_log_replay_interval_ms);
 DECLARE_mBool(enable_evaluate_shadow_queue_diff);
 

be/src/io/cache/block_file_cache.cpp

@@ -59,24 +59,53 @@
 namespace doris::io {
 #include "common/compile_check_begin.h"
 
+namespace {
+
+constexpr std::array<FileCacheType, 4> LRU_LOG_REPLAY_TYPES = {
+        FileCacheType::TTL, FileCacheType::INDEX, FileCacheType::NORMAL, FileCacheType::DISPOSABLE};
+
+size_t file_cache_type_index(FileCacheType type) {
+    return static_cast<size_t>(type);
+}
+
+} // namespace
+
 // Insert a block pointer into one shard while swallowing allocation failures.
-bool NeedUpdateLRUBlocks::insert(FileBlockSPtr block) {
-    if (!block) {
+bool NeedUpdateLRUBlocks::insert(FileBlockSPtr block, size_t max_queue_size) {
+    if (!block || max_queue_size == 0) {
         return false;
     }
+    bool reserved = false;
     try {
         auto* raw_ptr = block.get();
         auto idx = shard_index(raw_ptr);
         auto& shard = _shards[idx];
         std::lock_guard lock(shard.mutex);
-        auto [_, inserted] = shard.entries.emplace(raw_ptr, std::move(block));
-        if (inserted) {
-            _size.fetch_add(1, std::memory_order_relaxed);
+        if (shard.entries.contains(raw_ptr)) {
+            return false;
         }
-        return inserted;
+        size_t cur_size = _size.load(std::memory_order_relaxed);
+        while (cur_size < max_queue_size) {
+            if (_size.compare_exchange_weak(cur_size, cur_size + 1, std::memory_order_relaxed)) {
+                reserved = true;
+                break;
+            }
+        }
+        if (!reserved) {
+            return false;
+        }
+        auto [_, inserted] = shard.entries.emplace(raw_ptr, std::move(block));
+        DORIS_CHECK(inserted);
+        return true;
     } catch (const std::exception& e) {
+        if (reserved) {
+            decrease_size(1);
+        }
         LOG(WARNING) << "Failed to enqueue block for LRU update: " << e.what();
     } catch (...) {
+        if (reserved) {
+            decrease_size(1);
+        }
         LOG(WARNING) << "Failed to enqueue block for LRU update: unknown error";
     }
     return false;
@@ -103,7 +132,7 @@ size_t NeedUpdateLRUBlocks::drain(size_t limit, std::vector<FileBlockSPtr>* outp
                 ++shard_drained;
             }
             if (shard_drained > 0) {
-                _size.fetch_sub(shard_drained, std::memory_order_relaxed);
+                decrease_size(shard_drained);
                 drained += shard_drained;
             }
         }
@@ -123,7 +152,7 @@ void NeedUpdateLRUBlocks::clear() {
             if (!shard.entries.empty()) {
                 auto removed = shard.entries.size();
                 shard.entries.clear();
-                _size.fetch_sub(removed, std::memory_order_relaxed);
+                decrease_size(removed);
             }
         }
     } catch (const std::exception& e) {
@@ -133,6 +162,16 @@ void NeedUpdateLRUBlocks::clear() {
     }
 }
 
+void NeedUpdateLRUBlocks::decrease_size(size_t delta) {
+    size_t cur_size = _size.load(std::memory_order_relaxed);
+    while (true) {
+        DORIS_CHECK(cur_size >= delta);
+        if (_size.compare_exchange_weak(cur_size, cur_size - delta, std::memory_order_relaxed)) {
+            return;
+        }
+    }
+}
+
 size_t NeedUpdateLRUBlocks::shard_index(FileBlock* ptr) const {
     DCHECK(ptr != nullptr);
     return std::hash<FileBlock*> {}(ptr)&kShardMask;
@@ -348,12 +387,30 @@ BlockFileCache::BlockFileCache(const std::string& cache_base_path,
             _cache_base_path.c_str(), "file_cache_recycle_keys_length");
     _need_update_lru_blocks_length_recorder = std::make_shared<bvar::LatencyRecorder>(
             _cache_base_path.c_str(), "file_cache_need_update_lru_blocks_length");
+    _need_update_lru_blocks_produce_metrics = std::make_shared<bvar::Adder<size_t>>(
+            _cache_base_path.c_str(), "file_cache_need_update_lru_blocks_produce");
+    _need_update_lru_blocks_consume_metrics = std::make_shared<bvar::Adder<size_t>>(
+            _cache_base_path.c_str(), "file_cache_need_update_lru_blocks_consume");
     _update_lru_blocks_latency_us = std::make_shared<bvar::LatencyRecorder>(
             _cache_base_path.c_str(), "file_cache_update_lru_blocks_latency_us");
     _ttl_gc_latency_us = std::make_shared<bvar::LatencyRecorder>(_cache_base_path.c_str(),
                                                                  "file_cache_ttl_gc_latency_us");
     _shadow_queue_levenshtein_distance = std::make_shared<bvar::LatencyRecorder>(
             _cache_base_path.c_str(), "file_cache_shadow_queue_levenshtein_distance");
+    for (FileCacheType type : {FileCacheType::DISPOSABLE, FileCacheType::NORMAL,
+                               FileCacheType::INDEX, FileCacheType::TTL}) {
+        size_t idx = file_cache_type_index(type);
+        std::string metric_prefix =
+                "file_cache_lru_recorder_" + cache_type_to_string(type) + "_record_queue";
+        _lru_recorder_queue_length_recorder[idx] = std::make_shared<bvar::LatencyRecorder>(
+                _cache_base_path.c_str(), metric_prefix + "_length");
+        _lru_recorder_queue_produce_metrics[idx] = std::make_shared<bvar::Adder<size_t>>(
+                _cache_base_path.c_str(), metric_prefix + "_produce");
+        _lru_recorder_queue_consume_metrics[idx] = std::make_shared<bvar::Adder<size_t>>(
+                _cache_base_path.c_str(), metric_prefix + "_consume");
+    }
+    _lru_recorder_log_replay_idle_metrics = std::make_shared<bvar::Adder<size_t>>(
+            _cache_base_path.c_str(), "file_cache_lru_recorder_log_replay_idle");
 
     _disposable_queue = LRUQueue(cache_settings.disposable_queue_size,
                                  cache_settings.disposable_queue_elements, 60 * 60);
@@ -648,7 +705,10 @@ FileBlocks BlockFileCache::get_impl(const UInt128Wrapper& hash, const CacheConte
 }
 
 void BlockFileCache::add_need_update_lru_block(FileBlockSPtr block) {
-    if (_need_update_lru_blocks.insert(std::move(block))) {
+    int64_t queue_limit = config::file_cache_background_block_lru_update_queue_max_size;
+    size_t max_queue_size = queue_limit <= 0 ? 0 : static_cast<size_t>(queue_limit);
+    if (_need_update_lru_blocks.insert(std::move(block), max_queue_size)) {
+        *_need_update_lru_blocks_produce_metrics << 1;
         *_need_update_lru_blocks_length_recorder << _need_update_lru_blocks.size();
     }
 }
@@ -2093,6 +2153,7 @@ void BlockFileCache::run_background_block_lru_update() {
             *_need_update_lru_blocks_length_recorder << _need_update_lru_blocks.size();
             continue;
         }
+        *_need_update_lru_blocks_consume_metrics << drained;
 
         int64_t duration_ns = 0;
         {
@@ -2307,19 +2368,28 @@ void BlockFileCache::run_background_lru_log_replay() {
             }
         }
 
-        _lru_recorder->replay_queue_event(FileCacheType::TTL);
-        _lru_recorder->replay_queue_event(FileCacheType::INDEX);
-        _lru_recorder->replay_queue_event(FileCacheType::NORMAL);
-        _lru_recorder->replay_queue_event(FileCacheType::DISPOSABLE);
+        replay_lru_logs_once();
+    }
+}
+
+size_t BlockFileCache::replay_lru_logs_once() {
+    size_t replayed = 0;
+    for (FileCacheType type : LRU_LOG_REPLAY_TYPES) {
+        replayed += _lru_recorder->replay_queue_event(type);
+    }
 
-        if (config::enable_evaluate_shadow_queue_diff) {
-            SCOPED_CACHE_LOCK(_mutex, this);
-            _lru_recorder->evaluate_queue_diff(_ttl_queue, "ttl", cache_lock);
-            _lru_recorder->evaluate_queue_diff(_index_queue, "index", cache_lock);
-            _lru_recorder->evaluate_queue_diff(_normal_queue, "normal", cache_lock);
-            _lru_recorder->evaluate_queue_diff(_disposable_queue, "disposable", cache_lock);
-        }
+    if (replayed == 0) {
+        *_lru_recorder_log_replay_idle_metrics << 1;
+    }
+
+    if (config::enable_evaluate_shadow_queue_diff) {
+        SCOPED_CACHE_LOCK(_mutex, this);
+        _lru_recorder->evaluate_queue_diff(_ttl_queue, "ttl", cache_lock);
+        _lru_recorder->evaluate_queue_diff(_index_queue, "index", cache_lock);
+        _lru_recorder->evaluate_queue_diff(_normal_queue, "normal", cache_lock);
+        _lru_recorder->evaluate_queue_diff(_disposable_queue, "disposable", cache_lock);
     }
+    return replayed;
 }
 
 void BlockFileCache::dump_lru_queues(bool force) {

be/src/io/cache/block_file_cache.h

@@ -25,6 +25,7 @@
 #include <atomic>
 #include <boost/lockfree/spsc_queue.hpp>
 #include <functional>
+#include <limits>
 #include <memory>
 #include <mutex>
 #include <optional>
@@ -91,7 +92,7 @@ class NeedUpdateLRUBlocks {
 
     // Insert a block into the pending set. Returns true only when the block
     // was not already queued. Null inputs are ignored.
-    bool insert(FileBlockSPtr block);
+    bool insert(FileBlockSPtr block, size_t max_queue_size = std::numeric_limits<size_t>::max());
 
     // Drain up to `limit` unique blocks into `output`. The method returns how
     // many blocks were actually drained and shrinks the internal size
@@ -114,6 +115,7 @@ class NeedUpdateLRUBlocks {
     };
 
     size_t shard_index(FileBlock* ptr) const;
+    void decrease_size(size_t delta);
 
     std::array<Shard, kShardCount> _shards;
     std::atomic<size_t> _size {0};
@@ -464,6 +466,7 @@ class BlockFileCache {
     void run_background_monitor();
     void run_background_gc();
     void run_background_lru_log_replay();
+    size_t replay_lru_logs_once();
     void run_background_lru_dump();
     void restore_lru_queues_from_disk(std::lock_guard<std::mutex>& cache_lock);
     void run_background_evict_in_advance();
@@ -611,9 +614,15 @@ class BlockFileCache {
     std::shared_ptr<bvar::LatencyRecorder> _recycle_keys_length_recorder;
     std::shared_ptr<bvar::LatencyRecorder> _update_lru_blocks_latency_us;
     std::shared_ptr<bvar::LatencyRecorder> _need_update_lru_blocks_length_recorder;
+    std::shared_ptr<bvar::Adder<size_t>> _need_update_lru_blocks_produce_metrics;
+    std::shared_ptr<bvar::Adder<size_t>> _need_update_lru_blocks_consume_metrics;
     std::shared_ptr<bvar::LatencyRecorder> _ttl_gc_latency_us;
 
     std::shared_ptr<bvar::LatencyRecorder> _shadow_queue_levenshtein_distance;
+    std::array<std::shared_ptr<bvar::LatencyRecorder>, 4> _lru_recorder_queue_length_recorder;
+    std::array<std::shared_ptr<bvar::Adder<size_t>>, 4> _lru_recorder_queue_produce_metrics;
+    std::array<std::shared_ptr<bvar::Adder<size_t>>, 4> _lru_recorder_queue_consume_metrics;
+    std::shared_ptr<bvar::Adder<size_t>> _lru_recorder_log_replay_idle_metrics;
     // keep _storage last so it will deconstruct first
     // otherwise, load_cache_info_into_memory might crash
     // coz it will use other members of BlockFileCache

be/src/io/cache/lru_queue_recorder.cpp

@@ -17,27 +17,53 @@
 
 #include "io/cache/lru_queue_recorder.h"
 
+#include "common/config.h"
+#include "common/status.h"
 #include "io/cache/block_file_cache.h"
 #include "io/cache/file_cache_common.h"
 
 namespace doris::io {
 
+namespace {
+
+size_t file_cache_type_index(FileCacheType type) {
+    return static_cast<size_t>(type);
+}
+
+} // namespace
+
 void LRUQueueRecorder::record_queue_event(FileCacheType type, CacheLRULogType log_type,
                                           const UInt128Wrapper hash, const size_t offset,
                                           const size_t size) {
-    CacheLRULogQueue& log_queue = get_lru_log_queue(type);
-    log_queue.enqueue(std::make_unique<CacheLRULog>(log_type, hash, offset, size));
+    if (config::file_cache_background_lru_dump_tail_record_num <= 0) {
+        return;
+    }
     ++(_lru_queue_update_cnt_from_last_dump[type]);
+    auto log = std::make_unique<CacheLRULog>(log_type, hash, offset, size);
+    if (!reserve_lru_log_queue_slot(type)) {
+        return;
+    }
+    CacheLRULogQueue& log_queue = get_lru_log_queue(type);
+    if (!log_queue.enqueue(std::move(log))) {
+        release_lru_log_queue_slot(type);
+        return;
+    }
+    size_t idx = file_cache_type_index(type);
+    *(_mgr->_lru_recorder_queue_produce_metrics[idx]) << 1;
+    *(_mgr->_lru_recorder_queue_length_recorder[idx]) << lru_log_queue_size(type);
 }
 
-void LRUQueueRecorder::replay_queue_event(FileCacheType type) {
+size_t LRUQueueRecorder::replay_queue_event(FileCacheType type) {
     // we don't need the real cache lock for the shadow queue, but we do need a lock to prevent read/write contension
     CacheLRULogQueue& log_queue = get_lru_log_queue(type);
     LRUQueue& shadow_queue = get_shadow_queue(type);
 
     std::lock_guard<std::mutex> lru_log_lock(_mutex_lru_log);
     std::unique_ptr<CacheLRULog> log;
+    size_t replayed = 0;
     while (log_queue.try_dequeue(log)) {
+        release_lru_log_queue_slot(type);
+        ++replayed;
         try {
             switch (log->type) {
             case CacheLRULogType::ADD: {
@@ -79,6 +105,12 @@ void LRUQueueRecorder::replay_queue_event(FileCacheType type) {
             LOG(WARNING) << "Failed to replay queue event: " << e.what();
         }
     }
+    size_t idx = file_cache_type_index(type);
+    if (replayed > 0) {
+        *(_mgr->_lru_recorder_queue_consume_metrics[idx]) << replayed;
+    }
+    *(_mgr->_lru_recorder_queue_length_recorder[idx]) << lru_log_queue_size(type);
+    return replayed;
 }
 
 // we evaluate the diff between two queue by calculate how many operation is
@@ -137,4 +169,34 @@ void LRUQueueRecorder::reset_lru_queue_update_cnt_from_last_dump(FileCacheType t
     _lru_queue_update_cnt_from_last_dump[type] = 0;
 }
 
+size_t LRUQueueRecorder::lru_log_queue_size(FileCacheType type) const {
+    return _lru_log_queue_size[file_cache_type_index(type)].load(std::memory_order_relaxed);
+}
+
+bool LRUQueueRecorder::reserve_lru_log_queue_slot(FileCacheType type) {
+    int64_t queue_limit = config::file_cache_background_lru_log_queue_max_size;
+    if (queue_limit <= 0) {
+        return false;
+    }
+    auto& queue_size = _lru_log_queue_size[file_cache_type_index(type)];
+    size_t cur_size = queue_size.load(std::memory_order_relaxed);
+    while (cur_size < static_cast<size_t>(queue_limit)) {
+        if (queue_size.compare_exchange_weak(cur_size, cur_size + 1, std::memory_order_relaxed)) {
+            return true;
+        }
+    }
+    return false;
+}
+
+void LRUQueueRecorder::release_lru_log_queue_slot(FileCacheType type) {
+    auto& queue_size = _lru_log_queue_size[file_cache_type_index(type)];
+    size_t cur_size = queue_size.load(std::memory_order_relaxed);
+    while (true) {
+        DORIS_CHECK(cur_size > 0);
+        if (queue_size.compare_exchange_weak(cur_size, cur_size - 1, std::memory_order_relaxed)) {
+            return;
+        }
+    }
+}
+
 } // end of namespace doris::io