• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In
Build has been canceled!

taosdata / TDengine / #3991

29 Apr 2025 09:40AM UTC coverage: 62.455% (-0.01%) from 62.469%
#3991

push

travis-ci

web-flow
Merge pull request #30938 from taosdata/fix/taosdCmdArgs

fix: taosd cmd args

155487 of 317399 branches covered (48.99%)

Branch coverage included in aggregate %.

1 of 4 new or added lines in 1 file covered. (25.0%)

345 existing lines in 99 files now uncovered.

240414 of 316504 relevant lines covered (75.96%)

6321441.34 hits per line

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

62.81
/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
#define TRANS_ACTION_TIMEOUT 1000 * 1000 * 60 * 15
33

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

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

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

58
static inline bool mndTransIsInSyncContext(bool topHalf) { return !topHalf; }
393,410✔
59

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

67
static inline char *mndStrExecutionContext(bool topHalf) { return topHalf ? "transContext" : "syncContext"; }
459,318✔
68

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

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

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

90
  mndSetMsgHandle(pMnode, TDMT_MND_TRANS_TIMER, mndProcessTransTimer);
2,029✔
91
  mndSetMsgHandle(pMnode, TDMT_MND_KILL_TRANS, mndProcessKillTransReq);
2,029✔
92

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

99
void mndCleanupTrans(SMnode *pMnode) {}
2,028✔
100

101
static int32_t mndTransGetActionsSize(SArray *pArray) {
554,652✔
102
  int32_t actionNum = taosArrayGetSize(pArray);
554,652✔
103
  int32_t rawDataLen = 0;
554,652✔
104

105
  for (int32_t i = 0; i < actionNum; ++i) {
1,706,160✔
106
    STransAction *pAction = taosArrayGet(pArray, i);
1,151,508✔
107
    if (pAction->actionType == TRANS_ACTION_RAW) {
1,151,508✔
108
      rawDataLen += (sizeof(STransAction) + sdbGetRawTotalSize(pAction->pRaw));
774,896✔
109
    } else if (pAction->actionType == TRANS_ACTION_MSG) {
376,612!
110
      rawDataLen += (sizeof(STransAction) + pAction->contLen);
376,612✔
111
    } else {
112
      // empty
113
    }
114
    rawDataLen += sizeof(int8_t);
1,151,508✔
115
  }
116

117
  return rawDataLen;
554,652✔
118
}
119

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

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

154
_OVER:
554,652✔
155
  *offset = dataPos;
554,652✔
156
  return ret;
554,652✔
157
}
158

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

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

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

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

191
  int32_t prepareActionNum = taosArrayGetSize(pTrans->prepareActions);
138,663✔
192
  int32_t redoActionNum = taosArrayGetSize(pTrans->redoActions);
138,663✔
193
  int32_t undoActionNum = taosArrayGetSize(pTrans->undoActions);
138,663✔
194
  int32_t commitActionNum = taosArrayGetSize(pTrans->commitActions);
138,663✔
195

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

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

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

215
  SDB_SET_BINARY(pRaw, dataPos, pTrans->opername, TSDB_TRANS_OPER_LEN, _OVER)
138,663!
216

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

227
  if (sver > TRANS_VER1_NUMBER) {
138,663!
228
    SDB_SET_INT8(pRaw, dataPos, pTrans->ableToBeKilled, _OVER)
138,663!
229
    SDB_SET_INT32(pRaw, dataPos, pTrans->killMode, _OVER)
138,663!
230
  }
231

232
  SDB_SET_RESERVE(pRaw, dataPos, TRANS_RESERVE_SIZE, _OVER)
138,663!
233
  SDB_SET_DATALEN(pRaw, dataPos, _OVER)
138,663!
234

235
  terrno = 0;
138,663✔
236

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

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

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

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

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

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

320
  if (sdbGetRawSoftVer(pRaw, &sver) != 0) goto _OVER;
264,491!
321

322
  if (sver > TRANS_VER2_NUMBER) {
264,491!
323
    terrno = TSDB_CODE_SDB_INVALID_DATA_VER;
×
324
    goto _OVER;
×
325
  }
326

327
  pRow = sdbAllocRow(sizeof(STrans));
264,491✔
328
  if (pRow == NULL) goto _OVER;
264,491!
329

330
  pTrans = sdbGetRowObj(pRow);
264,491✔
331
  if (pTrans == NULL) goto _OVER;
264,491!
332

333
  SDB_GET_INT32(pRaw, dataPos, &pTrans->id, _OVER)
264,491!
334

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

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

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

371
  if (pTrans->prepareActions == NULL) goto _OVER;
264,491!
372
  if (pTrans->redoActions == NULL) goto _OVER;
264,491!
373
  if (pTrans->undoActions == NULL) goto _OVER;
264,491!
374
  if (pTrans->commitActions == NULL) goto _OVER;
264,491!
375

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

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

390
  SDB_GET_BINARY(pRaw, dataPos, pTrans->opername, TSDB_TRANS_OPER_LEN, _OVER);
264,491!
391

392
  pTrans->arbGroupIds = taosHashInit(16, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), false, HASH_ENTRY_LOCK);
264,491✔
393

394
  SDB_GET_INT32(pRaw, dataPos, &arbgroupIdNum, _OVER)
264,491!
395
  for (int32_t i = 0; i < arbgroupIdNum; ++i) {
264,546✔
396
    int32_t arbGroupId = 0;
55✔
397
    SDB_GET_INT32(pRaw, dataPos, &arbGroupId, _OVER)
55!
398
    if ((terrno = taosHashPut(pTrans->arbGroupIds, &arbGroupId, sizeof(int32_t), NULL, 0)) != 0) goto _OVER;
55!
399
  }
400

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

408
  SDB_GET_RESERVE(pRaw, dataPos, TRANS_RESERVE_SIZE, _OVER)
264,491!
409

410
  terrno = 0;
264,491✔
411

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

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

426
static const char *mndTransStr(ETrnStage stage) {
2,039,396✔
427
  switch (stage) {
2,039,396!
428
    case TRN_STAGE_PREPARE:
138,121✔
429
      return "prepare";
138,121✔
430
    case TRN_STAGE_REDO_ACTION:
777,751✔
431
      return "redoAction";
777,751✔
432
    case TRN_STAGE_ROLLBACK:
23✔
433
      return "rollback";
23✔
434
    case TRN_STAGE_UNDO_ACTION:
42,512✔
435
      return "undoAction";
42,512✔
436
    case TRN_STAGE_COMMIT:
226,266✔
437
      return "commit";
226,266✔
438
    case TRN_STAGE_COMMIT_ACTION:
522,197✔
439
      return "commitAction";
522,197✔
440
    case TRN_STAGE_FINISH:
332,507✔
441
      return "finished";
332,507✔
442
    case TRN_STAGE_PRE_FINISH:
19✔
443
      return "pre-finish";
19✔
444
    default:
×
445
      return "invalid";
×
446
  }
447
}
448

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

460
static void mndSetTransLastAction(STrans *pTrans, STransAction *pAction) {
591,097✔
461
  if (pAction != NULL) {
591,097✔
462
    if (pAction->errCode != TSDB_CODE_ACTION_IN_PROGRESS) {
477,635✔
463
      pTrans->lastAction = pAction->id;
335,545✔
464
      pTrans->lastMsgType = pAction->msgType;
335,545✔
465
      pTrans->lastEpset = pAction->epSet;
335,545✔
466
      pTrans->lastErrorNo = pAction->errCode;
335,545✔
467
    }
468
  } else {
469
    pTrans->lastAction = 0;
113,462✔
470
    pTrans->lastMsgType = 0;
113,462✔
471
    memset(&pTrans->lastEpset, 0, sizeof(pTrans->lastEpset));
113,462✔
472
    pTrans->lastErrorNo = 0;
113,462✔
473
  }
474
}
591,097✔
475

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

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

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

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

503
  (void)taosThreadMutexInit(&pTrans->mutex, NULL);
50,911✔
504

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

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

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

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

528
  return 0;
50,911✔
529
}
530

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

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

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

580
  mndTransDropData(pTrans);
157,684✔
581
  return 0;
157,684✔
582
}
583

584
static void mndTransUpdateActions(SArray *pOldArray, SArray *pNewArray) {
224,380✔
585
  for (int32_t i = 0; i < taosArrayGetSize(pOldArray); ++i) {
719,671✔
586
    STransAction *pOldAction = taosArrayGet(pOldArray, i);
495,291✔
587
    STransAction *pNewAction = taosArrayGet(pNewArray, i);
495,291✔
588
    pOldAction->rawWritten = pNewAction->rawWritten;
495,291✔
589
    pOldAction->msgSent = pNewAction->msgSent;
495,291✔
590
    pOldAction->msgReceived = pNewAction->msgReceived;
495,291✔
591
    pOldAction->errCode = pNewAction->errCode;
495,291✔
592
  }
593
}
224,380✔
594

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

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

608
  mndTransUpdateActions(pOld->prepareActions, pNew->prepareActions);
56,095✔
609
  mndTransUpdateActions(pOld->redoActions, pNew->redoActions);
56,095✔
610
  mndTransUpdateActions(pOld->undoActions, pNew->undoActions);
56,095✔
611
  mndTransUpdateActions(pOld->commitActions, pNew->commitActions);
56,095✔
612
  pOld->stage = pNew->stage;
56,095✔
613
  pOld->actionPos = pNew->actionPos;
56,095✔
614

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

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

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

630
  return 0;
56,095✔
631
}
632

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

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

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

656
  if (opername != NULL) {
42,127!
657
    tstrncpy(pTrans->opername, opername, TSDB_TRANS_OPER_LEN);
42,127✔
658
  }
659

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

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

688
  if (pReq != NULL) {
42,127✔
689
    if (taosArrayPush(pTrans->pRpcArray, &pReq->info) == NULL) {
50,440!
690
      terrno = TSDB_CODE_OUT_OF_MEMORY;
×
691
      return NULL;
×
692
    }
693
    pTrans->originRpcType = pReq->msgType;
25,220✔
694
  }
695

696
  mInfo("trans:%d, create transaction:%s, origin:%s", pTrans->id, pTrans->opername, opername);
42,127!
697

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

702
static void mndTransDropActions(SArray *pArray) {
1,226,472✔
703
  int32_t size = taosArrayGetSize(pArray);
1,226,472✔
704
  for (int32_t i = 0; i < size; ++i) {
3,708,986✔
705
    STransAction *pAction = taosArrayGet(pArray, i);
2,482,514✔
706
    if (pAction->actionType == TRANS_ACTION_RAW) {
2,482,514✔
707
      taosMemoryFreeClear(pAction->pRaw);
1,638,180!
708
    } else if (pAction->actionType == TRANS_ACTION_MSG) {
844,334!
709
      taosMemoryFreeClear(pAction->pCont);
844,334!
710
    } else {
711
      // nothing
712
    }
713
  }
714

715
  taosArrayDestroy(pArray);
1,226,472✔
716
}
1,226,472✔
717

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

726
static int32_t mndTransAppendAction(SArray *pArray, STransAction *pAction) {
333,142✔
727
  pAction->id = taosArrayGetSize(pArray);
333,142✔
728

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

734
  return 0;
333,142✔
735
}
736

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

848
void mndTransSetOper(STrans *pTrans, EOperType oper) { pTrans->oper = oper; }
4,173✔
849

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

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

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

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

882
static bool mndCheckStbConflict(const char *conflict, STrans *pTrans) {
17,904✔
883
  if (conflict[0] == 0) return false;
17,904✔
884
  if (taosStrcasecmp(conflict, pTrans->stbname) == 0) return true;
16,550✔
885
  return false;
16,485✔
886
}
887

