fix: The number of element updates in RedisCache is configurable (OpenAtomFoundation#3043)

* The number of element updates in RedisCache is configurable

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

Author: Mixficsol

conf/pika.conf

@@ -570,6 +570,12 @@ cache-model : 1
 # cache-type: string, set, zset, list, hash, bit
 cache-type: string, set, zset, list, hash, bit
 
+# Set the maximum number of elements in the cache of the Set, list, Zset data types
+cache-value-item-max-size: 1024
+
+# Sets the maximum number of bytes for Key when the String data type is updated in the cache
+max-key-size-in-cache: 1048576
+
 # Maximum number of keys in the zset redis cache
 # On the disk DB, a zset field may have many fields. In the memory cache, we limit the maximum
 # number of keys that can exist in a zset,  which is zset-zset-cache-field-num-per-key, with a

include/pika_command.h

@@ -539,7 +539,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; }
+  virtual bool IsTooLargeKey(const size_t &max_sz) { return false; }
 
   int8_t SubCmdIndex(const std::string& cmdName);  // if the command no subCommand，return -1；
 

include/pika_conf.h

@@ -959,6 +959,7 @@ class PikaConf : public pstd::BaseConf {
   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 value_item_max_size_in_cache() { return cache_value_item_max_size_; }
   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_; }
@@ -1120,7 +1121,8 @@ 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_value_item_max_size_ = 1024;
+  std::atomic_int max_key_size_in_cache_ = 1024 * 1024;
   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,7 +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); }
+  bool IsTooLargeKey(const size_t &max_sz) override { return key_.size() > max_sz; }
   Cmd* Clone() override { return new HGetCmd(*this); }
 
  private:

include/pika_kv.h

@@ -29,7 +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); }
+  bool IsTooLargeKey(const size_t &max_sz) override { return key_.size() > max_sz; }
   Cmd* Clone() override { return new SetCmd(*this); }
 
  private:
@@ -65,7 +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); }
+  bool IsTooLargeKey(const size_t &max_sz) override { return key_.size() > max_sz; }
   Cmd* Clone() override { return new GetCmd(*this); }
 
  private:

src/cache/include/config.h

@@ -40,13 +40,21 @@ constexpr int CACHE_START_FROM_END = -1;
 #define DEFAULT_CACHE_ITEMS_PER_KEY 512
 #define DEFAULT_CACHE_MAX_KEY_SIZE 512
 
+/*
+ * cache value item default size
+ */
+#define DEFAULT_CACHE_ITEMS_SIZE 1024
+#define MAX_CACHE_ITEMS_SIZE 2048
+
 struct CacheConfig {
   uint64_t maxmemory;                      /* Can used max memory */
   int32_t  maxmemory_policy;               /* Policy for key eviction */
   int32_t  maxmemory_samples;              /* Precision of random sampling */
   int32_t  lfu_decay_time;                 /* LFU counter decay factor. */
   int32_t  zset_cache_start_direction;
   int32_t  zset_cache_field_num_per_key;
+  int32_t  cache_value_item_max_size;
+
 
   CacheConfig()
     : maxmemory(CACHE_DEFAULT_MAXMEMORY)

src/pika_cache_load_thread.cc

@@ -10,6 +10,7 @@
 #include "pstd/include/scope_record_lock.h"
 
 extern PikaServer* g_pika_server;
+extern std::unique_ptr<PikaConf> g_pika_conf;
 
 PikaCacheLoadThread::PikaCacheLoadThread(int zset_cache_start_direction, int zset_cache_field_num_per_key)
     : should_exit_(false)
@@ -70,7 +71,7 @@ bool PikaCacheLoadThread::LoadHash(std::string& key, const std::shared_ptr<DB>&
   db->storage()->HLen(key, &len);
   // If the Hash type contains more than 2048 data members, 
   // it will not be updated to RedisCache
-  if (0 >= len || CACHE_VALUE_ITEM_MAX_SIZE < len) {
+  if (0 >= len || g_pika_conf->value_item_max_size_in_cache() < len) {
     return false;
   }
 
@@ -90,9 +91,9 @@ bool PikaCacheLoadThread::LoadList(std::string& key, const std::shared_ptr<DB>&
   db->storage()->LLen(key, &len);
   // If the List type contains more than 2048 data members, 
   // it will not be updated to RedisCache
-  if (len <= 0 || CACHE_VALUE_ITEM_MAX_SIZE < len) {
+  if (len <= 0 || g_pika_conf->value_item_max_size_in_cache() < len) {
     LOG(WARNING) << "can not load key, because item size:" << len
-                 << " beyond max item size:" << CACHE_VALUE_ITEM_MAX_SIZE;
+                 << " beyond max item size:" << g_pika_conf->value_item_max_size_in_cache();
     return false;
   }
 
@@ -112,9 +113,9 @@ bool PikaCacheLoadThread::LoadSet(std::string& key, const std::shared_ptr<DB>& d
   db->storage()->SCard(key, &len);
   // If the Set type contains more than 2048 data members, 
   // it will not be updated to RedisCache
-  if (0 >= len || CACHE_VALUE_ITEM_MAX_SIZE < len) {
+  if (0 >= len || g_pika_conf->value_item_max_size_in_cache() < len) {
     LOG(WARNING) << "can not load key, because item size:" << len
-                 << " beyond max item size:" << CACHE_VALUE_ITEM_MAX_SIZE;
+                 << " beyond max item size:" << g_pika_conf->value_item_max_size_in_cache();
     return false;
   }
 

src/pika_conf.cc

@@ -683,6 +683,15 @@ int PikaConf::Load() {
   }
   zset_cache_field_num_per_key_ = zset_cache_field_num_per_key;
 
+  int cache_value_item_max_size = DEFAULT_CACHE_ITEMS_SIZE;
+  GetConfInt("cache-value-item-max-size", &cache_value_item_max_size);
+  if (cache_value_item_max_size <= 0) {
+    cache_value_item_max_size = DEFAULT_CACHE_ITEMS_SIZE;
+  } else if (cache_value_item_max_size > MAX_CACHE_ITEMS_SIZE) {
+    cache_value_item_max_size = MAX_CACHE_ITEMS_SIZE;
+  }
+  cache_value_item_max_size_ = cache_value_item_max_size;
+
   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) {