sync refactor

source/libs/sync/inc/syncInt.h

@@ -193,6 +193,7 @@ 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);

source/libs/sync/inc/syncUtil.h

@@ -47,6 +47,7 @@ int32_t     syncUtilElectRandomMS();
 int32_t     syncUtilQuorum(int32_t replicaNum);
 cJSON*      syncUtilNodeInfo2Json(const SNodeInfo* p);
 cJSON*      syncUtilRaftId2Json(const SRaftId* p);
+char*       syncUtilRaftId2Str(const SRaftId* p);
 const char* syncUtilState2String(ESyncState state);
 
 #ifdef __cplusplus

source/libs/sync/src/syncMain.c

@@ -291,6 +291,13 @@ cJSON* syncNode2Json(const SSyncNode* pSyncNode) {
   return pJson;
 }
 
+char* syncNode2Str(const SSyncNode* pSyncNode) {
+  cJSON* pJson = syncNode2Json(pSyncNode);
+  char*  serialized = cJSON_Print(pJson);
+  cJSON_Delete(pJson);
+  return serialized;
+}
+
 int32_t syncNodeSendMsgById(const SRaftId* destRaftId, SSyncNode* pSyncNode, SRpcMsg* pMsg) {
   SEpSet epSet;
   syncUtilraftId2EpSet(destRaftId, &epSet);

source/libs/sync/src/syncUtil.c

@@ -131,6 +131,13 @@ cJSON* syncUtilRaftId2Json(const SRaftId* p) {
   return pJson;
 }
 
+char* syncUtilRaftId2Str(const SRaftId* p) {
+  cJSON* pJson = syncUtilRaftId2Json(p);
+  char*  serialized = cJSON_Print(pJson);
+  cJSON_Delete(pJson);
+  return serialized;
+}
+
 const char* syncUtilState2String(ESyncState state) {
   if (state == TAOS_SYNC_STATE_FOLLOWER) {
     return "TAOS_SYNC_STATE_FOLLOWER";

source/libs/sync/src/syncVoteMgr.c

@@ -63,6 +63,7 @@ void voteGrantedVote(SVotesGranted *pVotesGranted, SyncRequestVoteReply *pMsg) {
     }
   }
   assert(j != -1);
+  assert(j >= 0 && j < pVotesGranted->replicaNum);
 
   if (pVotesGranted->isGranted[j] != true) {
     ++(pVotesGranted->votes);
@@ -87,6 +88,14 @@ cJSON *voteGranted2Json(SVotesGranted *pVotesGranted) {
   for (int i = 0; i < pVotesGranted->replicaNum; ++i) {
     cJSON_AddItemToArray(pReplicas, syncUtilRaftId2Json(&(*(pVotesGranted->replicas)[i])));
   }
+  int *arr = (int *)malloc(sizeof(int) * pVotesGranted->replicaNum);
+  for (int i = 0; i < pVotesGranted->replicaNum; ++i) {
+    arr[i] = pVotesGranted->isGranted[i];
+  }
+  cJSON *pIsGranted = cJSON_CreateIntArray(arr, pVotesGranted->replicaNum);
+  free(arr);
+  cJSON_AddItemToObject(pRoot, "isGranted", pIsGranted);
+
   cJSON_AddNumberToObject(pRoot, "votes", pVotesGranted->votes);
   snprintf(u64buf, sizeof(u64buf), "%lu", pVotesGranted->term);
   cJSON_AddStringToObject(pRoot, "term", u64buf);
@@ -95,6 +104,9 @@ cJSON *voteGranted2Json(SVotesGranted *pVotesGranted) {
   snprintf(u64buf, sizeof(u64buf), "%p", pVotesGranted->pSyncNode);
   cJSON_AddStringToObject(pRoot, "pSyncNode", u64buf);
 
+  bool majority = voteGrantedMajority(pVotesGranted);
+  cJSON_AddNumberToObject(pRoot, "majority", majority);
+
   cJSON *pJson = cJSON_CreateObject();
   cJSON_AddItemToObject(pJson, "SVotesGranted", pRoot);
   return pJson;
@@ -102,7 +114,7 @@ cJSON *voteGranted2Json(SVotesGranted *pVotesGranted) {
 
 char *voteGranted2Str(SVotesGranted *pVotesGranted) {
   cJSON *pJson = voteGranted2Json(pVotesGranted);
-  char * serialized = cJSON_Print(pJson);
+  char  *serialized = cJSON_Print(pJson);
   cJSON_Delete(pJson);
   return serialized;
 }

source/libs/sync/test/syncInitTest.cpp

 #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,13 +15,16 @@ 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 = 5;
+int32_t  myIndex = 0;
 
-SSyncNode* syncInitTest() {
-  SSyncFSM* pFsm;
+SRaftId   ids[TSDB_MAX_REPLICA];
+SSyncInfo syncInfo;
+SSyncFSM* pFsm;
 
-  SSyncInfo syncInfo;
-  syncInfo.vgId = 1;
+SSyncNode* syncNodeInit() {
+  syncInfo.vgId = 1234;
   syncInfo.rpcClient = gSyncIO->clientRpc;
   syncInfo.FpSendMsg = syncIOSendMsg;
   syncInfo.queue = gSyncIO->pMsgQ;
@@ -29,46 +34,66 @@ SSyncNode* syncInitTest() {
   snprintf(syncInfo.walPath, sizeof(syncInfo.walPath), "%s", "./test_wal_path");
 
   SSyncCfg* pCfg = &syncInfo.syncCfg;
-  pCfg->myIndex = 0;
-  pCfg->replicaNum = 3;
+  pCfg->myIndex = myIndex;
+  pCfg->replicaNum = replicaNum;
 
-  pCfg->nodeInfo[0].nodePort = ports[0];
-  snprintf(pCfg->nodeInfo[0].nodeFqdn, sizeof(pCfg->nodeInfo[0].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);
+  for (int i = 0; i < replicaNum; ++i) {
+    pCfg->nodeInfo[i].nodePort = ports[i];
+    snprintf(pCfg->nodeInfo[i].nodeFqdn, sizeof(pCfg->nodeInfo[i].nodeFqdn), "%s", "127.0.0.1");
+    // taosGetFqdn(pCfg->nodeInfo[0].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->FpOnSyncPing = pSyncNode->FpOnPing;
+  gSyncIO->FpOnSyncPingReply = pSyncNode->FpOnPingReply;
   gSyncIO->pSyncNode = pSyncNode;
 
   return pSyncNode;
 }
 
-int main() {
+SSyncNode* syncInitTest() { return syncNodeInit(); }
+
+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 *)"syncTest.log", 100000, 10);
   tsAsyncLog = 0;
   sDebugFlag = 143 + 64;
 
-  int32_t ret = syncIOStart((char*)"127.0.0.1", ports[0]);
+  myIndex = 0;
+  if (argc >= 2) {
+    myIndex = atoi(argv[1]);
+  }
+
+  int32_t ret = syncIOStart((char*)"127.0.0.1", ports[myIndex]);
+  assert(ret == 0);
+
+  ret = syncEnvStart();
   assert(ret == 0);
 
   SSyncNode* pSyncNode = syncInitTest();
   assert(pSyncNode != NULL);
 
-  cJSON* pJson = syncNode2Json(pSyncNode);
-  char*  serialized = cJSON_Print(pJson);
+  char* serialized = syncNode2Str(pSyncNode);
   printf("%s\n", serialized);
   free(serialized);
-  cJSON_Delete(pJson);
+
+  initRaftId(pSyncNode);
 
   return 0;
 }

source/libs/sync/test/syncPingTest.cpp

@@ -25,7 +25,8 @@ SSyncNode* doSync(int myIndex) {
   syncInfo.queue = gSyncIO->pMsgQ;
   syncInfo.FpEqMsg = syncIOEqMsg;
   syncInfo.pFsm = pFsm;
-  snprintf(syncInfo.path, sizeof(syncInfo.path), "%s", "./test_sync_ping");
+  snprintf(syncInfo.path, sizeof(syncInfo.path), "%s", "./path");
+  snprintf(syncInfo.walPath, sizeof(syncInfo.walPath), "%s", "./wal_path");
 
   SSyncCfg* pCfg = &syncInfo.syncCfg;
   pCfg->myIndex = myIndex;

source/libs/sync/test/syncVotesGrantedTest.cpp

@@ -4,6 +4,7 @@
 #include "syncIO.h"
 #include "syncInt.h"
 #include "syncRaftStore.h"
+#include "syncUtil.h"
 #include "syncVoteMgr.h"
 
 void logTest() {
@@ -15,66 +16,141 @@ 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) {
-  SSyncFSM* pFsm;
+SRaftId    ids[TSDB_MAX_REPLICA];
+SSyncInfo  syncInfo;
+SSyncFSM*  pFsm;
+SSyncNode* pSyncNode;
 
-  SSyncInfo syncInfo;
-  syncInfo.vgId = 1;
+SSyncNode* syncNodeInit() {
+  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", "./test_sync_ping");
+  snprintf(syncInfo.path, sizeof(syncInfo.path), "%s", "./test_path");
+  snprintf(syncInfo.walPath, sizeof(syncInfo.walPath), "%s", "./test_wal_path");
 
   SSyncCfg* pCfg = &syncInfo.syncCfg;
   pCfg->myIndex = myIndex;
-  pCfg->replicaNum = 3;
+  pCfg->replicaNum = replicaNum;
 
-  pCfg->nodeInfo[0].nodePort = ports[0];
-  snprintf(pCfg->nodeInfo[0].nodeFqdn, sizeof(pCfg->nodeInfo[0].nodeFqdn), "%s", "127.0.0.1");
-  // taosGetFqdn(pCfg->nodeInfo[0].nodeFqdn);
+  for (int i = 0; i < replicaNum; ++i) {
+    pCfg->nodeInfo[i].nodePort = ports[i];
+    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);
+  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->FpOnSyncPing = pSyncNode->FpOnPing;
+  gSyncIO->FpOnSyncPingReply = pSyncNode->FpOnPingReply;
   gSyncIO->pSyncNode = pSyncNode;
 
   return pSyncNode;
 }
 
-int main() {
+SSyncNode* syncInitTest() { return syncNodeInit(); }
+
+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 *)"syncTest.log", 100000, 10);
   tsAsyncLog = 0;
   sDebugFlag = 143 + 64;
-  logTest();
 
-  int     myIndex = 0;
+  myIndex = 0;
+  if (argc >= 2) {
+    myIndex = atoi(argv[1]);
+  }
+
   int32_t ret = syncIOStart((char*)"127.0.0.1", ports[myIndex]);
   assert(ret == 0);
 
   ret = syncEnvStart();
   assert(ret == 0);
 
-  SSyncNode*     pSyncNode = doSync(myIndex);
-  SVotesGranted* pVotesGranted = voteGrantedCreate(pSyncNode);
-  assert(pVotesGranted != NULL);
+  SSyncNode* pSyncNode = syncInitTest();
+  assert(pSyncNode != NULL);
 
-  char* serialized = voteGranted2Str(pVotesGranted);
-  assert(serialized != NULL);
+  char* serialized = syncNode2Str(pSyncNode);
   printf("%s\n", serialized);
   free(serialized);
 
+  initRaftId(pSyncNode);
+
+  SVotesGranted* pVotesGranted = voteGrantedCreate(pSyncNode);
+  assert(pVotesGranted != NULL);
+
+  printf("---------------------------------------\n");
+  {
+    char* serialized = voteGranted2Str(pVotesGranted);
+    assert(serialized != NULL);
+    printf("%s\n", serialized);
+    free(serialized);
+  }
+
+  SyncTerm term = 1234;
+  printf("---------------------------------------\n");
+  voteGrantedReset(pVotesGranted, term);
+  {
+    char* serialized = voteGranted2Str(pVotesGranted);
+    assert(serialized != NULL);
+    printf("%s\n", serialized);
+    free(serialized);
+  }
+
+  for (int i = 0; i < replicaNum; ++i) {
+    SyncRequestVoteReply* reply = SyncRequestVoteReplyBuild();
+    reply->destId = pSyncNode->myRaftId;
+    reply->srcId = ids[i];
+    reply->term = term;
+    reply->voteGranted = true;
+
+    voteGrantedVote(pVotesGranted, reply);
+    {
+      char* serialized = voteGranted2Str(pVotesGranted);
+      assert(serialized != NULL);
+      printf("%s\n", serialized);
+      free(serialized);
+    }
+
+    voteGrantedVote(pVotesGranted, reply);
+    {
+      char* serialized = voteGranted2Str(pVotesGranted);
+      assert(serialized != NULL);
+      printf("%s\n", serialized);
+      free(serialized);
+    }
+  }
+
+  printf("---------------------------------------\n");
+  voteGrantedReset(pVotesGranted, 123456789);
+  {
+    char* serialized = voteGranted2Str(pVotesGranted);
+    assert(serialized != NULL);
+    printf("%s\n", serialized);
+    free(serialized);
+  }
+
+  voteGrantedDestroy(pVotesGranted);
   return 0;
 }