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

enh(sync): raft config change

This commit is contained in:
Minghao Li 2022-05-16 14:10:18 +08:00
parent 3e65caccaa
commit 956ab45a30
6 changed files with 24 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
source/libs/sync/inc/syncIndexMgr.h
View File

@ -35,6 +35,7 @@ typedef struct SSyncIndexMgr {
} SSyncIndexMgr;
SSyncIndexMgr *syncIndexMgrCreate(SSyncNode *pSyncNode);
void syncIndexMgrUpdate(SSyncIndexMgr *pSyncIndexMgr, SSyncNode *pSyncNode);
void syncIndexMgrDestroy(SSyncIndexMgr *pSyncIndexMgr);
void syncIndexMgrClear(SSyncIndexMgr *pSyncIndexMgr);
void syncIndexMgrSetIndex(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId, SyncIndex index);

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

@ -342,6 +342,11 @@ int32_t syncNodeOnAppendEntriesCb(SSyncNode* ths, SyncAppendEntries* pMsg) {
ASSERT(ret == 0);
syncNodeUpdateConfig(ths, &newSyncCfg);
if (ths->state == TAOS_SYNC_STATE_LEADER) {
syncNodeBecomeLeader(ths);
} else {
syncNodeBecomeFollower(ths);
}
}
rpcFreeCont(rpcMsg.pCont);

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

@ -120,6 +120,11 @@ void syncMaybeAdvanceCommitIndex(SSyncNode* pSyncNode) {
ASSERT(ret == 0);
syncNodeUpdateConfig(pSyncNode, &newSyncCfg);
if (pSyncNode->state == TAOS_SYNC_STATE_LEADER) {
syncNodeBecomeLeader(pSyncNode);
} else {
syncNodeBecomeFollower(pSyncNode);
}
}
rpcFreeCont(rpcMsg.pCont);

7
source/libs/sync/src/syncIndexMgr.c
View File

@ -31,6 +31,13 @@ SSyncIndexMgr *syncIndexMgrCreate(SSyncNode *pSyncNode) {
return pSyncIndexMgr;
}
void syncIndexMgrUpdate(SSyncIndexMgr *pSyncIndexMgr, SSyncNode *pSyncNode) {
pSyncIndexMgr->replicas = &(pSyncNode->replicasId);
pSyncIndexMgr->replicaNum = pSyncNode->replicaNum;
pSyncIndexMgr->pSyncNode = pSyncNode;
syncIndexMgrClear(pSyncIndexMgr);
}
void syncIndexMgrDestroy(SSyncIndexMgr *pSyncIndexMgr) {
if (pSyncIndexMgr != NULL) {
taosMemoryFree(pSyncIndexMgr);

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

@ -905,6 +905,11 @@ void syncNodeUpdateConfig(SSyncNode* pSyncNode, SSyncCfg* newConfig) {
for (int i = 0; i < pSyncNode->pRaftCfg->cfg.replicaNum; ++i) {
syncUtilnodeInfo2raftId(&pSyncNode->pRaftCfg->cfg.nodeInfo[i], pSyncNode->vgId, &pSyncNode->replicasId[i]);
}
syncIndexMgrUpdate(pSyncNode->pNextIndex, pSyncNode);
syncIndexMgrUpdate(pSyncNode->pMatchIndex, pSyncNode);
syncNodeLog2("==syncNodeUpdateConfig==", pSyncNode);
}
SSyncNode* syncNodeAcquire(int64_t rid) {

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

@ -220,7 +220,7 @@ int main(int argc, char** argv) {
assert(pSyncNode != NULL);
if (isConfigChange) {
configChange(rid, replicaNum, myIndex);
configChange(rid, 3, myIndex);
}
//---------------------------