Feature: sync all snapshot data in Pika process without rsync subprocess

.github/workflows/operator.yml

@@ -29,7 +29,7 @@ jobs:
 
  - name: Build
  run: |
- cd tools/pika_operator && make 
+ cd tools/pika_operator && make
 
  - name: Unit Test
  run: |

CMakeLists.txt

@@ -732,7 +732,7 @@ set(PIKA_BUILD_VERSION_CC ${CMAKE_BINARY_DIR}/pika_build_version.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)
 
-set(PROTO_FILES ${CMAKE_CURRENT_SOURCE_DIR}/src/pika_inner_message.proto)
+set(PROTO_FILES ${CMAKE_CURRENT_SOURCE_DIR}/src/pika_inner_message.proto ${CMAKE_CURRENT_SOURCE_DIR}/src/rsync_service.proto)
 custom_protobuf_generate_cpp(PROTO_SRCS PROTO_HDRS ${PROTO_FILES})
 message("pika PROTO_SRCS = ${PROTO_SRCS}")
 message("pika PROTO_HDRS = ${PROTO_HDRS}")

include/pika_define.h

@@ -34,6 +34,8 @@ class PikaServer;
 /* Port shift */
 const int kPortShiftRSync = 1000;
 const int kPortShiftReplServer = 2000;
+// todo 待移除，使用 kPortShiftRSync
+const int kPortShiftRsync2 = 10001;
 
 const std::string kPikaPidFile = "pika.pid";
 const std::string kPikaSecretFile = "rsync.secret";

include/pika_rm.h

@@ -21,6 +21,7 @@
 #include "include/pika_repl_server.h"
 #include "include/pika_slave_node.h"
 #include "include/pika_stable_log.h"
+#include "include/rsync_client.h"
 
 #define kBinlogSendPacketNum 40
 #define kBinlogSendBatchNum 100
