added inject_pause framework for unit testing

Summary:
This change adds the inject_pause points which provide a mechanism to do a rendezvous style
synchronization. This is required for unit testing to control the executions of the code from the
unit tests or install hooks to insert some custom callbacks in the code path.

Reviewed By: therealgymmy

Differential Revision: D48706614

fbshipit-source-id: 44b4bfa5ce88b53c096c907310e494a066e7dc1a

Author: Jaesoo Lee

cachelib/common/inject_pause.cpp

@@ -0,0 +1,228 @@
+/*
+ * Copyright (c) Meta Platforms, Inc. and affiliates.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include "inject_pause.h"
+
+#include <folly/Indestructible.h>
+#include <folly/Synchronized.h>
+#include <folly/fibers/Baton.h>
+#include <folly/logging/xlog.h>
+
+#include <condition_variable>
+#include <deque>
+#include <mutex>
+#include <string>
+
+namespace facebook {
+namespace cachelib {
+
+namespace {
+struct BatonSet {
+  std::deque<folly::fibers::Baton*> batons;
+  std::condition_variable cv;
+  bool enabled{false};
+  size_t limit{0};
+  std::shared_ptr<PauseCallback> callback;
+};
+
+using PausePointsMap =
+    folly::Synchronized<std::unordered_map<std::string, BatonSet>, std::mutex>;
+PausePointsMap& pausePoints() {
+  static folly::Indestructible<PausePointsMap> pausePoints;
+  return *pausePoints;
+}
+
+size_t wakeupBatonSet(BatonSet& batonSet, size_t numThreads) {
+  size_t numWaked = 0;
+  if (numThreads == 0) {
+    numThreads = batonSet.batons.size();
+  }
+
+  while (!batonSet.batons.empty() && numWaked < numThreads) {
+    auto* b = batonSet.batons.front();
+    batonSet.batons.pop_front();
+    b->post();
+    numWaked++;
+  }
+
+  return numWaked;
+}
+
+} // namespace
+
+bool& injectPauseEnabled() {
+  static bool enabled = false;
+  return enabled;
+}
+
+// Flag controls the debug logging
+bool& injectPauseLogEnabled() {
+  static bool enabled = false;
+  return enabled;
+}
+
+namespace detail {
+
+void injectPause(folly::StringPiece name) {
+  if (!injectPauseEnabled()) {
+    return;
+  }
+  folly::fibers::Baton baton;
+  std::shared_ptr<PauseCallback> callback;
+  {
+    auto ptr = pausePoints().lock();
+    auto it = ptr->find(name.str());
+    if (it == ptr->end() || !it->second.enabled ||
+        (it->second.limit > 0 &&
+         it->second.batons.size() == it->second.limit)) {
+      if (injectPauseLogEnabled()) {
+        XLOGF(ERR, "[{}] injectPause not set", name);
+      }
+      return;
+    }
+    if (injectPauseLogEnabled()) {
+      XLOGF(ERR, "[{}] injectPause begin", name);
+    }
+    callback = it->second.callback;
+    if (!callback) {
+      it->second.batons.push_back(&baton);
+      it->second.cv.notify_one();
+    }
+  }
+
+  if (callback) {
+    (*callback)();
+  } else {
+    baton.wait();
+  }
+
+  /* Avoid potential protect-its-own-lifetime bug:
+     Wait for the post() to finish before destroying the Baton. */
+  auto ptr = pausePoints().lock();
+  if (injectPauseLogEnabled()) {
+    XLOGF(ERR, "[{}] injectPause end", name);
+  }
+}
+
+} // namespace detail
+
+void injectPauseSet(folly::StringPiece name, size_t numThreads) {
+  if (!injectPauseEnabled()) {
+    return;
+  }
+  auto ptr = pausePoints().lock();
+  if (injectPauseLogEnabled()) {
+    XLOGF(ERR, "[{}] injectPauseSet threads {}", name, numThreads);
+  }
+  auto res = ptr->emplace(
+      std::piecewise_construct, std::make_tuple(name.str()), std::make_tuple());
+  res.first->second.limit = numThreads;
+  res.first->second.enabled = true;
+}
+
+void injectPauseSet(folly::StringPiece name, PauseCallback&& callback) {
+  if (!injectPauseEnabled()) {
+    return;
+  }
+  auto ptr = pausePoints().lock();
+  if (injectPauseLogEnabled()) {
+    XLOGF(ERR, "[{}] injectPauseSet callback", name);
+  }
+  auto res = ptr->emplace(
+      std::piecewise_construct, std::make_tuple(name.str()), std::make_tuple());
+  res.first->second.limit = 0;
+  res.first->second.enabled = true;
+  res.first->second.callback =
+      std::make_shared<PauseCallback>(std::move(callback));
+}
+
+bool injectPauseWait(folly::StringPiece name,
+                     size_t numThreads,
+                     bool wakeup,
+                     uint32_t timeoutMs) {
+  if (!injectPauseEnabled() || !numThreads) {
+    return false;
+  }
+  auto ptr = pausePoints().lock();
+  if (injectPauseLogEnabled()) {
+    XLOGF(ERR, "[{}] injectPauseWait begin {}", name, numThreads);
+  }
+  auto it = ptr->find(name.str());
+  if (it == ptr->end() || !it->second.enabled) {
+    if (injectPauseLogEnabled()) {
+      XLOGF(ERR, "[{}] injectPauseWait: ERROR not set", name);
+    }
+    return false;
+  }
+
+  if (!timeoutMs || timeoutMs > kInjectPauseMaxWaitTimeoutMs) {
+    timeoutMs = kInjectPauseMaxWaitTimeoutMs;
+  }
+
+  auto& batonSet = it->second;
+  bool status = batonSet.cv.wait_for(
+      ptr.as_lock(),
+      std::chrono::milliseconds(timeoutMs),
+      [&batonSet, numThreads]() {
+        return !batonSet.enabled || batonSet.batons.size() >= numThreads;
+      });
+
+  if (status && wakeup) {
+    wakeupBatonSet(batonSet, numThreads);
+  }
+
+  if (injectPauseLogEnabled()) {
+    std::string errStr;
+    if (!status) {
+      errStr = fmt::format(" with ERR (paused {})", batonSet.batons.size());
+    }
+    XLOGF(ERR, "[{}] injectPauseWait end {}", name, errStr);
+  }
+  return status;
+}
+
+size_t injectPauseClear(folly::StringPiece name) {
+  if (!injectPauseEnabled()) {
+    return false;
+  }
+
+  size_t numPaused = 0;
+  auto ptr = pausePoints().lock();
+  if (injectPauseLogEnabled()) {
+    XLOGF(ERR, "[{}] injectPauseClear ", name);
+  }
+
+  if (name.empty()) {
+    for (auto& it : *ptr) {
+      auto& batonSet = it.second;
+      numPaused += wakeupBatonSet(batonSet, 0);
+      batonSet.enabled = false;
+      batonSet.cv.notify_all();
+    }
+  } else {
+    auto it = ptr->find(name.str());
+    if (it != ptr->end()) {
+      auto& batonSet = it->second;
+      numPaused += wakeupBatonSet(batonSet, 0) > 0 ? 1 : 0;
+      batonSet.enabled = false;
+      batonSet.cv.notify_all();
+    }
+  }
+  return numPaused;
+}
+
+} // namespace cachelib
+} // namespace facebook

