add continue transferring from breakpoint for dbsync

include/pika_trysync_thread.h

@@ -19,7 +19,7 @@ class PikaTrysyncThread : public pink::Thread {
 
   bool Send();
   bool RecvProc();
-  bool PrepareRsync();
+  void PrepareRsync();
   bool TryUpdateMasterOffset();
 
   virtual void* ThreadMain();

src/pika_server.cc

@@ -423,38 +423,41 @@ void PikaServer::DBSyncSendFile(const std::string& ip, int port) {
   }
 
   // Iterate to send files
+  int ret = 0;
   std::string target_path;
   std::vector<std::string>::iterator it = descendant.begin();
-  int failed_count = 0;
   for (; it != descendant.end(); ++it) {
     target_path = (*it).substr(bgsave_info_.path.size() + 1);
     if (target_path == kBgsaveInfoFile) {
       continue;
     }
     // We need specify the speed limit for every single file
     slash::RsyncRemote remote(ip, port, kDBSyncModule, g_pika_conf->db_sync_speed() * 1024);
-    int ret = slash::RsyncSendFile(*it, target_path, remote);
+    ret = slash::RsyncSendFile(*it, target_path, remote);
     if (0 != ret) {
       LOG(ERROR) << "rsync send file failed! From: " << *it
         << ", To: " << target_path
         << ", At: " << ip << ":" << port
         << ", Error: " << ret;
-      failed_count++;
+      break;
     }
   }
   // Send info file at last
-  slash::RsyncRemote remote(ip, port, kDBSyncModule, g_pika_conf->db_sync_speed() * 1024);
-  if (0 != slash::RsyncSendFile(bgsave_info_.path + "/" + kBgsaveInfoFile, kBgsaveInfoFile, remote)) {
+  if (0 == ret) {
+    slash::RsyncRemote remote(ip, port, kDBSyncModule, g_pika_conf->db_sync_speed() * 1024);
+    if (0 != (ret = slash::RsyncSendFile(bgsave_info_.path + "/" + kBgsaveInfoFile, kBgsaveInfoFile, remote))) {
       LOG(ERROR) << "send info file failed";
-      failed_count++;
+    }
   }
 
   // remove slave
   char buf[10];
   slash::ll2string(buf, sizeof(buf), port);
   std::string ip_port(ip + ":" + buf);
   db_sync_slaves.erase(ip_port);
-  LOG(INFO) << "rsync send files finished, failed:" << failed_count++;
+  if (0 == ret) {
+    LOG(INFO) << "rsync send files success";
+  }
 }
 
 /*

src/pika_trysync_thread.cc

@@ -74,7 +74,7 @@ bool PikaTrysyncThread::RecvProc() {
       }
       is_authed = true;
     } else {
-      if (reply.size() == 1 &&
+      if (cli_->argv_.size() == 1 &&
           slash::string2l(reply.data(), reply.size(), &sid_)) {
         // Luckly, I got your point, the sync is comming
         DLOG(INFO) << "Recv sid from master: " << sid_;
@@ -162,18 +162,14 @@ bool PikaTrysyncThread::TryUpdateMasterOffset() {
   return true;
 }
 
-bool PikaTrysyncThread::PrepareRsync() {
+void PikaTrysyncThread::PrepareRsync() {
   std::string db_sync_path = g_pika_conf->db_sync_path();
   slash::StopRsync(db_sync_path);
-  if (!slash::DeleteDirIfExist(db_sync_path)) {
-    LOG(WARNING) << "Failed to delete rsync dir:" << db_sync_path;
-  }
-  int ret = slash::CreatePath(db_sync_path + "kv");
-  ret += slash::CreatePath(db_sync_path + "hash");
-  ret += slash::CreatePath(db_sync_path + "list");
-  ret += slash::CreatePath(db_sync_path + "set");
-  ret += slash::CreatePath(db_sync_path + "zset");
-  return ret == 0 ? true : false;
+  slash::CreatePath(db_sync_path + "kv");
+  slash::CreatePath(db_sync_path + "hash");
+  slash::CreatePath(db_sync_path + "list");
+  slash::CreatePath(db_sync_path + "set");
+  slash::CreatePath(db_sync_path + "zset");
 }
 
 // TODO maybe use RedisCli
@@ -193,9 +189,7 @@ void* PikaTrysyncThread::ThreadMain() {
 
     //Start rsync
     std::string dbsync_path = g_pika_conf->db_sync_path();
-    if (!PrepareRsync()) {
-      LOG(ERROR) << "Failed to prepare env for rsync, path:" << dbsync_path;;
-    }
+    PrepareRsync();
     int ret = slash::StartRsync(dbsync_path, kDBSyncModule, g_pika_conf->port() + 300);
     if (0 != ret) {
       LOG(ERROR) << "Failed to start rsync, path:" << dbsync_path << " error : " << ret;