Additional HISTORY.md update for 11.8.1

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

This change reduces DB open latency on remote storage by avoiding foreground deletion of obsolete `OPTIONS-*` files during open when `avoid_unnecessary_blocking_io` is enabled. DB open now avoids foreground deletion of obsolete `OPTIONS-*` files on the pre-open `WriteOptionsFile()` path. `RenameTempFileToOptionsFile()` records the new OPTIONS file and chooses an explicit cleanup mode: skip while open is still in progress, schedule deletion when background work is available, or delete immediately when blocking IO is allowed or background jobs are rejected.

`DBImpl::Open()` makes the deferred cleanup decision once final open status is known. A successful open schedules obsolete OPTIONS files on the purge queue, while a failed open deletes them synchronously before tearing down the partially opened DB. The default synchronous behavior remains unchanged when `avoid_unnecessary_blocking_io` is disabled, and the latest-two OPTIONS retention policy is preserved.

Reviewed By: pdillinger

Differential Revision: D113593680

fbshipit-source-id: 8aee99b3f38a729647d0bcb644d212181fe86895

HISTORY.md

@@ -1,6 +1,25 @@
 # Rocksdb Change Log
 > NOTE: Entries for next release do not go here. Follow instructions in `unreleased_history/README.txt`
 
+## 11.8.1 (07/28/2026)
+### Public API Changes
+* 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.
+
+### Performance Improvements
+* When `DBOptions::avoid_unnecessary_blocking_io` is true, obsolete `OPTIONS-*` files found during DB open are deleted by background purge instead of synchronously on the opening thread.
+
+## 11.8.0 (07/28/2026)
+### Public API Changes
+* Add callback-based asynchronous read APIs, `DB::GetAsync()` and `DB::MultiGetAsync()`. The idea is that when IO is required, RocksDB can suspend its internal coroutine read path, allowing the read executor thread to do other work. When the IO is complete, RocksDB invokes the user callback. This requires filesystem support for full performance benefits. A new filesystem API `FSRandomAccessFile::SubmitReadAsync()` is introduced for this. Unlike `ReadAsync`, the filesystem is responsible for eventually calling the callback. `FileSystem::GetReadExecutor()` returns the IO executor whose EventBases run coroutine read processing. In the Posix filesystem, an IO-uring based event loop is used to complete the IO.
+* Added `FlushOptions::listener_wait` (default `false`). When set together with `FlushOptions::wait == true`, `DB::Flush()` will not return until the registered `EventListener::OnFlushCompleted` callbacks for the flushed memtables have finished running. By default (`false`), `Flush(wait=true)` may return as soon as the flush result is committed, which can be before (or while) the `OnFlushCompleted` callbacks execute on the background flush thread. Also added the corresponding C API `rocksdb_flushoptions_set_listener_wait()` / `rocksdb_flushoptions_get_listener_wait()`.
+* Deprecated `PinnableWideColumns::serialized_size()` in favor of the new `PinnableWideColumns::payload_size()`, which returns the total size of the columns' names and values (for a plain value this equals the value size). Relatedly, wide-column read accounting changed subtly: read statistics for entities (`BYTES_READ`, `BYTES_PER_READ`, and the per-key byte counts for `GetEntity`/`MultiGetEntity`) and the `ReadOptions::value_size_soft_limit` threshold now measure entities by this payload size instead of the serialized entity size. This drops serialization framing bytes from those measurements (so reported/limited sizes for entity reads are slightly smaller than before) and makes plain-value and wide-column accounting consistent.
+
+### Behavior Changes
+* `ReadOptions::value_size_soft_limit` now also bounds the blob values resolved from pre-flush blob direct write references during `MultiGet`/`MultiGetEntity` (previously these were resolved after the limit check and read without bound). Enforcement is "always make progress": at least one key is read even if its value alone exceeds the limit, and only once the returned size exceeds the limit do subsequent keys get `Status::Aborted` -- so a caller retrying the aborted keys cannot loop forever on a single value that by itself exceeds the limit.
+
+### Performance Improvements
+* Reduced copying when reading wide-column entities that have blob-referenced columns, whether stored in separate blob files or embedded (same-file) as blob records in the same SST. On the point-lookup (`GetEntity`/`MultiGetEntity`) and blob-backed memtable/direct-write read paths, resolving these columns no longer re-serializes the whole entity into a fresh buffer: inline columns are referenced in place in the pinned entity and each resolved blob value is referenced in place from its fetched buffer. This is enabled by an internal change to `PinnableWideColumns` that lets its columns reference multiple backing buffers.
+
 ## 11.7.0 (07/16/2026)
 ### Public API Changes
 * Added `RangeLockManagerHandle::SetIsKilledCallback()`, a `std::function<bool()>` predicate polled during range-lock waits. When it returns true, a blocked `GetRangeLock()` wait returns promptly with `Status::Aborted()` instead of waiting for the lock timeout. Range-lock users that install no callback keep the original wait-until-grant-or-timeout behavior.

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_);
@@ -5830,7 +5835,7 @@ void DeleteOptionsFilesHelper(const std::map<uint64_t, std::string>& filenames,
 }
 }  // namespace
 
