Skip to content

Commit

Permalink
Detect
Browse files Browse the repository at this point in the history
  • Loading branch information
@hx235
hx235 committed Dec 12, 2024
1 parent 5fb3f47 commit 86dbf40
Show file tree
Hide file tree
Showing 16 changed files with 850 additions and 51 deletions.
25 changes: 17 additions & 8 deletions db/db_impl/db_impl.h
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2040,15 +2040,18 @@ class DBImpl : public DB {
bool read_only, int job_id, SequenceNumber* next_sequence,
bool* stop_replay_for_corruption, bool* stop_replay_by_wal_filter,
uint64_t* corrupted_wal_number, bool* corrupted_wal_found,
std::unordered_map<int, VersionEdit>* version_edits, bool* flushed);
std::unordered_map<int, VersionEdit>* version_edits, bool* flushed,
PredecessorWALInfo& predecessor_wal_info);

void SetupLogFileProcess(uint64_t wal_number);

Status InitializeLogReader(uint64_t wal_number, bool is_retry,
std::string& fname, bool* const old_log_record,
Status* const reporter_status,
DBOpenLogReporter* reporter,
std::unique_ptr<log::Reader>& reader);
Status InitializeLogReader(
uint64_t wal_number, bool is_retry, std::string& fname,

bool stop_replay_for_corruption, uint64_t min_wal_number,
const PredecessorWALInfo& predecessor_wal_info,
bool* const old_log_record, Status* const reporter_status,
DBOpenLogReporter* reporter, std::unique_ptr<log::Reader>& reader);
Status ProcessLogRecord(
Slice record, const std::unique_ptr<log::Reader>& reader,
const UnorderedMap<uint32_t, size_t>& running_ts_sz, uint64_t wal_number,
Expand Down Expand Up @@ -2085,8 +2088,13 @@ class DBImpl : public DB {
bool* stop_replay_for_corruption, uint64_t* corrupted_wal_number,
bool* corrupted_wal_found);

void FinishLogFileProcess(SequenceNumber const* const next_sequence,
const Status& status);
Status UpdatePredecessorWALInfo(uint64_t wal_number,
SequenceNumber* next_sequence,
const std::string& fname,
PredecessorWALInfo& predecessor_wal_info);

void FinishLogFileProcess(const Status& status,
SequenceNumber const* const next_sequence);

// Return `Status::Corruption()` when `stop_replay_for_corruption == true` and
// exits inconsistency between SST and WAL data
Expand Down Expand Up @@ -2511,6 +2519,7 @@ class DBImpl : public DB {

IOStatus CreateWAL(const WriteOptions& write_options, uint64_t log_file_num,
uint64_t recycle_log_number, size_t preallocate_block_size,
const PredecessorWALInfo& predecessor_wal_info,
log::Writer** new_log);

// Validate self-consistency of DB options
Expand Down
94 changes: 73 additions & 21 deletions db/db_impl/db_impl_open.cc
View file
Original file line number Diff line number Diff line change
Expand Up @@ -754,6 +754,11 @@ Status DBImpl::Recover(
}
}

if (!is_new_db && immutable_db_options_.track_and_verify_wals &&
wal_files.empty()) {
return Status::Corruption("Opening an existing DB with no WAL files");
}

if (immutable_db_options_.track_and_verify_wals_in_manifest) {
if (!immutable_db_options_.best_efforts_recovery) {
// Verify WALs in MANIFEST.
Expand Down Expand Up @@ -1190,14 +1195,15 @@ Status DBImpl::ProcessLogFiles(
bool stop_replay_for_corruption = false;
bool flushed = false;
uint64_t corrupted_wal_number = kMaxSequenceNumber;
PredecessorWALInfo predecessor_wal_info;

for (auto wal_number : wal_numbers) {
if (status.ok()) {
status =
ProcessLogFile(wal_number, min_wal_number, is_retry, read_only,
job_id, next_sequence, &stop_replay_for_corruption,
&stop_replay_by_wal_filter, &corrupted_wal_number,
corrupted_wal_found, version_edits, &flushed);
status = ProcessLogFile(
wal_number, min_wal_number, is_retry, read_only, job_id,
next_sequence, &stop_replay_for_corruption,
&stop_replay_by_wal_filter, &corrupted_wal_number,
corrupted_wal_found, version_edits, &flushed, predecessor_wal_info);
}
}

Expand All @@ -1218,7 +1224,8 @@ Status DBImpl::ProcessLogFile(
int job_id, SequenceNumber* next_sequence, bool* stop_replay_for_corruption,
bool* stop_replay_by_wal_filter, uint64_t* corrupted_wal_number,
bool* corrupted_wal_found,
std::unordered_map<int, VersionEdit>* version_edits, bool* flushed) {
std::unordered_map<int, VersionEdit>* version_edits, bool* flushed,
PredecessorWALInfo& predecessor_wal_info) {
assert(stop_replay_by_wal_filter);

// Variable initialization starts
Expand Down Expand Up @@ -1265,13 +1272,17 @@ Status DBImpl::ProcessLogFile(
}

Status init_status = InitializeLogReader(
wal_number, is_retry, fname, &old_log_record, &status, &reporter, reader);
wal_number, is_retry, fname, *stop_replay_for_corruption, min_wal_number,
predecessor_wal_info, &old_log_record, &status, &reporter, reader);
// FIXME(hx235): Consolidate `!init_status.ok()` and `reader == nullptr` cases
if (!init_status.ok()) {
assert(status.ok());
status.PermitUncheckedError();
return init_status;
} else if (reader == nullptr) {
assert(status.ok());
// Fail initializing log reader for one log file with an ok status.
// Try next one.
return status;
}

Expand Down Expand Up @@ -1305,13 +1316,19 @@ Status DBImpl::ProcessLogFile(
}
}

if (status.ok()) {
status = UpdatePredecessorWALInfo(wal_number, next_sequence, fname,
predecessor_wal_info);
}

if (!status.ok() || old_log_record) {
status = HandleNonOkStatusOrOldLogRecord(
wal_number, next_sequence, status, &old_log_record,
stop_replay_for_corruption, corrupted_wal_number, corrupted_wal_found);
}

FinishLogFileProcess(next_sequence, status);
FinishLogFileProcess(status, next_sequence);

return status;
}

Expand All @@ -1326,12 +1343,12 @@ void DBImpl::SetupLogFileProcess(uint64_t wal_number) {
static_cast<int>(immutable_db_options_.wal_recovery_mode));
}

Status DBImpl::InitializeLogReader(uint64_t wal_number, bool is_retry,
std::string& fname,
bool* const old_log_record,
Status* const reporter_status,
DBOpenLogReporter* reporter,
std::unique_ptr<log::Reader>& reader) {
Status DBImpl::InitializeLogReader(
uint64_t wal_number, bool is_retry, std::string& fname,
bool stop_replay_for_corruption, uint64_t min_wal_number,
const PredecessorWALInfo& predecessor_wal_info, bool* const old_log_record,
Status* const reporter_status, DBOpenLogReporter* reporter,
std::unique_ptr<log::Reader>& reader) {
assert(old_log_record);
assert(reporter_status);
assert(reporter);
Expand Down Expand Up @@ -1369,9 +1386,11 @@ Status DBImpl::InitializeLogReader(uint64_t wal_number, bool is_retry,
// paranoid_checks==false so that corruptions cause entire commits
// to be skipped instead of propagating bad information (like overly
// large sequence numbers).
reader.reset(new log::Reader(immutable_db_options_.info_log,
std::move(file_reader), reporter,
true /*checksum*/, wal_number));
reader.reset(new log::Reader(
immutable_db_options_.info_log, std::move(file_reader), reporter,
true /*checksum*/, wal_number,
immutable_db_options_.track_and_verify_wals, stop_replay_for_corruption,
min_wal_number, predecessor_wal_info));
return status;
}

Expand Down Expand Up @@ -1631,12 +1650,32 @@ Status DBImpl::HandleNonOkStatusOrOldLogRecord(
return status;
}
}
void DBImpl::FinishLogFileProcess(SequenceNumber const* const next_sequence,
const Status& status) {

Status DBImpl::UpdatePredecessorWALInfo(
uint64_t wal_number, SequenceNumber* next_sequence,
const std::string& fname, PredecessorWALInfo& predecessor_wal_info) {
uint64_t bytes;

Status s = env_->GetFileSize(fname, &bytes);
if (!s.ok()) {
return s;
}

assert(next_sequence);
predecessor_wal_info = PredecessorWALInfo(
wal_number, bytes,
*next_sequence == kMaxSequenceNumber ? 0 : *next_sequence - 1);
return s;
}

void DBImpl::FinishLogFileProcess(const Status& status,
SequenceNumber const* const next_sequence) {
if (status.ok()) {
assert(next_sequence);

flush_scheduler_.Clear();
trim_history_scheduler_.Clear();

auto last_sequence = *next_sequence - 1;
if ((*next_sequence != kMaxSequenceNumber) &&
(versions_->LastSequence() <= last_sequence)) {
Expand Down Expand Up @@ -2193,6 +2232,7 @@ Status DB::OpenAndTrimHistory(
IOStatus DBImpl::CreateWAL(const WriteOptions& write_options,
uint64_t log_file_num, uint64_t recycle_log_number,
size_t preallocate_block_size,
const PredecessorWALInfo& predecessor_wal_info,
log::Writer** new_log) {
IOStatus io_s;
std::unique_ptr<FSWritableFile> lfile;
Expand Down Expand Up @@ -2236,9 +2276,15 @@ IOStatus DBImpl::CreateWAL(const WriteOptions& write_options,
*new_log = new log::Writer(std::move(file_writer), log_file_num,
immutable_db_options_.recycle_log_file_num > 0,
immutable_db_options_.manual_wal_flush,
immutable_db_options_.wal_compression);
immutable_db_options_.wal_compression,
immutable_db_options_.track_and_verify_wals);
io_s = (*new_log)->AddCompressionTypeRecord(write_options);
if (io_s.ok()) {
io_s = (*new_log)->MaybeAddPredecessorWALInfo(write_options,
predecessor_wal_info);
}
}

return io_s;
}

Expand Down Expand Up @@ -2331,8 +2377,14 @@ Status DBImpl::Open(const DBOptions& db_options, const std::string& dbname,
log::Writer* new_log = nullptr;
const size_t preallocate_block_size =
impl->GetWalPreallocateBlockSize(max_write_buffer_size);
// TODO(hx235): Pass in correct `predecessor_wal_info` for the first WAL
// created during DB open with predecessor WALs from previous DB session due
// to `avoid_flush_during_recovery == true`. This can protect the last WAL
// recovered.
s = impl->CreateWAL(write_options, new_log_number, 0 /*recycle_log_number*/,
preallocate_block_size, &new_log);
preallocate_block_size,
PredecessorWALInfo() /* predecessor_wal_info */,
&new_log);
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.
Expand Down
13 changes: 11 additions & 2 deletions db/db_impl/db_impl_write.cc
View file
Original file line number Diff line number Diff line change
Expand Up @@ -1406,7 +1406,8 @@ IOStatus DBImpl::WriteToWAL(const WriteBatch& merged_batch,
if (!io_s.ok()) {
return io_s;
}
io_s = log_writer->AddRecord(write_options, log_entry);
io_s = log_writer->AddRecord(write_options, log_entry,
WriteBatchInternal::Sequence(&merged_batch));

if (UNLIKELY(needs_locking)) {
log_write_mutex_.Unlock();
Expand Down Expand Up @@ -2269,8 +2270,16 @@ Status DBImpl::SwitchMemtable(ColumnFamilyData* cfd, WriteContext* context) {
if (creating_new_log) {
// TODO: Write buffer size passed in should be max of all CF's instead
// of mutable_cf_options.write_buffer_size.
PredecessorWALInfo info;
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());
}

io_s = CreateWAL(write_options, new_log_number, recycle_log_number,
preallocate_block_size, &new_log);
preallocate_block_size, info, &new_log);
if (s.ok()) {
s = io_s;
}
Expand Down
Loading

0 comments on commit 86dbf40

Please sign in to comment.