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

Merge pull request #18099 from taosdata/fix/TD-20052 

react: sync msg builder
This commit is contained in:
Shengliang Guan 2022-11-13 18:00:20 +08:00 committed by GitHub
parent 6da0b736ba d37d83ee26
commit b82c70e90d
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
7 changed files with 688 additions and 640 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

82
source/libs/sync/inc/syncMessage.h
View File

@ -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
}

4
source/libs/sync/inc/syncSnapshot.h
View File

@ -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

26
source/libs/sync/src/syncMain.c
View File

@ -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;

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

@ -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) {

96
source/libs/sync/src/syncSnapshot.c
View File

@ -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");

68
source/libs/sync/test/sync_test_lib/inc/syncTest.h
View File

@ -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

514
source/libs/sync/test/sync_test_lib/src/syncMessageDebug.c
View File

@ -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;
}