Merge pull request #18062 from taosdata/fix/TD-20052

refact: adjust head file and sync log

include/libs/sync/sync.h

@@ -58,7 +58,6 @@ typedef int64_t  SyncIndex;
 typedef uint64_t SyncTerm;
 
 typedef struct SSyncNode      SSyncNode;
-typedef struct SSyncBuffer    SSyncBuffer;
 typedef struct SWal           SWal;
 typedef struct SSyncRaftEntry SSyncRaftEntry;
 

source/libs/sync/inc/syncAppendEntries.h

@@ -21,7 +21,6 @@ extern "C" {
 #endif
 
 #include "syncInt.h"
-#include "syncMessage.h"
 
 // TLA+ Spec
 // HandleAppendEntriesRequest(i, j, m) ==

source/libs/sync/inc/syncAppendEntriesReply.h

@@ -21,7 +21,6 @@ extern "C" {
 #endif
 
 #include "syncInt.h"
-#include "syncMessage.h"
 
 // TLA+ Spec
 // HandleAppendEntriesResponse(i, j, m) ==

source/libs/sync/inc/syncIndexMgr.h

@@ -41,22 +41,13 @@ 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);
 
-void    syncIndexMgrSetStartTime(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId, int64_t startTime);
-int64_t syncIndexMgrGetStartTime(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId);
-void    syncIndexMgrSetRecvTime(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId, int64_t recvTime);
-int64_t syncIndexMgrGetRecvTime(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId);
-
-// void     syncIndexMgrSetTerm(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId, SyncTerm term);
-// SyncTerm syncIndexMgrGetTerm(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId);
-
-// for debug -------------------
-void syncIndexMgrPrint(SSyncIndexMgr *pObj);
-void syncIndexMgrPrint2(char *s, SSyncIndexMgr *pObj);
-void syncIndexMgrLog(SSyncIndexMgr *pObj);
-void syncIndexMgrLog2(char *s, SSyncIndexMgr *pObj);
+void     syncIndexMgrSetStartTime(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId, int64_t startTime);
+int64_t  syncIndexMgrGetStartTime(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId);
+void     syncIndexMgrSetRecvTime(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId, int64_t recvTime);
+int64_t  syncIndexMgrGetRecvTime(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId);
+void     syncIndexMgrSetTerm(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId, SyncTerm term);
+SyncTerm syncIndexMgrGetTerm(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId);
 
 #ifdef __cplusplus
 }

source/libs/sync/inc/syncInt.h

