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

refactor(sync): pre snapshot on message

This commit is contained in:
Minghao Li 2022-11-01 15:20:08 +08:00
parent 21d60cb5be
commit 59448cd18e
5 changed files with 60 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -550,7 +550,7 @@ typedef struct SyncPreSnapshotReply {
// private data
SyncTerm term;
SyncIndex matchIndex;
SyncIndex snapStart;
} SyncPreSnapshotReply;

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

@ -3702,8 +3702,9 @@ void syncLogSendSyncPreSnapshotReply(SSyncNode* pSyncNode, const SyncPreSnapshot
uint16_t port;
syncUtilU642Addr(pMsg->destId.addr, host, sizeof(host), &port);
char logBuf[256];
snprintf(logBuf, sizeof(logBuf), "send sync-pre-snapshot-reply to %s:%d {term:%" PRIu64 ", match:%" PRId64 "}, %s",
host, port, pMsg->term, pMsg->matchIndex, s);
snprintf(logBuf, sizeof(logBuf),
"send sync-pre-snapshot-reply to %s:%d {term:%" PRIu64 ", snap-start:%" PRId64 "}, %s", host, port,
pMsg->term, pMsg->snapStart, s);
syncNodeEventLog(pSyncNode, logBuf);
}
@ -3712,7 +3713,8 @@ void syncLogRecvSyncPreSnapshotReply(SSyncNode* pSyncNode, const SyncPreSnapshot
uint16_t port;
syncUtilU642Addr(pMsg->srcId.addr, host, sizeof(host), &port);
char logBuf[256];
snprintf(logBuf, sizeof(logBuf), "recv sync-pre-snapshot-reply from %s:%d {term:%" PRIu64 ", match:%" PRId64 "}, %s",
host, port, pMsg->term, pMsg->matchIndex, s);
snprintf(logBuf, sizeof(logBuf),
"recv sync-pre-snapshot-reply from %s:%d {term:%" PRIu64 ", snap-start:%" PRId64 "}, %s", host, port,
pMsg->term, pMsg->snapStart, s);
syncNodeEventLog(pSyncNode, logBuf);
}

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

@ -2568,8 +2568,8 @@ cJSON* syncPreSnapshotReply2Json(const SyncPreSnapshotReply* pMsg) {
snprintf(u64buf, sizeof(u64buf), "%" PRIu64, pMsg->term);
cJSON_AddStringToObject(pRoot, "term", u64buf);
snprintf(u64buf, sizeof(u64buf), "%" PRId64, pMsg->matchIndex);
cJSON_AddStringToObject(pRoot, "match-index", u64buf);
snprintf(u64buf, sizeof(u64buf), "%" PRId64, pMsg->snapStart);
cJSON_AddStringToObject(pRoot, "snap-start", u64buf);
}
cJSON* pJson = cJSON_CreateObject();

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

@ -925,10 +925,60 @@ int32_t syncNodeOnSnapshotReply(SSyncNode *pSyncNode, SyncSnapshotRsp *pMsg) {
int32_t syncNodeOnPreSnapshot(SSyncNode *ths, SyncPreSnapshot *pMsg) {
syncLogRecvSyncPreSnapshot(ths, pMsg, "");
SyncPreSnapshotReply *pMsgReply = syncPreSnapshotReplyBuild(ths->vgId);
pMsgReply->srcId = ths->myRaftId;
pMsgReply->destId = pMsg->srcId;
pMsgReply->term = ths->pRaftStore->currentTerm;
SSyncLogStoreData *pData = ths->pLogStore->data;
SWal *pWal = pData->pWal;
if (syncNodeIsMnode(ths)) {
pMsgReply->snapStart = SYNC_INDEX_BEGIN;
} else {
bool isEmpty = ths->pLogStore->syncLogIsEmpty(ths->pLogStore);
int64_t walCommitVer = walGetCommittedVer(pWal);
if (!isEmpty && ths->commitIndex != walCommitVer) {
char logBuf[128];
snprintf(logBuf, sizeof(logBuf), "commit not same, wal-commit:%" PRId64 ", commit:%" PRId64 ", ignore",
walCommitVer, ths->commitIndex);
syncNodeErrorLog(ths, logBuf);
goto _IGNORE;
}
pMsgReply->snapStart = ths->commitIndex + 1;
// make local log clean
int32_t code = ths->pLogStore->syncLogTruncate(ths->pLogStore, pMsgReply->snapStart);
if (code != 0) {
syncNodeErrorLog(ths, "truncate wal error");
goto _IGNORE;
}
}
// can not write behind _RESPONSE
SRpcMsg rpcMsg;
_RESPONSE:
syncPreSnapshotReply2RpcMsg(pMsgReply, &rpcMsg);
syncNodeSendMsgById(&pMsgReply->destId, ths, &rpcMsg);
syncPreSnapshotReplyDestroy(pMsgReply);
return 0;
_IGNORE:
syncPreSnapshotReplyDestroy(pMsgReply);
return 0;
}
int32_t syncNodeOnPreSnapshotReply(SSyncNode *ths, SyncPreSnapshotReply *pMsg) {
syncLogRecvSyncPreSnapshotReply(ths, pMsg, "");
// start snapshot
return 0;
}

2
source/libs/sync/test/syncPreSnapshotReplyTest.cpp
View File

@ -22,7 +22,7 @@ SyncPreSnapshotReply *createMsg() {
pMsg->destId.addr = syncUtilAddr2U64("127.0.0.1", 5678);
pMsg->destId.vgId = 100;
pMsg->term = 9527;
pMsg->matchIndex = 12306;
pMsg->snapStart = 12306;
return pMsg;
}