feat:cache layer for pika

.github/workflows/codeql.yml

@@ -13,10 +13,10 @@ name: "CodeQL"
 
 on:
   push:
-    branches: [ "unstable", "3.5.1" ]
+    branches: [ "unstable", "3.5.1", "*" ]
   pull_request:
     # The branches below must be a subset of the branches above
-    branches: [ "unstable", "3.5.1" ]
+    branches: [ "unstable", "3.5.1", "*" ]
   schedule:
     - cron: '25 19 * * 6'
 

.github/workflows/codis.yml

@@ -5,9 +5,9 @@ name: Codis
 
 on:
   push:
-    branches: [ "unstable", "3.5.1" ]
+    branches: [ "unstable", "3.5.1", "*" ]
   pull_request:
-    branches: [ "unstable", "3.5.1" ]
+    branches: [ "unstable", "3.5.1", "*" ]
 
 jobs:
 

.github/workflows/operator.yml

@@ -5,9 +5,9 @@ name: Operator
 
 on:
   push:
-    branches: [ "unstable", "3.5.1" ]
+    branches: [ "unstable", "3.5.1", "*" ]
   pull_request:
-    branches: [ "unstable", "3.5.1" ]
+    branches: [ "unstable", "3.5.1", "*" ]
 
 jobs:
 

.github/workflows/pika.yml

@@ -2,9 +2,9 @@ name: Pika
 
 on:
   push:
-    branches: [ "unstable", "3.5.1" ]
+    branches: [ "unstable", "3.5.1", "*" ]
   pull_request:
-    branches: [ "unstable", "3.5.1" ]
+    branches: [ "unstable", "3.5.1", "*" ]
 
 env:
   # Customize the CMake build type here (Release, Debug, RelWithDebInfo, etc.)

.github/workflows/tools_go.yml

@@ -2,11 +2,11 @@ name: Tools_go_build
 
 on:
   push:
-    branches: [ "unstable", "3.5.1" ]
+    branches: [ "unstable", "3.5.1", "*" ]
     paths:
       - 'tools/**'
   pull_request:
-    branches: [ "unstable", "3.5.1" ]
+    branches: [ "unstable", "3.5.1", "*" ]
     paths:
       - 'tools/**'
 

CMakeLists.txt

@@ -52,6 +52,8 @@ elseif(${BUILD_TYPE} STREQUAL RELWITHDEBINFO)
 else()
   set(LIB_BUILD_TYPE RELEASE)
   set(CMAKE_CXX_FLAGS_RELEASE "-O2 -DNDEBUG")
+
+
 endif()
 
 if(CMAKE_SYSTEM_NAME MATCHES "Darwin")
@@ -626,6 +628,35 @@ ExternalProject_Add(rocksdb
   make -j${CPU_CORE}
 )
 
