diff --git a/source/dnode/mnode/impl/src/mndSync.c b/source/dnode/mnode/impl/src/mndSync.c index ac95dd2795b5fb90b4cd1cf657aa151850035af8..cd6fe380e13bfbcdc7af6e58c0d5adb5ed95c337 100644 --- a/source/dnode/mnode/impl/src/mndSync.c +++ b/source/dnode/mnode/impl/src/mndSync.c @@ -71,8 +71,8 @@ void mndSyncCommitMsg(struct SSyncFSM *pFsm, const SRpcMsg *pMsg, SFsmCbMeta cbM mInfo("trans:%d, is proposed and post sem", transId); } pMgmt->transId = 0; - taosWUnLockLatch(&pMgmt->lock); tsem_post(&pMgmt->syncSem); + taosWUnLockLatch(&pMgmt->lock); } else { taosWUnLockLatch(&pMgmt->lock); STrans *pTrans = mndAcquireTrans(pMnode, transId); @@ -113,27 +113,7 @@ void mndRestoreFinish(struct SSyncFSM *pFsm) { } } -void mndReConfig(struct SSyncFSM *pFsm, const SRpcMsg *pMsg, SReConfigCbMeta *cbMeta) { - SMnode *pMnode = pFsm->data; - SSyncMgmt *pMgmt = &pMnode->syncMgmt; - - pMgmt->errCode = cbMeta->code; - mInfo("trans:-1, sync reconfig is proposed, saved:%d code:0x%x, index:%" PRId64 " term:%" PRId64, pMgmt->transId, - cbMeta->code, cbMeta->index, cbMeta->term); - - taosWLockLatch(&pMgmt->lock); - if (pMgmt->transId == -1) { - if (pMgmt->errCode != 0) { - mError("trans:-1, failed to propose sync reconfig since %s, post sem", tstrerror(pMgmt->errCode)); - } else { - mInfo("trans:-1, sync reconfig is proposed, saved:%d code:0x%x, index:%" PRId64 " term:%" PRId64 " post sem", - pMgmt->transId, cbMeta->code, cbMeta->index, cbMeta->term); - } - pMgmt->transId = 0; - tsem_post(&pMgmt->syncSem); - } - taosWUnLockLatch(&pMgmt->lock); -} +void mndReConfig(struct SSyncFSM *pFsm, const SRpcMsg *pMsg, SReConfigCbMeta *cbMeta) {} int32_t mndSnapshotStartRead(struct SSyncFSM *pFsm, void *pParam, void **ppReader) { mInfo("start to read snapshot from sdb"); @@ -179,11 +159,14 @@ void mndLeaderTransfer(struct SSyncFSM *pFsm, const SRpcMsg *pMsg, SFsmCbMeta cb static void mndBecomeFollower(struct SSyncFSM *pFsm) { SMnode *pMnode = pFsm->data; - mInfo("vgId:1, become follower and post sem"); + mInfo("vgId:1, become follower"); taosWLockLatch(&pMnode->syncMgmt.lock); if (pMnode->syncMgmt.transId != 0) { + mInfo("vgId:1, become follower and post sem, trans:%d, failed to propose since not leader", + pMnode->syncMgmt.transId); pMnode->syncMgmt.transId = 0; + pMnode->syncMgmt.errCode = TSDB_CODE_SYN_NOT_LEADER; tsem_post(&pMnode->syncMgmt.syncSem); } taosWUnLockLatch(&pMnode->syncMgmt.lock); @@ -292,6 +275,7 @@ int32_t mndSyncPropose(SMnode *pMnode, SSdbRaw *pRaw, int32_t transId) { int32_t code = syncPropose(pMgmt->sync, &req, isWeak); if (code == 0) { + mInfo("trans:%d, is proposing and wait sem", pMgmt->transId); tsem_wait(&pMgmt->syncSem); } else if (code > 0) { mInfo("trans:%d, confirm at once since replica is 1, continue execute", transId); @@ -301,12 +285,16 @@ int32_t mndSyncPropose(SMnode *pMnode, SSdbRaw *pRaw, int32_t transId) { sdbWriteWithoutFree(pMnode->pSdb, pRaw); sdbSetApplyInfo(pMnode->pSdb, req.info.conn.applyIndex, req.info.conn.applyTerm, SYNC_INDEX_INVALID); code = 0; - } else if (code == -1 && terrno == TSDB_CODE_SYN_NOT_LEADER) { - terrno = TSDB_CODE_APP_NOT_READY; - } else if (code == -1 && terrno == TSDB_CODE_SYN_INTERNAL_ERROR) { - terrno = TSDB_CODE_SYN_INTERNAL_ERROR; } else { - terrno = TSDB_CODE_APP_ERROR; + taosWLockLatch(&pMgmt->lock); + mInfo("trans:%d, failed to proposed since %s", transId, terrstr()); + pMgmt->transId = 0; + taosWUnLockLatch(&pMgmt->lock); + if (terrno == TSDB_CODE_SYN_NOT_LEADER) { + terrno = TSDB_CODE_APP_NOT_READY; + } else { + terrno = TSDB_CODE_APP_ERROR; + } } rpcFreeCont(req.pCont); @@ -315,6 +303,7 @@ int32_t mndSyncPropose(SMnode *pMnode, SSdbRaw *pRaw, int32_t transId) { return code; } + if (pMgmt->errCode != 0) terrno = pMgmt->errCode; return pMgmt->errCode; } @@ -328,6 +317,7 @@ void mndSyncStart(SMnode *pMnode) { void mndSyncStop(SMnode *pMnode) { taosWLockLatch(&pMnode->syncMgmt.lock); if (pMnode->syncMgmt.transId != 0) { + mInfo("vgId:1, is stopped and post sem, trans:%d", pMnode->syncMgmt.transId); pMnode->syncMgmt.transId = 0; tsem_post(&pMnode->syncMgmt.syncSem); } diff --git a/source/dnode/mnode/impl/src/mndTrans.c b/source/dnode/mnode/impl/src/mndTrans.c index a64b07fdbd6491bb16b7854758029bde06b14808..92e9d755aa62855e5ce2d27cac7a91b367ac288e 100644 --- a/source/dnode/mnode/impl/src/mndTrans.c +++ b/source/dnode/mnode/impl/src/mndTrans.c @@ -778,7 +778,7 @@ static int32_t mndTransSync(SMnode *pMnode, STrans *pTrans) { mInfo("trans:%d, sync to other mnodes, stage:%s", pTrans->id, mndTransStr(pTrans->stage)); int32_t code = mndSyncPropose(pMnode, pRaw, pTrans->id); if (code != 0) { - mError("trans:%d, failed to sync since %s", pTrans->id, terrstr()); + mError("trans:%d, failed to sync, errno:%s code:%s", pTrans->id, terrstr(), tstrerror(code)); sdbFreeRaw(pRaw); return -1; } diff --git a/source/libs/sync/src/syncMain.c b/source/libs/sync/src/syncMain.c index b30717a1effb33cf8c057efdd780fa97168528c4..ba4ce8def37e38876db6a3b633a662d965b6230f 100644 --- a/source/libs/sync/src/syncMain.c +++ b/source/libs/sync/src/syncMain.c @@ -1115,6 +1115,9 @@ SSyncNode* syncNodeOpen(SSyncInfo* pOldSyncInfo) { sError("failed to create raft cfg file. configPath: %s", pSyncNode->configPath); goto _error; } + if (pSyncInfo->syncCfg.replicaNum == 0) { + pSyncInfo->syncCfg = pSyncNode->pRaftCfg->cfg; + } } else { // update syncCfg by raft_config.json pSyncNode->pRaftCfg = raftCfgOpen(pSyncNode->configPath); @@ -2047,8 +2050,27 @@ inline bool syncNodeInConfig(SSyncNode* pSyncNode, const SSyncCfg* config) { return b1; } +static bool syncIsConfigChanged(const SSyncCfg* pOldCfg, const SSyncCfg* pNewCfg) { + if (pOldCfg->replicaNum != pNewCfg->replicaNum) return true; + if (pOldCfg->myIndex != pNewCfg->myIndex) return true; + for (int32_t i = 0; i < pOldCfg->replicaNum; ++i) { + const SNodeInfo* pOldInfo = &pOldCfg->nodeInfo[i]; + const SNodeInfo* pNewInfo = &pNewCfg->nodeInfo[i]; + if (strcmp(pOldInfo->nodeFqdn, pNewInfo->nodeFqdn) != 0) return true; + if (pOldInfo->nodePort != pNewInfo->nodePort) return true; + } + + return false; +} + void syncNodeDoConfigChange(SSyncNode* pSyncNode, SSyncCfg* pNewConfig, SyncIndex lastConfigChangeIndex) { SSyncCfg oldConfig = pSyncNode->pRaftCfg->cfg; +#if 0 + if (!syncIsConfigChanged(&oldConfig, pNewConfig)) { + sInfo("vgId:1, sync not reconfig since not changed"); + return; + } +#endif pSyncNode->pRaftCfg->cfg = *pNewConfig; pSyncNode->pRaftCfg->lastConfigIndex = lastConfigChangeIndex;