add big key detection

Author: YuCai18

conf/pika.conf

@@ -494,7 +494,7 @@ default-slot-num : 1024
 # The value option is [yes | no]
 # enable-blob-garbage-collection : no
 
-# the cutoff that the GC logic uses to determine which blob files should be considered “old“.
+# the cutoff that the GC logic uses to determine which blob files should be considered "old".
 # This parameter can be tuned to adjust the trade-off between write amplification and space amplification.
 # blob-garbage-collection-age-cutoff : 0.25
 
@@ -527,6 +527,23 @@ max-rsync-parallel-num : 4
 # The synchronization mode of Pika primary/secondary replication is determined by ReplicationID. ReplicationID in one replication_cluster are the same
 # replication-id :
 
+# The maximum number of big keys to output in 'info' and log output.
+# This controls how many big key entries are shown at most for each type.
+# Default: 10
+BIGKEYS_SHOW_LIMIT = 10
+
+# The threshold for member count to trigger big key detection
+# Default: 10000
+bigkeys_member_threshold : 10000
+
+# The threshold for key and value length (in bytes) 
+# to trigger big key detection (for string type).
+# Default: 1048576 (1MB)
+bigkeys_key_value_length_threshold : 1048576
+
+# The interval (in minutes) for outputting big key statistics to the log.
+# Default: 1
+bigkeys_log_interval : 1
 ###################
 ## Cache Settings
 ###################
@@ -671,3 +688,9 @@ dont-compact-sst-created-in-seconds : 20
 # According to the number of sst files in rocksdb, 
 # compact every `compact-every-num-of-files` file.
 best-delete-min-ratio : 10
+
+# Bigkeys related
+bigkeys-show-limit : 10
+bigkeys-member-threshold : 10000
+bigkeys-key-value-length-threshold : 1048576
+bigkeys-log-interval : 1

include/pika_admin.h

@@ -267,7 +267,8 @@ class InfoCmd : public Cmd {
     kInfoAll,
     kInfoDebug,
     kInfoCommandStats,
-    kInfoCache
+    kInfoCache,
+    kInfoBigKeys
   };
   InfoCmd(const std::string& name, int arity, uint32_t flag) : Cmd(name, arity, flag) {}
   void Do() override;
