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,9 @@ 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;
@@ -295,6 +297,7 @@ 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 {
@@ -304,6 +307,7 @@ class InfoCmd : public Cmd {
   }
 
   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(int 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_define.h

@@ -12,6 +12,13 @@
 
 #include "net/include/redis_cli.h"
 
+/*
+ * Big keys related
+ */
+#define BIGKEYS_SHOW_LIMIT 10
+#define BIGKEYS_MEMBER_THRESHOLD 10000
+#define BIGKEYS_KEY_VALUE_LENGTH_THRESHOLD 1048576
+#define BIGKEYS_LOG_INTERVAL 1
 /*
  * TTL type
  */

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();