sync append entries (sync-io)

source/libs/sync/inc/syncRaftStore.h

@@ -48,6 +48,8 @@ void raftStoreVote(SRaftStore *pRaftStore, SRaftId *pRaftId);
 void   raftStoreClearVote(SRaftStore *pRaftStore);
 void   raftStoreNextTerm(SRaftStore *pRaftStore);
 void   raftStoreSetTerm(SRaftStore *pRaftStore, SyncTerm term);
+cJSON *raftStore2Json(SRaftStore *pRaftStore);
+char  *raftStore2Str(SRaftStore *pRaftStore);
 
 // for debug -------------------
 void raftStorePrint(SRaftStore *pObj);

source/libs/sync/src/syncIndexMgr.c

@@ -70,6 +70,7 @@ cJSON *syncIndexMgr2Json(SSyncIndexMgr *pSyncIndexMgr) {
   char   u64buf[128];
   cJSON *pRoot = cJSON_CreateObject();
 
+  if (pSyncIndexMgr != NULL) {
     cJSON_AddNumberToObject(pRoot, "replicaNum", pSyncIndexMgr->replicaNum);
     cJSON *pReplicas = cJSON_CreateArray();
     cJSON_AddItemToObject(pRoot, "replicas", pReplicas);
@@ -86,6 +87,7 @@ cJSON *syncIndexMgr2Json(SSyncIndexMgr *pSyncIndexMgr) {
     cJSON_AddItemToObject(pRoot, "index", pIndex);
     snprintf(u64buf, sizeof(u64buf), "%p", pSyncIndexMgr->pSyncNode);
     cJSON_AddStringToObject(pRoot, "pSyncNode", u64buf);
+  }
 
   cJSON *pJson = cJSON_CreateObject();
   cJSON_AddItemToObject(pJson, "pSyncIndexMgr", pRoot);

source/libs/sync/src/syncMain.c

@@ -355,6 +355,7 @@ cJSON* syncNode2Json(const SSyncNode* pSyncNode) {
   char   u64buf[128];
   cJSON* pRoot = cJSON_CreateObject();
 
+  if (pSyncNode != NULL) {
     // init by SSyncInfo
     cJSON_AddNumberToObject(pRoot, "vgId", pSyncNode->vgId);
     cJSON_AddStringToObject(pRoot, "path", pSyncNode->path);
@@ -406,9 +407,8 @@ cJSON* syncNode2Json(const SSyncNode* pSyncNode) {
     // tla+ server vars
     cJSON_AddNumberToObject(pRoot, "state", pSyncNode->state);
     cJSON_AddStringToObject(pRoot, "state_str", syncUtilState2String(pSyncNode->state));
-  char tmpBuf[RAFT_STORE_BLOCK_SIZE];
+    cJSON* pRaftStore = raftStore2Json(pSyncNode->pRaftStore);
-  raftStoreSerialize(pSyncNode->pRaftStore, tmpBuf, sizeof(tmpBuf));
+    cJSON_AddItemToObject(pRoot, "pRaftStore", pRaftStore);
-  cJSON_AddStringToObject(pRoot, "pRaftStore", tmpBuf);
 
     // tla+ candidate vars
     cJSON_AddItemToObject(pRoot, "pVotesGranted", voteGranted2Json(pSyncNode->pVotesGranted));
@@ -477,6 +477,7 @@ cJSON* syncNode2Json(const SSyncNode* pSyncNode) {
     cJSON_AddStringToObject(pRoot, "FpOnAppendEntriesReply", u64buf);
     snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->FpOnTimeout);
     cJSON_AddStringToObject(pRoot, "FpOnTimeout", u64buf);
+  }
 
   cJSON* pJson = cJSON_CreateObject();
   cJSON_AddItemToObject(pJson, "SSyncNode", pRoot);

source/libs/sync/src/syncMessage.c

@@ -213,8 +213,9 @@ SyncTimeout* syncTimeoutFromRpcMsg2(const SRpcMsg* pRpcMsg) {
 
 cJSON* syncTimeout2Json(const SyncTimeout* pMsg) {
   char   u64buf[128];
-
   cJSON* pRoot = cJSON_CreateObject();
+
+  if (pMsg != NULL) {
     cJSON_AddNumberToObject(pRoot, "bytes", pMsg->bytes);
     cJSON_AddNumberToObject(pRoot, "msgType", pMsg->msgType);
     cJSON_AddNumberToObject(pRoot, "timeoutType", pMsg->timeoutType);
@@ -223,6 +224,7 @@ cJSON* syncTimeout2Json(const SyncTimeout* pMsg) {
     cJSON_AddNumberToObject(pRoot, "timerMS", pMsg->timerMS);
     snprintf(u64buf, sizeof(u64buf), "%p", pMsg->data);
     cJSON_AddStringToObject(pRoot, "data", u64buf);
+  }
 
   cJSON* pJson = cJSON_CreateObject();
   cJSON_AddItemToObject(pJson, "SyncTimeout", pRoot);
@@ -343,8 +345,9 @@ SyncPing* syncPingFromRpcMsg2(const SRpcMsg* pRpcMsg) {
 
 cJSON* syncPing2Json(const SyncPing* pMsg) {
   char   u64buf[128];
-
   cJSON* pRoot = cJSON_CreateObject();
+
+  if (pMsg != NULL) {
     cJSON_AddNumberToObject(pRoot, "bytes", pMsg->bytes);
     cJSON_AddNumberToObject(pRoot, "msgType", pMsg->msgType);
 
@@ -386,6 +389,7 @@ cJSON* syncPing2Json(const SyncPing* pMsg) {
     s = syncUtilprintBin2((char*)(pMsg->data), pMsg->dataLen);
     cJSON_AddStringToObject(pRoot, "data2", s);
     free(s);
+  }
 
   cJSON* pJson = cJSON_CreateObject();
   cJSON_AddItemToObject(pJson, "SyncPing", pRoot);
@@ -506,8 +510,9 @@ SyncPingReply* syncPingReplyFromRpcMsg2(const SRpcMsg* pRpcMsg) {
 
 cJSON* syncPingReply2Json(const SyncPingReply* pMsg) {
   char   u64buf[128];
-
   cJSON* pRoot = cJSON_CreateObject();
+
+  if (pMsg != NULL) {
     cJSON_AddNumberToObject(pRoot, "bytes", pMsg->bytes);
     cJSON_AddNumberToObject(pRoot, "msgType", pMsg->msgType);
 
@@ -549,6 +554,7 @@ cJSON* syncPingReply2Json(const SyncPingReply* pMsg) {
     s = syncUtilprintBin2((char*)(pMsg->data), pMsg->dataLen);
     cJSON_AddStringToObject(pRoot, "data2", s);
     free(s);
+  }
 
   cJSON* pJson = cJSON_CreateObject();
   cJSON_AddItemToObject(pJson, "SyncPingReply", pRoot);
@@ -665,8 +671,9 @@ SyncClientRequest* syncClientRequestFromRpcMsg2(const SRpcMsg* pRpcMsg) {
 
 cJSON* syncClientRequest2Json(const SyncClientRequest* pMsg) {
   char   u64buf[128];
-
   cJSON* pRoot = cJSON_CreateObject();
+
+  if (pMsg != NULL) {
     cJSON_AddNumberToObject(pRoot, "bytes", pMsg->bytes);
     cJSON_AddNumberToObject(pRoot, "msgType", pMsg->msgType);
     cJSON_AddNumberToObject(pRoot, "originalRpcType", pMsg->originalRpcType);
@@ -682,6 +689,7 @@ cJSON* syncClientRequest2Json(const SyncClientRequest* pMsg) {
     s = syncUtilprintBin2((char*)(pMsg->data), pMsg->dataLen);
     cJSON_AddStringToObject(pRoot, "data2", s);
     free(s);
+  }
 
   cJSON* pJson = cJSON_CreateObject();
   cJSON_AddItemToObject(pJson, "SyncClientRequest", pRoot);
@@ -786,8 +794,9 @@ SyncRequestVote* syncRequestVoteFromRpcMsg2(const SRpcMsg* pRpcMsg) {
 
 cJSON* syncRequestVote2Json(const SyncRequestVote* pMsg) {
   char   u64buf[128];
-
   cJSON* pRoot = cJSON_CreateObject();
+
+  if (pMsg != NULL) {
     cJSON_AddNumberToObject(pRoot, "bytes", pMsg->bytes);
     cJSON_AddNumberToObject(pRoot, "msgType", pMsg->msgType);
 
@@ -826,6 +835,7 @@ cJSON* syncRequestVote2Json(const SyncRequestVote* pMsg) {
     cJSON_AddStringToObject(pRoot, "lastLogIndex", u64buf);
     snprintf(u64buf, sizeof(u64buf), "%lu", pMsg->lastLogTerm);
     cJSON_AddStringToObject(pRoot, "lastLogTerm", u64buf);
+  }
 
   cJSON* pJson = cJSON_CreateObject();
   cJSON_AddItemToObject(pJson, "SyncRequestVote", pRoot);
@@ -930,8 +940,9 @@ SyncRequestVoteReply* syncRequestVoteReplyFromRpcMsg2(const SRpcMsg* pRpcMsg) {
 
 cJSON* syncRequestVoteReply2Json(const SyncRequestVoteReply* pMsg) {
   char   u64buf[128];
-
   cJSON* pRoot = cJSON_CreateObject();
+
+  if (pMsg != NULL) {
     cJSON_AddNumberToObject(pRoot, "bytes", pMsg->bytes);
     cJSON_AddNumberToObject(pRoot, "msgType", pMsg->msgType);
 
@@ -967,6 +978,7 @@ cJSON* syncRequestVoteReply2Json(const SyncRequestVoteReply* pMsg) {
     snprintf(u64buf, sizeof(u64buf), "%lu", pMsg->term);
     cJSON_AddStringToObject(pRoot, "term", u64buf);
     cJSON_AddNumberToObject(pRoot, "vote_granted", pMsg->voteGranted);
+  }
 
   cJSON* pJson = cJSON_CreateObject();
   cJSON_AddItemToObject(pJson, "SyncRequestVoteReply", pRoot);
@@ -1073,8 +1085,9 @@ SyncAppendEntries* syncAppendEntriesFromRpcMsg2(const SRpcMsg* pRpcMsg) {
 
 cJSON* syncAppendEntries2Json(const SyncAppendEntries* pMsg) {
   char   u64buf[128];
-
   cJSON* pRoot = cJSON_CreateObject();
+
+  if (pMsg != NULL) {
     cJSON_AddNumberToObject(pRoot, "bytes", pMsg->bytes);
     cJSON_AddNumberToObject(pRoot, "msgType", pMsg->msgType);
 
@@ -1128,6 +1141,7 @@ cJSON* syncAppendEntries2Json(const SyncAppendEntries* pMsg) {
     s = syncUtilprintBin2((char*)(pMsg->data), pMsg->dataLen);
     cJSON_AddStringToObject(pRoot, "data2", s);
     free(s);
+  }
 
   cJSON* pJson = cJSON_CreateObject();
   cJSON_AddItemToObject(pJson, "SyncAppendEntries", pRoot);
@@ -1232,8 +1246,9 @@ SyncAppendEntriesReply* syncAppendEntriesReplyFromRpcMsg2(const SRpcMsg* pRpcMsg
 
 cJSON* syncAppendEntriesReply2Json(const SyncAppendEntriesReply* pMsg) {
   char   u64buf[128];
-
   cJSON* pRoot = cJSON_CreateObject();
+
+  if (pMsg != NULL) {
     cJSON_AddNumberToObject(pRoot, "bytes", pMsg->bytes);
     cJSON_AddNumberToObject(pRoot, "msgType", pMsg->msgType);
 
@@ -1272,6 +1287,7 @@ cJSON* syncAppendEntriesReply2Json(const SyncAppendEntriesReply* pMsg) {
     cJSON_AddNumberToObject(pRoot, "success", pMsg->success);
     snprintf(u64buf, sizeof(u64buf), "%lu", pMsg->matchIndex);
     cJSON_AddStringToObject(pRoot, "matchIndex", u64buf);
+  }
 
   cJSON* pJson = cJSON_CreateObject();
   cJSON_AddItemToObject(pJson, "SyncAppendEntriesReply", pRoot);

source/libs/sync/src/syncRaftEntry.c

@@ -69,8 +69,9 @@ SSyncRaftEntry* syncEntryDeserialize(const char* buf, uint32_t len) {
 
 cJSON* syncEntry2Json(const SSyncRaftEntry* pEntry) {
   char   u64buf[128];
-
   cJSON* pRoot = cJSON_CreateObject();
+
+  if (pEntry != NULL) {
     cJSON_AddNumberToObject(pRoot, "bytes", pEntry->bytes);
     cJSON_AddNumberToObject(pRoot, "msgType", pEntry->msgType);
     cJSON_AddNumberToObject(pRoot, "originalRpcType", pEntry->originalRpcType);
@@ -91,6 +92,7 @@ cJSON* syncEntry2Json(const SSyncRaftEntry* pEntry) {
     s = syncUtilprintBin2((char*)(pEntry->data), pEntry->dataLen);
     cJSON_AddStringToObject(pRoot, "data2", s);
     free(s);
+  }
 
   cJSON* pJson = cJSON_CreateObject();
   cJSON_AddItemToObject(pJson, "SSyncRaftEntry", pRoot);

source/libs/sync/src/syncRaftLog.c

@@ -123,9 +123,10 @@ SSyncRaftEntry* logStoreGetLastEntry(SSyncLogStore* pLogStore) {
 
 cJSON* logStore2Json(SSyncLogStore* pLogStore) {
   char               u64buf[128];
-
   SSyncLogStoreData* pData = (SSyncLogStoreData*)pLogStore->data;
   cJSON*             pRoot = cJSON_CreateObject();
+
+  if (pData != NULL && pData->pWal != NULL) {
     snprintf(u64buf, sizeof(u64buf), "%p", pData->pSyncNode);
     cJSON_AddStringToObject(pRoot, "pSyncNode", u64buf);
     snprintf(u64buf, sizeof(u64buf), "%p", pData->pWal);
@@ -143,6 +144,7 @@ cJSON* logStore2Json(SSyncLogStore* pLogStore) {
       cJSON_AddItemToArray(pEntries, syncEntry2Json(pEntry));
       syncEntryDestory(pEntry);
     }
+  }
 
   cJSON* pJson = cJSON_CreateObject();
   cJSON_AddItemToObject(pJson, "SSyncLogStore", pRoot);

source/libs/sync/src/syncRaftStore.c

@@ -164,30 +164,63 @@ void raftStoreSetTerm(SRaftStore *pRaftStore, SyncTerm term) {
   raftStorePersist(pRaftStore);
 }
 
+cJSON *raftStore2Json(SRaftStore *pRaftStore) {
+  char   u64buf[128];
+  cJSON *pRoot = cJSON_CreateObject();
+
+  if (pRaftStore != NULL) {
+    snprintf(u64buf, sizeof(u64buf), "%lu", pRaftStore->currentTerm);
+    cJSON_AddStringToObject(pRoot, "currentTerm", u64buf);
+
+    cJSON *pVoteFor = cJSON_CreateObject();
+    snprintf(u64buf, sizeof(u64buf), "%lu", pRaftStore->voteFor.addr);
+    cJSON_AddStringToObject(pVoteFor, "addr", u64buf);
+    {
+      uint64_t u64 = pRaftStore->voteFor.addr;
+      char     host[128];
+      uint16_t port;
+      syncUtilU642Addr(u64, host, sizeof(host), &port);
+      cJSON_AddStringToObject(pVoteFor, "addr_host", host);
+      cJSON_AddNumberToObject(pVoteFor, "addr_port", port);
+    }
+    cJSON_AddNumberToObject(pVoteFor, "vgId", pRaftStore->voteFor.vgId);
+    cJSON_AddItemToObject(pRoot, "voteFor", pVoteFor);
+  }
+
+  cJSON *pJson = cJSON_CreateObject();
+  cJSON_AddItemToObject(pJson, "SRaftStore", pRoot);
+  return pJson;
+}
+
+char *raftStore2Str(SRaftStore *pRaftStore) {
+  cJSON *pJson = raftStore2Json(pRaftStore);
+  char  *serialized = cJSON_Print(pJson);
+  cJSON_Delete(pJson);
+  return serialized;
+}
+
 // for debug -------------------
 void raftStorePrint(SRaftStore *pObj) {
-  char serialized[RAFT_STORE_BLOCK_SIZE];
+  char *serialized = raftStore2Str(pObj);
-  raftStoreSerialize(pObj, serialized, sizeof(serialized));
   printf("raftStorePrint | len:%lu | %s \n", strlen(serialized), serialized);
   fflush(NULL);
+  free(serialized);
 }
 
 void raftStorePrint2(char *s, SRaftStore *pObj) {
-  char serialized[RAFT_STORE_BLOCK_SIZE];
+  char *serialized = raftStore2Str(pObj);
-  raftStoreSerialize(pObj, serialized, sizeof(serialized));
   printf("raftStorePrint2 | len:%lu | %s | %s \n", strlen(serialized), s, serialized);
   fflush(NULL);
+  free(serialized);
 }
 void raftStoreLog(SRaftStore *pObj) {
-  char serialized[RAFT_STORE_BLOCK_SIZE];
+  char *serialized = raftStore2Str(pObj);
-  raftStoreSerialize(pObj, serialized, sizeof(serialized));
   sTrace("raftStoreLog | len:%lu | %s", strlen(serialized), serialized);
-  fflush(NULL);
+  free(serialized);
 }
 
 void raftStoreLog2(char *s, SRaftStore *pObj) {
-  char serialized[RAFT_STORE_BLOCK_SIZE];
+  char *serialized = raftStore2Str(pObj);
-  raftStoreSerialize(pObj, serialized, sizeof(serialized));
   sTrace("raftStoreLog2 | len:%lu | %s | %s", strlen(serialized), s, serialized);
-  fflush(NULL);
+  free(serialized);
 }

source/libs/sync/src/syncVoteMgr.c

@@ -82,6 +82,7 @@ cJSON *voteGranted2Json(SVotesGranted *pVotesGranted) {
   char   u64buf[128];
   cJSON *pRoot = cJSON_CreateObject();
 
+  if (pVotesGranted != NULL) {
     cJSON_AddNumberToObject(pRoot, "replicaNum", pVotesGranted->replicaNum);
     cJSON *pReplicas = cJSON_CreateArray();
     cJSON_AddItemToObject(pRoot, "replicas", pReplicas);
@@ -106,6 +107,7 @@ cJSON *voteGranted2Json(SVotesGranted *pVotesGranted) {
 
     bool majority = voteGrantedMajority(pVotesGranted);
     cJSON_AddNumberToObject(pRoot, "majority", majority);
+  }
 
   cJSON *pJson = cJSON_CreateObject();
   cJSON_AddItemToObject(pJson, "SVotesGranted", pRoot);
@@ -203,6 +205,7 @@ cJSON *votesRespond2Json(SVotesRespond *pVotesRespond) {
   char   u64buf[128];
   cJSON *pRoot = cJSON_CreateObject();
 
+  if (pVotesRespond != NULL) {
     cJSON_AddNumberToObject(pRoot, "replicaNum", pVotesRespond->replicaNum);
     cJSON *pReplicas = cJSON_CreateArray();
     cJSON_AddItemToObject(pRoot, "replicas", pReplicas);
@@ -226,6 +229,7 @@ cJSON *votesRespond2Json(SVotesRespond *pVotesRespond) {
     cJSON_AddStringToObject(pRoot, "term", u64buf);
     snprintf(u64buf, sizeof(u64buf), "%p", pVotesRespond->pSyncNode);
     cJSON_AddStringToObject(pRoot, "pSyncNode", u64buf);
+  }
 
   cJSON *pJson = cJSON_CreateObject();
   cJSON_AddItemToObject(pJson, "SVotesRespond", pRoot);

source/libs/sync/test/syncInitTest.cpp

@@ -47,6 +47,7 @@ SSyncNode* syncNodeInit() {
 
   gSyncIO->FpOnSyncPing = pSyncNode->FpOnPing;
   gSyncIO->FpOnSyncPingReply = pSyncNode->FpOnPingReply;
+  gSyncIO->FpOnSyncClientRequest = pSyncNode->FpOnClientRequest;
   gSyncIO->FpOnSyncRequestVote = pSyncNode->FpOnRequestVote;
   gSyncIO->FpOnSyncRequestVoteReply = pSyncNode->FpOnRequestVoteReply;
   gSyncIO->FpOnSyncAppendEntries = pSyncNode->FpOnAppendEntries;