Make periodic worker start and stop util functions

Summary: This change is to avoid code duplication inside CacheAllocator and ObjectCache.

Reviewed By: jaesoo-fb

Differential Revision: D47372992

fbshipit-source-id: 0e365700a38ded6202ec42b7e5c621c77814a021

Author: Jiayue Bao

cachelib/allocator/CacheAllocator-inl.h

@@ -3419,17 +3419,7 @@ bool CacheAllocator<CacheTrait>::stopWorker(folly::StringPiece name,
                                             std::unique_ptr<T>& worker,
                                             std::chrono::seconds timeout) {
   std::lock_guard<std::mutex> l(workersMutex_);
-  if (!worker) {
-    return true;
-  }
-
-  bool ret = worker->stop(timeout);
-  if (ret) {
-    XLOGF(DBG1, "Stopped worker '{}'", name);
-  } else {
-    XLOGF(ERR, "Couldn't stop worker '{}', timeout: {} seconds", name,
-          timeout.count());
-  }
+  auto ret = util::stopPeriodicWorker(name, worker, timeout);
   worker.reset();
   return ret;
 }
@@ -3441,29 +3431,22 @@ bool CacheAllocator<CacheTrait>::startNewWorker(
     std::unique_ptr<T>& worker,
     std::chrono::milliseconds interval,
     Args&&... args) {
-  if (!stopWorker(name, worker)) {
+  if (worker && !stopWorker(name, worker)) {
     return false;
   }
 
   std::lock_guard<std::mutex> l(workersMutex_);
-  worker = std::make_unique<T>(*this, std::forward<Args>(args)...);
-  bool ret = worker->start(interval, name);
-  if (ret) {
-    XLOGF(DBG1, "Started worker '{}'", name);
-  } else {
-    XLOGF(ERR, "Couldn't start worker '{}', interval: {} milliseconds", name,
-          interval.count());
-  }
-  return ret;
+  return util::startPeriodicWorker(name, worker, interval,
+                                   std::forward<Args>(args)...);
 }
 
 template <typename CacheTrait>
 bool CacheAllocator<CacheTrait>::startNewPoolRebalancer(
     std::chrono::milliseconds interval,
     std::shared_ptr<RebalanceStrategy> strategy,
     unsigned int freeAllocThreshold) {
-  if (!startNewWorker("PoolRebalancer", poolRebalancer_, interval, strategy,
-                      freeAllocThreshold)) {
+  if (!startNewWorker("PoolRebalancer", poolRebalancer_, interval, *this,
+                      strategy, freeAllocThreshold)) {
     return false;
   }
 
@@ -3479,7 +3462,7 @@ bool CacheAllocator<CacheTrait>::startNewPoolResizer(
     std::chrono::milliseconds interval,
     unsigned int poolResizeSlabsPerIter,
     std::shared_ptr<RebalanceStrategy> strategy) {
-  if (!startNewWorker("PoolResizer", poolResizer_, interval,
+  if (!startNewWorker("PoolResizer", poolResizer_, interval, *this,
                       poolResizeSlabsPerIter, strategy)) {
     return false;
   }
@@ -3500,8 +3483,8 @@ bool CacheAllocator<CacheTrait>::startNewPoolOptimizer(
   // it should do actual size optimization. Probably need to move to using
   // the same interval for both, with confirmation of further experiments.
   const auto workerInterval = std::chrono::seconds(1);
-  if (!startNewWorker("PoolOptimizer", poolOptimizer_, workerInterval, strategy,
-                      regularInterval.count(), ccacheInterval.count(),
+  if (!startNewWorker("PoolOptimizer", poolOptimizer_, workerInterval, *this,
+                      strategy, regularInterval.count(), ccacheInterval.count(),
                       ccacheStepSizePercent)) {
     return false;
   }
@@ -3519,7 +3502,7 @@ bool CacheAllocator<CacheTrait>::startNewMemMonitor(
     std::chrono::milliseconds interval,
     MemoryMonitor::Config config,
     std::shared_ptr<RebalanceStrategy> strategy) {
-  if (!startNewWorker("MemoryMonitor", memMonitor_, interval, config,
+  if (!startNewWorker("MemoryMonitor", memMonitor_, interval, *this, config,
                       strategy)) {
     return false;
   }
@@ -3534,7 +3517,8 @@ template <typename CacheTrait>
 bool CacheAllocator<CacheTrait>::startNewReaper(
     std::chrono::milliseconds interval,
     util::Throttler::Config reaperThrottleConfig) {
-  if (!startNewWorker("Reaper", reaper_, interval, reaperThrottleConfig)) {
+  if (!startNewWorker("Reaper", reaper_, interval, *this,
+                      reaperThrottleConfig)) {
     return false;
   }
 

cachelib/common/PeriodicWorker.h

@@ -17,6 +17,7 @@
 #pragma once
 
 #include <folly/Range.h>
+#include <folly/logging/xlog.h>
 
 #include <atomic>
 #include <condition_variable>
@@ -140,5 +141,62 @@ class PeriodicWorker {
   /* The main worker loop that handles the work periodically */
   void loop(void);
 };
+
+namespace util {
+// Stop a periodic worker
+//
+// @param name       name of the worker
+// @param worker     unique pointer of the worker to stop
+// @param timeout    timeout for the worker stopping
+// @return true if size controller has been successfully stopped
+template <typename WorkerT>
+bool stopPeriodicWorker(folly::StringPiece name,
+                        std::unique_ptr<WorkerT>& worker,
+                        std::chrono::seconds timeout = std::chrono::seconds{
+                            0}) {
+  if (!worker) {
+    XLOGF(INFO, "Worker '{}' has not been started. No need to stop.", name);
+    return true;
+  }
+
+  bool ret = worker->stop(timeout);
+  if (ret) {
+    XLOGF(INFO, "Stopped worker '{}'", name);
+  } else {
+    XLOGF(ERR, "Couldn't stop worker '{}', timeout: {} seconds", name,
+          timeout.count());
+  }
+  return ret;
+}
+
+// Start a periodic worker
+//
+// @param name       name of the worker
+// @param worker     unique pointer of the worker to start
+// @param interval   the period this worker fires
+// @param args...    the rest of the arguments to initialize the worker
+// @return true if the worker has been successfully started
+template <typename WorkerT, typename... Args>
+bool startPeriodicWorker(folly::StringPiece name,
+                         std::unique_ptr<WorkerT>& worker,
+                         std::chrono::milliseconds interval,
+                         Args&&... args) {
+  if (worker && !stopPeriodicWorker(name, worker)) {
+    XLOGF(ERR, "Couldn't restart worker '{}' because it couldn't be stopped",
+          name);
+    return false;
+  }
+  worker = std::make_unique<WorkerT>(std::forward<Args>(args)...);
+  bool ret = worker->start(interval, name);
+  if (ret) {
+    XLOGF(INFO, "Started worker '{}'", name);
+  } else {
+    XLOGF(ERR, "Couldn't start worker '{}', interval: {} milliseconds", name,
+          interval.count());
+  }
+  return ret;
+}
+
+} // namespace util
 } // namespace cachelib
 } // namespace facebook

cachelib/experimental/objcache2/ObjectCache-inl.h

@@ -110,15 +110,17 @@ template <typename AllocatorT>
 void ObjectCache<AllocatorT>::initWorkers() {
   if (config_.objectSizeTrackingEnabled &&
       config_.sizeControllerIntervalMs != 0) {
-    startWorker(kSizeControllerName, sizeController_,
-                std::chrono::milliseconds{config_.sizeControllerIntervalMs},
-                config_.sizeControllerThrottlerConfig);
+    util::startPeriodicWorker(
+        kSizeControllerName, sizeController_,
+        std::chrono::milliseconds{config_.sizeControllerIntervalMs}, *this,
+        config_.sizeControllerThrottlerConfig);
   }
 
   if (config_.objectSizeTrackingEnabled &&
       config_.objectSizeDistributionTrackingEnabled) {
-    startWorker(kSizeDistTrackerName, sizeDistTracker_,
-                std::chrono::seconds{60} /*default interval to be 60s*/);
+    util::startPeriodicWorker(
+        kSizeDistTrackerName, sizeDistTracker_,
+        std::chrono::seconds{60} /*default interval to be 60s*/, *this);
   }
 }
 
@@ -371,48 +373,6 @@ ObjectCache<AllocatorT>::serializeConfigParams() const {
   return config;
 }
 
-template <typename AllocatorT>
-template <typename WorkerT, typename... Args>
-bool ObjectCache<AllocatorT>::startWorker(folly::StringPiece name,
-                                          std::unique_ptr<WorkerT>& worker,
-                                          std::chrono::milliseconds interval,
-                                          Args&&... args) {
-  if (!stopWorker(name, worker)) {
-    XLOGF(ERR, "Worker '{}' is already running. Cannot start it again.", name);
-    return false;
-  }
-
-  worker = std::make_unique<WorkerT>(*this, std::forward<Args>(args)...);
-  bool ret = worker->start(interval, name);
-  if (ret) {
-    XLOGF(INFO, "Started worker '{}'", name);
-  } else {
-    XLOGF(ERR, "Couldn't start worker '{}', interval: {} milliseconds", name,
-          interval.count());
-  }
-  return ret;
-}
-
-template <typename AllocatorT>
-template <typename WorkerT>
-bool ObjectCache<AllocatorT>::stopWorker(folly::StringPiece name,
-                                         std::unique_ptr<WorkerT>& worker,
-                                         std::chrono::seconds timeout) {
-  if (!worker) {
-    XLOGF(INFO, "Worker '{}' has not been started. No need to stop.", name);
-    return true;
-  }
-
-  bool ret = worker->stop(timeout);
-  if (ret) {
-    XLOGF(INFO, "Stopped worker '{}'", name);
-  } else {
-    XLOGF(ERR, "Couldn't stop worker '{}', timeout: {} seconds", name,
-          timeout.count());
-  }
-  return ret;
-}
-
 template <typename AllocatorT>
 bool ObjectCache<AllocatorT>::persist() {
   if (config_.persistBaseFilePath.empty() || !config_.serializeCb) {

cachelib/experimental/objcache2/ObjectCache.h

@@ -406,8 +406,10 @@ class ObjectCache : public ObjectCacheBase<AllocatorT> {
   // @return true if all workers have been successfully stopped
   bool stopAllWorkers(std::chrono::seconds timeout = std::chrono::seconds{0}) {
     bool success = true;
-    success &= stopWorker(kSizeControllerName, sizeController_, timeout);
-    success &= stopWorker(kSizeDistTrackerName, sizeDistTracker_, timeout);
+    success &=
+        util::stopPeriodicWorker(kSizeControllerName, sizeController_, timeout);
+    success &= util::stopPeriodicWorker(
+        kSizeDistTrackerName, sizeDistTracker_, timeout);
     success &= this->l1Cache_->stopWorkers(timeout);
     return success;
   }
@@ -443,30 +445,6 @@ class ObjectCache : public ObjectCacheBase<AllocatorT> {
   // Returns the total number of placeholders
   size_t getNumPlaceholders() const { return placeholders_.size(); }
 
-  // Start a periodic worker
-  //
-  // @param name       name of the worker
-  // @param worker     unique pointer of the worker to start
-  // @param interval   the period this worker fires
-  // @param args...    the rest of the arguments to initialize the worker
-  // @return true if the worker has been successfully started
-  template <typename WorkerT, typename... Args>
-  bool startWorker(folly::StringPiece name,
-                   std::unique_ptr<WorkerT>& worker,
-                   std::chrono::milliseconds interval,
-                   Args&&... args);
-
-  // Stop a periodic worker
-  //
-  // @param name       name of the worker
-  // @param worker     unique pointer of the worker to stop
-  // @param timeout    timeout for the worker stopping
-  // @return true if size controller has been successfully stopped
-  template <typename WorkerT>
-  bool stopWorker(folly::StringPiece name,
-                  std::unique_ptr<WorkerT>& worker,
-                  std::chrono::seconds timeout = std::chrono::seconds{0});
-
   // Get a ReadHandle reference from the object shared_ptr
   template <typename T>
   typename AllocatorT::ReadHandle& getReadHandleRefInternal(