xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

refactor(sync): add syncHeartbeatReply

This commit is contained in:
Minghao Li 2022-09-07 18:02:10 +08:00
parent d9363214b7
commit c5ec1741bc
4 changed files with 292 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
include/libs/sync/syncTools.h
View File

@ -490,6 +490,24 @@ typedef struct SyncHeartbeatReply {
int64_t startTime;
} SyncHeartbeatReply;
SyncHeartbeatReply* syncHeartbeatReplyBuild(int32_t vgId);
void syncHeartbeatReplyDestroy(SyncHeartbeatReply* pMsg);
void syncHeartbeatReplySerialize(const SyncHeartbeatReply* pMsg, char* buf, uint32_t bufLen);
void syncHeartbeatReplyDeserialize(const char* buf, uint32_t len, SyncHeartbeatReply* pMsg);
char* syncHeartbeatReplySerialize2(const SyncHeartbeatReply* pMsg, uint32_t* len);
SyncHeartbeatReply* syncHeartbeatReplyDeserialize2(const char* buf, uint32_t len);
void syncHeartbeatReply2RpcMsg(const SyncHeartbeatReply* pMsg, SRpcMsg* pRpcMsg);
void syncHeartbeatReplyFromRpcMsg(const SRpcMsg* pRpcMsg, SyncHeartbeatReply* pMsg);
SyncHeartbeatReply* syncHeartbeatReplyFromRpcMsg2(const SRpcMsg* pRpcMsg);
cJSON* syncHeartbeatReply2Json(const SyncHeartbeatReply* pMsg);
char* syncHeartbeatReply2Str(const SyncHeartbeatReply* pMsg);
// for debug ----------------------
void syncHeartbeatReplyPrint(const SyncHeartbeatReply* pMsg);
void syncHeartbeatReplyPrint2(char* s, const SyncHeartbeatReply* pMsg);
void syncHeartbeatReplyLog(const SyncHeartbeatReply* pMsg);
void syncHeartbeatReplyLog2(char* s, const SyncHeartbeatReply* pMsg);
// ---------------------------------------------
typedef struct SyncApplyMsg {
uint32_t bytes;

156
source/libs/sync/src/syncMessage.c
View File

@ -1999,7 +1999,7 @@ SyncHeartbeat* syncHeartbeatBuild(int32_t vgId) {
memset(pMsg, 0, bytes);
pMsg->bytes = bytes;
pMsg->vgId = vgId;
pMsg->msgType = TDMT_SYNC_APPEND_ENTRIES;
pMsg->msgType = TDMT_SYNC_HEARTBEAT;
return pMsg;
}
@ -2145,6 +2145,160 @@ void syncHeartbeatLog2(char* s, const SyncHeartbeat* pMsg) {
}
}
// ---- message process SyncHeartbeatReply----
SyncHeartbeatReply* syncHeartbeatReplyBuild(int32_t vgId) {
uint32_t bytes = sizeof(SyncHeartbeatReply);
SyncHeartbeatReply* pMsg = taosMemoryMalloc(bytes);
memset(pMsg, 0, bytes);
pMsg->bytes = bytes;
pMsg->vgId = vgId;
pMsg->msgType = TDMT_SYNC_HEARTBEAT_REPLY;
return pMsg;
}
void syncHeartbeatReplyDestroy(SyncHeartbeatReply* pMsg) {
if (pMsg != NULL) {
taosMemoryFree(pMsg);
}
}
void syncHeartbeatReplySerialize(const SyncHeartbeatReply* pMsg, char* buf, uint32_t bufLen) {
ASSERT(pMsg->bytes <= bufLen);
memcpy(buf, pMsg, pMsg->bytes);
}
void syncHeartbeatReplyDeserialize(const char* buf, uint32_t len, SyncHeartbeatReply* pMsg) {
memcpy(pMsg, buf, len);
ASSERT(len == pMsg->bytes);
}
char* syncHeartbeatReplySerialize2(const SyncHeartbeatReply* pMsg, uint32_t* len) {
char* buf = taosMemoryMalloc(pMsg->bytes);
ASSERT(buf != NULL);
syncHeartbeatReplySerialize(pMsg, buf, pMsg->bytes);
if (len != NULL) {
*len = pMsg->bytes;
}
return buf;
}
SyncHeartbeatReply* syncHeartbeatReplyDeserialize2(const char* buf, uint32_t len) {
uint32_t bytes = *((uint32_t*)buf);
SyncHeartbeatReply* pMsg = taosMemoryMalloc(bytes);
ASSERT(pMsg != NULL);
syncHeartbeatReplyDeserialize(buf, len, pMsg);
ASSERT(len == pMsg->bytes);
return pMsg;
}
void syncHeartbeatReply2RpcMsg(const SyncHeartbeatReply* pMsg, SRpcMsg* pRpcMsg) {
memset(pRpcMsg, 0, sizeof(*pRpcMsg));
pRpcMsg->msgType = pMsg->msgType;
pRpcMsg->contLen = pMsg->bytes;
pRpcMsg->pCont = rpcMallocCont(pRpcMsg->contLen);
syncHeartbeatReplySerialize(pMsg, pRpcMsg->pCont, pRpcMsg->contLen);
}
void syncHeartbeatReplyFromRpcMsg(const SRpcMsg* pRpcMsg, SyncHeartbeatReply* pMsg) {
syncHeartbeatReplyDeserialize(pRpcMsg->pCont, pRpcMsg->contLen, pMsg);
}
SyncHeartbeatReply* syncHeartbeatReplyFromRpcMsg2(const SRpcMsg* pRpcMsg) {
SyncHeartbeatReply* pMsg = syncHeartbeatReplyDeserialize2(pRpcMsg->pCont, pRpcMsg->contLen);
ASSERT(pMsg != NULL);
return pMsg;
}
cJSON* syncHeartbeatReply2Json(const SyncHeartbeatReply* 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);
snprintf(u64buf, sizeof(u64buf), "%" PRIu64, pMsg->privateTerm);
cJSON_AddStringToObject(pRoot, "privateTerm", u64buf);
snprintf(u64buf, sizeof(u64buf), "%" PRIu64, pMsg->term);
cJSON_AddStringToObject(pRoot, "term", u64buf);
cJSON_AddStringToObject(pRoot, "matchIndex", u64buf);
snprintf(u64buf, sizeof(u64buf), "%" PRId64, pMsg->startTime);
cJSON_AddStringToObject(pRoot, "startTime", u64buf);
}
cJSON* pJson = cJSON_CreateObject();
cJSON_AddItemToObject(pJson, "SyncHeartbeatReply", pRoot);
return pJson;
}
char* syncHeartbeatReply2Str(const SyncHeartbeatReply* pMsg) {
cJSON* pJson = syncHeartbeatReply2Json(pMsg);
char* serialized = cJSON_Print(pJson);
cJSON_Delete(pJson);
return serialized;
}
void syncHeartbeatReplyPrint(const SyncHeartbeatReply* pMsg) {
char* serialized = syncHeartbeatReply2Str(pMsg);
printf("syncHeartbeatReplyPrint | len:%" PRIu64 " | %s \n", strlen(serialized), serialized);
fflush(NULL);
taosMemoryFree(serialized);
}
void syncHeartbeatReplyPrint2(char* s, const SyncHeartbeatReply* pMsg) {
char* serialized = syncHeartbeatReply2Str(pMsg);
printf("syncHeartbeatReplyPrint2 | len:%" PRIu64 " | %s | %s \n", strlen(serialized), s, serialized);
fflush(NULL);
taosMemoryFree(serialized);
}
void syncHeartbeatReplyLog(const SyncHeartbeatReply* pMsg) {
char* serialized = syncHeartbeatReply2Str(pMsg);
sTrace("syncHeartbeatReplyLog | len:%" PRIu64 " | %s", strlen(serialized), serialized);
taosMemoryFree(serialized);
}
void syncHeartbeatReplyLog2(char* s, const SyncHeartbeatReply* pMsg) {
if (gRaftDetailLog) {
char* serialized = syncHeartbeatReply2Str(pMsg);
sTrace("syncHeartbeatReplyLog2 | len:%" PRIu64 " | %s | %s", strlen(serialized), s, serialized);
taosMemoryFree(serialized);
}
}
// ---- message process SyncApplyMsg----
SyncApplyMsg* syncApplyMsgBuild(uint32_t dataLen) {
uint32_t bytes = sizeof(SyncApplyMsg) + dataLen;

14
source/libs/sync/test/CMakeLists.txt
View File

@ -58,6 +58,7 @@ add_executable(syncReconfigFinishTest "")
add_executable(syncRestoreFromSnapshot "")
add_executable(syncRaftCfgIndexTest "")
add_executable(syncHeartbeatTest "")
add_executable(syncHeartbeatReplyTest "")
target_sources(syncTest
@ -300,6 +301,10 @@ target_sources(syncHeartbeatTest
PRIVATE
"syncHeartbeatTest.cpp"
)
target_sources(syncHeartbeatReplyTest
PRIVATE
"syncHeartbeatReplyTest.cpp"
)
target_include_directories(syncTest
@ -602,6 +607,11 @@ target_include_directories(syncHeartbeatTest
"${TD_SOURCE_DIR}/include/libs/sync"
"${CMAKE_CURRENT_SOURCE_DIR}/../inc"
)
target_include_directories(syncHeartbeatReplyTest
PUBLIC
"${TD_SOURCE_DIR}/include/libs/sync"
"${CMAKE_CURRENT_SOURCE_DIR}/../inc"
)
target_link_libraries(syncTest
@ -844,6 +854,10 @@ target_link_libraries(syncHeartbeatTest
sync
gtest_main
)
target_link_libraries(syncHeartbeatReplyTest
sync
gtest_main
)
enable_testing()

105
source/libs/sync/test/syncHeartbeatReplyTest.cpp Normal file
View File

@ -0,0 +1,105 @@
#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");
}
SyncHeartbeatReply *createMsg() {
SyncHeartbeatReply *pMsg = syncHeartbeatReplyBuild(1000);
pMsg->srcId.addr = syncUtilAddr2U64("127.0.0.1", 1234);
pMsg->srcId.vgId = 100;
pMsg->destId.addr = syncUtilAddr2U64("127.0.0.1", 5678);
pMsg->destId.vgId = 100;
pMsg->term = 33;
pMsg->privateTerm = 44;
pMsg->startTime = taosGetTimestampMs();
return pMsg;
}
void test1() {
SyncHeartbeatReply *pMsg = createMsg();
syncHeartbeatReplyLog2((char *)"test1:", pMsg);
syncHeartbeatReplyDestroy(pMsg);
}
void test2() {
SyncHeartbeatReply *pMsg = createMsg();
uint32_t len = pMsg->bytes;
char * serialized = (char *)taosMemoryMalloc(len);
syncHeartbeatReplySerialize(pMsg, serialized, len);
SyncHeartbeatReply *pMsg2 = syncHeartbeatReplyBuild(1000);
syncHeartbeatReplyDeserialize(serialized, len, pMsg2);
syncHeartbeatReplyLog2((char *)"test2: syncHeartbeatReplySerialize -> syncHeartbeatReplyDeserialize ",
pMsg2);
taosMemoryFree(serialized);
syncHeartbeatReplyDestroy(pMsg);
syncHeartbeatReplyDestroy(pMsg2);
}
void test3() {
SyncHeartbeatReply *pMsg = createMsg();
uint32_t len;
char * serialized = syncHeartbeatReplySerialize2(pMsg, &len);
SyncHeartbeatReply *pMsg2 = syncHeartbeatReplyDeserialize2(serialized, len);
syncHeartbeatReplyLog2((char *)"test3: syncHeartbeatReplySerialize3 -> syncHeartbeatReplyDeserialize2 ",
pMsg2);
taosMemoryFree(serialized);
syncHeartbeatReplyDestroy(pMsg);
syncHeartbeatReplyDestroy(pMsg2);
}
void test4() {
SyncHeartbeatReply *pMsg = createMsg();
SRpcMsg rpcMsg;
syncHeartbeatReply2RpcMsg(pMsg, &rpcMsg);
SyncHeartbeatReply *pMsg2 = syncHeartbeatReplyBuild(1000);
syncHeartbeatReplyFromRpcMsg(&rpcMsg, pMsg2);
syncHeartbeatReplyLog2((char *)"test4: syncHeartbeatReply2RpcMsg -> syncHeartbeatReplyFromRpcMsg ",
pMsg2);
rpcFreeCont(rpcMsg.pCont);
syncHeartbeatReplyDestroy(pMsg);
syncHeartbeatReplyDestroy(pMsg2);
}
void test5() {
SyncHeartbeatReply *pMsg = createMsg();
SRpcMsg rpcMsg;
syncHeartbeatReply2RpcMsg(pMsg, &rpcMsg);
SyncHeartbeatReply *pMsg2 = syncHeartbeatReplyFromRpcMsg2(&rpcMsg);
syncHeartbeatReplyLog2((char *)"test5: syncHeartbeatReply2RpcMsg -> syncHeartbeatReplyFromRpcMsg2 ",
pMsg2);
rpcFreeCont(rpcMsg.pCont);
syncHeartbeatReplyDestroy(pMsg);
syncHeartbeatReplyDestroy(pMsg2);
}
int main() {
gRaftDetailLog = true;
tsAsyncLog = 0;
sDebugFlag = DEBUG_TRACE + DEBUG_SCREEN + DEBUG_FILE;
logTest();
test1();
test2();
test3();
test4();
test5();
return 0;
}