Added AsyncDevice based on io_uring

Summary:
This change implements the AsyncDevice which performs the IOs asynchronously using io_uring.
To do so, this change uses the folly's io_uring implementation called IoUring which supports
handling the completion using folly's EventBase mechanism asynchronously. This change supports the
running IOs in either of dedicated NavyThread called IoThread or caller's NavyThread depending on
option. When the Device API is called on unknown context other than NavyThread, the execution is
fall back to the synchronous mode using legacy pread/pwrite instead of io_uring.

Reviewed By: therealgymmy

Differential Revision: D45932967

fbshipit-source-id: 3ea0b911768cd6839d5d07635980320150aa0370

Author: Jaesoo Lee

cachelib/allocator/nvmcache/NavyConfig.cpp

@@ -63,7 +63,17 @@ RandomAPConfig& RandomAPConfig::setAdmProbability(double admProbability) {
   return *this;
 }
 
-// file settings
+// device settings
+void NavyConfig::setIoThreads(unsigned int numIoThreads,
+                              unsigned int qDepthPerThread) {
+  if (usesRaidFiles()) {
+    throw std::invalid_argument(
+        "AsyncDevice is not yet supported for RAID files");
+  }
+  numIoThreads_ = numIoThreads;
+  qDepthPerThread_ = qDepthPerThread;
+}
+
 void NavyConfig::setSimpleFile(const std::string& fileName,
                                uint64_t fileSize,
                                bool truncateFile) {
@@ -219,6 +229,9 @@ std::map<std::string, std::string> NavyConfig::serialize() const {
   configMap["navyConfig::navyReqOrderingShards"] =
       folly::to<std::string>(navyReqOrderingShards_);
 
+  configMap["navyConfig::numIoThreads"] = folly::to<std::string>(numIoThreads_);
+  configMap["navyConfig::QDepthPerThread"] =
+      folly::to<std::string>(qDepthPerThread_);
   // Other settings
   configMap["navyConfig::maxConcurrentInserts"] =
       folly::to<std::string>(maxConcurrentInserts_);

cachelib/allocator/nvmcache/NavyConfig.h

@@ -478,6 +478,8 @@ class NavyConfig {
   const RandomAPConfig& randomAdmPolicy() const { return randomAPConfig_; }
 
   // ============ Device settings =============
+  unsigned int getNumIoThreads() const { return numIoThreads_; }
+  unsigned int getQDepthPerThread() const { return qDepthPerThread_; }
   uint64_t getBlockSize() const { return blockSize_; }
   const std::string& getFileName() const;
   const std::vector<std::string>& getRaidPaths() const;
@@ -520,6 +522,12 @@ class NavyConfig {
   RandomAPConfig& enableRandomAdmPolicy();
 
   // ============ Device settings =============
+  // Set the number of IO threads and queue depth per thread.
+  // If set, AsyncDevice instead of Device will be created with given
+  // configurations. For now, AsyncDevice is not supported for RAID.
+  void setIoThreads(unsigned int numIoThreads, unsigned int qDepthPerThread);
+
+  // Set the device block size, i.e., minimum unit of IO
   void setBlockSize(uint64_t blockSize) noexcept { blockSize_ = blockSize; }
   // Set the parameters for a simple file.
   // @throw std::invalid_argument if RAID files have been already set.
@@ -629,6 +637,10 @@ class NavyConfig {
   // This value needs to be non-zero.
   uint64_t navyReqOrderingShards_{20};
 
+  // Number of IO threads for AsyncDevice.
+  unsigned int numIoThreads_{0};
+  // Number of queue depth per thread for AsyncDevice.
+  unsigned int qDepthPerThread_{64};
   // ============ Other settings =============
   // Maximum number of concurrent inserts we allow globally for Navy.
   // 0 means unlimited.

cachelib/allocator/nvmcache/NavySetup.cpp

@@ -324,6 +324,18 @@ std::unique_ptr<cachelib::navy::Device> createDevice(
         std::move(encryptor),
         maxDeviceWriteSize > 0 ? alignDown(maxDeviceWriteSize, blockSize) : 0);
   } else if (config.usesSimpleFile()) {
+    if (config.getNumIoThreads() > 0) {
+      return cachelib::navy::createAsyncFileDevice(
+          config.getFileName(),
+          config.getFileSize(),
+          config.getTruncateFile(),
+          blockSize,
+          config.getNumIoThreads(),
+          config.getQDepthPerThread(),
+          std::move(encryptor),
+          maxDeviceWriteSize > 0 ? alignDown(maxDeviceWriteSize, blockSize)
+                                 : 0);
+    }
     return cachelib::navy::createFileDevice(
         config.getFileName(),
         config.getFileSize(),

cachelib/allocator/nvmcache/tests/NavyConfigTest.cpp

@@ -183,6 +183,8 @@ TEST(NavyConfigTest, Serialization) {
   expectedConfigMap["navyConfig::fileSize"] = "10485760";
   expectedConfigMap["navyConfig::truncateFile"] = "false";
   expectedConfigMap["navyConfig::deviceMaxWriteSize"] = "4194304";
+  expectedConfigMap["navyConfig::numIoThreads"] = "0";
+  expectedConfigMap["navyConfig::QDepthPerThread"] = "64";
 
   expectedConfigMap["navyConfig::blockCacheLru"] = "false";
   expectedConfigMap["navyConfig::blockCacheRegionSize"] = "16777216";

cachelib/cachebench/cache/Cache-inl.h

@@ -212,6 +212,10 @@ Cache<Allocator>::Cache(const CacheConfig& config,
 
     nvmConfig.navyConfig.setMaxParcelMemoryMB(config_.navyParcelMemoryMB);
 
+    if (config_.navyNumIoThreads > 0) {
+      nvmConfig.navyConfig.setIoThreads(config_.navyNumIoThreads,
+                                        config_.navyQDepthPerThread);
+    }
     nvmConfig.navyConfig.setReaderAndWriterThreads(config_.navyReaderThreads,
                                                    config_.navyWriterThreads);
 

cachelib/cachebench/util/CacheConfig.cpp

@@ -74,6 +74,8 @@ CacheConfig::CacheConfig(const folly::dynamic& configJson) {
   JSONSetVal(configJson, navyProbabilityReinsertionThreshold);
   JSONSetVal(configJson, navyReaderThreads);
   JSONSetVal(configJson, navyWriterThreads);
+  JSONSetVal(configJson, navyNumIoThreads);
+  JSONSetVal(configJson, navyQDepthPerThread);
   JSONSetVal(configJson, navyCleanRegions);
   JSONSetVal(configJson, navyAdmissionWriteRateMB);
   JSONSetVal(configJson, navyMaxConcurrentInserts);
@@ -105,7 +107,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, 728>();
+  checkCorrectSize<CacheConfig, 736>();
 
   if (numPools != poolSizes.size()) {
     throw std::invalid_argument(folly::sformat(

cachelib/cachebench/util/CacheConfig.h

@@ -184,6 +184,10 @@ struct CacheConfig : public JSONConfig {
   // number of asynchronous worker thread for navy write operation,
   uint32_t navyWriterThreads{32};
 
+  uint32_t navyNumIoThreads{0};
+
+  uint32_t navyQDepthPerThread{1024};
+
   // buffer of clean regions to be maintained free to ensure writes
   // into navy don't queue behind a reclaim of region.
   uint32_t navyCleanRegions{1};

cachelib/navy/Factory.cpp

@@ -27,6 +27,7 @@
 #include "cachelib/navy/block_cache/BlockCache.h"
 #include "cachelib/navy/block_cache/FifoPolicy.h"
 #include "cachelib/navy/block_cache/LruPolicy.h"
+#include "cachelib/navy/common/AsyncDevice.h"
 #include "cachelib/navy/driver/Driver.h"
 #include "cachelib/navy/serialization/RecordIO.h"
 
@@ -494,6 +495,27 @@ std::unique_ptr<Device> createFileDevice(
                                   std::move(encryptor), maxDeviceWriteSize);
 }
 
+std::unique_ptr<Device> createAsyncFileDevice(
+    std::string fileName,
+    uint64_t singleFileSize,
+    bool truncateFile,
+    uint32_t blockSize,
+    uint32_t numIoThreads,
+    uint32_t qDepthPerThread,
+    std::shared_ptr<DeviceEncryptor> encryptor,
+    uint32_t maxDeviceWriteSize) {
+  folly::File f;
+  try {
+    f = openCacheFile(fileName, singleFileSize, truncateFile);
+  } catch (const std::exception& e) {
+    XLOG(ERR) << "Exception in openCacheFile: " << e.what();
+    throw;
+  }
+  return createAsyncIoFileDevice(std::move(f), singleFileSize, blockSize,
+                                 numIoThreads, qDepthPerThread,
+                                 std::move(encryptor), maxDeviceWriteSize);
+}
+
 } // namespace navy
 } // namespace cachelib
 } // namespace facebook

cachelib/navy/Factory.h

@@ -228,6 +228,27 @@ std::unique_ptr<Device> createFileDevice(
     std::shared_ptr<DeviceEncryptor> encryptor,
     uint32_t maxDeviceWriteSize);
 
+// Creates a direct IO single file device.
+//
+// @param fileName              name of the file
+// @param singleFileSize        size of the file
+// @param truncateFile          whether to truncate the file
+// @param blockSize             device block size
+// @param numIoThreads          the number of IO threads. If 0 and EventBase is
+//                              not available, IOs will be fall back to sync IO
+// @param qDeptherThread        queue depth per each IO thread
+// @param encryptor             encryption object
+// @param maxDeviceWriteSize    device maximum granularity of writes
+std::unique_ptr<Device> createAsyncFileDevice(
+    std::string fileName,
+    uint64_t singleFileSize,
+    bool truncateFile,
+    uint32_t blockSize,
+    uint32_t numIoThreads,
+    uint32_t qDepthPerThread,
+    std::shared_ptr<DeviceEncryptor> encryptor,
+    uint32_t maxDeviceWriteSize);
+
 } // namespace navy
 } // namespace cachelib
 } // namespace facebook