Optimize LogAndApply in-mutex duration (#422)

 

Signed-off-by: Connor1996 <zbk602423539@gmail.com>

Author: Connor1996

CMakeLists.txt

@@ -755,6 +755,7 @@ set(SOURCES
         db/version_edit.cc
         db/version_edit_handler.cc
         db/version_set.cc
+        db/version_set_deletion_scheduler.cc
         db/wal_edit.cc
         db/wal_manager.cc
         db/wide/wide_column_serialization.cc

TARGETS

@@ -99,6 +99,7 @@ cpp_library_wrapper(name="rocksdb_lib", srcs=[
         "db/version_edit.cc",
         "db/version_edit_handler.cc",
         "db/version_set.cc",
+        "db/version_set_deletion_scheduler.cc",
         "db/wal_edit.cc",
         "db/wal_manager.cc",
         "db/wide/wide_column_serialization.cc",

db/listener_test.cc

@@ -1585,6 +1585,7 @@ TEST_F(EventListenerTest, BlobDBFileTest) {
   // delete the oldest blob file and create new blob file during compaction.
   ASSERT_OK(db_->CompactRange(CompactRangeOptions(), begin, end));
   ASSERT_OK(dbfull()->TEST_WaitForCompact());
+  db_->Close();
 
   blob_event_listener->CheckCounters();
 }

db/version_set.cc

@@ -46,6 +46,7 @@
 #include "db/table_cache.h"
 #include "db/version_builder.h"
 #include "db/version_edit.h"
+#include "db/version_set_deletion_scheduler.h"
 #include "db/write_controller.h"
 #include "env/file_system_tracer.h"
 #if USE_COROUTINES
@@ -994,7 +995,7 @@ class Version {
       std::shared_ptr<const TableProperties>* tp, int level = -1);
 
   uint64_t GetEstimatedActiveKeys() {
-    return storage_info_.GetEstimatedActiveKeys();
+    return storage_info_->GetEstimatedActiveKeys();
   }
 
   size_t GetMemoryUsageByTableReaders(const ReadOptions& read_options);
@@ -1006,8 +1007,8 @@ class Version {
 
   int TEST_refs() const { return refs_; }
 
-  VersionStorageInfo* storage_info() { return &storage_info_; }
-  const VersionStorageInfo* storage_info() const { return &storage_info_; }
+  VersionStorageInfo* storage_info() { return storage_info_; }
+  const VersionStorageInfo* storage_info() const { return storage_info_; }
 
   VersionSet* version_set() { return vset_; }
 
@@ -1038,10 +1039,10 @@ class Version {
   friend class VersionEditHandlerPointInTime;
 
   const InternalKeyComparator* internal_comparator() const {
-    return storage_info_.internal_comparator_;
+    return storage_info_->internal_comparator_;
   }
   const Comparator* user_comparator() const {
-    return storage_info_.user_comparator_;
+    return storage_info_->user_comparator_;
   }
 
   // Returns true if the filter blocks in the specified level will not be
@@ -1097,7 +1098,7 @@ class Version {
   BlobSource* blob_source_;
   const MergeOperator* merge_operator_;
 
-  VersionStorageInfo storage_info_;
+  VersionStorageInfo* storage_info_;
   VersionSet* vset_;  // VersionSet to which this Version belongs
   Version* next_;     // Next version in linked list
   Version* prev_;     // Previous version in linked list
@@ -1678,6 +1679,8 @@ class VersionSet {
 
   // Pointer to the DB's ErrorHandler.
   ErrorHandler* const error_handler_;
+  // Dedicated background thread scheduler for storage deletion operations
+  std::unique_ptr<VersionSetDeletionScheduler> deletion_scheduler_;
 
  private:
   // REQUIRES db mutex at beginning. may release and re-acquire db mutex

db/version_set.h

@@ -0,0 +1,96 @@
+//  Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
+//  This source code is licensed under both the GPLv2 (found in the
+//  COPYING file in the root directory) and Apache 2.0 License
+//  (found in the LICENSE.Apache file in the root directory).
+
+#include "db/version_set_deletion_scheduler.h"
+
+#include <memory>
+
+#include "db/version_set.h"
+#include "logging/logging.h"
+#include "port/port.h"
+#include "rocksdb/env.h"
+#include "util/mutexlock.h"
+
+namespace ROCKSDB_NAMESPACE {
+
+VersionSetDeletionScheduler::VersionSetDeletionScheduler(Logger* info_log)
+    : version_delete_mutex_(),
+      cv_(&version_delete_mutex_),
+      deletion_queue_(),
+      bg_thread_(nullptr),
+      shutting_down_(false),
+      pending_deletion_count_(0),
+      info_log_(info_log) {
+  // Start the background thread
+  bg_thread_.reset(new port::Thread(
+      &VersionSetDeletionScheduler::BackgroundDeletionThread, this));
+
+  ROCKS_LOG_INFO(info_log_,
+                 "VersionSetDeletionScheduler: Created dedicated background "
+                 "thread for storage deletion");
+}
+
+VersionSetDeletionScheduler::~VersionSetDeletionScheduler() { Shutdown(); }
+
+void VersionSetDeletionScheduler::ScheduleDeletion(
+    VersionStorageInfo* storage_info) {
+  MutexLock lock(&version_delete_mutex_);
+
+  if (shutting_down_) {
+    return;
+  }
+
+  deletion_queue_.push(storage_info);
+  pending_deletion_count_++;
+
+  // Notify the background thread that work is available
+  cv_.Signal();
+}
+
+void VersionSetDeletionScheduler::Shutdown() {
+  {
+    MutexLock lock(&version_delete_mutex_);
+    if (shutting_down_) {
+      return;  // Already shutting down
+    }
+    shutting_down_ = true;
+    cv_.Signal();  // Wake up the background thread
+  }
+
+  // Wait for the background thread to finish
+  if (bg_thread_ != nullptr) {
+    bg_thread_->join();
+    bg_thread_.reset();
+  }
+}
+
+void VersionSetDeletionScheduler::BackgroundDeletionThread() {
+  version_delete_mutex_.Lock();
+
+  while (!shutting_down_) {
+    // Wait for work or shutdown signal
+    while (deletion_queue_.empty() && !shutting_down_) {
+      cv_.Wait();
+    }
+
+    // Process all pending deletion operations
+    while (!deletion_queue_.empty() && !shutting_down_) {
+      VersionStorageInfo* storage_info = deletion_queue_.front();
+      deletion_queue_.pop();
+      pending_deletion_count_--;
+
+      // Unlock mutex while executing the deletion operation
+      version_delete_mutex_.Unlock();
+      delete storage_info;
+
+      // Re-acquire the mutex for the next iteration
+      version_delete_mutex_.Lock();
+    }
+  }
+
+  version_delete_mutex_.Unlock();
+}
+
+}  // namespace ROCKSDB_NAMESPACE

db/version_set_deletion_scheduler.cc

@@ -0,0 +1,68 @@
+//  Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
+//  This source code is licensed under both the GPLv2 (found in the
+//  COPYING file in the root directory) and Apache 2.0 License
+//  (found in the LICENSE.Apache file in the root directory).
+
+#pragma once
+
+#include <memory>
+#include <queue>
+
+#include "port/port.h"
+#include "rocksdb/rocksdb_namespace.h"
+
+namespace ROCKSDB_NAMESPACE {
+
+class VersionSet;
+class VersionStorageInfo;
+class Logger;
+
+// VersionSetDeletionScheduler provides a dedicated background thread for
+// handling VersionStorageInfo deletion operations that were previously executed
+// via env_->Schedule() in the LOW priority thread pool.
+//
+// This dedicated thread helps to:
+// 1. Isolate version storage deletion operations from other background work
+// 2. Avoid potential interference with other low-priority operations
+// 3. Provide better control over deletion timing and parallelism
+//
+class VersionSetDeletionScheduler {
+ public:
+  explicit VersionSetDeletionScheduler(Logger* info_log);
+  ~VersionSetDeletionScheduler();
+
+  // Schedule a VersionStorageInfo deletion to be executed in the background
+  // thread
+  void ScheduleDeletion(VersionStorageInfo* storage_info);
+
+  // Wait for all pending operations to complete and shutdown the background
+  // thread
+  void Shutdown();
+
+ private:
+  // Entry point for the background thread
+  void BackgroundDeletionThread();
+
+  // Mutex to protect internal state
+  mutable port::Mutex version_delete_mutex_;
+
+  // Condition variable for thread coordination
+  port::CondVar cv_;
+
+  // Queue of deletion operations
+  std::queue<VersionStorageInfo*> deletion_queue_;
+
+  // Background thread for executing deletion operations
+  std::unique_ptr<port::Thread> bg_thread_;
+
+  // Flag to indicate shutdown
+  bool shutting_down_;
+
+  // Number of pending deletion operations
+  int pending_deletion_count_;
+
+  // Logger for debug/info messages
+  Logger* info_log_;
+};
+
+}  // namespace ROCKSDB_NAMESPACE

db/version_set_deletion_scheduler.h

@@ -92,6 +92,7 @@ LIB_SOURCES =                                                   \
   db/version_edit.cc                                            \
   db/version_edit_handler.cc                                    \
   db/version_set.cc                                             \
+  db/version_set_deletion_scheduler.cc                          \
   db/wal_edit.cc                                                \
   db/wal_manager.cc                                             \
   db/wide/wide_column_serialization.cc                          \