• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

taosdata / TDengine / #3824

01 Apr 2025 12:26PM UTC coverage: 34.064% (-0.001%) from 34.065%
#3824

push

travis-ci

happyguoxy
test:alter gcda dir

148483 of 599532 branches covered (24.77%)

Branch coverage included in aggregate %.

222466 of 489445 relevant lines covered (45.45%)

762427.9 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

48.28
/source/dnode/mnode/impl/src/mndTrans.c
1
/*
2
 * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
3
 *
4
 * This program is free software: you can use, redistribute, and/or modify
5
 * it under the terms of the GNU Affero General Public License, version 3
6
 * or later ("AGPL"), as published by the Free Software Foundation.
7
 *
8
 * This program is distributed in the hope that it will be useful, but WITHOUT
9
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
10
 * FITNESS FOR A PARTICULAR PURPOSE.
11
 *
12
 * You should have received a copy of the GNU Affero General Public License
13
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
14
 */
15

16
#define _DEFAULT_SOURCE
17
#include "mndTrans.h"
18
#include "mndDb.h"
19
#include "mndPrivilege.h"
20
#include "mndShow.h"
21
#include "mndStb.h"
22
#include "mndSubscribe.h"
23
#include "mndSync.h"
24
#include "mndUser.h"
25
#include "mndVgroup.h"
26
#include "osTime.h"
27

28
#define TRANS_VER1_NUMBER  1
29
#define TRANS_VER2_NUMBER  2
30
#define TRANS_ARRAY_SIZE   8
31
#define TRANS_RESERVE_SIZE 42
32

33
static int32_t mndTransActionInsert(SSdb *pSdb, STrans *pTrans);
34
static int32_t mndTransActionUpdate(SSdb *pSdb, STrans *OldTrans, STrans *pOld);
35
static int32_t mndTransDelete(SSdb *pSdb, STrans *pTrans, bool callFunc);
36

37
static int32_t mndTransAppendLog(SArray *pArray, SSdbRaw *pRaw);
38
static int32_t mndTransAppendAction(SArray *pArray, STransAction *pAction);
39
static void    mndTransDropLogs(SArray *pArray);
40
static void    mndTransDropActions(SArray *pArray);
41

42
static int32_t mndTransExecuteActions(SMnode *pMnode, STrans *pTrans, SArray *pArray, bool topHalf);
43
static int32_t mndTransExecuteRedoLogs(SMnode *pMnode, STrans *pTrans, bool topHalf);
44
static int32_t mndTransExecuteUndoLogs(SMnode *pMnode, STrans *pTrans, bool topHalf);
45
static int32_t mndTransExecuteRedoActions(SMnode *pMnode, STrans *pTrans, bool topHalf);
46
static int32_t mndTransExecuteUndoActions(SMnode *pMnode, STrans *pTrans, bool topHalf);
47
static int32_t mndTransExecuteCommitActions(SMnode *pMnode, STrans *pTrans, bool topHalf);
48
static bool    mndTransPerformRedoLogStage(SMnode *pMnode, STrans *pTrans, bool topHalf);
49
static bool    mndTransPerformRedoActionStage(SMnode *pMnode, STrans *pTrans, bool topHalf);
50
static bool    mndTransPerformUndoLogStage(SMnode *pMnode, STrans *pTrans, bool topHalf);
51
static bool    mndTransPerformUndoActionStage(SMnode *pMnode, STrans *pTrans, bool topHalf);
52
static bool    mndTransPerformCommitActionStage(SMnode *pMnode, STrans *pTrans, bool topHalf);
53
static bool    mndTransPerformCommitStage(SMnode *pMnode, STrans *pTrans, bool topHalf);
54
static bool    mndTransPerformRollbackStage(SMnode *pMnode, STrans *pTrans, bool topHalf);
55
static bool    mndTransPerformFinishStage(SMnode *pMnode, STrans *pTrans, bool topHalf);
56

57
static inline bool mndTransIsInSyncContext(bool topHalf) { return !topHalf; }
9,212✔
58

59
static bool mndCannotExecuteTrans(SMnode *pMnode, bool topHalf) {
8,440✔
60
  bool isLeader = mndIsLeader(pMnode);
8,440✔
61
  bool ret = (!pMnode->deploy && !isLeader) || mndTransIsInSyncContext(topHalf);
8,440✔
62
  if (ret) mDebug("cannot execute trans action, deploy:%d, isLeader:%d, topHalf:%d", pMnode->deploy, isLeader, topHalf);
8,440✔
63
  return ret;
8,440✔
64
}
65

66
static inline char *mndStrExecutionContext(bool topHalf) { return topHalf ? "transContext" : "syncContext"; }
12,148✔
67

68
static void    mndTransSendRpcRsp(SMnode *pMnode, STrans *pTrans);
69
static int32_t mndProcessTransTimer(SRpcMsg *pReq);
70
static int32_t mndProcessTtl(SRpcMsg *pReq);
71
static int32_t mndProcessKillTransReq(SRpcMsg *pReq);
72

73
static int32_t mndRetrieveTrans(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows);
74
static void    mndCancelGetNextTrans(SMnode *pMnode, void *pIter);
75
static int32_t mndRetrieveTransDetail(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows);
76
static int32_t tsMaxTransId = 0;
77

78
int32_t mndInitTrans(SMnode *pMnode) {
63✔
79
  SSdbTable table = {
63✔
80
      .sdbType = SDB_TRANS,
81
      .keyType = SDB_KEY_INT32,
82
      .encodeFp = (SdbEncodeFp)mndTransEncode,
83
      .decodeFp = (SdbDecodeFp)mndTransDecode,
84
      .insertFp = (SdbInsertFp)mndTransActionInsert,
85
      .updateFp = (SdbUpdateFp)mndTransActionUpdate,
86
      .deleteFp = (SdbDeleteFp)mndTransDelete,
87
  };
88

89
  mndSetMsgHandle(pMnode, TDMT_MND_TRANS_TIMER, mndProcessTransTimer);
63✔
90
  mndSetMsgHandle(pMnode, TDMT_MND_KILL_TRANS, mndProcessKillTransReq);
63✔
91

92
  mndAddShowRetrieveHandle(pMnode, TSDB_MGMT_TABLE_TRANS, mndRetrieveTrans);
63✔
93
  mndAddShowFreeIterHandle(pMnode, TSDB_MGMT_TABLE_TRANS, mndCancelGetNextTrans);
63✔
94
  mndAddShowRetrieveHandle(pMnode, TSDB_MGMT_TABLE_TRANSACTION_DETAIL, mndRetrieveTransDetail);
63✔
95
  return sdbSetTable(pMnode->pSdb, table);
63✔
96
}
97

98
void mndCleanupTrans(SMnode *pMnode) {}
60✔
99

100
static int32_t mndTransGetActionsSize(SArray *pArray) {
14,620✔
101
  int32_t actionNum = taosArrayGetSize(pArray);
14,620✔
102
  int32_t rawDataLen = 0;
14,620✔
103

104
  for (int32_t i = 0; i < actionNum; ++i) {
42,269✔
105
    STransAction *pAction = taosArrayGet(pArray, i);
27,649✔
106
    if (pAction->actionType == TRANS_ACTION_RAW) {
27,649✔
107
      rawDataLen += (sizeof(STransAction) + sdbGetRawTotalSize(pAction->pRaw));
19,312✔
108
    } else if (pAction->actionType == TRANS_ACTION_MSG) {
8,337!
109
      rawDataLen += (sizeof(STransAction) + pAction->contLen);
8,337✔
110
    } else {
111
      // empty
112
    }
113
    rawDataLen += sizeof(int8_t);
27,649✔
114
  }
115

116
  return rawDataLen;
14,620✔
117
}
118

119
static int32_t mndTransEncodeAction(SSdbRaw *pRaw, int32_t *offset, SArray *pActions, int32_t actionsNum) {
14,620✔
120
  int32_t code = 0;
14,620✔
121
  int32_t lino = 0;
14,620✔
122
  int32_t dataPos = *offset;
14,620✔
123
  int8_t  unused = 0;
14,620✔
124
  int32_t ret = -1;
14,620✔
125

126
  for (int32_t i = 0; i < actionsNum; ++i) {
42,269✔
127
    STransAction *pAction = taosArrayGet(pActions, i);
27,649✔
128
    SDB_SET_INT32(pRaw, dataPos, pAction->id, _OVER)
27,649!
129
    SDB_SET_INT32(pRaw, dataPos, pAction->errCode, _OVER)
27,649!
130
    SDB_SET_INT32(pRaw, dataPos, pAction->acceptableCode, _OVER)
27,649!
131
    SDB_SET_INT32(pRaw, dataPos, pAction->retryCode, _OVER)
27,649!
132
    SDB_SET_INT8(pRaw, dataPos, pAction->actionType, _OVER)
27,649!
133
    SDB_SET_INT8(pRaw, dataPos, pAction->stage, _OVER)
27,649!
134
    SDB_SET_INT8(pRaw, dataPos, pAction->reserved, _OVER)
27,649!
135
    if (pAction->actionType == TRANS_ACTION_RAW) {
27,649✔
136
      int32_t len = sdbGetRawTotalSize(pAction->pRaw);
19,312✔
137
      SDB_SET_INT8(pRaw, dataPos, unused /*pAction->rawWritten*/, _OVER)
19,312!
138
      SDB_SET_INT32(pRaw, dataPos, len, _OVER)
19,312!
139
      SDB_SET_BINARY(pRaw, dataPos, (void *)pAction->pRaw, len, _OVER)
19,312!
140
    } else if (pAction->actionType == TRANS_ACTION_MSG) {
8,337!
141
      SDB_SET_BINARY(pRaw, dataPos, (void *)&pAction->epSet, sizeof(SEpSet), _OVER)
8,337!
142
      SDB_SET_INT16(pRaw, dataPos, pAction->msgType, _OVER)
8,337!
143
      SDB_SET_INT8(pRaw, dataPos, unused /*pAction->msgSent*/, _OVER)
8,337!
144
      SDB_SET_INT8(pRaw, dataPos, unused /*pAction->msgReceived*/, _OVER)
8,337!
145
      SDB_SET_INT32(pRaw, dataPos, pAction->contLen, _OVER)
8,337!
146
      SDB_SET_BINARY(pRaw, dataPos, pAction->pCont, pAction->contLen, _OVER)
8,337!
147
    } else {
148
      // nothing
149
    }
150
  }
151
  ret = 0;
14,620✔
152

153
_OVER:
14,620✔
154
  *offset = dataPos;
14,620✔
155
  return ret;
14,620✔
156
}
157

158
SSdbRaw *mndTransEncode(STrans *pTrans) {
3,655✔
159
  int32_t code = 0;
3,655✔
160
  int32_t lino = 0;
3,655✔
161
  terrno = TSDB_CODE_INVALID_MSG;
3,655✔
162
  int8_t sver = TRANS_VER2_NUMBER;
3,655✔
163

164
  int32_t rawDataLen = sizeof(STrans) + TRANS_RESERVE_SIZE + pTrans->paramLen;
3,655✔
165
  rawDataLen += mndTransGetActionsSize(pTrans->prepareActions);
3,655✔
166
  rawDataLen += mndTransGetActionsSize(pTrans->redoActions);
3,655✔
167
  rawDataLen += mndTransGetActionsSize(pTrans->undoActions);
3,655✔
168
  rawDataLen += mndTransGetActionsSize(pTrans->commitActions);
3,655✔
169

170
  SSdbRaw *pRaw = sdbAllocRaw(SDB_TRANS, sver, rawDataLen);
3,655✔
171
  if (pRaw == NULL) {
3,655!
172
    mError("trans:%d, failed to alloc raw since %s", pTrans->id, terrstr());
×
173
    return NULL;
×
174
  }
175

176
  int32_t dataPos = 0;
3,655✔
177
  SDB_SET_INT32(pRaw, dataPos, pTrans->id, _OVER)
3,655!
178
  SDB_SET_INT8(pRaw, dataPos, pTrans->stage, _OVER)
3,655!
179
  SDB_SET_INT8(pRaw, dataPos, pTrans->policy, _OVER)
3,655!
180
  SDB_SET_INT8(pRaw, dataPos, pTrans->conflict, _OVER)
3,655!
181
  SDB_SET_INT8(pRaw, dataPos, pTrans->exec, _OVER)
3,655!
182
  SDB_SET_INT8(pRaw, dataPos, pTrans->oper, _OVER)
3,655!
183
  SDB_SET_INT8(pRaw, dataPos, 0, _OVER)
3,655!
184
  SDB_SET_INT16(pRaw, dataPos, pTrans->originRpcType, _OVER)
3,655!
185
  SDB_SET_INT64(pRaw, dataPos, pTrans->createdTime, _OVER)
3,655!
186
  SDB_SET_BINARY(pRaw, dataPos, pTrans->dbname, TSDB_TABLE_FNAME_LEN, _OVER)
3,655!
187
  SDB_SET_BINARY(pRaw, dataPos, pTrans->stbname, TSDB_TABLE_FNAME_LEN, _OVER)
3,655!
188
  SDB_SET_INT32(pRaw, dataPos, pTrans->actionPos, _OVER)
3,655!
189

190
  int32_t prepareActionNum = taosArrayGetSize(pTrans->prepareActions);
3,655✔
191
  int32_t redoActionNum = taosArrayGetSize(pTrans->redoActions);
3,655✔
192
  int32_t undoActionNum = taosArrayGetSize(pTrans->undoActions);
3,655✔
193
  int32_t commitActionNum = taosArrayGetSize(pTrans->commitActions);
3,655✔
194

195
  if (sver > TRANS_VER1_NUMBER) {
3,655!
196
    SDB_SET_INT32(pRaw, dataPos, prepareActionNum, _OVER)
3,655!
197
  }
198
  SDB_SET_INT32(pRaw, dataPos, redoActionNum, _OVER)
3,655!
199
  SDB_SET_INT32(pRaw, dataPos, undoActionNum, _OVER)
3,655!
200
  SDB_SET_INT32(pRaw, dataPos, commitActionNum, _OVER)
3,655!
201

202
  if (mndTransEncodeAction(pRaw, &dataPos, pTrans->prepareActions, prepareActionNum) < 0) goto _OVER;
3,655!
203
  if (mndTransEncodeAction(pRaw, &dataPos, pTrans->redoActions, redoActionNum) < 0) goto _OVER;
3,655!
204
  if (mndTransEncodeAction(pRaw, &dataPos, pTrans->undoActions, undoActionNum) < 0) goto _OVER;
3,655!
205
  if (mndTransEncodeAction(pRaw, &dataPos, pTrans->commitActions, commitActionNum) < 0) goto _OVER;
3,655!
206

207
  SDB_SET_INT32(pRaw, dataPos, pTrans->startFunc, _OVER)
3,655!
208
  SDB_SET_INT32(pRaw, dataPos, pTrans->stopFunc, _OVER)
3,655!
209
  SDB_SET_INT32(pRaw, dataPos, pTrans->paramLen, _OVER)
3,655!
210
  if (pTrans->param != NULL) {
3,655!
211
    SDB_SET_BINARY(pRaw, dataPos, pTrans->param, pTrans->paramLen, _OVER)
×
212
  }
213

214
  SDB_SET_BINARY(pRaw, dataPos, pTrans->opername, TSDB_TRANS_OPER_LEN, _OVER)
3,655!
215

216
  int32_t arbGroupNum = taosHashGetSize(pTrans->arbGroupIds);
3,655✔
217
  SDB_SET_INT32(pRaw, dataPos, arbGroupNum, _OVER)
3,655!
218
  void *pIter = NULL;
3,655✔
219
  pIter = taosHashIterate(pTrans->arbGroupIds, NULL);
3,655✔
220
  while (pIter) {
3,655!
221
    int32_t arbGroupId = *(int32_t *)pIter;
×
222
    SDB_SET_INT32(pRaw, dataPos, arbGroupId, _OVER)
×
223
    pIter = taosHashIterate(pTrans->arbGroupIds, pIter);
×
224
  }
225

226
  if (sver > TRANS_VER1_NUMBER) {
3,655!
227
    SDB_SET_INT8(pRaw, dataPos, pTrans->ableToBeKilled, _OVER)
3,655!
228
    SDB_SET_INT32(pRaw, dataPos, pTrans->killMode, _OVER)
3,655!
229
  }
230

231
  SDB_SET_RESERVE(pRaw, dataPos, TRANS_RESERVE_SIZE, _OVER)
3,655!
232
  SDB_SET_DATALEN(pRaw, dataPos, _OVER)
3,655!
233

234
  terrno = 0;
3,655✔
235

236
_OVER:
3,655✔
237
  if (terrno != 0) {
3,655!
238
    mError("trans:%d, failed to encode to raw:%p maxlen:%d len:%d since %s", pTrans->id, pRaw, sdbGetRawTotalSize(pRaw),
×
239
           dataPos, terrstr());
240
    sdbFreeRaw(pRaw);
×
241
    return NULL;
×
242
  }
243

244
  mTrace("trans:%d, encode to raw:%p, row:%p len:%d", pTrans->id, pRaw, pTrans, dataPos);
3,655✔
245
  return pRaw;
3,655✔
246
}
247

