Move size distribution tracking logic out of size-controller

Summary: Use a separate worker to track and upload size distribution stats.

Reviewed By: jaesoo-fb

Differential Revision: D47028003

fbshipit-source-id: 931f17a09d79365a02771ac18dde165a56bd46e8

Author: Jiayue Bao

cachelib/experimental/objcache2/ObjectCache-inl.h

@@ -103,11 +103,22 @@ void ObjectCache<AllocatorT>::init() {
     }
   }
 
+  initWorkers();
+}
+
+template <typename AllocatorT>
+void ObjectCache<AllocatorT>::initWorkers() {
   if (config_.objectSizeTrackingEnabled &&
       config_.sizeControllerIntervalMs != 0) {
-    startSizeController(
-        std::chrono::milliseconds{config_.sizeControllerIntervalMs},
-        config_.sizeControllerThrottlerConfig);
+    startWorker(kSizeControllerName, sizeController_,
+                std::chrono::milliseconds{config_.sizeControllerIntervalMs},
+                config_.sizeControllerThrottlerConfig);
+  }
+
+  if (config_.objectSizeTrackingEnabled &&
+      config_.objectSizeDistributionTrackingEnabled) {
+    startWorker(kSizeDistTrackerName, sizeDistTracker_,
+                std::chrono::seconds{60} /*default interval to be 60s*/);
   }
 }
 