888
static void mndTransLogConflict(STrans *pNew, STrans *pTrans, bool conflict, bool *globalConflict) {
19,537✔
889
  if (conflict) {
19,537✔
890
    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,
413!
891
           pNew->dbname, pNew->stbname, pNew->conflict, pTrans->id, pTrans->dbname, pTrans->stbname, pTrans->conflict);
892
    *globalConflict = true;
413✔
893
  } else {
894
    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,124!
895
          pNew->stbname, pNew->conflict, pTrans->id, pTrans->dbname, pTrans->stbname, pTrans->conflict);
896
  }
897
}
19,537✔
898

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

904
  if (pNew->conflict == TRN_CONFLICT_NOTHING) return conflict;
116,871✔
905

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

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

913
    if (pNew->conflict == TRN_CONFLICT_GLOBAL) conflict = true;
18,561✔
914

915
    if (pNew->conflict == TRN_CONFLICT_DB) {
18,561✔
916
      if (pTrans->conflict == TRN_CONFLICT_GLOBAL) conflict = true;
1,728✔
917
      if (pTrans->conflict == TRN_CONFLICT_DB || pTrans->conflict == TRN_CONFLICT_DB_INSIDE) {
1,728!
918
        mndTransLogConflict(pNew, pTrans, mndCheckDbConflict(pNew->dbname, pTrans), &conflict);
1,354✔
919
        mndTransLogConflict(pNew, pTrans, mndCheckStbConflict(pNew->stbname, pTrans), &conflict);
1,354✔
920
      }
921
    }
922

923
    if (pNew->conflict == TRN_CONFLICT_DB_INSIDE) {
18,561✔
924
      if (pTrans->conflict == TRN_CONFLICT_GLOBAL) conflict = true;
16,810✔
925
      if (pTrans->conflict == TRN_CONFLICT_DB) {
16,810✔
926
        mndTransLogConflict(pNew, pTrans, mndCheckDbConflict(pNew->dbname, pTrans), &conflict);
266✔
927
        mndTransLogConflict(pNew, pTrans, mndCheckStbConflict(pNew->stbname, pTrans), &conflict);
266✔
928
      }
929
      if (pTrans->conflict == TRN_CONFLICT_DB_INSIDE) {
16,810✔
930
        mndTransLogConflict(pNew, pTrans, mndCheckStbConflict(pNew->stbname, pTrans), &conflict);  // for stb
16,284✔
931
      }
932
    }
933

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

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

960
    sdbRelease(pMnode->pSdb, pTrans);
18,561✔
961
  }
962

963
  return conflict;
80,396✔
964
}
965

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

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

982
  TAOS_RETURN(code);
116,411✔
983
}
984

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

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

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

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

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

1020
  TAOS_RETURN(code);
223✔
1021
}
1022

1023
static bool mndTransActionsOfSameType(SArray *pActions) {
97,281✔
1024
  int32_t size = taosArrayGetSize(pActions);
97,281✔
1025
  ETrnAct lastActType = TRANS_ACTION_NULL;
97,281✔
1026
  bool    same = true;
97,281✔
1027
  for (int32_t i = 0; i < size; ++i) {
379,170✔
1028
    STransAction *pAction = taosArrayGet(pActions, i);
281,889✔
1029
    if (i > 0) {
281,889✔
1030
      if (lastActType != pAction->actionType) {
212,590!
1031
        same = false;
×
1032
        break;
×
1033
      }
1034
    }
1035
    lastActType = pAction->actionType;
281,889✔
1036
  }
1037
  return same;
97,281✔
1038
}
1039

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

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

1058
  TAOS_RETURN(code);
41,636✔
1059
}
1060

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

1074
  TAOS_RETURN(code);
41,636✔
1075
}
1076

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

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

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

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

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

1115

1116
  TAOS_CHECK_RETURN(mndTransCheckConflict(pMnode, pTrans));
41,636!
1117

1118
  TAOS_CHECK_RETURN(mndTransCheckParallelActions(pMnode, pTrans));
41,636!
1119

1120
  TAOS_CHECK_RETURN(mndTransCheckCommitActions(pMnode, pTrans));
41,636!
1121

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

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

1140
  pNew->pRpcArray = pTrans->pRpcArray;
41,628✔
1141
  pNew->rpcRsp = pTrans->rpcRsp;
41,628✔
1142
  pNew->rpcRspLen = pTrans->rpcRspLen;
41,628✔
1143
  pNew->mTraceId = pTrans->mTraceId;
41,628✔
1144
  pTrans->pRpcArray = NULL;
41,628✔
1145
  pTrans->rpcRsp = NULL;
41,628✔
1146
  pTrans->rpcRspLen = 0;
41,628✔
1147

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

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

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

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

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

1192
  if (pTrans->stage == TRN_STAGE_FINISH) {
221,899✔
1193
    sendRsp = true;
92,309✔
1194
  }
1195

1196
  if (pTrans->policy == TRN_POLICY_ROLLBACK) {
221,899✔
1197
    if (pTrans->stage == TRN_STAGE_UNDO_ACTION || pTrans->stage == TRN_STAGE_ROLLBACK) {
72,496✔
1198
      if (code == 0) code = TSDB_CODE_MND_TRANS_UNKNOW_ERROR;
26✔
1199
      sendRsp = true;
26✔
1200
    }
1201
  } else {
1202
    if (pTrans->stage == TRN_STAGE_REDO_ACTION) {
149,403✔
1203
      if (code == TSDB_CODE_SYN_NOT_LEADER || code == TSDB_CODE_SYN_RESTORING || code == TSDB_CODE_APP_IS_STARTING ||
91,924!
1204
          code == TSDB_CODE_SYN_PROPOSE_NOT_READY) {
1205
        if (pTrans->failedTimes > 60) sendRsp = true;
×
1206
      } else {
1207
        if (pTrans->failedTimes > 6) sendRsp = true;
91,924✔
1208
      }
1209
      if (code == 0) code = TSDB_CODE_MND_TRANS_UNKNOW_ERROR;
91,924✔
1210
    }
1211
  }
1212

1213
  if (!sendRsp) {
221,899✔
1214
    return;
129,562✔
1215
  } else {
1216
    mInfo("vgId:1, trans:%d, start to send rsp, stage:%s failedTimes:%d code:0x%x", pTrans->id,
92,337!
1217
          mndTransStr(pTrans->stage), pTrans->failedTimes, code);
1218
  }
1219

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

1228
  for (int32_t i = 0; i < size; ++i) {
49,402✔
1229
    SRpcHandleInfo *pInfo = taosArrayGet(pTrans->pRpcArray, i);
24,701✔
1230
    if (pInfo->handle != NULL) {
24,701✔
1231
      if (code == TSDB_CODE_RPC_NETWORK_UNAVAIL || code == TSDB_CODE_RPC_SOMENODE_NOT_CONNECTED) {
23,272!
1232
        code = TSDB_CODE_MND_TRANS_NETWORK_UNAVAILL;
1✔
1233
      }
1234
      if (code == TSDB_CODE_SYN_TIMEOUT) {
23,272!
1235
        code = TSDB_CODE_MND_TRANS_SYNC_TIMEOUT;
×
1236
      }
1237

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

1244
      SRpcMsg rspMsg = {.code = code, .info = *pInfo};
23,272✔
1245

1246
      if (pTrans->originRpcType == TDMT_MND_CREATE_DB) {
23,272✔
1247
        mInfo("trans:%d, origin msgtype:%s", pTrans->id, TMSG_INFO(pTrans->originRpcType));
4,155!
1248
        SDbObj *pDb = mndAcquireDb(pMnode, pTrans->dbname);
4,155✔
1249
        if (pDb != NULL) {
4,155!
1250
          for (int32_t j = 0; j < 12; j++) {
5,016✔
1251
            bool ready = mndIsDbReady(pMnode, pDb);
5,009✔
1252
            if (!ready) {
5,009✔
1253
              mInfo("trans:%d, db:%s not ready yet, wait %d times", pTrans->id, pTrans->dbname, j);
861!
1254
              taosMsleep(1000);
861✔
1255
            } else {
1256
              break;
4,148✔
1257
            }
1258
          }
1259
        }
1260
        mndReleaseDb(pMnode, pDb);
4,155✔
1261
      } else if (pTrans->originRpcType == TDMT_MND_CREATE_STB) {
19,117✔
1262
        void   *pCont = NULL;
6,547✔
1263
        int32_t contLen = 0;
6,547✔
1264
        if (0 == mndBuildSMCreateStbRsp(pMnode, pTrans->dbname, pTrans->stbname, &pCont, &contLen)) {
6,547✔
1265
          mndTransSetRpcRsp(pTrans, pCont, contLen);
6,544✔
1266
        }
1267
      } else if (pTrans->originRpcType == TDMT_MND_CREATE_INDEX) {
12,570✔
1268
        void   *pCont = NULL;
9✔
1269
        int32_t contLen = 0;
9✔
1270
        if (0 == mndBuildSMCreateStbRsp(pMnode, pTrans->dbname, pTrans->stbname, &pCont, &contLen)) {
9!
1271
          mndTransSetRpcRsp(pTrans, pCont, contLen);
9✔
1272
        }
1273
      }
1274

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

1284
      tmsgSendRsp(&rspMsg);
23,272✔
1285

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

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

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

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

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

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

1340
    // pTrans->lastErrorNo = pRsp->code;
1341
    mndSetTransLastAction(pTrans, pAction);
68,770✔
1342

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

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

1353
_OVER:
68,770✔
1354
  mndReleaseTrans(pMnode, pTrans);
68,770✔
1355
  TAOS_RETURN(code);
68,770✔
1356
}
1357

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

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

1376
  for (int32_t action = 0; action < numOfActions; ++action) {
38✔
1377
    STransAction *pAction = taosArrayGet(pArray, action);
32✔
1378
    if (pAction->msgSent && pAction->msgReceived &&
32!
1379
        (pAction->errCode == 0 || pAction->errCode == pAction->acceptableCode))
32!
1380
      continue;
26✔
1381
    if (pAction->msgSent && !pAction->msgReceived) {
6!
1382
      int64_t timestamp = taosGetTimestampMs();
×
1383
      if (timestamp - pAction->startTime <= TRANS_ACTION_TIMEOUT) continue;
×
1384
    }
1385

1386
    if (pAction->rawWritten && (pAction->errCode == 0 || pAction->errCode == pAction->acceptableCode)) continue;
6!
1387

1388
    mndTransResetAction(pMnode, pTrans, pAction);
6✔
1389
    mInfo("trans:%d, %s:%d reset, errCode:%d, startTime:%" PRId64, pTrans->id, mndTransStr(pAction->stage), pAction->id,
6!
1390
          pAction->errCode, pAction->startTime);
1391
  }
1392
}
6✔
1393

