issue #934: multi-thread compaction support

Author: caijieming

src/leveldb/db/db_impl.cc

@@ -97,6 +97,11 @@ class DBImpl : public DB {
   friend class DBTable;
   struct CompactionState;
   struct Writer;
+  struct CompactionTask {
+    int64_t id;
+    double score;
+    DBImpl* db;
+  };
 
   Iterator* NewInternalIterator(const ReadOptions&,
                                 SequenceNumber* latest_snapshot);
@@ -111,10 +116,10 @@ class DBImpl : public DB {
 
   // Compact the in-memory write buffer to disk.  Switches to a new
   // log-file/memtable and writes a new descriptor iff successful.
-  Status CompactMemTable()
+  Status CompactMemTable(bool* sched_idle = NULL)
       EXCLUSIVE_LOCKS_REQUIRED(mutex_);
 
-  Status WriteLevel0Table(MemTable* mem, VersionEdit* edit, Version* base)
+  Status WriteLevel0Table(MemTable* mem, VersionEdit* edit, Version* base, uint64_t* number = NULL)
       EXCLUSIVE_LOCKS_REQUIRED(mutex_);
 
   Status MakeRoomForWrite(bool force /* compact even if there is room? */)
@@ -123,8 +128,8 @@ class DBImpl : public DB {
 
   void MaybeScheduleCompaction() EXCLUSIVE_LOCKS_REQUIRED(mutex_);
   static void BGWork(void* db);
-  void BackgroundCall();
-  Status BackgroundCompaction() EXCLUSIVE_LOCKS_REQUIRED(mutex_);
+  void BackgroundCall(CompactionTask* task);
+  Status BackgroundCompaction(bool* sched_idle) EXCLUSIVE_LOCKS_REQUIRED(mutex_);
   void CleanupCompaction(CompactionState* compact)
       EXCLUSIVE_LOCKS_REQUIRED(mutex_);
   Status DoCompactionWork(CompactionState* compact)
@@ -193,17 +198,25 @@ class DBImpl : public DB {
   std::set<uint64_t> pending_outputs_;
 
   // Has a background compaction been scheduled or is running?
-  bool bg_compaction_scheduled_;
-  double bg_compaction_score_;
-  int64_t bg_schedule_id_;
+  std::set<CompactionTask*> bg_compaction_tasks_;
+  int bg_compaction_scheduled_;
+  std::vector<double> bg_compaction_score_;
+  std::vector<int64_t> bg_schedule_id_;
 
   // Information for a manual compaction
+  enum ManualCompactState {
+    kManualCompactIdle,
+    kManualCompactConflict,
+    kManualCompactWakeup,
+  };
   struct ManualCompaction {
     int level;
     bool done;
+    bool being_sched;
     const InternalKey* begin;   // NULL means beginning of key range
     const InternalKey* end;     // NULL means end of key range
     InternalKey tmp_storage;    // Used to keep track of compaction progress
+    int compaction_conflict; // 0 == idle, 1 == conflict, 2 == wake
   };
   ManualCompaction* manual_compaction_;
 

src/leveldb/db/db_impl.h

@@ -292,22 +292,6 @@ Status DBTable::Init() {
         uint32_t i = *it;
         DBImpl* impl = lg_list_[i];
         s = impl->RecoverLastDumpToLevel0(lg_edits[i]);
-
-        // LogAndApply to lg's manifest
-        if (s.ok()) {
-            MutexLock lock(&impl->mutex_);
-            s = impl->versions_->LogAndApply(lg_edits[i], &impl->mutex_);
-            if (s.ok()) {
-                impl->DeleteObsoleteFiles();
-                impl->MaybeScheduleCompaction();
-            } else {
-                Log(options_.info_log, "[%s] Fail to modify manifest of lg %d",
-                    dbname_.c_str(),
-                    i);
-            }
-        } else {
-            Log(options_.info_log, "[%s] Fail to dump log to level 0", dbname_.c_str());
-        }
         delete lg_edits[i];
     }
 
@@ -820,7 +804,6 @@ Status DBTable::RecoverLogFile(uint64_t log_number, uint64_t recover_limit,
             if (this->status != NULL && this->status->ok()) *this->status = s;
         }
     };
-
     mutex_.AssertHeld();
 
     // Open the log file
@@ -860,7 +843,7 @@ Status DBTable::RecoverLogFile(uint64_t log_number, uint64_t recover_limit,
         //    dbname_.c_str(), batch_seq, last_sequence_, WriteBatchInternal::Count(&batch));
         if (last_seq >= recover_limit) {
             Log(options_.info_log, "[%s] exceed limit %lu, ignore %lu ~ %lu",
-                        dbname_.c_str(), recover_limit, first_seq, last_seq);
+                dbname_.c_str(), recover_limit, first_seq, last_seq);
             continue;
         }
 
@@ -915,7 +898,6 @@ Status DBTable::RecoverLogFile(uint64_t log_number, uint64_t recover_limit,
         }
     }
     delete file;
-
     return status;
 }
 

src/leveldb/db/db_table.cc

@@ -26,6 +26,7 @@ MemTable::MemTable(const InternalKeyComparator& cmp, CompactStrategyFactory* com
     : last_seq_(0),
       comparator_(cmp),
       refs_(0),
+      being_flushed_(false),
       table_(comparator_, &arena_),
       empty_(true),
       compact_strategy_factory_(compact_strategy_factory) {

src/leveldb/db/memtable.cc

@@ -79,6 +79,13 @@ class MemTable {
     empty_ = false;
   }
 
+  bool BeingFlushed() { return being_flushed_;}
+  void SetBeingFlushed(bool flag) {
+    assert(flag ? !being_flushed_
+                : being_flushed_);
+    being_flushed_ = flag;
+  }
+
   virtual ~MemTable();
 
  protected:
@@ -97,6 +104,7 @@ class MemTable {
 
   KeyComparator comparator_;
   int refs_;
+  bool being_flushed_;
 
   Arena arena_;
   Table table_;

src/leveldb/db/memtable.h

@@ -30,14 +30,16 @@ struct FileMetaData {
   InternalKey largest;        // Largest internal key served by table
   bool smallest_fake;         // smallest is not real, have out-of-range keys
   bool largest_fake;          // largest is not real, have out-of-range keys
+  bool being_compacted;       // Is this file undergoing compaction?
 
   FileMetaData() :
       refs(0),
       allowed_seeks(1 << 30),
       file_size(0),
       data_size(0),
       smallest_fake(false),
-      largest_fake(false) { }
+      largest_fake(false),
+      being_compacted(false) { }
 };
 
 class VersionEdit {