Shard OpenedFileCache to avoid lock contention (#51341)

* shard OpenedFileCache to avoid lock contention

* Update OpenedFileCache.h

* fix build

---------

Co-authored-by: Alexey Milovidov <milovidov@clickhouse.com>

src/Common/ProfileEvents.cpp

@@ -45,6 +45,7 @@
     M(MMappedFileCacheMisses, "Number of times a file has not been found in the MMap cache (for the 'mmap' read_method), so we had to mmap it again.") \
     M(OpenedFileCacheHits, "Number of times a file has been found in the opened file cache, so we didn't have to open it again.") \
     M(OpenedFileCacheMisses, "Number of times a file has been found in the opened file cache, so we had to open it again.") \
+    M(OpenedFileCacheMicroseconds, "Amount of time spent executing OpenedFileCache methods.") \
     M(AIOWrite, "Number of writes with Linux or FreeBSD AIO interface") \
     M(AIOWriteBytes, "Number of bytes written with Linux or FreeBSD AIO interface") \
     M(AIORead, "Number of reads with Linux or FreeBSD AIO interface") \

src/IO/OpenedFileCache.h

@@ -4,14 +4,18 @@
 #include <mutex>
 
 #include <Core/Types.h>
-#include <Common/ProfileEvents.h>
 #include <IO/OpenedFile.h>
+#include <Common/ElapsedTimeProfileEventIncrement.h>
+#include <Common/ProfileEvents.h>
+
+#include <city.h>
 
 
 namespace ProfileEvents
 {
     extern const Event OpenedFileCacheHits;
     extern const Event OpenedFileCacheMisses;
+    extern const Event OpenedFileCacheMicroseconds;
 }
 
 namespace DB
@@ -26,57 +30,79 @@ namespace DB
   */
 class OpenedFileCache
 {
-private:
-    using Key = std::pair<std::string /* path */, int /* flags */>;
+    class OpenedFileMap
+    {
+        using Key = std::pair<std::string /* path */, int /* flags */>;
 
-    using OpenedFileWeakPtr = std::weak_ptr<OpenedFile>;
-    using Files = std::map<Key, OpenedFileWeakPtr>;
+        using OpenedFileWeakPtr = std::weak_ptr<OpenedFile>;
+        using Files = std::map<Key, OpenedFileWeakPtr>;
 
-    Files files;
-    std::mutex mutex;
+        Files files;
+        std::mutex mutex;
+
+    public:
+        using OpenedFilePtr = std::shared_ptr<OpenedFile>;
+
+        OpenedFilePtr get(const std::string & path, int flags)
+        {
+            Key key(path, flags);
+
+            std::lock_guard lock(mutex);
+
+            auto [it, inserted] = files.emplace(key, OpenedFilePtr{});
+            if (!inserted)
+            {
+                if (auto res = it->second.lock())
+                {
+                    ProfileEvents::increment(ProfileEvents::OpenedFileCacheHits);
+                    return res;
+                }
+            }
+            ProfileEvents::increment(ProfileEvents::OpenedFileCacheMisses);
+
+            OpenedFilePtr res
+            {
+                new OpenedFile(path, flags),
+                [key, this](auto ptr)
+                {
+                    {
+                        std::lock_guard another_lock(mutex);
+                        files.erase(key);
+                    }
+                    delete ptr;
+                }
+            };
+
+            it->second = res;
+            return res;
+        }
+
+        void remove(const std::string & path, int flags)
+        {
+            Key key(path, flags);
+            std::lock_guard lock(mutex);
+            files.erase(key);
+        }
+    };
+
+    static constexpr size_t buckets = 1024;
+    std::vector<OpenedFileMap> impls{buckets};
 
 public:
-    using OpenedFilePtr = std::shared_ptr<OpenedFile>;
+    using OpenedFilePtr = OpenedFileMap::OpenedFilePtr;
 
     OpenedFilePtr get(const std::string & path, int flags)
     {
-        Key key(path, flags);
-
-        std::lock_guard lock(mutex);
-
-        auto [it, inserted] = files.emplace(key, OpenedFilePtr{});
-        if (!inserted)
-        {
-            if (auto res = it->second.lock())
-            {
-                ProfileEvents::increment(ProfileEvents::OpenedFileCacheHits);
-                return res;
-            }
-        }
-        ProfileEvents::increment(ProfileEvents::OpenedFileCacheMisses);
-
-        OpenedFilePtr res
-        {
-            new OpenedFile(path, flags),
-            [key, this](auto ptr)
-            {
-                {
-                    std::lock_guard another_lock(mutex);
-                    files.erase(key);
-                }
-                delete ptr;
-            }
-        };
-
-        it->second = res;
-        return res;
+        ProfileEventTimeIncrement<Microseconds> watch(ProfileEvents::OpenedFileCacheMicroseconds);
+        const auto bucket = CityHash_v1_0_2::CityHash64(path.data(), path.length()) % buckets;
+        return impls[bucket].get(path, flags);
     }
 
     void remove(const std::string & path, int flags)
     {
-        Key key(path, flags);
-        std::lock_guard lock(mutex);
-        files.erase(key);
+        ProfileEventTimeIncrement<Microseconds> watch(ProfileEvents::OpenedFileCacheMicroseconds);
+        const auto bucket = CityHash_v1_0_2::CityHash64(path.data(), path.length()) % buckets;
+        impls[bucket].remove(path, flags);
     }
 
     static OpenedFileCache & instance()
@@ -87,5 +113,4 @@ public:
 };
 
 using OpenedFileCachePtr = std::shared_ptr<OpenedFileCache>;
-
 }