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

taosdata / TDengine / #3849

18 Apr 2025 06:24AM UTC coverage: 62.513% (-0.1%) from 62.61%
#3849

push

travis-ci

GitHub
feat: taosBenchmark restore RESTFUL (3.0) (#30699)

154684 of 316122 branches covered (48.93%)

Branch coverage included in aggregate %.

240354 of 315808 relevant lines covered (76.11%)

10508512.55 hits per line

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

63.76
/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, bool notSend);
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, bool notSend);
46
static int32_t mndTransExecuteUndoActions(SMnode *pMnode, STrans *pTrans, bool topHalf, bool notSend);
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, bool notSend);
50
static bool    mndTransPerformUndoLogStage(SMnode *pMnode, STrans *pTrans, bool topHalf);
51
static bool    mndTransPerformUndoActionStage(SMnode *pMnode, STrans *pTrans, bool topHalf, bool notSend);
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; }
452,156✔
58

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

66
static inline char *mndStrExecutionContext(bool topHalf) { return topHalf ? "transContext" : "syncContext"; }
473,461✔
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) {
1,829✔
79
  SSdbTable table = {
1,829✔
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);
1,829✔
90
  mndSetMsgHandle(pMnode, TDMT_MND_KILL_TRANS, mndProcessKillTransReq);
1,829✔
91

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

98
void mndCleanupTrans(SMnode *pMnode) {}
1,828✔
99

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

104
  for (int32_t i = 0; i < actionNum; ++i) {
1,707,631✔
105
    STransAction *pAction = taosArrayGet(pArray, i);
1,160,379✔
106
    if (pAction->actionType == TRANS_ACTION_RAW) {
1,160,379✔
107
      rawDataLen += (sizeof(STransAction) + sdbGetRawTotalSize(pAction->pRaw));
722,649✔
108
    } else if (pAction->actionType == TRANS_ACTION_MSG) {
437,730!
109
      rawDataLen += (sizeof(STransAction) + pAction->contLen);
437,730✔
110
    } else {
111
      // empty
112
    }
113
    rawDataLen += sizeof(int8_t);
1,160,379✔
114
  }
115

116
  return rawDataLen;
547,252✔
117
}
118

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

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

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

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

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

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

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

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

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

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

207
  SDB_SET_INT32(pRaw, dataPos, pTrans->startFunc, _OVER)
136,813!
208
  SDB_SET_INT32(pRaw, dataPos, pTrans->stopFunc, _OVER)
136,813!
209
  SDB_SET_INT32(pRaw, dataPos, pTrans->paramLen, _OVER)
136,813!
210
  if (pTrans->param != NULL) {
136,813!
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)
136,813!
215

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

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

231
  SDB_SET_RESERVE(pRaw, dataPos, TRANS_RESERVE_SIZE, _OVER)
136,813!
232
  SDB_SET_DATALEN(pRaw, dataPos, _OVER)
136,813!
233

234
  terrno = 0;
136,813✔
235

236
_OVER:
136,813✔
237
  if (terrno != 0) {
136,813!
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);
136,813✔
245
  return pRaw;
136,813✔
246
}
247

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

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

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

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

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

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

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

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

332
  SDB_GET_INT32(pRaw, dataPos, &pTrans->id, _OVER)
261,085!
333

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

358
  if (sver > TRANS_VER1_NUMBER) {
261,085!
359
    SDB_GET_INT32(pRaw, dataPos, &prepareActionNum, _OVER)
261,085!
360
  }
361
  SDB_GET_INT32(pRaw, dataPos, &redoActionNum, _OVER)
261,085!
362
  SDB_GET_INT32(pRaw, dataPos, &undoActionNum, _OVER)
261,085!
363
  SDB_GET_INT32(pRaw, dataPos, &commitActionNum, _OVER)
261,085!
364

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

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

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

380
  SDB_GET_INT32(pRaw, dataPos, &pTrans->startFunc, _OVER)
261,085!
381
  SDB_GET_INT32(pRaw, dataPos, &pTrans->stopFunc, _OVER)
261,085!
382
  SDB_GET_INT32(pRaw, dataPos, &pTrans->paramLen, _OVER)
261,085!
383
  if (pTrans->paramLen != 0) {
261,085!
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);
261,085!
390

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

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

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

407
  SDB_GET_RESERVE(pRaw, dataPos, TRANS_RESERVE_SIZE, _OVER)
261,085!
408

409
  terrno = 0;
261,085✔
410

411
_OVER:
261,085✔
412
  if (terrno != 0 && pTrans != NULL) {
261,085!
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) {
261,085!
420
    mTrace("trans:%d, decode from raw:%p, row:%p", pTrans->id, pRaw, pTrans);
261,085✔
421
  }
422
  return pRow;
261,085✔
423
}
424

425
static const char *mndTransStr(ETrnStage stage) {
2,129,278✔
426
  switch (stage) {
2,129,278!
427
    case TRN_STAGE_PREPARE:
132,885✔
428
      return "prepare";
132,885✔
429
    case TRN_STAGE_REDO_ACTION:
914,677✔
430
      return "redoAction";
914,677✔
431
    case TRN_STAGE_ROLLBACK:
41✔
432
      return "rollback";
41✔
433
    case TRN_STAGE_UNDO_ACTION:
44,063✔
434
      return "undoAction";
44,063✔
435
    case TRN_STAGE_COMMIT:
221,955✔
436
      return "commit";
221,955✔
437
    case TRN_STAGE_COMMIT_ACTION:
487,810✔
438
      return "commitAction";
487,810✔
439
    case TRN_STAGE_FINISH:
327,810✔
440
      return "finished";
327,810✔
441
    case TRN_STAGE_PRE_FINISH:
37✔
442
      return "pre-finish";
37✔
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) {
630,763✔
460
  if (pAction != NULL) {
630,763✔
461
    if (pAction->errCode != TSDB_CODE_ACTION_IN_PROGRESS) {
518,168✔
462
      pTrans->lastAction = pAction->id;
338,041✔
463
      pTrans->lastMsgType = pAction->msgType;
338,041✔
464
      pTrans->lastEpset = pAction->epSet;
338,041✔
465
      pTrans->lastErrorNo = pAction->errCode;
338,041✔
466
    }
467
  } else {
468
    pTrans->lastAction = 0;
112,595✔
469
    pTrans->lastMsgType = 0;
112,595✔
470
    memset(&pTrans->lastEpset, 0, sizeof(pTrans->lastEpset));
112,595✔
471
    pTrans->lastErrorNo = 0;
112,595✔
472
  }
473
}
630,763✔
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) {
2,046✔
484
  switch (ftype) {
2,046!
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:
1,023✔
490
      return mndRebCntInc;
1,023✔
491
    case TRANS_STOP_FUNC_MQ_REB:
1,023✔
492
      return mndRebCntDec;
1,023✔
493
    default:
×
494
      return NULL;
×
495
  }
496
}
497

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

502
  (void)taosThreadMutexInit(&pTrans->mutex, NULL);
49,977✔
503

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

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

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

522
  if (pTrans->stage == TRN_STAGE_PRE_FINISH) {
49,977!
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;
49,977✔
528
}
529

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

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

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

579
  mndTransDropData(pTrans);
155,515✔
580
  return 0;
155,515✔
581
}
582

583
static void mndTransUpdateActions(SArray *pOldArray, SArray *pNewArray) {
223,208✔
584
  for (int32_t i = 0; i < taosArrayGetSize(pOldArray); ++i) {
722,274✔
585
    STransAction *pOldAction = taosArrayGet(pOldArray, i);
499,066✔
586
    STransAction *pNewAction = taosArrayGet(pNewArray, i);
499,066✔
587
    pOldAction->rawWritten = pNewAction->rawWritten;
499,066✔
588
    pOldAction->msgSent = pNewAction->msgSent;
499,066✔
589
    pOldAction->msgReceived = pNewAction->msgReceived;
499,066✔
590
    pOldAction->errCode = pNewAction->errCode;
499,066✔
591
  }
592
}
223,208✔
593

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

598
  if (pOld->createdTime != pNew->createdTime) {
55,802!
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);
55,802✔
608
  mndTransUpdateActions(pOld->redoActions, pNew->redoActions);
55,802✔
609
  mndTransUpdateActions(pOld->undoActions, pNew->undoActions);
55,802✔
610
  mndTransUpdateActions(pOld->commitActions, pNew->commitActions);
55,802✔
611
  pOld->stage = pNew->stage;
55,802✔
612
  pOld->actionPos = pNew->actionPos;
55,802✔
613

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

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

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

629
  return 0;
55,802✔
630
}
631

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

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

646
STrans *mndTransCreate(SMnode *pMnode, ETrnPolicy policy, ETrnConflct conflict, const SRpcMsg *pReq,
41,262✔
647
                       const char *opername) {
648
  STrans *pTrans = taosMemoryCalloc(1, sizeof(STrans));
41,262!
649
  if (pTrans == NULL) {
41,262!
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) {
41,262!
656
    tstrncpy(pTrans->opername, opername, TSDB_TRANS_OPER_LEN);
41,262✔
657
  }
658

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

679
  if (pTrans->redoActions == NULL || pTrans->undoActions == NULL || pTrans->commitActions == NULL ||
41,262!
680
      pTrans->pRpcArray == NULL) {
41,262!
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) {
41,262✔
688
    if (taosArrayPush(pTrans->pRpcArray, &pReq->info) == NULL) {
51,078!
689
      terrno = TSDB_CODE_OUT_OF_MEMORY;
×
690
      return NULL;
×
691
    }
692
    pTrans->originRpcType = pReq->msgType;
25,539✔
693
  }
694

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

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

701
static void mndTransDropActions(SArray *pArray) {
1,209,388✔
702
  int32_t size = taosArrayGetSize(pArray);
1,209,388✔
703
  for (int32_t i = 0; i < size; ++i) {
3,707,505✔
704
    STransAction *pAction = taosArrayGet(pArray, i);
2,498,117✔
705
    if (pAction->actionType == TRANS_ACTION_RAW) {
2,498,117✔
706
      taosMemoryFreeClear(pAction->pRaw);
1,532,759!
707
    } else if (pAction->actionType == TRANS_ACTION_MSG) {
965,358!
708
      taosMemoryFreeClear(pAction->pCont);
965,358!
709
    } else {
710
      // nothing
711
    }
712
  }
713

714
  taosArrayDestroy(pArray);
1,209,388✔
715
}
1,209,388✔
716

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

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

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

733
  return 0;
335,536✔
734
}
735

736
int32_t mndTransAppendRedolog(STrans *pTrans, SSdbRaw *pRaw) {
1,655✔
737
  STransAction action = {
1,655✔
738
      .stage = TRN_STAGE_REDO_ACTION, .actionType = TRANS_ACTION_RAW, .pRaw = pRaw, .mTraceId = pTrans->mTraceId};
1,655✔
739
  return mndTransAppendAction(pTrans->redoActions, &action);
1,655✔
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) {
13,083✔
748
  STransAction action = {.stage = TRN_STAGE_UNDO_ACTION, .actionType = TRANS_ACTION_RAW, .pRaw = pRaw};
13,083✔
749
  return mndTransAppendAction(pTrans->undoActions, &action);
13,083✔
750
}
751

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

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

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

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

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

781
void mndTransSetCb(STrans *pTrans, ETrnFunc startFunc, ETrnFunc stopFunc, void *param, int32_t paramLen) {
1,011✔
782
  pTrans->startFunc = startFunc;
1,011✔
783
  pTrans->stopFunc = stopFunc;
1,011✔
784
  pTrans->param = param;
1,011✔
785
  pTrans->paramLen = paramLen;
1,011✔
786
}
1,011✔
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) {
25,834✔
820
  if (dbname != NULL) {
25,834!
821
    tstrncpy(pTrans->dbname, dbname, TSDB_DB_FNAME_LEN);
25,834✔
822
  }
823
  if (stbname != NULL) {
25,834✔
824
    tstrncpy(pTrans->stbname, stbname, TSDB_TABLE_FNAME_LEN);
18,780✔
825
  }
826
}
25,834✔
827

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

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

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

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

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

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

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

