add beut

Signed-off-by: zhengyu <[email protected]>

Author: freemandealer

be/src/common/config.cpp

@@ -1163,6 +1163,7 @@ DEFINE_mBool(enable_reader_dryrun_when_download_file_cache, "true");
 DEFINE_mInt64(file_cache_background_monitor_interval_ms, "5000");
 DEFINE_mInt64(file_cache_background_ttl_gc_interval_ms, "180000");
 DEFINE_mInt64(file_cache_background_ttl_info_update_interval_ms, "180000");
+DEFINE_mInt64(file_cache_background_tablet_id_flush_interval_ms, "1000");
 DEFINE_mInt64(file_cache_background_ttl_gc_batch, "1000");
 DEFINE_mInt64(file_cache_background_lru_dump_interval_ms, "60000");
 // dump queue only if the queue update specific times through several dump intervals

be/src/common/config.h

@@ -1184,6 +1184,7 @@ DECLARE_mBool(enable_reader_dryrun_when_download_file_cache);
 DECLARE_mInt64(file_cache_background_monitor_interval_ms);
 DECLARE_mInt64(file_cache_background_ttl_gc_interval_ms);
 DECLARE_mInt64(file_cache_background_ttl_info_update_interval_ms);
+DECLARE_mInt64(file_cache_background_tablet_id_flush_interval_ms);
 DECLARE_mInt64(file_cache_background_ttl_gc_batch);
 DECLARE_Int32(file_cache_downloader_thread_num_min);
 DECLARE_Int32(file_cache_downloader_thread_num_max);

be/src/io/cache/block_file_cache_ttl_mgr.cpp

@@ -17,12 +17,12 @@
 
 #include "io/cache/block_file_cache_ttl_mgr.h"
 
-#include <atomic>
 #include <chrono>
 #include <memory>
 #include <mutex>
 #include <thread>
 #include <unordered_set>
+#include <vector>
 
 #include "common/config.h"
 #include "common/logging.h"
