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

Merge branch 'fix/mnode' into fix/dnode

This commit is contained in:
Shengliang Guan 2022-05-26 21:55:46 +08:00
parent 2ed001ce68 3ba372a836
commit e58b6f37c5
17 changed files with 466 additions and 152 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -80,6 +80,7 @@ typedef struct SFsmCbMeta {
uint64_t seqNum; uint64_t seqNum;
SyncTerm term; SyncTerm term;
SyncTerm currentTerm; SyncTerm currentTerm;
uint64_t flag;
} SFsmCbMeta; } SFsmCbMeta;
typedef struct SReConfigCbMeta { typedef struct SReConfigCbMeta {
@ -87,6 +88,9 @@ typedef struct SReConfigCbMeta {
SyncIndex index; SyncIndex index;
SyncTerm term; SyncTerm term;
SyncTerm currentTerm; SyncTerm currentTerm;
SSyncCfg oldCfg;
bool isDrop;
uint64_t flag;
} SReConfigCbMeta; } SReConfigCbMeta;
typedef struct SSyncFSM { typedef struct SSyncFSM {
@ -162,6 +166,7 @@ void syncCleanUp();
int64_t syncOpen(const SSyncInfo* pSyncInfo); int64_t syncOpen(const SSyncInfo* pSyncInfo);
void syncStart(int64_t rid); void syncStart(int64_t rid);
void syncStop(int64_t rid); void syncStop(int64_t rid);
int32_t syncSetStandby(int64_t rid);
int32_t syncReconfig(int64_t rid, const SSyncCfg* pSyncCfg); int32_t syncReconfig(int64_t rid, const SSyncCfg* pSyncCfg);
ESyncState syncGetMyRole(int64_t rid); ESyncState syncGetMyRole(int64_t rid);
const char* syncGetMyRoleStr(int64_t rid); const char* syncGetMyRoleStr(int64_t rid);

7
source/dnode/mgmt/mgmt_dnode/src/dmHandle.c
View File

@ -92,6 +92,13 @@ void dmSendStatusReq(SDnodeMgmt *pMgmt) {
SEpSet epSet = {0}; SEpSet epSet = {0};
dmGetMnodeEpSet(pMgmt->pData, &epSet); dmGetMnodeEpSet(pMgmt->pData, &epSet);
rpcSendRecv(pMgmt->msgCb.clientRpc, &epSet, &rpcMsg, &rpcRsp); rpcSendRecv(pMgmt->msgCb.clientRpc, &epSet, &rpcMsg, &rpcRsp);
if (rpcRsp.code != 0) {
dError("failed to send status msg since %s, numOfEps:%d inUse:%d", tstrerror(rpcRsp.code), epSet.numOfEps,
epSet.inUse);
for (int32_t i = 0; i < epSet.numOfEps; ++i) {
dDebug("index:%d, mnode ep:%s:%u", i, epSet.eps[i].fqdn, epSet.eps[i].port);
}
}
dmProcessStatusRsp(pMgmt, &rpcRsp); dmProcessStatusRsp(pMgmt, &rpcRsp);
} }

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

@ -144,6 +144,7 @@ _OVER:
static int32_t mndClusterActionInsert(SSdb *pSdb, SClusterObj *pCluster) { static int32_t mndClusterActionInsert(SSdb *pSdb, SClusterObj *pCluster) {
mTrace("cluster:%" PRId64 ", perform insert action, row:%p", pCluster->id, pCluster); mTrace("cluster:%" PRId64 ", perform insert action, row:%p", pCluster->id, pCluster);
pSdb->pMnode->clusterId = pCluster->id;
return 0; return 0;
} }

2
source/dnode/mnode/impl/src/mndDnode.c
View File

@ -441,7 +441,7 @@ static int32_t mndProcessStatusReq(SRpcMsg *pReq) {
pDnode->numOfSupportVnodes = statusReq.numOfSupportVnodes; pDnode->numOfSupportVnodes = statusReq.numOfSupportVnodes;
SStatusRsp statusRsp = {0}; SStatusRsp statusRsp = {0};
statusRsp.dnodeVer = sdbGetTableVer(pMnode->pSdb, SDB_DNODE); statusRsp.dnodeVer = sdbGetTableVer(pMnode->pSdb, SDB_DNODE) + sdbGetTableVer(pMnode->pSdb, SDB_MNODE);
statusRsp.dnodeCfg.dnodeId = pDnode->id; statusRsp.dnodeCfg.dnodeId = pDnode->id;
statusRsp.dnodeCfg.clusterId = pMnode->clusterId; statusRsp.dnodeCfg.clusterId = pMnode->clusterId;
statusRsp.pDnodeEps = taosArrayInit(mndGetDnodeSize(pMnode), sizeof(SDnodeEp)); statusRsp.pDnodeEps = taosArrayInit(mndGetDnodeSize(pMnode), sizeof(SDnodeEp));

2
source/dnode/mnode/impl/src/mndMnode.c
View File

@ -233,7 +233,7 @@ void mndGetMnodeEpSet(SMnode *pMnode, SEpSet *pEpSet) {
if (pObj->pDnode == NULL) { if (pObj->pDnode == NULL) {
mError("mnode:%d, no corresponding dnode exists", pObj->id); mError("mnode:%d, no corresponding dnode exists", pObj->id);
} else { } else {
if (pObj->state == TAOS_SYNC_STATE_LEADER) { if (pObj->id == pMnode->selfDnodeId || pObj->state == TAOS_SYNC_STATE_LEADER) {
pEpSet->inUse = pEpSet->numOfEps; pEpSet->inUse = pEpSet->numOfEps;
} }
addEpIntoEpSet(pEpSet, pObj->pDnode->fqdn, pObj->pDnode->port); addEpIntoEpSet(pEpSet, pObj->pDnode->fqdn, pObj->pDnode->port);

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

@ -51,6 +51,7 @@ int32_t mndSyncGetSnapshot(struct SSyncFSM *pFsm, SSnapshot *pSnapshot) {
void mndRestoreFinish(struct SSyncFSM *pFsm) { void mndRestoreFinish(struct SSyncFSM *pFsm) {
SMnode *pMnode = pFsm->data; SMnode *pMnode = pFsm->data;
if (!pMnode->deploy) { if (!pMnode->deploy) {
mInfo("mnode sync restore finished");
mndTransPullup(pMnode); mndTransPullup(pMnode);
pMnode->syncMgmt.restored = true; pMnode->syncMgmt.restored = true;
} }

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

@ -182,7 +182,7 @@ int32_t syncNodeSendMsgByInfo(const SNodeInfo* nodeInfo, SSyncNode* pSyncNode, S
cJSON* syncNode2Json(const SSyncNode* pSyncNode); cJSON* syncNode2Json(const SSyncNode* pSyncNode);
char* syncNode2Str(const SSyncNode* pSyncNode); char* syncNode2Str(const SSyncNode* pSyncNode);
char* syncNode2SimpleStr(const SSyncNode* pSyncNode); char* syncNode2SimpleStr(const SSyncNode* pSyncNode);
void syncNodeUpdateConfig(SSyncNode* pSyncNode, SSyncCfg* newConfig); void syncNodeUpdateConfig(SSyncNode* pSyncNode, SSyncCfg* newConfig, bool* isDrop);
SSyncNode* syncNodeAcquire(int64_t rid); SSyncNode* syncNodeAcquire(int64_t rid);
void syncNodeRelease(SSyncNode* pNode); void syncNodeRelease(SSyncNode* pNode);

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

@ -333,7 +333,7 @@ int32_t syncNodeOnAppendEntriesCb(SSyncNode* ths, SyncAppendEntries* pMsg) {
cbMeta.seqNum = pEntry->seqNum; cbMeta.seqNum = pEntry->seqNum;
cbMeta.term = pEntry->term; cbMeta.term = pEntry->term;
cbMeta.currentTerm = ths->pRaftStore->currentTerm; cbMeta.currentTerm = ths->pRaftStore->currentTerm;
ths->pFsm->FpCommitCb(ths->pFsm, &rpcMsg, cbMeta); cbMeta.flag = 0x11;
bool needExecute = true; bool needExecute = true;
if (ths->pSnapshot != NULL && cbMeta.index <= ths->pSnapshot->lastApplyIndex) { if (ths->pSnapshot != NULL && cbMeta.index <= ths->pSnapshot->lastApplyIndex) {
@ -347,24 +347,55 @@ int32_t syncNodeOnAppendEntriesCb(SSyncNode* ths, SyncAppendEntries* pMsg) {
// config change // config change
if (pEntry->originalRpcType == TDMT_VND_SYNC_CONFIG_CHANGE) { if (pEntry->originalRpcType == TDMT_VND_SYNC_CONFIG_CHANGE) {
SSyncCfg oldSyncCfg = ths->pRaftCfg->cfg;
SSyncCfg newSyncCfg; SSyncCfg newSyncCfg;
int32_t ret = syncCfgFromStr(rpcMsg.pCont, &newSyncCfg); int32_t ret = syncCfgFromStr(rpcMsg.pCont, &newSyncCfg);
ASSERT(ret == 0); ASSERT(ret == 0);
syncNodeUpdateConfig(ths, &newSyncCfg); // update new config myIndex
if (ths->state == TAOS_SYNC_STATE_LEADER) { bool hit = false;
syncNodeBecomeLeader(ths); for (int i = 0; i < newSyncCfg.replicaNum; ++i) {
} else { if (strcmp(ths->myNodeInfo.nodeFqdn, (newSyncCfg.nodeInfo)[i].nodeFqdn) == 0 &&
syncNodeBecomeFollower(ths); ths->myNodeInfo.nodePort == (newSyncCfg.nodeInfo)[i].nodePort) {
newSyncCfg.myIndex = i;
hit = true;
break;
}
} }
// maybe newSyncCfg.myIndex is updated in syncNodeUpdateConfig SReConfigCbMeta cbMeta = {0};
if (ths->pFsm->FpReConfigCb != NULL) { bool isDrop;
SReConfigCbMeta cbMeta = {0};
// I am in newConfig
if (hit) {
syncNodeUpdateConfig(ths, &newSyncCfg, &isDrop);
// change isStandBy to normal
if (!isDrop) {
if (ths->state == TAOS_SYNC_STATE_LEADER) {
syncNodeBecomeLeader(ths);
} else {
syncNodeBecomeFollower(ths);
}
}
char* sOld = syncCfg2Str(&oldSyncCfg);
char* sNew = syncCfg2Str(&newSyncCfg);
sInfo("==config change== 0x11 old:%s new:%s isDrop:%d \n", sOld, sNew, isDrop);
taosMemoryFree(sOld);
taosMemoryFree(sNew);
}
// always call FpReConfigCb
if (ths->pFsm->FpReConfigCb != NULL) {
cbMeta.code = 0; cbMeta.code = 0;
cbMeta.currentTerm = ths->pRaftStore->currentTerm; cbMeta.currentTerm = ths->pRaftStore->currentTerm;
cbMeta.index = pEntry->index; cbMeta.index = pEntry->index;
cbMeta.term = pEntry->term; cbMeta.term = pEntry->term;
cbMeta.oldCfg = oldSyncCfg;
cbMeta.flag = 0x11;
cbMeta.isDrop = isDrop;
ths->pFsm->FpReConfigCb(ths->pFsm, newSyncCfg, cbMeta); ths->pFsm->FpReConfigCb(ths->pFsm, newSyncCfg, cbMeta);
} }
} }

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

@ -111,6 +111,7 @@ void syncMaybeAdvanceCommitIndex(SSyncNode* pSyncNode) {
cbMeta.seqNum = pEntry->seqNum; cbMeta.seqNum = pEntry->seqNum;
cbMeta.term = pEntry->term; cbMeta.term = pEntry->term;
cbMeta.currentTerm = pSyncNode->pRaftStore->currentTerm; cbMeta.currentTerm = pSyncNode->pRaftStore->currentTerm;
cbMeta.flag = 0x1;
bool needExecute = true; bool needExecute = true;
if (pSyncNode->pSnapshot != NULL && cbMeta.index <= pSyncNode->pSnapshot->lastApplyIndex) { if (pSyncNode->pSnapshot != NULL && cbMeta.index <= pSyncNode->pSnapshot->lastApplyIndex) {
@ -124,24 +125,54 @@ void syncMaybeAdvanceCommitIndex(SSyncNode* pSyncNode) {
// config change // config change
if (pEntry->originalRpcType == TDMT_VND_SYNC_CONFIG_CHANGE) { if (pEntry->originalRpcType == TDMT_VND_SYNC_CONFIG_CHANGE) {
SSyncCfg oldSyncCfg = pSyncNode->pRaftCfg->cfg;
SSyncCfg newSyncCfg; SSyncCfg newSyncCfg;
int32_t ret = syncCfgFromStr(rpcMsg.pCont, &newSyncCfg); int32_t ret = syncCfgFromStr(rpcMsg.pCont, &newSyncCfg);
ASSERT(ret == 0); ASSERT(ret == 0);
syncNodeUpdateConfig(pSyncNode, &newSyncCfg); // update new config myIndex
if (pSyncNode->state == TAOS_SYNC_STATE_LEADER) { bool hit = false;
syncNodeBecomeLeader(pSyncNode); for (int i = 0; i < newSyncCfg.replicaNum; ++i) {
} else { if (strcmp(pSyncNode->myNodeInfo.nodeFqdn, (newSyncCfg.nodeInfo)[i].nodeFqdn) == 0 &&
syncNodeBecomeFollower(pSyncNode); pSyncNode->myNodeInfo.nodePort == (newSyncCfg.nodeInfo)[i].nodePort) {
newSyncCfg.myIndex = i;
hit = true;
break;
}
} }
// maybe newSyncCfg.myIndex is updated in syncNodeUpdateConfig if (pSyncNode->state == TAOS_SYNC_STATE_LEADER) {
ASSERT(hit == true);
}
bool isDrop;
syncNodeUpdateConfig(pSyncNode, &newSyncCfg, &isDrop);
// change isStandBy to normal
if (!isDrop) {
if (pSyncNode->state == TAOS_SYNC_STATE_LEADER) {
syncNodeBecomeLeader(pSyncNode);
} else {
syncNodeBecomeFollower(pSyncNode);
}
}
char* sOld = syncCfg2Str(&oldSyncCfg);
char* sNew = syncCfg2Str(&newSyncCfg);
sInfo("==config change== 0x1 old:%s new:%s isDrop:%d \n", sOld, sNew, isDrop);
taosMemoryFree(sOld);
taosMemoryFree(sNew);
if (pSyncNode->pFsm->FpReConfigCb != NULL) { if (pSyncNode->pFsm->FpReConfigCb != NULL) {
SReConfigCbMeta cbMeta = {0}; SReConfigCbMeta cbMeta = {0};
cbMeta.code = 0; cbMeta.code = 0;
cbMeta.currentTerm = pSyncNode->pRaftStore->currentTerm; cbMeta.currentTerm = pSyncNode->pRaftStore->currentTerm;
cbMeta.index = pEntry->index; cbMeta.index = pEntry->index;
cbMeta.term = pEntry->term; cbMeta.term = pEntry->term;
cbMeta.oldCfg = oldSyncCfg;
cbMeta.flag = 0x1;
cbMeta.isDrop = isDrop;
pSyncNode->pFsm->FpReConfigCb(pSyncNode->pFsm, newSyncCfg, cbMeta); pSyncNode->pFsm->FpReConfigCb(pSyncNode->pFsm, newSyncCfg, cbMeta);
} }
} }

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

@ -141,9 +141,38 @@ void syncStop(int64_t rid) {
taosRemoveRef(tsNodeRefId, rid); taosRemoveRef(tsNodeRefId, rid);
} }
int32_t syncSetStandby(int64_t rid) {
SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid);
if (pSyncNode == NULL) {
return -1;
}
if (pSyncNode->state != TAOS_SYNC_STATE_LEADER) {
taosReleaseRef(tsNodeRefId, pSyncNode->rid);
return -1;
}
// state change
pSyncNode->state = TAOS_SYNC_STATE_FOLLOWER;
syncNodeStopHeartbeatTimer(pSyncNode);
// reset elect timer, long enough
int32_t electMS = TIMER_MAX_MS;
int32_t ret = syncNodeRestartElectTimer(pSyncNode, electMS);
ASSERT(ret == 0);
pSyncNode->pRaftCfg->isStandBy = 1;
raftCfgPersist(pSyncNode->pRaftCfg);
taosReleaseRef(tsNodeRefId, pSyncNode->rid);
return 0;
}
int32_t syncReconfig(int64_t rid, const SSyncCfg* pSyncCfg) { int32_t syncReconfig(int64_t rid, const SSyncCfg* pSyncCfg) {
int32_t ret = 0; int32_t ret = 0;
char* configChange = syncCfg2Str((SSyncCfg*)pSyncCfg); char* configChange = syncCfg2Str((SSyncCfg*)pSyncCfg);
sInfo("==syncReconfig== newconfig:%s", configChange);
SRpcMsg rpcMsg = {0}; SRpcMsg rpcMsg = {0};
rpcMsg.msgType = TDMT_VND_SYNC_CONFIG_CHANGE; rpcMsg.msgType = TDMT_VND_SYNC_CONFIG_CHANGE;
rpcMsg.info.noResp = 1; rpcMsg.info.noResp = 1;
@ -941,30 +970,19 @@ char* syncNode2SimpleStr(const SSyncNode* pSyncNode) {
int len = 256; int len = 256;
char* s = (char*)taosMemoryMalloc(len); char* s = (char*)taosMemoryMalloc(len);
snprintf(s, len, snprintf(s, len,
"syncNode2SimpleStr vgId:%d currentTerm:%lu, commitIndex:%ld, state:%d %s, electTimerLogicClock:%lu, " "syncNode2SimpleStr vgId:%d currentTerm:%lu, commitIndex:%ld, state:%d %s, isStandBy:%d, "
"electTimerLogicClock:%lu, "
"electTimerLogicClockUser:%lu, " "electTimerLogicClockUser:%lu, "
"electTimerMS:%d", "electTimerMS:%d",
pSyncNode->vgId, pSyncNode->pRaftStore->currentTerm, pSyncNode->commitIndex, pSyncNode->state, pSyncNode->vgId, pSyncNode->pRaftStore->currentTerm, pSyncNode->commitIndex, pSyncNode->state,
syncUtilState2String(pSyncNode->state), pSyncNode->electTimerLogicClock, pSyncNode->electTimerLogicClockUser, syncUtilState2String(pSyncNode->state), pSyncNode->pRaftCfg->isStandBy, pSyncNode->electTimerLogicClock,
pSyncNode->electTimerMS); pSyncNode->electTimerLogicClockUser, pSyncNode->electTimerMS);
return s; return s;
} }
void syncNodeUpdateConfig(SSyncNode* pSyncNode, SSyncCfg* newConfig) { void syncNodeUpdateConfig(SSyncNode* pSyncNode, SSyncCfg* newConfig, bool* isDrop) {
bool hit = false;
for (int i = 0; i < newConfig->replicaNum; ++i) {
if (strcmp(pSyncNode->myNodeInfo.nodeFqdn, (newConfig->nodeInfo)[i].nodeFqdn) == 0 &&
pSyncNode->myNodeInfo.nodePort == (newConfig->nodeInfo)[i].nodePort) {
newConfig->myIndex = i;
hit = true;
break;
}
}
ASSERT(hit == true);
pSyncNode->pRaftCfg->cfg = *newConfig; pSyncNode->pRaftCfg->cfg = *newConfig;
int32_t ret = raftCfgPersist(pSyncNode->pRaftCfg); int32_t ret = 0;
ASSERT(ret == 0);
// init internal // init internal
pSyncNode->myNodeInfo = pSyncNode->pRaftCfg->cfg.nodeInfo[pSyncNode->pRaftCfg->cfg.myIndex]; pSyncNode->myNodeInfo = pSyncNode->pRaftCfg->cfg.nodeInfo[pSyncNode->pRaftCfg->cfg.myIndex];
@ -994,9 +1012,22 @@ void syncNodeUpdateConfig(SSyncNode* pSyncNode, SSyncCfg* newConfig) {
voteGrantedUpdate(pSyncNode->pVotesGranted, pSyncNode); voteGrantedUpdate(pSyncNode->pVotesGranted, pSyncNode);
votesRespondUpdate(pSyncNode->pVotesRespond, pSyncNode); votesRespondUpdate(pSyncNode->pVotesRespond, pSyncNode);
pSyncNode->pRaftCfg->isStandBy = 0; // isDrop
raftCfgPersist(pSyncNode->pRaftCfg); *isDrop = true;
for (int i = 0; i < newConfig->replicaNum; ++i) {
if (strcmp((newConfig->nodeInfo)[i].nodeFqdn, pSyncNode->myNodeInfo.nodeFqdn) == 0 &&
(newConfig->nodeInfo)[i].nodePort == pSyncNode->myNodeInfo.nodePort) {
*isDrop = false;
break;
}
}
if (!(*isDrop)) {
// change isStandBy to normal
pSyncNode->pRaftCfg->isStandBy = 0;
}
raftCfgPersist(pSyncNode->pRaftCfg);
syncNodeLog2("==syncNodeUpdateConfig==", pSyncNode); syncNodeLog2("==syncNodeUpdateConfig==", pSyncNode);
} }

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

@ -43,8 +43,9 @@ void CommitCb(struct SSyncFSM* pFsm, const SRpcMsg* pMsg, SFsmCbMeta cbMeta) {
if (cbMeta.index > beginIndex) { if (cbMeta.index > beginIndex) {
char logBuf[256]; char logBuf[256];
snprintf(logBuf, sizeof(logBuf), "==callback== ==CommitCb== pFsm:%p, index:%ld, isWeak:%d, code:%d, state:%d %s \n", snprintf(logBuf, sizeof(logBuf),
pFsm, cbMeta.index, cbMeta.isWeak, cbMeta.code, cbMeta.state, syncUtilState2String(cbMeta.state)); "==callback== ==CommitCb== pFsm:%p, index:%ld, isWeak:%d, code:%d, state:%d %s flag:%lu\n", pFsm,
cbMeta.index, cbMeta.isWeak, cbMeta.code, cbMeta.state, syncUtilState2String(cbMeta.state), cbMeta.flag);
syncRpcMsgLog2(logBuf, (SRpcMsg*)pMsg); syncRpcMsgLog2(logBuf, (SRpcMsg*)pMsg);
} else { } else {
sTrace("==callback== ==CommitCb== do not apply again %ld", cbMeta.index); sTrace("==callback== ==CommitCb== do not apply again %ld", cbMeta.index);
@ -54,15 +55,16 @@ void CommitCb(struct SSyncFSM* pFsm, const SRpcMsg* pMsg, SFsmCbMeta cbMeta) {
void PreCommitCb(struct SSyncFSM* pFsm, const SRpcMsg* pMsg, SFsmCbMeta cbMeta) { void PreCommitCb(struct SSyncFSM* pFsm, const SRpcMsg* pMsg, SFsmCbMeta cbMeta) {
char logBuf[256]; char logBuf[256];
snprintf(logBuf, sizeof(logBuf), snprintf(logBuf, sizeof(logBuf),
"==callback== ==PreCommitCb== pFsm:%p, index:%ld, isWeak:%d, code:%d, state:%d %s \n", pFsm, cbMeta.index, "==callback== ==PreCommitCb== pFsm:%p, index:%ld, isWeak:%d, code:%d, state:%d %s flag:%lu\n", pFsm,
cbMeta.isWeak, cbMeta.code, cbMeta.state, syncUtilState2String(cbMeta.state)); cbMeta.index, cbMeta.isWeak, cbMeta.code, cbMeta.state, syncUtilState2String(cbMeta.state), cbMeta.flag);
syncRpcMsgLog2(logBuf, (SRpcMsg*)pMsg); syncRpcMsgLog2(logBuf, (SRpcMsg*)pMsg);
} }
void RollBackCb(struct SSyncFSM* pFsm, const SRpcMsg* pMsg, SFsmCbMeta cbMeta) { void RollBackCb(struct SSyncFSM* pFsm, const SRpcMsg* pMsg, SFsmCbMeta cbMeta) {
char logBuf[256]; char logBuf[256];
snprintf(logBuf, sizeof(logBuf), "==callback== ==RollBackCb== pFsm:%p, index:%ld, isWeak:%d, code:%d, state:%d %s \n", snprintf(logBuf, sizeof(logBuf),
pFsm, cbMeta.index, cbMeta.isWeak, cbMeta.code, cbMeta.state, syncUtilState2String(cbMeta.state)); "==callback== ==RollBackCb== pFsm:%p, index:%ld, isWeak:%d, code:%d, state:%d %s flag:%lu\n", pFsm,
cbMeta.index, cbMeta.isWeak, cbMeta.code, cbMeta.state, syncUtilState2String(cbMeta.state), cbMeta.flag);
syncRpcMsgLog2(logBuf, (SRpcMsg*)pMsg); syncRpcMsgLog2(logBuf, (SRpcMsg*)pMsg);
} }
@ -75,13 +77,23 @@ int32_t GetSnapshotCb(struct SSyncFSM* pFsm, SSnapshot* pSnapshot) {
void RestoreFinishCb(struct SSyncFSM* pFsm) { sTrace("==callback== ==RestoreFinishCb=="); } void RestoreFinishCb(struct SSyncFSM* pFsm) { sTrace("==callback== ==RestoreFinishCb=="); }
void ReConfigCb(struct SSyncFSM* pFsm, SSyncCfg newCfg, SReConfigCbMeta cbMeta) {
sTrace("==callback== ==ReConfigCb== flag:0x%lX, isDrop:%d, index:%ld, code:%d, currentTerm:%lu, term:%lu", cbMeta.flag, cbMeta.isDrop, cbMeta.index, cbMeta.code, cbMeta.currentTerm, cbMeta.term);
}
SSyncFSM* createFsm() { SSyncFSM* createFsm() {
SSyncFSM* pFsm = (SSyncFSM*)taosMemoryMalloc(sizeof(SSyncFSM)); SSyncFSM* pFsm = (SSyncFSM*)taosMemoryMalloc(sizeof(SSyncFSM));
pFsm->FpCommitCb = CommitCb; pFsm->FpCommitCb = CommitCb;
pFsm->FpPreCommitCb = PreCommitCb; pFsm->FpPreCommitCb = PreCommitCb;
pFsm->FpRollBackCb = RollBackCb; pFsm->FpRollBackCb = RollBackCb;
pFsm->FpGetSnapshot = GetSnapshotCb; pFsm->FpGetSnapshot = GetSnapshotCb;
pFsm->FpRestoreFinishCb = RestoreFinishCb; pFsm->FpRestoreFinishCb = RestoreFinishCb;
pFsm->FpSnapshotApply = NULL;
pFsm->FpSnapshotRead = NULL;
pFsm->FpReConfigCb = ReConfigCb;
return pFsm; return pFsm;
} }
@ -109,6 +121,7 @@ int64_t createSyncNode(int32_t replicaNum, int32_t myIndex, int32_t vgId, SWal*
syncInfo.pFsm = createFsm(); syncInfo.pFsm = createFsm();
snprintf(syncInfo.path, sizeof(syncInfo.path), "%s_sync_replica%d_index%d", path, replicaNum, myIndex); snprintf(syncInfo.path, sizeof(syncInfo.path), "%s_sync_replica%d_index%d", path, replicaNum, myIndex);
syncInfo.pWal = pWal; syncInfo.pWal = pWal;
syncInfo.isStandBy = isStandBy;
SSyncCfg* pCfg = &syncInfo.syncCfg; SSyncCfg* pCfg = &syncInfo.syncCfg;
@ -180,7 +193,7 @@ SRpcMsg* createRpcMsg(int i, int count, int myIndex) {
int main(int argc, char** argv) { int main(int argc, char** argv) {
tsAsyncLog = 0; tsAsyncLog = 0;
sDebugFlag = DEBUG_TRACE + DEBUG_SCREEN + DEBUG_FILE; sDebugFlag = DEBUG_TRACE + DEBUG_SCREEN + DEBUG_FILE + DEBUG_INFO;
if (argc != 7) { if (argc != 7) {
usage(argv[0]); usage(argv[0]);
exit(-1); exit(-1);
@ -212,17 +225,21 @@ int main(int argc, char** argv) {
int64_t rid = createSyncNode(replicaNum, myIndex, gVgId, pWal, (char*)gDir, isStandBy); int64_t rid = createSyncNode(replicaNum, myIndex, gVgId, pWal, (char*)gDir, isStandBy);
assert(rid > 0); assert(rid > 0);
if (isStandBy) { syncStart(rid);
syncStartStandBy(rid);
} else { /*
syncStart(rid); if (isStandBy) {
} syncStartStandBy(rid);
} else {
syncStart(rid);
}
*/
SSyncNode* pSyncNode = (SSyncNode*)syncNodeAcquire(rid); SSyncNode* pSyncNode = (SSyncNode*)syncNodeAcquire(rid);
assert(pSyncNode != NULL); assert(pSyncNode != NULL);
if (isConfigChange) { if (isConfigChange) {
configChange(rid, 3, myIndex); configChange(rid, 2, myIndex);
} }
//--------------------------- //---------------------------

1
source/libs/transport/inc/transComm.h
View File

@ -318,6 +318,7 @@ void transDQDestroy(SDelayQueue* queue);
int transDQSched(SDelayQueue* queue, void (*func)(void* arg), void* arg, uint64_t timeoutMs); int transDQSched(SDelayQueue* queue, void (*func)(void* arg), void* arg, uint64_t timeoutMs);
void transPrintEpSet(SEpSet* pEpSet);
/* /*
* init global func * init global func
*/ */

12
source/libs/transport/src/transCli.c
View File

@ -921,16 +921,18 @@ int cliAppCb(SCliConn* pConn, STransMsg* pResp, SCliMsg* pMsg) {
STransConnCtx* pCtx = pMsg->ctx; STransConnCtx* pCtx = pMsg->ctx;
SEpSet* pEpSet = &pCtx->epSet; SEpSet* pEpSet = &pCtx->epSet;
transPrintEpSet(pEpSet);
/* /*
* upper layer handle retry if code equal TSDB_CODE_RPC_NETWORK_UNAVAIL * upper layer handle retry if code equal TSDB_CODE_RPC_NETWORK_UNAVAIL
*/ */
tmsg_t msgType = pCtx->msgType; tmsg_t msgType = pCtx->msgType;
if ((pTransInst->retry != NULL && (pTransInst->retry(pResp->code))) || if ((pTransInst->retry != NULL && (pTransInst->retry(pResp->code))) ||
((pResp->code == TSDB_CODE_RPC_NETWORK_UNAVAIL) && msgType == TDMT_MND_CONNECT)) { (pResp->code == TSDB_CODE_RPC_NETWORK_UNAVAIL)) {
pMsg->sent = 0; pMsg->sent = 0;
pMsg->st = taosGetTimestampUs(); pMsg->st = taosGetTimestampUs();
pCtx->retryCount += 1; pCtx->retryCount += 1;
if (msgType == TDMT_MND_CONNECT && pResp->code == TSDB_CODE_RPC_NETWORK_UNAVAIL) { if (pResp->code == TSDB_CODE_RPC_NETWORK_UNAVAIL) {
if (pCtx->retryCount < pEpSet->numOfEps) { if (pCtx->retryCount < pEpSet->numOfEps) {
pEpSet->inUse = (++pEpSet->inUse) % pEpSet->numOfEps; pEpSet->inUse = (++pEpSet->inUse) % pEpSet->numOfEps;
@ -972,7 +974,11 @@ int cliAppCb(SCliConn* pConn, STransMsg* pResp, SCliMsg* pMsg) {
pCtx->pRsp = NULL; pCtx->pRsp = NULL;
} else { } else {
tTrace("%s cli conn %p handle resp", pTransInst->label, pConn); tTrace("%s cli conn %p handle resp", pTransInst->label, pConn);
pTransInst->cfp(pTransInst->parent, pResp, pEpSet); if (pResp->code != 0) {
pTransInst->cfp(pTransInst->parent, pResp, NULL);
} else {
pTransInst->cfp(pTransInst->parent, pResp, pEpSet);
}
} }
return 0; return 0;
} }

12
source/libs/transport/src/transComm.c
View File

@ -446,4 +446,16 @@ int transDQSched(SDelayQueue* queue, void (*func)(void* arg), void* arg, uint64_
uv_timer_start(queue->timer, transDQTimeout, timeoutMs, 0); uv_timer_start(queue->timer, transDQTimeout, timeoutMs, 0);
return 0; return 0;
} }
void transPrintEpSet(SEpSet* pEpSet) {
if (pEpSet == NULL) {
tTrace("NULL epset");
return;
}
tTrace("epset begin: inUse: %d", pEpSet->inUse);
for (int i = 0; i < pEpSet->numOfEps; i++) {
tTrace("ip: %s, port: %d", pEpSet->eps[i].fqdn, pEpSet->eps[i].port);
}
tTrace("epset end");
}
#endif #endif

149
tests/script/tsim/mnode/basic1.sim
View File

@ -6,15 +6,6 @@ system sh/exec.sh -n dnode2 -s start
sql connect sql connect
print =============== show dnodes print =============== show dnodes
sql show dnodes;
if $rows != 1 then
return -1
endi
if $data00 != 1 then
return -1
endi
sql show mnodes; sql show mnodes;
if $rows != 1 then if $rows != 1 then
return -1 return -1
@ -30,63 +21,55 @@ endi
print =============== create dnodes print =============== create dnodes
sql create dnode $hostname port 7200 sql create dnode $hostname port 7200
sleep 2000 $x = 0
step1:
sql show dnodes; $x = $x + 1
if $rows != 2 then sleep 500
return -1 if $x == 20 then
return -1
endi
sql show dnodes -x step1
if $data(1)[4] != ready then
goto step1
endi endi
if $data(2)[4] != ready then
if $data00 != 1 then goto step1
return -1
endi
if $data10 != 2 then
return -1
endi
print $data02
if $data02 != 0 then
return -1
endi
if $data12 != 0 then
return -1
endi
if $data04 != ready then
return -1
endi
if $data14 != ready then
return -1
endi
sql show mnodes;
if $rows != 1 then
return -1
endi
if $data00 != 1 then
return -1
endi
if $data02 != LEADER then
return -1
endi endi
print =============== create drop mnode 1 print =============== create drop mnode 1
sql_error create mnode on dnode 1 sql_error create mnode on dnode 1
sql_error drop mnode on dnode 1 sql_error drop mnode on dnode 1
print =============== create drop mnode 2
sql create mnode on dnode 2 sql create mnode on dnode 2
$x = 0
step1:
$x = $x + 1
sleep 1000
if $x == 20 then
return -1
endi
sql show mnodes sql show mnodes
print $data(1)[0] $data(1)[1] $data(1)[2]
print $data(2)[0] $data(2)[1] $data(2)[2]
if $rows != 2 then if $rows != 2 then
return -1 return -1
endi endi
sql_error create mnode on dnode 2 if $data(1)[0] != 1 then
return -1
endi
if $data(1)[2] != LEADER then
return -1
endi
if $data(2)[0] != 2 then
return -1
endi
if $data(2)[2] != FOLLOWER then
goto step1
endi
sleep 2000
print ============ drop mnodes
sql drop mnode on dnode 2 sql drop mnode on dnode 2
sql show mnodes sql show mnodes
if $rows != 1 then if $rows != 1 then
@ -94,6 +77,35 @@ if $rows != 1 then
endi endi
sql_error drop mnode on dnode 2 sql_error drop mnode on dnode 2
$x = 0
step2:
$x = $x + 1
sleep 1000
if $x == 20 then
return -1
endi
sql show mnodes
print $data(1)[0] $data(1)[1] $data(1)[2]
print $data(2)[0] $data(2)[1] $data(2)[2]
if $rows != 2 then
return -1
endi
if $data(1)[0] != 1 then
return -1
endi
if $data(1)[2] != LEADER then
return -1
endi
if $data(2)[0] != NULL then
goto step2
endi
if $data(2)[2] != NULL then
goto step2
endi
sleep 2000
print =============== create drop mnodes print =============== create drop mnodes
sql create mnode on dnode 2 sql create mnode on dnode 2
sql show mnodes sql show mnodes
@ -101,17 +113,32 @@ if $rows != 2 then
return -1 return -1
endi endi
print =============== restart $x = 0
system sh/exec.sh -n dnode1 -s stop -x SIGINT step3:
system sh/exec.sh -n dnode2 -s stop -x SIGINT $x = $x + 1
system sh/exec.sh -n dnode1 -s start sleep 1000
system sh/exec.sh -n dnode2 -s start if $x == 20 then
return -1
sleep 2000 endi
sql show mnodes sql show mnodes
print $data(1)[0] $data(1)[1] $data(1)[2]
print $data(2)[0] $data(2)[1] $data(2)[2]
if $rows != 2 then if $rows != 2 then
return -1 return -1
endi endi
if $data(1)[0] != 1 then
return -1
endi
if $data(1)[2] != LEADER then
return -1
endi
if $data(2)[0] != 2 then
return -1
endi
if $data(2)[2] != FOLLOWER then
goto step3
endi
system sh/exec.sh -n dnode1 -s stop -x SIGINT system sh/exec.sh -n dnode1 -s stop -x SIGINT
system sh/exec.sh -n dnode2 -s stop -x SIGINT system sh/exec.sh -n dnode2 -s stop -x SIGINT

83
tests/script/tsim/mnode/basic2.sim
View File

@ -21,29 +21,31 @@ endi
print =============== create dnodes print =============== create dnodes
sql create dnode $hostname port 7200 sql create dnode $hostname port 7200
sql create dnode $hostname port 7300 $x = 0
sleep 2000 step1:
$x = $x + 1
sql show dnodes; sleep 500
if $rows != 3 then if $x == 20 then
return -1 return -1
endi
sql show dnodes -x step1
if $data(1)[4] != ready then
goto step1
endi endi
if $data(2)[4] != ready then
sql show mnodes; goto step1
if $rows != 1 then
return -1
endi
if $data00 != 1 then
return -1
endi
if $data02 != LEADER then
return -1
endi endi
print =============== create mnode 2 print =============== create mnode 2
sql create mnode on dnode 2 sql create mnode on dnode 2
$x = 0
step1:
$x = $x + 1
sleep 1000
if $x == 20 then
return -1
endi
sql show mnodes sql show mnodes
print $data(1)[0] $data(1)[1] $data(1)[2] print $data(1)[0] $data(1)[1] $data(1)[2]
print $data(2)[0] $data(2)[1] $data(2)[2] print $data(2)[0] $data(2)[1] $data(2)[2]
@ -60,8 +62,8 @@ endi
if $data(2)[0] != 2 then if $data(2)[0] != 2 then
return -1 return -1
endi endi
if $data(2)[2] == LEADER then if $data(2)[2] != FOLLOWER then
return -1 goto step1
endi endi
print =============== create user print =============== create user
@ -71,42 +73,47 @@ if $rows != 2 then
return -1 return -1
endi endi
#sql create database db sql create database db
#sql show databases sql show databases
#if $rows != 3 then if $rows != 3 then
# return -1 return -1
#endi endi
sleep 5000
print =============== restart
system sh/exec.sh -n dnode1 -s stop system sh/exec.sh -n dnode1 -s stop
system sh/exec.sh -n dnode2 -s stop system sh/exec.sh -n dnode2 -s stop
sleep 100
system sh/exec.sh -n dnode1 -s start system sh/exec.sh -n dnode1 -s start
system sh/exec.sh -n dnode2 -s start system sh/exec.sh -n dnode2 -s start
sql connect sql connect
sql show mnodes sql show mnodes
if $rows != 2 then if $rows != 2 then
return -1 return -1
endi endi
if $data(1)[0] != 1 then
return -1
endi
if $data(1)[2] != LEADER then
return -1
endi
sql show users sql show users
if $rows != 2 then if $rows != 2 then
return -1 return -1
endi endi
#sql show databases sql show databases
#if $rows != 3 then if $rows != 3 then
# return -1 return -1
#endi endi
return sql show dnodes
if $data(1)[4] != ready then
return -1
endi
if $data(2)[4] != ready then
return -1
endi
print =============== insert data
sql create table db.stb (ts timestamp, i int) tags (j int)
sql create table db.ctb using db.stb tags(1);
system sh/exec.sh -n dnode1 -s stop system sh/exec.sh -n dnode1 -s stop
system sh/exec.sh -n dnode2 -s stop system sh/exec.sh -n dnode2 -s stop

137
tests/script/tsim/mnode/basic3.sim Normal file
View File

@ -0,0 +1,137 @@
system sh/stop_dnodes.sh
system sh/deploy.sh -n dnode1 -i 1
system sh/deploy.sh -n dnode2 -i 2
system sh/deploy.sh -n dnode3 -i 3
system sh/exec.sh -n dnode1 -s start
system sh/exec.sh -n dnode2 -s start
system sh/exec.sh -n dnode3 -s start
sql connect
print =============== step1: create dnodes
sql create dnode $hostname port 7200
sql create dnode $hostname port 7300
$x = 0
step1:
$x = $x + 1
sleep 1000
if $x == 20 then
return -1
endi
sql show dnodes -x step1
if $data(1)[4] != ready then
goto step1
endi
if $data(2)[4] != ready then
goto step1
endi
if $data(3)[4] != ready then
goto step1
endi
print =============== step2: create mnode 2
sql create mnode on dnode 2
sql create mnode on dnode 3
$x = 0
step2:
$x = $x + 1
sleep 1000
if $x == 20 then
return -1
endi
sql show mnodes -x step2
if $data(1)[2] != LEADER then
goto step2
endi
if $data(2)[2] != FOLLOWER then
goto step2
endi
if $data(3)[2] != FOLLOWER then
goto step2
endi
print =============== step3: create user
sql create user user1 PASS 'user1'
sql show users
if $rows != 2 then
return -1
endi
# wait mnode2 mnode3 recv data finish
sleep 10000
print =============== step4: stop dnode1
system sh/exec.sh -n dnode1 -s stop
$x = 0
step4:
$x = $x + 1
sleep 1000
if $x == 20 then
return -1
endi
sql show mnodes -x step4
print $data(1)[0] $data(1)[1] $data(1)[2]
print $data(2)[0] $data(2)[1] $data(2)[2]
print $data(3)[0] $data(3)[1] $data(3)[2]
sql show users
if $rows != 2 then
return -1
endi
sleep 1000
sql show dnodes
if $data(2)[4] != ready then
return -1
endi
if $data(3)[4] != ready then
return -1
endi
print =============== step5: stop dnode1
system sh/exec.sh -n dnode1 -s start
system sh/exec.sh -n dnode2 -s stop
$x = 0
step5:
$x = $x + 1
sleep 1000
if $x == 20 then
return -1
endi
sql show mnodes -x step5
print $data(1)[0] $data(1)[1] $data(1)[2]
print $data(2)[0] $data(2)[1] $data(2)[2]
print $data(3)[0] $data(3)[1] $data(3)[2]
sql show users
if $rows != 2 then
return -1
endi
print =============== step6: stop dnode1
system sh/exec.sh -n dnode2 -s start
system sh/exec.sh -n dnode3 -s stop
$x = 0
step6:
$x = $x + 1
sleep 1000
if $x == 20 then
return -1
endi
sql show mnodes -x step6
print $data(1)[0] $data(1)[1] $data(1)[2]
print $data(2)[0] $data(2)[1] $data(2)[2]
print $data(3)[0] $data(3)[1] $data(3)[2]
sql show users
if $rows != 2 then
return -1
endi
system sh/exec.sh -n dnode1 -s stop
system sh/exec.sh -n dnode2 -s stop
system sh/exec.sh -n dnode3 -s stop