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

FpReConfigCb

This commit is contained in:
Minghao Li 2022-05-25 14:43:45 +08:00
parent 81d0798f5f
commit 104a207579
6 changed files with 28 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
include/libs/sync/sync.h
View File

@ -105,7 +105,8 @@ typedef struct SSyncFSM {
// *ppIter.next(); // *ppIter.next();
// //
// if success, return 0. else return error code // if success, return 0. else return error code
int32_t (*FpSnapshotRead)(struct SSyncFSM* pFsm, const SSnapshot* pSnapshot, void** ppIter, char** ppBuf, int32_t* len); int32_t (*FpSnapshotRead)(struct SSyncFSM* pFsm, const SSnapshot* pSnapshot, void** ppIter, char** ppBuf,
int32_t* len);
// apply data into fsm // apply data into fsm
int32_t (*FpSnapshotApply)(struct SSyncFSM* pFsm, const SSnapshot* pSnapshot, char* pBuf, int32_t len); int32_t (*FpSnapshotApply)(struct SSyncFSM* pFsm, const SSnapshot* pSnapshot, char* pBuf, int32_t len);

1
source/dnode/mnode/impl/src/mndSync.c
View File

@ -74,6 +74,7 @@ int32_t mndSnapshotApply(struct SSyncFSM* pFsm, const SSnapshot* pSnapshot, char
} }
void mndReConfig(struct SSyncFSM* pFsm, SSyncCfg newCfg, SReConfigCbMeta cbMeta) { void mndReConfig(struct SSyncFSM* pFsm, SSyncCfg newCfg, SReConfigCbMeta cbMeta) {
mInfo("mndReConfig cbMeta.code:%d, cbMeta.currentTerm:%ld, cbMeta.term:%ld, cbMeta.index:%ld", cbMeta.code, cbMeta.currentTerm, cbMeta.term, cbMeta.index);
if (cbMeta.code == 0) { if (cbMeta.code == 0) {
// config change success // config change success
} else { } else {

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

@ -357,6 +357,16 @@ int32_t syncNodeOnAppendEntriesCb(SSyncNode* ths, SyncAppendEntries* pMsg) {
} else { } else {
syncNodeBecomeFollower(ths); syncNodeBecomeFollower(ths);
} }
// maybe newSyncCfg.myIndex is updated in syncNodeUpdateConfig
if (ths->pFsm->FpReConfigCb != NULL) {
SReConfigCbMeta cbMeta = {0};
cbMeta.code = 0;
cbMeta.currentTerm = ths->pRaftStore->currentTerm;
cbMeta.index = pEntry->index;
cbMeta.term = pEntry->term;
ths->pFsm->FpReConfigCb(ths->pFsm, newSyncCfg, cbMeta);
}
} }
// restore finish // restore finish

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

@ -134,6 +134,16 @@ void syncMaybeAdvanceCommitIndex(SSyncNode* pSyncNode) {
} else { } else {
syncNodeBecomeFollower(pSyncNode); syncNodeBecomeFollower(pSyncNode);
} }
// maybe newSyncCfg.myIndex is updated in syncNodeUpdateConfig
if (pSyncNode->pFsm->FpReConfigCb != NULL) {
SReConfigCbMeta cbMeta = {0};
cbMeta.code = 0;
cbMeta.currentTerm = pSyncNode->pRaftStore->currentTerm;
cbMeta.index = pEntry->index;
cbMeta.term = pEntry->term;
pSyncNode->pFsm->FpReConfigCb(pSyncNode->pFsm, newSyncCfg, cbMeta);
}
} }
// restore finish // restore finish