diff --git a/source/libs/sync/inc/syncMessage.h b/source/libs/sync/inc/syncMessage.h index a38b7c10ff..8b98cf0e69 100644 --- a/source/libs/sync/inc/syncMessage.h +++ b/source/libs/sync/inc/syncMessage.h @@ -172,27 +172,6 @@ typedef struct SyncApplyMsg { char data[]; // user RpcMsg.pCont } SyncApplyMsg; -SyncApplyMsg* syncApplyMsgBuild(uint32_t dataLen); -SyncApplyMsg* syncApplyMsgBuild2(const SRpcMsg* pOriginalRpcMsg, int32_t vgId, SFsmCbMeta* pMeta); -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); -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 -SyncApplyMsg* syncApplyMsgFromRpcMsg2(const SRpcMsg* pRpcMsg); -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 syncApplyMsgLog(const SyncApplyMsg* pMsg); -void syncApplyMsgLog2(char* s, const SyncApplyMsg* pMsg); - -// --------------------------------------------- typedef struct SyncSnapshotSend { uint32_t bytes; int32_t vgId; @@ -212,25 +191,6 @@ typedef struct SyncSnapshotSend { char data[]; } SyncSnapshotSend; -SyncSnapshotSend* syncSnapshotSendBuild(uint32_t dataLen, int32_t vgId); -void syncSnapshotSendDestroy(SyncSnapshotSend* pMsg); -void syncSnapshotSendSerialize(const SyncSnapshotSend* pMsg, char* buf, uint32_t bufLen); -void syncSnapshotSendDeserialize(const char* buf, uint32_t len, SyncSnapshotSend* pMsg); -char* syncSnapshotSendSerialize2(const SyncSnapshotSend* pMsg, uint32_t* len); -SyncSnapshotSend* syncSnapshotSendDeserialize2(const char* buf, uint32_t len); -void syncSnapshotSend2RpcMsg(const SyncSnapshotSend* pMsg, SRpcMsg* pRpcMsg); -void syncSnapshotSendFromRpcMsg(const SRpcMsg* pRpcMsg, SyncSnapshotSend* pMsg); -SyncSnapshotSend* syncSnapshotSendFromRpcMsg2(const SRpcMsg* pRpcMsg); -cJSON* syncSnapshotSend2Json(const SyncSnapshotSend* pMsg); -char* syncSnapshotSend2Str(const SyncSnapshotSend* pMsg); - -// for debug ---------------------- -void syncSnapshotSendPrint(const SyncSnapshotSend* pMsg); -void syncSnapshotSendPrint2(char* s, const SyncSnapshotSend* pMsg); -void syncSnapshotSendLog(const SyncSnapshotSend* pMsg); -void syncSnapshotSendLog2(char* s, const SyncSnapshotSend* pMsg); - -// --------------------------------------------- typedef struct SyncSnapshotRsp { uint32_t bytes; int32_t vgId; @@ -247,25 +207,6 @@ typedef struct SyncSnapshotRsp { SyncIndex snapBeginIndex; // when ack = SYNC_SNAPSHOT_SEQ_BEGIN, it's valid } SyncSnapshotRsp; -SyncSnapshotRsp* syncSnapshotRspBuild(int32_t vgId); -void syncSnapshotRspDestroy(SyncSnapshotRsp* pMsg); -void syncSnapshotRspSerialize(const SyncSnapshotRsp* pMsg, char* buf, uint32_t bufLen); -void syncSnapshotRspDeserialize(const char* buf, uint32_t len, SyncSnapshotRsp* pMsg); -char* syncSnapshotRspSerialize2(const SyncSnapshotRsp* pMsg, uint32_t* len); -SyncSnapshotRsp* syncSnapshotRspDeserialize2(const char* buf, uint32_t len); -void syncSnapshotRsp2RpcMsg(const SyncSnapshotRsp* pMsg, SRpcMsg* pRpcMsg); -void syncSnapshotRspFromRpcMsg(const SRpcMsg* pRpcMsg, SyncSnapshotRsp* pMsg); -SyncSnapshotRsp* syncSnapshotRspFromRpcMsg2(const SRpcMsg* pRpcMsg); -cJSON* syncSnapshotRsp2Json(const SyncSnapshotRsp* pMsg); -char* syncSnapshotRsp2Str(const SyncSnapshotRsp* pMsg); - -// for debug ---------------------- -void syncSnapshotRspPrint(const SyncSnapshotRsp* pMsg); -void syncSnapshotRspPrint2(char* s, const SyncSnapshotRsp* pMsg); -void syncSnapshotRspLog(const SyncSnapshotRsp* pMsg); -void syncSnapshotRspLog2(char* s, const SyncSnapshotRsp* pMsg); - -// --------------------------------------------- typedef struct SyncLeaderTransfer { uint32_t bytes; int32_t vgId; @@ -278,18 +219,6 @@ typedef struct SyncLeaderTransfer { SRaftId newLeaderId; } SyncLeaderTransfer; -SyncLeaderTransfer* syncLeaderTransferBuild(int32_t vgId); -void syncLeaderTransferDestroy(SyncLeaderTransfer* pMsg); -void syncLeaderTransferSerialize(const SyncLeaderTransfer* pMsg, char* buf, uint32_t bufLen); -void syncLeaderTransferDeserialize(const char* buf, uint32_t len, SyncLeaderTransfer* pMsg); -char* syncLeaderTransferSerialize2(const SyncLeaderTransfer* pMsg, uint32_t* len); -SyncLeaderTransfer* syncLeaderTransferDeserialize2(const char* buf, uint32_t len); -void syncLeaderTransfer2RpcMsg(const SyncLeaderTransfer* pMsg, SRpcMsg* pRpcMsg); -void syncLeaderTransferFromRpcMsg(const SRpcMsg* pRpcMsg, SyncLeaderTransfer* pMsg); -SyncLeaderTransfer* syncLeaderTransferFromRpcMsg2(const SRpcMsg* pRpcMsg); -cJSON* syncLeaderTransfer2Json(const SyncLeaderTransfer* pMsg); -char* syncLeaderTransfer2Str(const SyncLeaderTransfer* pMsg); - typedef enum { SYNC_LOCAL_CMD_STEP_DOWN = 100, SYNC_LOCAL_CMD_FOLLOWER_CMT, @@ -333,9 +262,8 @@ int32_t syncNodeOnRequestVote(SSyncNode* pNode, const SRpcMsg* pMsg); int32_t syncNodeOnRequestVoteReply(SSyncNode* pNode, const SRpcMsg* pMsg); int32_t syncNodeOnAppendEntries(SSyncNode* pNode, const SRpcMsg* pMsg); int32_t syncNodeOnAppendEntriesReply(SSyncNode* ths, const SRpcMsg* pMsg); - -int32_t syncNodeOnSnapshot(SSyncNode* ths, SyncSnapshotSend* pMsg); -int32_t syncNodeOnSnapshotReply(SSyncNode* ths, SyncSnapshotRsp* pMsg); +int32_t syncNodeOnSnapshot(SSyncNode* ths, const SRpcMsg* pMsg); +int32_t syncNodeOnSnapshotReply(SSyncNode* ths, const SRpcMsg* pMsg); int32_t syncNodeOnHeartbeat(SSyncNode* ths, const SRpcMsg* pMsg); int32_t syncNodeOnHeartbeatReply(SSyncNode* ths, const SRpcMsg* pMsg); @@ -352,7 +280,7 @@ ESyncStrategy syncNodeStrategy(SSyncNode* pSyncNode); const char* syncTimerTypeStr(enum ESyncTimeoutType timerType); int32_t syncBuildTimeout(SRpcMsg* pMsg, ESyncTimeoutType ttype, uint64_t logicClock, int32_t ms, SSyncNode* pNode); -int32_t syncBuildClientRequest(SRpcMsg* pMsg, const SRpcMsg* pOriginalRpc, uint64_t seq, bool isWeak, int32_t vgId); +int32_t syncBuildClientRequest(SRpcMsg* pMsg, const SRpcMsg* pOriginal, uint64_t seq, bool isWeak, int32_t vgId); int32_t syncBuildClientRequestFromNoopEntry(SRpcMsg* pMsg, const SSyncRaftEntry* pEntry, int32_t vgId); int32_t syncBuildRequestVote(SRpcMsg* pMsg, int32_t vgId); int32_t syncBuildRequestVoteReply(SRpcMsg* pMsg, int32_t vgId); @@ -362,6 +290,10 @@ int32_t syncBuildHeartbeat(SRpcMsg* pMsg, int32_t vgId); int32_t syncBuildHeartbeatReply(SRpcMsg* pMsg, int32_t vgId); int32_t syncBuildPreSnapshot(SRpcMsg* pMsg, int32_t vgId); int32_t syncBuildPreSnapshotReply(SRpcMsg* pMsg, int32_t vgId); +int32_t syncBuildApplyMsg(SRpcMsg* pMsg, const SRpcMsg* pOriginal, int32_t vgId, SFsmCbMeta* pMeta); +int32_t syncBuildSnapshotSend(SRpcMsg* pMsg, int32_t dataLen, int32_t vgId); +int32_t syncBuildSnapshotSendRsp(SRpcMsg* pMsg, int32_t vgId); +int32_t syncBuildLeaderTransfer(SRpcMsg* pMsg, int32_t vgId); #ifdef __cplusplus } diff --git a/source/libs/sync/inc/syncSnapshot.h b/source/libs/sync/inc/syncSnapshot.h index 93b2531235..7b8e768391 100644 --- a/source/libs/sync/inc/syncSnapshot.h +++ b/source/libs/sync/inc/syncSnapshot.h @@ -84,8 +84,8 @@ bool snapshotReceiverIsStart(SSyncSnapshotReceiver *pReceiver) void snapshotReceiverForceStop(SSyncSnapshotReceiver *pReceiver); // on message -int32_t syncNodeOnSnapshot(SSyncNode *ths, SyncSnapshotSend *pMsg); -int32_t syncNodeOnSnapshotReply(SSyncNode *ths, SyncSnapshotRsp *pMsg); +int32_t syncNodeOnSnapshot(SSyncNode *ths, const SRpcMsg *pMsg); +int32_t syncNodeOnSnapshotReply(SSyncNode *ths, const SRpcMsg *pMsg); // start diff --git a/source/libs/sync/src/syncMain.c b/source/libs/sync/src/syncMain.c index 0952f87ed7..75d3b34f21 100644 --- a/source/libs/sync/src/syncMain.c +++ b/source/libs/sync/src/syncMain.c @@ -150,13 +150,9 @@ int32_t syncProcessMsg(int64_t rid, SRpcMsg* pMsg) { } else if (pMsg->msgType == TDMT_SYNC_APPEND_ENTRIES_REPLY) { code = syncNodeOnAppendEntriesReply(pSyncNode, pMsg); } else if (pMsg->msgType == TDMT_SYNC_SNAPSHOT_SEND) { - SyncSnapshotSend* pSyncMsg = syncSnapshotSendFromRpcMsg2(pMsg); - code = syncNodeOnSnapshot(pSyncNode, pSyncMsg); - syncSnapshotSendDestroy(pSyncMsg); + code = syncNodeOnSnapshot(pSyncNode, pMsg); } else if (pMsg->msgType == TDMT_SYNC_SNAPSHOT_RSP) { - SyncSnapshotRsp* pSyncMsg = syncSnapshotRspFromRpcMsg2(pMsg); - code = syncNodeOnSnapshotReply(pSyncNode, pSyncMsg); - syncSnapshotRspDestroy(pSyncMsg); + code = syncNodeOnSnapshotReply(pSyncNode, pMsg); } else if (pMsg->msgType == TDMT_SYNC_LOCAL_CMD) { SyncLocalCmd* pSyncMsg = syncLocalCmdFromRpcMsg2(pMsg); code = syncNodeOnLocalCmd(pSyncNode, pSyncMsg); @@ -415,17 +411,15 @@ int32_t syncNodeLeaderTransferTo(SSyncNode* pSyncNode, SNodeInfo newLeader) { sNTrace(pSyncNode, "begin leader transfer to %s:%u", newLeader.nodeFqdn, newLeader.nodePort); - SyncLeaderTransfer* pMsg = syncLeaderTransferBuild(pSyncNode->vgId); + SRpcMsg rpcMsg = {0}; + (void)syncBuildLeaderTransfer(&rpcMsg, pSyncNode->vgId); + + SyncLeaderTransfer* pMsg = rpcMsg.pCont; pMsg->newLeaderId.addr = syncUtilAddr2U64(newLeader.nodeFqdn, newLeader.nodePort); pMsg->newLeaderId.vgId = pSyncNode->vgId; pMsg->newNodeInfo = newLeader; - ASSERT(pMsg != NULL); - SRpcMsg rpcMsg = {0}; - syncLeaderTransfer2RpcMsg(pMsg, &rpcMsg); - syncLeaderTransferDestroy(pMsg); - int32_t ret = syncNodePropose(pSyncNode, &rpcMsg, false); - return ret; + return syncNodePropose(pSyncNode, &rpcMsg, false); } SSyncState syncGetState(int64_t rid) { @@ -2214,6 +2208,7 @@ const char* syncStr(ESyncState state) { } } +#if 0 int32_t syncDoLeaderTransfer(SSyncNode* ths, SRpcMsg* pRpcMsg, SSyncRaftEntry* pEntry) { if (ths->state != TAOS_SYNC_STATE_FOLLOWER) { sNTrace(ths, "I am not follower, can not do leader transfer"); @@ -2242,7 +2237,7 @@ int32_t syncDoLeaderTransfer(SSyncNode* ths, SRpcMsg* pRpcMsg, SSyncRaftEntry* p } */ - SyncLeaderTransfer* pSyncLeaderTransfer = syncLeaderTransferFromRpcMsg2(pRpcMsg); + SyncLeaderTransfer* pSyncLeaderTransfer = pRpcMsg->pCont; sNTrace(ths, "do leader transfer, index:%" PRId64, pEntry->index); bool sameId = syncUtilSameId(&(pSyncLeaderTransfer->newLeaderId), &(ths->myRaftId)); @@ -2275,10 +2270,11 @@ int32_t syncDoLeaderTransfer(SSyncNode* ths, SRpcMsg* pRpcMsg, SSyncRaftEntry* p ths->pFsm->FpLeaderTransferCb(ths->pFsm, pRpcMsg, &cbMeta); } - syncLeaderTransferDestroy(pSyncLeaderTransfer); return 0; } +#endif + int32_t syncNodeUpdateNewConfigIndex(SSyncNode* ths, SSyncCfg* pNewCfg) { for (int32_t i = 0; i < pNewCfg->replicaNum; ++i) { SRaftId raftId; diff --git a/source/libs/sync/src/syncMessage.c b/source/libs/sync/src/syncMessage.c index 26bedb2b66..2b5b205f56 100644 --- a/source/libs/sync/src/syncMessage.c +++ b/source/libs/sync/src/syncMessage.c @@ -223,518 +223,56 @@ int32_t syncBuildPreSnapshotReply(SRpcMsg* pMsg, int32_t vgId) { return 0; } -// ---- message process SyncApplyMsg---- -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_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) { - 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) { - 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) { - 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] = {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_AddNumberToObject(pRoot, "originalRpcType", pMsg->originalRpcType); - - snprintf(u64buf, sizeof(u64buf), "%" PRId64, 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", syncStr(pMsg->fsmMeta.state)); - snprintf(u64buf, sizeof(u64buf), "%" PRIu64, 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) { - char* serialized = syncApplyMsg2Str(pMsg); - printf("syncApplyMsgPrint | len:%d | %s \n", (int32_t)strlen(serialized), serialized); - fflush(NULL); - taosMemoryFree(serialized); -} - -void syncApplyMsgPrint2(char* s, const SyncApplyMsg* pMsg) { - char* serialized = syncApplyMsg2Str(pMsg); - printf("syncApplyMsgPrint2 | len:%d | %s | %s \n", (int32_t)strlen(serialized), s, serialized); - fflush(NULL); - taosMemoryFree(serialized); -} - -void syncApplyMsgLog(const SyncApplyMsg* pMsg) { - char* serialized = syncApplyMsg2Str(pMsg); - sTrace("ssyncApplyMsgLog | len:%d | %s", (int32_t)strlen(serialized), serialized); - taosMemoryFree(serialized); -} - -void syncApplyMsgLog2(char* s, const SyncApplyMsg* pMsg) { - if (gRaftDetailLog) { - char* serialized = syncApplyMsg2Str(pMsg); - sTrace("syncApplyMsgLog2 | len:%d | %s | %s", (int32_t)strlen(serialized), s, serialized); - taosMemoryFree(serialized); - } -} - -// --------------------------------------------- -SyncSnapshotSend* syncSnapshotSendBuild(uint32_t dataLen, int32_t vgId) { - uint32_t bytes = sizeof(SyncSnapshotSend) + dataLen; - SyncSnapshotSend* pMsg = taosMemoryMalloc(bytes); - memset(pMsg, 0, bytes); - pMsg->bytes = bytes; - pMsg->vgId = vgId; +int32_t syncBuildSnapshotSend(SRpcMsg* pMsg, int32_t dataLen, int32_t vgId) { + int32_t bytes = sizeof(SyncSnapshotSend) + dataLen; + pMsg->pCont = rpcMallocCont(bytes); pMsg->msgType = TDMT_SYNC_SNAPSHOT_SEND; - pMsg->dataLen = dataLen; - return pMsg; -} - -void syncSnapshotSendDestroy(SyncSnapshotSend* pMsg) { - if (pMsg != NULL) { - taosMemoryFree(pMsg); + pMsg->contLen = bytes; + if (pMsg->pCont == NULL) { + terrno = TSDB_CODE_OUT_OF_MEMORY; + return -1; } + + SyncSnapshotSend* pSnapshotSend = pMsg->pCont; + pSnapshotSend->bytes = bytes; + pSnapshotSend->vgId = vgId; + pSnapshotSend->msgType = TDMT_SYNC_SNAPSHOT_SEND; + pSnapshotSend->dataLen = dataLen; + return 0; } -void syncSnapshotSendSerialize(const SyncSnapshotSend* pMsg, char* buf, uint32_t bufLen) { - ASSERT(pMsg->bytes <= bufLen); - memcpy(buf, pMsg, pMsg->bytes); -} - -void syncSnapshotSendDeserialize(const char* buf, uint32_t len, SyncSnapshotSend* pMsg) { - memcpy(pMsg, buf, len); - ASSERT(len == pMsg->bytes); - ASSERT(pMsg->bytes == sizeof(SyncSnapshotSend) + pMsg->dataLen); -} - -char* syncSnapshotSendSerialize2(const SyncSnapshotSend* pMsg, uint32_t* len) { - char* buf = taosMemoryMalloc(pMsg->bytes); - ASSERT(buf != NULL); - syncSnapshotSendSerialize(pMsg, buf, pMsg->bytes); - if (len != NULL) { - *len = pMsg->bytes; - } - return buf; -} - -SyncSnapshotSend* syncSnapshotSendDeserialize2(const char* buf, uint32_t len) { - uint32_t bytes = *((uint32_t*)buf); - SyncSnapshotSend* pMsg = taosMemoryMalloc(bytes); - ASSERT(pMsg != NULL); - syncSnapshotSendDeserialize(buf, len, pMsg); - ASSERT(len == pMsg->bytes); - return pMsg; -} - -void syncSnapshotSend2RpcMsg(const SyncSnapshotSend* pMsg, SRpcMsg* pRpcMsg) { - memset(pRpcMsg, 0, sizeof(*pRpcMsg)); - pRpcMsg->msgType = pMsg->msgType; - pRpcMsg->contLen = pMsg->bytes; - pRpcMsg->pCont = rpcMallocCont(pRpcMsg->contLen); - syncSnapshotSendSerialize(pMsg, pRpcMsg->pCont, pRpcMsg->contLen); -} - -void syncSnapshotSendFromRpcMsg(const SRpcMsg* pRpcMsg, SyncSnapshotSend* pMsg) { - syncSnapshotSendDeserialize(pRpcMsg->pCont, pRpcMsg->contLen, pMsg); -} - -SyncSnapshotSend* syncSnapshotSendFromRpcMsg2(const SRpcMsg* pRpcMsg) { - SyncSnapshotSend* pMsg = syncSnapshotSendDeserialize2(pRpcMsg->pCont, pRpcMsg->contLen); - ASSERT(pMsg != NULL); - return pMsg; -} - -SyncSnapshotRsp* syncSnapshotRspBuild(int32_t vgId) { - uint32_t bytes = sizeof(SyncSnapshotRsp); - SyncSnapshotRsp* pMsg = taosMemoryMalloc(bytes); - memset(pMsg, 0, bytes); - pMsg->bytes = bytes; - pMsg->vgId = vgId; +int32_t syncBuildSnapshotSendRsp(SRpcMsg* pMsg, int32_t vgId) { + int32_t bytes = sizeof(SyncSnapshotRsp); + pMsg->pCont = rpcMallocCont(bytes); pMsg->msgType = TDMT_SYNC_SNAPSHOT_RSP; - return pMsg; -} - -void syncSnapshotRspDestroy(SyncSnapshotRsp* pMsg) { - if (pMsg != NULL) { - taosMemoryFree(pMsg); - } -} - -void syncSnapshotRspSerialize(const SyncSnapshotRsp* pMsg, char* buf, uint32_t bufLen) { - ASSERT(pMsg->bytes <= bufLen); - memcpy(buf, pMsg, pMsg->bytes); -} - -void syncSnapshotRspDeserialize(const char* buf, uint32_t len, SyncSnapshotRsp* pMsg) { - memcpy(pMsg, buf, len); - ASSERT(len == pMsg->bytes); -} - -char* syncSnapshotRspSerialize2(const SyncSnapshotRsp* pMsg, uint32_t* len) { - char* buf = taosMemoryMalloc(pMsg->bytes); - ASSERT(buf != NULL); - syncSnapshotRspSerialize(pMsg, buf, pMsg->bytes); - if (len != NULL) { - *len = pMsg->bytes; - } - return buf; -} - -SyncSnapshotRsp* syncSnapshotRspDeserialize2(const char* buf, uint32_t len) { - uint32_t bytes = *((uint32_t*)buf); - SyncSnapshotRsp* pMsg = taosMemoryMalloc(bytes); - ASSERT(pMsg != NULL); - syncSnapshotRspDeserialize(buf, len, pMsg); - ASSERT(len == pMsg->bytes); - return pMsg; -} - -void syncSnapshotRsp2RpcMsg(const SyncSnapshotRsp* pMsg, SRpcMsg* pRpcMsg) { - memset(pRpcMsg, 0, sizeof(*pRpcMsg)); - pRpcMsg->msgType = pMsg->msgType; - pRpcMsg->contLen = pMsg->bytes; - pRpcMsg->pCont = rpcMallocCont(pRpcMsg->contLen); - syncSnapshotRspSerialize(pMsg, pRpcMsg->pCont, pRpcMsg->contLen); -} - -void syncSnapshotRspFromRpcMsg(const SRpcMsg* pRpcMsg, SyncSnapshotRsp* pMsg) { - syncSnapshotRspDeserialize(pRpcMsg->pCont, pRpcMsg->contLen, pMsg); -} - -SyncSnapshotRsp* syncSnapshotRspFromRpcMsg2(const SRpcMsg* pRpcMsg) { - SyncSnapshotRsp* pMsg = syncSnapshotRspDeserialize2(pRpcMsg->pCont, pRpcMsg->contLen); - ASSERT(pMsg != NULL); - return pMsg; -} - -cJSON* syncSnapshotRsp2Json(const SyncSnapshotRsp* 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* 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]; - 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]; - 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->term); - cJSON_AddStringToObject(pRoot, "term", u64buf); - - snprintf(u64buf, sizeof(u64buf), "%" PRId64, pMsg->startTime); - cJSON_AddStringToObject(pRoot, "startTime", u64buf); - - snprintf(u64buf, sizeof(u64buf), "%" PRId64, pMsg->lastIndex); - cJSON_AddStringToObject(pRoot, "lastIndex", u64buf); - - snprintf(u64buf, sizeof(u64buf), "%" PRIu64, pMsg->lastTerm); - cJSON_AddStringToObject(pRoot, "lastTerm", u64buf); - - cJSON_AddNumberToObject(pRoot, "ack", pMsg->ack); - cJSON_AddNumberToObject(pRoot, "code", pMsg->code); - - snprintf(u64buf, sizeof(u64buf), "%" PRId64, pMsg->snapBeginIndex); - cJSON_AddStringToObject(pRoot, "snap-begin", u64buf); + pMsg->contLen = bytes; + if (pMsg->pCont == NULL) { + terrno = TSDB_CODE_OUT_OF_MEMORY; + return -1; } - cJSON* pJson = cJSON_CreateObject(); - cJSON_AddItemToObject(pJson, "SyncSnapshotRsp", pRoot); - return pJson; + SyncSnapshotRsp* pPreSnapshotRsp = pMsg->pCont; + pPreSnapshotRsp->bytes = bytes; + pPreSnapshotRsp->msgType = TDMT_SYNC_SNAPSHOT_RSP; + pPreSnapshotRsp->vgId = vgId; + return 0; } -char* syncSnapshotRsp2Str(const SyncSnapshotRsp* pMsg) { - cJSON* pJson = syncSnapshotRsp2Json(pMsg); - char* serialized = cJSON_Print(pJson); - cJSON_Delete(pJson); - return serialized; -} - -// for debug ---------------------- -void syncSnapshotRspPrint(const SyncSnapshotRsp* pMsg) { - char* serialized = syncSnapshotRsp2Str(pMsg); - printf("syncSnapshotRspPrint | len:%d | %s \n", (int32_t)strlen(serialized), serialized); - fflush(NULL); - taosMemoryFree(serialized); -} - -void syncSnapshotRspPrint2(char* s, const SyncSnapshotRsp* pMsg) { - char* serialized = syncSnapshotRsp2Str(pMsg); - printf("syncSnapshotRspPrint2 | len:%d | %s | %s \n", (int32_t)strlen(serialized), s, serialized); - fflush(NULL); - taosMemoryFree(serialized); -} - -void syncSnapshotRspLog(const SyncSnapshotRsp* pMsg) { - char* serialized = syncSnapshotRsp2Str(pMsg); - sTrace("syncSnapshotRspLog | len:%d | %s", (int32_t)strlen(serialized), serialized); - taosMemoryFree(serialized); -} - -void syncSnapshotRspLog2(char* s, const SyncSnapshotRsp* pMsg) { - if (gRaftDetailLog) { - char* serialized = syncSnapshotRsp2Str(pMsg); - sTrace("syncSnapshotRspLog2 | len:%d | %s | %s", (int32_t)strlen(serialized), s, serialized); - taosMemoryFree(serialized); - } -} - -// --------------------------------------------- -SyncLeaderTransfer* syncLeaderTransferBuild(int32_t vgId) { - uint32_t bytes = sizeof(SyncLeaderTransfer); - SyncLeaderTransfer* pMsg = taosMemoryMalloc(bytes); - memset(pMsg, 0, bytes); - pMsg->bytes = bytes; - pMsg->vgId = vgId; +int32_t syncBuildLeaderTransfer(SRpcMsg* pMsg, int32_t vgId) { + int32_t bytes = sizeof(SyncLeaderTransfer); + pMsg->pCont = rpcMallocCont(bytes); pMsg->msgType = TDMT_SYNC_LEADER_TRANSFER; - return pMsg; -} - -void syncLeaderTransferDestroy(SyncLeaderTransfer* pMsg) { - if (pMsg != NULL) { - taosMemoryFree(pMsg); - } -} - -void syncLeaderTransferSerialize(const SyncLeaderTransfer* pMsg, char* buf, uint32_t bufLen) { - ASSERT(pMsg->bytes <= bufLen); - memcpy(buf, pMsg, pMsg->bytes); -} - -void syncLeaderTransferDeserialize(const char* buf, uint32_t len, SyncLeaderTransfer* pMsg) { - memcpy(pMsg, buf, len); - ASSERT(len == pMsg->bytes); -} - -char* syncLeaderTransferSerialize2(const SyncLeaderTransfer* pMsg, uint32_t* len) { - char* buf = taosMemoryMalloc(pMsg->bytes); - ASSERT(buf != NULL); - syncLeaderTransferSerialize(pMsg, buf, pMsg->bytes); - if (len != NULL) { - *len = pMsg->bytes; - } - return buf; -} - -SyncLeaderTransfer* syncLeaderTransferDeserialize2(const char* buf, uint32_t len) { - uint32_t bytes = *((uint32_t*)buf); - SyncLeaderTransfer* pMsg = taosMemoryMalloc(bytes); - ASSERT(pMsg != NULL); - syncLeaderTransferDeserialize(buf, len, pMsg); - ASSERT(len == pMsg->bytes); - return pMsg; -} - -void syncLeaderTransfer2RpcMsg(const SyncLeaderTransfer* pMsg, SRpcMsg* pRpcMsg) { - memset(pRpcMsg, 0, sizeof(*pRpcMsg)); - pRpcMsg->msgType = pMsg->msgType; - pRpcMsg->contLen = pMsg->bytes; - pRpcMsg->pCont = rpcMallocCont(pRpcMsg->contLen); - syncLeaderTransferSerialize(pMsg, pRpcMsg->pCont, pRpcMsg->contLen); -} - -void syncLeaderTransferFromRpcMsg(const SRpcMsg* pRpcMsg, SyncLeaderTransfer* pMsg) { - syncLeaderTransferDeserialize(pRpcMsg->pCont, pRpcMsg->contLen, pMsg); -} - -SyncLeaderTransfer* syncLeaderTransferFromRpcMsg2(const SRpcMsg* pRpcMsg) { - SyncLeaderTransfer* pMsg = syncLeaderTransferDeserialize2(pRpcMsg->pCont, pRpcMsg->contLen); - ASSERT(pMsg != NULL); - return pMsg; -} - -cJSON* syncLeaderTransfer2Json(const SyncLeaderTransfer* 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* 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]; - 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]; - 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* pNewerId = cJSON_CreateObject(); - snprintf(u64buf, sizeof(u64buf), "%" PRIu64, pMsg->newLeaderId.addr); - cJSON_AddStringToObject(pNewerId, "addr", u64buf); - { - uint64_t u64 = pMsg->newLeaderId.addr; - cJSON* pTmp = pNewerId; - char host[128]; - uint16_t port; - syncUtilU642Addr(u64, host, sizeof(host), &port); - cJSON_AddStringToObject(pTmp, "addr_host", host); - cJSON_AddNumberToObject(pTmp, "addr_port", port); - } - cJSON_AddNumberToObject(pNewerId, "vgId", pMsg->newLeaderId.vgId); - cJSON_AddItemToObject(pRoot, "newLeaderId", pNewerId); + pMsg->contLen = bytes; + if (pMsg->pCont == NULL) { + terrno = TSDB_CODE_OUT_OF_MEMORY; + return -1; } - cJSON* pJson = cJSON_CreateObject(); - cJSON_AddItemToObject(pJson, "SyncLeaderTransfer", pRoot); - return pJson; -} - -char* syncLeaderTransfer2Str(const SyncLeaderTransfer* pMsg) { - cJSON* pJson = syncLeaderTransfer2Json(pMsg); - char* serialized = cJSON_Print(pJson); - cJSON_Delete(pJson); - return serialized; + SyncLeaderTransfer* pLeaderTransfer = pMsg->pCont; + pLeaderTransfer->bytes = bytes; + pLeaderTransfer->msgType = TDMT_SYNC_LEADER_TRANSFER; + pLeaderTransfer->vgId = vgId; + return 0; } const char* syncLocalCmdGetStr(int32_t cmd) { diff --git a/source/libs/sync/src/syncSnapshot.c b/source/libs/sync/src/syncSnapshot.c index 256be35f05..9bc0a07d48 100644 --- a/source/libs/sync/src/syncSnapshot.c +++ b/source/libs/sync/src/syncSnapshot.c @@ -105,7 +105,10 @@ int32_t snapshotSenderStart(SSyncSnapshotSender *pSender) { pSender->finish = false; // build begin msg - SyncSnapshotSend *pMsg = syncSnapshotSendBuild(0, pSender->pSyncNode->vgId); + SRpcMsg rpcMsg = {0}; + (void)syncBuildSnapshotSend(&rpcMsg, 0, pSender->pSyncNode->vgId); + + SyncSnapshotSend *pMsg = rpcMsg.pCont; pMsg->srcId = pSender->pSyncNode->myRaftId; pMsg->destId = (pSender->pSyncNode->replicasId)[pSender->replicaIndex]; pMsg->term = pSender->pSyncNode->pRaftStore->currentTerm; @@ -118,11 +121,8 @@ int32_t snapshotSenderStart(SSyncSnapshotSender *pSender) { pMsg->seq = SYNC_SNAPSHOT_SEQ_PRE_SNAPSHOT; // send msg - SRpcMsg rpcMsg; - syncSnapshotSend2RpcMsg(pMsg, &rpcMsg); - syncNodeSendMsgById(&(pMsg->destId), pSender->pSyncNode, &rpcMsg); + syncNodeSendMsgById(&pMsg->destId, pSender->pSyncNode, &rpcMsg); syncLogSendSyncSnapshotSend(pSender->pSyncNode, pMsg, ""); - syncSnapshotSendDestroy(pMsg); // event log sSTrace(pSender, "snapshot sender start"); @@ -176,7 +176,10 @@ int32_t snapshotSend(SSyncSnapshotSender *pSender) { } // build msg - SyncSnapshotSend *pMsg = syncSnapshotSendBuild(pSender->blockLen, pSender->pSyncNode->vgId); + SRpcMsg rpcMsg = {0}; + (void)syncBuildSnapshotSend(&rpcMsg, pSender->blockLen, pSender->pSyncNode->vgId); + + SyncSnapshotSend *pMsg = rpcMsg.pCont; pMsg->srcId = pSender->pSyncNode->myRaftId; pMsg->destId = (pSender->pSyncNode->replicasId)[pSender->replicaIndex]; pMsg->term = pSender->pSyncNode->pRaftStore->currentTerm; @@ -192,11 +195,8 @@ int32_t snapshotSend(SSyncSnapshotSender *pSender) { memcpy(pMsg->data, pSender->pCurrentBlock, pSender->blockLen); // send msg - SRpcMsg rpcMsg; - syncSnapshotSend2RpcMsg(pMsg, &rpcMsg); - syncNodeSendMsgById(&(pMsg->destId), pSender->pSyncNode, &rpcMsg); + syncNodeSendMsgById(&pMsg->destId, pSender->pSyncNode, &rpcMsg); syncLogSendSyncSnapshotSend(pSender->pSyncNode, pMsg, ""); - syncSnapshotSendDestroy(pMsg); // event log if (pSender->seq == SYNC_SNAPSHOT_SEQ_END) { @@ -212,7 +212,10 @@ int32_t snapshotReSend(SSyncSnapshotSender *pSender) { // send current block data if (pSender->pCurrentBlock != NULL && pSender->blockLen > 0) { // build msg - SyncSnapshotSend *pMsg = syncSnapshotSendBuild(pSender->blockLen, pSender->pSyncNode->vgId); + SRpcMsg rpcMsg = {0}; + (void)syncBuildSnapshotSend(&rpcMsg, pSender->blockLen, pSender->pSyncNode->vgId); + + SyncSnapshotSend *pMsg = rpcMsg.pCont; pMsg->srcId = pSender->pSyncNode->myRaftId; pMsg->destId = (pSender->pSyncNode->replicasId)[pSender->replicaIndex]; pMsg->term = pSender->pSyncNode->pRaftStore->currentTerm; @@ -224,15 +227,11 @@ int32_t snapshotReSend(SSyncSnapshotSender *pSender) { pMsg->seq = pSender->seq; // pMsg->privateTerm = pSender->privateTerm; - memcpy(pMsg->data, pSender->pCurrentBlock, pSender->blockLen); // send msg - SRpcMsg rpcMsg; - syncSnapshotSend2RpcMsg(pMsg, &rpcMsg); - syncNodeSendMsgById(&(pMsg->destId), pSender->pSyncNode, &rpcMsg); + syncNodeSendMsgById(&pMsg->destId, pSender->pSyncNode, &rpcMsg); syncLogSendSyncSnapshotSend(pSender->pSyncNode, pMsg, ""); - syncSnapshotSendDestroy(pMsg); // event log sSTrace(pSender, "snapshot sender resend"); @@ -546,7 +545,11 @@ _START_RECEIVER: _SEND_REPLY: // build msg ; // make complier happy - SyncSnapshotRsp *pRspMsg = syncSnapshotRspBuild(pSyncNode->vgId); + + SRpcMsg rpcMsg = {0}; + (void)syncBuildSnapshotSendRsp(&rpcMsg, pSyncNode->vgId); + + SyncSnapshotRsp *pRspMsg = rpcMsg.pCont; pRspMsg->srcId = pSyncNode->myRaftId; pRspMsg->destId = pMsg->srcId; pRspMsg->term = pSyncNode->pRaftStore->currentTerm; @@ -558,12 +561,8 @@ _SEND_REPLY: pRspMsg->snapBeginIndex = syncNodeGetSnapBeginIndex(pSyncNode); // send msg - SRpcMsg rpcMsg; - syncSnapshotRsp2RpcMsg(pRspMsg, &rpcMsg); - syncNodeSendMsgById(&(pRspMsg->destId), pSyncNode, &rpcMsg); + syncNodeSendMsgById(&pRspMsg->destId, pSyncNode, &rpcMsg); syncLogSendSyncSnapshotRsp(pSyncNode, pRspMsg, ""); - syncSnapshotRspDestroy(pRspMsg); - return 0; } @@ -585,7 +584,10 @@ static int32_t syncNodeOnSnapshotBegin(SSyncNode *pSyncNode, SyncSnapshotSend *p snapshotReceiverStartWriter(pReceiver, pMsg); // build msg - SyncSnapshotRsp *pRspMsg = syncSnapshotRspBuild(pSyncNode->vgId); + SRpcMsg rpcMsg = {0}; + (void)syncBuildSnapshotSendRsp(&rpcMsg, pSyncNode->vgId); + + SyncSnapshotRsp *pRspMsg = rpcMsg.pCont; pRspMsg->srcId = pSyncNode->myRaftId; pRspMsg->destId = pMsg->srcId; pRspMsg->term = pSyncNode->pRaftStore->currentTerm; @@ -597,12 +599,8 @@ static int32_t syncNodeOnSnapshotBegin(SSyncNode *pSyncNode, SyncSnapshotSend *p pRspMsg->snapBeginIndex = pReceiver->snapshotParam.start; // send msg - SRpcMsg rpcMsg; - syncSnapshotRsp2RpcMsg(pRspMsg, &rpcMsg); - syncNodeSendMsgById(&(pRspMsg->destId), pSyncNode, &rpcMsg); + syncNodeSendMsgById(&pRspMsg->destId, pSyncNode, &rpcMsg); syncLogSendSyncSnapshotRsp(pSyncNode, pRspMsg, ""); - syncSnapshotRspDestroy(pRspMsg); - return 0; } @@ -624,7 +622,10 @@ static int32_t syncNodeOnSnapshotTransfering(SSyncNode *pSyncNode, SyncSnapshotS } // build msg - SyncSnapshotRsp *pRspMsg = syncSnapshotRspBuild(pSyncNode->vgId); + SRpcMsg rpcMsg = {0}; + (void)syncBuildSnapshotSendRsp(&rpcMsg, pSyncNode->vgId); + + SyncSnapshotRsp *pRspMsg = rpcMsg.pCont; pRspMsg->srcId = pSyncNode->myRaftId; pRspMsg->destId = pMsg->srcId; pRspMsg->term = pSyncNode->pRaftStore->currentTerm; @@ -636,12 +637,8 @@ static int32_t syncNodeOnSnapshotTransfering(SSyncNode *pSyncNode, SyncSnapshotS pRspMsg->snapBeginIndex = pReceiver->snapshotParam.start; // send msg - SRpcMsg rpcMsg; - syncSnapshotRsp2RpcMsg(pRspMsg, &rpcMsg); - syncNodeSendMsgById(&(pRspMsg->destId), pSyncNode, &rpcMsg); + syncNodeSendMsgById(&pRspMsg->destId, pSyncNode, &rpcMsg); syncLogSendSyncSnapshotRsp(pSyncNode, pRspMsg, ""); - syncSnapshotRspDestroy(pRspMsg); - return 0; } @@ -664,7 +661,10 @@ static int32_t syncNodeOnSnapshotEnd(SSyncNode *pSyncNode, SyncSnapshotSend *pMs } // build msg - SyncSnapshotRsp *pRspMsg = syncSnapshotRspBuild(pSyncNode->vgId); + SRpcMsg rpcMsg = {0}; + (void)syncBuildSnapshotSendRsp(&rpcMsg, pSyncNode->vgId); + + SyncSnapshotRsp *pRspMsg = rpcMsg.pCont; pRspMsg->srcId = pSyncNode->myRaftId; pRspMsg->destId = pMsg->srcId; pRspMsg->term = pSyncNode->pRaftStore->currentTerm; @@ -676,12 +676,8 @@ static int32_t syncNodeOnSnapshotEnd(SSyncNode *pSyncNode, SyncSnapshotSend *pMs pRspMsg->snapBeginIndex = pReceiver->snapshotParam.start; // send msg - SRpcMsg rpcMsg; - syncSnapshotRsp2RpcMsg(pRspMsg, &rpcMsg); - syncNodeSendMsgById(&(pRspMsg->destId), pSyncNode, &rpcMsg); + syncNodeSendMsgById(&pRspMsg->destId, pSyncNode, &rpcMsg); syncLogSendSyncSnapshotRsp(pSyncNode, pRspMsg, ""); - syncSnapshotRspDestroy(pRspMsg); - return 0; } @@ -705,7 +701,9 @@ static int32_t syncNodeOnSnapshotEnd(SSyncNode *pSyncNode, SyncSnapshotSend *pMs // // condition 5, got data, update ack // -int32_t syncNodeOnSnapshot(SSyncNode *pSyncNode, SyncSnapshotSend *pMsg) { +int32_t syncNodeOnSnapshot(SSyncNode *pSyncNode, const SRpcMsg *pRpcMsg) { + SyncSnapshotSend *pMsg = pRpcMsg->pCont; + // if already drop replica, do not process if (!syncNodeInRaftGroup(pSyncNode, &(pMsg->srcId))) { syncLogRecvSyncSnapshotSend(pSyncNode, pMsg, "not in my config"); @@ -797,13 +795,16 @@ int32_t syncNodeOnSnapshotReplyPre(SSyncNode *pSyncNode, SyncSnapshotRsp *pMsg) } // update next index - syncIndexMgrSetIndex(pSyncNode->pNextIndex, &(pMsg->srcId), snapshot.lastApplyIndex + 1); + syncIndexMgrSetIndex(pSyncNode->pNextIndex, &pMsg->srcId, snapshot.lastApplyIndex + 1); // update seq pSender->seq = SYNC_SNAPSHOT_SEQ_BEGIN; // build begin msg - SyncSnapshotSend *pSendMsg = syncSnapshotSendBuild(0, pSender->pSyncNode->vgId); + SRpcMsg rpcMsg = {0}; + (void)syncBuildSnapshotSend(&rpcMsg, 0, pSender->pSyncNode->vgId); + + SyncSnapshotSend *pSendMsg = rpcMsg.pCont; pSendMsg->srcId = pSender->pSyncNode->myRaftId; pSendMsg->destId = (pSender->pSyncNode->replicasId)[pSender->replicaIndex]; pSendMsg->term = pSender->pSyncNode->pRaftStore->currentTerm; @@ -816,11 +817,8 @@ int32_t syncNodeOnSnapshotReplyPre(SSyncNode *pSyncNode, SyncSnapshotRsp *pMsg) pSendMsg->seq = SYNC_SNAPSHOT_SEQ_BEGIN; // send msg - SRpcMsg rpcMsg; - syncSnapshotSend2RpcMsg(pSendMsg, &rpcMsg); - syncNodeSendMsgById(&(pSendMsg->destId), pSender->pSyncNode, &rpcMsg); + syncNodeSendMsgById(&pSendMsg->destId, pSender->pSyncNode, &rpcMsg); syncLogSendSyncSnapshotSend(pSyncNode, pSendMsg, ""); - syncSnapshotSendDestroy(pSendMsg); return 0; } @@ -831,7 +829,9 @@ int32_t syncNodeOnSnapshotReplyPre(SSyncNode *pSyncNode, SyncSnapshotRsp *pMsg) // condition 2 sender receives ack, set seq = ack + 1, send msg from seq // condition 3 sender receives error msg, just print error log // -int32_t syncNodeOnSnapshotReply(SSyncNode *pSyncNode, SyncSnapshotRsp *pMsg) { +int32_t syncNodeOnSnapshotReply(SSyncNode *pSyncNode, const SRpcMsg *pRpcMsg) { + SyncSnapshotRsp *pMsg = pRpcMsg->pCont; + // if already drop replica, do not process if (!syncNodeInRaftGroup(pSyncNode, &(pMsg->srcId))) { syncLogRecvSyncSnapshotRsp(pSyncNode, pMsg, "maybe replica already dropped"); 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 0df4c69d64..4784587f4a 100644 --- a/source/libs/sync/test/sync_test_lib/inc/syncTest.h +++ b/source/libs/sync/test/sync_test_lib/inc/syncTest.h @@ -389,6 +389,74 @@ void syncPreSnapshotReplyLog2(char* s, const SyncPreSnapshotReply* pMsg); int32_t syncNodeOnPreSnapshot(SSyncNode* ths, SyncPreSnapshot* pMsg); int32_t syncNodeOnPreSnapshotReply(SSyncNode* ths, SyncPreSnapshotReply* pMsg); +SyncApplyMsg* syncApplyMsgBuild(uint32_t dataLen); +SyncApplyMsg* syncApplyMsgBuild2(const SRpcMsg* pOriginalRpcMsg, int32_t vgId, SFsmCbMeta* pMeta); +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); +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 +SyncApplyMsg* syncApplyMsgFromRpcMsg2(const SRpcMsg* pRpcMsg); +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 syncApplyMsgLog(const SyncApplyMsg* pMsg); +void syncApplyMsgLog2(char* s, const SyncApplyMsg* pMsg); + +SyncSnapshotSend* syncSnapshotSendBuild(uint32_t dataLen, int32_t vgId); +void syncSnapshotSendDestroy(SyncSnapshotSend* pMsg); +void syncSnapshotSendSerialize(const SyncSnapshotSend* pMsg, char* buf, uint32_t bufLen); +void syncSnapshotSendDeserialize(const char* buf, uint32_t len, SyncSnapshotSend* pMsg); +char* syncSnapshotSendSerialize2(const SyncSnapshotSend* pMsg, uint32_t* len); +SyncSnapshotSend* syncSnapshotSendDeserialize2(const char* buf, uint32_t len); +void syncSnapshotSend2RpcMsg(const SyncSnapshotSend* pMsg, SRpcMsg* pRpcMsg); +void syncSnapshotSendFromRpcMsg(const SRpcMsg* pRpcMsg, SyncSnapshotSend* pMsg); +SyncSnapshotSend* syncSnapshotSendFromRpcMsg2(const SRpcMsg* pRpcMsg); +cJSON* syncSnapshotSend2Json(const SyncSnapshotSend* pMsg); +char* syncSnapshotSend2Str(const SyncSnapshotSend* pMsg); + +// for debug ---------------------- +void syncSnapshotSendPrint(const SyncSnapshotSend* pMsg); +void syncSnapshotSendPrint2(char* s, const SyncSnapshotSend* pMsg); +void syncSnapshotSendLog(const SyncSnapshotSend* pMsg); +void syncSnapshotSendLog2(char* s, const SyncSnapshotSend* pMsg); + +SyncSnapshotRsp* syncSnapshotRspBuild(int32_t vgId); +void syncSnapshotRspDestroy(SyncSnapshotRsp* pMsg); +void syncSnapshotRspSerialize(const SyncSnapshotRsp* pMsg, char* buf, uint32_t bufLen); +void syncSnapshotRspDeserialize(const char* buf, uint32_t len, SyncSnapshotRsp* pMsg); +char* syncSnapshotRspSerialize2(const SyncSnapshotRsp* pMsg, uint32_t* len); +SyncSnapshotRsp* syncSnapshotRspDeserialize2(const char* buf, uint32_t len); +void syncSnapshotRsp2RpcMsg(const SyncSnapshotRsp* pMsg, SRpcMsg* pRpcMsg); +void syncSnapshotRspFromRpcMsg(const SRpcMsg* pRpcMsg, SyncSnapshotRsp* pMsg); +SyncSnapshotRsp* syncSnapshotRspFromRpcMsg2(const SRpcMsg* pRpcMsg); +cJSON* syncSnapshotRsp2Json(const SyncSnapshotRsp* pMsg); +char* syncSnapshotRsp2Str(const SyncSnapshotRsp* pMsg); + +// for debug ---------------------- +void syncSnapshotRspPrint(const SyncSnapshotRsp* pMsg); +void syncSnapshotRspPrint2(char* s, const SyncSnapshotRsp* pMsg); +void syncSnapshotRspLog(const SyncSnapshotRsp* pMsg); +void syncSnapshotRspLog2(char* s, const SyncSnapshotRsp* pMsg); + +SyncLeaderTransfer* syncLeaderTransferBuild(int32_t vgId); +void syncLeaderTransferDestroy(SyncLeaderTransfer* pMsg); +void syncLeaderTransferSerialize(const SyncLeaderTransfer* pMsg, char* buf, uint32_t bufLen); +void syncLeaderTransferDeserialize(const char* buf, uint32_t len, SyncLeaderTransfer* pMsg); +char* syncLeaderTransferSerialize2(const SyncLeaderTransfer* pMsg, uint32_t* len); +SyncLeaderTransfer* syncLeaderTransferDeserialize2(const char* buf, uint32_t len); +void syncLeaderTransfer2RpcMsg(const SyncLeaderTransfer* pMsg, SRpcMsg* pRpcMsg); +void syncLeaderTransferFromRpcMsg(const SRpcMsg* pRpcMsg, SyncLeaderTransfer* pMsg); +SyncLeaderTransfer* syncLeaderTransferFromRpcMsg2(const SRpcMsg* pRpcMsg); +cJSON* syncLeaderTransfer2Json(const SyncLeaderTransfer* pMsg); +char* syncLeaderTransfer2Str(const SyncLeaderTransfer* pMsg); + #ifdef __cplusplus } #endif 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 d1125c66d9..5a32e809bf 100644 --- a/source/libs/sync/test/sync_test_lib/src/syncMessageDebug.c +++ b/source/libs/sync/test/sync_test_lib/src/syncMessageDebug.c @@ -2239,3 +2239,517 @@ void syncPreSnapshotReplyLog2(char* s, const SyncPreSnapshotReply* pMsg) { taosMemoryFree(serialized); } } + +// ---- message process SyncApplyMsg---- +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_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) { + 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) { + 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) { + 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] = {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_AddNumberToObject(pRoot, "originalRpcType", pMsg->originalRpcType); + + snprintf(u64buf, sizeof(u64buf), "%" PRId64, 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", syncStr(pMsg->fsmMeta.state)); + snprintf(u64buf, sizeof(u64buf), "%" PRIu64, 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) { + char* serialized = syncApplyMsg2Str(pMsg); + printf("syncApplyMsgPrint | len:%d | %s \n", (int32_t)strlen(serialized), serialized); + fflush(NULL); + taosMemoryFree(serialized); +} + +void syncApplyMsgPrint2(char* s, const SyncApplyMsg* pMsg) { + char* serialized = syncApplyMsg2Str(pMsg); + printf("syncApplyMsgPrint2 | len:%d | %s | %s \n", (int32_t)strlen(serialized), s, serialized); + fflush(NULL); + taosMemoryFree(serialized); +} + +void syncApplyMsgLog(const SyncApplyMsg* pMsg) { + char* serialized = syncApplyMsg2Str(pMsg); + sTrace("ssyncApplyMsgLog | len:%d | %s", (int32_t)strlen(serialized), serialized); + taosMemoryFree(serialized); +} + +void syncApplyMsgLog2(char* s, const SyncApplyMsg* pMsg) { + if (gRaftDetailLog) { + char* serialized = syncApplyMsg2Str(pMsg); + sTrace("syncApplyMsgLog2 | len:%d | %s | %s", (int32_t)strlen(serialized), s, serialized); + taosMemoryFree(serialized); + } +} + +// --------------------------------------------- +SyncSnapshotSend* syncSnapshotSendBuild(uint32_t dataLen, int32_t vgId) { + uint32_t bytes = sizeof(SyncSnapshotSend) + dataLen; + SyncSnapshotSend* pMsg = taosMemoryMalloc(bytes); + memset(pMsg, 0, bytes); + pMsg->bytes = bytes; + pMsg->vgId = vgId; + pMsg->msgType = TDMT_SYNC_SNAPSHOT_SEND; + pMsg->dataLen = dataLen; + return pMsg; +} + +void syncSnapshotSendDestroy(SyncSnapshotSend* pMsg) { + if (pMsg != NULL) { + taosMemoryFree(pMsg); + } +} + +void syncSnapshotSendSerialize(const SyncSnapshotSend* pMsg, char* buf, uint32_t bufLen) { + ASSERT(pMsg->bytes <= bufLen); + memcpy(buf, pMsg, pMsg->bytes); +} + +void syncSnapshotSendDeserialize(const char* buf, uint32_t len, SyncSnapshotSend* pMsg) { + memcpy(pMsg, buf, len); + ASSERT(len == pMsg->bytes); + ASSERT(pMsg->bytes == sizeof(SyncSnapshotSend) + pMsg->dataLen); +} + +char* syncSnapshotSendSerialize2(const SyncSnapshotSend* pMsg, uint32_t* len) { + char* buf = taosMemoryMalloc(pMsg->bytes); + ASSERT(buf != NULL); + syncSnapshotSendSerialize(pMsg, buf, pMsg->bytes); + if (len != NULL) { + *len = pMsg->bytes; + } + return buf; +} + +SyncSnapshotSend* syncSnapshotSendDeserialize2(const char* buf, uint32_t len) { + uint32_t bytes = *((uint32_t*)buf); + SyncSnapshotSend* pMsg = taosMemoryMalloc(bytes); + ASSERT(pMsg != NULL); + syncSnapshotSendDeserialize(buf, len, pMsg); + ASSERT(len == pMsg->bytes); + return pMsg; +} + +void syncSnapshotSend2RpcMsg(const SyncSnapshotSend* pMsg, SRpcMsg* pRpcMsg) { + memset(pRpcMsg, 0, sizeof(*pRpcMsg)); + pRpcMsg->msgType = pMsg->msgType; + pRpcMsg->contLen = pMsg->bytes; + pRpcMsg->pCont = rpcMallocCont(pRpcMsg->contLen); + syncSnapshotSendSerialize(pMsg, pRpcMsg->pCont, pRpcMsg->contLen); +} + +void syncSnapshotSendFromRpcMsg(const SRpcMsg* pRpcMsg, SyncSnapshotSend* pMsg) { + syncSnapshotSendDeserialize(pRpcMsg->pCont, pRpcMsg->contLen, pMsg); +} + +SyncSnapshotSend* syncSnapshotSendFromRpcMsg2(const SRpcMsg* pRpcMsg) { + SyncSnapshotSend* pMsg = syncSnapshotSendDeserialize2(pRpcMsg->pCont, pRpcMsg->contLen); + ASSERT(pMsg != NULL); + return pMsg; +} + +SyncSnapshotRsp* syncSnapshotRspBuild(int32_t vgId) { + uint32_t bytes = sizeof(SyncSnapshotRsp); + SyncSnapshotRsp* pMsg = taosMemoryMalloc(bytes); + memset(pMsg, 0, bytes); + pMsg->bytes = bytes; + pMsg->vgId = vgId; + pMsg->msgType = TDMT_SYNC_SNAPSHOT_RSP; + return pMsg; +} + +void syncSnapshotRspDestroy(SyncSnapshotRsp* pMsg) { + if (pMsg != NULL) { + taosMemoryFree(pMsg); + } +} + +void syncSnapshotRspSerialize(const SyncSnapshotRsp* pMsg, char* buf, uint32_t bufLen) { + ASSERT(pMsg->bytes <= bufLen); + memcpy(buf, pMsg, pMsg->bytes); +} + +void syncSnapshotRspDeserialize(const char* buf, uint32_t len, SyncSnapshotRsp* pMsg) { + memcpy(pMsg, buf, len); + ASSERT(len == pMsg->bytes); +} + +char* syncSnapshotRspSerialize2(const SyncSnapshotRsp* pMsg, uint32_t* len) { + char* buf = taosMemoryMalloc(pMsg->bytes); + ASSERT(buf != NULL); + syncSnapshotRspSerialize(pMsg, buf, pMsg->bytes); + if (len != NULL) { + *len = pMsg->bytes; + } + return buf; +} + +SyncSnapshotRsp* syncSnapshotRspDeserialize2(const char* buf, uint32_t len) { + uint32_t bytes = *((uint32_t*)buf); + SyncSnapshotRsp* pMsg = taosMemoryMalloc(bytes); + ASSERT(pMsg != NULL); + syncSnapshotRspDeserialize(buf, len, pMsg); + ASSERT(len == pMsg->bytes); + return pMsg; +} + +void syncSnapshotRsp2RpcMsg(const SyncSnapshotRsp* pMsg, SRpcMsg* pRpcMsg) { + memset(pRpcMsg, 0, sizeof(*pRpcMsg)); + pRpcMsg->msgType = pMsg->msgType; + pRpcMsg->contLen = pMsg->bytes; + pRpcMsg->pCont = rpcMallocCont(pRpcMsg->contLen); + syncSnapshotRspSerialize(pMsg, pRpcMsg->pCont, pRpcMsg->contLen); +} + +void syncSnapshotRspFromRpcMsg(const SRpcMsg* pRpcMsg, SyncSnapshotRsp* pMsg) { + syncSnapshotRspDeserialize(pRpcMsg->pCont, pRpcMsg->contLen, pMsg); +} + +SyncSnapshotRsp* syncSnapshotRspFromRpcMsg2(const SRpcMsg* pRpcMsg) { + SyncSnapshotRsp* pMsg = syncSnapshotRspDeserialize2(pRpcMsg->pCont, pRpcMsg->contLen); + ASSERT(pMsg != NULL); + return pMsg; +} + +cJSON* syncSnapshotRsp2Json(const SyncSnapshotRsp* 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* 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]; + 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]; + 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->term); + cJSON_AddStringToObject(pRoot, "term", u64buf); + + snprintf(u64buf, sizeof(u64buf), "%" PRId64, pMsg->startTime); + cJSON_AddStringToObject(pRoot, "startTime", u64buf); + + snprintf(u64buf, sizeof(u64buf), "%" PRId64, pMsg->lastIndex); + cJSON_AddStringToObject(pRoot, "lastIndex", u64buf); + + snprintf(u64buf, sizeof(u64buf), "%" PRIu64, pMsg->lastTerm); + cJSON_AddStringToObject(pRoot, "lastTerm", u64buf); + + cJSON_AddNumberToObject(pRoot, "ack", pMsg->ack); + cJSON_AddNumberToObject(pRoot, "code", pMsg->code); + + snprintf(u64buf, sizeof(u64buf), "%" PRId64, pMsg->snapBeginIndex); + cJSON_AddStringToObject(pRoot, "snap-begin", u64buf); + } + + cJSON* pJson = cJSON_CreateObject(); + cJSON_AddItemToObject(pJson, "SyncSnapshotRsp", pRoot); + return pJson; +} + +char* syncSnapshotRsp2Str(const SyncSnapshotRsp* pMsg) { + cJSON* pJson = syncSnapshotRsp2Json(pMsg); + char* serialized = cJSON_Print(pJson); + cJSON_Delete(pJson); + return serialized; +} + +// for debug ---------------------- +void syncSnapshotRspPrint(const SyncSnapshotRsp* pMsg) { + char* serialized = syncSnapshotRsp2Str(pMsg); + printf("syncSnapshotRspPrint | len:%d | %s \n", (int32_t)strlen(serialized), serialized); + fflush(NULL); + taosMemoryFree(serialized); +} + +void syncSnapshotRspPrint2(char* s, const SyncSnapshotRsp* pMsg) { + char* serialized = syncSnapshotRsp2Str(pMsg); + printf("syncSnapshotRspPrint2 | len:%d | %s | %s \n", (int32_t)strlen(serialized), s, serialized); + fflush(NULL); + taosMemoryFree(serialized); +} + +void syncSnapshotRspLog(const SyncSnapshotRsp* pMsg) { + char* serialized = syncSnapshotRsp2Str(pMsg); + sTrace("syncSnapshotRspLog | len:%d | %s", (int32_t)strlen(serialized), serialized); + taosMemoryFree(serialized); +} + +void syncSnapshotRspLog2(char* s, const SyncSnapshotRsp* pMsg) { + if (gRaftDetailLog) { + char* serialized = syncSnapshotRsp2Str(pMsg); + sTrace("syncSnapshotRspLog2 | len:%d | %s | %s", (int32_t)strlen(serialized), s, serialized); + taosMemoryFree(serialized); + } +} + +// --------------------------------------------- +SyncLeaderTransfer* syncLeaderTransferBuild(int32_t vgId) { + uint32_t bytes = sizeof(SyncLeaderTransfer); + SyncLeaderTransfer* pMsg = taosMemoryMalloc(bytes); + memset(pMsg, 0, bytes); + pMsg->bytes = bytes; + pMsg->vgId = vgId; + pMsg->msgType = TDMT_SYNC_LEADER_TRANSFER; + return pMsg; +} + +void syncLeaderTransferDestroy(SyncLeaderTransfer* pMsg) { + if (pMsg != NULL) { + taosMemoryFree(pMsg); + } +} + +void syncLeaderTransferSerialize(const SyncLeaderTransfer* pMsg, char* buf, uint32_t bufLen) { + ASSERT(pMsg->bytes <= bufLen); + memcpy(buf, pMsg, pMsg->bytes); +} + +void syncLeaderTransferDeserialize(const char* buf, uint32_t len, SyncLeaderTransfer* pMsg) { + memcpy(pMsg, buf, len); + ASSERT(len == pMsg->bytes); +} + +char* syncLeaderTransferSerialize2(const SyncLeaderTransfer* pMsg, uint32_t* len) { + char* buf = taosMemoryMalloc(pMsg->bytes); + ASSERT(buf != NULL); + syncLeaderTransferSerialize(pMsg, buf, pMsg->bytes); + if (len != NULL) { + *len = pMsg->bytes; + } + return buf; +} + +SyncLeaderTransfer* syncLeaderTransferDeserialize2(const char* buf, uint32_t len) { + uint32_t bytes = *((uint32_t*)buf); + SyncLeaderTransfer* pMsg = taosMemoryMalloc(bytes); + ASSERT(pMsg != NULL); + syncLeaderTransferDeserialize(buf, len, pMsg); + ASSERT(len == pMsg->bytes); + return pMsg; +} + +void syncLeaderTransfer2RpcMsg(const SyncLeaderTransfer* pMsg, SRpcMsg* pRpcMsg) { + memset(pRpcMsg, 0, sizeof(*pRpcMsg)); + pRpcMsg->msgType = pMsg->msgType; + pRpcMsg->contLen = pMsg->bytes; + pRpcMsg->pCont = rpcMallocCont(pRpcMsg->contLen); + syncLeaderTransferSerialize(pMsg, pRpcMsg->pCont, pRpcMsg->contLen); +} + +void syncLeaderTransferFromRpcMsg(const SRpcMsg* pRpcMsg, SyncLeaderTransfer* pMsg) { + syncLeaderTransferDeserialize(pRpcMsg->pCont, pRpcMsg->contLen, pMsg); +} + +SyncLeaderTransfer* syncLeaderTransferFromRpcMsg2(const SRpcMsg* pRpcMsg) { + SyncLeaderTransfer* pMsg = syncLeaderTransferDeserialize2(pRpcMsg->pCont, pRpcMsg->contLen); + ASSERT(pMsg != NULL); + return pMsg; +} + +cJSON* syncLeaderTransfer2Json(const SyncLeaderTransfer* 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* 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]; + 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]; + 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* pNewerId = cJSON_CreateObject(); + snprintf(u64buf, sizeof(u64buf), "%" PRIu64, pMsg->newLeaderId.addr); + cJSON_AddStringToObject(pNewerId, "addr", u64buf); + { + uint64_t u64 = pMsg->newLeaderId.addr; + cJSON* pTmp = pNewerId; + char host[128]; + uint16_t port; + syncUtilU642Addr(u64, host, sizeof(host), &port); + cJSON_AddStringToObject(pTmp, "addr_host", host); + cJSON_AddNumberToObject(pTmp, "addr_port", port); + } + cJSON_AddNumberToObject(pNewerId, "vgId", pMsg->newLeaderId.vgId); + cJSON_AddItemToObject(pRoot, "newLeaderId", pNewerId); + } + + cJSON* pJson = cJSON_CreateObject(); + cJSON_AddItemToObject(pJson, "SyncLeaderTransfer", pRoot); + return pJson; +} + +char* syncLeaderTransfer2Str(const SyncLeaderTransfer* pMsg) { + cJSON* pJson = syncLeaderTransfer2Json(pMsg); + char* serialized = cJSON_Print(pJson); + cJSON_Delete(pJson); + return serialized; +}