diff --git a/include/libs/sync/sync.h b/include/libs/sync/sync.h index 4741f7f3d8227c28b1979ca6925c52ee02863a4e..1fa862b145f4e4c97b1dc28a96bda8121fd97bdc 100644 --- a/include/libs/sync/sync.h +++ b/include/libs/sync/sync.h @@ -138,8 +138,8 @@ typedef struct SSnapshotMeta { typedef struct SSyncFSM { void* data; - void (*FpCommitCb)(const struct SSyncFSM* pFsm, SRpcMsg* pMsg, const SFsmCbMeta* pMeta); - void (*FpPreCommitCb)(const struct SSyncFSM* pFsm, SRpcMsg* pMsg, const SFsmCbMeta* pMeta); + int32_t (*FpCommitCb)(const struct SSyncFSM* pFsm, SRpcMsg* pMsg, const SFsmCbMeta* pMeta); + int32_t (*FpPreCommitCb)(const struct SSyncFSM* pFsm, SRpcMsg* pMsg, const SFsmCbMeta* pMeta); void (*FpRollBackCb)(const struct SSyncFSM* pFsm, SRpcMsg* pMsg, const SFsmCbMeta* pMeta); void (*FpRestoreFinishCb)(const struct SSyncFSM* pFsm); diff --git a/source/dnode/mnode/impl/src/mndSync.c b/source/dnode/mnode/impl/src/mndSync.c index 10cd6416b44d1e47528ed0911e9707b546599f36..2e3c04eb786f39601f9f82ba0f87e483b3f8da07 100644 --- a/source/dnode/mnode/impl/src/mndSync.c +++ b/source/dnode/mnode/impl/src/mndSync.c @@ -72,7 +72,7 @@ static int32_t mndSyncSendMsg(const SEpSet *pEpSet, SRpcMsg *pMsg) { return code; } -void mndProcessWriteMsg(const SSyncFSM *pFsm, SRpcMsg *pMsg, const SFsmCbMeta *pMeta) { +int32_t mndProcessWriteMsg(const SSyncFSM *pFsm, SRpcMsg *pMsg, const SFsmCbMeta *pMeta) { SMnode *pMnode = pFsm->data; SSyncMgmt *pMgmt = &pMnode->syncMgmt; SSdbRaw *pRaw = pMsg->pCont; @@ -114,12 +114,15 @@ void mndProcessWriteMsg(const SSyncFSM *pFsm, SRpcMsg *pMsg, const SFsmCbMeta *p mError("trans:%d, not found while execute in mnode since %s", transId, terrstr()); } } + + return 0; } -void mndSyncCommitMsg(const SSyncFSM *pFsm, SRpcMsg *pMsg, const SFsmCbMeta *pMeta) { - mndProcessWriteMsg(pFsm, pMsg, pMeta); +int32_t mndSyncCommitMsg(const SSyncFSM *pFsm, SRpcMsg *pMsg, const SFsmCbMeta *pMeta) { + int32_t code = mndProcessWriteMsg(pFsm, pMsg, pMeta); rpcFreeCont(pMsg->pCont); pMsg->pCont = NULL; + return code; } int32_t mndSyncGetSnapshot(const SSyncFSM *pFsm, SSnapshot *pSnapshot, void *pReaderParam, void **ppReader) { diff --git a/source/dnode/vnode/src/vnd/vnodeSync.c b/source/dnode/vnode/src/vnd/vnodeSync.c index bd8b00edf0472b424198f42d8101d0f93403df68..55ba0d3e6f4a297fc4e0578af424a50ecf5aa47b 100644 --- a/source/dnode/vnode/src/vnd/vnodeSync.c +++ b/source/dnode/vnode/src/vnd/vnodeSync.c @@ -295,7 +295,7 @@ static int32_t vnodeSyncGetSnapshot(const SSyncFSM *pFsm, SSnapshot *pSnapshot) return 0; } -static void vnodeSyncApplyMsg(const SSyncFSM *pFsm, SRpcMsg *pMsg, const SFsmCbMeta *pMeta) { +static int32_t vnodeSyncApplyMsg(const SSyncFSM *pFsm, SRpcMsg *pMsg, const SFsmCbMeta *pMeta) { SVnode *pVnode = pFsm->data; pMsg->info.conn.applyIndex = pMeta->index; pMsg->info.conn.applyTerm = pMeta->term; @@ -306,17 +306,18 @@ static void vnodeSyncApplyMsg(const SSyncFSM *pFsm, SRpcMsg *pMsg, const SFsmCbM pVnode->config.vgId, pFsm, pMeta->index, pMeta->term, pMsg->info.conn.applyIndex, pMeta->isWeak, pMeta->code, pMeta->state, syncStr(pMeta->state), TMSG_INFO(pMsg->msgType)); - tmsgPutToQueue(&pVnode->msgCb, APPLY_QUEUE, pMsg); + return tmsgPutToQueue(&pVnode->msgCb, APPLY_QUEUE, pMsg); } -static void vnodeSyncCommitMsg(const SSyncFSM *pFsm, SRpcMsg *pMsg, const SFsmCbMeta *pMeta) { - vnodeSyncApplyMsg(pFsm, pMsg, pMeta); +static int32_t vnodeSyncCommitMsg(const SSyncFSM *pFsm, SRpcMsg *pMsg, const SFsmCbMeta *pMeta) { + return vnodeSyncApplyMsg(pFsm, pMsg, pMeta); } -static void vnodeSyncPreCommitMsg(const SSyncFSM *pFsm, SRpcMsg *pMsg, const SFsmCbMeta *pMeta) { +static int32_t vnodeSyncPreCommitMsg(const SSyncFSM *pFsm, SRpcMsg *pMsg, const SFsmCbMeta *pMeta) { if (pMeta->isWeak == 1) { - vnodeSyncApplyMsg(pFsm, pMsg, pMeta); + return vnodeSyncApplyMsg(pFsm, pMsg, pMeta); } + return 0; } static void vnodeSyncRollBackMsg(const SSyncFSM *pFsm, SRpcMsg *pMsg, const SFsmCbMeta *pMeta) { diff --git a/source/libs/sync/src/syncPipeline.c b/source/libs/sync/src/syncPipeline.c index fcb0d6320954f4aba254ffff898f01cfbfb10f8f..d5d77274a9528d04aa7c16ec32efcc1d837d53c9 100644 --- a/source/libs/sync/src/syncPipeline.c +++ b/source/libs/sync/src/syncPipeline.c @@ -457,9 +457,9 @@ int32_t syncLogFsmExecute(SSyncNode* pNode, SSyncFSM* pFsm, ESyncState role, Syn cbMeta.flag = -1; (void)syncRespMgrGetAndDel(pNode->pSyncRespMgr, cbMeta.seqNum, &rpcMsg.info); - pFsm->FpCommitCb(pFsm, &rpcMsg, &cbMeta); + int32_t code = pFsm->FpCommitCb(pFsm, &rpcMsg, &cbMeta); ASSERT(rpcMsg.pCont == NULL); - return 0; + return code; } int32_t syncLogBufferValidate(SSyncLogBuffer* pBuf) { @@ -516,8 +516,9 @@ int32_t syncLogBufferCommit(SSyncLogBuffer* pBuf, SSyncNode* pNode, int64_t comm } if (syncLogFsmExecute(pNode, pFsm, role, term, pEntry) != 0) { - sError("vgId:%d, failed to execute sync log entry in FSM. log index:%" PRId64 ", term:%" PRId64 "", vgId, - pEntry->index, pEntry->term); + sError("vgId:%d, failed to commit sync log entry. index:%" PRId64 ", term:%" PRId64 + ", role: %d, current term: %" PRId64, + vgId, pEntry->index, pEntry->term, role, term); goto _out; } pBuf->commitIndex = index;