849
static int32_t mndTransSync(SMnode *pMnode, STrans *pTrans) {
86,751✔
850
  int32_t  code = 0;
86,751✔
851
  SSdbRaw *pRaw = mndTransEncode(pTrans);
86,751✔
852
  if (pRaw == NULL) {
86,751!
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));
86,751!
859

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

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

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

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

887
static void mndTransLogConflict(STrans *pNew, STrans *pTrans, bool conflict, bool *globalConflict) {
19,634✔
888
  if (conflict) {
19,634✔
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,
370!
890
           pNew->dbname, pNew->stbname, pNew->conflict, pTrans->id, pTrans->dbname, pTrans->stbname, pTrans->conflict);
891
    *globalConflict = true;
370✔
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,
19,264!
894
          pNew->stbname, pNew->conflict, pTrans->id, pTrans->dbname, pTrans->stbname, pTrans->conflict);
895
  }
896
}
19,634✔
897

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

903
  if (pNew->conflict == TRN_CONFLICT_NOTHING) return conflict;
116,971✔
904

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

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

912
    if (pNew->conflict == TRN_CONFLICT_GLOBAL) conflict = true;
19,819!
913

914
    if (pNew->conflict == TRN_CONFLICT_DB) {
19,819✔
915
      if (pTrans->conflict == TRN_CONFLICT_GLOBAL) conflict = true;
2,390✔
916
      if (pTrans->conflict == TRN_CONFLICT_DB || pTrans->conflict == TRN_CONFLICT_DB_INSIDE) {
2,390!
917
        mndTransLogConflict(pNew, pTrans, mndCheckDbConflict(pNew->dbname, pTrans), &conflict);
1,308✔
918
        mndTransLogConflict(pNew, pTrans, mndCheckStbConflict(pNew->stbname, pTrans), &conflict);
1,308✔
919
      }
920
    }
921

922
    if (pNew->conflict == TRN_CONFLICT_DB_INSIDE) {
19,819✔
923
      if (pTrans->conflict == TRN_CONFLICT_GLOBAL) conflict = true;
17,426✔
924
      if (pTrans->conflict == TRN_CONFLICT_DB) {
17,426✔
925
        mndTransLogConflict(pNew, pTrans, mndCheckDbConflict(pNew->dbname, pTrans), &conflict);
258✔
926
        mndTransLogConflict(pNew, pTrans, mndCheckStbConflict(pNew->stbname, pTrans), &conflict);
258✔
927
      }
928
      if (pTrans->conflict == TRN_CONFLICT_DB_INSIDE) {
17,426✔
929
        mndTransLogConflict(pNew, pTrans, mndCheckStbConflict(pNew->stbname, pTrans), &conflict);  // for stb
16,499✔
930
      }
931
    }
932

933
    if (pNew->conflict == TRN_CONFLICT_ARBGROUP) {
19,819!
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) {
19,819✔
952
      if (pTrans->conflict == TRN_CONFLICT_GLOBAL || pTrans->conflict == TRN_CONFLICT_TSMA) {
3!
953
        mndTransLogConflict(pNew, pTrans, true, &conflict);
×
954
      } else {
955
        mndTransLogConflict(pNew, pTrans, false, &conflict);
3✔
956
      }
957
    }
958

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

962
  return conflict;
78,212✔
963
}
964

965
int32_t mndTransCheckConflict(SMnode *pMnode, STrans *pTrans) {
116,971✔
966
  int32_t code = 0;
116,971✔
967
  if (pTrans->conflict == TRN_CONFLICT_DB || pTrans->conflict == TRN_CONFLICT_DB_INSIDE) {
116,971✔
968
    if (strlen(pTrans->dbname) == 0 && strlen(pTrans->stbname) == 0) {
68,576!
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)) {
116,971✔
976
    code = TSDB_CODE_MND_TRANS_CONFLICT;
392✔
977
    mError("trans:%d, failed to check tran conflict since %s", pTrans->id, tstrerror(code));
392!
978
    TAOS_RETURN(code);
392✔
979
  }
980

981
  TAOS_RETURN(code);
116,579✔
982
}
983

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

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

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

1002
    if (thisConflict) {
4!
1003
      mError("trans:%d, db:%s stb:%s type:%d, can't execute since conflict with compact:%d db:%s", pTrans->id,
4!
1004
             pTrans->dbname, pTrans->stbname, pTrans->conflict, pCompact->compactId, pCompact->dbname);
1005
      conflict = true;
4✔
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);
4✔
1011
  }
1012

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

1019
  TAOS_RETURN(code);
169✔
1020
}
1021

1022
static bool mndTransActionsOfSameType(SArray *pActions) {
94,526✔
1023
  int32_t size = taosArrayGetSize(pActions);
94,526✔
1024
  ETrnAct lastActType = TRANS_ACTION_NULL;
94,526✔
1025
  bool    same = true;
94,526✔
1026
  for (int32_t i = 0; i < size; ++i) {
378,294✔
1027
    STransAction *pAction = taosArrayGet(pActions, i);
283,768✔
1028
    if (i > 0) {
283,768✔
1029
      if (lastActType != pAction->actionType) {
214,426!
1030
        same = false;
×
1031
        break;
×
1032
      }
1033
    }
1034
    lastActType = pAction->actionType;
283,768✔
1035
  }
1036
  return same;
94,526✔
1037
}
1038

