add SyncApplyMsg and test

include/common/tmsgdef.h

@@ -206,6 +206,7 @@ enum {
   TD_DEF_MSG_TYPE(TDMT_VND_CANCEL_SMA, "vnode-cancel-sma", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_VND_DROP_SMA, "vnode-drop-sma", NULL, NULL)
 
+  // sync integration
   TD_DEF_MSG_TYPE(TDMT_VND_SYNC_TIMEOUT, "vnode-sync-timeout", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_VND_SYNC_PING, "vnode-sync-ping", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_VND_SYNC_PING_REPLY, "vnode-sync-ping-reply", NULL, NULL)
@@ -218,6 +219,7 @@ enum {
   TD_DEF_MSG_TYPE(TDMT_VND_SYNC_NOOP, "vnode-sync-noop", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_VND_SYNC_UNKNOWN, "vnode-sync-unknown", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_VND_SYNC_COMMON_RESPONSE, "vnode-sync-common-response", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_VND_SYNC_APPLY_MSG, "vnode-sync-apply-msg", NULL, NULL)
 
   // Requests handled by QNODE
   TD_NEW_MSG_SEG(TDMT_QND_MSG)

include/libs/sync/syncTools.h

@@ -383,15 +383,15 @@ char*         syncApplyMsgSerialize2(const SyncApplyMsg* pMsg, uint32_t* len);
 SyncApplyMsg* syncApplyMsgDeserialize2(const char* buf, uint32_t len);
 void syncApplyMsg2RpcMsg(const SyncApplyMsg* pMsg, SRpcMsg* pRpcMsg);     // SyncApplyMsg to SRpcMsg, put it into ApplyQ
 void syncApplyMsgFromRpcMsg(const SRpcMsg* pRpcMsg, SyncApplyMsg* pMsg);  // get SRpcMsg from ApplyQ, to SyncApplyMsg
-void syncApplyMsg2OriginalRpcMsg(const SyncApplyMsg* pMsg, SRpcMsg* pOriginalRpcMsg);  // SyncApplyMsg to OriginalRpcMsg
 SyncApplyMsg* syncApplyMsgFromRpcMsg2(const SRpcMsg* pRpcMsg);
-cJSON*        syncApplyMsg2Json(const SyncApplyMsg* pMsg);
-char*         syncApplyMsg2Str(const SyncApplyMsg* pMsg);
+void syncApplyMsg2OriginalRpcMsg(const SyncApplyMsg* pMsg, SRpcMsg* pOriginalRpcMsg);  // SyncApplyMsg to OriginalRpcMsg
+cJSON* syncApplyMsg2Json(const SyncApplyMsg* pMsg);
+char*  syncApplyMsg2Str(const SyncApplyMsg* pMsg);
 
 // for debug ----------------------
 void syncApplyMsgPrint(const SyncApplyMsg* pMsg);
 void syncApplyMsgPrint2(char* s, const SyncApplyMsg* pMsg);
-void ssyncApplyMsgLog(const SyncApplyMsg* pMsg);
+void syncApplyMsgLog(const SyncApplyMsg* pMsg);
 void syncApplyMsgLog2(char* s, const SyncApplyMsg* pMsg);
 
 // on message ----------------------

source/libs/sync/src/syncMessage.c

@@ -1533,29 +1533,152 @@ void syncAppendEntriesReplyLog2(char* s, const SyncAppendEntriesReply* pMsg) {
 }
 
 // ---- message process SyncApplyMsg----
-SyncApplyMsg* syncApplyMsgBuild(uint32_t dataLen) { return NULL; }
-SyncApplyMsg* syncApplyMsgBuild2(const SRpcMsg* pOriginalRpcMsg, int32_t vgId, SFsmCbMeta* pMeta) { return NULL; }
-void          syncApplyMsgDestroy(SyncApplyMsg* pMsg) {}
-void          syncApplyMsgSerialize(const SyncApplyMsg* pMsg, char* buf, uint32_t bufLen) {}
-void          syncApplyMsgDeserialize(const char* buf, uint32_t len, SyncApplyMsg* pMsg) {}
-char*         syncApplyMsgSerialize2(const SyncApplyMsg* pMsg, uint32_t* len) { return NULL; }
-SyncApplyMsg* syncApplyMsgDeserialize2(const char* buf, uint32_t len) { return NULL; }
+SyncApplyMsg* syncApplyMsgBuild(uint32_t dataLen) {
+  uint32_t      bytes = sizeof(SyncApplyMsg) + dataLen;
+  SyncApplyMsg* pMsg = taosMemoryMalloc(bytes);
+  memset(pMsg, 0, bytes);
+  pMsg->bytes = bytes;
+  pMsg->msgType = TDMT_VND_SYNC_APPLY_MSG;
+  pMsg->dataLen = dataLen;
+  return pMsg;
+}
+
+SyncApplyMsg* syncApplyMsgBuild2(const SRpcMsg* pOriginalRpcMsg, int32_t vgId, SFsmCbMeta* pMeta) {
+  SyncApplyMsg* pMsg = syncApplyMsgBuild(pOriginalRpcMsg->contLen);
+  pMsg->vgId = vgId;
+  pMsg->originalRpcType = pOriginalRpcMsg->msgType;
+  pMsg->fsmMeta = *pMeta;
+  memcpy(pMsg->data, pOriginalRpcMsg->pCont, pOriginalRpcMsg->contLen);
+  return pMsg;
+}
+
+void syncApplyMsgDestroy(SyncApplyMsg* pMsg) {
+  if (pMsg != NULL) {
+    taosMemoryFree(pMsg);
+  }
+}
+
+void syncApplyMsgSerialize(const SyncApplyMsg* pMsg, char* buf, uint32_t bufLen) {
+  assert(pMsg->bytes <= bufLen);
+  memcpy(buf, pMsg, pMsg->bytes);
+}
+
+void syncApplyMsgDeserialize(const char* buf, uint32_t len, SyncApplyMsg* pMsg) {
+  memcpy(pMsg, buf, len);
+  assert(len == pMsg->bytes);
+}
+
+char* syncApplyMsgSerialize2(const SyncApplyMsg* pMsg, uint32_t* len) {
+  char* buf = taosMemoryMalloc(pMsg->bytes);
+  assert(buf != NULL);
+  syncApplyMsgSerialize(pMsg, buf, pMsg->bytes);
+  if (len != NULL) {
+    *len = pMsg->bytes;
+  }
+  return buf;
+}
+
+SyncApplyMsg* syncApplyMsgDeserialize2(const char* buf, uint32_t len) {
+  uint32_t      bytes = *((uint32_t*)buf);
+  SyncApplyMsg* pMsg = taosMemoryMalloc(bytes);
+  assert(pMsg != NULL);
+  syncApplyMsgDeserialize(buf, len, pMsg);
+  assert(len == pMsg->bytes);
+  return pMsg;
+}
 
 // SyncApplyMsg to SRpcMsg, put it into ApplyQ
-void syncApplyMsg2RpcMsg(const SyncApplyMsg* pMsg, SRpcMsg* pRpcMsg) {}
+void syncApplyMsg2RpcMsg(const SyncApplyMsg* pMsg, SRpcMsg* pRpcMsg) {
+  memset(pRpcMsg, 0, sizeof(*pRpcMsg));
+  pRpcMsg->msgType = pMsg->msgType;
+  pRpcMsg->contLen = pMsg->bytes;
+  pRpcMsg->pCont = rpcMallocCont(pRpcMsg->contLen);
+  syncApplyMsgSerialize(pMsg, pRpcMsg->pCont, pRpcMsg->contLen);
+}
 
 // get SRpcMsg from ApplyQ, to SyncApplyMsg
-void syncApplyMsgFromRpcMsg(const SRpcMsg* pRpcMsg, SyncApplyMsg* pMsg) {}
+void syncApplyMsgFromRpcMsg(const SRpcMsg* pRpcMsg, SyncApplyMsg* pMsg) {
+  syncApplyMsgDeserialize(pRpcMsg->pCont, pRpcMsg->contLen, pMsg);
+}
+
+SyncApplyMsg* syncApplyMsgFromRpcMsg2(const SRpcMsg* pRpcMsg) {
+  SyncApplyMsg* pMsg = syncApplyMsgDeserialize2(pRpcMsg->pCont, pRpcMsg->contLen);
+  return pMsg;
+}
 
 // SyncApplyMsg to OriginalRpcMsg
-void syncApplyMsg2OriginalRpcMsg(const SyncApplyMsg* pMsg, SRpcMsg* pOriginalRpcMsg) {}
+void syncApplyMsg2OriginalRpcMsg(const SyncApplyMsg* pMsg, SRpcMsg* pOriginalRpcMsg) {
+  memset(pOriginalRpcMsg, 0, sizeof(*pOriginalRpcMsg));
+  pOriginalRpcMsg->msgType = pMsg->originalRpcType;
+  pOriginalRpcMsg->contLen = pMsg->dataLen;
+  pOriginalRpcMsg->pCont = rpcMallocCont(pOriginalRpcMsg->contLen);
+  memcpy(pOriginalRpcMsg->pCont, pMsg->data, pOriginalRpcMsg->contLen);
+}
+
+cJSON* syncApplyMsg2Json(const SyncApplyMsg* pMsg) {
+  char   u64buf[128];
+  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_AddNumberToObject(pRoot, "originalRpcType", pMsg->originalRpcType);
+
+    snprintf(u64buf, sizeof(u64buf), "%ld", pMsg->fsmMeta.index);
+    cJSON_AddStringToObject(pRoot, "fsmMeta.index", u64buf);
+    cJSON_AddNumberToObject(pRoot, "fsmMeta.isWeak", pMsg->fsmMeta.isWeak);
+    cJSON_AddNumberToObject(pRoot, "fsmMeta.code", pMsg->fsmMeta.code);
+    cJSON_AddNumberToObject(pRoot, "fsmMeta.state", pMsg->fsmMeta.state);
+    cJSON_AddStringToObject(pRoot, "fsmMeta.state.str", syncUtilState2String(pMsg->fsmMeta.state));
+    snprintf(u64buf, sizeof(u64buf), "%lu", pMsg->fsmMeta.seqNum);
+    cJSON_AddStringToObject(pRoot, "fsmMeta.seqNum", u64buf);
+
+    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);
+  }
 
-SyncApplyMsg* syncApplyMsgFromRpcMsg2(const SRpcMsg* pRpcMsg) { return NULL; }
-cJSON*        syncApplyMsg2Json(const SyncApplyMsg* pMsg) { return NULL; }
-char*         syncApplyMsg2Str(const SyncApplyMsg* pMsg) { return NULL; }
+  cJSON* pJson = cJSON_CreateObject();
+  cJSON_AddItemToObject(pJson, "SyncApplyMsg", pRoot);
+  return pJson;
+}
+
+char* syncApplyMsg2Str(const SyncApplyMsg* pMsg) {
+  cJSON* pJson = syncApplyMsg2Json(pMsg);
+  char*  serialized = cJSON_Print(pJson);
+  cJSON_Delete(pJson);
+  return serialized;
+}
 
 // for debug ----------------------
