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

enh: rename syncLogToAppendEntries to syncBuildAppendEntriesFromRaftLog

This commit is contained in:
Benguang Zhao 2022-11-18 10:09:31 +08:00
parent 03b88ff41c
commit c9c05761d6
4 changed files with 32 additions and 28 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -243,6 +243,8 @@ int32_t syncBuildRequestVote(SRpcMsg* pMsg, int32_t vgId);
int32_t syncBuildRequestVoteReply(SRpcMsg* pMsg, int32_t vgId); int32_t syncBuildRequestVoteReply(SRpcMsg* pMsg, int32_t vgId);
int32_t syncBuildAppendEntries(SRpcMsg* pMsg, int32_t dataLen, int32_t vgId); int32_t syncBuildAppendEntries(SRpcMsg* pMsg, int32_t dataLen, int32_t vgId);
int32_t syncBuildAppendEntriesReply(SRpcMsg* pMsg, int32_t vgId); int32_t syncBuildAppendEntriesReply(SRpcMsg* pMsg, int32_t vgId);
int32_t syncBuildAppendEntriesFromRaftLog(SSyncNode* pNode, SSyncRaftEntry* pEntry, SyncTerm prevLogTerm,
SRpcMsg* pRpcMsg);
int32_t syncBuildHeartbeat(SRpcMsg* pMsg, int32_t vgId); int32_t syncBuildHeartbeat(SRpcMsg* pMsg, int32_t vgId);
int32_t syncBuildHeartbeatReply(SRpcMsg* pMsg, int32_t vgId); int32_t syncBuildHeartbeatReply(SRpcMsg* pMsg, int32_t vgId);
int32_t syncBuildPreSnapshot(SRpcMsg* pMsg, int32_t vgId); int32_t syncBuildPreSnapshot(SRpcMsg* pMsg, int32_t vgId);

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

@ -2313,33 +2313,6 @@ int32_t syncNodeOnLocalCmd(SSyncNode* ths, const SRpcMsg* pRpcMsg) {
return 0; return 0;
} }
int32_t syncLogToAppendEntries(SSyncNode* pNode, SSyncRaftEntry* pEntry, SyncTerm prevLogTerm, SRpcMsg* pRpcMsg) {
uint32_t dataLen = pEntry->bytes;
uint32_t bytes = sizeof(SyncAppendEntries) + dataLen;
pRpcMsg->contLen = bytes;
pRpcMsg->pCont = rpcMallocCont(pRpcMsg->contLen);
if (pRpcMsg->pCont == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
SyncAppendEntries* pMsg = pRpcMsg->pCont;
pMsg->bytes = pRpcMsg->contLen;
pMsg->msgType = pRpcMsg->msgType = TDMT_SYNC_APPEND_ENTRIES;
pMsg->dataLen = dataLen;
(void)memcpy(pMsg->data, pEntry, dataLen);
pMsg->prevLogIndex = pEntry->index - 1;
pMsg->prevLogTerm = prevLogTerm;
pMsg->vgId = pNode->vgId;
pMsg->srcId = pNode->myRaftId;
pMsg->term = pNode->pRaftStore->currentTerm;
pMsg->commitIndex = pNode->commitIndex;
pMsg->privateTerm = 0;
return 0;
}
// TLA+ Spec // TLA+ Spec
// ClientRequest(i, v) == // ClientRequest(i, v) ==
// /\ state[i] = Leader // /\ state[i] = Leader

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

@ -16,6 +16,7 @@
#define _DEFAULT_SOURCE #define _DEFAULT_SOURCE
#include "syncMessage.h" #include "syncMessage.h"
#include "syncRaftEntry.h" #include "syncRaftEntry.h"
#include "syncRaftStore.h"
int32_t syncBuildTimeout(SRpcMsg* pMsg, ESyncTimeoutType timeoutType, uint64_t logicClock, int32_t timerMS, int32_t syncBuildTimeout(SRpcMsg* pMsg, ESyncTimeoutType timeoutType, uint64_t logicClock, int32_t timerMS,
SSyncNode* pNode) { SSyncNode* pNode) {
@ -152,6 +153,34 @@ int32_t syncBuildAppendEntriesReply(SRpcMsg* pMsg, int32_t vgId) {
return 0; return 0;
} }
int32_t syncBuildAppendEntriesFromRaftLog(SSyncNode* pNode, SSyncRaftEntry* pEntry, SyncTerm prevLogTerm,
SRpcMsg* pRpcMsg) {
uint32_t dataLen = pEntry->bytes;
uint32_t bytes = sizeof(SyncAppendEntries) + dataLen;
pRpcMsg->contLen = bytes;
pRpcMsg->pCont = rpcMallocCont(pRpcMsg->contLen);
if (pRpcMsg->pCont == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
SyncAppendEntries* pMsg = pRpcMsg->pCont;
pMsg->bytes = pRpcMsg->contLen;
pMsg->msgType = pRpcMsg->msgType = TDMT_SYNC_APPEND_ENTRIES;
pMsg->dataLen = dataLen;
(void)memcpy(pMsg->data, pEntry, dataLen);
pMsg->prevLogIndex = pEntry->index - 1;
pMsg->prevLogTerm = prevLogTerm;
pMsg->vgId = pNode->vgId;
pMsg->srcId = pNode->myRaftId;
pMsg->term = pNode->pRaftStore->currentTerm;
pMsg->commitIndex = pNode->commitIndex;
pMsg->privateTerm = 0;
return 0;
}
int32_t syncBuildHeartbeat(SRpcMsg* pMsg, int32_t vgId) { int32_t syncBuildHeartbeat(SRpcMsg* pMsg, int32_t vgId) {
int32_t bytes = sizeof(SyncHeartbeat); int32_t bytes = sizeof(SyncHeartbeat);
pMsg->pCont = rpcMallocCont(bytes); pMsg->pCont = rpcMallocCont(bytes);

2
source/libs/sync/src/syncPipeline.c
View File

@ -911,7 +911,7 @@ int32_t syncLogBufferReplicateOneTo(SSyncLogReplMgr* pMgr, SSyncNode* pNode, Syn
} }
if (pTerm) *pTerm = pEntry->term; if (pTerm) *pTerm = pEntry->term;
int32_t code = syncLogToAppendEntries(pNode, pEntry, prevLogTerm, &msgOut); int32_t code = syncBuildAppendEntriesFromRaftLog(pNode, pEntry, prevLogTerm, &msgOut);
if (code < 0) { if (code < 0) {
sError("vgId:%d, failed to get append entries for index:%" PRId64 "", pNode->vgId, index); sError("vgId:%d, failed to get append entries for index:%" PRId64 "", pNode->vgId, index);
goto _err; goto _err;