(PR2) Adds the support for Flexible Data Placement(FDP) over NVMe into the Cachelib

cachelib/allocator/nvmcache/NavyConfig.cpp

@@ -278,6 +278,7 @@ std::map<std::string, std::string> NavyConfig::serialize() const {
       folly::to<std::string>(deviceMaxWriteSize_);
   configMap["navyConfig::ioEngine"] = getIoEngineName(ioEngine_).str();
   configMap["navyConfig::QDepth"] = folly::to<std::string>(qDepth_);
+  configMap["navyConfig::enableFDP"] = folly::to<std::string>(enableFDP_);
 
   // Job scheduler settings
   configMap["navyConfig::readerThreads"] =

cachelib/allocator/nvmcache/NavyConfig.h

@@ -485,6 +485,8 @@ class NavyConfig {
   bool isBigHashEnabled() const {
     return enginesConfigs_[0].bigHash().getSizePct() > 0;
   }
+  bool isFDPEnabled() const { return enableFDP_; }
+
   std::map<std::string, std::string> serialize() const;
 
   // Getters:
@@ -549,6 +551,8 @@ class NavyConfig {
   // ============ Device settings =============
   // Set the device block size, i.e., minimum unit of IO
   void setBlockSize(uint64_t blockSize) noexcept { blockSize_ = blockSize; }
+  // Set the NVMe FDP Device data placement mode in the Cachelib
+  void setEnableFDP(bool enable) noexcept { enableFDP_ = enable; }
   // Set the parameters for a simple file.
   // @throw std::invalid_argument if RAID files have been already set.
   void setSimpleFile(const std::string& fileName,
@@ -694,6 +698,9 @@ class NavyConfig {
   // Whether to use write size (instead of parcel size) for Navy admission
   // policy.
   bool useEstimatedWriteSize_{false};
+  // Whether Navy support the NVMe FDP data placement(TP4146) directives or not.
+  // Reference: https://nvmexpress.org/nvmeflexible-data-placement-fdp-blog/
+  bool enableFDP_{false};
 };
 } // namespace navy
 } // namespace cachelib

cachelib/allocator/nvmcache/NavySetup.cpp

@@ -349,6 +349,7 @@ std::unique_ptr<cachelib::navy::Device> createDevice(
         maxDeviceWriteSize > 0 ? alignDown(maxDeviceWriteSize, blockSize) : 0,
         config.getIoEngine(),
         config.getQDepth(),
+        config.isFDPEnabled(),
         std::move(encryptor));
   } else {
     return cachelib::navy::createMemoryDevice(config.getFileSize(),

cachelib/allocator/nvmcache/tests/NavyConfigTest.cpp

@@ -189,6 +189,7 @@ TEST(NavyConfigTest, Serialization) {
   expectedConfigMap["navyConfig::deviceMaxWriteSize"] = "4194304";
   expectedConfigMap["navyConfig::ioEngine"] = "io_uring";
   expectedConfigMap["navyConfig::QDepth"] = "64";
+  expectedConfigMap["navyConfig::enableFDP"] = "0";
 
   expectedConfigMap["navyConfig::blockCacheLru"] = "false";
   expectedConfigMap["navyConfig::blockCacheRegionSize"] = "16777216";

cachelib/cachebench/cache/Cache-inl.h

@@ -173,6 +173,7 @@ Cache<Allocator>::Cache(const CacheConfig& config,
           config_.navyReqOrderShardsPower);
     }
     nvmConfig.navyConfig.setBlockSize(config_.navyBlockSize);
+    nvmConfig.navyConfig.setEnableFDP(config_.deviceEnableFDP);
 
     // configure BlockCache
     auto& bcConfig = nvmConfig.navyConfig.blockCache()

cachelib/cachebench/util/CacheConfig.cpp

@@ -88,6 +88,7 @@ CacheConfig::CacheConfig(const folly::dynamic& configJson) {
   JSONSetVal(configJson, truncateItemToOriginalAllocSizeInNvm);
   JSONSetVal(configJson, navyEncryption);
   JSONSetVal(configJson, deviceMaxWriteSize);
+  JSONSetVal(configJson, deviceEnableFDP);
 
   JSONSetVal(configJson, memoryOnlyTTL);
 
@@ -111,7 +112,7 @@ CacheConfig::CacheConfig(const folly::dynamic& configJson) {
   // if you added new fields to the configuration, update the JSONSetVal
   // to make them available for the json configs and increment the size
   // below
-  checkCorrectSize<CacheConfig, 752>();
+  checkCorrectSize<CacheConfig, 760>();
 
   if (numPools != poolSizes.size()) {
     throw std::invalid_argument(folly::sformat(

cachelib/cachebench/util/CacheConfig.h

@@ -227,6 +227,9 @@ struct CacheConfig : public JSONConfig {
   // Navy will split it into multiple IOs.
   uint32_t deviceMaxWriteSize{1024 * 1024};
 
+  // Enable the FDP Data placement mode in the device, if it is capable.
+  bool deviceEnableFDP{false};
+
   // Don't write to flash if cache TTL is smaller than this value.
   // Not used when its value is 0.  In seconds.
   uint32_t memoryOnlyTTL{0};

cachelib/navy/CMakeLists.txt

@@ -31,6 +31,7 @@ add_library (cachelib_navy
   block_cache/RegionManager.cpp
   common/Buffer.cpp
   common/Device.cpp
+  common/FdpNvme.cpp
   common/Hash.cpp
   common/SizeDistribution.cpp
   common/Types.cpp

cachelib/navy/Factory.cpp

@@ -454,6 +454,7 @@ std::unique_ptr<Device> createFileDevice(
     uint32_t maxDeviceWriteSize,
     IoEngine ioEngine,
     uint32_t qDepth,
+    bool isFDPEnabled,
     std::shared_ptr<navy::DeviceEncryptor> encryptor) {
   // File paths are opened in the increasing order of the
   // path string. This ensures that RAID0 stripes aren't
@@ -476,12 +477,14 @@ std::unique_ptr<Device> createFileDevice(
   }
 
   return createDirectIoFileDevice(std::move(fileVec),
+                                  std::move(filePaths),
                                   fdSize,
                                   blockSize,
                                   stripeSize,
                                   maxDeviceWriteSize,
                                   ioEngine,
                                   qDepth,
+                                  isFDPEnabled,
                                   std::move(encryptor));
 }
 

cachelib/navy/Factory.h

@@ -209,6 +209,7 @@ std::unique_ptr<AbstractCache> createCache(std::unique_ptr<CacheProto> proto);
 // @param maxDeviceWriteSize    device maximum granularity of writes
 // @param ioEngine              IoEngine to be used for IO
 // @param qDepth                queue depth for async IO; 0 for sync IO
+// @param isFDPEnabled          whether FDP placement mode enabled or not
 // @param encryptor             encryption object
 std::unique_ptr<Device> createFileDevice(
     std::vector<std::string> filePaths,
@@ -219,6 +220,7 @@ std::unique_ptr<Device> createFileDevice(
     uint32_t maxDeviceWriteSize,
     IoEngine ioEngine,
     uint32_t qDepth,
+    bool isFDPEnabled,
     std::shared_ptr<navy::DeviceEncryptor> encryptor);
 
 } // namespace navy

cachelib/navy/bighash/BigHash.cpp

@@ -88,7 +88,8 @@ BigHash::BigHash(Config&& config, ValidConfigTag)
       cacheBaseOffset_{config.cacheBaseOffset},
       numBuckets_{config.numBuckets()},
       bloomFilter_{std::move(config.bloomFilter)},
-      device_{*config.device} {
+      device_{*config.device},
+      placementHandle_{device_.allocatePlacementHandle()} {
   XLOGF(INFO,
         "BigHash created: buckets: {}, bucket size: {}, base offset: {}",
         numBuckets_,
@@ -550,6 +551,7 @@ Buffer BigHash::readBucket(BucketId bid) {
 bool BigHash::writeBucket(BucketId bid, Buffer buffer) {
   auto* bucket = reinterpret_cast<Bucket*>(buffer.data());
   bucket->setChecksum(Bucket::computeChecksum(buffer.view()));
-  return device_.write(getBucketOffset(bid), std::move(buffer));
+  return device_.write(
+      getBucketOffset(bid), std::move(buffer), placementHandle_);
 }
 } // namespace facebook::cachelib::navy

cachelib/navy/bighash/BigHash.h

@@ -212,6 +212,8 @@ class BigHash final : public Engine {
   std::unique_ptr<BloomFilter> bloomFilter_;
   std::chrono::nanoseconds generationTime_{};
   Device& device_;
+  // handle for data placement technologies like FDP
+  int placementHandle_;
   std::unique_ptr<SharedMutex[]> mutex_{new SharedMutex[kNumMutexes]};
   // Spinlocks for bloom filter operations
   // We use spinlock in addition to the mutex to avoid contentions of

cachelib/navy/bighash/tests/BigHashTest.cpp

@@ -213,7 +213,7 @@ TEST(BigHash, DeviceErrorStats) {
   BigHash bh(std::move(config));
 
   EXPECT_EQ(Status::Ok, bh.insert(makeHK("key1"), makeView("1")));
-  EXPECT_CALL(*device, writeImpl(0, 64, _)).WillOnce(Return(false));
+  EXPECT_CALL(*device, writeImpl(0, 64, _, _)).WillOnce(Return(false));
   EXPECT_EQ(Status::DeviceError, bh.insert(makeHK("key2"), makeView("1")));
   {
     MockCounterVisitor helper;
@@ -351,12 +351,13 @@ TEST(BigHash, WriteInTwoBuckets) {
       config.cacheBaseOffset + config.cacheSize, 128);
   {
     InSequence inSeq;
+    EXPECT_CALL(*device, allocatePlacementHandle());
     EXPECT_CALL(*device, readImpl(256, 128, _));
-    EXPECT_CALL(*device, writeImpl(256, 128, _));
+    EXPECT_CALL(*device, writeImpl(256, 128, _, _));
     EXPECT_CALL(*device, readImpl(384, 128, _));
-    EXPECT_CALL(*device, writeImpl(384, 128, _));
+    EXPECT_CALL(*device, writeImpl(384, 128, _, _));
     EXPECT_CALL(*device, readImpl(256, 128, _));
-    EXPECT_CALL(*device, writeImpl(256, 128, _));
+    EXPECT_CALL(*device, writeImpl(256, 128, _, _));
   }
   config.device = device.get();
 
@@ -375,10 +376,11 @@ TEST(BigHash, RemoveNotFound) {
   auto device = std::make_unique<StrictMock<MockDevice>>(config.cacheSize, 128);
   {
     InSequence inSeq;
+    EXPECT_CALL(*device, allocatePlacementHandle());
     EXPECT_CALL(*device, readImpl(0, 128, _));
-    EXPECT_CALL(*device, writeImpl(0, 128, _));
+    EXPECT_CALL(*device, writeImpl(0, 128, _, _));
     EXPECT_CALL(*device, readImpl(0, 128, _));
-    EXPECT_CALL(*device, writeImpl(0, 128, _));
+    EXPECT_CALL(*device, writeImpl(0, 128, _, _));
     EXPECT_CALL(*device, readImpl(0, 128, _));
   }
   config.device = device.get();
@@ -541,6 +543,7 @@ TEST(BigHash, BloomFilterRecoveryFail) {
   BigHash::Config config;
   setLayout(config, 128, 2);
   auto device = std::make_unique<StrictMock<MockDevice>>(config.cacheSize, 128);
+  EXPECT_CALL(*device, allocatePlacementHandle());
   EXPECT_CALL(*device, readImpl(_, _, _)).Times(0);
   config.device = device.get();
   config.bloomFilter = std::make_unique<BloomFilter>(2, 1, 4);
@@ -635,8 +638,9 @@ TEST(BigHash, BloomFilterRecovery) {
     setLayout(config, 128, 2);
     auto device =
         std::make_unique<StrictMock<MockDevice>>(config.cacheSize, 128);
+    EXPECT_CALL(*device, allocatePlacementHandle());
     EXPECT_CALL(*device, readImpl(0, 128, _));
-    EXPECT_CALL(*device, writeImpl(0, 128, _));
+    EXPECT_CALL(*device, writeImpl(0, 128, _, _));
     config.device = device.get();
     config.bloomFilter = std::make_unique<BloomFilter>(2, 1, 4);
 

cachelib/navy/block_cache/RegionManager.cpp

@@ -45,7 +45,8 @@ RegionManager::RegionManager(uint32_t numRegions,
       numCleanRegions_{numCleanRegions},
       evictCb_{evictCb},
       cleanupCb_{cleanupCb},
-      numInMemBuffers_{numInMemBuffers} {
+      numInMemBuffers_{numInMemBuffers},
+      placementHandle_{device_.allocatePlacementHandle()} {
   XLOGF(INFO, "{} regions, {} bytes each", numRegions_, regionSize_);
   for (uint32_t i = 0; i < numRegions; i++) {
     regions_[i] = std::make_unique<Region>(RegionId{i}, regionSize_);
@@ -526,7 +527,7 @@ bool RegionManager::deviceWrite(RelAddress addr, Buffer buf) {
   const auto bufSize = buf.size();
   XDCHECK(isValidIORange(addr.offset(), bufSize));
   auto physOffset = physicalOffset(addr);
-  if (!device_.write(physOffset, std::move(buf))) {
+  if (!device_.write(physOffset, std::move(buf), placementHandle_)) {
     return false;
   }
   physicalWrittenCount_.add(bufSize);
@@ -537,7 +538,7 @@ bool RegionManager::deviceWrite(RelAddress addr, BufferView view) {
   const auto bufSize = view.size();
   XDCHECK(isValidIORange(addr.offset(), bufSize));
   auto physOffset = physicalOffset(addr);
-  if (!device_.write(physOffset, view)) {
+  if (!device_.write(physOffset, view, placementHandle_)) {
     return false;
   }
   physicalWrittenCount_.add(bufSize);

cachelib/navy/block_cache/RegionManager.h

@@ -339,6 +339,7 @@ class RegionManager {
   mutable TimedMutex bufferMutex_;
   mutable util::ConditionVariable bufferCond_;
   std::vector<std::unique_ptr<Buffer>> buffers_;
+  int placementHandle_;
 };
 } // namespace navy
 } // namespace cachelib

cachelib/navy/block_cache/tests/BlockCacheTest.cpp

@@ -592,14 +592,14 @@ TEST(BlockCache, ReclaimCorruption) {
   auto driver = makeDriver(std::move(engine), std::move(ex));
 
   // Allow any number of writes in between and after our expected writes
-  EXPECT_CALL(*device, writeImpl(_, _, _)).Times(testing::AtLeast(0));
+  EXPECT_CALL(*device, writeImpl(_, _, _, _)).Times(testing::AtLeast(0));
 
   // Note even tho this item's value is corrupted, we would have aborted
   // the reclaim before we got here. So we will not bump the value checksum
   // error stat on this.
-  EXPECT_CALL(*device, writeImpl(0, 16384, _))
+  EXPECT_CALL(*device, writeImpl(0, 16384, _, _))
       .WillOnce(testing::Invoke(
-          [&device](uint64_t offset, uint32_t size, const void* data) {
+          [&device](uint64_t offset, uint32_t size, const void* data, int) {
             // Note that all items are aligned to 512 bytes in in-mem buffer
             // stacked mode, and we write around 800 bytes, so each is aligned
             // to 1024 bytes
@@ -703,7 +703,7 @@ TEST(BlockCache, RegionUnderflow) {
   std::vector<uint32_t> hits(4);
   auto policy = std::make_unique<NiceMock<MockPolicy>>(&hits);
   auto device = std::make_unique<NiceMock<MockDevice>>(kDeviceSize, 1024);
-  EXPECT_CALL(*device, writeImpl(0, 16 * 1024, _));
+  EXPECT_CALL(*device, writeImpl(0, 16 * 1024, _, _));
   // Although 2k read buffer, shouldn't underflow the region!
   EXPECT_CALL(*device, readImpl(0, 1024, _));
   auto ex = makeJobScheduler();
@@ -730,7 +730,7 @@ TEST(BlockCache, SmallReadBuffer) {
   auto policy = std::make_unique<NiceMock<MockPolicy>>(&hits);
   auto device = std::make_unique<NiceMock<MockDevice>>(
       kDeviceSize, 4096 /* io alignment size */);
-  EXPECT_CALL(*device, writeImpl(0, 16 * 1024, _));
+  EXPECT_CALL(*device, writeImpl(0, 16 * 1024, _, _));
   EXPECT_CALL(*device, readImpl(0, 8192, _));
   auto ex = makeJobScheduler();
   auto config = makeConfig(*ex, std::move(policy), *device);
@@ -1057,10 +1057,11 @@ TEST(BlockCache, DeviceFailure) {
   auto device = std::make_unique<NiceMock<MockDevice>>(kDeviceSize, 1024);
   {
     testing::InSequence seq;
-    EXPECT_CALL(*device, writeImpl(0, kRegionSize, _)).WillOnce(Return(false));
-    EXPECT_CALL(*device, writeImpl(0, kRegionSize, _));
-    EXPECT_CALL(*device, writeImpl(kRegionSize, kRegionSize, _));
-    EXPECT_CALL(*device, writeImpl(kRegionSize * 2, kRegionSize, _));
+    EXPECT_CALL(*device, writeImpl(0, kRegionSize, _, _))
+        .WillOnce(Return(false));
+    EXPECT_CALL(*device, writeImpl(0, kRegionSize, _, _));
+    EXPECT_CALL(*device, writeImpl(kRegionSize, kRegionSize, _, _));
+    EXPECT_CALL(*device, writeImpl(kRegionSize * 2, kRegionSize, _, _));
 
     EXPECT_CALL(*device, readImpl(0, 1024, _));
     EXPECT_CALL(*device, readImpl(kRegionSize, 1024, _))
@@ -1116,7 +1117,7 @@ namespace {
 std::unique_ptr<Device> setupResetTestDevice(uint32_t size) {
   auto device = std::make_unique<NiceMock<MockDevice>>(size, 512);
   for (uint32_t i = 0; i < 2; i++) {
-    EXPECT_CALL(*device, writeImpl(i * 16 * 1024, 16 * 1024, _));
+    EXPECT_CALL(*device, writeImpl(i * 16 * 1024, 16 * 1024, _, _));
   }
   return device;
 }
@@ -2080,7 +2081,7 @@ TEST(BlockCache, DeviceFlushFailureSync) {
   auto device = std::make_unique<MockDevice>(kDeviceSize, 1024);
 
   testing::InSequence inSeq;
-  EXPECT_CALL(*device, writeImpl(_, _, _)).WillRepeatedly(Return(false));
+  EXPECT_CALL(*device, writeImpl(_, _, _, _)).WillRepeatedly(Return(false));
 
   auto ex = makeJobScheduler();
   auto config = makeConfig(*ex, std::move(policy), *device);
@@ -2122,7 +2123,7 @@ TEST(BlockCache, DeviceFlushFailureAsync) {
   auto device = std::make_unique<MockDevice>(kDeviceSize, 1024);
 
   testing::InSequence inSeq;
-  EXPECT_CALL(*device, writeImpl(_, _, _)).WillRepeatedly(Return(false));
+  EXPECT_CALL(*device, writeImpl(_, _, _, _)).WillRepeatedly(Return(false));
 
   auto ex = makeJobScheduler();
   auto config = makeConfig(*ex, std::move(policy), *device);

cachelib/navy/block_cache/tests/RegionManagerTest.cpp

@@ -396,7 +396,7 @@ TEST(RegionManager, cleanupRegionFailureSync) {
 
   std::thread flushThread{[&sp, &device, &rm, &rid] {
     // Make sure flush will fail
-    EXPECT_CALL(*device, writeImpl(_, _, _)).WillRepeatedly(Return(false));
+    EXPECT_CALL(*device, writeImpl(_, _, _, _)).WillRepeatedly(Return(false));
     sp.wait(0); // Flush after active reader
     rm->doFlush(rid, false /* async */);
   }};
@@ -505,7 +505,7 @@ TEST(RegionManager, cleanupRegionFailureAsync) {
 
   std::thread flushThread{[&sp, &device, &rm, &rid] {
     // Make sure flush will fail
-    EXPECT_CALL(*device, writeImpl(_, _, _)).WillRepeatedly(Return(false));
+    EXPECT_CALL(*device, writeImpl(_, _, _, _)).WillRepeatedly(Return(false));
     sp.wait(0); // Flush after active reader
     rm->doFlush(rid, true /* async */);
   }};