1394
// execute in sync context
1395
static int32_t mndTransWriteSingleLog(SMnode *pMnode, STrans *pTrans, STransAction *pAction, bool topHalf) {
458,147✔
1396
  if (pAction->rawWritten) return 0;
458,147✔
1397
  if (topHalf) {
252,189✔
1398
    TAOS_RETURN(TSDB_CODE_MND_TRANS_CTX_SWITCH);
10✔
1399
  }
1400

1401
  if (pAction->pRaw->type >= SDB_MAX) {
252,179!
1402
    pAction->rawWritten = true;
×
1403
    pAction->errCode = 0;
×
1404
    mndSetTransLastAction(pTrans, pAction);
×
1405
    mInfo("skip sdb raw type:%d since it is not supported", pAction->pRaw->type);
×
1406
    TAOS_RETURN(TSDB_CODE_SUCCESS);
×
1407
  }
1408

1409
  int32_t code = sdbWriteWithoutFree(pMnode->pSdb, pAction->pRaw);
252,179✔
1410
  if (code == 0 || terrno == TSDB_CODE_SDB_OBJ_NOT_THERE) {
252,179!
1411
    pAction->rawWritten = true;
252,179✔
1412
    pAction->errCode = 0;
252,179✔
1413
    code = 0;
252,179✔
1414
    mInfo("trans:%d, %s:%d write to sdb, type:%s status:%s", pTrans->id, mndTransStr(pAction->stage), pAction->id,
252,179!
1415
          sdbTableName(pAction->pRaw->type), sdbStatusName(pAction->pRaw->status));
1416

1417
    mndSetTransLastAction(pTrans, pAction);
252,179✔
1418
  } else {
1419
    pAction->errCode = (terrno != 0) ? terrno : code;
×
1420
    mError("trans:%d, %s:%d failed to write sdb since %s, type:%s status:%s", pTrans->id, mndTransStr(pAction->stage),
×
1421
           pAction->id, tstrerror(code), sdbTableName(pAction->pRaw->type), sdbStatusName(pAction->pRaw->status));
1422
    mndSetTransLastAction(pTrans, pAction);
×
1423
  }
1424

1425
  TAOS_RETURN(code);
252,179✔
1426
}
1427

1428
// execute in trans context
1429
static int32_t mndTransSendSingleMsg(SMnode *pMnode, STrans *pTrans, STransAction *pAction, bool topHalf,
403,880✔
1430
                                     bool notSend) {
1431
  if (pAction->msgSent) return 0;
403,880✔
1432
  if (mndCannotExecuteTrans(pMnode, topHalf)) {
100,329✔
1433
    TAOS_RETURN(TSDB_CODE_MND_TRANS_CTX_SWITCH);
31,277✔
1434
  }
1435

1436
  if (notSend) {
69,052✔
1437
    mInfo("trans:%d, action:%d skip to execute msg action", pTrans->id, pAction->id);
179!
1438
    return 0;
179✔
1439
  }
1440

1441
#ifndef TD_ASTRA_32
1442
  int64_t signature = pTrans->id;
68,873✔
1443
  signature = (signature << 32);
68,873✔
1444
  signature += pAction->id;
68,873✔
1445

1446
  SRpcMsg rpcMsg = {.msgType = pAction->msgType, .contLen = pAction->contLen, .info.ahandle = (void *)signature};
68,873✔
1447
#else
1448
  SRpcMsg rpcMsg = {.msgType = pAction->msgType,
1449
                    .contLen = pAction->contLen,
1450
                    .info.ahandle = (void *)pTrans->id,
1451
                    .info.ahandleEx = (void *)pAction->id};
1452
#endif
1453
  rpcMsg.pCont = rpcMallocCont(pAction->contLen);
68,873✔
1454
  if (rpcMsg.pCont == NULL) {
68,873!
1455
    terrno = TSDB_CODE_OUT_OF_MEMORY;
×
1456
    return -1;
×
1457
  }
1458
  rpcMsg.info.traceId.rootId = pTrans->mTraceId;
68,873✔
1459
  rpcMsg.info.notFreeAhandle = 1;
68,873✔
1460

1461
  memcpy(rpcMsg.pCont, pAction->pCont, pAction->contLen);
68,873✔
1462

1463
  char    detail[1024] = {0};
68,873✔
1464
  int32_t len = tsnprintf(detail, sizeof(detail), "msgType:%s numOfEps:%d inUse:%d", TMSG_INFO(pAction->msgType),
68,873!
1465
                          pAction->epSet.numOfEps, pAction->epSet.inUse);
68,873✔
1466
  for (int32_t i = 0; i < pAction->epSet.numOfEps; ++i) {
143,657✔
1467
    len += tsnprintf(detail + len, sizeof(detail) - len, " ep:%d-%s:%u", i, pAction->epSet.eps[i].fqdn,
74,784✔
1468
                     pAction->epSet.eps[i].port);
74,784✔
1469
  }
1470

1471
  int32_t code = tmsgSendReq(&pAction->epSet, &rpcMsg);
68,873✔
1472
  if (code == 0) {
68,873✔
1473
    pAction->msgSent = 1;
68,871✔
1474
    // pAction->msgReceived = 0;
1475
    pAction->errCode = TSDB_CODE_ACTION_IN_PROGRESS;
68,871✔
1476
    pAction->startTime = taosGetTimestampMs();
68,871✔
1477
    pAction->endTime = 0;
68,871✔
1478
    mInfo("trans:%d, %s:%d is sent, %s", pTrans->id, mndTransStr(pAction->stage), pAction->id, detail);
68,871!
1479

1480
    mndSetTransLastAction(pTrans, pAction);
68,871✔
1481
  } else {
1482
    pAction->msgSent = 0;
2✔
1483
    pAction->msgReceived = 0;
2✔
1484
    pAction->errCode = (terrno != 0) ? terrno : code;
2!
1485
    mError("trans:%d, %s:%d not send since %s, %s", pTrans->id, mndTransStr(pAction->stage), pAction->id, terrstr(),
2!
1486
           detail);
1487

1488
    mndSetTransLastAction(pTrans, pAction);
2✔
1489
  }
1490

1491
  TAOS_RETURN(code);
68,873✔
1492
}
1493

1494
static int32_t mndTransExecNullMsg(SMnode *pMnode, STrans *pTrans, STransAction *pAction, bool topHalf) {
×
1495
  if (!topHalf) return TSDB_CODE_MND_TRANS_CTX_SWITCH;
×
1496
  pAction->rawWritten = 0;
×
1497
  pAction->errCode = 0;
×
1498
  mInfo("trans:%d, %s:%d confirm action executed", pTrans->id, mndTransStr(pAction->stage), pAction->id);
×
1499

1500
  mndSetTransLastAction(pTrans, pAction);
×
1501
  return 0;
×
1502
}
1503

1504
static int32_t mndTransExecSingleAction(SMnode *pMnode, STrans *pTrans, STransAction *pAction, bool topHalf,
862,027✔
1505
                                        bool notSend) {
1506
  if (pAction->actionType == TRANS_ACTION_RAW) {
862,027✔
1507
    return mndTransWriteSingleLog(pMnode, pTrans, pAction, topHalf);
458,147✔
1508
  } else if (pAction->actionType == TRANS_ACTION_MSG) {
403,880!
1509
    return mndTransSendSingleMsg(pMnode, pTrans, pAction, topHalf, notSend);
403,880✔
1510
  } else {
1511
    return mndTransExecNullMsg(pMnode, pTrans, pAction, topHalf);
×
1512
  }
1513
}
1514

1515
static int32_t mndTransExecSingleActions(SMnode *pMnode, STrans *pTrans, SArray *pArray, bool topHalf, bool notSend) {
203,388✔
1516
  int32_t numOfActions = taosArrayGetSize(pArray);
203,388✔
1517
  int32_t code = 0;
203,388✔
1518

1519
  for (int32_t action = 0; action < numOfActions; ++action) {
987,441✔
1520
    STransAction *pAction = taosArrayGet(pArray, action);
811,380✔
1521
    code = mndTransExecSingleAction(pMnode, pTrans, pAction, topHalf, notSend);
811,380✔
1522
    if (code != 0) {
811,380✔
1523
      mInfo("trans:%d, action:%d not executed since %s. numOfActions:%d", pTrans->id, action, tstrerror(code),
27,327!
1524
            numOfActions);
1525
      break;
27,327✔
1526
    }
1527
  }
1528

1529
  return code;
203,388✔
1530
}
1531

1532
static int32_t mndTransExecuteActions(SMnode *pMnode, STrans *pTrans, SArray *pArray, bool topHalf, bool notSend) {
245,212✔
1533
  int32_t numOfActions = taosArrayGetSize(pArray);
245,212✔
1534
  int32_t code = 0;
245,212✔
1535
  if (numOfActions == 0) return 0;
245,212✔
1536

1537
  if ((code = mndTransExecSingleActions(pMnode, pTrans, pArray, topHalf, notSend)) != 0) {
203,388✔
1538
    return code;
27,327✔
1539
  }
1540

1541
  int32_t       numOfExecuted = 0;
176,061✔
1542
  int32_t       errCode = 0;
176,061✔
1543
  STransAction *pErrAction = NULL;
176,061✔
1544
  for (int32_t action = 0; action < numOfActions; ++action) {
960,114✔
1545
    STransAction *pAction = taosArrayGet(pArray, action);
784,053✔
1546
    if (pAction->msgReceived || pAction->rawWritten) {
784,053✔
1547
      numOfExecuted++;
607,498✔
1548
      if (pAction->errCode != 0 && pAction->errCode != pAction->acceptableCode) {
607,498✔
1549
        errCode = pAction->errCode;
32✔
1550
        pErrAction = pAction;
32✔
1551
      }
1552
    } else {
1553
      pErrAction = pAction;
176,555✔
1554
    }
1555
  }
1556

1557
  mndSetTransLastAction(pTrans, pErrAction);
176,061✔
1558

1559
  if (numOfExecuted == numOfActions) {
176,061✔
1560
    if (errCode == 0) {
113,468✔
1561
      mInfo("trans:%d, all %d actions execute successfully", pTrans->id, numOfActions);
113,462!
1562
      return 0;
113,462✔
1563
    } else {
1564
      mError("trans:%d, all %d actions executed, code:0x%x", pTrans->id, numOfActions, errCode & 0XFFFF);
6!
1565
      mndTransResetActions(pMnode, pTrans, pArray);
6✔
1566
      terrno = errCode;
6✔
1567
      return errCode;
6✔
1568
    }
1569
  } else {
1570
    mInfo("trans:%d, %d of %d actions executed", pTrans->id, numOfExecuted, numOfActions);
62,593!
1571

1572
    for (int32_t action = 0; action < numOfActions; ++action) {
355,854✔
1573
      STransAction *pAction = taosArrayGet(pArray, action);
293,261✔
1574
      mDebug("trans:%d, %s:%d Sent:%d, Received:%d, errCode:0x%x, acceptableCode:0x%x, retryCode:0x%x", pTrans->id,
293,261✔
1575
             mndTransStr(pAction->stage), pAction->id, pAction->msgSent, pAction->msgReceived, pAction->errCode,
1576
             pAction->acceptableCode, pAction->retryCode);
1577
      if (pAction->msgSent) {
293,261✔
1578
        bool reset = false;
293,082✔
1579
        if (pAction->msgReceived) {
293,082✔
1580
          if (pAction->errCode != 0 && pAction->errCode != pAction->acceptableCode) reset = true;
116,706✔
1581
        } else {
1582
          int64_t timestamp = taosGetTimestampMs();
176,376✔
1583
          if (timestamp - pAction->startTime > TRANS_ACTION_TIMEOUT) reset = true;
176,376!
1584
        }
1585
        if (reset) {
293,082✔
1586
          mndTransResetAction(pMnode, pTrans, pAction);
26✔
1587
          mInfo("trans:%d, %s:%d reset, errCode:%d, startTime:%" PRId64, pTrans->id, mndTransStr(pAction->stage),
26!
1588
                pAction->id, pAction->errCode, pAction->startTime);
1589
        }
1590
      }
1591
    }
1592
    return TSDB_CODE_ACTION_IN_PROGRESS;
62,593✔
1593
  }
1594
}
1595

