From ff91282bee95865907a8cd71f635c3a1f92fae9a Mon Sep 17 00:00:00 2001 From: Shengliang Guan Date: Fri, 20 May 2022 10:51:53 +0800 Subject: [PATCH] refactor: adjust vnode sync --- source/dnode/vnode/src/vnd/vnodeOpen.c | 1 - source/dnode/vnode/src/vnd/vnodeSync.c | 91 ++++++++++++-------------- 2 files changed, 41 insertions(+), 51 deletions(-) diff --git a/source/dnode/vnode/src/vnd/vnodeOpen.c b/source/dnode/vnode/src/vnd/vnodeOpen.c index ef86ac86e4..f0af677641 100644 --- a/source/dnode/vnode/src/vnd/vnodeOpen.c +++ b/source/dnode/vnode/src/vnd/vnodeOpen.c @@ -180,7 +180,6 @@ void vnodeClose(SVnode *pVnode) { // start the sync timer after the queue is ready int32_t vnodeStart(SVnode *pVnode) { - vnodeSyncSetMsgCb(pVnode); vnodeSyncStart(pVnode); return 0; } diff --git a/source/dnode/vnode/src/vnd/vnodeSync.c b/source/dnode/vnode/src/vnd/vnodeSync.c index bcef95baff..8659c41807 100644 --- a/source/dnode/vnode/src/vnd/vnodeSync.c +++ b/source/dnode/vnode/src/vnd/vnodeSync.c @@ -13,71 +13,62 @@ * along with this program. If not, see . */ +#define _DEFAULT_SOURCE #include "vnd.h" -int32_t vnodeSyncOpen(SVnode *pVnode, char *path) { - SSyncInfo syncInfo; - syncInfo.vgId = pVnode->config.vgId; - SSyncCfg *pCfg = &(syncInfo.syncCfg); - pCfg->replicaNum = pVnode->config.syncCfg.replicaNum; - pCfg->myIndex = pVnode->config.syncCfg.myIndex; - memcpy(pCfg->nodeInfo, pVnode->config.syncCfg.nodeInfo, sizeof(pCfg->nodeInfo)); - - snprintf(syncInfo.path, sizeof(syncInfo.path), "%s/sync", path); - syncInfo.pWal = pVnode->pWal; +static int32_t vnodeSyncEqMsg(const SMsgCb *msgcb, SRpcMsg *pMsg); +static int32_t vnodeSyncSendMsg(const SEpSet *pEpSet, SRpcMsg *pMsg); +static SSyncFSM *vnodeSyncMakeFsm(SVnode *pVnode); +static void vnodeSyncCommitMsg(SSyncFSM *pFsm, const SRpcMsg *pMsg, SFsmCbMeta cbMeta); +static void vnodeSyncPreCommitMsg(SSyncFSM *pFsm, const SRpcMsg *pMsg, SFsmCbMeta cbMeta); +static void vnodeSyncRollBackMsg(SSyncFSM *pFsm, const SRpcMsg *pMsg, SFsmCbMeta cbMeta); +static int32_t vnodeSyncGetSnapshot(SSyncFSM *pFsm, SSnapshot *pSnapshot); - syncInfo.pFsm = syncVnodeMakeFsm(pVnode); - syncInfo.msgcb = NULL; - syncInfo.FpSendMsg = vnodeSyncSendMsg; - syncInfo.FpEqMsg = vnodeSyncEqMsg; +int32_t vnodeSyncOpen(SVnode *pVnode, char *path) { + SSyncInfo syncInfo = { + .vgId = pVnode->config.vgId, + .syncCfg = pVnode->config.syncCfg, + .pWal = pVnode->pWal, + .msgcb = NULL, + .FpSendMsg = vnodeSyncSendMsg, + .FpEqMsg = vnodeSyncEqMsg, + }; + + snprintf(syncInfo.path, sizeof(syncInfo.path), "%s%ssync", path, TD_DIRSEP); + syncInfo.pFsm = vnodeSyncMakeFsm(pVnode); pVnode->sync = syncOpen(&syncInfo); - assert(pVnode->sync > 0); + if (pVnode->sync <= 0) { + vError("vgId:%d, failed to open sync since %s", pVnode->config.vgId, terrstr()); + return -1; + } - // for test setPingTimerMS(pVnode->sync, 3000); setElectTimerMS(pVnode->sync, 500); setHeartbeatTimerMS(pVnode->sync, 100); - return 0; } -int32_t vnodeSyncStart(SVnode *pVnode) { +void vnodeSyncStart(SVnode *pVnode) { + syncSetMsgCb(pVnode->sync, &pVnode->msgCb); syncStart(pVnode->sync); - return 0; } -void vnodeSyncClose(SVnode *pVnode) { - // stop by ref id - syncStop(pVnode->sync); -} - -void vnodeSyncSetMsgCb(SVnode *pVnode) { syncSetMsgCb(pVnode->sync, &pVnode->msgCb); } +void vnodeSyncClose(SVnode *pVnode) { syncStop(pVnode->sync); } int32_t vnodeSyncEqMsg(const SMsgCb *msgcb, SRpcMsg *pMsg) { return tmsgPutToQueue(msgcb, SYNC_QUEUE, pMsg); } -int32_t vnodeSyncSendMsg(const SEpSet *pEpSet, SRpcMsg *pMsg) { - pMsg->info.noResp = 1; - return tmsgSendReq(pEpSet, pMsg); -} - -int32_t vnodeSyncGetSnapshotCb(struct SSyncFSM *pFsm, SSnapshot *pSnapshot) { - SVnode *pVnode = (SVnode *)(pFsm->data); - vnodeGetSnapshot(pVnode, pSnapshot); - - /* - pSnapshot->data = NULL; - pSnapshot->lastApplyIndex = 0; - pSnapshot->lastApplyTerm = 0; - */ +int32_t vnodeSyncSendMsg(const SEpSet *pEpSet, SRpcMsg *pMsg) { return tmsgSendReq(pEpSet, pMsg); } +int32_t vnodeSyncGetSnapshot(SSyncFSM *pFsm, SSnapshot *pSnapshot) { + vnodeGetSnapshot(pFsm->data, pSnapshot); return 0; } -void vnodeSyncCommitCb(struct SSyncFSM *pFsm, const SRpcMsg *pMsg, SFsmCbMeta cbMeta) { +void vnodeSyncCommitMsg(SSyncFSM *pFsm, const SRpcMsg *pMsg, SFsmCbMeta cbMeta) { SyncIndex beginIndex = SYNC_INDEX_INVALID; if (pFsm->FpGetSnapshot != NULL) { - SSnapshot snapshot; + SSnapshot snapshot = {0}; pFsm->FpGetSnapshot(pFsm, &snapshot); beginIndex = snapshot.lastApplyIndex; } @@ -128,7 +119,7 @@ void vnodeSyncCommitCb(struct SSyncFSM *pFsm, const SRpcMsg *pMsg, SFsmCbMeta cb } } -void vnodeSyncPreCommitCb(struct SSyncFSM *pFsm, const SRpcMsg *pMsg, SFsmCbMeta cbMeta) { +void vnodeSyncPreCommitMsg(SSyncFSM *pFsm, const SRpcMsg *pMsg, SFsmCbMeta cbMeta) { char logBuf[256]; snprintf(logBuf, sizeof(logBuf), "==callback== ==PreCommitCb== pFsm:%p, index:%ld, isWeak:%d, code:%d, state:%d %s \n", pFsm, cbMeta.index, @@ -136,19 +127,19 @@ void vnodeSyncPreCommitCb(struct SSyncFSM *pFsm, const SRpcMsg *pMsg, SFsmCbMeta syncRpcMsgLog2(logBuf, (SRpcMsg *)pMsg); } -void vnodeSyncRollBackCb(struct SSyncFSM *pFsm, const SRpcMsg *pMsg, SFsmCbMeta cbMeta) { +void vnodeSyncRollBackMsg(SSyncFSM *pFsm, const SRpcMsg *pMsg, SFsmCbMeta cbMeta) { char logBuf[256]; snprintf(logBuf, sizeof(logBuf), "==callback== ==RollBackCb== pFsm:%p, index:%ld, isWeak:%d, code:%d, state:%d %s \n", pFsm, cbMeta.index, cbMeta.isWeak, cbMeta.code, cbMeta.state, syncUtilState2String(cbMeta.state)); syncRpcMsgLog2(logBuf, (SRpcMsg *)pMsg); } -SSyncFSM *syncVnodeMakeFsm(SVnode *pVnode) { - SSyncFSM *pFsm = (SSyncFSM *)taosMemoryMalloc(sizeof(SSyncFSM)); +SSyncFSM *vnodeSyncMakeFsm(SVnode *pVnode) { + SSyncFSM *pFsm = taosMemoryCalloc(1, sizeof(SSyncFSM)); pFsm->data = pVnode; - pFsm->FpCommitCb = vnodeSyncCommitCb; - pFsm->FpPreCommitCb = vnodeSyncPreCommitCb; - pFsm->FpRollBackCb = vnodeSyncRollBackCb; - pFsm->FpGetSnapshot = vnodeSyncGetSnapshotCb; + pFsm->FpCommitCb = vnodeSyncCommitMsg; + pFsm->FpPreCommitCb = vnodeSyncPreCommitMsg; + pFsm->FpRollBackCb = vnodeSyncRollBackMsg; + pFsm->FpGetSnapshot = vnodeSyncGetSnapshot; return pFsm; -} +} \ No newline at end of file -- GitLab