fix GC bug completetly!

db/db_impl.cc

@@ -158,13 +158,21 @@ DBImpl::DBImpl(const Options& raw_options, const std::string& dbname)
 
 DBImpl::~DBImpl() {
   // Wait for background work to finish.
+  gc_mutex_.Lock();
+  while(background_garbage_collect_scheduled_){
+    background_gc_finished_signal_.Wait();
+  }
+  background_garbage_collect_scheduled_=true;
+  gc_mutex_.Unlock();
   mutex_.Lock();
   shutting_down_.store(true, std::memory_order_release);
   while (background_compaction_scheduled_) {
     background_work_finished_signal_.Wait();
   }
   mutex_.Unlock();
-
+  gc_mutex_.Lock();
+  background_garbage_collect_scheduled_=false;
+  gc_mutex_.Unlock();
   if (db_lock_ != nullptr) {
     env_->UnlockFile(db_lock_);
   }
@@ -1732,18 +1740,23 @@ Status DBImpl::ReadValueLog(uint64_t file_id, uint64_t offset, Slice* key,
 // 垃圾回收实现
 void DBImpl::GarbageCollect() {
   // 遍历数据库目录，找到所有 valuelog 文件
-  Log(options_.info_log, "start gc ");
+  
   std::vector<std::string> filenames;
   Status s = env_->GetChildren(dbname_, &filenames);
+  Log(options_.info_log, "start gc ");
   assert(s.ok());
   std::set<std::string> valuelog_set;
   for (const auto& filename:filenames) {
-    if (IsValueLogFile(filename)) {
-      valuelog_set.emplace(filename);
+    if (IsValueLogFile(filename)){
+      uint64_t cur_log_number = GetValueLogID(filename);
+      auto tmp_name = ValueLogFileName(dbname_, cur_log_number);
+      
+      if(!versions_->checkOldValueLog(tmp_name))valuelog_set.emplace(filename);
     }
   }
   //bool tmp_judge=false;//only clean one file
   for (std::string valuelog_name : valuelog_set) {
+    Log(options_.info_log, ("gc processing: "+valuelog_name).data());
     // if(tmp_judge){
     //   break;
     // }
@@ -1937,13 +1950,12 @@ void DBImpl::GarbageCollect() {
     cur_valuelog.close();
 
     mutex_.Lock();
-    versions_->current()->addOldValueLog(valuelog_name);
+    versions_->AddOldValueLogFile(valuelog_name);
 
     mutex_.Unlock();
 
-
-    Log(options_.info_log, "remove file during gc %s", valuelog_name.c_str());
   }
+  Log(options_.info_log, "end gc ");
 }
 
 // Default implementations of convenience methods that subclasses of DB
@@ -2031,6 +2043,12 @@ Status DestroyDB(const std::string& dbname, const Options& options) {
           result = del;
         }
       }
+      else if(IsValueLogFile(filenames[i])){
+        Status del = env->RemoveFile(dbname + "/" + filenames[i]);
+        if (result.ok() && !del.ok()) {
+          result = del;
+        }
+      }
     }
     env->UnlockFile(lock);  // Ignore error since state is already gone
     env->RemoveFile(lockname);

db/version_set.cc

@@ -6,7 +6,7 @@
 
 #include <algorithm>
 #include <cstdio>
-
+#include <iostream>
 #include "db/filename.h"
 #include "db/log_reader.h"
 #include "db/log_writer.h"
@@ -84,7 +84,12 @@ Version::~Version() {
   }
 
   for(auto valuelog_name:old_valuelog_names){
-    vset_->env_->RemoveFile(valuelog_name);
+    vset_->valuelogmap_mutex.Lock();
+    int res=vset_->old_valuelog_map[valuelog_name]--;
+    if(res==1){
+      vset_->env_->RemoveFile(valuelog_name);
+    }
+    vset_->valuelogmap_mutex.Unlock();
   }
 }
 
@@ -1166,6 +1171,16 @@ void VersionSet::AddLiveFiles(std::set* live) {
   }
 }
 
+void VersionSet::AddOldValueLogFile(std::string valuelog_name){
+  valuelogmap_mutex.Lock();
+  old_valuelog_map[valuelog_name]=0;
+  for (Version* v = dummy_versions_.next_; v != &dummy_versions_; v = v->next_) {
+    old_valuelog_map[valuelog_name]++;
+    v->addOldValueLog(valuelog_name);//when all these version were deleted, the value log will be free to delete
+  }
+  valuelogmap_mutex.Unlock();
+}
+
 int64_t VersionSet::NumLevelBytes(int level) const {
   assert(level >= 0);
   assert(level < config::kNumLevels);

db/version_set.h

@@ -18,7 +18,7 @@
 #include <map>
 #include <set>
 #include <vector>
-
+#include <atomic>
 #include "db/dbformat.h"
 #include "db/version_edit.h"
 #include "port/port.h"
@@ -263,6 +263,15 @@ class VersionSet {
   // May also mutate some internal state.
   void AddLiveFiles(std::set<uint64_t>* live);
 
+  void AddOldValueLogFile(std::string valuelog_name);
+
+  bool checkOldValueLog(std::string valuelog_name){
+    valuelogmap_mutex.Lock();
+    auto res=old_valuelog_map.count(valuelog_name);
+    valuelogmap_mutex.Unlock();
+    return res;
+  }
+
   // Return the approximate offset in the database of the data for
   // "key" as of version "v".
   uint64_t ApproximateOffsetOf(Version* v, const InternalKey& key);
@@ -309,6 +318,10 @@ class VersionSet {
   uint64_t log_number_;
   uint64_t prev_log_number_;  // 0 or backing store for memtable being compacted
 
+  std::map<std::string,int> old_valuelog_map;
+
+  port::Mutex valuelogmap_mutex;
+
   // Opened lazily
   WritableFile* descriptor_file_;
   log::Writer* descriptor_log_;