1039
static int32_t mndTransCheckParallelActions(SMnode *pMnode, STrans *pTrans) {
40,831✔
1040
  int32_t code = 0;
40,831✔
1041
  if (pTrans->exec == TRN_EXEC_PARALLEL) {
40,831✔
1042
    if (mndTransActionsOfSameType(pTrans->redoActions) == false) {
39,598!
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) {
39,598✔
1049
      if (mndTransActionsOfSameType(pTrans->undoActions) == false) {
14,097!
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);
40,831✔
1058
}
1059

1060
static int32_t mndTransCheckCommitActions(SMnode *pMnode, STrans *pTrans) {
40,831✔
1061
  int32_t code = 0;
40,831✔
1062
  if (!pTrans->changeless && taosArrayGetSize(pTrans->commitActions) <= 0) {
40,831!
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) {
40,831!
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);
40,831✔
1074
}
1075

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

1082
  mInfo("trans:%d, action list:", pTrans->id);
40,831!
1083
  int32_t index = 0;
40,831✔
1084
  for (int32_t i = 0; i < taosArrayGetSize(pTrans->prepareActions); ++i, ++index) {
59,478✔
1085
    STransAction *pAction = taosArrayGet(pTrans->prepareActions, i);
18,647✔
1086
    mInfo("trans:%d, action:%d, %s:%d sdbType:%s, sdbStatus:%s", pTrans->id, index,
18,647!
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) {
125,063✔
1091
    STransAction *pAction = taosArrayGet(pTrans->redoActions, i);
84,232✔
1092
    mInfo("trans:%d, action:%d, %s:%d msgType:%s", pTrans->id, index,
84,232!
1093
          mndTransStr(pAction->stage), pAction->id, TMSG_INFO(pAction->msgType));;
1094
  }
1095

1096
  for (int32_t i = 0; i < taosArrayGetSize(pTrans->commitActions); ++i, ++index) {
229,314✔
1097
    STransAction *pAction = taosArrayGet(pTrans->commitActions, i);
188,483✔
1098
    mInfo("trans:%d, action:%d, %s:%d sdbType:%s, sdbStatus:%s", pTrans->id, index,
188,483!
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) {
84,541✔
1103
    STransAction *pAction = taosArrayGet(pTrans->undoActions, i);
43,710✔
1104
    if(pAction->actionType == TRANS_ACTION_MSG){
43,710✔
1105
      mInfo("trans:%d, action:%d, %s:%d msgType:%s", pTrans->id, index,
30,627!
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,
13,083!
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));
40,831!
1116

1117
  TAOS_CHECK_RETURN(mndTransCheckParallelActions(pMnode, pTrans));
40,831!
1118

1119
  TAOS_CHECK_RETURN(mndTransCheckCommitActions(pMnode, pTrans));
40,831!
1120

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

1131
  STrans *pNew = mndAcquireTrans(pMnode, pTrans->id);
40,823✔
1132
  if (pNew == NULL) {
40,823!
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;
40,823✔
1140
  pNew->rpcRsp = pTrans->rpcRsp;
40,823✔
1141
  pNew->rpcRspLen = pTrans->rpcRspLen;
40,823✔
1142
  pNew->mTraceId = pTrans->mTraceId;
40,823✔
1143
  pTrans->pRpcArray = NULL;
40,823✔
1144
  pTrans->rpcRsp = NULL;
40,823✔
1145
  pTrans->rpcRspLen = 0;
40,823✔
1146

1147
  mInfo("trans:%d, execute transaction in prepare", pTrans->id);
40,823!
1148
  mndTransExecute(pMnode, pNew, false);
40,823✔
1149
  mndReleaseTrans(pMnode, pNew);
40,823✔
1150
  // TDOD change to TAOS_RETURN(code);
1151
  return 0;
40,823✔
1152
}
1153

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

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

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

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

1191
  if (pTrans->stage == TRN_STAGE_FINISH) {
239,048✔
1192
    sendRsp = true;
90,559✔
1193
  }
1194

1195
  if (pTrans->policy == TRN_POLICY_ROLLBACK) {
239,048✔
1196
    if (pTrans->stage == TRN_STAGE_UNDO_ACTION || pTrans->stage == TRN_STAGE_ROLLBACK) {
70,829✔
1197
      if (code == 0) code = TSDB_CODE_MND_TRANS_UNKNOW_ERROR;
58✔
1198
      sendRsp = true;
58✔
1199
    }
1200
  } else {
1201
    if (pTrans->stage == TRN_STAGE_REDO_ACTION) {
168,219✔
1202
      if (code == TSDB_CODE_SYN_NOT_LEADER || code == TSDB_CODE_SYN_RESTORING || code == TSDB_CODE_APP_IS_STARTING ||
111,084!
1203
          code == TSDB_CODE_SYN_PROPOSE_NOT_READY) {
1204
        if (pTrans->failedTimes > 60) sendRsp = true;
×
1205
      } else {
1206
        if (pTrans->failedTimes > 6) sendRsp = true;
111,084✔
1207
      }
1208
      if (code == 0) code = TSDB_CODE_MND_TRANS_UNKNOW_ERROR;
111,084✔
1209
    }
1210
  }
1211

1212
  if (!sendRsp) {
239,048✔
1213
    return;
148,429✔
1214
  } else {
1215
    mInfo("vgId:1, trans:%d, start to send rsp, stage:%s failedTimes:%d code:0x%x", pTrans->id,
90,619!
1216
          mndTransStr(pTrans->stage), pTrans->failedTimes, code);
1217
  }
1218

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

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

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

1243
      SRpcMsg rspMsg = {.code = code, .info = *pInfo};
23,616✔
1244

1245
      if (pTrans->originRpcType == TDMT_MND_CREATE_DB) {
23,616✔
1246
        mInfo("trans:%d, origin msgtype:%s", pTrans->id, TMSG_INFO(pTrans->originRpcType));
3,768!
1247
        SDbObj *pDb = mndAcquireDb(pMnode, pTrans->dbname);
3,768✔
1248
        if (pDb != NULL) {
3,768!
1249
          for (int32_t j = 0; j < 12; j++) {
4,578✔
1250
            bool ready = mndIsDbReady(pMnode, pDb);
4,575✔
1251
            if (!ready) {
4,575✔
1252
              mInfo("trans:%d, db:%s not ready yet, wait %d times", pTrans->id, pTrans->dbname, j);
810!
1253
              taosMsleep(1000);
810✔
1254
            } else {
1255
              break;
3,765✔
1256
            }
1257
          }
1258
        }
1259
        mndReleaseDb(pMnode, pDb);
3,768✔
1260
      } else if (pTrans->originRpcType == TDMT_MND_CREATE_STB) {
19,848✔
1261
        void   *pCont = NULL;
6,513✔
1262
        int32_t contLen = 0;
6,513✔
1263
        if (0 == mndBuildSMCreateStbRsp(pMnode, pTrans->dbname, pTrans->stbname, &pCont, &contLen)) {
6,513✔
1264
          mndTransSetRpcRsp(pTrans, pCont, contLen);
6,508✔
1265
        }
1266
      } else if (pTrans->originRpcType == TDMT_MND_CREATE_INDEX) {
13,335✔
1267
        void   *pCont = NULL;
8✔
1268
        int32_t contLen = 0;
8✔
1269
        if (0 == mndBuildSMCreateStbRsp(pMnode, pTrans->dbname, pTrans->stbname, &pCont, &contLen)) {
8!
1270
          mndTransSetRpcRsp(pTrans, pCont, contLen);
8✔
1271
        }
1272
      }
1273

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

1283
      tmsgSendRsp(&rspMsg);
23,616✔
1284

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

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

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

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

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

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

1339
    // pTrans->lastErrorNo = pRsp->code;
1340
    mndSetTransLastAction(pTrans, pAction);
87,655✔
1341

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

1349
  mInfo("trans:%d, execute transaction in process response", pTrans->id);
87,655!
1350
  mndTransExecute(pMnode, pTrans, true);
87,655✔
1351

1352
_OVER:
87,655✔
1353
  mndReleaseTrans(pMnode, pTrans);
87,655✔
1354
  TAOS_RETURN(code);
87,655✔
1355
}
1356

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

1372
static void mndTransResetActions(SMnode *pMnode, STrans *pTrans, SArray *pArray) {
6✔
1373
  int32_t numOfActions = taosArrayGetSize(pArray);
6✔
1374

1375
  for (int32_t action = 0; action < numOfActions; ++action) {
30✔
1376
    STransAction *pAction = taosArrayGet(pArray, action);
24✔
1377
    if (pAction->msgSent && pAction->msgReceived &&
24!
1378
        (pAction->errCode == 0 || pAction->errCode == pAction->acceptableCode))
24✔
1379
      continue;
18✔
1380
    if (pAction->rawWritten && (pAction->errCode == 0 || pAction->errCode == pAction->acceptableCode)) continue;
6!
1381

1382
    mndTransResetAction(pMnode, pTrans, pAction);
6✔
1383
  }
1384
}
6✔
1385

1386
// execute in sync context
1387
static int32_t mndTransWriteSingleLog(SMnode *pMnode, STrans *pTrans, STransAction *pAction, bool topHalf) {
424,295✔
1388
  if (pAction->rawWritten) return 0;
424,295✔
1389
  if (topHalf) {
234,293✔
1390
    TAOS_RETURN(TSDB_CODE_MND_TRANS_CTX_SWITCH);
10✔
1391
  }
1392

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

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

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

1417
  TAOS_RETURN(code);
234,283✔
1418
}
1419

1420
// execute in trans context
1421
static int32_t mndTransSendSingleMsg(SMnode *pMnode, STrans *pTrans, STransAction *pAction, bool topHalf,
632,690✔
1422
                                     bool notSend) {
1423
  if (pAction->msgSent) return 0;
632,690✔
1424
  if (mndCannotExecuteTrans(pMnode, topHalf)) {
120,376✔
1425
    TAOS_RETURN(TSDB_CODE_MND_TRANS_CTX_SWITCH);
32,448✔
1426
  }
1427

1428
  if (notSend) {
87,928✔
1429
    mInfo("trans:%d, action:%d skip to execute msg action", pTrans->id, pAction->id);
180!
1430
    return 0;
180✔
1431
  }
1432

1433
#ifndef TD_ASTRA_32
1434
  int64_t signature = pTrans->id;
87,748✔
1435
  signature = (signature << 32);
87,748✔
1436
  signature += pAction->id;
87,748✔
1437

1438
  SRpcMsg rpcMsg = {.msgType = pAction->msgType, .contLen = pAction->contLen, .info.ahandle = (void *)signature};
87,748✔
1439
#else
1440
  SRpcMsg rpcMsg = {.msgType = pAction->msgType,
1441
                    .contLen = pAction->contLen,
1442
                    .info.ahandle = (void *)pTrans->id,
1443
                    .info.ahandleEx = (void *)pAction->id};
1444
#endif
1445
  rpcMsg.pCont = rpcMallocCont(pAction->contLen);
87,748✔
1446
  if (rpcMsg.pCont == NULL) {
87,748!
1447
    terrno = TSDB_CODE_OUT_OF_MEMORY;
×
1448
    return -1;
×
1449
  }
1450
  rpcMsg.info.traceId.rootId = pTrans->mTraceId;
87,748✔
1451
  rpcMsg.info.notFreeAhandle = 1;
87,748✔
1452

1453
  memcpy(rpcMsg.pCont, pAction->pCont, pAction->contLen);
87,748✔
1454

1455
  char    detail[1024] = {0};
87,748✔
1456
  int32_t len = tsnprintf(detail, sizeof(detail), "msgType:%s numOfEps:%d inUse:%d", TMSG_INFO(pAction->msgType),
87,748!
1457
                          pAction->epSet.numOfEps, pAction->epSet.inUse);
87,748✔
1458
  for (int32_t i = 0; i < pAction->epSet.numOfEps; ++i) {
181,458✔
1459
    len += tsnprintf(detail + len, sizeof(detail) - len, " ep:%d-%s:%u", i, pAction->epSet.eps[i].fqdn,
93,710✔
1460
                     pAction->epSet.eps[i].port);
93,710✔
1461
  }
1462

1463
  int32_t code = tmsgSendReq(&pAction->epSet, &rpcMsg);
87,748✔
1464
  if (code == 0) {
87,748✔
1465
    pAction->msgSent = 1;
87,747✔
1466
    // pAction->msgReceived = 0;
1467
    pAction->errCode = TSDB_CODE_ACTION_IN_PROGRESS;
87,747✔
1468
    pAction->startTime = taosGetTimestampMs();
87,747✔
1469
    pAction->endTime = 0;
87,747✔
1470
    mInfo("trans:%d, %s:%d is sent, %s", pTrans->id, mndTransStr(pAction->stage), pAction->id, detail);
87,747!
1471

1472
    mndSetTransLastAction(pTrans, pAction);
87,747✔
1473
  } else {
1474
    pAction->msgSent = 0;
1✔
1475
    pAction->msgReceived = 0;
1✔
1476
    pAction->errCode = (terrno != 0) ? terrno : code;
1!
1477
    mError("trans:%d, %s:%d not send since %s, %s", pTrans->id, mndTransStr(pAction->stage), pAction->id, terrstr(),
1!
1478
           detail);
1479

1480
    mndSetTransLastAction(pTrans, pAction);
1✔
1481
  }
1482

1483
  TAOS_RETURN(code);
87,748✔
1484
}
1485

1486
static int32_t mndTransExecNullMsg(SMnode *pMnode, STrans *pTrans, STransAction *pAction, bool topHalf) {
×
1487
  if (!topHalf) return TSDB_CODE_MND_TRANS_CTX_SWITCH;
×
1488
  pAction->rawWritten = 0;
×
1489
  pAction->errCode = 0;
×
1490
  mInfo("trans:%d, %s:%d confirm action executed", pTrans->id, mndTransStr(pAction->stage), pAction->id);
×
1491

1492
  mndSetTransLastAction(pTrans, pAction);
×
1493
  return 0;
×
1494
}
1495

1496
static int32_t mndTransExecSingleAction(SMnode *pMnode, STrans *pTrans, STransAction *pAction, bool topHalf,
1,056,985✔
1497
                                        bool notSend) {
1498
  if (pAction->actionType == TRANS_ACTION_RAW) {
1,056,985✔
1499
    return mndTransWriteSingleLog(pMnode, pTrans, pAction, topHalf);
424,295✔
1500
  } else if (pAction->actionType == TRANS_ACTION_MSG) {
632,690!
1501
    return mndTransSendSingleMsg(pMnode, pTrans, pAction, topHalf, notSend);
632,690✔
1502
  } else {
1503
    return mndTransExecNullMsg(pMnode, pTrans, pAction, topHalf);
×
1504
  }
1505
}
1506

1507
static int32_t mndTransExecSingleActions(SMnode *pMnode, STrans *pTrans, SArray *pArray, bool topHalf, bool notSend) {
221,940✔
1508
  int32_t numOfActions = taosArrayGetSize(pArray);
221,940✔
1509
  int32_t code = 0;
221,940✔
1510

1511
  for (int32_t action = 0; action < numOfActions; ++action) {
1,199,853✔
1512
    STransAction *pAction = taosArrayGet(pArray, action);
1,006,061✔
1513
    code = mndTransExecSingleAction(pMnode, pTrans, pAction, topHalf, notSend);
1,006,061✔
1514
    if (code != 0) {
1,006,061✔
1515
      mInfo("trans:%d, action:%d not executed since %s. numOfActions:%d", pTrans->id, action, tstrerror(code),
28,148!
1516
            numOfActions);
1517
      break;
28,148✔
1518
    }
1519
  }
1520

1521
  return code;
221,940✔
1522
}
1523

1524
static int32_t mndTransExecuteActions(SMnode *pMnode, STrans *pTrans, SArray *pArray, bool topHalf, bool notSend) {
259,723✔
1525
  int32_t numOfActions = taosArrayGetSize(pArray);
259,723✔
1526
  int32_t code = 0;
259,723✔
1527
  if (numOfActions == 0) return 0;
259,723✔
1528

1529
  if ((code = mndTransExecSingleActions(pMnode, pTrans, pArray, topHalf, notSend)) != 0) {
221,940✔
1530
    return code;
28,148✔
1531
  }
1532

1533
  int32_t       numOfExecuted = 0;
193,792✔
1534
  int32_t       errCode = 0;
193,792✔
1535
  STransAction *pErrAction = NULL;
193,792✔
1536
  for (int32_t action = 0; action < numOfActions; ++action) {
1,171,704✔
1537
    STransAction *pAction = taosArrayGet(pArray, action);
977,912✔
1538
    if (pAction->msgReceived || pAction->rawWritten) {
977,912✔
1539
      numOfExecuted++;
688,133✔
1540
      if (pAction->errCode != 0 && pAction->errCode != pAction->acceptableCode) {
688,133✔
1541
        errCode = pAction->errCode;
31✔
1542
        pErrAction = pAction;
31✔
1543
      }
1544
    } else {
1545
      pErrAction = pAction;
289,779✔
1546
    }
1547
  }
1548

1549
  mndSetTransLastAction(pTrans, pErrAction);
193,792✔
1550

1551
  if (numOfExecuted == numOfActions) {
193,792✔
1552
    if (errCode == 0) {
112,601✔
1553
      mInfo("trans:%d, all %d actions execute successfully", pTrans->id, numOfActions);
112,595!
1554
      return 0;
112,595✔
1555
    } else {
1556
      mError("trans:%d, all %d actions executed, code:0x%x", pTrans->id, numOfActions, errCode & 0XFFFF);
6!
1557
      mndTransResetActions(pMnode, pTrans, pArray);
6✔
1558
      terrno = errCode;
6✔
1559
      return errCode;
6✔
1560
    }
1561
  } else {
1562
    mInfo("trans:%d, %d of %d actions executed", pTrans->id, numOfExecuted, numOfActions);
81,191!
1563

1564
    for (int32_t action = 0; action < numOfActions; ++action) {
582,679✔
1565
      STransAction *pAction = taosArrayGet(pArray, action);
501,488✔
1566
      mDebug("trans:%d, %s:%d Sent:%d, Received:%d, errCode:0x%x, acceptableCode:0x%x, retryCode:0x%x", pTrans->id,
501,488✔
1567
             mndTransStr(pAction->stage), pAction->id, pAction->msgSent, pAction->msgReceived, pAction->errCode,
1568
             pAction->acceptableCode, pAction->retryCode);
1569
      if (pAction->msgSent) {
501,488✔
1570
        if (pAction->msgReceived) {
501,308✔
1571
          if (pAction->errCode != 0 && pAction->errCode != pAction->acceptableCode) {
211,709✔
1572
            mndTransResetAction(pMnode, pTrans, pAction);
25✔
1573
            mInfo("trans:%d, %s:%d reset", pTrans->id, mndTransStr(pAction->stage), pAction->id);
25!
1574
          }
1575
        }
1576
      }
1577
    }
1578
    return TSDB_CODE_ACTION_IN_PROGRESS;
81,191✔
1579
  }
1580
}
1581

1582
static int32_t mndTransExecuteRedoActions(SMnode *pMnode, STrans *pTrans, bool topHalf, bool notSend) {
169,113✔
1583
  int32_t code = mndTransExecuteActions(pMnode, pTrans, pTrans->redoActions, topHalf, notSend);
169,113✔
1584
  if (code != 0 && code != TSDB_CODE_ACTION_IN_PROGRESS && code != TSDB_CODE_MND_TRANS_CTX_SWITCH) {
169,113✔
1585
    mError("trans:%d, failed to execute redoActions since:%s, code:0x%x, in %s", pTrans->id, terrstr(), terrno,
5!
1586
           mndStrExecutionContext(topHalf));
1587
  }
1588
  return code;
169,113✔
1589
}
1590

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

1600
static int32_t mndTransExecuteCommitActions(SMnode *pMnode, STrans *pTrans, bool topHalf) {
90,548✔
1601
  int32_t code = mndTransExecuteActions(pMnode, pTrans, pTrans->commitActions, topHalf, true);
90,548✔
1602
  if (code != 0 && code != TSDB_CODE_ACTION_IN_PROGRESS && code != TSDB_CODE_MND_TRANS_CTX_SWITCH) {
90,548!
1603
    mError("trans:%d, failed to execute commitActions since %s. in %s", pTrans->id, terrstr(),
×
1604
           mndStrExecutionContext(topHalf));
1605
  }
1606
  return code;
90,548✔
1607
}
1608

1609
static int32_t mndTransExecuteActionsSerial(SMnode *pMnode, STrans *pTrans, SArray *pActions, bool topHalf) {
20,136✔
1610
  int32_t code = 0;
20,136✔
1611
  int32_t numOfActions = taosArrayGetSize(pActions);
20,136✔
1612
  if (numOfActions == 0) return code;
20,136✔
1613

1614
  if (pTrans->actionPos >= numOfActions) {
20,134✔
1615
    return code;
1,519✔
1616
  }
1617

1618
  mInfo("trans:%d, execute %d actions serial, begin at action:%d, stage:%s", pTrans->id, numOfActions,
18,615!
1619
        pTrans->actionPos, mndTransStr(pTrans->stage));
1620

1621
  for (int32_t action = pTrans->actionPos; action < numOfActions; ++action) {
28,516✔
1622
    STransAction *pAction = taosArrayGet(pActions, action);
27,285✔
1623

1624
    mInfo("trans:%d, current action:%d, stage:%s, actionType(1:msg,2:log):%d, msgSent:%d, msgReceived:%d", 
27,285!
1625
          pTrans->id, pTrans->actionPos, mndTransStr(pAction->stage), pAction->actionType, pAction->msgSent,
1626
          pAction->msgReceived);
1627

1628
    code = mndTransExecSingleAction(pMnode, pTrans, pAction, topHalf, false);
27,285✔
1629
    if (code == 0) {
27,285✔
1630
      if (pAction->msgSent) {
22,974✔
1631
        if (pAction->msgReceived) {
19,599✔
1632
          if (pAction->errCode != 0 && pAction->errCode != pAction->acceptableCode) {
8,389✔
1633
            code = pAction->errCode;
4,839✔
1634
            mndTransResetAction(pMnode, pTrans, pAction);
4,839✔
1635
          } else {
1636
            mInfo("trans:%d, %s:%d execute successfully", pTrans->id, mndTransStr(pAction->stage), action);
3,550!
1637
          }
1638
        } else {
1639
          code = TSDB_CODE_ACTION_IN_PROGRESS;
11,210✔
1640
        }
1641
      } else if (pAction->rawWritten) {
3,375!
1642
        if (pAction->errCode != 0 && pAction->errCode != pAction->acceptableCode) {
3,375!
1643
          code = pAction->errCode;
×
1644
        } else {
1645
          mInfo("trans:%d, %s:%d write successfully", pTrans->id, mndTransStr(pAction->stage), action);
3,375!
1646
        }
1647
      } else {
1648
      }
1649
    }
1650

1651
    if (code == 0) {
27,285✔
1652
      pTrans->failedTimes = 0;
6,925✔
1653
    }
1654
    mndSetTransLastAction(pTrans, pAction);
27,285✔
1655

1656
    if (mndCannotExecuteTrans(pMnode, topHalf)) {
27,285✔
1657
      pTrans->lastErrorNo = code;
6,156✔
1658
      pTrans->code = code;
6,156✔
1659
      mInfo("trans:%d, %s:%d, cannot execute next action in %s, code:%s", pTrans->id, mndTransStr(pAction->stage),
6,156!
1660
            action, mndStrExecutionContext(topHalf), tstrerror(code));
1661
      break;
6,156✔
1662
    }
1663

1664
    if (code == 0) {
21,129✔
1665
      pTrans->code = 0;
5,080✔
1666
      pTrans->actionPos++;
5,080✔
1667
      mInfo("trans:%d, %s:%d is executed and need sync to other mnodes", pTrans->id, mndTransStr(pAction->stage),
5,080!
1668
            pAction->id);
1669
      (void)taosThreadMutexUnlock(&pTrans->mutex);
5,080✔
1670
      code = mndTransSync(pMnode, pTrans);
5,080✔
1671
      (void)taosThreadMutexLock(&pTrans->mutex);
5,080✔
1672
      if (code != 0) {
5,080!
1673
        pTrans->actionPos--;
×
1674
        pTrans->code = terrno;
×
1675
        mError("trans:%d, %s:%d is executed and failed to sync to other mnodes since %s", pTrans->id,
×
1676
               mndTransStr(pAction->stage), pAction->id, terrstr());
1677
        break;
×
1678
      }
1679
    } else if (code == TSDB_CODE_ACTION_IN_PROGRESS) {
16,049✔
1680
      mInfo("trans:%d, %s:%d is in progress and wait it finish", pTrans->id, mndTransStr(pAction->stage), pAction->id);
11,210!
1681
      break;
11,210✔
1682
    } else if (code == pAction->retryCode || code == TSDB_CODE_SYN_PROPOSE_NOT_READY ||
4,839✔
1683
               code == TSDB_CODE_SYN_RESTORING || code == TSDB_CODE_SYN_NOT_LEADER) {
26✔
1684
      mInfo("trans:%d, %s:%d receive code:0x%x(%s) and retry", pTrans->id, mndTransStr(pAction->stage), pAction->id,
4,821!
1685
            code, tstrerror(code));
1686
      pTrans->lastErrorNo = code;
4,821✔
1687
      taosMsleep(300);
4,821✔
1688
      action--;
4,821✔
1689
      continue;
4,821✔
1690
    } else {
1691
      terrno = code;
18✔
1692
      pTrans->lastErrorNo = code;
18✔
1693
      pTrans->code = code;
18✔
1694
      mInfo("trans:%d, %s:%d receive code:0x%x(%s) and wait another schedule, failedTimes:%d", pTrans->id,
18!
1695
            mndTransStr(pAction->stage), pAction->id, code, tstrerror(code), pTrans->failedTimes);
1696
      break;
18✔
1697
    }
1698
  }
1699

1700
  return code;
18,615✔
1701
}
1702

1703
static int32_t mndTransExecuteRedoActionsSerial(SMnode *pMnode, STrans *pTrans, bool topHalf) {
20,136✔
1704
  int32_t code = TSDB_CODE_ACTION_IN_PROGRESS;
20,136✔
1705
  (void)taosThreadMutexLock(&pTrans->mutex);
20,136✔
1706
  if (pTrans->stage == TRN_STAGE_REDO_ACTION) {
20,136!
1707
    code = mndTransExecuteActionsSerial(pMnode, pTrans, pTrans->redoActions, topHalf);
20,136✔
1708
  }
1709
  (void)taosThreadMutexUnlock(&pTrans->mutex);
20,136✔
1710
  return code;
20,136✔
1711
}
1712

1713
static int32_t mndTransExecuteUndoActionsSerial(SMnode *pMnode, STrans *pTrans, bool topHalf) {
×
1714
  int32_t code = TSDB_CODE_ACTION_IN_PROGRESS;
×
1715
  (void)taosThreadMutexLock(&pTrans->mutex);
×
1716
  if (pTrans->stage == TRN_STAGE_UNDO_ACTION) {
×
1717
    code = mndTransExecuteActionsSerial(pMnode, pTrans, pTrans->undoActions, topHalf);
×
1718
  }
1719
  (void)taosThreadMutexUnlock(&pTrans->mutex);
×
1720
  return code;
×
1721
}
1722

1723
bool mndTransPerformPrepareStage(SMnode *pMnode, STrans *pTrans, bool topHalf) {
49,768✔
1724
  bool    continueExec = true;
49,768✔
1725
  int32_t code = 0;
49,768✔
1726
  terrno = 0;
49,768✔
1727

1728
  int32_t numOfActions = taosArrayGetSize(pTrans->prepareActions);
49,768✔
1729
  if (numOfActions == 0) goto _OVER;
49,768✔
1730

1731
  mInfo("trans:%d, execute %d prepare actions.", pTrans->id, numOfActions);
14,330!
1732

1733
  for (int32_t action = 0; action < numOfActions; ++action) {
37,969✔
1734
    STransAction *pAction = taosArrayGet(pTrans->prepareActions, action);
23,639✔
1735
    code = mndTransExecSingleAction(pMnode, pTrans, pAction, topHalf, true);
23,639✔
1736
    if (code != 0) {
23,639!
1737
      terrno = code;
×
1738
      mError("trans:%d, failed to execute prepare action:%d, numOfActions:%d, since %s", pTrans->id, action,
×
1739
             numOfActions, tstrerror(code));
1740
      return false;
×
1741
    }
1742
  }
1743

1744
_OVER:
14,330✔
1745
  pTrans->stage = TRN_STAGE_REDO_ACTION;
49,768✔
1746
  mInfo("trans:%d, stage from prepare to redoAction", pTrans->id);
49,768!
1747
  return continueExec;
49,768✔
1748
}
1749

1750
static bool mndTransPerformRedoActionStage(SMnode *pMnode, STrans *pTrans, bool topHalf, bool notSend) {
189,249✔
1751
  bool    continueExec = true;
189,249✔
1752
  int32_t code = 0;
189,249✔
1753
  terrno = 0;
189,249✔
1754

1755
  if (pTrans->exec == TRN_EXEC_SERIAL) {
189,249✔
1756
    code = mndTransExecuteRedoActionsSerial(pMnode, pTrans, topHalf);
20,136✔
1757
  } else {
1758
    code = mndTransExecuteRedoActions(pMnode, pTrans, topHalf, notSend);
169,113✔
1759
  }
1760

1761
  if (code != 0 && code != TSDB_CODE_MND_TRANS_CTX_SWITCH && mndTransIsInSyncContext(topHalf)) {
189,249!
1762
    pTrans->lastErrorNo = code;
×
1763
    pTrans->code = code;
×
1764
    mInfo(
×
1765
        "trans:%d, failed to execute, will retry redo action stage in 100 ms , in %s, "
1766
        "continueExec:%d, code:%s",
1767
        pTrans->id, mndStrExecutionContext(topHalf), continueExec, tstrerror(code));
1768
    taosMsleep(100);
×
1769
    return true;
×
1770
  } else {
1771
    if (mndCannotExecuteTrans(pMnode, topHalf)) {
189,249✔
1772
      mInfo("trans:%d, cannot continue to execute redo action stage in %s, continueExec:%d, code:%s", pTrans->id,
56,035!
1773
            mndStrExecutionContext(topHalf), continueExec, tstrerror(code));
1774
      return false;
56,035✔
1775
    }
1776
  }
1777
  terrno = code;
133,214✔
1778

1779
  if (code == 0) {
133,214✔
1780
    pTrans->code = 0;
40,830✔
1781
    pTrans->stage = TRN_STAGE_COMMIT;
40,830✔
1782
    mInfo("trans:%d, stage from redoAction to commit", pTrans->id);
40,830!
1783
    continueExec = true;
40,830✔
1784
  } else if (code == TSDB_CODE_ACTION_IN_PROGRESS || code == TSDB_CODE_MND_TRANS_CTX_SWITCH) {
92,384!
1785
    mInfo("trans:%d, stage keep on redoAction since %s", pTrans->id, tstrerror(code));
92,361!
1786
    continueExec = false;
92,361✔
1787
  } else {
1788
    pTrans->failedTimes++;
23✔
1789
    pTrans->code = terrno;
23✔
1790
    if (pTrans->policy == TRN_POLICY_ROLLBACK) {
23✔
1791
      if (pTrans->lastAction != 0) {
5✔
1792
        STransAction *pAction = taosArrayGet(pTrans->redoActions, pTrans->lastAction);
3✔
1793
        if (pAction->retryCode != 0 && pAction->retryCode == pAction->errCode) {
3!
1794
          if (pTrans->failedTimes < 6) {
×
1795
            mError("trans:%d, stage keep on redoAction since action:%d code:0x%x not 0x%x, failedTimes:%d", pTrans->id,
×
1796
                   pTrans->lastAction, pTrans->code, pAction->retryCode, pTrans->failedTimes);
1797
            taosMsleep(1000);
×
1798
            continueExec = true;
×
1799
            return true;
×
1800
          }
1801
        }
1802
      }
1803

1804
      pTrans->stage = TRN_STAGE_ROLLBACK;
5✔
1805
      pTrans->actionPos = 0;
5✔
1806
      mError("trans:%d, stage from redoAction to rollback since %s, and set actionPos to %d", pTrans->id, terrstr(),
5!
1807
             pTrans->actionPos);
1808
      continueExec = true;
5✔
1809
    } else {
1810
      mError("trans:%d, stage keep on redoAction since %s, failedTimes:%d", pTrans->id, terrstr(), pTrans->failedTimes);
18!
1811
      continueExec = false;
18✔
1812
    }
1813
  }
1814

1815
  return continueExec;
133,214✔
1816
}
1817

1818
// execute in trans context
1819
static bool mndTransPerformCommitStage(SMnode *pMnode, STrans *pTrans, bool topHalf) {
40,835✔
1820
  if (mndCannotExecuteTrans(pMnode, topHalf)) return false;
40,835✔
1821

1822
  bool    continueExec = true;
40,830✔
1823
  int32_t code = mndTransCommit(pMnode, pTrans);
40,830✔
1824

1825
  if (code == 0) {
40,830✔
1826
    pTrans->code = 0;
40,818✔
1827
    pTrans->stage = TRN_STAGE_COMMIT_ACTION;
40,818✔
1828
    mInfo("trans:%d, stage from commit to commitAction", pTrans->id);
40,818!
1829
    continueExec = true;
40,818✔
1830
  } else {
1831
    pTrans->code = terrno;
12✔
1832
    pTrans->failedTimes++;
12✔
1833
    mError("trans:%d, stage keep on commit since %s, failedTimes:%d", pTrans->id, terrstr(), pTrans->failedTimes);
12!
1834
    continueExec = false;
12✔
1835
  }
1836

1837
  return continueExec;
40,830✔
1838
}
1839

1840
static bool mndTransPerformCommitActionStage(SMnode *pMnode, STrans *pTrans, bool topHalf) {
90,548✔
1841
  bool    continueExec = true;
90,548✔
1842
  int32_t code = mndTransExecuteCommitActions(pMnode, pTrans, topHalf);
90,548✔
1843

1844
  if (code == 0) {
90,548✔
1845
    pTrans->code = 0;
90,543✔
1846
    pTrans->stage = TRN_STAGE_FINISH;  // TRN_STAGE_PRE_FINISH is not necessary
90,543✔
1847
    mInfo("trans:%d, stage from commitAction to finished", pTrans->id);
90,543!
1848
    continueExec = true;
90,543✔
1849
  } else if (code == TSDB_CODE_MND_TRANS_CTX_SWITCH && topHalf) {
5!
1850
    pTrans->code = 0;
5✔
1851
    pTrans->stage = TRN_STAGE_COMMIT;
5✔
1852
    mInfo("trans:%d, back to commit stage", pTrans->id);
5!
1853
    continueExec = true;
5✔
1854
  } else {
1855
    pTrans->code = terrno;
×
1856
    pTrans->failedTimes++;
×
1857
    mError("trans:%d, stage keep on commitAction since %s, failedTimes:%d", pTrans->id, terrstr(), pTrans->failedTimes);
×
1858
    continueExec = false;
×
1859
  }
1860

1861
  return continueExec;
90,548✔
1862
}
1863

1864
static bool mndTransPerformUndoActionStage(SMnode *pMnode, STrans *pTrans, bool topHalf, bool notSend) {
62✔
1865
  bool    continueExec = true;
62✔
1866
  int32_t code = 0;
62✔
1867

1868
  if (pTrans->exec == TRN_EXEC_SERIAL) {
62!
1869
    code = mndTransExecuteUndoActionsSerial(pMnode, pTrans, topHalf);
×
1870
  } else {
1871
    code = mndTransExecuteUndoActions(pMnode, pTrans, topHalf, notSend);
62✔
1872
  }
1873

1874
  if (mndCannotExecuteTrans(pMnode, topHalf)) return false;
62✔
1875
  terrno = code;
47✔
1876

1877
  if (code == 0) {
47✔
1878
    pTrans->stage = TRN_STAGE_PRE_FINISH;
5✔
1879
    mInfo("trans:%d, stage from undoAction to pre-finish", pTrans->id);
5!
1880
    continueExec = true;
5✔
1881
  } else if (code == TSDB_CODE_ACTION_IN_PROGRESS || code == TSDB_CODE_MND_TRANS_CTX_SWITCH) {
42!
1882
    mInfo("trans:%d, stage keep on undoAction since %s", pTrans->id, tstrerror(code));
40!
1883
    continueExec = false;
40✔
1884
  } else {
1885
    pTrans->failedTimes++;
2✔
1886
    mError("trans:%d, stage keep on undoAction since %s, failedTimes:%d", pTrans->id, terrstr(), pTrans->failedTimes);
2!
1887
    continueExec = false;
2✔
1888
  }
1889

1890
  return continueExec;
47✔
1891
}
1892

1893
// in trans context
1894
static bool mndTransPerformRollbackStage(SMnode *pMnode, STrans *pTrans, bool topHalf) {
5✔
1895
  if (mndCannotExecuteTrans(pMnode, topHalf)) return false;
5!
1896

1897
  bool    continueExec = true;
5✔
1898
  int32_t code = mndTransRollback(pMnode, pTrans);
5✔
1899

1900
  if (code == 0) {
5✔
1901
    pTrans->stage = TRN_STAGE_UNDO_ACTION;
4✔
1902
    continueExec = true;
4✔
1903
  } else {
1904
    pTrans->failedTimes++;
1✔
1905
    mError("trans:%d, stage keep on rollback since %s, failedTimes:%d", pTrans->id, terrstr(), pTrans->failedTimes);
1!
1906
    continueExec = false;
1✔
1907
  }
1908

1909
  return continueExec;
5✔
1910
}
1911

1912
// excute in trans context
1913
static bool mndTransPerformPreFinishStage(SMnode *pMnode, STrans *pTrans, bool topHalf) {
5✔
1914
  if (mndCannotExecuteTrans(pMnode, topHalf)) return false;
5!
1915

1916
  bool    continueExec = true;
5✔
1917
  int32_t code = mndTransPreFinish(pMnode, pTrans);
5✔
1918

1919
  if (code == 0) {
5!
1920
    pTrans->stage = TRN_STAGE_FINISH;
5✔
1921
    mInfo("trans:%d, stage from pre-finish to finish", pTrans->id);
5!
1922
    continueExec = true;
5✔
1923
  } else {
1924
    pTrans->failedTimes++;
×
1925
    mError("trans:%d, stage keep on pre-finish since %s, failedTimes:%d", pTrans->id, terrstr(), pTrans->failedTimes);
×
1926
    continueExec = false;
×
1927
  }
1928

1929
  return continueExec;
5✔
1930
}
1931

1932
static bool mndTransPerformFinishStage(SMnode *pMnode, STrans *pTrans, bool topHalf) {
90,559✔
1933
  bool continueExec = false;
90,559✔
1934
  if (topHalf) return continueExec;
90,559✔
1935

1936
  SSdbRaw *pRaw = mndTransEncode(pTrans);
49,736✔
1937
  if (pRaw == NULL) {
49,736!
1938
    mError("trans:%d, failed to encode while finish trans since %s", pTrans->id, terrstr());
×
1939
    return false;
×
1940
  }
1941
  TAOS_CHECK_RETURN(sdbSetRawStatus(pRaw, SDB_STATUS_DROPPED));
49,736!
1942

1943
  int32_t code = sdbWrite(pMnode->pSdb, pRaw);
49,736✔
1944
  if (code != 0) {
49,736!
1945
    mError("trans:%d, failed to write sdb since %s", pTrans->id, terrstr());
×
1946
  }
1947

1948
  mInfo("trans:%d, execute finished, code:0x%x, failedTimes:%d createTime:%" PRId64, pTrans->id, pTrans->code,
49,736!
1949
        pTrans->failedTimes, pTrans->createdTime);
1950
  return continueExec;
49,736✔
1951
}
1952

1953
void mndTransExecuteImp(SMnode *pMnode, STrans *pTrans, bool topHalf, bool notSend) {
239,048✔
1954
  bool continueExec = true;
239,048✔
1955

1956
  while (continueExec) {
650,311✔
1957
    mInfo("trans:%d, continue to execute stage:%s in %s, createTime:%" PRId64, pTrans->id,
411,263!
1958
          mndTransStr(pTrans->stage), mndStrExecutionContext(topHalf), pTrans->createdTime);
1959
    pTrans->lastExecTime = taosGetTimestampMs();
411,263✔
1960
    switch (pTrans->stage) {
411,263!
1961
      case TRN_STAGE_PREPARE:
×
1962
        continueExec = mndTransPerformPrepareStage(pMnode, pTrans, topHalf);
×
1963
        break;
×
1964
      case TRN_STAGE_REDO_ACTION:
189,249✔
1965
        continueExec = mndTransPerformRedoActionStage(pMnode, pTrans, topHalf, notSend);
189,249✔
1966
        break;
189,249✔
1967
      case TRN_STAGE_COMMIT:
40,835✔
1968
        continueExec = mndTransPerformCommitStage(pMnode, pTrans, topHalf);
40,835✔
1969
        break;
40,835✔
1970
      case TRN_STAGE_COMMIT_ACTION:
90,548✔
1971
        continueExec = mndTransPerformCommitActionStage(pMnode, pTrans, topHalf);
90,548✔
1972
        break;
90,548✔
1973
      case TRN_STAGE_ROLLBACK:
5✔
1974
        continueExec = mndTransPerformRollbackStage(pMnode, pTrans, topHalf);
5✔
1975
        break;
5✔
1976
      case TRN_STAGE_UNDO_ACTION:
62✔
1977
        continueExec = mndTransPerformUndoActionStage(pMnode, pTrans, topHalf, notSend);
62✔
1978
        break;
62✔
1979
      case TRN_STAGE_PRE_FINISH:
5✔
1980
        continueExec = mndTransPerformPreFinishStage(pMnode, pTrans, topHalf);
5✔
1981
        break;
5✔
1982
      case TRN_STAGE_FINISH:
90,559✔
1983
        continueExec = mndTransPerformFinishStage(pMnode, pTrans, topHalf);
90,559✔
1984
        break;
90,559✔
1985
      default:
×
1986
        continueExec = false;
×
1987
        break;
×
1988
    }
1989
  }
1990

1991
  mndTransSendRpcRsp(pMnode, pTrans);
239,048✔
1992
}
239,048✔
1993

1994
// start trans, pullup, receive rsp, kill
1995
void mndTransExecute(SMnode *pMnode, STrans *pTrans, bool notSend) {
133,478✔
1996
  bool topHalf = true;
133,478✔
1997
  mndTransExecuteImp(pMnode, pTrans, topHalf, notSend);
133,478✔
1998
}
133,478✔
1999

2000
// update trans
2001
void mndTransRefresh(SMnode *pMnode, STrans *pTrans) {
105,570✔
2002
  bool topHalf = false;
105,570✔
2003
  mndTransExecuteImp(pMnode, pTrans, topHalf, false);
105,570✔
2004
}
105,570✔
2005

2006
static int32_t mndProcessTransTimer(SRpcMsg *pReq) {
33,057✔
2007
  mTrace("start to process trans timer");
33,057✔
2008
  mndTransPullup(pReq->info.node);
33,057✔
2009
  return 0;
33,057✔
2010
}
2011

2012
int32_t mndKillTrans(SMnode *pMnode, STrans *pTrans) {
×
2013
  SArray *pArray = NULL;
×
2014
  if (pTrans->stage == TRN_STAGE_REDO_ACTION) {
×
2015
    pArray = pTrans->redoActions;
×
2016
  } else if (pTrans->stage == TRN_STAGE_UNDO_ACTION) {
×
2017
    pArray = pTrans->undoActions;
×
2018
  } else {
2019
    TAOS_RETURN(TSDB_CODE_MND_TRANS_INVALID_STAGE);
×
2020
  }
2021

2022
  if(pTrans->ableToBeKilled == false){
×
2023
    return TSDB_CODE_MND_TRANS_NOT_ABLE_TO_kILLED;
×
2024
  }
2025
  
2026
  if(pTrans->killMode == TRN_KILL_MODE_SKIP){
×
2027
    for (int32_t i = 0; i < taosArrayGetSize(pArray); ++i) {
×
2028
      STransAction *pAction = taosArrayGet(pArray, i);
×
2029
      mInfo("trans:%d, %s:%d set processed for kill msg received, errCode from %s to success", pTrans->id,
×
2030
            mndTransStr(pAction->stage), i, tstrerror(pAction->errCode));
2031
      pAction->msgSent = 1;
×
2032
      pAction->msgReceived = 1;
×
2033
      pAction->errCode = 0;
×
2034
    }
2035
  }
2036
  else if(pTrans->killMode == TRN_KILL_MODE_INTERUPT){
×
2037
    pTrans->stage = TRN_STAGE_PRE_FINISH;
×
2038
  }
2039
  else{
2040
    return TSDB_CODE_MND_TRANS_NOT_ABLE_TO_kILLED;
×
2041
  }
2042

2043
  mInfo("trans:%d, execute transaction in kill trans", pTrans->id);
×
2044
  mndTransExecute(pMnode, pTrans, true);
×
2045
  return 0;
×
2046
}
2047

2048
static int32_t mndProcessKillTransReq(SRpcMsg *pReq) {
1✔
2049
  SMnode       *pMnode = pReq->info.node;
1✔
2050
  SKillTransReq killReq = {0};
1✔
2051
  int32_t       code = -1;
1✔
2052
  STrans       *pTrans = NULL;
1✔
2053

2054
  if (tDeserializeSKillTransReq(pReq->pCont, pReq->contLen, &killReq) != 0) {
1!
2055
    code = TSDB_CODE_INVALID_MSG;
×
2056
    goto _OVER;
×
2057
  }
2058

2059
  mInfo("trans:%d, start to kill", killReq.transId);
1!
2060
  if ((code = mndCheckOperPrivilege(pMnode, pReq->info.conn.user, MND_OPER_KILL_TRANS)) != 0) {
1!
2061
    goto _OVER;
1✔
2062
  }
2063

2064
  pTrans = mndAcquireTrans(pMnode, killReq.transId);
×
2065
  if (pTrans == NULL) {
×
2066
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
2067
    if (terrno != 0) code = terrno;
×
2068
    goto _OVER;
×
2069
  }
2070

2071
  code = mndKillTrans(pMnode, pTrans);
×
2072

2073
_OVER:
1✔
2074
  if (code != 0) {
1!
2075
    mError("trans:%d, failed to kill since %s", killReq.transId, terrstr());
1!
2076
  }
2077

2078
  mndReleaseTrans(pMnode, pTrans);
1✔
2079
  TAOS_RETURN(code);
1✔
2080
}
2081

2082
static int32_t mndCompareTransId(int32_t *pTransId1, int32_t *pTransId2) { return *pTransId1 >= *pTransId2 ? 1 : 0; }
348✔
2083

2084
void mndTransPullup(SMnode *pMnode) {
33,551✔
2085
  SSdb   *pSdb = pMnode->pSdb;
33,551✔
2086
  SArray *pArray = taosArrayInit(sdbGetSize(pSdb, SDB_TRANS), sizeof(int32_t));
33,551✔
2087
  if (pArray == NULL) return;
33,551!
2088

2089
  void *pIter = NULL;
33,551✔
2090
  while (1) {
5,000✔
2091
    STrans *pTrans = NULL;
38,551✔
2092
    pIter = sdbFetch(pMnode->pSdb, SDB_TRANS, pIter, (void **)&pTrans);
38,551✔
2093
    if (pIter == NULL) break;
38,551✔
2094
    if (taosArrayPush(pArray, &pTrans->id) == NULL) {
10,000!
2095
      mError("failed to put trans into array, trans:%d, but pull up will continute", pTrans->id);
×
2096
    }
2097
    sdbRelease(pSdb, pTrans);
5,000✔
2098
  }
2099

2100
  taosArraySort(pArray, (__compar_fn_t)mndCompareTransId);
33,551✔
2101

2102
  for (int32_t i = 0; i < taosArrayGetSize(pArray); ++i) {
38,551✔
2103
    int32_t *pTransId = taosArrayGet(pArray, i);
5,000✔
2104
    STrans  *pTrans = mndAcquireTrans(pMnode, *pTransId);
5,000✔
2105
    if (pTrans != NULL) {
5,000!
2106
      mInfo("trans:%d, execute transaction in trans pullup", pTrans->id);
5,000!
2107
      mndTransExecute(pMnode, pTrans, false);
5,000✔
2108
    }
2109
    mndReleaseTrans(pMnode, pTrans);
5,000✔
2110
  }
2111
  taosArrayDestroy(pArray);
33,551✔
2112
}
2113

2114
static char *formatTimestamp(char *buf, int64_t val, int precision) {
11,862✔
2115
  time_t tt;
2116
  if (precision == TSDB_TIME_PRECISION_MICRO) {
11,862!
2117
    tt = (time_t)(val / 1000000);
×
2118
  }
2119
  if (precision == TSDB_TIME_PRECISION_NANO) {
11,862!
2120
    tt = (time_t)(val / 1000000000);
×
2121
  } else {
2122
    tt = (time_t)(val / 1000);
11,862✔
2123
  }
2124

2125
  struct tm tm;
2126
  if (taosLocalTime(&tt, &tm, NULL, 0, NULL) == NULL) {
11,862!
2127
    mError("failed to get local time");
×
2128
    return NULL;
×
2129
  }
2130
  size_t pos = taosStrfTime(buf, 32, "%Y-%m-%d %H:%M:%S", &tm);
11,862✔
2131

2132
  if (precision == TSDB_TIME_PRECISION_MICRO) {
11,862!
2133
    sprintf(buf + pos, ".%06d", (int)(val % 1000000));
×
2134
  } else if (precision == TSDB_TIME_PRECISION_NANO) {
11,862!
2135
    sprintf(buf + pos, ".%09d", (int)(val % 1000000000));
×
2136
  } else {
2137
    sprintf(buf + pos, ".%03d", (int)(val % 1000));
11,862✔
2138
  }
2139

2140
  return buf;
11,862✔
2141
}
2142

2143
static void mndTransLogAction(STrans *pTrans) {
167✔
2144
  char    detail[512] = {0};
167✔
2145
  int32_t len = 0;
167✔
2146
  int32_t index = 0;
167✔
2147

2148
  if (pTrans->stage == TRN_STAGE_PREPARE) {
167!
2149
    for (int32_t i = 0; i < taosArrayGetSize(pTrans->prepareActions); ++i, ++index) {
×
2150
      len = 0;
×
2151
      STransAction *pAction = taosArrayGet(pTrans->prepareActions, i);
×
2152
      len += snprintf(detail + len, sizeof(detail) - len, "action:%d, %s:%d sdbType:%s, sdbStatus:%s\n", index,
×
2153
                      mndTransStr(pAction->stage), pAction->id, sdbTableName(pAction->pRaw->type),
×
2154
                      sdbStatusName(pAction->pRaw->status));
×
2155
      mDebug("trans:%d, show tran action, detail:%s", pTrans->id, detail);
×
2156
    }
2157
  }
2158

2159
  if (pTrans->stage == TRN_STAGE_REDO_ACTION) {
167!
2160
    for (int32_t i = 0; i < taosArrayGetSize(pTrans->redoActions); ++i, ++index) {
7,528✔
2161
      len = 0;
7,361✔
2162
      STransAction *pAction = taosArrayGet(pTrans->redoActions, i);
7,361✔
2163
      if (pAction->actionType == TRANS_ACTION_MSG) {
7,361✔
2164
        char bufStart[40] = {0};
5,931✔
2165
        (void)formatTimestamp(bufStart, pAction->startTime, TSDB_TIME_PRECISION_MILLI);
5,931✔
2166

2167
        char endStart[40] = {0};
5,931✔
2168
        (void)formatTimestamp(endStart, pAction->endTime, TSDB_TIME_PRECISION_MILLI);
5,931✔
2169
        len += snprintf(detail + len, sizeof(detail) - len,
11,862!
2170
                        "action:%d, %s:%d msgType:%s,"
2171
                        "sent:%d, received:%d, startTime:%s, endTime:%s, ",
2172
                        index, mndTransStr(pAction->stage), pAction->id, TMSG_INFO(pAction->msgType), pAction->msgSent,
5,931✔
2173
                        pAction->msgReceived, bufStart, endStart);
5,931✔
2174

2175
        SEpSet epset = pAction->epSet;
5,931✔
2176
        if (epset.numOfEps > 0) {
5,931!
2177
          len += snprintf(detail + len, sizeof(detail) - len, "numOfEps:%d inUse:%d ", epset.numOfEps, epset.inUse);
5,931✔
2178
          for (int32_t i = 0; i < epset.numOfEps; ++i) {
13,743✔
2179
            len +=
7,812✔
2180
                snprintf(detail + len, sizeof(detail) - len, "ep:%d-%s:%u ", i, epset.eps[i].fqdn, epset.eps[i].port);
7,812✔
2181
          }
2182
        }
2183

2184
        len += snprintf(detail + len, sizeof(detail) - len, ", errCode:0x%x(%s)\n", pAction->errCode & 0xFFFF,
5,931✔
2185
                        tstrerror(pAction->errCode));
2186
      } else {
2187
        len += snprintf(detail + len, sizeof(detail) - len, "action:%d, %s:%d sdbType:%s, sdbStatus:%s, written:%d\n",
1,430✔
2188
                        index, mndTransStr(pAction->stage), pAction->id, sdbTableName(pAction->pRaw->type),
1,430✔
2189
                        sdbStatusName(pAction->pRaw->status), pAction->rawWritten);
1,430✔
2190
      }
2191
      mDebug("trans:%d, show tran action, detail:%s", pTrans->id, detail);
7,361✔
2192
    }
2193
  }
2194

2195
  if (pTrans->stage == TRN_STAGE_COMMIT_ACTION) {
167!
2196
    for (int32_t i = 0; i < taosArrayGetSize(pTrans->commitActions); ++i, ++index) {
×
2197
      len = 0;
×
2198
      STransAction *pAction = taosArrayGet(pTrans->commitActions, i);
×
2199
      len += snprintf(detail + len, sizeof(detail) - len, "action:%d, %s:%d sdbType:%s, sdbStatus:%s\n", index,
×
2200
                      mndTransStr(pAction->stage), i, sdbTableName(pAction->pRaw->type),
×
2201
                      sdbStatusName(pAction->pRaw->status));
×
2202
      mDebug("trans:%d, show tran action, detail:%s", pTrans->id, detail);
×
2203
    }
2204

2205
    for (int32_t i = 0; i < taosArrayGetSize(pTrans->undoActions); ++i, ++index) {
×
2206
      len = 0;
×
2207
      STransAction *pAction = taosArrayGet(pTrans->undoActions, i);
×
2208
      if (pAction->actionType == TRANS_ACTION_MSG) {
×
2209
        len += snprintf(detail + len, sizeof(detail) - len, "action:%d, %s:%d msgType:%s\n", index,
×
2210
                        mndTransStr(pAction->stage), pAction->id, TMSG_INFO(pAction->msgType));
×
2211
        ;
2212
      } else {
2213
        len += snprintf(detail + len, sizeof(detail) - len, "action:%d, %s:%d sdbType:%s, sdbStatus:%s\n", index,
×
2214
                        mndTransStr(pAction->stage), pAction->id, sdbTableName(pAction->pRaw->type),
×
2215
                        sdbStatusName(pAction->pRaw->status));
×
2216
      }
2217
      mDebug("trans:%d, show tran action, detail:%s", pTrans->id, detail);
×
2218
    }
2219
  }
2220
}
167✔
2221

2222
static int32_t mndRetrieveTrans(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows) {
272✔
2223
  SMnode *pMnode = pReq->info.node;
272✔
2224
  SSdb   *pSdb = pMnode->pSdb;
272✔
2225
  int32_t numOfRows = 0;
272✔
2226
  STrans *pTrans = NULL;
272✔
2227
  int32_t cols = 0;
272✔
2228
  int32_t code = 0;
272✔
2229
  int32_t lino = 0;
272✔
2230

2231
  while (numOfRows < rows) {
439!
2232
    pShow->pIter = sdbFetch(pSdb, SDB_TRANS, pShow->pIter, (void **)&pTrans);
439✔
2233
    if (pShow->pIter == NULL) break;
439✔
2234

2235
    cols = 0;
167✔
2236

2237
    SColumnInfoData *pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
167✔
2238
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)&pTrans->id, false), pTrans, &lino, _OVER);
167!
2239

2240
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
167✔
2241
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)&pTrans->createdTime, false), pTrans, &lino,
167!
2242
                        _OVER);
