feat: Support read local file async (#11869)

Summary:
Add struct FileSystemOptions to function registerLocalFileSystem, default readAhead false, Use an executor to submit the read async task, the number thread of pool is half of system concurrency.
The performance is same with before when device is SSD, can save about 50% time in HDD.

Pull Request resolved: #11869

Reviewed By: yuandagits

Differential Revision: D68311321

Pulled By: xiaoxmeng

fbshipit-source-id: d091363b8f826ae5fc47252b6c123edcc2ffbdfe

Author: jinchengchenghh

velox/common/file/File.cpp

@@ -129,8 +129,11 @@ uint64_t InMemoryWriteFile::size() const {
   return file_->size();
 }
 
-LocalReadFile::LocalReadFile(std::string_view path, bool bufferIo)
-    : path_(path) {
+LocalReadFile::LocalReadFile(
+    std::string_view path,
+    folly::Executor* executor,
+    bool bufferIo)
+    : executor_(executor), path_(path) {
   int32_t flags = O_RDONLY;
 #ifdef linux
   if (!bufferIo) {
@@ -160,7 +163,8 @@ LocalReadFile::LocalReadFile(std::string_view path, bool bufferIo)
   size_ = ret;
 }
 
-LocalReadFile::LocalReadFile(int32_t fd) : fd_(fd) {}
+LocalReadFile::LocalReadFile(int32_t fd, folly::Executor* executor)
+    : executor_(executor), fd_(fd) {}
 
 LocalReadFile::~LocalReadFile() {
   const int ret = close(fd_);
@@ -245,6 +249,23 @@ uint64_t LocalReadFile::preadv(
   return totalBytesRead;
 }
 
+folly::SemiFuture<uint64_t> LocalReadFile::preadvAsync(
+    uint64_t offset,
+    const std::vector<folly::Range<char*>>& buffers) const {
+  if (!executor_) {
+    return ReadFile::preadvAsync(offset, buffers);
+  }
+  auto [promise, future] = folly::makePromiseContract<uint64_t>();
+  executor_->add([this,
+                  _promise = std::move(promise),
+                  _offset = offset,
+                  _buffers = buffers]() mutable {
+    auto delegateFuture = ReadFile::preadvAsync(_offset, _buffers);
+    _promise.setTry(std::move(delegateFuture).getTry());
+  });
+  return std::move(future);
+}
+
 uint64_t LocalReadFile::size() const {
   return size_;
 }

velox/common/file/File.h

@@ -34,6 +34,7 @@
 #include <string>
 #include <string_view>
 
+#include <folly/Executor.h>
 #include <folly/Range.h>
 #include <folly/futures/Future.h>
 
@@ -266,11 +267,14 @@ class InMemoryWriteFile final : public WriteFile {
 /// files match against any filepath starting with '/'.
 class LocalReadFile final : public ReadFile {
  public:
-  explicit LocalReadFile(std::string_view path, bool bufferIo = true);
+  LocalReadFile(
+      std::string_view path,
+      folly::Executor* executor = nullptr,
+      bool bufferIo = true);
 
   /// TODO: deprecate this after creating local file all through velox fs
   /// interface.
-  explicit LocalReadFile(int32_t fd);
+  LocalReadFile(int32_t fd, folly::Executor* executor = nullptr);
 
   ~LocalReadFile();
 
@@ -283,6 +287,14 @@ class LocalReadFile final : public ReadFile {
       uint64_t offset,
       const std::vector<folly::Range<char*>>& buffers) const final;
 
+  folly::SemiFuture<uint64_t> preadvAsync(
+      uint64_t offset,
+      const std::vector<folly::Range<char*>>& buffers) const override;
+
+  bool hasPreadvAsync() const override {
+    return executor_ != nullptr;
+  }
+
   uint64_t memoryUsage() const final;
 
   bool shouldCoalesce() const final {
@@ -303,6 +315,7 @@ class LocalReadFile final : public ReadFile {
  private:
   void preadInternal(uint64_t offset, uint64_t length, char* pos) const;
 
+  folly::Executor* const executor_;
   std::string path_;
   int32_t fd_;
   long size_;

velox/common/file/FileSystems.cpp

@@ -15,6 +15,7 @@
  */
 
 #include "velox/common/file/FileSystems.h"
+#include <folly/executors/CPUThreadPoolExecutor.h>
 #include <folly/synchronization/CallOnce.h>
 #include "velox/common/base/Exceptions.h"
 #include "velox/common/file/File.h"
@@ -79,10 +80,27 @@ folly::once_flag localFSInstantiationFlag;
 // Implement Local FileSystem.
 class LocalFileSystem : public FileSystem {
  public:
-  explicit LocalFileSystem(std::shared_ptr<const config::ConfigBase> config)
-      : FileSystem(config) {}
+  LocalFileSystem(
+      std::shared_ptr<const config::ConfigBase> config,
+      const FileSystemOptions& options)
+      : FileSystem(config),
+        executor_(
+            options.readAheadEnabled
+                ? std::make_unique<folly::CPUThreadPoolExecutor>(
+                      std::max(
+                          1,
+                          static_cast<int32_t>(
+                              std::thread::hardware_concurrency() / 2)),
+                      std::make_shared<folly::NamedThreadFactory>(
+                          "LocalReadahead"))
+                : nullptr) {}
 
-  ~LocalFileSystem() override {}
+  ~LocalFileSystem() override {
+    if (executor_) {
+      executor_->stop();
+      LOG(INFO) << "Executor " << executor_->getName() << " stopped.";
+    }
+  }
 
   std::string name() const override {
     return "Local FS";
@@ -98,7 +116,8 @@ class LocalFileSystem : public FileSystem {
   std::unique_ptr<ReadFile> openFileForRead(
       std::string_view path,
       const FileOptions& options) override {
-    return std::make_unique<LocalReadFile>(extractPath(path), options.bufferIo);
+    return std::make_unique<LocalReadFile>(
+        extractPath(path), executor_.get(), options.bufferIo);
   }
 
   std::unique_ptr<WriteFile> openFileForWrite(
@@ -216,23 +235,28 @@ class LocalFileSystem : public FileSystem {
 
   static std::function<std::shared_ptr<
       FileSystem>(std::shared_ptr<const config::ConfigBase>, std::string_view)>
-  fileSystemGenerator() {
-    return [](std::shared_ptr<const config::ConfigBase> properties,
-              std::string_view filePath) {
+  fileSystemGenerator(const FileSystemOptions& options) {
+    return [options](
+               std::shared_ptr<const config::ConfigBase> properties,
+               std::string_view filePath) {
       // One instance of Local FileSystem is sufficient.
       // Initialize on first access and reuse after that.
       static std::shared_ptr<FileSystem> lfs;
-      folly::call_once(localFSInstantiationFlag, [&properties]() {
-        lfs = std::make_shared<LocalFileSystem>(properties);
+      folly::call_once(localFSInstantiationFlag, [properties, options]() {
+        lfs = std::make_shared<LocalFileSystem>(properties, options);
       });
       return lfs;
     };
   }
+
+ private:
+  const std::unique_ptr<folly::CPUThreadPoolExecutor> executor_;
 };
 } // namespace
 
-void registerLocalFileSystem() {
+void registerLocalFileSystem(const FileSystemOptions& options) {
   registerFileSystem(
-      LocalFileSystem::schemeMatcher(), LocalFileSystem::fileSystemGenerator());
+      LocalFileSystem::schemeMatcher(),
+      LocalFileSystem::fileSystemGenerator(options));
 }
 } // namespace facebook::velox::filesystems

velox/common/file/FileSystems.h

@@ -77,6 +77,13 @@ struct DirectoryOptions : FileOptions {
       "make-directory-config"};
 };
 
+struct FileSystemOptions {
+  /// As for now, only local file system respects this option. It implements
+  /// async read by using a background cpu executor. Some filesystem might has
+  /// native async read-ahead support.
+  bool readAheadEnabled{false};
+};
+
 /// An abstract FileSystem
 class FileSystem {
  public:
@@ -161,6 +168,7 @@ void registerFileSystem(
         std::string_view)> fileSystemGenerator);
 
 /// Register the local filesystem.
-void registerLocalFileSystem();
+void registerLocalFileSystem(
+    const FileSystemOptions& options = FileSystemOptions());
 
 } // namespace facebook::velox::filesystems

velox/common/file/tests/FileTest.cpp

@@ -15,6 +15,7 @@
  */
 
 #include <fcntl.h>
+#include <folly/executors/CPUThreadPoolExecutor.h>
 
 #include "velox/common/base/tests/GTestUtils.h"
 #include "velox/common/file/File.h"
@@ -66,7 +67,10 @@ void writeDataWithOffset(WriteFile* writeFile) {
   ASSERT_EQ(writeFile->size(), 15 + kOneMB);
 }
 
-void readData(ReadFile* readFile, bool checkFileSize = true) {
+void readData(
+    ReadFile* readFile,
+    bool checkFileSize = true,
+    bool testReadAsync = false) {
   if (checkFileSize) {
     ASSERT_EQ(readFile->size(), 15 + kOneMB);
   }
@@ -105,6 +109,30 @@ void readData(ReadFile* readFile, bool checkFileSize = true) {
   ASSERT_EQ(std::string_view(head, sizeof(head)), "aaaaabbbbbcc");
   ASSERT_EQ(std::string_view(middle, sizeof(middle)), "cccc");
   ASSERT_EQ(std::string_view(tail, sizeof(tail)), "ccddddd");
+  if (testReadAsync) {
+    std::vector<folly::Range<char*>> buffers1 = {
+        folly::Range<char*>(head, sizeof(head)),
+        folly::Range<char*>(nullptr, (char*)(uint64_t)500000)};
+    auto future1 = readFile->preadvAsync(0, buffers1);
+    const auto offset1 = sizeof(head) + 500000;
+    std::vector<folly::Range<char*>> buffers2 = {
+        folly::Range<char*>(middle, sizeof(middle)),
+        folly::Range<char*>(
+            nullptr,
+            (char*)(uint64_t)(15 + kOneMB - offset1 - sizeof(middle) -
+                              sizeof(tail)))};
+    auto future2 = readFile->preadvAsync(offset1, buffers2);
+    std::vector<folly::Range<char*>> buffers3 = {
+        folly::Range<char*>(tail, sizeof(tail))};
+    const auto offset2 = 15 + kOneMB - sizeof(tail);
+    auto future3 = readFile->preadvAsync(offset2, buffers3);
+    ASSERT_EQ(offset1, future1.wait().value());
+    ASSERT_EQ(offset2 - offset1, future2.wait().value());
+    ASSERT_EQ(sizeof(tail), future3.wait().value());
+    ASSERT_EQ(std::string_view(head, sizeof(head)), "aaaaabbbbbcc");
+    ASSERT_EQ(std::string_view(middle, sizeof(middle)), "cccc");
+    ASSERT_EQ(std::string_view(tail, sizeof(tail)), "ccddddd");
+  }
 }
 
 // We could templated this test, but that's kinda overkill for how simple it is.
@@ -157,6 +185,13 @@ class LocalFileTest : public ::testing::TestWithParam<bool> {
   }
 
   const bool useFaultyFs_;
+  const std::unique_ptr<folly::CPUThreadPoolExecutor> executor_ =
+      std::make_unique<folly::CPUThreadPoolExecutor>(
+          std::max(
+              1,
+              static_cast<int32_t>(std::thread::hardware_concurrency() / 2)),
+          std::make_shared<folly::NamedThreadFactory>(
+              "LocalFileReadAheadTest"));
 };
 
 TEST_P(LocalFileTest, writeAndRead) {
@@ -185,6 +220,14 @@ TEST_P(LocalFileTest, writeAndRead) {
       writeFile->close();
       ASSERT_EQ(writeFile->size(), 15 + kOneMB);
     }
+    // Test read async.
+    if (!useFaultyFs_) {
+      auto readFile =
+          std::make_shared<LocalReadFile>(filename, executor_.get());
+      readData(readFile.get(), true, true);
+      auto readFileWithoutExecutor = std::make_shared<LocalReadFile>(filename);
+      readData(readFileWithoutExecutor.get(), true, true);
+    }
     auto readFile = fs->openFileForRead(filename);
     readData(readFile.get());
   }