From f25fb41da64fa180718386848eb253dfbb016d05 Mon Sep 17 00:00:00 2001 From: Josh Kang Date: Mon, 2 Mar 2026 16:18:14 -0800 Subject: [PATCH] Add option to validate sst files in the background on DB open (#14322) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Summary: Add `open_files_async` option for faster DB startup. When enabled, SST file opening and validation is deferred to a background thread after `DB::Open` returns, reducing startup latency for databases with many SST files. WAL recovery remains synchronous. To support this, `FindTable` is extended with a pinning mechanism that stores the cache handle directly on `FileMetaData` via a new `PinnedTableReader` class, and sets the table reader atomically so subsequent reads skip cache lookups. `FileDescriptor::table_reader` is replaced with `PinnedTableReader pinned_reader` which wraps a `std::atomic` with acquire/release ordering to safely handle concurrent access between the background opener and read threads. Should validations fail, the background opener sets a `kAsyncFileOpen` background error. Future read requests will look up the table reader again via the cache, and if any validations fail there it will get propagated to the user (existing behavior when `max_open_files > 0`). This feature is most useful when `max_open_files=-1`, because otherwise file opening is already capped at 16 files and DB open should be fast. ## Restrictions - This feature also is incompatible with fifo compaction because fifo compaction requires reading table properties under DB mutex. When table reader is unpinned, this may cause a DB hang. - This feature is also incompatible with `skip_stats_update_on_db_open=false` because it will result in even longer DB open ## Key changes - New `open_files_async` DB option with C, Java, and `db_bench` bindings - `BGWorkAsyncFileOpen` background worker that opens all SST files post-`DB::Open`, with shutdown awareness via `shutting_down_` flag - New `PinnedTableReader` class in `version_edit.h` — thread-safe wrapper holding `std::atomic` and `Cache::Handle*` with proper acquire/release ordering. Replaces the old `FileDescriptor::table_reader` raw pointer and `FileMetaData::table_reader_handle` - Extract `LoadTableHandlersHelper` into `db/version_util.cc` — shared between `VersionBuilder::LoadTableHandlers` (for version edits during recovery) and `BGWorkAsyncFileOpen` (for base storage post-open) - `FindTable` extended with `pin_table_handle` and `out_table_reader` params — when pinning is enabled, the table reader is stored on `FileMetaData` so Get/MultiGet/Iterator skip redundant cache lookups. `FindTable` now performs the pinned-reader fast-path check internally instead of requiring callers to check `fd.table_reader` beforehand - Note: pinning is explicit (not default) because some callers create temporary `FileMetaData`s that would need to properly clean up table handles - `CompactedDBImpl` updated to use `FindTable` + pinning instead of raw `fd.table_reader` access for Get/MultiGet - New `kAsyncFileOpen` background error reason in `listener.h` and `error_handler.cc` - Add a check in ~DBImpl to ensure async file open task has not been forgotten to be scheduled in (future) subclasses of DBImpl. Certain subclasses that never use it will need to explicitly mark it. Pull Request resolved: https://github.com/facebook/rocksdb/pull/14322 Test Plan: - `OpenFilesAsyncTest` parameterized over `num_flushes` (1, 20), `ReadType` (Get, MultiGet, Iterator), `max_open_files` (-1, 10), and `read_only` (true, false) - **ConcurrentFileAccess**: concurrent reads and compactions race with async opener - **AfterRead**: reads happen before async opener, verifying lazy open and that the opener sees already-pinned readers - **BeforeRead**: async opener completes first, verifying reads use pre-loaded table readers - **Shutdown**: DB closes before async opener starts, verifying clean cancellation with 0 file opens - **Error**: corrupted SST files, verifying `kAsyncFileOpen` background error is set and reads return corruption - **DropColumnFamily**: CF dropped before async opener runs, verifying the opener gracefully skips dropped CFs - Added to crash test ### Benchmark To simulate a high-latency remote filesystem, I set up a virtual filesystem with dm-delay using 10ms reads, 0 ms writes. ``` # Generate a DB with many L0 files TEST_TMPDIR=/data/users/jkangs/dm-delay-test/mnt ./db_bench -benchmarks=fillseq -disable_auto_compactions=true -write_buffer_size=1000 -num=1000000 ``` ``` ./db_bench -use_existing_db=true -db=/data/users/jkangs/dm-delay-test/mnt/dbbench -benchmarks=readrandom -reads=1 -report_open_timing=true -open_files_async=true -use_direct_reads -file_opening_threads=1 -skip_stats_update_on_db_open OpenDb: 25.1419 milliseconds ``` ``` ./db_bench -use_existing_db=true -db=/data/users/jkangs/dm-delay-test/mnt/dbbench -benchmarks=readrandom -reads=1 -report_open_timing=true -open_files_async=false -use_direct_reads -file_opening_threads=1 -skip_stats_update_on_db_open OpenDb: 23109.4 milliseconds ``` ### No read regressions On main branch ``` ./db_bench -use_existing_db=true -db=/dev/shm/dbbench -benchmarks=readrandom -seed=1 -threads=8 -duration=30 readrandom : 4.827 micros/op 1657100 ops/sec 30.005 seconds 49720992 operations; 183.3 MB/s (6198999 of 6198999 found) ``` On this branch ``` ./db_bench -use_existing_db=true -db=/dev/shm/dbbench -benchmarks=readrandom -seed=1 -threads=8 -duration=30 readrandom : 4.863 micros/op 1644808 ops/sec 30.007 seconds 49354992 operations; 182.0 MB/s (6099999 of 6099999 found) ./db_bench -use_existing_db=true -db=/dev/shm/dbbench -benchmarks=readrandom -seed=1 -threads=8 -duration=30 -open_files_async=true readrandom : 4.803 micros/op 1665392 ops/sec 30.004 seconds 49968992 operations; 184.2 MB/s (6222999 of 6222999 found) ``` Reviewed By: pdillinger, xingbowang Differential Revision: D93538033 Pulled By: joshkang97 fbshipit-source-id: 32ac70c112cd733b7c1e1c1e2e7ce6422318a5ae --- BUCK | 1 + CMakeLists.txt | 1 + db/c.cc | 9 + db/column_family.cc | 22 + db/compaction/compaction_picker_fifo.cc | 11 +- db/compaction/compaction_picker_test.cc | 8 +- db/db_compaction_test.cc | 28 +- db/db_impl/compacted_db_impl.cc | 85 ++-- db/db_impl/compacted_db_impl.h | 1 + db/db_impl/db_impl.cc | 13 + db/db_impl/db_impl.h | 27 +- db/db_impl/db_impl_files.cc | 2 +- db/db_impl/db_impl_follower.cc | 1 + db/db_impl/db_impl_open.cc | 126 ++++++ db/db_impl/db_impl_readonly.cc | 6 + db/db_impl/db_impl_secondary.cc | 1 + db/db_sst_test.cc | 2 +- db/db_test.cc | 413 ++++++++++++++++++ db/db_test_util.cc | 22 + db/db_test_util.h | 6 + db/db_wal_test.cc | 4 +- db/error_handler.cc | 24 + db/table_cache.cc | 275 ++++++------ db/table_cache.h | 36 +- db/table_cache_sync_and_async.h | 14 +- db/version_builder.cc | 61 +-- db/version_builder.h | 3 +- db/version_edit.cc | 34 ++ db/version_edit.h | 75 +++- db/version_edit_handler.h | 14 +- db/version_set.cc | 26 +- db/version_util.cc | 96 ++++ db/version_util.h | 24 + db_stress_tool/db_stress_common.h | 1 + db_stress_tool/db_stress_gflags.cc | 3 + db_stress_tool/db_stress_test_base.cc | 7 + include/rocksdb/c.h | 4 + include/rocksdb/listener.h | 1 + include/rocksdb/options.h | 19 + java/rocksjni/options.cc | 45 ++ java/src/main/java/org/rocksdb/DBOptions.java | 15 + .../java/org/rocksdb/DBOptionsInterface.java | 20 + java/src/main/java/org/rocksdb/Options.java | 15 + options/db_options.cc | 7 + options/db_options.h | 1 + options/options_helper.cc | 1 + options/options_settable_test.cc | 1 + src.mk | 1 + tools/db_bench_tool.cc | 14 + tools/db_crashtest.py | 8 + .../new_features/async_file_open | 1 + 51 files changed, 1346 insertions(+), 289 deletions(-) create mode 100644 db/version_util.cc create mode 100644 unreleased_history/new_features/async_file_open diff --git a/BUCK b/BUCK index ee05bbc57a..60c067dd27 100644 --- a/BUCK +++ b/BUCK @@ -106,6 +106,7 @@ cpp_library_wrapper(name="rocksdb_lib", srcs=[ "db/version_edit.cc", "db/version_edit_handler.cc", "db/version_set.cc", + "db/version_util.cc", "db/wal_edit.cc", "db/wal_manager.cc", "db/wide/wide_column_serialization.cc", diff --git a/CMakeLists.txt b/CMakeLists.txt index f601dd2714..aa8a38f4d6 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -766,6 +766,7 @@ set(SOURCES db/version_edit.cc db/version_edit_handler.cc db/version_set.cc + db/version_util.cc db/wal_edit.cc db/wal_manager.cc db/wide/wide_column_serialization.cc diff --git a/db/c.cc b/db/c.cc index 6b78342e86..4547115710 100644 --- a/db/c.cc +++ b/db/c.cc @@ -4351,6 +4351,15 @@ unsigned char rocksdb_options_get_paranoid_checks(rocksdb_options_t* opt) { return opt->rep.paranoid_checks; } +void rocksdb_options_set_open_files_async(rocksdb_options_t* opt, + unsigned char v) { + opt->rep.open_files_async = v; +} + +unsigned char rocksdb_options_get_open_files_async(rocksdb_options_t* opt) { + return opt->rep.open_files_async; +} + void rocksdb_options_set_db_paths(rocksdb_options_t* opt, const rocksdb_dbpath_t** dbpath_values, size_t num_paths) { diff --git a/db/column_family.cc b/db/column_family.cc index bbf9f8210b..766b826985 100644 --- a/db/column_family.cc +++ b/db/column_family.cc @@ -1569,6 +1569,28 @@ Status ColumnFamilyData::ValidateOptions( "FIFO compaction only supported with max_open_files = -1."); } + if (db_options.open_files_async) { + // FIFO TTL picker relies on reading the files table properties inside a + // DB mutex. This can be slow if files are opened asynchronously, so we + // disable it. + // + // TODO: consider blocking fifo compaction until async file open task + // completes. + if (cf_options.compaction_style == kCompactionStyleFIFO) { + return Status::NotSupported( + "FIFO compaction is not supported with open_files_async = true."); + } + // Open files async is not useful if skip_stats_update_on_db_open=true + // because DB open will still block on IO. + // + // TODO: consider moving stats update inside async file open background + // task. + if (!db_options.skip_stats_update_on_db_open) { + return Status::InvalidArgument( + "open_files_async requires skip_stats_update_on_db_open = true."); + } + } + if (cf_options.compaction_options_fifo.use_kv_ratio_compaction) { if (cf_options.compaction_style != kCompactionStyleFIFO) { return Status::InvalidArgument( diff --git a/db/compaction/compaction_picker_fifo.cc b/db/compaction/compaction_picker_fifo.cc index e13c333856..9a680bb1eb 100644 --- a/db/compaction/compaction_picker_fifo.cc +++ b/db/compaction/compaction_picker_fifo.cc @@ -102,10 +102,10 @@ Compaction* FIFOCompactionPicker::PickTTLCompaction( for (auto ritr = level_files.rbegin(); ritr != level_files.rend(); ++ritr) { FileMetaData* f = *ritr; assert(f); - if (f->fd.table_reader && f->fd.table_reader->GetTableProperties()) { + TableReader* reader = f->fd.pinned_reader.Get(); + if (reader != nullptr && reader->GetTableProperties() != nullptr) { uint64_t newest_key_time = f->TryGetNewestKeyTime(); - uint64_t creation_time = - f->fd.table_reader->GetTableProperties()->creation_time; + uint64_t creation_time = reader->GetTableProperties()->creation_time; uint64_t est_newest_key_time = newest_key_time == kUnknownNewestKeyTime ? creation_time : newest_key_time; @@ -166,8 +166,9 @@ Compaction* FIFOCompactionPicker::PickTTLCompaction( assert(f); uint64_t newest_key_time = f->TryGetNewestKeyTime(); uint64_t creation_time = 0; - if (f->fd.table_reader && f->fd.table_reader->GetTableProperties()) { - creation_time = f->fd.table_reader->GetTableProperties()->creation_time; + TableReader* reader = f->fd.pinned_reader.Get(); + if (reader != nullptr && reader->GetTableProperties() != nullptr) { + creation_time = reader->GetTableProperties()->creation_time; } uint64_t est_newest_key_time = newest_key_time == kUnknownNewestKeyTime ? creation_time diff --git a/db/compaction/compaction_picker_test.cc b/db/compaction/compaction_picker_test.cc index 4dfa327ae1..5539b5e8dc 100644 --- a/db/compaction/compaction_picker_test.cc +++ b/db/compaction/compaction_picker_test.cc @@ -178,7 +178,8 @@ class CompactionPickerTestBase : public testing::Test { } TableProperties tp; tp.newest_key_time = newest_key_time; - f->fd.table_reader = new mock::MockTableReader(mock::KVVector{}, tp); + f->fd.pinned_reader.TEST_SetReader( + new mock::MockTableReader(mock::KVVector{}, tp)); vstorage->AddFile(level, f); files_.emplace_back(f); @@ -272,8 +273,9 @@ class CompactionPickerTestBase : public testing::Test { void ClearFiles() { for (auto& file : files_) { - if (file->fd.table_reader != nullptr) { - delete file->fd.table_reader; + TableReader* t = file->fd.pinned_reader.Get(); + if (t != nullptr) { + delete t; } } files_.clear(); diff --git a/db/db_compaction_test.cc b/db/db_compaction_test.cc index a97d346150..6bf200adca 100644 --- a/db/db_compaction_test.cc +++ b/db/db_compaction_test.cc @@ -11488,20 +11488,24 @@ TEST_F(DBCompactionTest, RecordNewestKeyTimeForTtlCompaction) { // Check that we are populating newest_key_time on flush std::vector file_metadatas = GetLevelFileMetadatas(0); ASSERT_EQ(file_metadatas.size(), 4); - uint64_t first_newest_key_time = - file_metadatas[0]->fd.table_reader->GetTableProperties()->newest_key_time; + uint64_t first_newest_key_time = file_metadatas[0] + ->fd.pinned_reader.Get() + ->GetTableProperties() + ->newest_key_time; ASSERT_NE(first_newest_key_time, kUnknownNewestKeyTime); // Check that the newest_key_times are in expected ordering uint64_t prev_newest_key_time = first_newest_key_time; for (size_t idx = 1; idx < file_metadatas.size(); idx++) { uint64_t newest_key_time = file_metadatas[idx] - ->fd.table_reader->GetTableProperties() + ->fd.pinned_reader.Get() + ->GetTableProperties() ->newest_key_time; ASSERT_LT(newest_key_time, prev_newest_key_time); prev_newest_key_time = newest_key_time; ASSERT_EQ(newest_key_time, file_metadatas[idx] - ->fd.table_reader->GetTableProperties() + ->fd.pinned_reader.Get() + ->GetTableProperties() ->creation_time); } // The delta between the first and last newest_key_times is 15s @@ -11516,14 +11520,18 @@ TEST_F(DBCompactionTest, RecordNewestKeyTimeForTtlCompaction) { ASSERT_EQ(NumTableFilesAtLevel(0), 1); file_metadatas = GetLevelFileMetadatas(0); ASSERT_EQ(file_metadatas.size(), 1); - ASSERT_EQ( - file_metadatas[0]->fd.table_reader->GetTableProperties()->newest_key_time, - first_newest_key_time); + ASSERT_EQ(file_metadatas[0] + ->fd.pinned_reader.Get() + ->GetTableProperties() + ->newest_key_time, + first_newest_key_time); // Contrast newest_key_time with creation_time, which records the oldest // ancestor time (15s older than newest_key_time) - ASSERT_EQ( - file_metadatas[0]->fd.table_reader->GetTableProperties()->creation_time, - last_newest_key_time); + ASSERT_EQ(file_metadatas[0] + ->fd.pinned_reader.Get() + ->GetTableProperties() + ->creation_time, + last_newest_key_time); ASSERT_EQ(file_metadatas[0]->oldest_ancester_time, last_newest_key_time); // Make sure TTL of 5s causes compaction diff --git a/db/db_impl/compacted_db_impl.cc b/db/db_impl/compacted_db_impl.cc index 611695be66..0a10ce5a95 100644 --- a/db/db_impl/compacted_db_impl.cc +++ b/db/db_impl/compacted_db_impl.cc @@ -91,8 +91,21 @@ Status CompactedDBImpl::Get(const ReadOptions& _read_options, /*b_has_ts=*/false) < 0) { return Status::NotFound(); } - Status s = f.fd.table_reader->Get(read_options, lkey.internal_key(), - &get_context, nullptr); + TableReader* t = nullptr; + TableCache::TypedHandle* handle = nullptr; + Status s = cfd_->table_cache()->FindTable( + read_options, cfd_->table_cache()->file_options(), + cfd_->internal_comparator(), *f.file_metadata, &handle, + version_->GetMutableCFOptions(), &t, false /* no_io */, + nullptr /* file_read_hist */, false /* skip_filters */, files_level_, + true /* prefetch_index_and_filter_in_cache */, + 0 /* max_file_size_for_l0_meta_pin */, f.file_metadata->temperature, + true /* pin_table_handle */); + if (s.ok()) { + assert(handle == nullptr); + // Use TableReader directly to avoid extra table_cache->Get() overheads + s = t->Get(read_options, lkey.internal_key(), &get_context, nullptr); + } if (!s.ok() && !s.IsNotFound()) { return s; } @@ -152,7 +165,6 @@ void CompactedDBImpl::MultiGet(const ReadOptions& _read_options, } GetWithTimestampReadCallback read_cb(kMaxSequenceNumber); - autovector reader_list; for (size_t i = 0; i < num_keys; ++i) { const Slice& key = keys[i]; LookupKey lkey(key, kMaxSequenceNumber, read_options.timestamp); @@ -162,38 +174,40 @@ void CompactedDBImpl::MultiGet(const ReadOptions& _read_options, ExtractUserKeyAndStripTimestamp(f.smallest_key, user_comparator_->timestamp_size()), /*b_has_ts=*/false) < 0) { - reader_list.push_back(nullptr); - } else { - f.fd.table_reader->Prepare(lkey.internal_key()); - reader_list.push_back(f.fd.table_reader); + statuses[i] = Status::NotFound(); + continue; } - } - for (size_t i = 0; i < num_keys; ++i) { - statuses[i] = Status::NotFound(); - } - int idx = 0; - for (auto* r : reader_list) { - if (r != nullptr) { - PinnableSlice& pinnable_val = values[idx]; - LookupKey lkey(keys[idx], kMaxSequenceNumber, read_options.timestamp); - std::string* timestamp = timestamps ? ×tamps[idx] : nullptr; - GetContext get_context( - user_comparator_, nullptr, nullptr, nullptr, GetContext::kNotFound, - lkey.user_key(), &pinnable_val, /*columns=*/nullptr, - user_comparator_->timestamp_size() > 0 ? timestamp : nullptr, nullptr, - nullptr, true, nullptr, nullptr, nullptr, nullptr, &read_cb); - Status status = - r->Get(read_options, lkey.internal_key(), &get_context, nullptr); - assert(static_cast(idx) < num_keys); - if (!status.ok() && !status.IsNotFound()) { - statuses[idx] = status; + PinnableSlice& pinnable_val = values[i]; + std::string* timestamp = timestamps ? ×tamps[i] : nullptr; + GetContext get_context( + user_comparator_, nullptr, nullptr, nullptr, GetContext::kNotFound, + lkey.user_key(), &pinnable_val, /*columns=*/nullptr, + user_comparator_->timestamp_size() > 0 ? timestamp : nullptr, nullptr, + nullptr, true, nullptr, nullptr, nullptr, nullptr, &read_cb); + TableReader* t = nullptr; + TableCache::TypedHandle* handle = nullptr; + Status status = cfd_->table_cache()->FindTable( + read_options, cfd_->table_cache()->file_options(), + cfd_->internal_comparator(), *f.file_metadata, &handle, + version_->GetMutableCFOptions(), &t, false /* no_io */, + nullptr /* file_read_hist */, false /* skip_filters */, files_level_, + true /* prefetch_index_and_filter_in_cache */, + 0 /* max_file_size_for_l0_meta_pin */, f.file_metadata->temperature, + true /* pin_table_handle */); + if (status.ok()) { + assert(handle == nullptr); + // Use TableReader directly to avoid extra table_cache->Get() overheads + status = t->Get(read_options, lkey.internal_key(), &get_context, nullptr); + } + if (!status.ok() && !status.IsNotFound()) { + statuses[i] = status; + } else { + if (get_context.State() == GetContext::kFound) { + statuses[i] = Status::OK(); } else { - if (get_context.State() == GetContext::kFound) { - statuses[idx] = Status::OK(); - } + statuses[i] = Status::NotFound(); } } - ++idx; } } @@ -230,6 +244,7 @@ Status CompactedDBImpl::Init(const Options& options) { return Status::NotSupported("Both L0 and other level contain files"); } files_ = l0; + files_level_ = 0; return Status::OK(); } @@ -242,6 +257,7 @@ Status CompactedDBImpl::Init(const Options& options) { int level = vstorage->num_non_empty_levels() - 1; if (vstorage->LevelFilesBrief(level).num_files > 0) { files_ = vstorage->LevelFilesBrief(level); + files_level_ = level; return Status::OK(); } return Status::NotSupported("no file exists"); @@ -261,6 +277,13 @@ Status CompactedDBImpl::Open(const Options& options, const std::string& dbname, std::unique_ptr db(new CompactedDBImpl(db_options, dbname)); Status s = db->Init(options); if (s.ok()) { + { + InstrumentedMutexLock l(&db->mutex_); + db->opened_successfully_ = true; + if (db->immutable_db_options_.open_files_async) { + db->ScheduleAsyncFileOpening(); + } + } s = db->StartPeriodicTaskScheduler(); } if (s.ok()) { diff --git a/db/db_impl/compacted_db_impl.h b/db/db_impl/compacted_db_impl.h index 098016f5aa..432f8c21f0 100644 --- a/db/db_impl/compacted_db_impl.h +++ b/db/db_impl/compacted_db_impl.h @@ -143,5 +143,6 @@ class CompactedDBImpl : public DBImpl { Version* version_; const Comparator* user_comparator_; LevelFilesBrief files_; + int files_level_{0}; }; } // namespace ROCKSDB_NAMESPACE diff --git a/db/db_impl/db_impl.cc b/db/db_impl/db_impl.cc index a144f11c53..afecfe619b 100644 --- a/db/db_impl/db_impl.cc +++ b/db/db_impl/db_impl.cc @@ -561,11 +561,17 @@ Status DBImpl::CloseHelper() { // Wait for background work to finish while (bg_bottom_compaction_scheduled_ || bg_compaction_scheduled_ || bg_flush_scheduled_ || bg_purge_scheduled_ || + bg_async_file_open_state_ == AsyncFileOpenState::kScheduled || pending_purge_obsolete_files_ || error_handler_.IsRecoveryInProgress()) { TEST_SYNC_POINT("DBImpl::~DBImpl:WaitJob"); bg_cv_.Wait(); } + + // Ensure subclasses don't forget to schedule async file opening + assert(!immutable_db_options_.open_files_async || !opened_successfully_ || + bg_async_file_open_state_ != AsyncFileOpenState::kNotScheduled); + TEST_SYNC_POINT_CALLBACK("DBImpl::CloseHelper:PendingPurgeFinished", &files_grabbed_for_purge_); EraseThreadStatusDbInfo(); @@ -1541,6 +1547,13 @@ Status DBImpl::SetDBOptions( table_cache_.get()->SetCapacity(new_options.max_open_files == -1 ? TableCache::kInfiniteCapacity : new_options.max_open_files - 10); + // Potential table cache capacity change requires updating if table + // handles should get pinned. + for (auto cfd : *versions_->GetColumnFamilySet()) { + if (!cfd->IsDropped()) { + cfd->table_cache()->UpdateShouldPinTableHandles(); + } + } wal_other_option_changed = mutable_db_options_.wal_bytes_per_sync != new_options.wal_bytes_per_sync; wal_size_option_changed = mutable_db_options_.max_total_wal_size != diff --git a/db/db_impl/db_impl.h b/db/db_impl/db_impl.h index ded8bfda17..520b89f1d8 100644 --- a/db/db_impl/db_impl.h +++ b/db/db_impl/db_impl.h @@ -1718,6 +1718,17 @@ class DBImpl : public DB { // recovery. Status LogAndApplyForRecovery(const RecoveryContext& recovery_ctx); + // Schedule background work to open and validate SST files asynchronously. + // Called when open_files_async is enabled. + void ScheduleAsyncFileOpening(); + + // Mark async file opening as not needed. Used by subclasses that load + // table files through a different mechanism (e.g., ReactiveVersionSet). + void MarkAsyncFileOpenNotNeeded(); + + // Background work function for async file opening. + static void BGWorkAsyncFileOpen(void* arg); + void InvokeWalFilterIfNeededOnColumnFamilyToWalNumberMap(); // Return true to proceed with current WAL record whose content is stored in @@ -1728,6 +1739,9 @@ class DBImpl : public DB { Status& status, bool& stop_replay, WriteBatch& batch); + // Indicate DB was opened successfully + bool opened_successfully_ = false; + private: friend class DB; friend class ErrorHandler; @@ -3067,6 +3081,16 @@ class DBImpl : public DB { // number of background obsolete file purge jobs, submitted to the HIGH pool int bg_purge_scheduled_ = 0; + enum class AsyncFileOpenState : uint8_t { + kNotScheduled = 0, // Async file opening has not been scheduled. + kScheduled, // Async file opening is in-flight in the HIGH pool. + kComplete, // Async file opening has finished (or was not needed). + }; + + // Tracks whether background async file opening has been scheduled/completed. + AsyncFileOpenState bg_async_file_open_state_ = + AsyncFileOpenState::kNotScheduled; + std::deque manual_compaction_dequeue_; // shall we disable deletion of obsolete files @@ -3122,9 +3146,6 @@ class DBImpl : public DB { // Guard against multiple concurrent refitting bool refitting_level_ = false; - // Indicate DB was opened successfully - bool opened_successfully_ = false; - // The min threshold to triggere bottommost compaction for removing // garbages, among all column families. SequenceNumber bottommost_files_mark_threshold_ = kMaxSequenceNumber; diff --git a/db/db_impl/db_impl_files.cc b/db/db_impl/db_impl_files.cc index d9d56a1f44..abf9178f9a 100644 --- a/db/db_impl/db_impl_files.cc +++ b/db/db_impl/db_impl_files.cc @@ -446,7 +446,7 @@ void DBImpl::PurgeObsoleteFiles(JobContext& state, bool schedule_only) { state.manifest_delete_files.size()); // We may ignore the dbname when generating the file names. for (auto& file : state.sst_delete_files) { - auto* handle = file.metadata->table_reader_handle; + auto* handle = file.metadata->fd.pinned_reader.GetCacheHandle(); if (file.only_delete_metadata) { if (handle) { // Simply release handle of file that is not being deleted diff --git a/db/db_impl/db_impl_follower.cc b/db/db_impl/db_impl_follower.cc index 1262c5bdfd..11ba6070e7 100644 --- a/db/db_impl/db_impl_follower.cc +++ b/db/db_impl/db_impl_follower.cc @@ -321,6 +321,7 @@ Status DB::OpenAsFollower( sv_context.NewSuperVersion(); cfd->InstallSuperVersion(&sv_context, &impl->mutex_); } + impl->MarkAsyncFileOpenNotNeeded(); } impl->mutex_.Unlock(); sv_context.Clean(); diff --git a/db/db_impl/db_impl_open.cc b/db/db_impl/db_impl_open.cc index 7b2e949789..48bfef0111 100644 --- a/db/db_impl/db_impl_open.cc +++ b/db/db_impl/db_impl_open.cc @@ -12,6 +12,7 @@ #include "db/db_impl/db_impl.h" #include "db/error_handler.h" #include "db/periodic_task_scheduler.h" +#include "db/version_util.h" #include "env/composite_env_wrapper.h" #include "file/filename.h" #include "file/read_write_util.h" @@ -2657,6 +2658,10 @@ Status DBImpl::Open(const DBOptions& db_options, const std::string& dbname, impl->DeleteObsoleteFiles(); TEST_SYNC_POINT("DBImpl::Open:AfterDeleteFiles"); impl->MaybeScheduleFlushOrCompaction(); + + if (impl->immutable_db_options_.open_files_async) { + impl->ScheduleAsyncFileOpening(); + } impl->mutex_.Unlock(); } @@ -2705,4 +2710,125 @@ Status DBImpl::Open(const DBOptions& db_options, const std::string& dbname, } return s; } + +struct AsyncFileOpenContext { + DBImpl* db = nullptr; + FileOptions file_options; + std::vector versions; + + AsyncFileOpenContext() = default; + AsyncFileOpenContext(const AsyncFileOpenContext&) = delete; + AsyncFileOpenContext& operator=(const AsyncFileOpenContext&) = delete; + AsyncFileOpenContext(AsyncFileOpenContext&&) = delete; + AsyncFileOpenContext& operator=(AsyncFileOpenContext&&) = delete; + + ~AsyncFileOpenContext() { + db->mutex()->AssertHeld(); + for (auto* v : versions) { + // must unref version before cfd + ColumnFamilyData* cfd = v->cfd(); + v->Unref(); + cfd->UnrefAndTryDelete(); + } + } +}; + +void DBImpl::ScheduleAsyncFileOpening() { + mutex_.AssertHeld(); + + auto* ctx = new AsyncFileOpenContext(); + ctx->db = this; + ctx->file_options = versions_->file_options(); + + for (auto cfd : *versions_->GetColumnFamilySet()) { + assert(!cfd->IsDropped()); + Version* current = cfd->current(); + VersionStorageInfo* vstorage = current->storage_info(); + bool has_files = false; + for (int level = 0; level < vstorage->num_levels() && !has_files; level++) { + has_files = !vstorage->LevelFiles(level).empty(); + } + if (has_files) { + cfd->Ref(); + current->Ref(); + ctx->versions.push_back(current); + } + } + + bg_async_file_open_state_ = AsyncFileOpenState::kScheduled; + + // since this is a one time job, best to schedule it with high priority + env_->Schedule(&DBImpl::BGWorkAsyncFileOpen, ctx, Env::Priority::HIGH, + nullptr); +} + +void DBImpl::MarkAsyncFileOpenNotNeeded() { + mutex_.AssertHeld(); + assert(bg_async_file_open_state_ == AsyncFileOpenState::kNotScheduled); + bg_async_file_open_state_ = AsyncFileOpenState::kComplete; +} + +void DBImpl::BGWorkAsyncFileOpen(void* arg) { + TEST_SYNC_POINT("DBImpl::BGWorkAsyncFileOpen::Start"); + + AsyncFileOpenContext* raw_ctx = static_cast(arg); + DBImpl* db = raw_ctx->db; + + auto deleter = [](AsyncFileOpenContext* p) { + auto* dbPtr = p->db; + InstrumentedMutexLock l(&dbPtr->mutex_); + delete p; + dbPtr->bg_async_file_open_state_ = AsyncFileOpenState::kComplete; + dbPtr->bg_cv_.SignalAll(); + }; + std::unique_ptr ctx(raw_ctx, + deleter); + + ReadOptions ro; + for (size_t i = 0; i < ctx->versions.size(); i++) { + auto* version = ctx->versions[i]; + ColumnFamilyData* cfd = version->cfd(); + + // Skip column families that were dropped after scheduling + if (cfd->IsDropped()) { + continue; + } + + VersionStorageInfo* vstorage = version->storage_info(); + + MutableCFOptions mutable_cf_options; + { + InstrumentedMutexLock l(&db->mutex_); + mutable_cf_options = cfd->GetLatestMutableCFOptions(); + } + size_t max_file_size_for_l0_meta_pin = + MaxFileSizeForL0MetaPin(mutable_cf_options); + + std::vector> files_meta; + for (int level = 0; level < vstorage->num_levels(); level++) { + for (FileMetaData* file_meta : vstorage->LevelFiles(level)) { + files_meta.emplace_back(file_meta, level); + } + } + + Status s = LoadTableHandlersHelper( + files_meta, cfd->table_cache(), ctx->file_options, + *vstorage->InternalComparator(), cfd->internal_stats(), + db->immutable_db_options_.max_file_opening_threads, + false /* prefetch_index_and_filter_in_cache */, mutable_cf_options, + max_file_size_for_l0_meta_pin, ro, &db->shutting_down_); + if (!s.ok()) { + ROCKS_LOG_ERROR( + db->immutable_db_options_.info_log, + "BGWorkAsyncFileOpen: LoadTableHandlers failed for CF %s: " + "%s", + cfd->GetName().c_str(), s.ToString().c_str()); + InstrumentedMutexLock l(&db->mutex_); + db->error_handler_.SetBGError(s, BackgroundErrorReason::kAsyncFileOpen); + break; + } + } + TEST_SYNC_POINT("DBImpl::BGWorkAsyncFileOpen:Done"); +} + } // namespace ROCKSDB_NAMESPACE diff --git a/db/db_impl/db_impl_readonly.cc b/db/db_impl/db_impl_readonly.cc index 31934ee192..6ff0a48ba9 100644 --- a/db/db_impl/db_impl_readonly.cc +++ b/db/db_impl/db_impl_readonly.cc @@ -357,6 +357,12 @@ Status DBImplReadOnly::OpenForReadOnlyWithoutCheck( sv_context.NewSuperVersion(); cfd->InstallSuperVersion(&sv_context, &impl->mutex_); } + + impl->opened_successfully_ = true; + + if (db_options.open_files_async) { + impl->ScheduleAsyncFileOpening(); + } } impl->mutex_.Unlock(); sv_context.Clean(); diff --git a/db/db_impl/db_impl_secondary.cc b/db/db_impl/db_impl_secondary.cc index 0db4820c39..2bc4072ef0 100644 --- a/db/db_impl/db_impl_secondary.cc +++ b/db/db_impl/db_impl_secondary.cc @@ -802,6 +802,7 @@ Status DB::OpenAsSecondary( sv_context.NewSuperVersion(); cfd->InstallSuperVersion(&sv_context, &impl->mutex_); } + impl->MarkAsyncFileOpenNotNeeded(); } impl->mutex_.Unlock(); sv_context.Clean(); diff --git a/db/db_sst_test.cc b/db/db_sst_test.cc index d186efd8c6..db3fa1cb66 100644 --- a/db/db_sst_test.cc +++ b/db/db_sst_test.cc @@ -1535,7 +1535,7 @@ TEST_F(DBSSTTest, OpenDBWithInfiniteMaxOpenFiles) { for (const auto& level : files) { for (const auto& file : level) { - ASSERT_TRUE(file.table_reader_handle != nullptr); + ASSERT_TRUE(file.fd.pinned_reader.Get() != nullptr); } } diff --git a/db/db_test.cc b/db/db_test.cc index 9c0dc9fe32..16c15aa0a8 100644 --- a/db/db_test.cc +++ b/db/db_test.cc @@ -7762,6 +7762,419 @@ TEST_F(DBTest, ShuttingDownNotBlockStalledWrites) { thd.join(); } +enum class ReadType : uint8_t { kGet, kMultiGet, kIterator, kNumReadTypes }; + +class OpenFilesAsyncTest + : public DBTest, + public testing::WithParamInterface> { + public: + OpenFilesAsyncTest() { + std::tie(num_flushes_, max_open_files_, read_only_) = GetParam(); + } + + std::string BeforeFindTableSyncPoint(ReadType rt) { + switch (rt) { + case ReadType::kGet: + return "TableCache::Get::BeforeFindTable"; + case ReadType::kMultiGet: + return "TableCache::MultiGet::BeforeFindTable"; + case ReadType::kIterator: + return "TableCache::NewIterator::BeforeFindTable"; + default: + assert(false); + return ""; + } + } + + std::string KeyFor(uint32_t f) { + char buf[32]; + snprintf(buf, sizeof(buf), "key%04u", f); + return buf; + } + + std::string ValueFor(uint32_t f) { + char buf[32]; + snprintf(buf, sizeof(buf), "value%04u", f); + return buf; + } + + void SetupData(Options& options, const std::vector& cfs = { + kDefaultColumnFamilyName}) { + options.create_if_missing = true; + options.max_open_files = max_open_files_; + + // disable auto compactions/flushes + options.disable_auto_compactions = true; + options.write_buffer_size = std::numeric_limits::max(); + + options.info_log_level = FATAL_LEVEL; // reduce LOG size + DestroyAndReopen(options); + + for (const auto& cf : cfs) { + if (cf != kDefaultColumnFamilyName) { + CreateColumnFamilies({cf}, options); + } + } + ASSERT_OK(TryReopenWithColumnFamilies(cfs, options)); + + for (uint32_t f = 0; f < num_flushes_; f++) { + for (int cf = 0; cf < static_cast(cfs.size()); cf++) { + ASSERT_OK(Put(cf, KeyFor(f), ValueFor(f))); + ASSERT_OK(Flush(cf)); + } + } + + Close(); + } + + void VerifyData(ReadType rt) { + switch (rt) { + case ReadType::kGet: + for (uint32_t f = 0; f < num_flushes_; f++) { + ASSERT_EQ(Get(KeyFor(f)), ValueFor(f)); + } + break; + case ReadType::kMultiGet: { + std::vector keys; + std::vector key_slices; + keys.reserve(num_flushes_); + key_slices.reserve(num_flushes_); + for (uint32_t f = 0; f < num_flushes_; f++) { + keys.push_back(KeyFor(f)); + } + for (auto& k : keys) { + key_slices.emplace_back(k); + } + std::vector values(num_flushes_); + std::vector statuses = + db_->MultiGet(ReadOptions(), key_slices, &values); + for (uint32_t f = 0; f < num_flushes_; f++) { + ASSERT_OK(statuses[f]); + ASSERT_EQ(values[f], ValueFor(f)); + } + break; + } + case ReadType::kIterator: { + std::unique_ptr iter(db_->NewIterator(ReadOptions())); + iter->SeekToFirst(); + for (uint32_t f = 0; f < num_flushes_; f++) { + ASSERT_TRUE(iter->Valid()); + ASSERT_EQ(iter->key().ToString(), KeyFor(f)); + ASSERT_EQ(iter->value().ToString(), ValueFor(f)); + iter->Next(); + } + ASSERT_OK(iter->status()); + break; + } + default: + assert(false); + } + } + + void OpenTestDB(Options& options) { + options.open_files_async = true; + options.skip_stats_update_on_db_open = true; + options.statistics = CreateDBStatistics(); + options.info_log_level = FATAL_LEVEL; // reduce LOG size + + std::vector cfs; + ASSERT_OK(DB::ListColumnFamilies(options, dbname_, &cfs)); + + if (read_only_) { + ASSERT_OK(TryReopenReadOnlyWithColumnFamilies(cfs, options)); + } else { + ASSERT_OK(TryReopenWithColumnFamilies(cfs, options)); + } + } + + void VerifyReadError(ReadType rt) { + switch (rt) { + case ReadType::kGet: { + PinnableSlice value; + Status read_s = db_->Get(ReadOptions(), db_->DefaultColumnFamily(), + KeyFor(0), &value); + ASSERT_TRUE(read_s.IsCorruption()) << read_s.ToString(); + break; + } + case ReadType::kMultiGet: { + std::vector keys = {KeyFor(0)}; + std::vector key_slices = {keys[0]}; + std::vector values(1); + std::vector statuses = + db_->MultiGet(ReadOptions(), key_slices, &values); + ASSERT_TRUE(statuses[0].IsCorruption()) << statuses[0].ToString(); + break; + } + case ReadType::kIterator: { + std::unique_ptr iter(db_->NewIterator(ReadOptions())); + iter->SeekToFirst(); + ASSERT_TRUE(iter->status().IsCorruption()) << iter->status().ToString(); + break; + } + default: + assert(false); + } + } + + uint32_t num_flushes_{}; + int max_open_files_{}; + bool read_only_{}; +}; + +INSTANTIATE_TEST_CASE_P(OpenFilesAsync, OpenFilesAsyncTest, + ::testing::Combine(::testing::Values(4), + ::testing::Values(-1, 10), + ::testing::Bool())); + +// Test mix of races with async file open, reads, compactions +TEST_P(OpenFilesAsyncTest, ConcurrentFileAccess) { + Options options = CurrentOptions(); + ASSERT_NO_FATAL_FAILURE(SetupData(options)); + + for (uint8_t i = 0; i < static_cast(ReadType::kNumReadTypes); i++) { + auto rt = static_cast(i); + OpenTestDB(options); + + port::Thread reader_thread([&]() { VerifyData(rt); }); + + port::Thread compaction_thread([&]() { + if (!read_only_) { + db_->CompactRange(CompactRangeOptions(), nullptr, nullptr) + .PermitUncheckedError(); + } + }); + + reader_thread.join(); + compaction_thread.join(); + + if (max_open_files_ == -1) { + EXPECT_EQ(TestGetTickerCount(options, NO_FILE_OPENS), num_flushes_); + } else { + EXPECT_GE(TestGetTickerCount(options, NO_FILE_OPENS), num_flushes_); + } + Close(); + } +} + +TEST_P(OpenFilesAsyncTest, AfterRead) { + Options options = CurrentOptions(); + ASSERT_NO_FATAL_FAILURE(SetupData(options)); + + for (uint8_t i = 0; i < static_cast(ReadType::kNumReadTypes); i++) { + auto rt = static_cast(i); + ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->LoadDependency({ + {"OpenFilesAsyncTest::AfterRead", + "VersionBuilder::Rep::LoadTableHandlers::BeforeFindTable"}, + {"DBImpl::BGWorkAsyncFileOpen:Done", + "OpenFilesAsyncTest::AfterRead::BeforeClose"}, + }); + + ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack( + BeforeFindTableSyncPoint(rt), [&](void* arg) { + FileDescriptor* fd = static_cast(arg); + if (max_open_files_ == -1) { + ASSERT_EQ(fd->pinned_reader.Get(), nullptr); + } + }); + + std::atomic file_opens = 0; + ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack( + "VersionBuilder::Rep::LoadTableHandlers::BeforeFindTable", + [&](void* arg) { + ++file_opens; + FileMetaData* file_meta = static_cast(arg); + if (max_open_files_ == -1) { + ASSERT_NE(file_meta->fd.pinned_reader.Get(), nullptr); + } + }); + + ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->EnableProcessing(); + + OpenTestDB(options); + + VerifyData(rt); + TEST_SYNC_POINT("OpenFilesAsyncTest::AfterRead"); + + TEST_SYNC_POINT("OpenFilesAsyncTest::AfterRead::BeforeClose"); + + if (max_open_files_ == -1) { + EXPECT_EQ(file_opens, num_flushes_); + EXPECT_EQ(TestGetTickerCount(options, NO_FILE_OPENS), num_flushes_); + } else { + EXPECT_GE(TestGetTickerCount(options, NO_FILE_OPENS), num_flushes_); + } + + ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->DisableProcessing(); + ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->ClearAllCallBacks(); + Close(); + } +} + +TEST_P(OpenFilesAsyncTest, BeforeRead) { + Options options = CurrentOptions(); + ASSERT_NO_FATAL_FAILURE(SetupData(options)); + + for (uint8_t i = 0; i < static_cast(ReadType::kNumReadTypes); i++) { + auto rt = static_cast(i); + ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->LoadDependency({ + {"DBImpl::BGWorkAsyncFileOpen:Done", "OpenFilesAsyncTest::BeforeRead"}, + }); + + ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack( + BeforeFindTableSyncPoint(rt), [&](void* arg) { + FileDescriptor* fd = static_cast(arg); + if (options.max_open_files == -1) { + ASSERT_NE(fd->pinned_reader.Get(), nullptr); + } + }); + + std::atomic file_opens = 0; + ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack( + "VersionBuilder::Rep::LoadTableHandlers::BeforeFindTable", + [&](void* arg) { + ++file_opens; + FileMetaData* file_meta = static_cast(arg); + ASSERT_EQ(file_meta->fd.pinned_reader.Get(), nullptr); + }); + + ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->EnableProcessing(); + + OpenTestDB(options); + + TEST_SYNC_POINT("OpenFilesAsyncTest::BeforeRead"); + + VerifyData(rt); + + if (max_open_files_ == -1) { + EXPECT_EQ(file_opens, num_flushes_); + EXPECT_EQ(TestGetTickerCount(options, NO_FILE_OPENS), num_flushes_); + } else { + EXPECT_GE(TestGetTickerCount(options, NO_FILE_OPENS), num_flushes_); + } + + ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->DisableProcessing(); + ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->ClearAllCallBacks(); + Close(); + } +} + +TEST_P(OpenFilesAsyncTest, Shutdown) { + Options options = CurrentOptions(); + ASSERT_NO_FATAL_FAILURE(SetupData(options)); + + std::atomic file_opens = 0; + ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack( + "VersionBuilder::Rep::LoadTableHandlers::BeforeFindTable", + [&](void* /*arg*/) { ++file_opens; }); + + ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->LoadDependency( + {{"DBImpl::~DBImpl:WaitJob", "DBImpl::BGWorkAsyncFileOpen::Start"}, + {"DBImpl::BGWorkAsyncFileOpen:Done", "OpenFilesAsyncTest::Shutdown"}}); + + ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->EnableProcessing(); + + OpenTestDB(options); + + Close(); + TEST_SYNC_POINT("OpenFilesAsyncTest::Shutdown"); + + ASSERT_EQ(file_opens, 0); + + ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->DisableProcessing(); + ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->ClearAllCallBacks(); +} + +TEST_P(OpenFilesAsyncTest, Error) { + Options options = CurrentOptions(); + ASSERT_NO_FATAL_FAILURE(SetupData(options)); + + // Get SST file paths and corrupt all of them + std::vector sst_files; + { + Reopen(options); + std::vector metadata; + db_->GetLiveFilesMetaData(&metadata); + ASSERT_GT(metadata.size(), 0); + for (const auto& m : metadata) { + sst_files.push_back(m.directory + "/" + m.name); + } + Close(); + } + for (const auto& f : sst_files) { + ASSERT_OK(test::CorruptFile(env_, f, 0, 100)); + } + + for (uint8_t i = 0; i < static_cast(ReadType::kNumReadTypes); i++) { + auto rt = static_cast(i); + ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->LoadDependency({ + {"DBImpl::BGWorkAsyncFileOpen:Done", "OpenFilesAsyncTest::Error"}, + }); + + ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->EnableProcessing(); + + OpenTestDB(options); + + TEST_SYNC_POINT("OpenFilesAsyncTest::Error"); + + ASSERT_TRUE(dbfull()->TEST_GetBGError().IsCorruption()); + + VerifyReadError(rt); + + ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->DisableProcessing(); + ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->ClearAllCallBacks(); + Close(); + } +} + +TEST_P(OpenFilesAsyncTest, DropColumnFamily) { + // Can't drop CFs in read-only mode + if (read_only_) { + return; + } + + for (uint8_t i = 0; i < static_cast(ReadType::kNumReadTypes); i++) { + auto rt = static_cast(i); + Options options = CurrentOptions(); + ASSERT_NO_FATAL_FAILURE( + SetupData(options, {kDefaultColumnFamilyName, "cf1"})); + + // Delay the async opener until after we drop the CF + ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->LoadDependency({ + {"OpenFilesAsyncTest::DropColumnFamily::AfterDrop", + "DBImpl::BGWorkAsyncFileOpen::Start"}, + {"DBImpl::BGWorkAsyncFileOpen:Done", + "OpenFilesAsyncTest::DropColumnFamily::BeforeClose"}, + }); + + ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->EnableProcessing(); + + OpenTestDB(options); + + // Drop the non-default CF before async opener runs + ASSERT_OK(db_->DropColumnFamily(handles_[1])); + ASSERT_OK(db_->DestroyColumnFamilyHandle(handles_[1])); + handles_[1] = nullptr; + + TEST_SYNC_POINT("OpenFilesAsyncTest::DropColumnFamily::AfterDrop"); + + // Default CF data should still be readable + VerifyData(rt); + + TEST_SYNC_POINT("OpenFilesAsyncTest::DropColumnFamily::BeforeClose"); + + if (max_open_files_ == -1) { + EXPECT_EQ(TestGetTickerCount(options, NO_FILE_OPENS), num_flushes_); + } else { + EXPECT_GE(TestGetTickerCount(options, NO_FILE_OPENS), num_flushes_); + } + + ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->DisableProcessing(); + ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->ClearAllCallBacks(); + + Close(); + } +} + } // namespace ROCKSDB_NAMESPACE int main(int argc, char** argv) { diff --git a/db/db_test_util.cc b/db/db_test_util.cc index d62807d265..b8a90d32d2 100644 --- a/db/db_test_util.cc +++ b/db/db_test_util.cc @@ -675,6 +675,28 @@ Status DBTestBase::TryReopenWithColumnFamilies( return TryReopenWithColumnFamilies(cfs, v_opts); } +Status DBTestBase::TryReopenReadOnlyWithColumnFamilies( + const std::vector& cfs, const std::vector& options) { + Close(); + EXPECT_EQ(cfs.size(), options.size()); + std::vector column_families; + for (size_t i = 0; i < cfs.size(); ++i) { + column_families.emplace_back(cfs[i], options[i]); + } + DBOptions db_opts = DBOptions(options[0]); + last_options_ = options[0]; + MaybeInstallTimeElapseOnlySleep(db_opts); + return DB::OpenForReadOnly(db_opts, dbname_, column_families, &handles_, + &db_); +} + +Status DBTestBase::TryReopenReadOnlyWithColumnFamilies( + const std::vector& cfs, const Options& options) { + Close(); + std::vector v_opts(cfs.size(), options); + return TryReopenReadOnlyWithColumnFamilies(cfs, v_opts); +} + void DBTestBase::Reopen(const Options& options) { ASSERT_OK(TryReopen(options)); } diff --git a/db/db_test_util.h b/db/db_test_util.h index 44768f1d1c..7bdc9bef9c 100644 --- a/db/db_test_util.h +++ b/db/db_test_util.h @@ -1177,6 +1177,12 @@ class DBTestBase : public testing::Test { Status TryReopenWithColumnFamilies(const std::vector& cfs, const Options& options); + Status TryReopenReadOnlyWithColumnFamilies( + const std::vector& cfs, const std::vector& options); + + Status TryReopenReadOnlyWithColumnFamilies( + const std::vector& cfs, const Options& options); + void Reopen(const Options& options); void Close(); diff --git a/db/db_wal_test.cc b/db/db_wal_test.cc index 1e9270db0d..d3e9d553b4 100644 --- a/db/db_wal_test.cc +++ b/db/db_wal_test.cc @@ -600,9 +600,9 @@ TEST_F(DBWALTest, RecoverWithTableHandle) { for (const auto& level : files) { for (const auto& file : level) { if (options.max_open_files == kSmallMaxOpenFiles) { - ASSERT_TRUE(file.table_reader_handle == nullptr); + ASSERT_TRUE(file.fd.pinned_reader.Get() == nullptr); } else { - ASSERT_TRUE(file.table_reader_handle != nullptr); + ASSERT_TRUE(file.fd.pinned_reader.Get() != nullptr); } } } diff --git a/db/error_handler.cc b/db/error_handler.cc index 1e777fd426..e15de116cb 100644 --- a/db/error_handler.cc +++ b/db/error_handler.cc @@ -201,6 +201,25 @@ std::map, {std::make_tuple(BackgroundErrorReason::kManifestWriteNoWAL, Status::Code::kIOError, false), Status::Severity::kFatalError}, + // Errors during BG async file open + {std::make_tuple(BackgroundErrorReason::kAsyncFileOpen, + Status::Code::kCorruption, true), + Status::Severity::kUnrecoverableError}, + {std::make_tuple(BackgroundErrorReason::kAsyncFileOpen, + Status::Code::kCorruption, false), + Status::Severity::kNoError}, + {std::make_tuple(BackgroundErrorReason::kAsyncFileOpen, + Status::Code::kIncomplete, true), + Status::Severity::kFatalError}, + {std::make_tuple(BackgroundErrorReason::kAsyncFileOpen, + Status::Code::kIncomplete, false), + Status::Severity::kNoError}, + {std::make_tuple(BackgroundErrorReason::kAsyncFileOpen, + Status::Code::kNotSupported, true), + Status::Severity::kFatalError}, + {std::make_tuple(BackgroundErrorReason::kAsyncFileOpen, + Status::Code::kNotSupported, false), + Status::Severity::kNoError}, }; std::map, Status::Severity> @@ -225,6 +244,11 @@ std::map, Status::Severity> Status::Severity::kFatalError}, {std::make_tuple(BackgroundErrorReason::kMemTable, false), Status::Severity::kFatalError}, + // Errors during BG async file open + {std::make_tuple(BackgroundErrorReason::kAsyncFileOpen, true), + Status::Severity::kSoftError}, + {std::make_tuple(BackgroundErrorReason::kAsyncFileOpen, false), + Status::Severity::kNoError}, }; void ErrorHandler::CancelErrorRecoveryForShutDown() { diff --git a/db/table_cache.cc b/db/table_cache.cc index 0e4e9f2e51..cb8792761e 100644 --- a/db/table_cache.cc +++ b/db/table_cache.cc @@ -72,6 +72,8 @@ TableCache::TableCache(const ImmutableOptions& ioptions, file_options_(*file_options), cache_(cache), immortal_tables_(false), + should_pin_table_handles_(cache_.get()->GetCapacity() >= + kInfiniteCapacity), block_cache_tracer_(block_cache_tracer), loader_mutex_(kLoadConcurency), io_tracer_(io_tracer), @@ -169,15 +171,29 @@ Cache::Handle* TableCache::Lookup(Cache* cache, uint64_t file_number) { return cache->Lookup(key); } +// TODO: consider making handle RAII. Status TableCache::FindTable( const ReadOptions& ro, const FileOptions& file_options, const InternalKeyComparator& internal_comparator, const FileMetaData& file_meta, TypedHandle** handle, - const MutableCFOptions& mutable_cf_options, const bool no_io, - HistogramImpl* file_read_hist, bool skip_filters, int level, - bool prefetch_index_and_filter_in_cache, - size_t max_file_size_for_l0_meta_pin, Temperature file_temperature) { + const MutableCFOptions& mutable_cf_options, TableReader** out_table_reader, + const bool no_io, HistogramImpl* file_read_hist, bool skip_filters, + int level, bool prefetch_index_and_filter_in_cache, + size_t max_file_size_for_l0_meta_pin, Temperature file_temperature, + bool pin_table_handle) { + assert(out_table_reader != nullptr && *out_table_reader == nullptr); + assert(handle != nullptr && *handle == nullptr); PERF_TIMER_GUARD_WITH_CLOCK(find_table_nanos, ioptions_.clock); + + // Fast path: if table reader is already pinned, return it directly without a + // cache lookup. + auto pinned_reader = file_meta.fd.pinned_reader.Get(); + if (pinned_reader != nullptr) { + *handle = nullptr; + *out_table_reader = pinned_reader; + return Status::OK(); + } + uint64_t number = file_meta.fd.GetNumber(); // NOTE: sharing same Cache with BlobFileCache Slice key = GetSliceForFileNumber(&number); @@ -185,39 +201,71 @@ Status TableCache::FindTable( TEST_SYNC_POINT_CALLBACK("TableCache::FindTable:0", const_cast(&no_io)); + Status s = Status::OK(); if (*handle == nullptr) { if (no_io) { - return Status::Incomplete("Table not found in table_cache, no_io is set"); + s = Status::Incomplete("Table not found in table_cache, no_io is set"); + return s; } MutexLock load_lock(&loader_mutex_.Get(key)); - // We check the cache again under loading mutex - *handle = cache_.Lookup(key); - if (*handle != nullptr) { - return Status::OK(); + + // Check if another thread has already pinned the table reader + pinned_reader = file_meta.fd.pinned_reader.Get(); + if (pinned_reader != nullptr) { + *handle = nullptr; + *out_table_reader = pinned_reader; + return s; } - std::unique_ptr table_reader; - Status s = GetTableReader(ro, file_options, internal_comparator, file_meta, - false /* sequential mode */, file_read_hist, - &table_reader, mutable_cf_options, skip_filters, - level, prefetch_index_and_filter_in_cache, - max_file_size_for_l0_meta_pin, file_temperature); - if (!s.ok()) { - assert(table_reader == nullptr); - RecordTick(ioptions_.stats, NO_FILE_ERRORS); - // We do not cache error results so that if the error is transient, - // or somebody repairs the file, we recover automatically. - IGNORE_STATUS_IF_ERROR(s); - } else { - s = cache_.Insert(key, table_reader.get(), 1, handle); - if (s.ok()) { - // Release ownership of table reader. - table_reader.release(); + // We check the cache again under loading mutex + *handle = cache_.Lookup(key); + if (*handle == nullptr) { + std::unique_ptr table_reader; + s = GetTableReader(ro, file_options, internal_comparator, file_meta, + false /* sequential mode */, file_read_hist, + &table_reader, mutable_cf_options, skip_filters, level, + prefetch_index_and_filter_in_cache, + max_file_size_for_l0_meta_pin, file_temperature); + if (!s.ok()) { + assert(table_reader == nullptr); + RecordTick(ioptions_.stats, NO_FILE_ERRORS); + // We do not cache error results so that if the error is transient, + // or somebody repairs the file, we recover automatically. + IGNORE_STATUS_IF_ERROR(s); + } else { + s = cache_.Insert(key, table_reader.get(), 1, handle); + if (s.ok()) { + // Release ownership of table reader. + (void)table_reader.release(); + } } } - return s; + + if (s.ok()) { + *out_table_reader = cache_.Value(*handle); + if (pin_table_handle) { + file_meta.fd.pinned_reader.Pin(*handle, *out_table_reader); + *handle = nullptr; + } + } + } else { + *out_table_reader = cache_.Value(*handle); + if (pin_table_handle) { + // handle is in cache but not pinned. This should happen fairly rarely, + // and once the reader is pinned, we will no longer need to go through + // these mutexes again. + MutexLock load_lock(&loader_mutex_.Get(key)); + if (file_meta.fd.pinned_reader.Get() != nullptr) { + // Another thread has pinned the handle; release our lookup ref. + cache_.Release(*handle); + } else { + file_meta.fd.pinned_reader.Pin(*handle, *out_table_reader); + } + *handle = nullptr; + } } - return Status::OK(); + + return s; } InternalIterator* TableCache::NewIterator( @@ -230,7 +278,8 @@ InternalIterator* TableCache::NewIterator( const InternalKey* smallest_compaction_key, const InternalKey* largest_compaction_key, bool allow_unprepared_value, const SequenceNumber* read_seqno, - std::unique_ptr* range_del_iter) { + std::unique_ptr* range_del_iter, + bool maybe_pin_table_handle) { PERF_TIMER_GUARD(new_table_iterator_nanos); Status s; @@ -240,19 +289,15 @@ InternalIterator* TableCache::NewIterator( *table_reader_ptr = nullptr; } bool for_compaction = caller == TableReaderCaller::kCompaction; - auto& fd = file_meta.fd; - table_reader = fd.table_reader; - if (table_reader == nullptr) { - s = FindTable(options, file_options, icomparator, file_meta, &handle, - mutable_cf_options, - options.read_tier == kBlockCacheTier /* no_io */, - file_read_hist, skip_filters, level, - true /* prefetch_index_and_filter_in_cache */, - max_file_size_for_l0_meta_pin, file_meta.temperature); - if (s.ok()) { - table_reader = cache_.Value(handle); - } - } + TEST_SYNC_POINT_CALLBACK("TableCache::NewIterator::BeforeFindTable", + const_cast(&file_meta.fd)); + s = FindTable(options, file_options, icomparator, file_meta, &handle, + mutable_cf_options, &table_reader, + options.read_tier == kBlockCacheTier /* no_io */, + file_read_hist, skip_filters, level, + true /* prefetch_index_and_filter_in_cache */, + max_file_size_for_l0_meta_pin, file_meta.temperature, + maybe_pin_table_handle && should_pin_table_handles_); InternalIterator* result = nullptr; if (s.ok()) { if (options.table_filter && @@ -293,7 +338,7 @@ InternalIterator* TableCache::NewIterator( } } if (range_del_agg != nullptr) { - if (range_del_agg->AddFile(fd.GetNumber())) { + if (range_del_agg->AddFile(file_meta.fd.GetNumber())) { std::unique_ptr new_range_del_iter( static_cast( table_reader->NewRangeTombstoneIterator(options))); @@ -332,17 +377,11 @@ Status TableCache::GetRangeTombstoneIterator( const FileMetaData& file_meta, const MutableCFOptions& mutable_cf_options, std::unique_ptr* out_iter) { assert(out_iter); - const FileDescriptor& fd = file_meta.fd; Status s; - TableReader* t = fd.table_reader; + TableReader* t = nullptr; TypedHandle* handle = nullptr; - if (t == nullptr) { - s = FindTable(options, file_options_, internal_comparator, file_meta, - &handle, mutable_cf_options); - if (s.ok()) { - t = cache_.Value(handle); - } - } + s = FindTable(options, file_options_, internal_comparator, file_meta, &handle, + mutable_cf_options, &t); if (s.ok()) { // Note: NewRangeTombstoneIterator could return nullptr out_iter->reset(t->NewRangeTombstoneIterator(options)); @@ -456,20 +495,18 @@ Status TableCache::Get(const ReadOptions& options, row_cache_entry = &row_cache_entry_buffer; } } - TableReader* t = fd.table_reader; + TEST_SYNC_POINT_CALLBACK("TableCache::Get::BeforeFindTable", + const_cast(&fd)); + TableReader* t = nullptr; TypedHandle* handle = nullptr; if (s.ok() && !done) { - if (t == nullptr) { - s = FindTable(options, file_options_, internal_comparator, file_meta, - &handle, mutable_cf_options, - options.read_tier == kBlockCacheTier /* no_io */, - file_read_hist, skip_filters, level, - true /* prefetch_index_and_filter_in_cache */, - max_file_size_for_l0_meta_pin, file_meta.temperature); - if (s.ok()) { - t = cache_.Value(handle); - } - } + s = FindTable(options, file_options_, internal_comparator, file_meta, + &handle, mutable_cf_options, &t, + options.read_tier == kBlockCacheTier /* no_io */, + file_read_hist, skip_filters, level, + true /* prefetch_index_and_filter_in_cache */, + max_file_size_for_l0_meta_pin, file_meta.temperature, + should_pin_table_handles_); SequenceNumber* max_covering_tombstone_seq = get_context->max_covering_tombstone_seq(); if (s.ok() && max_covering_tombstone_seq != nullptr && @@ -546,8 +583,8 @@ Status TableCache::MultiGetFilter( const InternalKeyComparator& internal_comparator, const FileMetaData& file_meta, const MutableCFOptions& mutable_cf_options, HistogramImpl* file_read_hist, int level, - MultiGetContext::Range* mget_range, TypedHandle** table_handle) { - auto& fd = file_meta.fd; + MultiGetContext::Range* mget_range, TypedHandle** handle) { + assert(*handle == nullptr); IterKey row_cache_key; std::string row_cache_entry_buffer; @@ -559,23 +596,17 @@ Status TableCache::MultiGetFilter( return Status::NotSupported(); } Status s; - TableReader* t = fd.table_reader; - TypedHandle* handle = nullptr; + TableReader* t = nullptr; MultiGetContext::Range tombstone_range(*mget_range, mget_range->begin(), mget_range->end()); - if (t == nullptr) { - s = FindTable(options, file_options_, internal_comparator, file_meta, - &handle, mutable_cf_options, - options.read_tier == kBlockCacheTier /* no_io */, - file_read_hist, - /*skip_filters=*/false, level, - true /* prefetch_index_and_filter_in_cache */, - /*max_file_size_for_l0_meta_pin=*/0, file_meta.temperature); - if (s.ok()) { - t = cache_.Value(handle); - } - *table_handle = handle; - } + s = FindTable(options, file_options_, internal_comparator, file_meta, handle, + mutable_cf_options, &t, + options.read_tier == kBlockCacheTier /* no_io */, + file_read_hist, + /*skip_filters=*/false, level, + true /* prefetch_index_and_filter_in_cache */, + /*max_file_size_for_l0_meta_pin=*/0, file_meta.temperature, + should_pin_table_handles_); if (s.ok()) { s = t->MultiGetFilter(options, mutable_cf_options.prefix_extractor.get(), mget_range); @@ -586,9 +617,9 @@ Status TableCache::MultiGetFilter( // is, it may be called with fewer keys in the rangedue to filtering. UpdateRangeTombstoneSeqnums(options, t, tombstone_range); } - if (mget_range->empty() && handle) { - cache_.Release(handle); - *table_handle = nullptr; + if (mget_range->empty() && *handle) { + cache_.Release(*handle); + *handle = nullptr; } return s; @@ -600,24 +631,18 @@ Status TableCache::GetTableProperties( const FileMetaData& file_meta, std::shared_ptr* properties, const MutableCFOptions& mutable_cf_options, bool no_io) { - auto table_reader = file_meta.fd.table_reader; - // table already been pre-loaded? - if (table_reader) { - *properties = table_reader->GetTableProperties(); - - return Status::OK(); - } - TypedHandle* table_handle = nullptr; - Status s = FindTable(read_options, file_options, internal_comparator, - file_meta, &table_handle, mutable_cf_options, no_io); + TableReader* table = nullptr; + Status s = + FindTable(read_options, file_options, internal_comparator, file_meta, + &table_handle, mutable_cf_options, &table, no_io); if (!s.ok()) { return s; } - assert(table_handle); - auto table = cache_.Value(table_handle); *properties = table->GetTableProperties(); - cache_.Release(table_handle); + if (table_handle) { + cache_.Release(table_handle); + } return s; } @@ -627,15 +652,10 @@ Status TableCache::ApproximateKeyAnchors( std::vector& anchors) { Status s; - TableReader* t = file_meta.fd.table_reader; + TableReader* t = nullptr; TypedHandle* handle = nullptr; - if (t == nullptr) { - s = FindTable(ro, file_options_, internal_comparator, file_meta, &handle, - mutable_cf_options); - if (s.ok()) { - t = cache_.Value(handle); - } - } + s = FindTable(ro, file_options_, internal_comparator, file_meta, &handle, + mutable_cf_options, &t); if (s.ok() && t != nullptr) { s = t->ApproximateKeyAnchors(ro, anchors); } @@ -649,23 +669,18 @@ size_t TableCache::GetMemoryUsageByTableReader( const FileOptions& file_options, const ReadOptions& read_options, const InternalKeyComparator& internal_comparator, const FileMetaData& file_meta, const MutableCFOptions& mutable_cf_options) { - auto table_reader = file_meta.fd.table_reader; - // table already been pre-loaded? - if (table_reader) { - return table_reader->ApproximateMemoryUsage(); - } - TypedHandle* table_handle = nullptr; + TableReader* table = nullptr; Status s = FindTable(read_options, file_options, internal_comparator, file_meta, - &table_handle, mutable_cf_options, true /* no_io */); + &table_handle, mutable_cf_options, &table, true /* no_io */); if (!s.ok()) { return 0; } - assert(table_handle); - auto table = cache_.Value(table_handle); auto ret = table->ApproximateMemoryUsage(); - cache_.Release(table_handle); + if (table_handle) { + cache_.Release(table_handle); + } return ret; } @@ -679,18 +694,13 @@ uint64_t TableCache::ApproximateOffsetOf( const InternalKeyComparator& internal_comparator, const MutableCFOptions& mutable_cf_options) { uint64_t result = 0; - TableReader* table_reader = file_meta.fd.table_reader; + TableReader* table_reader = nullptr; TypedHandle* table_handle = nullptr; - if (table_reader == nullptr) { - Status s = - FindTable(read_options, file_options_, internal_comparator, file_meta, - &table_handle, mutable_cf_options, false /* no_io */); - if (s.ok()) { - table_reader = cache_.Value(table_handle); - } - } + Status s = + FindTable(read_options, file_options_, internal_comparator, file_meta, + &table_handle, mutable_cf_options, &table_reader); - if (table_reader != nullptr) { + if (s.ok() && table_reader != nullptr) { result = table_reader->ApproximateOffsetOf(read_options, key, caller); } if (table_handle != nullptr) { @@ -706,18 +716,13 @@ uint64_t TableCache::ApproximateSize( const InternalKeyComparator& internal_comparator, const MutableCFOptions& mutable_cf_options) { uint64_t result = 0; - TableReader* table_reader = file_meta.fd.table_reader; + TableReader* table_reader = nullptr; TypedHandle* table_handle = nullptr; - if (table_reader == nullptr) { - Status s = - FindTable(read_options, file_options_, internal_comparator, file_meta, - &table_handle, mutable_cf_options, false /* no_io */); - if (s.ok()) { - table_reader = cache_.Value(table_handle); - } - } + Status s = + FindTable(read_options, file_options_, internal_comparator, file_meta, + &table_handle, mutable_cf_options, &table_reader); - if (table_reader != nullptr) { + if (s.ok() && table_reader != nullptr) { result = table_reader->ApproximateSize(read_options, start, end, caller); } if (table_handle != nullptr) { diff --git a/db/table_cache.h b/db/table_cache.h index 701f96f22a..4f13766866 100644 --- a/db/table_cache.h +++ b/db/table_cache.h @@ -99,7 +99,8 @@ class TableCache { const InternalKey* smallest_compaction_key, const InternalKey* largest_compaction_key, bool allow_unprepared_value, const SequenceNumber* range_del_read_seqno = nullptr, - std::unique_ptr* range_del_iter = nullptr); + std::unique_ptr* range_del_iter = nullptr, + bool maybe_pin_table_handle = false); // If a seek to internal key "k" in specified file finds an entry, // call get_context->SaveValue() repeatedly until @@ -172,19 +173,31 @@ class TableCache { // Return handle to an existing cache entry if there is one static Cache::Handle* Lookup(Cache* cache, uint64_t file_number); - // Find table reader - // @param skip_filters Disables loading/accessing the filter block - // @param level == -1 means not specified + // Look up the TableReader for the given file in the cache, or open the file + // and create a new TableReader if not cached. On success, sets *table_reader + // to point to the TableReader (owned by the cache) and *handle to the cache + // handle (caller must release via cache_.Release() unless pin_table_handle is + // true). If the table reader is already pinned on file_meta, returns it + // directly without a cache lookup. + // + // @param no_io If true, returns Status::Incomplete() when the table is not + // already in cache rather than reading from disk. + // @param skip_filters Disables loading/accessing the filter block. + // @param level The LSM level of this table, -1 if not specified. + // @param pin_table_handle If true, pins the table reader on file_meta so + // future lookups bypass the cache. *handle is set to nullptr + // on return in this case. Status FindTable(const ReadOptions& ro, const FileOptions& toptions, const InternalKeyComparator& internal_comparator, const FileMetaData& file_meta, TypedHandle**, const MutableCFOptions& mutable_cf_options, - const bool no_io = false, + TableReader** table_reader, const bool no_io = false, HistogramImpl* file_read_hist = nullptr, bool skip_filters = false, int level = -1, bool prefetch_index_and_filter_in_cache = true, size_t max_file_size_for_l0_meta_pin = 0, - Temperature file_temperature = Temperature::kUnknown); + Temperature file_temperature = Temperature::kUnknown, + bool pin_table_handle = false); // Get the table properties of a given table. // @no_io: indicates if we should load table to the cache if it is not present @@ -231,6 +244,8 @@ class TableCache { CacheInterface& get_cache() { return cache_; } + const FileOptions& file_options() const { return file_options_; } + // Capacity of the backing Cache that indicates infinite TableCache capacity. // For example when max_open_files is -1 we set the backing Cache to this. static const int kInfiniteCapacity = 0x400000; @@ -243,6 +258,14 @@ class TableCache { } } + // Re-evaluates should_pin_table_handles_ from the current cache capacity. + // Must be called after the underlying cache capacity changes (e.g. via + // SetDBOptions changing max_open_files). + void UpdateShouldPinTableHandles() { + should_pin_table_handles_ = + cache_.get()->GetCapacity() >= kInfiniteCapacity; + } + private: // Build a table reader Status GetTableReader(const ReadOptions& ro, const FileOptions& file_options, @@ -283,6 +306,7 @@ class TableCache { CacheInterface cache_; std::string row_cache_id_; bool immortal_tables_; + bool should_pin_table_handles_; BlockCacheTracer* const block_cache_tracer_; Striped> loader_mutex_; std::shared_ptr io_tracer_; diff --git a/db/table_cache_sync_and_async.h b/db/table_cache_sync_and_async.h index 68228485ce..24bcbdaa61 100644 --- a/db/table_cache_sync_and_async.h +++ b/db/table_cache_sync_and_async.h @@ -21,7 +21,9 @@ DEFINE_SYNC_AND_ASYNC(Status, TableCache::MultiGet) bool skip_filters, bool skip_range_deletions, int level, TypedHandle* handle) { auto& fd = file_meta.fd; Status s; - TableReader* t = fd.table_reader; + TEST_SYNC_POINT_CALLBACK("TableCache::MultiGet::BeforeFindTable", + const_cast(&fd)); + TableReader* t = fd.pinned_reader.Get(); MultiGetRange table_range(*mget_range, mget_range->begin(), mget_range->end()); if (handle != nullptr && t == nullptr) { @@ -70,16 +72,14 @@ DEFINE_SYNC_AND_ASYNC(Status, TableCache::MultiGet) if (t == nullptr) { assert(handle == nullptr); s = FindTable(options, file_options_, internal_comparator, file_meta, - &handle, mutable_cf_options, + &handle, mutable_cf_options, &t, options.read_tier == kBlockCacheTier /* no_io */, file_read_hist, skip_filters, level, true /* prefetch_index_and_filter_in_cache */, - 0 /*max_file_size_for_l0_meta_pin*/, file_meta.temperature); + 0 /*max_file_size_for_l0_meta_pin*/, file_meta.temperature, + should_pin_table_handles_); TEST_SYNC_POINT_CALLBACK("TableCache::MultiGet:FindTable", &s); - if (s.ok()) { - t = cache_.Value(handle); - assert(t); - } + assert(!s.ok() || t); } if (s.ok() && !options.ignore_range_deletions && !skip_range_deletions) { UpdateRangeTombstoneSeqnums(options, t, table_range); diff --git a/db/version_builder.cc b/db/version_builder.cc index aa7510c97f..05bd9d7b5e 100644 --- a/db/version_builder.cc +++ b/db/version_builder.cc @@ -32,8 +32,10 @@ #include "db/version_edit.h" #include "db/version_edit_handler.h" #include "db/version_set.h" +#include "db/version_util.h" #include "port/port.h" #include "table/table_reader.h" +#include "test_util/sync_point.h" #include "util/string_util.h" namespace ROCKSDB_NAMESPACE { @@ -430,12 +432,11 @@ class VersionBuilder::Rep { void UnrefFile(FileMetaData* f) { f->refs--; if (f->refs <= 0) { - if (f->table_reader_handle) { + if (f->fd.pinned_reader.Get() != nullptr) { assert(table_cache_ != nullptr); // NOTE: have to release in raw cache interface to avoid using a - // TypedHandle for FileMetaData::table_reader_handle - table_cache_->get_cache().get()->Release(f->table_reader_handle); - f->table_reader_handle = nullptr; + // TypedHandle for PinnedTableReader's internal handle + f->fd.pinned_reader.Release(table_cache_->get_cache().get()); } if (file_metadata_cache_res_mgr_) { @@ -1680,7 +1681,6 @@ class VersionBuilder::Rep { // std::vector> files_meta; - std::vector statuses; for (int level = 0; level < num_levels_; level++) { for (auto& file_meta_pair : levels_[level].added_files) { auto* file_meta = file_meta_pair.second; @@ -1690,9 +1690,8 @@ class VersionBuilder::Rep { continue; } // If the file has been opened before, just skip it. - if (!file_meta->table_reader_handle) { + if (file_meta->fd.pinned_reader.Get() == nullptr) { files_meta.emplace_back(file_meta, level); - statuses.emplace_back(Status::OK()); } if (files_meta.size() >= max_load) { break; @@ -1703,49 +1702,11 @@ class VersionBuilder::Rep { } } - std::atomic next_file_meta_idx(0); - std::function load_handlers_func([&]() { - while (true) { - size_t file_idx = next_file_meta_idx.fetch_add(1); - if (file_idx >= files_meta.size()) { - break; - } - - auto* file_meta = files_meta[file_idx].first; - int level = files_meta[file_idx].second; - TableCache::TypedHandle* handle = nullptr; - statuses[file_idx] = table_cache_->FindTable( - read_options, file_options_, - *(base_vstorage_->InternalComparator()), *file_meta, &handle, - mutable_cf_options, false /*no_io */, - internal_stats->GetFileReadHist(level), false, level, - prefetch_index_and_filter_in_cache, max_file_size_for_l0_meta_pin, - file_meta->temperature); - if (handle != nullptr) { - file_meta->table_reader_handle = handle; - // Load table_reader - file_meta->fd.table_reader = table_cache_->get_cache().Value(handle); - } - } - }); - - std::vector threads; - for (int i = 1; i < max_threads; i++) { - threads.emplace_back(load_handlers_func); - } - load_handlers_func(); - for (auto& t : threads) { - t.join(); - } - Status ret; - for (const auto& s : statuses) { - if (!s.ok()) { - if (ret.ok()) { - ret = s; - } - } - } - return ret; + return LoadTableHandlersHelper( + files_meta, table_cache_, file_options_, + *base_vstorage_->InternalComparator(), internal_stats, max_threads, + prefetch_index_and_filter_in_cache, mutable_cf_options, + max_file_size_for_l0_meta_pin, read_options); } }; diff --git a/db/version_builder.h b/db/version_builder.h index 55520b4304..ce5e1b1f41 100644 --- a/db/version_builder.h +++ b/db/version_builder.h @@ -53,7 +53,8 @@ class VersionBuilder { // Save the current Version to the provided `vstorage`. Status SaveTo(VersionStorageInfo* vstorage) const; - // Load all the table handlers for the current Version in the builder. + // Load table handlers for newly added files in the builder. This does not + // load any files in the base storage. Status LoadTableHandlers(InternalStats* internal_stats, int max_threads, bool prefetch_index_and_filter_in_cache, bool is_initial_load, diff --git a/db/version_edit.cc b/db/version_edit.cc index 252413f2f9..d310271e15 100644 --- a/db/version_edit.cc +++ b/db/version_edit.cc @@ -22,6 +22,40 @@ namespace ROCKSDB_NAMESPACE { namespace {} // anonymous namespace +// After copying, we rely on the caller to ensure no double releases. Fragile, +// but keeps copying cheap. +PinnedTableReader& PinnedTableReader::operator=( + const PinnedTableReader& other) { + TableReader* r = other.reader_.load(std::memory_order_acquire); + // Only read handle_ when reader_ is non-null. Pin() writes handle_ before + // reader_ (with release), so a non-null reader_ guarantees handle_ is stable. + // If reader_ is null, Pin() may be in progress — avoid reading handle_. + handle_ = (r != nullptr) ? other.handle_ : nullptr; + reader_.store(r, std::memory_order_release); + return *this; +} + +Cache::Handle* PinnedTableReader::GetCacheHandle() const { + (void)reader_.load(std::memory_order_acquire); + return handle_; +} + +void PinnedTableReader::Pin(Cache::Handle* handle, TableReader* reader) { + assert(handle != nullptr); + assert(reader != nullptr); + handle_ = handle; + reader_.store(reader, std::memory_order_release); +} + +void PinnedTableReader::Release(Cache* cache) { + (void)reader_.load(std::memory_order_acquire); + if (handle_ != nullptr) { + cache->Release(handle_); + handle_ = nullptr; + reader_.store(nullptr, std::memory_order_relaxed); + } +} + uint64_t PackFileNumberAndPathId(uint64_t number, uint64_t path_id) { assert(number <= kFileNumberMask); return number | (path_id * (kFileNumberMask + 1)); diff --git a/db/version_edit.h b/db/version_edit.h index ee6a6b01be..e0394e84e0 100644 --- a/db/version_edit.h +++ b/db/version_edit.h @@ -134,14 +134,57 @@ constexpr uint64_t kReservedEpochNumberForFileIngestedBehind = 1; uint64_t PackFileNumberAndPathId(uint64_t number, uint64_t path_id); +// PinnedTableReader is used to safely access a table reader in a multi-threaded +// context. It holds both a pointer to the table reader and a cache handle. +class PinnedTableReader { + public: + PinnedTableReader() : reader_(nullptr), handle_(nullptr) {} + ~PinnedTableReader() = default; + PinnedTableReader(const PinnedTableReader& other) + : reader_(nullptr), handle_(nullptr) { + *this = other; + } + PinnedTableReader& operator=(const PinnedTableReader& other); + PinnedTableReader(PinnedTableReader&&) = delete; + PinnedTableReader& operator=(PinnedTableReader&&) = delete; + + // Returns the pinned TableReader, or nullptr if not pinned. + TableReader* Get() const { return reader_.load(std::memory_order_acquire); } + + // Returns the cache handle that keeps TableReader alive, or nullptr if not + // pinned. + Cache::Handle* GetCacheHandle() const; + + // Pin a table reader with its cache handle. + void Pin(Cache::Handle* handle, TableReader* reader); + + // Release the pinned handle via the given cache and reset state. + void Release(Cache* cache); + + // Test-only: set a reader without a cache handle. + void TEST_SetReader(TableReader* reader) { + reader_.store(reader, std::memory_order_release); + } + + private: + // Internally, we need to ensure reads and writes to reader_ and handle_ are + // properly ordered. handle_ must be written to before reader_ is written to + // with release semantics. handle_ must be read after reader_ is read with + // acquire semantics. + std::atomic reader_; + Cache::Handle* handle_; +}; + // A copyable structure contains information needed to read data from an SST // file. It can contain a pointer to a table reader opened for the file, or // file number and size, which can be used to create a new table reader for it. // The behavior is undefined when a copied of the structure is used when the // file is not in any live version any more. struct FileDescriptor { - // Table reader in table_reader_handle - TableReader* table_reader; + // Fast access to table reader without cache lookup. Marked mutable because + // reads can pin the table reader, but can also be done safely in a + // multi-threaded context. + mutable PinnedTableReader pinned_reader; uint64_t packed_number_and_path_id; uint64_t file_size; // File size in bytes SequenceNumber smallest_seqno; // The smallest seqno in this file @@ -154,7 +197,7 @@ struct FileDescriptor { FileDescriptor(uint64_t number, uint32_t path_id, uint64_t _file_size, SequenceNumber _smallest_seqno, SequenceNumber _largest_seqno) - : table_reader(nullptr), + : pinned_reader(), packed_number_and_path_id(PackFileNumberAndPathId(number, path_id)), file_size(_file_size), smallest_seqno(_smallest_seqno), @@ -163,7 +206,7 @@ struct FileDescriptor { FileDescriptor(const FileDescriptor& fd) { *this = fd; } FileDescriptor& operator=(const FileDescriptor& fd) { - table_reader = fd.table_reader; + pinned_reader = fd.pinned_reader; packed_number_and_path_id = fd.packed_number_and_path_id; file_size = fd.file_size; smallest_seqno = fd.smallest_seqno; @@ -198,9 +241,6 @@ struct FileMetaData { InternalKey smallest; // Smallest internal key served by table InternalKey largest; // Largest internal key served by table - // Needs to be disposed when refs becomes 0. - Cache::Handle* table_reader_handle = nullptr; - FileSampledStats stats; // Stats for compensating deletion entries during compaction @@ -347,9 +387,10 @@ struct FileMetaData { uint64_t TryGetOldestAncesterTime() { if (oldest_ancester_time != kUnknownOldestAncesterTime) { return oldest_ancester_time; - } else if (fd.table_reader != nullptr && - fd.table_reader->GetTableProperties() != nullptr) { - return fd.table_reader->GetTableProperties()->creation_time; + } + TableReader* reader = fd.pinned_reader.Get(); + if (reader != nullptr && reader->GetTableProperties() != nullptr) { + return reader->GetTableProperties()->creation_time; } return kUnknownOldestAncesterTime; } @@ -357,9 +398,10 @@ struct FileMetaData { uint64_t TryGetFileCreationTime() { if (file_creation_time != kUnknownFileCreationTime) { return file_creation_time; - } else if (fd.table_reader != nullptr && - fd.table_reader->GetTableProperties() != nullptr) { - return fd.table_reader->GetTableProperties()->file_creation_time; + } + TableReader* reader = fd.pinned_reader.Get(); + if (reader != nullptr && reader->GetTableProperties() != nullptr) { + return reader->GetTableProperties()->file_creation_time; } return kUnknownFileCreationTime; } @@ -367,10 +409,9 @@ struct FileMetaData { // Tries to get the newest key time from the current file // Falls back on oldest ancestor time of previous (newer) file uint64_t TryGetNewestKeyTime(FileMetaData* prev_file = nullptr) { - if (fd.table_reader != nullptr && - fd.table_reader->GetTableProperties() != nullptr) { - uint64_t newest_key_time = - fd.table_reader->GetTableProperties()->newest_key_time; + TableReader* reader = fd.pinned_reader.Get(); + if (reader != nullptr && reader->GetTableProperties() != nullptr) { + uint64_t newest_key_time = reader->GetTableProperties()->newest_key_time; if (newest_key_time != kUnknownNewestKeyTime) { return newest_key_time; } diff --git a/db/version_edit_handler.h b/db/version_edit_handler.h index 1d4b22e3c1..6d440e86de 100644 --- a/db/version_edit_handler.h +++ b/db/version_edit_handler.h @@ -146,13 +146,13 @@ class VersionEditHandler : public VersionEditHandlerBase { const std::shared_ptr& io_tracer, const ReadOptions& read_options, bool allow_incomplete_valid_version, EpochNumberRequirement epoch_number_requirement = - EpochNumberRequirement::kMustPresent) - : VersionEditHandler(read_only, column_families, version_set, - track_found_and_missing_files, - no_error_if_files_missing, io_tracer, read_options, - /*skip_load_table_files=*/false, - allow_incomplete_valid_version, - epoch_number_requirement) {} + EpochNumberRequirement::kMustPresent, + bool skip_load_table_files = false) + : VersionEditHandler( + read_only, column_families, version_set, + track_found_and_missing_files, no_error_if_files_missing, io_tracer, + read_options, skip_load_table_files, allow_incomplete_valid_version, + epoch_number_requirement) {} ~VersionEditHandler() override {} diff --git a/db/version_set.cc b/db/version_set.cc index 6c9cbc82a1..ec69af461e 100644 --- a/db/version_set.cc +++ b/db/version_set.cc @@ -170,7 +170,7 @@ class FilePicker { if (!search_ended_) { // Prefetch Level 0 table data to avoid cache miss if possible. for (unsigned int i = 0; i < (*level_files_brief_)[0].num_files; ++i) { - auto* r = (*level_files_brief_)[0].files[i].fd.table_reader; + auto* r = (*level_files_brief_)[0].files[i].fd.pinned_reader.Get(); if (r) { r->Prepare(ikey); } @@ -395,7 +395,7 @@ class FilePickerMultiGet { // prefetching. This may not be necessary anymore once we implement // batching in those table readers for (unsigned int i = 0; i < (*level_files_brief_)[0].num_files; ++i) { - auto* r = (*level_files_brief_)[0].files[i].fd.table_reader; + auto* r = (*level_files_brief_)[0].files[i].fd.pinned_reader.Get(); if (r) { for (auto iter = range_.begin(); iter != range_.end(); ++iter) { r->Prepare(iter->ikey); @@ -1298,7 +1298,8 @@ class LevelIterator final : public InternalIterator { /*arena=*/nullptr, skip_filters_, level_, /*max_file_size_for_l0_meta_pin=*/0, smallest_compaction_key, largest_compaction_key, allow_unprepared_value_, &read_seq_, - range_tombstone_iter_); + range_tombstone_iter_, + /*maybe_pin_table_handle=*/true); } // Check if current file being fully within iterate_lower_bound. @@ -2221,7 +2222,7 @@ void Version::GetCreationTimeOfOldestFile(uint64_t* creation_time) { uint64_t oldest_time = std::numeric_limits::max(); for (int level = 0; level < storage_info_.num_non_empty_levels_; level++) { for (FileMetaData* meta : storage_info_.LevelFiles(level)) { - assert(meta->fd.table_reader != nullptr); + assert(meta->fd.pinned_reader.Get() != nullptr); uint64_t file_creation_time = meta->TryGetFileCreationTime(); if (file_creation_time == kUnknownFileCreationTime) { *creation_time = 0; @@ -2357,7 +2358,8 @@ void Version::AddIteratorsForLevel(const ReadOptions& read_options, /*skip_filters=*/false, /*level=*/0, max_file_size_for_l0_meta_pin_, /*smallest_compaction_key=*/nullptr, /*largest_compaction_key=*/nullptr, allow_unprepared_value, - /*range_del_read_seqno=*/nullptr, &tombstone_iter); + /*range_del_read_seqno=*/nullptr, &tombstone_iter, + /*maybe_pin_table_handle=*/true); if (read_options.ignore_range_deletions) { merge_iter_builder->AddIterator(table_iter); } else { @@ -5597,9 +5599,10 @@ VersionSet::~VersionSet() { // Using uncache_aggressiveness=0 overrides any previous marking to // attempt to uncache the file's blocks (which after cleaning up // column families could cause use-after-free) - TableCache::ReleaseObsolete(table_cache_, file.metadata->fd.GetNumber(), - file.metadata->table_reader_handle, - /*uncache_aggressiveness=*/0); + TableCache::ReleaseObsolete( + table_cache_, file.metadata->fd.GetNumber(), + file.metadata->fd.pinned_reader.GetCacheHandle(), + /*uncache_aggressiveness=*/0); file.DeleteMetadata(); } obsolete_files_.clear(); @@ -6537,7 +6540,8 @@ Status VersionSet::Recover( read_only, column_families, const_cast(this), /*track_found_and_missing_files=*/false, no_error_if_files_missing, io_tracer_, read_options, /*allow_incomplete_valid_version=*/false, - EpochNumberRequirement::kMightMissing); + EpochNumberRequirement::kMightMissing, + /*skip_load_table_files=*/db_options_->open_files_async); handler.Iterate(reader, &log_read_status); s = handler.status(); if (s.ok()) { @@ -7856,10 +7860,12 @@ Status VersionSet::VerifyFileMetadata(const ReadOptions& read_options, InternalStats* internal_stats = cfd->internal_stats(); TableCache::TypedHandle* handle = nullptr; + TableReader* table_reader = nullptr; FileMetaData meta_copy = meta; status = table_cache->FindTable( read_options, file_opts, *icmp, meta_copy, &handle, cf_opts, - /*no_io=*/false, internal_stats->GetFileReadHist(level), false, level, + &table_reader, /*no_io=*/false, internal_stats->GetFileReadHist(level), + false, level, /*prefetch_index_and_filter_in_cache*/ false, max_sz_for_l0_meta_pin, meta_copy.temperature); if (handle) { diff --git a/db/version_util.cc b/db/version_util.cc new file mode 100644 index 0000000000..2772e3a6c3 --- /dev/null +++ b/db/version_util.cc @@ -0,0 +1,96 @@ +// Copyright (c) Meta Platforms, Inc. and affiliates. +// This source code is licensed under both the GPLv2 (found in the +// COPYING file in the root directory) and Apache 2.0 License +// (found in the LICENSE.Apache file in the root directory). + +#include "db/version_util.h" + +#include +#include +#include +#include +#include + +#include "db/internal_stats.h" +#include "db/table_cache.h" +#include "port/port.h" +#include "test_util/sync_point.h" + +namespace ROCKSDB_NAMESPACE { + +Status LoadTableHandlersHelper( + const std::vector>& files_meta, + TableCache* table_cache, const FileOptions& file_options, + const InternalKeyComparator& internal_comparator, + InternalStats* internal_stats, int max_threads, + bool prefetch_index_and_filter_in_cache, + const MutableCFOptions& mutable_cf_options, + size_t max_file_size_for_l0_meta_pin, const ReadOptions& read_options, + std::atomic* stop) { + assert(table_cache != nullptr); + + std::atomic next_file_meta_idx(0); + std::atomic has_error(false); + Status ret; + std::function load_handlers_func([&]() { + while (true) { + size_t file_idx = next_file_meta_idx.fetch_add(1); + + if (has_error.load(std::memory_order_relaxed)) { + break; + } + + if (file_idx >= files_meta.size()) { + break; + } + + if (stop != nullptr && stop->load()) { + break; + } + + auto* cache = table_cache->get_cache().get(); + if (cache->GetCapacity() < TableCache::kInfiniteCapacity && + cache->GetUsage() >= cache->GetCapacity()) { + break; + } + + auto* file_meta = files_meta[file_idx].first; + int level = files_meta[file_idx].second; + + TEST_SYNC_POINT_CALLBACK( + "VersionBuilder::Rep::LoadTableHandlers::BeforeFindTable", file_meta); + + TableCache::TypedHandle* handle = nullptr; + TableReader* table_reader = nullptr; + auto status = table_cache->FindTable( + read_options, file_options, internal_comparator, *file_meta, &handle, + mutable_cf_options, &table_reader, false /* no_io */, + internal_stats->GetFileReadHist(level), false /* skip_filters */, + level, prefetch_index_and_filter_in_cache, + max_file_size_for_l0_meta_pin, file_meta->temperature, + true /* pin_table_handle */); + + TEST_SYNC_POINT_CALLBACK( + "VersionBuilder::Rep::LoadTableHandlers::AfterFindTable", &status); + + if (!status.ok()) { + bool expected = false; + if (has_error.compare_exchange_strong(expected, true)) { + ret = status; + } + } + } + }); + + std::vector threads; + for (int i = 1; i < max_threads; i++) { + threads.emplace_back(load_handlers_func); + } + load_handlers_func(); + for (auto& t : threads) { + t.join(); + } + return ret; +} + +} // namespace ROCKSDB_NAMESPACE diff --git a/db/version_util.h b/db/version_util.h index 7219d11c85..a426f94c4e 100644 --- a/db/version_util.h +++ b/db/version_util.h @@ -5,10 +5,34 @@ #pragma once +#include +#include +#include + #include "db/version_set.h" +#include "rocksdb/file_system.h" +#include "rocksdb/options.h" +#include "rocksdb/status.h" namespace ROCKSDB_NAMESPACE { +class TableCache; +struct FileMetaData; +class InternalKeyComparator; +class InternalStats; + +// Load table handlers (i.e., open SST files and populate the table cache) for +// the given set of files. Used during DB open to eagerly warm the table cache. +Status LoadTableHandlersHelper( + const std::vector>& files_meta, + TableCache* table_cache, const FileOptions& file_options, + const InternalKeyComparator& internal_comparator, + InternalStats* internal_stats, int max_threads, + bool prefetch_index_and_filter_in_cache, + const MutableCFOptions& mutable_cf_options, + size_t max_file_size_for_l0_meta_pin, const ReadOptions& read_options, + std::atomic* stop = nullptr); + // Instead of opening a `DB` to perform certain manifest updates, this // uses the underlying `VersionSet` API to read and modify the MANIFEST. This // allows us to use the user's real options, while not having to worry about diff --git a/db_stress_tool/db_stress_common.h b/db_stress_tool/db_stress_common.h index a43a36a99d..ece60788c4 100644 --- a/db_stress_tool/db_stress_common.h +++ b/db_stress_tool/db_stress_common.h @@ -113,6 +113,7 @@ DECLARE_bool(use_write_buffer_manager); DECLARE_double(memtable_prefix_bloom_size_ratio); DECLARE_bool(memtable_whole_key_filtering); DECLARE_int32(open_files); +DECLARE_bool(open_files_async); DECLARE_uint64(compressed_secondary_cache_size); DECLARE_int32(compressed_secondary_cache_numshardbits); DECLARE_int32(secondary_cache_update_interval); diff --git a/db_stress_tool/db_stress_gflags.cc b/db_stress_tool/db_stress_gflags.cc index c17e605f9b..d9d0bab23f 100644 --- a/db_stress_tool/db_stress_gflags.cc +++ b/db_stress_tool/db_stress_gflags.cc @@ -203,6 +203,9 @@ DEFINE_int32(open_files, ROCKSDB_NAMESPACE::Options().max_open_files, "Maximum number of files to keep open at the same time " "(use default if == 0)"); +DEFINE_bool(open_files_async, ROCKSDB_NAMESPACE::Options().open_files_async, + "Options.open_files_async"); + DEFINE_uint64(compressed_secondary_cache_size, 0, "Number of bytes to use as a cache of compressed data." " 0 means use default settings."); diff --git a/db_stress_tool/db_stress_test_base.cc b/db_stress_tool/db_stress_test_base.cc index 2b0f663ada..14c4000528 100644 --- a/db_stress_tool/db_stress_test_base.cc +++ b/db_stress_tool/db_stress_test_base.cc @@ -4429,6 +4429,13 @@ void InitializeOptionsFromFlags( } } options.max_open_files = FLAGS_open_files; + options.open_files_async = FLAGS_open_files_async; + if (FLAGS_open_files_async && !FLAGS_skip_stats_update_on_db_open) { + FLAGS_skip_stats_update_on_db_open = true; + fprintf(stderr, + "open_files_async requires skip_stats_update_on_db_open, " + "enabling it automatically\n"); + } if (FLAGS_statistics) { options.statistics = ROCKSDB_NAMESPACE::CreateDBStatistics(); } diff --git a/include/rocksdb/c.h b/include/rocksdb/c.h index 3bb086f021..7605f253f3 100644 --- a/include/rocksdb/c.h +++ b/include/rocksdb/c.h @@ -1479,6 +1479,10 @@ extern ROCKSDB_LIBRARY_API void rocksdb_options_set_paranoid_checks( rocksdb_options_t*, unsigned char); extern ROCKSDB_LIBRARY_API unsigned char rocksdb_options_get_paranoid_checks( rocksdb_options_t*); +extern ROCKSDB_LIBRARY_API void rocksdb_options_set_open_files_async( + rocksdb_options_t*, unsigned char); +extern ROCKSDB_LIBRARY_API unsigned char rocksdb_options_get_open_files_async( + rocksdb_options_t*); extern ROCKSDB_LIBRARY_API void rocksdb_options_set_db_paths( rocksdb_options_t*, const rocksdb_dbpath_t** path_values, size_t num_paths); extern ROCKSDB_LIBRARY_API void rocksdb_options_set_cf_paths( diff --git a/include/rocksdb/listener.h b/include/rocksdb/listener.h index 1b41ca81f3..ea018e6117 100644 --- a/include/rocksdb/listener.h +++ b/include/rocksdb/listener.h @@ -201,6 +201,7 @@ enum class BackgroundErrorReason { kManifestWrite, kFlushNoWAL, kManifestWriteNoWAL, + kAsyncFileOpen, }; struct WriteStallInfo { diff --git a/include/rocksdb/options.h b/include/rocksdb/options.h index 3c0898fdc8..c38c4f2d0a 100644 --- a/include/rocksdb/options.h +++ b/include/rocksdb/options.h @@ -783,6 +783,25 @@ struct DBOptions { // Default: 16 int max_file_opening_threads = 16; + // 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 + // when max_open_files = -1, as max_open_files != -1 usually has fast open + // times. See also `max_file_opening_threads` and + // `skip_stats_update_on_db_open` to improve file open latency. + // + // Note: This option is currently not compatible with FIFO compaction and + // requires skip_stats_update_on_db_open=true. + // + // Errors will no longer show up in DB::Open, but instead can show up as + // either background errors and/or operations that access the file (e.g. + // reads, compactions). + // + // When false (default), SST files are opened and validated during DB::Open. + // + // Default: false + bool open_files_async = false; + // Once write-ahead logs exceed this size, we will start forcing the flush of // column families whose memtables are backed by the oldest live WAL file // (i.e. the ones that are causing all the space amplification). If set to 0 diff --git a/java/rocksjni/options.cc b/java/rocksjni/options.cc index 3166e66250..f1f5e81d70 100644 --- a/java/rocksjni/options.cc +++ b/java/rocksjni/options.cc @@ -393,6 +393,28 @@ void Java_org_rocksdb_Options_setParanoidChecks(JNIEnv*, jclass, jlong jhandle, static_cast(paranoid_checks); } +/* + * Class: org_rocksdb_Options + * Method: openFilesAsync + * Signature: (J)Z + */ +jboolean Java_org_rocksdb_Options_openFilesAsync(JNIEnv*, jclass, + jlong jhandle) { + return reinterpret_cast(jhandle) + ->open_files_async; +} + +/* + * Class: org_rocksdb_Options + * Method: setOpenFilesAsync + * Signature: (JZ)V + */ +void Java_org_rocksdb_Options_setOpenFilesAsync(JNIEnv*, jclass, jlong jhandle, + jboolean open_files_async) { + reinterpret_cast(jhandle)->open_files_async = + static_cast(open_files_async); +} + /* * Class: org_rocksdb_Options * Method: setEnv @@ -5996,6 +6018,29 @@ jboolean Java_org_rocksdb_DBOptions_paranoidChecks(JNIEnv*, jclass, ->paranoid_checks; } +/* + * Class: org_rocksdb_DBOptions + * Method: setOpenFilesAsync + * Signature: (JZ)V + */ +void Java_org_rocksdb_DBOptions_setOpenFilesAsync(JNIEnv*, jclass, + jlong jhandle, + jboolean open_files_async) { + reinterpret_cast(jhandle)->open_files_async = + static_cast(open_files_async); +} + +/* + * Class: org_rocksdb_DBOptions + * Method: openFilesAsync + * Signature: (J)Z + */ +jboolean Java_org_rocksdb_DBOptions_openFilesAsync(JNIEnv*, jclass, + jlong jhandle) { + return reinterpret_cast(jhandle) + ->open_files_async; +} + /* * Class: org_rocksdb_DBOptions * Method: setRateLimiter diff --git a/java/src/main/java/org/rocksdb/DBOptions.java b/java/src/main/java/org/rocksdb/DBOptions.java index 12f5d4913c..c2661b1621 100644 --- a/java/src/main/java/org/rocksdb/DBOptions.java +++ b/java/src/main/java/org/rocksdb/DBOptions.java @@ -185,6 +185,19 @@ public class DBOptions extends RocksObject return paranoidChecks(nativeHandle_); } + @Override + public DBOptions setOpenFilesAsync(final boolean openFilesAsync) { + assert (isOwningHandle()); + setOpenFilesAsync(nativeHandle_, openFilesAsync); + return this; + } + + @Override + public boolean openFilesAsync() { + assert (isOwningHandle()); + return openFilesAsync(nativeHandle_); + } + @Override public DBOptions setEnv(final Env env) { setEnv(nativeHandle_, env.nativeHandle_); @@ -1244,6 +1257,8 @@ public class DBOptions extends RocksObject private static native boolean errorIfExists(long handle); private static native void setParanoidChecks(long handle, boolean paranoidChecks); private static native boolean paranoidChecks(long handle); + private static native void setOpenFilesAsync(long handle, boolean openFilesAsync); + private static native boolean openFilesAsync(long handle); private static native void setRateLimiter(long handle, long rateLimiterHandle); private static native void setSstFileManager(final long handle, final long sstFileManagerHandle); private static native void setLogger( diff --git a/java/src/main/java/org/rocksdb/DBOptionsInterface.java b/java/src/main/java/org/rocksdb/DBOptionsInterface.java index f40fc1a25c..5d04a1dc07 100644 --- a/java/src/main/java/org/rocksdb/DBOptionsInterface.java +++ b/java/src/main/java/org/rocksdb/DBOptionsInterface.java @@ -145,6 +145,26 @@ public interface DBOptionsInterface> { */ boolean paranoidChecks(); + /** + * 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. + * + * Default: false + * + * @param openFilesAsync true to enable async file opening. + * @return the reference to the current option. + */ + T setOpenFilesAsync(boolean openFilesAsync); + + /** + * If true, SST files are opened and validated asynchronously in the + * background after DB::Open returns. + * + * @return a boolean indicating whether async file opening is enabled. + */ + boolean openFilesAsync(); + /** * Use to control write rate of flush and compaction. Flush has higher * priority than compaction. Rate limiting is disabled if nullptr. diff --git a/java/src/main/java/org/rocksdb/Options.java b/java/src/main/java/org/rocksdb/Options.java index 3e7bf28405..969f9ec2ac 100644 --- a/java/src/main/java/org/rocksdb/Options.java +++ b/java/src/main/java/org/rocksdb/Options.java @@ -324,6 +324,19 @@ public class Options extends RocksObject return this; } + @Override + public boolean openFilesAsync() { + assert (isOwningHandle()); + return openFilesAsync(nativeHandle_); + } + + @Override + public Options setOpenFilesAsync(final boolean openFilesAsync) { + assert (isOwningHandle()); + setOpenFilesAsync(nativeHandle_, openFilesAsync); + return this; + } + @Override public int maxOpenFiles() { assert(isOwningHandle()); @@ -2137,6 +2150,8 @@ public class Options extends RocksObject private static native boolean errorIfExists(long handle); private static native void setParanoidChecks(long handle, boolean paranoidChecks); private static native boolean paranoidChecks(long handle); + private static native void setOpenFilesAsync(long handle, boolean openFilesAsync); + private static native boolean openFilesAsync(long handle); private static native void setRateLimiter(long handle, long rateLimiterHandle); private static native void setSstFileManager(final long handle, final long sstFileManagerHandle); private static native void setLogger( diff --git a/options/db_options.cc b/options/db_options.cc index 2384355264..0cefcb7be9 100644 --- a/options/db_options.cc +++ b/options/db_options.cc @@ -230,6 +230,10 @@ static std::unordered_map {offsetof(struct ImmutableDBOptions, paranoid_checks), OptionType::kBoolean, OptionVerificationType::kNormal, OptionTypeFlags::kNone}}, + {"open_files_async", + {offsetof(struct ImmutableDBOptions, open_files_async), + OptionType::kBoolean, OptionVerificationType::kNormal, + OptionTypeFlags::kNone}}, {"flush_verify_memtable_count", {offsetof(struct ImmutableDBOptions, flush_verify_memtable_count), OptionType::kBoolean, OptionVerificationType::kNormal, @@ -717,6 +721,7 @@ ImmutableDBOptions::ImmutableDBOptions(const DBOptions& options) create_missing_column_families(options.create_missing_column_families), error_if_exists(options.error_if_exists), paranoid_checks(options.paranoid_checks), + open_files_async(options.open_files_async), flush_verify_memtable_count(options.flush_verify_memtable_count), compaction_verify_record_count(options.compaction_verify_record_count), track_and_verify_wals_in_manifest( @@ -814,6 +819,8 @@ void ImmutableDBOptions::Dump(Logger* log) const { create_if_missing); ROCKS_LOG_HEADER(log, " Options.paranoid_checks: %d", paranoid_checks); + ROCKS_LOG_HEADER(log, " Options.open_files_async: %d", + open_files_async); ROCKS_LOG_HEADER(log, " Options.flush_verify_memtable_count: %d", flush_verify_memtable_count); ROCKS_LOG_HEADER(log, " Options.compaction_verify_record_count: %d", diff --git a/options/db_options.h b/options/db_options.h index cc978d907d..2192e099c3 100644 --- a/options/db_options.h +++ b/options/db_options.h @@ -24,6 +24,7 @@ struct ImmutableDBOptions { bool create_missing_column_families; bool error_if_exists; bool paranoid_checks; + bool open_files_async; bool flush_verify_memtable_count; bool compaction_verify_record_count; bool track_and_verify_wals_in_manifest; diff --git a/options/options_helper.cc b/options/options_helper.cc index addada94f9..50fe1da95a 100644 --- a/options/options_helper.cc +++ b/options/options_helper.cc @@ -65,6 +65,7 @@ void BuildDBOptions(const ImmutableDBOptions& immutable_db_options, immutable_db_options.create_missing_column_families; options.error_if_exists = immutable_db_options.error_if_exists; options.paranoid_checks = immutable_db_options.paranoid_checks; + options.open_files_async = immutable_db_options.open_files_async; options.flush_verify_memtable_count = immutable_db_options.flush_verify_memtable_count; options.compaction_verify_record_count = diff --git a/options/options_settable_test.cc b/options/options_settable_test.cc index b2c7444912..d63746f337 100644 --- a/options/options_settable_test.cc +++ b/options/options_settable_test.cc @@ -414,6 +414,7 @@ TEST_F(OptionsSettableTest, DBOptionsAllFieldsSettable) { "db_log_dir=path/to/db_log_dir;" "writable_file_max_buffer_size=1048576;" "paranoid_checks=true;" + "open_files_async=true;" "flush_verify_memtable_count=true;" "compaction_verify_record_count=true;" "track_and_verify_wals_in_manifest=true;" diff --git a/src.mk b/src.mk index add3ce44ff..a2d47125a4 100644 --- a/src.mk +++ b/src.mk @@ -96,6 +96,7 @@ LIB_SOURCES = \ db/version_edit.cc \ db/version_edit_handler.cc \ db/version_set.cc \ + db/version_util.cc \ db/wal_edit.cc \ db/wal_manager.cc \ db/wide/wide_column_serialization.cc \ diff --git a/tools/db_bench_tool.cc b/tools/db_bench_tool.cc index 73abb2a750..a382fdbdbf 100644 --- a/tools/db_bench_tool.cc +++ b/tools/db_bench_tool.cc @@ -1642,6 +1642,12 @@ DEFINE_bool(print_malloc_stats, false, DEFINE_bool(disable_auto_compactions, false, "Do not auto trigger compactions"); +DEFINE_bool(open_files_async, false, + "Open SST files asynchronously during DB open"); + +DEFINE_bool(skip_stats_update_on_db_open, false, + "Skip loading table properties to update stats during DB open"); + DEFINE_uint64(wal_ttl_seconds, 0, "Set the TTL for the WAL Files in seconds."); DEFINE_uint64(wal_size_limit_MB, 0, "Set the size limit for the WAL Files in MB."); @@ -4870,6 +4876,14 @@ class Benchmark { options.table_cache_numshardbits = FLAGS_table_cache_numshardbits; options.max_compaction_bytes = FLAGS_max_compaction_bytes; options.disable_auto_compactions = FLAGS_disable_auto_compactions; + options.open_files_async = FLAGS_open_files_async; + if (FLAGS_open_files_async && !FLAGS_skip_stats_update_on_db_open) { + FLAGS_skip_stats_update_on_db_open = true; + fprintf(stderr, + "open_files_async requires skip_stats_update_on_db_open, " + "enabling it automatically\n"); + } + options.skip_stats_update_on_db_open = FLAGS_skip_stats_update_on_db_open; options.optimize_filters_for_hits = FLAGS_optimize_filters_for_hits; options.paranoid_checks = FLAGS_paranoid_checks; options.force_consistency_checks = FLAGS_force_consistency_checks; diff --git a/tools/db_crashtest.py b/tools/db_crashtest.py index f795a248be..3c77b0c794 100644 --- a/tools/db_crashtest.py +++ b/tools/db_crashtest.py @@ -193,6 +193,7 @@ default_params = { # overwrites. "nooverwritepercent": 1, "open_files": lambda: random.choice([-1, -1, 100, 500000]), + "open_files_async": lambda: random.choice([0, 1]), "optimize_filters_for_memory": lambda: random.randint(0, 1), "partition_filters": lambda: random.randint(0, 1), "partition_pinning": lambda: random.randint(0, 3), @@ -992,6 +993,8 @@ def finalize_and_sanitize(src_params): # now assertion failures are triggered. dest_params["compaction_ttl"] = 0 dest_params["periodic_compaction_seconds"] = 0 + # FIFO compaction is not supported with open_files_async + dest_params["open_files_async"] = 0 # Disable irrelevant tiering options dest_params["preclude_last_level_data_seconds"] = 0 dest_params["last_level_temperature"] = "kUnknown" @@ -1286,6 +1289,11 @@ def finalize_and_sanitize(src_params): # which are not updated if skip_stats_update_on_db_open is true dest_params["skip_stats_update_on_db_open"] = 0 + # open_files_async requires skip_stats_update_on_db_open to avoid + # synchronous I/O in UpdateAccumulatedStats during DB open + if dest_params.get("skip_stats_update_on_db_open", 0) == 0: + dest_params["open_files_async"] = 0 + # inplace update and key checksum verification during seek would cause race condition # Therefore, when inplace_update_support is enabled, disable memtable_veirfy_per_key_checksum_on_seek if dest_params["inplace_update_support"] == 1: diff --git a/unreleased_history/new_features/async_file_open b/unreleased_history/new_features/async_file_open new file mode 100644 index 0000000000..28e32e09dc --- /dev/null +++ b/unreleased_history/new_features/async_file_open @@ -0,0 +1 @@ +Add a new option `open_files_async`. The existing behavior is on DB open, we open all sst files and do basic validations. For very large DBs on remote filesystems with many ssts, this may take very long. This option performs these validations instead in the background. Open errors found by this async background task are surfaced as a new background error kAsyncFileOpen.