fault_injection_test: add a unit test to allow parallel compactions and multiple levels

Summary: Add a new test case in fault_injection_test, which covers parallel compactions and multiple levels. Use MockEnv to run the new test case to speed it up. Improve MockEnv to avoid DestoryDB(), previously failed when deleting lock files.

Test Plan: Run ./fault_injection_test, including valgrind

Reviewers: rven, yhchiang, igor

Reviewed By: igor

Subscribers: leveldb, dhruba

Differential Revision: https://reviews.facebook.net/D32415

db/fault_injection_test.cc

@@ -23,6 +23,7 @@
 #include "rocksdb/table.h"
 #include "rocksdb/write_batch.h"
 #include "util/logging.h"
+#include "util/mock_env.h"
 #include "util/mutexlock.h"
 #include "util/testharness.h"
 #include "util/testutil.h"
@@ -66,9 +67,7 @@ static std::pair<std::string, std::string> GetDirAndName(
 }
 
 // A basic file truncation function suitable for this test.
-Status Truncate(const std::string& filename, uint64_t length) {
+Status Truncate(Env* env, const std::string& filename, uint64_t length) {
-  rocksdb::Env* env = rocksdb::Env::Default();
-
   unique_ptr<SequentialFile> orig_file;
   const EnvOptions options;
   Status s = env->NewSequentialFile(filename, &orig_file, options);
@@ -122,9 +121,9 @@ struct FileState {
 
   bool IsFullySynced() const { return pos_ <= 0 || pos_ == pos_at_last_sync_; }
 
-  Status DropUnsyncedData() const;
+  Status DropUnsyncedData(Env* env) const;
 
-  Status DropRandomUnsyncedData(Random* rand) const;
+  Status DropRandomUnsyncedData(Env* env, Random* rand) const;
 };
 
 }  // anonymous namespace
@@ -246,7 +245,7 @@ class FaultInjectionTestEnv : public EnvWrapper {
 
   // For every file that is not fully synced, make a call to `func` with
   // FileState of the file as the parameter.
-  Status DropFileData(std::function<Status(FileState)> func) {
+  Status DropFileData(std::function<Status(Env*, FileState)> func) {
     Status s;
     MutexLock l(&mutex_);
     for (std::map<std::string, FileState>::const_iterator it =
@@ -254,20 +253,21 @@ class FaultInjectionTestEnv : public EnvWrapper {
          s.ok() && it != db_file_state_.end(); ++it) {
       const FileState& state = it->second;
       if (!state.IsFullySynced()) {
-        s = func(state);
+        s = func(target(), state);
       }
     }
     return s;
   }
 
   Status DropUnsyncedFileData() {
-    return DropFileData(
+    return DropFileData([&](Env* env, const FileState& state) {
-        [&](const FileState& state) { return state.DropUnsyncedData(); });
+      return state.DropUnsyncedData(env);
+    });
   }
 
   Status DropRandomUnsyncedFileData(Random* rnd) {
-    return DropFileData([&](const FileState& state) {
+    return DropFileData([&](Env* env, const FileState& state) {
-      return state.DropRandomUnsyncedData(rnd);
+      return state.DropRandomUnsyncedData(env, rnd);
     });
   }
 
@@ -335,18 +335,18 @@ class FaultInjectionTestEnv : public EnvWrapper {
   bool filesystem_active_;  // Record flushes, syncs, writes
 };
 
-Status FileState::DropUnsyncedData() const {
+Status FileState::DropUnsyncedData(Env* env) const {
   ssize_t sync_pos = pos_at_last_sync_ == -1 ? 0 : pos_at_last_sync_;
-  return Truncate(filename_, sync_pos);
+  return Truncate(env, filename_, sync_pos);
 }
 
-Status FileState::DropRandomUnsyncedData(Random* rand) const {
+Status FileState::DropRandomUnsyncedData(Env* env, Random* rand) const {
   ssize_t sync_pos = pos_at_last_sync_ == -1 ? 0 : pos_at_last_sync_;
   assert(pos_ >= sync_pos);
   int range = static_cast<int>(pos_ - sync_pos);
   uint64_t truncated_size =
       static_cast<uint64_t>(sync_pos) + rand->Uniform(range);
-  return Truncate(filename_, truncated_size);
+  return Truncate(env, filename_, truncated_size);
 }
 
 Status TestDirectory::Fsync() {
@@ -413,6 +413,7 @@ class FaultInjectionTest {
     kWalDir,
     kSyncWal,
     kWalDirSyncWal,
+    kMultiLevels,
     kEnd,
   };
   int option_config_;
@@ -431,6 +432,7 @@ class FaultInjectionTest {
     kResetDropAndDeleteUnsynced
   };
 
+  std::unique_ptr<Env> base_env_;
   FaultInjectionTestEnv* env_;
   std::string dbname_;
   shared_ptr<Cache> tiny_cache_;
@@ -441,6 +443,7 @@ class FaultInjectionTest {
       : option_config_(kDefault),
         sync_use_wal_(false),
         sync_use_compact_(true),
+        base_env_(nullptr),
         env_(NULL),
         db_(NULL) {
     NewDB();
@@ -453,6 +456,9 @@ class FaultInjectionTest {
     if (option_config_ >= kEnd) {
       return false;
     } else {
+      if (option_config_ == kMultiLevels) {
+        base_env_.reset(new MockEnv(Env::Default()));
+      }
       return true;
     }
   }
@@ -479,6 +485,19 @@ class FaultInjectionTest {
         sync_use_wal_ = true;
         sync_use_compact_ = false;
         break;
+      case kMultiLevels:
+        options.write_buffer_size = 64 * 1024;
+        options.target_file_size_base = 64 * 1024;
+        options.level0_file_num_compaction_trigger = 2;
+        options.level0_slowdown_writes_trigger = 2;
+        options.level0_stop_writes_trigger = 4;
+        options.max_bytes_for_level_base = 128 * 1024;
+        options.max_write_buffer_number = 2;
+        options.max_background_compactions = 8;
+        options.max_background_flushes = 8;
+        sync_use_wal_ = true;
+        sync_use_compact_ = false;
+        break;
       default:
         break;
     }
@@ -490,7 +509,8 @@ class FaultInjectionTest {
     assert(tiny_cache_ == nullptr);
     assert(env_ == NULL);
 
-    env_ = new FaultInjectionTestEnv(Env::Default());
+    env_ =
+        new FaultInjectionTestEnv(base_env_ ? base_env_.get() : Env::Default());
 
     options_ = CurrentOptions();
     options_.env = env_;

util/mock_env.cc

@@ -19,9 +19,11 @@ namespace rocksdb {
 
 class MemFile {
  public:
-  explicit MemFile(const std::string& fn)
+  explicit MemFile(const std::string& fn, bool _is_lock_file = false)
       : fn_(fn),
         refs_(0),
+        is_lock_file_(_is_lock_file),
+        locked_(false),
         size_(0),
         modified_time_(Now()),
         rnd_(static_cast<uint32_t>(
@@ -33,6 +35,25 @@ class MemFile {
     ++refs_;
   }
 
+  bool is_lock_file() const { return is_lock_file_; }
+
+  bool Lock() {
+    assert(is_lock_file_);
+    MutexLock lock(&mutex_);
+    if (locked_) {
+      return false;
+    } else {
+      refs_ = true;
+      return true;
+    }
+  }
+
+  void Unlock() {
+    assert(is_lock_file_);
+    MutexLock lock(&mutex_);
+    locked_ = false;
+  }
+
   void Unref() {
     bool do_delete = false;
     {
@@ -132,6 +153,8 @@ class MemFile {
   const std::string fn_;
   mutable port::Mutex mutex_;
   int refs_;
+  bool is_lock_file_;
+  bool locked_;
 
   // Data written into this file, all bytes before fsynced_bytes are
   // persistent.
@@ -398,6 +421,9 @@ Status MockEnv::NewSequentialFile(const std::string& fname,
     return Status::IOError(fn, "File not found");
   }
   auto* f = file_map_[fn];
+  if (f->is_lock_file()) {
+    return Status::InvalidArgument(fn, "Cannot open a lock file.");
+  }
   result->reset(new SequentialFileImpl(f));
   return Status::OK();
 }
@@ -412,6 +438,9 @@ Status MockEnv::NewRandomAccessFile(const std::string& fname,
     return Status::IOError(fn, "File not found");
   }
   auto* f = file_map_[fn];
+  if (f->is_lock_file()) {
+    return Status::InvalidArgument(fn, "Cannot open a lock file.");
+  }
   result->reset(new RandomAccessFileImpl(f));
   return Status::OK();
 }
@@ -424,7 +453,7 @@ Status MockEnv::NewWritableFile(const std::string& fname,
   if (file_map_.find(fn) != file_map_.end()) {
     DeleteFileInternal(fn);
   }
-  MemFile* file = new MemFile(fn);
+  MemFile* file = new MemFile(fn, false);
   file->Ref();
   file_map_[fn] = file;
 
@@ -490,12 +519,11 @@ Status MockEnv::GetChildren(const std::string& dir,
 
 void MockEnv::DeleteFileInternal(const std::string& fname) {
   assert(fname == NormalizePath(fname));
-  if (file_map_.find(fname) == file_map_.end()) {
+  const auto& pair = file_map_.find(fname);
-    return;
+  if (pair != file_map_.end()) {
+    pair->second->Unref();
+    file_map_.erase(fname);
   }
-
-  file_map_[fname]->Unref();
-  file_map_.erase(fname);
 }
 
 Status MockEnv::DeleteFile(const std::string& fname) {
@@ -579,7 +607,7 @@ Status MockEnv::NewLogger(const std::string& fname,
   auto iter = file_map_.find(fn);
   MemFile* file = nullptr;
   if (iter == file_map_.end()) {
-    file = new MemFile(fn);
+    file = new MemFile(fn, false);
     file->Ref();
     file_map_[fn] = file;
   } else {
@@ -595,9 +623,18 @@ Status MockEnv::LockFile(const std::string& fname, FileLock** flock) {
   {
     MutexLock lock(&mutex_);
     if (file_map_.find(fn) != file_map_.end()) {
-      return Status::IOError(fn, "Lock file exists");
+      if (!file_map_[fn]->is_lock_file()) {
+        return Status::InvalidArgument(fname, "Not a lock file.");
+      }
+      if (!file_map_[fn]->Lock()) {
+        return Status::IOError(fn, "Lock is already held.");
+      }
+    } else {
+      auto* file = new MemFile(fname, true);
+      file->Ref();
+      file->Lock();
+      file_map_[fname] = file;
     }
-    file_map_[fn] = nullptr;
   }
   *flock = new MockEnvFileLock(fn);
   return Status::OK();
@@ -607,9 +644,11 @@ Status MockEnv::UnlockFile(FileLock* flock) {
   std::string fn = dynamic_cast<MockEnvFileLock*>(flock)->FileName();
   {
     MutexLock lock(&mutex_);
-    auto iter = file_map_.find(fn);
+    if (file_map_.find(fn) != file_map_.end()) {
-    if (iter != file_map_.end()) {
+      if (!file_map_[fn]->is_lock_file()) {
-      file_map_.erase(fn);
+        return Status::InvalidArgument(fn, "Not a lock file.");
+      }
+      file_map_[fn]->Unlock();
     }
   }
   delete flock;