Merge remote-tracking branch 'origin/feature/sync-mnode-integration' into fix/mnode

include/libs/sync/sync.h

@@ -88,6 +88,9 @@ typedef struct SReConfigCbMeta {
   SyncIndex index;
   SyncTerm  term;
   SyncTerm  currentTerm;
+  SSyncCfg  oldCfg;
+  bool      isDrop;
+  uint64_t  flag;
 } SReConfigCbMeta;
 
 typedef struct SSyncFSM {
@@ -163,6 +166,7 @@ void        syncCleanUp();
 int64_t     syncOpen(const SSyncInfo* pSyncInfo);
 void        syncStart(int64_t rid);
 void        syncStop(int64_t rid);
+int32_t     syncSetStandby(int64_t rid);
 int32_t     syncReconfig(int64_t rid, const SSyncCfg* pSyncCfg);
 ESyncState  syncGetMyRole(int64_t rid);
 const char* syncGetMyRoleStr(int64_t rid);

source/libs/sync/inc/syncInt.h

@@ -182,7 +182,7 @@ int32_t syncNodeSendMsgByInfo(const SNodeInfo* nodeInfo, SSyncNode* pSyncNode, S
 cJSON*  syncNode2Json(const SSyncNode* pSyncNode);
 char*   syncNode2Str(const SSyncNode* pSyncNode);
 char*   syncNode2SimpleStr(const SSyncNode* pSyncNode);
-void    syncNodeUpdateConfig(SSyncNode* pSyncNode, SSyncCfg* newConfig);
+void    syncNodeUpdateConfig(SSyncNode* pSyncNode, SSyncCfg* newConfig, bool* isDrop);
 
 SSyncNode* syncNodeAcquire(int64_t rid);
 void       syncNodeRelease(SSyncNode* pNode);

source/libs/sync/src/syncAppendEntries.c

@@ -333,7 +333,7 @@ int32_t syncNodeOnAppendEntriesCb(SSyncNode* ths, SyncAppendEntries* pMsg) {
                 cbMeta.seqNum = pEntry->seqNum;
                 cbMeta.term = pEntry->term;
                 cbMeta.currentTerm = ths->pRaftStore->currentTerm;
-                cbMeta.flag = 9;
+                cbMeta.flag = 0x11;
 
                 bool needExecute = true;
                 if (ths->pSnapshot != NULL && cbMeta.index <= ths->pSnapshot->lastApplyIndex) {
@@ -347,24 +347,51 @@ int32_t syncNodeOnAppendEntriesCb(SSyncNode* ths, SyncAppendEntries* pMsg) {
 
               // config change
               if (pEntry->originalRpcType == TDMT_VND_SYNC_CONFIG_CHANGE) {
+                SSyncCfg oldSyncCfg = ths->pRaftCfg->cfg;
+
                 SSyncCfg newSyncCfg;
                 int32_t  ret = syncCfgFromStr(rpcMsg.pCont, &newSyncCfg);
                 ASSERT(ret == 0);
 
-                syncNodeUpdateConfig(ths, &newSyncCfg);
+                // update new config myIndex
-                if (ths->state == TAOS_SYNC_STATE_LEADER) {
+                bool hit = false;
-                  syncNodeBecomeLeader(ths);
+                for (int i = 0; i < newSyncCfg.replicaNum; ++i) {
-                } else {
+                  if (strcmp(ths->myNodeInfo.nodeFqdn, (newSyncCfg.nodeInfo)[i].nodeFqdn) == 0 &&
-                  syncNodeBecomeFollower(ths);
+                      ths->myNodeInfo.nodePort == (newSyncCfg.nodeInfo)[i].nodePort) {
+                    newSyncCfg.myIndex = i;
+                    hit = true;
+                    break;
+                  }
+                }
+                ASSERT(hit == true);
+
+                bool isDrop;
+                syncNodeUpdateConfig(ths, &newSyncCfg, &isDrop);
+
+                // change isStandBy to normal
+                if (!isDrop) {
+                  if (ths->state == TAOS_SYNC_STATE_LEADER) {
+                    syncNodeBecomeLeader(ths);
+                  } else {
+                    syncNodeBecomeFollower(ths);
+                  }
                 }
 
-                // maybe newSyncCfg.myIndex is updated in syncNodeUpdateConfig
+                char* sOld = syncCfg2Str(&oldSyncCfg);
+                char* sNew = syncCfg2Str(&newSyncCfg);
+                sInfo("==config change== 0x11 old:%s new:%s isDrop:%d \n", sOld, sNew, isDrop);
+                taosMemoryFree(sOld);
+                taosMemoryFree(sNew);
+
                 if (ths->pFsm->FpReConfigCb != NULL) {
                   SReConfigCbMeta cbMeta = {0};
                   cbMeta.code = 0;
                   cbMeta.currentTerm = ths->pRaftStore->currentTerm;
                   cbMeta.index = pEntry->index;
                   cbMeta.term = pEntry->term;
+                  cbMeta.oldCfg = oldSyncCfg;
+                  cbMeta.flag = 0x11;
+                  cbMeta.isDrop = isDrop;
                   ths->pFsm->FpReConfigCb(ths->pFsm, newSyncCfg, cbMeta);
                 }
               }

source/libs/sync/src/syncCommit.c

@@ -111,7 +111,7 @@ void syncMaybeAdvanceCommitIndex(SSyncNode* pSyncNode) {
             cbMeta.seqNum = pEntry->seqNum;
             cbMeta.term = pEntry->term;
             cbMeta.currentTerm = pSyncNode->pRaftStore->currentTerm;
-            cbMeta.flag = 7;
+            cbMeta.flag = 0x1;
 
             bool needExecute = true;
             if (pSyncNode->pSnapshot != NULL && cbMeta.index <= pSyncNode->pSnapshot->lastApplyIndex) {
@@ -125,24 +125,51 @@ void syncMaybeAdvanceCommitIndex(SSyncNode* pSyncNode) {
 
           // config change
           if (pEntry->originalRpcType == TDMT_VND_SYNC_CONFIG_CHANGE) {
+            SSyncCfg oldSyncCfg = pSyncNode->pRaftCfg->cfg;
+
             SSyncCfg newSyncCfg;
             int32_t  ret = syncCfgFromStr(rpcMsg.pCont, &newSyncCfg);
             ASSERT(ret == 0);
 
-            syncNodeUpdateConfig(pSyncNode, &newSyncCfg);
+            // update new config myIndex
-            if (pSyncNode->state == TAOS_SYNC_STATE_LEADER) {
+            bool hit = false;
-              syncNodeBecomeLeader(pSyncNode);
+            for (int i = 0; i < newSyncCfg.replicaNum; ++i) {
-            } else {
+              if (strcmp(pSyncNode->myNodeInfo.nodeFqdn, (newSyncCfg.nodeInfo)[i].nodeFqdn) == 0 &&
-              syncNodeBecomeFollower(pSyncNode);
+                  pSyncNode->myNodeInfo.nodePort == (newSyncCfg.nodeInfo)[i].nodePort) {
+                newSyncCfg.myIndex = i;
+                hit = true;
+                break;
+              }
+            }
+            ASSERT(hit == true);
+
+            bool isDrop;
+            syncNodeUpdateConfig(pSyncNode, &newSyncCfg, &isDrop);
+
+            // change isStandBy to normal
+            if (!isDrop) {
+              if (pSyncNode->state == TAOS_SYNC_STATE_LEADER) {
+                syncNodeBecomeLeader(pSyncNode);
+              } else {
+                syncNodeBecomeFollower(pSyncNode);
+              }
             }
 
-            // maybe newSyncCfg.myIndex is updated in syncNodeUpdateConfig
+            char* sOld = syncCfg2Str(&oldSyncCfg);
+            char* sNew = syncCfg2Str(&newSyncCfg);
+            sInfo("==config change== 0x1 old:%s new:%s isDrop:%d \n", sOld, sNew, isDrop);
+            taosMemoryFree(sOld);
+            taosMemoryFree(sNew);
+
             if (pSyncNode->pFsm->FpReConfigCb != NULL) {
               SReConfigCbMeta cbMeta = {0};
               cbMeta.code = 0;
               cbMeta.currentTerm = pSyncNode->pRaftStore->currentTerm;
               cbMeta.index = pEntry->index;
               cbMeta.term = pEntry->term;
+              cbMeta.oldCfg = oldSyncCfg;
+              cbMeta.flag = 0x1;
+              cbMeta.isDrop = isDrop;
               pSyncNode->pFsm->FpReConfigCb(pSyncNode->pFsm, newSyncCfg, cbMeta);
             }
           }

source/libs/sync/src/syncMain.c

@@ -141,9 +141,38 @@ void syncStop(int64_t rid) {
   taosRemoveRef(tsNodeRefId, rid);
 }
 
+int32_t syncSetStandby(int64_t rid) {
+  SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid);
+  if (pSyncNode == NULL) {
+    return -1;
+  }
+
+  if (pSyncNode->state != TAOS_SYNC_STATE_LEADER) {
+    taosReleaseRef(tsNodeRefId, pSyncNode->rid);
+    return -1;
+  }
+
+  // state change
+  pSyncNode->state = TAOS_SYNC_STATE_FOLLOWER;
+  syncNodeStopHeartbeatTimer(pSyncNode);
+
+  // reset elect timer, long enough
+  int32_t electMS = TIMER_MAX_MS;
+  int32_t ret = syncNodeRestartElectTimer(pSyncNode, electMS);
+  ASSERT(ret == 0);
+
+  pSyncNode->pRaftCfg->isStandBy = 1;
+  raftCfgPersist(pSyncNode->pRaftCfg);
+
+  taosReleaseRef(tsNodeRefId, pSyncNode->rid);
+  return 0;
+}
+
 int32_t syncReconfig(int64_t rid, const SSyncCfg* pSyncCfg) {
   int32_t ret = 0;
   char*   configChange = syncCfg2Str((SSyncCfg*)pSyncCfg);
+  sInfo("==syncReconfig== newconfig:%s", configChange);
+
   SRpcMsg rpcMsg = {0};
   rpcMsg.msgType = TDMT_VND_SYNC_CONFIG_CHANGE;
   rpcMsg.info.noResp = 1;
@@ -951,21 +980,9 @@ char* syncNode2SimpleStr(const SSyncNode* pSyncNode) {
   return s;
 }
 
-void syncNodeUpdateConfig(SSyncNode* pSyncNode, SSyncCfg* newConfig) {
+void syncNodeUpdateConfig(SSyncNode* pSyncNode, SSyncCfg* newConfig, bool* isDrop) {
-  bool hit = false;
-  for (int i = 0; i < newConfig->replicaNum; ++i) {
-    if (strcmp(pSyncNode->myNodeInfo.nodeFqdn, (newConfig->nodeInfo)[i].nodeFqdn) == 0 &&
-        pSyncNode->myNodeInfo.nodePort == (newConfig->nodeInfo)[i].nodePort) {
-      newConfig->myIndex = i;
-      hit = true;
-      break;
-    }
-  }
-  ASSERT(hit == true);
-
   pSyncNode->pRaftCfg->cfg = *newConfig;
-  int32_t ret = raftCfgPersist(pSyncNode->pRaftCfg);
+  int32_t ret = 0;
-  ASSERT(ret == 0);
 
   // init internal
   pSyncNode->myNodeInfo = pSyncNode->pRaftCfg->cfg.nodeInfo[pSyncNode->pRaftCfg->cfg.myIndex];
@@ -995,9 +1012,22 @@ void syncNodeUpdateConfig(SSyncNode* pSyncNode, SSyncCfg* newConfig) {
   voteGrantedUpdate(pSyncNode->pVotesGranted, pSyncNode);
   votesRespondUpdate(pSyncNode->pVotesRespond, pSyncNode);
 
-  pSyncNode->pRaftCfg->isStandBy = 0;
+  // isDrop
-  raftCfgPersist(pSyncNode->pRaftCfg);
+  *isDrop = true;
+  for (int i = 0; i < newConfig->replicaNum; ++i) {
+    if (strcmp((newConfig->nodeInfo)[i].nodeFqdn, pSyncNode->myNodeInfo.nodeFqdn) == 0 &&
+        (newConfig->nodeInfo)[i].nodePort == pSyncNode->myNodeInfo.nodePort) {
+      *isDrop = false;
+      break;
+    }
+  }
 
+  if (!(*isDrop)) {
+    // change isStandBy to normal
+    pSyncNode->pRaftCfg->isStandBy = 0;
+  }
+
+  raftCfgPersist(pSyncNode->pRaftCfg);
   syncNodeLog2("==syncNodeUpdateConfig==", pSyncNode);
 }
 

source/libs/sync/test/syncConfigChangeTest.cpp

@@ -43,8 +43,9 @@ void CommitCb(struct SSyncFSM* pFsm, const SRpcMsg* pMsg, SFsmCbMeta cbMeta) {
 
   if (cbMeta.index > beginIndex) {
     char logBuf[256];
-    snprintf(logBuf, sizeof(logBuf), "==callback== ==CommitCb== pFsm:%p, index:%ld, isWeak:%d, code:%d, state:%d %s flag:%lu\n",
+    snprintf(logBuf, sizeof(logBuf),
-             pFsm, cbMeta.index, cbMeta.isWeak, cbMeta.code, cbMeta.state, syncUtilState2String(cbMeta.state), cbMeta.flag);
+             "==callback== ==CommitCb== pFsm:%p, index:%ld, isWeak:%d, code:%d, state:%d %s flag:%lu\n", pFsm,
+             cbMeta.index, cbMeta.isWeak, cbMeta.code, cbMeta.state, syncUtilState2String(cbMeta.state), cbMeta.flag);
     syncRpcMsgLog2(logBuf, (SRpcMsg*)pMsg);
   } else {
     sTrace("==callback== ==CommitCb== do not apply again %ld", cbMeta.index);
@@ -54,15 +55,16 @@ void CommitCb(struct SSyncFSM* pFsm, const SRpcMsg* pMsg, SFsmCbMeta cbMeta) {
 void PreCommitCb(struct SSyncFSM* pFsm, const SRpcMsg* pMsg, SFsmCbMeta cbMeta) {
   char logBuf[256];
   snprintf(logBuf, sizeof(logBuf),
-           "==callback== ==PreCommitCb== pFsm:%p, index:%ld, isWeak:%d, code:%d, state:%d %s flag:%lu\n", pFsm, cbMeta.index,
+           "==callback== ==PreCommitCb== pFsm:%p, index:%ld, isWeak:%d, code:%d, state:%d %s flag:%lu\n", pFsm,
-           cbMeta.isWeak, cbMeta.code, cbMeta.state, syncUtilState2String(cbMeta.state), cbMeta.flag);
+           cbMeta.index, cbMeta.isWeak, cbMeta.code, cbMeta.state, syncUtilState2String(cbMeta.state), cbMeta.flag);
   syncRpcMsgLog2(logBuf, (SRpcMsg*)pMsg);
 }
 
 void RollBackCb(struct SSyncFSM* pFsm, const SRpcMsg* pMsg, SFsmCbMeta cbMeta) {
   char logBuf[256];
-  snprintf(logBuf, sizeof(logBuf), "==callback== ==RollBackCb== pFsm:%p, index:%ld, isWeak:%d, code:%d, state:%d %s flag:%lu\n",
+  snprintf(logBuf, sizeof(logBuf),
-           pFsm, cbMeta.index, cbMeta.isWeak, cbMeta.code, cbMeta.state, syncUtilState2String(cbMeta.state), cbMeta.flag);
+           "==callback== ==RollBackCb== pFsm:%p, index:%ld, isWeak:%d, code:%d, state:%d %s flag:%lu\n", pFsm,
+           cbMeta.index, cbMeta.isWeak, cbMeta.code, cbMeta.state, syncUtilState2String(cbMeta.state), cbMeta.flag);
   syncRpcMsgLog2(logBuf, (SRpcMsg*)pMsg);
 }
 
@@ -75,13 +77,23 @@ int32_t GetSnapshotCb(struct SSyncFSM* pFsm, SSnapshot* pSnapshot) {
 
 void RestoreFinishCb(struct SSyncFSM* pFsm) { sTrace("==callback== ==RestoreFinishCb=="); }
 
+void ReConfigCb(struct SSyncFSM* pFsm, SSyncCfg newCfg, SReConfigCbMeta cbMeta) {
+  sTrace("==callback== ==ReConfigCb== flag:0x%lX, isDrop:%d, index:%ld, code:%d, currentTerm:%lu, term:%lu", cbMeta.flag, cbMeta.isDrop, cbMeta.index, cbMeta.code, cbMeta.currentTerm, cbMeta.term);
+}
+
 SSyncFSM* createFsm() {
   SSyncFSM* pFsm = (SSyncFSM*)taosMemoryMalloc(sizeof(SSyncFSM));
   pFsm->FpCommitCb = CommitCb;
   pFsm->FpPreCommitCb = PreCommitCb;
   pFsm->FpRollBackCb = RollBackCb;
+
   pFsm->FpGetSnapshot = GetSnapshotCb;
   pFsm->FpRestoreFinishCb = RestoreFinishCb;
+  pFsm->FpSnapshotApply = NULL;
+  pFsm->FpSnapshotRead = NULL;
+
+  pFsm->FpReConfigCb = ReConfigCb;
+
   return pFsm;
 }
 
@@ -181,7 +193,7 @@ SRpcMsg* createRpcMsg(int i, int count, int myIndex) {
 
 int main(int argc, char** argv) {
   tsAsyncLog = 0;
-  sDebugFlag = DEBUG_TRACE + DEBUG_SCREEN + DEBUG_FILE;
+  sDebugFlag = DEBUG_TRACE + DEBUG_SCREEN + DEBUG_FILE + DEBUG_INFO;
   if (argc != 7) {
     usage(argv[0]);
     exit(-1);
@@ -227,7 +239,7 @@ int main(int argc, char** argv) {
   assert(pSyncNode != NULL);
 
   if (isConfigChange) {
-    configChange(rid, 3, myIndex);
+    configChange(rid, 2, myIndex);
   }
 
   //---------------------------