1596
static int32_t mndTransExecuteRedoActions(SMnode *pMnode, STrans *pTrans, bool topHalf, bool notSend) {
152,883✔
1597
  int32_t code = mndTransExecuteActions(pMnode, pTrans, pTrans->redoActions, topHalf, notSend);
152,883✔
1598
  if (code != 0 && code != TSDB_CODE_ACTION_IN_PROGRESS && code != TSDB_CODE_MND_TRANS_CTX_SWITCH) {
152,883✔
1599
    mError("trans:%d, failed to execute redoActions since:%s, code:0x%x, in %s", pTrans->id, terrstr(), terrno,
8!
1600
           mndStrExecutionContext(topHalf));
1601
  }
1602
  return code;
152,883✔
1603
}
1604

1605
static int32_t mndTransExecuteUndoActions(SMnode *pMnode, STrans *pTrans, bool topHalf, bool notSend) {
28✔
1606
  int32_t code = mndTransExecuteActions(pMnode, pTrans, pTrans->undoActions, topHalf, notSend);
28✔
1607
  if (code != 0 && code != TSDB_CODE_ACTION_IN_PROGRESS && code != TSDB_CODE_MND_TRANS_CTX_SWITCH) {
28!
UNCOV
1608
    mError("trans:%d, failed to execute undoActions since %s. in %s", pTrans->id, terrstr(),
×
1609
           mndStrExecutionContext(topHalf));
1610
  }
1611
  return code;
28✔
1612
}
1613

1614
static int32_t mndTransExecuteCommitActions(SMnode *pMnode, STrans *pTrans, bool topHalf) {
92,301✔
1615
  int32_t code = mndTransExecuteActions(pMnode, pTrans, pTrans->commitActions, topHalf, true);
92,301✔
1616
  if (code != 0 && code != TSDB_CODE_ACTION_IN_PROGRESS && code != TSDB_CODE_MND_TRANS_CTX_SWITCH) {
92,301!
1617
    mError("trans:%d, failed to execute commitActions since %s. in %s", pTrans->id, terrstr(),
×
1618
           mndStrExecutionContext(topHalf));
1619
  }
1620
  return code;
92,301✔
1621
}
1622

1623
static int32_t mndTransExecuteActionsSerial(SMnode *pMnode, STrans *pTrans, SArray *pActions, bool topHalf) {
18,309✔
1624
  int32_t code = 0;
18,309✔
1625
  int32_t numOfActions = taosArrayGetSize(pActions);
18,309✔
1626
  if (numOfActions == 0) return code;
18,309✔
1627

1628
  if (pTrans->actionPos >= numOfActions) {
18,305✔
1629
    return code;
1,213✔
1630
  }
1631

1632
  mInfo("trans:%d, execute %d actions serial, begin at action:%d, stage:%s", pTrans->id, numOfActions,
17,092!
1633
        pTrans->actionPos, mndTransStr(pTrans->stage));
1634

1635
  for (int32_t action = pTrans->actionPos; action < numOfActions; ++action) {
26,139✔
1636
    STransAction *pAction = taosArrayGet(pActions, action);
25,214✔
1637

1638
    mInfo("trans:%d, current action:%d, stage:%s, actionType(1:msg,2:log):%d, msgSent:%d, msgReceived:%d", 
25,214!
1639
          pTrans->id, pTrans->actionPos, mndTransStr(pAction->stage), pAction->actionType, pAction->msgSent,
1640
          pAction->msgReceived);
1641

1642
    code = mndTransExecSingleAction(pMnode, pTrans, pAction, topHalf, false);
25,214✔
1643
    if (code == 0) {
25,214✔
1644
      if (pAction->msgSent) {
21,252✔
1645
        bool reset = false;
18,458✔
1646
        if (pAction->msgReceived) {
18,458✔
1647
          if (pAction->errCode != 0 && pAction->errCode != pAction->acceptableCode) {
7,814!
1648
            code = pAction->errCode;
4,611✔
1649
            reset = true;
4,611✔
1650
          } else {
1651
            mInfo("trans:%d, %s:%d execute successfully", pTrans->id, mndTransStr(pAction->stage), action);
3,203!
1652
          }
1653
        } else {
1654
          int64_t timestamp = taosGetTimestampMs();
10,644✔
1655
          if (timestamp - pAction->startTime > TRANS_ACTION_TIMEOUT) reset = true;
10,644!
1656
          code = TSDB_CODE_ACTION_IN_PROGRESS;
10,644✔
1657
        }
1658
        if (reset) {
18,458✔
1659
          mndTransResetAction(pMnode, pTrans, pAction);
4,611✔
1660
          mInfo("trans:%d, %s:%d reset, errCode:%d, startTime:%" PRId64, pTrans->id, mndTransStr(pAction->stage),
4,611!
1661
                pAction->id, pAction->errCode, pAction->startTime);
1662
        }
1663
      } else if (pAction->rawWritten) {
2,794!
1664
        if (pAction->errCode != 0 && pAction->errCode != pAction->acceptableCode) {
2,794!
1665
          code = pAction->errCode;
×
1666
        } else {
1667
          mInfo("trans:%d, %s:%d write successfully", pTrans->id, mndTransStr(pAction->stage), action);
2,794!
1668
        }
1669
      } else {
1670
      }
1671
    }
1672

1673
    if (code == 0) {
25,214✔
1674
      pTrans->failedTimes = 0;
5,997✔
1675
    }
1676
    mndSetTransLastAction(pTrans, pAction);
25,214✔
1677

1678
    if (mndCannotExecuteTrans(pMnode, topHalf)) {
25,214✔
1679
      pTrans->lastErrorNo = code;
5,516✔
1680
      pTrans->code = code;
5,516✔
1681
      mInfo("trans:%d, %s:%d, cannot execute next action in %s, code:%s", pTrans->id, mndTransStr(pAction->stage),
5,516!
1682
            action, mndStrExecutionContext(topHalf), tstrerror(code));
1683
      break;
5,516✔
1684
    }
1685

1686
    if (code == 0) {
19,698✔
1687
      pTrans->code = 0;
4,443✔
1688
      pTrans->actionPos++;
4,443✔
1689
      mInfo("trans:%d, %s:%d is executed and need sync to other mnodes", pTrans->id, mndTransStr(pAction->stage),
4,443!
1690
            pAction->id);
1691
      (void)taosThreadMutexUnlock(&pTrans->mutex);
4,443✔
1692
      code = mndTransSync(pMnode, pTrans);
4,443✔
1693
      (void)taosThreadMutexLock(&pTrans->mutex);
4,443✔
1694
      if (code != 0) {
4,443!
1695
        pTrans->actionPos--;
×
1696
        pTrans->code = terrno;
×
1697
        mError("trans:%d, %s:%d is executed and failed to sync to other mnodes since %s", pTrans->id,
×
1698
               mndTransStr(pAction->stage), pAction->id, terrstr());
1699
        break;
×
1700
      }
1701
    } else if (code == TSDB_CODE_ACTION_IN_PROGRESS) {
15,255✔
1702
      mInfo("trans:%d, %s:%d is in progress and wait it finish", pTrans->id, mndTransStr(pAction->stage), pAction->id);
10,644!
1703
      break;
10,644✔
1704
    } else if (code == pAction->retryCode || code == TSDB_CODE_SYN_PROPOSE_NOT_READY ||
4,611✔
1705
               code == TSDB_CODE_SYN_RESTORING || code == TSDB_CODE_SYN_NOT_LEADER) {
23✔
1706
      mInfo("trans:%d, %s:%d receive code:0x%x(%s) and retry", pTrans->id, mndTransStr(pAction->stage), pAction->id,
4,604!
1707
            code, tstrerror(code));
1708
      pTrans->lastErrorNo = code;
4,604✔
1709
      taosMsleep(300);
4,604✔
1710
      action--;
4,604✔
1711
      continue;
4,604✔
1712
    } else {
1713
      terrno = code;
7✔
1714
      pTrans->lastErrorNo = code;
7✔
1715
      pTrans->code = code;
7✔
1716
      mInfo("trans:%d, %s:%d receive code:0x%x(%s) and wait another schedule, failedTimes:%d", pTrans->id,
7!
1717
            mndTransStr(pAction->stage), pAction->id, code, tstrerror(code), pTrans->failedTimes);
1718
      break;
7✔
1719
    }
1720
  }
1721

1722
  return code;
17,092✔
1723
}
1724

1725
static int32_t mndTransExecuteRedoActionsSerial(SMnode *pMnode, STrans *pTrans, bool topHalf) {
18,309✔
1726
  int32_t code = TSDB_CODE_ACTION_IN_PROGRESS;
18,309✔
1727
  (void)taosThreadMutexLock(&pTrans->mutex);
18,309✔
1728
  if (pTrans->stage == TRN_STAGE_REDO_ACTION) {
18,309!
1729
    code = mndTransExecuteActionsSerial(pMnode, pTrans, pTrans->redoActions, topHalf);
18,309✔
1730
  }
1731
  (void)taosThreadMutexUnlock(&pTrans->mutex);
18,309✔
1732
  return code;
18,309✔
1733
}
1734

1735
static int32_t mndTransExecuteUndoActionsSerial(SMnode *pMnode, STrans *pTrans, bool topHalf) {
×
1736
  int32_t code = TSDB_CODE_ACTION_IN_PROGRESS;
×
1737
  (void)taosThreadMutexLock(&pTrans->mutex);
×
1738
  if (pTrans->stage == TRN_STAGE_UNDO_ACTION) {
×
1739
    code = mndTransExecuteActionsSerial(pMnode, pTrans, pTrans->undoActions, topHalf);
×
1740
  }
1741
  (void)taosThreadMutexUnlock(&pTrans->mutex);
×
1742
  return code;
×
1743
}
1744

1745
bool mndTransPerformPrepareStage(SMnode *pMnode, STrans *pTrans, bool topHalf) {
50,709✔
1746
  bool    continueExec = true;
50,709✔
1747
  int32_t code = 0;
50,709✔
1748
  terrno = 0;
50,709✔
1749

1750
  int32_t numOfActions = taosArrayGetSize(pTrans->prepareActions);
50,709✔
1751
  if (numOfActions == 0) goto _OVER;
50,709✔
1752

1753
  mInfo("trans:%d, execute %d prepare actions.", pTrans->id, numOfActions);
15,774!
1754

1755
  for (int32_t action = 0; action < numOfActions; ++action) {
41,207✔
1756
    STransAction *pAction = taosArrayGet(pTrans->prepareActions, action);
25,433✔
1757
    code = mndTransExecSingleAction(pMnode, pTrans, pAction, topHalf, true);
25,433✔
1758
    if (code != 0) {
25,433!
1759
      terrno = code;
×
1760
      mError("trans:%d, failed to execute prepare action:%d, numOfActions:%d, since %s", pTrans->id, action,
×
1761
             numOfActions, tstrerror(code));
1762
      return false;
×
1763
    }
1764
  }
1765

1766
_OVER:
15,774✔
1767
  pTrans->stage = TRN_STAGE_REDO_ACTION;
50,709✔
1768
  mInfo("trans:%d, stage from prepare to redoAction", pTrans->id);
50,709!
1769
  return continueExec;
50,709✔
1770
}
1771

