feat: Feature/floyd rebase unstable (OpenAtomFoundation#2413)

* feat: add floyd (OpenAtomFoundation#2347)

* floyd: refactor storage to support multi rocksdb instance && todis key format

---------

Co-authored-by: wangshaoyi <wangshaoyi@360.cn>

* fix floyd keys (OpenAtomFoundation#2364)

Co-authored-by: wuxianrong <wuxianrong@360.cn>

* floyd rebase unstable branch

* fix by review comments

---------

Co-authored-by: wangshaoyi <wangshaoyi@360.cn>
Co-authored-by: Mixficsol <838844609@qq.com>
Co-authored-by: wuxianrong <wuxianrong@360.cn>

Author: 4 people

CMakeLists.txt

@@ -52,7 +52,6 @@ elseif(${BUILD_TYPE} STREQUAL RELWITHDEBINFO)
 else()
   set(LIB_BUILD_TYPE RELEASE)
   set(CMAKE_CXX_FLAGS_RELEASE "-O2 -g -DNDEBUG")
-
 endif()
 
 if(CMAKE_SYSTEM_NAME MATCHES "Darwin")
@@ -172,7 +171,6 @@ set(GTEST_INCLUDE_DIR ${INSTALL_INCLUDEDIR})
 set(GTEST_MAIN_INCLUDE_DIR ${INSTALL_INCLUDEDIR})
 set(GMOCK_INCLUDE_DIR ${INSTALL_INCLUDEDIR})
 
-
 ExternalProject_Add(gflags
   URL
   https://github.com/gflags/gflags/archive/refs/tags/v2.2.2.tar.gz

conf/pika.conf

@@ -7,6 +7,10 @@
 # Port 10221 is used for Rsync, and port 11221 is used for Replication, while the listening port is 9221.
 port : 9221
 
+db-instance-num : 3
+rocksdb-ttl-second : 86400 * 7;
+rocksdb-periodic-second : 86400 * 3;
+
 # Random value identifying the Pika server, its string length must be 40.
 # If not set, Pika will generate a random string with a length of 40 random characters.
 # run-id :

include/pika_admin.h

@@ -536,7 +536,7 @@ class DiskRecoveryCmd : public Cmd {
 
  private:
   void DoInitial() override;
-  std::map<std::string, uint64_t> background_errors_;
+  std::map<int, uint64_t> background_errors_;
 };
 
 class ClearReplicationIDCmd : public Cmd {

include/pika_binlog.h

@@ -12,10 +12,8 @@
 #include "pstd/include/pstd_mutex.h"
 #include "pstd/include/pstd_status.h"
 #include "pstd/include/noncopyable.h"
-
 #include "include/pika_define.h"
 
-
 std::string NewFileName(const std::string& name, uint32_t current);
 
 class Version final : public pstd::noncopyable {

include/pika_cache.h

@@ -223,4 +223,4 @@ class PikaCache : public pstd::noncopyable, public std::enable_shared_from_this<
   std::vector<std::shared_ptr<pstd::Mutex>> cache_mutexs_;
 };
 
-#endif
+#endif

include/pika_cache_load_thread.h

@@ -38,7 +38,7 @@ class PikaCacheLoadThread : public net::Thread {
  private:
   std::atomic_bool should_exit_;
   std::deque<std::tuple<const char, std::string, const std::shared_ptr<DB>>> loadkeys_queue_;
-  
+
   pstd::CondVar loadkeys_cond_;
   pstd::Mutex loadkeys_mutex_;
 

include/pika_conf.h

@@ -11,13 +11,14 @@
 #include <set>
 #include <unordered_set>
 
+#include "rocksdb/compression_type.h"
+
 #include "pstd/include/base_conf.h"
 #include "pstd/include/pstd_mutex.h"
 #include "pstd/include/pstd_string.h"
 
 #include "acl.h"
 #include "include/pika_define.h"
-#include "include/pika_meta.h"
 #include "rocksdb/compression_type.h"
 
 #define kBinlogReadWinDefaultSize 9000
@@ -76,6 +77,15 @@ class PikaConf : public pstd::BaseConf {
     std::shared_lock l(rwlock_);
     return db_path_;
   }
+  int db_instance_num() {
+    return db_instance_num_;
+  }
+  uint64_t rocksdb_ttl_second() {
+    return rocksdb_ttl_second_.load();
+  }
+  uint64_t rocksdb_periodic_compaction_second() {
+    return rocksdb_periodic_second_.load();
+  }
   std::string db_sync_path() {
     std::shared_lock l(rwlock_);
     return db_sync_path_;
@@ -376,7 +386,6 @@ class PikaConf : public pstd::BaseConf {
   bool daemonize() { return daemonize_; }
   std::string pidfile() { return pidfile_; }
   int binlog_file_size() { return binlog_file_size_; }
-  PikaMeta* local_meta() { return local_meta_.get(); }
   std::vector<rocksdb::CompressionType> compression_per_level();
   std::string compression_all_levels() const { return compression_per_level_; };
   static rocksdb::CompressionType GetCompression(const std::string& value);
@@ -416,6 +425,15 @@ class PikaConf : public pstd::BaseConf {
     TryPushDiffCommands("slaveof", value);
     slaveof_ = value;
   }
+
+  void SetRocksdbTTLSecond(uint64_t ttl) {
+    rocksdb_ttl_second_.store(ttl);
+  }
+
+  void SetRocksdbPeriodicSecond(uint64_t value) {
+    rocksdb_periodic_second_.store(value);
+  }
+
   void SetReplicationID(const std::string& value) {
     std::lock_guard l(rwlock_);
     TryPushDiffCommands("replication-id", value);
@@ -655,6 +673,7 @@ class PikaConf : public pstd::BaseConf {
   int ConfigRewriteReplicationID();
 
  private:
+  // TODO: replace mutex with atomic value
   int port_ = 0;
   int slave_priority_ = 0;
   int thread_num_ = 0;
@@ -668,6 +687,7 @@ class PikaConf : public pstd::BaseConf {
   std::string log_path_;
   std::string log_level_;
   std::string db_path_;
+  int db_instance_num_ = 0;
   std::string db_sync_path_;
   std::string compact_cron_;
   std::string compact_interval_;
@@ -719,6 +739,8 @@ class PikaConf : public pstd::BaseConf {
   int max_background_compactions_ = 0;
   int max_background_jobs_ = 0;
   int max_cache_files_ = 0;
+  std::atomic<uint64_t> rocksdb_ttl_second_ = 0;
+  std::atomic<uint64_t> rocksdb_periodic_second_ = 0;
   int max_bytes_for_level_multiplier_ = 0;
   int64_t block_size_ = 0;
   int64_t block_cache_ = 0;
@@ -787,7 +809,6 @@ class PikaConf : public pstd::BaseConf {
   int64_t blob_file_size_ = 256 * 1024 * 1024;  // 256M
   std::string blob_compression_type_ = "none";
 
-  std::unique_ptr<PikaMeta> local_meta_;
   std::shared_mutex rwlock_;
 
   // Rsync Rate limiting configuration

include/pika_consensus.h

@@ -7,12 +7,12 @@
 
 #include <utility>
 
+#include "include/pika_define.h"
+#include "pstd/include/env.h"
 #include "include/pika_binlog_transverter.h"
 #include "include/pika_client_conn.h"
-#include "include/pika_define.h"
 #include "include/pika_slave_node.h"
 #include "include/pika_stable_log.h"
-#include "pstd/include/env.h"
 
 class Context : public pstd::noncopyable {
  public:

include/pika_define.h

@@ -43,16 +43,16 @@ const std::string kDefaultRsyncAuth = "default";
 const int kMaxRsyncParallelNum = 4;
 
 struct DBStruct {
-  DBStruct(std::string tn)
-      : db_name(std::move(tn)) {}
+  DBStruct(std::string tn, int32_t inst_num)
+      : db_name(std::move(tn)), db_instance_num(inst_num) {}
 
   bool operator==(const DBStruct& db_struct) const {
-    return db_name == db_struct.db_name;
+    return db_name == db_struct.db_name && db_instance_num == db_struct.db_instance_num;
   }
   std::string db_name;
+  int32_t db_instance_num = 0;
 };
 
-// slave item
 struct SlaveItem {
   std::string ip_port;
   std::string ip;