Allow configuring the number of allocators per

Summary: To avoid the mutex of region allocator being the single point of contention, introduce a config to create multiple region allocators.

Reviewed By: pbhandar2

Differential Revision: D73201432

fbshipit-source-id: f9395e23737d6b01fe93b7c4c124e5ee295ff359

Author: haowu14

cachelib/allocator/nvmcache/NavyConfig.h

@@ -273,9 +273,16 @@ class BlockCacheConfig {
   //             {1, 2, 3} gives the 1/6th of the items in the first segment (P0
   //             least important), 2/6th of the items in the second segment
   //             (P1), and finally 3/6th of the items in the third segment (P2).
+  // @param Number of allocators for each priority. If not set, each priority
+  // would have one allocator.
   BlockCacheConfig& enableSegmentedFifo(
-      std::vector<unsigned int> sFifoSegmentRatio) noexcept {
+      std::vector<unsigned int> sFifoSegmentRatio,
+      std::vector<uint32_t> allocatorCounts = {}) noexcept {
     sFifoSegmentRatio_ = std::move(sFifoSegmentRatio);
+    if (allocatorCounts.size() > 0) {
+      XDCHECK_EQ(sFifoSegmentRatio_.size(), allocatorCounts.size());
+      allocatorsPerPriority_ = std::move(allocatorCounts);
+    }
     lru_ = false;
     return *this;
   }
@@ -342,6 +349,11 @@ class BlockCacheConfig {
     return *this;
   }
 
+  BlockCacheConfig& setAllocatorCount(uint32_t numAllocators) noexcept {
+    allocatorsPerPriority_ = {numAllocators};
+    return *this;
+  }
+
   bool isLruEnabled() const { return lru_; }
 
   const std::vector<unsigned int>& getSFifoSegmentRatio() const {
@@ -368,6 +380,10 @@ class BlockCacheConfig {
 
   bool isPreciseRemove() const { return preciseRemove_; }
 
+  const std::vector<uint32_t>& getNumAllocatorsPerPriority() const {
+    return allocatorsPerPriority_;
+  }
+
  private:
   // Whether Navy BlockCache will use region-based LRU eviction policy.
   bool lru_{true};
@@ -400,6 +416,11 @@ class BlockCacheConfig {
   // Whether the region manager workers flushes asynchronously.
   bool regionManagerFlushAsync_{false};
 
+  // Number of allocators per priority.
+  // Do not set this directly. This should be configured by setAllocatorCount
+  // for FIFO and LRU, and enableSegmentedFifio for segmented FIFO.
+  std::vector<uint32_t> allocatorsPerPriority_{1};
+
   friend class NavyConfig;
 };
 

cachelib/allocator/nvmcache/NavySetup.cpp

@@ -174,6 +174,8 @@ uint64_t setupBlockCache(const navy::BlockCacheConfig& blockCacheConfig,
                                   blockCacheConfig.getCleanRegionThreads());
   blockCache->setRegionManagerFlushAsync(
       blockCacheConfig.isRegionManagerFlushAsync());
+  blockCache->setNumAllocatorsPerPriority(
+      blockCacheConfig.getNumAllocatorsPerPriority());
 
   blockCache->setReinsertionConfig(blockCacheConfig.getReinsertionConfig());
 

cachelib/navy/Factory.cpp

@@ -80,11 +80,15 @@ class BlockCacheProtoImpl final : public BlockCacheProto {
     if (config_.evictionPolicy) {
       throw std::invalid_argument("There's already an eviction policy set");
     }
-    config_.numPriorities = static_cast<uint16_t>(segmentRatio.size());
     config_.evictionPolicy =
         std::make_unique<SegmentedFifoPolicy>(std::move(segmentRatio));
   }
 
+  void setNumAllocatorsPerPriority(
+      std::vector<uint32_t> allocatorsPerPriority) override {
+    config_.allocatorsPerPriority = std::move(allocatorsPerPriority);
+  }
+
   void setReadBufferSize(uint32_t size) override {
     config_.readBufferSize = size;
   }

cachelib/navy/Factory.h

@@ -89,6 +89,10 @@ class BlockCacheProto {
 
   // (Optional) Set if the region manager worker flush should be async.
   virtual void setRegionManagerFlushAsync(bool flushAsync) = 0;
+
+  // Set number of allocators per priority.
+  virtual void setNumAllocatorsPerPriority(
+      std::vector<uint32_t> numAllocatorsPerPriority) = 0;
 };
 
 // BigHash engine proto. BigHash is used to cache small objects (under 2KB)

cachelib/navy/block_cache/Allocator.cpp

@@ -32,21 +32,27 @@ void RegionAllocator::setAllocationRegion(RegionId rid) {
 
 void RegionAllocator::reset() { rid_ = RegionId{}; }
 
-Allocator::Allocator(RegionManager& regionManager, uint16_t numPriorities)
+Allocator::Allocator(RegionManager& regionManager,
+                     const std::vector<uint32_t>& allocatorsPerPriority)
     : regionManager_{regionManager} {
+  size_t numPriorities = allocatorsPerPriority.size();
   XLOGF(INFO,
         "Enable priority-based allocation for Allocator. Number of "
         "priorities: {}",
         numPriorities);
   for (uint16_t i = 0; i < numPriorities; i++) {
-    allocators_.emplace_back(i /* priority */);
+    allocators_.emplace_back();
+    for (uint32_t j = 0; j < allocatorsPerPriority[i]; j++) {
+      allocators_.back().emplace_back(i /* priority */);
+    }
   }
 }
 
 std::tuple<RegionDescriptor, uint32_t, RelAddress> Allocator::allocate(
-    uint32_t size, uint16_t priority, bool canWait) {
+    uint32_t size, uint16_t priority, bool canWait, uint64_t keyHash) {
   XDCHECK_LT(priority, allocators_.size());
-  RegionAllocator* ra = &allocators_[priority];
+  RegionAllocator* ra =
+      &allocators_[priority][keyHash % allocators_[priority].size()];
   if (size == 0 || size > regionManager_.regionSize()) {
     return std::make_tuple(RegionDescriptor{OpenStatus::Error}, size,
                            RelAddress());
@@ -125,18 +131,22 @@ void Allocator::flushAndReleaseRegionFromRALocked(RegionAllocator& ra,
 }
 
 void Allocator::flush() {
-  for (auto& ra : allocators_) {
-    std::lock_guard<TimedMutex> lock{ra.getLock()};
-    flushAndReleaseRegionFromRALocked(ra, false /* async */);
+  for (auto& ras : allocators_) {
+    for (auto& ra : ras) {
+      std::lock_guard<TimedMutex> lock{ra.getLock()};
+      flushAndReleaseRegionFromRALocked(ra, false /* async */);
+    }
   }
 }
 
 void Allocator::reset() {
   allocRetryWaits_.set(0);
   regionManager_.reset();
-  for (auto& ra : allocators_) {
-    std::lock_guard<TimedMutex> lock{ra.getLock()};
-    ra.reset();
+  for (auto& ras : allocators_) {
+    for (auto& ra : ras) {
+      std::lock_guard<TimedMutex> lock{ra.getLock()};
+      ra.reset();
+    }
   }
 }
 

cachelib/navy/block_cache/Allocator.h

@@ -77,18 +77,19 @@ class Allocator {
  public:
   // Constructs an allocator.
   //
-  // @param regionManager     Used to get eviction information and for
-  //                          locking regions
-  // @param numPriorities     Specifies how many priorities this allocator
-  //                          supports
-  // Throws std::exception if invalid arguments
-  explicit Allocator(RegionManager& regionManager, uint16_t numPriorities);
+  // @param regionManager           Used to get eviction information and for
+  //                                locking regions
+  // @param allocatorsPerPriority   Number of allocators per priority
+  explicit Allocator(RegionManager& regionManager,
+                     const std::vector<uint32_t>& allocatorsPerPriority);
 
   // Allocates and opens for writing.
   //
   // @param size          Allocation size
   // @param priority      Specifies how important this allocation is
   // @param canWait       If true, wait until allocation can be retried
+  // @param keyHash       Hash of the key to allocate for. Used for deciding the
+  // allocator if there's more than one for the priority.
   //
   // Returns a tuple containing region descriptor, allocated slotSize and
   // allocated address
@@ -102,7 +103,8 @@ class Allocator {
   // this allocator.
   std::tuple<RegionDescriptor, uint32_t, RelAddress> allocate(uint32_t size,
                                                               uint16_t priority,
-                                                              bool canWait);
+                                                              bool canWait,
+                                                              uint64_t keyHash);
 
   // Closes the region.
   void close(RegionDescriptor&& rid);
@@ -132,7 +134,7 @@ class Allocator {
 
   RegionManager& regionManager_;
   // Multiple allocators when we use priority-based allocation
-  std::vector<RegionAllocator> allocators_;
+  std::vector<std::vector<RegionAllocator>> allocators_;
 
   mutable AtomicCounter allocRetryWaits_;
 };

cachelib/navy/block_cache/BlockCache.cpp

@@ -64,8 +64,14 @@ BlockCache::Config& BlockCache::Config::validate() {
   if (numInMemBuffers == 0) {
     throw std::invalid_argument("there must be at least one in-mem buffers");
   }
-  if (numPriorities == 0) {
-    throw std::invalid_argument("allocator must have at least one priority");
+  if (allocatorsPerPriority.size() == 0) {
+    throw std::invalid_argument("Allocators must have at least one priority");
+  }
+  for (const auto& ac : allocatorsPerPriority) {
+    if (ac == 0) {
+      throw std::invalid_argument(
+          "Each priority must have at least one allocator");
+    }
   }
 
   reinsertionConfig.validate();
@@ -115,7 +121,8 @@ BlockCache::BlockCache(Config&& config)
 
 BlockCache::BlockCache(Config&& config, ValidConfigTag)
     : config_{serializeConfig(config)},
-      numPriorities_{config.numPriorities},
+      numPriorities_{
+          static_cast<uint16_t>(config.allocatorsPerPriority.size())},
       checkExpired_{std::move(config.checkExpired)},
       destructorCb_{std::move(config.destructorCb)},
       checksumData_{config.checksum},
@@ -141,10 +148,10 @@ BlockCache::BlockCache(Config&& config, ValidConfigTag)
                      bindThis(&BlockCache::onRegionCleanup, *this),
                      std::move(config.evictionPolicy),
                      config.numInMemBuffers,
-                     config.numPriorities,
+                     static_cast<uint16_t>(config.allocatorsPerPriority.size()),
                      config.inMemBufFlushRetryLimit,
                      config.regionManagerFlushAsync},
-      allocator_{regionManager_, config.numPriorities},
+      allocator_{regionManager_, config.allocatorsPerPriority},
       reinsertionPolicy_{makeReinsertionPolicy(config.reinsertionConfig)} {
   validate(config);
   XLOG(INFO, "Block cache created");
@@ -181,8 +188,8 @@ Status BlockCache::insert(HashedKey hk, BufferView value) {
   }
 
   // All newly inserted items are assigned with the lowest priority
-  auto [desc, slotSize, addr] =
-      allocator_.allocate(size, kDefaultItemPriority, true /* canWait */);
+  auto [desc, slotSize, addr] = allocator_.allocate(
+      size, kDefaultItemPriority, true /* canWait */, hk.keyHash());
 
   switch (desc.status()) {
   case OpenStatus::Error:
@@ -634,7 +641,7 @@ BlockCache::ReinsertionRes BlockCache::reinsertOrRemoveItem(
 
   uint32_t size = serializedSize(hk.key().size(), value.size());
   auto [desc, slotSize, addr] =
-      allocator_.allocate(size, priority, false /* canWait */);
+      allocator_.allocate(size, priority, false /* canWait */, hk.keyHash());
 
   switch (desc.status()) {
   case OpenStatus::Ready:

cachelib/navy/block_cache/BlockCache.h

@@ -81,10 +81,12 @@ class BlockCache final : public Engine {
     // directly fail it.
     uint16_t inMemBufFlushRetryLimit{10};
 
-    // Number of priorities. Items of the same priority will be put into
-    // the same reigon. The effect of priorities will be up to the particular
+    // The effect of priorities will be up to the particular
     // eviction policy. There must be at least one priority.
-    uint16_t numPriorities{1};
+    // This vector is the number of allocators for each priority. Under the same
+    // priority, allocators will be picked by key hash and written into one of
+    // the regions.
+    std::vector<uint32_t> allocatorsPerPriority{1};
 
     // whether to remove an item by checking the full key.
     bool preciseRemove{false};