248
static int32_t mndTransDecodeAction(SSdbRaw *pRaw, int32_t *offset, SArray *pActions, int32_t actionNum) {
28,380✔
249
  int32_t      code = 0;
28,380✔
250
  int32_t      lino = 0;
28,380✔
251
  STransAction action = {0};
28,380✔
252
  int32_t      dataPos = *offset;
28,380✔
253
  int8_t       unused = 0;
28,380✔
254
  int8_t       stage = 0;
28,380✔
255
  int8_t       actionType = 0;
28,380✔
256
  int32_t      dataLen = 0;
28,380✔
257
  int32_t      ret = -1;
28,380✔
258

259
  for (int32_t i = 0; i < actionNum; ++i) {
95,070✔
260
    memset(&action, 0, sizeof(action));
66,690✔
261
    SDB_GET_INT32(pRaw, dataPos, &action.id, _OVER)
66,690!
262
    SDB_GET_INT32(pRaw, dataPos, &action.errCode, _OVER)
66,690!
263
    SDB_GET_INT32(pRaw, dataPos, &action.acceptableCode, _OVER)
66,690!
264
    SDB_GET_INT32(pRaw, dataPos, &action.retryCode, _OVER)
66,690!
265
    SDB_GET_INT8(pRaw, dataPos, &actionType, _OVER)
66,690!
266
    action.actionType = actionType;
66,690✔
267
    SDB_GET_INT8(pRaw, dataPos, &stage, _OVER)
66,690!
268
    action.stage = stage;
66,690✔
269
    SDB_GET_INT8(pRaw, dataPos, &action.reserved, _OVER)
66,690!
270
    if (action.actionType == TRANS_ACTION_RAW) {
66,690✔
271
      SDB_GET_INT8(pRaw, dataPos, &unused /*&action.rawWritten*/, _OVER)
39,977!
272
      SDB_GET_INT32(pRaw, dataPos, &dataLen, _OVER)
39,977!
273
      action.pRaw = taosMemoryMalloc(dataLen);
39,977!
274
      if (action.pRaw == NULL) goto _OVER;
39,977!
275
      mTrace("raw:%p, is created", action.pRaw);
39,977✔
276
      SDB_GET_BINARY(pRaw, dataPos, (void *)action.pRaw, dataLen, _OVER);
39,977!
277
      if (taosArrayPush(pActions, &action) == NULL) goto _OVER;
39,977!
278
      action.pRaw = NULL;
39,977✔
279
    } else if (action.actionType == TRANS_ACTION_MSG) {
26,713!
280
      SDB_GET_BINARY(pRaw, dataPos, (void *)&action.epSet, sizeof(SEpSet), _OVER);
26,713!
281
      tmsgUpdateDnodeEpSet(&action.epSet);
26,713✔
282
      SDB_GET_INT16(pRaw, dataPos, &action.msgType, _OVER)
26,713!
283
      SDB_GET_INT8(pRaw, dataPos, &unused /*&action.msgSent*/, _OVER)
26,713!
284
      SDB_GET_INT8(pRaw, dataPos, &unused /*&action.msgReceived*/, _OVER)
26,713!
285
      SDB_GET_INT32(pRaw, dataPos, &action.contLen, _OVER)
26,713!
286
      action.pCont = taosMemoryMalloc(action.contLen);
26,713!
287
      if (action.pCont == NULL) goto _OVER;
26,713!
288
      SDB_GET_BINARY(pRaw, dataPos, action.pCont, action.contLen, _OVER);
26,713!
289
      if (taosArrayPush(pActions, &action) == NULL) goto _OVER;
26,713!
290
      action.pCont = NULL;
26,713✔
291
    } else {
292
      if (taosArrayPush(pActions, &action) == NULL) goto _OVER;
×
293
    }
294
  }
295
  ret = 0;
28,380✔
296

297
_OVER:
28,380✔
298
  *offset = dataPos;
28,380✔
299
  taosMemoryFreeClear(action.pCont);
28,380!
300
  return ret;
28,380✔
301
}
302

303
SSdbRow *mndTransDecode(SSdbRaw *pRaw) {
7,095✔
304
  terrno = TSDB_CODE_INVALID_MSG;
7,095✔
305
  int32_t  code = 0;
7,095✔
306
  int32_t  lino = 0;
7,095✔
307
  SSdbRow *pRow = NULL;
7,095✔
308
  STrans  *pTrans = NULL;
7,095✔
309
  char    *pData = NULL;
7,095✔
310
  int32_t  dataLen = 0;
7,095✔
311
  int8_t   sver = 0;
7,095✔
312
  int32_t  prepareActionNum = 0;
7,095✔
313
  int32_t  redoActionNum = 0;
7,095✔
314
  int32_t  undoActionNum = 0;
7,095✔
315
  int32_t  commitActionNum = 0;
7,095✔
316
  int32_t  dataPos = 0;
7,095✔
317
  int32_t  arbgroupIdNum = 0;
7,095✔
318

319
  if (sdbGetRawSoftVer(pRaw, &sver) != 0) goto _OVER;
7,095!
320

321
  if (sver > TRANS_VER2_NUMBER) {
7,095!
322
    terrno = TSDB_CODE_SDB_INVALID_DATA_VER;
×
323
    goto _OVER;
×
324
  }
325

326
  pRow = sdbAllocRow(sizeof(STrans));
7,095✔
327
  if (pRow == NULL) goto _OVER;
7,095!
328

329
  pTrans = sdbGetRowObj(pRow);
7,095✔
330
  if (pTrans == NULL) goto _OVER;
7,095!
331

332
  SDB_GET_INT32(pRaw, dataPos, &pTrans->id, _OVER)
7,095!
333

334
  int8_t stage = 0;
7,095✔
335
  int8_t policy = 0;
7,095✔
336
  int8_t conflict = 0;
7,095✔
337
  int8_t exec = 0;
7,095✔
338
  int8_t oper = 0;
7,095✔
339
  int8_t reserved = 0;
7,095✔
340
  int8_t actionType = 0;
7,095✔
341
  SDB_GET_INT8(pRaw, dataPos, &stage, _OVER)
7,095!
342
  SDB_GET_INT8(pRaw, dataPos, &policy, _OVER)
7,095!
343
  SDB_GET_INT8(pRaw, dataPos, &conflict, _OVER)
7,095!
344
  SDB_GET_INT8(pRaw, dataPos, &exec, _OVER)
7,095!
345
  SDB_GET_INT8(pRaw, dataPos, &oper, _OVER)
7,095!
346
  SDB_GET_INT8(pRaw, dataPos, &reserved, _OVER)
7,095!
347
  pTrans->stage = stage;
7,095✔
348
  pTrans->policy = policy;
7,095✔
349
  pTrans->conflict = conflict;
7,095✔
350
  pTrans->exec = exec;
7,095✔
351
  pTrans->oper = oper;
7,095✔
352
  SDB_GET_INT16(pRaw, dataPos, &pTrans->originRpcType, _OVER)
7,095!
353
  SDB_GET_INT64(pRaw, dataPos, &pTrans->createdTime, _OVER)
7,095!
354
  SDB_GET_BINARY(pRaw, dataPos, pTrans->dbname, TSDB_TABLE_FNAME_LEN, _OVER)
7,095!
355
  SDB_GET_BINARY(pRaw, dataPos, pTrans->stbname, TSDB_TABLE_FNAME_LEN, _OVER)
7,095!
356
  SDB_GET_INT32(pRaw, dataPos, &pTrans->actionPos, _OVER)
7,095!
357

358
  if (sver > TRANS_VER1_NUMBER) {
7,095✔
359
    SDB_GET_INT32(pRaw, dataPos, &prepareActionNum, _OVER)
6,975!
360
  }
361
  SDB_GET_INT32(pRaw, dataPos, &redoActionNum, _OVER)
7,095!
362
  SDB_GET_INT32(pRaw, dataPos, &undoActionNum, _OVER)
7,095!
363
  SDB_GET_INT32(pRaw, dataPos, &commitActionNum, _OVER)
7,095!
364

365
  pTrans->prepareActions = taosArrayInit(prepareActionNum, sizeof(STransAction));
7,095✔
366
  pTrans->redoActions = taosArrayInit(redoActionNum, sizeof(STransAction));
7,095✔
367
  pTrans->undoActions = taosArrayInit(undoActionNum, sizeof(STransAction));
7,095✔
368
  pTrans->commitActions = taosArrayInit(commitActionNum, sizeof(STransAction));
7,095✔
369

370
  if (pTrans->prepareActions == NULL) goto _OVER;
7,095!
371
  if (pTrans->redoActions == NULL) goto _OVER;
7,095!
372
  if (pTrans->undoActions == NULL) goto _OVER;
7,095!
373
  if (pTrans->commitActions == NULL) goto _OVER;
7,095!
374

375
  if (mndTransDecodeAction(pRaw, &dataPos, pTrans->prepareActions, prepareActionNum) < 0) goto _OVER;
7,095!
376
  if (mndTransDecodeAction(pRaw, &dataPos, pTrans->redoActions, redoActionNum) < 0) goto _OVER;
7,095!
377
  if (mndTransDecodeAction(pRaw, &dataPos, pTrans->undoActions, undoActionNum) < 0) goto _OVER;
7,095!
378
  if (mndTransDecodeAction(pRaw, &dataPos, pTrans->commitActions, commitActionNum) < 0) goto _OVER;
7,095!
379

380
  SDB_GET_INT32(pRaw, dataPos, &pTrans->startFunc, _OVER)
7,095!
381
  SDB_GET_INT32(pRaw, dataPos, &pTrans->stopFunc, _OVER)
7,095!
382
  SDB_GET_INT32(pRaw, dataPos, &pTrans->paramLen, _OVER)
7,095!
383
  if (pTrans->paramLen != 0) {
7,095!
384
    pTrans->param = taosMemoryMalloc(pTrans->paramLen);
×
385
    if (pTrans->param == NULL) goto _OVER;
×
386
    SDB_GET_BINARY(pRaw, dataPos, pTrans->param, pTrans->paramLen, _OVER);
×
387
  }
388

389
  SDB_GET_BINARY(pRaw, dataPos, pTrans->opername, TSDB_TRANS_OPER_LEN, _OVER);
7,095!
390

391
  pTrans->arbGroupIds = taosHashInit(16, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), false, HASH_ENTRY_LOCK);
7,095✔
392

393
  SDB_GET_INT32(pRaw, dataPos, &arbgroupIdNum, _OVER)
7,095!
394
  for (int32_t i = 0; i < arbgroupIdNum; ++i) {
7,095!
395
    int32_t arbGroupId = 0;
×
396
    SDB_GET_INT32(pRaw, dataPos, &arbGroupId, _OVER)
×
397
    if ((terrno = taosHashPut(pTrans->arbGroupIds, &arbGroupId, sizeof(int32_t), NULL, 0)) != 0) goto _OVER;
×
398
  }
399

400
  int8_t ableKill = 0;
7,095✔
401
  int8_t killMode = 0;
7,095✔
402
  SDB_GET_INT8(pRaw, dataPos, &ableKill, _OVER)
7,095!
403
  SDB_GET_INT8(pRaw, dataPos, &killMode, _OVER)
7,095!
404
  pTrans->ableToBeKilled = ableKill;
7,095✔
405
  pTrans->killMode = killMode;
7,095✔
406

407
  SDB_GET_RESERVE(pRaw, dataPos, TRANS_RESERVE_SIZE, _OVER)
7,095!
408

409
  terrno = 0;
7,095✔
410

411
_OVER:
7,095✔
412
  if (terrno != 0 && pTrans != NULL) {
7,095!
413
    mError("trans:%d, failed to parse from raw:%p since %s", pTrans->id, pRaw, terrstr());
×
414
    mndTransDropData(pTrans);
×
415
    taosMemoryFreeClear(pRow);
×
416
    return NULL;
×
417
  }
418

419
  if (pTrans != NULL) {
7,095!
420
    mTrace("trans:%d, decode from raw:%p, row:%p", pTrans->id, pRaw, pTrans);
7,095✔
421
  }
422
  return pRow;
7,095✔
423
}
424

425
static const char *mndTransStr(ETrnStage stage) {
49,444✔
426
  switch (stage) {
49,444!
427
    case TRN_STAGE_PREPARE:
3,123✔
428
      return "prepare";
3,123✔
429
    case TRN_STAGE_REDO_ACTION:
18,614✔
430
      return "redoAction";
18,614✔
431
    case TRN_STAGE_ROLLBACK:
×
432
      return "rollback";
×
433
    case TRN_STAGE_UNDO_ACTION:
810✔
434
      return "undoAction";
810✔
435
    case TRN_STAGE_COMMIT:
5,860✔
436
      return "commit";
5,860✔
437
    case TRN_STAGE_COMMIT_ACTION:
12,673✔
438
      return "commitAction";
12,673✔
439
    case TRN_STAGE_FINISH:
8,364✔
440
      return "finished";
8,364✔
441
    case TRN_STAGE_PRE_FINISH:
×
442
      return "pre-finish";
×
443
    default:
×
444
      return "invalid";
×
445
  }
446
}
447

448
static const char *mndTransTypeStr(ETrnAct actionType) {
×
449
  switch (actionType) {
×
450
    case TRANS_ACTION_MSG:
×
451
      return "msg";
×
452
    case TRANS_ACTION_RAW:
×
453
      return "sdb";
×
454
    default:
×
455
      return "invalid";
×
456
  }
457
}
458

459
static void mndSetTransLastAction(STrans *pTrans, STransAction *pAction) {
14,015✔
460
  if (pAction != NULL) {
14,015✔
461
    if (pAction->errCode != TSDB_CODE_ACTION_IN_PROGRESS) {
11,050✔
462
      pTrans->lastAction = pAction->id;
8,112✔
463
      pTrans->lastMsgType = pAction->msgType;
8,112✔
464
      pTrans->lastEpset = pAction->epSet;
8,112✔
465
      pTrans->lastErrorNo = pAction->errCode;
8,112✔
466
    }
467
  } else {
468
    pTrans->lastAction = 0;
2,965✔
469
    pTrans->lastMsgType = 0;
2,965✔
470
    memset(&pTrans->lastEpset, 0, sizeof(pTrans->lastEpset));
2,965✔
471
    pTrans->lastErrorNo = 0;
2,965✔
472
  }
473
}
14,015✔
474

475
static void mndTransTestStartFunc(SMnode *pMnode, void *param, int32_t paramLen) {
×
476
  mInfo("test trans start, param:%s, len:%d", (char *)param, paramLen);
×
477
}
×
478

479
static void mndTransTestStopFunc(SMnode *pMnode, void *param, int32_t paramLen) {
×
480
  mInfo("test trans stop, param:%s, len:%d", (char *)param, paramLen);
×
481
}
×
482

483
static TransCbFp mndTransGetCbFp(ETrnFunc ftype) {
206✔
484
  switch (ftype) {
206!
485
    case TRANS_START_FUNC_TEST:
×
486
      return mndTransTestStartFunc;
×
487
    case TRANS_STOP_FUNC_TEST:
×
488
      return mndTransTestStopFunc;
×
489
    case TRANS_START_FUNC_MQ_REB:
103✔
490
      return mndRebCntInc;
103✔
491
    case TRANS_STOP_FUNC_MQ_REB:
103✔
492
      return mndRebCntDec;
103✔
493
    default:
×
494
      return NULL;
×
495
  }
496
}
497

498
static int32_t mndTransActionInsert(SSdb *pSdb, STrans *pTrans) {
1,279✔
499
  mInfo("trans:%d, perform insert action, row:%p stage:%s, callfunc:1, startFunc:%d", pTrans->id, pTrans,
1,279!
500
        mndTransStr(pTrans->stage), pTrans->startFunc);
501

502
  (void)taosThreadMutexInit(&pTrans->mutex, NULL);
1,279✔
503

504
  if (pTrans->startFunc > 0) {
1,279✔
505
    TransCbFp fp = mndTransGetCbFp(pTrans->startFunc);
103✔
506
    if (fp) {
103!
507
      (*fp)(pSdb->pMnode, pTrans->param, pTrans->paramLen);
103✔
508
    }
509
    // pTrans->startFunc = 0;
510
  }
511

512
  if (pTrans->stage == TRN_STAGE_COMMIT) {
1,279!
513
    pTrans->stage = TRN_STAGE_COMMIT_ACTION;
×
514
    mInfo("trans:%d, stage from commit to commitAction since perform update action", pTrans->id);
×
515
  }
516

517
  if (pTrans->stage == TRN_STAGE_ROLLBACK) {
1,279!
518
    pTrans->stage = TRN_STAGE_UNDO_ACTION;
×
519
    mInfo("trans:%d, stage from rollback to undoAction since perform update action", pTrans->id);
×
520
  }
521

522
  if (pTrans->stage == TRN_STAGE_PRE_FINISH) {
1,279!
523
    pTrans->stage = TRN_STAGE_FINISH;
×
524
    mInfo("trans:%d, stage from pre-finish to finished since perform update action", pTrans->id);
×
525
  }
526

527
  return 0;
1,279✔
528
}
529

530
void mndTransDropData(STrans *pTrans) {
8,200✔
531
  if (pTrans->prepareActions != NULL) {
8,200!
532
    mndTransDropActions(pTrans->prepareActions);
8,200✔
533
    pTrans->prepareActions = NULL;
8,200✔
534
  }
535
  if (pTrans->redoActions != NULL) {
8,200!
536
    mndTransDropActions(pTrans->redoActions);
8,200✔
537
    pTrans->redoActions = NULL;
8,200✔
538
  }
539
  if (pTrans->undoActions != NULL) {
8,200!
540
    mndTransDropActions(pTrans->undoActions);
8,200✔
541
    pTrans->undoActions = NULL;
8,200✔
542
  }
543
  if (pTrans->commitActions != NULL) {
8,200!
544
    mndTransDropActions(pTrans->commitActions);
8,200✔
545
    pTrans->commitActions = NULL;
8,200✔
546
  }
547
  if (pTrans->arbGroupIds != NULL) {
8,200!
548
    taosHashCleanup(pTrans->arbGroupIds);
8,200✔
549
  }
550
  if (pTrans->pRpcArray != NULL) {
8,200✔
551
    taosArrayDestroy(pTrans->pRpcArray);
1,105✔
552
    pTrans->pRpcArray = NULL;
1,105✔
553
  }
554
  if (pTrans->rpcRsp != NULL) {
8,200✔
555
    taosMemoryFree(pTrans->rpcRsp);
153!
556
    pTrans->rpcRsp = NULL;
153✔
557
    pTrans->rpcRspLen = 0;
153✔
558
  }
559
  if (pTrans->param != NULL) {
8,200!
560
    taosMemoryFree(pTrans->param);
×
561
    pTrans->param = NULL;
×
562
    pTrans->paramLen = 0;
×
563
  }
564
  (void)taosThreadMutexDestroy(&pTrans->mutex);
8,200✔
565
}
8,200✔
566

567
static int32_t mndTransDelete(SSdb *pSdb, STrans *pTrans, bool callFunc) {
4,186✔
568
  mInfo("trans:%d, perform delete action, row:%p stage:%s callfunc:%d, stopFunc:%d", pTrans->id, pTrans,
4,186!
569
        mndTransStr(pTrans->stage), callFunc, pTrans->stopFunc);
570

571
  if (pTrans->stopFunc > 0 && callFunc) {
4,186✔
572
    TransCbFp fp = mndTransGetCbFp(pTrans->stopFunc);
103✔
573
    if (fp) {
103!
574
      (*fp)(pSdb->pMnode, pTrans->param, pTrans->paramLen);
103✔
575
    }
576
    // pTrans->stopFunc = 0;
577
  }
578

579
  mndTransDropData(pTrans);
4,186✔
580
  return 0;
4,186✔
581
}
582

583
static void mndTransUpdateActions(SArray *pOldArray, SArray *pNewArray) {
6,532✔
584
  for (int32_t i = 0; i < taosArrayGetSize(pOldArray); ++i) {
26,870✔
585
    STransAction *pOldAction = taosArrayGet(pOldArray, i);
20,338✔
586
    STransAction *pNewAction = taosArrayGet(pNewArray, i);
20,338✔
587
    pOldAction->rawWritten = pNewAction->rawWritten;
20,338✔
588
    pOldAction->msgSent = pNewAction->msgSent;
20,338✔
589
    pOldAction->msgReceived = pNewAction->msgReceived;
20,338✔
590
    pOldAction->errCode = pNewAction->errCode;
20,338✔
591
  }
592
}
6,532✔
593

