feat: overload the PosixFileSystemAdaptor interface to generate real snapshot when follower installs snapshot

CMakeLists.txt

@@ -146,6 +146,9 @@ INCLUDE(braft)
 
 ENABLE_TESTING()
 
+SET(PROTO_OUTPUT_DIR "${CMAKE_BINARY_DIR}/generated_pb")
+FILE(MAKE_DIRECTORY "${PROTO_OUTPUT_DIR}" PARENT)
+
 ADD_SUBDIRECTORY(src/pstd)
 ADD_SUBDIRECTORY(src/net)
 ADD_SUBDIRECTORY(src/praft)

cmake/protobuf.cmake

@@ -132,6 +132,7 @@ FUNCTION(build_protobuf TARGET_NAME)
  UPDATE_COMMAND ""
  DEPENDS zlib
  URL "https:/protocolbuffers/protobuf/archive/v3.18.0.tar.gz"
+ URL_HASH SHA256=14e8042b5da37652c92ef6a2759e7d2979d295f60afd7767825e3de68c856c54
  CONFIGURE_COMMAND mv ../config.sh . COMMAND sh config.sh
  CMAKE_CACHE_ARGS
  -DCMAKE_INSTALL_PREFIX:PATH=${PROTOBUF_INSTALL_DIR}

pikiwidb.conf

@@ -347,7 +347,7 @@ backendpath dump
 # the frequency of dump to backend per second
 backendhz 10
 # the rocksdb number per db
-db-instance-num 5
+db-instance-num 1
 # default 86400 * 7
 rocksdb-ttl-second 604800
 # default 86400 * 3

save_load.sh

@@ -0,0 +1,16 @@
+#!/bin/bash
+killall -9 pikiwidb
+mkdir leader follower1
+
+cd leader && ulimit -n 99999 && rm -fr * && ../bin/pikiwidb ../pikiwidb.conf --port 7777 &
+cd follower1 && ulimit -n 99999 && rm -fr * && ../bin/pikiwidb ../pikiwidb.conf --port 8888 &
+sleep 5
+
+redis-cli -p 7777 raft.cluster init
+
+redis-benchmark -p 7777 -c 5 -n 10000 -r 10000 -d 1024 -t hset
+redis-cli -p 7777 raft.node DSS
+redis-benchmark -p 7777 -c 5 -n 10000 -r 10000 -d 1024 -t hset
+redis-cli -p 7777 raft.node DSS
+
+redis-cli -p 8888 raft.cluster join 127.0.0.1:7777

src/checkpoint_manager.cpp