2243

2244
    char stage[TSDB_TRANS_STAGE_LEN + VARSTR_HEADER_SIZE] = {0};
167✔
2245
    STR_WITH_MAXSIZE_TO_VARSTR(stage, mndTransStr(pTrans->stage), pShow->pMeta->pSchemas[cols].bytes);
167✔
2246
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
167✔
2247
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)stage, false), pTrans, &lino, _OVER);
167!
2248

2249
    char opername[TSDB_TRANS_OPER_LEN + VARSTR_HEADER_SIZE] = {0};
167✔
2250
    STR_WITH_MAXSIZE_TO_VARSTR(opername, pTrans->opername, pShow->pMeta->pSchemas[cols].bytes);
167✔
2251
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
167✔
2252
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)opername, false), pTrans, &lino, _OVER);
167!
2253

2254
    char dbname[TSDB_DB_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
167✔
2255
    STR_WITH_MAXSIZE_TO_VARSTR(dbname, mndGetDbStr(pTrans->dbname), pShow->pMeta->pSchemas[cols].bytes);
167✔
2256
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
167✔
2257
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)dbname, false), pTrans, &lino, _OVER);
167!
2258

2259
    char stbname[TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
167✔
2260
    STR_WITH_MAXSIZE_TO_VARSTR(stbname, mndGetDbStr(pTrans->stbname), pShow->pMeta->pSchemas[cols].bytes);
167✔
2261
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
167✔
2262
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)stbname, false), pTrans, &lino, _OVER);
167!
2263

