[fix](filecache) Fix LRU persist queue thread-safe (#53046)

### What problem does this PR solve?

Issue Number: close #xxx

Related PR: #xxx

Problem Summary:

1.fix LRU queue crash use after free
2.fix extra LRU queue info when 'need_to_move' flag unset
3.use concurrent queueu to record queueu change info for thread safety

```
ERROR: AddressSanitizer: heap-use-after-free on address 0x603005548c40 at pc 0x55f28e8c4785 bp 0x7f603582e1f0 sp 0x7f603582e1e8
READ of size 8 at 0x603005548c40 thread T201
    #0 0x55f28e8c4784 in std::_Head_base<0ul, doris::io::CacheLRULog*, false>::_Head_base<doris::io::CacheLRULog*>(doris::io::CacheLRULog*&&) /var/local/ldb-toolchain/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/tuple:190:17
    #1 0x55f28e8c4784 in std::_Tuple_impl<0ul, doris::io::CacheLRULog*, std::default_delete<doris::io::CacheLRULog>>::_Tuple_impl(std::_Tuple_impl<0ul, doris::io::CacheLRULog*, std::default_delete<doris::io::CacheLRULog>>&&) /var/local/ldb-toolchain/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/tuple:292:2
    #2 0x55f28e8c4784 in std::tuple<doris::io::CacheLRULog*, std::default_delete<doris::io::CacheLRULog>>::tuple(std::tuple<doris::io::CacheLRULog*, std::default_delete<doris::io::CacheLRULog>>&&) /var/local/ldb-toolchain/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/tuple:1079:17
    #3 0x55f28e8c4784 in std::_uniq_ptr_impl<doris::io::CacheLRULog, std::default_delete<doris::io::CacheLRULog>>::uniq_ptr_impl(std::_uniq_ptr_impl<doris::io::CacheLRULog, std::default_delete<doris::io::CacheLRULog>>&&) /var/local/ldb-toolchain/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/bits/unique_ptr.h:162:9
    #4 0x55f28e8c4784 in std::_uniq_ptr_data<doris::io::CacheLRULog, std::default_delete<doris::io::CacheLRULog>, true, true>::uniq_ptr_data(std::_uniq_ptr_data<doris::io::CacheLRULog, std::default_delete<doris::io::CacheLRULog>, true, true>&&) /var/local/ldb-toolchain/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/bits/unique_ptr.h:211:7
    #5 0x55f28e8c4784 in std::unique_ptr<doris::io::CacheLRULog, std::default_delete<doris::io::CacheLRULog>>::unique_ptr(std::unique_ptr<doris::io::CacheLRULog, std::default_delete<doris::io::CacheLRULog>>&&) /var/local/ldb-toolchain/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/bits/unique_ptr.h:327:7
    #6 0x55f28e8c4784 in doris::io::LRUQueueRecorder::replay_queue_event(doris::io::FileCacheType) /root/doris/be/src/io/cache/lru_queue_recorder.cpp:40:20
    #7 0x55f28e82d620 in doris::io::BlockFileCache::run_background_lru_log_replay() /root/doris/be/src/io/cache/block_file_cache.cpp:2242:24
    #8 0x55f2cdc2720f in execute_native_thread_routine /data/gcc-11.1.0/build/x86_64-pc-linux-gnu/libstdc+-v3/src/c11/../../../../../libstdc-v3/src/c+11/thread.cc:82:18
    #9 0x7f61f1842608 in start_thread /build/glibc-SzIz7B/glibc-2.31/nptl/pthread_create.c:477:8
    #10 0x7f61f1aef132 in __clone /build/glibc-SzIz7B/glibc-2.31/misc/../sysdeps/unix/sysv/linux/x86_64/clone.S:95

0x603005548c40 is located 16 bytes inside of 24-byte region [0x603005548c30,0x603005548c48)
freed by thread T201 here:
    #0 0x55f28e51680d in operator delete(void*) (/home/work/unlimit_teamcity/TeamCity/Agents/20250708205944agent_172.16.0.48_1/work/60183217f6ee2a9c/output/be/lib/doris_be+0x3975a80d) (BuildId: 8b6ba6101e736655)
    #1 0x55f28e8c3ce0 in std::__cxx11::list<std::unique_ptr<doris::io::CacheLRULog, std::default_delete<doris::io::CacheLRULog>>, std::allocator<std::unique_ptr<doris::io::CacheLRULog, std::default_delete<doris::io::CacheLRULog>>>>::pop_front() /var/local/ldb-toolchain/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/bits/stl_list.h:1198:15
    #2 0x55f28e8c3ce0 in doris::io::LRUQueueRecorder::replay_queue_event(doris::io::FileCacheType) /root/doris/be/src/io/cache/lru_queue_recorder.cpp:41:19
    #3 0x55f28e82d620 in doris::io::BlockFileCache::run_background_lru_log_replay() /root/doris/be/src/io/cache/block_file_cache.cpp:2242:24
    #4 0x55f2cdc2720f in execute_native_thread_routine /data/gcc-11.1.0/build/x86_64-pc-linux-gnu/libstdc+-v3/src/c11/../../../../../libstdc-v3/src/c+11/thread.cc:82:18

previously allocated by thread T607 (CumuCompactionT) here:
    #0 0x55f28e515fad in operator new(unsigned long) (/home/work/unlimit_teamcity/TeamCity/Agents/20250708205944agent_172.16.0.48_1/work/60183217f6ee2a9c/output/be/lib/doris_be+0x39759fad) (BuildId: 8b6ba6101e736655)
    #1 0x55f28e8c660d in __gnu_cxx::new_allocator<std::_List_node<std::unique_ptr<doris::io::CacheLRULog, std::default_delete<doris::io::CacheLRULog>>>>::allocate(unsigned long, void const*) /var/local/ldb-toolchain/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/ext/new_allocator.h:121:27
    #2 0x55f28e8c660d in std::allocator<std::_List_node<std::unique_ptr<doris::io::CacheLRULog, std::default_delete<doris::io::CacheLRULog>>>>::allocate(unsigned long) /var/local/ldb-toolchain/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/bits/allocator.h:173:32
    #3 0x55f28e8c660d in std::allocator_traits<std::allocator<std::_List_node<std::unique_ptr<doris::io::CacheLRULog, std::default_delete<doris::io::CacheLRULog>>>>>::allocate(std::allocator<std::_List_node<std::unique_ptr<doris::io::CacheLRULog, std::default_delete<doris::io::CacheLRULog>>>>&, unsigned long) /var/local/ldb-toolchain/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/bits/alloc_traits.h:460:20
    #4 0x55f28e8c660d in std::__cxx11::_List_base<std::unique_ptr<doris::io::CacheLRULog, std::default_delete<doris::io::CacheLRULog>>, std::allocator<std::unique_ptr<doris::io::CacheLRULog, std::default_delete<doris::io::CacheLRULog>>>>::_M_get_node() /var/local/ldb-toolchain/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/bits/stl_list.h:442:16
    #5 0x55f28e8c660d in std::List_node<std::unique_ptr<doris::io::CacheLRULog, std::default_delete<doris::io::CacheLRULog>>>* std::_cxx11::list<std::unique_ptr<doris::io::CacheLRULog, std::default_delete<doris::io::CacheLRULog>>, std::allocator<std::unique_ptr<doris::io::CacheLRULog, std::default_delete<doris::io::CacheLRULog>>>>::_M_create_node<std::unique_ptr<doris::io::CacheLRULog, std::default_delete<doris::io::CacheLRULog>>>(std::unique_ptr<doris::io::CacheLRULog, std::default_delete<doris::io::CacheLRULog>>&&) /var/local/ldb-toolchain/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/bits/stl_list.h:634:21
    #6 0x55f28e8c660d in void std::__cxx11::list<std::unique_ptr<doris::io::CacheLRULog, std::default_delete<doris::io::CacheLRULog>>, std::allocator<std::unique_ptr<doris::io::CacheLRULog, std::default_delete<doris::io::CacheLRULog>>>>::_M_insert<std::unique_ptr<doris::io::CacheLRULog, std::default_delete<doris::io::CacheLRULog>>>(std::_List_iterator<std::unique_ptr<doris::io::CacheLRULog, std::default_delete<doris::io::CacheLRULog>>>, std::unique_ptr<doris::io::CacheLRULog, std::default_delete<doris::io::CacheLRULog>>&&) /var/local/ldb-toolchain/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/bits/stl_list.h:1911:18
    #7 0x55f28e8c3522 in std::__cxx11::list<std::unique_ptr<doris::io::CacheLRULog, std::default_delete<doris::io::CacheLRULog>>, std::allocator<std::unique_ptr<doris::io::CacheLRULog, std::default_delete<doris::io::CacheLRULog>>>>::push_back(std::unique_ptr<doris::io::CacheLRULog, std::default_delete<doris::io::CacheLRULog>>&&) /var/local/ldb-toolchain/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/bits/stl_list.h:1217:15
    #8 0x55f28e8c3522 in doris::io::LRUQueueRecorder::record_queue_event(doris::io::FileCacheType, doris::io::CacheLRULogType, doris::io::UInt128Wrapper, unsigned long, unsigned long) /root/doris/be/src/io/cache/lru_queue_recorder.cpp:29:15
    #9 0x55f28e82f09b in doris::io::BlockFileCache::use_cell(doris::io::BlockFileCache::FileBlockCell const&, std::__cxx11::list<std::shared_ptr<doris::io::FileBlock>, std::allocator<std::shared_ptr<doris::io::FileBlock>>>*, bool, std::lock_guard<std::mutex>&) /root/doris/be/src/io/cache/block_file_cache.cpp:380:20
    #10 0x55f28e833d1b in doris::io::BlockFileCache::get_impl[abi:cxx11](doris::io::UInt128Wrapper const&, doris::io::CacheContext const&, doris::io::FileBlock::Range const&, std::lock_guard<std::mutex>&) /root/doris/be/src/io/cache/block_file_cache.cpp:572:13
    #11 0x55f28e83b4ef in doris::io::BlockFileCache::get_or_set(doris::io::UInt128Wrapper const&, unsigned long, unsigned long, doris::io::CacheContext&) /root/doris/be/src/io/cache/block_file_cache.cpp:762:27
    #12 0x55f28e7ffcee in doris::io::CachedRemoteFileReader::read_at_impl(unsigned long, doris::Slice, unsigned long*, doris::io::IOContext const*) /root/doris/be/src/io/cache/cached_remote_file_reader.cpp:191:21
    #13 0x55f28e7f8017 in doris::io::FileReader::read_at(unsigned long, doris::Slice, unsigned long*, doris::io::IOContext const*) /root/doris/be/src/io/fs/file_reader.cpp:34:17
```

### Release note

None

### Check List (For Author)

- Test <!-- At least one of them must be included. -->
    - [ ] Regression test
    - [ ] Unit Test
    - [ ] Manual test (add detailed scripts or steps below)
    - [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
        - [x] Previous test can cover this change.
        - [ ] No code files have been changed.
        - [ ] Other reason <!-- Add your reason?  -->

- Behavior changed:
    - [x] No.
    - [ ] Yes. <!-- Explain the behavior change -->

- Does this need documentation?
    - [x] No.
- [ ] Yes. <!-- Add document PR link here. eg:
apache/doris-website#1214 -->

### Check List (For Reviewer who merge this PR)

- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label <!-- Add branch pick label that this PR
should merge into -->

---------

Signed-off-by: zhengyu <zhangzhengyu@selectdb.com>

Author: freemandealer

be/src/io/cache/block_file_cache.cpp

@@ -376,10 +376,10 @@ void BlockFileCache::use_cell(const FileBlockCell& cell, FileBlocks* result, boo
     /// Move to the end of the queue. The iterator remains valid.
     if (cell.queue_iterator && move_iter_flag) {
         queue.move_to_end(*cell.queue_iterator, cache_lock);
+        _lru_recorder->record_queue_event(cell.file_block->cache_type(),
+                                          CacheLRULogType::MOVETOBACK, cell.file_block->_key.hash,
+                                          cell.file_block->_key.offset, cell.size());
     }
-    _lru_recorder->record_queue_event(cell.file_block->cache_type(), CacheLRULogType::MOVETOBACK,
-                                      cell.file_block->_key.hash, cell.file_block->_key.offset,
-                                      cell.size());
 
     cell.update_atime();
 }
@@ -1546,7 +1546,11 @@ bool LRUQueue::contains(const UInt128Wrapper& hash, size_t offset,
 
 LRUQueue::Iterator LRUQueue::get(const UInt128Wrapper& hash, size_t offset,
                                  std::lock_guard<std::mutex>& /* cache_lock */) const {
-    return map.find(std::make_pair(hash, offset))->second;
+    auto itr = map.find(std::make_pair(hash, offset));
+    if (itr != map.end()) {
+        return itr->second;
+    }
+    return std::list<FileKeyAndOffset>::iterator();
 }
 
 std::string LRUQueue::to_string(std::lock_guard<std::mutex>& /* cache_lock */) const {

be/src/io/cache/lru_queue_recorder.cpp

@@ -26,7 +26,7 @@ void LRUQueueRecorder::record_queue_event(FileCacheType type, CacheLRULogType lo
                                           const UInt128Wrapper hash, const size_t offset,
                                           const size_t size) {
     CacheLRULogQueue& log_queue = get_lru_log_queue(type);
-    log_queue.push_back(std::make_unique<CacheLRULog>(log_type, hash, offset, size));
+    log_queue.enqueue(std::make_unique<CacheLRULog>(log_type, hash, offset, size));
     ++(_lru_queue_update_cnt_from_last_dump[type]);
 }
 
@@ -36,9 +36,8 @@ void LRUQueueRecorder::replay_queue_event(FileCacheType type) {
     LRUQueue& shadow_queue = get_shadow_queue(type);
 
     std::lock_guard<std::mutex> lru_log_lock(_mutex_lru_log);
-    while (!log_queue.empty()) {
-        auto log = std::move(log_queue.front());
-        log_queue.pop_front();
+    std::unique_ptr<CacheLRULog> log;
+    while (log_queue.try_dequeue(log)) {
         try {
             switch (log->type) {
             case CacheLRULogType::ADD: {
@@ -47,16 +46,17 @@ void LRUQueueRecorder::replay_queue_event(FileCacheType type) {
             }
             case CacheLRULogType::REMOVE: {
                 auto it = shadow_queue.get(log->hash, log->offset, lru_log_lock);
-                if (it != shadow_queue.end()) {
+                if (it != std::list<LRUQueue::FileKeyAndOffset>::iterator()) {
                     shadow_queue.remove(it, lru_log_lock);
                 } else {
                     LOG(WARNING) << "REMOVE failed, doesn't exist in shadow queue";
                 }
                 break;
             }
             case CacheLRULogType::MOVETOBACK: {
+                LOG(INFO) << "MOVETOBACK" << log->hash.to_string() << " " << log->offset;
                 auto it = shadow_queue.get(log->hash, log->offset, lru_log_lock);
-                if (it != shadow_queue.end()) {
+                if (it != std::list<LRUQueue::FileKeyAndOffset>::iterator()) {
                     shadow_queue.move_to_end(it, lru_log_lock);
                 } else {
                     LOG(WARNING) << "MOVETOBACK failed, doesn't exist in shadow queue";

be/src/io/cache/lru_queue_recorder.h

@@ -17,6 +17,10 @@
 
 #pragma once
 
+#include <concurrentqueue.h>
+
+#include <boost/lockfree/spsc_queue.hpp>
+
 #include "io/cache/file_cache_common.h"
 
 namespace doris::io {
@@ -40,7 +44,7 @@ struct CacheLRULog {
             : type(t), hash(h), offset(o), size(s) {}
 };
 
-using CacheLRULogQueue = std::list<std::unique_ptr<CacheLRULog>>;
+using CacheLRULogQueue = moodycamel::ConcurrentQueue<std::unique_ptr<CacheLRULog>>;
 
 class LRUQueueRecorder {
 public:

be/test/io/cache/block_file_cache_test.cpp

@@ -3909,6 +3909,7 @@ TEST_F(BlockFileCacheTest, cached_remote_file_reader_error_handle) {
     }
     EXPECT_TRUE(reader.close().ok());
     EXPECT_TRUE(reader.closed());
+    std::this_thread::sleep_for(std::chrono::seconds(1));
     if (fs::exists(cache_base_path)) {
         fs::remove_all(cache_base_path);
     }
@@ -3969,6 +3970,7 @@ TEST_F(BlockFileCacheTest, cached_remote_file_reader_init) {
         CachedRemoteFileReader reader(local_reader, opts);
         EXPECT_EQ(reader._cache->get_base_path(), cache_base_path);
     }
+    std::this_thread::sleep_for(std::chrono::seconds(1));
     if (fs::exists(cache_base_path)) {
         fs::remove_all(cache_base_path);
     }

be/test/io/cache/block_file_cache_test_lru_dump.cpp

@@ -156,10 +156,10 @@ TEST_F(BlockFileCacheTest, test_lru_log_record_replay_dump_restore) {
     ASSERT_EQ(cache.get_stats_unsafe()["normal_queue_curr_size"], 500000);
 
     // all queue are filled, let's check the lru log records
-    ASSERT_EQ(cache._lru_recorder->_ttl_lru_log_queue.size(), 5);
-    ASSERT_EQ(cache._lru_recorder->_index_lru_log_queue.size(), 5);
-    ASSERT_EQ(cache._lru_recorder->_normal_lru_log_queue.size(), 5);
-    ASSERT_EQ(cache._lru_recorder->_disposable_lru_log_queue.size(), 5);
+    ASSERT_EQ(cache._lru_recorder->_ttl_lru_log_queue.size_approx(), 5);
+    ASSERT_EQ(cache._lru_recorder->_index_lru_log_queue.size_approx(), 5);
+    ASSERT_EQ(cache._lru_recorder->_normal_lru_log_queue.size_approx(), 5);
+    ASSERT_EQ(cache._lru_recorder->_disposable_lru_log_queue.size_approx(), 5);
 
     // then check the log replay
     std::this_thread::sleep_for(std::chrono::milliseconds(
@@ -175,10 +175,10 @@ TEST_F(BlockFileCacheTest, test_lru_log_record_replay_dump_restore) {
                                        context2); // move index queue 3rd element to the end
         cache.remove_if_cached(key3);             // remove all element from ttl queue
     }
-    ASSERT_EQ(cache._lru_recorder->_ttl_lru_log_queue.size(), 5);
-    ASSERT_EQ(cache._lru_recorder->_index_lru_log_queue.size(), 1);
-    ASSERT_EQ(cache._lru_recorder->_normal_lru_log_queue.size(), 0);
-    ASSERT_EQ(cache._lru_recorder->_disposable_lru_log_queue.size(), 0);
+    ASSERT_EQ(cache._lru_recorder->_ttl_lru_log_queue.size_approx(), 5);
+    ASSERT_EQ(cache._lru_recorder->_index_lru_log_queue.size_approx(), 1);
+    ASSERT_EQ(cache._lru_recorder->_normal_lru_log_queue.size_approx(), 0);
+    ASSERT_EQ(cache._lru_recorder->_disposable_lru_log_queue.size_approx(), 0);
 
     std::this_thread::sleep_for(std::chrono::milliseconds(
             2 * config::file_cache_background_lru_log_replay_interval_ms));