-void syncApplyMsgPrint(const SyncApplyMsg* pMsg) {}
-void syncApplyMsgPrint2(char* s, const SyncApplyMsg* pMsg) {}
-void ssyncApplyMsgLog(const SyncApplyMsg* pMsg) {}
-void syncApplyMsgLog2(char* s, const SyncApplyMsg* pMsg) {}
+void syncApplyMsgPrint(const SyncApplyMsg* pMsg) {
+  char* serialized = syncApplyMsg2Str(pMsg);
+  printf("syncApplyMsgPrint | len:%lu | %s \n", strlen(serialized), serialized);
+  fflush(NULL);
+  taosMemoryFree(serialized);
+}
+
+void syncApplyMsgPrint2(char* s, const SyncApplyMsg* pMsg) {
+  char* serialized = syncApplyMsg2Str(pMsg);
+  printf("syncApplyMsgPrint2 | len:%lu | %s | %s \n", strlen(serialized), s, serialized);
+  fflush(NULL);
+  taosMemoryFree(serialized);
+}
+
+void syncApplyMsgLog(const SyncApplyMsg* pMsg) {
+  char* serialized = syncApplyMsg2Str(pMsg);
+  sTrace("ssyncApplyMsgLog | len:%lu | %s", strlen(serialized), serialized);
+  taosMemoryFree(serialized);
+}
+
+void syncApplyMsgLog2(char* s, const SyncApplyMsg* pMsg) {
+  char* serialized = syncApplyMsg2Str(pMsg);
+  sTrace("syncApplyMsgLog2 | len:%lu | %s | %s", strlen(serialized), s, serialized);
+  taosMemoryFree(serialized);
+}