@@ -304,7 +315,7 @@ uint32_t ObjectCache<AllocatorT>::getL1AllocSize(uint8_t maxKeySizeBytes) {
 
 template <typename AllocatorT>
 ObjectCache<AllocatorT>::~ObjectCache() {
-  stopSizeController();
+  stopAllWorkers();
 
   for (auto itr = this->l1Cache_->begin(); itr != this->l1Cache_->end();
        ++itr) {
@@ -339,6 +350,10 @@ void ObjectCache<AllocatorT>::getObjectCacheCounters(
   if (sizeController_) {
     sizeController_->getCounters(visitor);
   }
+
+  if (sizeDistTracker_) {
+    sizeDistTracker_->getCounters(visitor);
+  }
 }
 
 template <typename AllocatorT>
@@ -357,41 +372,44 @@ ObjectCache<AllocatorT>::serializeConfigParams() const {
 }
 
 template <typename AllocatorT>
-bool ObjectCache<AllocatorT>::startSizeController(
-    std::chrono::milliseconds interval, const util::Throttler::Config& config) {
-  if (!stopSizeController()) {
-    XLOG(ERR) << "Size controller is already running. Cannot start it again.";
+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;
   }
 
-  sizeController_ =
-      std::make_unique<ObjectCacheSizeController<AllocatorT>>(*this, config);
-  bool ret = sizeController_->start(interval, "ObjectCache-SizeController");
+  worker = std::make_unique<WorkerT>(*this, std::forward<Args>(args)...);
+  bool ret = worker->start(interval, name);
   if (ret) {
-    XLOG(DBG) << "Started ObjectCache SizeController";
+    XLOGF(INFO, "Started worker '{}'", name);
   } else {
-    XLOGF(
-        ERR,
-        "Couldn't start ObjectCache SizeController, interval: {} milliseconds",
-        interval.count());
+    XLOGF(ERR, "Couldn't start worker '{}', interval: {} milliseconds", name,
+          interval.count());
   }
   return ret;
 }
 
 template <typename AllocatorT>
-bool ObjectCache<AllocatorT>::stopSizeController(std::chrono::seconds timeout) {
-  if (!sizeController_) {
+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 = sizeController_->stop(timeout);
+  bool ret = worker->stop(timeout);
   if (ret) {
-    XLOG(DBG) << "Stopped ObjectCache SizeController";
+    XLOGF(INFO, "Stopped worker '{}'", name);
   } else {
-    XLOGF(ERR, "Couldn't stop ObjectCache SizeController, timeout: {} seconds",
+    XLOGF(ERR, "Couldn't stop worker '{}', timeout: {} seconds", name,
           timeout.count());
   }
-  sizeController_.reset();
   return ret;
 }
 
@@ -402,11 +420,7 @@ bool ObjectCache<AllocatorT>::persist() {
   }
 
   // Stop all the other workers before persist
-  if (!stopSizeController()) {
-    return false;
-  }
-
-  if (!this->l1Cache_->stopWorkers()) {
+  if (!stopAllWorkers()) {
     return false;
   }
 

cachelib/experimental/objcache2/ObjectCache.h

@@ -37,6 +37,7 @@
 #include "cachelib/experimental/objcache2/ObjectCacheBase.h"
 #include "cachelib/experimental/objcache2/ObjectCacheConfig.h"
 #include "cachelib/experimental/objcache2/ObjectCacheSizeController.h"
+#include "cachelib/experimental/objcache2/ObjectCacheSizeDistTracker.h"
 #include "cachelib/experimental/objcache2/persistence/Persistence.h"
 #include "cachelib/experimental/objcache2/persistence/gen-cpp2/persistent_data_types.h"
 #include "cachelib/experimental/objcache2/util/ThreadMemoryTracker.h"
@@ -159,6 +160,7 @@ class ObjectCache : public ObjectCacheBase<AllocatorT> {
   using Persistor = Persistor<ObjectCache<AllocatorT>>;
   using Restorer = Restorer<ObjectCache<AllocatorT>>;
   using EvictionIterator = typename AllocatorT::EvictionIterator;
+  using AccessIterator = typename AllocatorT::AccessIterator;
 
   enum class AllocStatus { kSuccess, kAllocError, kKeyAlreadyExists };
 
@@ -271,6 +273,11 @@ class ObjectCache : public ObjectCacheBase<AllocatorT> {
   // This is only used in tests.
   AllocatorT& getL1Cache() { return *this->l1Cache_; }
 
+  // Get an iterator to iterate over all items in object-cache.
+  AccessIterator begin() { return this->l1Cache_->begin(); }
+
+  AccessIterator end() { return this->l1Cache_->end(); }
+
   // Get the default l1 allocation size in bytes.
   static uint32_t getL1AllocSize(uint8_t maxKeySizeBytes);
 
@@ -386,6 +393,25 @@ class ObjectCache : public ObjectCacheBase<AllocatorT> {
         ->objectSize;
   }
 
+  size_t getObjectSize(AccessIterator& itr) const {
+    if (!itr.asHandle()) {
+      return 0;
+    }
+    return reinterpret_cast<const ObjectCacheItem*>(itr.asHandle()->getMemory())
+        ->objectSize;
+  }
+
+  // Stop all workers
+  //
+  // @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 &= this->l1Cache_->stopWorkers(timeout);
+    return success;
+  }
+
  protected:
   // Serialize cache allocator config for exporting to Scuba
   std::map<std::string, std::string> serializeConfigParams() const override;
@@ -395,8 +421,14 @@ class ObjectCache : public ObjectCacheBase<AllocatorT> {
   static constexpr uint32_t kL1AllocSizeMin = 64;
   static constexpr const char* kPlaceholderKey = "_cl_ph";
 
+  // Names of periodic workers
+  static constexpr folly::StringPiece kSizeControllerName{"SizeController"};
+  static constexpr folly::StringPiece kSizeDistTrackerName{"SizeDistTracker"};
+
   void init();
 
+  void initWorkers();
+
   // Allocate an item handle from the interal cache allocator. This item's
   // storage is used to cache pointer to objects in object-cache.
   typename AllocatorT::WriteHandle allocateFromL1(folly::StringPiece key,
@@ -411,19 +443,29 @@ class ObjectCache : public ObjectCacheBase<AllocatorT> {
   // Returns the total number of placeholders
   size_t getNumPlaceholders() const { return placeholders_.size(); }
 
-  // Start size controller
+  // 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 config     throttling config
-  // @return true if size controller has been successfully started
-  bool startSizeController(std::chrono::milliseconds interval,
-                           const util::Throttler::Config& config);
-
-  // Stop size controller
+  // @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
-  bool stopSizeController(std::chrono::seconds timeout = std::chrono::seconds{
-                              0});
+  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>
@@ -462,6 +504,10 @@ class ObjectCache : public ObjectCacheBase<AllocatorT> {
   // cache size limit
   std::unique_ptr<ObjectCacheSizeController<AllocatorT>> sizeController_;
 
+  // A periodic worker that tracks the object size distribution stats.
+  std::unique_ptr<ObjectCacheSizeDistTracker<ObjectCache<AllocatorT>>>
+      sizeDistTracker_;
+
   // Actual object size in total
   std::atomic<size_t> totalObjectSizeBytes_{0};
 

cachelib/experimental/objcache2/ObjectCacheSizeController-inl.h

@@ -72,10 +72,6 @@ void ObjectCacheSizeController<AllocatorT>::work() {
                    totalObjSize, currentNumEntries, averageObjSize,
                    newEntriesLimit, currentEntriesLimit_);
   }
-
-  if (objCache_.config_.objectSizeDistributionTrackingEnabled) {
-    trackObjectSizeDistributionStats();
-  }
 }
 
 template <typename AllocatorT>
@@ -131,17 +127,6 @@ void ObjectCacheSizeController<AllocatorT>::getCounters(
     visitor("objcache.jemalloc_active_bytes", jemallocActiveBytes);
     visitor("objcache.jemalloc_allocated_bytes", jemallocAllocatedBytes);
   }
-  if (objCache_.config_.objectSizeDistributionTrackingEnabled) {
-    constexpr folly::StringPiece fmt = "{}_p{}";
-    constexpr folly::StringPiece prefix =
-        "objcache.size_distribution.object_size_bytes";
-
-    for (auto quantile : objectSizeBytesHist_.quantiles()) {
-      visitor(
-          folly::sformat(fmt, prefix, static_cast<uint32_t>(quantile * 100)),
-          objectSizeBytesHist_.estimateQuantile(quantile));
-    }
-  }
 }
 
 template <typename AllocatorT>

cachelib/experimental/objcache2/ObjectCacheSizeController.h

@@ -17,7 +17,6 @@
 #pragma once
 
 #include <folly/memory/Malloc.h>
-#include <folly/stats/QuantileHistogram.h>
 
 #include "cachelib/common/PeriodicWorker.h"
 
@@ -58,18 +57,6 @@ class ObjectCacheSizeController : public PeriodicWorker {
     return {jemallocAllocatedBytes, jemallocActiveBytes};
   }
 
-  void trackObjectSizeDistributionStats() {
-    // scan the cache to get the object size
-    for (auto itr = objCache_.l1Cache_->begin();
-         itr != objCache_.l1Cache_->end();
-         ++itr) {
-      size_t objectSize = reinterpret_cast<const typename ObjectCache::Item*>(
-                              itr.asHandle()->getMemory())
-                              ->objectSize;
-      objectSizeBytesHist_.addValue(objectSize);
-    }
-  }
-
   // threshold in percentage to determine whether the size-controller should do
   // the calculation
   const size_t kSizeControllerThresholdPct = 50;
@@ -81,8 +68,6 @@ class ObjectCacheSizeController : public PeriodicWorker {
 
   // will be adjusted to control the cache size limit
   std::atomic<size_t> currentEntriesLimit_;
-
-  folly::QuantileHistogram<> objectSizeBytesHist_{};
 };
 
 } // namespace objcache2

cachelib/experimental/objcache2/ObjectCacheSizeDistTracker.h

@@ -0,0 +1,68 @@
+/*
+ * 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/stats/QuantileHistogram.h>
+
+#include "cachelib/common/PeriodicWorker.h"
+
+namespace facebook {
+namespace cachelib {
+namespace objcache2 {
+
+template <typename ObjectCache>
+class ObjectCacheSizeDistTracker : public PeriodicWorker {
+ public:
+  explicit ObjectCacheSizeDistTracker(ObjectCache& objCache)
+      : objCache_(objCache),
+        objectSizeBytesHist_{std::make_shared<folly::QuantileHistogram<>>()} {}
+
+  void getCounters(const util::CounterVisitor& visitor) const {
+    std::shared_ptr<folly::QuantileHistogram<>> curHist;
+    {
+      // lock the mutex before accessing objectSizeBytesHist_
+      std::lock_guard<std::mutex> l(mutex_);
+      curHist = objectSizeBytesHist_;
+    }
+
+    for (auto quantile : curHist->quantiles()) {
+      visitor(fmt::format("objcache.size_distribution.object_size_bytes_p{}",
+                          static_cast<uint32_t>(quantile * 100)),
+              curHist->estimateQuantile(quantile));
+    }
+  }
+
+ private:
+  void work() override final {
+    auto newHist = std::make_shared<folly::QuantileHistogram<>>();
+    // scan the cache to get the object size
+    for (auto itr = objCache_.begin(); itr != objCache_.end(); ++itr) {
+      newHist->addValue(objCache_.getObjectSize(itr));
+    }
+    // lock the mutex before updating objectSizeBytesHist_
+    std::lock_guard<std::mutex> l(mutex_);
+    objectSizeBytesHist_ = newHist;
+  }
+
+  ObjectCache& objCache_;
+  mutable std::mutex mutex_;
+  std::shared_ptr<folly::QuantileHistogram<>> objectSizeBytesHist_{};
+};
+
+} // namespace objcache2
+} // namespace cachelib
+} // namespace facebook