enh(sync): add term into index mgr

source/libs/sync/inc/syncIndexMgr.h

@@ -30,6 +30,7 @@ extern "C" {
 typedef struct SSyncIndexMgr {
   SRaftId (*replicas)[TSDB_MAX_REPLICA];
   SyncIndex  index[TSDB_MAX_REPLICA];
+  SyncTerm   privateTerm[TSDB_MAX_REPLICA];  // for advanced function
   int32_t    replicaNum;
   SSyncNode *pSyncNode;
 } SSyncIndexMgr;
@@ -40,8 +41,11 @@ void           syncIndexMgrDestroy(SSyncIndexMgr *pSyncIndexMgr);
 void           syncIndexMgrClear(SSyncIndexMgr *pSyncIndexMgr);
 void           syncIndexMgrSetIndex(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId, SyncIndex index);
 SyncIndex      syncIndexMgrGetIndex(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId);
-cJSON *        syncIndexMgr2Json(SSyncIndexMgr *pSyncIndexMgr);
-char *         syncIndexMgr2Str(SSyncIndexMgr *pSyncIndexMgr);
+cJSON         *syncIndexMgr2Json(SSyncIndexMgr *pSyncIndexMgr);
+char          *syncIndexMgr2Str(SSyncIndexMgr *pSyncIndexMgr);
+
+void     syncIndexMgrSetTerm(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId, SyncTerm term);
+SyncTerm syncIndexMgrGetTerm(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId);
 
 // for debug -------------------
 void syncIndexMgrPrint(SSyncIndexMgr *pObj);

source/libs/sync/src/syncIndexMgr.c

@@ -46,6 +46,7 @@ void syncIndexMgrDestroy(SSyncIndexMgr *pSyncIndexMgr) {
 
 void syncIndexMgrClear(SSyncIndexMgr *pSyncIndexMgr) {
   memset(pSyncIndexMgr->index, 0, sizeof(pSyncIndexMgr->index));
+  memset(pSyncIndexMgr->privateTerm, 0, sizeof(pSyncIndexMgr->privateTerm));
   /*
   for (int i = 0; i < pSyncIndexMgr->replicaNum; ++i) {
     pSyncIndexMgr->index[i] = 0;
@@ -62,7 +63,7 @@ void syncIndexMgrSetIndex(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId,
   }
 
   // maybe config change
-  // assert(0);
+  assert(0);
 }
 
 SyncIndex syncIndexMgrGetIndex(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId) {
@@ -86,14 +87,27 @@ cJSON *syncIndexMgr2Json(SSyncIndexMgr *pSyncIndexMgr) {
     for (int i = 0; i < pSyncIndexMgr->replicaNum; ++i) {
       cJSON_AddItemToArray(pReplicas, syncUtilRaftId2Json(&(*(pSyncIndexMgr->replicas))[i]));
     }
-    int  respondNum = 0;
-    int *arr = (int *)taosMemoryMalloc(sizeof(int) * pSyncIndexMgr->replicaNum);
-    for (int i = 0; i < pSyncIndexMgr->replicaNum; ++i) {
-      arr[i] = pSyncIndexMgr->index[i];
+
+    {
+      int *arr = (int *)taosMemoryMalloc(sizeof(int) * pSyncIndexMgr->replicaNum);
+      for (int i = 0; i < pSyncIndexMgr->replicaNum; ++i) {
+        arr[i] = pSyncIndexMgr->index[i];
+      }
+      cJSON *pIndex = cJSON_CreateIntArray(arr, pSyncIndexMgr->replicaNum);
+      taosMemoryFree(arr);
+      cJSON_AddItemToObject(pRoot, "index", pIndex);
     }
-    cJSON *pIndex = cJSON_CreateIntArray(arr, pSyncIndexMgr->replicaNum);
-    taosMemoryFree(arr);
-    cJSON_AddItemToObject(pRoot, "index", pIndex);
+
+    {
+      int *arr = (int *)taosMemoryMalloc(sizeof(int) * pSyncIndexMgr->replicaNum);
+      for (int i = 0; i < pSyncIndexMgr->replicaNum; ++i) {
+        arr[i] = pSyncIndexMgr->privateTerm[i];
+      }
+      cJSON *pIndex = cJSON_CreateIntArray(arr, pSyncIndexMgr->replicaNum);
+      taosMemoryFree(arr);
+      cJSON_AddItemToObject(pRoot, "privateTerm", pIndex);
+    }
+
     snprintf(u64buf, sizeof(u64buf), "%p", pSyncIndexMgr->pSyncNode);
     cJSON_AddStringToObject(pRoot, "pSyncNode", u64buf);
   }
@@ -105,7 +119,7 @@ cJSON *syncIndexMgr2Json(SSyncIndexMgr *pSyncIndexMgr) {
 
 char *syncIndexMgr2Str(SSyncIndexMgr *pSyncIndexMgr) {
   cJSON *pJson = syncIndexMgr2Json(pSyncIndexMgr);
-  char * serialized = cJSON_Print(pJson);
+  char  *serialized = cJSON_Print(pJson);
   cJSON_Delete(pJson);
   return serialized;
 }
@@ -135,4 +149,26 @@ void syncIndexMgrLog2(char *s, SSyncIndexMgr *pObj) {
   char *serialized = syncIndexMgr2Str(pObj);
   sTrace("syncIndexMgrLog2 | len:%lu | %s | %s", strlen(serialized), s, serialized);
   taosMemoryFree(serialized);
+}
+
+void syncIndexMgrSetTerm(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId, SyncTerm term) {
+  for (int i = 0; i < pSyncIndexMgr->replicaNum; ++i) {
+    if (syncUtilSameId(&((*(pSyncIndexMgr->replicas))[i]), pRaftId)) {
+      (pSyncIndexMgr->privateTerm)[i] = term;
+      return;
+    }
+  }
+
+  // maybe config change
+  assert(0);
+}
+
+SyncTerm syncIndexMgrGetTerm(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId) {
+  for (int i = 0; i < pSyncIndexMgr->replicaNum; ++i) {
+    if (syncUtilSameId(&((*(pSyncIndexMgr->replicas))[i]), pRaftId)) {
+      SyncTerm term = (pSyncIndexMgr->privateTerm)[i];
+      return term;
+    }
+  }
+  assert(0);
 }
\ No newline at end of file

source/libs/sync/test/syncIndexMgrTest.cpp

@@ -22,55 +22,24 @@ int32_t  replicaNum = 3;
 int32_t  myIndex = 0;
 
 SRaftId    ids[TSDB_MAX_REPLICA];
-SSyncInfo  syncInfo;
-SSyncFSM*  pFsm;
 SSyncNode* pSyncNode;
 
 SSyncNode* syncNodeInit() {
-  syncInfo.vgId = 1234;
-  syncInfo.msgcb = &gSyncIO->msgcb;
-  syncInfo.FpSendMsg = syncIOSendMsg;
-  syncInfo.FpEqMsg = syncIOEqMsg;
-  syncInfo.pFsm = pFsm;
-  snprintf(syncInfo.path, sizeof(syncInfo.path), "%s", "./");
-
-  SSyncCfg* pCfg = &syncInfo.syncCfg;
-  pCfg->myIndex = myIndex;
-  pCfg->replicaNum = replicaNum;
 
+  pSyncNode = (SSyncNode*)taosMemoryMalloc(sizeof(SSyncNode));
+  memset(pSyncNode, 0, sizeof(SSyncNode));
+  pSyncNode->replicaNum = replicaNum;
   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);
-  }
-
-  pSyncNode = syncNodeOpen(&syncInfo);
-  assert(pSyncNode != NULL);
+    pSyncNode->replicasId[i].addr = syncUtilAddr2U64("127.0.0.1", ports[i]);
+    pSyncNode->replicasId[i].vgId = 1234;
 
-  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;
+    ids[i].addr = pSyncNode->replicasId[i].addr;
+    ids[i].vgId = pSyncNode->replicasId[i].vgId;
+  }
 
   return pSyncNode;
 }
 
-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);
-    taosMemoryFree(s);
-  }
-}
-
 int main(int argc, char** argv) {
   tsAsyncLog = 0;
   sDebugFlag = DEBUG_TRACE + DEBUG_SCREEN + DEBUG_FILE;
@@ -80,58 +49,52 @@ int main(int argc, char** argv) {
     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();
+  SSyncNode* pSyncNode = syncNodeInit();
   assert(pSyncNode != NULL);
 
-  char* serialized = syncNode2Str(pSyncNode);
-  printf("%s\n", serialized);
-  taosMemoryFree(serialized);
-
-  initRaftId(pSyncNode);
-
+  printf("---------------------------------------\n");
   SSyncIndexMgr* pSyncIndexMgr = syncIndexMgrCreate(pSyncNode);
   assert(pSyncIndexMgr != NULL);
-
-  printf("---------------------------------------\n");
   {
     char* serialized = syncIndexMgr2Str(pSyncIndexMgr);
     assert(serialized != NULL);
     printf("%s\n", serialized);
     taosMemoryFree(serialized);
   }
+  printf("---------------------------------------\n");
 
+  printf("---------------------------------------\n");
   syncIndexMgrSetIndex(pSyncIndexMgr, &ids[0], 100);
   syncIndexMgrSetIndex(pSyncIndexMgr, &ids[1], 200);
   syncIndexMgrSetIndex(pSyncIndexMgr, &ids[2], 300);
-
-  printf("---------------------------------------\n");
+  syncIndexMgrSetTerm(pSyncIndexMgr, &ids[0], 700);
+  syncIndexMgrSetTerm(pSyncIndexMgr, &ids[1], 800);
+  syncIndexMgrSetTerm(pSyncIndexMgr, &ids[2], 900);
   {
     char* serialized = syncIndexMgr2Str(pSyncIndexMgr);
     assert(serialized != NULL);
     printf("%s\n", serialized);
     taosMemoryFree(serialized);
   }
+  printf("---------------------------------------\n");
 
   printf("---------------------------------------\n");
   for (int i = 0; i < pSyncIndexMgr->replicaNum; ++i) {
     SyncIndex idx = syncIndexMgrGetIndex(pSyncIndexMgr, &ids[i]);
-    printf("index %d : %lu \n", i, idx);
+    SyncTerm term = syncIndexMgrGetTerm(pSyncIndexMgr, &ids[i]);
+    printf("%d: index:%ld term:%lu \n", i, idx, term);
   }
+  printf("---------------------------------------\n");
 
-  syncIndexMgrClear(pSyncIndexMgr);
   printf("---------------------------------------\n");
+  syncIndexMgrClear(pSyncIndexMgr);
   {
     char* serialized = syncIndexMgr2Str(pSyncIndexMgr);
     assert(serialized != NULL);
     printf("%s\n", serialized);
     taosMemoryFree(serialized);
   }
+  printf("---------------------------------------\n");
 
   syncIndexMgrDestroy(pSyncIndexMgr);
   return 0;