Merge pull request ClickHouse#87979 from ClickHouse/backport/25.8/87914

Backport ClickHouse#87914 to 25.8: fs cache: reuse cache priority iterator among threads concurrently reserving space in cache

Author: clickhouse-gh[bot]

src/Common/CurrentMetrics.cpp

@@ -299,6 +299,7 @@
     M(FilesystemCacheDelayedCleanupElements, "Filesystem cache elements in background cleanup queue") \
     M(FilesystemCacheHoldFileSegments, "Filesystem cache file segment which are currently hold as unreleasable") \
     M(FilesystemCacheKeys, "Number of keys in filesystem cache") \
+    M(FilesystemCacheReserveThreads, "Threads number trying to reserve space in cache") \
     M(AsyncInsertCacheSize, "Number of async insert hash id in cache") \
     M(IcebergMetadataFilesCacheBytes, "Size of the iceberg metadata cache in bytes") \
     M(IcebergMetadataFilesCacheFiles, "Number of cached files in the iceberg metadata cache") \

src/Common/ProfileEvents.cpp

@@ -668,13 +668,17 @@ The server successfully detected this situation and will download merged part fr
     M(FilesystemCacheEvictionSkippedFileSegments, "Number of file segments skipped for eviction because of being in unreleasable state", ValueType::Number) \
     M(FilesystemCacheEvictionSkippedEvictingFileSegments, "Number of file segments skipped for eviction because of being in evicting state", ValueType::Number) \
     M(FilesystemCacheEvictionTries, "Number of filesystem cache eviction attempts", ValueType::Number) \
+    M(FilesystemCacheEvictionReusedIterator, "Number of filesystem cache iterator reusing", ValueType::Number) \
     M(FilesystemCacheLockKeyMicroseconds, "Lock cache key time", ValueType::Microseconds) \
     M(FilesystemCacheLockMetadataMicroseconds, "Lock filesystem cache metadata time", ValueType::Microseconds) \
     M(FilesystemCacheLockCacheMicroseconds, "Lock filesystem cache time", ValueType::Microseconds) \
     M(FilesystemCacheReserveMicroseconds, "Filesystem cache space reservation time", ValueType::Microseconds) \
+    M(FilesystemCacheReserveAttempts, "Filesystem cache space reservation attempt", ValueType::Number) \
     M(FilesystemCacheEvictMicroseconds, "Filesystem cache eviction time", ValueType::Microseconds) \
     M(FilesystemCacheGetOrSetMicroseconds, "Filesystem cache getOrSet() time", ValueType::Microseconds) \
     M(FilesystemCacheGetMicroseconds, "Filesystem cache get() time", ValueType::Microseconds) \
+    M(FilesystemCacheBackgroundEvictedFileSegments, "Number of file segments evicted by background thread", ValueType::Number) \
+    M(FilesystemCacheBackgroundEvictedBytes, "Number of bytes evicted by background thread", ValueType::Number) \
     M(FileSegmentWaitMicroseconds, "Wait on DOWNLOADING state", ValueType::Microseconds) \
     M(FileSegmentCompleteMicroseconds, "Duration of FileSegment::complete() in filesystem cache", ValueType::Microseconds) \
     M(FileSegmentLockMicroseconds, "Lock file segment time", ValueType::Microseconds) \

src/Interpreters/Cache/EvictionCandidates.cpp

@@ -86,6 +86,7 @@ void EvictionCandidates::add(
     it->second.candidates.push_back(candidate);
     candidate->setEvictingFlag(locked_key, lock);
     ++candidates_size;
+    candidates_bytes += candidate->size();
 }
 
 void EvictionCandidates::removeQueueEntries(const CachePriorityGuard::Lock & lock)

src/Interpreters/Cache/EvictionCandidates.h

@@ -58,6 +58,7 @@ class EvictionCandidates : private boost::noncopyable
     FailedCandidates getFailedCandidates() const { return failed_candidates; }
 
     size_t size() const { return candidates_size; }
+    size_t bytes() const { return candidates_bytes; }
 
     auto begin() const { return candidates.begin(); }
 
@@ -73,6 +74,7 @@ class EvictionCandidates : private boost::noncopyable
 
     std::unordered_map<FileCacheKey, KeyCandidates> candidates;
     size_t candidates_size = 0;
+    size_t candidates_bytes = 0;
     FailedCandidates failed_candidates;
 
     std::vector<FinalizeEvictionFunc> on_finalize;

src/Interpreters/Cache/FileCache.cpp

@@ -31,17 +31,21 @@ namespace ProfileEvents
     extern const Event FilesystemCacheLoadMetadataMicroseconds;
     extern const Event FilesystemCacheLockCacheMicroseconds;
     extern const Event FilesystemCacheReserveMicroseconds;