594
static int32_t mndTransActionUpdate(SSdb *pSdb, STrans *pOld, STrans *pNew) {
1,633✔
595
  mInfo("trans:%d, perform update action, old row:%p stage:%s create:%" PRId64 ", new row:%p stage:%s create:%" PRId64,
1,633!
596
        pOld->id, pOld, mndTransStr(pOld->stage), pOld->createdTime, pNew, mndTransStr(pNew->stage), pNew->createdTime);
597

598
  if (pOld->createdTime != pNew->createdTime) {
1,633!
599
    mError("trans:%d, failed to perform update action since createTime not match, old row:%p stage:%s create:%" PRId64
×
600
           ", new row:%p stage:%s create:%" PRId64,
601
           pOld->id, pOld, mndTransStr(pOld->stage), pOld->createdTime, pNew, mndTransStr(pNew->stage),
602
           pNew->createdTime);
603
    // only occured while sync timeout
604
    TAOS_RETURN(TSDB_CODE_MND_TRANS_SYNC_TIMEOUT);
×
605
  }
606

607
  mndTransUpdateActions(pOld->prepareActions, pNew->prepareActions);
1,633✔
608
  mndTransUpdateActions(pOld->redoActions, pNew->redoActions);
1,633✔
609
  mndTransUpdateActions(pOld->undoActions, pNew->undoActions);
1,633✔
610
  mndTransUpdateActions(pOld->commitActions, pNew->commitActions);
1,633✔
611
  pOld->stage = pNew->stage;
1,633✔
612
  pOld->actionPos = pNew->actionPos;
1,633✔
613

614
  if (pOld->stage == TRN_STAGE_COMMIT) {
1,633✔
615
    pOld->stage = TRN_STAGE_COMMIT_ACTION;
1,274✔
616
    mInfo("trans:%d, stage from commit to commitAction since perform update action", pNew->id);
1,274!
617
  }
618

619
  if (pOld->stage == TRN_STAGE_ROLLBACK) {
1,633!
620
    pOld->stage = TRN_STAGE_UNDO_ACTION;
×
621
    mInfo("trans:%d, stage from rollback to undoAction since perform update action", pNew->id);
×
622
  }
623

624
  if (pOld->stage == TRN_STAGE_PRE_FINISH) {
1,633!
625
    pOld->stage = TRN_STAGE_FINISH;
×
626
    mInfo("trans:%d, stage from pre-finish to finished since perform update action", pNew->id);
×
627
  }
628

629
  return 0;
1,633✔
630
}
631

632
STrans *mndAcquireTrans(SMnode *pMnode, int32_t transId) {
5,566✔
633
  STrans *pTrans = sdbAcquire(pMnode->pSdb, SDB_TRANS, &transId);
5,566✔
634
  if (pTrans == NULL) {
5,566✔
635
    terrno = TSDB_CODE_MND_TRANS_NOT_EXIST;
5✔
636
  }
637
  return pTrans;
5,566✔
638
}
639

640
void mndReleaseTrans(SMnode *pMnode, STrans *pTrans) {
5,561✔
641
  SSdb *pSdb = pMnode->pSdb;
5,561✔
642
  if (pTrans != NULL) mInfo("vgId:1, trans:%d, release transaction", pTrans->id);
5,561!
643
  sdbRelease(pSdb, pTrans);
5,561✔
644
}
5,561✔
645

646
STrans *mndTransCreate(SMnode *pMnode, ETrnPolicy policy, ETrnConflct conflict, const SRpcMsg *pReq,
1,105✔
647
                       const char *opername) {
648
  STrans *pTrans = taosMemoryCalloc(1, sizeof(STrans));
1,105!
649
  if (pTrans == NULL) {
1,105!
650
    terrno = TSDB_CODE_OUT_OF_MEMORY;
×
651
    mError("failed to create transaction since %s", terrstr());
×
652
    return NULL;
×
653
  }
654

655
  if (opername != NULL) {
1,105!
656
    tstrncpy(pTrans->opername, opername, TSDB_TRANS_OPER_LEN);
1,105✔
657
  }
658

659
  int32_t sdbMaxId = sdbGetMaxId(pMnode->pSdb, SDB_TRANS);
1,105✔
660
  sdbReadLock(pMnode->pSdb, SDB_TRANS);
1,105✔
661
  pTrans->id = TMAX(sdbMaxId, tsMaxTransId + 1);
1,105✔
662
  sdbUnLock(pMnode->pSdb, SDB_TRANS);
1,105✔
663
  pTrans->stage = TRN_STAGE_PREPARE;
1,105✔
664
  pTrans->policy = policy;
1,105✔
665
  pTrans->conflict = conflict;
1,105✔
666
  pTrans->exec = TRN_EXEC_PARALLEL;
1,105✔
667
  pTrans->ableToBeKilled = false;
1,105✔
668
  pTrans->createdTime = taosGetTimestampMs();
1,105✔
669
  pTrans->prepareActions = taosArrayInit(TRANS_ARRAY_SIZE, sizeof(STransAction));
1,105✔
670
  pTrans->redoActions = taosArrayInit(TRANS_ARRAY_SIZE, sizeof(STransAction));
1,105✔
671
  pTrans->undoActions = taosArrayInit(TRANS_ARRAY_SIZE, sizeof(STransAction));
1,105✔
672
  pTrans->commitActions = taosArrayInit(TRANS_ARRAY_SIZE, sizeof(STransAction));
1,105✔
673
  pTrans->arbGroupIds = taosHashInit(16, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), false, HASH_ENTRY_LOCK);
1,105✔
674
  pTrans->pRpcArray = taosArrayInit(1, sizeof(SRpcHandleInfo));
1,105✔
675
  pTrans->mTraceId = pReq ? TRACE_GET_ROOTID(&pReq->info.traceId) : tGenIdPI64();
1,105✔
676
  taosInitRWLatch(&pTrans->lockRpcArray);
1,105✔
677
  (void)taosThreadMutexInit(&pTrans->mutex, NULL);
1,105✔
678

679
  if (pTrans->redoActions == NULL || pTrans->undoActions == NULL || pTrans->commitActions == NULL ||
1,105!
680
      pTrans->pRpcArray == NULL) {
1,105!
681
    terrno = TSDB_CODE_OUT_OF_MEMORY;
×
682
    mError("failed to create transaction since %s", terrstr());
×
683
    mndTransDrop(pTrans);
×
684
    return NULL;
×
685
  }
686

687
  if (pReq != NULL) {
1,105✔
688
    if (taosArrayPush(pTrans->pRpcArray, &pReq->info) == NULL) {
1,348!
689
      terrno = TSDB_CODE_OUT_OF_MEMORY;
×
690
      return NULL;
×
691
    }
692
    pTrans->originRpcType = pReq->msgType;
674✔
693
  }
694

695
  mInfo("trans:%d, create transaction:%s, origin:%s", pTrans->id, pTrans->opername, opername);
1,105!
696

697
  mTrace("trans:%d, local object is created, data:%p", pTrans->id, pTrans);
1,105✔
698
  return pTrans;
1,105✔
699
}
700

701
static void mndTransDropActions(SArray *pArray) {
32,800✔
702
  int32_t size = taosArrayGetSize(pArray);
32,800✔
703
  for (int32_t i = 0; i < size; ++i) {
106,914✔
704
    STransAction *pAction = taosArrayGet(pArray, i);
74,114✔
705
    if (pAction->actionType == TRANS_ACTION_RAW) {
74,114✔
706
      taosMemoryFreeClear(pAction->pRaw);
45,685!
707
    } else if (pAction->actionType == TRANS_ACTION_MSG) {
28,429!
708
      taosMemoryFreeClear(pAction->pCont);
28,429!
709
    } else {
710
      // nothing
711
    }
712
  }
713

714
  taosArrayDestroy(pArray);
32,800✔
715
}
32,800✔
716

717
void mndTransDrop(STrans *pTrans) {
1,169✔
718
  if (pTrans != NULL) {
1,169✔
719
    mndTransDropData(pTrans);
1,105✔
720
    mTrace("trans:%d, local object is freed, data:%p", pTrans->id, pTrans);
1,105✔
721
    taosMemoryFreeClear(pTrans);
1,105!
722
  }
723
}
1,169✔
724

725
static int32_t mndTransAppendAction(SArray *pArray, STransAction *pAction) {
7,424✔
726
  pAction->id = taosArrayGetSize(pArray);
7,424✔
727

728
  void *ptr = taosArrayPush(pArray, pAction);
7,424✔
729
  if (ptr == NULL) {
7,424!
730
    TAOS_RETURN(terrno);
×
731
  }
732

733
  return 0;
7,424✔
734
}
735

736
int32_t mndTransAppendRedolog(STrans *pTrans, SSdbRaw *pRaw) {
153✔
737
  STransAction action = {
153✔
738
      .stage = TRN_STAGE_REDO_ACTION, .actionType = TRANS_ACTION_RAW, .pRaw = pRaw, .mTraceId = pTrans->mTraceId};
153✔
739
  return mndTransAppendAction(pTrans->redoActions, &action);
153✔
740
}
741

742
int32_t mndTransAppendNullLog(STrans *pTrans) {
×
743
  STransAction action = {.stage = TRN_STAGE_REDO_ACTION, .actionType = TRANS_ACTION_NULL};
×
744
  return mndTransAppendAction(pTrans->redoActions, &action);
×
745
}
746

747
int32_t mndTransAppendUndolog(STrans *pTrans, SSdbRaw *pRaw) {
378✔
748
  STransAction action = {.stage = TRN_STAGE_UNDO_ACTION, .actionType = TRANS_ACTION_RAW, .pRaw = pRaw};
378✔
749
  return mndTransAppendAction(pTrans->undoActions, &action);
378✔
750
}
751

752
int32_t mndTransAppendCommitlog(STrans *pTrans, SSdbRaw *pRaw) {
4,837✔
753
  STransAction action = {.stage = TRN_STAGE_COMMIT_ACTION, .actionType = TRANS_ACTION_RAW, .pRaw = pRaw};
4,837✔
754
  return mndTransAppendAction(pTrans->commitActions, &action);
4,837✔
755
}
756

757
int32_t mndTransAppendPrepareLog(STrans *pTrans, SSdbRaw *pRaw) {
340✔
758
  STransAction action = {
340✔
759
      .pRaw = pRaw, .stage = TRN_STAGE_PREPARE, .actionType = TRANS_ACTION_RAW, .mTraceId = pTrans->mTraceId};
340✔
760
  return mndTransAppendAction(pTrans->prepareActions, &action);
340✔
761
}
762

763
int32_t mndTransAppendRedoAction(STrans *pTrans, STransAction *pAction) {
1,284✔
764
  pAction->stage = TRN_STAGE_REDO_ACTION;
1,284✔
765
  pAction->actionType = TRANS_ACTION_MSG;
1,284✔
766
  pAction->mTraceId = pTrans->mTraceId;
1,284✔
767
  return mndTransAppendAction(pTrans->redoActions, pAction);
1,284✔
768
}
769

770
int32_t mndTransAppendUndoAction(STrans *pTrans, STransAction *pAction) {
432✔
771
  pAction->stage = TRN_STAGE_UNDO_ACTION;
432✔
772
  pAction->actionType = TRANS_ACTION_MSG;
432✔
773
  return mndTransAppendAction(pTrans->undoActions, pAction);
432✔
774
}
775

776
void mndTransSetRpcRsp(STrans *pTrans, void *pCont, int32_t contLen) {
153✔
777
  pTrans->rpcRsp = pCont;
153✔
778
  pTrans->rpcRspLen = contLen;
153✔
779
}
153✔
780

781
void mndTransSetCb(STrans *pTrans, ETrnFunc startFunc, ETrnFunc stopFunc, void *param, int32_t paramLen) {
101✔
782
  pTrans->startFunc = startFunc;
101✔
783
  pTrans->stopFunc = stopFunc;
101✔
784
  pTrans->param = param;
101✔
785
  pTrans->paramLen = paramLen;
101✔
786
}
101✔
787

788
int32_t mndSetRpcInfoForDbTrans(SMnode *pMnode, SRpcMsg *pMsg, EOperType oper, const char *dbname) {
×
789
  STrans *pTrans = NULL;
×
790
  void   *pIter = NULL;
×
791
  int32_t code = -1;
×
792

793
  while (1) {
794
    pIter = sdbFetch(pMnode->pSdb, SDB_TRANS, pIter, (void **)&pTrans);
×
795
    if (pIter == NULL) break;
×
796

797
    if (pTrans->oper == oper) {
×
798
      if (taosStrcasecmp(dbname, pTrans->dbname) == 0) {
×
799
        mInfo("trans:%d, db:%s oper:%d matched with input", pTrans->id, dbname, oper);
×
800
        taosWLockLatch(&pTrans->lockRpcArray);
×
801
        if (pTrans->pRpcArray == NULL) {
×
802
          pTrans->pRpcArray = taosArrayInit(4, sizeof(SRpcHandleInfo));
×
803
        }
804
        if (pTrans->pRpcArray != NULL && taosArrayPush(pTrans->pRpcArray, &pMsg->info) != NULL) {
×
805
          code = 0;
×
806
        }
807
        taosWUnLockLatch(&pTrans->lockRpcArray);
×
808

809
        sdbRelease(pMnode->pSdb, pTrans);
×
810
        break;
×
811
      }
812
    }
813

814
    sdbRelease(pMnode->pSdb, pTrans);
×
815
  }
816
  return code;
×
817
}
818

819
void mndTransSetDbName(STrans *pTrans, const char *dbname, const char *stbname) {
445✔
820
  if (dbname != NULL) {
445!
821
    tstrncpy(pTrans->dbname, dbname, TSDB_DB_FNAME_LEN);
445✔
822
  }
823
  if (stbname != NULL) {
445✔
824
    tstrncpy(pTrans->stbname, stbname, TSDB_TABLE_FNAME_LEN);
315✔
825
  }
826
}
445✔
827

828
void mndTransAddArbGroupId(STrans *pTrans, int32_t groupId) {
×
829
  if (taosHashPut(pTrans->arbGroupIds, &groupId, sizeof(int32_t), NULL, 0) != 0) {
×
830
    mError("trans:%d, failed to put groupid into hash, groupId:%d", pTrans->id, groupId);
×
831
  }
832
}
×
833

834
void mndTransSetSerial(STrans *pTrans) { pTrans->exec = TRN_EXEC_SERIAL; }
51✔
835

836
void mndTransSetBeKilled(STrans *pTrans, bool ableToBeKilled) { pTrans->ableToBeKilled = ableToBeKilled; }
×
837

838
void mndTransSetKillMode(STrans *pTrans, ETrnKillMode killMode) {
×
839
  pTrans->ableToBeKilled = true; 
×
840
  pTrans->killMode = killMode; 
×
841
}
×
842

843
void mndTransSetParallel(STrans *pTrans) { pTrans->exec = TRN_EXEC_PARALLEL; }
×
844

845
void mndTransSetChangeless(STrans *pTrans) { pTrans->changeless = true; }
×
846

847
void mndTransSetOper(STrans *pTrans, EOperType oper) { pTrans->oper = oper; }
74✔
848

849
static int32_t mndTransSync(SMnode *pMnode, STrans *pTrans) {
2,378✔
850
  int32_t  code = 0;
2,378✔
851
  SSdbRaw *pRaw = mndTransEncode(pTrans);
2,378✔
852
  if (pRaw == NULL) {
2,378!
853
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
854
    if (terrno != 0) code = terrno;
×
855
    mError("trans:%d, failed to encode while sync trans since %s", pTrans->id, tstrerror(code));
×
856
    TAOS_RETURN(code);
×
857
  }
858
  TAOS_CHECK_RETURN(sdbSetRawStatus(pRaw, SDB_STATUS_READY));
2,378!
859

860
  mInfo("trans:%d, sync to other mnodes, stage:%s createTime:%" PRId64, pTrans->id, mndTransStr(pTrans->stage),
2,378!
861
        pTrans->createdTime);
862
  code = mndSyncPropose(pMnode, pRaw, pTrans->id);
2,378✔
863
  if (code != 0) {
2,378✔
864
    mError("trans:%d, failed to sync, errno:%s code:0x%x createTime:%" PRId64 " saved trans:%d", pTrans->id,
1!
865
           tstrerror(code), code, pTrans->createdTime, pMnode->syncMgmt.transId);
866
    sdbFreeRaw(pRaw);
1✔
867
    TAOS_RETURN(code);
1✔
868
  }
869

870
  sdbFreeRaw(pRaw);
2,377✔
871
  mInfo("trans:%d, sync finished, createTime:%" PRId64, pTrans->id, pTrans->createdTime);
2,377!
872
  TAOS_RETURN(code);
2,377✔
873
}
874

875
static bool mndCheckDbConflict(const char *conflict, STrans *pTrans) {
×
876
  if (conflict[0] == 0) return false;
×
877
  if (taosStrcasecmp(conflict, pTrans->dbname) == 0) return true;
×
878
  return false;
×
879
}
880

881
static bool mndCheckStbConflict(const char *conflict, STrans *pTrans) {
141✔
882
  if (conflict[0] == 0) return false;
141!
883
  if (taosStrcasecmp(conflict, pTrans->stbname) == 0) return true;
141!
884
  return false;
141✔
885
}
886

887
static void mndTransLogConflict(STrans *pNew, STrans *pTrans, bool conflict, bool *globalConflict) {
141✔
888
  if (conflict) {
141!
889
    mError("trans:%d, db:%s stb:%s type:%d, can't execute since conflict with trans:%d db:%s stb:%s type:%d", pNew->id,
×
890
           pNew->dbname, pNew->stbname, pNew->conflict, pTrans->id, pTrans->dbname, pTrans->stbname, pTrans->conflict);
891
    *globalConflict = true;
×
892
  } else {
893
    mInfo("trans:%d, db:%s stb:%s type:%d, not conflict with trans:%d db:%s stb:%s type:%d", pNew->id, pNew->dbname,
141!
894
          pNew->stbname, pNew->conflict, pTrans->id, pTrans->dbname, pTrans->stbname, pTrans->conflict);
895
  }
896
}
141✔
897