source/libs/sync/test/CMakeLists.txt

@@ -41,6 +41,7 @@ add_executable(syncLogStoreCheck "")
 add_executable(syncRaftCfgTest "")
 add_executable(syncRespMgrTest "")
 add_executable(syncSnapshotTest "")
+add_executable(syncApplyMsgTest "")
 
 
 target_sources(syncTest
@@ -215,6 +216,10 @@ target_sources(syncSnapshotTest
     PRIVATE
     "syncSnapshotTest.cpp"
 )
+target_sources(syncApplyMsgTest
+    PRIVATE
+    "syncApplyMsgTest.cpp"
+)
 
 
 target_include_directories(syncTest
@@ -432,6 +437,11 @@ target_include_directories(syncSnapshotTest
     "${CMAKE_SOURCE_DIR}/include/libs/sync"
     "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
 )
+target_include_directories(syncApplyMsgTest
+    PUBLIC
+    "${CMAKE_SOURCE_DIR}/include/libs/sync"
+    "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
+)
 
 
 target_link_libraries(syncTest
@@ -606,6 +616,10 @@ target_link_libraries(syncSnapshotTest
     sync
     gtest_main
 )
+target_link_libraries(syncApplyMsgTest
+    sync
+    gtest_main
+)
 
 
 enable_testing()

source/libs/sync/test/syncApplyMsgTest.cpp

+#include <gtest/gtest.h>
+#include <stdio.h>
+#include "syncIO.h"
+#include "syncInt.h"
+#include "syncMessage.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");
+}
+
+SyncApplyMsg *createMsg() {
+  SRpcMsg rpcMsg;
+  memset(&rpcMsg, 0, sizeof(rpcMsg));
+  rpcMsg.msgType = 12345;
+  rpcMsg.contLen = 20;
+  rpcMsg.pCont = rpcMallocCont(rpcMsg.contLen);
+  strcpy((char *)rpcMsg.pCont, "hello rpc");
+
+  SFsmCbMeta meta;
+  meta.code = 11;
+  meta.index = 22;
+  meta.isWeak = 1;
+  meta.seqNum = 33;
+  meta.state = TAOS_SYNC_STATE_LEADER;
+
+  SyncApplyMsg *pMsg = syncApplyMsgBuild2(&rpcMsg, 123, &meta);
+  rpcFreeCont(rpcMsg.pCont);
+  return pMsg;
+}
+
+void test1() {
+  SyncApplyMsg *pMsg = createMsg();
+  syncApplyMsgLog2((char *)"test1:", pMsg);
+  syncApplyMsgDestroy(pMsg);
+}
+
+void test2() {
+  SyncApplyMsg *pMsg = createMsg();
+  uint32_t           len = pMsg->bytes;
+  char *             serialized = (char *)taosMemoryMalloc(len);
+  syncApplyMsgSerialize(pMsg, serialized, len);
+  SyncApplyMsg *pMsg2 = syncApplyMsgBuild(pMsg->dataLen);
+  syncApplyMsgDeserialize(serialized, len, pMsg2);
+  syncApplyMsgLog2((char *)"test2: syncApplyMsgSerialize -> syncApplyMsgDeserialize ", pMsg2);
+
+  taosMemoryFree(serialized);
+  syncApplyMsgDestroy(pMsg);
+  syncApplyMsgDestroy(pMsg2);
+}
+
+void test3() {
+  SyncApplyMsg *pMsg = createMsg();
+  uint32_t           len;
+  char *             serialized = syncApplyMsgSerialize2(pMsg, &len);
+  SyncApplyMsg *pMsg2 = syncApplyMsgDeserialize2(serialized, len);
+  syncApplyMsgLog2((char *)"test3: syncApplyMsgSerialize2 -> syncApplyMsgDeserialize2 ", pMsg2);
+
+  taosMemoryFree(serialized);
+  syncApplyMsgDestroy(pMsg);
+  syncApplyMsgDestroy(pMsg2);
+}
+
+void test4() {
+  SyncApplyMsg *pMsg = createMsg();
+  SRpcMsg            rpcMsg;
+  syncApplyMsg2RpcMsg(pMsg, &rpcMsg);
+  SyncApplyMsg *pMsg2 = (SyncApplyMsg *)taosMemoryMalloc(rpcMsg.contLen);
+  syncApplyMsgFromRpcMsg(&rpcMsg, pMsg2);
+  syncApplyMsgLog2((char *)"test4: syncApplyMsg2RpcMsg -> syncApplyMsgFromRpcMsg ", pMsg2);
+
+  rpcFreeCont(rpcMsg.pCont);
+  syncApplyMsgDestroy(pMsg);
+  syncApplyMsgDestroy(pMsg2);
+}
+
+void test5() {
+  SyncApplyMsg *pMsg = createMsg();
+  SRpcMsg            rpcMsg;
+  syncApplyMsg2RpcMsg(pMsg, &rpcMsg);
+  SyncApplyMsg *pMsg2 = syncApplyMsgFromRpcMsg2(&rpcMsg);
+  syncApplyMsgLog2((char *)"test5: syncClientRequest2RpcMsg -> syncApplyMsgFromRpcMsg2 ", pMsg2);
+
+  rpcFreeCont(rpcMsg.pCont);
+  syncApplyMsgDestroy(pMsg);
+  syncApplyMsgDestroy(pMsg2);
+}
+
+void test6() {
+  SyncApplyMsg *pMsg = createMsg();
+  SRpcMsg            rpcMsg;
+  syncApplyMsg2RpcMsg(pMsg, &rpcMsg);
+  SyncApplyMsg *pMsg2 = syncApplyMsgFromRpcMsg2(&rpcMsg);
+
+  SRpcMsg originalRpcMsg;
+  syncApplyMsg2OriginalRpcMsg(pMsg2, &originalRpcMsg);
+  syncRpcMsgLog2((char*)"test6", &originalRpcMsg);
+
+  rpcFreeCont(originalRpcMsg.pCont);
+  rpcFreeCont(rpcMsg.pCont);
+  syncApplyMsgDestroy(pMsg);
+  syncApplyMsgDestroy(pMsg2);
+}
+
+int main() {
+  tsAsyncLog = 0;
+  sDebugFlag = DEBUG_TRACE + DEBUG_SCREEN + DEBUG_FILE;
+  logTest();
+
+  test1();
+  test2();
+  test3();
+  test4();
+  test5();
+  test6();
+
+  return 0;
+}