@@ -157,7 +158,12 @@ class SyncSlaveSlot : public SyncSlot {
 
  std::string LocalIp();
 
+ void ActivateRsync();
+
+ bool IsRsyncRunning() {return rsync_cli_->IsRunning();}
+
  private:
+ std::unique_ptr<rsync::RsyncClient> rsync_cli_;
  pstd::Mutex slot_mu_;
  RmNode m_info_;
  ReplState repl_state_{kNoConnect};

include/pika_server.h

@@ -36,6 +36,7 @@
 #include "include/pika_repl_client.h"
 #include "include/pika_repl_server.h"
 #include "include/pika_rsync_service.h"
+#include "include/rsync_server.h"
 #include "include/pika_statistic.h"
 #include "include/pika_slot_command.h"
 #include "include/pika_migrate_thread.h"
@@ -261,6 +262,8 @@ class PikaServer : public pstd::noncopyable {
  /*
  * DBSync used
  */
+ pstd::Status GetDumpUUID(const std::string& db_name, const uint32_t slot_id, std::string* snapshot_uuid);
+ pstd::Status GetDumpMeta(const std::string& db_name, const uint32_t slot_id, std::vector<std::string>* files, std::string* snapshot_uuid);
  void DBSync(const std::string& ip, int port, const std::string& db_name, uint32_t slot_id);
  void TryDBSync(const std::string& ip, int port, const std::string& db_name, uint32_t slot_id, int32_t top);
  void DbSyncSendFile(const std::string& ip, int port, const std::string& db_name, uint32_t slot_id);
@@ -587,6 +590,7 @@ class PikaServer : public pstd::noncopyable {
  * Rsync used
  */
  std::unique_ptr<PikaRsyncService> pika_rsync_service_;
+ std::unique_ptr<rsync::RsyncServer> rsync_server_;
 
  /*
  * Pubsub used

include/pika_slot.h

@@ -76,6 +76,8 @@ class Slot : public std::enable_shared_from_this<Slot>,public pstd::noncopyable
  bool IsBgSaving();
  void BgSaveSlot();
  BgSaveInfo bgsave_info();
+ void GetBgSaveMetaData(std::vector<std::string>* fileNames, std::string* snapshot_uuid);
+ pstd::Status GetBgSaveUUID(std::string* snapshot_uuid);
 
  // FlushDB & FlushSubDB use
  bool FlushDB();
@@ -93,6 +95,7 @@ class Slot : public std::enable_shared_from_this<Slot>,public pstd::noncopyable
  private:
  std::string db_name_;
  uint32_t slot_id_ = 0;
+ std::string snapshot_uuid_;
 
  std::string db_path_;
  std::string bgsave_sub_path_;

include/rsync_client.h

@@ -0,0 +1,156 @@
+// Copyright (c) 2023-present, Qihoo, Inc. All rights reserved.
+// 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.
+
+#ifndef RSYNC_CLIENT_H_
+#define RSYNC_CLIENT_H_
+#include <glog/logging.h>
+#include <fcntl.h>
+#include <sys/stat.h>
+#include <sys/types.h>
+#include <list>
+#include <atomic>
+#include <memory>
+#include <thread>
+#include <condition_variable>
+
+#include "include/rsync_client_thread.h"
+#include "net/include/bg_thread.h"
+#include "pstd/include/pstd_status.h"
+#include "include/rsync_client_thread.h"
+#include "net/include/net_cli.h"
+#include "pstd/include/env.h"
+#include "pstd/include/pstd_hash.h"
+#include "pstd/include/pstd_string.h"
+#include "pstd/include/pstd_status.h"
+#include "rsync_service.pb.h"
+#include "include/throttle.h"
+
+using namespace pstd;
+using namespace net;
+using namespace RsyncService;
+
+const std::string kDumpMetaFileName = "DUMP_META_DATA";
+const std::string kUuidPrefix = "snapshot-uuid:";
+
+namespace rsync {
+
+class RsyncWriter;
+class Session;
+class WaitObject;
+
+class RsyncClient : public net::Thread {
+public:
+ enum State {
+ IDLE,
+ RUNNING,
+ STOP,
+ };
+ RsyncClient(const std::string& dir, const std::string& db_name, const uint32_t slot_id);
+ void* ThreadMain() override;
+ bool Init();
+ Status Start();
+ Status Stop();
+ bool IsRunning() {
+ return state_.load() == RUNNING;
+ }
+ bool IsIdle() { return state_.load() == IDLE;}
+ void OnReceive(RsyncResponse* resp);
+
+private:
+ bool Recover();
+ Status Wait(RsyncResponse*& resp);
+ Status CopyRemoteFile(const std::string& filename);
+ Status CopyRemoteMeta(std::string* snapshot_uuid, std::set<std::string>* file_set);
+ Status LoadLocalMeta(std::string* snapshot_uuid, std::map<std::string, std::string>* file_map);
+ std::string GetLocalMetaFilePath();
+ Status FlushMetaTable();
+ Status CleanUpExpiredFiles(bool need_reset_path, const std::set<std::string>& files);
+ Status UpdateLocalMeta(const std::string& snapshot_uuid, const std::set<std::string>& expired_files, std::map<std::string, std::string>* localFileMap);
+ void HandleRsyncMetaResponse(RsyncResponse* response);
+
+private:
+ std::map<std::string, std::string> meta_table_;
+ int flush_period_;
+ std::set<std::string> file_set_;
+ std::string snapshot_uuid_;
+ std::string dir_;
+ std::string db_name_;
+ uint32_t slot_id_;
+ std::unique_ptr<RsyncClientThread> client_thread_;
+ std::atomic<State> state_;
+ int max_retries_;
+ std::unique_ptr<WaitObject> wo_;
+ std::condition_variable cond_;
+ std::mutex mu_;
+ std::unique_ptr<Throttle> throttle_;
+ std::string master_ip_;
+ int master_port_;
+};
+
+class RsyncWriter {
+public:
+ RsyncWriter(const std::string& filepath) {
+ filepath_ = filepath;
+ fd_ = open(filepath.c_str(), O_RDWR | O_APPEND | O_CREAT, 0644);
+ }
+ ~RsyncWriter() {}
+ Status Write(uint64_t offset, size_t n, const char* data) {
+ const char* ptr = data;
+ size_t left = n;
+ Status s;
+ while (left != 0) {
+ ssize_t done = write(fd_, ptr, left);
+ if (done < 0) {
+ if (errno == EINTR) {
+ continue;
+ }
+ LOG(WARNING) << "pwrite failed, filename: " << filepath_ << "errno: " << strerror(errno) << "n: " << n;
+ return Status::IOError(filepath_, "pwrite failed");
+ }
+ left -= done;
+ ptr += done;
+ offset += done;
+ }
+ return Status::OK();
+ }
+ Status Close() {
+ close(fd_);
+ return Status::OK();
+ }
+ Status Fsync() {
+ fsync(fd_);
+ return Status::OK();
+ }
+
+private:
+ std::string filepath_;
+ int fd_;
+};
+
+class WaitObject {
+public:
+ WaitObject() : filename_(""), type_(RsyncService::kRsyncMeta), offset_(0), resp_(nullptr) {}
+ ~WaitObject() {}
+ void Reset(const std::string& filename, RsyncService::Type t, size_t offset) {
+ resp_ = nullptr;
+ filename_ = filename;
+ type_ = t;
+ offset_ = offset;
+ }
+ void Reset(RsyncService::Type t) {
+ resp_ = nullptr;
+ filename_ = "";
+ type_ = t;
+ offset_ = 0xFFFFFFFF;
+ }
+ std::string filename_;
+ RsyncService::Type type_;
+ size_t offset_;
+ RsyncResponse* resp_;
+};
+
+} // end namespace rsync
+
+#endif

include/rsync_client_thread.h

@@ -0,0 +1,55 @@
+// Copyright (c) 2023-present, Qihoo, Inc. All rights reserved.
+// 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.
+
+#ifndef RSYNC_CLIENT_THREAD_H_
+#define RSYNC_CLIENT_THREAD_H_
+
+#include "net/include/client_thread.h"
+#include "net/include/net_conn.h"
+#include "net/include/pb_conn.h"
+#include "rsync_service.pb.h"
+
+using namespace pstd;
+using namespace net;
+
+namespace rsync {
+
+class RsyncClientConn : public PbConn {
+public:
+ RsyncClientConn(int fd, const std::string& ip_port,
+ net::Thread* thread, void* cb_handler,
+ NetMultiplexer* mpx);
+ ~RsyncClientConn() override;
+ int DealMessage() override;
+
+private:
+ void* cb_handler_;
+};
+
+class RsyncClientConnFactory : public ConnFactory {
+public:
+ RsyncClientConnFactory(void* scheduler) : cb_handler_(scheduler) {}
+ std::shared_ptr<net::NetConn> NewNetConn(int connfd, const std::string& ip_port,
+ net::Thread* thread, void* cb_handler,
+ net::NetMultiplexer* net) const override {
+ return std::static_pointer_cast<net::NetConn>(
+ std::make_shared<RsyncClientConn>(connfd, ip_port, thread, cb_handler_, net));
+ }
+private:
+ void* cb_handler_;
+};
+
+class RsyncClientThread : public ClientThread {
+public:
+ RsyncClientThread(int cron_interval, int keepalive_timeout, void* scheduler);
+ ~RsyncClientThread() override;
+private:
+ RsyncClientConnFactory conn_factory_;
+ ClientHandle handle_;
+};
+
+} //end namespace rsync
+
+#endif