fast-slow-separate

Author: Z-G-H1

include/pika_define.h

@@ -422,6 +422,7 @@ const int64_t CACHE_LOAD_NUM_ONE_TIME = 256;
  */
 enum TaskPoolType {
   kFastCmdPool,
-  kSlowCmdPool
+  kSlowCmdPool,
+  kAdminCmdPool
 };
 #endif

include/pika_server.h

@@ -84,7 +84,7 @@ void DoBgslotsreload(void* arg);
 // threadpool metrics
 struct ThreadPoolMetrics {
   std::atomic<uint64_t> tasks_scheduled{0};
-  std::atomic<uint64_t> tasks_completed{0};
+  std::atomic<uint64_t> active_tasks{0};
   std::atomic<uint64_t> borrow_attempts{0};
 
   // latency（1ms, 5ms, 10ms, 50ms, 100ms, 500ms, 1s, 5s, >5s）
@@ -567,6 +567,7 @@ class PikaServer : public pstd::noncopyable {
   void AutoDeleteExpiredDump();
   void AutoUpdateNetworkMetric();
   void PrintThreadPoolQueueStatus();
+  void AutoUpdateIdlePoolEMA();
   void StatDiskUsage();
   int64_t GetLastSaveTime(const std::string& dump_dir);
 
@@ -708,22 +709,31 @@ class PikaServer : public pstd::noncopyable {
   PoolLatencyStats slow_pool_stats_;
 
   // EMA configuration parameters (atomic for dynamic updates)
-  std::atomic<uint32_t> ema_alpha_numerator_{5};      // EMA alpha numerator (alpha = 5/100 = 0.05)
-  std::atomic<uint32_t> ema_alpha_denominator_{100};  // EMA alpha denominator
-  std::atomic<uint64_t> fast_busy_threshold_us_{2000}; // Fast pool busy threshold (2ms)
-  std::atomic<uint64_t> fast_idle_threshold_us_{500};  // Fast pool idle threshold (0.5ms)
-  std::atomic<uint64_t> slow_busy_threshold_us_{5000}; // Slow pool busy threshold (5ms)
-  std::atomic<uint64_t> slow_idle_threshold_us_{1000}; // Slow pool idle threshold (1ms)
+  std::atomic<uint32_t> ema_alpha_numerator_{5};
+  std::atomic<uint32_t> ema_alpha_denominator_{100};
+  std::atomic<uint64_t> fast_busy_threshold_us_{2000};
+  std::atomic<uint64_t> fast_idle_threshold_us_{500};
+  std::atomic<uint64_t> slow_busy_threshold_us_{5000};
+  std::atomic<uint64_t> slow_idle_threshold_us_{1000};
 
   // Get EMA queue wait time (us)
   uint64_t GetEMAQueueWait(TaskPoolType pool_type) const;
+
+  // EMA decay helper functions
+  uint64_t CalculateDecayedEMA(uint64_t old_ema, uint64_t last_update, uint64_t now);
+  void UpdateSinglePoolEMA(PoolLatencyStats& stats, uint64_t new_sample, uint64_t now);
+  void DecaySinglePool(PoolLatencyStats& stats, uint64_t now);
 
   // Busy/Idle determination based on EMA (internal helpers)
   bool IsFastPoolBusyByEMA() const;
   bool IsFastPoolIdleByEMA() const;
   bool IsSlowPoolBusyByEMA() const;
   bool IsSlowPoolIdleByEMA() const;
 
+  // Borrow idle window
+  std::atomic<uint64_t> threadpool_idle_window_us_{1000000};
+  std::atomic<uint64_t> fast_pool_last_active_us_{0};
+  std::atomic<uint64_t> slow_pool_last_active_us_{0};
   /*
    * acl
    */

src/pika_client_conn.cc

@@ -233,7 +233,6 @@ std::shared_ptr<Cmd> PikaClientConn::DoCmd(const PikaCmdArgsType& argv, const st
                                 : g_pika_server->GetFastPoolMetrics();
   if (metrics) {
     metrics->RecordLatency(duration_us);
-    metrics->tasks_completed.fetch_add(1, std::memory_order_relaxed);
   }
   if (g_pika_conf->slowlog_slower_than() >= 0) {
     ProcessSlowlog(argv, c_ptr, pipeline_idx, pipeline_total);
@@ -351,15 +350,28 @@ void PikaClientConn::DoBackgroundTask(void* arg) {
   std::unique_ptr<BgTaskArg> bg_arg(static_cast<BgTaskArg*>(arg));
   std::shared_ptr<PikaClientConn> conn_ptr = bg_arg->conn_ptr;
   conn_ptr->task_pool_type_ = bg_arg->pool_type;
-  
+
+  ThreadPoolMetrics* metrics = nullptr;
+  if (bg_arg->pool_type == TaskPoolType::kFastCmdPool) {
+    metrics = g_pika_server->GetFastPoolMetrics();
+  } else if (bg_arg->pool_type == TaskPoolType::kSlowCmdPool) {
+    metrics = g_pika_server->GetSlowPoolMetrics();
+  }
+
+  if (metrics) {
+    metrics->active_tasks.fetch_add(1, std::memory_order_relaxed);
+  }
+  DEFER {
+    if (metrics) {
+      metrics->active_tasks.fetch_sub(1, std::memory_order_relaxed);
+    }
+  };
   // Record dequeue time and calculate queue wait time
   uint64_t now = pstd::NowMicros();
   conn_ptr->time_stat_->dequeue_ts_ = now;
   uint64_t queue_wait = now - conn_ptr->time_stat_->enqueue_ts_;
-  
   // Update EMA statistics for queue wait time
   g_pika_server->UpdateQueueWaitStats(bg_arg->pool_type, queue_wait);
-  
   if (conn_ptr->IsClose()) {
     LOG(INFO) << "conn " << conn_ptr->ip_port() << " has already been closed, skip processing command";
     return;