1772
static bool mndTransPerformRedoActionStage(SMnode *pMnode, STrans *pTrans, bool topHalf, bool notSend) {
171,192✔
1773
  bool    continueExec = true;
171,192✔
1774
  int32_t code = 0;
171,192✔
1775
  terrno = 0;
171,192✔
1776

1777
  if (pTrans->exec == TRN_EXEC_SERIAL) {
171,192✔
1778
    code = mndTransExecuteRedoActionsSerial(pMnode, pTrans, topHalf);
18,309✔
1779
  } else {
1780
    code = mndTransExecuteRedoActions(pMnode, pTrans, topHalf, notSend);
152,883✔
1781
  }
1782

1783
  if (code != 0 && code != TSDB_CODE_MND_TRANS_CTX_SWITCH && mndTransIsInSyncContext(topHalf)) {
171,192!
1784
    pTrans->lastErrorNo = code;
×
1785
    pTrans->code = code;
×
1786
    mInfo(
×
1787
        "trans:%d, failed to execute, will retry redo action stage in 100 ms , in %s, "
1788
        "continueExec:%d, code:%s",
1789
        pTrans->id, mndStrExecutionContext(topHalf), continueExec, tstrerror(code));
1790
    taosMsleep(100);
×
1791
    return true;
×
1792
  } else {
1793
    if (mndCannotExecuteTrans(pMnode, topHalf)) {
171,192✔
1794
      mInfo("trans:%d, cannot continue to execute redo action stage in %s, continueExec:%d, code:%s", pTrans->id,
56,320!
1795
            mndStrExecutionContext(topHalf), continueExec, tstrerror(code));
1796
      return false;
56,320✔
1797
    }
1798
  }
1799
  terrno = code;
114,872✔
1800

1801
  if (code == 0) {
114,872✔
1802
    pTrans->code = 0;
41,638✔
1803
    pTrans->stage = TRN_STAGE_COMMIT;
41,638✔
1804
    mInfo("trans:%d, stage from redoAction to commit", pTrans->id);
41,638!
1805
    continueExec = true;
41,638✔
1806
  } else if (code == TSDB_CODE_ACTION_IN_PROGRESS || code == TSDB_CODE_MND_TRANS_CTX_SWITCH) {
73,234!
1807
    mInfo("trans:%d, stage keep on redoAction since %s", pTrans->id, tstrerror(code));
73,219!
1808
    continueExec = false;
73,219✔
1809
  } else {
1810
    pTrans->failedTimes++;
15✔
1811
    pTrans->code = terrno;
15✔
1812
    if (pTrans->policy == TRN_POLICY_ROLLBACK) {
15✔
1813
      if (pTrans->lastAction != 0) {
3✔
1814
        STransAction *pAction = taosArrayGet(pTrans->redoActions, pTrans->lastAction);
2✔
1815
        if (pAction->retryCode != 0 && pAction->retryCode == pAction->errCode) {
2!
1816
          if (pTrans->failedTimes < 6) {
×
1817
            mError("trans:%d, stage keep on redoAction since action:%d code:0x%x not 0x%x, failedTimes:%d", pTrans->id,
×
1818
                   pTrans->lastAction, pTrans->code, pAction->retryCode, pTrans->failedTimes);
1819
            taosMsleep(1000);
×
1820
            continueExec = true;
×
1821
            return true;
×
1822
          }
1823
        }
1824
      }
1825

1826
      pTrans->stage = TRN_STAGE_ROLLBACK;
3✔
1827
      pTrans->actionPos = 0;
3✔
1828
      mError("trans:%d, stage from redoAction to rollback since %s, and set actionPos to %d", pTrans->id, terrstr(),
3!
1829
             pTrans->actionPos);
1830
      continueExec = true;
3✔
1831
    } else {
1832
      mError("trans:%d, stage keep on redoAction since %s, failedTimes:%d", pTrans->id, terrstr(), pTrans->failedTimes);
12!
1833
      continueExec = false;
12✔
1834
    }
1835
  }
1836

1837
  return continueExec;
114,872✔
1838
}
1839

1840
// execute in trans context
1841
static bool mndTransPerformCommitStage(SMnode *pMnode, STrans *pTrans, bool topHalf) {
41,641✔
1842
  if (mndCannotExecuteTrans(pMnode, topHalf)) return false;
41,641✔
1843

1844
  bool    continueExec = true;
41,638✔
1845
  int32_t code = mndTransCommit(pMnode, pTrans);
41,638✔
1846

1847
  if (code == 0) {
41,638✔
1848
    pTrans->code = 0;
41,625✔
1849
    pTrans->stage = TRN_STAGE_COMMIT_ACTION;
41,625✔
1850
    mInfo("trans:%d, stage from commit to commitAction", pTrans->id);
41,625!
1851
    continueExec = true;
41,625✔
1852
  } else {
1853
    pTrans->code = terrno;
13✔
1854
    pTrans->failedTimes++;
13✔
1855
    mError("trans:%d, stage keep on commit since %s, failedTimes:%d", pTrans->id, terrstr(), pTrans->failedTimes);
13!
1856
    continueExec = false;
13✔
1857
  }
1858

1859
  return continueExec;
41,638✔
1860
}
1861

1862
static bool mndTransPerformCommitActionStage(SMnode *pMnode, STrans *pTrans, bool topHalf) {
92,301✔
1863
  bool    continueExec = true;
92,301✔
1864
  int32_t code = mndTransExecuteCommitActions(pMnode, pTrans, topHalf);
92,301✔
1865

1866
  if (code == 0) {
92,301✔
1867
    pTrans->code = 0;
92,298✔
1868
    pTrans->stage = TRN_STAGE_FINISH;  // TRN_STAGE_PRE_FINISH is not necessary
92,298✔
1869
    mInfo("trans:%d, stage from commitAction to finished", pTrans->id);
92,298!
1870
    continueExec = true;
92,298✔
1871
  } else if (code == TSDB_CODE_MND_TRANS_CTX_SWITCH && topHalf) {
3!
1872
    pTrans->code = 0;
3✔
1873
    pTrans->stage = TRN_STAGE_COMMIT;
3✔
1874
    mInfo("trans:%d, back to commit stage", pTrans->id);
3!
1875
    continueExec = true;
3✔
1876
  } else {
1877
    pTrans->code = terrno;
×
1878
    pTrans->failedTimes++;
×
1879
    mError("trans:%d, stage keep on commitAction since %s, failedTimes:%d", pTrans->id, terrstr(), pTrans->failedTimes);
×
1880
    continueExec = false;
×
1881
  }
1882

1883
  return continueExec;
92,301✔
1884
}
1885

1886
static bool mndTransPerformUndoActionStage(SMnode *pMnode, STrans *pTrans, bool topHalf, bool notSend) {
28✔
1887
  bool    continueExec = true;
28✔
1888
  int32_t code = 0;
28✔
1889

1890
  if (pTrans->exec == TRN_EXEC_SERIAL) {
28!
1891
    code = mndTransExecuteUndoActionsSerial(pMnode, pTrans, topHalf);
×
1892
  } else {
1893
    code = mndTransExecuteUndoActions(pMnode, pTrans, topHalf, notSend);
28✔
1894
  }
1895

1896
  if (mndCannotExecuteTrans(pMnode, topHalf)) return false;
28✔
1897
  terrno = code;
21✔
1898

1899
  if (code == 0) {
21✔
1900
    pTrans->stage = TRN_STAGE_PRE_FINISH;
3✔
1901
    mInfo("trans:%d, stage from undoAction to pre-finish", pTrans->id);
3!
1902
    continueExec = true;
3✔
1903
  } else if (code == TSDB_CODE_ACTION_IN_PROGRESS || code == TSDB_CODE_MND_TRANS_CTX_SWITCH) {
18!
1904
    mInfo("trans:%d, stage keep on undoAction since %s", pTrans->id, tstrerror(code));
18!
1905
    continueExec = false;
18✔
1906
  } else {
UNCOV
1907
    pTrans->failedTimes++;
×
UNCOV
1908
    mError("trans:%d, stage keep on undoAction since %s, failedTimes:%d", pTrans->id, terrstr(), pTrans->failedTimes);
×
UNCOV
1909
    continueExec = false;
×
1910
  }
1911

1912
  return continueExec;
21✔
1913
}
1914

1915
// in trans context
1916
static bool mndTransPerformRollbackStage(SMnode *pMnode, STrans *pTrans, bool topHalf) {
3✔
1917
  if (mndCannotExecuteTrans(pMnode, topHalf)) return false;
3!
1918

1919
  bool    continueExec = true;
3✔
1920
  int32_t code = mndTransRollback(pMnode, pTrans);
3✔
1921

1922
  if (code == 0) {
3✔
1923
    pTrans->stage = TRN_STAGE_UNDO_ACTION;
2✔
1924
    continueExec = true;
2✔
1925
  } else {
1926
    pTrans->failedTimes++;
1✔
1927
    mError("trans:%d, stage keep on rollback since %s, failedTimes:%d", pTrans->id, terrstr(), pTrans->failedTimes);
1!
1928
    continueExec = false;
1✔
1929
  }
1930

1931
  return continueExec;
3✔
1932
}
1933

1934
// excute in trans context
1935
static bool mndTransPerformPreFinishStage(SMnode *pMnode, STrans *pTrans, bool topHalf) {
3✔
1936
  if (mndCannotExecuteTrans(pMnode, topHalf)) return false;
3!
1937

1938
  bool    continueExec = true;
3✔
1939
  int32_t code = mndTransPreFinish(pMnode, pTrans);
3✔
1940

1941
  if (code == 0) {
3!
1942
    pTrans->stage = TRN_STAGE_FINISH;
3✔
1943
    mInfo("trans:%d, stage from pre-finish to finish", pTrans->id);
3!
1944
    continueExec = true;
3✔
1945
  } else {
1946
    pTrans->failedTimes++;
×
1947
    mError("trans:%d, stage keep on pre-finish since %s, failedTimes:%d", pTrans->id, terrstr(), pTrans->failedTimes);
×
1948
    continueExec = false;
×
1949
  }
1950

1951
  return continueExec;
3✔
1952
}
1953

1954
static bool mndTransPerformFinishStage(SMnode *pMnode, STrans *pTrans, bool topHalf) {
92,306✔
1955
  bool continueExec = false;
92,306✔
1956
  if (topHalf) return continueExec;
92,306✔
1957

1958
  SSdbRaw *pRaw = mndTransEncode(pTrans);
50,678✔
1959
  if (pRaw == NULL) {
50,678!
1960
    mError("trans:%d, failed to encode while finish trans since %s", pTrans->id, terrstr());
×
1961
    return false;
×
1962
  }
1963
  TAOS_CHECK_RETURN(sdbSetRawStatus(pRaw, SDB_STATUS_DROPPED));
50,678!
1964

1965
  int32_t code = sdbWrite(pMnode->pSdb, pRaw);
50,678✔
1966
  if (code != 0) {
50,678!
1967
    mError("trans:%d, failed to write sdb since %s", pTrans->id, terrstr());
×
1968
  }
1969

1970
  mInfo("trans:%d, execute finished, code:0x%x, failedTimes:%d createTime:%" PRId64, pTrans->id, pTrans->code,
50,678!
1971
        pTrans->failedTimes, pTrans->createdTime);
1972
  return continueExec;
50,678✔
1973
}
1974

