jabberwocky238/rocksdb
1
0
Fork 0
mirror of https://github.com/facebook/rocksdb.git synced 2026-07-07 14:47:40 +08:00
Code Issues Packages Projects Releases Wiki Activity

Refactor wal related naming and more (#13490)

Browse Source 
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
This commit is contained in:
Peter Dillinger
2025-04-11 10:08:29 -07:00
committed by Facebook GitHub Bot
parent 56359da691
commit 2a0ee4ddd8
22 changed files with 407 additions and 427 deletions
Download Patch File Download Diff File Expand all files Collapse all files
db/column_family_test.cc
+1 -1
View File
@@ -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();
}
db/db_filesnapshot.cc
+4 -4
View File
@@ -183,14 +183,14 @@ Status DBImpl::GetSortedWalFilesImpl(VectorWalPtr& files, bool need_seqnos) {
return s;
}
Status DBImpl::GetCurrentWalFile(std::unique_ptr<WalFile>* current_log_file) {
Status DBImpl::GetCurrentWalFile(std::unique_ptr<WalFile>* 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();
db/db_impl/db_impl.cc
+29 -64
View File
@@ -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<uint64_t, uint64_t>& 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);
db/db_impl/db_impl.h
+135 -125
View File
@@ -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<uint64_t, uint64_t>& number_to_size);
Status GetCurrentWalFile(std::unique_ptr<WalFile>* current_log_file) override;
Status GetCurrentWalFile(std::unique_ptr<WalFile>* 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<const Cache*>* 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<Cache> 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<uint64_t> max_total_in_memory_state_;
std::atomic<uint64_t> 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<int> next_job_id_;
std::atomic<int> next_job_id_ = 1;
std::atomic<bool> shutting_down_;
std::atomic<bool> 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<uint64_t>& 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<uint64_t>& 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<int> manual_compaction_paused_;
std::atomic<int> 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<uint64_t> log_recycle_files_;
std::deque<uint64_t> 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<LogFileNumberSize> alive_log_files_;
std::deque<WalFileNumberSize> alive_wal_files_;
// Total size of all "alive" WALs (for easy access without synchronization)
RelaxedAtomic<uint64_t> 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<LogWriterNumber> 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<bool> cached_recoverable_state_empty_ = {true};
std::atomic<uint64_t> total_log_size_;
// If this is non-empty, we need to delete these log files in background
// threads. Protected by log_write_mutex_.
autovector<log::Writer*> logs_to_free_;
// threads. Protected by wal_write_mutex_.
autovector<log::Writer*> wals_to_free_;
bool is_snapshot_supported_;
bool is_snapshot_supported_ = true;
std::map<uint64_t, std::map<std::string, uint64_t>> 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<uint64_t> files_grabbed_for_purge_;
// A queue to store log writers to close. Protected by db mutex_.
std::deque<log::Writer*> logs_to_free_queue_;
std::deque<log::Writer*> wals_to_free_queue_;
std::deque<SuperVersion*> 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<ManualCompactionState*> 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<size_t> pending_memtable_writes_ = {};
std::atomic<size_t> 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<bool> has_unpersisted_data_;
std::atomic<bool> 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<bool> shutdown_initiated_;
std::atomic<bool> 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 {
db/db_impl/db_impl_compaction_flush.cc
+4 -4
View File
@@ -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
db/db_impl/db_impl_debug.cc
+3 -3
View File
@@ -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(
db/db_impl/db_impl_files.cc
+35 -35
View File
@@ -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<uint64_t> blob_live_set(state.blob_live.begin(),
state.blob_live.end());
std::unordered_set<uint64_t> log_recycle_files_set(
std::unordered_set<uint64_t> wal_recycle_files_set(
state.log_recycle_files.begin(), state.log_recycle_files.end());
std::unordered_set<uint64_t> 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<std::string> 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'
db/db_impl/db_impl_open.cc
+18 -18
View File
@@ -1119,7 +1119,7 @@ void DBOpenLogRecordReadReporter::Corruption(size_t bytes, const Status& s,
static_cast<int>(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<uint64_t>& 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);
db/db_impl/db_impl_write.cc
+130 -130
View File
@@ -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<WriteBatchWithIndex> 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<ColumnFamilyData*> 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();
db/db_kv_checksum_test.cc
+8 -8
View File
@@ -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++;
db/db_test.cc
+2 -2
View File
@@ -3407,7 +3407,7 @@ class ModelDB : public DB {
}
Status GetCurrentWalFile(
std::unique_ptr<LogFile>* /*current_log_file*/) override {
std::unique_ptr<LogFile>* /*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<size_t>(3));
}
db/db_wal_test.cc
+2 -2
View File
@@ -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"));
db/db_write_test.cc
+1 -1
View File
@@ -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
db/job_context.h
+7 -7
View File
@@ -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<SuperVersionContext> superversion_contexts;
autovector<log::Writer*> logs_to_free;
autovector<log::Writer*> 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);
}
};
db/memtable.h
+3 -3
View File
@@ -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;
db/write_thread.h
+3 -3
View File
@@ -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),
include/rocksdb/db.h
+3 -3
View File
@@ -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<WalFile>* current_log_file) = 0;
std::unique_ptr<WalFile>* current_wal_file) = 0;
// IngestExternalFile() will load a list of external SST files (1) into the DB
// Two primary modes are supported:
include/rocksdb/options.h
+11 -6
View File
@@ -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;
include/rocksdb/utilities/stackable_db.h
+2 -2
View File
@@ -517,8 +517,8 @@ class StackableDB : public DB {
}
Status GetCurrentWalFile(
std::unique_ptr<WalFile>* current_log_file) override {
return db_->GetCurrentWalFile(current_log_file);
std::unique_ptr<WalFile>* current_wal_file) override {
return db_->GetCurrentWalFile(current_wal_file);
}
Status GetCreationTimeOfOldestFile(uint64_t* creation_time) override {
monitoring/stats_history_test.cc
+1 -1
View File
@@ -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());
utilities/transactions/pessimistic_transaction.cc
+4 -4
View File
@@ -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<WriteBatchWithIndex>(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);
utilities/transactions/write_unprepared_txn.cc
+1 -1
View File
@@ -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,