dbslaveof support force option

include/pika_admin.h

@@ -44,13 +44,17 @@ class DbSlaveofCmd : public Cmd {
 
  private:
   std::string db_name_;
+  bool force_sync_;
   bool is_noone_;
+  bool have_offset_;
   int64_t filenum_;
   int64_t offset_;
   virtual void DoInitial() override;
   virtual void Clear() {
     db_name_.clear();
+    force_sync_ = false;
     is_noone_ = false;
+    have_offset_ = false;
   }
 };
 

include/pika_partition.h

@@ -82,9 +82,6 @@ class Partition : public std::enable_shared_from_this<Partition> {
   ReplState State();
   void SetReplState(const ReplState& state);
 
-  bool FullSync();
-  void SetFullSync(bool full_sync);
-
   void PrepareRsync();
   bool TryUpdateMasterOffset();
   bool ChangeDb(const std::string& new_path);
@@ -125,8 +122,6 @@ class Partition : public std::enable_shared_from_this<Partition> {
   slash::RecordMutex mutex_record_;
   std::shared_ptr<blackwidow::BlackWidow> db_;
 
-  bool full_sync_;
-
   /*
    * BgSave use
    */

src/pika_admin.cc

@@ -97,6 +97,8 @@ void SlaveofCmd::Do(std::shared_ptr<Partition> partition) {
 }
 
 /*
+ * dbslaveof db[0 ~ 7]
+ * dbslaveof db[0 ~ 7] force
  * dbslaveof db[0 ~ 7] no one
  * dbslaveof db[0 ~ 7] filenum offset
  */
@@ -121,19 +123,28 @@ void DbSlaveofCmd::DoInitial() {
     return;
   }
 
-  if (!strcasecmp(argv_[2].data(), "no")
-    && !strcasecmp(argv_[3].data(), "one")) {
-    is_noone_ = true;
+  if (argv_.size() == 3
+    && !strcasecmp(argv_[2].data(), "force")) {
+    force_sync_ = true;
     return;
   }
 
-  if (!slash::string2l(argv_[2].data(), argv_[2].size(), &filenum_) || filenum_ < 0) {
-    res_.SetRes(CmdRes::kInvalidInt);
-    return;
-  }
-  if (!slash::string2l(argv_[3].data(), argv_[3].size(), &offset_) || offset_ < 0) {
-    res_.SetRes(CmdRes::kInvalidInt);
-    return;
+  if (argv_.size() == 4) {
+    if (!strcasecmp(argv_[2].data(), "no")
+      && !strcasecmp(argv_[3].data(), "one")) {
+      is_noone_ = true;
+      return;
+    }
+
+    if (!slash::string2l(argv_[2].data(), argv_[2].size(), &filenum_) || filenum_ < 0) {
+      res_.SetRes(CmdRes::kInvalidInt);
+      return;
+    }
+    if (!slash::string2l(argv_[3].data(), argv_[3].size(), &offset_) || offset_ < 0) {
+      res_.SetRes(CmdRes::kInvalidInt);
+      return;
+    }
+    have_offset_ = true;
   }
 }
 
@@ -161,9 +172,15 @@ void DbSlaveofCmd::Do(std::shared_ptr<Partition> partition) {
     }
   } else {
     if (slave_partition->State() == ReplState::kNoConnect
-        || slave_partition->State() == ReplState::kError) {
-      db_partition->logger()->SetProducerStatus(filenum_, offset_);
-      slave_partition->SetReplState(ReplState::kTryConnect);
+      || slave_partition->State() == ReplState::kError) {
+      if (force_sync_) {
+        slave_partition->SetReplState(ReplState::kTryDBSync);
+      } else {
+        if (have_offset_) {
+          db_partition->logger()->SetProducerStatus(filenum_, offset_);
+        }
+        slave_partition->SetReplState(ReplState::kTryConnect);
+      }
       g_pika_server->SetLoopPartitionStateMachine(true);
     }
   }

src/pika_auxiliary_thread.cc

@@ -72,13 +72,7 @@ void PikaAuxiliaryThread::RunEveryPartitionStateMachine() {
         continue;
       }
       if (slave_partition->State() == ReplState::kTryConnect) {
-        // If partition need to FullSync, we Send DBSync Request
-        // directly, instead of TrySync first
-        if (partition->FullSync()) {
-          g_pika_server->SendPartitionDBSyncRequest(partition);
-        } else {
-          g_pika_server->SendPartitionTrySyncRequest(partition);
-        }
+        g_pika_server->SendPartitionTrySyncRequest(partition);
       } else if (slave_partition->State() == ReplState::kTryDBSync) {
         g_pika_server->SendPartitionDBSyncRequest(partition);
       } else if (slave_partition->State() == ReplState::kWaitReply) {

src/pika_command.cc

@@ -24,7 +24,7 @@ void InitCmdTable(std::unordered_map<std::string, Cmd*> *cmd_table) {
   ////Slaveof
   Cmd* slaveofptr = new SlaveofCmd(kCmdNameSlaveof, -3, kCmdFlagsRead | kCmdFlagsAdmin);
   cmd_table->insert(std::pair<std::string, Cmd*>(kCmdNameSlaveof, slaveofptr));
-  Cmd* dbslaveofptr = new DbSlaveofCmd(kCmdNameDbSlaveof, 4, kCmdFlagsRead | kCmdFlagsAdmin);
+  Cmd* dbslaveofptr = new DbSlaveofCmd(kCmdNameDbSlaveof, -2, kCmdFlagsRead | kCmdFlagsAdmin);
   cmd_table->insert(std::pair<std::string, Cmd*>(kCmdNameDbSlaveof, dbslaveofptr));
   Cmd* authptr = new AuthCmd(kCmdNameAuth, 2, kCmdFlagsRead | kCmdFlagsAdmin);
   cmd_table->insert(std::pair<std::string, Cmd*>(kCmdNameAuth, authptr));

src/pika_partition.cc

@@ -59,7 +59,6 @@ Partition::Partition(const std::string& table_name,
   table_name_(table_name),
   partition_id_(partition_id),
   binlog_io_error_(false),
-  full_sync_(false),
   bgsave_engine_(NULL),
   purging_(false) {
 
@@ -260,14 +259,6 @@ bool Partition::SetBinlogOffset(const BinlogOffset& boffset) {
   return false;
 }
 
-bool Partition::FullSync() {
-  return full_sync_;
-}
-
-void Partition::SetFullSync(bool full_sync) {
-  full_sync_ = full_sync;
-}
-
 void Partition::PrepareRsync() {
   slash::DeleteDirIfExist(dbsync_path_);
   slash::CreatePath(dbsync_path_ + "strings");
@@ -358,7 +349,6 @@ bool Partition::TryUpdateMasterOffset() {
 
   // Update master offset
   logger_->SetProducerStatus(filenum, offset);
-  full_sync_ = false;
   slave_partition->SetReplState(ReplState::kTryConnect);
   return true;
 }

src/pika_server.cc

@@ -515,13 +515,13 @@ Status PikaServer::DoSameThingSpecificTable(const TaskType& type, const std::set
 
 void PikaServer::PreparePartitionTrySync() {
   slash::RWLock rwl(&tables_rw_, false);
+  ReplState state = force_full_sync_ ?
+      ReplState::kTryDBSync : ReplState::kTryConnect;
   for (const auto& table_item : tables_) {
     for (const auto& partition_item : table_item.second->partitions_) {
-      partition_item.second->SetFullSync(force_full_sync_);
       Status s = g_pika_rm->SetSlaveReplState(
-          RmNode(table_item.second->GetTableName(),
-            partition_item.second->GetPartitionId()),
-          ReplState::kTryConnect);
+              RmNode(table_item.second->GetTableName(),
+                  partition_item.second->GetPartitionId()), state);
       if (!s.ok()) {
         LOG(WARNING) << s.ToString();
       }