2264
    const char *killableStr = pTrans->ableToBeKilled ? "yes" : "no";
167!
2265
    char        killableVstr[10 + VARSTR_HEADER_SIZE] = {0};
167✔
2266
    STR_WITH_MAXSIZE_TO_VARSTR(killableVstr, killableStr, 10 + VARSTR_HEADER_SIZE);
167✔
2267
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
167✔
2268
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)killableVstr, false), pTrans, &lino, _OVER);
167!
2269

2270
    /*
2271
    const char *killModeStr = pTrans->killMode == TRN_KILL_MODE_SKIP ? "skip" : "interrupt";
2272
    char        killModeVstr[10 + VARSTR_HEADER_SIZE] = {0};
2273
    STR_WITH_MAXSIZE_TO_VARSTR(killModeVstr, killModeStr, 24);
2274
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
2275
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)killModeVstr, false), pTrans, &lino, _OVER);
2276
    */
2277

2278
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
167✔
2279
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)&pTrans->failedTimes, false), pTrans, &lino,
167!
2280
                        _OVER);
2281

2282
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
167✔
2283
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)&pTrans->lastExecTime, false), pTrans, &lino,
167!
2284
                        _OVER);
2285

2286
    char    lastInfo[TSDB_TRANS_ERROR_LEN + VARSTR_HEADER_SIZE] = {0};
