fix: Implement periodic logging of task queue size (#1978)

* fix: Implement periodic logging of task queue size

Author: Tianpingan

include/pika_client_processor.h

@@ -21,6 +21,7 @@ class PikaClientProcessor {
   void SchedulePool(net::TaskFunc func, void* arg);
   void ScheduleBgThreads(net::TaskFunc func, void* arg, const std::string& hash_str);
   size_t ThreadPoolCurQueueSize();
+  size_t ThreadPoolMaxQueueSize();
 
  private:
   std::unique_ptr<net::ThreadPool> pool_;

include/pika_server.h

@@ -240,6 +240,7 @@ class PikaServer : public pstd::noncopyable {
   void ScheduleClientBgThreads(net::TaskFunc func, void* arg, const std::string& hash_str);
   // for info debug
   size_t ClientProcessorThreadPoolCurQueueSize();
+  size_t ClientProcessorThreadPoolMaxQueueSize();
 
   /*
    * BGSave used
@@ -518,7 +519,8 @@ class PikaServer : public pstd::noncopyable {
   void AutoDeleteExpiredDump();
   void AutoKeepAliveRSync();
   void AutoUpdateNetworkMetric();
-
+  void PrintThreadPoolQueueStatus();
+  
   std::string host_;
   int port_ = 0;
   time_t start_time_s_ = 0;

src/pika_client_processor.cc

@@ -54,3 +54,11 @@ size_t PikaClientProcessor::ThreadPoolCurQueueSize() {
   }
   return cur_size;
 }
+
+size_t PikaClientProcessor::ThreadPoolMaxQueueSize() {
+  size_t cur_size = 0;
+  if (pool_) {
+    cur_size = pool_->max_queue_size();
+  }
+  return cur_size;
+}

src/pika_server.cc

@@ -34,6 +34,8 @@ extern PikaServer* g_pika_server;
 extern std::unique_ptr<PikaReplicaManager> g_pika_rm;
 extern std::unique_ptr<PikaCmdTableManager> g_pika_cmd_table_manager;
 extern std::unique_ptr<net::NetworkStatistic> g_network_statistic;
+// QUEUE_SIZE_THRESHOLD_PERCENTAGE is used to represent a percentage value and should be within the range of 0 to 100.
+const size_t QUEUE_SIZE_THRESHOLD_PERCENTAGE = 75;
 
 void DoPurgeDir(void* arg) {
   std::unique_ptr<std::string> path(static_cast<std::string*>(arg));
@@ -854,6 +856,13 @@ size_t PikaServer::ClientProcessorThreadPoolCurQueueSize() {
   return pika_client_processor_->ThreadPoolCurQueueSize();
 }
 
+size_t PikaServer::ClientProcessorThreadPoolMaxQueueSize() {
+  if (!pika_client_processor_) {
+    return 0;
+  }
+  return pika_client_processor_->ThreadPoolMaxQueueSize();
+}
+
 void PikaServer::BGSaveTaskSchedule(net::TaskFunc func, void* arg) {
   bgsave_thread_.StartThread();
   bgsave_thread_.Schedule(func, arg);
@@ -1298,6 +1307,8 @@ void PikaServer::DoTimingTask() {
   ResetLastSecQuerynum();
   // Auto update network instantaneous metric
   AutoUpdateNetworkMetric();
+  // Print the queue status periodically
+  PrintThreadPoolQueueStatus();
 }
 
 void PikaServer::AutoCompactRange() {
@@ -1491,6 +1502,16 @@ void PikaServer::AutoUpdateNetworkMetric() {
                                      current_time, factor);
 }
 
+void PikaServer::PrintThreadPoolQueueStatus() {
+    // Print the current queue size if it exceeds QUEUE_SIZE_THRESHOLD_PERCENTAGE/100 of the maximum queue size.
+    size_t cur_size = ClientProcessorThreadPoolCurQueueSize();
+    size_t max_size = ClientProcessorThreadPoolMaxQueueSize();
+    size_t thread_hold = (max_size / 100) * QUEUE_SIZE_THRESHOLD_PERCENTAGE;
+    if (cur_size > thread_hold) {
+      LOG(INFO) << "The current queue size of the Pika Server's client thread processor thread pool: " << cur_size;
+    }
+}
+
 void PikaServer::InitStorageOptions() {
   std::lock_guard rwl(storage_options_rw_);