898
static bool mndCheckTransConflict(SMnode *pMnode, STrans *pNew) {
2,849✔
899
  STrans *pTrans = NULL;
2,849✔
900
  void   *pIter = NULL;
2,849✔
901
  bool    conflict = false;
2,849✔
902

903
  if (pNew->conflict == TRN_CONFLICT_NOTHING) return conflict;
2,849✔
904

905
  int32_t size = sdbGetSize(pMnode->pSdb, SDB_TRANS);
1,644✔
906
  mInfo("trans:%d, trans hash size %d", pNew->id, size);
1,644!
907

908
  while (1) {
909
    pIter = sdbFetch(pMnode->pSdb, SDB_TRANS, pIter, (void **)&pTrans);
1,785✔
910
    if (pIter == NULL) break;
1,785✔
911

912
    if (pNew->conflict == TRN_CONFLICT_GLOBAL) conflict = true;
141!
913

914
    if (pNew->conflict == TRN_CONFLICT_DB) {
141!
915
      if (pTrans->conflict == TRN_CONFLICT_GLOBAL) conflict = true;
×
916
      if (pTrans->conflict == TRN_CONFLICT_DB || pTrans->conflict == TRN_CONFLICT_DB_INSIDE) {
×
917
        mndTransLogConflict(pNew, pTrans, mndCheckDbConflict(pNew->dbname, pTrans), &conflict);
×
918
        mndTransLogConflict(pNew, pTrans, mndCheckStbConflict(pNew->stbname, pTrans), &conflict);
×
919
      }
920
    }
921

922
    if (pNew->conflict == TRN_CONFLICT_DB_INSIDE) {
141!
923
      if (pTrans->conflict == TRN_CONFLICT_GLOBAL) conflict = true;
141!
924
      if (pTrans->conflict == TRN_CONFLICT_DB) {
141!
925
        mndTransLogConflict(pNew, pTrans, mndCheckDbConflict(pNew->dbname, pTrans), &conflict);
×
926
        mndTransLogConflict(pNew, pTrans, mndCheckStbConflict(pNew->stbname, pTrans), &conflict);
×
927
      }
928
      if (pTrans->conflict == TRN_CONFLICT_DB_INSIDE) {
141!
929
        mndTransLogConflict(pNew, pTrans, mndCheckStbConflict(pNew->stbname, pTrans), &conflict);  // for stb
141✔
930
      }
931
    }
932

933
    if (pNew->conflict == TRN_CONFLICT_ARBGROUP) {
141!
934
      if (pTrans->conflict == TRN_CONFLICT_GLOBAL) conflict = true;
×
935
      if (pTrans->conflict == TRN_CONFLICT_ARBGROUP) {
×
936
        void *pGidIter = taosHashIterate(pNew->arbGroupIds, NULL);
×
937
        while (pGidIter != NULL) {
×
938
          int32_t groupId = *(int32_t *)pGidIter;
×
939
          if (taosHashGet(pTrans->arbGroupIds, &groupId, sizeof(int32_t)) != NULL) {
×
940
            taosHashCancelIterate(pNew->arbGroupIds, pGidIter);
×
941
            mndTransLogConflict(pNew, pTrans, true, &conflict);
×
942
            break;
×
943
          } else {
944
            mndTransLogConflict(pNew, pTrans, false, &conflict);
×
945
          }
946
          pGidIter = taosHashIterate(pNew->arbGroupIds, pGidIter);
×
947
        }
948
      }
949
    }
950

951
    if (pNew->conflict == TRN_CONFLICT_TSMA) {
141!
952
      if (pTrans->conflict == TRN_CONFLICT_GLOBAL || pTrans->conflict == TRN_CONFLICT_TSMA) {
×
953
        mndTransLogConflict(pNew, pTrans, true, &conflict);
×
954
      } else {
955
        mndTransLogConflict(pNew, pTrans, false, &conflict);
×
956
      }
957
    }
958

959
    sdbRelease(pMnode->pSdb, pTrans);
141✔
960
  }
961

962
  return conflict;
1,644✔
963
}
964

965
int32_t mndTransCheckConflict(SMnode *pMnode, STrans *pTrans) {
2,849✔
966
  int32_t code = 0;
2,849✔
967
  if (pTrans->conflict == TRN_CONFLICT_DB || pTrans->conflict == TRN_CONFLICT_DB_INSIDE) {
2,849✔
968
    if (strlen(pTrans->dbname) == 0 && strlen(pTrans->stbname) == 0) {
1,363!
969
      code = TSDB_CODE_MND_TRANS_CONFLICT;
×
970
      mError("trans:%d, failed to check tran conflict since db not set", pTrans->id);
×
971
      TAOS_RETURN(code);
×
972
    }
973
  }
974

975
  if (mndCheckTransConflict(pMnode, pTrans)) {
2,849!
976
    code = TSDB_CODE_MND_TRANS_CONFLICT;
×
977
    mError("trans:%d, failed to check tran conflict since %s", pTrans->id, tstrerror(code));
×
978
    TAOS_RETURN(code);
×
979
  }
980

981
  TAOS_RETURN(code);
2,849✔
982
}
983

984
int32_t mndTransCheckConflictWithCompact(SMnode *pMnode, STrans *pTrans) {
7✔
985
  int32_t      code = 0;
7✔
986
  void        *pIter = NULL;
7✔
987
  bool         conflict = false;
7✔
988
  SCompactObj *pCompact = NULL;
7✔
989

990
  while (1) {
×
991
    bool thisConflict = false;
7✔
992
    pIter = sdbFetch(pMnode->pSdb, SDB_COMPACT, pIter, (void **)&pCompact);
7✔
993
    if (pIter == NULL) break;
7!
994

995
    if (pTrans->conflict == TRN_CONFLICT_GLOBAL) {
×
996
      thisConflict = true;
×
997
    }
998
    if (pTrans->conflict == TRN_CONFLICT_DB || pTrans->conflict == TRN_CONFLICT_DB_INSIDE) {
×
999
      if (taosStrcasecmp(pTrans->dbname, pCompact->dbname) == 0) thisConflict = true;
×
1000
    }
1001

1002
    if (thisConflict) {
×
1003
      mError("trans:%d, db:%s stb:%s type:%d, can't execute since conflict with compact:%d db:%s", pTrans->id,
×
1004
             pTrans->dbname, pTrans->stbname, pTrans->conflict, pCompact->compactId, pCompact->dbname);
1005
      conflict = true;
×
1006
    } else {
1007
      mInfo("trans:%d, db:%s stb:%s type:%d, not conflict with compact:%d db:%s", pTrans->id, pTrans->dbname,
×
1008
            pTrans->stbname, pTrans->conflict, pCompact->compactId, pCompact->dbname);
1009
    }
1010
    sdbRelease(pMnode->pSdb, pCompact);
×
1011
  }
1012

1013
  if (conflict) {
7!
1014
    code = TSDB_CODE_MND_TRANS_CONFLICT_COMPACT;
×
1015
    mError("trans:%d, failed to check tran conflict with compact since %s", pTrans->id, tstrerror(code));
×
1016
    TAOS_RETURN(code);
×
1017
  }
1018

1019
  TAOS_RETURN(code);
7✔
1020
}
1021

1022
static bool mndTransActionsOfSameType(SArray *pActions) {
2,567✔
1023
  int32_t size = taosArrayGetSize(pActions);
2,567✔
1024
  ETrnAct lastActType = TRANS_ACTION_NULL;
2,567✔
1025
  bool    same = true;
2,567✔
1026
  for (int32_t i = 0; i < size; ++i) {
9,009✔
1027
    STransAction *pAction = taosArrayGet(pActions, i);
6,442✔
1028
    if (i > 0) {
6,442✔
1029
      if (lastActType != pAction->actionType) {
4,659!
1030
        same = false;
×
1031
        break;
×
1032
      }
1033
    }
1034
    lastActType = pAction->actionType;
6,442✔
1035
  }
1036
  return same;
2,567✔
1037
}
1038

1039
static int32_t mndTransCheckParallelActions(SMnode *pMnode, STrans *pTrans) {
1,105✔
1040
  int32_t code = 0;
1,105✔
1041
  if (pTrans->exec == TRN_EXEC_PARALLEL) {
1,105✔
1042
    if (mndTransActionsOfSameType(pTrans->redoActions) == false) {
1,060!
1043
      code = TSDB_CODE_MND_TRANS_INVALID_STAGE;
×
1044
      mError("trans:%d, types of parallel redo actions are not the same", pTrans->id);
×
1045
      TAOS_RETURN(code);
×
1046
    }
1047

1048
    if (pTrans->policy == TRN_POLICY_ROLLBACK) {
1,060✔
1049
      if (mndTransActionsOfSameType(pTrans->undoActions) == false) {
402!
1050
        code = TSDB_CODE_MND_TRANS_INVALID_STAGE;
×
1051
        mError("trans:%d, types of parallel undo actions are not the same", pTrans->id);
×
1052
        TAOS_RETURN(code);
×
1053
      }
1054
    }
1055
  }
1056

1057
  TAOS_RETURN(code);
1,105✔
1058
}
1059

1060
static int32_t mndTransCheckCommitActions(SMnode *pMnode, STrans *pTrans) {
1,105✔
1061
  int32_t code = 0;
1,105✔
1062
  if (!pTrans->changeless && taosArrayGetSize(pTrans->commitActions) <= 0) {
1,105!
1063
    code = TSDB_CODE_MND_TRANS_CLOG_IS_NULL;
×
1064
    mError("trans:%d, commit actions of non-changeless trans are empty", pTrans->id);
×
1065
    TAOS_RETURN(code);
×
1066
  }
1067
  if (mndTransActionsOfSameType(pTrans->commitActions) == false) {
1,105!
1068
    code = TSDB_CODE_MND_TRANS_INVALID_STAGE;
×
1069
    mError("trans:%d, types of commit actions are not the same", pTrans->id);
×
1070
    TAOS_RETURN(code);
×
1071
  }
1072

1073
  TAOS_RETURN(code);
1,105✔
1074
}
1075

1076
int32_t mndTransPrepare(SMnode *pMnode, STrans *pTrans) {
1,105✔
1077
  int32_t code = 0;
1,105✔
1078
  if (pTrans == NULL) {
1,105!
1079
    return TSDB_CODE_INVALID_PARA;
×
1080
  }
1081

1082
  mInfo("trans:%d, action list:", pTrans->id);
1,105!
1083
  int32_t index = 0;
1,105✔
1084
  for (int32_t i = 0; i < taosArrayGetSize(pTrans->prepareActions); ++i, ++index) {
1,445✔
1085
    STransAction *pAction = taosArrayGet(pTrans->prepareActions, i);
340✔
1086
    mInfo("trans:%d, action:%d, %s:%d sdbType:%s, sdbStatus:%s", pTrans->id, index,
340!
1087
          mndTransStr(pAction->stage), pAction->id, sdbTableName(pAction->pRaw->type), sdbStatusName(pAction->pRaw->status));
1088
  }
1089

1090
  for (int32_t i = 0; i < taosArrayGetSize(pTrans->redoActions); ++i, ++index) {
2,542✔
1091
    STransAction *pAction = taosArrayGet(pTrans->redoActions, i);
1,437✔
1092
    mInfo("trans:%d, action:%d, %s:%d msgType:%s", pTrans->id, index,
1,437!
1093
          mndTransStr(pAction->stage), pAction->id, TMSG_INFO(pAction->msgType));;
1094
  }
1095

1096
  for (int32_t i = 0; i < taosArrayGetSize(pTrans->commitActions); ++i, ++index) {
5,942✔
1097
    STransAction *pAction = taosArrayGet(pTrans->commitActions, i);
4,837✔
1098
    mInfo("trans:%d, action:%d, %s:%d sdbType:%s, sdbStatus:%s", pTrans->id, index,
4,837!
1099
          mndTransStr(pAction->stage), i, sdbTableName(pAction->pRaw->type), sdbStatusName(pAction->pRaw->status));
1100
  }
1101

1102
  for (int32_t i = 0; i < taosArrayGetSize(pTrans->undoActions); ++i, ++index) {
1,915✔
1103
    STransAction *pAction = taosArrayGet(pTrans->undoActions, i);
810✔
1104
    if(pAction->actionType == TRANS_ACTION_MSG){
810✔
1105
      mInfo("trans:%d, action:%d, %s:%d msgType:%s", pTrans->id, index,
432!
1106
            mndTransStr(pAction->stage), pAction->id, TMSG_INFO(pAction->msgType));;
1107
    }
1108
    else{
1109
      mInfo("trans:%d, action:%d, %s:%d sdbType:%s, sdbStatus:%s", pTrans->id, index,
378!
1110
            mndTransStr(pAction->stage), pAction->id, sdbTableName(pAction->pRaw->type), sdbStatusName(pAction->pRaw->status));
1111
    }
1112
  }
1113

1114

1115
  TAOS_CHECK_RETURN(mndTransCheckConflict(pMnode, pTrans));
1,105!
1116

1117
  TAOS_CHECK_RETURN(mndTransCheckParallelActions(pMnode, pTrans));
1,105!
1118

1119
  TAOS_CHECK_RETURN(mndTransCheckCommitActions(pMnode, pTrans));
1,105!
1120

1121
  mInfo("trans:%d, prepare transaction", pTrans->id);
1,105!
1122
  if ((code = mndTransSync(pMnode, pTrans)) != 0) {
1,105✔
1123
    mError("trans:%d, failed to prepare since %s", pTrans->id, tstrerror(code));
1!
1124
    sdbWriteLock(pMnode->pSdb, SDB_TRANS);
1✔
1125
    tsMaxTransId = TMAX(pTrans->id, tsMaxTransId);
1✔
1126
    sdbUnLock(pMnode->pSdb, SDB_TRANS);
1✔
1127
    TAOS_RETURN(code);
1✔
1128
  }
1129
  mInfo("trans:%d, prepare finished", pTrans->id);
1,104!
1130

1131
  STrans *pNew = mndAcquireTrans(pMnode, pTrans->id);
1,104✔
1132
  if (pNew == NULL) {
1,104!
1133
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
1134
    if (terrno != 0) code = terrno;
×
1135
    mError("trans:%d, failed to read from sdb since %s", pTrans->id, tstrerror(code));
×
1136
    TAOS_RETURN(code);
×
1137
  }
1138

1139
  pNew->pRpcArray = pTrans->pRpcArray;
1,104✔
1140
  pNew->rpcRsp = pTrans->rpcRsp;
1,104✔
1141
  pNew->rpcRspLen = pTrans->rpcRspLen;
1,104✔
1142
  pNew->mTraceId = pTrans->mTraceId;
1,104✔
1143
  pTrans->pRpcArray = NULL;
1,104✔
1144
  pTrans->rpcRsp = NULL;
1,104✔
1145
  pTrans->rpcRspLen = 0;
1,104✔
1146

1147
  mndTransExecute(pMnode, pNew);
1,104✔
1148
  mndReleaseTrans(pMnode, pNew);
1,104✔
1149
  // TDOD change to TAOS_RETURN(code);
1150
  return 0;
1,104✔
1151
}
1152

1153
static int32_t mndTransCommit(SMnode *pMnode, STrans *pTrans) {
1,104✔
1154
  int32_t code = 0;
1,104✔
1155
  mInfo("trans:%d, commit transaction", pTrans->id);
1,104!
1156
  if ((code = mndTransSync(pMnode, pTrans)) != 0) {
1,104!
1157
    mError("trans:%d, failed to commit since %s", pTrans->id, tstrerror(code));
×
1158
    TAOS_RETURN(code);
×
1159
  }
1160
  mInfo("trans:%d, commit finished", pTrans->id);
1,104!
1161
  TAOS_RETURN(code);
1,104✔
1162
}
1163

1164
static int32_t mndTransRollback(SMnode *pMnode, STrans *pTrans) {
×
1165
  int32_t code = 0;
×
1166
  mInfo("trans:%d, rollback transaction", pTrans->id);
×
1167
  if ((code = mndTransSync(pMnode, pTrans)) != 0) {
×
1168
    mError("trans:%d, failed to rollback since %s", pTrans->id, tstrerror(code));
×
1169
    TAOS_RETURN(code);
×
1170
  }
1171
  mInfo("trans:%d, rollback finished", pTrans->id);
×
1172
  TAOS_RETURN(code);
×
1173
}
1174

1175
static int32_t mndTransPreFinish(SMnode *pMnode, STrans *pTrans) {
×
1176
  int32_t code = 0;
×
1177
  mInfo("trans:%d, pre-finish transaction", pTrans->id);
×
1178
  if ((code = mndTransSync(pMnode, pTrans)) != 0) {
×
1179
    mError("trans:%d, failed to pre-finish since %s", pTrans->id, tstrerror(code));
×
1180
    TAOS_RETURN(code);
×
1181
  }
1182
  mInfo("trans:%d, pre-finish finished", pTrans->id);
×
1183
  TAOS_RETURN(code);
×
1184
}
1185

1186
static void mndTransSendRpcRsp(SMnode *pMnode, STrans *pTrans) {
5,561✔
1187
  bool    sendRsp = false;
5,561✔
1188
  int32_t code = pTrans->code;
5,561✔
1189

1190
  if (pTrans->stage == TRN_STAGE_FINISH) {
5,561✔
1191
    sendRsp = true;
2,378✔
1192
  }
1193

1194
  if (pTrans->policy == TRN_POLICY_ROLLBACK) {
5,561✔
1195
    if (pTrans->stage == TRN_STAGE_UNDO_ACTION || pTrans->stage == TRN_STAGE_ROLLBACK) {
1,662!
1196
      if (code == 0) code = TSDB_CODE_MND_TRANS_UNKNOW_ERROR;
×
1197
      sendRsp = true;
×
1198
    }
1199
  } else {
1200
    if (pTrans->stage == TRN_STAGE_REDO_ACTION) {
3,899✔
1201
      if (code == TSDB_CODE_SYN_NOT_LEADER || code == TSDB_CODE_SYN_RESTORING || code == TSDB_CODE_APP_IS_STARTING ||
2,437!
1202
          code == TSDB_CODE_SYN_PROPOSE_NOT_READY) {
1203
        if (pTrans->failedTimes > 60) sendRsp = true;
×
1204
      } else {
1205
        if (pTrans->failedTimes > 6) sendRsp = true;
2,437!
1206
      }
1207
      if (code == 0) code = TSDB_CODE_MND_TRANS_UNKNOW_ERROR;
2,437✔
1208
    }
1209
  }
1210

1211
  if (!sendRsp) {
5,561✔
1212
    return;
3,183✔
1213
  } else {
1214
    mInfo("vgId:1, trans:%d, start to send rsp, stage:%s failedTimes:%d code:0x%x", pTrans->id,
2,378!
1215
          mndTransStr(pTrans->stage), pTrans->failedTimes, code);
1216
  }
1217

1218
  mInfo("vgId:1, trans:%d, start to lock rpc array", pTrans->id);
2,378!
1219
  taosWLockLatch(&pTrans->lockRpcArray);
2,378✔
1220
  int32_t size = taosArrayGetSize(pTrans->pRpcArray);
2,378✔
1221
  if (size <= 0) {
2,378✔
1222
    taosWUnLockLatch(&pTrans->lockRpcArray);
1,704✔
1223
    return;
1,704✔
1224
  }
1225

1226
  for (int32_t i = 0; i < size; ++i) {
1,348✔
1227
    SRpcHandleInfo *pInfo = taosArrayGet(pTrans->pRpcArray, i);
674✔
1228
    if (pInfo->handle != NULL) {
674✔
1229
      if (code == TSDB_CODE_RPC_NETWORK_UNAVAIL || code == TSDB_CODE_RPC_SOMENODE_NOT_CONNECTED) {
530!
1230
        code = TSDB_CODE_MND_TRANS_NETWORK_UNAVAILL;
×
1231
      }
1232
      if (code == TSDB_CODE_SYN_TIMEOUT) {
530!
1233
        code = TSDB_CODE_MND_TRANS_SYNC_TIMEOUT;
×
1234
      }
1235

1236
      if (i != 0 && code == 0) {
530!
1237
        code = TSDB_CODE_MNODE_NOT_FOUND;
×
1238
      }
1239
      mInfo("vgId:1, trans:%d, client:%d start to send rsp, code:0x%x stage:%s app:%p", pTrans->id, i, code,
530!
1240
            mndTransStr(pTrans->stage), pInfo->ahandle);
1241

1242
      SRpcMsg rspMsg = {.code = code, .info = *pInfo};
530✔
1243

1244
      if (pTrans->originRpcType == TDMT_MND_CREATE_DB) {
530✔
1245
        mInfo("trans:%d, origin msgtype:%s", pTrans->id, TMSG_INFO(pTrans->originRpcType));
74!
1246
        SDbObj *pDb = mndAcquireDb(pMnode, pTrans->dbname);
74✔
1247
        if (pDb != NULL) {
74!
1248
          for (int32_t j = 0; j < 12; j++) {
96!
1249
            bool ready = mndIsDbReady(pMnode, pDb);
96✔
1250
            if (!ready) {
96✔
1251
              mInfo("trans:%d, db:%s not ready yet, wait %d times", pTrans->id, pTrans->dbname, j);
22!
1252
              taosMsleep(1000);
22✔
1253
            } else {
1254
              break;
74✔
1255
            }
1256
          }
1257
        }
1258
        mndReleaseDb(pMnode, pDb);
74✔
1259
      } else if (pTrans->originRpcType == TDMT_MND_CREATE_STB) {
456✔
1260
        void   *pCont = NULL;
91✔
1261
        int32_t contLen = 0;
91✔
1262
        if (0 == mndBuildSMCreateStbRsp(pMnode, pTrans->dbname, pTrans->stbname, &pCont, &contLen)) {
91!
1263
          mndTransSetRpcRsp(pTrans, pCont, contLen);
91✔
1264
        }
1265
      } else if (pTrans->originRpcType == TDMT_MND_CREATE_INDEX) {
365✔
1266
        void   *pCont = NULL;
14✔
1267
        int32_t contLen = 0;
14✔
1268
        if (0 == mndBuildSMCreateStbRsp(pMnode, pTrans->dbname, pTrans->stbname, &pCont, &contLen)) {
14!
1269
          mndTransSetRpcRsp(pTrans, pCont, contLen);
14✔
1270
        }
1271
      }
1272

1273
      if (pTrans->rpcRspLen != 0) {
530✔
1274
        void *rpcCont = rpcMallocCont(pTrans->rpcRspLen);
153✔
1275
        if (rpcCont != NULL) {
153!
1276
          memcpy(rpcCont, pTrans->rpcRsp, pTrans->rpcRspLen);
153✔
1277
          rspMsg.pCont = rpcCont;
153✔
1278
          rspMsg.contLen = pTrans->rpcRspLen;
153✔
1279
        }
1280
      }
1281

1282
      tmsgSendRsp(&rspMsg);
530✔
1283

1284
      mInfo("vgId:1, trans:%d, client:%d send rsp finished, code:0x%x stage:%s app:%p", pTrans->id, i, code,
530!
1285
            mndTransStr(pTrans->stage), pInfo->ahandle);
1286
    }
1287
  }
1288
  taosArrayClear(pTrans->pRpcArray);
674✔
1289
  taosWUnLockLatch(&pTrans->lockRpcArray);
674✔
1290
}
1291

