Better hash set and less locks

src/Coordination/KeeperStateMachine.cpp

@@ -212,14 +212,15 @@ void KeeperStateMachine::create_snapshot(
             }
 
             {
-                /// Must do it with lock (clearing elements from list)
-                std::lock_guard lock(storage_and_responses_lock);
+                LOG_TRACE(log, "Clearing grabage after snapshot");
                 /// Turn off "snapshot mode" and clear outdate part of storage state
-                storage->clearGarbageAfterSnapshot();
-                /// Destroy snapshot with lock
-                snapshot.reset();
+                storage->clearGarbageAfterSnapshot(snapshot->snapshot_container_size);
                 LOG_TRACE(log, "Cleared garbage after snapshot");
-
+                {
+                    /// Destroy snapshot with lock
+                    std::lock_guard lock(storage_and_responses_lock);
+                    snapshot.reset();
+                }
             }
         }
         catch (...)

src/Coordination/KeeperStorage.cpp

@@ -317,6 +317,7 @@ struct KeeperStorageCreateRequestProcessor final : public KeeperStorageRequestPr
         created_node.is_sequental = request.is_sequential;
 
         auto [map_key, _] = container.insert(path_created, std::move(created_node));
+        /// Take child path from key owned by map.
         auto child_path = getBaseName(map_key->getKey());
 
         int32_t parent_cversion = request.parent_cversion;
@@ -492,8 +493,7 @@ struct KeeperStorageRemoveRequestProcessor final : public KeeperStorageRequestPr
 
             storage.acl_map.removeUsage(prev_node.acl_id);
 
-            auto child_basename = getBaseName(it->key);
-            container.updateValue(parentPath(request.path), [&child_basename] (KeeperStorage::Node & parent)
+            container.updateValue(parentPath(request.path), [child_basename = getBaseName(it->key)] (KeeperStorage::Node & parent)
             {
                 --parent.stat.numChildren;
                 ++parent.stat.cversion;
@@ -502,7 +502,7 @@ struct KeeperStorageRemoveRequestProcessor final : public KeeperStorageRequestPr
             });
 
             response.error = Coordination::Error::ZOK;
-
+            /// Erase full path from container after child removed from parent
             container.erase(request.path);
 
             undo = [prev_node, &storage, path = request.path]
@@ -512,6 +512,8 @@ struct KeeperStorageRemoveRequestProcessor final : public KeeperStorageRequestPr
 
                 storage.acl_map.addUsage(prev_node.acl_id);
 
+                /// Dangerous place: we are adding StringRef to child into children unordered_hash set.
+                /// That's why we are taking getBaseName from inserted key, not from the path from request object.
                 auto [map_key, _] = storage.container.insert(path, prev_node);
                 storage.container.updateValue(parentPath(path), [child_name = getBaseName(map_key->getKey())] (KeeperStorage::Node & parent)
                 {

src/Coordination/KeeperStorage.h

@@ -10,6 +10,7 @@
 #include <unordered_map>
 #include <unordered_set>
 #include <vector>
+#include <absl/container/flat_hash_set.h>
 
 namespace DB
 {
@@ -17,7 +18,7 @@ namespace DB
 struct KeeperStorageRequestProcessor;
 using KeeperStorageRequestProcessorPtr = std::shared_ptr<KeeperStorageRequestProcessor>;
 using ResponseCallback = std::function<void(const Coordination::ZooKeeperResponsePtr &)>;
-using ChildrenSet = std::unordered_set<StringRef, StringRefHash>;
+using ChildrenSet = absl::flat_hash_set<StringRef, StringRefHash>;
 using SessionAndTimeout = std::unordered_map<int64_t, int64_t>;
 
 struct KeeperStorageSnapshot;
@@ -176,9 +177,9 @@ public:
     }
 
     /// Clear outdated data from internal container.
-    void clearGarbageAfterSnapshot()
+    void clearGarbageAfterSnapshot(size_t up_to_size)
     {
-        container.clearOutdatedNodes();
+        container.clearOutdatedNodes(up_to_size);
     }
 
     /// Get all active sessions

src/Coordination/SnapshotableHashTable.h

@@ -269,11 +269,11 @@ public:
         return it->getMapped()->value;
     }
 
-    void clearOutdatedNodes()
+    void clearOutdatedNodes(size_t up_to_size)
     {
         auto start = list.begin();
-        auto end = list.end();
-        for (auto itr = start; itr != end;)
+        size_t counter = 0;
+        for (auto itr = start; counter < up_to_size; ++counter)
         {
             if (!itr->active_in_map)
             {
@@ -288,6 +288,7 @@ public:
                 itr++;
             }
         }
+
     }
 
     void clear()