Transparent Item movements (#251)

Summary:
The new algorithm relies on the moving bit and does not require external synchronization. Data movement happens transparently for the client: if the client thread attempts to get a handle for the item being moved it will get a handle with wait context to wait till the movement is completed.

Pull Request resolved: #251

Reviewed By: jaesoo-fb

Differential Revision: D51402910

Pulled By: haowu14

fbshipit-source-id: 349933265fa8e888b5a99426bd7e15743b86f592

cachelib/allocator/CacheAllocator.h

@@ -1347,7 +1347,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
@@ -1471,26 +1471,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,
-                                          uint32_t size);
-
-  // Given an item and its parentKey, validate that the parentKey
-  // corresponds to an item that's the parent of the supplied item.
-  //
-  // @param item       item that we want to get the parent handle for
-  // @param parentKey  key of the item's parent
-  //
-  // @return  handle to the parent item if the validations pass
-  //          otherwise, an empty Handle is returned.
-  //
-  ReadHandle validateAndGetParentHandleForChainedMoveLocked(
-      const ChainedItem& item, const Key& parentKey);
+  WriteHandle allocateChainedItemInternal(const Item& parent, uint32_t size);
 
   // Given an existing item, allocate a new one for the
   // existing one to later be moved into.
@@ -1607,7 +1594,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, Item& newParent);
 
   // replace a chained item in the existing chain. This needs to be called
   // with the chained item lock held exclusive
@@ -1621,6 +1608,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.
   //
@@ -1729,6 +1734,19 @@ 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(folly::StringPiece key, 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
@@ -1810,50 +1828,21 @@ class CacheAllocator : public CacheBase {
 
   // "Move" (by copying) the content in this item to another memory
   // location by invoking the move callback.
-  //
-  //
-  // @param ctx         slab release context
   // @param item        old item to be moved elsewhere
-  // @param throttler   slow this function down as not to take too much cpu
-  //
   // @return    true  if the item has been moved
   //            false if we have exhausted moving attempts
-  bool moveForSlabRelease(const SlabReleaseContext& ctx,
-                          Item& item,
-                          util::Throttler& throttler);
-
-  // "Move" (by copying) the content in this item to another memory
-  // location by invoking the move callback.
-  //
-  // @param item         old item to be moved elsewhere
-  // @param newItemHdl   handle of new item to be moved into
-  //
-  // @return    true  if the item has been moved
-  //            false if we have exhausted moving attempts
-  bool tryMovingForSlabRelease(Item& item, WriteHandle& newItemHdl);
+  bool moveForSlabRelease(Item& item);
 
   // Evict an item from access and mm containers and
   // ensure it is safe for freeing.
   //
-  // @param ctx         slab release context
   // @param item        old item to be moved elsewhere
-  // @param throttler   slow this function down as not to take too much cpu
-  void evictForSlabRelease(const SlabReleaseContext& ctx,
-                           Item& item,
-                           util::Throttler& throttler);
+  void evictForSlabRelease(Item& item);
 
-  // Helper function to evict a normal item for slab release
+  // Helper function to create PutToken
   //
-  // @return last handle for corresponding to item on success. empty handle on
-  // failure. caller can retry if needed.
-  WriteHandle evictNormalItemForSlabRelease(Item& item);
-
-  // Helper function to evict a child item for slab release
-  // As a side effect, the parent item is also evicted
-  //
-  // @return  last handle to the parent item of the child on success. empty
-  // handle on failure. caller can retry.
-  WriteHandle evictChainedItemForSlabRelease(ChainedItem& item);
+  // @return valid token if the item should be written to NVM cache.
+  typename NvmCacheT::PutToken createPutToken(Item& item);
 
   // Helper function to remove a item if expired.
   //
@@ -2078,6 +2067,82 @@ class CacheAllocator : public CacheBase {
     return stats;
   }
 
+  bool tryGetHandleWithWaitContextForMovingItem(Item& item,
+                                                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 waiters.
+        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> acquireMoveLockForShard(size_t shard) {
+    return std::unique_lock<std::mutex>(moveLock_[shard].moveLock_);
+  }
+
   // BEGIN private members
 
   // Whether the memory allocator for this cache allocator was created on shared
@@ -2173,6 +2238,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/CacheItem-inl.h

@@ -252,11 +252,6 @@ bool CacheItem<CacheTrait>::isMoving() const noexcept {
   return ref_.isMoving();
 }
 
-template <typename CacheTrait>
-bool CacheItem<CacheTrait>::isOnlyMoving() const noexcept {
-  return ref_.isOnlyMoving();
-}
-
 template <typename CacheTrait>
 void CacheItem<CacheTrait>::markNvmClean() noexcept {
   ref_.markNvmClean();

cachelib/allocator/CacheItem.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) {
@@ -366,22 +366,18 @@ class CACHELIB_PACKED_ATTR CacheItem {
   /**
    * The following functions correspond to whether or not an item is
    * currently in the processed of being moved. When moving, ref count
-   * is always >= 1.
+   * is always == 1.
    *
    * An item can only be marked moving when `isInMMContainer` returns true
    * and item is not already exclusive nor moving.
    *
-   * User can also query if an item "isOnlyMoving". This returns true only
-   * if the refcount is one and only the exclusive bit is set.
-   *
    * Unmarking moving does not depend on `isInMMContainer`
    * Unmarking moving will also return the refcount at the moment of
    * unmarking.
    */
   bool markMoving();
   RefcountWithFlags::Value unmarkMoving() noexcept;
   bool isMoving() const noexcept;
-  bool isOnlyMoving() const noexcept;
 
   /** This function attempts to mark item as exclusive.
    * Can only be called on the item that is moving.*/

cachelib/allocator/Handle.h

@@ -26,6 +26,7 @@
 #include <mutex>
 
 #include "cachelib/allocator/nvmcache/WaitContext.h"
+#include "cachelib/common/Exceptions.h"
 
 namespace facebook {
 namespace cachelib {
@@ -71,7 +72,7 @@ struct ReadHandleImpl {
     try {
       alloc_->release(it_, isNascent());
     } catch (const std::exception& e) {
-      XLOGF(CRITICAL, "Failed to release {:#10x} : {}", static_cast<void*>(it_),
+      XLOGF(CRITICAL, "Failed to release {} : {}", static_cast<void*>(it_),
             e.what());
     }
     it_ = nullptr;

cachelib/allocator/MM2Q-inl.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([&fun]() { fun(); });
+}
+
 template <typename T, MM2Q::Hook<T> T::*HookPtr>
 void MM2Q::Container<T, HookPtr>::removeLocked(T& node,
                                                bool doRebalance) noexcept {

cachelib/allocator/MM2Q.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/MMLru-inl.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([&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.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/MMTinyLFU-inl.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.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