summary different stages duration && add rocksdb perfcontext (OpenAtomFoundation#3036)

Co-authored-by: wangshaoyi <[email protected]>

Author: wangshao1

include/pika_client_conn.h

@@ -130,7 +130,7 @@ class PikaClientConn : public net::RedisConn {
   std::shared_ptr<Cmd> DoCmd(const PikaCmdArgsType& argv, const std::string& opt,
                              const std::shared_ptr<std::string>& resp_ptr, bool cache_miss_in_rtc);
 
-  void ProcessSlowlog(const PikaCmdArgsType& argv, uint64_t do_duration);
+  void ProcessSlowlog(const PikaCmdArgsType& argv, std::shared_ptr<Cmd> c_ptr);
   void ProcessMonitor(const PikaCmdArgsType& argv);
 
   void ExecRedisCmd(const PikaCmdArgsType& argv, std::shared_ptr<std::string>& resp_ptr, bool cache_miss_in_rtc);

include/pika_command.h

@@ -17,9 +17,16 @@
 #include "net/include/net_conn.h"
 #include "net/include/redis_conn.h"
 #include "pstd/include/pstd_string.h"
+#include "pstd/include/stage_timer.h"
 
 #include "net/src/dispatch_thread.h"
 
+// Declare and set start time of the timer
+#define STAGE_TIMER_GUARD(metric, enabled)  \
+  pstd::StageTimer stage_timer_##metric(    \
+      &metric, enabled);                    \
+  stage_timer_##metric.Start();
+
 class SyncMasterDB;
 class SyncSlaveDB;
 class DB;
@@ -544,7 +551,7 @@ class Cmd : public std::enable_shared_from_this<Cmd> {
   bool IsNeedReadCache() const;
   bool IsNeedCacheDo() const;
   bool HashtagIsConsistent(const std::string& lhs, const std::string& rhs) const;
-  uint64_t GetDoDuration() const { return do_duration_; };
+  virtual std::string StagesDurationSummary(bool exclude_zero_value) const;
   std::shared_ptr<DB> GetDB() const { return db_; };
   uint32_t AclCategory() const;
   void AddAclCategory(uint32_t aclCategory);
@@ -599,7 +606,13 @@ class Cmd : public std::enable_shared_from_this<Cmd> {
   std::weak_ptr<net::NetConn> conn_;
   std::weak_ptr<std::string> resp_;
   CmdStage stage_ = kNone;
-  uint64_t do_duration_ = 0;
+
+  uint64_t acquire_lock_duration_ms = 0;
+  uint64_t command_duration_ms = 0;
+  uint64_t binlog_duration_ms = 0;
+  uint64_t storage_duration_ms = 0;
+  uint64_t cache_duration_ms = 0;
+
   uint32_t cmdId_ = 0;
   uint32_t aclCategory_ = 0;
   bool cache_missed_in_rtc_{false};

include/pika_conf.h

@@ -744,6 +744,30 @@ class PikaConf : public pstd::BaseConf {
     rsync_timeout_ms_.store(value);
   }
 
+  int RocksDBPerfLevel() const {
+    return rocksdb_perf_level_.load();
+  }
+
+  bool UpdateRocksDBPerfLevel(int perf_level) {
+    if (perf_level >= 6 || perf_level < 0) {
+      return false;
+    }
+    rocksdb_perf_level_.store(perf_level);
+    return true;
+  }
+
+  int RocksDBPerfPercent() const {
+    return rocksdb_perf_percent_.load();
+  }
+
+  bool UpdateRocksDBPerfPercent(int percent) {
+    if (percent > 100 || percent < 0) {
+      return false;
+    }
+    rocksdb_perf_percent_.store(percent);
+    return true;
+  }
+
   void SetAclPubsubDefault(const std::string& value) {
     std::lock_guard l(rwlock_);
     TryPushDiffCommands("acl-pubsub-default", value);
@@ -984,6 +1008,21 @@ class PikaConf : public pstd::BaseConf {
   int max_rsync_parallel_num_ = kMaxRsyncParallelNum;
   std::atomic_int64_t rsync_timeout_ms_ = 1000;
 
+  /*
+  kUninitialized = 0,             // unknown setting
+  kDisable = 1,                   // disable perf stats
+  kEnableCount = 2,               // enable only count stats
+  kEnableTimeExceptForMutex = 3,  // Other than count stats, also enable time
+                                  // stats except for mutexes
+  // Other than time, also measure CPU time counters. Still don't measure
+  // time (neither wall time nor CPU time) for mutexes.
+  kEnableTimeAndCPUTimeExceptForMutex = 4,
+  kEnableTime = 5,  // enable count and time stats
+  kOutOfBounds = 6  // N.B. Must always be the last value!
+  */
+  std::atomic_int rocksdb_perf_level_ = 2;
+  std::atomic_int rocksdb_perf_percent_ = 10;
+
   //Internal used metrics Persisted by pika.conf
   std::unordered_set<std::string> internal_used_unfinished_full_sync_;
 };

src/pika_admin.cc

@@ -2196,6 +2196,18 @@ void ConfigCmd::ConfigGet(std::string& ret) {
     EncodeNumber(&config_body, g_pika_conf->throttle_bytes_per_second());
   }
 
+  if (pstd::stringmatch(pattern.data(), "rocksdb-perf-level", 1) != 0) {
+    elements += 2;
+    EncodeString(&config_body, "rocksdb-perf-level");
+    EncodeNumber(&config_body, g_pika_conf->RocksDBPerfLevel());
+  }
+
+  if (pstd::stringmatch(pattern.data(), "rocksdb-perf-percent", 1) != 0) {
+    elements += 2;
+    EncodeString(&config_body, "rocksdb-perf-percent");
+    EncodeNumber(&config_body, g_pika_conf->RocksDBPerfPercent());
+  }
+
   if (pstd::stringmatch(pattern.data(), "max-rsync-parallel-num", 1) != 0) {
     elements += 2;
     EncodeString(&config_body, "max-rsync-parallel-num");
@@ -2740,6 +2752,32 @@ void ConfigCmd::ConfigSet(std::shared_ptr<DB> db) {
     }
     g_pika_conf->SetArenaBlockSize(static_cast<int>(ival));
     res_.AppendStringRaw("+OK\r\n");
+  } else if (set_item == "rocksdb-perf-level") {
+    if ((pstd::string2int(value.data(), value.size(), &ival) == 0) || ival <= 0) {
+      res_.AppendStringRaw("-ERR Invalid argument \'" + value + "\' for CONFIG SET 'rocksdb-perf-level'\r\n");
+      return;
+    }
+    bool success = g_pika_conf->UpdateRocksDBPerfLevel(int(ival));
+    LOG(INFO) << "update rocksdb-perf-level to " << ival
+              << (success ? " success" : " failed");
+    if (!success) {
+      res_.AppendStringRaw("-ERR Invalid argument \'" + value + "\' for CONFIG SET 'rocksdb-perf-level', should between 1 and 5\r\n");
+      return;
+    }
+    res_.AppendStringRaw("+OK\r\n");
+  } else if (set_item == "rocksdb-perf-percent") {
+    if ((pstd::string2int(value.data(), value.size(), &ival) == 0) || ival <= 0) {
+      res_.AppendStringRaw("-ERR Invalid argument \'" + value + "\' for CONFIG SET 'rocksdb-perf-percent'\r\n");
+      return;
+    }
+    bool success = g_pika_conf->UpdateRocksDBPerfPercent(int(ival));
+    LOG(INFO) << "update rocksdb-perf-percent to " << ival
+              << (success ? " success" : " failed");
+    if (!success) {
+      res_.AppendStringRaw("-ERR Invalid argument \'" + value + "\' for CONFIG SET 'rocksdb-perf-percent', should between 0 and 100\r\n");
+      return;
+    }
+    res_.AppendStringRaw("+OK\r\n");
   } else if (set_item == "throttle-bytes-per-second") {
     if ((pstd::string2int(value.data(), value.size(), &ival) == 0) || ival <= 0) {
       res_.AppendStringRaw("-ERR Invalid argument \'" + value + "\' for CONFIG SET 'throttle-bytes-per-second'\r\n");

src/pika_client_conn.cc

@@ -20,6 +20,10 @@
 #include "net/src/worker_thread.h"
 #include "src/pstd/include/scope_record_lock.h"
 
+#include "rocksdb/perf_context.h"
+#include "rocksdb/iostats_context.h"
+#include "util/random.h"
+
 extern std::unique_ptr<PikaConf> g_pika_conf;
 extern PikaServer* g_pika_server;
 extern std::unique_ptr<PikaReplicaManager> g_pika_rm;
@@ -46,6 +50,7 @@ std::shared_ptr<Cmd> PikaClientConn::DoCmd(const PikaCmdArgsType& argv, const st
     }
     return tmp_ptr;
   }
+
   c_ptr->SetCacheMissedInRtc(cache_miss_in_rtc);
   c_ptr->SetConn(shared_from_this());
   c_ptr->SetResp(resp_ptr);
@@ -202,21 +207,32 @@ std::shared_ptr<Cmd> PikaClientConn::DoCmd(const PikaCmdArgsType& argv, const st
     }
   }
 
+
+  // set rocksdb perflevel based on RocksDBPerfLevel and RocksDBPerfPercent
+  int rocksdb_perf_level = 2;
+  if (rocksdb::Random::GetTLSInstance()->PercentTrue(g_pika_conf->RocksDBPerfPercent())) {
+    rocksdb_perf_level = g_pika_conf->RocksDBPerfLevel();
+  }
+  rocksdb::SetPerfLevel(rocksdb::PerfLevel(rocksdb_perf_level));
+
+  // Perform some operations
+  rocksdb::get_perf_context()->Reset();
   // Process Command
   c_ptr->Execute();
+
   time_stat_->process_done_ts_ = pstd::NowMicros();
   auto cmdstat_map = g_pika_cmd_table_manager->GetCommandStatMap();
   (*cmdstat_map)[opt].cmd_count.fetch_add(1);
   (*cmdstat_map)[opt].cmd_time_consuming.fetch_add(time_stat_->total_time());
 
   if (g_pika_conf->slowlog_slower_than() >= 0) {
-    ProcessSlowlog(argv, c_ptr->GetDoDuration());
+    ProcessSlowlog(argv, c_ptr);
   }
 
   return c_ptr;
 }
 
-void PikaClientConn::ProcessSlowlog(const PikaCmdArgsType& argv, uint64_t do_duration) {
+void PikaClientConn::ProcessSlowlog(const PikaCmdArgsType& argv, std::shared_ptr<Cmd> c_ptr) {
   if (time_stat_->total_time() > g_pika_conf->slowlog_slower_than()) {
     g_pika_server->SlowlogPushEntry(argv, time_stat_->start_ts() / 1000000, time_stat_->total_time());
     if (g_pika_conf->slowlog_write_errorlog()) {
@@ -241,7 +257,8 @@ void PikaClientConn::ProcessSlowlog(const PikaCmdArgsType& argv, uint64_t do_dur
                  << ", before_queue_time(ms): " << time_stat_->before_queue_time() / 1000
                  << ", queue_time(ms): " << time_stat_->queue_time() / 1000
                  << ", process_time(ms): " << time_stat_->process_time() / 1000
-                 << ", cmd_time(ms): " << do_duration / 1000;
+                 << ", " << c_ptr->StagesDurationSummary(true /*skip zero counter*/)
+                 << ", " << rocksdb::get_perf_context()->ToString(true);
     }
   }
 }

src/pika_command.cc

@@ -4,6 +4,7 @@
 // of patent rights can be found in the PATENTS file in the same directory.
 
 #include <memory>
+#include <sstream>
 #include <utility>
 
 #include <glog/logging.h>
@@ -864,23 +865,22 @@ void Cmd::ProcessCommand(const HintKeys& hint_keys) {
 }
 
 void Cmd::InternalProcessCommand(const HintKeys& hint_keys) {
+  uint64_t start_us = pstd::NowMicros();
   pstd::lock::MultiRecordLock record_lock(db_->LockMgr());
   if (is_write()) {
     record_lock.Lock(current_key());
   }
-  uint64_t start_us = 0;
-  if (g_pika_conf->slowlog_slower_than() >= 0) {
-    start_us = pstd::NowMicros();
-  }
 
   if (!IsSuspend()) {
     db_->DBLockShared();
   }
 
+  uint64_t before_do_command_us = pstd::NowMicros();
+  this->acquire_lock_duration_ms = (before_do_command_us - start_us) / 1000;
   DoCommand(hint_keys);
-  if (g_pika_conf->slowlog_slower_than() >= 0) {
-    do_duration_ += pstd::NowMicros() - start_us;
-  }
+
+  uint64_t before_do_binlog_us = pstd::NowMicros();
+  this->command_duration_ms = (before_do_binlog_us - before_do_command_us) / 1000;
   DoBinlog();
 
   if (!IsSuspend()) {
@@ -889,6 +889,9 @@ void Cmd::InternalProcessCommand(const HintKeys& hint_keys) {
   if (is_write()) {
     record_lock.Unlock(current_key());
   }
+
+  uint64_t end_us = pstd::NowMicros();
+  this->binlog_duration_ms = (end_us - before_do_binlog_us) / 1000;
 }
 
 void Cmd::DoCommand(const HintKeys& hint_keys) {
@@ -966,6 +969,23 @@ void Cmd::DoBinlog() {
   }
 }
 
+#define PIKA_STAGE_DURATION_OUTPUT(duration)      \
+  if (!exclude_zero_value || duration > 0) {      \
+    ss << #duration << " = " << duration << ", "; \
+  }
+
+std::string Cmd::StagesDurationSummary(bool exclude_zero_value) const {
+  std::ostringstream ss;
+  PIKA_STAGE_DURATION_OUTPUT(acquire_lock_duration_ms);
+  PIKA_STAGE_DURATION_OUTPUT(command_duration_ms);
+  PIKA_STAGE_DURATION_OUTPUT(binlog_duration_ms);
+  PIKA_STAGE_DURATION_OUTPUT(storage_duration_ms);
+  PIKA_STAGE_DURATION_OUTPUT(cache_duration_ms);
+  std::string str = ss.str();
+  str.erase(str.find_last_not_of(", ") + 1);
+  return str;
+}
+
 bool Cmd::hasFlag(uint32_t flag) const { return (flag_ & flag); }
 bool Cmd::is_read() const { return (flag_ & kCmdFlagsRead); }
 bool Cmd::is_write() const { return (flag_ & kCmdFlagsWrite); }