@@ -23,15 +23,6 @@ void CheckpointManager::Init(int instNum, DB* db) {
 
 void CheckpointManager::CreateCheckpoint(const std::string& path) {
  res_.clear();
-
- if (!pstd::FileExists(path)) {
- if (0 != pstd::CreatePath(path)) {
- WARN("Create Dir {} fail!", path);
- return;
- }
- INFO("Create Dir {} success!", path);
- }
-
  std::lock_guard Lock(shared_mutex_);
  for (int i = 0; i < checkpoint_num_; ++i) {
  checkpoint_infoes_[i].checkpoint_in_process = true;

src/client.cc

@@ -12,10 +12,10 @@
 #include "config.h"
 #include "log.h"
 #include "pikiwidb.h"
+#include "praft.h"
 #include "pstd_string.h"
 #include "slow_log.h"
 #include "store.h"
-#include "praft.h"
 
 namespace pikiwidb {
 

src/client.h

@@ -13,10 +13,10 @@
 #include <unordered_set>
 
 #include "common.h"
+#include "net/tcp_connection.h"
 #include "proto_parser.h"
 #include "replication.h"
 #include "storage/storage.h"
-#include "tcp_connection.h"
 
 namespace pikiwidb {
 
@@ -250,4 +250,4 @@ class PClient : public std::enable_shared_from_this<PClient>, public CmdRes {
 
  static thread_local PClient* s_current;
 };
-} // namespace pikiwidb
+} // namespace pikiwidb

src/cmd_admin.cc

@@ -6,9 +6,9 @@
  */
 
 #include "cmd_admin.h"
-#include "store.h"
 #include "braft/raft.h"
 #include "praft.h"
+#include "store.h"
 
 namespace pikiwidb {
 
@@ -80,7 +80,7 @@ void SelectCmd::DoCmd(PClient* client) {
  client->SetRes(CmdRes::kOK);
 }
 
-InfoCmd::InfoCmd(const std::string& name, int16_t arity) 
+InfoCmd::InfoCmd(const std::string& name, int16_t arity)
  : BaseCmd(name, arity, kCmdFlagsAdmin | kCmdFlagsReadonly, kAclCategoryAdmin) {}
 
 bool InfoCmd::DoInitial(PClient* client) { return true; }
@@ -127,7 +127,7 @@ void InfoCmd::DoCmd(PClient* client) {
  message += "raft_state:up\r\n";
  } else {
  message += "raft_state:down\r\n";
- } 
+ }
  message += "raft_role:" + std::string(braft::state2str(node_status.state)) + "\r\n";
  // message += "raft_is_voting:" + node_status.is_voting + "\r\n";
  message += "raft_leader_id:" + node_status.leader_id.to_string() + "\r\n";
@@ -141,9 +141,10 @@ void InfoCmd::DoCmd(PClient* client) {
  if (!status.ok()) {
  return client->SetRes(CmdRes::kErrOther, status.error_str());
  }
- 
+
  for (int i = 0; i < peers.size(); i++) {
- message += "raft_node" + std::to_string(i) + ":addr=" + butil::ip2str(peers[i].addr.ip).c_str() + ",port=" + std::to_string(peers[i].addr.port) + "\r\n";
+ message += "raft_node" + std::to_string(i) + ":addr=" + butil::ip2str(peers[i].addr.ip).c_str() +
+ ",port=" + std::to_string(peers[i].addr.port) + "\r\n";
  }
  }
 

src/cmd_admin.h

@@ -85,14 +85,14 @@ class SelectCmd : public BaseCmd {
 };
 
 class InfoCmd : public BaseCmd {
-  public:
-  InfoCmd(const std::string& name, int16_t arity);
+ public:
+ InfoCmd(const std::string& name, int16_t arity);
 
-  protected:
-  bool DoInitial(PClient* client) override;
+ protected:
+ bool DoInitial(PClient* client) override;
 
-  private:
-  void DoCmd(PClient* client) override;
+ private:
+ void DoCmd(PClient* client) override;
 };
 
 } // namespace pikiwidb

src/cmd_raft.cc

@@ -36,6 +36,8 @@ void RaftNodeCmd::DoCmd(PClient* client) {
  DoCmdAdd(client);
  } else if (!strcasecmp(cmd.c_str(), "REMOVE")) {
  DoCmdRemove(client);
+ } else if (!strcasecmp(cmd.c_str(), "DSS")) {
+ DoCmdSnapshot(client);
  } else {
  client->SetRes(CmdRes::kErrOther, "RAFT.NODE supports ADD / REMOVE only");
  }
@@ -70,6 +72,13 @@ void RaftNodeCmd::DoCmdRemove(PClient* client) {
  }
 }
 
+void RaftNodeCmd::DoCmdSnapshot(PClient* client) {
+ auto s = PRAFT.DoSnapshot();
+ if (s.ok()) {
+ client->SetRes(CmdRes::kOK);
+ }
+}
+
 RaftClusterCmd::RaftClusterCmd(const std::string& name, int16_t arity)
  : BaseCmd(name, arity, kCmdFlagsRaft, kAclCategoryRaft) {}
 

src/cmd_raft.h

@@ -25,7 +25,7 @@ namespace pikiwidb {
  * :<new node id>
  * :<dbid>
  *
- * RAFT.NODE REMOVE [id] 
+ * RAFT.NODE REMOVE [id]
  * Remove an existing node from the cluster.
  * Reply:
  * -NOCLUSTER ||
@@ -45,6 +45,7 @@ class RaftNodeCmd : public BaseCmd {
  void DoCmd(PClient *client) override;
  void DoCmdAdd(PClient *client);
  void DoCmdRemove(PClient *client);
+ void DoCmdSnapshot(PClient *client);
 
  static constexpr std::string_view kAddCmd = "ADD";
  static constexpr std::string_view kRemoveCmd = "REMOVE";

src/cmd_table_manager.cc

@@ -12,8 +12,8 @@
 #include "cmd_keys.h"
 #include "cmd_kv.h"
 #include "cmd_list.h"
-#include "cmd_set.h"
 #include "cmd_raft.h"
+#include "cmd_set.h"
 #include "cmd_table_manager.h"
 #include "cmd_zset.h"
 
@@ -46,7 +46,7 @@ void CmdTableManager::InitCmdTable() {
 
  // info
  ADD_COMMAND(Info, -1);
- 
+
  // raft
  ADD_COMMAND(RaftCluster, -1);
  ADD_COMMAND(RaftNode, -2);

src/db.cpp

@@ -30,6 +30,8 @@ DB::DB(int db_index, const std::string& db_path)
  ERROR("Storage open failed! {}", s.ToString());
  abort();
  }
+ checkpoint_manager_ = std::make_unique<CheckpointManager>();
+ checkpoint_manager_->Init(g_config.db_instance_num, this);
  opened_ = true;
  INFO("Open DB{} success!", db_index_);
 }
@@ -45,8 +47,14 @@ void DB::DoBgSave(CheckpointInfo& checkpoint_info, const std::string& path, int
  checkpoint_info.checkpoint_in_process = false;
 }
 
-void DB::CreateCheckpoint(const std::string& path) { checkpoint_manager_->CreateCheckpoint(path); }
+void DB::CreateCheckpoint(const std::string& path) {
+ if (0 != pstd::CreatePath(path + '/' + std::to_string(db_index_))) {
+ WARN("Create dir {} fail !", path + '/' + std::to_string(db_index_));
+ return;
+ }
+ checkpoint_manager_->CreateCheckpoint(path);
+}
 
 void DB::WaitForCheckpointDone() { checkpoint_manager_->WaitForCheckpointDone(); }
- 
+
 } // namespace pikiwidb

src/db.h

@@ -11,7 +11,7 @@
 #include <string>
 
 #include "checkpoint_manager.h"
-#include "log.h"
+#include "pstd/log.h"
 #include "pstd/noncopyable.h"
 #include "storage/storage.h"
 namespace pikiwidb {
@@ -55,7 +55,6 @@ class DB {
  bool opened_ = false;
 
  std::unique_ptr<CheckpointManager> checkpoint_manager_;
-
 };
 } // namespace pikiwidb
 

src/pikiwidb.h

@@ -7,9 +7,8 @@
 
 #include "cmd_table_manager.h"
 #include "common.h"
-#include "event_loop.h"
 #include "io_thread_pool.h"
-#include "tcp_connection.h"
+#include "net/tcp_connection.h"
 #include "praft/praft.h"
 
 #define kPIKIWIDB_VERSION "4.0.0"

src/praft/CMakeLists.txt

@@ -2,30 +2,28 @@
 # This source code is licensed under the BSD-style license found in the
 # LICENSE file in the root directory of this source tree. An additional grant
 # of patent rights can be found in the PATENTS file in the same directory.
-
-FILE(GLOB PRAFT_PROTO "${CMAKE_CURRENT_SOURCE_DIR}/*.proto")
-EXECUTE_PROCESS(
- COMMAND ${PROTOBUF_PROTOC_EXECUTABLE} --cpp_out=${CMAKE_CURRENT_SOURCE_DIR} -I=${CMAKE_CURRENT_SOURCE_DIR} ${PRAFT_PROTO}
+ADD_CUSTOM_COMMAND(
+ OUTPUT "${PROTO_OUTPUT_DIR}/binlog.pb.cc"
+ DEPENDS extern_protobuf
+ COMMAND ${PROTOBUF_PROTOC_EXECUTABLE}
+ ARGS -I ${CMAKE_CURRENT_SOURCE_DIR}
+ --cpp_out ${PROTO_OUTPUT_DIR}
+ ${CMAKE_CURRENT_SOURCE_DIR}/binlog.proto
 )
+ADD_LIBRARY(binlog_pb STATIC "${PROTO_OUTPUT_DIR}/binlog.pb.cc")
 
 FILE(GLOB PRAFT_SRC
  "${CMAKE_CURRENT_SOURCE_DIR}/*.cc"
- "${CMAKE_CURRENT_SOURCE_DIR}/*.h"
 )
 SET(LIBRARY_OUTPUT_PATH ${PROJECT_SOURCE_DIR}/bin)
 ADD_LIBRARY(praft ${PRAFT_SRC})
 
 TARGET_INCLUDE_DIRECTORIES(praft
  PRIVATE ${PROJECT_SOURCE_DIR}/src
- PRIVATE ${PROJECT_SOURCE_DIR}/src/pstd
- PRIVATE ${PROJECT_SOURCE_DIR}/src/net
- PRIVATE ${PROJECT_SOURCE_DIR}/src/storage/include
- PRIVATE ${rocksdb_SOURCE_DIR}/
  PRIVATE ${rocksdb_SOURCE_DIR}/include
  PRIVATE ${BRAFT_INCLUDE_DIR}
  PRIVATE ${BRPC_INCLUDE_DIR}
- PRIVATE ${GFLAGS_INCLUDE_PATH}
- PRIVATE ${PROJECT_SOURCE_DIR}/src/praft
+ PRIVATE ${PROTO_OUTPUT_DIR}
 )
 
 IF(CMAKE_SYSTEM_NAME STREQUAL "Linux")
@@ -34,4 +32,4 @@ ENDIF()
 
 TARGET_LINK_LIBRARIES(praft net; dl; fmt; storage; pstd braft brpc ssl crypto zlib protobuf leveldb gflags rocksdb z ${PRAFT_LIB})
 
-SET_TARGET_PROPERTIES(praft PROPERTIES LINKER_LANGUAGE CXX)
+SET_TARGET_PROPERTIES(praft PROPERTIES LINKER_LANGUAGE CXX)

src/praft/binlog.proto

@@ -0,0 +1,22 @@
+syntax = "proto3";
+package pikiwidb;
+// option optimize_for = LITE_RUNTIME;
+
+enum OperateType {
+ kNoOperate = 0;
+ kPut = 1;
+ kDelete = 2;
+}
+
+message BinlogEntry {
+ uint32 cf_idx = 1;
+ OperateType op_type = 2;
+ bytes key = 3;
+ optional bytes value = 4;
+}
+
+message Binlog {
+ uint32 db_id = 1;
+ uint32 slot_idx = 2;
+ repeated BinlogEntry entries = 3;
+}