1292
int32_t mndTransProcessRsp(SRpcMsg *pRsp) {
1,393✔
1293
  int32_t code = 0;
1,393✔
1294
  SMnode *pMnode = pRsp->info.node;
1,393✔
1295
#ifndef TD_ASTRA_32
1296
  int64_t signature = (int64_t)(pRsp->info.ahandle);
1,393✔
1297
  int32_t transId = (int32_t)(signature >> 32);
1,393✔
1298
  int32_t action = (int32_t)((signature << 32) >> 32);
1,393✔
1299
#else
1300
  int32_t transId = (int32_t)(pRsp->info.ahandle);
1301
  int32_t action = (int32_t)(pRsp->info.ahandleEx);
1302
#endif
1303
  STrans *pTrans = mndAcquireTrans(pMnode, transId);
1,393✔
1304
  if (pTrans == NULL) {
1,393!
1305
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
1306
    if (terrno != 0) code = terrno;
×
1307
    mError("trans:%d, failed to get transId from vnode rsp since %s", transId, tstrerror(code));
×
1308
    goto _OVER;
×
1309
  }
1310

1311
  SArray *pArray = NULL;
1,393✔
1312
  if (pTrans->stage == TRN_STAGE_REDO_ACTION) {
1,393!
1313
    pArray = pTrans->redoActions;
1,393✔
1314
  } else if (pTrans->stage == TRN_STAGE_UNDO_ACTION) {
×
1315
    pArray = pTrans->undoActions;
×
1316
  } else {
1317
    mError("trans:%d, invalid trans stage:%d while recv action rsp", pTrans->id, pTrans->stage);
×
1318
    goto _OVER;
×
1319
  }
1320

1321
  if (pArray == NULL) {
1,393!
1322
    mError("trans:%d, invalid trans stage:%d", transId, pTrans->stage);
×
1323
    goto _OVER;
×
1324
  }
1325

1326
  int32_t actionNum = taosArrayGetSize(pArray);
1,393✔
1327
  if (action < 0 || action >= actionNum) {
1,393!
1328
    mError("trans:%d, invalid action:%d", transId, action);
×
1329
    goto _OVER;
×
1330
  }
1331

1332
  STransAction *pAction = taosArrayGet(pArray, action);
1,393✔
1333
  if (pAction != NULL) {
1,393!
1334
    pAction->msgReceived = 1;
1,393✔
1335
    pAction->errCode = pRsp->code;
1,393✔
1336
    pAction->endTime = taosGetTimestampMs();
1,393✔
1337

1338
    // pTrans->lastErrorNo = pRsp->code;
1339
    mndSetTransLastAction(pTrans, pAction);
1,393✔
1340

1341
    mInfo("trans:%d, %s:%d response is received, received code:0x%x(%s), accept:0x%x(%s) retry:0x%x(%s)", transId,
1,393!
1342
          mndTransStr(pAction->stage), action, pRsp->code, tstrerror(pRsp->code), pAction->acceptableCode,
1343
          tstrerror(pAction->acceptableCode), pAction->retryCode, tstrerror(pAction->retryCode));
1344
  } else {
1345
    mInfo("trans:%d, invalid action, index:%d, code:0x%x", transId, action, pRsp->code);
×
1346
  }
1347

1348
  mndTransExecute(pMnode, pTrans);
1,393✔
1349

1350
_OVER:
1,393✔
1351
  mndReleaseTrans(pMnode, pTrans);
1,393✔
1352
  TAOS_RETURN(code);
1,393✔
1353
}
1354

1355
static void mndTransResetAction(SMnode *pMnode, STrans *pTrans, STransAction *pAction) {
109✔
1356
  pAction->rawWritten = 0;
109✔
1357
  pAction->msgSent = 0;
109✔
1358
  pAction->msgReceived = 0;
109✔
1359
  if (pAction->errCode == TSDB_CODE_SYN_NEW_CONFIG_ERROR || pAction->errCode == TSDB_CODE_SYN_INTERNAL_ERROR ||
109!
1360
      pAction->errCode == TSDB_CODE_SYN_NOT_LEADER) {
109!
1361
    pAction->epSet.inUse = (pAction->epSet.inUse + 1) % pAction->epSet.numOfEps;
×
1362
    mInfo("trans:%d, %s:%d execute status is reset and set epset inuse:%d", pTrans->id, mndTransStr(pAction->stage),
×
1363
          pAction->id, pAction->epSet.inUse);
1364
  } else {
1365
    mInfo("trans:%d, %s:%d execute status is reset", pTrans->id, mndTransStr(pAction->stage), pAction->id);
109!
1366
  }
1367
  pAction->errCode = 0;
109✔
1368
}
109✔
1369

1370
static void mndTransResetActions(SMnode *pMnode, STrans *pTrans, SArray *pArray) {
×
1371
  int32_t numOfActions = taosArrayGetSize(pArray);
×
1372

1373
  for (int32_t action = 0; action < numOfActions; ++action) {
×
1374
    STransAction *pAction = taosArrayGet(pArray, action);
×
1375
    if (pAction->msgSent && pAction->msgReceived &&
×
1376
        (pAction->errCode == 0 || pAction->errCode == pAction->acceptableCode))
×
1377
      continue;
×
1378
    if (pAction->rawWritten && (pAction->errCode == 0 || pAction->errCode == pAction->acceptableCode)) continue;
×
1379

1380
    mndTransResetAction(pMnode, pTrans, pAction);
×
1381
  }
1382
}
×
1383

1384
// execute in sync context
1385
static int32_t mndTransWriteSingleLog(SMnode *pMnode, STrans *pTrans, STransAction *pAction, bool topHalf) {
11,067✔
1386
  if (pAction->rawWritten) return 0;
11,067✔
1387
  if (topHalf) {
6,078!
1388
    TAOS_RETURN(TSDB_CODE_MND_TRANS_CTX_SWITCH);
×
1389
  }
1390

1391
  if (pAction->pRaw->type >= SDB_MAX) {
6,078!
1392
    pAction->rawWritten = true;
×
1393
    pAction->errCode = 0;
×
1394
    mndSetTransLastAction(pTrans, pAction);
×
1395
    mInfo("skip sdb raw type:%d since it is not supported", pAction->pRaw->type);
×
1396
    TAOS_RETURN(TSDB_CODE_SUCCESS);
×
1397
  }
1398

1399
  int32_t code = sdbWriteWithoutFree(pMnode->pSdb, pAction->pRaw);
6,078✔
1400
  if (code == 0 || terrno == TSDB_CODE_SDB_OBJ_NOT_THERE) {
6,078!
1401
    pAction->rawWritten = true;
6,078✔
1402
    pAction->errCode = 0;
6,078✔
1403
    code = 0;
6,078✔
1404
    mInfo("trans:%d, %s:%d write to sdb, type:%s status:%s", pTrans->id, mndTransStr(pAction->stage), pAction->id,
6,078!
1405
          sdbTableName(pAction->pRaw->type), sdbStatusName(pAction->pRaw->status));
1406

1407
    mndSetTransLastAction(pTrans, pAction);
6,078✔
1408
  } else {
1409
    pAction->errCode = (terrno != 0) ? terrno : code;
×
1410
    mError("trans:%d, %s:%d failed to write sdb since %s, type:%s status:%s", pTrans->id, mndTransStr(pAction->stage),
×
1411
           pAction->id, tstrerror(code), sdbTableName(pAction->pRaw->type), sdbStatusName(pAction->pRaw->status));
1412
    mndSetTransLastAction(pTrans, pAction);
×
1413
  }
1414

1415
  TAOS_RETURN(code);
6,078✔
1416
}
1417

1418
// execute in trans context
1419
static int32_t mndTransSendSingleMsg(SMnode *pMnode, STrans *pTrans, STransAction *pAction, bool topHalf) {
8,251✔
1420
  if (pAction->msgSent) return 0;
8,251✔
1421
  if (mndCannotExecuteTrans(pMnode, topHalf)) {
2,088✔
1422
    TAOS_RETURN(TSDB_CODE_MND_TRANS_CTX_SWITCH);
695✔
1423
  }
1424

1425
#ifndef TD_ASTRA_32
1426
  int64_t signature = pTrans->id;
1,393✔
1427
  signature = (signature << 32);
1,393✔
1428
  signature += pAction->id;
1,393✔
1429

1430
  SRpcMsg rpcMsg = {.msgType = pAction->msgType, .contLen = pAction->contLen, .info.ahandle = (void *)signature};
1,393✔
1431
#else
1432
  SRpcMsg rpcMsg = {.msgType = pAction->msgType,
1433
                    .contLen = pAction->contLen,
1434
                    .info.ahandle = (void *)pTrans->id,
1435
                    .info.ahandleEx = (void *)pAction->id};
1436
#endif
1437
  rpcMsg.pCont = rpcMallocCont(pAction->contLen);
1,393✔
1438
  if (rpcMsg.pCont == NULL) {
1,393!
1439
    terrno = TSDB_CODE_OUT_OF_MEMORY;
×
1440
    return -1;
×
1441
  }
1442
  rpcMsg.info.traceId.rootId = pTrans->mTraceId;
1,393✔
1443
  rpcMsg.info.notFreeAhandle = 1;
1,393✔
1444

1445
  memcpy(rpcMsg.pCont, pAction->pCont, pAction->contLen);
1,393✔
1446

1447
  char    detail[1024] = {0};
1,393✔
1448
  int32_t len = tsnprintf(detail, sizeof(detail), "msgType:%s numOfEps:%d inUse:%d", TMSG_INFO(pAction->msgType),
1,393!
1449
                          pAction->epSet.numOfEps, pAction->epSet.inUse);
1,393✔
1450
  for (int32_t i = 0; i < pAction->epSet.numOfEps; ++i) {
3,609✔
1451
    len += tsnprintf(detail + len, sizeof(detail) - len, " ep:%d-%s:%u", i, pAction->epSet.eps[i].fqdn,
2,216✔
1452
                     pAction->epSet.eps[i].port);
2,216✔
1453
  }
1454

1455
  int32_t code = tmsgSendReq(&pAction->epSet, &rpcMsg);
1,393✔
1456
  if (code == 0) {
1,393!
1457
    pAction->msgSent = 1;
1,393✔
1458
    // pAction->msgReceived = 0;
1459
    pAction->errCode = TSDB_CODE_ACTION_IN_PROGRESS;
1,393✔
1460
    pAction->startTime = taosGetTimestampMs();
1,393✔
1461
    pAction->endTime = 0;
1,393✔
1462
    mInfo("trans:%d, %s:%d is sent, %s", pTrans->id, mndTransStr(pAction->stage), pAction->id, detail);
1,393!
1463

1464
    mndSetTransLastAction(pTrans, pAction);
1,393✔
1465
  } else {
1466
    pAction->msgSent = 0;
×
1467
    pAction->msgReceived = 0;
×
1468
    pAction->errCode = (terrno != 0) ? terrno : code;
×
1469
    mError("trans:%d, %s:%d not send since %s, %s", pTrans->id, mndTransStr(pAction->stage), pAction->id, terrstr(),
×
1470
           detail);
1471

1472
    mndSetTransLastAction(pTrans, pAction);
×
1473
  }
1474

1475
  TAOS_RETURN(code);
1,393✔
1476
}
1477

1478
static int32_t mndTransExecNullMsg(SMnode *pMnode, STrans *pTrans, STransAction *pAction, bool topHalf) {
×
1479
  if (!topHalf) return TSDB_CODE_MND_TRANS_CTX_SWITCH;
×
1480
  pAction->rawWritten = 0;
×
1481
  pAction->errCode = 0;
×
1482
  mInfo("trans:%d, %s:%d confirm action executed", pTrans->id, mndTransStr(pAction->stage), pAction->id);
×
1483

1484
  mndSetTransLastAction(pTrans, pAction);
×
1485
  return 0;
×
1486
}
1487

1488
static int32_t mndTransExecSingleAction(SMnode *pMnode, STrans *pTrans, STransAction *pAction, bool topHalf) {
19,318✔
1489
  if (pAction->actionType == TRANS_ACTION_RAW) {
19,318✔
1490
    return mndTransWriteSingleLog(pMnode, pTrans, pAction, topHalf);
11,067✔
1491
  } else if (pAction->actionType == TRANS_ACTION_MSG) {
8,251!
1492
    return mndTransSendSingleMsg(pMnode, pTrans, pAction, topHalf);
8,251✔
1493
  } else {
1494
    return mndTransExecNullMsg(pMnode, pTrans, pAction, topHalf);
×
1495
  }
1496
}
1497

1498
static int32_t mndTransExecSingleActions(SMnode *pMnode, STrans *pTrans, SArray *pArray, bool topHalf) {
4,596✔
1499
  int32_t numOfActions = taosArrayGetSize(pArray);
4,596✔
1500
  int32_t code = 0;
4,596✔
1501

1502
  for (int32_t action = 0; action < numOfActions; ++action) {
22,145✔
1503
    STransAction *pAction = taosArrayGet(pArray, action);
17,955✔
1504
    code = mndTransExecSingleAction(pMnode, pTrans, pAction, topHalf);
17,955✔
1505
    if (code != 0) {
17,955✔
1506
      mInfo("trans:%d, action:%d not executed since %s. numOfActions:%d", pTrans->id, action, tstrerror(code),
406!
1507
            numOfActions);
1508
      break;
406✔
1509
    }
1510
  }
1511

1512
  return code;
4,596✔
1513
}
1514

1515
static int32_t mndTransExecuteActions(SMnode *pMnode, STrans *pTrans, SArray *pArray, bool topHalf) {
5,878✔
1516
  int32_t numOfActions = taosArrayGetSize(pArray);
5,878✔
1517
  int32_t code = 0;
5,878✔
1518
  if (numOfActions == 0) return 0;
5,878✔
1519

1520
  if ((code = mndTransExecSingleActions(pMnode, pTrans, pArray, topHalf)) != 0) {
4,596✔
1521
    return code;
406✔
1522
  }
1523

1524
  int32_t       numOfExecuted = 0;
4,190✔
1525
  int32_t       errCode = 0;
4,190✔
1526
  STransAction *pErrAction = NULL;
4,190✔
1527
  for (int32_t action = 0; action < numOfActions; ++action) {
21,717✔
1528
    STransAction *pAction = taosArrayGet(pArray, action);
17,527✔
1529
    if (pAction->msgReceived || pAction->rawWritten) {
17,527✔
1530
      numOfExecuted++;
13,988✔
1531
      if (pAction->errCode != 0 && pAction->errCode != pAction->acceptableCode) {
13,988!
1532
        errCode = pAction->errCode;
×
1533
        pErrAction = pAction;
×
1534
      }
1535
    } else {
1536
      pErrAction = pAction;
3,539✔
1537
    }
1538
  }
1539

1540
  mndSetTransLastAction(pTrans, pErrAction);
4,190✔
1541

1542
  if (numOfExecuted == numOfActions) {
4,190✔
1543
    if (errCode == 0) {
2,965!
1544
      mInfo("trans:%d, all %d actions execute successfully", pTrans->id, numOfActions);
2,965!
1545
      return 0;
2,965✔
1546
    } else {
1547
      mError("trans:%d, all %d actions executed, code:0x%x", pTrans->id, numOfActions, errCode & 0XFFFF);
×
1548
      mndTransResetActions(pMnode, pTrans, pArray);
×
1549
      terrno = errCode;
×
1550
      return errCode;
×
1551
    }
1552
  } else {
1553
    mInfo("trans:%d, %d of %d actions executed", pTrans->id, numOfExecuted, numOfActions);
1,225!
1554

1555
    for (int32_t action = 0; action < numOfActions; ++action) {
7,068✔
1556
      STransAction *pAction = taosArrayGet(pArray, action);
5,843✔
1557
      mDebug("trans:%d, %s:%d Sent:%d, Received:%d, errCode:0x%x, acceptableCode:0x%x, retryCode:0x%x", pTrans->id,
5,843✔
1558
             mndTransStr(pAction->stage), pAction->id, pAction->msgSent, pAction->msgReceived, pAction->errCode,
1559
             pAction->acceptableCode, pAction->retryCode);
1560
      if (pAction->msgSent) {
5,843!
1561
        if (pAction->msgReceived) {
5,843✔
1562
          if (pAction->errCode != 0 && pAction->errCode != pAction->acceptableCode) {
2,304!
1563
            mndTransResetAction(pMnode, pTrans, pAction);
×
1564
            mInfo("trans:%d, %s:%d reset", pTrans->id, mndTransStr(pAction->stage), pAction->id);
×
1565
          }
1566
        }
1567
      }
1568
    }
1569
    return TSDB_CODE_ACTION_IN_PROGRESS;
1,225✔
1570
  }
1571
}
1572

