Fix IoUringBackend crash in ReadIOExecutor worker threads (#15020)

Summary:
Pull Request resolved: https://github.com/facebook/rocksdb/pull/15020

Each read executor worker creates its own `IoUringBackend`, and Folly's I/O thread pool creates its worker threads and EventBases lazily. A successful calling-thread probe or `DB::Open()` therefore does not validate every worker. A later read can start a new worker; previously, an `IoUringBackend` initialization exception on that worker could escape startup and terminate the process. The previous 512-submit setup also allowed Folly's capacity retry to reduce the completion queue to 512 while retaining a 1024-entry submission queue, which the kernel rejects with `EINVAL`.

Use a 256-submit limit so the 512-entry retry remains valid. Configure the shared read executor during `DB::Open()` through `DBOptions::read_io_executor_threads`, defaulting to one thread. The backend factory catches per-worker io_uring initialization failures and uses Folly's default backend. Each worker's thread-local availability marker starts false and is set true only after `IoUringBackend` construction succeeds. When construction fails, the marker remains false and `SubmitReadAsync()` performs the read synchronously. The executor's maximum thread capacity can be raised or lowered; `db_stress` requests eight threads for async DB APIs.

To expose this degradation, `FSRandomAccessFile::SubmitReadAsync()` now reports whether the asynchronous path was used. `RandomAccessFileReader` increments the `FILE_SUBMIT_ASYNC_READ_FALLBACK` ticker for every submission that falls back to synchronous I/O, making the fallback visible through the DB's `Statistics`.

Reviewed By: xingbowang

Differential Revision: D114031018

fbshipit-source-id: 1b7f6c1cea693e5dd743ea1ada0fca1cf757b3f9

c_api_gen/c_generated_option_structs_auto.cc.inc

@@ -141,6 +141,15 @@ int rocksdb_options_get_max_file_opening_threads(rocksdb_options_t* opt) {
   return opt->rep.max_file_opening_threads;
 }
 
+void rocksdb_options_set_read_io_executor_threads(rocksdb_options_t* opt,
+                                                  int v) {
+  opt->rep.read_io_executor_threads = v;
+}
+
+int rocksdb_options_get_read_io_executor_threads(rocksdb_options_t* opt) {
+  return opt->rep.read_io_executor_threads;
+}
+
 void rocksdb_options_set_open_files_async(rocksdb_options_t* opt,
                                           unsigned char v) {
   opt->rep.open_files_async = v;

c_api_gen/c_generated_option_structs_auto.h.inc

@@ -55,6 +55,12 @@ rocksdb_options_set_verify_sst_unique_id_in_manifest(rocksdb_options_t* opt,
 extern ROCKSDB_LIBRARY_API unsigned char
 rocksdb_options_get_verify_sst_unique_id_in_manifest(rocksdb_options_t* opt);
 
+extern ROCKSDB_LIBRARY_API void rocksdb_options_set_read_io_executor_threads(
+    rocksdb_options_t* opt, int v);
+
+extern ROCKSDB_LIBRARY_API int rocksdb_options_get_read_io_executor_threads(
+    rocksdb_options_t* opt);
+
 extern ROCKSDB_LIBRARY_API const char* rocksdb_options_get_db_log_dir(
     rocksdb_options_t* opt, size_t* size);
 

c_api_gen/c_generated_roundtrip_tests.c.inc

@@ -849,6 +849,10 @@ static void rocksdb_roundtrip_test_rocksdb_options(void) {
   CheckCondition(rocksdb_options_get_preserve_internal_time_seconds(obj) == 1);
   rocksdb_options_set_preserve_internal_time_seconds(obj, 0);
   CheckCondition(rocksdb_options_get_preserve_internal_time_seconds(obj) == 0);
+  rocksdb_options_set_read_io_executor_threads(obj, 1);
+  CheckCondition(rocksdb_options_get_read_io_executor_threads(obj) == 1);
+  rocksdb_options_set_read_io_executor_threads(obj, 0);
+  CheckCondition(rocksdb_options_get_read_io_executor_threads(obj) == 0);
   rocksdb_options_set_reuse_manifest_on_open(obj, 1);
   CheckCondition(rocksdb_options_get_reuse_manifest_on_open(obj) == 1);
   rocksdb_options_set_reuse_manifest_on_open(obj, 0);

db/c.cc

@@ -9107,6 +9107,15 @@ int rocksdb_options_get_max_file_opening_threads(rocksdb_options_t* opt) {
   return opt->rep.max_file_opening_threads;
 }
 
+void rocksdb_options_set_read_io_executor_threads(rocksdb_options_t* opt,
+                                                  int v) {
+  opt->rep.read_io_executor_threads = v;
+}
+
+int rocksdb_options_get_read_io_executor_threads(rocksdb_options_t* opt) {
+  return opt->rep.read_io_executor_threads;
+}
+
 void rocksdb_options_set_open_files_async(rocksdb_options_t* opt,
                                           unsigned char v) {
   opt->rep.open_files_async = v;

db/db_impl/db_impl.cc

@@ -244,6 +244,11 @@ DBImpl::DBImpl(const DBOptions& options, const std::string& dbname,
       blob_callback_(immutable_db_options_.sst_file_manager.get(), &mutex_,
                      &error_handler_, &event_logger_,
                      immutable_db_options_.listeners, dbname_) {
+#if USE_COROUTINES
+  immutable_db_options_.fs->SetReadIOExecutorThreads(
+      immutable_db_options_.read_io_executor_threads);
+#endif  // USE_COROUTINES
+
   // !batch_per_trx_ implies seq_per_batch_ because it is only unset for
   // WriteUnprepared, which should use seq_per_batch_.
   assert(batch_per_txn_ || seq_per_batch_);
@@ -7607,8 +7612,7 @@ void DBImpl::GetAsync(const ReadOptions& options,
                       Status& status, AsyncCallback& callback) {
   const bool stats_enabled = callback.EnableStats();
 #if USE_COROUTINES
-  auto* read_executor =
-      immutable_db_options_.env->GetFileSystem()->GetReadExecutor();
+  auto* read_executor = immutable_db_options_.fs->GetReadExecutor();
   if (read_executor != nullptr) {
     auto* read_event_base = read_executor->getEventBase();
     assert(read_event_base != nullptr);
@@ -7659,8 +7663,7 @@ void DBImpl::MultiGetAsync(const ReadOptions& options, const size_t num_keys,
                            const bool sorted_input, AsyncCallback& callback) {
   const bool stats_enabled = callback.EnableStats();
 #if USE_COROUTINES
-  auto* read_executor =
-      immutable_db_options_.env->GetFileSystem()->GetReadExecutor();
+  auto* read_executor = immutable_db_options_.fs->GetReadExecutor();
   if (read_executor != nullptr) {
     auto* read_event_base = read_executor->getEventBase();
     assert(read_event_base != nullptr);

db/db_impl/db_impl_open.cc

@@ -243,6 +243,11 @@ Status DBImpl::ValidateOptions(
 }
 
 Status DBImpl::ValidateOptions(const DBOptions& db_options) {
+  if (db_options.read_io_executor_threads <= 0) {
+    return Status::InvalidArgument(
+        "read_io_executor_threads must be greater than zero");
+  }
+
   if (db_options.db_paths.size() > 4) {
     return Status::NotSupported(
         "More than four DB paths are not supported yet. ");

db/db_impl/db_impl_sync_and_async.h

@@ -42,8 +42,7 @@ DEFINE_SYNC_AND_ASYNC(Status, DBImpl::Get)
   }
 
 #ifdef WITH_COROUTINES
-  auto* read_executor =
-      immutable_db_options_.env->GetFileSystem()->GetReadExecutor();
+  auto* read_executor = immutable_db_options_.fs->GetReadExecutor();
   if (read_executor != nullptr) {
     auto* read_event_base = read_executor->getEventBase();
     assert(read_event_base != nullptr);
@@ -808,8 +807,7 @@ DEFINE_SYNC_AND_ASYNC(void, DBImpl::MultiGet)
     read_options.io_activity = Env::IOActivity::kMultiGet;
   }
 #ifdef WITH_COROUTINES
-  auto* read_executor =
-      immutable_db_options_.env->GetFileSystem()->GetReadExecutor();
+  auto* read_executor = immutable_db_options_.fs->GetReadExecutor();
   if (read_executor != nullptr) {
     auto* read_event_base = read_executor->getEventBase();
     assert(read_event_base != nullptr);

db_stress_tool/db_stress_env_wrapper.h

@@ -92,11 +92,11 @@ class DbStressRandomAccessFileWrapper : public FSRandomAccessFileOwnerWrapper {
                                dbg);
   }
 
-  void SubmitReadAsync(FSReadRequest& req, const IOOptions& options,
+  bool SubmitReadAsync(FSReadRequest& req, const IOOptions& options,
                        std::function<void(FSReadRequest&)> cb,
                        IODebugContext* dbg) override {
     CheckIOActivity(options);
-    target()->SubmitReadAsync(req, options, std::move(cb), dbg);
+    return target()->SubmitReadAsync(req, options, std::move(cb), dbg);
   }
 };
 

db_stress_tool/db_stress_test_base.cc

@@ -5841,6 +5841,11 @@ void InitializeOptionsGeneral(
     Options& options) {
   options.create_missing_column_families = true;
   options.create_if_missing = true;
+#if USE_COROUTINES
+  if (FLAGS_use_async_db_api) {
+    options.read_io_executor_threads = 8;
+  }
+#endif  // USE_COROUTINES
 
   if (FLAGS_statistics) {
     options.statistics = ROCKSDB_NAMESPACE::CreateDBStatistics();

env/file_system_tracer.cc

@@ -363,11 +363,11 @@ IOStatus FSRandomAccessFileTracingWrapper::ReadAsync(
   return s;
 }
 
-void FSRandomAccessFileTracingWrapper::SubmitReadAsync(
+bool FSRandomAccessFileTracingWrapper::SubmitReadAsync(
     FSReadRequest& req, const IOOptions& opts,
     std::function<void(FSReadRequest&)> cb, IODebugContext* dbg) {
   const uint64_t start_time = clock_->NowNanos();
-  target()->SubmitReadAsync(
+  return target()->SubmitReadAsync(
       req, opts,
       [this, start_time, cb = std::move(cb)](FSReadRequest& completed_req) {
         const uint64_t elapsed = clock_->NowNanos() - start_time;

env/file_system_tracer.h

@@ -233,7 +233,7 @@ class FSRandomAccessFileTracingWrapper : public FSRandomAccessFileOwnerWrapper {
                      void* cb_arg, void** io_handle, IOHandleDeleter* del_fn,
                      IODebugContext* dbg) override;
 
-  void SubmitReadAsync(FSReadRequest& req, const IOOptions& opts,
+  bool SubmitReadAsync(FSReadRequest& req, const IOOptions& opts,
                        std::function<void(FSReadRequest&)> cb,
                        IODebugContext* dbg) override;
 

env/fs_posix.cc

@@ -34,6 +34,7 @@
 
 #include <algorithm>
 #include <ctime>
+#include <exception>
 #include <stdexcept>
 // Get nano time includes
 #if defined(OS_LINUX) || defined(OS_FREEBSD)
@@ -46,7 +47,6 @@
 #endif
 #include <deque>
 #include <set>
-#include <thread>
 #include <vector>
 
 #include "env/composite_env_wrapper.h"
@@ -60,22 +60,22 @@
 #include "rocksdb/slice.h"
 #include "rocksdb/utilities/object_registry.h"
 #include "test_util/sync_point.h"
+#include "util/atomic.h"
 #include "util/coding.h"
 #include "util/compression_context_cache.h"
+#include "util/mutexlock.h"
 #include "util/random.h"
 #include "util/string_util.h"
 #include "util/thread_local.h"
 #include "util/threadpool_imp.h"
 
-#if USE_COROUTINES
+#if USE_COROUTINES && FOLLY_HAS_LIBURING
 #include "folly/executors/IOThreadPoolExecutor.h"
-#include "folly/io/async/EventBaseManager.h"
-#if FOLLY_HAS_LIBURING
 #include "folly/executors/thread_factory/NamedThreadFactory.h"
 #include "folly/io/async/EventBase.h"
+#include "folly/io/async/EventBaseManager.h"
 #include "folly/io/async/IoUringBackend.h"
-#endif  // FOLLY_HAS_LIBURING
-#endif
+#endif  // USE_COROUTINES && FOLLY_HAS_LIBURING
 
 #if !defined(TMPFS_MAGIC)
 #define TMPFS_MAGIC 0x01021994
@@ -97,11 +97,10 @@ inline mode_t GetDBFileMode(bool allow_non_owner_access) {
   return allow_non_owner_access ? 0644 : 0600;
 }
 
-#if USE_COROUTINES
-#if FOLLY_HAS_LIBURING
+#if USE_COROUTINES && FOLLY_HAS_LIBURING
 folly::IoUringOptions GetReadIOUringOptions() {
   folly::IoUringOptions options;
-  options.setMaxSubmit(512);
+  options.setMaxSubmit(256);
   options.setCapacity(1024);
   options.setMinCapacity(512);
   options.setRegisterRingFd(true);
@@ -110,42 +109,69 @@ folly::IoUringOptions GetReadIOUringOptions() {
   return options;
 }
 
-#endif  // FOLLY_HAS_LIBURING
-
 struct ReadIOExecutorState {
-  ReadIOExecutorState() {
-#if FOLLY_HAS_LIBURING
-    try {
-      // Older Linux kernels can support basic io_uring while rejecting some
-      // of the required setup flags. Validate the worker options at runtime
-      // before exposing the lazy executor.
-      std::make_unique<folly::IoUringBackend>(GetReadIOUringOptions());
-      folly::EventBase::Options event_base_options;
-      event_base_options.setBackendFactory([]() {
-        return std::make_unique<folly::IoUringBackend>(GetReadIOUringOptions());
-      });
-      event_base_manager_ =
-          std::make_unique<folly::EventBaseManager>(event_base_options);
-      executor_ = std::make_unique<folly::IOThreadPoolExecutor>(
-          std::max(1u, std::thread::hardware_concurrency()),
-          /*minThreads=*/0,
-          std::make_shared<folly::NamedThreadFactory>("RocksDBAsyncRead"),
-          event_base_manager_.get());
-    } catch (const std::runtime_error&) {
-      executor_.reset();
-      event_base_manager_.reset();
-      return;
-    }
-#else
-    event_base_manager_ = nullptr;
-    executor_ = nullptr;
-#endif  // FOLLY_HAS_LIBURING
+  ReadIOExecutorState() = default;
+  ReadIOExecutorState(const ReadIOExecutorState&) = delete;
+  ReadIOExecutorState& operator=(const ReadIOExecutorState&) = delete;
+  ReadIOExecutorState(ReadIOExecutorState&&) = delete;
+  ReadIOExecutorState& operator=(ReadIOExecutorState&&) = delete;
+
+  ~ReadIOExecutorState() {
+    delete executor_.LoadRelaxed();
+    delete event_base_manager_.LoadRelaxed();
   }
 
-  std::unique_ptr<folly::EventBaseManager> event_base_manager_;
-  std::unique_ptr<folly::IOThreadPoolExecutor> executor_;
+  folly::IOExecutor* GetExecutor() { return executor_.Load(); }
+
+  folly::EventBaseManager* GetEventBaseManager() {
+    return event_base_manager_.Load();
+  }
+
+  void SetThreads(int num) {
+    assert(num > 0);
+    MutexLock lock(&mutex_);
+    const size_t requested_threads = static_cast<size_t>(num);
+    auto* executor = executor_.Load();
+    if (executor != nullptr) {
+      if (requested_threads > executor->numThreads()) {
+        try {
+          executor->setNumThreads(requested_threads);
+        } catch (const std::exception&) {
+        }
+      }
+      return;
+    }
+    try {
+      folly::EventBase::Options event_base_options;
+      event_base_options.setBackendFactory(
+          []() -> std::unique_ptr<folly::EventBaseBackendBase> {
+            try {
+              auto backend = std::make_unique<folly::IoUringBackend>(
+                  GetReadIOUringOptions());
+              SetCurrentThreadReadIOUringBackendAvailable();
+              return backend;
+            } catch (const std::exception&) {
+              return folly::EventBase::getDefaultBackend();
+            }
+          });
+      auto event_base_manager =
+          std::make_unique<folly::EventBaseManager>(event_base_options);
+      auto new_executor = std::make_unique<folly::IOThreadPoolExecutor>(
+          requested_threads, 0,
+          std::make_shared<folly::NamedThreadFactory>("RocksDBAsyncRead"),
+          event_base_manager.get());
+      event_base_manager_.Store(event_base_manager.release());
+      executor_.Store(new_executor.release());
+    } catch (const std::exception&) {
+      return;
+    }
+  }
+
+  port::Mutex mutex_;
+  Atomic<folly::EventBaseManager*> event_base_manager_{nullptr};
+  Atomic<folly::IOThreadPoolExecutor*> executor_{nullptr};
 };
-#endif  // USE_COROUTINES
+#endif  // USE_COROUTINES && FOLLY_HAS_LIBURING
 
 // list of pathnames that are locked
 // Only used for error message.
@@ -207,22 +233,15 @@ class PosixFileSystem : public FileSystem {
   const char* Name() const override { return kClassName(); }
   const char* NickName() const override { return kDefaultName(); }
 
-#if USE_COROUTINES
+#if USE_COROUTINES && FOLLY_HAS_LIBURING
   folly::IOExecutor* GetReadExecutor() override {
-    return read_io_executor_state_.executor_.get();
+    return read_io_executor_state_.GetExecutor();
   }
 
   void SetReadIOExecutorThreads(int num) override {
-    assert(num > 0);
-    if (num <= 0) {
-      return;
-    }
-    if (read_io_executor_state_.executor_ != nullptr) {
-      read_io_executor_state_.executor_->setNumThreads(
-          static_cast<size_t>(num));
-    }
+    read_io_executor_state_.SetThreads(num);
   }
-#endif  // USE_COROUTINES
+#endif  // USE_COROUTINES && FOLLY_HAS_LIBURING
 
   ~PosixFileSystem() override = default;
   bool IsInstanceOf(const std::string& name) const override {
@@ -350,7 +369,7 @@ class PosixFileSystem : public FileSystem {
           options
 #if USE_COROUTINES && FOLLY_HAS_LIBURING
           ,
-          read_io_executor_state_.event_base_manager_.get()
+          read_io_executor_state_.GetEventBaseManager()
 #endif
 #if defined(ROCKSDB_IOURING_PRESENT)
               ,
@@ -1391,9 +1410,9 @@ class PosixFileSystem : public FileSystem {
     supported_ops |= (1 << FSSupportedOps::kFSPrefetch);
   }
 
-#if USE_COROUTINES
+#if USE_COROUTINES && FOLLY_HAS_LIBURING
   ReadIOExecutorState read_io_executor_state_;
-#endif  // USE_COROUTINES
+#endif  // USE_COROUTINES && FOLLY_HAS_LIBURING
 
 #if defined(ROCKSDB_IOURING_PRESENT)
   // io_uring instance

env/io_posix.cc

@@ -87,6 +87,20 @@ IOStatus IOError(const std::string& context, const std::string& file_name,
   }
 }
 
+#if USE_COROUTINES && FOLLY_HAS_LIBURING
+namespace {
+thread_local bool current_thread_read_io_uring_backend_available = false;
+}  // namespace
+
+void SetCurrentThreadReadIOUringBackendAvailable() {
+  current_thread_read_io_uring_backend_available = true;
+}
+
+bool IsCurrentThreadReadIOUringBackendAvailable() {
+  return current_thread_read_io_uring_backend_available;
+}
+#endif  // USE_COROUTINES && FOLLY_HAS_LIBURING
+
 // A wrapper for fadvise, if the platform doesn't support fadvise,
 // it will simply return 0.
 int Fadvise(int fd, off_t offset, size_t len, int advice) {
@@ -701,9 +715,9 @@ struct PosixSubmitReadAsyncState {
 
 }  // namespace
 
-void PosixRandomAccessFile::SubmitReadAsync(
-    FSReadRequest& req, const IOOptions& /*opts*/,
-    std::function<void(FSReadRequest&)> cb, IODebugContext* /*dbg*/) {
+bool PosixRandomAccessFile::SubmitReadAsync(
+    FSReadRequest& req, const IOOptions& opts,
+    std::function<void(FSReadRequest&)> cb, IODebugContext* dbg) {
   if (use_direct_io()) {
     assert(IsSectorAligned(req.offset, GetRequiredBufferAlignment()));
     assert(IsSectorAligned(req.len, GetRequiredBufferAlignment()));
@@ -711,6 +725,14 @@ void PosixRandomAccessFile::SubmitReadAsync(
   }
   assert(req.len == 0 || req.scratch != nullptr);
 
+  // Workers using the default EventBase backend read synchronously.
+  if (!IsCurrentThreadReadIOUringBackendAvailable()) {
+    req.status =
+        Read(req.offset, req.len, opts, &(req.result), req.scratch, dbg);
+    cb(req);
+    return false;
+  }
+
   assert(read_event_base_manager_);
   folly::EventBase* event_base =
       read_event_base_manager_->getExistingEventBase();
@@ -735,6 +757,7 @@ void PosixRandomAccessFile::SubmitReadAsync(
   } else {
     state->Queue();
   }
+  return true;
 }
 #endif  // USE_COROUTINES && FOLLY_HAS_LIBURING
 

env/io_posix.h

@@ -77,6 +77,10 @@ std::string IOErrorMsg(const std::string& context,
 // file_name can be left empty if it is not unkown.
 IOStatus IOError(const std::string& context, const std::string& file_name,
                  int err_number);
+#if USE_COROUTINES && FOLLY_HAS_LIBURING
+void SetCurrentThreadReadIOUringBackendAvailable();
+bool IsCurrentThreadReadIOUringBackendAvailable();
+#endif  // USE_COROUTINES && FOLLY_HAS_LIBURING
 #if !defined(OS_WIN)
 bool PosixPositionedWrite(int fd, const char* buf, size_t nbyte, off_t offset);
 #endif
@@ -465,7 +469,7 @@ class PosixRandomAccessFile : public FSRandomAccessFile {
                              IODebugContext* dbg) override;
 
 #if USE_COROUTINES && FOLLY_HAS_LIBURING
-  void SubmitReadAsync(FSReadRequest& req, const IOOptions& opts,
+  bool SubmitReadAsync(FSReadRequest& req, const IOOptions& opts,
                        std::function<void(FSReadRequest&)> cb,
                        IODebugContext* dbg) override;
 #endif  // USE_COROUTINES && FOLLY_HAS_LIBURING

env/io_posix_test.cc

@@ -12,7 +12,6 @@
 #include "rocksdb/file_system.h"
 
 #if USE_COROUTINES && FOLLY_HAS_LIBURING
-#include "folly/ScopeGuard.h"
 #include "folly/executors/IOThreadPoolExecutor.h"
 #endif  // USE_COROUTINES && FOLLY_HAS_LIBURING
 
@@ -21,24 +20,22 @@ namespace ROCKSDB_NAMESPACE {
 #if USE_COROUTINES && FOLLY_HAS_LIBURING
 class PosixFileSystemTest : public testing::Test {};
 
-TEST_F(PosixFileSystemTest, SetsReadIOExecutorThreads) {
+TEST_F(PosixFileSystemTest, OnlyIncreasesReadIOExecutor) {
   auto fs = FileSystem::Default();
+  fs->SetReadIOExecutorThreads(1);
   auto* read_executor = fs->GetReadExecutor();
   if (read_executor == nullptr) {
     return;
   }
   auto* thread_pool = dynamic_cast<folly::IOThreadPoolExecutor*>(read_executor);
   ASSERT_NE(thread_pool, nullptr);
-  ASSERT_NE(read_executor->getEventBase(), nullptr);
-  const auto original_threads = thread_pool->numThreads();
-  auto restore_threads = folly::makeGuard([&] {
-    fs->SetReadIOExecutorThreads(static_cast<int>(original_threads));
-  });
+  const size_t original_threads = thread_pool->numThreads();
 
-  fs->SetReadIOExecutorThreads(1);
-  ASSERT_EQ(thread_pool->numThreads(), static_cast<size_t>(1));
-  fs->SetReadIOExecutorThreads(2);
-  ASSERT_EQ(thread_pool->numThreads(), static_cast<size_t>(2));
+  fs->SetReadIOExecutorThreads(static_cast<int>(original_threads + 1));
+  EXPECT_EQ(thread_pool->numThreads(), original_threads + 1);
+
+  fs->SetReadIOExecutorThreads(static_cast<int>(original_threads));
+  EXPECT_EQ(thread_pool->numThreads(), original_threads + 1);
 }
 
 #endif  // USE_COROUTINES && FOLLY_HAS_LIBURING

file/random_access_file_reader_sync_and_async.h

@@ -20,11 +20,9 @@
 namespace ROCKSDB_NAMESPACE {
 
 #if defined(WITH_COROUTINES)
-inline folly::coro::Task<void> SubmitMultiReadAsync(FSRandomAccessFile* file,
-                                                    FSReadRequest* fs_reqs,
-                                                    size_t num_fs_reqs,
-                                                    const IOOptions& opts,
-                                                    IODebugContext* dbg) {
+inline folly::coro::Task<void> SubmitMultiReadAsync(
+    FSRandomAccessFile* file, FSReadRequest* fs_reqs, size_t num_fs_reqs,
+    const IOOptions& opts, Statistics* stats, IODebugContext* dbg) {
   if (num_fs_reqs == 0) {
     co_return;
   }
@@ -39,14 +37,16 @@ inline folly::coro::Task<void> SubmitMultiReadAsync(FSRandomAccessFile* file,
       [&] { assert(pending.load(std::memory_order_acquire) == 0); });
 #endif  // NDEBUG
   for (size_t i = 0; i < num_fs_reqs; ++i) {
-    file->SubmitReadAsync(
-        fs_reqs[i], opts,
-        [executor, &baton, &pending](FSReadRequest& /*req*/) {
-          if (pending.fetch_sub(1) == 1) {
-            executor->add([&baton] { baton.post(); });
-          }
-        },
-        dbg);
+    if (!file->SubmitReadAsync(
+            fs_reqs[i], opts,
+            [executor, &baton, &pending](FSReadRequest& /*req*/) {
+              if (pending.fetch_sub(1) == 1) {
+                executor->add([&baton] { baton.post(); });
+              }
+            },
+            dbg)) {
+      RecordTick(stats, FILE_SUBMIT_ASYNC_READ_FALLBACK, 1);
+    }
   }
 
   co_await baton;
@@ -145,7 +145,8 @@ DEFINE_SYNC_AND_ASYNC(IOStatus, RandomAccessFileReader::Read)
         fs_req.status.PermitUncheckedError();
         TEST_SYNC_POINT_CALLBACK(
             "RandomAccessFileReader::ReadCoroutine:SubmitReadAsync", &fs_req);
-        co_await SubmitMultiReadAsync(file_.get(), &fs_req, 1, opts, dbg);
+        co_await SubmitMultiReadAsync(file_.get(), &fs_req, 1, opts, stats_,
+                                      dbg);
         io_s = fs_req.status;
         tmp = fs_req.result;
 #else
@@ -224,7 +225,8 @@ DEFINE_SYNC_AND_ASYNC(IOStatus, RandomAccessFileReader::Read)
         if (fs_req.scratch != nullptr) {
           TEST_SYNC_POINT_CALLBACK(
               "RandomAccessFileReader::ReadCoroutine:SubmitReadAsync", &fs_req);
-          co_await SubmitMultiReadAsync(file_.get(), &fs_req, 1, opts, dbg);
+          co_await SubmitMultiReadAsync(file_.get(), &fs_req, 1, opts, stats_,
+                                        dbg);
           io_s = fs_req.status;
           tmp_result = fs_req.result;
         } else {
@@ -408,7 +410,7 @@ DEFINE_SYNC_AND_ASYNC(IOStatus, RandomAccessFileReader::MultiRead)
           const_cast<void*>(static_cast<void*>(dbg)));
 #if defined(USE_COROUTINES) && defined(WITH_COROUTINES)
       co_await SubmitMultiReadAsync(file_.get(), fs_reqs, num_fs_reqs, opts,
-                                    dbg);
+                                    stats_, dbg);
 #else
       io_s = file_->MultiRead(fs_reqs, num_fs_reqs, opts, dbg);
 #endif

file/random_access_file_reader_test.cc

@@ -16,6 +16,14 @@
 #include "test_util/testutil.h"
 #include "util/random.h"
 
+#if defined(USE_COROUTINES)
+#include "folly/Executor.h"
+#include "folly/coro/BlockingWait.h"
+#include "folly/coro/Task.h"
+#include "folly/executors/IOThreadPoolExecutor.h"
+#include "rocksdb/statistics.h"
+#endif
+
 namespace ROCKSDB_NAMESPACE {
 
 class RandomAccessFileReaderTest : public testing::Test {
@@ -101,8 +109,39 @@ class ScratchObservingRandomAccessFile : public FSRandomAccessFileOwnerWrapper {
  private:
   bool* saw_null_scratch_;
 };
+
 }  // namespace
 
+#if defined(USE_COROUTINES)
+TEST_F(RandomAccessFileReaderTest, CountsCoroutineReadSyncFallback) {
+  const std::string fname = "async-read-fallback";
+  const std::string content = "async read fallback";
+  Write(fname, content);
+
+  const std::string fpath = Path(fname);
+  std::unique_ptr<FSRandomAccessFile> file;
+  ASSERT_OK(fs_->NewRandomAccessFile(fpath, FileOptions(), &file, nullptr));
+
+  auto statistics = CreateDBStatistics();
+  RandomAccessFileReader reader(std::move(file), fpath,
+                                env_->GetSystemClock().get(), nullptr,
+                                statistics.get());
+  std::string scratch(content.size(), '\0');
+  Slice result;
+
+  folly::IOThreadPoolExecutor executor(1);
+  folly::EventBase* event_base = executor.getEventBase();
+  ASSERT_NE(event_base, nullptr);
+  ASSERT_OK(folly::coro::blockingWait(folly::coro::co_withExecutor(
+      folly::Executor::getKeepAliveToken(event_base),
+      reader.ReadCoroutine(IOOptions(), 0, content.size(), &result,
+                           scratch.data(), nullptr, nullptr))));
+
+  ASSERT_EQ(content, result.ToString());
+  EXPECT_EQ(1, statistics->getTickerCount(FILE_SUBMIT_ASYNC_READ_FALLBACK));
+}
+#endif  // USE_COROUTINES
+
 // Skip the following tests in lite mode since direct I/O is unsupported.
 
 TEST_F(RandomAccessFileReaderTest, ReadDirectIO) {

include/rocksdb/c.h

@@ -4993,6 +4993,12 @@ rocksdb_options_set_verify_sst_unique_id_in_manifest(rocksdb_options_t* opt,
 extern ROCKSDB_LIBRARY_API unsigned char
 rocksdb_options_get_verify_sst_unique_id_in_manifest(rocksdb_options_t* opt);
 
+extern ROCKSDB_LIBRARY_API void rocksdb_options_set_read_io_executor_threads(
+    rocksdb_options_t* opt, int v);
+
+extern ROCKSDB_LIBRARY_API int rocksdb_options_get_read_io_executor_threads(
+    rocksdb_options_t* opt);
+
 extern ROCKSDB_LIBRARY_API const char* rocksdb_options_get_db_log_dir(
     rocksdb_options_t* opt, size_t* size);
 

include/rocksdb/db.h

@@ -1113,7 +1113,9 @@ class DB {
   // is available, RocksDB can run an internal read coroutine on the read
   // executor, suspend while async file IO is outstanding, and resume when the
   // filesystem signals completion. The callback is invoked after the requested
-  // statuses and output buffers have been populated.
+  // statuses and output buffers have been populated. Applications can set
+  // `DBOptions::read_io_executor_threads` before opening the DB to configure
+  // executor parallelism for their workload.
   //
   // Callers must keep the DB, callback, inputs, and output buffers alive until
   // the callback returns. The callback may run inline before the async method

include/rocksdb/file_system.h

@@ -466,8 +466,9 @@ class FileSystem : public Customizable {
   // coroutine read execution is not supported.
   virtual folly::IOExecutor* GetReadExecutor() { return nullptr; }
 
-  // Set the maximum number of threads used by the FileSystem-owned read IO
-  // executor. The number must be positive.
+  // Increase the maximum number of threads used by the FileSystem-owned read
+  // IO executor to at least the requested number. The executor is not reduced
+  // when its current maximum is larger. The number must be positive.
   virtual void SetReadIOExecutorThreads(int /*number*/) {}
 
   // Create a brand new sequentially-readable file with the specified name.
@@ -1133,14 +1134,18 @@ class FSRandomAccessFile {
   // completion executor and must update req.status and invoke cb(req) exactly
   // once.
   //
+  // Returns true when the asynchronous path is used and false when the request
+  // is completed using the synchronous Read fallback.
+  //
   // Default implementation is to read the data synchronously and invoke the
   // callback before returning.
-  virtual void SubmitReadAsync(FSReadRequest& req, const IOOptions& opts,
+  virtual bool SubmitReadAsync(FSReadRequest& req, const IOOptions& opts,
                                std::function<void(FSReadRequest&)> cb,
                                IODebugContext* dbg) {
     req.status =
         Read(req.offset, req.len, opts, &(req.result), req.scratch, dbg);
     cb(req);
+    return false;
   }
 
   // EXPERIMENTAL
@@ -1919,10 +1924,10 @@ class FSRandomAccessFileWrapper : public FSRandomAccessFile {
                      IODebugContext* dbg) override {
     return target()->ReadAsync(req, opts, cb, cb_arg, io_handle, del_fn, dbg);
   }
-  void SubmitReadAsync(FSReadRequest& req, const IOOptions& opts,
+  bool SubmitReadAsync(FSReadRequest& req, const IOOptions& opts,
                        std::function<void(FSReadRequest&)> cb,
                        IODebugContext* dbg) override {
-    target()->SubmitReadAsync(req, opts, std::move(cb), dbg);
+    return target()->SubmitReadAsync(req, opts, std::move(cb), dbg);
   }
   Temperature GetTemperature() const override {
     return target_->GetTemperature();

include/rocksdb/options.h

@@ -780,6 +780,11 @@ struct DBOptions {
   // Default: 16
   int max_file_opening_threads = 16;
 
+  // Requested maximum number of threads in the shared read I/O executor. A DB
+  // open can increase the executor to this size but cannot reduce it. Used
+  // exclusively for asynchronous read requests (e.g. GetAsync, MultiGetAsync).
+  int read_io_executor_threads = 1;
+
   // If true, SST files are opened and validated asynchronously in the
   // background after DB::Open returns. This reduces DB open time for
   // databases with many SST files and high latency file systems. Mostly useful

include/rocksdb/statistics.h

@@ -653,6 +653,9 @@ enum Tickers : uint32_t {
   CHECKPOINT_READ_BYTES,
   CHECKPOINT_WRITE_BYTES,
 
+  // Number of SubmitReadAsync calls that fell back to a synchronous read
+  FILE_SUBMIT_ASYNC_READ_FALLBACK,
+
   TICKER_ENUM_MAX
 };
 

java/rocksjni/portal.h

@@ -5357,6 +5357,8 @@ class TickerTypeJni {
         return -0x7E;
       case ROCKSDB_NAMESPACE::Tickers::ATOMIC_FLUSH_REQUEST_REASON_OTHER:
         return -0x7F;
+      case ROCKSDB_NAMESPACE::Tickers::FILE_SUBMIT_ASYNC_READ_FALLBACK:
+        return -0x80;
       case ROCKSDB_NAMESPACE::Tickers::TICKER_ENUM_MAX:
         // -0x54 is the max value at this time. Since these values are exposed
         // directly to Java clients, we'll keep the value the same till the next
@@ -5904,6 +5906,8 @@ class TickerTypeJni {
             ATOMIC_FLUSH_REQUEST_REASON_MEMTABLE_MAX_RANGE_DELETIONS;
       case -0x7F:
         return ROCKSDB_NAMESPACE::Tickers::ATOMIC_FLUSH_REQUEST_REASON_OTHER;
+      case -0x80:
+        return ROCKSDB_NAMESPACE::Tickers::FILE_SUBMIT_ASYNC_READ_FALLBACK;
       case -0x54:
         // -0x54 is the max value at this time. Since these values are exposed
         // directly to Java clients, we'll keep the value the same till the next

java/src/main/java/org/rocksdb/TickerType.java

@@ -1077,6 +1077,11 @@ public enum TickerType {
      */
     ATOMIC_FLUSH_REQUEST_REASON_OTHER((byte) -0x7F),
 
+    /**
+     * Number of SubmitReadAsync calls that fell back to a synchronous read
+     */
+    FILE_SUBMIT_ASYNC_READ_FALLBACK((byte) -0x80),
+
     TICKER_ENUM_MAX((byte) -0x54);
 
     private final byte value;

monitoring/statistics.cc

@@ -336,6 +336,8 @@ const std::vector<std::pair<Tickers, std::string>> TickersNameMap = {
      "rocksdb.atomic_flush.request.reason.other"},
     {CHECKPOINT_READ_BYTES, "rocksdb.checkpoint.read.bytes"},
     {CHECKPOINT_WRITE_BYTES, "rocksdb.checkpoint.write.bytes"},
+    {FILE_SUBMIT_ASYNC_READ_FALLBACK,
+     "rocksdb.file.submit.async.read.fallback"},
 };
 
 const std::vector<std::pair<Histograms, std::string>> HistogramsNameMap = {

options/db_options.cc

@@ -304,6 +304,10 @@ static std::unordered_map<std::string, OptionTypeInfo>
          {offsetof(struct ImmutableDBOptions, max_file_opening_threads),
           OptionType::kInt, OptionVerificationType::kNormal,
           OptionTypeFlags::kNone}},
+        {"read_io_executor_threads",
+         {offsetof(struct ImmutableDBOptions, read_io_executor_threads),
+          OptionType::kInt, OptionVerificationType::kNormal,
+          OptionTypeFlags::kNone}},
         {"table_cache_numshardbits",
          {offsetof(struct ImmutableDBOptions, table_cache_numshardbits),
           OptionType::kInt, OptionVerificationType::kNormal,
@@ -772,6 +776,7 @@ ImmutableDBOptions::ImmutableDBOptions(const DBOptions& options)
       info_log(options.info_log),
       info_log_level(options.info_log_level),
       max_file_opening_threads(options.max_file_opening_threads),
+      read_io_executor_threads(options.read_io_executor_threads),
       statistics(options.statistics),
       use_fsync(options.use_fsync),
       db_paths(options.db_paths),
@@ -886,6 +891,8 @@ void ImmutableDBOptions::Dump(Logger* log) const {
                    info_log.get());
   ROCKS_LOG_HEADER(log, "               Options.max_file_opening_threads: %d",
                    max_file_opening_threads);
+  ROCKS_LOG_HEADER(log, "               Options.read_io_executor_threads: %d",
+                   read_io_executor_threads);
   ROCKS_LOG_HEADER(log, "                             Options.statistics: %p",
                    stats);
   if (stats) {

options/db_options.h

@@ -36,6 +36,7 @@ struct ImmutableDBOptions {
   std::shared_ptr<Logger> info_log;
   InfoLogLevel info_log_level;
   int max_file_opening_threads;
+  int read_io_executor_threads;
   std::shared_ptr<Statistics> statistics;
   bool use_fsync;
   std::vector<DbPath> db_paths;

options/options_helper.cc

@@ -84,6 +84,8 @@ void BuildDBOptions(const ImmutableDBOptions& immutable_db_options,
   options.max_open_files = mutable_db_options.max_open_files;
   options.max_file_opening_threads =
       immutable_db_options.max_file_opening_threads;
+  options.read_io_executor_threads =
+      immutable_db_options.read_io_executor_threads;
   options.max_total_wal_size = mutable_db_options.max_total_wal_size;
   options.statistics = immutable_db_options.statistics;
   options.use_fsync = immutable_db_options.use_fsync;

options/options_settable_test.cc

@@ -406,6 +406,7 @@ TEST_F(OptionsSettableTest, DBOptionsAllFieldsSettable) {
       "table_cache_numshardbits=28;"
       "max_open_files=72;"
       "max_file_opening_threads=35;"
+      "read_io_executor_threads=7;"
       "max_background_jobs=8;"
       "max_background_compactions=33;"
       "use_fsync=true;"

unreleased_history/public_api_changes/read_io_executor_threads.md

@@ -0,0 +1 @@
+Add `DBOptions::read_io_executor_threads` to increase the maximum thread count of the shared filesystem read I/O executor used for asynchronous reads when opening a DB. Opening a DB never reduces the shared executor thread count. Change the experimental `FSRandomAccessFile::SubmitReadAsync()` API to return whether the asynchronous path was used, and add the `FILE_SUBMIT_ASYNC_READ_FALLBACK` ticker to count calls that fall back to synchronous reads.