Add running_flushes to delay count

speedb-io · Feb 19, 2024 · 9b5cbd1 · 9b5cbd1
1 parent 6a30334
commit 9b5cbd1
Show file tree

Hide file tree

Showing 3 changed files with 22 additions and 5 deletions.
diff --git a/db/column_family.cc b/db/column_family.cc
@@ -1064,8 +1064,8 @@ ColumnFamilyData::CalculateWriteDelayDividerAndMaybeUpdateWriteStallCause(
   // linear step, and num_delay_steps is num_steps_till_stop - 1.
   double l0_divider = 1;
   const int kGoalMbStep = 1;
-  const auto extra_l0_ssts =
-      vstorage->l0_delay_trigger_count() - file_to_start_delay;
+  const auto extra_l0_ssts = vstorage->l0_delay_trigger_count() +
+                             running_flushes_ - file_to_start_delay;
   if (extra_l0_ssts > 0) {
     const auto num_steps_till_stop = stop - file_to_start_delay;
     assert(num_steps_till_stop > 0);
@@ -1093,7 +1093,7 @@ ColumnFamilyData::CalculateWriteDelayDividerAndMaybeUpdateWriteStallCause(
       assert(delay_percent > 0 && delay_percent < 1);
       // since extra_l0_ssts == num_steps_till_stop then we're in a stop
       // condition.
-      assert(extra_l0_ssts < num_steps_till_stop);
+      assert(extra_l0_ssts - running_flushes_ < num_steps_till_stop);
       auto l0_rate = last_linear_step_mbs *
                      std::pow(delay_percent, (extra_l0_ssts - linear_steps));
       l0_divider = initial_linear_rate / l0_rate;
@@ -1147,7 +1147,8 @@ WriteStallCondition ColumnFamilyData::RecalculateWriteStallConditions(
         vstorage->estimated_compaction_needed_bytes();
 
     auto write_stall_condition_and_cause = GetWriteStallConditionAndCause(
-        imm()->NumNotFlushed(), vstorage->l0_delay_trigger_count(),
+        imm()->NumNotFlushed(),
+        vstorage->l0_delay_trigger_count() + running_flushes_,
         vstorage->estimated_compaction_needed_bytes(), mutable_cf_options,
         *ioptions());
     write_stall_condition = write_stall_condition_and_cause.first;

diff --git a/db/column_family.h b/db/column_family.h
@@ -497,6 +497,16 @@ class ColumnFamilyData {
 
   bool IsLastLevelWithData(int level) const;
 
+  void DecrRunningFlushes() {
+    assert(running_flushes_ > 0);
+    --running_flushes_;
+  }
+
+  void IncrRunningFlushes() {
+    assert(running_flushes_ >= 0);
+    ++running_flushes_;
+  }
+
   // REQUIREMENT: db mutex must be held
   double TEST_CalculateWriteDelayDivider(
       uint64_t compaction_needed_bytes,
@@ -542,6 +552,8 @@ class ColumnFamilyData {
       const MutableCFOptions& mutable_cf_options,
       WriteStallCause& write_stall_cause);
 
+  std::atomic<int> running_flushes_ = 0;
+
  public:
   void set_initialized() { initialized_.store(true); }
 

diff --git a/db/flush_job.cc b/db/flush_job.cc
@@ -158,9 +158,13 @@ FlushJob::FlushJob(
   TEST_SYNC_POINT("FlushJob::FlushJob()");
 }
 
-FlushJob::~FlushJob() { ThreadStatusUtil::ResetThreadStatus(); }
+FlushJob::~FlushJob() {
+  cfd_->DecrRunningFlushes();
+  ThreadStatusUtil::ResetThreadStatus();
+}
 
 void FlushJob::ReportStartedFlush() {
+  cfd_->IncrRunningFlushes();
   ThreadStatusUtil::SetEnableTracking(db_options_.enable_thread_tracking);
   ThreadStatusUtil::SetColumnFamily(cfd_);
   ThreadStatusUtil::SetThreadOperation(ThreadStatus::OP_FLUSH);