Transparent Item movements

Author: vinser52

cachelib/allocator/CacheAllocator-inl.h

@@ -1349,7 +1349,7 @@ class CacheAllocator : public CacheBase {
 
  private:
   // wrapper around Item's refcount and active handle tracking
-  FOLLY_ALWAYS_INLINE bool incRef(Item& it);
+  FOLLY_ALWAYS_INLINE RefcountWithFlags::incResult incRef(Item& it);
   FOLLY_ALWAYS_INLINE RefcountWithFlags::Value decRef(Item& it);
 
   // drops the refcount and if needed, frees the allocation back to the memory
@@ -1473,13 +1473,13 @@ class CacheAllocator : public CacheBase {
   // The parent handle parameter here is mainly used to find the
   // correct pool to allocate memory for this chained item
   //
-  // @param parent    handle to the cache item
+  // @param parent    the parent item
   // @param size      the size for the chained allocation
   //
   // @return    handle to the chained allocation
   // @throw     std::invalid_argument if the size requested is invalid or
   //            if the item is invalid
-  WriteHandle allocateChainedItemInternal(const ReadHandle& parent,
+  WriteHandle allocateChainedItemInternal(const Item& parent,
                                           uint32_t size);
 
   // Given an item and its parentKey, validate that the parentKey
@@ -1609,7 +1609,7 @@ class CacheAllocator : public CacheBase {
   // @param newParent the new parent for the chain
   //
   // @throw if any of the conditions for parent or newParent are not met.
-  void transferChainLocked(WriteHandle& parent, WriteHandle& newParent);
+  void transferChainLocked(Item& parent, WriteHandle& newParent);
 
   // replace a chained item in the existing chain. This needs to be called
   // with the chained item lock held exclusive
@@ -1623,6 +1623,24 @@ class CacheAllocator : public CacheBase {
                                        WriteHandle newItemHdl,
                                        const Item& parent);
 
+  //
+  // Performs the actual inplace replace - it is called from
+  // moveChainedItem and replaceChainedItemLocked
+  // must hold chainedItemLock
+  //
+  // @param oldItem  the item we are replacing in the chain
+  // @param newItem  the item we are replacing it with
+  // @param parent   the parent for the chain
+  // @param fromMove used to determine if the replaced was called from
+  //                 moveChainedItem - we avoid the handle destructor
+  //                 in this case.
+  //
+  // @return handle to the oldItem
+  void replaceInChainLocked(Item& oldItem,
+                            WriteHandle& newItemHdl,
+                            const Item& parent,
+                            bool fromMove);
+
   // Insert an item into MM container. The caller must hold a valid handle for
   // the item.
   //
@@ -1731,6 +1749,18 @@ class CacheAllocator : public CacheBase {
 
   using EvictionIterator = typename MMContainer::LockedIterator;
 
+  // Wakes up waiters if there are any
+  //
+  // @param item    wakes waiters that are waiting on that item
+  // @param handle  handle to pass to the waiters
+  void wakeUpWaiters(Item& item, WriteHandle handle);
+
+  // Unmarks item as moving and wakes up any waiters waiting on that item
+  //
+  // @param item    wakes waiters that are waiting on that item
+  // @param handle  handle to pass to the waiters
+  typename RefcountWithFlags::Value unmarkMovingAndWakeUpWaiters(Item &item, WriteHandle handle);
+
   // Deserializer CacheAllocatorMetadata and verify the version
   //
   // @param  deserializer   Deserializer object
@@ -2082,6 +2112,87 @@ class CacheAllocator : public CacheBase {
 
   // BEGIN private members
 
+  bool tryGetHandleWithWaitContextForMovingItem(Item& item, WriteHandle& handle);
+
+  size_t wakeUpWaitersLocked(folly::StringPiece key, WriteHandle&& handle);
+
+  class MoveCtx {
+   public:
+    MoveCtx() {}
+
+    ~MoveCtx() {
+      // prevent any further enqueue to waiters
+      // Note: we don't need to hold locks since no one can enqueue
+      // after this point.
+      wakeUpWaiters();
+    }
+
+    // record the item handle. Upon destruction we will wake up the waiters
+    // and pass a clone of the handle to the callBack. By default we pass
+    // a null handle
+    void setItemHandle(WriteHandle _it) { it = std::move(_it); }
+
+    // enqueue a waiter into the waiter list
+    // @param  waiter       WaitContext
+    void addWaiter(std::shared_ptr<WaitContext<ReadHandle>> waiter) {
+      XDCHECK(waiter);
+      waiters.push_back(std::move(waiter));
+    }
+
+    size_t numWaiters() const { return waiters.size(); }
+
+   private:
+    // notify all pending waiters that are waiting for the fetch.
+    void wakeUpWaiters() {
+      bool refcountOverflowed = false;
+      for (auto& w : waiters) {
+        // If refcount overflowed earlier, then we will return miss to
+        // all subsequent waitors.
+        if (refcountOverflowed) {
+          w->set(WriteHandle{});
+          continue;
+        }
+
+        try {
+          w->set(it.clone());
+        } catch (const exception::RefcountOverflow&) {
+          // We'll return a miss to the user's pending read,
+          // so we should enqueue a delete via NvmCache.
+          // TODO: cache.remove(it);
+          refcountOverflowed = true;
+        }
+      }
+    }
+
+    WriteHandle it; // will be set when Context is being filled
+    std::vector<std::shared_ptr<WaitContext<ReadHandle>>> waiters; // list of
+                                                                   // waiters
+  };
+  using MoveMap =
+      folly::F14ValueMap<folly::StringPiece,
+                         std::unique_ptr<MoveCtx>,
+                         folly::HeterogeneousAccessHash<folly::StringPiece>>;
+
+  static size_t getShardForKey(folly::StringPiece key) {
+    return folly::Hash()(key) % kShards;
+  }
+
+  MoveMap& getMoveMapForShard(size_t shard) {
+    return movesMap_[shard].movesMap_;
+  }
+
+  MoveMap& getMoveMap(folly::StringPiece key) {
+    return getMoveMapForShard(getShardForKey(key));
+  }
+
+  std::unique_lock<std::mutex> getMoveLockForShard(size_t shard) {
+    return std::unique_lock<std::mutex>(moveLock_[shard].moveLock_);
+  }
+
+  std::unique_lock<std::mutex> getMoveLock(folly::StringPiece key) {
+    return getMoveLockForShard(getShardForKey(key));
+  }
+
   // Whether the memory allocator for this cache allocator was created on shared
   // memory. The hash table, chained item hash table etc is also created on
   // shared memory except for temporary shared memory mode when they're created
@@ -2175,6 +2286,22 @@ class CacheAllocator : public CacheBase {
   // poolResizer_, poolOptimizer_, memMonitor_, reaper_
   mutable std::mutex workersMutex_;
 
+  static constexpr size_t kShards = 8192; // TODO: need to define right value
+
+  struct MovesMapShard {
+    alignas(folly::hardware_destructive_interference_size) MoveMap movesMap_;
+  };
+
+  struct MoveLock {
+    alignas(folly::hardware_destructive_interference_size) std::mutex moveLock_;
+  };
+
+  // a map of all pending moves
+  std::vector<MovesMapShard> movesMap_;
+
+  // a map of move locks for each shard
+  std::vector<MoveLock> moveLock_;
+
   // time when the ram cache was first created
   const uint32_t cacheCreationTime_{0};
 

cachelib/allocator/CacheAllocator.h

@@ -309,7 +309,7 @@ class CACHELIB_PACKED_ATTR CacheItem {
   //
   // @return true on success, failure if item is marked as exclusive
   // @throw exception::RefcountOverflow on ref count overflow
-  FOLLY_ALWAYS_INLINE bool incRef() {
+  FOLLY_ALWAYS_INLINE RefcountWithFlags::incResult incRef() {
     try {
       return ref_.incRef();
     } catch (exception::RefcountOverflow& e) {

cachelib/allocator/CacheItem.h

@@ -258,6 +258,12 @@ void MM2Q::Container<T, HookPtr>::withEvictionIterator(F&& fun) {
   }
 }
 
+template <typename T, MM2Q::Hook<T> T::*HookPtr>
+template <typename F>
+void MM2Q::Container<T, HookPtr>::withContainerLock(F&& fun) {
+  lruMutex_->lock_combine([this, &fun]() { fun(); });
+}
+
 template <typename T, MM2Q::Hook<T> T::*HookPtr>
 void MM2Q::Container<T, HookPtr>::removeLocked(T& node,
                                                bool doRebalance) noexcept {

cachelib/allocator/MM2Q-inl.h

@@ -502,6 +502,10 @@ class MM2Q {
     template <typename F>
     void withEvictionIterator(F&& f);
 
+    // Execute provided function under container lock.
+    template <typename F>
+    void withContainerLock(F&& f);
+
     // get the current config as a copy
     Config getConfig() const;
 

cachelib/allocator/MM2Q.h

@@ -229,6 +229,12 @@ void MMLru::Container<T, HookPtr>::withEvictionIterator(F&& fun) {
   }
 }
 
+template <typename T, MMLru::Hook<T> T::*HookPtr>
+template <typename F>
+void MMLru::Container<T, HookPtr>::withContainerLock(F&& fun) {
+  lruMutex_->lock_combine([this, &fun]() { fun(); });
+}
+
 template <typename T, MMLru::Hook<T> T::*HookPtr>
 void MMLru::Container<T, HookPtr>::ensureNotInsertionPoint(T& node) noexcept {
   // If we are removing the insertion point node, grow tail before we remove

cachelib/allocator/MMLru-inl.h

@@ -376,6 +376,10 @@ class MMLru {
     template <typename F>
     void withEvictionIterator(F&& f);
 
+    // Execute provided function under container lock.
+    template <typename F>
+    void withContainerLock(F&& f);
+
     // get copy of current config
     Config getConfig() const;
 

cachelib/allocator/MMLru.h

@@ -227,6 +227,13 @@ void MMTinyLFU::Container<T, HookPtr>::withEvictionIterator(F&& fun) {
   fun(getEvictionIterator());
 }
 
+template <typename T, MMTinyLFU::Hook<T> T::*HookPtr>
+template <typename F>
+void MMTinyLFU::Container<T, HookPtr>::withContainerLock(F&& fun) {
+  LockHolder l(lruMutex_);
+  fun();
+}
+
 template <typename T, MMTinyLFU::Hook<T> T::*HookPtr>
 void MMTinyLFU::Container<T, HookPtr>::removeLocked(T& node) noexcept {
   if (isTiny(node)) {

cachelib/allocator/MMTinyLFU-inl.h

@@ -497,6 +497,10 @@ class MMTinyLFU {
     template <typename F>
     void withEvictionIterator(F&& f);
 
+    // Execute provided function under container lock.
+    template <typename F>
+    void withContainerLock(F&& f);
+
     // for saving the state of the lru
     //
     // precondition:  serialization must happen without any reader or writer

cachelib/allocator/MMTinyLFU.h

@@ -130,30 +130,41 @@ class FOLLY_PACK_ATTR RefcountWithFlags {
   RefcountWithFlags& operator=(const RefcountWithFlags&) = delete;
   RefcountWithFlags(RefcountWithFlags&&) = delete;
   RefcountWithFlags& operator=(RefcountWithFlags&&) = delete;
-
+  enum incResult {
+     incOk,
+     incFailedMoving,
+     incFailedEviction
+   };
   // Bumps up the reference count only if the new count will be strictly less
   // than or equal to the maxCount and the item is not exclusive
   // @return true if refcount is bumped. false otherwise (if item is exclusive)
   // @throw  exception::RefcountOverflow if new count would be greater than
   // maxCount
-  FOLLY_ALWAYS_INLINE bool incRef() {
-    auto predicate = [](const Value curValue) {
-      Value bitMask = getAdminRef<kExclusive>();
-
-      const bool exlusiveBitIsSet = curValue & bitMask;
-      if (UNLIKELY((curValue & kAccessRefMask) == (kAccessRefMask))) {
-        throw exception::RefcountOverflow("Refcount maxed out.");
-      }
-
-      // Check if the item is not marked for eviction
-      return !exlusiveBitIsSet || ((curValue & kAccessRefMask) != 0);
-    };
-
-    auto newValue = [](const Value curValue) {
-      return (curValue + static_cast<Value>(1));
-    };
-
-    return atomicUpdateValue(predicate, newValue);
+  FOLLY_ALWAYS_INLINE incResult incRef() {
+    incResult res = incOk;
+    auto predicate = [&res](const Value curValue) {
+       Value bitMask = getAdminRef<kExclusive>();
+
+       const bool exlusiveBitIsSet = curValue & bitMask;
+       if (UNLIKELY((curValue & kAccessRefMask) == (kAccessRefMask))) {
+         throw exception::RefcountOverflow("Refcount maxed out.");
+       } else if (exlusiveBitIsSet && (curValue & kAccessRefMask) == 0) {
+         res = incFailedEviction;
+         return false;
+       } else if (exlusiveBitIsSet) {
+         res = incFailedMoving;
+         return false;
+       }
+       res = incOk;
+       return true;
+     };
+
+     auto newValue = [](const Value curValue) {
+       return (curValue + static_cast<Value>(1));
+     };
+
+     atomicUpdateValue(predicate, newValue);
+     return res;
   }
 
   // Bumps down the reference count

cachelib/allocator/Refcount.h

@@ -341,6 +341,7 @@ class RWBucketLocks : public BaseBucketLocks<LockType, LockAlignmentType> {
   using Lock = LockType;
   using ReadLockHolder = ReadLockHolderType;
   using WriteLockHolder = WriteLockHolderType;
+  using LockHolder = std::unique_lock<Lock>;
 
   RWBucketLocks(uint32_t locksPower, std::shared_ptr<Hash> hasher)
       : Base::BaseBucketLocks(locksPower, std::move(hasher)) {}
@@ -357,6 +358,11 @@ class RWBucketLocks : public BaseBucketLocks<LockType, LockAlignmentType> {
     return WriteLockHolder{Base::getLock(args...)};
   }
 
+  template <typename... Args>
+  LockHolder tryLockExclusive(Args... args) noexcept {
+    return LockHolder(Base::getLock(args...), std::try_to_lock);
+  }
+
   // try to grab the reader lock for a limit _timeout_ duration
   template <typename... Args>
   ReadLockHolder lockShared(const std::chrono::microseconds& timeout,