@@ -42,6 +42,8 @@ BlockFileCacheTtlMgr::BlockFileCacheTtlMgr(BlockFileCache* mgr, CacheBlockMetaSt
             std::thread(&BlockFileCacheTtlMgr::run_backgroud_update_ttl_info_map, this);
     _expiration_check_thread =
             std::thread(&BlockFileCacheTtlMgr::run_backgroud_expiration_check, this);
+    _tablet_id_flush_thread =
+            std::thread(&BlockFileCacheTtlMgr::run_background_tablet_id_flush, this);
 }
 
 BlockFileCacheTtlMgr::~BlockFileCacheTtlMgr() {
@@ -54,11 +56,65 @@ BlockFileCacheTtlMgr::~BlockFileCacheTtlMgr() {
     if (_expiration_check_thread.joinable()) {
         _expiration_check_thread.join();
     }
+
+    if (_tablet_id_flush_thread.joinable()) {
+        _tablet_id_flush_thread.join();
+    }
 }
 
 void BlockFileCacheTtlMgr::register_tablet_id(int64_t tablet_id) {
-    std::lock_guard<std::mutex> lock(_tablet_id_mutex);
-    _tablet_id_set.insert(tablet_id);
+    _tablet_id_queue.enqueue(tablet_id);
+}
+
+void BlockFileCacheTtlMgr::run_background_tablet_id_flush() {
+    Thread::set_self_name("ttl_mgr_flush");
+
+    static constexpr size_t kBatchSize = 1024;
+    std::vector<int64_t> pending;
+    pending.reserve(kBatchSize);
+
+    auto flush_pending = [this](std::vector<int64_t>* items) {
+        if (items->empty()) {
+            return;
+        }
+        std::lock_guard<std::mutex> lock(_tablet_id_mutex);
+        _tablet_id_set.insert(items->begin(), items->end());
+        items->clear();
+    };
+
+    while (!_stop_background.load(std::memory_order_acquire)) {
+        bool drained = false;
+        int64_t tablet_id = 0;
+        while (_tablet_id_queue.try_dequeue(tablet_id)) {
+            drained = true;
+            pending.push_back(tablet_id);
+            if (pending.size() >= kBatchSize) {
+                flush_pending(&pending);
+            }
+        }
+
+        flush_pending(&pending);
+
+        if (!drained) {
+            std::this_thread::sleep_for(std::chrono::milliseconds(
+                    config::file_cache_background_tablet_id_flush_interval_ms));
+        }
+    }
+
+    // Drain remaining items before exit
+    int64_t tablet_id = 0;
+    while (_tablet_id_queue.try_dequeue(tablet_id)) {
+        pending.push_back(tablet_id);
+        if (pending.size() >= kBatchSize) {
+            std::lock_guard<std::mutex> lock(_tablet_id_mutex);
+            _tablet_id_set.insert(pending.begin(), pending.end());
+            pending.clear();
+        }
+    }
+    if (!pending.empty()) {
+        std::lock_guard<std::mutex> lock(_tablet_id_mutex);
+        _tablet_id_set.insert(pending.begin(), pending.end());
+    }
 }
 
 FileBlocks BlockFileCacheTtlMgr::get_file_blocks_from_tablet_id(int64_t tablet_id) {
@@ -101,8 +157,6 @@ void BlockFileCacheTtlMgr::run_backgroud_update_ttl_info_map() {
     while (!_stop_background.load(std::memory_order_acquire)) {
         try {
             std::unordered_set<int64_t> tablet_ids_to_process;
-
-            // Copy tablet IDs to process
             {
                 std::lock_guard<std::mutex> lock(_tablet_id_mutex);
                 tablet_ids_to_process = _tablet_id_set;
@@ -202,7 +256,6 @@ void BlockFileCacheTtlMgr::run_backgroud_expiration_check() {
                 }
             }
 
-            // Sleep for configured interval (use existing TTL GC interval)
             std::this_thread::sleep_for(
                     std::chrono::milliseconds(config::file_cache_background_ttl_gc_interval_ms));
 

be/src/io/cache/block_file_cache_ttl_mgr.h

@@ -19,10 +19,11 @@
 
 #pragma once
 
+#include <concurrentqueue.h>
+
 #include <atomic>
 #include <map>
 #include <mutex>
-#include <set>
 #include <thread>
 #include <unordered_set>
 
@@ -50,22 +51,27 @@ class BlockFileCacheTtlMgr {
     void run_backgroud_update_ttl_info_map();
     // Background thread to find expired tablet and evict from ttl queue
     void run_backgroud_expiration_check();
+    // Background thread to drain the concurrent tablet-id queue into the dedup set
+    void run_background_tablet_id_flush();
 
 private:
     FileBlocks get_file_blocks_from_tablet_id(int64_t tablet_id);
 
 private:
-    // the set contains all the tablet ids which has cache data
-    std::unordered_set<int64_t> _tablet_id_set; //TODO(zhengyu): clean up old tablet ids
+    // Tablet ids waiting to be deduplicated + set of unique ids known to have cached data
+    moodycamel::ConcurrentQueue<int64_t> _tablet_id_queue;
+    std::unordered_set<int64_t> _tablet_id_set; // TODO(zhengyu): clean up old tablet ids
+    std::mutex _tablet_id_mutex;
+
     std::map<int64_t /* tablet_id */, TtlInfo> _ttl_info_map;
     BlockFileCache* _mgr;
     CacheBlockMetaStore* _meta_store;
 
     std::atomic<bool> _stop_background;
     std::thread _update_ttl_thread;
     std::thread _expiration_check_thread;
+    std::thread _tablet_id_flush_thread;
 
-    std::mutex _tablet_id_mutex;
     std::mutex _ttl_info_mutex;
 };
 

be/src/io/cache/cache_block_meta_store.cpp

@@ -74,6 +74,10 @@ size_t CacheBlockMetaStore::get_write_queue_size() const {
 }
 
 Status CacheBlockMetaStore::init() {
+    if (_initialized.load(std::memory_order_acquire)) {
+        return Status::OK();
+    }
+
     std::filesystem::create_directories(_db_path);
 
     _options.create_if_missing = true;
@@ -119,6 +123,7 @@ Status CacheBlockMetaStore::init() {
     }
 
     _write_thread = std::thread(&CacheBlockMetaStore::async_write_worker, this);
+    _initialized.store(true, std::memory_order_release);
 
     return Status::OK();
 }

be/src/io/cache/cache_block_meta_store.h

@@ -118,6 +118,7 @@ class CacheBlockMetaStore {
     std::unique_ptr<rocksdb::DB> _db;
     rocksdb::Options _options;
     std::unique_ptr<rocksdb::ColumnFamilyHandle> _file_cache_meta_cf_handle;
+    std::atomic<bool> _initialized {false};
 
     enum class OperationType { PUT, DELETE };
     struct WriteOperation {