sync refactor

source/libs/sync/inc/syncRaftLog.h

@@ -63,6 +63,9 @@ cJSON* logStore2Json(SSyncLogStore* pLogStore);
 
 char* logStore2Str(SSyncLogStore* pLogStore);
 
+// for debug
+void logStorePrint(SSyncLogStore* pLogStore);
+
 #ifdef __cplusplus
 }
 #endif

source/libs/sync/src/syncRaftLog.c

@@ -115,11 +115,44 @@ SSyncRaftEntry* logStoreGetLastEntry(SSyncLogStore* pLogStore) {
   return pEntry;
 }
 
-cJSON* logStore2Json(SSyncLogStore* pLogStore) {}
+cJSON* logStore2Json(SSyncLogStore* pLogStore) {
+  char u64buf[128];
+
+  SSyncLogStoreData* pData = (SSyncLogStoreData*)pLogStore->data;
+  cJSON*             pRoot = cJSON_CreateObject();
+  snprintf(u64buf, sizeof(u64buf), "%p", pData->pSyncNode);
+  cJSON_AddStringToObject(pRoot, "pSyncNode", u64buf);
+  snprintf(u64buf, sizeof(u64buf), "%p", pData->pWal);
+  cJSON_AddStringToObject(pRoot, "pWal", u64buf);
+  snprintf(u64buf, sizeof(u64buf), "%lu", logStoreLastIndex(pLogStore));
+  cJSON_AddStringToObject(pRoot, "LastIndex", u64buf);
+  snprintf(u64buf, sizeof(u64buf), "%lu", logStoreLastTerm(pLogStore));
+  cJSON_AddStringToObject(pRoot, "LastTerm", u64buf);
+
+  cJSON* pEntries = cJSON_CreateArray();
+  cJSON_AddItemToObject(pRoot, "pEntries", pEntries);
+  SyncIndex lastIndex = logStoreLastIndex(pLogStore);
+  for (SyncIndex i = 1; i <= lastIndex; ++i) {
+    SSyncRaftEntry* pEntry = logStoreGetEntry(pLogStore, i);
+    cJSON_AddItemToArray(pEntries, syncEntry2Json(pEntry));
+    syncEntryDestory(pEntry);
+  }
+
+  cJSON* pJson = cJSON_CreateObject();
+  cJSON_AddItemToObject(pJson, "SSyncLogStore", pRoot);
+  return pJson;
+}
 
 char* logStore2Str(SSyncLogStore* pLogStore) {
   cJSON* pJson = logStore2Json(pLogStore);
   char*  serialized = cJSON_Print(pJson);
   cJSON_Delete(pJson);
   return serialized;
+}
+
+// for debug
+void logStorePrint(SSyncLogStore* pLogStore) {
+  char* s = logStore2Str(pLogStore);
+  sTrace("%s", s);
+  free(s);
 }

source/libs/sync/test/CMakeLists.txt

@@ -15,6 +15,7 @@ add_executable(syncUtilTest "")
 add_executable(syncVotesGrantedTest "")
 add_executable(syncVotesRespondTest "")
 add_executable(syncIndexMgrTest "")
