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

fix(sync): when receive snapshot rsp from a dropped replica

This commit is contained in:
Minghao Li 2022-06-14 10:24:05 +08:00
parent e4897e5f46
commit 3887e940f1
2 changed files with 18 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -1968,8 +1968,9 @@ static int32_t syncNodeConfigChange(SSyncNode* ths, SRpcMsg* pRpcMsg, SSyncRaftE
bool isDrop; bool isDrop;
if (IamInNew || (!IamInNew && ths->state != TAOS_SYNC_STATE_LEADER)) { if (IamInNew || (!IamInNew && ths->state != TAOS_SYNC_STATE_LEADER)) {
syncNodeUpdateConfig(ths, &newSyncCfg, pEntry->index, &isDrop);
syncNodeUpdateConfig(ths, &newSyncCfg, pEntry->index, &isDrop);
// change isStandBy to normal // change isStandBy to normal
if (!isDrop) { if (!isDrop) {
if (ths->state == TAOS_SYNC_STATE_LEADER) { if (ths->state == TAOS_SYNC_STATE_LEADER) {
@ -1978,14 +1979,16 @@ static int32_t syncNodeConfigChange(SSyncNode* ths, SRpcMsg* pRpcMsg, SSyncRaftE
syncNodeBecomeFollower(ths, "config change"); syncNodeBecomeFollower(ths, "config change");
} }
} }
} else {
if (gRaftDetailLog) { syncNodeBecomeFollower(ths, "config change");
char* sOld = syncCfg2Str(&oldSyncCfg); }
char* sNew = syncCfg2Str(&newSyncCfg);
sInfo("==config change== 0x11 old:%s new:%s isDrop:%d \n", sOld, sNew, isDrop); if (gRaftDetailLog) {
taosMemoryFree(sOld); char* sOld = syncCfg2Str(&oldSyncCfg);
taosMemoryFree(sNew); char* sNew = syncCfg2Str(&newSyncCfg);
} sInfo("==config change== 0x11 old:%s new:%s isDrop:%d index:%ld IamInNew:%d \n", sOld, sNew, isDrop, pEntry->index, IamInNew);
taosMemoryFree(sOld);
taosMemoryFree(sNew);
} }
// always call FpReConfigCb // always call FpReConfigCb

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

@ -755,6 +755,12 @@ int32_t syncNodeOnSnapshotSendCb(SSyncNode *pSyncNode, SyncSnapshotSend *pMsg) {
// sender receives ack, set seq = ack + 1, send msg from seq // sender receives ack, set seq = ack + 1, send msg from seq
// if ack == SYNC_SNAPSHOT_SEQ_END, stop sender // if ack == SYNC_SNAPSHOT_SEQ_END, stop sender
int32_t syncNodeOnSnapshotRspCb(SSyncNode *pSyncNode, SyncSnapshotRsp *pMsg) { int32_t syncNodeOnSnapshotRspCb(SSyncNode *pSyncNode, SyncSnapshotRsp *pMsg) {
// if already drop replica, do not process
if (!syncNodeInRaftGroup(pSyncNode, &(pMsg->srcId)) && pSyncNode->state == TAOS_SYNC_STATE_LEADER) {
sInfo("recv SyncSnapshotRsp maybe replica already dropped");
return 0;
}
// get sender // get sender
SSyncSnapshotSender *pSender = syncNodeGetSnapshotSender(pSyncNode, &(pMsg->srcId)); SSyncSnapshotSender *pSender = syncNodeGetSnapshotSender(pSyncNode, &(pMsg->srcId));
ASSERT(pSender != NULL); ASSERT(pSender != NULL);