+ExternalProject_Add(rediscache
+  URL
+  https://github.com/pikiwidb/rediscache/archive/refs/tags/v1.0.5.tar.gz
+  URL_HASH
+  MD5=99e4d0bde20811a6058a6aa482c18711
+  DOWNLOAD_NO_PROGRESS
+  1
+  UPDATE_COMMAND
+  ""
+  LOG_BUILD
+  1
+  BUILD_IN_SOURCE
+  1
+  SOURCE_SUBDIR
+  ""
+  CMAKE_ARGS
+  -DCMAKE_INSTALL_PREFIX=${STAGED_INSTALL_PREFIX}
+  -DCMAKE_INSTALL_INCLUDEDIR=${INSTALL_INCLUDEDIR}
+  -DCMAKE_INSTALL_LIBDIR=${INSTALL_LIBDIR}
+  -DCMAKE_BUILD_TYPE=${LIB_BUILD_TYPE}
+  BUILD_ALWAYS
+  1
+  BUILD_COMMAND
+  make -j${CPU_CORE}
+)
+set(REDISCACHE_INCLUDE_DIR ${INSTALL_INCLUDEDIR})
+set(REDISCACHE_LIBRARY ${INSTALL_LIBDIR}/librediscache.a)
+
+
 option(USE_PIKA_TOOLS "compile pika-tools" OFF)
 if (USE_PIKA_TOOLS)
   ExternalProject_Add(hiredis
@@ -697,6 +728,7 @@ set(ROCKSDB_SOURCE_DIR ${CMAKE_CURRENT_SOURCE_DIR}/${EP_BASE_SUFFIX}/Source/rock
 add_subdirectory(src/pstd)
 add_subdirectory(src/net)
 add_subdirectory(src/storage)
+add_subdirectory(src/cache)
 if (USE_PIKA_TOOLS)
   add_subdirectory(tools)
 endif()
@@ -731,7 +763,9 @@ message("pika GIT_DATE = ${PIKA_GIT_DATE}")
 message("pika GIT_TAG = ${PIKA_GIT_TAG}")
 message("pika BUILD_DATE = ${PIKA_BUILD_DATE}")
 
-set(PIKA_BUILD_VERSION_CC ${CMAKE_BINARY_DIR}/pika_build_version.cc)
+set(PIKA_BUILD_VERSION_CC ${CMAKE_BINARY_DIR}/pika_build_version.cc
+        src/pika_cache_load_thread.cc
+        )
 message("PIKA_BUILD_VERSION_CC : " ${PIKA_BUILD_VERSION_CC})
 configure_file(${CMAKE_CURRENT_SOURCE_DIR}/src/build_version.cc.in ${PIKA_BUILD_VERSION_CC} @ONLY)
 
@@ -766,7 +800,10 @@ add_dependencies(${PROJECT_NAME}
   protobuf
   pstd
   net
-  storage)
+  rediscache
+  storage
+  cache
+)
 
 target_include_directories(${PROJECT_NAME}
   PUBLIC ${CMAKE_CURRENT_BINARY_DIR}
@@ -775,6 +812,7 @@ target_include_directories(${PROJECT_NAME}
 )
 
 target_link_libraries(${PROJECT_NAME}
+  cache
   storage
   net
   pstd
@@ -787,6 +825,7 @@ target_link_libraries(${PROJECT_NAME}
   libzstd.a
   liblz4.a
   libz.a
+  librediscache.a
   ${LIBUNWIND_LIBRARY}
   ${JEMALLOC_LIBRARY})
 

conf/pika.conf

@@ -422,3 +422,37 @@ 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 :
 
+###################
+## Cache Settings
+###################
+# cache-num
+cache-num : 16
+
+# cache-model 0:cache_none 1:cache_read
+cache-model : 1
+# cache-type: string, set, zset, list, hash, bit
+cache-type: string, set, zset, list, hash
+# cache-start-direction 0:cache first ${cache-items-per-key} items; -1:cache last ${cache-items-per-key} items
+cache-start-direction : 0
+cache-items-per-key : 512
+
+# cache-maxmemory, unit bytes
+cache-maxmemory : 10737418240
+
+# cache-maxmemory-policy
+# 0: volatile-lru -> Evict using approximated LRU among the keys with an expire set.
+# 1: allkeys-lru -> Evict any key using approximated LRU.
+# 2: volatile-lfu -> Evict using approximated LFU among the keys with an expire set.
+# 3: allkeys-lfu -> Evict any key using approximated LFU.
+# 4: volatile-random -> Remove a random key among the ones with an expire set.
+# 5: allkeys-random -> Remove a random key, any key.
+# 6: volatile-ttl -> Remove the key with the nearest expire time (minor TTL)
+# 7: noeviction -> Don't evict anything, just return an error on write operations.
+cache-maxmemory-policy : 1
+
+# cache-maxmemory-samples
+cache-maxmemory-samples: 5
+
+# cache-lfu-decay-time
+cache-lfu-decay-time: 1
+

include/pika_admin.h

@@ -155,6 +155,8 @@ class FlushallCmd : public Cmd {
  public:
   FlushallCmd(const std::string& name, int arity, uint16_t flag) : Cmd(name, arity, flag) {}
   void Do(std::shared_ptr<Slot> slot = nullptr) override;
+  void DoThroughDB(std::shared_ptr<Slot> slot = nullptr) override;
+  void DoUpdateCache(std::shared_ptr<Slot> slot = nullptr) override;
   void Split(std::shared_ptr<Slot> slot, const HintKeys& hint_keys) override{};
   void Merge() override{};
   Cmd* Clone() override { return new FlushallCmd(*this); }
@@ -169,6 +171,8 @@ class FlushdbCmd : public Cmd {
  public:
   FlushdbCmd(const std::string& name, int arity, uint16_t flag) : Cmd(name, arity, flag) {}
   void Do(std::shared_ptr<Slot> slot = nullptr) override;
+  void DoThroughDB(std::shared_ptr<Slot> slot = nullptr) override;
+  void DoUpdateCache(std::shared_ptr<Slot> slot = nullptr) override;
   void Split(std::shared_ptr<Slot> slot, const HintKeys& hint_keys) override{};
   void Merge() override{};
   Cmd* Clone() override { return new FlushdbCmd(*this); }
@@ -211,7 +215,8 @@ class InfoCmd : public Cmd {
     kInfo,
     kInfoAll,
     kInfoDebug,
-    kInfoCommandStats
+    kInfoCommandStats,
+    kInfoCache
   };
 
   InfoCmd(const std::string& name, int arity, uint16_t flag) : Cmd(name, arity, flag) {}
@@ -239,6 +244,7 @@ class InfoCmd : public Cmd {
   const static std::string kRocksDBSection;
   const static std::string kDebugSection;
   const static std::string kCommandStatsSection;
+  const static std::string kCacheSection;
 
   void DoInitial() override;
   void Clear() override {
@@ -259,6 +265,9 @@ class InfoCmd : public Cmd {
   void InfoRocksDB(std::string& info);
   void InfoDebug(std::string& info);
   void InfoCommandStats(std::string& info);
+  void InfoCache(std::string& info, std::shared_ptr<Slot> slot = nullptr);
+
+  std::string CacheStatusToString(int status);
 };
 
 class ShutdownCmd : public Cmd {
@@ -285,7 +294,7 @@ class ConfigCmd : public Cmd {
   std::vector<std::string> config_args_v_;
   void DoInitial() override;
   void ConfigGet(std::string& ret);
-  void ConfigSet(std::string& ret);
+  void ConfigSet(std::string& ret, std::shared_ptr<Slot> slot);
   void ConfigRewrite(std::string& ret);
   void ConfigResetstat(std::string& ret);
   void ConfigRewriteReplicationID(std::string& ret);
@@ -487,6 +496,25 @@ class DisableWalCmd : public Cmd {
   void DoInitial() override;
 };
 
+class CacheCmd : public Cmd {
+ public:
+  enum CacheCondition {kCLEAR_DB, kCLEAR_HITRATIO, kDEL_KEYS, kRANDOM_KEY};
+  CacheCmd(const std::string& name, int arity, uint16_t flag) : Cmd(name, arity, flag) {}
+  void Do(std::shared_ptr<Slot> slot = nullptr) override;
+  void Split(std::shared_ptr<Slot> slot, const HintKeys& hint_keys) override{};
+  void Merge() override{};
+  Cmd* Clone() override { return new CacheCmd(*this); }
+
+ private:
+  CacheCondition condition_;
+  std::vector<std::string> keys_;
+  rocksdb::Status s_;
+  void DoInitial() override;
+  void Clear() override {
+    keys_.clear();
+  }
+};
+
 #ifdef WITH_COMMAND_DOCS
 class CommandCmd : public Cmd {
  public:

include/pika_bit.h

@@ -24,13 +24,17 @@ class BitGetCmd : public Cmd {
     return res;
   }
   void Do(std::shared_ptr<Slot> slot = nullptr) override;
+  void ReadCache(std::shared_ptr<Slot> slot = nullptr) override;
+  void DoUpdateCache(std::shared_ptr<Slot> slot = nullptr) override;
+  void DoThroughDB(std::shared_ptr<Slot> slot = nullptr) override;
   void Split(std::shared_ptr<Slot> slot, const HintKeys& hint_keys) override {};
   void Merge() override {};
   Cmd* Clone() override { return new BitGetCmd(*this); }
 
  private:
   std::string key_;
   int64_t bit_offset_ = -1;
+  rocksdb::Status s_;
   void Clear() override {
     key_ = "";
     bit_offset_ = -1;
@@ -47,6 +51,8 @@ class BitSetCmd : public Cmd {
     return res;
   }
   void Do(std::shared_ptr<Slot> slot = nullptr) override;
+  void DoUpdateCache(std::shared_ptr<Slot> slot = nullptr) override;
+  void DoThroughDB(std::shared_ptr<Slot> slot = nullptr) override;
   void Split(std::shared_ptr<Slot> slot, const HintKeys& hint_keys) override {};
   void Merge() override {};
   Cmd* Clone() override { return new BitSetCmd(*this); }
@@ -55,6 +61,7 @@ class BitSetCmd : public Cmd {
   std::string key_;
   int64_t bit_offset_;
   int64_t on_;
+  rocksdb::Status s_;
   void Clear() override {
     key_ = "";
     bit_offset_ = -1;
@@ -72,6 +79,9 @@ class BitCountCmd : public Cmd {
     return res;
   }
   void Do(std::shared_ptr<Slot> slot = nullptr) override;
+  void ReadCache(std::shared_ptr<Slot> slot = nullptr) override;
+  void DoUpdateCache(std::shared_ptr<Slot> slot = nullptr) override;
+  void DoThroughDB(std::shared_ptr<Slot> slot = nullptr) override;
   void Split(std::shared_ptr<Slot> slot, const HintKeys& hint_keys) override {};
   void Merge() override {};
   Cmd* Clone() override { return new BitCountCmd(*this); }
@@ -81,6 +91,7 @@ class BitCountCmd : public Cmd {
   bool count_all_;
   int64_t start_offset_;
   int64_t end_offset_;
+  rocksdb::Status s_;
   void Clear() override {
     key_ = "";
     count_all_ = false;
@@ -99,6 +110,9 @@ class BitPosCmd : public Cmd {
     return res;
   }
   void Do(std::shared_ptr<Slot> slot = nullptr) override;
+  void ReadCache(std::shared_ptr<Slot> slot = nullptr) override;
+  void DoUpdateCache(std::shared_ptr<Slot> slot = nullptr) override;
+  void DoThroughDB(std::shared_ptr<Slot> slot = nullptr) override;
   void Split(std::shared_ptr<Slot> slot, const HintKeys& hint_keys) override {};
   void Merge() override {};
   Cmd* Clone() override { return new BitPosCmd(*this); }
@@ -110,6 +124,7 @@ class BitPosCmd : public Cmd {
   int64_t bit_val_;
   int64_t start_offset_;
   int64_t end_offset_;
+  rocksdb::Status s_;
   void Clear() override {
     key_ = "";
     pos_all_ = false;
@@ -139,13 +154,16 @@ class BitOpCmd : public Cmd {
     return {dest_key_};
   }
   void Do(std::shared_ptr<Slot> slot = nullptr) override;
+  void DoUpdateCache(std::shared_ptr<Slot> slot = nullptr) override;
+  void DoThroughDB(std::shared_ptr<Slot> slot = nullptr) override;
   void Split(std::shared_ptr<Slot> slot, const HintKeys& hint_keys) override{};
   void Merge() override{};
   Cmd* Clone() override { return new BitOpCmd(*this); }
   void DoBinlog(const std::shared_ptr<SyncMasterSlot>& slot) override;
 
  private:
   std::string dest_key_;
+  rocksdb::Status s_;
   std::vector<std::string> src_keys_;
   storage::BitOpType op_;
   void Clear() override {