deletions in concurrent hash

Author: williamnixon20

cachelib/allocator/MMS3FIFO.h

@@ -33,7 +33,6 @@
 #include "cachelib/allocator/memory/serialize/gen-cpp2/objects_types.h"
 #include "cachelib/common/CompilerUtils.h"
 #include "cachelib/common/FIFOHashSet.h"
-#include "cachelib/common/FIFOAtomicHashSet.h"
 #include "cachelib/common/FIFOConcurrentHashSet.h"
 
 namespace facebook::cachelib {
@@ -386,10 +385,15 @@ class MMS3FIFO {
     void maybeResizeGhostLocked() noexcept;
 
     // Returns the hash of node's key
-    static size_t hashNode(const T& node) noexcept {
+    static size_t hashNode64(const T& node) noexcept {
       return folly::hasher<folly::StringPiece>()(node.getKey());
     }
 
+    static uint32_t hashNode(const T& node) noexcept {
+      return static_cast<uint32_t>(
+          folly::hasher<folly::StringPiece>()(node.getKey()));
+    }
+
     void removeLocked(T& node) noexcept;
 
     static bool isTiny(const T& node) noexcept {
@@ -429,9 +433,8 @@ class MMS3FIFO {
     // the lru
     LruList lru_;
 
-    facebook::cachelib::util::detail::FIFOConcurrentHashSet ghostQueue_;
-    // facebook::cachelib::util::detail::FIFOAtomicHashSet ghostQueue_;
-    // facebook::cachelib::util::FIFOHashSet ghostQueue_;
+    facebook::cachelib::util::FIFOConcurrentHashSet32 ghostQueue_;
+    // facebook::cachelib::util::FIFOHashSet32 ghostQueue_;
 
     // Config for this lru.
     // Write access to the MMS3FIFO Config is serialized.
@@ -482,7 +485,7 @@ bool MMS3FIFO::Container<T, HookPtr>::recordAccess(T& node,
   }
   const auto currTime = static_cast<Time>(util::getCurrentTimeSec());
 
-  // Remove lock from record access wince we only set bits
+  // Remove lock from record access since we only set bits
   if (node.isInMMContainer()) {
     if (!isAccessed(node)) {
       markAccessed(node);
@@ -643,13 +646,12 @@ void MMS3FIFO::Container<T, HookPtr>::withContainerLock(F&& fun) {
   lruMutex_->lock_combine([&fun]() { fun(); });
 }
 
+// Ghost queue insertion done on callee, outside lock
 template <typename T, MMS3FIFO::Hook<T> T::* HookPtr>
 void MMS3FIFO::Container<T, HookPtr>::removeLocked(T& node) noexcept {
   if (isTiny(node)) {
     lru_.getList(LruType::Tiny).remove(node); 
     unmarkTiny(node);
-    // Insert into ghost queue upon eviction from tiny queue
-    ghostQueue_.insert(hashNode(node));
   } else {
     lru_.getList(LruType::Main).remove(node);
   }
@@ -661,13 +663,19 @@ void MMS3FIFO::Container<T, HookPtr>::removeLocked(T& node) noexcept {
 
 template <typename T, MMS3FIFO::Hook<T> T::* HookPtr>
 bool MMS3FIFO::Container<T, HookPtr>::remove(T& node) noexcept {
-  return lruMutex_->lock_combine([this, &node]() {
+  bool isTiny_ = isTiny(node);
+  auto result = lruMutex_->lock_combine([this, &node]() {
     if (!node.isInMMContainer()) {
       return false;
     }
     removeLocked(node);
     return true;
   });
+  if (result && isTiny_) {
+    // Insert to ghost queue
+    ghostQueue_.insert(hashNode(node));
+  }
+  return result;
 }
 
 template <typename T, MMS3FIFO::Hook<T> T::* HookPtr>

cachelib/common/FIFOAtomicHashSet.h

@@ -9,15 +9,16 @@
 namespace facebook::cachelib::util {
 namespace detail {
 
-class FIFOConcurrentHashSet {
+template <typename KeyT = uint32_t>
+class FIFOConcurrentHashSetBase {
  public:
-  using Key = uint64_t; // hashed object id
+  using Key = KeyT; // hashed object id
   using TS = uint64_t;  // logical time / FIFO order
 
-  explicit FIFOConcurrentHashSet(size_t capacity)
+  explicit FIFOConcurrentHashSetBase(size_t capacity)
       : ghost_(capacity), seq_(0), queueSize_(0) {}
 
-  FIFOConcurrentHashSet() : seq_(0), queueSize_(0) {}
+  FIFOConcurrentHashSetBase() : seq_(0), queueSize_(0) {}
 
   struct PendingEntry {
     Key k;
@@ -27,16 +28,26 @@ class FIFOConcurrentHashSet {
   inline void insert(Key k) {
     TS ts = seq_.fetch_add(1, std::memory_order_relaxed);
     ghost_.insert_or_assign(k, ts);
+    // If it overflows and wraps around, we reset the entire structure.
+    if (UNLIKELY(ts == std::numeric_limits<TS>::max())) {
+      ghost_.clear();
+    }
   }
 
-  inline bool contains(Key k) const {
+  inline bool contains(Key k)  {
     auto it = ghost_.find(k);
     if (it == ghost_.end())
       return false;
     // Seq - current must be larger
     auto currDiff = seq_.load(std::memory_order_relaxed) - it->second;
-    return currDiff <=
-           queueSize_.load(std::memory_order_relaxed); // stale == not present
+    auto isValid = currDiff <= queueSize_.load(std::memory_order_relaxed);
+
+    if (!isValid) {
+      // remove stale entry
+      ghost_.erase(k);
+    }
+
+    return isValid;
   }
 
   inline void resize(TS limitTS) {
@@ -53,4 +64,7 @@ class FIFOConcurrentHashSet {
 };
 
 } // namespace detail
+using FIFOConcurrentHashSet32 = detail::FIFOConcurrentHashSetBase<uint32_t>;
+using FIFOConcurrentHashSet64 = detail::FIFOConcurrentHashSetBase<uint64_t>;
+  
 } // namespace facebook::cachelib::util

cachelib/common/FIFOConcurrentHashSet.h

@@ -3,11 +3,11 @@
 #include <folly/Format.h>
 #include <folly/Math.h>
 #include <folly/container/F14Set.h>
+#include <folly/synchronization/DistributedMutex.h>
 
 #include <cstdint>
 #include <deque>
 #include <stdexcept>
-#include <unordered_set>
 
 namespace facebook::cachelib::util {
 namespace detail {
@@ -26,94 +26,63 @@ class FIFOHashSetBase {
   FIFOHashSetBase(const FIFOHashSetBase&) = delete;
   FIFOHashSetBase& operator=(const FIFOHashSetBase&) = delete;
 
-  FIFOHashSetBase(FIFOHashSetBase&& other) noexcept {
-    maxSize_ = other.maxSize_;
-    fifo_ = std::move(other.fifo_);
-    set_ = std::move(other.set_);
-    other.maxSize_ = 0;
-  }
-
-  FIFOHashSetBase& operator=(FIFOHashSetBase&& other) noexcept {
-    if (this != &other) {
-      maxSize_ = other.maxSize_;
-      fifo_ = std::move(other.fifo_);
-      set_ = std::move(other.set_);
-      other.maxSize_ = 0;
-    }
-    return *this;
-  }
 
   bool contains(KeyT key) {
-    return set_.contains(key);
-  }
-
-  // Enforce size variant of contains
-  bool containsEnforce(KeyT key) {
-    enforceCapacityLocked();
-    return set_.contains(key);
+    return mutex_.lock_combine([&]() { return set_.contains(key); });
   }
 
   size_t size() {
-    return set_.size();
+    return mutex_.lock_combine([&]() { return set_.size(); });
   }
 
   size_t capacity() {
-    return maxSize_;
+    return mutex_.lock_combine([&]() { return maxSize_; });
   }
 
   void reserve(size_t newCap) {
-    set_.reserve(newCap);
+    mutex_.lock_combine([&]() { set_.reserve(newCap); });
   }
 
   void insert(KeyT key) {
-    auto [it, inserted] = set_.insert(key);
-    if (inserted) {
-      fifo_.push_back(key); 
-    }
-    enforceCapacityLocked();
-  }
-
-  // No enforce variant of insert
-  void insertNoEnforce(KeyT key) {
-    auto [it, inserted] = set_.insert(key);
-    if (inserted) {
-      fifo_.push_back(key);
-    }
+    mutex_.lock_combine([&]() {
+      auto [it, inserted] = set_.insert(key);
+      if (inserted) {
+        fifo_.push_back(key);
+      }
+      enforceCapacityLocked();
+    });
   }
 
   void resize(size_t newSize) {
-    maxSize_ = newSize;
-    enforceCapacityLocked();
-  }
-
-  // No enforce variant of resize
-  void resizeNoEnforce(size_t newSize) {
-    maxSize_ = newSize;
+    mutex_.lock_combine([&]() {
+      maxSize_ = newSize;
+    });
   }
 
  private:
-  // To prevent tail latency, limits max elements removed per call.
+  // Enforce max removal to control tail latency
   void enforceCapacityLocked() {
-    constexpr int max_rem = 5;
+    constexpr int max_rem = 2;
     int removed = 0;
 
     while (fifo_.size() > maxSize_ && removed < max_rem) {
       KeyT k = fifo_.front();
       fifo_.pop_front();
       set_.erase(k);
-      removed++;
+      ++removed;
     }
   }
 
  private:
-  // mutable folly::SpinLock lock_;
+  folly::DistributedMutex mutex_;
+
   size_t maxSize_{0};
   std::deque<KeyT> fifo_;
   folly::F14FastSet<KeyT> set_;
 };
 
 } // namespace detail
 
-using FIFOHashSet = detail::FIFOHashSetBase<uint64_t>;
+using FIFOHashSet32 = detail::FIFOHashSetBase<uint32_t>;
 
-} // namespace facebook::cachelib::util
+} // namespace facebook::cachelib::util