1573
static int32_t mndTransExecuteRedoActions(SMnode *pMnode, STrans *pTrans, bool topHalf) {
3,500✔
1574
  int32_t code = mndTransExecuteActions(pMnode, pTrans, pTrans->redoActions, topHalf);
3,500✔
1575
  if (code != 0 && code != TSDB_CODE_ACTION_IN_PROGRESS && code != TSDB_CODE_MND_TRANS_CTX_SWITCH) {
3,500!
1576
    mError("trans:%d, failed to execute redoActions since:%s, code:0x%x, in %s", pTrans->id, terrstr(), terrno,
×
1577
           mndStrExecutionContext(topHalf));
1578
  }
1579
  return code;
3,500✔
1580
}
1581

1582
static int32_t mndTransExecuteUndoActions(SMnode *pMnode, STrans *pTrans, bool topHalf) {
×
1583
  int32_t code = mndTransExecuteActions(pMnode, pTrans, pTrans->undoActions, topHalf);
×
1584
  if (code != 0 && code != TSDB_CODE_ACTION_IN_PROGRESS && code != TSDB_CODE_MND_TRANS_CTX_SWITCH) {
×
1585
    mError("trans:%d, failed to execute undoActions since %s. in %s", pTrans->id, terrstr(),
×
1586
           mndStrExecutionContext(topHalf));
1587
  }
1588
  return code;
×
1589
}
1590

1591
static int32_t mndTransExecuteCommitActions(SMnode *pMnode, STrans *pTrans, bool topHalf) {
2,378✔
1592
  int32_t code = mndTransExecuteActions(pMnode, pTrans, pTrans->commitActions, topHalf);
2,378✔
1593
  if (code != 0 && code != TSDB_CODE_ACTION_IN_PROGRESS && code != TSDB_CODE_MND_TRANS_CTX_SWITCH) {
2,378!
1594
    mError("trans:%d, failed to execute commitActions since %s. in %s", pTrans->id, terrstr(),
×
1595
           mndStrExecutionContext(topHalf));
1596
  }
1597
  return code;
2,378✔
1598
}
1599

1600
static int32_t mndTransExecuteActionsSerial(SMnode *pMnode, STrans *pTrans, SArray *pActions, bool topHalf) {
787✔
1601
  int32_t code = 0;
787✔
1602
  int32_t numOfActions = taosArrayGetSize(pActions);
787✔
1603
  if (numOfActions == 0) return code;
787!
1604

1605
  if (pTrans->actionPos >= numOfActions) {
787✔
1606
    return code;
59✔
1607
  }
1608

1609
  mInfo("trans:%d, execute %d actions serial, begin at action:%d, stage:%s", pTrans->id, numOfActions,
728!
1610
        pTrans->actionPos, mndTransStr(pTrans->stage));
1611

1612
  for (int32_t action = pTrans->actionPos; action < numOfActions; ++action) {
1,006✔
1613
    STransAction *pAction = taosArrayGet(pActions, action);
961✔
1614

1615
    mInfo("trans:%d, current action:%d, stage:%s, actionType(1:msg,2:log):%d, msgSent:%d, msgReceived:%d", 
961!
1616
          pTrans->id, pTrans->actionPos, mndTransStr(pAction->stage), pAction->actionType, pAction->msgSent,
1617
          pAction->msgReceived);
1618

1619
    code = mndTransExecSingleAction(pMnode, pTrans, pAction, topHalf);
961✔
1620
    if (code == 0) {
961✔
1621
      if (pAction->msgSent) {
672✔
1622
        if (pAction->msgReceived) {
567✔
1623
          if (pAction->errCode != 0 && pAction->errCode != pAction->acceptableCode) {
247!
1624
            code = pAction->errCode;
109✔
1625
            mndTransResetAction(pMnode, pTrans, pAction);
109✔
1626
          } else {
1627
            mInfo("trans:%d, %s:%d execute successfully", pTrans->id, mndTransStr(pAction->stage), action);
138!
1628
          }
1629
        } else {
1630
          code = TSDB_CODE_ACTION_IN_PROGRESS;
320✔
1631
        }
1632
      } else if (pAction->rawWritten) {
105!
1633
        if (pAction->errCode != 0 && pAction->errCode != pAction->acceptableCode) {
105!
1634
          code = pAction->errCode;
×
1635
        } else {
1636
          mInfo("trans:%d, %s:%d write successfully", pTrans->id, mndTransStr(pAction->stage), action);
105!
1637
        }
1638
      } else {
1639
      }
1640
    }
1641

1642
    if (code == 0) {
961✔
1643
      pTrans->failedTimes = 0;
243✔
1644
    }
1645
    mndSetTransLastAction(pTrans, pAction);
961✔
1646

1647
    if (mndCannotExecuteTrans(pMnode, topHalf)) {
961✔
1648
      pTrans->lastErrorNo = code;
363✔
1649
      pTrans->code = code;
363✔
1650
      mInfo("trans:%d, %s:%d, cannot execute next action in %s, code:%s", pTrans->id, mndTransStr(pAction->stage),
363!
1651
            action, mndStrExecutionContext(topHalf), tstrerror(code));
1652
      break;
363✔
1653
    }
1654

1655
    if (code == 0) {
598✔
1656
      pTrans->code = 0;
169✔
1657
      pTrans->actionPos++;
169✔
1658
      mInfo("trans:%d, %s:%d is executed and need sync to other mnodes", pTrans->id, mndTransStr(pAction->stage),
169!
1659
            pAction->id);
1660
      (void)taosThreadMutexUnlock(&pTrans->mutex);
169✔
1661
      code = mndTransSync(pMnode, pTrans);
169✔
1662
      (void)taosThreadMutexLock(&pTrans->mutex);
169✔
1663
      if (code != 0) {
169!
1664
        pTrans->actionPos--;
×
1665
        pTrans->code = terrno;
×
1666
        mError("trans:%d, %s:%d is executed and failed to sync to other mnodes since %s", pTrans->id,
×
1667
               mndTransStr(pAction->stage), pAction->id, terrstr());
1668
        break;
×
1669
      }
1670
    } else if (code == TSDB_CODE_ACTION_IN_PROGRESS) {
429✔
1671
      mInfo("trans:%d, %s:%d is in progress and wait it finish", pTrans->id, mndTransStr(pAction->stage), pAction->id);
320!
1672
      break;
320✔
1673
    } else if (code == pAction->retryCode || code == TSDB_CODE_SYN_PROPOSE_NOT_READY ||
109!
1674
               code == TSDB_CODE_SYN_RESTORING || code == TSDB_CODE_SYN_NOT_LEADER) {
×
1675
      mInfo("trans:%d, %s:%d receive code:0x%x(%s) and retry", pTrans->id, mndTransStr(pAction->stage), pAction->id,
109!
1676
            code, tstrerror(code));
1677
      pTrans->lastErrorNo = code;
109✔
1678
      taosMsleep(300);
109✔
1679
      action--;
109✔
1680
      continue;
109✔
1681
    } else {
1682
      terrno = code;
×
1683
      pTrans->lastErrorNo = code;
×
1684
      pTrans->code = code;
×
1685
      mInfo("trans:%d, %s:%d receive code:0x%x(%s) and wait another schedule, failedTimes:%d", pTrans->id,
×
1686
            mndTransStr(pAction->stage), pAction->id, code, tstrerror(code), pTrans->failedTimes);
1687
      break;
×
1688
    }
1689
  }
1690

1691
  return code;
728✔
1692
}
1693

1694
static int32_t mndTransExecuteRedoActionsSerial(SMnode *pMnode, STrans *pTrans, bool topHalf) {
787✔
1695
  int32_t code = TSDB_CODE_ACTION_IN_PROGRESS;
787✔
1696
  (void)taosThreadMutexLock(&pTrans->mutex);
787✔
1697
  if (pTrans->stage == TRN_STAGE_REDO_ACTION) {
787!
1698
    code = mndTransExecuteActionsSerial(pMnode, pTrans, pTrans->redoActions, topHalf);
787✔
1699
  }
1700
  (void)taosThreadMutexUnlock(&pTrans->mutex);
787✔
1701
  return code;
787✔
1702
}
1703

1704
static int32_t mndTransExecuteUndoActionsSerial(SMnode *pMnode, STrans *pTrans, bool topHalf) {
×
1705
  int32_t code = TSDB_CODE_ACTION_IN_PROGRESS;
×
1706
  (void)taosThreadMutexLock(&pTrans->mutex);
×
1707
  if (pTrans->stage == TRN_STAGE_UNDO_ACTION) {
×
1708
    code = mndTransExecuteActionsSerial(pMnode, pTrans, pTrans->undoActions, topHalf);
×
1709
  }
1710
  (void)taosThreadMutexUnlock(&pTrans->mutex);
×
1711
  return code;
×
1712
}
1713

1714
bool mndTransPerformPrepareStage(SMnode *pMnode, STrans *pTrans, bool topHalf) {
1,276✔
1715
  bool    continueExec = true;
1,276✔
1716
  int32_t code = 0;
1,276✔
1717
  terrno = 0;
1,276✔
1718

1719
  int32_t numOfActions = taosArrayGetSize(pTrans->prepareActions);
1,276✔
1720
  if (numOfActions == 0) goto _OVER;
1,276✔
1721

1722
  mInfo("trans:%d, execute %d prepare actions.", pTrans->id, numOfActions);
192!
1723

1724
  for (int32_t action = 0; action < numOfActions; ++action) {
594✔
1725
    STransAction *pAction = taosArrayGet(pTrans->prepareActions, action);
402✔
1726
    code = mndTransExecSingleAction(pMnode, pTrans, pAction, topHalf);
402✔
1727
    if (code != 0) {
402!
1728
      terrno = code;
×
1729
      mError("trans:%d, failed to execute prepare action:%d, numOfActions:%d, since %s", pTrans->id, action,
×
1730
             numOfActions, tstrerror(code));
1731
      return false;
×
1732
    }
1733
  }
1734

1735
_OVER:
192✔
1736
  pTrans->stage = TRN_STAGE_REDO_ACTION;
1,276✔
1737
  mInfo("trans:%d, stage from prepare to redoAction", pTrans->id);
1,276!
1738
  return continueExec;
1,276✔
1739
}
1740

1741
static bool mndTransPerformRedoActionStage(SMnode *pMnode, STrans *pTrans, bool topHalf) {
4,287✔
1742
  bool    continueExec = true;
4,287✔
1743
  int32_t code = 0;
4,287✔
1744
  terrno = 0;
4,287✔
1745

1746
  if (pTrans->exec == TRN_EXEC_SERIAL) {
4,287✔
1747
    code = mndTransExecuteRedoActionsSerial(pMnode, pTrans, topHalf);
787✔
1748
  } else {
1749
    code = mndTransExecuteRedoActions(pMnode, pTrans, topHalf);
3,500✔
1750
  }
1751

1752
  if (code != 0 && code != TSDB_CODE_MND_TRANS_CTX_SWITCH && mndTransIsInSyncContext(topHalf)) {
4,287!
1753
    pTrans->lastErrorNo = code;
×
1754
    pTrans->code = code;
×
1755
    mInfo(
×
1756
        "trans:%d, failed to execute, will retry redo action stage in 100 ms , in %s, "
1757
        "continueExec:%d, code:%s",
1758
        pTrans->id, mndStrExecutionContext(topHalf), continueExec, tstrerror(code));
1759
    taosMsleep(100);
×
1760
    return true;
×
1761
  } else {
1762
    if (mndCannotExecuteTrans(pMnode, topHalf)) {
4,287✔
1763
      mInfo("trans:%d, cannot continue to execute redo action stage in %s, continueExec:%d, code:%s", pTrans->id,
1,638!
1764
            mndStrExecutionContext(topHalf), continueExec, tstrerror(code));
1765
      return false;
1,638✔
1766
    }
1767
  }
1768
  terrno = code;
2,649✔
1769

1770
  if (code == 0) {
2,649✔
1771
    pTrans->code = 0;
1,104✔
1772
    pTrans->stage = TRN_STAGE_COMMIT;
1,104✔
1773
    mInfo("trans:%d, stage from redoAction to commit", pTrans->id);
1,104!
1774
    continueExec = true;
1,104✔
1775
  } else if (code == TSDB_CODE_ACTION_IN_PROGRESS || code == TSDB_CODE_MND_TRANS_CTX_SWITCH) {
1,545!
1776
    mInfo("trans:%d, stage keep on redoAction since %s", pTrans->id, tstrerror(code));
1,545!
1777
    continueExec = false;
1,545✔
1778
  } else {
1779
    pTrans->failedTimes++;
×
1780
    pTrans->code = terrno;
×
1781
    if (pTrans->policy == TRN_POLICY_ROLLBACK) {
×
1782
      if (pTrans->lastAction != 0) {
×
1783
        STransAction *pAction = taosArrayGet(pTrans->redoActions, pTrans->lastAction);
×
1784
        if (pAction->retryCode != 0 && pAction->retryCode == pAction->errCode) {
×
1785
          if (pTrans->failedTimes < 6) {
×
1786
            mError("trans:%d, stage keep on redoAction since action:%d code:0x%x not 0x%x, failedTimes:%d", pTrans->id,
×
1787
                   pTrans->lastAction, pTrans->code, pAction->retryCode, pTrans->failedTimes);
1788
            taosMsleep(1000);
×
1789
            continueExec = true;
×
1790
            return true;
×
1791
          }
1792
        }
1793
      }
1794

1795
      pTrans->stage = TRN_STAGE_ROLLBACK;
×
1796
      pTrans->actionPos = 0;
×
1797
      mError("trans:%d, stage from redoAction to rollback since %s, and set actionPos to %d", pTrans->id, terrstr(),
×
1798
             pTrans->actionPos);
1799
      continueExec = true;
×
1800
    } else {
1801
      mError("trans:%d, stage keep on redoAction since %s, failedTimes:%d", pTrans->id, terrstr(), pTrans->failedTimes);
×
1802
      continueExec = false;
×
1803
    }
1804
  }
1805

1806
  return continueExec;
2,649✔
1807
}
1808

1809
// execute in trans context
1810
static bool mndTransPerformCommitStage(SMnode *pMnode, STrans *pTrans, bool topHalf) {
1,104✔
1811
  if (mndCannotExecuteTrans(pMnode, topHalf)) return false;
1,104!
1812

1813
  bool    continueExec = true;
1,104✔
1814
  int32_t code = mndTransCommit(pMnode, pTrans);
1,104✔
1815

1816
  if (code == 0) {
1,104!
1817
    pTrans->code = 0;
1,104✔
1818
    pTrans->stage = TRN_STAGE_COMMIT_ACTION;
1,104✔
1819
    mInfo("trans:%d, stage from commit to commitAction", pTrans->id);
1,104!
1820
    continueExec = true;
1,104✔
1821
  } else {
1822
    pTrans->code = terrno;
×
1823
    pTrans->failedTimes++;
×
1824
    mError("trans:%d, stage keep on commit since %s, failedTimes:%d", pTrans->id, terrstr(), pTrans->failedTimes);
×
1825
    continueExec = false;
×
1826
  }
1827

1828
  return continueExec;
1,104✔
1829
}
1830

1831
static bool mndTransPerformCommitActionStage(SMnode *pMnode, STrans *pTrans, bool topHalf) {
2,378✔
1832
  bool    continueExec = true;
2,378✔
1833
  int32_t code = mndTransExecuteCommitActions(pMnode, pTrans, topHalf);
2,378✔
1834

1835
  if (code == 0) {
2,378!
1836
    pTrans->code = 0;
2,378✔
1837
    pTrans->stage = TRN_STAGE_FINISH;  // TRN_STAGE_PRE_FINISH is not necessary
2,378✔
1838
    mInfo("trans:%d, stage from commitAction to finished", pTrans->id);
2,378!
1839
    continueExec = true;
2,378✔
1840
  } else if (code == TSDB_CODE_MND_TRANS_CTX_SWITCH && topHalf) {
×
1841
    pTrans->code = 0;
×
1842
    pTrans->stage = TRN_STAGE_COMMIT;
×
1843
    mInfo("trans:%d, back to commit stage", pTrans->id);
×
1844
    continueExec = true;
×
1845
  } else {
1846
    pTrans->code = terrno;
×
1847
    pTrans->failedTimes++;
×
1848
    mError("trans:%d, stage keep on commitAction since %s, failedTimes:%d", pTrans->id, terrstr(), pTrans->failedTimes);
×
1849
    continueExec = false;
×
1850
  }
1851

1852
  return continueExec;
2,378✔
1853
}
1854

1855
static bool mndTransPerformUndoActionStage(SMnode *pMnode, STrans *pTrans, bool topHalf) {
×
1856
  bool    continueExec = true;
×
1857
  int32_t code = 0;
×
1858

1859
  if (pTrans->exec == TRN_EXEC_SERIAL) {
×
1860
    code = mndTransExecuteUndoActionsSerial(pMnode, pTrans, topHalf);
×
1861
  } else {
1862
    code = mndTransExecuteUndoActions(pMnode, pTrans, topHalf);
×
1863
  }
1864

1865
  if (mndCannotExecuteTrans(pMnode, topHalf)) return false;
×
1866
  terrno = code;
×
1867

1868
  if (code == 0) {
×
1869
    pTrans->stage = TRN_STAGE_PRE_FINISH;
×
1870
    mInfo("trans:%d, stage from undoAction to pre-finish", pTrans->id);
×
1871
    continueExec = true;
×
1872
  } else if (code == TSDB_CODE_ACTION_IN_PROGRESS || code == TSDB_CODE_MND_TRANS_CTX_SWITCH) {
×
1873
    mInfo("trans:%d, stage keep on undoAction since %s", pTrans->id, tstrerror(code));
×
1874
    continueExec = false;
×
1875
  } else {
1876
    pTrans->failedTimes++;
×
1877
    mError("trans:%d, stage keep on undoAction since %s, failedTimes:%d", pTrans->id, terrstr(), pTrans->failedTimes);
×
1878
    continueExec = false;
×
1879
  }
1880

1881
  return continueExec;
×
1882
}
1883

