diff --git a/c_api_gen/c_generated_option_structs_auto.cc.inc b/c_api_gen/c_generated_option_structs_auto.cc.inc index eab2ece854..26f722647c 100644 --- a/c_api_gen/c_generated_option_structs_auto.cc.inc +++ b/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; diff --git a/c_api_gen/c_generated_option_structs_auto.h.inc b/c_api_gen/c_generated_option_structs_auto.h.inc index fef3d31baf..74e71eed69 100644 --- a/c_api_gen/c_generated_option_structs_auto.h.inc +++ b/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); diff --git a/c_api_gen/c_generated_roundtrip_tests.c.inc b/c_api_gen/c_generated_roundtrip_tests.c.inc index e3ffa048e8..b1c9318061 100644 --- a/c_api_gen/c_generated_roundtrip_tests.c.inc +++ b/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); diff --git a/db/c.cc b/db/c.cc index 2daf8a660c..bff4deba77 100644 --- a/db/c.cc +++ b/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; diff --git a/db/db_impl/db_impl.cc b/db/db_impl/db_impl.cc index 37c2b87a7e..ca93353c65 100644 --- a/db/db_impl/db_impl.cc +++ b/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); diff --git a/db/db_impl/db_impl_open.cc b/db/db_impl/db_impl_open.cc index 1450635597..cc385bfffd 100644 --- a/db/db_impl/db_impl_open.cc +++ b/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. "); diff --git a/db/db_impl/db_impl_sync_and_async.h b/db/db_impl/db_impl_sync_and_async.h index 65bbf80bcd..ecdf775c45 100644 --- a/db/db_impl/db_impl_sync_and_async.h +++ b/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); diff --git a/db_stress_tool/db_stress_env_wrapper.h b/db_stress_tool/db_stress_env_wrapper.h index e31a218c5f..2273fecf15 100644 --- a/db_stress_tool/db_stress_env_wrapper.h +++ b/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 cb, IODebugContext* dbg) override { CheckIOActivity(options); - target()->SubmitReadAsync(req, options, std::move(cb), dbg); + return target()->SubmitReadAsync(req, options, std::move(cb), dbg); } }; diff --git a/db_stress_tool/db_stress_test_base.cc b/db_stress_tool/db_stress_test_base.cc index 3e66ffce28..b2ffc0f54a 100644 --- a/db_stress_tool/db_stress_test_base.cc +++ b/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(); diff --git a/env/file_system_tracer.cc b/env/file_system_tracer.cc index 09b7dac003..82a3c59b56 100644 --- a/env/file_system_tracer.cc +++ b/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 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; diff --git a/env/file_system_tracer.h b/env/file_system_tracer.h index 2a93ea2c7c..588906205b 100644 --- a/env/file_system_tracer.h +++ b/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 cb, IODebugContext* dbg) override; diff --git a/env/fs_posix.cc b/env/fs_posix.cc index 268fcd3b2c..e36cb2c854 100644 --- a/env/fs_posix.cc +++ b/env/fs_posix.cc @@ -34,6 +34,7 @@ #include #include +#include #include // Get nano time includes #if defined(OS_LINUX) || defined(OS_FREEBSD) @@ -46,7 +47,6 @@ #endif #include #include -#include #include #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(GetReadIOUringOptions()); - folly::EventBase::Options event_base_options; - event_base_options.setBackendFactory([]() { - return std::make_unique(GetReadIOUringOptions()); - }); - event_base_manager_ = - std::make_unique(event_base_options); - executor_ = std::make_unique( - std::max(1u, std::thread::hardware_concurrency()), - /*minThreads=*/0, - std::make_shared("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 event_base_manager_; - std::unique_ptr 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(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 { + try { + auto backend = std::make_unique( + GetReadIOUringOptions()); + SetCurrentThreadReadIOUringBackendAvailable(); + return backend; + } catch (const std::exception&) { + return folly::EventBase::getDefaultBackend(); + } + }); + auto event_base_manager = + std::make_unique(event_base_options); + auto new_executor = std::make_unique( + requested_threads, 0, + std::make_shared("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 event_base_manager_{nullptr}; + Atomic 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(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 diff --git a/env/io_posix.cc b/env/io_posix.cc index cd83356a59..9c7e675595 100644 --- a/env/io_posix.cc +++ b/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 cb, IODebugContext* /*dbg*/) { +bool PosixRandomAccessFile::SubmitReadAsync( + FSReadRequest& req, const IOOptions& opts, + std::function 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 diff --git a/env/io_posix.h b/env/io_posix.h index 7f2f0f85a0..15502f53c4 100644 --- a/env/io_posix.h +++ b/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 cb, IODebugContext* dbg) override; #endif // USE_COROUTINES && FOLLY_HAS_LIBURING diff --git a/env/io_posix_test.cc b/env/io_posix_test.cc index 23d08b5039..12403151f4 100644 --- a/env/io_posix_test.cc +++ b/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(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(original_threads)); - }); + const size_t original_threads = thread_pool->numThreads(); - fs->SetReadIOExecutorThreads(1); - ASSERT_EQ(thread_pool->numThreads(), static_cast(1)); - fs->SetReadIOExecutorThreads(2); - ASSERT_EQ(thread_pool->numThreads(), static_cast(2)); + fs->SetReadIOExecutorThreads(static_cast(original_threads + 1)); + EXPECT_EQ(thread_pool->numThreads(), original_threads + 1); + + fs->SetReadIOExecutorThreads(static_cast(original_threads)); + EXPECT_EQ(thread_pool->numThreads(), original_threads + 1); } #endif // USE_COROUTINES && FOLLY_HAS_LIBURING diff --git a/file/random_access_file_reader_sync_and_async.h b/file/random_access_file_reader_sync_and_async.h index 0db0f2da73..0a7088b81a 100644 --- a/file/random_access_file_reader_sync_and_async.h +++ b/file/random_access_file_reader_sync_and_async.h @@ -20,11 +20,9 @@ namespace ROCKSDB_NAMESPACE { #if defined(WITH_COROUTINES) -inline folly::coro::Task SubmitMultiReadAsync(FSRandomAccessFile* file, - FSReadRequest* fs_reqs, - size_t num_fs_reqs, - const IOOptions& opts, - IODebugContext* dbg) { +inline folly::coro::Task 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 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(static_cast(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 diff --git a/file/random_access_file_reader_test.cc b/file/random_access_file_reader_test.cc index 63ff867e33..9441e4d09a 100644 --- a/file/random_access_file_reader_test.cc +++ b/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 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) { diff --git a/include/rocksdb/c.h b/include/rocksdb/c.h index c7a29b4aa0..f1e42b283f 100644 --- a/include/rocksdb/c.h +++ b/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); diff --git a/include/rocksdb/db.h b/include/rocksdb/db.h index 982e6b2f10..9f68f17297 100644 --- a/include/rocksdb/db.h +++ b/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 diff --git a/include/rocksdb/file_system.h b/include/rocksdb/file_system.h index 7e3e1e81df..f7222cf53c 100644 --- a/include/rocksdb/file_system.h +++ b/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 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 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(); diff --git a/include/rocksdb/options.h b/include/rocksdb/options.h index a62806ef43..cb7c4709ba 100644 --- a/include/rocksdb/options.h +++ b/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 diff --git a/include/rocksdb/statistics.h b/include/rocksdb/statistics.h index a71db29b15..90954b865e 100644 --- a/include/rocksdb/statistics.h +++ b/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 }; diff --git a/java/rocksjni/portal.h b/java/rocksjni/portal.h index a1b0941250..42894c45d6 100644 --- a/java/rocksjni/portal.h +++ b/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 diff --git a/java/src/main/java/org/rocksdb/TickerType.java b/java/src/main/java/org/rocksdb/TickerType.java index 2a040bf20e..b7e5af1b86 100644 --- a/java/src/main/java/org/rocksdb/TickerType.java +++ b/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; diff --git a/monitoring/statistics.cc b/monitoring/statistics.cc index 5b11358892..f1059da725 100644 --- a/monitoring/statistics.cc +++ b/monitoring/statistics.cc @@ -336,6 +336,8 @@ const std::vector> 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> HistogramsNameMap = { diff --git a/options/db_options.cc b/options/db_options.cc index 13f7b8c137..cba8cec38b 100644 --- a/options/db_options.cc +++ b/options/db_options.cc @@ -304,6 +304,10 @@ static std::unordered_map {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) { diff --git a/options/db_options.h b/options/db_options.h index 56e3145fe3..9b25c5671f 100644 --- a/options/db_options.h +++ b/options/db_options.h @@ -36,6 +36,7 @@ struct ImmutableDBOptions { std::shared_ptr info_log; InfoLogLevel info_log_level; int max_file_opening_threads; + int read_io_executor_threads; std::shared_ptr statistics; bool use_fsync; std::vector db_paths; diff --git a/options/options_helper.cc b/options/options_helper.cc index c107622033..f861fba3f3 100644 --- a/options/options_helper.cc +++ b/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; diff --git a/options/options_settable_test.cc b/options/options_settable_test.cc index 7b03e2aadb..4ccc790209 100644 --- a/options/options_settable_test.cc +++ b/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;" diff --git a/unreleased_history/public_api_changes/read_io_executor_threads.md b/unreleased_history/public_api_changes/read_io_executor_threads.md new file mode 100644 index 0000000000..64ac4cf10f --- /dev/null +++ b/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.