diff --git a/include/common/tmsgdef.h b/include/common/tmsgdef.h index 24d259b0f2..b3c47353b8 100644 --- a/include/common/tmsgdef.h +++ b/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) diff --git a/include/libs/sync/syncTools.h b/include/libs/sync/syncTools.h index 327450c10a..0438d6d436 100644 --- a/include/libs/sync/syncTools.h +++ b/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 ---------------------- diff --git a/source/libs/sync/src/syncMessage.c b/source/libs/sync/src/syncMessage.c index 1299b37769..2b1b40997e 100644 --- a/source/libs/sync/src/syncMessage.c +++ b/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); +} -SyncApplyMsg* syncApplyMsgFromRpcMsg2(const SRpcMsg* pRpcMsg) { return NULL; } -cJSON* syncApplyMsg2Json(const SyncApplyMsg* pMsg) { return NULL; } -char* syncApplyMsg2Str(const SyncApplyMsg* pMsg) { return NULL; } +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); + } + + 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); +} diff --git a/source/libs/sync/test/CMakeLists.txt b/source/libs/sync/test/CMakeLists.txt index e2ac38150f..5d43d47358 100644 --- a/source/libs/sync/test/CMakeLists.txt +++ b/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() diff --git a/source/libs/sync/test/syncApplyMsgTest.cpp b/source/libs/sync/test/syncApplyMsgTest.cpp new file mode 100644 index 0000000000..7544613c5d --- /dev/null +++ b/source/libs/sync/test/syncApplyMsgTest.cpp @@ -0,0 +1,123 @@ +#include +#include +#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; +}