Add num-running-compaction-iterators stat

Author: archang19

db/db_impl/db_impl.h

@@ -2430,6 +2430,8 @@ class DBImpl : public DB {
                                const std::vector<CompactionInputFiles>& inputs,
                                bool* sfm_bookkeeping, LogBuffer* log_buffer);
 
+  int GetNumberCompactionIterators(Compaction* c);
+
   // Request compaction tasks token from compaction thread limiter.
   // It always succeeds if force = true or limiter is disable.
   bool RequestCompactionToken(ColumnFamilyData* cfd, bool force,
@@ -2963,6 +2965,9 @@ class DBImpl : public DB {
   // stores the number of compactions are currently running
   int num_running_compactions_;
 
+  // stores the number of iterators required for currently running compactions
+  int num_running_compaction_iterators_;
+
   // number of background memtable flush jobs, submitted to the HIGH pool
   int bg_flush_scheduled_;
 

db/db_impl/db_impl_compaction_flush.cc

@@ -61,6 +61,20 @@ bool DBImpl::EnoughRoomForCompaction(
   return enough_room;
 }
 
+int DBImpl::GetNumberCompactionIterators(Compaction* c) {
+  assert(c);
+  int num_l0_files = 0;
+  int num_non_l0_levels = 0;
+  for (auto& each_level : *c->inputs()) {
+    if (each_level.level == 0) {
+      num_l0_files += each_level.files.size();
+    } else {
+      num_non_l0_levels++;
+    }
+  }
+  return num_l0_files + num_non_l0_levels;
+}
+
 bool DBImpl::RequestCompactionToken(ColumnFamilyData* cfd, bool force,
                                     std::unique_ptr<TaskLimiterToken>* token,
                                     LogBuffer* log_buffer) {
@@ -3410,6 +3424,9 @@ void DBImpl::BackgroundCallCompaction(PrepickedCompaction* prepicked_compaction,
     InstrumentedMutexLock l(&mutex_);
 
     num_running_compactions_++;
+    int num_compaction_iterators =
+        GetNumberCompactionIterators(prepicked_compaction->compaction);
+    num_running_compaction_iterators_ += num_compaction_iterators;
 
     std::unique_ptr<std::list<uint64_t>::iterator>
         pending_outputs_inserted_elem(new std::list<uint64_t>::iterator(
@@ -3484,6 +3501,8 @@ void DBImpl::BackgroundCallCompaction(PrepickedCompaction* prepicked_compaction,
 
     assert(num_running_compactions_ > 0);
     num_running_compactions_--;
+    assert(num_running_compaction_iterators_ >= num_compaction_iterators);
+    num_running_compaction_iterators_ -= num_compaction_iterators;
 
     if (bg_thread_pri == Env::Priority::LOW) {
       bg_compaction_scheduled_--;

db/internal_stats.cc

@@ -301,6 +301,8 @@ static const std::string aggregated_table_properties =
 static const std::string aggregated_table_properties_at_level =
     aggregated_table_properties + "-at-level";
 static const std::string num_running_compactions = "num-running-compactions";
+static const std::string num_running_compaction_iterators =
+    "num-running-compaction-iterators";
 static const std::string num_running_flushes = "num-running-flushes";
 static const std::string actual_delayed_write_rate =
     "actual-delayed-write-rate";
@@ -351,6 +353,8 @@ const std::string DB::Properties::kCompactionPending =
     rocksdb_prefix + compaction_pending;
 const std::string DB::Properties::kNumRunningCompactions =
     rocksdb_prefix + num_running_compactions;
+const std::string DB::Properties::kNumRunningCompactionIterators =
+    rocksdb_prefix + num_running_compaction_iterators;
 const std::string DB::Properties::kNumRunningFlushes =
     rocksdb_prefix + num_running_flushes;
 const std::string DB::Properties::kBackgroundErrors =
@@ -580,6 +584,9 @@ const UnorderedMap<std::string, DBPropertyInfo>
         {DB::Properties::kNumRunningCompactions,
          {false, nullptr, &InternalStats::HandleNumRunningCompactions, nullptr,
           nullptr}},
+        {DB::Properties::kNumRunningCompactionIterators,
+         {false, nullptr, &InternalStats::HandleNumRunningCompactionIterators,
+          nullptr, nullptr}},
         {DB::Properties::kActualDelayedWriteRate,
          {false, nullptr, &InternalStats::HandleActualDelayedWriteRate, nullptr,
           nullptr}},
@@ -1265,6 +1272,13 @@ bool InternalStats::HandleNumRunningCompactions(uint64_t* value, DBImpl* db,
   return true;
 }
 
+bool InternalStats::HandleNumRunningCompactionIterators(uint64_t* value,
+                                                        DBImpl* db,
+                                                        Version* /*version*/) {
+  *value = db->num_running_compaction_iterators_;
+  return true;
+}
+
 bool InternalStats::HandleBackgroundErrors(uint64_t* value, DBImpl* /*db*/,
                                            Version* /*version*/) {
   // Accumulated number of  errors in background flushes or compactions.

db/internal_stats.h

@@ -847,6 +847,8 @@ class InternalStats {
   bool HandleCompactionPending(uint64_t* value, DBImpl* db, Version* version);
   bool HandleNumRunningCompactions(uint64_t* value, DBImpl* db,
                                    Version* version);
+  bool HandleNumRunningCompactionIterators(uint64_t* value, DBImpl* db,
+                                           Version* version);
   bool HandleBackgroundErrors(uint64_t* value, DBImpl* db, Version* version);
   bool HandleCurSizeActiveMemTable(uint64_t* value, DBImpl* db,
                                    Version* version);

include/rocksdb/db.h

@@ -1210,6 +1210,10 @@ class DB {
     //      running compactions.
     static const std::string kNumRunningCompactions;
 
+    //  "rocksdb.num-running-compaction-iterators" - returns the number of
+    //  iterators required for currently running compactions.
+    static const std::string kNumRunningCompactionIterators;
+
     //  "rocksdb.background-errors" - returns accumulated number of background
     //      errors.
     static const std::string kBackgroundErrors;