Merge pull request #21136 from taosdata/fix/transaction-multithread

transaction multithread

source/dnode/mnode/impl/inc/mndDef.h

@@ -177,6 +177,7 @@ typedef struct {
   SArray*     pRpcArray;
   SRWLatch    lockRpcArray;
   int64_t     mTraceId;
+  TdThreadMutex    mutex;
 } STrans;
 
 typedef struct {

source/dnode/mnode/impl/src/mndTrans.c

@@ -546,6 +546,7 @@ static void mndTransDropData(STrans *pTrans) {
     pTrans->param = NULL;
     pTrans->paramLen = 0;
   }
+  (void)taosThreadMutexDestroy(&pTrans->mutex);
 }
 
 static int32_t mndTransActionDelete(SSdb *pSdb, STrans *pTrans, bool callFunc) {
@@ -651,6 +652,7 @@ STrans *mndTransCreate(SMnode *pMnode, ETrnPolicy policy, ETrnConflct conflict,
   pTrans->pRpcArray = taosArrayInit(1, sizeof(SRpcHandleInfo));
   pTrans->mTraceId = pReq ? TRACE_GET_ROOTID(&pReq->info.traceId) : 0;
   taosInitRWLatch(&pTrans->lockRpcArray);
+  taosThreadMutexInit(&pTrans->mutex, NULL);
 
   if (pTrans->redoActions == NULL || pTrans->undoActions == NULL || pTrans->commitActions == NULL ||
       pTrans->pRpcArray == NULL) {
@@ -1309,7 +1311,13 @@ static int32_t mndTransExecuteRedoActionsSerial(SMnode *pMnode, STrans *pTrans)
   int32_t code = 0;
   int32_t numOfActions = taosArrayGetSize(pTrans->redoActions);
   if (numOfActions == 0) return code;
-  if (pTrans->redoActionPos >= numOfActions) return code;
+
+  taosThreadMutexLock(&pTrans->mutex);
+
+  if (pTrans->redoActionPos >= numOfActions) {
+    taosThreadMutexUnlock(&pTrans->mutex);
+    return code;
+  }
 
   mInfo("trans:%d, execute %d actions serial, current redoAction:%d", pTrans->id, numOfActions, pTrans->redoActionPos);
 
@@ -1379,6 +1387,8 @@ static int32_t mndTransExecuteRedoActionsSerial(SMnode *pMnode, STrans *pTrans)
     }
   }
 
+  taosThreadMutexUnlock(&pTrans->mutex);
+
   return code;
 }