167✔
2287
    char    detail[TSDB_TRANS_ERROR_LEN + 1] = {0};
167✔
2288
    int32_t len = tsnprintf(detail, sizeof(detail), "action:%d code:0x%x(%s) ", pTrans->lastAction,
501✔
2289
                            pTrans->lastErrorNo & 0xFFFF, tstrerror(pTrans->lastErrorNo));
167✔
2290
    SEpSet  epset = pTrans->lastEpset;
167✔
2291
    if (epset.numOfEps > 0) {
167!
2292
      len += tsnprintf(detail + len, sizeof(detail) - len, "msgType:%s numOfEps:%d inUse:%d ",
334!
2293
                       TMSG_INFO(pTrans->lastMsgType), epset.numOfEps, epset.inUse);
334✔
2294
      for (int32_t i = 0; i < pTrans->lastEpset.numOfEps; ++i) {
462✔
2295
        len += snprintf(detail + len, sizeof(detail) - len, "ep:%d-%s:%u ", i, epset.eps[i].fqdn, epset.eps[i].port);
295✔
2296
      }
2297
    }
2298
    STR_WITH_MAXSIZE_TO_VARSTR(lastInfo, detail, pShow->pMeta->pSchemas[cols].bytes);
167✔
2299
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
167✔
2300
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)lastInfo, false), pTrans, &lino, _OVER);
167!
2301

