andreacavalli/rocksdb
0
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Moving sequence number compaction variables from SubCompactionState to CompactionJob

Browse Source 
Summary:
It was pointed out to me that the members of SubCompactionState
'earliest_snapshot', 'latest_snapshot' and 'visible_at_tip' are never
modified by the subcompactions, so they can stay as global varaibles
instead to make things simpler.

Test Plan: make all && make check

Reviewers: sdong, igor, noetzli, anthony, yhchiang

Reviewed By: yhchiang

Subscribers: dhruba

Differential Revision: https://reviews.facebook.net/D45477
This commit is contained in:
Ari Ekmekji 2015-08-25 14:03:10 -07:00
parent bab9934d9e
commit 2f8d71ec05
3 changed files with 29 additions and 44 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
HISTORY.md
View File

@ -7,7 +7,7 @@
### Public API Changes ### Public API Changes
* Removed class Env::RandomRWFile and Env::NewRandomRWFile(). * Removed class Env::RandomRWFile and Env::NewRandomRWFile().
* Renamed DBOptions.num_subcompactions to DBOptions.max_subcompactions to make the name better match the actual funcionality of the option. * Renamed DBOptions.num_subcompactions to DBOptions.max_subcompactions to make the name better match the actual functionality of the option.
## 3.13.0 (8/6/2015) ## 3.13.0 (8/6/2015)
### New Features ### New Features

63
db/compaction_job.cc
View File