cachelib/common/inject_pause.h

@@ -0,0 +1,156 @@
+/*
+ * Copyright (c) Meta Platforms, Inc. and affiliates.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#pragma once
+
+#include <folly/Function.h>
+#include <folly/Range.h>
+
+/*
+ * inject_pause framework provides a rendezvous-style syncronization
+ * for debugging and testing purposes. This allows us to control
+ * executions of multiple threads deterministically.
+ *
+ * The basic idea is to have a named point in code where we want to
+ * block execution of some threads. We then allow other threads,
+ * probably, the test threads, to synchronize with the main threads
+ * at the pause points and signal the threads to proceed.
+ *
+ * The basic usage is: (e.g., RegionManager.ReadWrite unit test)
+ * 1. Insert a call to INJECT_PAUSE() in the code
+ *    void RegionManager::doReclaim() {
+ *      ...
+ *      INJECT_PAUSE("pause_reclaim_done");
+ *    }
+ *
+ * 2. Enable INJECT_PAUSE() using injectPauseEnabled() in unit test
+ *    TEST(RegionManager, ReadWrite) {
+ *        ...
+ *        injectPauseEnabled() = true;
+ *
+ * 3. In the test, we can wait for the threads to reach the pause point
+ *
+ *        ...
+ *        rm->startReclaim();
+ *        EXPECT_TRUE(injectPauseWait("pause_reclaim_done"));
+ *        ASSERT_EQ(OpenStatus::Ready, rm->getCleanRegion(rid, false).first);
+ *        ASSERT_EQ(0, rid.index());
+ *
+ * 4. Once done, clear the pause point
+ *
+ *        ASSERT_EQ(0, injectPauseClear("pause_reclaim_done"));
+ *        ...
+ *    }
+ *
+ * In above example, the region reclaim thread and test thread are synchronized
+ * at the pause point and injectPauseWait() and continue.
+ *
+ * The inject_pause framework also supports setting a callback function where
+ * the threads can register a callback function to be called when they hit the
+ * INJECT_PAUSE() point.
+ *
+ * The example usage with the callback function is as follows:
+ *
+ *     folly::fibers::TimedMutex mutex;
+ *     bool reclaimStarted = false;
+ *     size_t numCleanRegions = 0;
+ *     util::ConditionVariable cv;
+ *
+ *     injectPauseSet("pause_blockcache_clean_alloc", [&]() {
+ *       std::unique_lock<folly::fibers::TimedMutex> lk(mutex);
+ *       XDCHECK_GT(numCleanRegions, 0u);
+ *       numCleanRegions--;
+ *     });
+ *
+ *     injectPauseSet("pause_blockcache_clean_free", [&]() {
+ *       std::unique_lock<folly::fibers::TimedMutex> lk(mutex);
+ *       if (numCleanRegions++ == 0u) {
+ *         cv.notifyAll();
+ *       }
+ *       reclaimStarted = true;
+ *     });
+ *
+ *     injectPauseSet("pause_block_cache_insert_entry", [&]() {
+ *       std::unique_lock<folly::fibers::TimedMutex> lk(mutex);
+ *       if (numCleanRegions == 0u && reclaimStarted) {
+ *         cv.wait(lk);
+ *       }
+ *     });
+ *
+ * In the above example, the callbacks count the number of clean regions
+ * and allows pause_blockcache_clean_alloc to continue only if there is
+ * at least one clean region.
+ */
+namespace facebook {
+namespace cachelib {
+namespace detail {
+void injectPause(folly::StringPiece name);
+}
+
+/**
+ * Allow stopping the process at this named point.
+ */
+#define INJECT_PAUSE(name) facebook::cachelib::detail::injectPause(#name)
+
+// Callback that can be executed optionally for INJECT_PAUSE point
+using PauseCallback = folly::Function<void()>;
+
+// Default timeout for INJECT_PAUSE wait
+constexpr uint32_t kInjectPauseMaxWaitTimeoutMs = 60000;
+
+/**
+ * Toggle INJECT_PAUSE() logic on and off
+ */
+bool& injectPauseEnabled();
+
+/**
+ * Stop any thread at this named INJECT_PAUSE point from now on.
+ * @param numThreads  If nonzero, only stop the first N threads.
+ */
+void injectPauseSet(folly::StringPiece name, size_t numThreads = 0);
+
+/**
+ * The thread will execute the callback at INJECT_PAUSE point and go.
+ * @param callback Callback to be executed at INJECT_PAUSE point
+ */
+void injectPauseSet(folly::StringPiece name, PauseCallback&& callback);
+
+/**
+ * If the named INJECT_PAUSE point was set, blocks until the requested
+ * number of threads is stopped at it including those already stopped
+ * and returns true.
+ * @param name       Name of the INJECT_PAUSE point
+ * @param numThreads The number of threads to wait for
+ * @param wakeup     If true, wakes up any threads blocked at the point
+ * @param timeoutMs  If nonzero, waits no longer than this many milliseconds
+ * @return           true if waited for requested number of threads successfully
+ *                   false otherwise or timed out
+ */
+bool injectPauseWait(folly::StringPiece name,
+                     size_t numThreads = 1,
+                     bool wakeup = true,
+                     uint32_t timeoutMs = 0);
+
+/**
+ * Stop blocking threads at this INJECT_PAUSE point and unblock any
+ * currently waiting threads. If name is not given, all of
+ * INJECT_PAUSE points are cleared.
+ * Returns the number of INJECT_PAUSE points where threads were stopped
+ */
+size_t injectPauseClear(folly::StringPiece name = "");
+
+} // namespace cachelib
+} // namespace facebook