2302
    mndTransLogAction(pTrans);
167✔
2303

2304
    numOfRows++;
167✔
2305
    sdbRelease(pSdb, pTrans);
167✔
2306
  }
2307

2308
_OVER:
×
2309
  if (code != 0) mError("failed to retrieve at line:%d, since %s", lino, tstrerror(code));
272!
2310
  pShow->numOfRows += numOfRows;
272✔
2311
  return numOfRows;
272✔
2312
}
2313

2314
static int32_t mndShowTransCommonColumns(SShowObj *pShow, SSDataBlock *pBlock, STransAction *pAction,
×
2315
                                         int32_t transactionId, int32_t curActionId, int32_t numOfRows, int32_t *cols) {
2316
  int32_t code = 0;
×
2317
  int32_t lino = 0;
×
2318
  int32_t len = 0;
×
2319

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

2323
  char action[30 + 1] = {0};
×
2324
  if (curActionId == pAction->id) {
×
2325
    len += snprintf(action + len, sizeof(action) - len, "%s:%d(%s)<-last", mndTransStr(pAction->stage), pAction->id,
×
2326
                    mndTransTypeStr(pAction->actionType));
2327
  } else {
2328
    len += snprintf(action + len, sizeof(action) - len, "%s:%d(%s)", mndTransStr(pAction->stage), pAction->id,
×
2329
                    mndTransTypeStr(pAction->actionType));
2330
  }
2331
  char actionVStr[30 + VARSTR_HEADER_SIZE] = {0};
×
2332
  STR_WITH_MAXSIZE_TO_VARSTR(actionVStr, action, pShow->pMeta->pSchemas[*cols].bytes);
×
2333
  pColInfo = taosArrayGet(pBlock->pDataBlock, (*cols)++);
×
2334
  TAOS_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)actionVStr, false), &lino, _OVER);
×
2335
_OVER:
×
2336
  if (code != 0) mError("failed to retrieve at line:%d, since %s", lino, tstrerror(code));
×
2337
  return code;
×
2338
}
2339

2340
static void mndShowTransAction(SShowObj *pShow, SSDataBlock *pBlock, STransAction *pAction, int32_t transactionId,
×
2341
                               int32_t curActionId, int32_t rows, int32_t numOfRows) {
2342
  int32_t code = 0;
×
2343
  int32_t lino = 0;
×
2344
  int32_t len = 0;
×
2345
  int32_t cols = 0;
×
2346

2347
  cols = 0;
×
2348

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

2351
  if (pAction->actionType == TRANS_ACTION_MSG) {
×
2352
    int32_t len = 0;
×
2353

2354
    char objType[TSDB_TRANS_OBJTYPE_LEN + 1] = {0};
×
2355
    len += snprintf(objType + len, sizeof(objType) - len, "%s(s:%d,r:%d)", TMSG_INFO(pAction->msgType),
×
2356
                    pAction->msgSent, pAction->msgReceived);
×
2357
    char objTypeVStr[TSDB_TRANS_OBJTYPE_LEN + VARSTR_HEADER_SIZE] = {0};
×
2358
    STR_WITH_MAXSIZE_TO_VARSTR(objTypeVStr, objType, pShow->pMeta->pSchemas[cols].bytes);
×
2359
    SColumnInfoData *pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
×
2360
    TAOS_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)objTypeVStr, false), &lino, _OVER);
