diff --git a/db/column_family_test.cc b/db/column_family_test.cc index 02ed3b0049..d34169868f 100644 --- a/db/column_family_test.cc +++ b/db/column_family_test.cc @@ -3179,6 +3179,8 @@ TEST_P(ColumnFamilyTest, IteratorCloseWALFile1) { SpecialEnv env(Env::Default()); db_options_.env = &env; db_options_.max_background_flushes = 1; + // When this option is removed, the test will need re-engineering + db_options_.background_close_inactive_wals = true; column_family_options_.memtable_factory.reset( test::NewSpecialSkipListFactory(2)); Open(); @@ -3231,6 +3233,8 @@ TEST_P(ColumnFamilyTest, IteratorCloseWALFile2) { env.SetBackgroundThreads(2, Env::HIGH); db_options_.env = &env; db_options_.max_background_flushes = 1; + // When this option is removed, the test will need re-engineering + db_options_.background_close_inactive_wals = true; column_family_options_.memtable_factory.reset( test::NewSpecialSkipListFactory(2)); Open(); @@ -3288,6 +3292,8 @@ TEST_P(ColumnFamilyTest, ForwardIteratorCloseWALFile) { env.SetBackgroundThreads(2, Env::HIGH); db_options_.env = &env; db_options_.max_background_flushes = 1; + // When this option is removed, the test will need re-engineering + db_options_.background_close_inactive_wals = true; column_family_options_.memtable_factory.reset( test::NewSpecialSkipListFactory(3)); column_family_options_.level0_file_num_compaction_trigger = 2; diff --git a/db/db_filesnapshot.cc b/db/db_filesnapshot.cc index 526e8fcd22..f22c77e0d0 100644 --- a/db/db_filesnapshot.cc +++ b/db/db_filesnapshot.cc @@ -401,7 +401,7 @@ Status DBImpl::GetLiveFilesStorageInfo( // fully synced. Although the output DB of a Checkpoint or Backup needs // to be fully synced on return, we don't strictly need to sync this // DB (the input DB). If we allow Checkpoint to hard link an inactive - // WAL that isn't fully synced, that could result in an unsufficiently + // WAL that isn't fully synced, that could result in an insufficiently // sync-ed Checkpoint. Here we get the set of WALs that are potentially // unsynced or still being written to, to prevent them from being hard // linked. Enforcing max_log_num from above ensures any new WALs after diff --git a/db/db_impl/db_impl.cc b/db/db_impl/db_impl.cc index aa2ecd8801..2dc18f1522 100644 --- a/db/db_impl/db_impl.cc +++ b/db/db_impl/db_impl.cc @@ -1550,6 +1550,7 @@ bool DBImpl::WALBufferIsEmpty() { } Status DBImpl::GetOpenWalSizes(std::map& number_to_size) { + assert(number_to_size.empty()); InstrumentedMutexLock l(&log_write_mutex_); for (auto& log : logs_) { auto* open_file = log.writer->file(); @@ -1627,6 +1628,7 @@ IOStatus DBImpl::SyncWalImpl(bool include_current_wal, RecordTick(stats_, WAL_FILE_SYNCED); IOOptions opts; IOStatus io_s = WritableFileWriter::PrepareIOOptions(write_options, opts); + std::list wals_internally_closed; if (io_s.ok()) { for (log::Writer* log : wals_to_sync) { if (job_context) { @@ -1647,15 +1649,22 @@ IOStatus DBImpl::SyncWalImpl(bool include_current_wal, if (!io_s.ok()) { break; } - // Normally the log file is closed when purging obsolete file, but if - // log recycling is enabled, the log file is closed here so that it - // can be reused. + // WALs can be closed when purging obsolete files, but if recycling is + // enabled, the log file is closed here so that it can be reused. And + // immediate closure here upon final sync makes it easier to guarantee + // that Checkpoint doesn't LinkFile on a WAL still open for write, which + // might be unsupported for some FileSystem implementations. Close here + // should be inexpensive because flush and sync are done, so the kill + // switch background_close_inactive_wals is expected to be removed in + // the future. if (log->get_log_number() < maybe_active_number && - immutable_db_options_.recycle_log_file_num > 0) { + (immutable_db_options_.recycle_log_file_num > 0 || + !immutable_db_options_.background_close_inactive_wals)) { if (error_recovery_in_prog) { log->file()->reset_seen_error(); } - io_s = log->Close(write_options); + io_s = log->file()->Close(opts); + wals_internally_closed.push_back(log); if (!io_s.ok()) { break; } @@ -1684,6 +1693,12 @@ IOStatus DBImpl::SyncWalImpl(bool include_current_wal, } { InstrumentedMutexLock l(&log_write_mutex_); + for (auto* wal : wals_internally_closed) { + // We can only modify the state of log::Writer under the mutex + bool was_closed = wal->PublishIfClosed(); + assert(was_closed); + (void)was_closed; + } if (io_s.ok()) { MarkLogsSynced(up_to_number, need_wal_dir_sync, synced_wals); } else { @@ -1808,16 +1823,19 @@ void DBImpl::MarkLogsSynced(uint64_t up_to, bool synced_dir, wal.GetPreSyncSize() > 0) { synced_wals->AddWal(wal.number, WalMetadata(wal.GetPreSyncSize())); } - // Check if the file has been closed, i.e wal.writer->file() == nullptr - // which can happen if log recycling is enabled, or if all the data in - // the log has been synced + // Reclaim closed WALs (wal.writer->file() == nullptr), and if we don't + // need to close before that (background_close_inactive_wals) we can + // opportunistically reclaim WALs that happen to be fully synced. + // (Probably not worth extra code and mutex release to opportunistically + // close WALs that became eligible since last holding the mutex. + // FindObsoleteFiles can take care of it.) if (wal.writer->file() == nullptr || - wal.GetPreSyncSize() == wal.writer->file()->GetFlushedSize()) { + (immutable_db_options_.background_close_inactive_wals && + wal.GetPreSyncSize() == wal.writer->file()->GetFlushedSize())) { // Fully synced logs_to_free_.push_back(wal.ReleaseWriter()); it = logs_.erase(it); } else { - assert(wal.GetPreSyncSize() < wal.writer->file()->GetFlushedSize()); wal.FinishSync(); ++it; } diff --git a/db/db_impl/db_impl_files.cc b/db/db_impl/db_impl_files.cc index 8ade307316..dd4bf411cd 100644 --- a/db/db_impl/db_impl_files.cc +++ b/db/db_impl/db_impl_files.cc @@ -312,9 +312,26 @@ void DBImpl::FindObsoleteFiles(JobContext* job_context, bool force, // logs_ could have changed while we were waiting. continue; } - // This WAL file is not live, so it's OK if we never sync the rest of it - // or if we close it *after* removing from `logs_`. If it's already - // closed, then it's been fully synced. + // This WAL file is not live, so it's OK if we never sync the rest of it. + // If it's already closed, then it's been fully synced. If + // !background_close_inactive_wals then we need to Close it before + // removing from logs_ but not blocking while holding log_write_mutex_. + if (!immutable_db_options_.background_close_inactive_wals && + log.writer->file()) { + // We are taking ownership of and pinning the front entry, so we can + // expect it to be the same after releasing and re-acquiring the lock + log.PrepareForSync(); + log_write_mutex_.Unlock(); + // TODO: maybe check the return value of Close. + // TODO: plumb Env::IOActivity, Env::IOPriority + auto s = log.writer->file()->Close({}); + s.PermitUncheckedError(); + log_write_mutex_.Lock(); + log.writer->PublishIfClosed(); + assert(&log == &logs_.front()); + log.FinishSync(); + log_sync_cv_.SignalAll(); + } logs_to_free_.push_back(log.ReleaseWriter()); logs_.pop_front(); } @@ -506,7 +523,7 @@ void DBImpl::PurgeObsoleteFiles(JobContext& state, bool schedule_only) { for (const auto w : state.logs_to_free) { // TODO: maybe check the return value of Close. // TODO: plumb Env::IOActivity, Env::IOPriority - auto s = w->Close(WriteOptions()); + auto s = w->Close({}); s.PermitUncheckedError(); } diff --git a/db/log_writer.cc b/db/log_writer.cc index 2cd6bbd788..5c532859e5 100644 --- a/db/log_writer.cc +++ b/db/log_writer.cc @@ -74,6 +74,15 @@ IOStatus Writer::Close(const WriteOptions& write_options) { return s; } +bool Writer::PublishIfClosed() { + if (dest_->IsClosed()) { + dest_.reset(); + return true; + } else { + return false; + } +} + IOStatus Writer::AddRecord(const WriteOptions& write_options, const Slice& slice) { if (dest_->seen_error()) { diff --git a/db/log_writer.h b/db/log_writer.h index 48fd3db7c2..7cae52dd51 100644 --- a/db/log_writer.h +++ b/db/log_writer.h @@ -107,6 +107,11 @@ class Writer { IOStatus Close(const WriteOptions& write_options); + // If closing the writer through file(), call this afterwards to modify + // this object's state to reflect that. Returns true if the destination file + // has been closed. If it hasn't been closed, returns false with no change. + bool PublishIfClosed(); + bool BufferIsEmpty(); size_t TEST_block_offset() const { return block_offset_; } diff --git a/file/writable_file_writer.h b/file/writable_file_writer.h index ff1302d711..a2cdfdf39f 100644 --- a/file/writable_file_writer.h +++ b/file/writable_file_writer.h @@ -278,7 +278,9 @@ class WritableFileWriter { bool use_direct_io() { return writable_file_->use_direct_io(); } - bool BufferIsEmpty() { return buf_.CurrentSize() == 0; } + bool BufferIsEmpty() const { return buf_.CurrentSize() == 0; } + + bool IsClosed() const { return writable_file_.get() == nullptr; } void TEST_SetFileChecksumGenerator( FileChecksumGenerator* checksum_generator) { diff --git a/include/rocksdb/options.h b/include/rocksdb/options.h index b023a86475..06c9248c7c 100644 --- a/include/rocksdb/options.h +++ b/include/rocksdb/options.h @@ -1347,6 +1347,15 @@ struct DBOptions { // the WAL is read. CompressionType wal_compression = kNoCompression; + // Set to true to re-instate an old behavior of keeping complete, synced WAL + // files open for write until they are collected for deletion by a + // background thread. This should not be needed unless there is a + // performance issue with file Close(), but setting it to true means that + // Checkpoint might call LinkFile on a WAL still open for write, which might + // be unsupported on some FileSystem implementations. As this is intended as + // a temporary kill switch, it is already DEPRECATED. + bool background_close_inactive_wals = false; + // If true, RocksDB supports flushing multiple column families and committing // their results atomically to MANIFEST. Note that it is not // necessary to set atomic_flush to true if WAL is always enabled since WAL diff --git a/options/db_options.cc b/options/db_options.cc index 6da12a1215..8eb28c1edc 100644 --- a/options/db_options.cc +++ b/options/db_options.cc @@ -388,6 +388,10 @@ static std::unordered_map {offsetof(struct ImmutableDBOptions, wal_compression), OptionType::kCompressionType, OptionVerificationType::kNormal, OptionTypeFlags::kNone}}, + {"background_close_inactive_wals", + {offsetof(struct ImmutableDBOptions, background_close_inactive_wals), + OptionType::kBoolean, OptionVerificationType::kNormal, + OptionTypeFlags::kNone}}, {"seq_per_batch", {0, OptionType::kBoolean, OptionVerificationType::kDeprecated, OptionTypeFlags::kNone}}, @@ -755,6 +759,7 @@ ImmutableDBOptions::ImmutableDBOptions(const DBOptions& options) two_write_queues(options.two_write_queues), manual_wal_flush(options.manual_wal_flush), wal_compression(options.wal_compression), + background_close_inactive_wals(options.background_close_inactive_wals), atomic_flush(options.atomic_flush), avoid_unnecessary_blocking_io(options.avoid_unnecessary_blocking_io), persist_stats_to_disk(options.persist_stats_to_disk), @@ -921,6 +926,9 @@ void ImmutableDBOptions::Dump(Logger* log) const { manual_wal_flush); ROCKS_LOG_HEADER(log, " Options.wal_compression: %d", wal_compression); + ROCKS_LOG_HEADER(log, + " Options.background_close_inactive_wals: %d", + background_close_inactive_wals); ROCKS_LOG_HEADER(log, " Options.atomic_flush: %d", atomic_flush); ROCKS_LOG_HEADER(log, " Options.avoid_unnecessary_blocking_io: %d", diff --git a/options/db_options.h b/options/db_options.h index ff7ddc880f..4bd80cbab0 100644 --- a/options/db_options.h +++ b/options/db_options.h @@ -84,6 +84,7 @@ struct ImmutableDBOptions { bool two_write_queues; bool manual_wal_flush; CompressionType wal_compression; + bool background_close_inactive_wals; bool atomic_flush; bool avoid_unnecessary_blocking_io; bool persist_stats_to_disk; diff --git a/options/options_settable_test.cc b/options/options_settable_test.cc index b71fb0a265..1fec8c20ab 100644 --- a/options/options_settable_test.cc +++ b/options/options_settable_test.cc @@ -353,6 +353,7 @@ TEST_F(OptionsSettableTest, DBOptionsAllFieldsSettable) { "two_write_queues=false;" "manual_wal_flush=false;" "wal_compression=kZSTD;" + "background_close_inactive_wals=true;" "seq_per_batch=false;" "atomic_flush=false;" "avoid_unnecessary_blocking_io=false;" diff --git a/unreleased_history/behavior_changes/aggressive_close_inactive_wals.md b/unreleased_history/behavior_changes/aggressive_close_inactive_wals.md new file mode 100644 index 0000000000..c75f598496 --- /dev/null +++ b/unreleased_history/behavior_changes/aggressive_close_inactive_wals.md @@ -0,0 +1 @@ +* Inactive WALs are immediately closed upon being fully sync-ed rather than in a background thread. This is to ensure LinkFile() is not called on files still open for write, which might not be supported by some FileSystem implementations. This should not be a performance issue, but an opt-out is available with with new DB option `background_close_inactive_wals`. diff --git a/utilities/checkpoint/checkpoint_test.cc b/utilities/checkpoint/checkpoint_test.cc index 111e81f322..3b867bfa6d 100644 --- a/utilities/checkpoint/checkpoint_test.cc +++ b/utilities/checkpoint/checkpoint_test.cc @@ -762,27 +762,38 @@ TEST_F(CheckpointTest, CheckpointWithParallelWrites) { class CheckpointTestWithWalParams : public CheckpointTest, - public testing::WithParamInterface> { + public testing::WithParamInterface< + std::tuple> { public: uint64_t GetLogSizeForFlush() { return std::get<0>(GetParam()); } bool GetWalsInManifest() { return std::get<1>(GetParam()); } bool GetManualWalFlush() { return std::get<2>(GetParam()); } + bool GetBackgroundCloseInactiveWals() { return std::get<3>(GetParam()); } }; -INSTANTIATE_TEST_CASE_P(CheckpointTestWithWalParams, - CheckpointTestWithWalParams, +INSTANTIATE_TEST_CASE_P(NormalWalParams, CheckpointTestWithWalParams, ::testing::Combine(::testing::Values(0U, 100000000U), - ::testing::Bool(), - ::testing::Bool())); + ::testing::Bool(), ::testing::Bool(), + ::testing::Values(false))); + +INSTANTIATE_TEST_CASE_P(DeprecatedWalParams, CheckpointTestWithWalParams, + ::testing::Values(std::make_tuple(100000000U, true, + false, true))); TEST_P(CheckpointTestWithWalParams, CheckpointWithUnsyncedDataDropped) { Options options = CurrentOptions(); options.max_write_buffer_number = 4; options.track_and_verify_wals_in_manifest = GetWalsInManifest(); options.manual_wal_flush = GetManualWalFlush(); + options.background_close_inactive_wals = GetBackgroundCloseInactiveWals(); auto fault_fs = std::make_shared(FileSystem::Default()); std::unique_ptr fault_fs_env(NewCompositeEnv(fault_fs)); + if (options.background_close_inactive_wals) { + // Disable this hygiene check when the fix is disabled + fault_fs->SetAllowLinkOpenFile(); + } + options.env = fault_fs_env.get(); Reopen(options); ASSERT_OK(Put("key1", "val1")); diff --git a/utilities/fault_injection_fs.cc b/utilities/fault_injection_fs.cc index 4244b0a4eb..1cdc1256f2 100644 --- a/utilities/fault_injection_fs.cc +++ b/utilities/fault_injection_fs.cc @@ -17,6 +17,7 @@ #include "utilities/fault_injection_fs.h" #include +#include #include #include @@ -886,6 +887,12 @@ IOStatus FaultInjectionTestFS::LinkFile(const std::string& s, if (io_s.ok()) { { MutexLock l(&mutex_); + if (!allow_link_open_file_ && + open_managed_files_.find(s) != open_managed_files_.end()) { + fprintf(stderr, "Attempt to LinkFile while open for write: %s\n", + s.c_str()); + abort(); + } if (db_file_state_.find(s) != db_file_state_.end()) { db_file_state_[t] = db_file_state_[s]; } diff --git a/utilities/fault_injection_fs.h b/utilities/fault_injection_fs.h index b49a69d2c3..ea76bf5386 100644 --- a/utilities/fault_injection_fs.h +++ b/utilities/fault_injection_fs.h @@ -203,6 +203,7 @@ class FaultInjectionTestFS : public FileSystemWrapper { filesystem_active_(true), filesystem_writable_(false), read_unsynced_data_(true), + allow_link_open_file_(false), thread_local_error_(new ThreadLocalPtr(DeleteThreadLocalErrorContext)), enable_write_error_injection_(false), enable_metadata_write_error_injection_(false), @@ -365,6 +366,14 @@ class FaultInjectionTestFS : public FileSystemWrapper { read_unsynced_data_ = read_unsynced_data; } bool ReadUnsyncedData() const { return read_unsynced_data_; } + + // FaultInjectionTestFS normally includes a hygiene check for FileSystem + // implementations that only support LinkFile() on closed files (not open + // for write). Setting this to true bypasses the check. + void SetAllowLinkOpenFile(bool allow_link_open_file = true) { + allow_link_open_file_ = allow_link_open_file; + } + void AssertNoOpenFile() { assert(open_managed_files_.empty()); } IOStatus GetError() { return error_; } @@ -565,6 +574,7 @@ class FaultInjectionTestFS : public FileSystemWrapper { bool filesystem_writable_; // Bypass FaultInjectionTestFS and go directly // to underlying FS for writable files bool read_unsynced_data_; // See SetReadUnsyncedData() + bool allow_link_open_file_; // See SetAllowLinkOpenFile() IOStatus error_; enum ErrorType : int {