1884
// in trans context
1885
static bool mndTransPerformRollbackStage(SMnode *pMnode, STrans *pTrans, bool topHalf) {
×
1886
  if (mndCannotExecuteTrans(pMnode, topHalf)) return false;
×
1887

1888
  bool    continueExec = true;
×
1889
  int32_t code = mndTransRollback(pMnode, pTrans);
×
1890

1891
  if (code == 0) {
×
1892
    pTrans->stage = TRN_STAGE_UNDO_ACTION;
×
1893
    continueExec = true;
×
1894
  } else {
1895
    pTrans->failedTimes++;
×
1896
    mError("trans:%d, stage keep on rollback since %s, failedTimes:%d", pTrans->id, terrstr(), pTrans->failedTimes);
×
1897
    continueExec = false;
×
1898
  }
1899

1900
  return continueExec;
×
1901
}
1902

1903
// excute in trans context
1904
static bool mndTransPerformPreFinishStage(SMnode *pMnode, STrans *pTrans, bool topHalf) {
×
1905
  if (mndCannotExecuteTrans(pMnode, topHalf)) return false;
×
1906

1907
  bool    continueExec = true;
×
1908
  int32_t code = mndTransPreFinish(pMnode, pTrans);
×
1909

1910
  if (code == 0) {
×
1911
    pTrans->stage = TRN_STAGE_FINISH;
×
1912
    mInfo("trans:%d, stage from pre-finish to finish", pTrans->id);
×
1913
    continueExec = true;
×
1914
  } else {
1915
    pTrans->failedTimes++;
×
1916
    mError("trans:%d, stage keep on pre-finish since %s, failedTimes:%d", pTrans->id, terrstr(), pTrans->failedTimes);
×
1917
    continueExec = false;
×
1918
  }
1919

1920
  return continueExec;
×
1921
}
1922

1923
static bool mndTransPerformFinishStage(SMnode *pMnode, STrans *pTrans, bool topHalf) {
2,378✔
1924
  bool continueExec = false;
2,378✔
1925
  if (topHalf) return continueExec;
2,378✔
1926

1927
  SSdbRaw *pRaw = mndTransEncode(pTrans);
1,274✔
1928
  if (pRaw == NULL) {
1,274!
1929
    mError("trans:%d, failed to encode while finish trans since %s", pTrans->id, terrstr());
×
1930
    return false;
×
1931
  }
1932
  TAOS_CHECK_RETURN(sdbSetRawStatus(pRaw, SDB_STATUS_DROPPED));
1,274!
1933

1934
  int32_t code = sdbWrite(pMnode->pSdb, pRaw);
1,274✔
1935
  if (code != 0) {
1,274!
1936
    mError("trans:%d, failed to write sdb since %s", pTrans->id, terrstr());
×
1937
  }
1938

1939
  mInfo("trans:%d, execute finished, code:0x%x, failedTimes:%d createTime:%" PRId64, pTrans->id, pTrans->code,
1,274!
1940
        pTrans->failedTimes, pTrans->createdTime);
1941
  return continueExec;
1,274✔
1942
}
1943

1944
void mndTransExecuteImp(SMnode *pMnode, STrans *pTrans, bool topHalf) {
5,561✔
1945
  bool continueExec = true;
5,561✔
1946

1947
  while (continueExec) {
15,708✔
1948
    mInfo("trans:%d, continue to execute stage:%s in %s, createTime:%" PRId64, pTrans->id,
10,147!
1949
          mndTransStr(pTrans->stage), mndStrExecutionContext(topHalf), pTrans->createdTime);
1950
    pTrans->lastExecTime = taosGetTimestampMs();
10,147✔
1951
    switch (pTrans->stage) {
10,147!
1952
      case TRN_STAGE_PREPARE:
×
1953
        continueExec = mndTransPerformPrepareStage(pMnode, pTrans, topHalf);
×
1954
        break;
×
1955
      case TRN_STAGE_REDO_ACTION:
4,287✔
1956
        continueExec = mndTransPerformRedoActionStage(pMnode, pTrans, topHalf);
4,287✔
1957
        break;
4,287✔
1958
      case TRN_STAGE_COMMIT:
1,104✔
1959
        continueExec = mndTransPerformCommitStage(pMnode, pTrans, topHalf);
1,104✔
1960
        break;
1,104✔
1961
      case TRN_STAGE_COMMIT_ACTION:
2,378✔
1962
        continueExec = mndTransPerformCommitActionStage(pMnode, pTrans, topHalf);
2,378✔
1963
        break;
2,378✔
1964
      case TRN_STAGE_ROLLBACK:
×
1965
        continueExec = mndTransPerformRollbackStage(pMnode, pTrans, topHalf);
×
1966
        break;
×
1967
      case TRN_STAGE_UNDO_ACTION:
×
1968
        continueExec = mndTransPerformUndoActionStage(pMnode, pTrans, topHalf);
×
1969
        break;
×
1970
      case TRN_STAGE_PRE_FINISH:
×
1971
        continueExec = mndTransPerformPreFinishStage(pMnode, pTrans, topHalf);
×
1972
        break;
×
1973
      case TRN_STAGE_FINISH:
2,378✔
1974
        continueExec = mndTransPerformFinishStage(pMnode, pTrans, topHalf);
2,378✔
1975
        break;
2,378✔
1976
      default:
×
1977
        continueExec = false;
×
1978
        break;
×
1979
    }
1980
  }
1981

1982
  mndTransSendRpcRsp(pMnode, pTrans);
5,561✔
1983
}
5,561✔
1984

1985
// start trans, pullup, receive rsp, kill
1986
void mndTransExecute(SMnode *pMnode, STrans *pTrans) {
2,652✔
1987
  bool topHalf = true;
2,652✔
1988
  mndTransExecuteImp(pMnode, pTrans, topHalf);
2,652✔
1989
}
2,652✔
1990

1991
// update trans
1992
void mndTransRefresh(SMnode *pMnode, STrans *pTrans) {
2,909✔
1993
  bool topHalf = false;
2,909✔
1994
  mndTransExecuteImp(pMnode, pTrans, topHalf);
2,909✔
1995
}
2,909✔
1996

1997
static int32_t mndProcessTransTimer(SRpcMsg *pReq) {
842✔
1998
  mTrace("start to process trans timer");
842✔
1999
  mndTransPullup(pReq->info.node);
842✔
2000
  return 0;
842✔
2001
}
2002

2003
int32_t mndKillTrans(SMnode *pMnode, STrans *pTrans) {
×
2004
  SArray *pArray = NULL;
×
2005
  if (pTrans->stage == TRN_STAGE_REDO_ACTION) {
×
2006
    pArray = pTrans->redoActions;
×
2007
  } else if (pTrans->stage == TRN_STAGE_UNDO_ACTION) {
×
2008
    pArray = pTrans->undoActions;
×
2009
  } else {
2010
    TAOS_RETURN(TSDB_CODE_MND_TRANS_INVALID_STAGE);
×
2011
  }
2012

2013
  if(pTrans->ableToBeKilled == false){
×
2014
    return TSDB_CODE_MND_TRANS_NOT_ABLE_TO_kILLED;
×
2015
  }
2016
  
2017
  if(pTrans->killMode == TRN_KILL_MODE_SKIP){
×
2018
    for (int32_t i = 0; i < taosArrayGetSize(pArray); ++i) {
×
2019
      STransAction *pAction = taosArrayGet(pArray, i);
×
2020
      mInfo("trans:%d, %s:%d set processed for kill msg received, errCode from %s to success", pTrans->id,
×
2021
            mndTransStr(pAction->stage), i, tstrerror(pAction->errCode));
2022
      pAction->msgSent = 1;
×
2023
      pAction->msgReceived = 1;
×
2024
      pAction->errCode = 0;
×
2025
    }
2026
  }
2027
  else if(pTrans->killMode == TRN_KILL_MODE_INTERUPT){
×
2028
    pTrans->stage = TRN_STAGE_PRE_FINISH;
×
2029
  }
2030
  else{
2031
    return TSDB_CODE_MND_TRANS_NOT_ABLE_TO_kILLED;
×
2032
  }
2033

2034
  mndTransExecute(pMnode, pTrans);
×
2035
  return 0;
×
2036
}
2037

2038
static int32_t mndProcessKillTransReq(SRpcMsg *pReq) {
×
2039
  SMnode       *pMnode = pReq->info.node;
×
2040
  SKillTransReq killReq = {0};
×
2041
  int32_t       code = -1;
×
2042
  STrans       *pTrans = NULL;
×
2043

2044
  if (tDeserializeSKillTransReq(pReq->pCont, pReq->contLen, &killReq) != 0) {
×
2045
    code = TSDB_CODE_INVALID_MSG;
×
2046
    goto _OVER;
×
2047
  }
2048

2049
  mInfo("trans:%d, start to kill", killReq.transId);
×
2050
  if ((code = mndCheckOperPrivilege(pMnode, pReq->info.conn.user, MND_OPER_KILL_TRANS)) != 0) {
×
2051
    goto _OVER;
×
2052
  }
2053

2054
  pTrans = mndAcquireTrans(pMnode, killReq.transId);
×
2055
  if (pTrans == NULL) {
×
2056
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
2057
    if (terrno != 0) code = terrno;
×
2058
    goto _OVER;
×
2059
  }
2060

2061
  code = mndKillTrans(pMnode, pTrans);
×
2062

2063
_OVER:
×
2064
  if (code != 0) {
×
2065
    mError("trans:%d, failed to kill since %s", killReq.transId, terrstr());
×
2066
  }
2067

2068
  mndReleaseTrans(pMnode, pTrans);
×
2069
  TAOS_RETURN(code);
×
2070
}
2071

2072
static int32_t mndCompareTransId(int32_t *pTransId1, int32_t *pTransId2) { return *pTransId1 >= *pTransId2 ? 1 : 0; }
×
2073

2074
void mndTransPullup(SMnode *pMnode) {
859✔
2075
  SSdb   *pSdb = pMnode->pSdb;
859✔
2076
  SArray *pArray = taosArrayInit(sdbGetSize(pSdb, SDB_TRANS), sizeof(int32_t));
859✔
2077
  if (pArray == NULL) return;
859!
2078

2079
  void *pIter = NULL;
859✔
2080
  while (1) {
155✔
2081
    STrans *pTrans = NULL;
1,014✔
2082
    pIter = sdbFetch(pMnode->pSdb, SDB_TRANS, pIter, (void **)&pTrans);
1,014✔
2083
    if (pIter == NULL) break;
1,014✔
2084
    if (taosArrayPush(pArray, &pTrans->id) == NULL) {
310!
2085
      mError("failed to put trans into array, trans:%d, but pull up will continute", pTrans->id);
×
2086
    }
2087
    sdbRelease(pSdb, pTrans);
155✔
2088
  }
2089

2090
  taosArraySort(pArray, (__compar_fn_t)mndCompareTransId);
859✔
2091

2092
  for (int32_t i = 0; i < taosArrayGetSize(pArray); ++i) {
1,014✔
2093
    int32_t *pTransId = taosArrayGet(pArray, i);
155✔
2094
    STrans  *pTrans = mndAcquireTrans(pMnode, *pTransId);
155✔
2095
    if (pTrans != NULL) {
155!
2096
      mndTransExecute(pMnode, pTrans);
155✔
2097
    }
2098
    mndReleaseTrans(pMnode, pTrans);
155✔
2099
  }
2100
  taosArrayDestroy(pArray);
859✔
2101
}
2102

2103
static char *formatTimestamp(char *buf, int64_t val, int precision) {
×
2104
  time_t tt;
2105
  if (precision == TSDB_TIME_PRECISION_MICRO) {
×
2106
    tt = (time_t)(val / 1000000);
×
2107
  }
2108
  if (precision == TSDB_TIME_PRECISION_NANO) {
×
2109
    tt = (time_t)(val / 1000000000);
×
2110
  } else {
2111
    tt = (time_t)(val / 1000);
×
2112
  }
2113

2114
  struct tm tm;
2115
  if (taosLocalTime(&tt, &tm, NULL, 0, NULL) == NULL) {
×
2116
    mError("failed to get local time");
×
2117
    return NULL;
×
2118
  }
2119
  size_t pos = taosStrfTime(buf, 32, "%Y-%m-%d %H:%M:%S", &tm);
×
2120

2121
  if (precision == TSDB_TIME_PRECISION_MICRO) {
×
2122
    sprintf(buf + pos, ".%06d", (int)(val % 1000000));
×
2123
  } else if (precision == TSDB_TIME_PRECISION_NANO) {
×
2124
    sprintf(buf + pos, ".%09d", (int)(val % 1000000000));
×
2125
  } else {
2126
    sprintf(buf + pos, ".%03d", (int)(val % 1000));
×
2127
  }
2128

2129
  return buf;
×
2130
}
2131

2132
static void mndTransLogAction(STrans *pTrans) {
×
2133
  char    detail[512] = {0};
×
2134
  int32_t len = 0;
×
2135
  int32_t index = 0;
×
2136

2137
  if (pTrans->stage == TRN_STAGE_PREPARE) {
×
2138
    for (int32_t i = 0; i < taosArrayGetSize(pTrans->prepareActions); ++i, ++index) {
×
2139
      len = 0;
×
2140
      STransAction *pAction = taosArrayGet(pTrans->prepareActions, i);
×
2141
      len += snprintf(detail + len, sizeof(detail) - len, "action:%d, %s:%d sdbType:%s, sdbStatus:%s\n", index,
×
2142
                      mndTransStr(pAction->stage), pAction->id, sdbTableName(pAction->pRaw->type),
×
2143
                      sdbStatusName(pAction->pRaw->status));
×
2144
      mDebug("trans:%d, show tran action, detail:%s", pTrans->id, detail);
×
2145
    }
2146
  }
2147

2148
  if (pTrans->stage == TRN_STAGE_REDO_ACTION) {
×
2149
    for (int32_t i = 0; i < taosArrayGetSize(pTrans->redoActions); ++i, ++index) {
×
2150
      len = 0;
×
2151
      STransAction *pAction = taosArrayGet(pTrans->redoActions, i);
×
2152
      if (pAction->actionType == TRANS_ACTION_MSG) {
×
2153
        char bufStart[40] = {0};
×
2154
        (void)formatTimestamp(bufStart, pAction->startTime, TSDB_TIME_PRECISION_MILLI);
×
2155

2156
        char endStart[40] = {0};
×
2157
        (void)formatTimestamp(endStart, pAction->endTime, TSDB_TIME_PRECISION_MILLI);
×
2158
        len += snprintf(detail + len, sizeof(detail) - len,
×
2159
                        "action:%d, %s:%d msgType:%s,"
2160
                        "sent:%d, received:%d, startTime:%s, endTime:%s, ",
2161
                        index, mndTransStr(pAction->stage), pAction->id, TMSG_INFO(pAction->msgType), pAction->msgSent,
×
2162
                        pAction->msgReceived, bufStart, endStart);
×
2163

2164
        SEpSet epset = pAction->epSet;
×
2165
        if (epset.numOfEps > 0) {
×
2166
          len += snprintf(detail + len, sizeof(detail) - len, "numOfEps:%d inUse:%d ", epset.numOfEps, epset.inUse);
×
2167
          for (int32_t i = 0; i < epset.numOfEps; ++i) {
×
2168
            len +=
×
2169
                snprintf(detail + len, sizeof(detail) - len, "ep:%d-%s:%u ", i, epset.eps[i].fqdn, epset.eps[i].port);
×
2170
          }
2171
        }
2172

2173
        len += snprintf(detail + len, sizeof(detail) - len, ", errCode:0x%x(%s)\n", pAction->errCode & 0xFFFF,
×
2174
                        tstrerror(pAction->errCode));
2175
      } else {
2176
        len += snprintf(detail + len, sizeof(detail) - len, "action:%d, %s:%d sdbType:%s, sdbStatus:%s, written:%d\n",
×
2177
                        index, mndTransStr(pAction->stage), pAction->id, sdbTableName(pAction->pRaw->type),
×
2178
                        sdbStatusName(pAction->pRaw->status), pAction->rawWritten);
×
2179
      }
2180
      mDebug("trans:%d, show tran action, detail:%s", pTrans->id, detail);
×
2181
    }
2182
  }
2183

2184
  if (pTrans->stage == TRN_STAGE_COMMIT_ACTION) {
×
2185
    for (int32_t i = 0; i < taosArrayGetSize(pTrans->commitActions); ++i, ++index) {
×
2186
      len = 0;
×
2187
      STransAction *pAction = taosArrayGet(pTrans->commitActions, i);
×
2188
      len += snprintf(detail + len, sizeof(detail) - len, "action:%d, %s:%d sdbType:%s, sdbStatus:%s\n", index,
×
2189
                      mndTransStr(pAction->stage), i, sdbTableName(pAction->pRaw->type),
×
2190
                      sdbStatusName(pAction->pRaw->status));
×
2191
      mDebug("trans:%d, show tran action, detail:%s", pTrans->id, detail);
×
2192
    }
2193

2194
    for (int32_t i = 0; i < taosArrayGetSize(pTrans->undoActions); ++i, ++index) {
×
2195
      len = 0;
×
2196
      STransAction *pAction = taosArrayGet(pTrans->undoActions, i);
×
2197
      if (pAction->actionType == TRANS_ACTION_MSG) {
×
2198
        len += snprintf(detail + len, sizeof(detail) - len, "action:%d, %s:%d msgType:%s\n", index,
×
2199
                        mndTransStr(pAction->stage), pAction->id, TMSG_INFO(pAction->msgType));
×
2200
        ;
2201
      } else {
2202
        len += snprintf(detail + len, sizeof(detail) - len, "action:%d, %s:%d sdbType:%s, sdbStatus:%s\n", index,
×
2203
                        mndTransStr(pAction->stage), pAction->id, sdbTableName(pAction->pRaw->type),
×
2204
                        sdbStatusName(pAction->pRaw->status));
×
2205
      }
2206
      mDebug("trans:%d, show tran action, detail:%s", pTrans->id, detail);
×
2207
    }
2208
  }
2209
}
×
2210

2211
static int32_t mndRetrieveTrans(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows) {
1✔
2212
  SMnode *pMnode = pReq->info.node;
1✔
2213
  SSdb   *pSdb = pMnode->pSdb;
1✔
2214
  int32_t numOfRows = 0;
1✔
2215
  STrans *pTrans = NULL;
1✔
2216
  int32_t cols = 0;
1✔
2217
  int32_t code = 0;
1✔
2218
  int32_t lino = 0;
1✔
2219

2220
  while (numOfRows < rows) {
1!
2221
    pShow->pIter = sdbFetch(pSdb, SDB_TRANS, pShow->pIter, (void **)&pTrans);
1✔
2222
    if (pShow->pIter == NULL) break;
1!
2223

2224
    cols = 0;
×
2225

2226
    SColumnInfoData *pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
×
2227
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)&pTrans->id, false), pTrans, &lino, _OVER);
×
2228

2229
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
×
2230
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)&pTrans->createdTime, false), pTrans, &lino,
×
2231
                        _OVER);
2232

2233
    char stage[TSDB_TRANS_STAGE_LEN + VARSTR_HEADER_SIZE] = {0};
×
2234
    STR_WITH_MAXSIZE_TO_VARSTR(stage, mndTransStr(pTrans->stage), pShow->pMeta->pSchemas[cols].bytes);
