feat: dont store cmd(Get/Set/Hget/Hset) with too large key in cache (#…

…2849)

* feat: dont store cmd(Get/Set/Hget/Hset) with too large key in cache

* format code

include/pika_command.h

@@ -529,6 +529,7 @@ class Cmd : public std::enable_shared_from_this<Cmd> {
   // used for execute multikey command into different slots
   virtual void Split(const HintKeys& hint_keys) = 0;
   virtual void Merge() = 0;
+  virtual bool IsTooLargeKey(const int &max_sz) { return false; }
 
   int8_t SubCmdIndex(const std::string& cmdName);  // if the command no subCommand，return -1；
 

include/pika_conf.h

@@ -427,6 +427,7 @@ class PikaConf : public pstd::BaseConf {
   void SetCacheMode(const int value) { cache_mode_ = value; }
   void SetCacheStartDirection(const int value) { zset_cache_start_direction_ = value; }
   void SetCacheItemsPerKey(const int value) { zset_cache_field_num_per_key_ = value; }
+  void SetCacheMaxKeySize(const int value) { max_key_size_in_cache_ = value; }
   void SetCacheMaxmemory(const int64_t value) { cache_maxmemory_ = value; }
   void SetCacheMaxmemoryPolicy(const int value) { cache_maxmemory_policy_ = value; }
   void SetCacheMaxmemorySamples(const int value) { cache_maxmemory_samples_ = value; }
@@ -897,6 +898,7 @@ class PikaConf : public pstd::BaseConf {
   void SetCacheDisableFlag() { tmp_cache_disable_flag_ = true; }
   int zset_cache_start_direction() { return zset_cache_start_direction_; }
   int zset_cache_field_num_per_key() { return zset_cache_field_num_per_key_; }
+  int max_key_size_in_cache() { return max_key_size_in_cache_; }
   int cache_maxmemory_policy() { return cache_maxmemory_policy_; }
   int cache_maxmemory_samples() { return cache_maxmemory_samples_; }
   int cache_lfu_decay_time() { return cache_lfu_decay_time_; }
@@ -1049,6 +1051,7 @@ class PikaConf : public pstd::BaseConf {
   std::atomic_int cache_bit_ = 1;
   std::atomic_int zset_cache_start_direction_ = 0;
   std::atomic_int zset_cache_field_num_per_key_ = 512;
+  std::atomic_int max_key_size_in_cache_ = 512;
   std::atomic_int cache_maxmemory_policy_ = 1;
   std::atomic_int cache_maxmemory_samples_ = 5;
   std::atomic_int cache_lfu_decay_time_ = 1;

include/pika_hash.h

@@ -54,6 +54,7 @@ class HGetCmd : public Cmd {
   void DoUpdateCache() override;
   void Split(const HintKeys& hint_keys) override {};
   void Merge() override {};
+  bool IsTooLargeKey(const int &max_sz) override { return key_.size() > static_cast<uint32_t>(max_sz); }
   Cmd* Clone() override { return new HGetCmd(*this); }
 
  private:

include/pika_kv.h

@@ -29,6 +29,7 @@ class SetCmd : public Cmd {
   void DoThroughDB() override;
   void Split(const HintKeys& hint_keys) override{};
   void Merge() override{};
+  bool IsTooLargeKey(const int& max_sz) override { return key_.size() > static_cast<uint32_t>(max_sz); }
   Cmd* Clone() override { return new SetCmd(*this); }
 
  private:
@@ -64,6 +65,7 @@ class GetCmd : public Cmd {
   void ReadCache() override;
   void Split(const HintKeys& hint_keys) override{};
   void Merge() override{};
+  bool IsTooLargeKey(const int &max_sz) override { return key_.size() > static_cast<uint32_t>(max_sz); }
   Cmd* Clone() override { return new GetCmd(*this); }
 
  private:

src/cache/include/config.h

@@ -38,6 +38,7 @@ constexpr int CACHE_START_FROM_END = -1;
  * cache items per key
  */
 #define DEFAULT_CACHE_ITEMS_PER_KEY 512
+#define DEFAULT_CACHE_MAX_KEY_SIZE 512
 
 struct CacheConfig {
   uint64_t maxmemory;                      /* Can used max memory */

src/pika_client_conn.cc

@@ -352,6 +352,12 @@ bool PikaClientConn::ReadCmdInCache(const net::RedisCmdArgsType& argv, const std
   }
   // Initial
   c_ptr->Initial(argv, current_db_);
+  // dont store cmd with too large key(only Get/HGet cmd can reach here)
+  // the cmd with large key should be non-exist in cache, except for pre-stored
+  if (c_ptr->IsTooLargeKey(g_pika_conf->max_key_size_in_cache())) {
+    resp_num--;
+    return false;
+  }
   //acl check
   int8_t subCmdIndex = -1;
   std::string errKey;

src/pika_conf.cc

@@ -640,6 +640,13 @@ int PikaConf::Load() {
   }
   zset_cache_field_num_per_key_ = zset_cache_field_num_per_key;
 
+  int max_key_size_in_cache = DEFAULT_CACHE_MAX_KEY_SIZE;
+  GetConfInt("max-key-size-in-cache", &max_key_size_in_cache);
+  if (max_key_size_in_cache <= 0) {
+    max_key_size_in_cache = DEFAULT_CACHE_MAX_KEY_SIZE;
+  }
+  max_key_size_in_cache_ = max_key_size_in_cache;
+
   int64_t cache_maxmemory = PIKA_CACHE_SIZE_DEFAULT;
   GetConfInt64("cache-maxmemory", &cache_maxmemory);
   cache_maxmemory_ = (PIKA_CACHE_SIZE_MIN > cache_maxmemory) ? PIKA_CACHE_SIZE_DEFAULT : cache_maxmemory;

src/pika_hash.cc

@@ -75,6 +75,10 @@ void HSetCmd::DoThroughDB() {
 }
 
 void HSetCmd::DoUpdateCache() {
+  // HSetIfKeyExist() can void storing large key, but IsTooLargeKey() can speed up it
+  if (IsTooLargeKey(g_pika_conf->max_key_size_in_cache())) {
+    return;
+  }
   if (s_.ok()) {
     db_->cache()->HSetIfKeyExist(key_, field_, value_);
   }
@@ -123,6 +127,9 @@ void HGetCmd::DoThroughDB() {
 }
 
 void HGetCmd::DoUpdateCache() {
+  if (IsTooLargeKey(g_pika_conf->max_key_size_in_cache())) {
+    return;
+  }
   if (s_.ok()) {
     db_->cache()->PushKeyToAsyncLoadQueue(PIKA_KEY_TYPE_HASH, key_, db_);
   }

src/pika_kv.cc

@@ -106,7 +106,7 @@ void SetCmd::DoThroughDB() {
 }
 
 void SetCmd::DoUpdateCache() {
-  if (SetCmd::kNX == condition_) {
+  if (SetCmd::kNX == condition_ || IsTooLargeKey(g_pika_conf->max_key_size_in_cache())) {
     return;
   }
   if (s_.ok()) {
@@ -187,6 +187,9 @@ void GetCmd::DoThroughDB() {
 }
 
 void GetCmd::DoUpdateCache() {
+  if (IsTooLargeKey(g_pika_conf->max_key_size_in_cache())) {
+    return;
+  }
   if (s_.ok()) {
     db_->cache()->WriteKVToCache(key_, value_, ttl_millsec_ / 1000);
   }