@ -100,9 +100,6 @@ struct CompactionJob::SubCompactionState {
uint64_t total_bytes; uint64_t total_bytes;
uint64_t num_input_records; uint64_t num_input_records;
uint64_t num_output_records; uint64_t num_output_records;
SequenceNumber earliest_snapshot;
SequenceNumber visible_at_tip;
SequenceNumber latest_snapshot;
CompactionJobStats compaction_job_stats; CompactionJobStats compaction_job_stats;
// "level_ptrs" holds indices that remember which file of an associated // "level_ptrs" holds indices that remember which file of an associated
@ -113,20 +110,15 @@ struct CompactionJob::SubCompactionState {
// is in or beyond the last file checked during the previous call // is in or beyond the last file checked during the previous call
std::vector<size_t> level_ptrs; std::vector<size_t> level_ptrs;
SubCompactionState(Compaction* c, Slice* _start, Slice* _end, SubCompactionState(Compaction* c, Slice* _start, Slice* _end)
SequenceNumber earliest, SequenceNumber visible, : compaction(c),
SequenceNumber latest) start(_start),
: compaction(c), end(_end),
start(_start), outfile(nullptr),
end(_end), builder(nullptr),
outfile(nullptr), total_bytes(0),
builder(nullptr), num_input_records(0),
total_bytes(0), num_output_records(0) {
num_input_records(0),
num_output_records(0),
earliest_snapshot(earliest),
visible_at_tip(visible),
latest_snapshot(latest) {
assert(compaction != nullptr); assert(compaction != nullptr);
level_ptrs = std::vector<size_t>(compaction->number_levels(), 0); level_ptrs = std::vector<size_t>(compaction->number_levels(), 0);
} }
@ -146,9 +138,6 @@ struct CompactionJob::SubCompactionState {
total_bytes = std::move(o.total_bytes); total_bytes = std::move(o.total_bytes);
num_input_records = std::move(o.num_input_records); num_input_records = std::move(o.num_input_records);
num_output_records = std::move(o.num_output_records); num_output_records = std::move(o.num_output_records);
earliest_snapshot = std::move(o.earliest_snapshot);
visible_at_tip = std::move(o.visible_at_tip);
latest_snapshot = std::move(o.latest_snapshot);
level_ptrs = std::move(o.level_ptrs); level_ptrs = std::move(o.level_ptrs);
return *this; return *this;
} }
@ -321,32 +310,29 @@ void CompactionJob::Prepare() {
bottommost_level_ = c->bottommost_level(); bottommost_level_ = c->bottommost_level();
// Initialize subcompaction states // Initialize subcompaction states
SequenceNumber earliest_snapshot; latest_snapshot_ = 0;
SequenceNumber latest_snapshot = 0; visible_at_tip_ = 0;
SequenceNumber visible_at_tip = 0;
if (existing_snapshots_.size() == 0) { if (existing_snapshots_.size() == 0) {
// optimize for fast path if there are no snapshots // optimize for fast path if there are no snapshots
visible_at_tip = versions_->LastSequence(); visible_at_tip_ = versions_->LastSequence();
earliest_snapshot = visible_at_tip; earliest_snapshot_ = visible_at_tip_;
} else { } else {
latest_snapshot = existing_snapshots_.back(); latest_snapshot_ = existing_snapshots_.back();
// Add the current seqno as the 'latest' virtual // Add the current seqno as the 'latest' virtual
// snapshot to the end of this list. // snapshot to the end of this list.
existing_snapshots_.push_back(versions_->LastSequence()); existing_snapshots_.push_back(versions_->LastSequence());
earliest_snapshot = existing_snapshots_[0]; earliest_snapshot_ = existing_snapshots_[0];
} }
InitializeSubCompactions(earliest_snapshot, visible_at_tip, latest_snapshot); InitializeSubCompactions();
} }
// For L0-L1 compaction, iterators work in parallel by processing // For L0-L1 compaction, iterators work in parallel by processing
// different subsets of the full key range. This function sets up // different subsets of the full key range. This function sets up
// the local states used by each of these subcompactions during // the local states used by each of these subcompactions during
// their execution // their execution
void CompactionJob::InitializeSubCompactions(const SequenceNumber& earliest, void CompactionJob::InitializeSubCompactions() {
const SequenceNumber& visible,
const SequenceNumber& latest) {
Compaction* c = compact_->compaction; Compaction* c = compact_->compaction;
auto& bounds = sub_compaction_boundaries_; auto& bounds = sub_compaction_boundaries_;
if (c->IsSubCompaction()) { if (c->IsSubCompaction()) {
@ -410,8 +396,7 @@ void CompactionJob::InitializeSubCompactions(const SequenceNumber& earliest,
for (size_t i = 0; i <= bounds.size(); i++) { for (size_t i = 0; i <= bounds.size(); i++) {
Slice *start = i == 0 ? nullptr : &bounds[i - 1]; Slice *start = i == 0 ? nullptr : &bounds[i - 1];
Slice *end = i == bounds.size() ? nullptr : &bounds[i]; Slice *end = i == bounds.size() ? nullptr : &bounds[i];
compact_->sub_compact_states.emplace_back(compact_->compaction, start, compact_->sub_compact_states.emplace_back(compact_->compaction, start, end);
end, earliest, visible, latest);
} }
} }
@ -658,8 +643,7 @@ void CompactionJob::ProcessKeyValueCompaction(SubCompactionState* sub_compact) {
visible_in_snapshot = kMaxSequenceNumber; visible_in_snapshot = kMaxSequenceNumber;
// apply the compaction filter to the first occurrence of the user key // apply the compaction filter to the first occurrence of the user key
if (compaction_filter && ikey.type == kTypeValue && if (compaction_filter && ikey.type == kTypeValue &&
(sub_compact->visible_at_tip || (visible_at_tip_ || ikey.sequence > latest_snapshot_)) {
ikey.sequence > sub_compact->latest_snapshot)) {
// If the user has specified a compaction filter and the sequence // If the user has specified a compaction filter and the sequence
// number is greater than any external snapshot, then invoke the // number is greater than any external snapshot, then invoke the
// filter. If the return value of the compaction filter is true, // filter. If the return value of the compaction filter is true,
@ -695,9 +679,8 @@ void CompactionJob::ProcessKeyValueCompaction(SubCompactionState* sub_compact) {
// the earlist snapshot that is affected by this kv. // the earlist snapshot that is affected by this kv.
SequenceNumber prev_snapshot = 0; // 0 means no previous snapshot SequenceNumber prev_snapshot = 0; // 0 means no previous snapshot
SequenceNumber visible = SequenceNumber visible =
sub_compact->visible_at_tip visible_at_tip_ ? visible_at_tip_ : findEarliestVisibleSnapshot(
? sub_compact->visible_at_tip ikey.sequence, &prev_snapshot);
: findEarliestVisibleSnapshot(ikey.sequence, &prev_snapshot);
if (visible_in_snapshot == visible) { if (visible_in_snapshot == visible) {
// If the earliest snapshot is which this key is visible in // If the earliest snapshot is which this key is visible in
@ -709,7 +692,7 @@ void CompactionJob::ProcessKeyValueCompaction(SubCompactionState* sub_compact) {
++key_drop_newer_entry; ++key_drop_newer_entry;
input->Next(); // (A) input->Next(); // (A)
} else if (ikey.type == kTypeDeletion && } else if (ikey.type == kTypeDeletion &&
ikey.sequence <= sub_compact->earliest_snapshot && ikey.sequence <= earliest_snapshot_ &&
sub_compact->compaction->KeyNotExistsBeyondOutputLevel( sub_compact->compaction->KeyNotExistsBeyondOutputLevel(
ikey.user_key, &sub_compact->level_ptrs)) { ikey.user_key, &sub_compact->level_ptrs)) {
// For this user key: // For this user key:
@ -815,7 +798,7 @@ Status CompactionJob::WriteKeyValue(const Slice& key, const Slice& value,
// If this is the bottommost level (no files in lower levels) // If this is the bottommost level (no files in lower levels)
// and the earliest snapshot is larger than this seqno // and the earliest snapshot is larger than this seqno
// then we can squash the seqno to zero. // then we can squash the seqno to zero.
if (bottommost_level_ && ikey.sequence < sub_compact->earliest_snapshot && if (bottommost_level_ && ikey.sequence < earliest_snapshot_ &&
ikey.type != kTypeMerge) { ikey.type != kTypeMerge) {
assert(ikey.type != kTypeDeletion); assert(ikey.type != kTypeDeletion);
// make a copy because updating in place would cause problems // make a copy because updating in place would cause problems

8
db/compaction_job.h
View File

@ -83,9 +83,7 @@ class CompactionJob {
void AggregateStatistics(); void AggregateStatistics();
// Set up the individual states used by each subcompaction // Set up the individual states used by each subcompaction
void InitializeSubCompactions(const SequenceNumber& earliest, void InitializeSubCompactions();
const SequenceNumber& visible,
const SequenceNumber& latest);
// update the thread status for starting a compaction. // update the thread status for starting a compaction.
void ReportStartedCompaction(Compaction* compaction); void ReportStartedCompaction(Compaction* compaction);
@ -132,6 +130,10 @@ class CompactionJob {
InternalStats::CompactionStats compaction_stats_; InternalStats::CompactionStats compaction_stats_;
SequenceNumber earliest_snapshot_;
SequenceNumber latest_snapshot_;
SequenceNumber visible_at_tip_;
// DBImpl state // DBImpl state
const std::string& dbname_; const std::string& dbname_;
const DBOptions& db_options_; const DBOptions& db_options_;