Do not block reader if a child item is moving

This would lead to deadlock (.e.g in forEachChainedItem)
if the child is moving (e.g. marked by Slab Release thread).

Instead treat moving bit only to prevent freeing the item and
do all synchronization on parent.

cachelib/allocator/CacheAllocator-inl.h

@@ -987,7 +987,8 @@ CacheAllocator<CacheTrait>::acquire(Item* it) {
 
   SCOPE_FAIL { stats_.numRefcountOverflow.inc(); };
 
-  auto failIfMoving = getNumTiers() > 1;
+  // TODO: do not block incRef for child items to avoid deadlock
+  auto failIfMoving = getNumTiers() > 1 && !it->isChainedItem();
   auto incRes = incRef(*it, failIfMoving);
   if (LIKELY(incRes == RefcountWithFlags::incResult::incOk)) {
     return WriteHandle{it, *this};
@@ -3051,7 +3052,8 @@ bool CacheAllocator<CacheTrait>::tryMovingForSlabRelease(
   // a regular item or chained item is synchronized with any potential
   // user-side mutation.
   std::unique_ptr<SyncObj> syncObj;
-  if (config_.movingSync) {
+  if (config_.movingSync && getNumTiers() == 1) {
+    // TODO: use moving-bit synchronization for single tier as well
     if (!oldItem.isChainedItem()) {
       syncObj = config_.movingSync(oldItem.getKey());
     } else {
@@ -3149,47 +3151,51 @@ void CacheAllocator<CacheTrait>::evictForSlabRelease(
     Item* evicted;
     if (item.isChainedItem()) {
       auto& expectedParent = item.asChainedItem().getParentItem(compressor_);
-      const std::string parentKey = expectedParent.getKey().str();
-      auto l = chainedItemLocks_.lockExclusive(parentKey);
-
-      // check if the child is still in mmContainer and the expected parent is
-      // valid under the chained item lock.
-      if (expectedParent.getKey() != parentKey || !item.isInMMContainer() ||
-          item.isOnlyMoving() ||
-          &expectedParent != &item.asChainedItem().getParentItem(compressor_) ||
-          !expectedParent.isAccessible() || !expectedParent.hasChainedItem()) {
-        continue;
-      }
 
-      // search if the child is present in the chain
-      {
-        auto parentHandle = findInternal(parentKey);
-        if (!parentHandle || parentHandle != &expectedParent) {
+      if (getNumTiers() == 1) {
+        // TODO: unify this with multi-tier implementation
+        // right now, taking a chained item lock here would lead to deadlock
+        const std::string parentKey = expectedParent.getKey().str();
+        auto l = chainedItemLocks_.lockExclusive(parentKey);
+
+        // check if the child is still in mmContainer and the expected parent is
+        // valid under the chained item lock.
+        if (expectedParent.getKey() != parentKey || !item.isInMMContainer() ||
+            item.isOnlyMoving() ||
+            &expectedParent != &item.asChainedItem().getParentItem(compressor_) ||
+            !expectedParent.isAccessible() || !expectedParent.hasChainedItem()) {
           continue;
         }
 
-        ChainedItem* head = nullptr;
-        { // scope for the handle
-          auto headHandle = findChainedItem(expectedParent);
-          head = headHandle ? &headHandle->asChainedItem() : nullptr;
-        }
+        // search if the child is present in the chain
+        {
+          auto parentHandle = findInternal(parentKey);
+          if (!parentHandle || parentHandle != &expectedParent) {
+            continue;
+          }
 
-        bool found = false;
-        while (head) {
-          if (head == &item) {
-            found = true;
-            break;
+          ChainedItem* head = nullptr;
+          { // scope for the handle
+            auto headHandle = findChainedItem(expectedParent);
+            head = headHandle ? &headHandle->asChainedItem() : nullptr;
           }
-          head = head->getNext(compressor_);
-        }
 
-        if (!found) {
-          continue;
+          bool found = false;
+          while (head) {
+            if (head == &item) {
+              found = true;
+              break;
+            }
+            head = head->getNext(compressor_);
+          }
+
+          if (!found) {
+            continue;
+          }
         }
       }
 
       evicted = &expectedParent;
-
       token = createPutToken(*evicted);
       if (evicted->markForEviction()) {
         // unmark the child so it will be freed
@@ -3200,6 +3206,9 @@ void CacheAllocator<CacheTrait>::evictForSlabRelease(
         // no other reader can be added to the waiters list
         wakeUpWaiters(*evicted, {});
       } else {
+        // TODO: potential deadlock with markUseful for parent item
+        // for now, we do not block any reader on child items but this
+        // should probably be fixed
         continue;
       }
     } else {
@@ -3231,7 +3240,17 @@ void CacheAllocator<CacheTrait>::evictForSlabRelease(
     XDCHECK(evicted->getRefCount() == 0);
     const auto res =
         releaseBackToAllocator(*evicted, RemoveContext::kEviction, false);
-    XDCHECK(res == ReleaseRes::kReleased);
+
+    if (getNumTiers() == 1) {
+      XDCHECK(res == ReleaseRes::kReleased);
+    } else {
+      const bool isAlreadyFreed =
+          !markMovingForSlabRelease(ctx, &item, throttler);
+      if (!isAlreadyFreed) {
+        continue;
+      }
+    }
+
     return;
   }
 }
@@ -3279,11 +3298,15 @@ bool CacheAllocator<CacheTrait>::markMovingForSlabRelease(
   bool itemFreed = true;
   bool markedMoving = false;
   TierId tid = getTierId(alloc);
-  const auto fn = [&markedMoving, &itemFreed](void* memory) {
+  auto numTiers = getNumTiers();
+  const auto fn = [&markedMoving, &itemFreed, numTiers](void* memory) {
     // Since this callback is executed, the item is not yet freed
     itemFreed = false;
     Item* item = static_cast<Item*>(memory);
-    if (item->markMoving(false)) {
+    // TODO: for chained items, moving bit is only used to avoid
+    // freeing the item prematurely
+    auto failIfRefNotZero = numTiers > 1 && !item->isChainedItem();
+    if (item->markMoving(failIfRefNotZero)) {
       markedMoving = true;
     }
   };