1975
void mndTransExecuteImp(SMnode *pMnode, STrans *pTrans, bool topHalf, bool notSend) {
221,899✔
1976
  bool continueExec = true;
221,899✔
1977

1978
  while (continueExec) {
619,373✔
1979
    mInfo("trans:%d, continue to execute stage:%s in %s, createTime:%" PRId64, pTrans->id,
397,474!
1980
          mndTransStr(pTrans->stage), mndStrExecutionContext(topHalf), pTrans->createdTime);
1981
    pTrans->lastExecTime = taosGetTimestampMs();
397,474✔
1982
    switch (pTrans->stage) {
397,474!
1983
      case TRN_STAGE_PREPARE:
×
1984
        continueExec = mndTransPerformPrepareStage(pMnode, pTrans, topHalf);
×
1985
        break;
×
1986
      case TRN_STAGE_REDO_ACTION:
171,192✔
1987
        continueExec = mndTransPerformRedoActionStage(pMnode, pTrans, topHalf, notSend);
171,192✔
1988
        break;
171,192✔
1989
      case TRN_STAGE_COMMIT:
41,641✔
1990
        continueExec = mndTransPerformCommitStage(pMnode, pTrans, topHalf);
41,641✔
1991
        break;
41,641✔
1992
      case TRN_STAGE_COMMIT_ACTION:
92,301✔
1993
        continueExec = mndTransPerformCommitActionStage(pMnode, pTrans, topHalf);
92,301✔
1994
        break;
92,301✔
1995
      case TRN_STAGE_ROLLBACK:
3✔
1996
        continueExec = mndTransPerformRollbackStage(pMnode, pTrans, topHalf);
3✔
1997
        break;
3✔
1998
      case TRN_STAGE_UNDO_ACTION:
28✔
1999
        continueExec = mndTransPerformUndoActionStage(pMnode, pTrans, topHalf, notSend);
28✔
2000
        break;
28✔
2001
      case TRN_STAGE_PRE_FINISH:
3✔
2002
        continueExec = mndTransPerformPreFinishStage(pMnode, pTrans, topHalf);
3✔
2003
        break;
3✔
2004
      case TRN_STAGE_FINISH:
92,306✔
2005
        continueExec = mndTransPerformFinishStage(pMnode, pTrans, topHalf);
92,306✔
2006
        break;
92,306✔
2007
      default:
×
2008
        continueExec = false;
×
2009
        break;
×
2010
    }
2011
  }
2012

2013
  mndTransSendRpcRsp(pMnode, pTrans);
221,899✔
2014
}
221,899✔
2015

2016
// start trans, pullup, receive rsp, kill
2017
void mndTransExecute(SMnode *pMnode, STrans *pTrans, bool notSend) {
115,092✔
2018
  bool topHalf = true;
115,092✔
2019
  mndTransExecuteImp(pMnode, pTrans, topHalf, notSend);
115,092✔
2020
}
115,092✔
2021

2022
// update trans
2023
void mndTransRefresh(SMnode *pMnode, STrans *pTrans) {
106,807✔
2024
  bool topHalf = false;
106,807✔
2025
  mndTransExecuteImp(pMnode, pTrans, topHalf, false);
106,807✔
2026
}
106,807✔
2027

2028
static int32_t mndProcessTransTimer(SRpcMsg *pReq) {
25,105✔
2029
  mTrace("start to process trans timer");
25,105✔
2030
  mndTransPullup(pReq->info.node);
25,105✔
2031
  return 0;
25,105✔
2032
}
2033

2034
int32_t mndKillTrans(SMnode *pMnode, STrans *pTrans) {
×
2035
  SArray *pArray = NULL;
×
2036
  if (pTrans->stage == TRN_STAGE_REDO_ACTION) {
×
2037
    pArray = pTrans->redoActions;
×
2038
  } else if (pTrans->stage == TRN_STAGE_UNDO_ACTION) {
×
2039
    pArray = pTrans->undoActions;
×
2040
  } else {
2041
    TAOS_RETURN(TSDB_CODE_MND_TRANS_INVALID_STAGE);
×
2042
  }
2043

2044
  if(pTrans->ableToBeKilled == false){
×
2045
    return TSDB_CODE_MND_TRANS_NOT_ABLE_TO_kILLED;
×
2046
  }
2047
  
2048
  if(pTrans->killMode == TRN_KILL_MODE_SKIP){
×
2049
    for (int32_t i = 0; i < taosArrayGetSize(pArray); ++i) {
×
2050
      STransAction *pAction = taosArrayGet(pArray, i);
×
2051
      mInfo("trans:%d, %s:%d set processed for kill msg received, errCode from %s to success", pTrans->id,
×
2052
            mndTransStr(pAction->stage), i, tstrerror(pAction->errCode));
2053
      pAction->msgSent = 1;
×
2054
      pAction->msgReceived = 1;
×
2055
      pAction->errCode = 0;
×
2056
    }
2057
  }
2058
  else if(pTrans->killMode == TRN_KILL_MODE_INTERUPT){
×
2059
    pTrans->stage = TRN_STAGE_PRE_FINISH;
×
2060
  }
2061
  else{
2062
    return TSDB_CODE_MND_TRANS_NOT_ABLE_TO_kILLED;
×
2063
  }
2064

2065
  mInfo("trans:%d, execute transaction in kill trans", pTrans->id);
×
2066
  mndTransExecute(pMnode, pTrans, true);
×
2067
  return 0;
×
2068
}
2069

2070
static int32_t mndProcessKillTransReq(SRpcMsg *pReq) {
×
2071
  SMnode       *pMnode = pReq->info.node;
×
2072
  SKillTransReq killReq = {0};
×
2073
  int32_t       code = -1;
×
2074
  STrans       *pTrans = NULL;
×
2075

2076
  if (tDeserializeSKillTransReq(pReq->pCont, pReq->contLen, &killReq) != 0) {
×
2077
    code = TSDB_CODE_INVALID_MSG;
×
2078
    goto _OVER;
×
2079
  }
2080

2081
  mInfo("trans:%d, start to kill", killReq.transId);
×
2082
  if ((code = mndCheckOperPrivilege(pMnode, pReq->info.conn.user, MND_OPER_KILL_TRANS)) != 0) {
×
2083
    goto _OVER;
×
2084
  }
2085

2086
  pTrans = mndAcquireTrans(pMnode, killReq.transId);
×
2087
  if (pTrans == NULL) {
×
2088
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
2089
    if (terrno != 0) code = terrno;
×
2090
    goto _OVER;
×
2091
  }
2092

2093
  code = mndKillTrans(pMnode, pTrans);
×
2094

2095
_OVER:
×
2096
  if (code != 0) {
×
2097
    mError("trans:%d, failed to kill since %s", killReq.transId, terrstr());
×
2098
  }
2099

2100
  mndReleaseTrans(pMnode, pTrans);
×
2101
  TAOS_RETURN(code);
×
2102
}
2103

2104
static int32_t mndCompareTransId(int32_t *pTransId1, int32_t *pTransId2) { return *pTransId1 >= *pTransId2 ? 1 : 0; }
219✔
2105

2106
void mndTransPullup(SMnode *pMnode) {
25,598✔
2107
  SSdb   *pSdb = pMnode->pSdb;
25,598✔
2108
  SArray *pArray = taosArrayInit(sdbGetSize(pSdb, SDB_TRANS), sizeof(int32_t));
25,598✔
2109
  if (pArray == NULL) return;
25,598!
2110

2111
  void *pIter = NULL;
25,598✔
2112
  while (1) {
4,694✔
2113
    STrans *pTrans = NULL;
30,292✔
2114
    pIter = sdbFetch(pMnode->pSdb, SDB_TRANS, pIter, (void **)&pTrans);
30,292✔
2115
    if (pIter == NULL) break;
30,292✔
2116
    if (taosArrayPush(pArray, &pTrans->id) == NULL) {
9,388!
2117
      mError("failed to put trans into array, trans:%d, but pull up will continute", pTrans->id);
×
2118
    }
2119
    sdbRelease(pSdb, pTrans);
4,694✔
2120
  }
2121

2122
  taosArraySort(pArray, (__compar_fn_t)mndCompareTransId);
25,598✔
2123

2124
  for (int32_t i = 0; i < taosArrayGetSize(pArray); ++i) {
30,292✔
2125
    int32_t *pTransId = taosArrayGet(pArray, i);
4,694✔
2126
    STrans  *pTrans = mndAcquireTrans(pMnode, *pTransId);
4,694✔
2127
    if (pTrans != NULL) {
4,694!
2128
      mInfo("trans:%d, execute transaction in trans pullup", pTrans->id);
4,694!
2129
      mndTransExecute(pMnode, pTrans, false);
4,694✔
2130
    }
2131
    mndReleaseTrans(pMnode, pTrans);
4,694✔
2132
  }
2133
  taosArrayDestroy(pArray);
25,598✔
2134
}
2135

2136
static char *formatTimestamp(char *buf, int64_t val, int precision) {
13,038✔
2137
  time_t tt;
2138
  if (precision == TSDB_TIME_PRECISION_MICRO) {
13,038!
2139
    tt = (time_t)(val / 1000000);
×
2140
  }
2141
  if (precision == TSDB_TIME_PRECISION_NANO) {
13,038!
2142
    tt = (time_t)(val / 1000000000);
×
2143
  } else {
2144
    tt = (time_t)(val / 1000);
13,038✔
2145
  }
2146

2147
  struct tm tm;
2148
  if (taosLocalTime(&tt, &tm, NULL, 0, NULL) == NULL) {
13,038!
2149
    mError("failed to get local time");
×
2150
    return NULL;
×
2151
  }
2152
  size_t pos = taosStrfTime(buf, 32, "%Y-%m-%d %H:%M:%S", &tm);
13,038✔
2153

2154
  if (precision == TSDB_TIME_PRECISION_MICRO) {
13,038!
2155
    sprintf(buf + pos, ".%06d", (int)(val % 1000000));
×
2156
  } else if (precision == TSDB_TIME_PRECISION_NANO) {
13,038!
2157
    sprintf(buf + pos, ".%09d", (int)(val % 1000000000));
×
2158
  } else {
2159
    sprintf(buf + pos, ".%03d", (int)(val % 1000));
13,038✔
2160
  }
2161

2162
  return buf;
13,038✔
2163
}
2164

