Fix false positive lock order inversion

src/Interpreters/Cache/FileCache.cpp

@@ -129,14 +129,14 @@ void FileCache::initialize()
         throw;
     }
 
-    is_initialized = true;
-
     for (size_t i = 0; i < background_download_threads; ++i)
          download_threads.emplace_back([this] { metadata.downloadThreadFunc(); });
 
     cleanup_task = Context::getGlobalContextInstance()->getSchedulePool().createTask("FileCacheCleanup", [this]{ cleanupThreadFunc(); });
     cleanup_task->activate();
     cleanup_task->scheduleAfter(delayed_cleanup_interval_ms);
+
+    is_initialized = true;
 }
 
 CacheGuard::Lock FileCache::lockCache() const
@@ -973,7 +973,10 @@ void FileCache::loadMetadataForKeys(const fs::path & keys_dir)
         }
 
         const auto key = Key(unhexUInt<UInt128>(key_directory.filename().string().data()));
-        auto locked_key = metadata.lockKeyMetadata(key, CacheMetadata::KeyNotFoundPolicy::CREATE_EMPTY, /* is_initial_load */true);
+        auto key_metadata = metadata.getKeyMetadata(key, CacheMetadata::KeyNotFoundPolicy::CREATE_EMPTY, /* is_initial_load */true);
+
+        const size_t size_limit = main_priority->getSizeLimit();
+        const size_t elements_limit = main_priority->getElementsLimit();
 
         for (fs::directory_iterator offset_it{key_directory}; offset_it != fs::directory_iterator(); ++offset_it)
         {
@@ -1017,13 +1020,16 @@ void FileCache::loadMetadataForKeys(const fs::path & keys_dir)
             IFileCachePriority::Iterator cache_it;
             {
                 auto lock = lockCache();
-                limits_satisfied = (main_priority->getSizeLimit() == 0
-                                            || main_priority->getSize(lock) + size <= main_priority->getSizeLimit())
-                    && (main_priority->getElementsLimit() == 0
-                        || main_priority->getElementsCount(lock) + 1 <= main_priority->getElementsLimit());
+                limits_satisfied = (size_limit == 0 || main_priority->getSize(lock) + size <= size_limit)
+                    && (elements_limit == 0 || main_priority->getElementsCount(lock) + 1 <= elements_limit);
 
                 if (limits_satisfied)
-                    cache_it = main_priority->add(locked_key->getKeyMetadata(), offset, size, lock);
+                    cache_it = main_priority->add(key_metadata, offset, size, lock);
+
+                /// TODO: we can get rid of this lockCache() if we first load everything in parallel
+                /// without any mutual lock between loading threads, and only after do removeOverflow().
+                /// This will be better because overflow here may
+                /// happen only if cache configuration changed and max_size because less than it was.
             }
 
             if (limits_satisfied)
@@ -1032,7 +1038,7 @@ void FileCache::loadMetadataForKeys(const fs::path & keys_dir)
                 try
                 {
                     file_segment_metadata_it = addFileSegment(
-                        *locked_key, offset, size, FileSegment::State::DOWNLOADED, CreateFileSegmentSettings(segment_kind), cache_it, nullptr);
+                        *key_metadata->lock(), offset, size, FileSegment::State::DOWNLOADED, CreateFileSegmentSettings(segment_kind), cache_it, nullptr);
                 }
                 catch (...)
                 {

src/Interpreters/Cache/Metadata.cpp

@@ -183,25 +183,7 @@ LockedKeyPtr CacheMetadata::lockKeyMetadata(
     KeyNotFoundPolicy key_not_found_policy,
     bool is_initial_load)
 {
-    KeyMetadataPtr key_metadata;
-    {
-        auto lock = lockMetadata();
-
-        auto it = find(key);
-        if (it == end())
-        {
-            if (key_not_found_policy == KeyNotFoundPolicy::THROW)
-                throw Exception(ErrorCodes::LOGICAL_ERROR, "No such key `{}` in cache", key);
-            else if (key_not_found_policy == KeyNotFoundPolicy::RETURN_NULL)
-                return nullptr;
-
-            it = emplace(
-                key, std::make_shared<KeyMetadata>(
-                    key, getPathForKey(key), *cleanup_queue, *download_queue, log, is_initial_load)).first;
-        }
-
-        key_metadata = it->second;
-    }
+    auto key_metadata = getKeyMetadata(key, key_not_found_policy, is_initial_load);
 
     {
         LockedKeyPtr locked_metadata;
@@ -237,6 +219,29 @@ LockedKeyPtr CacheMetadata::lockKeyMetadata(
     return lockKeyMetadata(key, key_not_found_policy);
 }
 
+KeyMetadataPtr CacheMetadata::getKeyMetadata(
+    const Key & key,
+    KeyNotFoundPolicy key_not_found_policy,
+    bool is_initial_load)
+{
+    auto lock = lockMetadata();
+
+    auto it = find(key);
+    if (it == end())
+    {
+        if (key_not_found_policy == KeyNotFoundPolicy::THROW)
+            throw Exception(ErrorCodes::LOGICAL_ERROR, "No such key `{}` in cache", key);
+        else if (key_not_found_policy == KeyNotFoundPolicy::RETURN_NULL)
+            return nullptr;
+
+        it = emplace(
+            key, std::make_shared<KeyMetadata>(
+                key, getPathForKey(key), *cleanup_queue, *download_queue, log, is_initial_load)).first;
+    }
+
+    return it->second;
+}
+
 void CacheMetadata::iterate(IterateCacheMetadataFunc && func)
 {
     auto lock = lockMetadata();

src/Interpreters/Cache/Metadata.h

@@ -115,6 +115,11 @@ public:
         KeyNotFoundPolicy key_not_found_policy,
         bool is_initial_load = false);
 
+    KeyMetadataPtr getKeyMetadata(
+        const Key & key,
+        KeyNotFoundPolicy key_not_found_policy,
+        bool is_initial_load = false);
+
     void doCleanup();
 
     void downloadThreadFunc();