+add_executable(syncLogStoreTest "")
 
 
 target_sources(syncTest
@@ -85,6 +86,10 @@ target_sources(syncIndexMgrTest
     PRIVATE
     "syncIndexMgrTest.cpp"
 )
+target_sources(syncLogStoreTest
+    PRIVATE
+    "syncLogStoreTest.cpp"
+)
 
 
 target_include_directories(syncTest
@@ -172,6 +177,11 @@ target_include_directories(syncIndexMgrTest
     "${CMAKE_SOURCE_DIR}/include/libs/sync"
     "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
 )
+target_include_directories(syncLogStoreTest
+    PUBLIC
+    "${CMAKE_SOURCE_DIR}/include/libs/sync"
+    "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
+)
 
 
 target_link_libraries(syncTest
@@ -242,6 +252,10 @@ target_link_libraries(syncIndexMgrTest
     sync
     gtest_main
 )
+target_link_libraries(syncLogStoreTest
+    sync
+    gtest_main
+)
 
 
 enable_testing()

source/libs/sync/test/syncInitTest.cpp

@@ -53,6 +53,7 @@ SSyncNode* syncNodeInit() {
   gSyncIO->FpOnSyncAppendEntriesReply = pSyncNode->FpOnAppendEntriesReply;
   gSyncIO->FpOnSyncPing = pSyncNode->FpOnPing;
   gSyncIO->FpOnSyncPingReply = pSyncNode->FpOnPingReply;
+  gSyncIO->FpOnSyncTimeout = pSyncNode->FpOnTimeout;
   gSyncIO->pSyncNode = pSyncNode;
 
   return pSyncNode;
@@ -88,11 +89,11 @@ int main(int argc, char** argv) {
   SSyncNode* pSyncNode = syncInitTest();
   assert(pSyncNode != NULL);
 
-  char* serialized = syncNode2Str(pSyncNode);
+  syncNodePrint((char*)"syncInitTest", pSyncNode);
-  printf("%s\n", serialized);
-  free(serialized);
 
   initRaftId(pSyncNode);
 
+  //--------------------------------------------------------------
+
   return 0;
-}
+}

source/libs/sync/test/syncLogStoreTest.cpp

@@ -0,0 +1,132 @@
+#include <gtest/gtest.h>
+#include <stdio.h>
+#include "syncEnv.h"
+#include "syncIO.h"
+#include "syncInt.h"
+#include "syncRaftLog.h"
+#include "syncRaftStore.h"
+#include "syncUtil.h"
+
+void logTest() {
+  sTrace("--- sync log test: trace");
+  sDebug("--- sync log test: debug");
+  sInfo("--- sync log test: info");
+  sWarn("--- sync log test: warn");
+  sError("--- sync log test: error");
+  sFatal("--- sync log test: fatal");
+}
+
+uint16_t ports[] = {7010, 7110, 7210, 7310, 7410};
+int32_t  replicaNum = 1;
+int32_t  myIndex = 0;
+
+SRaftId    ids[TSDB_MAX_REPLICA];
+SSyncInfo  syncInfo;
+SSyncFSM*  pFsm;
+SWal*      pWal;
+SSyncNode* pSyncNode;
+
+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", "./");
+
+  SWalCfg walCfg;
+  memset(&walCfg, 0, sizeof(SWalCfg));
+  walCfg.vgId = syncInfo.vgId;
+  walCfg.fsyncPeriod = 1000;
+  walCfg.retentionPeriod = 1000;
+  walCfg.rollPeriod = 1000;
+  walCfg.retentionSize = 100000;
+  walCfg.segSize = 100000;
+  walCfg.level = TAOS_WAL_WRITE;
+  pWal = walOpen("./wal_test", &walCfg);
+
+  syncInfo.pWal = pWal;
+
+  SSyncCfg* pCfg = &syncInfo.syncCfg;
+  pCfg->myIndex = myIndex;
+  pCfg->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);
+
+  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->FpOnSyncTimeout = pSyncNode->FpOnTimeout;
+  gSyncIO->pSyncNode = pSyncNode;
+
+  return pSyncNode;
+}
+
+SSyncNode* syncInitTest() { return syncNodeInit(); }
+
+void logStoreTest() {
+  logStorePrint(pSyncNode->pLogStore);
+  for (int i = 0; i < 5; ++i) {
+    SSyncRaftEntry* pEntry;
+    pSyncNode->pLogStore->appendEntry(pSyncNode->pLogStore, pEntry);
+  }
+  logStorePrint(pSyncNode->pLogStore);
+
+  pSyncNode->pLogStore->truncate(pSyncNode->pLogStore, 3);
+  logStorePrint(pSyncNode->pLogStore);
+}
+
+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;
+
+  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);
+
+  pSyncNode = syncInitTest();
+  assert(pSyncNode != NULL);
+
+  syncNodePrint((char*)"syncLogStoreTest", pSyncNode);
+
+  initRaftId(pSyncNode);
+
+  //--------------------------------------------------------------
+
+  logStoreTest();
+
+  //--------------------------------------------------------------
+  // walClose(pWal);
+
+  return 0;
+}