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

include/pika_define.h

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

include/rsync_client.h

@@ -16,21 +16,16 @@
 #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_status.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/rsync_client_thread.h"
 #include "include/throttle.h"
-
-using namespace pstd;
-using namespace net;
-using namespace RsyncService;
+#include "rsync_service.pb.h"
 
 const std::string kDumpMetaFileName = "DUMP_META_DATA";
 const std::string kUuidPrefix = "snapshot-uuid:";
@@ -57,19 +52,19 @@ class RsyncClient : public net::Thread {
     return state_.load() == RUNNING;
   }
   bool IsIdle() { return state_.load() == IDLE;}
-  void OnReceive(RsyncResponse* resp);
+  void OnReceive(RsyncService::RsyncResponse* resp);
 
 private:
   bool Recover();
-  Status Wait(RsyncResponse*& resp);
+  Status Wait(RsyncService::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);
+  void HandleRsyncMetaResponse(RsyncService::RsyncResponse* response);
 
 private:
   std::map<std::string, std::string> meta_table_;
@@ -149,7 +144,7 @@ class WaitObject {
   std::string filename_;
   RsyncService::Type type_;
   size_t offset_ = 0xFFFFFFFF;
-  RsyncResponse* resp_ = nullptr;
+  RsyncService::RsyncResponse* resp_ = nullptr;
 };
 
 } // end namespace rsync

src/pika_server.cc

@@ -3,8 +3,6 @@
 // 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.
 
-#include "include/pika_server.h"
-
 #include <arpa/inet.h>
 #include <ifaddrs.h>
 #include <netinet/in.h>
@@ -25,10 +23,10 @@
 
 #include "include/pika_cmd_table_manager.h"
 #include "include/pika_dispatch_thread.h"
-#include "include/pika_rm.h"
-#include "pstd_hash.h"
 #include "include/pika_monotonic_time.h"
 #include "include/pika_instant.h"
+#include "include/pika_server.h"
+#include "include/pika_rm.h"
 
 using pstd::Status;
 extern PikaServer* g_pika_server;

src/pika_slot.cc

@@ -3,17 +3,16 @@
 // 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.
 
-#include "include/pika_slot.h"
-
 #include <fstream>
 #include <memory>
 
 #include "include/pika_conf.h"
 #include "include/pika_rm.h"
 #include "include/pika_server.h"
+#include "include/pika_slot.h"
 
 #include "pstd/include/mutex_impl.h"
-#include "pstd_hash.h"
+#include "pstd/include/pstd_hash.h"
 
 using pstd::Status;
 
@@ -143,7 +142,6 @@ void Slot::PrepareRsync() {
 // 3, Update master offset, and the PikaAuxiliaryThread cron will connect and do slaveof task with master
 bool Slot::TryUpdateMasterOffset() {
   std::string info_path = dbsync_path_ + kBgsaveInfoFile;
-  // todo 这里要改动，定期向 master 发送 meta_rsync 的请求
   if (!pstd::FileExists(info_path)) {
     LOG(WARNING) << "info path: " << info_path << " not exist";
     return false;
@@ -331,7 +329,6 @@ Status Slot::GetBgSaveUUID(std::string* snapshot_uuid) {
   if (snapshot_uuid_.empty()) {
     std::string info_data;
     const std::string infoPath = bgsave_info().path + "/info";
-    // todo 这里待替换
     rocksdb::Status s = rocksdb::ReadFileToString(rocksdb::Env::Default(), infoPath, &info_data);
     if (!s.ok()) {
       LOG(WARNING) << "read dump meta info failed! error:" << s.ToString();

src/rsync_client.cc

@@ -14,7 +14,6 @@
 using namespace net;
 using namespace pstd;
 using namespace RsyncService;
-using namespace pstd;
 
 extern PikaServer* g_pika_server;
 

src/rsync_client_thread.cc

@@ -42,3 +42,4 @@ RsyncClientThread::RsyncClientThread(int cron_interval, int keepalive_timeout, v
 
 RsyncClientThread::~RsyncClientThread() {}
 } //end namespace rsync
+

src/rsync_service.proto

@@ -46,3 +46,4 @@ message RsyncResponse {
     optional MetaResponse meta_resp = 6;
     optional FileResponse file_resp = 7;
 }
+