×
2361

2362
    char result[TSDB_TRANS_RESULT_LEN + 1] = {0};
×
2363
    len = 0;
×
2364
    len += snprintf(result + len, sizeof(result) - len, "errCode:0x%x(%s)", pAction->errCode & 0xFFFF,
×
2365
                    tstrerror(pAction->errCode));
2366
    char resultVStr[TSDB_TRANS_RESULT_LEN + VARSTR_HEADER_SIZE] = {0};
×
2367
    STR_WITH_MAXSIZE_TO_VARSTR(resultVStr, result, pShow->pMeta->pSchemas[cols].bytes);
×
2368
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
×
2369
    TAOS_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)resultVStr, false), &lino, _OVER);
×
2370

2371
    char target[TSDB_TRANS_TARGET_LEN] = {0};
×
2372
    len = 0;
×
2373
    SEpSet epset = pAction->epSet;
×
2374
    if (epset.numOfEps > 0) {
×
2375
      for (int32_t i = 0; i < epset.numOfEps; ++i) {
×
2376
        len += snprintf(target + len, sizeof(target) - len, "ep:%d-%s:%u,", i, epset.eps[i].fqdn, epset.eps[i].port);
×
2377
      }
2378
      len += snprintf(target + len, sizeof(target) - len, "(%d:%d) ", epset.numOfEps, epset.inUse);
×
2379
    }
2380
    char targetVStr[TSDB_TRANS_TARGET_LEN + VARSTR_HEADER_SIZE] = {0};
×
2381
    STR_WITH_MAXSIZE_TO_VARSTR(targetVStr, target, pShow->pMeta->pSchemas[cols].bytes);
×
2382
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
×
2383
    TAOS_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)targetVStr, false), &lino, _OVER);
×
2384

2385
    char detail[TSDB_TRANS_DETAIL_LEN] = {0};
×
2386
    len = 0;
×
2387
    char bufStart[40] = {0};
×
2388
    if (pAction->startTime > 0) (void)formatTimestamp(bufStart, pAction->startTime, TSDB_TIME_PRECISION_MILLI);
×
2389
    char bufEnd[40] = {0};
×
2390
    if (pAction->endTime > 0) (void)formatTimestamp(bufEnd, pAction->endTime, TSDB_TIME_PRECISION_MILLI);
×
2391
    len += snprintf(detail + len, sizeof(detail) - len, "startTime:%s, endTime:%s, ", bufStart, bufEnd);
×
2392
    char detailVStr[TSDB_TRANS_DETAIL_LEN + VARSTR_HEADER_SIZE] = {0};
×
2393
    STR_WITH_MAXSIZE_TO_VARSTR(detailVStr, detail, pShow->pMeta->pSchemas[cols].bytes);
×
2394
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
×
2395
    TAOS_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)detailVStr, false), &lino, _OVER);
×
2396

2397
  } else {
2398
    int32_t len = 0;
×
2399

2400
    char objType[TSDB_TRANS_OBJTYPE_LEN + 1] = {0};
×
2401
    if (pAction->pRaw->type == SDB_VGROUP) {
×
2402
      SSdbRow *pRow = mndVgroupActionDecode(pAction->pRaw);
×
2403
      SVgObj  *pVgroup = sdbGetRowObj(pRow);
×
2404
      len += snprintf(objType + len, sizeof(objType) - len, "%s(%d)", sdbTableName(pAction->pRaw->type), pVgroup->vgId);
×
2405
      taosMemoryFreeClear(pRow);
×
2406
    } else {
2407
      strcpy(objType, sdbTableName(pAction->pRaw->type));
×
2408
    }
2409
    char objTypeVStr[TSDB_TRANS_OBJTYPE_LEN + VARSTR_HEADER_SIZE] = {0};
×
2410
    STR_WITH_MAXSIZE_TO_VARSTR(objTypeVStr, objType, pShow->pMeta->pSchemas[cols].bytes);
×
2411
    SColumnInfoData *pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
×
2412
    TAOS_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)objTypeVStr, false), &lino, _OVER);
×
2413

2414
    char result[TSDB_TRANS_RESULT_LEN + 1] = {0};
×
2415
    len = 0;
×
2416
    len += snprintf(result + len, sizeof(result) - len, "rawWritten:%d", pAction->rawWritten);
×
2417
    char resultVStr[TSDB_TRANS_RESULT_LEN + VARSTR_HEADER_SIZE] = {0};
×
2418
    STR_WITH_MAXSIZE_TO_VARSTR(resultVStr, result, pShow->pMeta->pSchemas[cols].bytes);
×
2419
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
×
2420
    TAOS_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)resultVStr, false), &lino, _OVER);
×
2421

2422
    char target[TSDB_TRANS_TARGET_LEN] = "";
×
2423
    char targetVStr[TSDB_TRANS_TARGET_LEN + VARSTR_HEADER_SIZE] = {0};
×
2424
    STR_WITH_MAXSIZE_TO_VARSTR(targetVStr, target, pShow->pMeta->pSchemas[cols].bytes);
×
2425
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
×
2426
    TAOS_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)targetVStr, false), &lino, _OVER);
×
2427

2428
    char detail[TSDB_TRANS_DETAIL_LEN] = {0};
×
2429
    len = 0;
×
2430
    len += snprintf(detail + len, sizeof(detail) - len, "sdbStatus:%s", sdbStatusName(pAction->pRaw->status));
×
2431
    char detailVStr[TSDB_TRANS_DETAIL_LEN + VARSTR_HEADER_SIZE] = {0};
×
2432
    STR_WITH_MAXSIZE_TO_VARSTR(detailVStr, detail, pShow->pMeta->pSchemas[cols].bytes);
×
2433
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
×
2434
    TAOS_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)detailVStr, false), &lino, _OVER);
×
2435
  }
2436

2437
_OVER:
×
2438
  if (code != 0) mError("failed to retrieve at line:%d, since %s", lino, tstrerror(code));
×
2439
}
2440

2441
static SArray *mndTransGetAction(STrans *pTrans, ETrnStage stage) {
×
2442
  if (stage == TRN_STAGE_PREPARE) {
×
2443
    return pTrans->prepareActions;
×
2444
  }
2445
  if (stage == TRN_STAGE_REDO_ACTION) {
×
2446
    return pTrans->redoActions;
×
2447
  }
2448
  if (stage == TRN_STAGE_COMMIT_ACTION) {
×
2449
    return pTrans->commitActions;
×
2450
  }
2451
  if (stage == TRN_STAGE_UNDO_ACTION) {
×
2452
    return pTrans->undoActions;
×
2453
  }
2454
  return NULL;
×
2455
}
2456

2457
typedef struct STransDetailIter {
2458
  void     *pIter;
2459
  STrans   *pTrans;
2460
  ETrnStage stage;
2461
  int32_t   num;
2462
} STransDetailIter;
2463

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

2469
  for (int32_t i = start; i < actionNum; ++i) {
×
2470
    STransAction *pAction = taosArrayGet(pShowIter->pTrans->redoActions, i);
×
2471
    mndShowTransAction(pShow, pBlock, pAction, pShowIter->pTrans->id, pShowIter->pTrans->lastAction, rows, *numOfRows);
×
2472
    (*numOfRows)++;
×
2473
    if (*numOfRows >= rows) break;
×
2474
  }
2475

2476
  if (*numOfRows == end) {
×
2477
    sdbRelease(pSdb, pShowIter->pTrans);
×
2478
    pShowIter->pTrans = NULL;
×
2479
    pShowIter->num = 0;
×
2480
  } else {
2481
    pShowIter->pTrans = pShowIter->pTrans;
×
2482
    pShowIter->stage = pShowIter->pTrans->stage;
×
2483
    pShowIter->num += (*numOfRows);
×
2484
  }
2485
}
×
2486

2487
static int32_t mndRetrieveTransDetail(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows) {
×
2488
  SMnode *pMnode = pReq->info.node;
×
2489
  SSdb   *pSdb = pMnode->pSdb;
×
2490
  int32_t numOfRows = 0;
×
2491

2492
  int32_t code = 0;
×
2493
  int32_t lino = 0;
×
2494

2495
  mInfo("start to mndRetrieveTransDetail, rows:%d, pShow->numOfRows:%d, pShow->pIter:%p", rows, pShow->numOfRows,
×
2496
        pShow->pIter);
2497

2498
  if (pShow->pIter == NULL) {
×
2499
    pShow->pIter = taosMemoryMalloc(sizeof(STransDetailIter));
×
2500
    if (pShow->pIter == NULL) {
×
2501
      mError("failed to malloc for pShow->pIter");
×
2502
      return 0;
×
2503
    }
2504
    memset(pShow->pIter, 0, sizeof(STransDetailIter));
×
2505
  }
2506

2507
  STransDetailIter *pShowIter = (STransDetailIter *)pShow->pIter;
×
2508

2509
  while (numOfRows < rows) {
×
2510
    if (pShowIter->pTrans == NULL) {
×
2511
      pShowIter->pIter = sdbFetch(pSdb, SDB_TRANS, pShowIter->pIter, (void **)&(pShowIter->pTrans));
×
2512
      mDebug("retrieve trans detail from fetch, pShow->pIter:%p, pTrans:%p", pShowIter->pIter, pShowIter->pTrans);
×
2513
      if (pShowIter->pIter == NULL) break;
×
2514
      mInfo("retrieve trans detail from fetch, id:%d, trans stage:%d, IterNum:%d", pShowIter->pTrans->id,
×
2515
            pShowIter->pTrans->stage, pShowIter->num);
2516

2517
      SArray *pActions = mndTransGetAction(pShowIter->pTrans, pShowIter->pTrans->stage);
×
2518

2519
      mndTransShowActions(pSdb, pShowIter, pShow, pBlock, rows, &numOfRows, pActions, taosArrayGetSize(pActions), 0);
×
2520
      break;
×
2521
    } else {
2522
      mInfo("retrieve trans detail from iter, id:%d, iterStage:%d, IterNum:%d", pShowIter->pTrans->id, pShowIter->stage,
×
2523
            pShowIter->num);
2524
      SArray *pActions = mndTransGetAction(pShowIter->pTrans, pShowIter->stage);
×
2525

2526
      mndTransShowActions(pSdb, pShowIter, pShow, pBlock, rows, &numOfRows, pActions,
×
2527
                          taosArrayGetSize(pActions) - pShowIter->num, pShowIter->num);
×
2528
      break;
×
2529
    }
2530
  }
2531

2532
_OVER:
×
2533
  pShow->numOfRows += numOfRows;
×
2534

2535
  if (code != 0) {
×
2536
    mError("failed to retrieve at line:%d, since %s", lino, tstrerror(code));
×
2537
  } else {
2538
    mInfo("retrieve trans detail, numOfRows:%d, pShow->numOfRows:%d", numOfRows, pShow->numOfRows)
×
2539
  }
2540
  if (numOfRows == 0) {
×
2541
    taosMemoryFree(pShow->pIter);
×
2542
    pShow->pIter = NULL;
×
2543
  }
2544
  return numOfRows;
×
2545
}
2546

2547
static void mndCancelGetNextTrans(SMnode *pMnode, void *pIter) {
×
2548
  SSdb *pSdb = pMnode->pSdb;
×
2549
  sdbCancelFetchByType(pSdb, pIter, SDB_TRANS);
×
2550
}
×
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