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

refactor(sync): add local-cmd:follower-commit

This commit is contained in:
Minghao Li 2022-11-03 10:25:38 +08:00
parent 1a4b7622e4
commit 102969086d
5 changed files with 33 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -729,6 +729,7 @@ void syncReconfigFinishLog2(char* s, const SyncReconfigFinish* pMsg);
typedef enum {
SYNC_LOCAL_CMD_STEP_DOWN = 100,
SYNC_LOCAL_CMD_FOLLOWER_CMT,
} ESyncLocalCmd;
const char* syncLocalCmdGetStr(int32_t cmd);
@ -742,6 +743,7 @@ typedef struct SyncLocalCmd {
int32_t cmd;
SyncTerm sdNewTerm; // step down new term
SyncIndex fcIndex;// follower commit index
} SyncLocalCmd;

5
source/libs/sync/src/syncAppendEntries.c
View File

@ -90,6 +90,11 @@
//
int32_t syncNodeFollowerCommit(SSyncNode* ths, SyncIndex newCommitIndex) {
if (ths->state != TAOS_SYNC_STATE_FOLLOWER) {
syncNodeEventLog(ths, "can not do follower commit");
return -1;
}
// maybe update commit index, leader notice me
if (newCommitIndex > ths->commitIndex) {
// has commit entry in local

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

@ -2842,11 +2842,25 @@ int32_t syncNodeOnHeartbeat(SSyncNode* ths, SyncHeartbeat* pMsg) {
syncNodeResetElectTimer(ths);
ths->minMatchIndex = pMsg->minMatchIndex;
#if 0
if (ths->state == TAOS_SYNC_STATE_FOLLOWER) {
syncNodeFollowerCommit(ths, pMsg->commitIndex);
// syncNodeFollowerCommit(ths, pMsg->commitIndex);
SyncLocalCmd* pSyncMsg = syncLocalCmdBuild(ths->vgId);
pSyncMsg->cmd = SYNC_LOCAL_CMD_FOLLOWER_CMT;
pSyncMsg->fcIndex = pMsg->commitIndex;
SRpcMsg rpcMsgLocalCmd;
syncLocalCmd2RpcMsg(pSyncMsg, &rpcMsgLocalCmd);
if (ths->syncEqMsg != NULL && ths->msgcb != NULL) {
int32_t code = ths->syncEqMsg(ths->msgcb, &rpcMsgLocalCmd);
if (code != 0) {
sError("vgId:%d, sync enqueue fc-commit msg error, code:%d", ths->vgId, code);
rpcFreeCont(rpcMsgLocalCmd.pCont);
} else {
sTrace("vgId:%d, sync enqueue fc-commit msg, fc-index: %" PRIu64, ths->vgId, pSyncMsg->fcIndex);
}
}
}
#endif
}
if (pMsg->term >= ths->pRaftStore->currentTerm && ths->state != TAOS_SYNC_STATE_FOLLOWER) {
@ -2900,6 +2914,9 @@ int32_t syncNodeOnLocalCmd(SSyncNode* ths, SyncLocalCmd* pMsg) {
if (pMsg->cmd == SYNC_LOCAL_CMD_STEP_DOWN) {
syncNodeStepDown(ths, pMsg->sdNewTerm);
} else if (pMsg->cmd == SYNC_LOCAL_CMD_FOLLOWER_CMT) {
syncNodeFollowerCommit(ths, pMsg->fcIndex);
} else {
syncNodeErrorLog(ths, "error local cmd");
}

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

@ -3400,6 +3400,8 @@ void syncReconfigFinishLog2(char* s, const SyncReconfigFinish* pMsg) {
const char* syncLocalCmdGetStr(int32_t cmd) {
if (cmd == SYNC_LOCAL_CMD_STEP_DOWN) {
return "step-down";
} else if (cmd == SYNC_LOCAL_CMD_FOLLOWER_CMT) {
return "follower-commit";
}
return "unknown-local-cmd";
@ -3511,6 +3513,9 @@ cJSON* syncLocalCmd2Json(const SyncLocalCmd* pMsg) {
snprintf(u64buf, sizeof(u64buf), "%" PRIu64, pMsg->sdNewTerm);
cJSON_AddStringToObject(pRoot, "sd-new-term", u64buf);
snprintf(u64buf, sizeof(u64buf), "%" PRId64, pMsg->fcIndex);
cJSON_AddStringToObject(pRoot, "fc-index", u64buf);
}
cJSON* pJson = cJSON_CreateObject();

1
source/libs/sync/test/syncLocalCmdTest.cpp
View File

@ -21,6 +21,7 @@ SyncLocalCmd *createMsg() {
pMsg->destId.addr = syncUtilAddr2U64("127.0.0.1", 5678);
pMsg->destId.vgId = 100;
pMsg->sdNewTerm = 123;
pMsg->fcIndex = 456;
pMsg->cmd = SYNC_LOCAL_CMD_STEP_DOWN;
return pMsg;