×
2235
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
×
2236
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)stage, false), pTrans, &lino, _OVER);
×
2237

2238
    char opername[TSDB_TRANS_OPER_LEN + VARSTR_HEADER_SIZE] = {0};
×
2239
    STR_WITH_MAXSIZE_TO_VARSTR(opername, pTrans->opername, pShow->pMeta->pSchemas[cols].bytes);
×
2240
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
×
2241
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)opername, false), pTrans, &lino, _OVER);
×
2242

2243
    char dbname[TSDB_DB_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
×
2244
    STR_WITH_MAXSIZE_TO_VARSTR(dbname, mndGetDbStr(pTrans->dbname), pShow->pMeta->pSchemas[cols].bytes);
×
2245
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
×
2246
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)dbname, false), pTrans, &lino, _OVER);
×
2247

2248
    char stbname[TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
×
2249
    STR_WITH_MAXSIZE_TO_VARSTR(stbname, mndGetDbStr(pTrans->stbname), pShow->pMeta->pSchemas[cols].bytes);
×
2250
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
×
2251
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)stbname, false), pTrans, &lino, _OVER);
×
2252

2253
    const char *killableStr = pTrans->ableToBeKilled ? "yes" : "no";
×
2254
    char        killableVstr[10 + VARSTR_HEADER_SIZE] = {0};
×
2255
    STR_WITH_MAXSIZE_TO_VARSTR(killableVstr, killableStr, 10 + VARSTR_HEADER_SIZE);
×
2256
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
×
2257
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)killableVstr, false), pTrans, &lino, _OVER);
×
2258

2259
    /*
2260
    const char *killModeStr = pTrans->killMode == TRN_KILL_MODE_SKIP ? "skip" : "interrupt";
2261
    char        killModeVstr[10 + VARSTR_HEADER_SIZE] = {0};
2262
    STR_WITH_MAXSIZE_TO_VARSTR(killModeVstr, killModeStr, 24);
2263
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
2264
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)killModeVstr, false), pTrans, &lino, _OVER);
2265
    */
2266

2267
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
×
2268
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)&pTrans->failedTimes, false), pTrans, &lino,
×
2269
                        _OVER);
2270

2271
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
×
2272
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)&pTrans->lastExecTime, false), pTrans, &lino,
×
2273
                        _OVER);
2274

2275
    char    lastInfo[TSDB_TRANS_ERROR_LEN + VARSTR_HEADER_SIZE] = {0};
×
2276
    char    detail[TSDB_TRANS_ERROR_LEN + 1] = {0};
×
2277
    int32_t len = tsnprintf(detail, sizeof(detail), "action:%d code:0x%x(%s) ", pTrans->lastAction,
×
2278
                            pTrans->lastErrorNo & 0xFFFF, tstrerror(pTrans->lastErrorNo));
×
2279
    SEpSet  epset = pTrans->lastEpset;
×
2280
    if (epset.numOfEps > 0) {
×
2281
      len += tsnprintf(detail + len, sizeof(detail) - len, "msgType:%s numOfEps:%d inUse:%d ",
×
2282
                       TMSG_INFO(pTrans->lastMsgType), epset.numOfEps, epset.inUse);
×
2283
      for (int32_t i = 0; i < pTrans->lastEpset.numOfEps; ++i) {
×
2284
        len += snprintf(detail + len, sizeof(detail) - len, "ep:%d-%s:%u ", i, epset.eps[i].fqdn, epset.eps[i].port);
×
2285
      }
2286
    }
2287
    STR_WITH_MAXSIZE_TO_VARSTR(lastInfo, detail, pShow->pMeta->pSchemas[cols].bytes);
×
2288
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
×
2289
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)lastInfo, false), pTrans, &lino, _OVER);
×
2290

2291
    mndTransLogAction(pTrans);
×
2292

2293
    numOfRows++;
×
2294
    sdbRelease(pSdb, pTrans);
×
2295
  }
2296

2297
_OVER:
×
2298
  if (code != 0) mError("failed to retrieve at line:%d, since %s", lino, tstrerror(code));
1!
2299
  pShow->numOfRows += numOfRows;
1✔
2300
  return numOfRows;
1✔
2301
}
2302

2303
static int32_t mndShowTransCommonColumns(SShowObj *pShow, SSDataBlock *pBlock, STransAction *pAction,
×
2304
                                         int32_t transactionId, int32_t curActionId, int32_t numOfRows, int32_t *cols) {
2305
  int32_t code = 0;
×
2306
  int32_t lino = 0;
×
2307
  int32_t len = 0;
×
2308

2309
  SColumnInfoData *pColInfo = taosArrayGet(pBlock->pDataBlock, (*cols)++);
×
2310
  TAOS_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)&transactionId, false), &lino, _OVER);
×
2311

2312
  char action[30 + 1] = {0};
×
2313
  if (curActionId == pAction->id) {
×
2314
    len += snprintf(action + len, sizeof(action) - len, "%s:%d(%s)<-last", mndTransStr(pAction->stage), pAction->id,
×
2315
                    mndTransTypeStr(pAction->actionType));
2316
  } else {
2317
    len += snprintf(action + len, sizeof(action) - len, "%s:%d(%s)", mndTransStr(pAction->stage), pAction->id,
×
2318
                    mndTransTypeStr(pAction->actionType));
2319
  }
2320
  char actionVStr[30 + VARSTR_HEADER_SIZE] = {0};
×
2321
  STR_WITH_MAXSIZE_TO_VARSTR(actionVStr, action, pShow->pMeta->pSchemas[*cols].bytes);
×
2322
  pColInfo = taosArrayGet(pBlock->pDataBlock, (*cols)++);
×
2323
  TAOS_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)actionVStr, false), &lino, _OVER);
×
2324
_OVER:
×
2325
  if (code != 0) mError("failed to retrieve at line:%d, since %s", lino, tstrerror(code));
×
2326
  return code;
×
2327
}
2328

2329
static void mndShowTransAction(SShowObj *pShow, SSDataBlock *pBlock, STransAction *pAction, int32_t transactionId,
×
2330
                               int32_t curActionId, int32_t rows, int32_t numOfRows) {
2331
  int32_t code = 0;
×
2332
  int32_t lino = 0;
×
2333
  int32_t len = 0;
×
2334
  int32_t cols = 0;
×
2335

2336
  cols = 0;
×
2337

2338
  if (mndShowTransCommonColumns(pShow, pBlock, pAction, transactionId, curActionId, numOfRows, &cols) != 0) return;
×
2339

2340
  if (pAction->actionType == TRANS_ACTION_MSG) {
×
2341
    int32_t len = 0;
×
2342

2343
    char objType[TSDB_TRANS_OBJTYPE_LEN + 1] = {0};
×
2344
    len += snprintf(objType + len, sizeof(objType) - len, "%s(s:%d,r:%d)", TMSG_INFO(pAction->msgType),
×
2345
                    pAction->msgSent, pAction->msgReceived);
×
2346
    char objTypeVStr[TSDB_TRANS_OBJTYPE_LEN + VARSTR_HEADER_SIZE] = {0};
×
2347
    STR_WITH_MAXSIZE_TO_VARSTR(objTypeVStr, objType, pShow->pMeta->pSchemas[cols].bytes);
×
2348
    SColumnInfoData *pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
×
2349
    TAOS_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)objTypeVStr, false), &lino, _OVER);
×
2350

2351
    char result[TSDB_TRANS_RESULT_LEN + 1] = {0};
×
2352
    len = 0;
×
2353
    len += snprintf(result + len, sizeof(result) - len, "errCode:0x%x(%s)", pAction->errCode & 0xFFFF,
×
2354
                    tstrerror(pAction->errCode));
2355
    char resultVStr[TSDB_TRANS_RESULT_LEN + VARSTR_HEADER_SIZE] = {0};
×
2356
    STR_WITH_MAXSIZE_TO_VARSTR(resultVStr, result, pShow->pMeta->pSchemas[cols].bytes);
×
2357
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
×
2358
    TAOS_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)resultVStr, false), &lino, _OVER);
×
2359

2360
    char target[TSDB_TRANS_TARGET_LEN] = {0};
×
2361
    len = 0;
×
2362
    SEpSet epset = pAction->epSet;
×
2363
    if (epset.numOfEps > 0) {
×
2364
      for (int32_t i = 0; i < epset.numOfEps; ++i) {
×
2365
        len += snprintf(target + len, sizeof(target) - len, "ep:%d-%s:%u,", i, epset.eps[i].fqdn, epset.eps[i].port);
×
2366
      }
2367
      len += snprintf(target + len, sizeof(target) - len, "(%d:%d) ", epset.numOfEps, epset.inUse);
×
2368
    }
2369
    char targetVStr[TSDB_TRANS_TARGET_LEN + VARSTR_HEADER_SIZE] = {0};
×
2370
    STR_WITH_MAXSIZE_TO_VARSTR(targetVStr, target, pShow->pMeta->pSchemas[cols].bytes);
×
2371
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
×
2372
    TAOS_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)targetVStr, false), &lino, _OVER);
×
2373

2374
    char detail[TSDB_TRANS_DETAIL_LEN] = {0};
×
2375
    len = 0;
×
2376
    char bufStart[40] = {0};
×
2377
    if (pAction->startTime > 0) (void)formatTimestamp(bufStart, pAction->startTime, TSDB_TIME_PRECISION_MILLI);
×
2378
    char bufEnd[40] = {0};
×
2379
    if (pAction->endTime > 0) (void)formatTimestamp(bufEnd, pAction->endTime, TSDB_TIME_PRECISION_MILLI);
×
2380
    len += snprintf(detail + len, sizeof(detail) - len, "startTime:%s, endTime:%s, ", bufStart, bufEnd);
×
2381
    char detailVStr[TSDB_TRANS_DETAIL_LEN + VARSTR_HEADER_SIZE] = {0};
×
2382
    STR_WITH_MAXSIZE_TO_VARSTR(detailVStr, detail, pShow->pMeta->pSchemas[cols].bytes);
×
2383
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
×
2384
    TAOS_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)detailVStr, false), &lino, _OVER);
×
2385

2386
  } else {
2387
    int32_t len = 0;
×
2388

2389
    char objType[TSDB_TRANS_OBJTYPE_LEN + 1] = {0};
×
2390
    if (pAction->pRaw->type == SDB_VGROUP) {
×
2391
      SSdbRow *pRow = mndVgroupActionDecode(pAction->pRaw);
×
2392
      SVgObj  *pVgroup = sdbGetRowObj(pRow);
×
2393
      len += snprintf(objType + len, sizeof(objType) - len, "%s(%d)", sdbTableName(pAction->pRaw->type), pVgroup->vgId);
×
2394
      taosMemoryFreeClear(pRow);
×
2395
    } else {
2396
      strcpy(objType, sdbTableName(pAction->pRaw->type));
×
2397
    }
2398
    char objTypeVStr[TSDB_TRANS_OBJTYPE_LEN + VARSTR_HEADER_SIZE] = {0};
×
2399
    STR_WITH_MAXSIZE_TO_VARSTR(objTypeVStr, objType, pShow->pMeta->pSchemas[cols].bytes);
×
2400
    SColumnInfoData *pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
×
2401
    TAOS_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)objTypeVStr, false), &lino, _OVER);
×
2402

2403
    char result[TSDB_TRANS_RESULT_LEN + 1] = {0};
×
2404
    len = 0;
×
2405
    len += snprintf(result + len, sizeof(result) - len, "rawWritten:%d", pAction->rawWritten);
×
2406
    char resultVStr[TSDB_TRANS_RESULT_LEN + VARSTR_HEADER_SIZE] = {0};
×
2407
    STR_WITH_MAXSIZE_TO_VARSTR(resultVStr, result, pShow->pMeta->pSchemas[cols].bytes);
×
2408
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
×
2409
    TAOS_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)resultVStr, false), &lino, _OVER);
×
2410

2411
    char target[TSDB_TRANS_TARGET_LEN] = "";
×
2412
    char targetVStr[TSDB_TRANS_TARGET_LEN + VARSTR_HEADER_SIZE] = {0};
×
2413
    STR_WITH_MAXSIZE_TO_VARSTR(targetVStr, target, pShow->pMeta->pSchemas[cols].bytes);
×
2414
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
×
2415
    TAOS_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)targetVStr, false), &lino, _OVER);
×
2416

2417
    char detail[TSDB_TRANS_DETAIL_LEN] = {0};
×
2418
    len = 0;
×
2419
    len += snprintf(detail + len, sizeof(detail) - len, "sdbStatus:%s", sdbStatusName(pAction->pRaw->status));
×
2420
    char detailVStr[TSDB_TRANS_DETAIL_LEN + VARSTR_HEADER_SIZE] = {0};
×
2421
    STR_WITH_MAXSIZE_TO_VARSTR(detailVStr, detail, pShow->pMeta->pSchemas[cols].bytes);
×
2422
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
×
2423
    TAOS_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)detailVStr, false), &lino, _OVER);
×
2424
  }
2425

2426
_OVER:
×
2427
  if (code != 0) mError("failed to retrieve at line:%d, since %s", lino, tstrerror(code));
×
2428
}
2429

2430
static SArray *mndTransGetAction(STrans *pTrans, ETrnStage stage) {
×
2431
  if (stage == TRN_STAGE_PREPARE) {
×
2432
    return pTrans->prepareActions;
×
2433
  }
2434
  if (stage == TRN_STAGE_REDO_ACTION) {
×
2435
    return pTrans->redoActions;
×
2436
  }
2437
  if (stage == TRN_STAGE_COMMIT_ACTION) {
×
2438
    return pTrans->commitActions;
×
2439
  }
2440
  if (stage == TRN_STAGE_UNDO_ACTION) {
×
2441
    return pTrans->undoActions;
×
2442
  }
2443
  return NULL;
×
2444
}
2445

2446
typedef struct STransDetailIter {
2447
  void     *pIter;
2448
  STrans   *pTrans;
2449
  ETrnStage stage;
2450
  int32_t   num;
2451
} STransDetailIter;
2452

2453
static void mndTransShowActions(SSdb *pSdb, STransDetailIter *pShowIter, SShowObj *pShow, SSDataBlock *pBlock,
×
2454
                                int32_t rows, int32_t *numOfRows, SArray *pActions, int32_t end, int32_t start) {
2455
  int32_t actionNum = taosArrayGetSize(pActions);
×
2456
  mInfo("stage:%s, Actions num:%d", mndTransStr(pShowIter->stage), actionNum);
×
2457

2458
  for (int32_t i = start; i < actionNum; ++i) {
×
2459
    STransAction *pAction = taosArrayGet(pShowIter->pTrans->redoActions, i);
×
2460
    mndShowTransAction(pShow, pBlock, pAction, pShowIter->pTrans->id, pShowIter->pTrans->lastAction, rows, *numOfRows);
×
2461
    (*numOfRows)++;
×
2462
    if (*numOfRows >= rows) break;
×
2463
  }
2464

2465
  if (*numOfRows == end) {
×
2466
    sdbRelease(pSdb, pShowIter->pTrans);
×
2467
    pShowIter->pTrans = NULL;
×
2468
    pShowIter->num = 0;
×
2469
  } else {
2470
    pShowIter->pTrans = pShowIter->pTrans;
×
2471
    pShowIter->stage = pShowIter->pTrans->stage;
×
2472
    pShowIter->num += (*numOfRows);
×
2473
  }
2474
}
×
2475

2476
static int32_t mndRetrieveTransDetail(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows) {
×
2477
  SMnode *pMnode = pReq->info.node;
×
2478
  SSdb   *pSdb = pMnode->pSdb;
×
2479
  int32_t numOfRows = 0;
×
2480

2481
  int32_t code = 0;
×
2482
  int32_t lino = 0;
×
2483

2484
  mInfo("start to mndRetrieveTransDetail, rows:%d, pShow->numOfRows:%d, pShow->pIter:%p", rows, pShow->numOfRows,
×
2485
        pShow->pIter);
2486

2487
  if (pShow->pIter == NULL) {
×
2488
    pShow->pIter = taosMemoryMalloc(sizeof(STransDetailIter));
×
2489
    if (pShow->pIter == NULL) {
×
2490
      mError("failed to malloc for pShow->pIter");
×
2491
      return 0;
×
2492
    }
2493
    memset(pShow->pIter, 0, sizeof(STransDetailIter));
×
2494
  }
2495

2496
  STransDetailIter *pShowIter = (STransDetailIter *)pShow->pIter;
×
2497

2498
  while (numOfRows < rows) {
×
2499
    if (pShowIter->pTrans == NULL) {
×
2500
      pShowIter->pIter = sdbFetch(pSdb, SDB_TRANS, pShowIter->pIter, (void **)&(pShowIter->pTrans));
×
2501
      mDebug("retrieve trans detail from fetch, pShow->pIter:%p, pTrans:%p", pShowIter->pIter, pShowIter->pTrans);
×
2502
      if (pShowIter->pIter == NULL) break;
×
2503
      mInfo("retrieve trans detail from fetch, id:%d, trans stage:%d, IterNum:%d", pShowIter->pTrans->id,
×
2504
            pShowIter->pTrans->stage, pShowIter->num);
2505

2506
      SArray *pActions = mndTransGetAction(pShowIter->pTrans, pShowIter->pTrans->stage);
×
2507

2508
      mndTransShowActions(pSdb, pShowIter, pShow, pBlock, rows, &numOfRows, pActions, taosArrayGetSize(pActions), 0);
×
2509
      break;
×
2510
    } else {
2511
      mInfo("retrieve trans detail from iter, id:%d, iterStage:%d, IterNum:%d", pShowIter->pTrans->id, pShowIter->stage,
×
2512
            pShowIter->num);
2513
      SArray *pActions = mndTransGetAction(pShowIter->pTrans, pShowIter->stage);
×
2514

2515
      mndTransShowActions(pSdb, pShowIter, pShow, pBlock, rows, &numOfRows, pActions,
×
2516
                          taosArrayGetSize(pActions) - pShowIter->num, pShowIter->num);
×
2517
      break;
×
2518
    }
2519
  }
2520

2521
_OVER:
×
2522
  pShow->numOfRows += numOfRows;
×
2523

2524
  if (code != 0) {
×
2525
    mError("failed to retrieve at line:%d, since %s", lino, tstrerror(code));
×
2526
  } else {
2527
    mInfo("retrieve trans detail, numOfRows:%d, pShow->numOfRows:%d", numOfRows, pShow->numOfRows)
×
2528
  }
2529
  if (numOfRows == 0) {
×
2530
    taosMemoryFree(pShow->pIter);
×
2531
    pShow->pIter = NULL;
×
2532
  }
2533
  return numOfRows;
×
2534
}
2535

2536
static void mndCancelGetNextTrans(SMnode *pMnode, void *pIter) {
×
2537
  SSdb *pSdb = pMnode->pSdb;
×
2538
  sdbCancelFetchByType(pSdb, pIter, SDB_TRANS);
×
2539
}
×
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2026 Coveralls, Inc