New stat rocksdb.cf-write-stall-stats exposed in structural map

CMakeLists.txt

@@ -782,6 +782,7 @@ set(SOURCES
         memtable/skiplistrep.cc
         memtable/vectorrep.cc
         memtable/write_buffer_manager.cc
+        memtable/write_stall_stats.cc
         monitoring/histogram.cc
         monitoring/histogram_windowing.cc
         monitoring/in_memory_stats_history.cc

HISTORY.md

@@ -13,6 +13,8 @@
 * Add statistics rocksdb.secondary.cache.filter.hits, rocksdb.secondary.cache.index.hits, and rocksdb.secondary.cache.filter.hits
 * Added a new PerfContext counter `internal_merge_point_lookup_count` which tracks the number of Merge operands applied while serving point lookup queries.
 * Add new statistics rocksdb.table.open.prefetch.tail.read.bytes, rocksdb.table.open.prefetch.tail.{miss|hit}
+* Add new db property `rocksdb.cf-write-stall-stats` and APIs to examine it in a structured way.
+In particular, users of `GetMapProperty()` with property `kCFWriteStallStats` can now use the functions in `CFWriteStallStatsMapKeys` to find stats in the map.
 
 ## 8.0.0 (02/19/2023)
 ### Behavior changes

TARGETS

@@ -143,6 +143,7 @@ cpp_library_wrapper(name="rocksdb_lib", srcs=[
         "memtable/skiplistrep.cc",
         "memtable/vectorrep.cc",
         "memtable/write_buffer_manager.cc",
+        "memtable/write_stall_stats.cc",
         "monitoring/histogram.cc",
         "monitoring/histogram_windowing.cc",
         "monitoring/in_memory_stats_history.cc",
@@ -487,6 +488,7 @@ cpp_library_wrapper(name="rocksdb_whole_archive_lib", srcs=[
         "memtable/skiplistrep.cc",
         "memtable/vectorrep.cc",
         "memtable/write_buffer_manager.cc",
+        "memtable/write_stall_stats.cc",
         "monitoring/histogram.cc",
         "monitoring/histogram_windowing.cc",
         "monitoring/in_memory_stats_history.cc",

db/column_family.cc

@@ -869,7 +869,7 @@ int GetL0ThresholdSpeedupCompaction(int level0_file_num_compaction_trigger,
 }
 }  // anonymous namespace
 
-std::pair<WriteStallCondition, ColumnFamilyData::WriteStallCause>
+std::pair<WriteStallCondition, WriteStallCause>
 ColumnFamilyData::GetWriteStallConditionAndCause(
     int num_unflushed_memtables, int num_l0_files,
     uint64_t num_compaction_needed_bytes,
@@ -939,10 +939,12 @@ WriteStallCondition ColumnFamilyData::RecalculateWriteStallConditions(
     } else if (write_stall_condition == WriteStallCondition::kStopped &&
                write_stall_cause == WriteStallCause::kL0FileCountLimit) {
       write_controller_token_ = write_controller->GetStopToken();
-      internal_stats_->AddCFStats(InternalStats::L0_FILE_COUNT_LIMIT_STOPS, 1);
+      internal_stats_->AddCFStats(
+          InternalStats::L0_FILE_COUNT_LIMIT_STOPS_WITH_ONGOING_COMPACTION, 1);
       if (compaction_picker_->IsLevel0CompactionInProgress()) {
         internal_stats_->AddCFStats(
-            InternalStats::LOCKED_L0_FILE_COUNT_LIMIT_STOPS, 1);
+            InternalStats::L0_FILE_COUNT_LIMIT_STOPS_WITH_ONGOING_COMPACTION,
+            1);
       }
       ROCKS_LOG_WARN(ioptions_.logger,
                      "[%s] Stopping writes because we have %d level-0 files",
@@ -985,7 +987,9 @@ WriteStallCondition ColumnFamilyData::RecalculateWriteStallConditions(
                                   1);
       if (compaction_picker_->IsLevel0CompactionInProgress()) {
         internal_stats_->AddCFStats(
-            InternalStats::LOCKED_L0_FILE_COUNT_LIMIT_SLOWDOWNS, 1);
+            InternalStats::
+                L0_FILE_COUNT_LIMIT_SLOWDOWNS_WITH_ONGOING_COMPACTION,
+            1);
       }
       ROCKS_LOG_WARN(ioptions_.logger,
                      "[%s] Stalling writes because we have %d level-0 files "

db/column_family.h

@@ -462,12 +462,6 @@ class ColumnFamilyData {
   bool queued_for_flush() { return queued_for_flush_; }
   bool queued_for_compaction() { return queued_for_compaction_; }
 
-  enum class WriteStallCause {
-    kNone,
-    kMemtableLimit,
-    kL0FileCountLimit,
-    kPendingCompactionBytes,
-  };
   static std::pair<WriteStallCondition, WriteStallCause>
   GetWriteStallConditionAndCause(
       int num_unflushed_memtables, int num_l0_files,

db/db_impl/db_impl_write.cc

@@ -1211,6 +1211,18 @@ Status DBImpl::PreprocessWrite(const WriteOptions& write_options,
   // exceeded at this point so no new write (including current one) will go
   // through until memory usage is decreased.
   if (UNLIKELY(status.ok() && write_buffer_manager_->ShouldStall())) {
+    {
+      InstrumentedMutexLock l(&mutex_);
+      ColumnFamilySet* column_families = versions_->GetColumnFamilySet();
+      assert(column_families);
+      for (auto const cfd : *column_families) {
+        if (cfd->IsDropped() || !cfd->initialized()) {
+          continue;
+        }
+        cfd->internal_stats()->AddCFStats(
+            InternalStats::WRITE_BUFFER_MANAGER_LIMIT_STOPS, 1);
+      }
+    }
     if (write_options.no_slowdown) {
       status = Status::Incomplete("Write stall");
     } else {

db/db_properties_test.cc

@@ -2106,6 +2106,97 @@ TEST_F(DBPropertiesTest, GetMapPropertyBlockCacheEntryStats) {
   ASSERT_EQ(3 * kNumCacheEntryRoles + 4, values.size());
 }
 
+TEST_F(DBPropertiesTest, GetMapPropertyCFWriteStallStats) {
+  // Test 1: stats of write stall caused by WriteBufferManager
+  Options options = CurrentOptions();
+  options.write_buffer_manager.reset(
+      new WriteBufferManager(100000, nullptr, true));
+  CreateAndReopenWithCF({"stall_cf"}, options);
+
+  for (int cf = 0; cf <= 1; ++cf) {
+    std::map<std::string, std::string> values;
+    ASSERT_TRUE(db_->GetMapProperty(
+        handles_[cf], DB::Properties::kCFWriteStallStats, &values));
+    ASSERT_EQ(std::stoi(values[CFWriteStallStatsMapKeys::TotalStop()]), 0);
+    ASSERT_EQ(std::stoi(values[CFWriteStallStatsMapKeys::TotalSlowDown()]), 0);
+  }
+
+  ASSERT_OK(Put(0, Key(1), DummyString(100000)));
+  ASSERT_OK(Put(1, Key(1), DummyString(100000)));
+
+  // Write stall caused by WriteBufferManager affects and increases stats from
+  // all CFs
+  for (int cf = 0; cf <= 1; ++cf) {
+    std::map<std::string, std::string> values;
+    ASSERT_TRUE(db_->GetMapProperty(
+        handles_[cf], DB::Properties::kCFWriteStallStats, &values));
+    ASSERT_EQ(std::stoi(values[CFWriteStallStatsMapKeys::TotalStop()]), 1);
+    ASSERT_EQ(std::stoi(values[CFWriteStallStatsMapKeys::CauseConditionCount(
+                  WriteStallCause::kWriteBufferManagerLimit,
+                  WriteStallCondition::kStopped)]),
+              1);
+    ASSERT_EQ(std::stoi(values[CFWriteStallStatsMapKeys::TotalSlowDown()]), 0);
+    ASSERT_EQ(std::stoi(values[CFWriteStallStatsMapKeys::CauseConditionCount(
+                  WriteStallCause::kWriteBufferManagerLimit,
+                  WriteStallCondition::kDelayed)]),
+              0);
+  }
+
+  Close();
+
+  // Test 2: stats of write stall caused by memtable limit in `stall_cf` only
+  options.write_buffer_manager.reset();
+  options.max_write_buffer_number = 2;
+  options.disable_auto_compactions = true;
+  ReopenWithColumnFamilies({"default", "stall_cf"}, options);
+
+  for (int cf = 0; cf <= 1; ++cf) {
+    std::map<std::string, std::string> values;
+    ASSERT_TRUE(db_->GetMapProperty(
+        handles_[cf], DB::Properties::kCFWriteStallStats, &values));
+    ASSERT_EQ(std::stoi(values[CFWriteStallStatsMapKeys::TotalStop()]), 0);
+    ASSERT_EQ(std::stoi(values[CFWriteStallStatsMapKeys::TotalSlowDown()]), 0);
+  }
+
+  // Pause flush thread so we have enough unflushed memtables to coerce write
+  // stall caused by memtable limit later
+  std::unique_ptr<test::SleepingBackgroundTask> sleeping_task(
+      new test::SleepingBackgroundTask());
+  env_->SetBackgroundThreads(1, Env::HIGH);
+  env_->Schedule(&test::SleepingBackgroundTask::DoSleepTask,
+                 sleeping_task.get(), Env::Priority::HIGH);
+  sleeping_task->WaitUntilSleeping();
+
+  // Create 2 unflushed memtables to reach the memtable limit
+  FlushOptions fo;
+  fo.allow_write_stall = true;
+  fo.wait = false;
+  ASSERT_OK(Put(1, Key(1), DummyString(1)));
+  ASSERT_OK(dbfull()->Flush(fo, handles_[1]));
+  ASSERT_OK(Put(1, Key(2), DummyString(1)));
+  ASSERT_OK(dbfull()->Flush(fo, handles_[1]));
+
+  for (int cf = 0; cf <= 1; ++cf) {
+    std::map<std::string, std::string> values;
+    ASSERT_TRUE(db_->GetMapProperty(
+        handles_[cf], DB::Properties::kCFWriteStallStats, &values));
+    ASSERT_EQ(std::stoi(values[CFWriteStallStatsMapKeys::TotalStop()]),
+              cf == 1 ? 1 : 0);
+    ASSERT_EQ(
+        std::stoi(values[CFWriteStallStatsMapKeys::CauseConditionCount(
+            WriteStallCause::kMemtableLimit, WriteStallCondition::kStopped)]),
+        cf == 1 ? 1 : 0);
+    ASSERT_EQ(std::stoi(values[CFWriteStallStatsMapKeys::TotalSlowDown()]), 0);
+    ASSERT_EQ(
+        std::stoi(values[CFWriteStallStatsMapKeys::CauseConditionCount(
+            WriteStallCause::kMemtableLimit, WriteStallCondition::kDelayed)]),
+        0);
+  }
+
+  sleeping_task->WakeUp();
+  sleeping_task->WaitUntilDone();
+}
+
 namespace {
 std::string PopMetaIndexKey(InternalIterator* meta_iter) {
   Status s = meta_iter->status();