Merge branch 'master' into columnfamilies

Conflicts:
	db/version_set.h

db/db_impl.cc

@@ -1781,7 +1781,7 @@ void DBImpl::MaybeScheduleFlushOrCompaction() {
     // max_background_compactions hasn't been reached and, in case
     // bg_manual_only_ > 0, if it's a manual compaction.
     if ((manual_compaction_ ||
-         versions_->NeedsCompaction() ||
+         versions_->current()->NeedsCompaction() ||
          (is_flush_pending && (options_.max_background_flushes <= 0))) &&
         bg_compaction_scheduled_ < options_.max_background_compactions &&
         (!bg_manual_only_ || manual_compaction_)) {
@@ -3408,12 +3408,11 @@ Status DBImpl::MakeRoomForWrite(bool force,
       RecordTick(options_.statistics.get(), STALL_L0_NUM_FILES_MICROS, stall);
       stall_level0_num_files_ += stall;
       stall_level0_num_files_count_++;
-    } else if (
-        allow_hard_rate_limit_delay &&
-        options_.hard_rate_limit > 1.0 &&
-        (score = versions_->MaxCompactionScore()) > options_.hard_rate_limit) {
+    } else if (allow_hard_rate_limit_delay && options_.hard_rate_limit > 1.0 &&
+               (score = versions_->current()->MaxCompactionScore()) >
+                   options_.hard_rate_limit) {
       // Delay a write when the compaction score for any level is too large.
-      int max_level = versions_->MaxCompactionScoreLevel();
+      int max_level = versions_->current()->MaxCompactionScoreLevel();
       mutex_.Unlock();
       uint64_t delayed;
       {
@@ -3435,10 +3434,9 @@ Status DBImpl::MakeRoomForWrite(bool force,
         allow_hard_rate_limit_delay = false;
       }
       mutex_.Lock();
-    } else if (
-        allow_soft_rate_limit_delay &&
-        options_.soft_rate_limit > 0.0 &&
-        (score = versions_->MaxCompactionScore()) > options_.soft_rate_limit) {
+    } else if (allow_soft_rate_limit_delay && options_.soft_rate_limit > 0.0 &&
+               (score = versions_->current()->MaxCompactionScore()) >
+                   options_.soft_rate_limit) {
       // Delay a write when the compaction score for any level is too large.
       // TODO: add statistics
       mutex_.Unlock();

db/memtablelist.cc

@@ -31,6 +31,7 @@ MemTableListVersion::MemTableListVersion(MemTableListVersion* old) {
 void MemTableListVersion::Ref() { ++refs_; }
 
 void MemTableListVersion::Unref(std::vector<MemTable*>* to_delete) {
+  assert(refs_ >= 1);
   --refs_;
   if (refs_ == 0) {
     // if to_delete is equal to nullptr it means we're confident
@@ -75,17 +76,16 @@ void MemTableListVersion::AddIterators(const ReadOptions& options,
   }
 }
 
+// caller is responsible for referencing m
 void MemTableListVersion::Add(MemTable* m) {
   assert(refs_ == 1);  // only when refs_ == 1 is MemTableListVersion mutable
-  m->Ref();
   memlist_.push_front(m);
   ++size_;
 }
 
+// caller is responsible for unreferencing m
 void MemTableListVersion::Remove(MemTable* m) {
   assert(refs_ == 1);  // only when refs_ == 1 is MemTableListVersion mutable
-  MemTable* x __attribute__((unused)) = m->Unref();
-  assert(x == nullptr);  // it still needs to be alive!
   memlist_.remove(m);
   --size_;
 }
@@ -231,6 +231,11 @@ Status MemTableList::InstallMemtableFlushResults(
 void MemTableList::Add(MemTable* m) {
   assert(current_->size_ >= num_flush_not_started_);
   InstallNewVersion();
+  // this method is used to move mutable memtable into an immutable list.
+  // since mutable memtable is already refcounted by the DBImpl,
+  // and when moving to the imutable list we don't unref it,
+  // we don't have to ref the memtable here. we just take over the
+  // reference from the DBImpl.
   current_->Add(m);
   m->MarkImmutable();
   num_flush_not_started_++;
@@ -255,6 +260,7 @@ void MemTableList::InstallNewVersion() {
     // somebody else holds the current version, we need to create new one
     MemTableListVersion* version = current_;
     current_ = new MemTableListVersion(current_);
+    current_->Ref();
     version->Unref();
   }
 }

db/memtablelist.h

@@ -40,16 +40,16 @@ class MemTableListVersion {
   void AddIterators(const ReadOptions& options,
                     std::vector<Iterator*>* iterator_list);
 
+ private:
   // REQUIRE: m is mutable memtable
   void Add(MemTable* m);
   // REQUIRE: m is mutable memtable
   void Remove(MemTable* m);
 
- private:
   friend class MemTableList;
   std::list<MemTable*> memlist_;
   int size_ = 0;
-  int refs_ = 1;
+  int refs_ = 0;
 };
 
 // This class stores references to all the immutable memtables.

db/version_set.cc

@@ -759,6 +759,28 @@ void Version::Unref() {
   }
 }
 
+bool Version::NeedsCompaction() const {
+  if (file_to_compact_ != nullptr) {
+    return true;
+  }
+  // In universal compaction case, this check doesn't really
+  // check the compaction condition, but checks num of files threshold
+  // only. We are not going to miss any compaction opportunity
+  // but it's likely that more compactions are scheduled but
+  // ending up with nothing to do. We can improve it later.
+  // TODO(sdong): improve this function to be accurate for universal
+  //              compactions.
+  int num_levels_to_check =
+    (vset_->options_->compaction_style != kCompactionStyleUniversal) ?
+    NumberLevels() - 1 : 1;
+  for (int i = 0; i < num_levels_to_check; i++) {
+    if (compaction_score_[i] >= 1) {
+      return true;
+    }
+  }
+  return false;
+}
+
 bool Version::OverlapInLevel(int level,
                              const Slice* smallest_user_key,
                              const Slice* largest_user_key) {

db/version_set.h

@@ -103,6 +103,15 @@ class Version {
   void Ref();
   void Unref();
 
+  // Returns true iff some level needs a compaction.
+  bool NeedsCompaction() const;
+
+  // Returns the maxmimum compaction score for levels 1 to max
+  double MaxCompactionScore() const { return max_compaction_score_; }
+
+  // See field declaration
+  int MaxCompactionScoreLevel() const { return max_compaction_score_level_; }
+
   void GetOverlappingInputs(
       int level,
       const InternalKey* begin,         // nullptr means before all keys
@@ -390,49 +399,6 @@ class VersionSet {
   // The caller should delete the iterator when no longer needed.
   Iterator* MakeInputIterator(Compaction* c);
 
-  // Returns true iff some level needs a compaction because it has
-  // exceeded its target size.
-  bool NeedsSizeCompaction() const {
-    // In universal compaction case, this check doesn't really
-    // check the compaction condition, but checks num of files threshold
-    // only. We are not going to miss any compaction opportunity
-    // but it's likely that more compactions are scheduled but
-    // ending up with nothing to do. We can improve it later.
-    // TODO: improve this function to be accurate for universal
-    //       compactions.
-    // TODO this only works for default column family now
-    Version* version = column_family_set_->GetDefault()->current;
-    int num_levels_to_check =
-        (options_->compaction_style != kCompactionStyleUniversal) ?
-            NumberLevels() - 1 : 1;
-    for (int i = 0; i < num_levels_to_check; i++) {
-      if (version->compaction_score_[i] >= 1) {
-        return true;
-      }
-    }
-    return false;
-  }
-  // Returns true iff some level needs a compaction.
-  bool NeedsCompaction() const {
-    // TODO this only works for default column family now
-    Version* version = column_family_set_->GetDefault()->current;
-    return ((version->file_to_compact_ != nullptr) || NeedsSizeCompaction());
-  }
-
-  // Returns the maxmimum compaction score for levels 1 to max
-  double MaxCompactionScore() const {
-    // TODO this only works for default column family now
-    Version* version = column_family_set_->GetDefault()->current;
-    return version->max_compaction_score_;
-  }
-
-  // See field declaration
-  int MaxCompactionScoreLevel() const {
-    // TODO this only works for default column family now
-    Version* version = column_family_set_->GetDefault()->current;
-    return version->max_compaction_score_level_;
-  }
-
   // Add all files listed in any live version to *live.
   void AddLiveFiles(std::vector<uint64_t>* live_list);