fix: run DestroyTerminated periodically without starvation (#308)

Bundle DestroyTerminated together with ProcessSleep and run both of them periodically with predictable frequency but not too frequent. Signed-off-by: Roman Gershman <[email protected]>
romange · Aug 15, 2024 · 50da303 · 50da303
1 parent f9e28c7
commit 50da303
Show file tree

Hide file tree

Showing 5 changed files with 34 additions and 26 deletions.
diff --git a/util/fibers/detail/scheduler.cc b/util/fibers/detail/scheduler.cc
@@ -294,6 +294,7 @@ void Scheduler::ScheduleTermination(FiberInterface* cntx) {
 }
 
 void Scheduler::DestroyTerminated() {
+  unsigned i = 0;
   while (!terminate_queue_.empty()) {
     FiberInterface* tfi = &terminate_queue_.front();
     terminate_queue_.pop_front();
@@ -303,6 +304,10 @@ void Scheduler::DestroyTerminated() {
 
     // maybe someone holds a Fiber handle and waits for the fiber to join.
     intrusive_ptr_release(tfi);
+    ++i;
+  }
+  if (i > 10) {
+    DVLOG(1) << "Destroyed " << i << " fibers";
   }
 }
 

diff --git a/util/fibers/epoll_proactor.cc b/util/fibers/epoll_proactor.cc
@@ -294,9 +294,7 @@ void EpollProactor::MainLoop(detail::Scheduler* scheduler) {
       };
     }
 
-    if (scheduler->HasSleepingFibers()) {
-      ProcessSleepFibers(scheduler);
-    }
+    RunL2Tasks(scheduler);
 
     // must be if and not while - see uring_proactor.cc for more details.
     if (scheduler->HasReady()) {

diff --git a/util/fibers/proactor_base.cc b/util/fibers/proactor_base.cc
@@ -256,30 +256,30 @@ void ProactorBase::RegisterSignal(std::initializer_list<uint16_t> l, std::functi
 }
 
 // The threshold is set to ~2.5ms.
-bool ProactorBase::HasSleepFibersStarved() const {
+bool ProactorBase::ShouldPollL2Tasks() const {
   uint64_t now = GetCPUCycleCount();
-  return now > last_sleep_cycle_ + 256 * cycles_per_10us;
+  return now > last_level2_cycle_ + 256 * cycles_per_10us;
 }
 
-unsigned ProactorBase::ProcessSleepFibers(detail::Scheduler* scheduler) {
-  if (!scheduler->HasSleepingFibers())
-    return 0;
-
+bool ProactorBase::RunL2Tasks(detail::Scheduler* scheduler) {
   // avoid calling steady_clock::now() too much.
   // Cycles count can reset, for example when CPU is suspended, therefore we also allow
   // "returning  into past". False positive is possible but it's not a big deal.
   uint64_t now = GetCPUCycleCount();
-  if (ABSL_PREDICT_FALSE(now < last_sleep_cycle_)) {
+  if (ABSL_PREDICT_FALSE(now < last_level2_cycle_)) {
     // LOG_FIRST_N - otherwise every adjustment will trigger num-threads messages.
     LOG_FIRST_N(WARNING, 1) << "The cycle clock was adjusted backwards by "
-                            << last_sleep_cycle_ - now << " cycles";
-    now = last_sleep_cycle_ + cycles_per_10us;
+                            << last_level2_cycle_ - now << " cycles";
+    now = last_level2_cycle_ + cycles_per_10us;
   }
 
-  unsigned result = 0;
-  if (now >= last_sleep_cycle_ + cycles_per_10us) {
-    last_sleep_cycle_ = now;
-    result = scheduler->ProcessSleep();
+  bool result = false;
+  if (now >= last_level2_cycle_ + cycles_per_10us) {
+    last_level2_cycle_ = now;
+    scheduler->DestroyTerminated();
+    if (scheduler->HasSleepingFibers()) {
+      result = scheduler->ProcessSleep() > 0;
+    }
   }
   return result;
 }

diff --git a/util/fibers/proactor_base.h b/util/fibers/proactor_base.h
@@ -240,11 +240,13 @@ class ProactorBase {
     return absl::GetCurrentTimeNanos();
   }
 
-  // Returns true if we have not processed sleep fibers for too long.
-  bool HasSleepFibersStarved() const;
+  // Returns true if we should poll scheduler tasks that run periodically but not too often.
+  bool ShouldPollL2Tasks() const;
 
-  // Returns number of sleeping fibers being activated.
-  unsigned ProcessSleepFibers(detail::Scheduler* scheduler);
+  // Runs all the tasks that should run periodically but not too often. Skips the run if
+  // they recently run.
+  // Returns true if there are fibers that became ready as a result.
+  bool RunL2Tasks(detail::Scheduler* scheduler);
 
   pthread_t thread_id_ = 0U;
   int sys_thread_id_ = 0;
@@ -303,7 +305,7 @@ class ProactorBase {
     return false;
   }
 
-  uint64_t last_sleep_cycle_ = 0;
+  uint64_t last_level2_cycle_ = 0;
 };
 
 class ProactorDispatcher : public DispatchPolicy {

diff --git a/util/fibers/uring_proactor.cc b/util/fibers/uring_proactor.cc
@@ -691,8 +691,8 @@ void UringProactor::MainLoop(detail::Scheduler* scheduler) {
       // and reap the same amount.
       ReapCompletions(cqe_count, cqes, dispatcher);
 
-      if (HasSleepFibersStarved()) {
-        ProcessSleepFibers(scheduler);
+      if (ShouldPollL2Tasks()) {
+        RunL2Tasks(scheduler);
       }
       continue;
     }
@@ -704,12 +704,15 @@ void UringProactor::MainLoop(detail::Scheduler* scheduler) {
     ///
     /// End of the tight loop that processes tasks, ready fibers, and submits sqes.
     ///
-    unsigned activated = ProcessSleepFibers(scheduler);
-    if (activated > 0) {  // If we have ready fibers - restart the loop.
+    bool activated = RunL2Tasks(scheduler);
+    if (activated) {  // If we have ready fibers - restart the loop.
       continue;
     }
 
-    if (has_cpu_work || io_uring_sq_ready(&ring_) > 0)
+    DCHECK(!has_cpu_work);
+    DCHECK_EQ(io_uring_sq_ready(&ring_), 0u);
+
+    if (io_uring_sq_ready(&ring_) > 0)
       continue;
 
     DCHECK(!scheduler->HasReady());