From a18c731cd06d0f43a5742f5dd0eef3a2cdcc8b92 Mon Sep 17 00:00:00 2001 From: Shengliang Guan Date: Fri, 11 Nov 2022 17:35:04 +0800 Subject: [PATCH] refact: remove sync ping and pingreply --- source/libs/sync/inc/syncInt.h | 31 -- source/libs/sync/inc/syncMessage.h | 56 ---- source/libs/sync/src/syncMain.c | 47 --- source/libs/sync/src/syncMessage.c | 303 ------------------ .../sync/test/sync_test_lib/inc/syncTest.h | 72 +++++ .../test/sync_test_lib/src/syncMainDebug.c | 26 ++ .../test/sync_test_lib/src/syncMessageDebug.c | 302 +++++++++++++++++ 7 files changed, 400 insertions(+), 437 deletions(-) diff --git a/source/libs/sync/inc/syncInt.h b/source/libs/sync/inc/syncInt.h index 46bbb14421..d9eefab33f 100644 --- a/source/libs/sync/inc/syncInt.h +++ b/source/libs/sync/inc/syncInt.h @@ -67,8 +67,6 @@ extern "C" { typedef struct SyncTimeout SyncTimeout; typedef struct SyncClientRequest SyncClientRequest; -typedef struct SyncPing SyncPing; -typedef struct SyncPingReply SyncPingReply; typedef struct SyncRequestVote SyncRequestVote; typedef struct SyncRequestVoteReply SyncRequestVoteReply; typedef struct SyncAppendEntries SyncAppendEntries; @@ -93,17 +91,6 @@ typedef struct SyncHeartbeatReply SyncHeartbeatReply; typedef struct SyncHeartbeat SyncHeartbeat; typedef struct SyncPreSnapshot SyncPreSnapshot; -typedef int32_t (*FpOnPingCb)(SSyncNode* ths, SyncPing* pMsg); -typedef int32_t (*FpOnPingReplyCb)(SSyncNode* ths, SyncPingReply* pMsg); -typedef int32_t (*FpOnClientRequestCb)(SSyncNode* ths, SRpcMsg* pMsg, SyncIndex* pRetIndex); -typedef int32_t (*FpOnRequestVoteCb)(SSyncNode* ths, SyncRequestVote* pMsg); -typedef int32_t (*FpOnRequestVoteReplyCb)(SSyncNode* ths, SyncRequestVoteReply* pMsg); -typedef int32_t (*FpOnAppendEntriesCb)(SSyncNode* ths, SyncAppendEntries* pMsg); -typedef int32_t (*FpOnAppendEntriesReplyCb)(SSyncNode* ths, SyncAppendEntriesReply* pMsg); -typedef int32_t (*FpOnTimeoutCb)(SSyncNode* pSyncNode, SyncTimeout* pMsg); -typedef int32_t (*FpOnSnapshotCb)(SSyncNode* ths, SyncSnapshotSend* pMsg); -typedef int32_t (*FpOnSnapshotReplyCb)(SSyncNode* ths, SyncSnapshotRsp* pMsg); - extern bool gRaftDetailLog; typedef struct SRaftId { @@ -220,18 +207,6 @@ typedef struct SSyncNode { // peer heartbeat timer SSyncTimer peerHeartbeatTimerArr[TSDB_MAX_REPLICA]; - // callback - FpOnPingCb FpOnPing; - FpOnPingReplyCb FpOnPingReply; - FpOnClientRequestCb FpOnClientRequest; - FpOnTimeoutCb FpOnTimeout; - FpOnRequestVoteCb FpOnRequestVote; - FpOnRequestVoteReplyCb FpOnRequestVoteReply; - FpOnAppendEntriesCb FpOnAppendEntries; - FpOnAppendEntriesReplyCb FpOnAppendEntriesReply; - FpOnSnapshotCb FpOnSnapshot; - FpOnSnapshotReplyCb FpOnSnapshotReply; - // tools SSyncRespMgr* pSyncRespMgr; @@ -269,12 +244,6 @@ bool syncNodeSnapshotEnable(SSyncNode* pSyncNode); ESyncStrategy syncNodeStrategy(SSyncNode* pSyncNode); SyncIndex syncNodeGetSnapshotConfigIndex(SSyncNode* pSyncNode, SyncIndex snapshotLastApplyIndex); -// ping -------------- -int32_t syncNodePing(SSyncNode* pSyncNode, const SRaftId* destRaftId, SyncPing* pMsg); -int32_t syncNodePingSelf(SSyncNode* pSyncNode); -int32_t syncNodePingPeers(SSyncNode* pSyncNode); -int32_t syncNodePingAll(SSyncNode* pSyncNode); - // timer control -------------- int32_t syncNodeStartPingTimer(SSyncNode* pSyncNode); int32_t syncNodeStopPingTimer(SSyncNode* pSyncNode); diff --git a/source/libs/sync/inc/syncMessage.h b/source/libs/sync/inc/syncMessage.h index 7dc04d0b86..2110c80df3 100644 --- a/source/libs/sync/inc/syncMessage.h +++ b/source/libs/sync/inc/syncMessage.h @@ -22,60 +22,6 @@ extern "C" { #include "syncInt.h" -// --------------------------------------------- -typedef struct SyncPing { - uint32_t bytes; - int32_t vgId; - uint32_t msgType; - SRaftId srcId; - SRaftId destId; - // private data - uint32_t dataLen; - char data[]; -} SyncPing; - - -void syncPingDestroy(SyncPing* pMsg); -void syncPingSerialize(const SyncPing* pMsg, char* buf, uint32_t bufLen); -void syncPingDeserialize(const char* buf, uint32_t len, SyncPing* pMsg); -SyncPing* syncPingDeserialize2(const char* buf, uint32_t len); -SyncPing* syncPingFromRpcMsg2(const SRpcMsg* pRpcMsg); - -// --------------------------------------------- -typedef struct SyncPingReply { - uint32_t bytes; - int32_t vgId; - uint32_t msgType; - SRaftId srcId; - SRaftId destId; - // private data - uint32_t dataLen; - char data[]; -} SyncPingReply; - -SyncPingReply* syncPingReplyBuild(uint32_t dataLen); -SyncPingReply* syncPingReplyBuild2(const SRaftId* srcId, const SRaftId* destId, int32_t vgId, const char* str); -SyncPingReply* syncPingReplyBuild3(const SRaftId* srcId, const SRaftId* destId, int32_t vgId); -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); -int32_t syncPingReplySerialize3(const SyncPingReply* pMsg, char* buf, int32_t bufLen); -SyncPingReply* syncPingReplyDeserialize3(void* buf, int32_t bufLen); -void syncPingReply2RpcMsg(const SyncPingReply* pMsg, SRpcMsg* pRpcMsg); -void syncPingReplyFromRpcMsg(const SRpcMsg* pRpcMsg, SyncPingReply* pMsg); -SyncPingReply* syncPingReplyFromRpcMsg2(const SRpcMsg* pRpcMsg); -cJSON* syncPingReply2Json(const SyncPingReply* pMsg); -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 enum ESyncTimeoutType { SYNC_TIMEOUT_PING = 100, SYNC_TIMEOUT_ELECTION, @@ -656,8 +602,6 @@ void syncLocalCmdLog(const SyncLocalCmd* pMsg); void syncLocalCmdLog2(char* s, const SyncLocalCmd* pMsg); // on message ---------------------- -int32_t syncNodeOnPing(SSyncNode* ths, SyncPing* pMsg); -int32_t syncNodeOnPingReply(SSyncNode* ths, SyncPingReply* pMsg); int32_t syncNodeOnRequestVote(SSyncNode* ths, SyncRequestVote* pMsg); int32_t syncNodeOnRequestVoteReply(SSyncNode* ths, SyncRequestVoteReply* pMsg); diff --git a/source/libs/sync/src/syncMain.c b/source/libs/sync/src/syncMain.c index 25f6c5e6f1..c9e6885e50 100644 --- a/source/libs/sync/src/syncMain.c +++ b/source/libs/sync/src/syncMain.c @@ -145,14 +145,6 @@ int32_t syncProcessMsg(int64_t rid, SRpcMsg* pMsg) { SyncTimeout* pSyncMsg = syncTimeoutFromRpcMsg2(pMsg); code = syncNodeOnTimer(pSyncNode, pSyncMsg); syncTimeoutDestroy(pSyncMsg); - } else if (pMsg->msgType == TDMT_SYNC_PING) { - SyncPing* pSyncMsg = syncPingFromRpcMsg2(pMsg); - code = syncNodeOnPing(pSyncNode, pSyncMsg); - syncPingDestroy(pSyncMsg); - } else if (pMsg->msgType == TDMT_SYNC_PING_REPLY) { - SyncPingReply* pSyncMsg = syncPingReplyFromRpcMsg2(pMsg); - code = syncNodeOnPingReply(pSyncNode, pSyncMsg); - syncPingReplyDestroy(pSyncMsg); } else if (pMsg->msgType == TDMT_SYNC_CLIENT_REQUEST) { code = syncNodeOnClientRequest(pSyncNode, pMsg, NULL); } else if (pMsg->msgType == TDMT_SYNC_REQUEST_VOTE) { @@ -906,18 +898,6 @@ SSyncNode* syncNodeOpen(SSyncInfo* pSyncInfo) { syncHbTimerInit(pSyncNode, &(pSyncNode->peerHeartbeatTimerArr[i]), (pSyncNode->replicasId)[i]); } - // init callback - pSyncNode->FpOnPing = syncNodeOnPing; - pSyncNode->FpOnPingReply = syncNodeOnPingReply; - pSyncNode->FpOnClientRequest = syncNodeOnClientRequest; - pSyncNode->FpOnTimeout = syncNodeOnTimer; - pSyncNode->FpOnSnapshot = syncNodeOnSnapshot; - pSyncNode->FpOnSnapshotReply = syncNodeOnSnapshotReply; - pSyncNode->FpOnRequestVote = syncNodeOnRequestVote; - pSyncNode->FpOnRequestVoteReply = syncNodeOnRequestVoteReply; - pSyncNode->FpOnAppendEntries = syncNodeOnAppendEntries; - pSyncNode->FpOnAppendEntriesReply = syncNodeOnAppendEntriesReply; - // tools pSyncNode->pSyncRespMgr = syncRespMgrCreate(pSyncNode, SYNC_RESP_TTL_MS); if (pSyncNode->pSyncRespMgr == NULL) { @@ -2062,33 +2042,6 @@ static int32_t syncNodeAppendNoop(SSyncNode* ths) { return ret; } -// on message ---- -int32_t syncNodeOnPing(SSyncNode* ths, SyncPing* pMsg) { - sTrace("vgId:%d, recv sync-ping", ths->vgId); - - SyncPingReply* pMsgReply = syncPingReplyBuild3(&ths->myRaftId, &pMsg->srcId, ths->vgId); - SRpcMsg rpcMsg; - syncPingReply2RpcMsg(pMsgReply, &rpcMsg); - - /* - // htonl - SMsgHead* pHead = rpcMsg.pCont; - pHead->contLen = htonl(pHead->contLen); - pHead->vgId = htonl(pHead->vgId); - */ - - syncNodeSendMsgById(&pMsgReply->destId, ths, &rpcMsg); - syncPingReplyDestroy(pMsgReply); - - return 0; -} - -int32_t syncNodeOnPingReply(SSyncNode* ths, SyncPingReply* pMsg) { - int32_t ret = 0; - sTrace("vgId:%d, recv sync-ping-reply", ths->vgId); - return ret; -} - int32_t syncNodeOnHeartbeat(SSyncNode* ths, SyncHeartbeat* pMsg) { syncLogRecvHeartbeat(ths, pMsg, ""); diff --git a/source/libs/sync/src/syncMessage.c b/source/libs/sync/src/syncMessage.c index 565b21eb55..74f4a19f81 100644 --- a/source/libs/sync/src/syncMessage.c +++ b/source/libs/sync/src/syncMessage.c @@ -152,309 +152,6 @@ void syncTimeoutLog2(char* s, const SyncTimeout* pMsg) { } -void syncPingDestroy(SyncPing* pMsg) { - if (pMsg != NULL) { - taosMemoryFree(pMsg); - } -} - -void syncPingSerialize(const SyncPing* pMsg, char* buf, uint32_t bufLen) { - ASSERT(pMsg->bytes <= bufLen); - memcpy(buf, pMsg, pMsg->bytes); -} - -void syncPingDeserialize(const char* buf, uint32_t len, SyncPing* pMsg) { - memcpy(pMsg, buf, len); - ASSERT(len == pMsg->bytes); - ASSERT(pMsg->bytes == sizeof(SyncPing) + pMsg->dataLen); -} - -SyncPing* syncPingDeserialize2(const char* buf, uint32_t len) { - uint32_t bytes = *((uint32_t*)buf); - SyncPing* pMsg = taosMemoryMalloc(bytes); - ASSERT(pMsg != NULL); - syncPingDeserialize(buf, len, pMsg); - ASSERT(len == pMsg->bytes); - return pMsg; -} - - -SyncPing* syncPingFromRpcMsg2(const SRpcMsg* pRpcMsg) { - SyncPing* pMsg = syncPingDeserialize2(pRpcMsg->pCont, pRpcMsg->contLen); - ASSERT(pMsg != NULL); - return pMsg; -} - -// ---- message process SyncPingReply---- -SyncPingReply* syncPingReplyBuild(uint32_t dataLen) { - uint32_t bytes = sizeof(SyncPingReply) + dataLen; - SyncPingReply* pMsg = taosMemoryMalloc(bytes); - memset(pMsg, 0, bytes); - pMsg->bytes = bytes; - pMsg->msgType = TDMT_SYNC_PING_REPLY; - pMsg->dataLen = dataLen; - return pMsg; -} - -SyncPingReply* syncPingReplyBuild2(const SRaftId* srcId, const SRaftId* destId, int32_t vgId, const char* str) { - uint32_t dataLen = strlen(str) + 1; - SyncPingReply* pMsg = syncPingReplyBuild(dataLen); - pMsg->vgId = vgId; - pMsg->srcId = *srcId; - pMsg->destId = *destId; - snprintf(pMsg->data, pMsg->dataLen, "%s", str); - return pMsg; -} - -SyncPingReply* syncPingReplyBuild3(const SRaftId* srcId, const SRaftId* destId, int32_t vgId) { - SyncPingReply* pMsg = syncPingReplyBuild2(srcId, destId, vgId, "pang"); - return pMsg; -} - -void syncPingReplyDestroy(SyncPingReply* pMsg) { - if (pMsg != NULL) { - taosMemoryFree(pMsg); - } -} - -void syncPingReplySerialize(const SyncPingReply* pMsg, char* buf, uint32_t bufLen) { - ASSERT(pMsg->bytes <= bufLen); - memcpy(buf, pMsg, pMsg->bytes); -} - -void syncPingReplyDeserialize(const char* buf, uint32_t len, SyncPingReply* pMsg) { - memcpy(pMsg, buf, len); - ASSERT(len == pMsg->bytes); - ASSERT(pMsg->bytes == sizeof(SyncPingReply) + pMsg->dataLen); -} - -char* syncPingReplySerialize2(const SyncPingReply* pMsg, uint32_t* len) { - char* buf = taosMemoryMalloc(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 = taosMemoryMalloc(bytes); - ASSERT(pMsg != NULL); - syncPingReplyDeserialize(buf, len, pMsg); - ASSERT(len == pMsg->bytes); - return pMsg; -} - -int32_t syncPingReplySerialize3(const SyncPingReply* pMsg, char* buf, int32_t bufLen) { - SEncoder encoder = {0}; - tEncoderInit(&encoder, buf, bufLen); - if (tStartEncode(&encoder) < 0) { - return -1; - } - - if (tEncodeU32(&encoder, pMsg->bytes) < 0) { - return -1; - } - if (tEncodeI32(&encoder, pMsg->vgId) < 0) { - return -1; - } - if (tEncodeU32(&encoder, pMsg->msgType) < 0) { - return -1; - } - if (tEncodeU64(&encoder, pMsg->srcId.addr) < 0) { - return -1; - } - if (tEncodeI32(&encoder, pMsg->srcId.vgId) < 0) { - return -1; - } - if (tEncodeU64(&encoder, pMsg->destId.addr) < 0) { - return -1; - } - if (tEncodeI32(&encoder, pMsg->destId.vgId) < 0) { - return -1; - } - if (tEncodeU32(&encoder, pMsg->dataLen) < 0) { - return -1; - } - if (tEncodeBinary(&encoder, pMsg->data, pMsg->dataLen)) { - return -1; - } - - tEndEncode(&encoder); - int32_t tlen = encoder.pos; - tEncoderClear(&encoder); - return tlen; -} - -SyncPingReply* syncPingReplyDeserialize3(void* buf, int32_t bufLen) { - SDecoder decoder = {0}; - tDecoderInit(&decoder, buf, bufLen); - if (tStartDecode(&decoder) < 0) { - return NULL; - } - - SyncPingReply* pMsg = NULL; - uint32_t bytes; - if (tDecodeU32(&decoder, &bytes) < 0) { - return NULL; - } - - pMsg = taosMemoryMalloc(bytes); - ASSERT(pMsg != NULL); - pMsg->bytes = bytes; - - if (tDecodeI32(&decoder, &pMsg->vgId) < 0) { - taosMemoryFree(pMsg); - return NULL; - } - if (tDecodeU32(&decoder, &pMsg->msgType) < 0) { - taosMemoryFree(pMsg); - return NULL; - } - if (tDecodeU64(&decoder, &pMsg->srcId.addr) < 0) { - taosMemoryFree(pMsg); - return NULL; - } - if (tDecodeI32(&decoder, &pMsg->srcId.vgId) < 0) { - taosMemoryFree(pMsg); - return NULL; - } - if (tDecodeU64(&decoder, &pMsg->destId.addr) < 0) { - taosMemoryFree(pMsg); - return NULL; - } - if (tDecodeI32(&decoder, &pMsg->destId.vgId) < 0) { - taosMemoryFree(pMsg); - return NULL; - } - if (tDecodeU32(&decoder, &pMsg->dataLen) < 0) { - taosMemoryFree(pMsg); - return NULL; - } - uint32_t len; - char* data = NULL; - if (tDecodeBinary(&decoder, (uint8_t**)(&data), &len) < 0) { - taosMemoryFree(pMsg); - return NULL; - } - ASSERT(len == pMsg->dataLen); - memcpy(pMsg->data, data, len); - - tEndDecode(&decoder); - tDecoderClear(&decoder); - return pMsg; -} - -void syncPingReply2RpcMsg(const SyncPingReply* pMsg, SRpcMsg* pRpcMsg) { - memset(pRpcMsg, 0, sizeof(*pRpcMsg)); - pRpcMsg->msgType = pMsg->msgType; - pRpcMsg->contLen = pMsg->bytes; - pRpcMsg->pCont = rpcMallocCont(pRpcMsg->contLen); - syncPingReplySerialize(pMsg, pRpcMsg->pCont, pRpcMsg->contLen); -} - -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); - ASSERT(pMsg != NULL); - return pMsg; -} - -cJSON* syncPingReply2Json(const SyncPingReply* 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); - - 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, "SyncPingReply", pRoot); - return pJson; -} - -char* syncPingReply2Str(const SyncPingReply* pMsg) { - cJSON* pJson = syncPingReply2Json(pMsg); - char* serialized = cJSON_Print(pJson); - cJSON_Delete(pJson); - return serialized; -} - -// for debug ---------------------- -void syncPingReplyPrint(const SyncPingReply* pMsg) { - char* serialized = syncPingReply2Str(pMsg); - printf("syncPingReplyPrint | len:%zu | %s \n", strlen(serialized), serialized); - fflush(NULL); - taosMemoryFree(serialized); -} - -void syncPingReplyPrint2(char* s, const SyncPingReply* pMsg) { - char* serialized = syncPingReply2Str(pMsg); - printf("syncPingReplyPrint2 | len:%zu | %s | %s \n", strlen(serialized), s, serialized); - fflush(NULL); - taosMemoryFree(serialized); -} - -void syncPingReplyLog(const SyncPingReply* pMsg) { - char* serialized = syncPingReply2Str(pMsg); - sTrace("syncPingReplyLog | len:%zu | %s", strlen(serialized), serialized); - taosMemoryFree(serialized); -} - -void syncPingReplyLog2(char* s, const SyncPingReply* pMsg) { - if (gRaftDetailLog) { - char* serialized = syncPingReply2Str(pMsg); - sTrace("syncPingReplyLog2 | len:%zu | %s | %s", strlen(serialized), s, serialized); - taosMemoryFree(serialized); - } -} - // ---- message process SyncClientRequest---- SyncClientRequest* syncClientRequestAlloc(uint32_t dataLen) { uint32_t bytes = sizeof(SyncClientRequest) + dataLen; diff --git a/source/libs/sync/test/sync_test_lib/inc/syncTest.h b/source/libs/sync/test/sync_test_lib/inc/syncTest.h index 93da2df5a2..c02e3013d1 100644 --- a/source/libs/sync/test/sync_test_lib/inc/syncTest.h +++ b/source/libs/sync/test/sync_test_lib/inc/syncTest.h @@ -42,6 +42,20 @@ extern "C" { extern void addEpIntoEpSet(SEpSet* pEpSet, const char* fqdn, uint16_t port); +typedef struct SyncPing SyncPing; +typedef struct SyncPingReply SyncPingReply; + +typedef int32_t (*FpOnPingCb)(SSyncNode* ths, SyncPing* pMsg); +typedef int32_t (*FpOnPingReplyCb)(SSyncNode* ths, SyncPingReply* pMsg); +typedef int32_t (*FpOnClientRequestCb)(SSyncNode* ths, SRpcMsg* pMsg, SyncIndex* pRetIndex); +typedef int32_t (*FpOnRequestVoteCb)(SSyncNode* ths, SyncRequestVote* pMsg); +typedef int32_t (*FpOnRequestVoteReplyCb)(SSyncNode* ths, SyncRequestVoteReply* pMsg); +typedef int32_t (*FpOnAppendEntriesCb)(SSyncNode* ths, SyncAppendEntries* pMsg); +typedef int32_t (*FpOnAppendEntriesReplyCb)(SSyncNode* ths, SyncAppendEntriesReply* pMsg); +typedef int32_t (*FpOnTimeoutCb)(SSyncNode* pSyncNode, SyncTimeout* pMsg); +typedef int32_t (*FpOnSnapshotCb)(SSyncNode* ths, SyncSnapshotSend* pMsg); +typedef int32_t (*FpOnSnapshotReplyCb)(SSyncNode* ths, SyncSnapshotRsp* pMsg); + cJSON* syncEntry2Json(const SSyncRaftEntry* pEntry); char* syncEntry2Str(const SSyncRaftEntry* pEntry); void syncEntryPrint(const SSyncRaftEntry* pObj); @@ -127,6 +141,18 @@ void syncRpcMsgLog2(char* s, SRpcMsg* pMsg); // origin syncMessage +typedef struct SyncPing { + uint32_t bytes; + int32_t vgId; + uint32_t msgType; + SRaftId srcId; + SRaftId destId; + // private data + uint32_t dataLen; + char data[]; +} SyncPing; + + SyncPing* syncPingBuild(uint32_t dataLen); SyncPing* syncPingBuild2(const SRaftId* srcId, const SRaftId* destId, int32_t vgId, const char* str); SyncPing* syncPingBuild3(const SRaftId* srcId, const SRaftId* destId, int32_t vgId); @@ -141,6 +167,52 @@ void syncPingPrint(const SyncPing* pMsg); void syncPingPrint2(char* s, const SyncPing* pMsg); void syncPingLog(const SyncPing* pMsg); void syncPingLog2(char* s, const SyncPing* pMsg); +void syncPingDestroy(SyncPing* pMsg); +void syncPingSerialize(const SyncPing* pMsg, char* buf, uint32_t bufLen); +void syncPingDeserialize(const char* buf, uint32_t len, SyncPing* pMsg); +SyncPing* syncPingDeserialize2(const char* buf, uint32_t len); +SyncPing* syncPingFromRpcMsg2(const SRpcMsg* pRpcMsg); + +typedef struct SyncPingReply { + uint32_t bytes; + int32_t vgId; + uint32_t msgType; + SRaftId srcId; + SRaftId destId; + // private data + uint32_t dataLen; + char data[]; +} SyncPingReply; + +SyncPingReply* syncPingReplyBuild(uint32_t dataLen); +SyncPingReply* syncPingReplyBuild2(const SRaftId* srcId, const SRaftId* destId, int32_t vgId, const char* str); +SyncPingReply* syncPingReplyBuild3(const SRaftId* srcId, const SRaftId* destId, int32_t vgId); +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); +int32_t syncPingReplySerialize3(const SyncPingReply* pMsg, char* buf, int32_t bufLen); +SyncPingReply* syncPingReplyDeserialize3(void* buf, int32_t bufLen); +void syncPingReply2RpcMsg(const SyncPingReply* pMsg, SRpcMsg* pRpcMsg); +void syncPingReplyFromRpcMsg(const SRpcMsg* pRpcMsg, SyncPingReply* pMsg); +SyncPingReply* syncPingReplyFromRpcMsg2(const SRpcMsg* pRpcMsg); +cJSON* syncPingReply2Json(const SyncPingReply* pMsg); +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); + +int32_t syncNodeOnPing(SSyncNode* ths, SyncPing* pMsg); +int32_t syncNodeOnPingReply(SSyncNode* ths, SyncPingReply* pMsg); +int32_t syncNodePing(SSyncNode* pSyncNode, const SRaftId* destRaftId, SyncPing* pMsg); +int32_t syncNodePingSelf(SSyncNode* pSyncNode); +int32_t syncNodePingPeers(SSyncNode* pSyncNode); +int32_t syncNodePingAll(SSyncNode* pSyncNode); + #ifdef __cplusplus } diff --git a/source/libs/sync/test/sync_test_lib/src/syncMainDebug.c b/source/libs/sync/test/sync_test_lib/src/syncMainDebug.c index 5e1a9be164..b7a9d86ecc 100644 --- a/source/libs/sync/test/sync_test_lib/src/syncMainDebug.c +++ b/source/libs/sync/test/sync_test_lib/src/syncMainDebug.c @@ -253,3 +253,29 @@ int32_t syncNodePingAll(SSyncNode* pSyncNode) { return ret; } +// on message ---- +int32_t syncNodeOnPing(SSyncNode* ths, SyncPing* pMsg) { + sTrace("vgId:%d, recv sync-ping", ths->vgId); + + SyncPingReply* pMsgReply = syncPingReplyBuild3(&ths->myRaftId, &pMsg->srcId, ths->vgId); + SRpcMsg rpcMsg; + syncPingReply2RpcMsg(pMsgReply, &rpcMsg); + + /* + // htonl + SMsgHead* pHead = rpcMsg.pCont; + pHead->contLen = htonl(pHead->contLen); + pHead->vgId = htonl(pHead->vgId); + */ + + syncNodeSendMsgById(&pMsgReply->destId, ths, &rpcMsg); + syncPingReplyDestroy(pMsgReply); + + return 0; +} + +int32_t syncNodeOnPingReply(SSyncNode* ths, SyncPingReply* pMsg) { + int32_t ret = 0; + sTrace("vgId:%d, recv sync-ping-reply", ths->vgId); + return ret; +} diff --git a/source/libs/sync/test/sync_test_lib/src/syncMessageDebug.c b/source/libs/sync/test/sync_test_lib/src/syncMessageDebug.c index cdd2ae045d..c56d983e6e 100644 --- a/source/libs/sync/test/sync_test_lib/src/syncMessageDebug.c +++ b/source/libs/sync/test/sync_test_lib/src/syncMessageDebug.c @@ -254,6 +254,308 @@ void syncPingLog2(char* s, const SyncPing* pMsg) { } } +void syncPingDestroy(SyncPing* pMsg) { + if (pMsg != NULL) { + taosMemoryFree(pMsg); + } +} + +void syncPingSerialize(const SyncPing* pMsg, char* buf, uint32_t bufLen) { + ASSERT(pMsg->bytes <= bufLen); + memcpy(buf, pMsg, pMsg->bytes); +} + +void syncPingDeserialize(const char* buf, uint32_t len, SyncPing* pMsg) { + memcpy(pMsg, buf, len); + ASSERT(len == pMsg->bytes); + ASSERT(pMsg->bytes == sizeof(SyncPing) + pMsg->dataLen); +} + +SyncPing* syncPingDeserialize2(const char* buf, uint32_t len) { + uint32_t bytes = *((uint32_t*)buf); + SyncPing* pMsg = taosMemoryMalloc(bytes); + ASSERT(pMsg != NULL); + syncPingDeserialize(buf, len, pMsg); + ASSERT(len == pMsg->bytes); + return pMsg; +} + +SyncPing* syncPingFromRpcMsg2(const SRpcMsg* pRpcMsg) { + SyncPing* pMsg = syncPingDeserialize2(pRpcMsg->pCont, pRpcMsg->contLen); + ASSERT(pMsg != NULL); + return pMsg; +} + +// ---- message process SyncPingReply---- +SyncPingReply* syncPingReplyBuild(uint32_t dataLen) { + uint32_t bytes = sizeof(SyncPingReply) + dataLen; + SyncPingReply* pMsg = taosMemoryMalloc(bytes); + memset(pMsg, 0, bytes); + pMsg->bytes = bytes; + pMsg->msgType = TDMT_SYNC_PING_REPLY; + pMsg->dataLen = dataLen; + return pMsg; +} + +SyncPingReply* syncPingReplyBuild2(const SRaftId* srcId, const SRaftId* destId, int32_t vgId, const char* str) { + uint32_t dataLen = strlen(str) + 1; + SyncPingReply* pMsg = syncPingReplyBuild(dataLen); + pMsg->vgId = vgId; + pMsg->srcId = *srcId; + pMsg->destId = *destId; + snprintf(pMsg->data, pMsg->dataLen, "%s", str); + return pMsg; +} + +SyncPingReply* syncPingReplyBuild3(const SRaftId* srcId, const SRaftId* destId, int32_t vgId) { + SyncPingReply* pMsg = syncPingReplyBuild2(srcId, destId, vgId, "pang"); + return pMsg; +} + +void syncPingReplyDestroy(SyncPingReply* pMsg) { + if (pMsg != NULL) { + taosMemoryFree(pMsg); + } +} + +void syncPingReplySerialize(const SyncPingReply* pMsg, char* buf, uint32_t bufLen) { + ASSERT(pMsg->bytes <= bufLen); + memcpy(buf, pMsg, pMsg->bytes); +} + +void syncPingReplyDeserialize(const char* buf, uint32_t len, SyncPingReply* pMsg) { + memcpy(pMsg, buf, len); + ASSERT(len == pMsg->bytes); + ASSERT(pMsg->bytes == sizeof(SyncPingReply) + pMsg->dataLen); +} + +char* syncPingReplySerialize2(const SyncPingReply* pMsg, uint32_t* len) { + char* buf = taosMemoryMalloc(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 = taosMemoryMalloc(bytes); + ASSERT(pMsg != NULL); + syncPingReplyDeserialize(buf, len, pMsg); + ASSERT(len == pMsg->bytes); + return pMsg; +} + +int32_t syncPingReplySerialize3(const SyncPingReply* pMsg, char* buf, int32_t bufLen) { + SEncoder encoder = {0}; + tEncoderInit(&encoder, buf, bufLen); + if (tStartEncode(&encoder) < 0) { + return -1; + } + + if (tEncodeU32(&encoder, pMsg->bytes) < 0) { + return -1; + } + if (tEncodeI32(&encoder, pMsg->vgId) < 0) { + return -1; + } + if (tEncodeU32(&encoder, pMsg->msgType) < 0) { + return -1; + } + if (tEncodeU64(&encoder, pMsg->srcId.addr) < 0) { + return -1; + } + if (tEncodeI32(&encoder, pMsg->srcId.vgId) < 0) { + return -1; + } + if (tEncodeU64(&encoder, pMsg->destId.addr) < 0) { + return -1; + } + if (tEncodeI32(&encoder, pMsg->destId.vgId) < 0) { + return -1; + } + if (tEncodeU32(&encoder, pMsg->dataLen) < 0) { + return -1; + } + if (tEncodeBinary(&encoder, pMsg->data, pMsg->dataLen)) { + return -1; + } + + tEndEncode(&encoder); + int32_t tlen = encoder.pos; + tEncoderClear(&encoder); + return tlen; +} + +SyncPingReply* syncPingReplyDeserialize3(void* buf, int32_t bufLen) { + SDecoder decoder = {0}; + tDecoderInit(&decoder, buf, bufLen); + if (tStartDecode(&decoder) < 0) { + return NULL; + } + + SyncPingReply* pMsg = NULL; + uint32_t bytes; + if (tDecodeU32(&decoder, &bytes) < 0) { + return NULL; + } + + pMsg = taosMemoryMalloc(bytes); + ASSERT(pMsg != NULL); + pMsg->bytes = bytes; + + if (tDecodeI32(&decoder, &pMsg->vgId) < 0) { + taosMemoryFree(pMsg); + return NULL; + } + if (tDecodeU32(&decoder, &pMsg->msgType) < 0) { + taosMemoryFree(pMsg); + return NULL; + } + if (tDecodeU64(&decoder, &pMsg->srcId.addr) < 0) { + taosMemoryFree(pMsg); + return NULL; + } + if (tDecodeI32(&decoder, &pMsg->srcId.vgId) < 0) { + taosMemoryFree(pMsg); + return NULL; + } + if (tDecodeU64(&decoder, &pMsg->destId.addr) < 0) { + taosMemoryFree(pMsg); + return NULL; + } + if (tDecodeI32(&decoder, &pMsg->destId.vgId) < 0) { + taosMemoryFree(pMsg); + return NULL; + } + if (tDecodeU32(&decoder, &pMsg->dataLen) < 0) { + taosMemoryFree(pMsg); + return NULL; + } + uint32_t len; + char* data = NULL; + if (tDecodeBinary(&decoder, (uint8_t**)(&data), &len) < 0) { + taosMemoryFree(pMsg); + return NULL; + } + ASSERT(len == pMsg->dataLen); + memcpy(pMsg->data, data, len); + + tEndDecode(&decoder); + tDecoderClear(&decoder); + return pMsg; +} + +void syncPingReply2RpcMsg(const SyncPingReply* pMsg, SRpcMsg* pRpcMsg) { + memset(pRpcMsg, 0, sizeof(*pRpcMsg)); + pRpcMsg->msgType = pMsg->msgType; + pRpcMsg->contLen = pMsg->bytes; + pRpcMsg->pCont = rpcMallocCont(pRpcMsg->contLen); + syncPingReplySerialize(pMsg, pRpcMsg->pCont, pRpcMsg->contLen); +} + +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); + ASSERT(pMsg != NULL); + return pMsg; +} + +cJSON* syncPingReply2Json(const SyncPingReply* 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); + + 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, "SyncPingReply", pRoot); + return pJson; +} + +char* syncPingReply2Str(const SyncPingReply* pMsg) { + cJSON* pJson = syncPingReply2Json(pMsg); + char* serialized = cJSON_Print(pJson); + cJSON_Delete(pJson); + return serialized; +} + +// for debug ---------------------- +void syncPingReplyPrint(const SyncPingReply* pMsg) { + char* serialized = syncPingReply2Str(pMsg); + printf("syncPingReplyPrint | len:%zu | %s \n", strlen(serialized), serialized); + fflush(NULL); + taosMemoryFree(serialized); +} + +void syncPingReplyPrint2(char* s, const SyncPingReply* pMsg) { + char* serialized = syncPingReply2Str(pMsg); + printf("syncPingReplyPrint2 | len:%zu | %s | %s \n", strlen(serialized), s, serialized); + fflush(NULL); + taosMemoryFree(serialized); +} + +void syncPingReplyLog(const SyncPingReply* pMsg) { + char* serialized = syncPingReply2Str(pMsg); + sTrace("syncPingReplyLog | len:%zu | %s", strlen(serialized), serialized); + taosMemoryFree(serialized); +} + +void syncPingReplyLog2(char* s, const SyncPingReply* pMsg) { + if (gRaftDetailLog) { + char* serialized = syncPingReply2Str(pMsg); + sTrace("syncPingReplyLog2 | len:%zu | %s | %s", strlen(serialized), s, serialized); + taosMemoryFree(serialized); + } +} + // --------------------------------------------- cJSON* syncRpcMsg2Json(SRpcMsg* pRpcMsg) { cJSON* pRoot; -- GitLab