Rate limit deletes issued by DestroyDB

Summary: Update DestroyDB so that all SST files in the first path id go through DeleteScheduler instead of being deleted immediately

Test Plan: added a unittest

Reviewers: igor, yhchiang, anthony, kradhakrishnan, rven, sdong

Reviewed By: sdong

Subscribers: jeanxu2012, dhruba

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

db/db_impl.cc

@@ -750,9 +750,8 @@ void DBImpl::PurgeObsoleteFiles(const JobContext& state) {
     }
 #endif  // !ROCKSDB_LITE
     Status file_deletion_status;
-    if (db_options_.delete_scheduler != nullptr && type == kTableFile &&
+    if (type == kTableFile && path_id == 0) {
-        path_id == 0) {
+      file_deletion_status = DeleteOrMoveToTrash(&db_options_, fname);
-      file_deletion_status = db_options_.delete_scheduler->DeleteFile(fname);
     } else {
       file_deletion_status = env_->DeleteFile(fname);
     }
@@ -4518,10 +4517,13 @@ Status DestroyDB(const std::string& dbname, const Options& options) {
       if (ParseFileName(filenames[i], &number, info_log_prefix.prefix, &type) &&
           type != kDBLockFile) {  // Lock file will be deleted at end
         Status del;
+        std::string path_to_delete = dbname + "/" + filenames[i];
         if (type == kMetaDatabase) {
-          del = DestroyDB(dbname + "/" + filenames[i], options);
+          del = DestroyDB(path_to_delete, options);
+        } else if (type == kTableFile) {
+          del = DeleteOrMoveToTrash(&options, path_to_delete);
         } else {
-          del = env->DeleteFile(dbname + "/" + filenames[i]);
+          del = env->DeleteFile(path_to_delete);
         }
         if (result.ok() && !del.ok()) {
           result = del;
@@ -4529,12 +4531,19 @@ Status DestroyDB(const std::string& dbname, const Options& options) {
       }
     }
 
-    for (auto& db_path : options.db_paths) {
+    for (size_t path_id = 0; path_id < options.db_paths.size(); path_id++) {
+      const auto& db_path = options.db_paths[path_id];
       env->GetChildren(db_path.path, &filenames);
       for (size_t i = 0; i < filenames.size(); i++) {
         if (ParseFileName(filenames[i], &number, &type) &&
             type == kTableFile) {  // Lock file will be deleted at end
-          Status del = env->DeleteFile(db_path.path + "/" + filenames[i]);
+          Status del;
+          std::string table_path = db_path.path + "/" + filenames[i];
+          if (path_id == 0) {
+            del = DeleteOrMoveToTrash(&options, table_path);
+          } else {
+            del = env->DeleteFile(table_path);
+          }
           if (result.ok() && !del.ok()) {
             result = del;
           }

db/db_test.cc

@@ -60,7 +60,6 @@
 #include "utilities/merge_operators.h"
 #include "util/logging.h"
 #include "util/compression.h"
-#include "util/delete_scheduler_impl.h"
 #include "util/mutexlock.h"
 #include "util/rate_limiter.h"
 #include "util/statistics.h"
@@ -8388,12 +8387,10 @@ TEST_F(DBTest, RateLimitedDelete) {
   ASSERT_OK(db_->CompactRange(CompactRangeOptions(), nullptr, nullptr));
   ASSERT_EQ("0,1", FilesPerLevel(0));
 
+  uint64_t delete_start_time = env_->NowMicros();
   // Hold BackgroundEmptyTrash
   TEST_SYNC_POINT("DBTest::RateLimitedDelete:1");
-
+  options.delete_scheduler->WaitForEmptyTrash();
-  uint64_t delete_start_time = env_->NowMicros();
-  reinterpret_cast<DeleteSchedulerImpl*>(options.delete_scheduler.get())
-      ->TEST_WaitForEmptyTrash();
   uint64_t time_spent_deleting = env_->NowMicros() - delete_start_time;
 
   uint64_t total_files_size = 0;
@@ -8465,8 +8462,7 @@ TEST_F(DBTest, DeleteSchedulerMultipleDBPaths) {
   ASSERT_OK(db_->CompactRange(compact_options, &begin, &end));
   ASSERT_EQ("0,2", FilesPerLevel(0));
 
-  reinterpret_cast<DeleteSchedulerImpl*>(options.delete_scheduler.get())
+  options.delete_scheduler->WaitForEmptyTrash();
-      ->TEST_WaitForEmptyTrash();
   ASSERT_EQ(bg_delete_file, 8);
 
   compact_options.bottommost_level_compaction =
@@ -8474,13 +8470,47 @@ TEST_F(DBTest, DeleteSchedulerMultipleDBPaths) {
   ASSERT_OK(db_->CompactRange(compact_options, nullptr, nullptr));
   ASSERT_EQ("0,1", FilesPerLevel(0));
 
-  reinterpret_cast<DeleteSchedulerImpl*>(options.delete_scheduler.get())
+  options.delete_scheduler->WaitForEmptyTrash();
-      ->TEST_WaitForEmptyTrash();
   ASSERT_EQ(bg_delete_file, 8);
 
   rocksdb::SyncPoint::GetInstance()->DisableProcessing();
 }
 
+TEST_F(DBTest, DestroyDBWithRateLimitedDelete) {
+  int bg_delete_file = 0;
+  rocksdb::SyncPoint::GetInstance()->SetCallBack(
+      "DeleteSchedulerImpl::DeleteTrashFile:DeleteFile",
+      [&](void* arg) { bg_delete_file++; });
+  rocksdb::SyncPoint::GetInstance()->EnableProcessing();
+
+  Options options = CurrentOptions();
+  options.disable_auto_compactions = true;
+  options.env = env_;
+  DestroyAndReopen(options);
+
+  // Create 4 files in L0
+  for (int i = 0; i < 4; i++) {
+    ASSERT_OK(Put("Key" + ToString(i), DummyString(1024, 'A')));
+    ASSERT_OK(Flush());
+  }
+  // We created 4 sst files in L0
+  ASSERT_EQ("4", FilesPerLevel(0));
+
+  // Close DB and destory it using DeleteScheduler
+  Close();
+  std::string trash_dir = test::TmpDir(env_) + "/trash";
+  int64_t rate_bytes_per_sec = 1024 * 1024;  // 1 Mb / Sec
+  Status s;
+  options.delete_scheduler.reset(NewDeleteScheduler(
+      env_, trash_dir, rate_bytes_per_sec, nullptr, false, &s));
+  ASSERT_OK(s);
+  ASSERT_OK(DestroyDB(dbname_, options));
+
+  options.delete_scheduler->WaitForEmptyTrash();
+  // We have deleted the 4 sst files in the delete_scheduler
+  ASSERT_EQ(bg_delete_file, 4);
+}
+
 TEST_F(DBTest, UnsupportedManualSync) {
   DestroyAndReopen(CurrentOptions());
   env_->is_wal_sync_thread_safe_.store(false);

include/rocksdb/delete_scheduler.h

@@ -35,6 +35,10 @@ class DeleteScheduler {
   // Return a map containing errors that happened in the background thread
   // file_path => error status
   virtual std::map<std::string, Status> GetBackgroundErrors() = 0;
+
+  // Wait for all files being deleteing in the background to finish or for
+  // destructor to be called.
+  virtual void WaitForEmptyTrash() = 0;
 };
 
 // Create a new DeleteScheduler that can be shared among multiple RocksDB

util/delete_scheduler_impl.cc

@@ -3,12 +3,13 @@
 //  LICENSE file in the root directory of this source tree. An additional grant
 //  of patent rights can be found in the PATENTS file in the same directory.
 
+#include "util/delete_scheduler_impl.h"
+
 #include <thread>
 #include <vector>
 
 #include "port/port.h"
 #include "rocksdb/env.h"
-#include "util/delete_scheduler_impl.h"
 #include "util/mutexlock.h"
 #include "util/sync_point.h"
 
@@ -155,7 +156,7 @@ void DeleteSchedulerImpl::BackgroundEmptyTrash() {
 
       pending_files_--;
       if (pending_files_ == 0) {
-        // Unblock TEST_WaitForEmptyTrash since there are no more files waiting
+        // Unblock WaitForEmptyTrash since there are no more files waiting
         // to be deleted
         cv_.SignalAll();
       }
@@ -185,7 +186,7 @@ Status DeleteSchedulerImpl::DeleteTrashFile(const std::string& path_in_trash,
   return s;
 }
 
-void DeleteSchedulerImpl::TEST_WaitForEmptyTrash() {
+void DeleteSchedulerImpl::WaitForEmptyTrash() {
   MutexLock l(&mu_);
   while (pending_files_ > 0 && !closing_) {
     cv_.Wait();

util/delete_scheduler_impl.h

@@ -36,7 +36,7 @@ class DeleteSchedulerImpl : public DeleteScheduler {
 
   // Wait for all files being deleteing in the background to finish or for
   // destructor to be called.
-  void TEST_WaitForEmptyTrash();
+  void WaitForEmptyTrash();
 
   // Return a map containing errors that happened in BackgroundEmptyTrash
   // file_path => error status

util/delete_scheduler_test.cc

@@ -15,7 +15,6 @@
 #include "rocksdb/delete_scheduler.h"
 #include "rocksdb/env.h"
 #include "rocksdb/options.h"
-#include "util/delete_scheduler_impl.h"
 #include "util/string_util.h"
 #include "util/sync_point.h"
 #include "util/testharness.h"
@@ -38,11 +37,6 @@ class DeleteSchedulerTest : public testing::Test {
     DestroyDir(dummy_files_dir_);
   }
 
-  void WaitForEmptyTrash() {
-    reinterpret_cast<DeleteSchedulerImpl*>(delete_scheduler_.get())
-        ->TEST_WaitForEmptyTrash();
-  }
-
   void DestroyDir(const std::string& dir) {
     if (env_->FileExists(dir).IsNotFound()) {
       return;
@@ -133,7 +127,7 @@ TEST_F(DeleteSchedulerTest, BasicRateLimiting) {
 
     uint64_t delete_start_time = env_->NowMicros();
     TEST_SYNC_POINT("DeleteSchedulerTest::BasicRateLimiting:1");
-    WaitForEmptyTrash();
+    delete_scheduler_->WaitForEmptyTrash();
     uint64_t time_spent_deleting = env_->NowMicros() - delete_start_time;
 
     uint64_t total_files_size = 0;
@@ -213,7 +207,7 @@ TEST_F(DeleteSchedulerTest, RateLimitingMultiThreaded) {
 
     uint64_t delete_start_time = env_->NowMicros();
     TEST_SYNC_POINT("DeleteSchedulerTest::RateLimitingMultiThreaded:1");
-    WaitForEmptyTrash();
+    delete_scheduler_->WaitForEmptyTrash();
     uint64_t time_spent_deleting = env_->NowMicros() - delete_start_time;
 
     uint64_t total_files_size = 0;
@@ -289,7 +283,7 @@ TEST_F(DeleteSchedulerTest, ConflictNames) {
 
   // Hold BackgroundEmptyTrash
   TEST_SYNC_POINT("DeleteSchedulerTest::ConflictNames:1");
-  WaitForEmptyTrash();
+  delete_scheduler_->WaitForEmptyTrash();
   ASSERT_EQ(CountFilesInDir(trash_dir_), 0);
 
   auto bg_errors = delete_scheduler_->GetBackgroundErrors();
@@ -333,7 +327,7 @@ TEST_F(DeleteSchedulerTest, BackgroundError) {
 
   // Hold BackgroundEmptyTrash
   TEST_SYNC_POINT("DeleteSchedulerTest::BackgroundError:1");
-  WaitForEmptyTrash();
+  delete_scheduler_->WaitForEmptyTrash();
   auto bg_errors = delete_scheduler_->GetBackgroundErrors();
   ASSERT_EQ(bg_errors.size(), 10);
 
@@ -359,7 +353,7 @@ TEST_F(DeleteSchedulerTest, TrashWithExistingFiles) {
       env_, trash_dir_, rate_bytes_per_sec_, nullptr, true, &s));
   ASSERT_OK(s);
 
-  WaitForEmptyTrash();
+  delete_scheduler_->WaitForEmptyTrash();
   ASSERT_EQ(CountFilesInDir(trash_dir_), 0);
 
   auto bg_errors = delete_scheduler_->GetBackgroundErrors();
@@ -390,7 +384,7 @@ TEST_F(DeleteSchedulerTest, StartBGEmptyTrashMultipleTimes) {
       ASSERT_OK(delete_scheduler_->DeleteFile(NewDummyFile(file_name)));
     }
     ASSERT_EQ(CountFilesInDir(dummy_files_dir_), 0);
-    WaitForEmptyTrash();
+    delete_scheduler_->WaitForEmptyTrash();
     ASSERT_EQ(bg_delete_file, 10 * run);
     ASSERT_EQ(CountFilesInDir(trash_dir_), 0);
 

util/file_util.cc

@@ -3,10 +3,14 @@
 //  LICENSE file in the root directory of this source tree. An additional grant
 //  of patent rights can be found in the PATENTS file in the same directory.
 //
+#include "util/file_util.h"
+
 #include <string>
 #include <algorithm>
-#include "util/file_util.h"
+
+#include "rocksdb/delete_scheduler.h"
 #include "rocksdb/env.h"
+#include "rocksdb/options.h"
 #include "db/filename.h"
 #include "util/file_reader_writer.h"
 
@@ -64,4 +68,13 @@ Status CopyFile(Env* env, const std::string& source,
   return Status::OK();
 }
 
+Status DeleteOrMoveToTrash(const DBOptions* db_options,
+                           const std::string& fname) {
+  if (db_options->delete_scheduler == nullptr) {
+    return db_options->env->DeleteFile(fname);
+  } else {
+    return db_options->delete_scheduler->DeleteFile(fname);
+  }
+}
+
 }  // namespace rocksdb

util/file_util.h

@@ -3,16 +3,20 @@
 //  LICENSE file in the root directory of this source tree. An additional grant
 //  of patent rights can be found in the PATENTS file in the same directory.
 //
+#pragma once
 #include <string>
 
-#pragma once
 #include "rocksdb/status.h"
 #include "rocksdb/types.h"
 #include "rocksdb/env.h"
+#include "rocksdb/options.h"
 
 namespace rocksdb {
 
 extern Status CopyFile(Env* env, const std::string& source,
                        const std::string& destination, uint64_t size = 0);
 
+extern Status DeleteOrMoveToTrash(const DBOptions* db_options,
+                                  const std::string& fname);
+
 }  // namespace rocksdb