2165
static void mndTransLogAction(STrans *pTrans) {
172✔
2166
  char    detail[512] = {0};
172✔
2167
  int32_t len = 0;
172✔
2168
  int32_t index = 0;
172✔
2169

2170
  if (pTrans->stage == TRN_STAGE_PREPARE) {
172!
2171
    for (int32_t i = 0; i < taosArrayGetSize(pTrans->prepareActions); ++i, ++index) {
×
2172
      len = 0;
×
2173
      STransAction *pAction = taosArrayGet(pTrans->prepareActions, i);
×
2174
      len += snprintf(detail + len, sizeof(detail) - len, "action:%d, %s:%d sdbType:%s, sdbStatus:%s\n", index,
×
2175
                      mndTransStr(pAction->stage), pAction->id, sdbTableName(pAction->pRaw->type),
×
2176
                      sdbStatusName(pAction->pRaw->status));
×
2177
      mDebug("trans:%d, show tran action, detail:%s", pTrans->id, detail);
×
2178
    }
2179
  }
2180

2181
  if (pTrans->stage == TRN_STAGE_REDO_ACTION) {
172!
2182
    for (int32_t i = 0; i < taosArrayGetSize(pTrans->redoActions); ++i, ++index) {
8,226✔
2183
      len = 0;
8,054✔
2184
      STransAction *pAction = taosArrayGet(pTrans->redoActions, i);
8,054✔
2185
      if (pAction->actionType == TRANS_ACTION_MSG) {
8,054✔
2186
        char bufStart[40] = {0};
6,519✔
2187
        (void)formatTimestamp(bufStart, pAction->startTime, TSDB_TIME_PRECISION_MILLI);
6,519✔
2188

2189
        char endStart[40] = {0};
6,519✔
2190
        (void)formatTimestamp(endStart, pAction->endTime, TSDB_TIME_PRECISION_MILLI);
6,519✔
2191
        len += snprintf(detail + len, sizeof(detail) - len,
13,038!
2192
                        "action:%d, %s:%d msgType:%s,"
2193
                        "sent:%d, received:%d, startTime:%s, endTime:%s, ",
2194
                        index, mndTransStr(pAction->stage), pAction->id, TMSG_INFO(pAction->msgType), pAction->msgSent,
6,519✔
2195
                        pAction->msgReceived, bufStart, endStart);
6,519✔
2196

2197
        SEpSet epset = pAction->epSet;
6,519✔
2198
        if (epset.numOfEps > 0) {
6,519!
2199
          len += snprintf(detail + len, sizeof(detail) - len, "numOfEps:%d inUse:%d ", epset.numOfEps, epset.inUse);
6,519✔
2200
          for (int32_t i = 0; i < epset.numOfEps; ++i) {
15,137✔
2201
            len +=
8,618✔
2202
                snprintf(detail + len, sizeof(detail) - len, "ep:%d-%s:%u ", i, epset.eps[i].fqdn, epset.eps[i].port);
8,618✔
2203
          }
2204
        }
2205

2206
        len += snprintf(detail + len, sizeof(detail) - len, ", errCode:0x%x(%s)\n", pAction->errCode & 0xFFFF,
6,519✔
2207
                        tstrerror(pAction->errCode));
2208
      } else {
2209
        len += snprintf(detail + len, sizeof(detail) - len, "action:%d, %s:%d sdbType:%s, sdbStatus:%s, written:%d\n",
1,535✔
2210
                        index, mndTransStr(pAction->stage), pAction->id, sdbTableName(pAction->pRaw->type),
1,535✔
2211
                        sdbStatusName(pAction->pRaw->status), pAction->rawWritten);
1,535✔
2212
      }
2213
      mDebug("trans:%d, show tran action, detail:%s", pTrans->id, detail);
8,054!
2214
    }
2215
  }
2216

2217
  if (pTrans->stage == TRN_STAGE_COMMIT_ACTION) {
172!
2218
    for (int32_t i = 0; i < taosArrayGetSize(pTrans->commitActions); ++i, ++index) {
×
2219
      len = 0;
×
2220
      STransAction *pAction = taosArrayGet(pTrans->commitActions, i);
×
2221
      len += snprintf(detail + len, sizeof(detail) - len, "action:%d, %s:%d sdbType:%s, sdbStatus:%s\n", index,
×
2222
                      mndTransStr(pAction->stage), i, sdbTableName(pAction->pRaw->type),
×
2223
                      sdbStatusName(pAction->pRaw->status));
×
2224
      mDebug("trans:%d, show tran action, detail:%s", pTrans->id, detail);
×
2225
    }
2226

2227
    for (int32_t i = 0; i < taosArrayGetSize(pTrans->undoActions); ++i, ++index) {
×
2228
      len = 0;
×
2229
      STransAction *pAction = taosArrayGet(pTrans->undoActions, i);
×
2230
      if (pAction->actionType == TRANS_ACTION_MSG) {
×
2231
        len += snprintf(detail + len, sizeof(detail) - len, "action:%d, %s:%d msgType:%s\n", index,
×
2232
                        mndTransStr(pAction->stage), pAction->id, TMSG_INFO(pAction->msgType));
×
2233
        ;
2234
      } else {
2235
        len += snprintf(detail + len, sizeof(detail) - len, "action:%d, %s:%d sdbType:%s, sdbStatus:%s\n", index,
×
2236
                        mndTransStr(pAction->stage), pAction->id, sdbTableName(pAction->pRaw->type),
×
2237
                        sdbStatusName(pAction->pRaw->status));
×
2238
      }
2239
      mDebug("trans:%d, show tran action, detail:%s", pTrans->id, detail);
×
2240
    }
2241
  }
2242
}
172✔
2243

2244
static int32_t mndRetrieveTrans(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows) {
184✔
2245
  SMnode *pMnode = pReq->info.node;
184✔
2246
  SSdb   *pSdb = pMnode->pSdb;
184✔
2247
  int32_t numOfRows = 0;
184✔
2248
  STrans *pTrans = NULL;
184✔
2249
  int32_t cols = 0;
184✔
2250
  int32_t code = 0;
184✔
2251
  int32_t lino = 0;
184✔
2252

2253
  while (numOfRows < rows) {
356!
2254
    pShow->pIter = sdbFetch(pSdb, SDB_TRANS, pShow->pIter, (void **)&pTrans);
356✔
2255
    if (pShow->pIter == NULL) break;
356✔
2256

2257
    cols = 0;
172✔
2258

2259
    SColumnInfoData *pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
172✔
2260
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)&pTrans->id, false), pTrans, &lino, _OVER);
172!
2261

2262
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
172✔
2263
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)&pTrans->createdTime, false), pTrans, &lino,
172!
2264
                        _OVER);
2265

2266
    char stage[TSDB_TRANS_STAGE_LEN + VARSTR_HEADER_SIZE] = {0};
172✔
2267
    STR_WITH_MAXSIZE_TO_VARSTR(stage, mndTransStr(pTrans->stage), pShow->pMeta->pSchemas[cols].bytes);
172✔
2268
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
172✔
2269
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)stage, false), pTrans, &lino, _OVER);
172!
2270

2271
    char opername[TSDB_TRANS_OPER_LEN + VARSTR_HEADER_SIZE] = {0};
172✔
2272
    STR_WITH_MAXSIZE_TO_VARSTR(opername, pTrans->opername, pShow->pMeta->pSchemas[cols].bytes);
172✔
2273
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
172✔
2274
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)opername, false), pTrans, &lino, _OVER);
172!
2275

2276
    char dbname[TSDB_DB_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
172✔
2277
    STR_WITH_MAXSIZE_TO_VARSTR(dbname, mndGetDbStr(pTrans->dbname), pShow->pMeta->pSchemas[cols].bytes);
172✔
2278
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
172✔
2279
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)dbname, false), pTrans, &lino, _OVER);
172!
2280

2281
    char stbname[TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
172✔
2282
    STR_WITH_MAXSIZE_TO_VARSTR(stbname, mndGetDbStr(pTrans->stbname), pShow->pMeta->pSchemas[cols].bytes);
172✔
2283
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
172✔
2284
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)stbname, false), pTrans, &lino, _OVER);
172!
2285

2286
    const char *killableStr = pTrans->ableToBeKilled ? "yes" : "no";
172!
2287
    char        killableVstr[10 + VARSTR_HEADER_SIZE] = {0};
172✔
2288
    STR_WITH_MAXSIZE_TO_VARSTR(killableVstr, killableStr, 10 + VARSTR_HEADER_SIZE);
172✔
2289
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
172✔
2290
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)killableVstr, false), pTrans, &lino, _OVER);
172!
2291

2292
    /*
2293
    const char *killModeStr = pTrans->killMode == TRN_KILL_MODE_SKIP ? "skip" : "interrupt";
2294
    char        killModeVstr[10 + VARSTR_HEADER_SIZE] = {0};
2295
    STR_WITH_MAXSIZE_TO_VARSTR(killModeVstr, killModeStr, 24);
2296
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
2297
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)killModeVstr, false), pTrans, &lino, _OVER);
2298
    */
2299

2300
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
172✔
2301
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)&pTrans->failedTimes, false), pTrans, &lino,
172!
2302
                        _OVER);
2303

2304
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
172✔
2305
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)&pTrans->lastExecTime, false), pTrans, &lino,
172!
2306
                        _OVER);
2307

2308
    char    lastInfo[TSDB_TRANS_ERROR_LEN + VARSTR_HEADER_SIZE] = {0};
172✔
2309
    char    detail[TSDB_TRANS_ERROR_LEN + 1] = {0};
172✔
2310
    int32_t len = tsnprintf(detail, sizeof(detail), "action:%d code:0x%x(%s) ", pTrans->lastAction,
516✔
2311
                            pTrans->lastErrorNo & 0xFFFF, tstrerror(pTrans->lastErrorNo));
172✔
2312
    SEpSet  epset = pTrans->lastEpset;
172✔
2313
    if (epset.numOfEps > 0) {
172!
2314
      len += tsnprintf(detail + len, sizeof(detail) - len, "msgType:%s numOfEps:%d inUse:%d ",
344!
2315
                       TMSG_INFO(pTrans->lastMsgType), epset.numOfEps, epset.inUse);
344✔
2316
      for (int32_t i = 0; i < pTrans->lastEpset.numOfEps; ++i) {
479✔
2317
        len += snprintf(detail + len, sizeof(detail) - len, "ep:%d-%s:%u ", i, epset.eps[i].fqdn, epset.eps[i].port);
307✔
2318
      }
2319
    }
2320
    STR_WITH_MAXSIZE_TO_VARSTR(lastInfo, detail, pShow->pMeta->pSchemas[cols].bytes);
172✔
2321
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
172✔
2322
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)lastInfo, false), pTrans, &lino, _OVER);
172!
2323

2324
    mndTransLogAction(pTrans);
172✔
2325

2326
    numOfRows++;
172✔
2327
    sdbRelease(pSdb, pTrans);
172✔
2328
  }
2329

2330
_OVER:
×
2331
  if (code != 0) mError("failed to retrieve at line:%d, since %s", lino, tstrerror(code));
184!
2332
  pShow->numOfRows += numOfRows;
184✔
2333
  return numOfRows;
184✔
2334
}
2335

2336
static int32_t mndShowTransCommonColumns(SShowObj *pShow, SSDataBlock *pBlock, STransAction *pAction,
×
2337
                                         int32_t transactionId, int32_t curActionId, int32_t numOfRows, int32_t *cols) {
2338
  int32_t code = 0;
×
2339
  int32_t lino = 0;
×
2340
  int32_t len = 0;
×
2341

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

2345
  char action[30 + 1] = {0};
×
2346
  if (curActionId == pAction->id) {
×
2347
    len += snprintf(action + len, sizeof(action) - len, "%s:%d(%s)<-last", mndTransStr(pAction->stage), pAction->id,
×
2348
                    mndTransTypeStr(pAction->actionType));
2349
  } else {
2350
    len += snprintf(action + len, sizeof(action) - len, "%s:%d(%s)", mndTransStr(pAction->stage), pAction->id,
×
2351
                    mndTransTypeStr(pAction->actionType));
2352
  }
2353
  char actionVStr[30 + VARSTR_HEADER_SIZE] = {0};
×
2354
  STR_WITH_MAXSIZE_TO_VARSTR(actionVStr, action, pShow->pMeta->pSchemas[*cols].bytes);
×
2355
  pColInfo = taosArrayGet(pBlock->pDataBlock, (*cols)++);
×
2356
  TAOS_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)actionVStr, false), &lino, _OVER);
×
2357
_OVER:
×
2358
  if (code != 0) mError("failed to retrieve at line:%d, since %s", lino, tstrerror(code));
×
2359
  return code;
×
2360
}
2361

2362
static void mndShowTransAction(SShowObj *pShow, SSDataBlock *pBlock, STransAction *pAction, int32_t transactionId,
×
2363
                               int32_t curActionId, int32_t rows, int32_t numOfRows) {
2364
  int32_t code = 0;
×
2365
  int32_t lino = 0;
×
2366
  int32_t len = 0;
×
2367
  int32_t cols = 0;
×
2368

2369
  cols = 0;
×
2370

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

2373
  if (pAction->actionType == TRANS_ACTION_MSG) {
×
2374
    int32_t len = 0;
×
2375

2376
    char objType[TSDB_TRANS_OBJTYPE_LEN + 1] = {0};
×
2377
    len += snprintf(objType + len, sizeof(objType) - len, "%s(s:%d,r:%d)", TMSG_INFO(pAction->msgType),
×
2378
                    pAction->msgSent, pAction->msgReceived);
×
2379
    char objTypeVStr[TSDB_TRANS_OBJTYPE_LEN + VARSTR_HEADER_SIZE] = {0};
×
2380
    STR_WITH_MAXSIZE_TO_VARSTR(objTypeVStr, objType, pShow->pMeta->pSchemas[cols].bytes);
×
2381
    SColumnInfoData *pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
×
2382
    TAOS_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)objTypeVStr, false), &lino, _OVER);
