From eedd73323057d139268d5d86770177bc1c7f6fe2 Mon Sep 17 00:00:00 2001 From: Shengliang Guan Date: Tue, 26 Apr 2022 14:14:28 +0800 Subject: [PATCH] enh: add cb function for transaction --- source/dnode/mnode/impl/inc/mndDef.h | 4 +++ source/dnode/mnode/impl/inc/mndTrans.h | 1 + source/dnode/mnode/impl/src/mndTrans.c | 36 +++++++++++++++++--------- tests/script/sh/deploy.sh | 2 +- 4 files changed, 30 insertions(+), 13 deletions(-) diff --git a/source/dnode/mnode/impl/inc/mndDef.h b/source/dnode/mnode/impl/inc/mndDef.h index cf1cd58540..63dd9989c0 100644 --- a/source/dnode/mnode/impl/inc/mndDef.h +++ b/source/dnode/mnode/impl/inc/mndDef.h @@ -126,6 +126,8 @@ typedef enum { DND_REASON_OTHERS } EDndReason; +typedef void (*TransCbFp)(SMnode* pMnode, void* param); + typedef struct { int32_t id; ETrnStage stage; @@ -148,6 +150,8 @@ typedef struct { int64_t dbUid; char dbname[TSDB_DB_FNAME_LEN]; char lastError[TSDB_TRANS_ERROR_LEN]; + TransCbFp transCbFp; + void* transCbParam; } STrans; typedef struct { diff --git a/source/dnode/mnode/impl/inc/mndTrans.h b/source/dnode/mnode/impl/inc/mndTrans.h index 5c1b0991be..2fcc82e861 100644 --- a/source/dnode/mnode/impl/inc/mndTrans.h +++ b/source/dnode/mnode/impl/inc/mndTrans.h @@ -44,6 +44,7 @@ int32_t mndTransAppendCommitlog(STrans *pTrans, SSdbRaw *pRaw); int32_t mndTransAppendRedoAction(STrans *pTrans, STransAction *pAction); int32_t mndTransAppendUndoAction(STrans *pTrans, STransAction *pAction); void mndTransSetRpcRsp(STrans *pTrans, void *pCont, int32_t contLen); +void mndTransSetCb(STrans *pTrans, TransCbFp fp, void *param); void mndTransSetDbInfo(STrans *pTrans, SDbObj *pDb); int32_t mndTransPrepare(SMnode *pMnode, STrans *pTrans); diff --git a/source/dnode/mnode/impl/src/mndTrans.c b/source/dnode/mnode/impl/src/mndTrans.c index 9e9a8b56c9..6e221fb3df 100644 --- a/source/dnode/mnode/impl/src/mndTrans.c +++ b/source/dnode/mnode/impl/src/mndTrans.c @@ -193,9 +193,9 @@ TRANS_ENCODE_OVER: static SSdbRow *mndTransActionDecode(SSdbRaw *pRaw) { terrno = TSDB_CODE_OUT_OF_MEMORY; - SSdbRow * pRow = NULL; - STrans * pTrans = NULL; - char * pData = NULL; + SSdbRow *pRow = NULL; + STrans *pTrans = NULL; + char *pData = NULL; int32_t dataLen = 0; int8_t sver = 0; int32_t redoLogNum = 0; @@ -456,7 +456,7 @@ static int32_t mndTransActionUpdate(SSdb *pSdb, STrans *pOld, STrans *pNew) { } static STrans *mndAcquireTrans(SMnode *pMnode, int32_t transId) { - SSdb * pSdb = pMnode->pSdb; + SSdb *pSdb = pMnode->pSdb; STrans *pTrans = sdbAcquire(pSdb, SDB_TRANS, &transId); if (pTrans == NULL) { terrno = TSDB_CODE_MND_TRANS_NOT_EXIST; @@ -574,6 +574,11 @@ void mndTransSetRpcRsp(STrans *pTrans, void *pCont, int32_t contLen) { pTrans->rpcRspLen = contLen; } +void mndTransSetCb(STrans *pTrans, TransCbFp fp, void *param) { + pTrans->transCbFp = fp; + pTrans->transCbParam = param; +} + void mndTransSetDbInfo(STrans *pTrans, SDbObj *pDb) { pTrans->dbUid = pDb->uid; memcpy(pTrans->dbname, pDb->name, TSDB_DB_FNAME_LEN); @@ -626,7 +631,7 @@ static int32_t mndCheckTransCanBeStartedInParallel(SMnode *pMnode, STrans *pNewT if (mndIsBasicTrans(pNewTrans)) return 0; STrans *pTrans = NULL; - void * pIter = NULL; + void *pIter = NULL; int32_t code = 0; while (1) { @@ -707,6 +712,8 @@ int32_t mndTransPrepare(SMnode *pMnode, STrans *pTrans) { pNew->rpcRefId = pTrans->rpcRefId; pNew->rpcRsp = pTrans->rpcRsp; pNew->rpcRspLen = pTrans->rpcRspLen; + pNew->transCbFp = pTrans->transCbFp; + pNew->transCbParam = pTrans->transCbParam; pTrans->rpcRsp = NULL; pTrans->rpcRspLen = 0; @@ -830,7 +837,7 @@ HANDLE_ACTION_RSP_OVER: } static int32_t mndTransExecuteLogs(SMnode *pMnode, SArray *pArray) { - SSdb * pSdb = pMnode->pSdb; + SSdb *pSdb = pMnode->pSdb; int32_t arraySize = taosArrayGetSize(pArray); if (arraySize == 0) return 0; @@ -1117,6 +1124,11 @@ static bool mndTransPerfromFinishedStage(SMnode *pMnode, STrans *pTrans) { } mDebug("trans:%d, finished, code:0x%04x, failedTimes:%d", pTrans->id, pTrans->code, pTrans->failedTimes); + + if (pTrans->transCbFp != NULL) { + (*pTrans->transCbFp)(pMnode, pTrans->transCbParam); + } + return continueExec; } @@ -1205,11 +1217,11 @@ static int32_t mndKillTrans(SMnode *pMnode, STrans *pTrans) { } static int32_t mndProcessKillTransReq(SNodeMsg *pReq) { - SMnode * pMnode = pReq->pNode; + SMnode *pMnode = pReq->pNode; SKillTransReq killReq = {0}; int32_t code = -1; - SUserObj * pUser = NULL; - STrans * pTrans = NULL; + SUserObj *pUser = NULL; + STrans *pTrans = NULL; if (tDeserializeSKillTransReq(pReq->rpcMsg.pCont, pReq->rpcMsg.contLen, &killReq) != 0) { terrno = TSDB_CODE_INVALID_MSG; @@ -1249,7 +1261,7 @@ KILL_OVER: void mndTransPullup(SMnode *pMnode) { STrans *pTrans = NULL; - void * pIter = NULL; + void *pIter = NULL; while (1) { pIter = sdbFetch(pMnode->pSdb, SDB_TRANS, pIter, (void **)&pTrans); @@ -1264,11 +1276,11 @@ void mndTransPullup(SMnode *pMnode) { static int32_t mndRetrieveTrans(SNodeMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows) { SMnode *pMnode = pReq->pNode; - SSdb * pSdb = pMnode->pSdb; + SSdb *pSdb = pMnode->pSdb; int32_t numOfRows = 0; STrans *pTrans = NULL; int32_t cols = 0; - char * pWrite; + char *pWrite; while (numOfRows < rows) { pShow->pIter = sdbFetch(pSdb, SDB_TRANS, pShow->pIter, (void **)&pTrans); diff --git a/tests/script/sh/deploy.sh b/tests/script/sh/deploy.sh index b8abd4ff10..ec847dedbb 100755 --- a/tests/script/sh/deploy.sh +++ b/tests/script/sh/deploy.sh @@ -135,7 +135,7 @@ echo "qDebugFlag 143" >> $TAOS_CFG echo "rpcDebugFlag 143" >> $TAOS_CFG echo "tmrDebugFlag 131" >> $TAOS_CFG echo "uDebugFlag 143" >> $TAOS_CFG -echo "sDebugFlag 143" >> $TAOS_CFG +echo "sDebugFlag 135" >> $TAOS_CFG echo "wDebugFlag 143" >> $TAOS_CFG echo "numOfLogLines 20000000" >> $TAOS_CFG echo "statusInterval 1" >> $TAOS_CFG -- GitLab