@@ -277,6 +278,7 @@ class InfoCmd : public Cmd {
   void Execute() override;
 
  private:
+  int bigkeys_limit_ = 0;
   InfoSection info_section_;
   bool rescan_ = false;  // whether to rescan the keyspace
   bool off_ = false;
@@ -295,15 +297,19 @@ class InfoCmd : public Cmd {
   const static std::string kDebugSection;
   const static std::string kCommandStatsSection;
   const static std::string kCacheSection;
+  const static std::string kBigKeysSection;
 
   void DoInitial() override;
   void Clear() override {
     rescan_ = false;
     off_ = false;
     keyspace_scan_dbs_.clear();
+    info_section_ = kInfoErr;
+    bigkeys_limit_ = 0;
   }
 
   void InfoServer(std::string& info);
+  void InfoBigKeys(std::string& info);
   void InfoClients(std::string& info);
   void InfoStats(std::string& info);
   void InfoExecCount(std::string& info);

include/pika_conf.h

@@ -160,6 +160,23 @@ class PikaConf : public pstd::BaseConf {
     std::shared_lock l(rwlock_);
     return binlog_writer_num_;
   }
+  //big keys
+  int bigkeys_show_limit() {
+    std::shared_lock l(rwlock_);
+    return bigkeys_show_limit_;
+  }
+  int bigkeys_member_threshold() {
+    std::shared_lock l(rwlock_);
+    return bigkeys_member_threshold_;
+  }
+  int bigkeys_key_value_length_threshold() {
+    std::shared_lock l(rwlock_);
+    return bigkeys_key_value_length_threshold_;
+  }
+  int bigkeys_log_interval() {
+    std::shared_lock l(rwlock_);
+    return bigkeys_log_interval_;
+  }
   bool slotmigrate() {
     std::shared_lock l(rwlock_);
     return slotmigrate_;
@@ -729,7 +746,23 @@ class PikaConf : public pstd::BaseConf {
       log_net_activities_.store(false);
     }
   }
-
+  //big keys
+  void SetBigkeysShowLimit(const int value) {
+    std::lock_guard l(rwlock_);
+    bigkeys_show_limit_ = value;
+  }
+  void SetBigkeysKeyValueLengthThreshold(const int value) {
+    std::lock_guard l(rwlock_);
+    bigkeys_key_value_length_threshold_ = value;
+  }
+  void SetBigkeysMemberCountThreshold(const int value) {
+    std::lock_guard l(rwlock_);
+    bigkeys_member_threshold_ = value;
+  }
+  void SetBigkeysLogInterval(const int value) {
+    std::lock_guard l(rwlock_);
+    bigkeys_log_interval_ = value;
+  }
   // Rsync Rate limiting configuration
   void SetThrottleBytesPerSecond(const int value) {
     std::lock_guard l(rwlock_);
@@ -895,6 +928,11 @@ class PikaConf : public pstd::BaseConf {
   int thread_pool_size_ = 0;
   int slow_cmd_thread_pool_size_ = 0;
   int admin_thread_pool_size_ = 0;
+  //big keys
+  int bigkeys_show_limit_ = 10;
+  int bigkeys_member_threshold_ = 10000;
+  int bigkeys_key_value_length_threshold_ = 1048576;
+  int bigkeys_log_interval_ = 60;
   std::unordered_set<std::string> slow_cmd_set_;
   // Because the exporter of Pika_exporter implements Auth authentication
   // with the Exporter of Pika, and the Exporter authenticates the Auth when 

include/pika_db.h

@@ -128,6 +128,9 @@ class DB : public std::enable_shared_from_this<DB>, public pstd::noncopyable {
 
   void SetCompactRangeOptions(const bool is_canceled);
 
+  // Update big keys configuration
+  void UpdateStorageBigKeysConfig(uint32_t log_interval, uint64_t member_threshold, uint64_t key_value_length_threshold, size_t show_limit);
+
   std::shared_ptr<pstd::lock::LockMgr> LockMgr();
   /*
    * Cache used

include/pika_server.h

@@ -145,6 +145,7 @@ class PikaServer : public pstd::noncopyable {
   void DBSetMaxCacheStatisticKeys(uint32_t max_cache_statistic_keys);
   void DBSetSmallCompactionThreshold(uint32_t small_compaction_threshold);
   void DBSetSmallCompactionDurationThreshold(uint32_t small_compaction_duration_threshold);
+  void UpdateDBBigKeysConfig();
   bool GetDBBinlogOffset(const std::string& db_name, BinlogOffset* boffset);
   pstd::Status DoSameThingEveryDB(const TaskType& type);
 

src/net/include/net_define.h

@@ -48,7 +48,8 @@ enum EventStatus {
   kNone = 0,
   kReadable = 0x1,
   kWritable = 0x1 << 1,
-  kErrorEvent = 0x1 << 2,
+  kPeerClose = 0x1 << 2,
+  kErrorEvent = 0x1 << 3,
 };
 
 enum ConnStatus {

src/net/src/net_epoll.cc

@@ -14,6 +14,10 @@
 #include "net/include/net_define.h"
 #include "pstd/include/xdebug.h"
 
+#ifndef EPOLLRDHUP
+#define EPOLLRDHUP 0x2000
+#endif
+
 namespace net {
 
 NetMultiplexer* CreateNetMultiplexer(int limit) { return new NetEpoll(limit); }
@@ -45,7 +49,7 @@ int NetEpoll::NetAddEvent(int fd, int mask) {
   }
   if (mask & kWritable) {
     ee.events |= EPOLLOUT;
-    }
+  }
 
   return epoll_ctl(multiplexer_, EPOLL_CTL_ADD, fd, &ee);
 }
@@ -62,6 +66,10 @@ int NetEpoll::NetModEvent(int fd, int old_mask, int mask) {
   if ((old_mask | mask) & kWritable) {
     ee.events |= EPOLLOUT;
   }
+
+  if ((old_mask | mask) & kPeerClose) {
+    ee.events |= EPOLLRDHUP;
+  }
   return epoll_ctl(multiplexer_, EPOLL_CTL_MOD, fd, &ee);
 }
 
@@ -93,6 +101,10 @@ int NetEpoll::NetPoll(int timeout) {
       ev.mask |= kWritable;
     }
 
+    if (events_[i].events & EPOLLRDHUP) {
+      ev.mask |= kPeerClose;
+    }
+
     if (events_[i].events & (EPOLLERR | EPOLLHUP)) {
       ev.mask |= kErrorEvent;
     }

src/net/src/worker_thread.cc

@@ -189,7 +189,7 @@ void* WorkerThread::ThreadMain() {
           ReadStatus read_status = in_conn->GetRequest();
           in_conn->set_last_interaction(now);
           if (read_status == kReadAll) {
-            net_multiplexer_->NetModEvent(pfe->fd, 0, 0);
+            net_multiplexer_->NetModEvent(pfe->fd, 0, kPeerClose);
             // Wait for the conn complete asynchronous task and
             // Mod Event to kWritable
           } else if (read_status == kReadHalf) {
@@ -199,8 +199,15 @@ void* WorkerThread::ThreadMain() {
           }
         }
 
+        if ((should_close == 0) && ((pfe->mask & kPeerClose) != 0)) {
+          should_close = 1;
+        }
+
         if (((pfe->mask & kErrorEvent) != 0) || (should_close != 0)) {
           net_multiplexer_->NetDelEvent(pfe->fd, 0);
+          // TODO: in_conn may live longer than fd.
+          // eg. in_conn are being transferred to net_pubsub
+          // while peer client closing this connection
           CloseFd(in_conn);
           in_conn = nullptr;
           {

src/pika.cc

@@ -211,6 +211,7 @@ int main(int argc, char* argv[]) {
   g_pika_rm = std::make_unique<PikaReplicaManager>();
   g_network_statistic = std::make_unique<net::NetworkStatistic>();
   g_pika_server->InitDBStruct();
+  g_pika_server->UpdateDBBigKeysConfig();
   //the cmd table of g_pika_cmd_table_manager must be inited before calling PikaServer::InitStatistic(CmdTable* )
   g_pika_server->InitStatistic(g_pika_cmd_table_manager->GetCmdTable());
   auto status = g_pika_server->InitAcl();

src/pika_admin.cc

@@ -884,7 +884,7 @@ const std::string InfoCmd::kRocksDBSection = "rocksdb";
 const std::string InfoCmd::kDebugSection = "debug";
 const std::string InfoCmd::kCommandStatsSection = "commandstats";
 const std::string InfoCmd::kCacheSection = "cache";
-
+const std::string InfoCmd::kBigKeysSection = "bigkeys";
 
 const std::string ClientCmd::KILLTYPE_NORMAL = "normal";
 const std::string ClientCmd::KILLTYPE_PUBSUB = "pubsub";
@@ -910,6 +910,8 @@ void InfoCmd::DoInitial() {
     keyspace_scan_dbs_ = g_pika_server->GetAllDBName();
   } else if (strcasecmp(argv_[1].data(), kServerSection.data()) == 0) {
     info_section_ = kInfoServer;
+  } else if (strcasecmp(argv_[1].data(), "bigkeys") == 0) {
+    info_section_ = kInfoBigKeys;
   } else if (strcasecmp(argv_[1].data(), kClientsSection.data()) == 0) {
     info_section_ = kInfoClients;
   } else if (strcasecmp(argv_[1].data(), kStatsSection.data()) == 0) {
@@ -994,6 +996,9 @@ void InfoCmd::Do() {
       InfoReplication(info);
       info.append("\r\n");
       InfoKeyspace(info);
+      info.append("\r\n");
+      InfoBigKeys(info);
+      info.append("\r\n");
       break;
     case kInfoAll:
       InfoServer(info);
@@ -1017,6 +1022,9 @@ void InfoCmd::Do() {
       InfoKeyspace(info);
       info.append("\r\n");
       InfoRocksDB(info);
+      info.append("\r\n");
+      InfoBigKeys(info);
+      info.append("\r\n");
       break;
     case kInfoServer:
       InfoServer(info);
@@ -1054,6 +1062,9 @@ void InfoCmd::Do() {
     case kInfoCache:
       InfoCache(info, db_);
       break;
+    case kInfoBigKeys:
+      InfoBigKeys(info);
+      break;
     default:
       // kInfoErr is nothing
       break;
@@ -1094,7 +1105,21 @@ void InfoCmd::InfoServer(std::string& info) {
 
   info.append(tmp_stream.str());
 }
-
+void InfoCmd::InfoBigKeys(std::string& info) {
+  std::stringstream tmp_stream;
+  std::shared_lock db_rwl(g_pika_server->dbs_rw_);
+  for (const auto& db_item : g_pika_server->dbs_) {
+    if (!db_item.second) {
+      continue;
+    }
+    std::vector<storage::BigKeyInfo> bigkeys;
+    db_item.second->storage()->GetBigKeyStatistics(&bigkeys);
+    std::string bigkey_info;
+    storage::FormatBigKeyStatistics(bigkeys, &bigkey_info, g_pika_conf->bigkeys_show_limit());
+    tmp_stream << bigkey_info;
+  }
+  info.append(tmp_stream.str());
+}
 void InfoCmd::InfoClients(std::string& info) {
   std::stringstream tmp_stream;
   tmp_stream << "# Clients"
@@ -1833,7 +1858,27 @@ void ConfigCmd::ConfigGet(std::string& ret) {
     EncodeString(&config_body, "max-cache-statistic-keys");
     EncodeNumber(&config_body, g_pika_conf->max_cache_statistic_keys());
   }
-
+  //big keys
+  if (pstd::stringmatch(pattern.data(), "BIGKEYS_SHOW_LIMIT", 1)) {
+    elements += 2;
+    EncodeString(&config_body, "BIGKEYS_SHOW_LIMIT");
+    EncodeNumber(&config_body, g_pika_conf->bigkeys_show_limit());
+  }
+  if (pstd::stringmatch(pattern.data(), "bigkeys_member_threshold", 1)) {
+    elements += 2;
+    EncodeString(&config_body, "bigkeys_member_threshold");
+    EncodeNumber(&config_body, g_pika_conf->bigkeys_member_threshold());
+  }
+  if (pstd::stringmatch(pattern.data(), "bigkeys_key_value_length_threshold", 1)) {
+    elements += 2;
+    EncodeString(&config_body, "bigkeys_key_value_length_threshold");
+    EncodeNumber(&config_body, g_pika_conf->bigkeys_key_value_length_threshold());
+  }
+  if (pstd::stringmatch(pattern.data(), "bigkeys_log_interval", 1)) {
+    elements += 2;
+    EncodeString(&config_body, "bigkeys_log_interval");
+    EncodeNumber(&config_body, g_pika_conf->bigkeys_log_interval());
+  }
   if (pstd::stringmatch(pattern.data(), "small-compaction-threshold", 1) != 0) {
     elements += 2;
     EncodeString(&config_body, "small-compaction-threshold");
@@ -2963,6 +3008,35 @@ void ConfigCmd::ConfigSet(std::shared_ptr<DB> db) {
     }
     g_pika_conf->SetMaxConnRbufSize(static_cast<int>(ival));
     res_.AppendStringRaw("+OK\r\n");
+  //big keys
+  } else if (set_item == "BIGKEYS_SHOW_LIMIT") {
+    if (!pstd::string2int(value.data(), value.size(), &ival) || ival < 0) {
+      res_.AppendStringRaw("-ERR Invalid argument '" + value + "' for CONFIG SET 'BIGKEYS_SHOW_LIMIT'\r\n");
+      return;
+    }
+    g_pika_conf->SetBigkeysShowLimit(ival);
+    res_.AppendStringRaw("+OK\r\n");
+  } else if (set_item == "bigkeys_member_threshold") {
+    if (!pstd::string2int(value.data(), value.size(), &ival) || ival < 0) {
+      res_.AppendStringRaw("-ERR Invalid argument '" + value + "' for CONFIG SET 'bigkeys_member_threshold'\r\n");
+      return;
+    }
+    g_pika_conf->SetBigkeysMemberCountThreshold(ival);
+    res_.AppendStringRaw("+OK\r\n");
+  } else if (set_item == "bigkeys_key_value_length_threshold") {
+    if (!pstd::string2int(value.data(), value.size(), &ival) || ival < 0) {
+      res_.AppendStringRaw("-ERR Invalid argument '" + value + "' for CONFIG SET 'bigkeys_key_value_length_threshold'\r\n");
+      return;
+    }
+    g_pika_conf->SetBigkeysKeyValueLengthThreshold(ival);
+    res_.AppendStringRaw("+OK\r\n");
+  } else if (set_item == "bigkeys_log_interval") {
+    if (!pstd::string2int(value.data(), value.size(), &ival) || ival < 0) {
+      res_.AppendStringRaw("-ERR Invalid argument '" + value + "' for CONFIG SET 'bigkeys_log_interval'\r\n");
+      return;
+    }
+    g_pika_conf->SetBigkeysLogInterval(ival);
+    res_.AppendStringRaw("+OK\r\n");
   } else {
     res_.AppendStringRaw("-ERR Unsupported CONFIG parameter: " + set_item + "\r\n");
   }