+    extern const Event FilesystemCacheReserveAttempts;
     extern const Event FilesystemCacheGetOrSetMicroseconds;
     extern const Event FilesystemCacheGetMicroseconds;
     extern const Event FilesystemCacheFailToReserveSpaceBecauseOfLockContention;
     extern const Event FilesystemCacheFreeSpaceKeepingThreadRun;
     extern const Event FilesystemCacheFreeSpaceKeepingThreadWorkMilliseconds;
     extern const Event FilesystemCacheFailToReserveSpaceBecauseOfCacheResize;
+    extern const Event FilesystemCacheBackgroundEvictedFileSegments;
+    extern const Event FilesystemCacheBackgroundEvictedBytes;
 }
 
 namespace CurrentMetrics
 {
     extern const Metric FilesystemCacheDownloadQueueElements;
+    extern const Metric FilesystemCacheReserveThreads;
 }
 
 namespace DB
@@ -941,7 +945,9 @@ bool FileCache::tryReserve(
     size_t lock_wait_timeout_milliseconds,
     std::string & failure_reason)
 {
+    CurrentMetrics::Increment increment(CurrentMetrics::FilesystemCacheReserveThreads);
     ProfileEventTimeIncrement<Microseconds> watch(ProfileEvents::FilesystemCacheReserveMicroseconds);
+    ProfileEvents::increment(ProfileEvents::FilesystemCacheReserveAttempts);
 
     assertInitialized();
 
@@ -956,6 +962,11 @@ bool FileCache::tryReserve(
         return false;
     }
 
+    cache_reserve_active_threads.fetch_add(1, std::memory_order_relaxed);
+    SCOPE_EXIT({
+        cache_reserve_active_threads.fetch_sub(1, std::memory_order_relaxed);
+    });
+
     auto cache_lock = tryLockCache(std::chrono::milliseconds(lock_wait_timeout_milliseconds));
     if (!cache_lock)
     {
@@ -1021,7 +1032,14 @@ bool FileCache::tryReserve(
     if (query_priority)
     {
         if (!query_priority->collectCandidatesForEviction(
-                size, required_elements_num, reserve_stat, eviction_candidates, {}, user.user_id, cache_lock))
+                size,
+                required_elements_num,
+                reserve_stat,
+                eviction_candidates,
+                {},
+                /* continue_from_last_eviction_pos */false,
+                user.user_id,
+                cache_lock))
         {
             const auto & stat = reserve_stat.total_stat;
             failure_reason = fmt::format(
@@ -1041,8 +1059,19 @@ bool FileCache::tryReserve(
         reserve_stat.stat_by_kind.clear();
     }
 
+    bool continue_from_last_eviction_pos = cache_reserve_active_threads.load(std::memory_order_relaxed) > 1;
+    if (!continue_from_last_eviction_pos)
+        main_priority->resetEvictionPos(cache_lock);
+
     if (!main_priority->collectCandidatesForEviction(
-            size, required_elements_num, reserve_stat, eviction_candidates, queue_iterator, user.user_id, cache_lock))
+            size,
+            required_elements_num,
+            reserve_stat,
+            eviction_candidates,
+            queue_iterator,
+            continue_from_last_eviction_pos,
+            user.user_id,
+            cache_lock))
     {
         const auto & stat = reserve_stat.total_stat;
         failure_reason = fmt::format(
@@ -1130,6 +1159,11 @@ bool FileCache::tryReserve(
     if (main_priority->getSize(cache_lock) > (1ull << 63))
         throw Exception(ErrorCodes::LOGICAL_ERROR, "Cache became inconsistent. There must be a bug");
 
+    if (cache_reserve_active_threads.load(std::memory_order_relaxed) == 1)
+    {
+        main_priority->resetEvictionPos(cache_lock);
+    }
+
     cache_lock.unlock();
     if (!file_segment.getKeyMetadata()->createBaseDirectory())
     {
@@ -1187,7 +1221,12 @@ void FileCache::freeSpaceRatioKeepingThreadFunc()
         /// (we use batches to make sure we do not block cache for too long,
         /// by default the batch size is quite small).
         desired_size_status = main_priority->collectCandidatesForEviction(
-            desired_size, desired_elements_num, keep_up_free_space_remove_batch, stat, eviction_candidates, lock);
+            desired_size,
+            desired_elements_num,
+            keep_up_free_space_remove_batch,
+            stat,
+            eviction_candidates,
+            lock);
 
 #ifdef DEBUG_OR_SANITIZER_BUILD
         /// Let's make sure that we correctly processed the limits.
@@ -1229,6 +1268,9 @@ void FileCache::freeSpaceRatioKeepingThreadFunc()
             /// e.g. to update the in-memory state.
             lock.lock();
             eviction_candidates.finalize(nullptr, lock);
+
+            ProfileEvents::increment(ProfileEvents::FilesystemCacheBackgroundEvictedFileSegments, eviction_candidates.size());
+            ProfileEvents::increment(ProfileEvents::FilesystemCacheBackgroundEvictedBytes, eviction_candidates.bytes());
         }
     }
     catch (...)

src/Interpreters/Cache/FileCache.h

@@ -238,6 +238,8 @@ class FileCache : private boost::noncopyable
     std::atomic<bool> shutdown = false;
     std::atomic<bool> cache_is_being_resized = false;
 
+    std::atomic<size_t> cache_reserve_active_threads = 0;
+
     std::mutex apply_settings_mutex;
 
     CacheMetadata metadata;

src/Interpreters/Cache/IFileCachePriority.h

@@ -159,9 +159,12 @@ class IFileCachePriority : private boost::noncopyable
         FileCacheReserveStat & stat,
         EvictionCandidates & res,
         IteratorPtr reservee,
+        bool continue_from_last_eviction_pos,
         const UserID & user_id,
         const CachePriorityGuard::Lock &) = 0;
 
+    virtual void resetEvictionPos(const CachePriorityGuard::Lock & lock) = 0;
+
     /// Collect eviction candidates sufficient to have `desired_size`
     /// and `desired_elements_num` as current cache state.
     /// Collect no more than `max_candidates_to_evict` elements.

src/Interpreters/Cache/LRUFileCachePriority.cpp

@@ -18,6 +18,7 @@ namespace ProfileEvents
     extern const Event FilesystemCacheEvictionSkippedFileSegments;
     extern const Event FilesystemCacheEvictionTries;
     extern const Event FilesystemCacheEvictionSkippedEvictingFileSegments;
+    extern const Event FilesystemCacheEvictionReusedIterator;
 }
 
 namespace DB
@@ -36,6 +37,7 @@ LRUFileCachePriority::LRUFileCachePriority(
     : IFileCachePriority(max_size_, max_elements_)
     , description(description_)
     , log(getLogger("LRUFileCachePriority" + (description.empty() ? "" : "(" + description + ")")))
+    , eviction_pos(queue.end())
 {
     if (state_)
         state = state_;
@@ -100,6 +102,13 @@ LRUFileCachePriority::LRUIterator LRUFileCachePriority::add(EntryPtr entry, cons
     return LRUIterator(this, iterator);
 }
 
+void LRUFileCachePriority::increasePriority(LRUQueue::iterator it, const CachePriorityGuard::Lock &)
+{
+    if (eviction_pos == it)
+        eviction_pos = std::next(it);
+    queue.splice(queue.end(), queue, it);
+}
+
 LRUFileCachePriority::LRUQueue::iterator
 LRUFileCachePriority::remove(LRUQueue::iterator it, const CachePriorityGuard::Lock &)
 {
@@ -115,6 +124,8 @@ LRUFileCachePriority::remove(LRUQueue::iterator it, const CachePriorityGuard::Lo
         log, "Removed entry from LRU queue, key: {}, offset: {}, size: {}",
         entry.key, entry.offset, entry.size.load());
 
+    if (eviction_pos == it)
+        eviction_pos = std::next(it);
     return queue.erase(it);
 }
 
@@ -166,8 +177,20 @@ bool LRUFileCachePriority::LRUIterator::operator ==(const LRUIterator & other) c
 
 void LRUFileCachePriority::iterate(IterateFunc func, const CachePriorityGuard::Lock & lock)
 {
-    for (auto it = queue.begin(); it != queue.end();)
+    iterateImpl(queue.begin(), func, lock);
+}
+
+LRUFileCachePriority::LRUQueue::iterator
+LRUFileCachePriority::iterateImpl(LRUQueue::iterator start_pos, IterateFunc func, const CachePriorityGuard::Lock & lock)
+{
+    const size_t max_elements_to_iterate = queue.size();
+    auto it = start_pos;
+
+    for (size_t iterated_elements = 0; iterated_elements < max_elements_to_iterate; ++iterated_elements)
     {
+        if (it == queue.end())
+            it = queue.begin();
+
         const auto & entry = **it;
 
         if (entry.size == 0)
@@ -223,7 +246,7 @@ void LRUFileCachePriority::iterate(IterateFunc func, const CachePriorityGuard::L
         {
             case IterationResult::BREAK:
             {
-                return;
+                return it;
             }
             case IterationResult::CONTINUE:
             {
@@ -237,6 +260,7 @@ void LRUFileCachePriority::iterate(IterateFunc func, const CachePriorityGuard::L
             }
         }
     }
+    return queue.end();
 }
 
 bool LRUFileCachePriority::canFit( /// NOLINT
@@ -270,6 +294,7 @@ bool LRUFileCachePriority::collectCandidatesForEviction(
     FileCacheReserveStat & stat,
     EvictionCandidates & res,
     IFileCachePriority::IteratorPtr /* reservee */,
+    bool continue_from_last_eviction_pos,
     const UserID &,
     const CachePriorityGuard::Lock & lock)
 {
@@ -283,7 +308,7 @@ bool LRUFileCachePriority::collectCandidatesForEviction(
         return canFit(size, elements, stat.total_stat.releasable_size, stat.total_stat.releasable_count, lock);
     };
 
-    iterateForEviction(res, stat, can_fit, lock);
+    iterateForEviction(res, stat, can_fit, continue_from_last_eviction_pos, lock);
 
     if (can_fit())
     {
@@ -349,7 +374,7 @@ IFileCachePriority::CollectStatus LRUFileCachePriority::collectCandidatesForEvic
         }
         return false;
     };
-    iterateForEviction(res, stat, stop_condition, lock);
+    iterateForEviction(res, stat, stop_condition, false, lock);
     chassert(status != CollectStatus::SUCCESS || stop_condition());
     return status;
 }
@@ -358,14 +383,15 @@ void LRUFileCachePriority::iterateForEviction(
     EvictionCandidates & res,
     FileCacheReserveStat & stat,
     StopConditionFunc stop_condition,
+    bool continue_from_last_eviction_pos,
     const CachePriorityGuard::Lock & lock)
 {
     if (stop_condition())
         return;
 
     ProfileEvents::increment(ProfileEvents::FilesystemCacheEvictionTries);
 
-    IterateFunc iterate_func = [&](LockedKey & locked_key, const FileSegmentMetadataPtr & segment_metadata)
+    auto iterate_func = [&](LockedKey & locked_key, const FileSegmentMetadataPtr & segment_metadata)
     {
         const auto & file_segment = segment_metadata->file_segment;
         chassert(file_segment->assertCorrectness());
@@ -380,14 +406,27 @@ void LRUFileCachePriority::iterateForEviction(
             ProfileEvents::increment(ProfileEvents::FilesystemCacheEvictionSkippedFileSegments);
             stat.update(segment_metadata->size(), file_segment->getKind(), false);
         }
-
-        return IterationResult::CONTINUE;
     };
 
-    iterate([&](LockedKey & locked_key, const FileSegmentMetadataPtr & segment_metadata)
+    auto start_pos = queue.begin();
+    if (continue_from_last_eviction_pos && eviction_pos != queue.end() && start_pos != eviction_pos)
     {
-        return stop_condition() ? IterationResult::BREAK : iterate_func(locked_key, segment_metadata);
+        ProfileEvents::increment(ProfileEvents::FilesystemCacheEvictionReusedIterator);
+        start_pos = eviction_pos;
+    }
+
+    auto iteration_pos = iterateImpl(
+        start_pos,
+        [&](LockedKey & locked_key, const FileSegmentMetadataPtr & segment_metadata)
+    {
+        if (stop_condition())
+            return IterationResult::BREAK;
+        iterate_func(locked_key, segment_metadata);
+        return stop_condition() ? IterationResult::BREAK : IterationResult::CONTINUE;
     }, lock);
+
+    if (continue_from_last_eviction_pos)
+        eviction_pos = iteration_pos;
 }
 
 LRUFileCachePriority::LRUIterator LRUFileCachePriority::move(
@@ -415,6 +454,8 @@ LRUFileCachePriority::LRUIterator LRUFileCachePriority::move(
     }
 #endif
 
+    if (other.eviction_pos == it.iterator)
+        other.eviction_pos = std::next(it.iterator);
     queue.splice(queue.end(), other.queue, it.iterator);
 
     updateSize(entry.size);
@@ -534,7 +575,7 @@ void LRUFileCachePriority::LRUIterator::decrementSize(size_t size)
 size_t LRUFileCachePriority::LRUIterator::increasePriority(const CachePriorityGuard::Lock & lock)
 {
     assertValid();
-    cache_priority->queue.splice(cache_priority->queue.end(), cache_priority->queue, iterator);
+    cache_priority->increasePriority(iterator, lock);
     cache_priority->check(lock);
     return ++((*iterator)->hits);
 }
@@ -547,6 +588,7 @@ void LRUFileCachePriority::LRUIterator::assertValid() const
 
 void LRUFileCachePriority::shuffle(const CachePriorityGuard::Lock &)
 {
+    chassert(eviction_pos == queue.end());
     std::vector<LRUQueue::iterator> its;
     its.reserve(queue.size());
     for (auto it = queue.begin(); it != queue.end(); ++it)