sync refactor

source/libs/sync/inc/syncInt.h

@@ -116,6 +116,7 @@ typedef struct SSyncNode {
   SyncGroupId vgId;
   SSyncCfg    syncCfg;
   char        path[TSDB_FILENAME_LEN];
+  char        raftStorePath[TSDB_FILENAME_LEN * 2];
   SWal*       pWal;
   void*       rpcClient;
   int32_t (*FpSendMsg)(void* rpcClient, const SEpSet* pEpSet, SRpcMsg* pMsg);
@@ -195,8 +196,6 @@ typedef struct SSyncNode {
 
 SSyncNode* syncNodeOpen(const SSyncInfo* pSyncInfo);
 void       syncNodeClose(SSyncNode* pSyncNode);
-cJSON*     syncNode2Json(const SSyncNode* pSyncNode);
-char*      syncNode2Str(const SSyncNode* pSyncNode);
 
 int32_t syncNodeSendMsgById(const SRaftId* destRaftId, SSyncNode* pSyncNode, SRpcMsg* pMsg);
 int32_t syncNodeSendMsgByInfo(const SNodeInfo* nodeInfo, SSyncNode* pSyncNode, SRpcMsg* pMsg);
@@ -213,6 +212,11 @@ int32_t syncNodeRestartElectTimer(SSyncNode* pSyncNode, int32_t ms);
 int32_t syncNodeStartHeartbeatTimer(SSyncNode* pSyncNode);
 int32_t syncNodeStopHeartbeatTimer(SSyncNode* pSyncNode);
 
+// for debug
+cJSON* syncNode2Json(const SSyncNode* pSyncNode);
+char*  syncNode2Str(const SSyncNode* pSyncNode);
+void   syncNodePrint(char* s, const SSyncNode* pSyncNode);
+
 #ifdef __cplusplus
 }
 #endif

source/libs/sync/inc/syncMessage.h

@@ -46,6 +46,7 @@ typedef enum ESyncMessageType {
 // ---------------------------------------------
 cJSON* syncRpcMsg2Json(SRpcMsg* pRpcMsg);
 cJSON* syncRpcUnknownMsg2Json();
+char*  syncRpcMsg2Str(SRpcMsg* pRpcMsg);
 
 // ---------------------------------------------
 typedef enum ESyncTimeoutType {

source/libs/sync/src/syncIO.c

@@ -220,12 +220,17 @@ static void *syncIOConsumerFunc(void *param) {
   while (1) {
     int numOfMsgs = taosReadAllQitemsFromQset(io->pQset, qall, NULL, NULL);
     sTrace("syncIOConsumerFunc %d msgs are received", numOfMsgs);
-    if (numOfMsgs <= 0) break;
+    if (numOfMsgs <= 0) {
+      break;
+    }
 
     for (int i = 0; i < numOfMsgs; ++i) {
       taosGetQitem(qall, (void **)&pRpcMsg);
+
+      char *s = syncRpcMsg2Str(pRpcMsg);
       sTrace("syncIOConsumerFunc get item from queue: msgType:%d contLen:%d msg:%s", pRpcMsg->msgType, pRpcMsg->contLen,
-             (char *)(pRpcMsg->pCont));
+             s);
+      free(s);
 
       if (pRpcMsg->msgType == SYNC_PING) {
         if (io->FpOnSyncPing != NULL) {
@@ -247,7 +252,7 @@ static void *syncIOConsumerFunc(void *param) {
         }
 
       } else if (pRpcMsg->msgType == SYNC_REQUEST_VOTE) {
-        if (io->FpOnSyncRequestVote) {
+        if (io->FpOnSyncRequestVote != NULL) {
           SyncRequestVote *pSyncMsg;
           pSyncMsg = syncRequestVoteBuild(pRpcMsg->contLen);
           syncRequestVoteFromRpcMsg(pRpcMsg, pSyncMsg);
@@ -256,7 +261,7 @@ static void *syncIOConsumerFunc(void *param) {
         }
 
       } else if (pRpcMsg->msgType == SYNC_REQUEST_VOTE_REPLY) {
-        if (io->FpOnSyncRequestVoteReply) {
+        if (io->FpOnSyncRequestVoteReply != NULL) {
           SyncRequestVoteReply *pSyncMsg;
           pSyncMsg = SyncRequestVoteReplyBuild();
           syncRequestVoteReplyFromRpcMsg(pRpcMsg, pSyncMsg);
@@ -265,7 +270,7 @@ static void *syncIOConsumerFunc(void *param) {
         }
 
       } else if (pRpcMsg->msgType == SYNC_APPEND_ENTRIES) {
-        if (io->FpOnSyncAppendEntries) {
+        if (io->FpOnSyncAppendEntries != NULL) {
           SyncAppendEntries *pSyncMsg;
           pSyncMsg = syncAppendEntriesBuild(pRpcMsg->contLen);
           syncAppendEntriesFromRpcMsg(pRpcMsg, pSyncMsg);
@@ -274,7 +279,7 @@ static void *syncIOConsumerFunc(void *param) {
         }
 
       } else if (pRpcMsg->msgType == SYNC_APPEND_ENTRIES_REPLY) {
-        if (io->FpOnSyncAppendEntriesReply) {
+        if (io->FpOnSyncAppendEntriesReply != NULL) {
           SyncAppendEntriesReply *pSyncMsg;
           pSyncMsg = syncAppendEntriesReplyBuild();
           syncAppendEntriesReplyFromRpcMsg(pRpcMsg, pSyncMsg);

source/libs/sync/src/syncMain.c

@@ -98,6 +98,7 @@ SSyncNode* syncNodeOpen(const SSyncInfo* pSyncInfo) {
   pSyncNode->vgId = pSyncInfo->vgId;
   pSyncNode->syncCfg = pSyncInfo->syncCfg;
   memcpy(pSyncNode->path, pSyncInfo->path, sizeof(pSyncNode->path));
+  snprintf(pSyncNode->raftStorePath, sizeof(pSyncNode->raftStorePath), "%s/raft_store.json", pSyncInfo->path);
   pSyncNode->pWal = pSyncInfo->pWal;
   pSyncNode->rpcClient = pSyncInfo->rpcClient;
   pSyncNode->FpSendMsg = pSyncInfo->FpSendMsg;
@@ -136,6 +137,7 @@ SSyncNode* syncNodeOpen(const SSyncInfo* pSyncInfo) {
 
   // init TLA+ server vars
   pSyncNode->state = TAOS_SYNC_STATE_FOLLOWER;
+  pSyncNode->pRaftStore = raftStoreOpen(pSyncNode->raftStorePath);
   assert(pSyncNode->pRaftStore != NULL);
 
   // init TLA+ candidate vars
@@ -325,6 +327,13 @@ char* syncNode2Str(const SSyncNode* pSyncNode) {
   return serialized;
 }
 
+void syncNodePrint(char* s, const SSyncNode* pSyncNode) {
+  char* ss = syncNode2Str(pSyncNode);
+  // sTrace("syncNodePrint: %s [len:%lu]| %s", s, strlen(ss), ss);
+  fprintf(stderr, "syncNodePrint: %s [len:%lu]| %s", s, strlen(ss), ss);
+  free(ss);
+}
+
 int32_t syncNodeSendMsgById(const SRaftId* destRaftId, SSyncNode* pSyncNode, SRpcMsg* pMsg) {
   SEpSet epSet;
   syncUtilraftId2EpSet(destRaftId, &epSet);
@@ -499,6 +508,8 @@ static int32_t syncNodeOnPingReplyCb(SSyncNode* ths, SyncPingReply* pMsg) {
 }
 
 static void syncNodeEqPingTimer(void* param, void* tmrId) {
+  sTrace("<-- syncNodeEqPingTimer -->");
+
   SSyncNode* pSyncNode = (SSyncNode*)param;
   if (atomic_load_64(&pSyncNode->pingTimerLogicClockUser) <= atomic_load_64(&pSyncNode->pingTimerLogicClock)) {
     SyncTimeout* pSyncMsg = syncTimeoutBuild2(SYNC_TIMEOUT_PING, atomic_load_64(&pSyncNode->pingTimerLogicClock),
@@ -511,7 +522,7 @@ static void syncNodeEqPingTimer(void* param, void* tmrId) {
     // reset timer ms
     // pSyncNode->pingTimerMS += 100;
 
-    taosTmrReset(syncNodeEqPingTimer, pSyncNode->pingTimerMS, pSyncNode, &gSyncEnv->pTimerManager,
+    taosTmrReset(syncNodeEqPingTimer, pSyncNode->pingTimerMS, pSyncNode, gSyncEnv->pTimerManager,
                  &pSyncNode->pPingTimer);
   } else {
     sTrace("syncNodeEqPingTimer: pingTimerLogicClock:%lu, pingTimerLogicClockUser:%lu", pSyncNode->pingTimerLogicClock,
@@ -557,7 +568,7 @@ static void syncNodeEqHeartbeatTimer(void* param, void* tmrId) {
     // reset timer ms
     // pSyncNode->heartbeatTimerMS += 100;
 
-    taosTmrReset(syncNodeEqHeartbeatTimer, pSyncNode->heartbeatTimerMS, pSyncNode, &gSyncEnv->pTimerManager,
+    taosTmrReset(syncNodeEqHeartbeatTimer, pSyncNode->heartbeatTimerMS, pSyncNode, gSyncEnv->pTimerManager,
                  &pSyncNode->pHeartbeatTimer);
   } else {
     sTrace("syncNodeEqHeartbeatTimer: heartbeatTimerLogicClock:%lu, heartbeatTimerLogicClockUser:%lu",

source/libs/sync/src/syncMessage.c

@@ -76,6 +76,13 @@ cJSON* syncRpcUnknownMsg2Json() {
   return pJson;
 }
 
+char* syncRpcMsg2Str(SRpcMsg* pRpcMsg) {
+  cJSON* pJson = syncRpcMsg2Json(pRpcMsg);
+  char*  serialized = cJSON_Print(pJson);
+  cJSON_Delete(pJson);
+  return serialized;
+}
+
 // ---- message process SyncTimeout----
 SyncTimeout* syncTimeoutBuild() {
   uint32_t     bytes = sizeof(SyncTimeout);

source/libs/sync/test/syncEnvTest.cpp

@@ -3,6 +3,7 @@
 #include "syncIO.h"
 #include "syncInt.h"
 #include "syncRaftStore.h"
+#include "ttime.h"
 
 void logTest() {
   sTrace("--- sync log test: trace");
@@ -13,24 +14,13 @@ void logTest() {
   sFatal("--- sync log test: fatal");
 }
 
-void doSync() {
+void *pTimer = NULL;
-  SSyncInfo syncInfo;
+void *pTimerMgr = NULL;
-  syncInfo.vgId = 1;
+int   g = 300;
 
-  SSyncCfg* pCfg = &syncInfo.syncCfg;
+static void timerFp(void *param, void *tmrId) {
-  pCfg->replicaNum = 3;
+  printf("param:%p, tmrId:%p, pTimer:%p, pTimerMgr:%p \n", param, tmrId, pTimer, pTimerMgr);
-
+  taosTmrReset(timerFp, 1000, param, pTimerMgr, &pTimer);
-  pCfg->nodeInfo[0].nodePort = 7010;
-  taosGetFqdn(pCfg->nodeInfo[0].nodeFqdn);
-
-  pCfg->nodeInfo[1].nodePort = 7110;
-  taosGetFqdn(pCfg->nodeInfo[1].nodeFqdn);
-
-  pCfg->nodeInfo[2].nodePort = 7210;
-  taosGetFqdn(pCfg->nodeInfo[2].nodeFqdn);
-
-  SSyncNode* pSyncNode = syncNodeOpen(&syncInfo);
-  assert(pSyncNode != NULL);
 }
 
 int main() {
@@ -41,13 +31,12 @@ int main() {
 
   logTest();
 
-  // ret = syncIOStart();
-  // assert(ret == 0);
-
   ret = syncEnvStart();
   assert(ret == 0);
 
-  // doSync();
+  // timer
+  pTimerMgr = taosTmrInit(1000, 50, 10000, "SYNC-ENV-TEST");
+  taosTmrStart(timerFp, 1000, &g, pTimerMgr);
 
   while (1) {
     taosMsleep(1000);

source/libs/sync/test/syncIndexMgrTest.cpp

@@ -33,7 +33,7 @@ SSyncNode* syncNodeInit() {
   syncInfo.queue = gSyncIO->pMsgQ;
   syncInfo.FpEqMsg = syncIOEqMsg;
   syncInfo.pFsm = pFsm;
-  snprintf(syncInfo.path, sizeof(syncInfo.path), "%s", "./test_path");
+  snprintf(syncInfo.path, sizeof(syncInfo.path), "%s", "./");
 
   SSyncCfg* pCfg = &syncInfo.syncCfg;
   pCfg->myIndex = myIndex;

source/libs/sync/test/syncInitTest.cpp

@@ -30,7 +30,7 @@ SSyncNode* syncNodeInit() {
   syncInfo.queue = gSyncIO->pMsgQ;
   syncInfo.FpEqMsg = syncIOEqMsg;
   syncInfo.pFsm = pFsm;
-  snprintf(syncInfo.path, sizeof(syncInfo.path), "%s", "./test_path");
+  snprintf(syncInfo.path, sizeof(syncInfo.path), "%s", "./");
 
   SSyncCfg* pCfg = &syncInfo.syncCfg;
   pCfg->myIndex = myIndex;

source/libs/sync/test/syncPingTest.cpp

@@ -1,8 +1,10 @@
+#include <gtest/gtest.h>
 #include <stdio.h>
 #include "syncEnv.h"
 #include "syncIO.h"
 #include "syncInt.h"
 #include "syncRaftStore.h"
+#include "syncUtil.h"
 
 void logTest() {
   sTrace("--- sync log test: trace");
@@ -13,58 +15,65 @@ void logTest() {
   sFatal("--- sync log test: fatal");
 }
 
-uint16_t ports[3] = {7010, 7110, 7210};
+uint16_t ports[] = {7010, 7110, 7210, 7310, 7410};
+int32_t  replicaNum = 3;
+int32_t  myIndex = 0;
 
-SSyncNode* doSync(int myIndex) {
+SRaftId   ids[TSDB_MAX_REPLICA];
-  SSyncFSM* pFsm;
+SSyncInfo syncInfo;
+SSyncFSM* pFsm;
 
-  SSyncInfo syncInfo;
+SSyncNode* syncNodeInit() {
-  syncInfo.vgId = 1;
+  syncInfo.vgId = 1234;
   syncInfo.rpcClient = gSyncIO->clientRpc;
   syncInfo.FpSendMsg = syncIOSendMsg;
   syncInfo.queue = gSyncIO->pMsgQ;
   syncInfo.FpEqMsg = syncIOEqMsg;
   syncInfo.pFsm = pFsm;
-  snprintf(syncInfo.path, sizeof(syncInfo.path), "%s", "./path");
+  snprintf(syncInfo.path, sizeof(syncInfo.path), "%s", "./");
 
   SSyncCfg* pCfg = &syncInfo.syncCfg;
   pCfg->myIndex = myIndex;
-  pCfg->replicaNum = 3;
+  pCfg->replicaNum = replicaNum;
 
-  pCfg->nodeInfo[0].nodePort = ports[0];
+  for (int i = 0; i < replicaNum; ++i) {
-  snprintf(pCfg->nodeInfo[0].nodeFqdn, sizeof(pCfg->nodeInfo[0].nodeFqdn), "%s", "127.0.0.1");
+    pCfg->nodeInfo[i].nodePort = ports[i];
-  // taosGetFqdn(pCfg->nodeInfo[0].nodeFqdn);
+    snprintf(pCfg->nodeInfo[i].nodeFqdn, sizeof(pCfg->nodeInfo[i].nodeFqdn), "%s", "127.0.0.1");
-
+    // taosGetFqdn(pCfg->nodeInfo[0].nodeFqdn);
-  pCfg->nodeInfo[1].nodePort = ports[1];
+  }
-  snprintf(pCfg->nodeInfo[1].nodeFqdn, sizeof(pCfg->nodeInfo[1].nodeFqdn), "%s", "127.0.0.1");
-  // taosGetFqdn(pCfg->nodeInfo[1].nodeFqdn);
-
-  pCfg->nodeInfo[2].nodePort = ports[2];
-  snprintf(pCfg->nodeInfo[2].nodeFqdn, sizeof(pCfg->nodeInfo[2].nodeFqdn), "%s", "127.0.0.1");
-  // taosGetFqdn(pCfg->nodeInfo[2].nodeFqdn);
 
   SSyncNode* pSyncNode = syncNodeOpen(&syncInfo);
   assert(pSyncNode != NULL);
 
   gSyncIO->FpOnSyncPing = pSyncNode->FpOnPing;
+  gSyncIO->FpOnSyncPingReply = pSyncNode->FpOnPingReply;
+  gSyncIO->FpOnSyncRequestVote = pSyncNode->FpOnRequestVote;
+  gSyncIO->FpOnSyncRequestVoteReply = pSyncNode->FpOnRequestVoteReply;
+  gSyncIO->FpOnSyncAppendEntries = pSyncNode->FpOnAppendEntries;
+  gSyncIO->FpOnSyncAppendEntriesReply = pSyncNode->FpOnAppendEntriesReply;
+  gSyncIO->FpOnSyncTimeout = pSyncNode->FpOnTimeout;
   gSyncIO->pSyncNode = pSyncNode;
 
   return pSyncNode;
 }
 
-void timerPingAll(void* param, void* tmrId) {
+SSyncNode* syncInitTest() { return syncNodeInit(); }
-  SSyncNode* pSyncNode = (SSyncNode*)param;
+
-  syncNodePingAll(pSyncNode);
+void initRaftId(SSyncNode* pSyncNode) {
+  for (int i = 0; i < replicaNum; ++i) {
+    ids[i] = pSyncNode->replicasId[i];
+    char* s = syncUtilRaftId2Str(&ids[i]);
+    printf("raftId[%d] : %s\n", i, s);
+    free(s);
+  }
 }
 
 int main(int argc, char** argv) {
-  // taosInitLog((char*)"syncPingTest.log", 100000, 10);
+  // taosInitLog((char *)"syncTest.log", 100000, 10);
   tsAsyncLog = 0;
   sDebugFlag = 143 + 64;
 
-  logTest();
+  myIndex = 0;
-
-  int myIndex = 0;
   if (argc >= 2) {
     myIndex = atoi(argv[1]);
   }
@@ -75,30 +84,45 @@ int main(int argc, char** argv) {
   ret = syncEnvStart();
   assert(ret == 0);
 
-  SSyncNode* pSyncNode = doSync(myIndex);
+  SSyncNode* pSyncNode = syncInitTest();
-  gSyncIO->FpOnSyncPing = pSyncNode->FpOnPing;
+  assert(pSyncNode != NULL);
-  gSyncIO->FpOnSyncPingReply = pSyncNode->FpOnPingReply;
+  syncNodePrint((char*)"----1", pSyncNode);
-  gSyncIO->FpOnSyncTimeout = pSyncNode->FpOnTimeout;
 
+  initRaftId(pSyncNode);
+
+  //---------------------------
+
+  sTrace("syncNodeStartPingTimer ...");
   ret = syncNodeStartPingTimer(pSyncNode);
   assert(ret == 0);
+  syncNodePrint((char*)"----2", pSyncNode);
 
+  sTrace("sleep ...");
   taosMsleep(10000);
 
+  sTrace("syncNodeStopPingTimer ...");
   ret = syncNodeStopPingTimer(pSyncNode);
   assert(ret == 0);
+  syncNodePrint((char*)"----3", pSyncNode);
 
-  taosMsleep(10000);
+  sTrace("sleep ...");
+  taosMsleep(5000);
 
+  sTrace("syncNodeStartPingTimer ...");
   ret = syncNodeStartPingTimer(pSyncNode);
   assert(ret == 0);
+  syncNodePrint((char*)"----4", pSyncNode);
 
+  sTrace("sleep ...");
   taosMsleep(10000);
 
+  sTrace("syncNodeStopPingTimer ...");
   ret = syncNodeStopPingTimer(pSyncNode);
   assert(ret == 0);
+  syncNodePrint((char*)"----5", pSyncNode);
 
   while (1) {
+    sTrace("while 1 sleep ...");
     taosMsleep(1000);
   }
 

source/libs/sync/test/syncVotesGrantedTest.cpp

@@ -32,7 +32,7 @@ SSyncNode* syncNodeInit() {
   syncInfo.queue = gSyncIO->pMsgQ;
   syncInfo.FpEqMsg = syncIOEqMsg;
   syncInfo.pFsm = pFsm;
-  snprintf(syncInfo.path, sizeof(syncInfo.path), "%s", "./test_path");
+  snprintf(syncInfo.path, sizeof(syncInfo.path), "%s", "./");
 
   SSyncCfg* pCfg = &syncInfo.syncCfg;
   pCfg->myIndex = myIndex;

source/libs/sync/test/syncVotesRespondTest.cpp

@@ -32,7 +32,7 @@ SSyncNode* syncNodeInit() {
   syncInfo.queue = gSyncIO->pMsgQ;
   syncInfo.FpEqMsg = syncIOEqMsg;
   syncInfo.pFsm = pFsm;
-  snprintf(syncInfo.path, sizeof(syncInfo.path), "%s", "./test_path");
+  snprintf(syncInfo.path, sizeof(syncInfo.path), "%s", "./");
 
   SSyncCfg* pCfg = &syncInfo.syncCfg;
   pCfg->myIndex = myIndex;