-Status DBImpl::DeleteObsoleteOptionsFiles() {
+Status DBImpl::DeleteObsoleteOptionsFiles(bool schedule_only) {
   std::vector<std::string> filenames;
   // use ordered map to store keep the filenames sorted from the newest
   // to the oldest.
@@ -5855,8 +5860,23 @@ Status DBImpl::DeleteObsoleteOptionsFiles() {
 
   // Keeps the latest 2 Options file
   const size_t kNumOptionsFilesKept = 2;
-  DeleteOptionsFilesHelper(options_filenames, kNumOptionsFilesKept,
-                           immutable_db_options_.info_log, GetEnv());
+  if (options_filenames.size() > kNumOptionsFilesKept) {
+    if (schedule_only) {
+      InstrumentedMutexLock l(&mutex_);
+      for (auto iter =
+               std::next(options_filenames.begin(), kNumOptionsFilesKept);
+           iter != options_filenames.end(); ++iter) {
+        const uint64_t file_number =
+            std::numeric_limits<uint64_t>::max() - iter->first;
+        SchedulePendingPurge(iter->second, GetName(), kOptionsFile, file_number,
+                             /*job_id=*/0);
+      }
+      SchedulePurge();
+    } else {
+      DeleteOptionsFilesHelper(options_filenames, kNumOptionsFilesKept,
+                               immutable_db_options_.info_log, GetEnv());
+    }
+  }
   return Status::OK();
 }
 
@@ -5896,18 +5916,42 @@ Status DBImpl::RenameTempFileToOptionsFile(const std::string& file_name,
   }
 
   if (s.ok()) {
-    int my_disable_delete_obsolete_files;
+    enum class ObsoleteOptionsFileCleanup {
+      kSkip,
+      kDeleteNow,
+      kSchedule,
+    };
+    ObsoleteOptionsFileCleanup obsolete_options_file_cleanup =
+        ObsoleteOptionsFileCleanup::kSkip;
 
     {
       InstrumentedMutexLock l(&mutex_);
       versions_->options_file_number_ = options_file_number;
       versions_->options_file_size_ = options_file_size;
-      my_disable_delete_obsolete_files = disable_delete_obsolete_files_;
+      if (!disable_delete_obsolete_files_ && !is_remote_compaction_enabled) {
+        if (immutable_db_options_.avoid_unnecessary_blocking_io &&
+            !reject_new_background_jobs_) {
+          // DB::Open() sets `opened_successfully_` after WriteOptionsFile()
+          // returns, then schedules the deferred OPTIONS-file purge.
+          obsolete_options_file_cleanup =
+              opened_successfully_ ? ObsoleteOptionsFileCleanup::kSchedule
+                                   : ObsoleteOptionsFileCleanup::kSkip;
+        } else {
+          obsolete_options_file_cleanup =
+              ObsoleteOptionsFileCleanup::kDeleteNow;
+        }
+      }
     }
 
-    if (!my_disable_delete_obsolete_files && !is_remote_compaction_enabled) {
-      // TODO: Should we check for errors here?
-      DeleteObsoleteOptionsFiles().PermitUncheckedError();
+    if (obsolete_options_file_cleanup != ObsoleteOptionsFileCleanup::kSkip) {
+      Status obsolete_options_status =
+          DeleteObsoleteOptionsFiles(obsolete_options_file_cleanup ==
+                                     ObsoleteOptionsFileCleanup::kSchedule);
+      if (!obsolete_options_status.ok()) {
+        ROCKS_LOG_WARN(immutable_db_options_.info_log,
+                       "Unable to delete obsolete OPTIONS files: %s",
+                       obsolete_options_status.ToString().c_str());
+      }
     }
   }
 
@@ -7568,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);
@@ -7620,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.h

@@ -1562,7 +1562,7 @@ class DBImpl : public DB {
   // 2. db_mutex is NOT held
   Status RenameTempFileToOptionsFile(const std::string& file_name,
                                      bool is_remote_compaction_enabled);
-  Status DeleteObsoleteOptionsFiles();
+  Status DeleteObsoleteOptionsFiles(bool schedule_only);
 
   void NotifyOnManualFlushScheduled(autovector<ColumnFamilyData*> cfds,
                                     FlushReason flush_reason);

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. ");
@@ -2872,11 +2877,16 @@ Status DBImpl::Open(const DBOptions& db_options, const std::string& dbname,
   }
   TEST_SYNC_POINT("DBImpl::Open:Opened");
   Status persist_options_status;
+  bool cleanup_obsolete_options_files = false;
   if (s.ok()) {
     // Persist RocksDB Options before scheduling the compaction.
     // The WriteOptionsFile() will release and lock the mutex internally.
     persist_options_status =
         impl->WriteOptionsFile(write_options, true /*db_mutex_already_held*/);
+    cleanup_obsolete_options_files =
+        impl->immutable_db_options_.avoid_unnecessary_blocking_io &&
+        impl->immutable_db_options_.compaction_service == nullptr &&
+        !impl->disable_delete_obsolete_files_;
     impl->opened_successfully_ = true;
   } else {
     persist_options_status.PermitUncheckedError();
@@ -2966,6 +2976,17 @@ Status DBImpl::Open(const DBOptions& db_options, const std::string& dbname,
     s = impl->RegisterRecordSeqnoTimeWorker();
   }
   impl->options_mutex_.Unlock();
+  if (cleanup_obsolete_options_files) {
+    Status obsolete_options_status =
+        impl->DeleteObsoleteOptionsFiles(/*schedule_only=*/s.ok());
+    if (!obsolete_options_status.ok()) {
+      ROCKS_LOG_WARN(impl->immutable_db_options_.info_log,
+                     "Unable to %s obsolete OPTIONS files%s: %s",
+                     s.ok() ? "schedule deletion of" : "delete",
+                     s.ok() ? "" : " after DB open failure",
+                     obsolete_options_status.ToString().c_str());
+    }
+  }
   if (s.ok()) {
     *dbptr = std::move(impl);
   } else {

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/deletefile_test.cc

@@ -418,6 +418,15 @@ TEST_F(DeleteFileTest, BackgroundPurgeCFDropTest) {
     // Expect 1 sst file.
     CheckFileTypeCounts(dbname_, 0, 1, 1);
 
+    if (bg_purge) {
+      SyncPoint::GetInstance()->DisableProcessing();
+      SyncPoint::GetInstance()->ClearAllCallBacks();
+      SyncPoint::GetInstance()->LoadDependency(
+          {{"DeleteFileTest::BackgroundPurgeCFDropTest:1",
+            "DBImpl::BGWorkPurge:start"}});
+      SyncPoint::GetInstance()->EnableProcessing();
+    }
+
     ASSERT_OK(db_->DropColumnFamily(cfh));
     // Still 1 file, it won't be deleted while ColumnFamilyHandle is alive.
     CheckFileTypeCounts(dbname_, 0, 1, 1);
@@ -447,13 +456,6 @@ TEST_F(DeleteFileTest, BackgroundPurgeCFDropTest) {
   Reopen(options);
   ASSERT_OK(dbfull()->TEST_WaitForPurge());
 
-  SyncPoint::GetInstance()->DisableProcessing();
-  SyncPoint::GetInstance()->ClearAllCallBacks();
-  SyncPoint::GetInstance()->LoadDependency(
-      {{"DeleteFileTest::BackgroundPurgeCFDropTest:1",
-        "DBImpl::BGWorkPurge:start"}});
-  SyncPoint::GetInstance()->EnableProcessing();
-
   {
     SCOPED_TRACE("avoid_unnecessary_blocking_io = true");
     do_test(true);

db/options_file_test.cc

@@ -3,13 +3,17 @@
 //  COPYING file in the root directory) and Apache 2.0 License
 //  (found in the LICENSE.Apache file in the root directory).
 
+#include <atomic>
+#include <memory>
 #include <string>
+#include <utility>
 
 #include "db/db_impl/db_impl.h"
 #include "db/db_test_util.h"
 #include "rocksdb/options.h"
 #include "rocksdb/table.h"
 #include "test_util/testharness.h"
+#include "util/cast_util.h"
 
 namespace ROCKSDB_NAMESPACE {
 class OptionsFileTest : public testing::Test {
@@ -20,6 +24,91 @@ class OptionsFileTest : public testing::Test {
 };
 
 namespace {
+std::string Basename(const std::string& fname) {
+  const size_t basename_pos = fname.find_last_of("/\\");
+  return basename_pos == std::string::npos ? fname
+                                           : fname.substr(basename_pos + 1);
+}
+
+bool IsOptionsFile(const std::string& fname) {
+  return Basename(fname).find(kOptionsFileNamePrefix) == 0;
+}
+
+class FailOptionsDirFsyncFileSystem : public FileSystemWrapper {
+ public:
+  explicit FailOptionsDirFsyncFileSystem(std::shared_ptr<FileSystem> target)
+      : FileSystemWrapper(std::move(target)) {}
+
+  static const char* kClassName() { return "FailOptionsDirFsyncFileSystem"; }
+  const char* Name() const override { return kClassName(); }
+
+  void FailOptionsDirFsyncAfterNextOptionsFile() {
+    observe_options_file_.store(true, std::memory_order_release);
+    fail_options_dir_fsync_.store(false, std::memory_order_release);
+    options_dir_fsync_failures_.store(0, std::memory_order_release);
+  }
+
+  int options_dir_fsync_failures() const {
+    return options_dir_fsync_failures_.load(std::memory_order_acquire);
+  }
+
+  IOStatus NewWritableFile(const std::string& fname,
+                           const FileOptions& file_opts,
+                           std::unique_ptr<FSWritableFile>* result,
+                           IODebugContext* dbg) override {
+    IOStatus s =
+        FileSystemWrapper::NewWritableFile(fname, file_opts, result, dbg);
+    if (!s.ok()) {
+      return s;
+    }
+    if (observe_options_file_.load(std::memory_order_acquire) &&
+        IsOptionsFile(fname)) {
+      fail_options_dir_fsync_.store(true, std::memory_order_release);
+      observe_options_file_.store(false, std::memory_order_release);
+    }
+    return s;
+  }
+
+  IOStatus NewDirectory(const std::string& name, const IOOptions& io_opts,
+                        std::unique_ptr<FSDirectory>* result,
+                        IODebugContext* dbg) override {
+    IOStatus s = FileSystemWrapper::NewDirectory(name, io_opts, result, dbg);
+    if (s.ok()) {
+      *result = std::make_unique<FailOptionsDirFsyncDirectory>(
+          std::move(*result), *this);
+    }
+    return s;
+  }
+
+ private:
+  class FailOptionsDirFsyncDirectory : public FSDirectoryWrapper {
+   public:
+    FailOptionsDirFsyncDirectory(std::unique_ptr<FSDirectory>&& dir,
+                                 FailOptionsDirFsyncFileSystem& fs)
+        : FSDirectoryWrapper(std::move(dir)), fs_(fs) {}
+
+    IOStatus FsyncWithDirOptions(
+        const IOOptions& options, IODebugContext* dbg,
+        const DirFsyncOptions& dir_fsync_options) override {
+      if (fs_.fail_options_dir_fsync_.load(std::memory_order_acquire) &&
+          IsOptionsFile(dir_fsync_options.renamed_new_name)) {
+        fs_.fail_options_dir_fsync_.store(false, std::memory_order_release);
+        fs_.options_dir_fsync_failures_.fetch_add(1, std::memory_order_acq_rel);
+        return IOStatus::IOError("Injected OPTIONS directory fsync failure");
+      }
+      return FSDirectoryWrapper::FsyncWithDirOptions(options, dbg,
+                                                     dir_fsync_options);
+    }
+
+   private:
+    FailOptionsDirFsyncFileSystem& fs_;
+  };
+
+  std::atomic<bool> observe_options_file_{false};
+  std::atomic<bool> fail_options_dir_fsync_{false};
+  std::atomic<int> options_dir_fsync_failures_{0};
+};
+
 void UpdateOptionsFiles(DB* db,
                         std::unordered_set<std::string>* filename_history,
                         int* options_files_count) {
@@ -58,6 +147,34 @@ void VerifyOptionsFileName(
     }
   }
 }
+
+int CountOptionsFiles(Env* env, const std::string& dbname) {
+  std::vector<std::string> filenames;
+  EXPECT_OK(env->GetChildren(dbname, &filenames));
+
+  int options_files_count = 0;
+  uint64_t number;
+  FileType type;
+  for (const auto& filename : filenames) {
+    if (ParseFileName(filename, &number, &type) && type == kOptionsFile) {
+      ++options_files_count;
+    }
+  }
+  return options_files_count;
+}
+
+int CountOptionsFiles(DB* db) {
+  return CountOptionsFiles(db->GetEnv(), db->GetName());
+}
+
+void GenerateStaleOptionsFiles(DB* db, int options_file_count) {
+  ASSERT_OK(db->DisableFileDeletions());
+  for (int i = 0; i < options_file_count; ++i) {
+    ASSERT_OK(db->SetOptions(
+        {{"level0_file_num_compaction_trigger", std::to_string(8 + i)}}));
+  }
+  ASSERT_GT(CountOptionsFiles(db), 2);
+}
 }  // anonymous namespace
 
 TEST_F(OptionsFileTest, NumberOfOptionsFiles) {
@@ -79,6 +196,80 @@ TEST_F(OptionsFileTest, NumberOfOptionsFiles) {
   }
 }
 
+TEST_F(OptionsFileTest, ObsoleteOptionsFilesPurgedSynchronouslyOnOpen) {
+  Options opt;
+  opt.create_if_missing = true;
+  ASSERT_OK(DestroyDB(dbname_, opt));
+
+  std::unique_ptr<DB> db;
+  ASSERT_OK(DB::Open(opt, dbname_, &db));
+  GenerateStaleOptionsFiles(db.get(), 5);
+  db.reset();
+
+  opt.create_if_missing = false;
+  opt.avoid_unnecessary_blocking_io = false;
+  ASSERT_OK(DB::Open(opt, dbname_, &db));
+  ASSERT_LE(CountOptionsFiles(db.get()), 2);
+}
+
+TEST_F(OptionsFileTest, ObsoleteOptionsFilesPurgedInBackgroundOnOpen) {
+  Options opt;
+  opt.create_if_missing = true;
+  ASSERT_OK(DestroyDB(dbname_, opt));
+
+  std::unique_ptr<DB> db;
+  ASSERT_OK(DB::Open(opt, dbname_, &db));
+  GenerateStaleOptionsFiles(db.get(), 5);
+  db.reset();
+
+  opt.create_if_missing = false;
+  opt.avoid_unnecessary_blocking_io = true;
+
+  SyncPoint::GetInstance()->DisableProcessing();
+  SyncPoint::GetInstance()->ClearAllCallBacks();
+  SyncPoint::GetInstance()->LoadDependency(
+      {{"OptionsFileTest::ObsoleteOptionsFilesPurgedInBackgroundOnOpen:"
+        "ReleasePurge",
+        "DBImpl::BGWorkPurge:start"}});
+  SyncPoint::GetInstance()->EnableProcessing();
+
+  ASSERT_OK(DB::Open(opt, dbname_, &db));
+  ASSERT_GT(CountOptionsFiles(db.get()), 2);
+
+  TEST_SYNC_POINT(
+      "OptionsFileTest::ObsoleteOptionsFilesPurgedInBackgroundOnOpen:"
+      "ReleasePurge");
+  ASSERT_OK(static_cast_with_check<DBImpl>(db.get())->TEST_WaitForPurge());
+  ASSERT_LE(CountOptionsFiles(db.get()), 2);
+
+  SyncPoint::GetInstance()->DisableProcessing();
+  SyncPoint::GetInstance()->ClearAllCallBacks();
+}
+
+TEST_F(OptionsFileTest, ObsoleteOptionsFilesPurgedSynchronouslyOnOpenFailure) {
+  auto fail_fs =
+      std::make_shared<FailOptionsDirFsyncFileSystem>(FileSystem::Default());
+  std::unique_ptr<Env> env(NewCompositeEnv(fail_fs));
+
+  Options opt;
+  opt.env = env.get();
+  opt.create_if_missing = true;
+  ASSERT_OK(DestroyDB(dbname_, opt));
+
+  std::unique_ptr<DB> db;
+  ASSERT_OK(DB::Open(opt, dbname_, &db));
+  GenerateStaleOptionsFiles(db.get(), 5);
+  db.reset();
+
+  opt.create_if_missing = false;
+  opt.avoid_unnecessary_blocking_io = true;
+
+  fail_fs->FailOptionsDirFsyncAfterNextOptionsFile();
+  ASSERT_NOK(DB::Open(opt, dbname_, &db));
+  ASSERT_EQ(fail_fs->options_dir_fsync_failures(), 1);
+  ASSERT_LE(CountOptionsFiles(opt.env, dbname_), 2);
+}
+
 TEST_F(OptionsFileTest, OptionsFileName) {
   const uint64_t kOptionsFileNum = 12345;
   uint64_t number;

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
 };
 

include/rocksdb/version.h

@@ -13,7 +13,7 @@
 // minor or major version number planned for release.
 #define ROCKSDB_MAJOR 11
 #define ROCKSDB_MINOR 8
-#define ROCKSDB_PATCH 0
+#define ROCKSDB_PATCH 1
 
 // Make it easy to do conditional compilation based on version checks, i.e.
 // #if ROCKSDB_VERSION_GE(4, 5, 6)

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/behavior_changes/multiget_value_size_soft_limit_blob_direct_write.md

@@ -1 +0,0 @@
-`ReadOptions::value_size_soft_limit` now also bounds the blob values resolved from pre-flush blob direct write references during `MultiGet`/`MultiGetEntity` (previously these were resolved after the limit check and read without bound). Enforcement is "always make progress": at least one key is read even if its value alone exceeds the limit, and only once the returned size exceeds the limit do subsequent keys get `Status::Aborted` -- so a caller retrying the aborted keys cannot loop forever on a single value that by itself exceeds the limit.

unreleased_history/performance_improvements/pinnable_wide_columns_zero_copy.md

@@ -1 +0,0 @@
-Reduced copying when reading wide-column entities that have blob-referenced columns, whether stored in separate blob files or embedded (same-file) as blob records in the same SST. On the point-lookup (`GetEntity`/`MultiGetEntity`) and blob-backed memtable/direct-write read paths, resolving these columns no longer re-serializes the whole entity into a fresh buffer: inline columns are referenced in place in the pinned entity and each resolved blob value is referenced in place from its fetched buffer. This is enabled by an internal change to `PinnableWideColumns` that lets its columns reference multiple backing buffers.

unreleased_history/public_api_changes/coroutine_read_executor.md

@@ -1 +0,0 @@
-Add callback-based asynchronous read APIs, `DB::GetAsync()` and `DB::MultiGetAsync()`. The idea is that when IO is required, RocksDB can suspend its internal coroutine read path, allowing the read executor thread to do other work. When the IO is complete, RocksDB invokes the user callback. This requires filesystem support for full performance benefits. A new filesystem API `FSRandomAccessFile::SubmitReadAsync()` is introduced for this. Unlike `ReadAsync`, the filesystem is responsible for eventually calling the callback. `FileSystem::GetReadExecutor()` returns the IO executor whose EventBases run coroutine read processing. In the Posix filesystem, an IO-uring based event loop is used to complete the IO.

unreleased_history/public_api_changes/flush_listener_wait.md

@@ -1 +0,0 @@
-Added `FlushOptions::listener_wait` (default `false`). When set together with `FlushOptions::wait == true`, `DB::Flush()` will not return until the registered `EventListener::OnFlushCompleted` callbacks for the flushed memtables have finished running. By default (`false`), `Flush(wait=true)` may return as soon as the flush result is committed, which can be before (or while) the `OnFlushCompleted` callbacks execute on the background flush thread. Also added the corresponding C API `rocksdb_flushoptions_set_listener_wait()` / `rocksdb_flushoptions_get_listener_wait()`.

unreleased_history/public_api_changes/pinnable_wide_columns_payload_size.md

@@ -1 +0,0 @@
-Deprecated `PinnableWideColumns::serialized_size()` in favor of the new `PinnableWideColumns::payload_size()`, which returns the total size of the columns' names and values (for a plain value this equals the value size). Relatedly, wide-column read accounting changed subtly: read statistics for entities (`BYTES_READ`, `BYTES_PER_READ`, and the per-key byte counts for `GetEntity`/`MultiGetEntity`) and the `ReadOptions::value_size_soft_limit` threshold now measure entities by this payload size instead of the serialized entity size. This drops serialization framing bytes from those measurements (so reported/limited sizes for entity reads are slightly smaller than before) and makes plain-value and wide-column accounting consistent.