×
2383

2384
    char result[TSDB_TRANS_RESULT_LEN + 1] = {0};
×
2385
    len = 0;
×
2386
    len += snprintf(result + len, sizeof(result) - len, "errCode:0x%x(%s)", pAction->errCode & 0xFFFF,
×
2387
                    tstrerror(pAction->errCode));
2388
    char resultVStr[TSDB_TRANS_RESULT_LEN + VARSTR_HEADER_SIZE] = {0};
×
2389
    STR_WITH_MAXSIZE_TO_VARSTR(resultVStr, result, pShow->pMeta->pSchemas[cols].bytes);
×
2390
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
×
2391
    TAOS_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)resultVStr, false), &lino, _OVER);
×
2392

2393
    char target[TSDB_TRANS_TARGET_LEN] = {0};
×
2394
    len = 0;
×
2395
    SEpSet epset = pAction->epSet;
×
2396
    if (epset.numOfEps > 0) {
×
2397
      for (int32_t i = 0; i < epset.numOfEps; ++i) {
×
2398
        len += snprintf(target + len, sizeof(target) - len, "ep:%d-%s:%u,", i, epset.eps[i].fqdn, epset.eps[i].port);
×
2399
      }
2400
      len += snprintf(target + len, sizeof(target) - len, "(%d:%d) ", epset.numOfEps, epset.inUse);
×
2401
    }
2402
    char targetVStr[TSDB_TRANS_TARGET_LEN + VARSTR_HEADER_SIZE] = {0};
×
2403
    STR_WITH_MAXSIZE_TO_VARSTR(targetVStr, target, pShow->pMeta->pSchemas[cols].bytes);
×
2404
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
×
2405
    TAOS_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)targetVStr, false), &lino, _OVER);
×
2406

2407
    char detail[TSDB_TRANS_DETAIL_LEN] = {0};
×
2408
    len = 0;
×
2409
    char bufStart[40] = {0};
×
2410
    if (pAction->startTime > 0) (void)formatTimestamp(bufStart, pAction->startTime, TSDB_TIME_PRECISION_MILLI);
×
2411
    char bufEnd[40] = {0};
×
2412
    if (pAction->endTime > 0) (void)formatTimestamp(bufEnd, pAction->endTime, TSDB_TIME_PRECISION_MILLI);
×
2413
    len += snprintf(detail + len, sizeof(detail) - len, "startTime:%s, endTime:%s, ", bufStart, bufEnd);
×
2414
    char detailVStr[TSDB_TRANS_DETAIL_LEN + VARSTR_HEADER_SIZE] = {0};
×
2415
    STR_WITH_MAXSIZE_TO_VARSTR(detailVStr, detail, pShow->pMeta->pSchemas[cols].bytes);
×
2416
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
×
2417
    TAOS_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)detailVStr, false), &lino, _OVER);
×
2418

2419
  } else {
2420
    int32_t len = 0;
×
2421

2422
    char objType[TSDB_TRANS_OBJTYPE_LEN + 1] = {0};
×
2423
    if (pAction->pRaw->type == SDB_VGROUP) {
×
2424
      SSdbRow *pRow = mndVgroupActionDecode(pAction->pRaw);
×
2425
      SVgObj  *pVgroup = sdbGetRowObj(pRow);
×
2426
      len += snprintf(objType + len, sizeof(objType) - len, "%s(%d)", sdbTableName(pAction->pRaw->type), pVgroup->vgId);
×
2427
      taosMemoryFreeClear(pRow);
×
2428
    } else {
2429
      strcpy(objType, sdbTableName(pAction->pRaw->type));
×
2430
    }
2431
    char objTypeVStr[TSDB_TRANS_OBJTYPE_LEN + VARSTR_HEADER_SIZE] = {0};
×
2432
    STR_WITH_MAXSIZE_TO_VARSTR(objTypeVStr, objType, pShow->pMeta->pSchemas[cols].bytes);
×
2433
    SColumnInfoData *pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
×
2434
    TAOS_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)objTypeVStr, false), &lino, _OVER);
×
2435

2436
    char result[TSDB_TRANS_RESULT_LEN + 1] = {0};
×
2437
    len = 0;
×
2438
    len += snprintf(result + len, sizeof(result) - len, "rawWritten:%d", pAction->rawWritten);
×
2439
    char resultVStr[TSDB_TRANS_RESULT_LEN + VARSTR_HEADER_SIZE] = {0};
×
2440
    STR_WITH_MAXSIZE_TO_VARSTR(resultVStr, result, pShow->pMeta->pSchemas[cols].bytes);
×
2441
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
×
2442
    TAOS_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)resultVStr, false), &lino, _OVER);
×
2443

2444
    char target[TSDB_TRANS_TARGET_LEN] = "";
×
2445
    char targetVStr[TSDB_TRANS_TARGET_LEN + VARSTR_HEADER_SIZE] = {0};
×
2446
    STR_WITH_MAXSIZE_TO_VARSTR(targetVStr, target, pShow->pMeta->pSchemas[cols].bytes);
×
2447
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
×
2448
    TAOS_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)targetVStr, false), &lino, _OVER);
×
2449

2450
    char detail[TSDB_TRANS_DETAIL_LEN] = {0};
×
2451
    len = 0;
×
2452
    len += snprintf(detail + len, sizeof(detail) - len, "sdbStatus:%s", sdbStatusName(pAction->pRaw->status));
×
2453
    char detailVStr[TSDB_TRANS_DETAIL_LEN + VARSTR_HEADER_SIZE] = {0};
×
2454
    STR_WITH_MAXSIZE_TO_VARSTR(detailVStr, detail, pShow->pMeta->pSchemas[cols].bytes);
×
2455
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
×
2456
    TAOS_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)detailVStr, false), &lino, _OVER);
×
2457
  }
2458

2459
_OVER:
×
2460
  if (code != 0) mError("failed to retrieve at line:%d, since %s", lino, tstrerror(code));
×
2461
}
2462

2463
static SArray *mndTransGetAction(STrans *pTrans, ETrnStage stage) {
×
2464
  if (stage == TRN_STAGE_PREPARE) {
×
2465
    return pTrans->prepareActions;
×
2466
  }
2467
  if (stage == TRN_STAGE_REDO_ACTION) {
×
2468
    return pTrans->redoActions;
×
2469
  }
2470
  if (stage == TRN_STAGE_COMMIT_ACTION) {
×
2471
    return pTrans->commitActions;
×
2472
  }
2473
  if (stage == TRN_STAGE_UNDO_ACTION) {
×
2474
    return pTrans->undoActions;
×
2475
  }
2476
  return NULL;
×
2477
}
2478

2479
typedef struct STransDetailIter {
2480
  void     *pIter;
2481
  STrans   *pTrans;
2482
  ETrnStage stage;
2483
  int32_t   num;
2484
} STransDetailIter;
2485

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

2491
  for (int32_t i = start; i < actionNum; ++i) {
×
2492
    STransAction *pAction = taosArrayGet(pShowIter->pTrans->redoActions, i);
×
2493
    mndShowTransAction(pShow, pBlock, pAction, pShowIter->pTrans->id, pShowIter->pTrans->lastAction, rows, *numOfRows);
×
2494
    (*numOfRows)++;
×
2495
    if (*numOfRows >= rows) break;
×
2496
  }
2497

2498
  if (*numOfRows == end) {
×
2499
    sdbRelease(pSdb, pShowIter->pTrans);
×
2500
    pShowIter->pTrans = NULL;
×
2501
    pShowIter->num = 0;
×
2502
  } else {
2503
    pShowIter->pTrans = pShowIter->pTrans;
×
2504
    pShowIter->stage = pShowIter->pTrans->stage;
×
2505
    pShowIter->num += (*numOfRows);
×
2506
  }
2507
}
×
2508

2509
static int32_t mndRetrieveTransDetail(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows) {
×
2510
  SMnode *pMnode = pReq->info.node;
×
2511
  SSdb   *pSdb = pMnode->pSdb;
×
2512
  int32_t numOfRows = 0;
×
2513

2514
  int32_t code = 0;
×
2515
  int32_t lino = 0;
×
2516

2517
  mInfo("start to mndRetrieveTransDetail, rows:%d, pShow->numOfRows:%d, pShow->pIter:%p", rows, pShow->numOfRows,
×
2518
        pShow->pIter);
2519

2520
  if (pShow->pIter == NULL) {
×
2521
    pShow->pIter = taosMemoryMalloc(sizeof(STransDetailIter));
×
2522
    if (pShow->pIter == NULL) {
×
2523
      mError("failed to malloc for pShow->pIter");
×
2524
      return 0;
×
2525
    }
2526
    memset(pShow->pIter, 0, sizeof(STransDetailIter));
×
2527
  }
2528

2529
  STransDetailIter *pShowIter = (STransDetailIter *)pShow->pIter;
×
2530

2531
  while (numOfRows < rows) {
×
2532
    if (pShowIter->pTrans == NULL) {
×
2533
      pShowIter->pIter = sdbFetch(pSdb, SDB_TRANS, pShowIter->pIter, (void **)&(pShowIter->pTrans));
×
2534
      mDebug("retrieve trans detail from fetch, pShow->pIter:%p, pTrans:%p", pShowIter->pIter, pShowIter->pTrans);
×
2535
      if (pShowIter->pIter == NULL) break;
×
2536
      mInfo("retrieve trans detail from fetch, id:%d, trans stage:%d, IterNum:%d", pShowIter->pTrans->id,
×
2537
            pShowIter->pTrans->stage, pShowIter->num);
2538

2539
      SArray *pActions = mndTransGetAction(pShowIter->pTrans, pShowIter->pTrans->stage);
×
2540

2541
      mndTransShowActions(pSdb, pShowIter, pShow, pBlock, rows, &numOfRows, pActions, taosArrayGetSize(pActions), 0);
×
2542
      break;
×
2543
    } else {
2544
      mInfo("retrieve trans detail from iter, id:%d, iterStage:%d, IterNum:%d", pShowIter->pTrans->id, pShowIter->stage,
×
2545
            pShowIter->num);
2546
      SArray *pActions = mndTransGetAction(pShowIter->pTrans, pShowIter->stage);
×
2547

2548
      mndTransShowActions(pSdb, pShowIter, pShow, pBlock, rows, &numOfRows, pActions,
×
2549
                          taosArrayGetSize(pActions) - pShowIter->num, pShowIter->num);
×
2550
      break;
×
2551
    }
2552
  }
2553

2554
_OVER:
×
2555
  pShow->numOfRows += numOfRows;
×
2556

2557
  if (code != 0) {
×
2558
    mError("failed to retrieve at line:%d, since %s", lino, tstrerror(code));
×
2559
  } else {
2560
    mInfo("retrieve trans detail, numOfRows:%d, pShow->numOfRows:%d", numOfRows, pShow->numOfRows)
×
2561
  }
2562
  if (numOfRows == 0) {
×
2563
    taosMemoryFree(pShow->pIter);
×
2564
    pShow->pIter = NULL;
×
2565
  }
2566
  return numOfRows;
×
2567
}
2568

2569
static void mndCancelGetNextTrans(SMnode *pMnode, void *pIter) {
×
2570
  SSdb *pSdb = pMnode->pSdb;
×
2571
  sdbCancelFetchByType(pSdb, pIter, SDB_TRANS);
×
2572
}
×
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