From 2a0ee4ddd841306f1cc4c6f79ca25f97012a6cf9 Mon Sep 17 00:00:00 2001 From: Peter Dillinger Date: Fri, 11 Apr 2025 10:08:29 -0700 Subject: [PATCH] Refactor wal related naming and more (#13490) Summary: * Clarify in API comments which `log_` options in DBOptions relate to WALs, info log, and/or manifest files. * Rename a bunch of "log" things to "wal" for clarity, especially in DBImpl. (More to go, especially some more challenging cases like `DBImpl::logs_`, but a step in the right direction IMHO) * Simplify DBImpl ctor by moving constant initializers to field definitions. * Use RelaxedAtomic for (renamed) `wals_total_size_` Pull Request resolved: https://github.com/facebook/rocksdb/pull/13490 Test Plan: existing tests Reviewed By: cbi42 Differential Revision: D71939382 Pulled By: pdillinger fbshipit-source-id: 852f4737eca83e6ad653010cc197ad1b6e6bae13 --- db/column_family_test.cc | 2 +- db/db_filesnapshot.cc | 8 +- db/db_impl/db_impl.cc | 93 ++----- db/db_impl/db_impl.h | 260 +++++++++--------- db/db_impl/db_impl_compaction_flush.cc | 8 +- db/db_impl/db_impl_debug.cc | 6 +- db/db_impl/db_impl_files.cc | 70 ++--- db/db_impl/db_impl_open.cc | 36 +-- db/db_impl/db_impl_write.cc | 260 +++++++++--------- db/db_kv_checksum_test.cc | 16 +- db/db_test.cc | 4 +- db/db_wal_test.cc | 4 +- db/db_write_test.cc | 2 +- db/job_context.h | 14 +- db/memtable.h | 6 +- db/write_thread.h | 6 +- include/rocksdb/db.h | 6 +- include/rocksdb/options.h | 17 +- include/rocksdb/utilities/stackable_db.h | 4 +- monitoring/stats_history_test.cc | 2 +- .../transactions/pessimistic_transaction.cc | 8 +- .../transactions/write_unprepared_txn.cc | 2 +- 22 files changed, 407 insertions(+), 427 deletions(-) diff --git a/db/column_family_test.cc b/db/column_family_test.cc index 1ec6cb81f2..3a2ca06176 100644 --- a/db/column_family_test.cc +++ b/db/column_family_test.cc @@ -2175,7 +2175,7 @@ TEST_P(ColumnFamilyTest, FlushStaleColumnFamilies) { ASSERT_TRUE(has_cf2_sst); ASSERT_OK(Flush(0)); - ASSERT_EQ(0, dbfull()->TEST_total_log_size()); + ASSERT_EQ(0, dbfull()->TEST_wals_total_size()); Close(); } diff --git a/db/db_filesnapshot.cc b/db/db_filesnapshot.cc index 9b8f602c03..d524487750 100644 --- a/db/db_filesnapshot.cc +++ b/db/db_filesnapshot.cc @@ -183,14 +183,14 @@ Status DBImpl::GetSortedWalFilesImpl(VectorWalPtr& files, bool need_seqnos) { return s; } -Status DBImpl::GetCurrentWalFile(std::unique_ptr* current_log_file) { +Status DBImpl::GetCurrentWalFile(std::unique_ptr* current_wal_file) { uint64_t current_logfile_number; { InstrumentedMutexLock l(&mutex_); - current_logfile_number = logfile_number_; + current_logfile_number = cur_wal_number_; } - return wal_manager_.GetLiveWalFile(current_logfile_number, current_log_file); + return wal_manager_.GetLiveWalFile(current_logfile_number, current_wal_file); } Status DBImpl::GetLiveFilesStorageInfo( @@ -330,7 +330,7 @@ Status DBImpl::GetLiveFilesStorageInfo( const uint64_t options_size = versions_->options_file_size_; const uint64_t min_log_num = MinLogNumberToKeep(); // Ensure consistency with manifest for track_and_verify_wals_in_manifest - const uint64_t max_log_num = logfile_number_; + const uint64_t max_log_num = cur_wal_number_; mutex_.Unlock(); diff --git a/db/db_impl/db_impl.cc b/db/db_impl/db_impl.cc index 35dc300b9f..571e1e2675 100644 --- a/db/db_impl/db_impl.cc +++ b/db/db_impl/db_impl.cc @@ -169,7 +169,6 @@ DBImpl::DBImpl(const DBOptions& options, const std::string& dbname, bool read_only) : dbname_(dbname), own_info_log_(options.info_log == nullptr), - init_logger_creation_s_(), initial_db_options_(SanitizeOptions(dbname, options, read_only, &init_logger_creation_s_)), env_(initial_db_options_.env), @@ -185,7 +184,6 @@ DBImpl::DBImpl(const DBOptions& options, const std::string& dbname, mutex_(stats_, immutable_db_options_.clock, DB_MUTEX_WAIT_MICROS, immutable_db_options_.use_adaptive_mutex), #endif // COERCE_CONTEXT_SWITCH - default_cf_handle_(nullptr), error_handler_(this, immutable_db_options_, &mutex_), event_logger_(immutable_db_options_.info_log.get()), max_total_in_memory_state_(0), @@ -194,45 +192,15 @@ DBImpl::DBImpl(const DBOptions& options, const std::string& dbname, file_options_, immutable_db_options_)), seq_per_batch_(seq_per_batch), batch_per_txn_(batch_per_txn), - next_job_id_(1), - shutting_down_(false), - reject_new_background_jobs_(false), - db_lock_(nullptr), - manual_compaction_paused_(false), bg_cv_(&mutex_), - logfile_number_(0), - log_dir_synced_(false), - log_empty_(true), - persist_stats_cf_handle_(nullptr), - log_sync_cv_(&log_write_mutex_), - total_log_size_(0), - is_snapshot_supported_(true), + wal_sync_cv_(&wal_write_mutex_), write_buffer_manager_(immutable_db_options_.write_buffer_manager.get()), write_thread_(immutable_db_options_), nonmem_write_thread_(immutable_db_options_), write_controller_(mutable_db_options_.delayed_write_rate), - last_batch_group_size_(0), - unscheduled_flushes_(0), - unscheduled_compactions_(0), - bg_bottom_compaction_scheduled_(0), - bg_compaction_scheduled_(0), - num_running_compactions_(0), - bg_flush_scheduled_(0), - num_running_flushes_(0), - bg_purge_scheduled_(0), - disable_delete_obsolete_files_(0), - pending_purge_obsolete_files_(0), delete_obsolete_files_last_run_(immutable_db_options_.clock->NowMicros()), - has_unpersisted_data_(false), - unable_to_release_oldest_log_(false), - num_running_ingest_file_(0), wal_manager_(immutable_db_options_, file_options_, io_tracer_, seq_per_batch), - bg_work_paused_(0), - bg_compaction_paused_(0), - refitting_level_(false), - opened_successfully_(false), - periodic_task_scheduler_(), two_write_queues_(options.two_write_queues), manual_wal_flush_(options.manual_wal_flush), // last_sequencee_ is always maintained by the main queue that also writes @@ -250,14 +218,11 @@ DBImpl::DBImpl(const DBOptions& options, const std::string& dbname, // requires a custom gc for compaction, we use that to set use_custom_gc_ // as well. use_custom_gc_(seq_per_batch), - shutdown_initiated_(false), own_sfm_(options.sst_file_manager == nullptr), - closed_(false), atomic_flush_install_cv_(&mutex_), blob_callback_(immutable_db_options_.sst_file_manager.get(), &mutex_, &error_handler_, &event_logger_, - immutable_db_options_.listeners, dbname_), - lock_wal_count_(0) { + immutable_db_options_.listeners, dbname_) { // !batch_per_trx_ implies seq_per_batch_ because it is only unset for // WriteUnprepared, which should use seq_per_batch_. assert(batch_per_txn_ || seq_per_batch_); @@ -636,8 +601,8 @@ Status DBImpl::CloseHelper() { mutex_.Lock(); } { - InstrumentedMutexLock lock(&log_write_mutex_); - for (auto l : logs_to_free_) { + InstrumentedMutexLock lock(&wal_write_mutex_); + for (auto l : wals_to_free_) { delete l; } for (auto& log : logs_) { @@ -1180,11 +1145,11 @@ Status DBImpl::TablesRangeTombstoneSummary(ColumnFamilyHandle* column_family, void DBImpl::ScheduleBgLogWriterClose(JobContext* job_context) { mutex_.AssertHeld(); - if (!job_context->logs_to_free.empty()) { - for (auto l : job_context->logs_to_free) { + if (!job_context->wals_to_free.empty()) { + for (auto l : job_context->wals_to_free) { AddToLogsToFreeQueue(l); } - job_context->logs_to_free.clear(); + job_context->wals_to_free.clear(); } } @@ -1443,7 +1408,7 @@ Status DBImpl::SetDBOptions( WriteThread::Writer w; write_thread_.EnterUnbatched(&w, &mutex_); if (wal_other_option_changed || - total_log_size_ > GetMaxTotalWalSize()) { + wals_total_size_.LoadRelaxed() > GetMaxTotalWalSize()) { Status purge_wal_status = SwitchWAL(&write_context); if (!purge_wal_status.ok()) { ROCKS_LOG_WARN(immutable_db_options_.info_log, @@ -1507,8 +1472,8 @@ Status DBImpl::FlushWAL(const WriteOptions& write_options, bool sync) { if (manual_wal_flush_) { IOStatus io_s; { - // We need to lock log_write_mutex_ since logs_ might change concurrently - InstrumentedMutexLock wl(&log_write_mutex_); + // We need to lock wal_write_mutex_ since logs_ might change concurrently + InstrumentedMutexLock wl(&wal_write_mutex_); log::Writer* cur_log_writer = logs_.back().writer; io_s = cur_log_writer->WriteBuffer(write_options); } @@ -1535,7 +1500,7 @@ Status DBImpl::FlushWAL(const WriteOptions& write_options, bool sync) { } bool DBImpl::WALBufferIsEmpty() { - InstrumentedMutexLock l(&log_write_mutex_); + InstrumentedMutexLock l(&wal_write_mutex_); log::Writer* cur_log_writer = logs_.back().writer; auto res = cur_log_writer->BufferIsEmpty(); return res; @@ -1543,7 +1508,7 @@ bool DBImpl::WALBufferIsEmpty() { Status DBImpl::GetOpenWalSizes(std::map& number_to_size) { assert(number_to_size.empty()); - InstrumentedMutexLock l(&log_write_mutex_); + InstrumentedMutexLock l(&wal_write_mutex_); for (auto& log : logs_) { auto* open_file = log.writer->file(); if (open_file) { @@ -1585,15 +1550,15 @@ IOStatus DBImpl::SyncWalImpl(bool include_current_wal, uint64_t up_to_number; { - InstrumentedMutexLock l(&log_write_mutex_); + InstrumentedMutexLock l(&wal_write_mutex_); assert(!logs_.empty()); - maybe_active_number = logfile_number_; + maybe_active_number = cur_wal_number_; up_to_number = include_current_wal ? maybe_active_number : maybe_active_number - 1; while (logs_.front().number <= up_to_number && logs_.front().IsSyncing()) { - log_sync_cv_.Wait(); + wal_sync_cv_.Wait(); } // First check that logs are safe to sync in background. if (include_current_wal && @@ -1617,7 +1582,7 @@ IOStatus DBImpl::SyncWalImpl(bool include_current_wal, } } - need_wal_dir_sync = !log_dir_synced_; + need_wal_dir_sync = !wal_dir_synced_; } if (include_current_wal) { @@ -1690,7 +1655,7 @@ IOStatus DBImpl::SyncWalImpl(bool include_current_wal, /*arg=*/nullptr); } { - InstrumentedMutexLock l(&log_write_mutex_); + InstrumentedMutexLock l(&wal_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(); @@ -1807,9 +1772,9 @@ Status DBImpl::UnlockWAL() { void DBImpl::MarkLogsSynced(uint64_t up_to, bool synced_dir, VersionEdit* synced_wals) { - log_write_mutex_.AssertHeld(); - if (synced_dir && logfile_number_ == up_to) { - log_dir_synced_ = true; + wal_write_mutex_.AssertHeld(); + if (synced_dir && cur_wal_number_ == up_to) { + wal_dir_synced_ = true; } for (auto it = logs_.begin(); it != logs_.end() && it->number <= up_to;) { auto& wal = *it; @@ -1831,7 +1796,7 @@ void DBImpl::MarkLogsSynced(uint64_t up_to, bool synced_dir, (immutable_db_options_.background_close_inactive_wals && wal.GetPreSyncSize() == wal.writer->file()->GetFlushedSize())) { // Fully synced - logs_to_free_.push_back(wal.ReleaseWriter()); + wals_to_free_.push_back(wal.ReleaseWriter()); it = logs_.erase(it); } else { wal.FinishSync(); @@ -1844,17 +1809,17 @@ void DBImpl::MarkLogsSynced(uint64_t up_to, bool synced_dir, ++it; } } - log_sync_cv_.SignalAll(); + wal_sync_cv_.SignalAll(); } void DBImpl::MarkLogsNotSynced(uint64_t up_to) { - log_write_mutex_.AssertHeld(); + wal_write_mutex_.AssertHeld(); for (auto it = logs_.begin(); it != logs_.end() && it->number <= up_to; ++it) { auto& wal = *it; wal.FinishSync(); } - log_sync_cv_.SignalAll(); + wal_sync_cv_.SignalAll(); } SequenceNumber DBImpl::GetLatestSequenceNumber() const { @@ -1923,10 +1888,10 @@ void DBImpl::BackgroundCallPurge() { TEST_SYNC_POINT("DBImpl::BackgroundCallPurge:beforeMutexLock"); mutex_.Lock(); - while (!logs_to_free_queue_.empty()) { - assert(!logs_to_free_queue_.empty()); - log::Writer* log_writer = *(logs_to_free_queue_.begin()); - logs_to_free_queue_.pop_front(); + while (!wals_to_free_queue_.empty()) { + assert(!wals_to_free_queue_.empty()); + log::Writer* log_writer = *(wals_to_free_queue_.begin()); + wals_to_free_queue_.pop_front(); mutex_.Unlock(); delete log_writer; mutex_.Lock(); @@ -3592,7 +3557,7 @@ Status DBImpl::CreateColumnFamilyImpl(const ReadOptions& read_options, edit.AddColumnFamily(column_family_name); uint32_t new_id = versions_->GetColumnFamilySet()->GetNextColumnFamilyID(); edit.SetColumnFamily(new_id); - edit.SetLogNumber(logfile_number_); + edit.SetLogNumber(cur_wal_number_); edit.SetComparatorName(cf_options.comparator->Name()); edit.SetPersistUserDefinedTimestamps( cf_options.persist_user_defined_timestamps); diff --git a/db/db_impl/db_impl.h b/db/db_impl/db_impl.h index 57fa0a5e83..5f617ac9c9 100644 --- a/db/db_impl/db_impl.h +++ b/db/db_impl/db_impl.h @@ -173,10 +173,10 @@ struct DBOpenLogRecordReadReporter : public log::Reader::Reporter { void OldLogRecord(size_t bytes) override; - uint64_t GetCorruptedLogNumber() const { return corrupted_log_number_; } + uint64_t GetCorruptedLogNumber() const { return corrupted_wal_number_; } private: - uint64_t corrupted_log_number_ = kMaxSequenceNumber; + uint64_t corrupted_wal_number_ = kMaxSequenceNumber; }; // While DB is the public interface of RocksDB, and DBImpl is the actual @@ -535,11 +535,11 @@ class DBImpl : public DB { // Get the known flushed sizes of WALs that might still be written to // or have pending sync. - // NOTE: unlike alive_log_files_, this function includes WALs that might + // NOTE: unlike alive_wal_files_, this function includes WALs that might // be obsolete (but not obsolete to a pending Checkpoint) and not yet fully // synced. Status GetOpenWalSizes(std::map& number_to_size); - Status GetCurrentWalFile(std::unique_ptr* current_log_file) override; + Status GetCurrentWalFile(std::unique_ptr* current_wal_file) override; Status GetCreationTimeOfOldestFile(uint64_t* creation_time) override; Status GetUpdatesSince( @@ -1073,7 +1073,7 @@ class DBImpl : public DB { void AddToLogsToFreeQueue(log::Writer* log_writer) { mutex_.AssertHeld(); - logs_to_free_queue_.push_back(log_writer); + wals_to_free_queue_.push_back(log_writer); } void AddSuperVersionsToFreeQueue(SuperVersion* sv) { @@ -1138,7 +1138,7 @@ class DBImpl : public DB { bool TEST_UnableToReleaseOldestLog() { return unable_to_release_oldest_log_; } bool TEST_IsLogGettingFlushed() { - return alive_log_files_.begin()->getting_flushed; + return alive_wal_files_.begin()->getting_flushed; } Status TEST_SwitchMemtable(ColumnFamilyData* cfd = nullptr); @@ -1218,7 +1218,9 @@ class DBImpl : public DB { uint64_t TEST_LogfileNumber(); - uint64_t TEST_total_log_size() const { return total_log_size_; } + uint64_t TEST_wals_total_size() const { + return wals_total_size_.LoadRelaxed(); + } void TEST_GetAllBlockCaches(std::unordered_set* cache_set); @@ -1376,16 +1378,16 @@ class DBImpl : public DB { // State below is protected by mutex_ // With two_write_queues enabled, some of the variables that accessed during - // WriteToWAL need different synchronization: log_empty_, alive_log_files_, - // logs_, logfile_number_. Refer to the definition of each variable below for + // WriteToWAL need different synchronization: wal_empty_, alive_wal_files_, + // logs_, cur_wal_number_. Refer to the definition of each variable below for // more description. // // `mutex_` can be a hot lock in some workloads, so it deserves dedicated // cachelines. mutable CacheAlignedInstrumentedMutex mutex_; - ColumnFamilyHandleImpl* default_cf_handle_; - InternalStats* default_cf_internal_stats_; + ColumnFamilyHandleImpl* default_cf_handle_ = nullptr; + InternalStats* default_cf_internal_stats_ = nullptr; // table_cache_ provides its own synchronization std::shared_ptr table_cache_; @@ -1397,7 +1399,7 @@ class DBImpl : public DB { // only used for dynamically adjusting max_total_wal_size. it is a sum of // [write_buffer_size * max_write_buffer_number] over all column families - std::atomic max_total_in_memory_state_; + std::atomic max_total_in_memory_state_ = 0; // The options to access storage files const FileOptions file_options_; @@ -1424,14 +1426,14 @@ class DBImpl : public DB { // Each flush or compaction gets its own job id. this counter makes sure // they're unique - std::atomic next_job_id_; + std::atomic next_job_id_ = 1; - std::atomic shutting_down_; + std::atomic shutting_down_ = false; // No new background jobs can be queued if true. This is used to prevent new // background jobs from being queued after WaitForCompact() completes waiting // all background jobs then attempts to close when close_db_ option is true. - bool reject_new_background_jobs_; + bool reject_new_background_jobs_ = false; // RecoveryContext struct stores the context about version edits along // with corresponding column_family_data and column_family_options. @@ -1563,7 +1565,7 @@ class DBImpl : public DB { Status WriteImpl(const WriteOptions& options, WriteBatch* updates, WriteCallback* callback = nullptr, UserWriteCallback* user_write_cb = nullptr, - uint64_t* log_used = nullptr, uint64_t log_ref = 0, + uint64_t* wal_used = nullptr, uint64_t log_ref = 0, bool disable_memtable = false, uint64_t* seq_used = nullptr, size_t batch_cnt = 0, PreReleaseCallback* pre_release_callback = nullptr, @@ -1574,7 +1576,7 @@ class DBImpl : public DB { Status PipelinedWriteImpl(const WriteOptions& options, WriteBatch* updates, WriteCallback* callback = nullptr, UserWriteCallback* user_write_cb = nullptr, - uint64_t* log_used = nullptr, uint64_t log_ref = 0, + uint64_t* wal_used = nullptr, uint64_t log_ref = 0, bool disable_memtable = false, uint64_t* seq_used = nullptr); @@ -1601,7 +1603,7 @@ class DBImpl : public DB { Status WriteImplWALOnly( WriteThread* write_thread, const WriteOptions& options, WriteBatch* updates, WriteCallback* callback, - UserWriteCallback* user_write_cb, uint64_t* log_used, + UserWriteCallback* user_write_cb, uint64_t* wal_used, const uint64_t log_ref, uint64_t* seq_used, const size_t sub_batch_cnt, PreReleaseCallback* pre_release_callback, const AssignOrder assign_order, const PublishLastSeq publish_last_seq, const bool disable_memtable); @@ -1762,9 +1764,9 @@ class DBImpl : public DB { } }; - struct LogFileNumberSize { - explicit LogFileNumberSize(uint64_t _number) : number(_number) {} - LogFileNumberSize() {} + struct WalFileNumberSize { + explicit WalFileNumberSize(uint64_t _number) : number(_number) {} + WalFileNumberSize() {} void AddSize(uint64_t new_size) { size += new_size; } uint64_t number; uint64_t size = 0; @@ -1848,13 +1850,13 @@ class DBImpl : public DB { uint64_t attempt_truncate_size = SIZE_MAX; }; - struct LogContext { - explicit LogContext(bool need_sync = false) - : need_log_sync(need_sync), need_log_dir_sync(need_sync) {} - bool need_log_sync = false; - bool need_log_dir_sync = false; + struct WalContext { + explicit WalContext(bool need_sync = false) + : need_wal_sync(need_sync), need_wal_dir_sync(need_sync) {} + bool need_wal_sync = false; + bool need_wal_dir_sync = false; log::Writer* writer = nullptr; - LogFileNumberSize* log_file_number_size = nullptr; + WalFileNumberSize* wal_file_number_size = nullptr; uint64_t prev_size = SIZE_MAX; }; @@ -2063,10 +2065,10 @@ class DBImpl : public DB { JobContext* job_context, LogBuffer* log_buffer, Env::Priority thread_pri); // REQUIRES: log_numbers are sorted in ascending order - // corrupted_log_found is set to true if we recover from a corrupted log file. + // corrupted_wal_found is set to true if we recover from a corrupted log file. Status RecoverLogFiles(const std::vector& log_numbers, SequenceNumber* next_sequence, bool read_only, - bool is_retry, bool* corrupted_log_found, + bool is_retry, bool* corrupted_wal_found, RecoveryContext* recovery_ctx); void SetupLogFilesRecovery( @@ -2174,12 +2176,12 @@ class DBImpl : public DB { // log file to its actual size, thereby freeing preallocated space. // Return success even if truncate fails Status GetLogSizeAndMaybeTruncate(uint64_t wal_number, bool truncate, - LogFileNumberSize* log); + WalFileNumberSize* log); - // Restore alive_log_files_ and total_log_size_ after recovery. + // Restore alive_wal_files_ and wals_total_size_ after recovery. // It needs to run only when there's no flush during recovery // (e.g. avoid_flush_during_recovery=true). May also trigger flush - // in case total_log_size > max_total_wal_size. + // in case wals_total_size > max_total_wal_size. Status RestoreAliveLogFiles(const std::vector& log_numbers); // num_bytes: for slowdown case, delay time is calculated based on @@ -2328,7 +2330,7 @@ class DBImpl : public DB { // REQUIRES: mutex locked Status PreprocessWrite(const WriteOptions& write_options, - LogContext* log_context, WriteContext* write_context); + WalContext* log_context, WriteContext* write_context); // Merge write batches in the write group into merged_batch. // Returns OK if merge is successful. @@ -2339,20 +2341,21 @@ class DBImpl : public DB { IOStatus WriteToWAL(const WriteBatch& merged_batch, const WriteOptions& write_options, - log::Writer* log_writer, uint64_t* log_used, + log::Writer* log_writer, uint64_t* wal_used, uint64_t* log_size, - LogFileNumberSize& log_file_number_size, + WalFileNumberSize& wal_file_number_size, SequenceNumber sequence); - IOStatus WriteToWAL(const WriteThread::WriteGroup& write_group, - log::Writer* log_writer, uint64_t* log_used, - bool need_log_sync, bool need_log_dir_sync, - SequenceNumber sequence, - LogFileNumberSize& log_file_number_size); + IOStatus WriteGroupToWAL(const WriteThread::WriteGroup& write_group, + log::Writer* log_writer, uint64_t* wal_used, + bool need_wal_sync, bool need_wal_dir_sync, + SequenceNumber sequence, + WalFileNumberSize& wal_file_number_size); - IOStatus ConcurrentWriteToWAL(const WriteThread::WriteGroup& write_group, - uint64_t* log_used, - SequenceNumber* last_sequence, size_t seq_inc); + IOStatus ConcurrentWriteGroupToWAL(const WriteThread::WriteGroup& write_group, + uint64_t* wal_used, + SequenceNumber* last_sequence, + size_t seq_inc); // Used by WriteImpl to update bg_error_ if paranoid check is enabled. // Caller must hold mutex_. @@ -2719,7 +2722,7 @@ class DBImpl : public DB { ErrorIteratorFuncType error_iterator_func); // Lock over the persistent DB state. Non-nullptr iff successfully acquired. - FileLock* db_lock_; + FileLock* db_lock_ = nullptr; // Guards changes to DB and CF options to ensure consistency between // * In-memory options objects @@ -2733,20 +2736,20 @@ class DBImpl : public DB { // Guards reads and writes to in-memory stats_history_. InstrumentedMutex stats_history_mutex_; - // In addition to mutex_, log_write_mutex_ protects writes to logs_ and - // logfile_number_. With two_write_queues it also protects alive_log_files_, - // and log_empty_. Refer to the definition of each variable below for more + // In addition to mutex_, wal_write_mutex_ protects writes to logs_ and + // cur_wal_number_. With two_write_queues it also protects alive_wal_files_, + // and wal_empty_. Refer to the definition of each variable below for more // details. - // Note: to avoid deadlock, if needed to acquire both log_write_mutex_ and - // mutex_, the order should be first mutex_ and then log_write_mutex_. - InstrumentedMutex log_write_mutex_; + // Note: to avoid deadlock, if needed to acquire both wal_write_mutex_ and + // mutex_, the order should be first mutex_ and then wal_write_mutex_. + InstrumentedMutex wal_write_mutex_; // If zero, manual compactions are allowed to proceed. If non-zero, manual // compactions may still be running, but will quickly fail with // `Status::Incomplete`. The value indicates how many threads have paused // manual compactions. It is accessed in read mode outside the DB mutex in // compaction code paths. - std::atomic manual_compaction_paused_; + std::atomic manual_compaction_paused_ = false; // This condition variable is signaled on these conditions: // * whenever bg_compaction_scheduled_ goes down to 0 @@ -2762,106 +2765,114 @@ class DBImpl : public DB { // * whenever SetOptions successfully updates options. // * whenever a column family is dropped. InstrumentedCondVar bg_cv_; - // Writes are protected by locking both mutex_ and log_write_mutex_, and reads - // must be under either mutex_ or log_write_mutex_. Since after ::Open, - // logfile_number_ is currently updated only in write_thread_, it can be read + + ColumnFamilyHandleImpl* persist_stats_cf_handle_ = nullptr; + + bool persistent_stats_cfd_exists_ = true; + + // Writes are protected by locking both mutex_ and wal_write_mutex_, and reads + // must be under either mutex_ or wal_write_mutex_. Since after ::Open, + // cur_wal_number_ is currently updated only in write_thread_, it can be read // from the same write_thread_ without any locks. - uint64_t logfile_number_; + uint64_t cur_wal_number_ = 0; + // Log files that we can recycle. Must be protected by db mutex_. - std::deque log_recycle_files_; + std::deque wal_recycle_files_; + // The minimum log file number taht can be recycled, if log recycling is // enabled. This is used to ensure that log files created by previous // instances of the database are not recycled, as we cannot be sure they // were created in the recyclable format. - uint64_t min_log_number_to_recycle_; - // Protected by log_write_mutex_. - bool log_dir_synced_; - // Without two_write_queues, read and writes to log_empty_ are protected by + uint64_t min_wal_number_to_recycle_ = 0; + + // Protected by wal_write_mutex_. + bool wal_dir_synced_ = false; + + // Without two_write_queues, read and writes to wal_empty_ are protected by // mutex_. Since it is currently updated/read only in write_thread_, it can be // accessed from the same write_thread_ without any locks. With // two_write_queues writes, where it can be updated in different threads, - // read and writes are protected by log_write_mutex_ instead. This is to avoid - // expensive mutex_ lock during WAL write, which update log_empty_. - bool log_empty_; - - ColumnFamilyHandleImpl* persist_stats_cf_handle_; - - bool persistent_stats_cfd_exists_ = true; + // read and writes are protected by wal_write_mutex_ instead. This is to avoid + // expensive mutex_ lock during WAL write, which update wal_empty_. + bool wal_empty_ = true; // The current WAL file and those that have not been found obsolete from // memtable flushes. A WAL not on this list might still be pending writer - // flush and/or sync and close and might still be in logs_. alive_log_files_ - // is protected by mutex_ and log_write_mutex_ with details as follows: + // flush and/or sync and close and might still be in logs_. alive_wal_files_ + // is protected by mutex_ and wal_write_mutex_ with details as follows: // 1. read by FindObsoleteFiles() which can be called in either application - // thread or RocksDB bg threads, both mutex_ and log_write_mutex_ are + // thread or RocksDB bg threads, both mutex_ and wal_write_mutex_ are // held. - // 2. pop_front() by FindObsoleteFiles(), both mutex_ and log_write_mutex_ + // 2. pop_front() by FindObsoleteFiles(), both mutex_ and wal_write_mutex_ // are held. // 3. push_back() by DBImpl::Open() and DBImpl::RestoreAliveLogFiles() // (actually called by Open()), only mutex_ is held because at this point, // the DB::Open() call has not returned success to application, and the // only other thread(s) that can conflict are bg threads calling - // FindObsoleteFiles() which ensure that both mutex_ and log_write_mutex_ - // are held when accessing alive_log_files_. + // FindObsoleteFiles() which ensure that both mutex_ and wal_write_mutex_ + // are held when accessing alive_wal_files_. // 4. read by DBImpl::Open() is protected by mutex_. - // 5. push_back() by SwitchMemtable(). Both mutex_ and log_write_mutex_ are + // 5. push_back() by SwitchMemtable(). Both mutex_ and wal_write_mutex_ are // held. This is done by the write group leader. Note that in the case of // two-write-queues, another WAL-only write thread can be writing to the // WAL concurrently. See 9. - // 6. read by SwitchWAL() with both mutex_ and log_write_mutex_ held. This is + // 6. read by SwitchWAL() with both mutex_ and wal_write_mutex_ held. This is // done by write group leader. // 7. read by ConcurrentWriteToWAL() by the write group leader in the case of - // two-write-queues. Only log_write_mutex_ is held to protect concurrent + // two-write-queues. Only wal_write_mutex_ is held to protect concurrent // pop_front() by FindObsoleteFiles(). - // 8. read by PreprocessWrite() by the write group leader. log_write_mutex_ + // 8. read by PreprocessWrite() by the write group leader. wal_write_mutex_ // is held to protect the data structure from concurrent pop_front() by // FindObsoleteFiles(). // 9. read by ConcurrentWriteToWAL() by a WAL-only write thread in the case - // of two-write-queues. Only log_write_mutex_ is held. This suffices to + // of two-write-queues. Only wal_write_mutex_ is held. This suffices to // protect the data structure from concurrent push_back() by current // write group leader as well as pop_front() by FindObsoleteFiles(). - std::deque alive_log_files_; + std::deque alive_wal_files_; + + // Total size of all "alive" WALs (for easy access without synchronization) + RelaxedAtomic wals_total_size_{0}; // Log files that aren't fully synced, and the current log file. // Synchronization: // 1. read by FindObsoleteFiles() which can be called either in application - // thread or RocksDB bg threads. log_write_mutex_ is always held, while + // thread or RocksDB bg threads. wal_write_mutex_ is always held, while // some reads are performed without mutex_. - // 2. pop_front() by FindObsoleteFiles() with only log_write_mutex_ held. - // 3. read by DBImpl::Open() with both mutex_ and log_write_mutex_. - // 4. emplace_back() by DBImpl::Open() with both mutex_ and log_write_mutex. + // 2. pop_front() by FindObsoleteFiles() with only wal_write_mutex_ held. + // 3. read by DBImpl::Open() with both mutex_ and wal_write_mutex_. + // 4. emplace_back() by DBImpl::Open() with both mutex_ and wal_write_mutex. // Note that at this point, DB::Open() has not returned success to // application, thus the only other thread(s) that can conflict are bg // threads calling FindObsoleteFiles(). See 1. - // 5. iteration and clear() from CloseHelper() always hold log_write_mutex + // 5. iteration and clear() from CloseHelper() always hold wal_write_mutex // and mutex_. // 6. back() called by APIs FlushWAL() and LockWAL() are protected by only - // log_write_mutex_. These two can be called by application threads after + // wal_write_mutex_. These two can be called by application threads after // DB::Open() returns success to applications. - // 7. read by SyncWAL(), another API, protected by only log_write_mutex_. + // 7. read by SyncWAL(), another API, protected by only wal_write_mutex_. // 8. read by MarkLogsNotSynced() and MarkLogsSynced() are protected by - // log_write_mutex_. - // 9. erase() by MarkLogsSynced() protected by log_write_mutex_. - // 10. read by SyncClosedWals() protected by only log_write_mutex_. This can + // wal_write_mutex_. + // 9. erase() by MarkLogsSynced() protected by wal_write_mutex_. + // 10. read by SyncClosedWals() protected by only wal_write_mutex_. This can // happen in bg flush threads after DB::Open() returns success to // applications. // 11. reads, e.g. front(), iteration, and back() called by PreprocessWrite() - // holds only the log_write_mutex_. This is done by the write group + // holds only the wal_write_mutex_. This is done by the write group // leader. A bg thread calling FindObsoleteFiles() or MarkLogsSynced() - // can happen concurrently. This is fine because log_write_mutex_ is used + // can happen concurrently. This is fine because wal_write_mutex_ is used // by all parties. See 2, 5, 9. // 12. reads, empty(), back() called by SwitchMemtable() hold both mutex_ and - // log_write_mutex_. This happens in the write group leader. + // wal_write_mutex_. This happens in the write group leader. // 13. emplace_back() by SwitchMemtable() hold both mutex_ and - // log_write_mutex_. This happens in the write group leader. Can conflict + // wal_write_mutex_. This happens in the write group leader. Can conflict // with bg threads calling FindObsoleteFiles(), MarkLogsSynced(), // SyncClosedWals(), etc. as well as application threads calling // FlushWAL(), SyncWAL(), LockWAL(). This is fine because all parties - // require at least log_write_mutex_. + // require at least wal_write_mutex_. // 14. iteration called in WriteToWAL(write_group) protected by - // log_write_mutex_. This is done by write group leader when + // wal_write_mutex_. This is done by write group leader when // two-write-queues is disabled and write needs to sync logs. - // 15. back() called in ConcurrentWriteToWAL() protected by log_write_mutex_. + // 15. back() called in ConcurrentWriteToWAL() protected by wal_write_mutex_. // This can be done by the write group leader if two-write-queues is // enabled. It can also be done by another WAL-only write thread. // @@ -2878,23 +2889,22 @@ class DBImpl : public DB { std::deque logs_; // Signaled when getting_synced becomes false for some of the logs_. - InstrumentedCondVar log_sync_cv_; + InstrumentedCondVar wal_sync_cv_; // This is the app-level state that is written to the WAL but will be used // only during recovery. Using this feature enables not writing the state to // memtable on normal writes and hence improving the throughput. Each new // write of the state will replace the previous state entirely even if the // keys in the two consecutive states do not overlap. - // It is protected by log_write_mutex_ when two_write_queues_ is enabled. + // It is protected by wal_write_mutex_ when two_write_queues_ is enabled. // Otherwise only the heaad of write_thread_ can access it. WriteBatch cached_recoverable_state_; std::atomic cached_recoverable_state_empty_ = {true}; - std::atomic total_log_size_; // If this is non-empty, we need to delete these log files in background - // threads. Protected by log_write_mutex_. - autovector logs_to_free_; + // threads. Protected by wal_write_mutex_. + autovector wals_to_free_; - bool is_snapshot_supported_; + bool is_snapshot_supported_ = true; std::map> stats_history_; @@ -2918,7 +2928,7 @@ class DBImpl : public DB { // sleep if it uses up the quota. // Note: This is to protect memtable and compaction. If the batch only writes // to the WAL its size need not to be included in this. - uint64_t last_batch_group_size_; + uint64_t last_batch_group_size_ = 0; FlushScheduler flush_scheduler_; @@ -2977,32 +2987,32 @@ class DBImpl : public DB { std::unordered_set files_grabbed_for_purge_; // A queue to store log writers to close. Protected by db mutex_. - std::deque logs_to_free_queue_; + std::deque wals_to_free_queue_; std::deque superversions_to_free_queue_; - int unscheduled_flushes_; + int unscheduled_flushes_ = 0; - int unscheduled_compactions_; + int unscheduled_compactions_ = 0; // count how many background compactions are running or have been scheduled in // the BOTTOM pool - int bg_bottom_compaction_scheduled_; + int bg_bottom_compaction_scheduled_ = 0; // count how many background compactions are running or have been scheduled - int bg_compaction_scheduled_; + int bg_compaction_scheduled_ = 0; // stores the number of compactions are currently running - int num_running_compactions_; + int num_running_compactions_ = 0; // number of background memtable flush jobs, submitted to the HIGH pool - int bg_flush_scheduled_; + int bg_flush_scheduled_ = 0; // stores the number of flushes are currently running - int num_running_flushes_; + int num_running_flushes_ = 0; // number of background obsolete file purge jobs, submitted to the HIGH pool - int bg_purge_scheduled_; + int bg_purge_scheduled_ = 0; std::deque manual_compaction_dequeue_; @@ -3012,11 +3022,11 @@ class DBImpl : public DB { // This enables two different threads to call // EnableFileDeletions() and DisableFileDeletions() // without any synchronization - int disable_delete_obsolete_files_; + int disable_delete_obsolete_files_ = 0; // Number of times FindObsoleteFiles has found deletable files and the // corresponding call to PurgeObsoleteFiles has not yet finished. - int pending_purge_obsolete_files_; + int pending_purge_obsolete_files_ = 0; // last time when DeleteObsoleteFiles with full scan was executed. Originally // initialized with startup time. @@ -3028,12 +3038,12 @@ class DBImpl : public DB { // The mutex used by switch_cv_. mutex_ should be acquired beforehand. std::mutex switch_mutex_; // Number of threads intending to write to memtable - std::atomic pending_memtable_writes_ = {}; + std::atomic pending_memtable_writes_{0}; // A flag indicating whether the current rocksdb database has any // data that is not yet persisted into either WAL or SST file. // Used when disableWAL is true. - std::atomic has_unpersisted_data_; + std::atomic has_unpersisted_data_{false}; // if an attempt was made to flush all column families that // the oldest log depends on but uncommitted data in the oldest @@ -3041,26 +3051,26 @@ class DBImpl : public DB { // We must attempt to free the dependent memtables again // at a later time after the transaction in the oldest // log is fully commited. - bool unable_to_release_oldest_log_; + bool unable_to_release_oldest_log_{false}; // Number of running IngestExternalFile() or CreateColumnFamilyWithImport() // calls. // REQUIRES: mutex held - int num_running_ingest_file_; + int num_running_ingest_file_ = 0; WalManager wal_manager_; // A value of > 0 temporarily disables scheduling of background work - int bg_work_paused_; + int bg_work_paused_ = 0; // A value of > 0 temporarily disables scheduling of background compaction - int bg_compaction_paused_; + int bg_compaction_paused_ = 0; // Guard against multiple concurrent refitting - bool refitting_level_; + bool refitting_level_ = false; // Indicate DB was opened successfully - bool opened_successfully_; + bool opened_successfully_ = false; // The min threshold to triggere bottommost compaction for removing // garbages, among all column families. @@ -3106,13 +3116,13 @@ class DBImpl : public DB { // error recovery from going on in parallel. The latter, shutting_down_, // is set a little later during the shutdown after scheduling memtable // flushes - std::atomic shutdown_initiated_; + std::atomic shutdown_initiated_{false}; // Flag to indicate whether sst_file_manager object was allocated in // DB::Open() or passed to us bool own_sfm_; // Flag to check whether Close() has been called on this DB - bool closed_; + bool closed_ = false; // save the closing status, for re-calling the close() Status closing_status_; // mutex for DB::Close() @@ -3148,7 +3158,7 @@ class DBImpl : public DB { // The number of LockWAL called without matching UnlockWAL call. // See also lock_wal_write_token_ - uint32_t lock_wal_count_; + uint32_t lock_wal_count_ = 0; }; class GetWithTimestampReadCallback : public ReadCallback { diff --git a/db/db_impl/db_impl_compaction_flush.cc b/db/db_impl/db_impl_compaction_flush.cc index d0b1074b3b..9ae28aa8db 100644 --- a/db/db_impl/db_impl_compaction_flush.cc +++ b/db/db_impl/db_impl_compaction_flush.cc @@ -168,7 +168,7 @@ Status DBImpl::FlushMemTableToOutputFile( // had not been committed yet. Make sure we sync them to keep the persisted // WAL state at least as new as the persisted SST state. const bool needs_to_sync_closed_wals = - logfile_number_ > 0 && + cur_wal_number_ > 0 && (versions_->GetColumnFamilySet()->NumberOfColumnFamilies() > 1 || allow_2pc()); @@ -224,7 +224,7 @@ Status DBImpl::FlushMemTableToOutputFile( bool need_cancel = false; IOStatus log_io_s = IOStatus::OK(); if (needs_to_sync_closed_wals) { - // SyncClosedWals() may unlock and re-lock the log_write_mutex multiple + // SyncClosedWals() may unlock and re-lock the wal_write_mutex multiple // times. VersionEdit synced_wals; bool error_recovery_in_prog = error_handler_.IsRecoveryInProgress(); @@ -512,7 +512,7 @@ Status DBImpl::AtomicFlushMemTablesToOutputFiles( job_context->job_id, flush_reason); } - if (logfile_number_ > 0) { + if (cur_wal_number_ > 0) { // TODO (yanqin) investigate whether we should sync the closed logs for // single column family case. VersionEdit synced_wals; @@ -528,7 +528,7 @@ Status DBImpl::AtomicFlushMemTablesToOutputFiles( if (!log_io_s.ok() && !log_io_s.IsShutdownInProgress() && !log_io_s.IsColumnFamilyDropped()) { - if (total_log_size_ > 0) { + if (wals_total_size_.LoadRelaxed() > 0) { error_handler_.SetBGError(log_io_s, BackgroundErrorReason::kFlush); } else { // If the WAL is empty, we use different error reason diff --git a/db/db_impl/db_impl_debug.cc b/db/db_impl/db_impl_debug.cc index 3f29d06341..ee48b07986 100644 --- a/db/db_impl/db_impl_debug.cc +++ b/db/db_impl/db_impl_debug.cc @@ -224,13 +224,13 @@ void DBImpl::TEST_EndWrite(void* w) { } size_t DBImpl::TEST_LogsToFreeSize() { - InstrumentedMutexLock l(&log_write_mutex_); - return logs_to_free_.size(); + InstrumentedMutexLock l(&wal_write_mutex_); + return wals_to_free_.size(); } uint64_t DBImpl::TEST_LogfileNumber() { InstrumentedMutexLock l(&mutex_); - return logfile_number_; + return cur_wal_number_; } void DBImpl::TEST_GetAllBlockCaches( diff --git a/db/db_impl/db_impl_files.cc b/db/db_impl/db_impl_files.cc index 2c4b3bfde9..e2dc53e7d4 100644 --- a/db/db_impl/db_impl_files.cc +++ b/db/db_impl/db_impl_files.cc @@ -28,7 +28,7 @@ uint64_t DBImpl::MinLogNumberToKeep() { return versions_->min_log_number_to_keep(); } -uint64_t DBImpl::MinLogNumberToRecycle() { return min_log_number_to_recycle_; } +uint64_t DBImpl::MinLogNumberToRecycle() { return min_wal_number_to_recycle_; } uint64_t DBImpl::MinObsoleteSstNumberToKeep() { mutex_.AssertHeld(); @@ -272,77 +272,77 @@ void DBImpl::FindObsoleteFiles(JobContext* job_context, bool force, // logs_ is empty when called during recovery, in which case there can't yet // be any tracked obsolete logs - log_write_mutex_.Lock(); + wal_write_mutex_.Lock(); - if (alive_log_files_.empty() || logs_.empty()) { + if (alive_wal_files_.empty() || logs_.empty()) { mutex_.AssertHeld(); // We may reach here if the db is DBImplSecondary - log_write_mutex_.Unlock(); + wal_write_mutex_.Unlock(); return; } bool mutex_unlocked = false; - if (!alive_log_files_.empty() && !logs_.empty()) { + if (!alive_wal_files_.empty() && !logs_.empty()) { uint64_t min_log_number = job_context->log_number; - size_t num_alive_log_files = alive_log_files_.size(); + size_t num_alive_wal_files = alive_wal_files_.size(); // find newly obsoleted log files - while (alive_log_files_.begin()->number < min_log_number) { - auto& earliest = *alive_log_files_.begin(); + while (alive_wal_files_.begin()->number < min_log_number) { + auto& earliest = *alive_wal_files_.begin(); if (immutable_db_options_.recycle_log_file_num > - log_recycle_files_.size() && + wal_recycle_files_.size() && earliest.number >= MinLogNumberToRecycle()) { ROCKS_LOG_INFO(immutable_db_options_.info_log, "adding log %" PRIu64 " to recycle list\n", earliest.number); - log_recycle_files_.push_back(earliest.number); + wal_recycle_files_.push_back(earliest.number); } else { job_context->log_delete_files.push_back(earliest.number); } if (job_context->size_log_to_delete == 0) { - job_context->prev_total_log_size = total_log_size_; - job_context->num_alive_log_files = num_alive_log_files; + job_context->prev_wals_total_size = wals_total_size_.LoadRelaxed(); + job_context->num_alive_wal_files = num_alive_wal_files; } job_context->size_log_to_delete += earliest.size; - total_log_size_ -= earliest.size; - alive_log_files_.pop_front(); + wals_total_size_.FetchSubRelaxed(earliest.size); + alive_wal_files_.pop_front(); // Current log should always stay alive since it can't have // number < MinLogNumber(). - assert(alive_log_files_.size()); + assert(alive_wal_files_.size()); } - log_write_mutex_.Unlock(); + wal_write_mutex_.Unlock(); mutex_.Unlock(); mutex_unlocked = true; TEST_SYNC_POINT_CALLBACK("FindObsoleteFiles::PostMutexUnlock", nullptr); - log_write_mutex_.Lock(); + wal_write_mutex_.Lock(); while (!logs_.empty() && logs_.front().number < min_log_number) { auto& log = logs_.front(); if (log.IsSyncing()) { - log_sync_cv_.Wait(); + wal_sync_cv_.Wait(); // 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. // 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_. + // removing from logs_ but not blocking while holding wal_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(); + wal_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(); + wal_write_mutex_.Lock(); log.writer->PublishIfClosed(); assert(&log == &logs_.front()); log.FinishSync(); - log_sync_cv_.SignalAll(); + wal_sync_cv_.SignalAll(); } - logs_to_free_.push_back(log.ReleaseWriter()); + wals_to_free_.push_back(log.ReleaseWriter()); logs_.pop_front(); } // Current log cannot be obsolete. @@ -350,16 +350,16 @@ void DBImpl::FindObsoleteFiles(JobContext* job_context, bool force, } // We're just cleaning up for DB::Write(). - assert(job_context->logs_to_free.empty()); - job_context->logs_to_free = logs_to_free_; + assert(job_context->wals_to_free.empty()); + job_context->wals_to_free = wals_to_free_; - logs_to_free_.clear(); - log_write_mutex_.Unlock(); + wals_to_free_.clear(); + wal_write_mutex_.Unlock(); if (mutex_unlocked) { mutex_.Lock(); } - job_context->log_recycle_files.assign(log_recycle_files_.begin(), - log_recycle_files_.end()); + job_context->log_recycle_files.assign(wal_recycle_files_.begin(), + wal_recycle_files_.end()); } // Delete obsolete files and log status and information of file deletion @@ -431,7 +431,7 @@ void DBImpl::PurgeObsoleteFiles(JobContext& state, bool schedule_only) { state.sst_live.end()); std::unordered_set blob_live_set(state.blob_live.begin(), state.blob_live.end()); - std::unordered_set log_recycle_files_set( + std::unordered_set wal_recycle_files_set( state.log_recycle_files.begin(), state.log_recycle_files.end()); std::unordered_set quarantine_files_set( state.files_to_quarantine.begin(), state.files_to_quarantine.end()); @@ -491,13 +491,13 @@ void DBImpl::PurgeObsoleteFiles(JobContext& state, bool schedule_only) { std::unique(candidate_files.begin(), candidate_files.end()), candidate_files.end()); - if (state.prev_total_log_size > 0) { + if (state.prev_wals_total_size > 0) { ROCKS_LOG_INFO(immutable_db_options_.info_log, "[JOB %d] Try to delete WAL files size %" PRIu64 ", prev total WAL file size %" PRIu64 ", number of live WAL files %" ROCKSDB_PRIszt ".\n", state.job_id, state.size_log_to_delete, - state.prev_total_log_size, state.num_alive_log_files); + state.prev_wals_total_size, state.num_alive_wal_files); } std::vector old_info_log_files; @@ -532,7 +532,7 @@ void DBImpl::PurgeObsoleteFiles(JobContext& state, bool schedule_only) { optsfile_num2 = std::min(optsfile_num2, state.min_options_file_number); // Close WALs before trying to delete them. - for (const auto w : state.logs_to_free) { + for (const auto w : state.wals_to_free) { // TODO: maybe check the return value of Close. // TODO: plumb Env::IOActivity, Env::IOPriority auto s = w->Close({}); @@ -559,8 +559,8 @@ void DBImpl::PurgeObsoleteFiles(JobContext& state, bool schedule_only) { case kWalFile: keep = ((number >= state.log_number) || (number == state.prev_log_number) || - (log_recycle_files_set.find(number) != - log_recycle_files_set.end())); + (wal_recycle_files_set.find(number) != + wal_recycle_files_set.end())); break; case kDescriptorFile: // Keep my manifest file, and any newer incarnations' diff --git a/db/db_impl/db_impl_open.cc b/db/db_impl/db_impl_open.cc index ab6902075f..0e365c9b34 100644 --- a/db/db_impl/db_impl_open.cc +++ b/db/db_impl/db_impl_open.cc @@ -1119,7 +1119,7 @@ void DBOpenLogRecordReadReporter::Corruption(size_t bytes, const Status& s, static_cast(bytes), s.ToString().c_str()); if (status != nullptr && status->ok()) { *status = s; - corrupted_log_number_ = log_number; + corrupted_wal_number_ = log_number; } } @@ -1902,8 +1902,8 @@ void DBImpl::FinishLogFilesRecovery(int job_id, const Status& status) { } Status DBImpl::GetLogSizeAndMaybeTruncate(uint64_t wal_number, bool truncate, - LogFileNumberSize* log_ptr) { - LogFileNumberSize log(wal_number); + WalFileNumberSize* log_ptr) { + WalFileNumberSize log(wal_number); std::string fname = LogFileName(immutable_db_options_.GetWalDir(), wal_number); Status s; @@ -1946,27 +1946,27 @@ Status DBImpl::RestoreAliveLogFiles(const std::vector& wal_numbers) { assert(immutable_db_options_.avoid_flush_during_recovery); // Mark these as alive so they'll be considered for deletion later by // FindObsoleteFiles() - total_log_size_ = 0; - log_empty_ = false; + wals_total_size_.StoreRelaxed(0); + wal_empty_ = false; uint64_t min_wal_with_unflushed_data = versions_->MinLogNumberWithUnflushedData(); for (auto wal_number : wal_numbers) { if (!allow_2pc() && wal_number < min_wal_with_unflushed_data) { // In non-2pc mode, the WAL files not backing unflushed data are not - // alive, thus should not be added to the alive_log_files_. + // alive, thus should not be added to the alive_wal_files_. continue; } // We preallocate space for wals, but then after a crash and restart, those // preallocated space are not needed anymore. It is likely only the last // log has such preallocated space, so we only truncate for the last log. - LogFileNumberSize log; + WalFileNumberSize log; s = GetLogSizeAndMaybeTruncate( wal_number, /*truncate=*/(wal_number == wal_numbers.back()), &log); if (!s.ok()) { break; } - total_log_size_ += log.size; - alive_log_files_.push_back(log); + wals_total_size_.FetchAddRelaxed(log.size); + alive_wal_files_.push_back(log); } return s; } @@ -2449,18 +2449,18 @@ Status DBImpl::Open(const DBOptions& db_options, const std::string& dbname, if (s.ok()) { // Prevent log files created by previous instance from being recycled. // They might be in alive_log_file_, and might get recycled otherwise. - impl->min_log_number_to_recycle_ = new_log_number; + impl->min_wal_number_to_recycle_ = new_log_number; } if (s.ok()) { - InstrumentedMutexLock wl(&impl->log_write_mutex_); - impl->logfile_number_ = new_log_number; + InstrumentedMutexLock wl(&impl->wal_write_mutex_); + impl->cur_wal_number_ = new_log_number; assert(new_log != nullptr); assert(impl->logs_.empty()); impl->logs_.emplace_back(new_log_number, new_log); } if (s.ok()) { - impl->alive_log_files_.emplace_back(impl->logfile_number_); + impl->alive_wal_files_.emplace_back(impl->cur_wal_number_); // In WritePrepared there could be gap in sequence numbers. This breaks // the trick we use in kPointInTimeRecovery which assumes the first seq in // the log right after the corrupted log is one larger than the last seq @@ -2473,14 +2473,14 @@ Status DBImpl::Open(const DBOptions& db_options, const std::string& dbname, if (recovered_seq != kMaxSequenceNumber) { WriteBatch empty_batch; WriteBatchInternal::SetSequence(&empty_batch, recovered_seq); - uint64_t log_used, log_size; + uint64_t wal_used, log_size; log::Writer* log_writer = impl->logs_.back().writer; - LogFileNumberSize& log_file_number_size = impl->alive_log_files_.back(); + WalFileNumberSize& wal_file_number_size = impl->alive_wal_files_.back(); - assert(log_writer->get_log_number() == log_file_number_size.number); + assert(log_writer->get_log_number() == wal_file_number_size.number); impl->mutex_.AssertHeld(); - s = impl->WriteToWAL(empty_batch, write_options, log_writer, &log_used, - &log_size, log_file_number_size, recovered_seq); + s = impl->WriteToWAL(empty_batch, write_options, log_writer, &wal_used, + &log_size, wal_file_number_size, recovered_seq); if (s.ok()) { // Need to fsync, otherwise it might get lost after a power reset. s = impl->FlushWAL(write_options, false); diff --git a/db/db_impl/db_impl_write.cc b/db/db_impl/db_impl_write.cc index 16a80a17bf..d6639a4b29 100644 --- a/db/db_impl/db_impl_write.cc +++ b/db/db_impl/db_impl_write.cc @@ -157,7 +157,7 @@ Status DBImpl::Write(const WriteOptions& write_options, WriteBatch* my_batch) { if (s.ok()) { s = WriteImpl(write_options, my_batch, /*callback=*/nullptr, /*user_write_cb=*/nullptr, - /*log_used=*/nullptr); + /*wal_used=*/nullptr); } return s; } @@ -314,7 +314,7 @@ Status DBImpl::IngestWBWI(std::shared_ptr wbwi, Status DBImpl::WriteImpl(const WriteOptions& write_options, WriteBatch* my_batch, WriteCallback* callback, - UserWriteCallback* user_write_cb, uint64_t* log_used, + UserWriteCallback* user_write_cb, uint64_t* wal_used, uint64_t log_ref, bool disable_memtable, uint64_t* seq_used, size_t batch_cnt, PreReleaseCallback* pre_release_callback, @@ -444,7 +444,7 @@ Status DBImpl::WriteImpl(const WriteOptions& write_options, // they don't consume sequence. return WriteImplWALOnly( &nonmem_write_thread_, write_options, my_batch, callback, user_write_cb, - log_used, log_ref, seq_used, batch_cnt, pre_release_callback, + wal_used, log_ref, seq_used, batch_cnt, pre_release_callback, assign_order, kDontPublishLastSeq, disable_memtable); } @@ -458,7 +458,7 @@ Status DBImpl::WriteImpl(const WriteOptions& write_options, // sequence in in increasing order, iii) call pre_release_callback serially Status status = WriteImplWALOnly( &write_thread_, write_options, my_batch, callback, user_write_cb, - log_used, log_ref, &seq, sub_batch_cnt, pre_release_callback, + wal_used, log_ref, &seq, sub_batch_cnt, pre_release_callback, kDoAssignOrder, kDoPublishLastSeq, disable_memtable); TEST_SYNC_POINT("DBImpl::WriteImpl:UnorderedWriteAfterWriteWAL"); if (!status.ok()) { @@ -477,7 +477,7 @@ Status DBImpl::WriteImpl(const WriteOptions& write_options, if (immutable_db_options_.enable_pipelined_write) { return PipelinedWriteImpl(write_options, my_batch, callback, user_write_cb, - log_used, log_ref, disable_memtable, seq_used); + wal_used, log_ref, disable_memtable, seq_used); } PERF_TIMER_GUARD(write_pre_and_post_process_time); @@ -535,8 +535,8 @@ Status DBImpl::WriteImpl(const WriteOptions& write_options, // STATE_COMPLETED conditional below handles exit } if (w.state == WriteThread::STATE_COMPLETED) { - if (log_used != nullptr) { - *log_used = w.log_used; + if (wal_used != nullptr) { + *wal_used = w.wal_used; } if (seq_used != nullptr) { *seq_used = w.sequence; @@ -552,7 +552,8 @@ Status DBImpl::WriteImpl(const WriteOptions& write_options, // when it finds suitable, and finish them in the same write batch. // This is how a write job could be done by the other writer. WriteContext write_context; - LogContext log_context(write_options.sync); + // FIXME: also check disableWAL like others? + WalContext wal_context(write_options.sync); WriteThread::WriteGroup write_group; bool in_parallel_group = false; uint64_t last_sequence = kMaxSequenceNumber; @@ -566,7 +567,7 @@ Status DBImpl::WriteImpl(const WriteOptions& write_options, // PreprocessWrite does its own perf timing. PERF_TIMER_STOP(write_pre_and_post_process_time); - status = PreprocessWrite(write_options, &log_context, &write_context); + status = PreprocessWrite(write_options, &wal_context, &write_context); if (!two_write_queues_) { // Assign it after ::PreprocessWrite since the sequence might advance // inside it by WriteRecoverableState @@ -692,23 +693,21 @@ Status DBImpl::WriteImpl(const WriteOptions& write_options, if (!two_write_queues_) { if (status.ok() && !write_options.disableWAL) { - assert(log_context.log_file_number_size); - log_context.prev_size = log_context.writer->file()->GetFileSize(); - LogFileNumberSize& log_file_number_size = - *(log_context.log_file_number_size); + assert(wal_context.wal_file_number_size); + wal_context.prev_size = wal_context.writer->file()->GetFileSize(); PERF_TIMER_GUARD(write_wal_time); - io_s = - WriteToWAL(write_group, log_context.writer, log_used, - log_context.need_log_sync, log_context.need_log_dir_sync, - last_sequence + 1, log_file_number_size); + io_s = WriteGroupToWAL(write_group, wal_context.writer, wal_used, + wal_context.need_wal_sync, + wal_context.need_wal_dir_sync, last_sequence + 1, + *wal_context.wal_file_number_size); } } else { if (status.ok() && !write_options.disableWAL) { PERF_TIMER_GUARD(write_wal_time); // LastAllocatedSequence is increased inside WriteToWAL under // wal_write_mutex_ to ensure ordered events in WAL - io_s = ConcurrentWriteToWAL(write_group, log_used, &last_sequence, - seq_inc); + io_s = ConcurrentWriteGroupToWAL(write_group, wal_used, &last_sequence, + seq_inc); } else { // Otherwise we inc seq number for memtable writes last_sequence = versions_->FetchAddLastAllocatedSequence(seq_inc); @@ -720,16 +719,16 @@ Status DBImpl::WriteImpl(const WriteOptions& write_options, last_sequence += seq_inc; // Seqno assigned to this write are [current_sequence, last_sequence] - if (log_context.need_log_sync) { + if (wal_context.need_wal_sync) { VersionEdit synced_wals; - log_write_mutex_.Lock(); + wal_write_mutex_.Lock(); if (status.ok()) { - MarkLogsSynced(logfile_number_, log_context.need_log_dir_sync, + MarkLogsSynced(cur_wal_number_, wal_context.need_wal_dir_sync, &synced_wals); } else { - MarkLogsNotSynced(logfile_number_); + MarkLogsNotSynced(cur_wal_number_); } - log_write_mutex_.Unlock(); + wal_write_mutex_.Unlock(); if (status.ok() && synced_wals.IsWalAddition()) { InstrumentedMutexLock l(&mutex_); // TODO: plumb Env::IOActivity, Env::IOPriority @@ -764,7 +763,7 @@ Status DBImpl::WriteImpl(const WriteOptions& write_options, writer->sequence = next_sequence; if (writer->pre_release_callback) { Status ws = writer->pre_release_callback->Callback( - writer->sequence, disable_memtable, writer->log_used, index++, + writer->sequence, disable_memtable, writer->wal_used, index++, pre_release_callback_cnt); if (!ws.ok()) { status = pre_release_cb_status = ws; @@ -882,10 +881,10 @@ Status DBImpl::WriteImpl(const WriteOptions& write_options, } } if (!w.status.ok()) { - if (log_context.prev_size < SIZE_MAX) { - InstrumentedMutexLock l(&log_write_mutex_); - if (logs_.back().number == log_context.log_file_number_size->number) { - logs_.back().SetAttemptTruncateSize(log_context.prev_size); + if (wal_context.prev_size < SIZE_MAX) { + InstrumentedMutexLock l(&wal_write_mutex_); + if (logs_.back().number == wal_context.wal_file_number_size->number) { + logs_.back().SetAttemptTruncateSize(wal_context.prev_size); } } HandleMemTableInsertFailure(w.status); @@ -902,7 +901,7 @@ Status DBImpl::WriteImpl(const WriteOptions& write_options, Status DBImpl::PipelinedWriteImpl(const WriteOptions& write_options, WriteBatch* my_batch, WriteCallback* callback, UserWriteCallback* user_write_cb, - uint64_t* log_used, uint64_t log_ref, + uint64_t* wal_used, uint64_t log_ref, bool disable_memtable, uint64_t* seq_used) { PERF_TIMER_GUARD(write_pre_and_post_process_time); StopWatch write_sw(immutable_db_options_.clock, stats_, DB_WRITE); @@ -919,10 +918,10 @@ Status DBImpl::PipelinedWriteImpl(const WriteOptions& write_options, if (w.callback && !w.callback->AllowWriteBatching()) { write_thread_.WaitForMemTableWriters(); } - LogContext log_context(!write_options.disableWAL && write_options.sync); + WalContext wal_context(!write_options.disableWAL && write_options.sync); // PreprocessWrite does its own perf timing. PERF_TIMER_STOP(write_pre_and_post_process_time); - w.status = PreprocessWrite(write_options, &log_context, &write_context); + w.status = PreprocessWrite(write_options, &wal_context, &write_context); PERF_TIMER_START(write_pre_and_post_process_time); // This can set non-OK status if callback fail. @@ -991,13 +990,13 @@ Status DBImpl::PipelinedWriteImpl(const WriteOptions& write_options, wal_write_group.size - 1); RecordTick(stats_, WRITE_DONE_BY_OTHER, wal_write_group.size - 1); } - assert(log_context.log_file_number_size); - LogFileNumberSize& log_file_number_size = - *(log_context.log_file_number_size); - io_s = - WriteToWAL(wal_write_group, log_context.writer, log_used, - log_context.need_log_sync, log_context.need_log_dir_sync, - current_sequence, log_file_number_size); + assert(wal_context.wal_file_number_size); + WalFileNumberSize& wal_file_number_size = + *(wal_context.wal_file_number_size); + io_s = WriteGroupToWAL(wal_write_group, wal_context.writer, wal_used, + wal_context.need_wal_sync, + wal_context.need_wal_dir_sync, current_sequence, + wal_file_number_size); w.status = io_s; } @@ -1009,13 +1008,13 @@ Status DBImpl::PipelinedWriteImpl(const WriteOptions& write_options, } VersionEdit synced_wals; - if (log_context.need_log_sync) { - InstrumentedMutexLock l(&log_write_mutex_); + if (wal_context.need_wal_sync) { + InstrumentedMutexLock l(&wal_write_mutex_); if (w.status.ok()) { - MarkLogsSynced(logfile_number_, log_context.need_log_dir_sync, + MarkLogsSynced(cur_wal_number_, wal_context.need_wal_dir_sync, &synced_wals); } else { - MarkLogsNotSynced(logfile_number_); + MarkLogsNotSynced(cur_wal_number_); } } if (w.status.ok() && synced_wals.IsWalAddition()) { @@ -1156,7 +1155,7 @@ Status DBImpl::UnorderedWriteMemtable(const WriteOptions& write_options, Status DBImpl::WriteImplWALOnly( WriteThread* write_thread, const WriteOptions& write_options, WriteBatch* my_batch, WriteCallback* callback, - UserWriteCallback* user_write_cb, uint64_t* log_used, + UserWriteCallback* user_write_cb, uint64_t* wal_used, const uint64_t log_ref, uint64_t* seq_used, const size_t sub_batch_cnt, PreReleaseCallback* pre_release_callback, const AssignOrder assign_order, const PublishLastSeq publish_last_seq, const bool disable_memtable) { @@ -1169,8 +1168,8 @@ Status DBImpl::WriteImplWALOnly( write_thread->JoinBatchGroup(&w); assert(w.state != WriteThread::STATE_PARALLEL_MEMTABLE_WRITER); if (w.state == WriteThread::STATE_COMPLETED) { - if (log_used != nullptr) { - *log_used = w.log_used; + if (wal_used != nullptr) { + *wal_used = w.wal_used; } if (seq_used != nullptr) { *seq_used = w.sequence; @@ -1186,10 +1185,10 @@ Status DBImpl::WriteImplWALOnly( // TODO(myabandeh): Make preliminary checks thread-safe so we could do them // without paying the cost of obtaining the mutex. - LogContext log_context; + WalContext wal_context; WriteContext write_context; Status status = - PreprocessWrite(write_options, &log_context, &write_context); + PreprocessWrite(write_options, &wal_context, &write_context); WriteStatusCheckOnLocked(status); if (!status.ok()) { @@ -1286,8 +1285,8 @@ Status DBImpl::WriteImplWALOnly( } Status status; if (!write_options.disableWAL) { - IOStatus io_s = - ConcurrentWriteToWAL(write_group, log_used, &last_sequence, seq_inc); + IOStatus io_s = ConcurrentWriteGroupToWAL(write_group, wal_used, + &last_sequence, seq_inc); status = io_s; // last_sequence may not be set if there is an error // This error checking and return is moved up to avoid using uninitialized @@ -1339,7 +1338,7 @@ Status DBImpl::WriteImplWALOnly( if (!writer->CallbackFailed() && writer->pre_release_callback) { assert(writer->sequence != kMaxSequenceNumber); Status ws = writer->pre_release_callback->Callback( - writer->sequence, disable_memtable, writer->log_used, index++, + writer->sequence, disable_memtable, writer->wal_used, index++, pre_release_callback_cnt); if (!ws.ok()) { status = ws; @@ -1421,9 +1420,9 @@ void DBImpl::HandleMemTableInsertFailure(const Status& status) { } Status DBImpl::PreprocessWrite(const WriteOptions& write_options, - LogContext* log_context, + WalContext* wal_context, WriteContext* write_context) { - assert(write_context != nullptr && log_context != nullptr); + assert(write_context != nullptr && wal_context != nullptr); Status status; if (error_handler_.IsDBStopped()) { @@ -1433,7 +1432,8 @@ Status DBImpl::PreprocessWrite(const WriteOptions& write_options, PERF_TIMER_GUARD(write_scheduling_flushes_compactions_time); - if (UNLIKELY(status.ok() && total_log_size_ > GetMaxTotalWalSize())) { + if (UNLIKELY(status.ok() && + wals_total_size_.LoadRelaxed() > GetMaxTotalWalSize())) { assert(versions_); InstrumentedMutexLock l(&mutex_); const ColumnFamilySet* const column_families = @@ -1502,17 +1502,17 @@ Status DBImpl::PreprocessWrite(const WriteOptions& write_options, WriteBufferManagerStallWrites(); } } - InstrumentedMutexLock l(&log_write_mutex_); - if (status.ok() && log_context->need_log_sync) { + InstrumentedMutexLock l(&wal_write_mutex_); + if (status.ok() && wal_context->need_wal_sync) { // Wait until the parallel syncs are finished. Any sync process has to sync // the front log too so it is enough to check the status of front() - // We do a while loop since log_sync_cv_ is signalled when any sync is + // We do a while loop since wal_sync_cv_ is signalled when any sync is // finished // Note: there does not seem to be a reason to wait for parallel sync at // this early step but it is not important since parallel sync (SyncWAL) and - // need_log_sync are usually not used together. + // need_wal_sync are usually not used together. while (logs_.front().IsSyncing()) { - log_sync_cv_.Wait(); + wal_sync_cv_.Wait(); } for (auto& log : logs_) { // This is just to prevent the logs to be synced by a parallel SyncWAL @@ -1523,12 +1523,12 @@ Status DBImpl::PreprocessWrite(const WriteOptions& write_options, log.PrepareForSync(); } } else { - log_context->need_log_sync = false; + wal_context->need_wal_sync = false; } - log_context->writer = logs_.back().writer; - log_context->need_log_dir_sync = - log_context->need_log_dir_sync && !log_dir_synced_; - log_context->log_file_number_size = std::addressof(alive_log_files_.back()); + wal_context->writer = logs_.back().writer; + wal_context->need_wal_dir_sync = + wal_context->need_wal_dir_sync && !wal_dir_synced_; + wal_context->wal_file_number_size = std::addressof(alive_wal_files_.back()); return status; } @@ -1579,12 +1579,12 @@ Status DBImpl::MergeBatch(const WriteThread::WriteGroup& write_group, } // When two_write_queues_ is disabled, this function is called from the only -// write thread. Otherwise this must be called holding log_write_mutex_. +// write thread. Otherwise this must be called holding wal_write_mutex_. IOStatus DBImpl::WriteToWAL(const WriteBatch& merged_batch, const WriteOptions& write_options, - log::Writer* log_writer, uint64_t* log_used, + log::Writer* log_writer, uint64_t* wal_used, uint64_t* log_size, - LogFileNumberSize& log_file_number_size, + WalFileNumberSize& wal_file_number_size, SequenceNumber sequence) { assert(log_size != nullptr); @@ -1596,7 +1596,7 @@ IOStatus DBImpl::WriteToWAL(const WriteBatch& merged_batch, } *log_size = log_entry.size(); // When two_write_queues_ WriteToWAL has to be protected from concurretn calls - // from the two queues anyway and log_write_mutex_ is already held. Otherwise + // from the two queues anyway and wal_write_mutex_ is already held. Otherwise // if manual_wal_flush_ is enabled we need to protect log_writer->AddRecord // from possible concurrent calls via the FlushWAL by the application. const bool needs_locking = manual_wal_flush_ && !two_write_queues_; @@ -1604,7 +1604,7 @@ IOStatus DBImpl::WriteToWAL(const WriteBatch& merged_batch, // manual_wal_flush_ feature (by UNLIKELY) instead of the more common case // when we do not need any locking. if (UNLIKELY(needs_locking)) { - log_write_mutex_.Lock(); + wal_write_mutex_.Lock(); } IOStatus io_s = log_writer->MaybeAddUserDefinedTimestampSizeRecord( write_options, versions_->GetColumnFamiliesTimestampSizeForRecord()); @@ -1614,24 +1614,24 @@ IOStatus DBImpl::WriteToWAL(const WriteBatch& merged_batch, io_s = log_writer->AddRecord(write_options, log_entry, sequence); if (UNLIKELY(needs_locking)) { - log_write_mutex_.Unlock(); + wal_write_mutex_.Unlock(); } - if (log_used != nullptr) { - *log_used = logfile_number_; - assert(*log_used == log_file_number_size.number); + if (wal_used != nullptr) { + *wal_used = cur_wal_number_; + assert(*wal_used == wal_file_number_size.number); } - total_log_size_ += log_entry.size(); - log_file_number_size.AddSize(*log_size); - log_empty_ = false; + wals_total_size_.FetchAddRelaxed(log_entry.size()); + wal_file_number_size.AddSize(*log_size); + wal_empty_ = false; return io_s; } -IOStatus DBImpl::WriteToWAL(const WriteThread::WriteGroup& write_group, - log::Writer* log_writer, uint64_t* log_used, - bool need_log_sync, bool need_log_dir_sync, - SequenceNumber sequence, - LogFileNumberSize& log_file_number_size) { +IOStatus DBImpl::WriteGroupToWAL(const WriteThread::WriteGroup& write_group, + log::Writer* log_writer, uint64_t* wal_used, + bool need_wal_sync, bool need_wal_dir_sync, + SequenceNumber sequence, + WalFileNumberSize& wal_file_number_size) { IOStatus io_s; assert(!two_write_queues_); assert(!write_group.leader->disable_wal); @@ -1646,10 +1646,10 @@ IOStatus DBImpl::WriteToWAL(const WriteThread::WriteGroup& write_group, } if (merged_batch == write_group.leader->batch) { - write_group.leader->log_used = logfile_number_; + write_group.leader->wal_used = cur_wal_number_; } else if (write_with_wal > 1) { for (auto writer : write_group) { - writer->log_used = logfile_number_; + writer->wal_used = cur_wal_number_; } } @@ -1661,14 +1661,14 @@ IOStatus DBImpl::WriteToWAL(const WriteThread::WriteGroup& write_group, WriteOptions write_options; write_options.rate_limiter_priority = write_group.leader->rate_limiter_priority; - io_s = WriteToWAL(*merged_batch, write_options, log_writer, log_used, - &log_size, log_file_number_size, sequence); + io_s = WriteToWAL(*merged_batch, write_options, log_writer, wal_used, + &log_size, wal_file_number_size, sequence); if (to_be_cached_state) { cached_recoverable_state_ = *to_be_cached_state; cached_recoverable_state_empty_ = false; } - if (io_s.ok() && need_log_sync) { + if (io_s.ok() && need_wal_sync) { StopWatch sw(immutable_db_options_.clock, stats_, WAL_FILE_SYNC_MICROS); // It's safe to access logs_ with unlocked mutex_ here because: // - we've set getting_synced=true for all logs, @@ -1678,15 +1678,15 @@ IOStatus DBImpl::WriteToWAL(const WriteThread::WriteGroup& write_group, // - as long as other threads don't modify it, it's safe to read // from std::deque from multiple threads concurrently. // - // Sync operation should work with locked log_write_mutex_, because: + // Sync operation should work with locked wal_write_mutex_, because: // when DBOptions.manual_wal_flush_ is set, // FlushWAL function will be invoked by another thread. - // if without locked log_write_mutex_, the log file may get data + // if without locked wal_write_mutex_, the log file may get data // corruption const bool needs_locking = manual_wal_flush_ && !two_write_queues_; if (UNLIKELY(needs_locking)) { - log_write_mutex_.Lock(); + wal_write_mutex_.Lock(); } if (io_s.ok()) { @@ -1709,10 +1709,10 @@ IOStatus DBImpl::WriteToWAL(const WriteThread::WriteGroup& write_group, } if (UNLIKELY(needs_locking)) { - log_write_mutex_.Unlock(); + wal_write_mutex_.Unlock(); } - if (io_s.ok() && need_log_dir_sync) { + if (io_s.ok() && need_wal_dir_sync) { // We only sync WAL directory the first time WAL syncing is // requested, so that in case users never turn on WAL sync, // we can avoid the disk I/O in the write code path. @@ -1727,7 +1727,7 @@ IOStatus DBImpl::WriteToWAL(const WriteThread::WriteGroup& write_group, } if (io_s.ok()) { auto stats = default_cf_internal_stats_; - if (need_log_sync) { + if (need_wal_sync) { stats->AddDBStats(InternalStats::kIntStatsWalFileSynced, 1); RecordTick(stats_, WAL_FILE_SYNCED); } @@ -1744,8 +1744,8 @@ IOStatus DBImpl::WriteToWAL(const WriteThread::WriteGroup& write_group, return io_s; } -IOStatus DBImpl::ConcurrentWriteToWAL( - const WriteThread::WriteGroup& write_group, uint64_t* log_used, +IOStatus DBImpl::ConcurrentWriteGroupToWAL( + const WriteThread::WriteGroup& write_group, uint64_t* wal_used, SequenceNumber* last_sequence, size_t seq_inc) { IOStatus io_s; @@ -1762,14 +1762,14 @@ IOStatus DBImpl::ConcurrentWriteToWAL( return io_s; } - // We need to lock log_write_mutex_ since logs_ and alive_log_files might be + // We need to lock wal_write_mutex_ since logs_ and alive_wal_files might be // pushed back concurrently - log_write_mutex_.Lock(); + wal_write_mutex_.Lock(); if (merged_batch == write_group.leader->batch) { - write_group.leader->log_used = logfile_number_; + write_group.leader->wal_used = cur_wal_number_; } else if (write_with_wal > 1) { for (auto writer : write_group) { - writer->log_used = logfile_number_; + writer->wal_used = cur_wal_number_; } } *last_sequence = versions_->FetchAddLastAllocatedSequence(seq_inc); @@ -1777,9 +1777,9 @@ IOStatus DBImpl::ConcurrentWriteToWAL( WriteBatchInternal::SetSequence(merged_batch, sequence); log::Writer* log_writer = logs_.back().writer; - LogFileNumberSize& log_file_number_size = alive_log_files_.back(); + WalFileNumberSize& wal_file_number_size = alive_wal_files_.back(); - assert(log_writer->get_log_number() == log_file_number_size.number); + assert(log_writer->get_log_number() == wal_file_number_size.number); uint64_t log_size; @@ -1787,13 +1787,13 @@ IOStatus DBImpl::ConcurrentWriteToWAL( WriteOptions write_options; write_options.rate_limiter_priority = write_group.leader->rate_limiter_priority; - io_s = WriteToWAL(*merged_batch, write_options, log_writer, log_used, - &log_size, log_file_number_size, sequence); + io_s = WriteToWAL(*merged_batch, write_options, log_writer, wal_used, + &log_size, wal_file_number_size, sequence); if (to_be_cached_state) { cached_recoverable_state_ = *to_be_cached_state; cached_recoverable_state_empty_ = false; } - log_write_mutex_.Unlock(); + wal_write_mutex_.Unlock(); if (io_s.ok()) { const bool concurrent = true; @@ -1821,7 +1821,7 @@ Status DBImpl::WriteRecoverableState() { bool dont_care_bool; SequenceNumber next_seq; if (two_write_queues_) { - log_write_mutex_.Lock(); + wal_write_mutex_.Lock(); } SequenceNumber seq; if (two_write_queues_) { @@ -1846,7 +1846,7 @@ Status DBImpl::WriteRecoverableState() { HandleMemTableInsertFailure(status); } if (two_write_queues_) { - log_write_mutex_.Unlock(); + wal_write_mutex_.Unlock(); } if (status.ok() && recoverable_state_pre_release_callback_) { const bool DISABLE_MEMTABLE = true; @@ -1927,11 +1927,11 @@ Status DBImpl::SwitchWAL(WriteContext* write_context) { assert(write_context != nullptr); Status status; - if (alive_log_files_.begin()->getting_flushed) { + if (alive_wal_files_.begin()->getting_flushed) { return status; } - auto oldest_alive_log = alive_log_files_.begin()->number; + auto oldest_alive_log = alive_wal_files_.begin()->number; bool flush_wont_release_oldest_log = false; if (allow_2pc()) { auto oldest_log_with_uncommitted_prep = @@ -1961,14 +1961,14 @@ Status DBImpl::SwitchWAL(WriteContext* write_context) { // transactions then we cannot flush this log until those transactions are // commited. unable_to_release_oldest_log_ = false; - alive_log_files_.begin()->getting_flushed = true; + alive_wal_files_.begin()->getting_flushed = true; } ROCKS_LOG_INFO( immutable_db_options_.info_log, "Flushing all column families with data in WAL number %" PRIu64 ". Total log size is %" PRIu64 " while max_total_wal_size is %" PRIu64, - oldest_alive_log, total_log_size_.load(), GetMaxTotalWalSize()); + oldest_alive_log, wals_total_size_.LoadRelaxed(), GetMaxTotalWalSize()); // no need to refcount because drop is happening in write thread, so can't // happen while we're in the write thread autovector cfds; @@ -2438,21 +2438,21 @@ Status DBImpl::SwitchMemtable(ColumnFamilyData* cfd, WriteContext* context, // Do this without holding the dbmutex lock. assert(versions_->prev_log_number() == 0); if (two_write_queues_) { - log_write_mutex_.Lock(); + wal_write_mutex_.Lock(); } - bool creating_new_log = !log_empty_; + bool creating_new_log = !wal_empty_; if (two_write_queues_) { - log_write_mutex_.Unlock(); + wal_write_mutex_.Unlock(); } uint64_t recycle_log_number = 0; // If file deletion is disabled, don't recycle logs since it'll result in // the file getting renamed if (creating_new_log && immutable_db_options_.recycle_log_file_num && - !log_recycle_files_.empty() && IsFileDeletionsEnabled()) { - recycle_log_number = log_recycle_files_.front(); + !wal_recycle_files_.empty() && IsFileDeletionsEnabled()) { + recycle_log_number = wal_recycle_files_.front(); } uint64_t new_log_number = - creating_new_log ? versions_->NewFileNumber() : logfile_number_; + creating_new_log ? versions_->NewFileNumber() : cur_wal_number_; // For use outside of holding DB mutex const MutableCFOptions mutable_cf_options_copy = cfd->GetLatestMutableCFOptions(); @@ -2478,14 +2478,14 @@ Status DBImpl::SwitchMemtable(ColumnFamilyData* cfd, WriteContext* context, mutex_.Unlock(); if (creating_new_log) { PredecessorWALInfo info; - log_write_mutex_.Lock(); + wal_write_mutex_.Lock(); if (!logs_.empty()) { log::Writer* cur_log_writer = logs_.back().writer; info = PredecessorWALInfo(cur_log_writer->get_log_number(), cur_log_writer->file()->GetFileSize(), cur_log_writer->GetLastSeqnoRecorded()); } - log_write_mutex_.Unlock(); + wal_write_mutex_.Unlock(); // TODO: Write buffer size passed in should be max of all CF's instead // of mutable_cf_options.write_buffer_size. io_s = CreateWAL(write_options, new_log_number, recycle_log_number, @@ -2526,11 +2526,11 @@ Status DBImpl::SwitchMemtable(ColumnFamilyData* cfd, WriteContext* context, // concurrent full purges don't delete the file while we're recycling it. // To achieve that we hold the old log number in the recyclable list until // after it has been renamed. - assert(log_recycle_files_.front() == recycle_log_number); - log_recycle_files_.pop_front(); + assert(wal_recycle_files_.front() == recycle_log_number); + wal_recycle_files_.pop_front(); } if (s.ok() && creating_new_log) { - InstrumentedMutexLock l(&log_write_mutex_); + InstrumentedMutexLock l(&wal_write_mutex_); assert(new_log != nullptr); if (!logs_.empty()) { // Alway flush the buffer of the last log before switching to a new one @@ -2552,11 +2552,11 @@ Status DBImpl::SwitchMemtable(ColumnFamilyData* cfd, WriteContext* context, } } if (s.ok()) { - logfile_number_ = new_log_number; - log_empty_ = true; - log_dir_synced_ = false; - logs_.emplace_back(logfile_number_, new_log); - alive_log_files_.emplace_back(logfile_number_); + cur_wal_number_ = new_log_number; + wal_empty_ = true; + wal_dir_synced_ = false; + logs_.emplace_back(cur_wal_number_, new_log); + alive_wal_files_.emplace_back(cur_wal_number_); } } @@ -2587,7 +2587,7 @@ Status DBImpl::SwitchMemtable(ColumnFamilyData* cfd, WriteContext* context, // obsolete. So we should track the WAL obsoletion event before actually // updating the empty CF's log number. uint64_t min_wal_number_to_keep = - versions_->PreComputeMinLogNumberWithUnflushedData(logfile_number_); + versions_->PreComputeMinLogNumberWithUnflushedData(cur_wal_number_); if (min_wal_number_to_keep > versions_->GetWalSet().GetMinWalNumberToKeep()) { // TODO: plumb Env::IOActivity, Env::IOPriority @@ -2622,7 +2622,7 @@ Status DBImpl::SwitchMemtable(ColumnFamilyData* cfd, WriteContext* context, for (auto cf : empty_cfs) { if (cf->IsEmpty()) { - cf->SetLogNumber(logfile_number_); + cf->SetLogNumber(cur_wal_number_); // MEMPURGE: No need to change this, because new adds // should still receive new sequence numbers. cf->mem()->SetCreationSeq(versions_->LastSequence()); @@ -2639,14 +2639,14 @@ Status DBImpl::SwitchMemtable(ColumnFamilyData* cfd, WriteContext* context, // advance the log number. no need to persist this in the manifest if (cf->IsEmpty()) { if (creating_new_log) { - cf->SetLogNumber(logfile_number_); + cf->SetLogNumber(cur_wal_number_); } cf->mem()->SetCreationSeq(versions_->LastSequence()); } } } - cfd->mem()->SetNextLogNumber(logfile_number_); + cfd->mem()->SetNextLogNumber(cur_wal_number_); assert(new_mem != nullptr); cfd->imm()->Add(cfd->mem(), &context->memtables_to_free_); if (new_imm) { @@ -2658,7 +2658,7 @@ Status DBImpl::SwitchMemtable(ColumnFamilyData* cfd, WriteContext* context, // we always try to flush all immutable memtable. For atomic flush, these // two memtables will be marked eligible for flush in the same call to // AssignAtomicFlushSeq(). - new_imm->SetNextLogNumber(logfile_number_); + new_imm->SetNextLogNumber(cur_wal_number_); cfd->imm()->Add(new_imm, &context->memtables_to_free_); } new_mem->Ref(); diff --git a/db/db_kv_checksum_test.cc b/db/db_kv_checksum_test.cc index 6eea6e5b4b..7d18688f07 100644 --- a/db/db_kv_checksum_test.cc +++ b/db/db_kv_checksum_test.cc @@ -312,12 +312,12 @@ TEST_P(DbKvChecksumTest, WriteToWALCorrupted) { // Corrupted write batch leads to read-only mode, so we have to // reopen for every attempt. Reopen(options); - auto log_size_pre_write = dbfull()->TEST_total_log_size(); + auto log_size_pre_write = dbfull()->TEST_wals_total_size(); SyncPoint::GetInstance()->EnableProcessing(); ASSERT_TRUE(ExecuteWrite(nullptr /* cf_handle */).IsCorruption()); // Confirm that nothing was written to WAL - ASSERT_EQ(log_size_pre_write, dbfull()->TEST_total_log_size()); + ASSERT_EQ(log_size_pre_write, dbfull()->TEST_wals_total_size()); ASSERT_TRUE(dbfull()->TEST_GetBGError().IsCorruption()); SyncPoint::GetInstance()->DisableProcessing(); @@ -350,12 +350,12 @@ TEST_P(DbKvChecksumTest, WriteToWALWithColumnFamilyCorrupted) { // Corrupted write batch leads to read-only mode, so we have to // reopen for every attempt. ReopenWithColumnFamilies({kDefaultColumnFamilyName, "pikachu"}, options); - auto log_size_pre_write = dbfull()->TEST_total_log_size(); + auto log_size_pre_write = dbfull()->TEST_wals_total_size(); SyncPoint::GetInstance()->EnableProcessing(); ASSERT_TRUE(ExecuteWrite(nullptr /* cf_handle */).IsCorruption()); // Confirm that nothing was written to WAL - ASSERT_EQ(log_size_pre_write, dbfull()->TEST_total_log_size()); + ASSERT_EQ(log_size_pre_write, dbfull()->TEST_wals_total_size()); ASSERT_TRUE(dbfull()->TEST_GetBGError().IsCorruption()); SyncPoint::GetInstance()->DisableProcessing(); @@ -487,7 +487,7 @@ TEST_P(DbKvChecksumTestMergedBatch, WriteToWALCorrupted) { // Reopen DB since it failed WAL write which lead to read-only mode Reopen(options); SyncPoint::GetInstance()->EnableProcessing(); - auto log_size_pre_write = dbfull()->TEST_total_log_size(); + auto log_size_pre_write = dbfull()->TEST_wals_total_size(); leader_batch_and_status = GetWriteBatch(GetCFHandleToUse(nullptr, op_type1_), 8 /* protection_bytes_per_key */, op_type1_); @@ -499,7 +499,7 @@ TEST_P(DbKvChecksumTestMergedBatch, WriteToWALCorrupted) { SyncPoint::GetInstance()->ClearCallBack("WriteThread::JoinBatchGroup:Wait"); ASSERT_EQ(1, leader_count); // Nothing should have been written to WAL - ASSERT_EQ(log_size_pre_write, dbfull()->TEST_total_log_size()); + ASSERT_EQ(log_size_pre_write, dbfull()->TEST_wals_total_size()); ASSERT_TRUE(dbfull()->TEST_GetBGError().IsCorruption()); corrupt_byte_offset++; @@ -599,7 +599,7 @@ TEST_P(DbKvChecksumTestMergedBatch, WriteToWALWithColumnFamilyCorrupted) { // Reopen DB since it failed WAL write which lead to read-only mode ReopenWithColumnFamilies({kDefaultColumnFamilyName, "ramen"}, options); SyncPoint::GetInstance()->EnableProcessing(); - auto log_size_pre_write = dbfull()->TEST_total_log_size(); + auto log_size_pre_write = dbfull()->TEST_wals_total_size(); leader_batch_and_status = GetWriteBatch(GetCFHandleToUse(handles_[1], op_type1_), 8 /* protection_bytes_per_key */, op_type1_); @@ -612,7 +612,7 @@ TEST_P(DbKvChecksumTestMergedBatch, WriteToWALWithColumnFamilyCorrupted) { ASSERT_EQ(1, leader_count); // Nothing should have been written to WAL - ASSERT_EQ(log_size_pre_write, dbfull()->TEST_total_log_size()); + ASSERT_EQ(log_size_pre_write, dbfull()->TEST_wals_total_size()); ASSERT_TRUE(dbfull()->TEST_GetBGError().IsCorruption()); corrupt_byte_offset++; diff --git a/db/db_test.cc b/db/db_test.cc index b1c181a1f3..b3511f3eec 100644 --- a/db/db_test.cc +++ b/db/db_test.cc @@ -3407,7 +3407,7 @@ class ModelDB : public DB { } Status GetCurrentWalFile( - std::unique_ptr* /*current_log_file*/) override { + std::unique_ptr* /*current_wal_file*/) override { return Status::OK(); } @@ -6414,7 +6414,7 @@ TEST_F(DBTest, TestLogCleanup) { for (int i = 0; i < 100000; ++i) { ASSERT_OK(Put(Key(i), "val")); - // only 2 memtables will be alive, so logs_to_free needs to always be below + // only 2 memtables will be alive, so wals_to_free needs to always be below // 2 ASSERT_LT(dbfull()->TEST_LogsToFreeSize(), static_cast(3)); } diff --git a/db/db_wal_test.cc b/db/db_wal_test.cc index f89cfe5946..da9ef31587 100644 --- a/db/db_wal_test.cc +++ b/db/db_wal_test.cc @@ -3024,13 +3024,13 @@ TEST_F(DBWALTest, GetCompressedWalsAfterSync) { options.wal_compression = kZSTD; DestroyAndReopen(options); - // Write something to memtable and WAL so that log_empty_ will be false after + // Write something to memtable and WAL so that wal_empty_ will be false after // next DB::Open(). ASSERT_OK(Put("a", "v")); Reopen(options); - // New WAL is created, thanks to !log_empty_. + // New WAL is created, thanks to !wal_empty_. ASSERT_OK(dbfull()->TEST_SwitchWAL()); ASSERT_OK(Put("b", "v")); diff --git a/db/db_write_test.cc b/db/db_write_test.cc index 2dfcd864f5..e3afd219dc 100644 --- a/db/db_write_test.cc +++ b/db/db_write_test.cc @@ -987,7 +987,7 @@ TEST_P(DBWriteTest, RecycleLogToggleTest) { options.recycle_log_file_num = 1; Reopen(options); - // 1.log is added to alive_log_files_ + // 1.log is added to alive_wal_files_ ASSERT_OK(Put(Key(2), "val1")); ASSERT_OK(Flush()); // 1.log should be deleted and not recycled, since it diff --git a/db/job_context.h b/db/job_context.h index 766502ca46..3d2fe933a5 100644 --- a/db/job_context.h +++ b/db/job_context.h @@ -123,7 +123,7 @@ struct JobContext { break; } } - return memtables_to_free.size() > 0 || logs_to_free.size() > 0 || + return memtables_to_free.size() > 0 || wals_to_free.size() > 0 || job_snapshot != nullptr || sv_have_sth; } @@ -193,7 +193,7 @@ struct JobContext { // contexts for installing superversions for multiple column families std::vector superversion_contexts; - autovector logs_to_free; + autovector wals_to_free; // the current manifest_file_number, log_number and prev_log_number // that corresponds to the set of files in 'live'. @@ -207,8 +207,8 @@ struct JobContext { uint64_t prev_log_number; uint64_t min_pending_output = 0; - uint64_t prev_total_log_size = 0; - size_t num_alive_log_files = 0; + uint64_t prev_wals_total_size = 0; + size_t num_alive_wal_files = 0; uint64_t size_log_to_delete = 0; // Snapshot taken before flush/compaction job. @@ -237,18 +237,18 @@ struct JobContext { for (auto m : memtables_to_free) { delete m; } - for (auto l : logs_to_free) { + for (auto l : wals_to_free) { delete l; } memtables_to_free.clear(); - logs_to_free.clear(); + wals_to_free.clear(); job_snapshot.reset(); } ~JobContext() { assert(memtables_to_free.size() == 0); - assert(logs_to_free.size() == 0); + assert(wals_to_free.size() == 0); } }; diff --git a/db/memtable.h b/db/memtable.h index 21532e4566..bd64499024 100644 --- a/db/memtable.h +++ b/db/memtable.h @@ -355,13 +355,13 @@ class ReadOnlyMemTable { // be flushed to storage // REQUIRES: external synchronization to prevent simultaneous // operations on the same MemTable. - uint64_t GetNextLogNumber() const { return mem_next_logfile_number_; } + uint64_t GetNextLogNumber() const { return mem_next_walfile_number_; } // Sets the next active logfile number when this memtable is about to // be flushed to storage // REQUIRES: external synchronization to prevent simultaneous // operations on the same MemTable. - void SetNextLogNumber(uint64_t num) { mem_next_logfile_number_ = num; } + void SetNextLogNumber(uint64_t num) { mem_next_walfile_number_ = num; } // REQUIRES: db_mutex held. void SetID(uint64_t id) { id_ = id; } @@ -516,7 +516,7 @@ class ReadOnlyMemTable { VersionEdit edit_; // The log files earlier than this number can be deleted. - uint64_t mem_next_logfile_number_{0}; + uint64_t mem_next_walfile_number_{0}; // Memtable id to track flush. uint64_t id_ = 0; diff --git a/db/write_thread.h b/db/write_thread.h index 42256970f4..7adf362dcb 100644 --- a/db/write_thread.h +++ b/db/write_thread.h @@ -132,7 +132,7 @@ class WriteThread { size_t protection_bytes_per_key; PreReleaseCallback* pre_release_callback; PostMemTableCallback* post_memtable_callback; - uint64_t log_used; // log number that this batch was inserted into + uint64_t wal_used; // log number that this batch was inserted into uint64_t log_ref; // log number that memtable insert should reference WriteCallback* callback; UserWriteCallback* user_write_cb; @@ -161,7 +161,7 @@ class WriteThread { protection_bytes_per_key(0), pre_release_callback(nullptr), post_memtable_callback(nullptr), - log_used(0), + wal_used(0), log_ref(0), callback(nullptr), user_write_cb(nullptr), @@ -190,7 +190,7 @@ class WriteThread { protection_bytes_per_key(_batch->GetProtectionBytesPerKey()), pre_release_callback(_pre_release_callback), post_memtable_callback(_post_memtable_callback), - log_used(0), + wal_used(0), log_ref(_log_ref), callback(_callback), user_write_cb(_user_write_cb), diff --git a/include/rocksdb/db.h b/include/rocksdb/db.h index 2727bbcb22..58eafe2dc4 100644 --- a/include/rocksdb/db.h +++ b/include/rocksdb/db.h @@ -1893,12 +1893,12 @@ class DB { // Retrieve information about the current wal file // // Note that the log might have rolled after this call in which case - // the current_log_file would not point to the current log file. + // the current_wal_file would not point to the current log file. // - // Additionally, for the sake of optimization current_log_file->StartSequence + // Additionally, for the sake of optimization current_wal_file->StartSequence // would always be set to 0 virtual Status GetCurrentWalFile( - std::unique_ptr* current_log_file) = 0; + std::unique_ptr* current_wal_file) = 0; // IngestExternalFile() will load a list of external SST files (1) into the DB // Two primary modes are supported: diff --git a/include/rocksdb/options.h b/include/rocksdb/options.h index a9b5bb373e..c4b19fea78 100644 --- a/include/rocksdb/options.h +++ b/include/rocksdb/options.h @@ -815,6 +815,7 @@ struct DBOptions { // If it is non empty, the log files will be in the specified dir, // and the db data dir's absolute path will be used as the log file // name's prefix. + // NOTE: not for WALs std::string db_log_dir = ""; // This specifies the absolute dir path for write-ahead logs (WAL). @@ -895,21 +896,24 @@ struct DBOptions { // be created. // If max_log_file_size == 0, all logs will be written to one // log file. + // NOTE: not for WALs size_t max_log_file_size = 0; // Time for the info log file to roll (in seconds). // If specified with non-zero value, log file will be rolled // if it has been active longer than `log_file_time_to_roll`. // Default: 0 (disabled) + // NOTE: not for WALs size_t log_file_time_to_roll = 0; // Maximal info log files to be kept. // Default: 1000 + // NOTE: not for WALs size_t keep_log_file_num = 1000; - // Recycle log files. - // If non-zero, we will reuse previously written log files for new - // logs, overwriting the old data. The value indicates how many + // Recycle WAL files. + // If non-zero, we will reuse previously written WAL files for new + // WALs, overwriting the old data. The value indicates how many // such files we will keep around at any point in time for later // use. This is more efficient because the blocks are already // allocated and fdatasync does not need to update the inode after @@ -1415,9 +1419,10 @@ struct DBOptions { // prefix_same_as_start=true can take advantage of prefix seek optimizations. bool prefix_seek_opt_in_only = false; - // The number of bytes to prefetch when reading the log. This is mostly useful - // for reading a remotely located log, as it can save the number of - // round-trips. If 0, then the prefetching is disabled. + // The number of bytes to prefetch when reading the DB manifest and WAL files + // during DB::Open (and variants). This is mostly useful for reading a + // remotely located log, as it can save the number of round-trips. If 0, then + // the prefetching is disabled. // // Default: 0 size_t log_readahead_size = 0; diff --git a/include/rocksdb/utilities/stackable_db.h b/include/rocksdb/utilities/stackable_db.h index bea4b0d133..b4fa0fc92d 100644 --- a/include/rocksdb/utilities/stackable_db.h +++ b/include/rocksdb/utilities/stackable_db.h @@ -517,8 +517,8 @@ class StackableDB : public DB { } Status GetCurrentWalFile( - std::unique_ptr* current_log_file) override { - return db_->GetCurrentWalFile(current_log_file); + std::unique_ptr* current_wal_file) override { + return db_->GetCurrentWalFile(current_wal_file); } Status GetCreationTimeOfOldestFile(uint64_t* creation_time) override { diff --git a/monitoring/stats_history_test.cc b/monitoring/stats_history_test.cc index 295e7bf3da..fab5914742 100644 --- a/monitoring/stats_history_test.cc +++ b/monitoring/stats_history_test.cc @@ -616,7 +616,7 @@ TEST_F(StatsHistoryTest, ForceManualFlushStatsCF) { // LogNumbers: default: 16, stats: 10, pikachu: 5 // Since in recovery process, cfd_stats column is created after WAL is // created, synced and MANIFEST is persisted, its log number which depends on - // logfile_number_ will be different. Since "pikachu" is never flushed, thus + // cur_wal_number_ will be different. Since "pikachu" is never flushed, thus // its log_number should be the smallest of the three. ASSERT_OK(Flush()); ASSERT_LT(cfd_test->GetLogNumber(), cfd_stats->GetLogNumber()); diff --git a/utilities/transactions/pessimistic_transaction.cc b/utilities/transactions/pessimistic_transaction.cc index e978ad8631..711de009d0 100644 --- a/utilities/transactions/pessimistic_transaction.cc +++ b/utilities/transactions/pessimistic_transaction.cc @@ -811,7 +811,7 @@ Status WriteCommittedTxn::CommitWithoutPrepareInternal() { } auto s = db_impl_->WriteImpl( write_options_, wb, - /*callback*/ nullptr, /*user_write_cb=*/nullptr, /*log_used*/ nullptr, + /*callback*/ nullptr, /*user_write_cb=*/nullptr, /*wal_used*/ nullptr, /*log_ref*/ 0, /*disable_memtable*/ false, &seq_used, /*batch_cnt=*/0, /*pre_release_callback=*/nullptr, post_mem_cb); assert(!s.ok() || seq_used != kMaxSequenceNumber); @@ -825,7 +825,7 @@ Status WriteCommittedTxn::CommitBatchInternal(WriteBatch* batch, size_t) { uint64_t seq_used = kMaxSequenceNumber; auto s = db_impl_->WriteImpl(write_options_, batch, /*callback*/ nullptr, /*user_write_cb=*/nullptr, - /*log_used*/ nullptr, /*log_ref*/ 0, + /*wal_used*/ nullptr, /*log_ref*/ 0, /*disable_memtable*/ false, &seq_used); assert(!s.ok() || seq_used != kMaxSequenceNumber); if (s.ok()) { @@ -917,7 +917,7 @@ Status WriteCommittedTxn::CommitInternal() { s = db_impl_->WriteImpl( write_options_, working_batch, /*callback*/ nullptr, /*user_write_cb=*/nullptr, - /*log_used*/ nullptr, /*log_ref*/ log_number_, + /*wal_used*/ nullptr, /*log_ref*/ log_number_, /*disable_memtable*/ false, &seq_used, /*batch_cnt=*/0, /*pre_release_callback=*/nullptr, post_mem_cb, /*wbwi=*/std::make_shared(std::move(write_batch_)), @@ -929,7 +929,7 @@ Status WriteCommittedTxn::CommitInternal() { } else { s = db_impl_->WriteImpl(write_options_, working_batch, /*callback*/ nullptr, /*user_write_cb=*/nullptr, - /*log_used*/ nullptr, /*log_ref*/ log_number_, + /*wal_used*/ nullptr, /*log_ref*/ log_number_, /*disable_memtable*/ false, &seq_used, /*batch_cnt=*/0, /*pre_release_callback=*/nullptr, post_mem_cb); diff --git a/utilities/transactions/write_unprepared_txn.cc b/utilities/transactions/write_unprepared_txn.cc index 8e9647b8c4..444c1c9b63 100644 --- a/utilities/transactions/write_unprepared_txn.cc +++ b/utilities/transactions/write_unprepared_txn.cc @@ -374,7 +374,7 @@ Status WriteUnpreparedTxn::FlushWriteBatchToDBInternal(bool prepared) { uint64_t seq_used = kMaxSequenceNumber; // log_number_ should refer to the oldest log containing uncommitted data // from the current transaction. This means that if log_number_ is set, - // WriteImpl should not overwrite that value, so set log_used to nullptr if + // WriteImpl should not overwrite that value, so set wal_used to nullptr if // log_number_ is already set. s = db_impl_->WriteImpl(write_options, GetWriteBatch()->GetWriteBatch(), /*callback*/ nullptr, /*user_write_cb=*/nullptr,