NextGraph
/
rocksdb
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Fix some variable naming in db/transaction_log_impl.* (#5112)

Browse Source 
Summary:
We follow Google C++ Style which indicates variable names should be
all underscore: https://google.github.io/styleguide/cppguide.html#Variable_Names
Fix some variable names under db/transaction_log_impl.*
Pull Request resolved: https://github.com/facebook/rocksdb/pull/5112

Differential Revision: D14631157

Pulled By: siying

fbshipit-source-id: 9525c9b0976b843bca377b03897700d87cc60af8
main
Siying Dong 6 years ago committed by Facebook Github Bot
parent 8c072044d2
commit 5f6adf3f6a
3 changed files with 92 additions and 93 deletions
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Show Stats Download Patch File Download Diff File
  1. 4
      db/perf_context_test.cc
  2. 158
      db/transaction_log_impl.cc
  3. 23
      db/transaction_log_impl.h

4
db/perf_context_test.cc
Unescape View File

@ -841,8 +841,8 @@ TEST_F(PerfContextTest, CPUTimer) {
ASSERT_EQ(value, "v0"); ASSERT_EQ(value, "v0");
if (FLAGS_verbose) { if (FLAGS_verbose) {
std::cout << "Get CPU time nanos: " std::cout << "Get CPU time nanos: " << get_perf_context()->get_cpu_nanos
<< get_perf_context()->get_cpu_nanos << "ns\n"; << "ns\n";
} }
// Iter // Iter

158
db/transaction_log_impl.cc
Unescape View File

@ -25,13 +25,13 @@ TransactionLogIteratorImpl::TransactionLogIteratorImpl(
options_(options), options_(options),
read_options_(read_options), read_options_(read_options),
soptions_(soptions), soptions_(soptions),
startingSequenceNumber_(seq), starting_sequence_number_(seq),
files_(std::move(files)), files_(std::move(files)),
started_(false), started_(false),
isValid_(false), is_valid_(false),
currentFileIndex_(0), current_file_index_(0),
currentBatchSeq_(0), current_batch_seq_(0),
currentLastSeq_(0), current_last_seq_(0),
versions_(versions), versions_(versions),
seq_per_batch_(seq_per_batch) { seq_per_batch_(seq_per_batch) {
assert(files_ != nullptr); assert(files_ != nullptr);
@ -43,23 +43,23 @@ TransactionLogIteratorImpl::TransactionLogIteratorImpl(
} }
Status TransactionLogIteratorImpl::OpenLogFile( Status TransactionLogIteratorImpl::OpenLogFile(
const LogFile* logFile, const LogFile* log_file,
std::unique_ptr<SequentialFileReader>* file_reader) { std::unique_ptr<SequentialFileReader>* file_reader) {
Env* env = options_->env; Env* env = options_->env;
std::unique_ptr<SequentialFile> file; std::unique_ptr<SequentialFile> file;
std::string fname; std::string fname;
Status s; Status s;
EnvOptions optimized_env_options = env->OptimizeForLogRead(soptions_); EnvOptions optimized_env_options = env->OptimizeForLogRead(soptions_);
if (logFile->Type() == kArchivedLogFile) { if (log_file->Type() == kArchivedLogFile) {
fname = ArchivedLogFileName(dir_, logFile->LogNumber()); fname = ArchivedLogFileName(dir_, log_file->LogNumber());
s = env->NewSequentialFile(fname, &file, optimized_env_options); s = env->NewSequentialFile(fname, &file, optimized_env_options);
} else { } else {
fname = LogFileName(dir_, logFile->LogNumber()); fname = LogFileName(dir_, log_file->LogNumber());
s = env->NewSequentialFile(fname, &file, optimized_env_options); s = env->NewSequentialFile(fname, &file, optimized_env_options);
if (!s.ok()) { if (!s.ok()) {
// If cannot open file in DB directory. // If cannot open file in DB directory.
// Try the archive dir, as it could have moved in the meanwhile. // Try the archive dir, as it could have moved in the meanwhile.
fname = ArchivedLogFileName(dir_, logFile->LogNumber()); fname = ArchivedLogFileName(dir_, log_file->LogNumber());
s = env->NewSequentialFile(fname, &file, optimized_env_options); s = env->NewSequentialFile(fname, &file, optimized_env_options);
} }
} }
@ -70,45 +70,41 @@ Status TransactionLogIteratorImpl::OpenLogFile(
} }
BatchResult TransactionLogIteratorImpl::GetBatch() { BatchResult TransactionLogIteratorImpl::GetBatch() {
assert(isValid_); // cannot call in a non valid state. assert(is_valid_); // cannot call in a non valid state.
BatchResult result; BatchResult result;
result.sequence = currentBatchSeq_; result.sequence = current_batch_seq_;
result.writeBatchPtr = std::move(currentBatch_); result.writeBatchPtr = std::move(current_batch_);
return result; return result;
} }
Status TransactionLogIteratorImpl::status() { Status TransactionLogIteratorImpl::status() { return current_status_; }
return currentStatus_;
}
bool TransactionLogIteratorImpl::Valid() { bool TransactionLogIteratorImpl::Valid() { return started_ && is_valid_; }
return started_ && isValid_;
}
bool TransactionLogIteratorImpl::RestrictedRead( bool TransactionLogIteratorImpl::RestrictedRead(
Slice* record, Slice* record,
std::string* scratch) { std::string* scratch) {
// Don't read if no more complete entries to read from logs // Don't read if no more complete entries to read from logs
if (currentLastSeq_ >= versions_->LastSequence()) { if (current_last_seq_ >= versions_->LastSequence()) {
return false; return false;
} }
return currentLogReader_->ReadRecord(record, scratch); return current_log_reader_->ReadRecord(record, scratch);
} }
void TransactionLogIteratorImpl::SeekToStartSequence( void TransactionLogIteratorImpl::SeekToStartSequence(uint64_t start_file_index,
uint64_t startFileIndex, bool strict) {
bool strict) {
std::string scratch; std::string scratch;
Slice record; Slice record;
started_ = false; started_ = false;
isValid_ = false; is_valid_ = false;
if (files_->size() <= startFileIndex) { if (files_->size() <= start_file_index) {
return; return;
} }
Status s = OpenLogReader(files_->at(static_cast<size_t>(startFileIndex)).get()); Status s =
OpenLogReader(files_->at(static_cast<size_t>(start_file_index)).get());
if (!s.ok()) { if (!s.ok()) {
currentStatus_ = s; current_status_ = s;
reporter_.Info(currentStatus_.ToString().c_str()); reporter_.Info(current_status_.ToString().c_str());
return; return;
} }
while (RestrictedRead(&record, &scratch)) { while (RestrictedRead(&record, &scratch)) {
@ -118,21 +114,22 @@ void TransactionLogIteratorImpl::SeekToStartSequence(
continue; continue;
} }
UpdateCurrentWriteBatch(record); UpdateCurrentWriteBatch(record);
if (currentLastSeq_ >= startingSequenceNumber_) { if (current_last_seq_ >= starting_sequence_number_) {
if (strict && currentBatchSeq_ != startingSequenceNumber_) { if (strict && current_batch_seq_ != starting_sequence_number_) {
currentStatus_ = Status::Corruption("Gap in sequence number. Could not " current_status_ = Status::Corruption(
"seek to required sequence number"); "Gap in sequence number. Could not "
reporter_.Info(currentStatus_.ToString().c_str()); "seek to required sequence number");
reporter_.Info(current_status_.ToString().c_str());
return; return;
} else if (strict) { } else if (strict) {
reporter_.Info("Could seek required sequence number. Iterator will " reporter_.Info("Could seek required sequence number. Iterator will "
"continue."); "continue.");
} }
isValid_ = true; is_valid_ = true;
started_ = true; // set started_ as we could seek till starting sequence started_ = true; // set started_ as we could seek till starting sequence
return; return;
} else { } else {
isValid_ = false; is_valid_ = false;
} }
} }
@ -141,13 +138,15 @@ void TransactionLogIteratorImpl::SeekToStartSequence(
// If strict is set, we want to seek exactly till the start sequence and it // If strict is set, we want to seek exactly till the start sequence and it
// should have been present in the file we scanned above // should have been present in the file we scanned above
if (strict) { if (strict) {
currentStatus_ = Status::Corruption("Gap in sequence number. Could not " current_status_ = Status::Corruption(
"seek to required sequence number"); "Gap in sequence number. Could not "
reporter_.Info(currentStatus_.ToString().c_str()); "seek to required sequence number");
reporter_.Info(current_status_.ToString().c_str());
} else if (files_->size() != 1) { } else if (files_->size() != 1) {
currentStatus_ = Status::Corruption("Start sequence was not found, " current_status_ = Status::Corruption(
"skipping to the next available"); "Start sequence was not found, "
reporter_.Info(currentStatus_.ToString().c_str()); "skipping to the next available");
reporter_.Info(current_status_.ToString().c_str());
// Let NextImpl find the next available entry. started_ remains false // Let NextImpl find the next available entry. started_ remains false
// because we don't want to check for gaps while moving to start sequence // because we don't want to check for gaps while moving to start sequence
NextImpl(true); NextImpl(true);
@ -161,15 +160,15 @@ void TransactionLogIteratorImpl::Next() {
void TransactionLogIteratorImpl::NextImpl(bool internal) { void TransactionLogIteratorImpl::NextImpl(bool internal) {
std::string scratch; std::string scratch;
Slice record; Slice record;
isValid_ = false; is_valid_ = false;
if (!internal && !started_) { if (!internal && !started_) {
// Runs every time until we can seek to the start sequence // Runs every time until we can seek to the start sequence
return SeekToStartSequence(); return SeekToStartSequence();
} }
while(true) { while(true) {
assert(currentLogReader_); assert(current_log_reader_);
if (currentLogReader_->IsEOF()) { if (current_log_reader_->IsEOF()) {
currentLogReader_->UnmarkEOF(); current_log_reader_->UnmarkEOF();
} }
while (RestrictedRead(&record, &scratch)) { while (RestrictedRead(&record, &scratch)) {
if (record.size() < WriteBatchInternal::kHeader) { if (record.size() < WriteBatchInternal::kHeader) {
@ -190,20 +189,20 @@ void TransactionLogIteratorImpl::NextImpl(bool internal) {
} }
// Open the next file // Open the next file
if (currentFileIndex_ < files_->size() - 1) { if (current_file_index_ < files_->size() - 1) {
++currentFileIndex_; ++current_file_index_;
Status s = OpenLogReader(files_->at(currentFileIndex_).get()); Status s = OpenLogReader(files_->at(current_file_index_).get());
if (!s.ok()) { if (!s.ok()) {
isValid_ = false; is_valid_ = false;
currentStatus_ = s; current_status_ = s;
return; return;
} }
} else { } else {
isValid_ = false; is_valid_ = false;
if (currentLastSeq_ == versions_->LastSequence()) { if (current_last_seq_ == versions_->LastSequence()) {
currentStatus_ = Status::OK(); current_status_ = Status::OK();
} else { } else {
currentStatus_ = Status::Corruption("NO MORE DATA LEFT"); current_status_ = Status::Corruption("NO MORE DATA LEFT");
} }
return; return;
} }
@ -211,17 +210,16 @@ void TransactionLogIteratorImpl::NextImpl(bool internal) {
} }
bool TransactionLogIteratorImpl::IsBatchExpected( bool TransactionLogIteratorImpl::IsBatchExpected(
const WriteBatch* batch, const WriteBatch* batch, const SequenceNumber expected_seq) {
const SequenceNumber expectedSeq) {
assert(batch); assert(batch);
SequenceNumber batchSeq = WriteBatchInternal::Sequence(batch); SequenceNumber batchSeq = WriteBatchInternal::Sequence(batch);
if (batchSeq != expectedSeq) { if (batchSeq != expected_seq) {
char buf[200]; char buf[200];
snprintf(buf, sizeof(buf), snprintf(buf, sizeof(buf),
"Discontinuity in log records. Got seq=%" PRIu64 "Discontinuity in log records. Got seq=%" PRIu64
", Expected seq=%" PRIu64 ", Last flushed seq=%" PRIu64 ", Expected seq=%" PRIu64 ", Last flushed seq=%" PRIu64
".Log iterator will reseek the correct batch.", ".Log iterator will reseek the correct batch.",
batchSeq, expectedSeq, versions_->LastSequence()); batchSeq, expected_seq, versions_->LastSequence());
reporter_.Info(buf); reporter_.Info(buf);
return false; return false;
} }
@ -232,25 +230,25 @@ void TransactionLogIteratorImpl::UpdateCurrentWriteBatch(const Slice& record) {
std::unique_ptr<WriteBatch> batch(new WriteBatch()); std::unique_ptr<WriteBatch> batch(new WriteBatch());
WriteBatchInternal::SetContents(batch.get(), record); WriteBatchInternal::SetContents(batch.get(), record);
SequenceNumber expectedSeq = currentLastSeq_ + 1; SequenceNumber expected_seq = current_last_seq_ + 1;
// If the iterator has started, then confirm that we get continuous batches // If the iterator has started, then confirm that we get continuous batches
if (started_ && !IsBatchExpected(batch.get(), expectedSeq)) { if (started_ && !IsBatchExpected(batch.get(), expected_seq)) {
// Seek to the batch having expected sequence number // Seek to the batch having expected sequence number
if (expectedSeq < files_->at(currentFileIndex_)->StartSequence()) { if (expected_seq < files_->at(current_file_index_)->StartSequence()) {
// Expected batch must lie in the previous log file // Expected batch must lie in the previous log file
// Avoid underflow. // Avoid underflow.
if (currentFileIndex_ != 0) { if (current_file_index_ != 0) {
currentFileIndex_--; current_file_index_--;
} }
} }
startingSequenceNumber_ = expectedSeq; starting_sequence_number_ = expected_seq;
// currentStatus_ will be set to Ok if reseek succeeds // currentStatus_ will be set to Ok if reseek succeeds
// Note: this is still ok in seq_pre_batch_ && two_write_queuesp_ mode // Note: this is still ok in seq_pre_batch_ && two_write_queuesp_ mode
// that allows gaps in the WAL since it will still skip over the gap. // that allows gaps in the WAL since it will still skip over the gap.
currentStatus_ = Status::NotFound("Gap in sequence numbers"); current_status_ = Status::NotFound("Gap in sequence numbers");
// In seq_per_batch_ mode, gaps in the seq are possible so the strict mode // In seq_per_batch_ mode, gaps in the seq are possible so the strict mode
// should be disabled // should be disabled
return SeekToStartSequence(currentFileIndex_, !seq_per_batch_); return SeekToStartSequence(current_file_index_, !seq_per_batch_);
} }
struct BatchCounter : public WriteBatch::Handler { struct BatchCounter : public WriteBatch::Handler {
@ -289,33 +287,33 @@ void TransactionLogIteratorImpl::UpdateCurrentWriteBatch(const Slice& record) {
Status MarkRollback(const Slice&) override { return Status::OK(); } Status MarkRollback(const Slice&) override { return Status::OK(); }
}; };
currentBatchSeq_ = WriteBatchInternal::Sequence(batch.get()); current_batch_seq_ = WriteBatchInternal::Sequence(batch.get());
if (seq_per_batch_) { if (seq_per_batch_) {
BatchCounter counter(currentBatchSeq_); BatchCounter counter(current_batch_seq_);
batch->Iterate(&counter); batch->Iterate(&counter);
currentLastSeq_ = counter.sequence_; current_last_seq_ = counter.sequence_;
} else { } else {
currentLastSeq_ = current_last_seq_ =
currentBatchSeq_ + WriteBatchInternal::Count(batch.get()) - 1; current_batch_seq_ + WriteBatchInternal::Count(batch.get()) - 1;
} }
// currentBatchSeq_ can only change here // currentBatchSeq_ can only change here
assert(currentLastSeq_ <= versions_->LastSequence()); assert(current_last_seq_ <= versions_->LastSequence());
currentBatch_ = std::move(batch); current_batch_ = std::move(batch);
isValid_ = true; is_valid_ = true;
currentStatus_ = Status::OK(); current_status_ = Status::OK();
} }
Status TransactionLogIteratorImpl::OpenLogReader(const LogFile* logFile) { Status TransactionLogIteratorImpl::OpenLogReader(const LogFile* log_file) {
std::unique_ptr<SequentialFileReader> file; std::unique_ptr<SequentialFileReader> file;
Status s = OpenLogFile(logFile, &file); Status s = OpenLogFile(log_file, &file);
if (!s.ok()) { if (!s.ok()) {
return s; return s;
} }
assert(file); assert(file);
currentLogReader_.reset( current_log_reader_.reset(
new log::Reader(options_->info_log, std::move(file), &reporter_, new log::Reader(options_->info_log, std::move(file), &reporter_,
read_options_.verify_checksums_, logFile->LogNumber())); read_options_.verify_checksums_, log_file->LogNumber()));
return Status::OK(); return Status::OK();
} }
} // namespace rocksdb } // namespace rocksdb

23
db/transaction_log_impl.h
Unescape View File

@ -78,15 +78,15 @@ class TransactionLogIteratorImpl : public TransactionLogIterator {
const ImmutableDBOptions* options_; const ImmutableDBOptions* options_;
const TransactionLogIterator::ReadOptions read_options_; const TransactionLogIterator::ReadOptions read_options_;
const EnvOptions& soptions_; const EnvOptions& soptions_;
SequenceNumber startingSequenceNumber_; SequenceNumber starting_sequence_number_;
std::unique_ptr<VectorLogPtr> files_; std::unique_ptr<VectorLogPtr> files_;
bool started_; bool started_;
bool isValid_; // not valid when it starts of. bool is_valid_; // not valid when it starts of.
Status currentStatus_; Status current_status_;
size_t currentFileIndex_; size_t current_file_index_;
std::unique_ptr<WriteBatch> currentBatch_; std::unique_ptr<WriteBatch> current_batch_;
std::unique_ptr<log::Reader> currentLogReader_; std::unique_ptr<log::Reader> current_log_reader_;
Status OpenLogFile(const LogFile* logFile, Status OpenLogFile(const LogFile* log_file,
std::unique_ptr<SequentialFileReader>* file); std::unique_ptr<SequentialFileReader>* file);
struct LogReporter : public log::Reader::Reporter { struct LogReporter : public log::Reader::Reporter {
@ -99,8 +99,9 @@ class TransactionLogIteratorImpl : public TransactionLogIterator {
virtual void Info(const char* s) { ROCKS_LOG_INFO(info_log, "%s", s); } virtual void Info(const char* s) { ROCKS_LOG_INFO(info_log, "%s", s); }
} reporter_; } reporter_;
SequenceNumber currentBatchSeq_; // sequence number at start of current batch SequenceNumber
SequenceNumber currentLastSeq_; // last sequence in the current batch current_batch_seq_; // sequence number at start of current batch
SequenceNumber current_last_seq_; // last sequence in the current batch
// Used only to get latest seq. num // Used only to get latest seq. num
// TODO(icanadi) can this be just a callback? // TODO(icanadi) can this be just a callback?
VersionSet const* const versions_; VersionSet const* const versions_;
@ -109,14 +110,14 @@ class TransactionLogIteratorImpl : public TransactionLogIterator {
bool RestrictedRead(Slice* record, std::string* scratch); bool RestrictedRead(Slice* record, std::string* scratch);
// Seeks to startingSequenceNumber reading from startFileIndex in files_. // Seeks to startingSequenceNumber reading from startFileIndex in files_.
// If strict is set,then must get a batch starting with startingSequenceNumber // If strict is set,then must get a batch starting with startingSequenceNumber
void SeekToStartSequence(uint64_t startFileIndex = 0, bool strict = false); void SeekToStartSequence(uint64_t start_file_index = 0, bool strict = false);
// Implementation of Next. SeekToStartSequence calls it internally with // Implementation of Next. SeekToStartSequence calls it internally with
// internal=true to let it find next entry even if it has to jump gaps because // internal=true to let it find next entry even if it has to jump gaps because
// the iterator may start off from the first available entry but promises to // the iterator may start off from the first available entry but promises to
// be continuous after that // be continuous after that
void NextImpl(bool internal = false); void NextImpl(bool internal = false);
// Check if batch is expected, else return false // Check if batch is expected, else return false
bool IsBatchExpected(const WriteBatch* batch, SequenceNumber expectedSeq); bool IsBatchExpected(const WriteBatch* batch, SequenceNumber expected_seq);
// Update current batch if a continuous batch is found, else return false // Update current batch if a continuous batch is found, else return false
void UpdateCurrentWriteBatch(const Slice& record); void UpdateCurrentWriteBatch(const Slice& record);
Status OpenLogReader(const LogFile* file); Status OpenLogReader(const LogFile* file);

Write Preview
Loading…
Cancel
Save