@@ -21,7 +21,6 @@ extern "C" {
 #endif
 
 #include "sync.h"
-#include "syncTools.h"
 #include "taosdef.h"
 #include "tlog.h"
 #include "trpc.h"
@@ -85,9 +84,33 @@ typedef struct SSyncSnapshotSender    SSyncSnapshotSender;
 typedef struct SSyncSnapshotReceiver  SSyncSnapshotReceiver;
 typedef struct SSyncTimer             SSyncTimer;
 typedef struct SSyncHbTimerData       SSyncHbTimerData;
+typedef struct SyncSnapshotSend       SyncSnapshotSend;
+typedef struct SyncSnapshotRsp        SyncSnapshotRsp;
+typedef struct SyncLocalCmd           SyncLocalCmd;
+typedef struct SyncAppendEntriesBatch SyncAppendEntriesBatch;
+typedef struct SyncPreSnapshotReply   SyncPreSnapshotReply;
+typedef struct SyncHeartbeatReply     SyncHeartbeatReply;
+typedef struct SyncHeartbeat          SyncHeartbeat;
+typedef struct SyncPreSnapshot        SyncPreSnapshot;
+
+typedef int32_t (*FpOnPingCb)(SSyncNode* ths, SyncPing* pMsg);
+typedef int32_t (*FpOnPingReplyCb)(SSyncNode* ths, SyncPingReply* pMsg);
+typedef int32_t (*FpOnClientRequestCb)(SSyncNode* ths, SRpcMsg* pMsg, SyncIndex* pRetIndex);
+typedef int32_t (*FpOnRequestVoteCb)(SSyncNode* ths, SyncRequestVote* pMsg);
+typedef int32_t (*FpOnRequestVoteReplyCb)(SSyncNode* ths, SyncRequestVoteReply* pMsg);
+typedef int32_t (*FpOnAppendEntriesCb)(SSyncNode* ths, SyncAppendEntries* pMsg);
+typedef int32_t (*FpOnAppendEntriesReplyCb)(SSyncNode* ths, SyncAppendEntriesReply* pMsg);
+typedef int32_t (*FpOnTimeoutCb)(SSyncNode* pSyncNode, SyncTimeout* pMsg);
+typedef int32_t (*FpOnSnapshotCb)(SSyncNode* ths, SyncSnapshotSend* pMsg);
+typedef int32_t (*FpOnSnapshotReplyCb)(SSyncNode* ths, SyncSnapshotRsp* pMsg);
 
 extern bool gRaftDetailLog;
 
+typedef struct SRaftId {
+  SyncNodeId  addr;
+  SyncGroupId vgId;
+} SRaftId;
+
 typedef struct SSyncHbTimerData {
   SSyncNode*  pSyncNode;
   SSyncTimer* pTimer;

source/libs/sync/src/syncAppendEntries.c

@@ -13,15 +13,11 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
+#define _DEFAULT_SOURCE
 #include "syncAppendEntries.h"
-#include "syncInt.h"
-#include "syncRaftCfg.h"
+#include "syncMessage.h"
 #include "syncRaftLog.h"
 #include "syncRaftStore.h"
-#include "syncSnapshot.h"
-#include "syncUtil.h"
-#include "syncVoteMgr.h"
-#include "wal.h"
 
 // TLA+ Spec
 // HandleAppendEntriesRequest(i, j, m) ==

source/libs/sync/src/syncAppendEntriesReply.c

@@ -13,17 +13,14 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
+#define _DEFAULT_SOURCE
 #include "syncAppendEntriesReply.h"
+#include "syncMessage.h"
 #include "syncCommit.h"
 #include "syncIndexMgr.h"
-#include "syncInt.h"
-#include "syncRaftCfg.h"
-#include "syncRaftLog.h"
 #include "syncRaftStore.h"
 #include "syncReplication.h"
 #include "syncSnapshot.h"
-#include "syncUtil.h"
-#include "syncVoteMgr.h"
 
 // TLA+ Spec
 // HandleAppendEntriesResponse(i, j, m) ==

source/libs/sync/src/syncCommit.c

@@ -13,10 +13,9 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
+#define _DEFAULT_SOURCE
 #include "syncCommit.h"
 #include "syncIndexMgr.h"
-#include "syncInt.h"
-#include "syncRaftCfg.h"
 #include "syncRaftLog.h"
 #include "syncRaftStore.h"
 #include "syncUtil.h"

source/libs/sync/src/syncElection.c

@@ -18,8 +18,8 @@
 #include "syncMessage.h"
 #include "syncRaftCfg.h"
 #include "syncRaftStore.h"
-#include "syncVoteMgr.h"
 #include "syncUtil.h"
+#include "syncVoteMgr.h"
 
 // TLA+ Spec
 // RequestVote(i, j) ==

source/libs/sync/src/syncEnv.c

@@ -105,6 +105,7 @@ void syncEnvStopTimer() {
 #endif
 
 static void syncEnvTick(void *param, void *tmrId) {
+#if 0
   SSyncEnv *pSyncEnv = param;
   if (atomic_load_64(&gSyncEnv.envTickTimerLogicClockUser) <= atomic_load_64(&gSyncEnv.envTickTimerLogicClock)) {
     gSyncEnv.envTickTimerCounter++;
@@ -121,4 +122,5 @@ static void syncEnvTick(void *param, void *tmrId) {
            gSyncEnv.envTickTimerLogicClockUser, gSyncEnv.envTickTimerLogicClock, gSyncEnv.envTickTimerCounter,
            gSyncEnv.envTickTimerMS, tmrId);
   }
+#endif
 }

source/libs/sync/src/syncIndexMgr.c

@@ -13,18 +13,16 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
+#define _DEFAULT_SOURCE
 #include "syncIndexMgr.h"
 #include "syncUtil.h"
 
-// SMatchIndex -----------------------------
-
 SSyncIndexMgr *syncIndexMgrCreate(SSyncNode *pSyncNode) {
-  SSyncIndexMgr *pSyncIndexMgr = taosMemoryMalloc(sizeof(SSyncIndexMgr));
+  SSyncIndexMgr *pSyncIndexMgr = taosMemoryCalloc(1, sizeof(SSyncIndexMgr));
   if (pSyncIndexMgr == NULL) {
     terrno = TSDB_CODE_OUT_OF_MEMORY;
     return NULL;
   }
-  memset(pSyncIndexMgr, 0, sizeof(SSyncIndexMgr));
 
   pSyncIndexMgr->replicas = &(pSyncNode->replicasId);
   pSyncIndexMgr->replicaNum = pSyncNode->replicaNum;
@@ -97,54 +95,6 @@ SyncIndex syncIndexMgrGetIndex(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaf
   return SYNC_INDEX_INVALID;
 }
 
-cJSON *syncIndexMgr2Json(SSyncIndexMgr *pSyncIndexMgr) {
-  char   u64buf[128] = {0};
-  cJSON *pRoot = cJSON_CreateObject();
-
-  if (pSyncIndexMgr != NULL) {
-    cJSON_AddNumberToObject(pRoot, "replicaNum", pSyncIndexMgr->replicaNum);
-    cJSON *pReplicas = cJSON_CreateArray();
-    cJSON_AddItemToObject(pRoot, "replicas", pReplicas);
-    for (int i = 0; i < pSyncIndexMgr->replicaNum; ++i) {
-      cJSON_AddItemToArray(pReplicas, syncUtilRaftId2Json(&(*(pSyncIndexMgr->replicas))[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);
-    }
-
-    {
-      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);
-  }
-
-  cJSON *pJson = cJSON_CreateObject();
-  cJSON_AddItemToObject(pJson, "pSyncIndexMgr", pRoot);
-  return pJson;
-}
-
-char *syncIndexMgr2Str(SSyncIndexMgr *pSyncIndexMgr) {
-  cJSON *pJson = syncIndexMgr2Json(pSyncIndexMgr);
-  char  *serialized = cJSON_Print(pJson);
-  cJSON_Delete(pJson);
-  return serialized;
-}
-
 void syncIndexMgrSetStartTime(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId, int64_t startTime) {
   for (int i = 0; i < pSyncIndexMgr->replicaNum; ++i) {
     if (syncUtilSameId(&((*(pSyncIndexMgr->replicas))[i]), pRaftId)) {
@@ -201,35 +151,6 @@ int64_t syncIndexMgrGetRecvTime(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRa
   return -1;
 }
 
-// for debug -------------------
-void syncIndexMgrPrint(SSyncIndexMgr *pObj) {
-  char *serialized = syncIndexMgr2Str(pObj);
-  printf("syncIndexMgrPrint | len:%" PRIu64 " | %s \n", (uint64_t)strlen(serialized), serialized);
-  fflush(NULL);
-  taosMemoryFree(serialized);
-}
-
-void syncIndexMgrPrint2(char *s, SSyncIndexMgr *pObj) {
-  char *serialized = syncIndexMgr2Str(pObj);
-  printf("syncIndexMgrPrint2 | len:%" PRIu64 " | %s | %s \n", (uint64_t)strlen(serialized), s, serialized);
-  fflush(NULL);
-  taosMemoryFree(serialized);
-}
-
-void syncIndexMgrLog(SSyncIndexMgr *pObj) {
-  char *serialized = syncIndexMgr2Str(pObj);
-  sTrace("syncIndexMgrLog | len:%" PRIu64 " | %s", (uint64_t)strlen(serialized), serialized);
-  taosMemoryFree(serialized);
-}
-
-void syncIndexMgrLog2(char *s, SSyncIndexMgr *pObj) {
-  if (gRaftDetailLog) {
-    char *serialized = syncIndexMgr2Str(pObj);
-    sTrace("syncIndexMgrLog2 | len:%" PRIu64 " | %s | %s", (uint64_t)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)) {

source/libs/sync/src/syncMessage.c

@@ -13,6 +13,7 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
+#define _DEFAULT_SOURCE
 #include "syncMessage.h"
 #include "syncRaftCfg.h"
 #include "syncRaftEntry.h"
@@ -150,31 +151,6 @@ void syncTimeoutLog2(char* s, const SyncTimeout* pMsg) {
   }
 }
 
-// ---- message process SyncPing----
-SyncPing* syncPingBuild(uint32_t dataLen) {
-  uint32_t  bytes = sizeof(SyncPing) + dataLen;
-  SyncPing* pMsg = taosMemoryMalloc(bytes);
-  memset(pMsg, 0, bytes);
-  pMsg->bytes = bytes;
-  pMsg->msgType = TDMT_SYNC_PING;
-  pMsg->dataLen = dataLen;
-  return pMsg;
-}
-
-SyncPing* syncPingBuild2(const SRaftId* srcId, const SRaftId* destId, int32_t vgId, const char* str) {
-  uint32_t  dataLen = strlen(str) + 1;
-  SyncPing* pMsg = syncPingBuild(dataLen);
-  pMsg->vgId = vgId;
-  pMsg->srcId = *srcId;
-  pMsg->destId = *destId;
-  snprintf(pMsg->data, pMsg->dataLen, "%s", str);
-  return pMsg;
-}
-
-SyncPing* syncPingBuild3(const SRaftId* srcId, const SRaftId* destId, int32_t vgId) {
-  SyncPing* pMsg = syncPingBuild2(srcId, destId, vgId, "ping");
-  return pMsg;
-}
 
 void syncPingDestroy(SyncPing* pMsg) {
   if (pMsg != NULL) {
@@ -193,16 +169,6 @@ void syncPingDeserialize(const char* buf, uint32_t len, SyncPing* pMsg) {
   ASSERT(pMsg->bytes == sizeof(SyncPing) + pMsg->dataLen);
 }
 
-char* syncPingSerialize2(const SyncPing* pMsg, uint32_t* len) {
-  char* buf = taosMemoryMalloc(pMsg->bytes);
-  ASSERT(buf != NULL);
-  syncPingSerialize(pMsg, buf, pMsg->bytes);
-  if (len != NULL) {
-    *len = pMsg->bytes;
-  }
-  return buf;
-}
-
 SyncPing* syncPingDeserialize2(const char* buf, uint32_t len) {
   uint32_t  bytes = *((uint32_t*)buf);
   SyncPing* pMsg = taosMemoryMalloc(bytes);
@@ -212,117 +178,6 @@ SyncPing* syncPingDeserialize2(const char* buf, uint32_t len) {
   return pMsg;
 }
 
-int32_t syncPingSerialize3(const SyncPing* pMsg, char* buf, int32_t bufLen) {
-  SEncoder encoder = {0};
-  tEncoderInit(&encoder, buf, bufLen);
-  if (tStartEncode(&encoder) < 0) {
-    return -1;
-  }
-
-  if (tEncodeU32(&encoder, pMsg->bytes) < 0) {
-    return -1;
-  }
-  if (tEncodeI32(&encoder, pMsg->vgId) < 0) {
-    return -1;
-  }
-  if (tEncodeU32(&encoder, pMsg->msgType) < 0) {
-    return -1;
-  }
-  if (tEncodeU64(&encoder, pMsg->srcId.addr) < 0) {
-    return -1;
-  }
-  if (tEncodeI32(&encoder, pMsg->srcId.vgId) < 0) {
-    return -1;
-  }
-  if (tEncodeU64(&encoder, pMsg->destId.addr) < 0) {
-    return -1;
-  }
-  if (tEncodeI32(&encoder, pMsg->destId.vgId) < 0) {
-    return -1;
-  }
-  if (tEncodeU32(&encoder, pMsg->dataLen) < 0) {
-    return -1;
-  }
-  if (tEncodeBinary(&encoder, pMsg->data, pMsg->dataLen)) {
-    return -1;
-  }
-
-  tEndEncode(&encoder);
-  int32_t tlen = encoder.pos;
-  tEncoderClear(&encoder);
-  return tlen;
-}
-
-SyncPing* syncPingDeserialize3(void* buf, int32_t bufLen) {
-  SDecoder decoder = {0};
-  tDecoderInit(&decoder, buf, bufLen);
-  if (tStartDecode(&decoder) < 0) {
-    return NULL;
-  }
-
-  SyncPing* pMsg = NULL;
-  uint32_t  bytes;
-  if (tDecodeU32(&decoder, &bytes) < 0) {
-    return NULL;
-  }
-
-  pMsg = taosMemoryMalloc(bytes);
-  ASSERT(pMsg != NULL);
-  pMsg->bytes = bytes;
-
-  if (tDecodeI32(&decoder, &pMsg->vgId) < 0) {
-    taosMemoryFree(pMsg);
-    return NULL;
-  }
-  if (tDecodeU32(&decoder, &pMsg->msgType) < 0) {
-    taosMemoryFree(pMsg);
-    return NULL;
-  }
-  if (tDecodeU64(&decoder, &pMsg->srcId.addr) < 0) {
-    taosMemoryFree(pMsg);
-    return NULL;
-  }
-  if (tDecodeI32(&decoder, &pMsg->srcId.vgId) < 0) {
-    taosMemoryFree(pMsg);
-    return NULL;
-  }
-  if (tDecodeU64(&decoder, &pMsg->destId.addr) < 0) {
-    taosMemoryFree(pMsg);
-    return NULL;
-  }
-  if (tDecodeI32(&decoder, &pMsg->destId.vgId) < 0) {
-    taosMemoryFree(pMsg);
-    return NULL;
-  }
-  if (tDecodeU32(&decoder, &pMsg->dataLen) < 0) {
-    taosMemoryFree(pMsg);
-    return NULL;
-  }
-  uint32_t len;
-  char*    data = NULL;
-  if (tDecodeBinary(&decoder, (uint8_t**)(&data), &len) < 0) {
-    taosMemoryFree(pMsg);
-    return NULL;
-  }
-  ASSERT(len == pMsg->dataLen);
-  memcpy(pMsg->data, data, len);
-
-  tEndDecode(&decoder);
-  tDecoderClear(&decoder);
-  return pMsg;
-}
-
-void syncPing2RpcMsg(const SyncPing* pMsg, SRpcMsg* pRpcMsg) {
-  memset(pRpcMsg, 0, sizeof(*pRpcMsg));
-  pRpcMsg->msgType = pMsg->msgType;
-  pRpcMsg->contLen = pMsg->bytes;
-  pRpcMsg->pCont = rpcMallocCont(pRpcMsg->contLen);
-  syncPingSerialize(pMsg, pRpcMsg->pCont, pRpcMsg->contLen);
-}
-
-void syncPingFromRpcMsg(const SRpcMsg* pRpcMsg, SyncPing* pMsg) {
-  syncPingDeserialize(pRpcMsg->pCont, pRpcMsg->contLen, pMsg);
-}
 
 SyncPing* syncPingFromRpcMsg2(const SRpcMsg* pRpcMsg) {
   SyncPing* pMsg = syncPingDeserialize2(pRpcMsg->pCont, pRpcMsg->contLen);
@@ -330,96 +185,6 @@ SyncPing* syncPingFromRpcMsg2(const SRpcMsg* pRpcMsg) {
   return pMsg;
 }
 
-cJSON* syncPing2Json(const SyncPing* pMsg) {
-  char   u64buf[128] = {0};
-  cJSON* pRoot = cJSON_CreateObject();
-
-  if (pMsg != NULL) {
-    cJSON_AddNumberToObject(pRoot, "bytes", pMsg->bytes);
-    cJSON_AddNumberToObject(pRoot, "vgId", pMsg->vgId);
-    cJSON_AddNumberToObject(pRoot, "msgType", pMsg->msgType);
-
-    cJSON* pSrcId = cJSON_CreateObject();
-    snprintf(u64buf, sizeof(u64buf), "%" PRIu64, pMsg->srcId.addr);
-    cJSON_AddStringToObject(pSrcId, "addr", u64buf);
-    {
-      uint64_t u64 = pMsg->srcId.addr;
-      cJSON*   pTmp = pSrcId;
-      char     host[128] = {0};
-      uint16_t port;
-      syncUtilU642Addr(u64, host, sizeof(host), &port);
-      cJSON_AddStringToObject(pTmp, "addr_host", host);
-      cJSON_AddNumberToObject(pTmp, "addr_port", port);
-    }
-    cJSON_AddNumberToObject(pSrcId, "vgId", pMsg->srcId.vgId);
-    cJSON_AddItemToObject(pRoot, "srcId", pSrcId);
-
-    cJSON* pDestId = cJSON_CreateObject();
-    snprintf(u64buf, sizeof(u64buf), "%" PRIu64, pMsg->destId.addr);
-    cJSON_AddStringToObject(pDestId, "addr", u64buf);
-    {
-      uint64_t u64 = pMsg->destId.addr;
-      cJSON*   pTmp = pDestId;
-      char     host[128] = {0};
-      uint16_t port;
-      syncUtilU642Addr(u64, host, sizeof(host), &port);
-      cJSON_AddStringToObject(pTmp, "addr_host", host);
-      cJSON_AddNumberToObject(pTmp, "addr_port", port);
-    }
-    cJSON_AddNumberToObject(pDestId, "vgId", pMsg->destId.vgId);
-    cJSON_AddItemToObject(pRoot, "destId", pDestId);
-
-    cJSON_AddNumberToObject(pRoot, "dataLen", pMsg->dataLen);
-    char* s;
-    s = syncUtilPrintBin((char*)(pMsg->data), pMsg->dataLen);
-    cJSON_AddStringToObject(pRoot, "data", s);
-    taosMemoryFree(s);
-    s = syncUtilPrintBin2((char*)(pMsg->data), pMsg->dataLen);
-    cJSON_AddStringToObject(pRoot, "data2", s);
-    taosMemoryFree(s);
-  }
-
-  cJSON* pJson = cJSON_CreateObject();
-  cJSON_AddItemToObject(pJson, "SyncPing", pRoot);
-  return pJson;
-}
-
-char* syncPing2Str(const SyncPing* pMsg) {
-  cJSON* pJson = syncPing2Json(pMsg);
-  char*  serialized = cJSON_Print(pJson);
-  cJSON_Delete(pJson);
-  return serialized;
-}
-
-// for debug ----------------------
-void syncPingPrint(const SyncPing* pMsg) {
-  char* serialized = syncPing2Str(pMsg);
-  printf("syncPingPrint | len:%d | %s \n", (int32_t)strlen(serialized), serialized);
-  fflush(NULL);
-  taosMemoryFree(serialized);
-}
-
-void syncPingPrint2(char* s, const SyncPing* pMsg) {
-  char* serialized = syncPing2Str(pMsg);
-  printf("syncPingPrint2 | len:%d | %s | %s \n", (int32_t)strlen(serialized), s, serialized);
-  fflush(NULL);
-  taosMemoryFree(serialized);
-}
-
-void syncPingLog(const SyncPing* pMsg) {
-  char* serialized = syncPing2Str(pMsg);
-  sTrace("syncPingLog | len:%d | %s", (int32_t)strlen(serialized), serialized);
-  taosMemoryFree(serialized);
-}
-
-void syncPingLog2(char* s, const SyncPing* pMsg) {
-  if (gRaftDetailLog) {
-    char* serialized = syncPing2Str(pMsg);
-    sTrace("syncPingLog2 | len:%d | %s | %s", (int32_t)strlen(serialized), s, serialized);
-    taosMemoryFree(serialized);
-  }
-}
-
 // ---- message process SyncPingReply----
 SyncPingReply* syncPingReplyBuild(uint32_t dataLen) {
   uint32_t       bytes = sizeof(SyncPingReply) + dataLen;

source/libs/sync/src/syncRequestVote.c

@@ -15,6 +15,7 @@
 
 #define _DEFAULT_SOURCE
 #include "syncRequestVote.h"
+#include "syncMessage.h"
 #include "syncRaftCfg.h"
 #include "syncRaftStore.h"
 #include "syncUtil.h"

source/libs/sync/src/syncRequestVoteReply.c

@@ -15,6 +15,7 @@
 
 #define _DEFAULT_SOURCE
 #include "syncRequestVoteReply.h"
+#include "syncMessage.h"
 #include "syncRaftStore.h"
 #include "syncVoteMgr.h"
 

source/libs/sync/src/syncVoteMgr.c

@@ -15,6 +15,7 @@
 
 #define _DEFAULT_SOURCE
 #include "syncVoteMgr.h"
+#include "syncMessage.h"
 #include "syncUtil.h"
 
 static void voteGrantedClearVotes(SVotesGranted *pVotesGranted) {

source/libs/sync/test/syncConfigChangeSnapshotTest.cpp

@@ -143,7 +143,7 @@ int32_t SnapshotDoWrite(struct SSyncFSM* pFsm, void* pWriter, void* pBuf, int32_
 void RestoreFinishCb(struct SSyncFSM* pFsm) { sTrace("==callback== ==RestoreFinishCb=="); }
 
 void ReConfigCb(struct SSyncFSM* pFsm, const SRpcMsg* pMsg, SReConfigCbMeta* cbMeta) {
-  sTrace("==callback== ==ReConfigCb== flag:0x%" PRIx64 ", index:%" PRId64 ", code:%d, currentTerm:%" PRIu64
+  sTrace("==callback== ==ReConfigCb== flag:%" PRIx64 ", index:%" PRId64 ", code:%d, currentTerm:%" PRIu64
          ", term:%" PRIu64,
          cbMeta->flag, cbMeta->index, cbMeta->code, cbMeta->currentTerm, cbMeta->term);
 }

source/libs/sync/test/sync_test_lib/inc/syncTest.h

@@ -110,7 +110,37 @@ char*  snapshotSender2Str(SSyncSnapshotSender* pSender);
 cJSON* snapshotReceiver2Json(SSyncSnapshotReceiver* pReceiver);
 char*  snapshotReceiver2Str(SSyncSnapshotReceiver* pReceiver);
 
-
+cJSON*   syncIndexMgr2Json(SSyncIndexMgr* pSyncIndexMgr);
+char*    syncIndexMgr2Str(SSyncIndexMgr* pSyncIndexMgr);
+void     syncIndexMgrPrint(SSyncIndexMgr* pObj);
+void     syncIndexMgrPrint2(char* s, SSyncIndexMgr* pObj);
+void     syncIndexMgrLog(SSyncIndexMgr* pObj);
+void     syncIndexMgrLog2(char* s, SSyncIndexMgr* pObj);
+
+cJSON* syncRpcMsg2Json(SRpcMsg* pRpcMsg);
+cJSON* syncRpcUnknownMsg2Json();
+char*  syncRpcMsg2Str(SRpcMsg* pRpcMsg);
+void   syncRpcMsgPrint(SRpcMsg* pMsg);
+void   syncRpcMsgPrint2(char* s, SRpcMsg* pMsg);
+void   syncRpcMsgLog(SRpcMsg* pMsg);
+void   syncRpcMsgLog2(char* s, SRpcMsg* pMsg);
+
+
+// origin syncMessage
+SyncPing* syncPingBuild(uint32_t dataLen);
+SyncPing* syncPingBuild2(const SRaftId* srcId, const SRaftId* destId, int32_t vgId, const char* str);
+SyncPing* syncPingBuild3(const SRaftId* srcId, const SRaftId* destId, int32_t vgId);
+char*     syncPingSerialize2(const SyncPing* pMsg, uint32_t* len);
+int32_t   syncPingSerialize3(const SyncPing* pMsg, char* buf, int32_t bufLen);
+SyncPing* syncPingDeserialize3(void* buf, int32_t bufLen);
+void      syncPing2RpcMsg(const SyncPing* pMsg, SRpcMsg* pRpcMsg);
+void      syncPingFromRpcMsg(const SRpcMsg* pRpcMsg, SyncPing* pMsg);
+cJSON*    syncPing2Json(const SyncPing* pMsg);
+char*     syncPing2Str(const SyncPing* pMsg);
+void      syncPingPrint(const SyncPing* pMsg);
+void      syncPingPrint2(char* s, const SyncPing* pMsg);
+void      syncPingLog(const SyncPing* pMsg);
+void      syncPingLog2(char* s, const SyncPing* pMsg);
 
 #ifdef __cplusplus
 }

source/libs/sync/test/sync_test_lib/src/syncIndexMgrDebug.c

+/*
+ * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
+ *
+ * This program is free software: you can use, redistribute, and/or modify
+ * it under the terms of the GNU Affero General Public License, version 3
+ * or later ("AGPL"), as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#define _DEFAULT_SOURCE
+#include "syncTest.h"
+
+void syncIndexMgrPrint(SSyncIndexMgr *pObj) {
+  char *serialized = syncIndexMgr2Str(pObj);
+  printf("syncIndexMgrPrint | len:%" PRIu64 " | %s \n", (uint64_t)strlen(serialized), serialized);
+  fflush(NULL);
+  taosMemoryFree(serialized);
+}
+
+void syncIndexMgrPrint2(char *s, SSyncIndexMgr *pObj) {
+  char *serialized = syncIndexMgr2Str(pObj);
+  printf("syncIndexMgrPrint2 | len:%" PRIu64 " | %s | %s \n", (uint64_t)strlen(serialized), s, serialized);
+  fflush(NULL);
+  taosMemoryFree(serialized);
+}
+
+void syncIndexMgrLog(SSyncIndexMgr *pObj) {
+  char *serialized = syncIndexMgr2Str(pObj);
+  sTrace("syncIndexMgrLog | len:%" PRIu64 " | %s", (uint64_t)strlen(serialized), serialized);
+  taosMemoryFree(serialized);
+}
+
+void syncIndexMgrLog2(char *s, SSyncIndexMgr *pObj) {
+  if (gRaftDetailLog) {
+    char *serialized = syncIndexMgr2Str(pObj);
+    sTrace("syncIndexMgrLog2 | len:%" PRIu64 " | %s | %s", (uint64_t)strlen(serialized), s, serialized);
+    taosMemoryFree(serialized);
+  }
+}
+
+cJSON *syncIndexMgr2Json(SSyncIndexMgr *pSyncIndexMgr) {
+  char   u64buf[128] = {0};
+  cJSON *pRoot = cJSON_CreateObject();
+
+  if (pSyncIndexMgr != NULL) {
+    cJSON_AddNumberToObject(pRoot, "replicaNum", pSyncIndexMgr->replicaNum);
+    cJSON *pReplicas = cJSON_CreateArray();
+    cJSON_AddItemToObject(pRoot, "replicas", pReplicas);
+    for (int i = 0; i < pSyncIndexMgr->replicaNum; ++i) {
+      cJSON_AddItemToArray(pReplicas, syncUtilRaftId2Json(&(*(pSyncIndexMgr->replicas))[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);
+    }
+
+    {
+      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);
+  }
+
+  cJSON *pJson = cJSON_CreateObject();
+  cJSON_AddItemToObject(pJson, "pSyncIndexMgr", pRoot);
+  return pJson;
+}
+
+char *syncIndexMgr2Str(SSyncIndexMgr *pSyncIndexMgr) {
+  cJSON *pJson = syncIndexMgr2Json(pSyncIndexMgr);
+  char  *serialized = cJSON_Print(pJson);
+  cJSON_Delete(pJson);
+  return serialized;
+}

source/libs/sync/test/sync_test_lib/src/syncMessageDebug.c

@@ -16,6 +16,244 @@
 #define _DEFAULT_SOURCE
 #include "syncTest.h"
 
+// ---- message process SyncPing----
+SyncPing* syncPingBuild(uint32_t dataLen) {
+  uint32_t  bytes = sizeof(SyncPing) + dataLen;
+  SyncPing* pMsg = taosMemoryMalloc(bytes);
+  memset(pMsg, 0, bytes);
+  pMsg->bytes = bytes;
+  pMsg->msgType = TDMT_SYNC_PING;
+  pMsg->dataLen = dataLen;
+  return pMsg;
+}
+
+SyncPing* syncPingBuild2(const SRaftId* srcId, const SRaftId* destId, int32_t vgId, const char* str) {
+  uint32_t  dataLen = strlen(str) + 1;
+  SyncPing* pMsg = syncPingBuild(dataLen);
+  pMsg->vgId = vgId;
+  pMsg->srcId = *srcId;
+  pMsg->destId = *destId;
+  snprintf(pMsg->data, pMsg->dataLen, "%s", str);
+  return pMsg;
+}
+
+SyncPing* syncPingBuild3(const SRaftId* srcId, const SRaftId* destId, int32_t vgId) {
+  SyncPing* pMsg = syncPingBuild2(srcId, destId, vgId, "ping");
+  return pMsg;
+}
+
+char* syncPingSerialize2(const SyncPing* pMsg, uint32_t* len) {
+  char* buf = taosMemoryMalloc(pMsg->bytes);
+  ASSERT(buf != NULL);
+  syncPingSerialize(pMsg, buf, pMsg->bytes);
+  if (len != NULL) {
+    *len = pMsg->bytes;
+  }
+  return buf;
+}
+
+void syncPing2RpcMsg(const SyncPing* pMsg, SRpcMsg* pRpcMsg) {
+  memset(pRpcMsg, 0, sizeof(*pRpcMsg));
+  pRpcMsg->msgType = pMsg->msgType;
+  pRpcMsg->contLen = pMsg->bytes;
+  pRpcMsg->pCont = rpcMallocCont(pRpcMsg->contLen);
+  syncPingSerialize(pMsg, pRpcMsg->pCont, pRpcMsg->contLen);
+}
+
+void syncPingFromRpcMsg(const SRpcMsg* pRpcMsg, SyncPing* pMsg) {
+  syncPingDeserialize(pRpcMsg->pCont, pRpcMsg->contLen, pMsg);
+}
+
+SyncPing* syncPingDeserialize3(void* buf, int32_t bufLen) {
+  SDecoder decoder = {0};
+  tDecoderInit(&decoder, buf, bufLen);
+  if (tStartDecode(&decoder) < 0) {
+    return NULL;
+  }
+
+  SyncPing* pMsg = NULL;
+  uint32_t  bytes;
+  if (tDecodeU32(&decoder, &bytes) < 0) {
+    return NULL;
+  }
+
+  pMsg = taosMemoryMalloc(bytes);
+  ASSERT(pMsg != NULL);
+  pMsg->bytes = bytes;
+
+  if (tDecodeI32(&decoder, &pMsg->vgId) < 0) {
+    taosMemoryFree(pMsg);
+    return NULL;
+  }
+  if (tDecodeU32(&decoder, &pMsg->msgType) < 0) {
+    taosMemoryFree(pMsg);
+    return NULL;
+  }
+  if (tDecodeU64(&decoder, &pMsg->srcId.addr) < 0) {
+    taosMemoryFree(pMsg);
+    return NULL;
+  }
+  if (tDecodeI32(&decoder, &pMsg->srcId.vgId) < 0) {
+    taosMemoryFree(pMsg);
+    return NULL;
+  }
+  if (tDecodeU64(&decoder, &pMsg->destId.addr) < 0) {
+    taosMemoryFree(pMsg);
+    return NULL;
+  }
+  if (tDecodeI32(&decoder, &pMsg->destId.vgId) < 0) {
+    taosMemoryFree(pMsg);
+    return NULL;
+  }
+  if (tDecodeU32(&decoder, &pMsg->dataLen) < 0) {
+    taosMemoryFree(pMsg);
+    return NULL;
+  }
+  uint32_t len;
+  char*    data = NULL;
+  if (tDecodeBinary(&decoder, (uint8_t**)(&data), &len) < 0) {
+    taosMemoryFree(pMsg);
+    return NULL;
+  }
+  ASSERT(len == pMsg->dataLen);
+  memcpy(pMsg->data, data, len);
+
+  tEndDecode(&decoder);
+  tDecoderClear(&decoder);
+  return pMsg;
+}
+
+int32_t syncPingSerialize3(const SyncPing* pMsg, char* buf, int32_t bufLen) {
+  SEncoder encoder = {0};
+  tEncoderInit(&encoder, buf, bufLen);
+  if (tStartEncode(&encoder) < 0) {
+    return -1;
+  }
+
+  if (tEncodeU32(&encoder, pMsg->bytes) < 0) {
+    return -1;
+  }
+  if (tEncodeI32(&encoder, pMsg->vgId) < 0) {
+    return -1;
+  }
+  if (tEncodeU32(&encoder, pMsg->msgType) < 0) {
+    return -1;
+  }
+  if (tEncodeU64(&encoder, pMsg->srcId.addr) < 0) {
+    return -1;
+  }
+  if (tEncodeI32(&encoder, pMsg->srcId.vgId) < 0) {
+    return -1;
+  }
+  if (tEncodeU64(&encoder, pMsg->destId.addr) < 0) {
+    return -1;
+  }
+  if (tEncodeI32(&encoder, pMsg->destId.vgId) < 0) {
+    return -1;
+  }
+  if (tEncodeU32(&encoder, pMsg->dataLen) < 0) {
+    return -1;
+  }
+  if (tEncodeBinary(&encoder, pMsg->data, pMsg->dataLen)) {
+    return -1;
+  }
+
+  tEndEncode(&encoder);
+  int32_t tlen = encoder.pos;
+  tEncoderClear(&encoder);
+  return tlen;
+}
+
+cJSON* syncPing2Json(const SyncPing* pMsg) {
+  char   u64buf[128] = {0};
+  cJSON* pRoot = cJSON_CreateObject();
+
+  if (pMsg != NULL) {
+    cJSON_AddNumberToObject(pRoot, "bytes", pMsg->bytes);
+    cJSON_AddNumberToObject(pRoot, "vgId", pMsg->vgId);
+    cJSON_AddNumberToObject(pRoot, "msgType", pMsg->msgType);
+
+    cJSON* pSrcId = cJSON_CreateObject();
+    snprintf(u64buf, sizeof(u64buf), "%" PRIu64, pMsg->srcId.addr);
+    cJSON_AddStringToObject(pSrcId, "addr", u64buf);
+    {
+      uint64_t u64 = pMsg->srcId.addr;
+      cJSON*   pTmp = pSrcId;
+      char     host[128] = {0};
+      uint16_t port;
+      syncUtilU642Addr(u64, host, sizeof(host), &port);
+      cJSON_AddStringToObject(pTmp, "addr_host", host);
+      cJSON_AddNumberToObject(pTmp, "addr_port", port);
+    }
+    cJSON_AddNumberToObject(pSrcId, "vgId", pMsg->srcId.vgId);
+    cJSON_AddItemToObject(pRoot, "srcId", pSrcId);
+
+    cJSON* pDestId = cJSON_CreateObject();
+    snprintf(u64buf, sizeof(u64buf), "%" PRIu64, pMsg->destId.addr);
+    cJSON_AddStringToObject(pDestId, "addr", u64buf);
+    {
+      uint64_t u64 = pMsg->destId.addr;
+      cJSON*   pTmp = pDestId;
+      char     host[128] = {0};
+      uint16_t port;
+      syncUtilU642Addr(u64, host, sizeof(host), &port);
+      cJSON_AddStringToObject(pTmp, "addr_host", host);
+      cJSON_AddNumberToObject(pTmp, "addr_port", port);
+    }
+    cJSON_AddNumberToObject(pDestId, "vgId", pMsg->destId.vgId);
+    cJSON_AddItemToObject(pRoot, "destId", pDestId);
+
+    cJSON_AddNumberToObject(pRoot, "dataLen", pMsg->dataLen);
+    char* s;
+    s = syncUtilPrintBin((char*)(pMsg->data), pMsg->dataLen);
+    cJSON_AddStringToObject(pRoot, "data", s);
+    taosMemoryFree(s);
+    s = syncUtilPrintBin2((char*)(pMsg->data), pMsg->dataLen);
+    cJSON_AddStringToObject(pRoot, "data2", s);
+    taosMemoryFree(s);
+  }
+
+  cJSON* pJson = cJSON_CreateObject();
+  cJSON_AddItemToObject(pJson, "SyncPing", pRoot);
+  return pJson;
+}
+
+char* syncPing2Str(const SyncPing* pMsg) {
+  cJSON* pJson = syncPing2Json(pMsg);
+  char*  serialized = cJSON_Print(pJson);
+  cJSON_Delete(pJson);
+  return serialized;
+}
+
+// for debug ----------------------
+void syncPingPrint(const SyncPing* pMsg) {
+  char* serialized = syncPing2Str(pMsg);
+  printf("syncPingPrint | len:%d | %s \n", (int32_t)strlen(serialized), serialized);
+  fflush(NULL);
+  taosMemoryFree(serialized);
+}
+
+void syncPingPrint2(char* s, const SyncPing* pMsg) {
+  char* serialized = syncPing2Str(pMsg);
+  printf("syncPingPrint2 | len:%d | %s | %s \n", (int32_t)strlen(serialized), s, serialized);
+  fflush(NULL);
+  taosMemoryFree(serialized);
+}
+
+void syncPingLog(const SyncPing* pMsg) {
+  char* serialized = syncPing2Str(pMsg);
+  sTrace("syncPingLog | len:%d | %s", (int32_t)strlen(serialized), serialized);
+  taosMemoryFree(serialized);
+}
+
+void syncPingLog2(char* s, const SyncPing* pMsg) {
+  if (gRaftDetailLog) {
+    char* serialized = syncPing2Str(pMsg);
+    sTrace("syncPingLog2 | len:%d | %s | %s", (int32_t)strlen(serialized), s, serialized);
+    taosMemoryFree(serialized);
+  }
+}
+
 // ---------------------------------------------
 cJSON* syncRpcMsg2Json(SRpcMsg* pRpcMsg) {
   cJSON* pRoot;