From 690dabd61795b178eec16dddb9a7a6a02a6827df Mon Sep 17 00:00:00 2001 From: Minghao Li Date: Fri, 11 Mar 2022 19:11:38 +0800 Subject: [PATCH] sync refactor --- source/libs/sync/inc/syncMessage.h | 31 +++- source/libs/sync/src/syncMessage.c | 174 +++++++++++++++--- source/libs/sync/test/CMakeLists.txt | 38 +++- .../libs/sync/test/syncAppendEntriesTest.cpp | 2 +- .../libs/sync/test/syncClientRequestTest.cpp | 4 +- source/libs/sync/test/syncPingReplyTest.cpp | 95 ++++++++++ source/libs/sync/test/syncPingTest.cpp | 159 +++++++--------- source/libs/sync/test/syncPingTimerTest.cpp | 130 +++++++++++++ source/libs/sync/test/syncTimeoutTest.cpp | 12 +- 9 files changed, 504 insertions(+), 141 deletions(-) create mode 100644 source/libs/sync/test/syncPingReplyTest.cpp create mode 100644 source/libs/sync/test/syncPingTimerTest.cpp diff --git a/source/libs/sync/inc/syncMessage.h b/source/libs/sync/inc/syncMessage.h index a3b7517bdc..3f19593e50 100644 --- a/source/libs/sync/inc/syncMessage.h +++ b/source/libs/sync/inc/syncMessage.h @@ -94,14 +94,24 @@ typedef struct SyncPing { } SyncPing; SyncPing* syncPingBuild(uint32_t dataLen); +SyncPing* syncPingBuild2(const SRaftId* srcId, const SRaftId* destId, const char* str); +SyncPing* syncPingBuild3(const SRaftId* srcId, const SRaftId* destId); void syncPingDestroy(SyncPing* pMsg); void syncPingSerialize(const SyncPing* pMsg, char* buf, uint32_t bufLen); void syncPingDeserialize(const char* buf, uint32_t len, SyncPing* pMsg); +char* syncPingSerialize2(const SyncPing* pMsg, uint32_t* len); +SyncPing* syncPingDeserialize2(const char* buf, uint32_t len); void syncPing2RpcMsg(const SyncPing* pMsg, SRpcMsg* pRpcMsg); void syncPingFromRpcMsg(const SRpcMsg* pRpcMsg, SyncPing* pMsg); +SyncPing* syncPingFromRpcMsg2(const SRpcMsg* pRpcMsg); cJSON* syncPing2Json(const SyncPing* pMsg); -SyncPing* syncPingBuild2(const SRaftId* srcId, const SRaftId* destId, const char* str); -SyncPing* syncPingBuild3(const SRaftId* srcId, const SRaftId* destId); +char* syncPing2Str(const SyncPing* pMsg); + +// for debug ---------------------- +void syncPingPrint(const SyncPing* pMsg); +void syncPingPrint2(char* s, const SyncPing* pMsg); +void syncPingLog(const SyncPing* pMsg); +void syncPingLog2(char* s, const SyncPing* pMsg); // --------------------------------------------- typedef struct SyncPingReply { @@ -114,18 +124,25 @@ typedef struct SyncPingReply { char data[]; } SyncPingReply; -#define SYNC_PING_REPLY_FIX_LEN \ - (sizeof(uint32_t) + sizeof(uint32_t) + sizeof(SRaftId) + sizeof(SRaftId) + sizeof(uint32_t)) - SyncPingReply* syncPingReplyBuild(uint32_t dataLen); +SyncPingReply* syncPingReplyBuild2(const SRaftId* srcId, const SRaftId* destId, const char* str); +SyncPingReply* syncPingReplyBuild3(const SRaftId* srcId, const SRaftId* destId); void syncPingReplyDestroy(SyncPingReply* pMsg); void syncPingReplySerialize(const SyncPingReply* pMsg, char* buf, uint32_t bufLen); void syncPingReplyDeserialize(const char* buf, uint32_t len, SyncPingReply* pMsg); +char* syncPingReplySerialize2(const SyncPingReply* pMsg, uint32_t* len); // +SyncPingReply* syncPingReplyDeserialize2(const char* buf, uint32_t len); // void syncPingReply2RpcMsg(const SyncPingReply* pMsg, SRpcMsg* pRpcMsg); void syncPingReplyFromRpcMsg(const SRpcMsg* pRpcMsg, SyncPingReply* pMsg); +SyncPingReply* syncPingReplyFromRpcMsg2(const SRpcMsg* pRpcMsg); // cJSON* syncPingReply2Json(const SyncPingReply* pMsg); -SyncPingReply* syncPingReplyBuild2(const SRaftId* srcId, const SRaftId* destId, const char* str); -SyncPingReply* syncPingReplyBuild3(const SRaftId* srcId, const SRaftId* destId); +char* syncPingReply2Str(const SyncPingReply* pMsg); // + +// for debug ---------------------- +void syncPingReplyPrint(const SyncPingReply* pMsg); +void syncPingReplyPrint2(char* s, const SyncPingReply* pMsg); +void syncPingReplyLog(const SyncPingReply* pMsg); +void syncPingReplyLog2(char* s, const SyncPingReply* pMsg); // --------------------------------------------- typedef struct SyncClientRequest { diff --git a/source/libs/sync/src/syncMessage.c b/source/libs/sync/src/syncMessage.c index 3d5d302559..477ba1929e 100644 --- a/source/libs/sync/src/syncMessage.c +++ b/source/libs/sync/src/syncMessage.c @@ -217,6 +217,20 @@ SyncPing* syncPingBuild(uint32_t dataLen) { return pMsg; } +SyncPing* syncPingBuild2(const SRaftId* srcId, const SRaftId* destId, const char* str) { + uint32_t dataLen = strlen(str) + 1; + SyncPing* pMsg = syncPingBuild(dataLen); + pMsg->srcId = *srcId; + pMsg->destId = *destId; + snprintf(pMsg->data, pMsg->dataLen, "%s", str); + return pMsg; +} + +SyncPing* syncPingBuild3(const SRaftId* srcId, const SRaftId* destId) { + SyncPing* pMsg = syncPingBuild2(srcId, destId, "ping"); + return pMsg; +} + void syncPingDestroy(SyncPing* pMsg) { if (pMsg != NULL) { free(pMsg); @@ -234,6 +248,25 @@ 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 = malloc(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 = malloc(bytes); + assert(pMsg != NULL); + syncPingDeserialize(buf, len, pMsg); + assert(len == pMsg->bytes); + return pMsg; +} + void syncPing2RpcMsg(const SyncPing* pMsg, SRpcMsg* pRpcMsg) { memset(pRpcMsg, 0, sizeof(*pRpcMsg)); pRpcMsg->msgType = pMsg->msgType; @@ -246,6 +279,11 @@ 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); + return pMsg; +} + cJSON* syncPing2Json(const SyncPing* pMsg) { char u64buf[128]; @@ -284,30 +322,56 @@ cJSON* syncPing2Json(const SyncPing* pMsg) { cJSON_AddItemToObject(pRoot, "destId", pDestId); cJSON_AddNumberToObject(pRoot, "dataLen", pMsg->dataLen); - cJSON_AddStringToObject(pRoot, "data", pMsg->data); + char* s; + s = syncUtilprintBin((char*)(pMsg->data), pMsg->dataLen); + cJSON_AddStringToObject(pRoot, "data", s); + free(s); + s = syncUtilprintBin2((char*)(pMsg->data), pMsg->dataLen); + cJSON_AddStringToObject(pRoot, "data2", s); + free(s); cJSON* pJson = cJSON_CreateObject(); cJSON_AddItemToObject(pJson, "SyncPing", pRoot); return pJson; } -SyncPing* syncPingBuild2(const SRaftId* srcId, const SRaftId* destId, const char* str) { - uint32_t dataLen = strlen(str) + 1; - SyncPing* pMsg = syncPingBuild(dataLen); - pMsg->srcId = *srcId; - pMsg->destId = *destId; - snprintf(pMsg->data, pMsg->dataLen, "%s", str); - return pMsg; +char* syncPing2Str(const SyncPing* pMsg) { + cJSON* pJson = syncPing2Json(pMsg); + char* serialized = cJSON_Print(pJson); + cJSON_Delete(pJson); + return serialized; } -SyncPing* syncPingBuild3(const SRaftId* srcId, const SRaftId* destId) { - SyncPing* pMsg = syncPingBuild2(srcId, destId, "ping"); - return pMsg; +// for debug ---------------------- +void syncPingPrint(const SyncPing* pMsg) { + char* serialized = syncPing2Str(pMsg); + printf("syncPingPrint | len:%lu | %s \n", strlen(serialized), serialized); + fflush(NULL); + free(serialized); +} + +void syncPingPrint2(char* s, const SyncPing* pMsg) { + char* serialized = syncPing2Str(pMsg); + printf("syncPingPrint2 | len:%lu | %s | %s \n", strlen(serialized), s, serialized); + fflush(NULL); + free(serialized); +} + +void syncPingLog(const SyncPing* pMsg) { + char* serialized = syncPing2Str(pMsg); + sTrace("syncPingLog | len:%lu | %s", strlen(serialized), serialized); + free(serialized); +} + +void syncPingLog2(char* s, const SyncPing* pMsg) { + char* serialized = syncPing2Str(pMsg); + sTrace("syncPingLog2 | len:%lu | %s | %s", strlen(serialized), s, serialized); + free(serialized); } // ---- message process SyncPingReply---- SyncPingReply* syncPingReplyBuild(uint32_t dataLen) { - uint32_t bytes = SYNC_PING_REPLY_FIX_LEN + dataLen; + uint32_t bytes = sizeof(SyncPingReply) + dataLen; SyncPingReply* pMsg = malloc(bytes); memset(pMsg, 0, bytes); pMsg->bytes = bytes; @@ -316,6 +380,20 @@ SyncPingReply* syncPingReplyBuild(uint32_t dataLen) { return pMsg; } +SyncPingReply* syncPingReplyBuild2(const SRaftId* srcId, const SRaftId* destId, const char* str) { + uint32_t dataLen = strlen(str) + 1; + SyncPingReply* pMsg = syncPingReplyBuild(dataLen); + pMsg->srcId = *srcId; + pMsg->destId = *destId; + snprintf(pMsg->data, pMsg->dataLen, "%s", str); + return pMsg; +} + +SyncPingReply* syncPingReplyBuild3(const SRaftId* srcId, const SRaftId* destId) { + SyncPingReply* pMsg = syncPingReplyBuild2(srcId, destId, "pang"); + return pMsg; +} + void syncPingReplyDestroy(SyncPingReply* pMsg) { if (pMsg != NULL) { free(pMsg); @@ -333,6 +411,25 @@ void syncPingReplyDeserialize(const char* buf, uint32_t len, SyncPingReply* pMsg assert(pMsg->bytes == sizeof(SyncPing) + pMsg->dataLen); } +char* syncPingReplySerialize2(const SyncPingReply* pMsg, uint32_t* len) { + char* buf = malloc(pMsg->bytes); + assert(buf != NULL); + syncPingReplySerialize(pMsg, buf, pMsg->bytes); + if (len != NULL) { + *len = pMsg->bytes; + } + return buf; +} + +SyncPingReply* syncPingReplyDeserialize2(const char* buf, uint32_t len) { + uint32_t bytes = *((uint32_t*)buf); + SyncPingReply* pMsg = malloc(bytes); + assert(pMsg != NULL); + syncPingReplyDeserialize(buf, len, pMsg); + assert(len == pMsg->bytes); + return pMsg; +} + void syncPingReply2RpcMsg(const SyncPingReply* pMsg, SRpcMsg* pRpcMsg) { memset(pRpcMsg, 0, sizeof(*pRpcMsg)); pRpcMsg->msgType = pMsg->msgType; @@ -345,6 +442,11 @@ void syncPingReplyFromRpcMsg(const SRpcMsg* pRpcMsg, SyncPingReply* pMsg) { syncPingReplyDeserialize(pRpcMsg->pCont, pRpcMsg->contLen, pMsg); } +SyncPingReply* syncPingReplyFromRpcMsg2(const SRpcMsg* pRpcMsg) { + SyncPingReply* pMsg = syncPingReplyDeserialize2(pRpcMsg->pCont, pRpcMsg->contLen); + return pMsg; +} + cJSON* syncPingReply2Json(const SyncPingReply* pMsg) { char u64buf[128]; @@ -383,25 +485,51 @@ cJSON* syncPingReply2Json(const SyncPingReply* pMsg) { cJSON_AddItemToObject(pRoot, "destId", pDestId); cJSON_AddNumberToObject(pRoot, "dataLen", pMsg->dataLen); - cJSON_AddStringToObject(pRoot, "data", pMsg->data); + char* s; + s = syncUtilprintBin((char*)(pMsg->data), pMsg->dataLen); + cJSON_AddStringToObject(pRoot, "data", s); + free(s); + s = syncUtilprintBin2((char*)(pMsg->data), pMsg->dataLen); + cJSON_AddStringToObject(pRoot, "data2", s); + free(s); cJSON* pJson = cJSON_CreateObject(); cJSON_AddItemToObject(pJson, "SyncPingReply", pRoot); return pJson; } -SyncPingReply* syncPingReplyBuild2(const SRaftId* srcId, const SRaftId* destId, const char* str) { - uint32_t dataLen = strlen(str) + 1; - SyncPingReply* pMsg = syncPingReplyBuild(dataLen); - pMsg->srcId = *srcId; - pMsg->destId = *destId; - snprintf(pMsg->data, pMsg->dataLen, "%s", str); - return pMsg; +char* syncPingReply2Str(const SyncPingReply* pMsg) { + cJSON* pJson = syncPingReply2Json(pMsg); + char* serialized = cJSON_Print(pJson); + cJSON_Delete(pJson); + return serialized; } -SyncPingReply* syncPingReplyBuild3(const SRaftId* srcId, const SRaftId* destId) { - SyncPingReply* pMsg = syncPingReplyBuild2(srcId, destId, "pang"); - return pMsg; +// for debug ---------------------- +void syncPingReplyPrint(const SyncPingReply* pMsg) { + char* serialized = syncPingReply2Str(pMsg); + printf("syncPingReplyPrint | len:%lu | %s \n", strlen(serialized), serialized); + fflush(NULL); + free(serialized); +} + +void syncPingReplyPrint2(char* s, const SyncPingReply* pMsg) { + char* serialized = syncPingReply2Str(pMsg); + printf("syncPingReplyPrint2 | len:%lu | %s | %s \n", strlen(serialized), s, serialized); + fflush(NULL); + free(serialized); +} + +void syncPingReplyLog(const SyncPingReply* pMsg) { + char* serialized = syncPingReply2Str(pMsg); + sTrace("syncPingReplyLog | len:%lu | %s", strlen(serialized), serialized); + free(serialized); +} + +void syncPingReplyLog2(char* s, const SyncPingReply* pMsg) { + char* serialized = syncPingReply2Str(pMsg); + sTrace("syncPingReplyLog2 | len:%lu | %s | %s", strlen(serialized), s, serialized); + free(serialized); } // ---- message process SyncClientRequest---- diff --git a/source/libs/sync/test/CMakeLists.txt b/source/libs/sync/test/CMakeLists.txt index 2ed4799467..3a87d3f2e2 100644 --- a/source/libs/sync/test/CMakeLists.txt +++ b/source/libs/sync/test/CMakeLists.txt @@ -1,6 +1,6 @@ add_executable(syncTest "") add_executable(syncEnvTest "") -add_executable(syncPingTest "") +add_executable(syncPingTimerTest "") add_executable(syncEncodeTest "") add_executable(syncIOTickQTest "") add_executable(syncIOTickPingTest "") @@ -23,6 +23,8 @@ add_executable(syncAppendEntriesTest "") add_executable(syncAppendEntriesReplyTest "") add_executable(syncClientRequestTest "") add_executable(syncTimeoutTest "") +add_executable(syncPingTest "") +add_executable(syncPingReplyTest "") target_sources(syncTest @@ -33,9 +35,9 @@ target_sources(syncEnvTest PRIVATE "syncEnvTest.cpp" ) -target_sources(syncPingTest +target_sources(syncPingTimerTest PRIVATE - "syncPingTest.cpp" + "syncPingTimerTest.cpp" ) target_sources(syncEncodeTest PRIVATE @@ -125,6 +127,14 @@ target_sources(syncTimeoutTest PRIVATE "syncTimeoutTest.cpp" ) +target_sources(syncPingTest + PRIVATE + "syncPingTest.cpp" +) +target_sources(syncPingReplyTest + PRIVATE + "syncPingReplyTest.cpp" +) target_include_directories(syncTest @@ -137,7 +147,7 @@ target_include_directories(syncEnvTest "${CMAKE_SOURCE_DIR}/include/libs/sync" "${CMAKE_CURRENT_SOURCE_DIR}/../inc" ) -target_include_directories(syncPingTest +target_include_directories(syncPingTimerTest PUBLIC "${CMAKE_SOURCE_DIR}/include/libs/sync" "${CMAKE_CURRENT_SOURCE_DIR}/../inc" @@ -252,6 +262,16 @@ target_include_directories(syncTimeoutTest "${CMAKE_SOURCE_DIR}/include/libs/sync" "${CMAKE_CURRENT_SOURCE_DIR}/../inc" ) +target_include_directories(syncPingTest + PUBLIC + "${CMAKE_SOURCE_DIR}/include/libs/sync" + "${CMAKE_CURRENT_SOURCE_DIR}/../inc" +) +target_include_directories(syncPingReplyTest + PUBLIC + "${CMAKE_SOURCE_DIR}/include/libs/sync" + "${CMAKE_CURRENT_SOURCE_DIR}/../inc" +) target_link_libraries(syncTest @@ -262,7 +282,7 @@ target_link_libraries(syncEnvTest sync gtest_main ) -target_link_libraries(syncPingTest +target_link_libraries(syncPingTimerTest sync gtest_main ) @@ -354,6 +374,14 @@ target_link_libraries(syncTimeoutTest sync gtest_main ) +target_link_libraries(syncPingTest + sync + gtest_main +) +target_link_libraries(syncPingReplyTest + sync + gtest_main +) enable_testing() diff --git a/source/libs/sync/test/syncAppendEntriesTest.cpp b/source/libs/sync/test/syncAppendEntriesTest.cpp index 6fa3041919..687d9bcb94 100644 --- a/source/libs/sync/test/syncAppendEntriesTest.cpp +++ b/source/libs/sync/test/syncAppendEntriesTest.cpp @@ -38,7 +38,7 @@ void test2() { uint32_t len = pMsg->bytes; char * serialized = (char *)malloc(len); syncAppendEntriesSerialize(pMsg, serialized, len); - SyncAppendEntries *pMsg2 = syncAppendEntriesBuild(pMsg2->dataLen); + SyncAppendEntries *pMsg2 = syncAppendEntriesBuild(pMsg->dataLen); syncAppendEntriesDeserialize(serialized, len, pMsg2); syncAppendEntriesPrint2((char *)"test2: syncAppendEntriesSerialize -> syncAppendEntriesDeserialize ", pMsg2); diff --git a/source/libs/sync/test/syncClientRequestTest.cpp b/source/libs/sync/test/syncClientRequestTest.cpp index 311a7e5fa3..6323f53a03 100644 --- a/source/libs/sync/test/syncClientRequestTest.cpp +++ b/source/libs/sync/test/syncClientRequestTest.cpp @@ -20,7 +20,7 @@ SyncClientRequest *createMsg() { rpcMsg.msgType = 12345; rpcMsg.contLen = 20; rpcMsg.pCont = rpcMallocCont(rpcMsg.contLen); - strcpy((char*)rpcMsg.pCont, "hello rpc"); + strcpy((char *)rpcMsg.pCont, "hello rpc"); SyncClientRequest *pMsg = syncClientRequestBuild2(&rpcMsg, 123, true); return pMsg; } @@ -36,7 +36,7 @@ void test2() { uint32_t len = pMsg->bytes; char * serialized = (char *)malloc(len); syncClientRequestSerialize(pMsg, serialized, len); - SyncClientRequest *pMsg2 = syncClientRequestBuild(pMsg2->dataLen); + SyncClientRequest *pMsg2 = syncClientRequestBuild(pMsg->dataLen); syncClientRequestDeserialize(serialized, len, pMsg2); syncClientRequestPrint2((char *)"test2: syncClientRequestSerialize -> syncClientRequestDeserialize ", pMsg2); diff --git a/source/libs/sync/test/syncPingReplyTest.cpp b/source/libs/sync/test/syncPingReplyTest.cpp new file mode 100644 index 0000000000..27f88b698c --- /dev/null +++ b/source/libs/sync/test/syncPingReplyTest.cpp @@ -0,0 +1,95 @@ +#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"); +} + +SyncPingReply *createMsg() { + SRaftId srcId, destId; + srcId.addr = syncUtilAddr2U64("127.0.0.1", 1234); + srcId.vgId = 100; + destId.addr = syncUtilAddr2U64("127.0.0.1", 5678); + destId.vgId = 100; + SyncPingReply *pMsg = syncPingReplyBuild3(&srcId, &destId); + return pMsg; +} + +void test1() { + SyncPingReply *pMsg = createMsg(); + syncPingReplyPrint2((char *)"test1:", pMsg); + syncPingReplyDestroy(pMsg); +} + +void test2() { + SyncPingReply *pMsg = createMsg(); + uint32_t len = pMsg->bytes; + char * serialized = (char *)malloc(len); + syncPingReplySerialize(pMsg, serialized, len); + SyncPingReply *pMsg2 = syncPingReplyBuild(pMsg->dataLen); + syncPingReplyDeserialize(serialized, len, pMsg2); + syncPingReplyPrint2((char *)"test2: syncPingReplySerialize -> syncPingReplyDeserialize ", pMsg2); + + free(serialized); + syncPingReplyDestroy(pMsg); + syncPingReplyDestroy(pMsg2); +} + +void test3() { + SyncPingReply *pMsg = createMsg(); + uint32_t len; + char * serialized = syncPingReplySerialize2(pMsg, &len); + SyncPingReply *pMsg2 = syncPingReplyDeserialize2(serialized, len); + syncPingReplyPrint2((char *)"test3: syncPingReplySerialize3 -> syncPingReplyDeserialize2 ", pMsg2); + + free(serialized); + syncPingReplyDestroy(pMsg); + syncPingReplyDestroy(pMsg2); +} + +void test4() { + SyncPingReply *pMsg = createMsg(); + SRpcMsg rpcMsg; + syncPingReply2RpcMsg(pMsg, &rpcMsg); + SyncPingReply *pMsg2 = (SyncPingReply *)malloc(rpcMsg.contLen); + syncPingReplyFromRpcMsg(&rpcMsg, pMsg2); + syncPingReplyPrint2((char *)"test4: syncPingReply2RpcMsg -> syncPingReplyFromRpcMsg ", pMsg2); + + syncPingReplyDestroy(pMsg); + syncPingReplyDestroy(pMsg2); +} + +void test5() { + SyncPingReply *pMsg = createMsg(); + SRpcMsg rpcMsg; + syncPingReply2RpcMsg(pMsg, &rpcMsg); + SyncPingReply *pMsg2 = syncPingReplyFromRpcMsg2(&rpcMsg); + syncPingReplyPrint2((char *)"test5: syncPingReply2RpcMsg -> syncPingReplyFromRpcMsg2 ", pMsg2); + + syncPingReplyDestroy(pMsg); + syncPingReplyDestroy(pMsg2); +} + +int main() { + // taosInitLog((char *)"syncTest.log", 100000, 10); + tsAsyncLog = 0; + sDebugFlag = 143 + 64; + logTest(); + + test1(); + test2(); + test3(); + test4(); + test5(); + + return 0; +} diff --git a/source/libs/sync/test/syncPingTest.cpp b/source/libs/sync/test/syncPingTest.cpp index e69878632f..8776d2de52 100644 --- a/source/libs/sync/test/syncPingTest.cpp +++ b/source/libs/sync/test/syncPingTest.cpp @@ -1,9 +1,8 @@ #include #include -#include "syncEnv.h" #include "syncIO.h" #include "syncInt.h" -#include "syncRaftStore.h" +#include "syncMessage.h" #include "syncUtil.h" void logTest() { @@ -15,116 +14,82 @@ void logTest() { sFatal("--- sync log test: fatal"); } -uint16_t ports[] = {7010, 7110, 7210, 7310, 7410}; -int32_t replicaNum = 3; -int32_t myIndex = 0; - -SRaftId ids[TSDB_MAX_REPLICA]; -SSyncInfo syncInfo; -SSyncFSM* pFsm; +SyncPing *createMsg() { + SRaftId srcId, destId; + srcId.addr = syncUtilAddr2U64("127.0.0.1", 1234); + srcId.vgId = 100; + destId.addr = syncUtilAddr2U64("127.0.0.1", 5678); + destId.vgId = 100; + SyncPing *pMsg = syncPingBuild3(&srcId, &destId); + return pMsg; +} -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", "./"); +void test1() { + SyncPing *pMsg = createMsg(); + syncPingPrint2((char *)"test1:", pMsg); + syncPingDestroy(pMsg); +} - SSyncCfg* pCfg = &syncInfo.syncCfg; - pCfg->myIndex = myIndex; - pCfg->replicaNum = replicaNum; +void test2() { + SyncPing *pMsg = createMsg(); + uint32_t len = pMsg->bytes; + char * serialized = (char *)malloc(len); + syncPingSerialize(pMsg, serialized, len); + SyncPing *pMsg2 = syncPingBuild(pMsg->dataLen); + syncPingDeserialize(serialized, len, pMsg2); + syncPingPrint2((char *)"test2: syncPingSerialize -> syncPingDeserialize ", pMsg2); + + free(serialized); + syncPingDestroy(pMsg); + syncPingDestroy(pMsg2); +} - 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); - } +void test3() { + SyncPing *pMsg = createMsg(); + uint32_t len; + char * serialized = syncPingSerialize2(pMsg, &len); + SyncPing *pMsg2 = syncPingDeserialize2(serialized, len); + syncPingPrint2((char *)"test3: syncPingSerialize3 -> syncPingDeserialize2 ", pMsg2); - SSyncNode* pSyncNode = syncNodeOpen(&syncInfo); - assert(pSyncNode != NULL); + free(serialized); + syncPingDestroy(pMsg); + syncPingDestroy(pMsg2); +} - 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->FpOnSyncTimeout = pSyncNode->FpOnTimeout; - gSyncIO->pSyncNode = pSyncNode; +void test4() { + SyncPing *pMsg = createMsg(); + SRpcMsg rpcMsg; + syncPing2RpcMsg(pMsg, &rpcMsg); + SyncPing *pMsg2 = (SyncPing *)malloc(rpcMsg.contLen); + syncPingFromRpcMsg(&rpcMsg, pMsg2); + syncPingPrint2((char *)"test4: syncPing2RpcMsg -> syncPingFromRpcMsg ", pMsg2); - return pSyncNode; + syncPingDestroy(pMsg); + syncPingDestroy(pMsg2); } -SSyncNode* syncInitTest() { return syncNodeInit(); } +void test5() { + SyncPing *pMsg = createMsg(); + SRpcMsg rpcMsg; + syncPing2RpcMsg(pMsg, &rpcMsg); + SyncPing *pMsg2 = syncPingFromRpcMsg2(&rpcMsg); + syncPingPrint2((char *)"test5: syncPing2RpcMsg -> syncPingFromRpcMsg2 ", pMsg2); -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); - } + syncPingDestroy(pMsg); + syncPingDestroy(pMsg2); } -int main(int argc, char** argv) { +int main() { // taosInitLog((char *)"syncTest.log", 100000, 10); tsAsyncLog = 0; sDebugFlag = 143 + 64; + logTest(); - 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); - - SSyncNode* pSyncNode = syncInitTest(); - assert(pSyncNode != NULL); - syncNodePrint2((char*)"----1", pSyncNode); - - initRaftId(pSyncNode); - - //--------------------------- - - sTrace("syncNodeStartPingTimer ..."); - ret = syncNodeStartPingTimer(pSyncNode); - assert(ret == 0); - syncNodePrint2((char*)"----2", pSyncNode); - - sTrace("sleep ..."); - taosMsleep(10000); - - sTrace("syncNodeStopPingTimer ..."); - ret = syncNodeStopPingTimer(pSyncNode); - assert(ret == 0); - syncNodePrint2((char*)"----3", pSyncNode); - - sTrace("sleep ..."); - taosMsleep(5000); - - sTrace("syncNodeStartPingTimer ..."); - ret = syncNodeStartPingTimer(pSyncNode); - assert(ret == 0); - syncNodePrint2((char*)"----4", pSyncNode); - - sTrace("sleep ..."); - taosMsleep(10000); - - sTrace("syncNodeStopPingTimer ..."); - ret = syncNodeStopPingTimer(pSyncNode); - assert(ret == 0); - syncNodePrint2((char*)"----5", pSyncNode); - - while (1) { - sTrace("while 1 sleep ..."); - taosMsleep(1000); - } + test1(); + test2(); + test3(); + test4(); + test5(); return 0; } diff --git a/source/libs/sync/test/syncPingTimerTest.cpp b/source/libs/sync/test/syncPingTimerTest.cpp new file mode 100644 index 0000000000..e69878632f --- /dev/null +++ b/source/libs/sync/test/syncPingTimerTest.cpp @@ -0,0 +1,130 @@ +#include +#include +#include "syncEnv.h" +#include "syncIO.h" +#include "syncInt.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 = 3; +int32_t myIndex = 0; + +SRaftId ids[TSDB_MAX_REPLICA]; +SSyncInfo syncInfo; +SSyncFSM* pFsm; + +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", "./"); + + 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); + } + + SSyncNode* 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->FpOnSyncTimeout = pSyncNode->FpOnTimeout; + gSyncIO->pSyncNode = pSyncNode; + + return pSyncNode; +} + +SSyncNode* syncInitTest() { return syncNodeInit(); } + +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); + + SSyncNode* pSyncNode = syncInitTest(); + assert(pSyncNode != NULL); + syncNodePrint2((char*)"----1", pSyncNode); + + initRaftId(pSyncNode); + + //--------------------------- + + sTrace("syncNodeStartPingTimer ..."); + ret = syncNodeStartPingTimer(pSyncNode); + assert(ret == 0); + syncNodePrint2((char*)"----2", pSyncNode); + + sTrace("sleep ..."); + taosMsleep(10000); + + sTrace("syncNodeStopPingTimer ..."); + ret = syncNodeStopPingTimer(pSyncNode); + assert(ret == 0); + syncNodePrint2((char*)"----3", pSyncNode); + + sTrace("sleep ..."); + taosMsleep(5000); + + sTrace("syncNodeStartPingTimer ..."); + ret = syncNodeStartPingTimer(pSyncNode); + assert(ret == 0); + syncNodePrint2((char*)"----4", pSyncNode); + + sTrace("sleep ..."); + taosMsleep(10000); + + sTrace("syncNodeStopPingTimer ..."); + ret = syncNodeStopPingTimer(pSyncNode); + assert(ret == 0); + syncNodePrint2((char*)"----5", pSyncNode); + + while (1) { + sTrace("while 1 sleep ..."); + taosMsleep(1000); + } + + return 0; +} diff --git a/source/libs/sync/test/syncTimeoutTest.cpp b/source/libs/sync/test/syncTimeoutTest.cpp index 4d2da97a00..3f46ab5c7c 100644 --- a/source/libs/sync/test/syncTimeoutTest.cpp +++ b/source/libs/sync/test/syncTimeoutTest.cpp @@ -29,8 +29,8 @@ void test1() { void test2() { SyncTimeout *pMsg = createMsg(); - uint32_t len = pMsg->bytes; - char * serialized = (char *)malloc(len); + uint32_t len = pMsg->bytes; + char * serialized = (char *)malloc(len); syncTimeoutSerialize(pMsg, serialized, len); SyncTimeout *pMsg2 = syncTimeoutBuild(); syncTimeoutDeserialize(serialized, len, pMsg2); @@ -43,8 +43,8 @@ void test2() { void test3() { SyncTimeout *pMsg = createMsg(); - uint32_t len; - char * serialized = syncTimeoutSerialize2(pMsg, &len); + uint32_t len; + char * serialized = syncTimeoutSerialize2(pMsg, &len); SyncTimeout *pMsg2 = syncTimeoutDeserialize2(serialized, len); syncTimeoutPrint2((char *)"test3: syncTimeoutSerialize3 -> syncTimeoutDeserialize2 ", pMsg2); @@ -55,7 +55,7 @@ void test3() { void test4() { SyncTimeout *pMsg = createMsg(); - SRpcMsg rpcMsg; + SRpcMsg rpcMsg; syncTimeout2RpcMsg(pMsg, &rpcMsg); SyncTimeout *pMsg2 = (SyncTimeout *)malloc(rpcMsg.contLen); syncTimeoutFromRpcMsg(&rpcMsg, pMsg2); @@ -67,7 +67,7 @@ void test4() { void test5() { SyncTimeout *pMsg = createMsg(); - SRpcMsg rpcMsg; + SRpcMsg rpcMsg; syncTimeout2RpcMsg(pMsg, &rpcMsg); SyncTimeout *pMsg2 = syncTimeoutFromRpcMsg2(&rpcMsg); syncTimeoutPrint2((char *)"test5: syncTimeout2RpcMsg -> syncTimeoutFromRpcMsg2 ", pMsg2); -- GitLab