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

#define _DEFAULT_SOURCE
S
Shengliang Guan 已提交
17
#include "mndTrans.h"
L
Liu Jicong 已提交
18
#include "mndConsumer.h"
S
Shengliang Guan 已提交
19
#include "mndDb.h"
20
#include "mndStb.h"
21
#include "mndPrivilege.h"
S
Shengliang Guan 已提交
22
#include "mndShow.h"
S
Shengliang Guan 已提交
23
#include "mndSync.h"
S
Shengliang Guan 已提交
24
#include "mndUser.h"
S
Shengliang Guan 已提交
25

26 27
#define TRANS_VER_NUMBER   1
#define TRANS_ARRAY_SIZE   8
S
Shengliang Guan 已提交
28
#define TRANS_RESERVE_SIZE 64
S
Shengliang Guan 已提交
29

S
Shengliang Guan 已提交
30 31 32
static SSdbRaw *mndTransActionEncode(STrans *pTrans);
static SSdbRow *mndTransActionDecode(SSdbRaw *pRaw);
static int32_t  mndTransActionInsert(SSdb *pSdb, STrans *pTrans);
S
Shengliang Guan 已提交
33
static int32_t  mndTransActionUpdate(SSdb *pSdb, STrans *OldTrans, STrans *pOld);
34
static int32_t  mndTransActionDelete(SSdb *pSdb, STrans *pTrans, bool callFunc);
S
Shengliang Guan 已提交
35

S
Shengliang Guan 已提交
36
static int32_t mndTransAppendLog(SArray *pArray, SSdbRaw *pRaw);
S
Shengliang Guan 已提交
37
static int32_t mndTransAppendAction(SArray *pArray, STransAction *pAction);
S
Shengliang Guan 已提交
38 39
static void    mndTransDropLogs(SArray *pArray);
static void    mndTransDropActions(SArray *pArray);
40
static void    mndTransDropData(STrans *pTrans);
41
static int32_t mndTransExecuteActions(SMnode *pMnode, STrans *pTrans, SArray *pArray);
S
Shengliang Guan 已提交
42 43 44 45
static int32_t mndTransExecuteRedoLogs(SMnode *pMnode, STrans *pTrans);
static int32_t mndTransExecuteUndoLogs(SMnode *pMnode, STrans *pTrans);
static int32_t mndTransExecuteRedoActions(SMnode *pMnode, STrans *pTrans);
static int32_t mndTransExecuteUndoActions(SMnode *pMnode, STrans *pTrans);
46
static int32_t mndTransExecuteCommitActions(SMnode *pMnode, STrans *pTrans);
S
Shengliang Guan 已提交
47 48 49 50 51
static bool    mndTransPerformPrepareStage(SMnode *pMnode, STrans *pTrans);
static bool    mndTransPerformRedoLogStage(SMnode *pMnode, STrans *pTrans);
static bool    mndTransPerformRedoActionStage(SMnode *pMnode, STrans *pTrans);
static bool    mndTransPerformUndoLogStage(SMnode *pMnode, STrans *pTrans);
static bool    mndTransPerformUndoActionStage(SMnode *pMnode, STrans *pTrans);
52
static bool    mndTransPerformCommitActionStage(SMnode *pMnode, STrans *pTrans);
S
Shengliang Guan 已提交
53 54 55
static bool    mndTransPerformCommitStage(SMnode *pMnode, STrans *pTrans);
static bool    mndTransPerformRollbackStage(SMnode *pMnode, STrans *pTrans);
static bool    mndTransPerfromFinishedStage(SMnode *pMnode, STrans *pTrans);
56
static bool    mndCannotExecuteTransAction(SMnode *pMnode) { return !pMnode->deploy && !mndIsMaster(pMnode); }
S
Shengliang Guan 已提交
57

58
static void    mndTransSendRpcRsp(SMnode *pMnode, STrans *pTrans);
59
static int32_t mndProcessTransTimer(SRpcMsg *pReq);
wmmhello's avatar
wmmhello 已提交
60
static int32_t mndProcessTtl(SRpcMsg *pReq);
S
Shengliang Guan 已提交
61
static int32_t mndProcessKillTransReq(SRpcMsg *pReq);
S
Shengliang Guan 已提交
62

S
Shengliang Guan 已提交
63
static int32_t mndRetrieveTrans(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows);
S
Shengliang Guan 已提交
64 65
static void    mndCancelGetNextTrans(SMnode *pMnode, void *pIter);

S
Shengliang Guan 已提交
66
int32_t mndInitTrans(SMnode *pMnode) {
S
Shengliang Guan 已提交
67 68 69 70 71 72 73 74 75
  SSdbTable table = {
      .sdbType = SDB_TRANS,
      .keyType = SDB_KEY_INT32,
      .encodeFp = (SdbEncodeFp)mndTransActionEncode,
      .decodeFp = (SdbDecodeFp)mndTransActionDecode,
      .insertFp = (SdbInsertFp)mndTransActionInsert,
      .updateFp = (SdbUpdateFp)mndTransActionUpdate,
      .deleteFp = (SdbDeleteFp)mndTransActionDelete,
  };
S
Shengliang Guan 已提交
76

77
  mndSetMsgHandle(pMnode, TDMT_MND_TRANS_TIMER, mndProcessTransTimer);
S
Shengliang Guan 已提交
78
  mndSetMsgHandle(pMnode, TDMT_MND_KILL_TRANS, mndProcessKillTransReq);
S
Shengliang Guan 已提交
79

S
Shengliang Guan 已提交
80
  mndAddShowRetrieveHandle(pMnode, TSDB_MGMT_TABLE_TRANS, mndRetrieveTrans);
S
Shengliang Guan 已提交
81
  mndAddShowFreeIterHandle(pMnode, TSDB_MGMT_TABLE_TRANS, mndCancelGetNextTrans);
S
Shengliang Guan 已提交
82 83 84 85 86
  return sdbSetTable(pMnode->pSdb, table);
}

void mndCleanupTrans(SMnode *pMnode) {}

87 88 89
static int32_t mndTransGetActionsSize(SArray *pArray) {
  int32_t actionNum = taosArrayGetSize(pArray);
  int32_t rawDataLen = 0;
S
Shengliang Guan 已提交
90

91 92
  for (int32_t i = 0; i < actionNum; ++i) {
    STransAction *pAction = taosArrayGet(pArray, i);
93
    if (pAction->actionType == TRANS_ACTION_RAW) {
S
Shengliang Guan 已提交
94
      rawDataLen += (sizeof(STransAction) + sdbGetRawTotalSize(pAction->pRaw));
95
    } else if (pAction->actionType == TRANS_ACTION_MSG) {
96
      rawDataLen += (sizeof(STransAction) + pAction->contLen);
97 98
    } else {
      // empty
99
    }
100
    rawDataLen += sizeof(int8_t);
S
Shengliang Guan 已提交
101 102
  }

103 104
  return rawDataLen;
}
S
Shengliang Guan 已提交
105

106 107
static SSdbRaw *mndTransActionEncode(STrans *pTrans) {
  terrno = TSDB_CODE_OUT_OF_MEMORY;
S
Shengliang Guan 已提交
108

S
Shengliang Guan 已提交
109
  int32_t rawDataLen = sizeof(STrans) + TRANS_RESERVE_SIZE + pTrans->paramLen;
110 111 112
  rawDataLen += mndTransGetActionsSize(pTrans->redoActions);
  rawDataLen += mndTransGetActionsSize(pTrans->undoActions);
  rawDataLen += mndTransGetActionsSize(pTrans->commitActions);
S
Shengliang Guan 已提交
113

S
Shengliang Guan 已提交
114
  SSdbRaw *pRaw = sdbAllocRaw(SDB_TRANS, TRANS_VER_NUMBER, rawDataLen);
S
Shengliang Guan 已提交
115
  if (pRaw == NULL) {
S
Shengliang Guan 已提交
116
    mError("trans:%d, failed to alloc raw since %s", pTrans->id, terrstr());
S
Shengliang Guan 已提交
117 118 119 120
    return NULL;
  }

  int32_t dataPos = 0;
121
  SDB_SET_INT32(pRaw, dataPos, pTrans->id, _OVER)
S
Shengliang Guan 已提交
122 123 124 125
  SDB_SET_INT8(pRaw, dataPos, pTrans->stage, _OVER)
  SDB_SET_INT8(pRaw, dataPos, pTrans->policy, _OVER)
  SDB_SET_INT8(pRaw, dataPos, pTrans->conflict, _OVER)
  SDB_SET_INT8(pRaw, dataPos, pTrans->exec, _OVER)
126 127
  SDB_SET_INT8(pRaw, dataPos, pTrans->oper, _OVER)
  SDB_SET_INT8(pRaw, dataPos, 0, _OVER)
128
  SDB_SET_INT16(pRaw, dataPos, pTrans->originRpcType, _OVER)
129
  SDB_SET_INT64(pRaw, dataPos, pTrans->createdTime, _OVER)
130 131
  SDB_SET_BINARY(pRaw, dataPos, pTrans->dbname1, TSDB_TABLE_FNAME_LEN, _OVER)
  SDB_SET_BINARY(pRaw, dataPos, pTrans->dbname2, TSDB_TABLE_FNAME_LEN, _OVER)
132
  SDB_SET_INT32(pRaw, dataPos, pTrans->redoActionPos, _OVER)
133 134 135 136

  int32_t redoActionNum = taosArrayGetSize(pTrans->redoActions);
  int32_t undoActionNum = taosArrayGetSize(pTrans->undoActions);
  int32_t commitActionNum = taosArrayGetSize(pTrans->commitActions);
137 138
  SDB_SET_INT32(pRaw, dataPos, redoActionNum, _OVER)
  SDB_SET_INT32(pRaw, dataPos, undoActionNum, _OVER)
139
  SDB_SET_INT32(pRaw, dataPos, commitActionNum, _OVER)
S
Shengliang Guan 已提交
140

S
Shengliang Guan 已提交
141 142
  for (int32_t i = 0; i < redoActionNum; ++i) {
    STransAction *pAction = taosArrayGet(pTrans->redoActions, i);
143 144 145
    SDB_SET_INT32(pRaw, dataPos, pAction->id, _OVER)
    SDB_SET_INT32(pRaw, dataPos, pAction->errCode, _OVER)
    SDB_SET_INT32(pRaw, dataPos, pAction->acceptableCode, _OVER)
146
    SDB_SET_INT32(pRaw, dataPos, pAction->retryCode, _OVER)
147
    SDB_SET_INT8(pRaw, dataPos, pAction->actionType, _OVER)
148
    SDB_SET_INT8(pRaw, dataPos, pAction->stage, _OVER)
149
    SDB_SET_INT8(pRaw, dataPos, pAction->reserved, _OVER)
150
    if (pAction->actionType == TRANS_ACTION_RAW) {
151
      int32_t len = sdbGetRawTotalSize(pAction->pRaw);
152
      SDB_SET_INT8(pRaw, dataPos, pAction->rawWritten, _OVER)
153 154
      SDB_SET_INT32(pRaw, dataPos, len, _OVER)
      SDB_SET_BINARY(pRaw, dataPos, (void *)pAction->pRaw, len, _OVER)
155
    } else if (pAction->actionType == TRANS_ACTION_MSG) {
156 157
      SDB_SET_BINARY(pRaw, dataPos, (void *)&pAction->epSet, sizeof(SEpSet), _OVER)
      SDB_SET_INT16(pRaw, dataPos, pAction->msgType, _OVER)
158 159
      SDB_SET_INT8(pRaw, dataPos, pAction->msgSent, _OVER)
      SDB_SET_INT8(pRaw, dataPos, pAction->msgReceived, _OVER)
160 161
      SDB_SET_INT32(pRaw, dataPos, pAction->contLen, _OVER)
      SDB_SET_BINARY(pRaw, dataPos, pAction->pCont, pAction->contLen, _OVER)
162 163
    } else {
      // nothing
164
    }
S
Shengliang Guan 已提交
165 166 167 168
  }

  for (int32_t i = 0; i < undoActionNum; ++i) {
    STransAction *pAction = taosArrayGet(pTrans->undoActions, i);
169 170 171
    SDB_SET_INT32(pRaw, dataPos, pAction->id, _OVER)
    SDB_SET_INT32(pRaw, dataPos, pAction->errCode, _OVER)
    SDB_SET_INT32(pRaw, dataPos, pAction->acceptableCode, _OVER)
172
    SDB_SET_INT32(pRaw, dataPos, pAction->retryCode, _OVER)
173
    SDB_SET_INT8(pRaw, dataPos, pAction->actionType, _OVER)
174
    SDB_SET_INT8(pRaw, dataPos, pAction->stage, _OVER)
175
    SDB_SET_INT8(pRaw, dataPos, pAction->reserved, _OVER)
176
    if (pAction->actionType == TRANS_ACTION_RAW) {
177
      int32_t len = sdbGetRawTotalSize(pAction->pRaw);
178
      SDB_SET_INT8(pRaw, dataPos, pAction->rawWritten, _OVER)
179 180
      SDB_SET_INT32(pRaw, dataPos, len, _OVER)
      SDB_SET_BINARY(pRaw, dataPos, (void *)pAction->pRaw, len, _OVER)
181
    } else if (pAction->actionType == TRANS_ACTION_MSG) {
182 183
      SDB_SET_BINARY(pRaw, dataPos, (void *)&pAction->epSet, sizeof(SEpSet), _OVER)
      SDB_SET_INT16(pRaw, dataPos, pAction->msgType, _OVER)
184 185
      SDB_SET_INT8(pRaw, dataPos, pAction->msgSent, _OVER)
      SDB_SET_INT8(pRaw, dataPos, pAction->msgReceived, _OVER)
186 187
      SDB_SET_INT32(pRaw, dataPos, pAction->contLen, _OVER)
      SDB_SET_BINARY(pRaw, dataPos, pAction->pCont, pAction->contLen, _OVER)
188 189
    } else {
      // nothing
190 191 192 193 194
    }
  }

  for (int32_t i = 0; i < commitActionNum; ++i) {
    STransAction *pAction = taosArrayGet(pTrans->commitActions, i);
195 196 197
    SDB_SET_INT32(pRaw, dataPos, pAction->id, _OVER)
    SDB_SET_INT32(pRaw, dataPos, pAction->errCode, _OVER)
    SDB_SET_INT32(pRaw, dataPos, pAction->acceptableCode, _OVER)
198
    SDB_SET_INT32(pRaw, dataPos, pAction->retryCode, _OVER)
199
    SDB_SET_INT8(pRaw, dataPos, pAction->actionType, _OVER)
200
    SDB_SET_INT8(pRaw, dataPos, pAction->stage, _OVER)
201
    SDB_SET_INT8(pRaw, dataPos, pAction->reserved, _OVER)
202
    if (pAction->actionType == TRANS_ACTION_RAW) {
203
      int32_t len = sdbGetRawTotalSize(pAction->pRaw);
204
      SDB_SET_INT8(pRaw, dataPos, pAction->rawWritten, _OVER)
205 206
      SDB_SET_INT32(pRaw, dataPos, len, _OVER)
      SDB_SET_BINARY(pRaw, dataPos, (void *)pAction->pRaw, len, _OVER)
207
    } else if (pAction->actionType == TRANS_ACTION_MSG) {
208 209
      SDB_SET_BINARY(pRaw, dataPos, (void *)&pAction->epSet, sizeof(SEpSet), _OVER)
      SDB_SET_INT16(pRaw, dataPos, pAction->msgType, _OVER)
210 211
      SDB_SET_INT8(pRaw, dataPos, pAction->msgSent, _OVER)
      SDB_SET_INT8(pRaw, dataPos, pAction->msgReceived, _OVER)
212 213
      SDB_SET_INT32(pRaw, dataPos, pAction->contLen, _OVER)
      SDB_SET_BINARY(pRaw, dataPos, pAction->pCont, pAction->contLen, _OVER)
214 215
    } else {
      // nothing
216
    }
217 218
  }

219 220 221
  SDB_SET_INT32(pRaw, dataPos, pTrans->startFunc, _OVER)
  SDB_SET_INT32(pRaw, dataPos, pTrans->stopFunc, _OVER)
  SDB_SET_INT32(pRaw, dataPos, pTrans->paramLen, _OVER)
222
  if (pTrans->param != NULL) {
223
    SDB_SET_BINARY(pRaw, dataPos, pTrans->param, pTrans->paramLen, _OVER)
224 225
  }

226 227
  SDB_SET_RESERVE(pRaw, dataPos, TRANS_RESERVE_SIZE, _OVER)
  SDB_SET_DATALEN(pRaw, dataPos, _OVER)
228 229 230

  terrno = 0;

231
_OVER:
232
  if (terrno != 0) {
S
Shengliang Guan 已提交
233 234
    mError("trans:%d, failed to encode to raw:%p maxlen:%d len:%d since %s", pTrans->id, pRaw, sdbGetRawTotalSize(pRaw),
           dataPos, terrstr());
235 236
    sdbFreeRaw(pRaw);
    return NULL;
S
Shengliang Guan 已提交
237 238
  }

S
Shengliang Guan 已提交
239
  mTrace("trans:%d, encode to raw:%p, row:%p len:%d", pTrans->id, pRaw, pTrans, dataPos);
S
Shengliang Guan 已提交
240 241 242
  return pRaw;
}

S
Shengliang Guan 已提交
243
static SSdbRow *mndTransActionDecode(SSdbRaw *pRaw) {
244 245
  terrno = TSDB_CODE_OUT_OF_MEMORY;

246 247 248
  SSdbRow     *pRow = NULL;
  STrans      *pTrans = NULL;
  char        *pData = NULL;
249 250 251 252
  int32_t      dataLen = 0;
  int8_t       sver = 0;
  int32_t      redoActionNum = 0;
  int32_t      undoActionNum = 0;
253
  int32_t      commitActionNum = 0;
254 255 256
  int32_t      dataPos = 0;
  STransAction action = {0};

S
Shengliang Guan 已提交
257
  if (sdbGetRawSoftVer(pRaw, &sver) != 0) goto _OVER;
S
Shengliang Guan 已提交
258

S
Shengliang Guan 已提交
259
  if (sver != TRANS_VER_NUMBER) {
S
Shengliang Guan 已提交
260
    terrno = TSDB_CODE_SDB_INVALID_DATA_VER;
S
Shengliang Guan 已提交
261
    goto _OVER;
S
Shengliang Guan 已提交
262 263
  }

264
  pRow = sdbAllocRow(sizeof(STrans));
S
Shengliang Guan 已提交
265
  if (pRow == NULL) goto _OVER;
266 267

  pTrans = sdbGetRowObj(pRow);
S
Shengliang Guan 已提交
268
  if (pTrans == NULL) goto _OVER;
S
Shengliang Guan 已提交
269

S
Shengliang Guan 已提交
270
  SDB_GET_INT32(pRaw, dataPos, &pTrans->id, _OVER)
S
Shengliang Guan 已提交
271

S
Shengliang Guan 已提交
272 273 274 275
  int8_t stage = 0;
  int8_t policy = 0;
  int8_t conflict = 0;
  int8_t exec = 0;
276 277
  int8_t oper = 0;
  int8_t reserved = 0;
S
Shengliang Guan 已提交
278 279 280 281 282
  int8_t actionType = 0;
  SDB_GET_INT8(pRaw, dataPos, &stage, _OVER)
  SDB_GET_INT8(pRaw, dataPos, &policy, _OVER)
  SDB_GET_INT8(pRaw, dataPos, &conflict, _OVER)
  SDB_GET_INT8(pRaw, dataPos, &exec, _OVER)
283 284
  SDB_GET_INT8(pRaw, dataPos, &oper, _OVER)
  SDB_GET_INT8(pRaw, dataPos, &reserved, _OVER)
S
Shengliang Guan 已提交
285
  pTrans->stage = stage;
286
  pTrans->policy = policy;
287 288
  pTrans->conflict = conflict;
  pTrans->exec = exec;
289
  pTrans->oper = oper;
290
  SDB_GET_INT16(pRaw, dataPos, &pTrans->originRpcType, _OVER)
S
Shengliang Guan 已提交
291
  SDB_GET_INT64(pRaw, dataPos, &pTrans->createdTime, _OVER)
292 293
  SDB_GET_BINARY(pRaw, dataPos, pTrans->dbname1, TSDB_TABLE_FNAME_LEN, _OVER)
  SDB_GET_BINARY(pRaw, dataPos, pTrans->dbname2, TSDB_TABLE_FNAME_LEN, _OVER)
294
  SDB_GET_INT32(pRaw, dataPos, &pTrans->redoActionPos, _OVER)
S
Shengliang Guan 已提交
295 296
  SDB_GET_INT32(pRaw, dataPos, &redoActionNum, _OVER)
  SDB_GET_INT32(pRaw, dataPos, &undoActionNum, _OVER)
297
  SDB_GET_INT32(pRaw, dataPos, &commitActionNum, _OVER)
S
Shengliang Guan 已提交
298

S
Shengliang Guan 已提交
299 300
  pTrans->redoActions = taosArrayInit(redoActionNum, sizeof(STransAction));
  pTrans->undoActions = taosArrayInit(undoActionNum, sizeof(STransAction));
301
  pTrans->commitActions = taosArrayInit(commitActionNum, sizeof(STransAction));
S
Shengliang Guan 已提交
302

S
Shengliang Guan 已提交
303 304
  if (pTrans->redoActions == NULL) goto _OVER;
  if (pTrans->undoActions == NULL) goto _OVER;
305
  if (pTrans->commitActions == NULL) goto _OVER;
S
Shengliang Guan 已提交
306 307

  for (int32_t i = 0; i < redoActionNum; ++i) {
308 309 310
    SDB_GET_INT32(pRaw, dataPos, &action.id, _OVER)
    SDB_GET_INT32(pRaw, dataPos, &action.errCode, _OVER)
    SDB_GET_INT32(pRaw, dataPos, &action.acceptableCode, _OVER)
311
    SDB_GET_INT32(pRaw, dataPos, &action.retryCode, _OVER)
312 313
    SDB_GET_INT8(pRaw, dataPos, &actionType, _OVER)
    action.actionType = actionType;
S
Shengliang Guan 已提交
314 315
    SDB_GET_INT8(pRaw, dataPos, &stage, _OVER)
    action.stage = stage;
316
    SDB_GET_INT8(pRaw, dataPos, &action.reserved, _OVER)
317
    if (action.actionType == TRANS_ACTION_RAW) {
318
      SDB_GET_INT8(pRaw, dataPos, &action.rawWritten, _OVER)
319
      SDB_GET_INT32(pRaw, dataPos, &dataLen, _OVER)
320 321
      action.pRaw = taosMemoryMalloc(dataLen);
      if (action.pRaw == NULL) goto _OVER;
S
Shengliang Guan 已提交
322
      // mTrace("raw:%p, is created", pData);
323 324 325
      SDB_GET_BINARY(pRaw, dataPos, (void *)action.pRaw, dataLen, _OVER);
      if (taosArrayPush(pTrans->redoActions, &action) == NULL) goto _OVER;
      action.pRaw = NULL;
326
    } else if (action.actionType == TRANS_ACTION_MSG) {
327 328
      SDB_GET_BINARY(pRaw, dataPos, (void *)&action.epSet, sizeof(SEpSet), _OVER);
      SDB_GET_INT16(pRaw, dataPos, &action.msgType, _OVER)
329 330
      SDB_GET_INT8(pRaw, dataPos, &action.msgSent, _OVER)
      SDB_GET_INT8(pRaw, dataPos, &action.msgReceived, _OVER)
331 332 333 334 335 336
      SDB_GET_INT32(pRaw, dataPos, &action.contLen, _OVER)
      action.pCont = taosMemoryMalloc(action.contLen);
      if (action.pCont == NULL) goto _OVER;
      SDB_GET_BINARY(pRaw, dataPos, action.pCont, action.contLen, _OVER);
      if (taosArrayPush(pTrans->redoActions, &action) == NULL) goto _OVER;
      action.pCont = NULL;
337 338
    } else {
      if (taosArrayPush(pTrans->redoActions, &action) == NULL) goto _OVER;
339
    }
S
Shengliang Guan 已提交
340 341 342
  }

  for (int32_t i = 0; i < undoActionNum; ++i) {
343 344 345
    SDB_GET_INT32(pRaw, dataPos, &action.id, _OVER)
    SDB_GET_INT32(pRaw, dataPos, &action.errCode, _OVER)
    SDB_GET_INT32(pRaw, dataPos, &action.acceptableCode, _OVER)
346
    SDB_GET_INT32(pRaw, dataPos, &action.retryCode, _OVER)
347 348
    SDB_GET_INT8(pRaw, dataPos, &actionType, _OVER)
    action.actionType = actionType;
S
Shengliang Guan 已提交
349 350
    SDB_GET_INT8(pRaw, dataPos, &stage, _OVER)
    action.stage = stage;
351
    SDB_GET_INT8(pRaw, dataPos, &action.reserved, _OVER)
352
    if (action.actionType == TRANS_ACTION_RAW) {
353
      SDB_GET_INT8(pRaw, dataPos, &action.rawWritten, _OVER)
354
      SDB_GET_INT32(pRaw, dataPos, &dataLen, _OVER)
355 356
      action.pRaw = taosMemoryMalloc(dataLen);
      if (action.pRaw == NULL) goto _OVER;
S
Shengliang Guan 已提交
357
      // mTrace("raw:%p, is created", action.pRaw);
358 359 360
      SDB_GET_BINARY(pRaw, dataPos, (void *)action.pRaw, dataLen, _OVER);
      if (taosArrayPush(pTrans->undoActions, &action) == NULL) goto _OVER;
      action.pRaw = NULL;
361
    } else if (action.actionType == TRANS_ACTION_MSG) {
362 363
      SDB_GET_BINARY(pRaw, dataPos, (void *)&action.epSet, sizeof(SEpSet), _OVER);
      SDB_GET_INT16(pRaw, dataPos, &action.msgType, _OVER)
364 365
      SDB_GET_INT8(pRaw, dataPos, &action.msgSent, _OVER)
      SDB_GET_INT8(pRaw, dataPos, &action.msgReceived, _OVER)
366 367 368 369 370 371
      SDB_GET_INT32(pRaw, dataPos, &action.contLen, _OVER)
      action.pCont = taosMemoryMalloc(action.contLen);
      if (action.pCont == NULL) goto _OVER;
      SDB_GET_BINARY(pRaw, dataPos, action.pCont, action.contLen, _OVER);
      if (taosArrayPush(pTrans->undoActions, &action) == NULL) goto _OVER;
      action.pCont = NULL;
372 373
    } else {
      if (taosArrayPush(pTrans->redoActions, &action) == NULL) goto _OVER;
374 375 376 377
    }
  }

  for (int32_t i = 0; i < commitActionNum; ++i) {
378 379 380
    SDB_GET_INT32(pRaw, dataPos, &action.id, _OVER)
    SDB_GET_INT32(pRaw, dataPos, &action.errCode, _OVER)
    SDB_GET_INT32(pRaw, dataPos, &action.acceptableCode, _OVER)
381
    SDB_GET_INT32(pRaw, dataPos, &action.retryCode, _OVER)
382 383
    SDB_GET_INT8(pRaw, dataPos, &actionType, _OVER)
    action.actionType = actionType;
S
Shengliang Guan 已提交
384 385
    SDB_GET_INT8(pRaw, dataPos, &stage, _OVER)
    action.stage = stage;
386
    SDB_GET_INT8(pRaw, dataPos, &action.reserved, _OVER)
387
    if (action.actionType) {
388
      SDB_GET_INT8(pRaw, dataPos, &action.rawWritten, _OVER)
389
      SDB_GET_INT32(pRaw, dataPos, &dataLen, _OVER)
390 391
      action.pRaw = taosMemoryMalloc(dataLen);
      if (action.pRaw == NULL) goto _OVER;
S
Shengliang Guan 已提交
392
      // mTrace("raw:%p, is created", action.pRaw);
393 394 395
      SDB_GET_BINARY(pRaw, dataPos, (void *)action.pRaw, dataLen, _OVER);
      if (taosArrayPush(pTrans->commitActions, &action) == NULL) goto _OVER;
      action.pRaw = NULL;
396
    } else if (action.actionType == TRANS_ACTION_MSG) {
397 398
      SDB_GET_BINARY(pRaw, dataPos, (void *)&action.epSet, sizeof(SEpSet), _OVER);
      SDB_GET_INT16(pRaw, dataPos, &action.msgType, _OVER)
399 400
      SDB_GET_INT8(pRaw, dataPos, &action.msgSent, _OVER)
      SDB_GET_INT8(pRaw, dataPos, &action.msgReceived, _OVER)
401 402 403 404 405 406
      SDB_GET_INT32(pRaw, dataPos, &action.contLen, _OVER)
      action.pCont = taosMemoryMalloc(action.contLen);
      if (action.pCont == NULL) goto _OVER;
      SDB_GET_BINARY(pRaw, dataPos, action.pCont, action.contLen, _OVER);
      if (taosArrayPush(pTrans->commitActions, &action) == NULL) goto _OVER;
      action.pCont = NULL;
407 408
    } else {
      if (taosArrayPush(pTrans->redoActions, &action) == NULL) goto _OVER;
409
    }
S
Shengliang Guan 已提交
410 411
  }

S
Shengliang Guan 已提交
412 413 414
  SDB_GET_INT32(pRaw, dataPos, &pTrans->startFunc, _OVER)
  SDB_GET_INT32(pRaw, dataPos, &pTrans->stopFunc, _OVER)
  SDB_GET_INT32(pRaw, dataPos, &pTrans->paramLen, _OVER)
415 416
  if (pTrans->paramLen != 0) {
    pTrans->param = taosMemoryMalloc(pTrans->paramLen);
S
Shengliang Guan 已提交
417
    SDB_GET_BINARY(pRaw, dataPos, pTrans->param, pTrans->paramLen, _OVER);
418 419
  }

S
Shengliang Guan 已提交
420
  SDB_GET_RESERVE(pRaw, dataPos, TRANS_RESERVE_SIZE, _OVER)
421 422

  terrno = 0;
S
Shengliang Guan 已提交
423

S
Shengliang Guan 已提交
424
_OVER:
425 426 427
  if (terrno != 0) {
    mError("trans:%d, failed to parse from raw:%p since %s", pTrans->id, pRaw, terrstr());
    mndTransDropData(pTrans);
wafwerar's avatar
wafwerar 已提交
428 429
    taosMemoryFreeClear(pRow);
    taosMemoryFreeClear(action.pCont);
S
Shengliang Guan 已提交
430 431 432
    return NULL;
  }

S
Shengliang Guan 已提交
433
  mTrace("trans:%d, decode from raw:%p, row:%p", pTrans->id, pRaw, pTrans);
S
Shengliang Guan 已提交
434 435 436
  return pRow;
}

S
Shengliang Guan 已提交
437 438 439 440 441 442 443 444
static const char *mndTransStr(ETrnStage stage) {
  switch (stage) {
    case TRN_STAGE_PREPARE:
      return "prepare";
    case TRN_STAGE_REDO_ACTION:
      return "redoAction";
    case TRN_STAGE_ROLLBACK:
      return "rollback";
445 446 447 448 449 450
    case TRN_STAGE_UNDO_ACTION:
      return "undoAction";
    case TRN_STAGE_COMMIT:
      return "commit";
    case TRN_STAGE_COMMIT_ACTION:
      return "commitAction";
S
Shengliang Guan 已提交
451 452 453 454 455 456 457
    case TRN_STAGE_FINISHED:
      return "finished";
    default:
      return "invalid";
  }
}

458
static void mndTransTestStartFunc(SMnode *pMnode, void *param, int32_t paramLen) {
459
  mInfo("test trans start, param:%s, len:%d", (char *)param, paramLen);
460 461 462
}

static void mndTransTestStopFunc(SMnode *pMnode, void *param, int32_t paramLen) {
463
  mInfo("test trans stop, param:%s, len:%d", (char *)param, paramLen);
464 465
}

S
Shengliang Guan 已提交
466
static TransCbFp mndTransGetCbFp(ETrnFunc ftype) {
467
  switch (ftype) {
S
Shengliang Guan 已提交
468
    case TRANS_START_FUNC_TEST:
469
      return mndTransTestStartFunc;
S
Shengliang Guan 已提交
470
    case TRANS_STOP_FUNC_TEST:
471
      return mndTransTestStopFunc;
S
Shengliang Guan 已提交
472
    case TRANS_START_FUNC_MQ_REB:
L
Liu Jicong 已提交
473
      return mndRebCntInc;
474
    case TRANS_STOP_FUNC_MQ_REB:
L
Liu Jicong 已提交
475
      return mndRebCntDec;
476 477 478 479 480
    default:
      return NULL;
  }
}

S
Shengliang Guan 已提交
481
static int32_t mndTransActionInsert(SSdb *pSdb, STrans *pTrans) {
S
Shengliang Guan 已提交
482
  mTrace("trans:%d, perform insert action, row:%p stage:%s", pTrans->id, pTrans, mndTransStr(pTrans->stage));
483 484 485 486 487 488

  if (pTrans->startFunc > 0) {
    TransCbFp fp = mndTransGetCbFp(pTrans->startFunc);
    if (fp) {
      (*fp)(pSdb->pMnode, pTrans->param, pTrans->paramLen);
    }
489
    pTrans->startFunc = 0;
490 491
  }

S
Shengliang Guan 已提交
492 493 494
  return 0;
}

495
static void mndTransDropData(STrans *pTrans) {
496 497 498 499 500 501 502 503 504 505 506 507
  if (pTrans->redoActions != NULL) {
    mndTransDropActions(pTrans->redoActions);
    pTrans->redoActions = NULL;
  }
  if (pTrans->undoActions != NULL) {
    mndTransDropActions(pTrans->undoActions);
    pTrans->undoActions = NULL;
  }
  if (pTrans->commitActions != NULL) {
    mndTransDropActions(pTrans->commitActions);
    pTrans->commitActions = NULL;
  }
508 509 510 511
  if (pTrans->pRpcArray != NULL) {
    taosArrayDestroy(pTrans->pRpcArray);
    pTrans->pRpcArray = NULL;
  }
S
Shengliang Guan 已提交
512
  if (pTrans->rpcRsp != NULL) {
wafwerar's avatar
wafwerar 已提交
513
    taosMemoryFree(pTrans->rpcRsp);
S
Shengliang Guan 已提交
514 515 516
    pTrans->rpcRsp = NULL;
    pTrans->rpcRspLen = 0;
  }
517 518 519 520 521
  if (pTrans->param != NULL) {
    taosMemoryFree(pTrans->param);
    pTrans->param = NULL;
    pTrans->paramLen = 0;
  }
522
}
S
Shengliang Guan 已提交
523

524
static int32_t mndTransActionDelete(SSdb *pSdb, STrans *pTrans, bool callFunc) {
525
  mTrace("trans:%d, perform delete action, row:%p stage:%s callfunc:%d", pTrans->id, pTrans, mndTransStr(pTrans->stage),
526 527 528 529 530 531
         callFunc);
  if (pTrans->stopFunc > 0 && callFunc) {
    TransCbFp fp = mndTransGetCbFp(pTrans->stopFunc);
    if (fp) {
      (*fp)(pSdb->pMnode, pTrans->param, pTrans->paramLen);
    }
532
    pTrans->stopFunc = 0;
533 534
  }

535
  mndTransDropData(pTrans);
S
Shengliang Guan 已提交
536 537 538
  return 0;
}

539 540 541 542 543 544 545 546
static void mndTransUpdateActions(SArray *pOldArray, SArray *pNewArray) {
  for (int32_t i = 0; i < taosArrayGetSize(pOldArray); ++i) {
    STransAction *pOldAction = taosArrayGet(pOldArray, i);
    STransAction *pNewAction = taosArrayGet(pNewArray, i);
    pOldAction->rawWritten = pNewAction->rawWritten;
    pOldAction->msgSent = pNewAction->msgSent;
    pOldAction->msgReceived = pNewAction->msgReceived;
    pOldAction->errCode = pNewAction->errCode;
547
  }
548
}
549

550
static int32_t mndTransActionUpdate(SSdb *pSdb, STrans *pOld, STrans *pNew) {
S
Shengliang Guan 已提交
551 552
  mTrace("trans:%d, perform update action, old row:%p stage:%s, new row:%p stage:%s", pOld->id, pOld,
         mndTransStr(pOld->stage), pNew, mndTransStr(pNew->stage));
553 554 555
  mndTransUpdateActions(pOld->redoActions, pNew->redoActions);
  mndTransUpdateActions(pOld->undoActions, pNew->undoActions);
  mndTransUpdateActions(pOld->commitActions, pNew->commitActions);
S
Shengliang Guan 已提交
556
  pOld->stage = pNew->stage;
557
  pOld->redoActionPos = pNew->redoActionPos;
558 559 560

  if (pOld->stage == TRN_STAGE_COMMIT) {
    pOld->stage = TRN_STAGE_COMMIT_ACTION;
561
    mTrace("trans:%d, stage from commit to commitAction since perform update action", pNew->id);
562 563 564
  }

  if (pOld->stage == TRN_STAGE_ROLLBACK) {
565 566
    pOld->stage = TRN_STAGE_REDO_ACTION;
    mTrace("trans:%d, stage from rollback to undoAction since perform update action", pNew->id);
567
  }
568

S
Shengliang Guan 已提交
569 570 571
  return 0;
}

572
STrans *mndAcquireTrans(SMnode *pMnode, int32_t transId) {
573
  STrans *pTrans = sdbAcquire(pMnode->pSdb, SDB_TRANS, &transId);
S
Shengliang Guan 已提交
574 575 576 577
  if (pTrans == NULL) {
    terrno = TSDB_CODE_MND_TRANS_NOT_EXIST;
  }
  return pTrans;
S
Shengliang Guan 已提交
578 579
}

580
void mndReleaseTrans(SMnode *pMnode, STrans *pTrans) {
S
Shengliang Guan 已提交
581 582 583 584
  SSdb *pSdb = pMnode->pSdb;
  sdbRelease(pSdb, pTrans);
}

585
STrans *mndTransCreate(SMnode *pMnode, ETrnPolicy policy, ETrnConflct conflict, const SRpcMsg *pReq) {
wafwerar's avatar
wafwerar 已提交
586
  STrans *pTrans = taosMemoryCalloc(1, sizeof(STrans));
S
Shengliang Guan 已提交
587 588 589 590 591 592
  if (pTrans == NULL) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    mError("failed to create transaction since %s", terrstr());
    return NULL;
  }

S
Shengliang Guan 已提交
593
  pTrans->id = sdbGetMaxId(pMnode->pSdb, SDB_TRANS);
S
Shengliang Guan 已提交
594 595
  pTrans->stage = TRN_STAGE_PREPARE;
  pTrans->policy = policy;
596 597
  pTrans->conflict = conflict;
  pTrans->exec = TRN_EXEC_PRARLLEL;
S
Shengliang Guan 已提交
598
  pTrans->createdTime = taosGetTimestampMs();
S
Shengliang Guan 已提交
599 600
  pTrans->redoActions = taosArrayInit(TRANS_ARRAY_SIZE, sizeof(STransAction));
  pTrans->undoActions = taosArrayInit(TRANS_ARRAY_SIZE, sizeof(STransAction));
601
  pTrans->commitActions = taosArrayInit(TRANS_ARRAY_SIZE, sizeof(STransAction));
602
  pTrans->pRpcArray = taosArrayInit(1, sizeof(SRpcHandleInfo));
S
Shengliang Guan 已提交
603

604 605
  if (pTrans->redoActions == NULL || pTrans->undoActions == NULL || pTrans->commitActions == NULL ||
      pTrans->pRpcArray == NULL) {
S
Shengliang Guan 已提交
606 607 608 609 610
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    mError("failed to create transaction since %s", terrstr());
    return NULL;
  }

611 612
  if (pReq != NULL) {
    taosArrayPush(pTrans->pRpcArray, &pReq->info);
613
    pTrans->originRpcType = pReq->msgType;
614
  }
615
  mTrace("trans:%d, local object is created, data:%p", pTrans->id, pTrans);
S
Shengliang Guan 已提交
616 617 618
  return pTrans;
}

S
Shengliang Guan 已提交
619
static void mndTransDropActions(SArray *pArray) {
S
Shengliang Guan 已提交
620 621
  int32_t size = taosArrayGetSize(pArray);
  for (int32_t i = 0; i < size; ++i) {
S
Shengliang Guan 已提交
622
    STransAction *pAction = taosArrayGet(pArray, i);
623
    if (pAction->actionType == TRANS_ACTION_RAW) {
624
      taosMemoryFreeClear(pAction->pRaw);
625
    } else if (pAction->actionType == TRANS_ACTION_MSG) {
626
      taosMemoryFreeClear(pAction->pCont);
627 628
    } else {
      // nothing
629
    }
S
Shengliang Guan 已提交
630 631 632 633 634
  }

  taosArrayDestroy(pArray);
}

S
Shengliang Guan 已提交
635
void mndTransDrop(STrans *pTrans) {
S
Shengliang 已提交
636 637
  if (pTrans != NULL) {
    mndTransDropData(pTrans);
638
    mTrace("trans:%d, local object is freed, data:%p", pTrans->id, pTrans);
wafwerar's avatar
wafwerar 已提交
639
    taosMemoryFreeClear(pTrans);
S
Shengliang 已提交
640
  }
S
Shengliang Guan 已提交
641 642
}

643
static int32_t mndTransAppendAction(SArray *pArray, STransAction *pAction) {
644 645
  pAction->id = taosArrayGetSize(pArray);

646
  void *ptr = taosArrayPush(pArray, pAction);
S
Shengliang Guan 已提交
647 648 649 650 651 652 653 654
  if (ptr == NULL) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return -1;
  }

  return 0;
}

655
int32_t mndTransAppendRedolog(STrans *pTrans, SSdbRaw *pRaw) {
656 657 658 659 660 661
  STransAction action = {.stage = TRN_STAGE_REDO_ACTION, .actionType = TRANS_ACTION_RAW, .pRaw = pRaw};
  return mndTransAppendAction(pTrans->redoActions, &action);
}

int32_t mndTransAppendNullLog(STrans *pTrans) {
  STransAction action = {.stage = TRN_STAGE_REDO_ACTION, .actionType = TRANS_ACTION_NULL};
662 663
  return mndTransAppendAction(pTrans->redoActions, &action);
}
S
Shengliang Guan 已提交
664

665
int32_t mndTransAppendUndolog(STrans *pTrans, SSdbRaw *pRaw) {
666
  STransAction action = {.stage = TRN_STAGE_UNDO_ACTION, .actionType = TRANS_ACTION_RAW, .pRaw = pRaw};
667 668
  return mndTransAppendAction(pTrans->undoActions, &action);
}
S
Shengliang Guan 已提交
669

670
int32_t mndTransAppendCommitlog(STrans *pTrans, SSdbRaw *pRaw) {
671
  STransAction action = {.stage = TRN_STAGE_COMMIT_ACTION, .actionType = TRANS_ACTION_RAW, .pRaw = pRaw};
672
  return mndTransAppendAction(pTrans->commitActions, &action);
S
Shengliang Guan 已提交
673 674
}

S
Shengliang Guan 已提交
675
int32_t mndTransAppendRedoAction(STrans *pTrans, STransAction *pAction) {
676
  pAction->stage = TRN_STAGE_REDO_ACTION;
677
  pAction->actionType = TRANS_ACTION_MSG;
S
Shengliang Guan 已提交
678
  return mndTransAppendAction(pTrans->redoActions, pAction);
S
Shengliang Guan 已提交
679 680
}

S
Shengliang Guan 已提交
681
int32_t mndTransAppendUndoAction(STrans *pTrans, STransAction *pAction) {
682
  pAction->stage = TRN_STAGE_UNDO_ACTION;
683
  pAction->actionType = TRANS_ACTION_MSG;
S
Shengliang Guan 已提交
684
  return mndTransAppendAction(pTrans->undoActions, pAction);
S
Shengliang Guan 已提交
685 686
}

S
Shengliang Guan 已提交
687 688 689 690 691
void mndTransSetRpcRsp(STrans *pTrans, void *pCont, int32_t contLen) {
  pTrans->rpcRsp = pCont;
  pTrans->rpcRspLen = contLen;
}

S
Shengliang Guan 已提交
692
void mndTransSetCb(STrans *pTrans, ETrnFunc startFunc, ETrnFunc stopFunc, void *param, int32_t paramLen) {
693 694 695 696
  pTrans->startFunc = startFunc;
  pTrans->stopFunc = stopFunc;
  pTrans->param = param;
  pTrans->paramLen = paramLen;
697 698
}

699 700 701 702 703 704 705 706 707 708 709
int32_t mndSetRpcInfoForDbTrans(SMnode *pMnode, SRpcMsg *pMsg, EOperType oper, const char *dbname) {
  STrans *pTrans = NULL;
  void   *pIter = NULL;
  int32_t code = -1;

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

    if (pTrans->oper == oper) {
      if (strcasecmp(dbname, pTrans->dbname1) == 0) {
710
        mInfo("trans:%d, db:%s oper:%d matched with input", pTrans->id, dbname, oper);
711 712
        if (pTrans->pRpcArray == NULL) {
          pTrans->pRpcArray = taosArrayInit(1, sizeof(SRpcHandleInfo));
713 714 715
        }
        if (pTrans->pRpcArray != NULL && taosArrayPush(pTrans->pRpcArray, &pMsg->info) != NULL) {
          code = 0;
716
        }
717

718 719 720 721 722 723 724 725 726 727
        sdbRelease(pMnode->pSdb, pTrans);
        break;
      }
    }

    sdbRelease(pMnode->pSdb, pTrans);
  }
  return code;
}

728 729
void mndTransSetDbName(STrans *pTrans, const char *dbname1, const char *dbname2) {
  if (dbname1 != NULL) {
730
    tstrncpy(pTrans->dbname1, dbname1, TSDB_TABLE_FNAME_LEN);
731 732
  }
  if (dbname2 != NULL) {
733
    tstrncpy(pTrans->dbname2, dbname2, TSDB_TABLE_FNAME_LEN);
734 735
  }
}
S
Shengliang Guan 已提交
736

737
void mndTransSetSerial(STrans *pTrans) { pTrans->exec = TRN_EXEC_SERIAL; }
738

739 740
void mndTransSetOper(STrans *pTrans, EOperType oper) { pTrans->oper = oper; }

S
Shengliang Guan 已提交
741
static int32_t mndTransSync(SMnode *pMnode, STrans *pTrans) {
S
Shengliang Guan 已提交
742
  SSdbRaw *pRaw = mndTransActionEncode(pTrans);
S
Shengliang Guan 已提交
743
  if (pRaw == NULL) {
S
Shengliang Guan 已提交
744
    mError("trans:%d, failed to encode while sync trans since %s", pTrans->id, terrstr());
S
Shengliang Guan 已提交
745 746
    return -1;
  }
S
Shengliang Guan 已提交
747
  sdbSetRawStatus(pRaw, SDB_STATUS_READY);
S
Shengliang Guan 已提交
748

749
  mInfo("trans:%d, sync to other mnodes, stage:%s", pTrans->id, mndTransStr(pTrans->stage));
S
Shengliang Guan 已提交
750
  int32_t code = mndSyncPropose(pMnode, pRaw, pTrans->id);
S
Shengliang Guan 已提交
751 752 753
  if (code != 0) {
    mError("trans:%d, failed to sync since %s", pTrans->id, terrstr());
    sdbFreeRaw(pRaw);
S
Shengliang Guan 已提交
754 755 756
    return -1;
  }

757
  sdbFreeRaw(pRaw);
758
  mInfo("trans:%d, sync finished", pTrans->id);
S
Shengliang Guan 已提交
759 760 761
  return 0;
}

S
Shengliang Guan 已提交
762 763
static bool mndCheckDbConflict(const char *db, STrans *pTrans) {
  if (db[0] == 0) return false;
764
  if (strcasecmp(db, pTrans->dbname1) == 0 || strcasecmp(db, pTrans->dbname2) == 0) return true;
S
Shengliang Guan 已提交
765 766 767
  return false;
}

768
static bool mndCheckTransConflict(SMnode *pMnode, STrans *pNew) {
S
Shengliang Guan 已提交
769
  STrans *pTrans = NULL;
770
  void   *pIter = NULL;
771
  bool    conflict = false;
772

773
  if (pNew->conflict == TRN_CONFLICT_NOTHING) return conflict;
S
Shengliang Guan 已提交
774 775 776 777 778

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

779 780 781
    if (pNew->conflict == TRN_CONFLICT_GLOBAL) conflict = true;
    if (pNew->conflict == TRN_CONFLICT_DB) {
      if (pTrans->conflict == TRN_CONFLICT_GLOBAL) conflict = true;
782
      if (pTrans->conflict == TRN_CONFLICT_DB || pTrans->conflict == TRN_CONFLICT_DB_INSIDE) {
S
Shengliang Guan 已提交
783 784
        if (mndCheckDbConflict(pNew->dbname1, pTrans)) conflict = true;
        if (mndCheckDbConflict(pNew->dbname2, pTrans)) conflict = true;
785
      }
786 787 788
    }
    if (pNew->conflict == TRN_CONFLICT_DB_INSIDE) {
      if (pTrans->conflict == TRN_CONFLICT_GLOBAL) conflict = true;
789
      if (pTrans->conflict == TRN_CONFLICT_DB) {
S
Shengliang Guan 已提交
790 791
        if (mndCheckDbConflict(pNew->dbname1, pTrans)) conflict = true;
        if (mndCheckDbConflict(pNew->dbname2, pTrans)) conflict = true;
792
      }
793 794 795
      if (pTrans->conflict == TRN_CONFLICT_DB_INSIDE) {
        if (mndCheckDbConflict(pNew->dbname2, pTrans)) conflict = true; // for stb
      }
S
Shengliang Guan 已提交
796
    }
S
Shengliang Guan 已提交
797

798 799 800 801 802 803 804 805 806
    if (conflict) {
      mError("trans:%d, db1:%s db2:%s type:%d, can't execute since conflict with trans:%d db1:%s db2:%s type:%d",
             pNew->id, pNew->dbname1, pNew->dbname2, pNew->conflict, pTrans->id, pTrans->dbname1, pTrans->dbname2,
             pTrans->conflict);
    } else {
      mDebug("trans:%d, db1:%s db2:%s type:%d, not conflict with trans:%d db1:%s db2:%s type:%d", pNew->id,
             pNew->dbname1, pNew->dbname2, pNew->conflict, pTrans->id, pTrans->dbname1, pTrans->dbname2,
             pTrans->conflict);
    }
S
Shengliang Guan 已提交
807 808 809
    sdbRelease(pMnode->pSdb, pTrans);
  }

810
  return conflict;
S
Shengliang Guan 已提交
811 812
}

S
Shengliang Guan 已提交
813
int32_t mndTransPrepare(SMnode *pMnode, STrans *pTrans) {
814
  if (pTrans->conflict == TRN_CONFLICT_DB || pTrans->conflict == TRN_CONFLICT_DB_INSIDE) {
815
    if (strlen(pTrans->dbname1) == 0 && strlen(pTrans->dbname2) == 0) {
816 817 818 819 820 821
      terrno = TSDB_CODE_MND_TRANS_CONFLICT;
      mError("trans:%d, failed to prepare conflict db not set", pTrans->id);
      return -1;
    }
  }

822 823
  if (mndCheckTransConflict(pMnode, pTrans)) {
    terrno = TSDB_CODE_MND_TRANS_CONFLICT;
S
Shengliang Guan 已提交
824 825 826 827
    mError("trans:%d, failed to prepare since %s", pTrans->id, terrstr());
    return -1;
  }

828
  if (taosArrayGetSize(pTrans->commitActions) <= 0) {
829 830 831 832 833
    terrno = TSDB_CODE_MND_TRANS_CLOG_IS_NULL;
    mError("trans:%d, failed to prepare since %s", pTrans->id, terrstr());
    return -1;
  }

834
  mInfo("trans:%d, prepare transaction", pTrans->id);
S
Shengliang Guan 已提交
835 836 837 838
  if (mndTransSync(pMnode, pTrans) != 0) {
    mError("trans:%d, failed to prepare since %s", pTrans->id, terrstr());
    return -1;
  }
839
  mInfo("trans:%d, prepare finished", pTrans->id);
S
Shengliang Guan 已提交
840

S
Shengliang Guan 已提交
841 842
  STrans *pNew = mndAcquireTrans(pMnode, pTrans->id);
  if (pNew == NULL) {
S
Shengliang Guan 已提交
843
    mError("trans:%d, failed to read from sdb since %s", pTrans->id, terrstr());
S
Shengliang Guan 已提交
844 845 846
    return -1;
  }

847
  pNew->pRpcArray = pTrans->pRpcArray;
S
Shengliang Guan 已提交
848 849
  pNew->rpcRsp = pTrans->rpcRsp;
  pNew->rpcRspLen = pTrans->rpcRspLen;
850
  pTrans->pRpcArray = NULL;
S
Shengliang Guan 已提交
851 852 853
  pTrans->rpcRsp = NULL;
  pTrans->rpcRspLen = 0;

S
Shengliang Guan 已提交
854 855
  mndTransExecute(pMnode, pNew);
  mndReleaseTrans(pMnode, pNew);
S
Shengliang Guan 已提交
856 857 858
  return 0;
}

S
Shengliang Guan 已提交
859
static int32_t mndTransCommit(SMnode *pMnode, STrans *pTrans) {
860
  mInfo("trans:%d, commit transaction", pTrans->id);
S
Shengliang Guan 已提交
861 862
  if (mndTransSync(pMnode, pTrans) != 0) {
    mError("trans:%d, failed to commit since %s", pTrans->id, terrstr());
S
Shengliang Guan 已提交
863
    return -1;
S
Shengliang Guan 已提交
864
  }
865
  mInfo("trans:%d, commit finished", pTrans->id);
S
Shengliang Guan 已提交
866 867 868
  return 0;
}

S
Shengliang Guan 已提交
869
static int32_t mndTransRollback(SMnode *pMnode, STrans *pTrans) {
870
  mInfo("trans:%d, rollback transaction", pTrans->id);
S
Shengliang Guan 已提交
871 872
  if (mndTransSync(pMnode, pTrans) != 0) {
    mError("trans:%d, failed to rollback since %s", pTrans->id, terrstr());
S
Shengliang Guan 已提交
873
    return -1;
S
Shengliang Guan 已提交
874
  }
875
  mInfo("trans:%d, rollback finished", pTrans->id);
S
Shengliang Guan 已提交
876
  return 0;
S
Shengliang Guan 已提交
877
}
S
Shengliang Guan 已提交
878

879
static void mndTransSendRpcRsp(SMnode *pMnode, STrans *pTrans) {
880 881
  bool    sendRsp = false;
  int32_t code = pTrans->code;
882 883 884 885 886 887

  if (pTrans->stage == TRN_STAGE_FINISHED) {
    sendRsp = true;
  }

  if (pTrans->policy == TRN_POLICY_ROLLBACK) {
888
    if (pTrans->stage == TRN_STAGE_UNDO_ACTION || pTrans->stage == TRN_STAGE_ROLLBACK) {
889
      if (code == 0) code = TSDB_CODE_MND_TRANS_UNKNOW_ERROR;
890
      sendRsp = true;
891
    }
892
  } else {
893
    if (pTrans->stage == TRN_STAGE_REDO_ACTION && pTrans->failedTimes > 6) {
894
      if (code == 0) code = TSDB_CODE_MND_TRANS_UNKNOW_ERROR;
895
      sendRsp = true;
896
    }
S
Shengliang Guan 已提交
897
  }
898

899 900 901 902 903 904 905 906
  if (!sendRsp) return;

  int32_t size = taosArrayGetSize(pTrans->pRpcArray);
  if (size <= 0) return;

  for (int32_t i = 0; i < size; ++i) {
    SRpcHandleInfo *pInfo = taosArrayGet(pTrans->pRpcArray, i);
    if (pInfo->handle != NULL) {
907
      mInfo("trans:%d, send rsp, code:0x%x stage:%s app:%p", pTrans->id, code, mndTransStr(pTrans->stage),
908 909 910 911 912 913
             pInfo->ahandle);
      if (code == TSDB_CODE_RPC_NETWORK_UNAVAIL) {
        code = TSDB_CODE_MND_TRANS_NETWORK_UNAVAILL;
      }
      SRpcMsg rspMsg = {.code = code, .info = *pInfo};

914
      if (pTrans->originRpcType == TDMT_MND_CREATE_DB) {
915
        mInfo("trans:%d, origin msgtype:%s", pTrans->id, TMSG_INFO(pTrans->originRpcType));
916 917 918 919 920
        SDbObj *pDb = mndAcquireDb(pMnode, pTrans->dbname1);
        if (pDb != NULL) {
          for (int32_t j = 0; j < 12; j++) {
            bool ready = mndIsDbReady(pMnode, pDb);
            if (!ready) {
921
              mInfo("trans:%d, db:%s not ready yet, wait %d times", pTrans->id, pTrans->dbname1, j);
922 923 924 925 926 927 928
              taosMsleep(1000);
            } else {
              break;
            }
          }
        }
        mndReleaseDb(pMnode, pDb);
929 930 931 932 933 934 935 936 937 938 939 940 941 942 943
      } else if (pTrans->originRpcType == TDMT_MND_CREATE_STB) {
        void   *pCont = NULL;
        int32_t contLen = 0;
        if (0 == mndBuildSMCreateStbRsp(pMnode, pTrans->dbname1, pTrans->dbname2, &pCont, &contLen) != 0) {
          mndTransSetRpcRsp(pTrans, pCont, contLen);
        }
      }

      if (pTrans->rpcRspLen != 0) {
        void *rpcCont = rpcMallocCont(pTrans->rpcRspLen);
        if (rpcCont != NULL) {
          memcpy(rpcCont, pTrans->rpcRsp, pTrans->rpcRspLen);
          rspMsg.pCont = rpcCont;
          rspMsg.contLen = pTrans->rpcRspLen;
        }
944 945
      }

946 947
      tmsgSendRsp(&rspMsg);
    }
948
  }
949
  taosArrayClear(pTrans->pRpcArray);
S
Shengliang Guan 已提交
950 951
}

952
int32_t mndTransProcessRsp(SRpcMsg *pRsp) {
S
Shengliang Guan 已提交
953 954
  SMnode *pMnode = pRsp->info.node;
  int64_t signature = (int64_t)(pRsp->info.ahandle);
S
Shengliang Guan 已提交
955 956
  int32_t transId = (int32_t)(signature >> 32);
  int32_t action = (int32_t)((signature << 32) >> 32);
957 958 959 960

  STrans *pTrans = mndAcquireTrans(pMnode, transId);
  if (pTrans == NULL) {
    mError("trans:%d, failed to get transId from vnode rsp since %s", transId, terrstr());
S
Shengliang Guan 已提交
961
    goto _OVER;
962 963 964
  }

  SArray *pArray = NULL;
S
Shengliang Guan 已提交
965
  if (pTrans->stage == TRN_STAGE_REDO_ACTION) {
966
    pArray = pTrans->redoActions;
S
Shengliang Guan 已提交
967
  } else if (pTrans->stage == TRN_STAGE_UNDO_ACTION) {
968 969
    pArray = pTrans->undoActions;
  } else {
S
Shengliang Guan 已提交
970
    mError("trans:%d, invalid trans stage:%d while recv action rsp", pTrans->id, pTrans->stage);
S
Shengliang Guan 已提交
971
    goto _OVER;
972 973 974
  }

  if (pArray == NULL) {
S
Shengliang Guan 已提交
975
    mError("trans:%d, invalid trans stage:%d", transId, pTrans->stage);
S
Shengliang Guan 已提交
976
    goto _OVER;
977 978 979
  }

  int32_t actionNum = taosArrayGetSize(pTrans->redoActions);
S
Shengliang Guan 已提交
980
  if (action < 0 || action >= actionNum) {
981
    mError("trans:%d, invalid action:%d", transId, action);
S
Shengliang Guan 已提交
982
    goto _OVER;
983 984 985 986 987
  }

  STransAction *pAction = taosArrayGet(pArray, action);
  if (pAction != NULL) {
    pAction->msgReceived = 1;
S
Shengliang Guan 已提交
988
    pAction->errCode = pRsp->code;
989 990
  }

991
  mInfo("trans:%d, %s:%d response is received, code:0x%x, accept:0x%x retry:0x%x", transId,
992
         mndTransStr(pAction->stage), action, pRsp->code, pAction->acceptableCode, pAction->retryCode);
993 994
  mndTransExecute(pMnode, pTrans);

S
Shengliang Guan 已提交
995
_OVER:
996
  mndReleaseTrans(pMnode, pTrans);
997
  return 0;
998 999
}

1000 1001 1002 1003 1004 1005 1006
static void mndTransResetAction(SMnode *pMnode, STrans *pTrans, STransAction *pAction) {
  pAction->rawWritten = 0;
  pAction->msgSent = 0;
  pAction->msgReceived = 0;
  if (pAction->errCode == TSDB_CODE_RPC_REDIRECT || pAction->errCode == TSDB_CODE_SYN_NEW_CONFIG_ERROR ||
      pAction->errCode == TSDB_CODE_SYN_INTERNAL_ERROR || pAction->errCode == TSDB_CODE_SYN_NOT_LEADER) {
    pAction->epSet.inUse = (pAction->epSet.inUse + 1) % pAction->epSet.numOfEps;
1007
    mInfo("trans:%d, %s:%d execute status is reset and set epset inuse:%d", pTrans->id, mndTransStr(pAction->stage),
1008 1009
           pAction->id, pAction->epSet.inUse);
  } else {
1010
    mInfo("trans:%d, %s:%d execute status is reset", pTrans->id, mndTransStr(pAction->stage), pAction->id);
1011 1012 1013 1014
  }
  pAction->errCode = 0;
}

S
Shengliang Guan 已提交
1015 1016 1017 1018 1019
static void mndTransResetActions(SMnode *pMnode, STrans *pTrans, SArray *pArray) {
  int32_t numOfActions = taosArrayGetSize(pArray);

  for (int32_t action = 0; action < numOfActions; ++action) {
    STransAction *pAction = taosArrayGet(pArray, action);
1020 1021 1022 1023
    if (pAction->msgSent && pAction->msgReceived &&
        (pAction->errCode == 0 || pAction->errCode == pAction->acceptableCode))
      continue;
    if (pAction->rawWritten && (pAction->errCode == 0 || pAction->errCode == pAction->acceptableCode)) continue;
S
Shengliang Guan 已提交
1024

1025
    mndTransResetAction(pMnode, pTrans, pAction);
S
Shengliang Guan 已提交
1026 1027 1028
  }
}

1029
static int32_t mndTransWriteSingleLog(SMnode *pMnode, STrans *pTrans, STransAction *pAction) {
1030 1031
  if (pAction->rawWritten) return 0;

1032
  int32_t code = sdbWriteWithoutFree(pMnode->pSdb, pAction->pRaw);
1033 1034 1035 1036
  if (code == 0 || terrno == TSDB_CODE_SDB_OBJ_NOT_THERE) {
    pAction->rawWritten = true;
    pAction->errCode = 0;
    code = 0;
1037
    mInfo("trans:%d, %s:%d write to sdb, type:%s status:%s", pTrans->id, mndTransStr(pAction->stage), pAction->id,
S
Shengliang Guan 已提交
1038
           sdbTableName(pAction->pRaw->type), sdbStatusName(pAction->pRaw->status));
1039 1040 1041 1042 1043

    pTrans->lastAction = pAction->id;
    pTrans->lastMsgType = pAction->msgType;
    pTrans->lastEpset = pAction->epSet;
    pTrans->lastErrorNo = 0;
1044
  } else {
1045
    pAction->errCode = (terrno != 0) ? terrno : code;
S
Shengliang Guan 已提交
1046 1047
    mError("trans:%d, %s:%d failed to write sdb since %s, type:%s status:%s", pTrans->id, mndTransStr(pAction->stage),
           pAction->id, terrstr(), sdbTableName(pAction->pRaw->type), sdbStatusName(pAction->pRaw->status));
1048 1049 1050 1051
    pTrans->lastAction = pAction->id;
    pTrans->lastMsgType = pAction->msgType;
    pTrans->lastEpset = pAction->epSet;
    pTrans->lastErrorNo = pAction->errCode;
1052 1053 1054 1055 1056 1057 1058
  }

  return code;
}

static int32_t mndTransSendSingleMsg(SMnode *pMnode, STrans *pTrans, STransAction *pAction) {
  if (pAction->msgSent) return 0;
1059
  if (mndCannotExecuteTransAction(pMnode)) return -1;
1060 1061 1062 1063 1064 1065 1066 1067 1068 1069 1070 1071 1072

  int64_t signature = pTrans->id;
  signature = (signature << 32);
  signature += pAction->id;

  SRpcMsg rpcMsg = {.msgType = pAction->msgType, .contLen = pAction->contLen, .info.ahandle = (void *)signature};
  rpcMsg.pCont = rpcMallocCont(pAction->contLen);
  if (rpcMsg.pCont == NULL) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return -1;
  }
  memcpy(rpcMsg.pCont, pAction->pCont, pAction->contLen);

S
Shengliang Guan 已提交
1073 1074 1075
  char    detail[1024] = {0};
  int32_t len = snprintf(detail, sizeof(detail), "msgType:%s numOfEps:%d inUse:%d", TMSG_INFO(pAction->msgType),
                         pAction->epSet.numOfEps, pAction->epSet.inUse);
S
Shengliang Guan 已提交
1076
  for (int32_t i = 0; i < pAction->epSet.numOfEps; ++i) {
S
Shengliang Guan 已提交
1077 1078 1079 1080
    len += snprintf(detail + len, sizeof(detail) - len, " ep:%d-%s:%u", i, pAction->epSet.eps[i].fqdn,
                    pAction->epSet.eps[i].port);
  }

1081 1082 1083 1084 1085
  int32_t code = tmsgSendReq(&pAction->epSet, &rpcMsg);
  if (code == 0) {
    pAction->msgSent = 1;
    pAction->msgReceived = 0;
    pAction->errCode = 0;
1086
    mInfo("trans:%d, %s:%d is sent, %s", pTrans->id, mndTransStr(pAction->stage), pAction->id, detail);
1087 1088 1089 1090 1091 1092 1093

    pTrans->lastAction = pAction->id;
    pTrans->lastMsgType = pAction->msgType;
    pTrans->lastEpset = pAction->epSet;
    if (pTrans->lastErrorNo == 0) {
      pTrans->lastErrorNo = TSDB_CODE_ACTION_IN_PROGRESS;
    }
1094 1095 1096 1097
  } else {
    pAction->msgSent = 0;
    pAction->msgReceived = 0;
    pAction->errCode = (terrno != 0) ? terrno : code;
S
Shengliang Guan 已提交
1098 1099
    mError("trans:%d, %s:%d not send since %s, %s", pTrans->id, mndTransStr(pAction->stage), pAction->id, terrstr(),
           detail);
1100 1101 1102 1103 1104

    pTrans->lastAction = pAction->id;
    pTrans->lastMsgType = pAction->msgType;
    pTrans->lastEpset = pAction->epSet;
    pTrans->lastErrorNo = pAction->errCode;
1105 1106 1107 1108 1109
  }

  return code;
}

1110 1111 1112
static int32_t mndTransExecNullMsg(SMnode *pMnode, STrans *pTrans, STransAction *pAction) {
  pAction->rawWritten = 0;
  pAction->errCode = 0;
1113
  mInfo("trans:%d, %s:%d confirm action executed", pTrans->id, mndTransStr(pAction->stage), pAction->id);
1114 1115 1116 1117

  pTrans->lastAction = pAction->id;
  pTrans->lastMsgType = pAction->msgType;
  pTrans->lastEpset = pAction->epSet;
wafwerar's avatar
wafwerar 已提交
1118
  pTrans->lastErrorNo = 0;
1119 1120 1121
  return 0;
}

1122
static int32_t mndTransExecSingleAction(SMnode *pMnode, STrans *pTrans, STransAction *pAction) {
1123
  if (pAction->actionType == TRANS_ACTION_RAW) {
1124
    return mndTransWriteSingleLog(pMnode, pTrans, pAction);
1125
  } else if (pAction->actionType == TRANS_ACTION_MSG) {
1126
    return mndTransSendSingleMsg(pMnode, pTrans, pAction);
1127
  } else {
1128
    return mndTransExecNullMsg(pMnode, pTrans, pAction);
1129 1130 1131
  }
}

1132
static int32_t mndTransExecSingleActions(SMnode *pMnode, STrans *pTrans, SArray *pArray) {
1133
  int32_t numOfActions = taosArrayGetSize(pArray);
1134
  int32_t code = 0;
1135 1136 1137

  for (int32_t action = 0; action < numOfActions; ++action) {
    STransAction *pAction = taosArrayGet(pArray, action);
1138
    code = mndTransExecSingleAction(pMnode, pTrans, pAction);
1139
    if (code != 0) break;
S
Shengliang Guan 已提交
1140 1141
  }

1142
  return code;
S
Shengliang Guan 已提交
1143 1144 1145 1146 1147 1148
}

static int32_t mndTransExecuteActions(SMnode *pMnode, STrans *pTrans, SArray *pArray) {
  int32_t numOfActions = taosArrayGetSize(pArray);
  if (numOfActions == 0) return 0;

1149
  if (mndTransExecSingleActions(pMnode, pTrans, pArray) != 0) {
S
Shengliang Guan 已提交
1150 1151 1152
    return -1;
  }

1153 1154 1155
  int32_t       numOfExecuted = 0;
  int32_t       errCode = 0;
  STransAction *pErrAction = NULL;
1156 1157
  for (int32_t action = 0; action < numOfActions; ++action) {
    STransAction *pAction = taosArrayGet(pArray, action);
1158
    if (pAction->msgReceived || pAction->rawWritten) {
1159
      numOfExecuted++;
1160
      if (pAction->errCode != 0 && pAction->errCode != pAction->acceptableCode) {
S
Shengliang Guan 已提交
1161
        errCode = pAction->errCode;
1162
        pErrAction = pAction;
1163 1164 1165 1166
      }
    }
  }

1167
  if (numOfExecuted == numOfActions) {
S
Shengliang Guan 已提交
1168
    if (errCode == 0) {
1169 1170 1171
      pTrans->lastAction = 0;
      pTrans->lastMsgType = 0;
      memset(&pTrans->lastEpset, 0, sizeof(pTrans->lastEpset));
S
Shengliang Guan 已提交
1172
      pTrans->lastErrorNo = 0;
1173
      mInfo("trans:%d, all %d actions execute successfully", pTrans->id, numOfActions);
S
Shengliang Guan 已提交
1174 1175
      return 0;
    } else {
S
Shengliang Guan 已提交
1176
      mError("trans:%d, all %d actions executed, code:0x%x", pTrans->id, numOfActions, errCode & 0XFFFF);
1177
      if (pErrAction != NULL) {
1178
        pTrans->lastAction = pErrAction->id;
S
Shengliang Guan 已提交
1179
        pTrans->lastMsgType = pErrAction->msgType;
1180
        pTrans->lastEpset = pErrAction->epSet;
S
Shengliang Guan 已提交
1181
        pTrans->lastErrorNo = pErrAction->errCode;
1182
      }
S
Shengliang Guan 已提交
1183 1184 1185 1186
      mndTransResetActions(pMnode, pTrans, pArray);
      terrno = errCode;
      return errCode;
    }
1187
  } else {
1188
    mInfo("trans:%d, %d of %d actions executed", pTrans->id, numOfExecuted, numOfActions);
S
Shengliang Guan 已提交
1189
    return TSDB_CODE_ACTION_IN_PROGRESS;
1190
  }
S
Shengliang Guan 已提交
1191 1192
}

S
Shengliang Guan 已提交
1193
static int32_t mndTransExecuteRedoActions(SMnode *pMnode, STrans *pTrans) {
1194
  int32_t code = mndTransExecuteActions(pMnode, pTrans, pTrans->redoActions);
S
Shengliang Guan 已提交
1195
  if (code != 0 && code != TSDB_CODE_ACTION_IN_PROGRESS) {
1196
    mError("failed to execute redoActions since:%s, code:0x%x", terrstr(), terrno);
1197 1198
  }
  return code;
S
Shengliang Guan 已提交
1199
}
S
Shengliang Guan 已提交
1200

S
Shengliang Guan 已提交
1201
static int32_t mndTransExecuteUndoActions(SMnode *pMnode, STrans *pTrans) {
1202
  int32_t code = mndTransExecuteActions(pMnode, pTrans, pTrans->undoActions);
S
Shengliang Guan 已提交
1203
  if (code != 0 && code != TSDB_CODE_ACTION_IN_PROGRESS) {
1204 1205 1206
    mError("failed to execute undoActions since %s", terrstr());
  }
  return code;
S
Shengliang Guan 已提交
1207
}
S
Shengliang Guan 已提交
1208

1209
static int32_t mndTransExecuteCommitActions(SMnode *pMnode, STrans *pTrans) {
1210
  int32_t code = mndTransExecuteActions(pMnode, pTrans, pTrans->commitActions);
1211
  if (code != 0 && code != TSDB_CODE_ACTION_IN_PROGRESS) {
1212 1213 1214 1215 1216
    mError("failed to execute commitActions since %s", terrstr());
  }
  return code;
}

1217
static int32_t mndTransExecuteRedoActionsSerial(SMnode *pMnode, STrans *pTrans) {
1218 1219 1220 1221
  int32_t code = 0;
  int32_t numOfActions = taosArrayGetSize(pTrans->redoActions);
  if (numOfActions == 0) return code;
  if (pTrans->redoActionPos >= numOfActions) return code;
S
Shengliang Guan 已提交
1222

1223 1224
  for (int32_t action = pTrans->redoActionPos; action < numOfActions; ++action) {
    STransAction *pAction = taosArrayGet(pTrans->redoActions, pTrans->redoActionPos);
S
Shengliang Guan 已提交
1225

1226 1227 1228 1229 1230 1231
    code = mndTransExecSingleAction(pMnode, pTrans, pAction);
    if (code == 0) {
      if (pAction->msgSent) {
        if (pAction->msgReceived) {
          if (pAction->errCode != 0 && pAction->errCode != pAction->acceptableCode) {
            code = pAction->errCode;
1232 1233
            mndTransResetAction(pMnode, pTrans, pAction);
          } else {
1234
            mInfo("trans:%d, %s:%d execute successfully", pTrans->id, mndTransStr(pAction->stage), action);
1235 1236 1237 1238
          }
        } else {
          code = TSDB_CODE_ACTION_IN_PROGRESS;
        }
S
Shengliang Guan 已提交
1239
      } else if (pAction->rawWritten) {
1240 1241
        if (pAction->errCode != 0 && pAction->errCode != pAction->acceptableCode) {
          code = pAction->errCode;
1242
        } else {
1243
          mInfo("trans:%d, %s:%d write successfully", pTrans->id, mndTransStr(pAction->stage), action);
1244
        }
S
Shengliang Guan 已提交
1245
      } else {
1246 1247
      }
    }
1248

1249
    if (code == 0) {
1250
      pTrans->failedTimes = 0;
S
Shengliang Guan 已提交
1251
      pTrans->lastAction = action;
1252
      pTrans->lastMsgType = 0;
S
Shengliang Guan 已提交
1253
      pTrans->lastErrorNo = 0;
1254
      memset(&pTrans->lastEpset, 0, sizeof(pTrans->lastEpset));
1255
    } else {
1256
      pTrans->lastAction = action;
S
Shengliang Guan 已提交
1257
      pTrans->lastMsgType = pAction->msgType;
1258 1259
      pTrans->lastErrorNo = code;
      pTrans->lastEpset = pAction->epSet;
1260 1261
    }

1262
    if (mndCannotExecuteTransAction(pMnode)) break;
1263

1264
    if (code == 0) {
1265
      pTrans->code = 0;
1266
      pTrans->redoActionPos++;
1267
      mInfo("trans:%d, %s:%d is executed and need sync to other mnodes", pTrans->id, mndTransStr(pAction->stage),
1268 1269 1270
             pAction->id);
      code = mndTransSync(pMnode, pTrans);
      if (code != 0) {
1271 1272 1273
        pTrans->code = terrno;
        mError("trans:%d, %s:%d is executed and failed to sync to other mnodes since %s", pTrans->id,
               mndTransStr(pAction->stage), pAction->id, terrstr());
1274 1275
      }
    } else if (code == TSDB_CODE_ACTION_IN_PROGRESS) {
1276
      mInfo("trans:%d, %s:%d is in progress and wait it finish", pTrans->id, mndTransStr(pAction->stage), pAction->id);
1277
      break;
1278
    } else if (code == pAction->retryCode) {
1279
      mInfo("trans:%d, %s:%d receive code:0x%x and retry", pTrans->id, mndTransStr(pAction->stage), pAction->id, code);
1280
      taosMsleep(300);
1281
      action--;
1282
      continue;
1283
    } else {
1284 1285
      terrno = code;
      pTrans->code = code;
1286
      mInfo("trans:%d, %s:%d receive code:0x%x and wait another schedule, failedTimes:%d", pTrans->id,
1287
             mndTransStr(pAction->stage), pAction->id, code, pTrans->failedTimes);
S
Shengliang Guan 已提交
1288
      break;
1289
    }
S
Shengliang Guan 已提交
1290
  }
S
Shengliang Guan 已提交
1291

1292 1293 1294 1295 1296 1297
  return code;
}

static bool mndTransPerformPrepareStage(SMnode *pMnode, STrans *pTrans) {
  bool continueExec = true;
  pTrans->stage = TRN_STAGE_REDO_ACTION;
1298
  mInfo("trans:%d, stage from prepare to redoAction", pTrans->id);
S
Shengliang Guan 已提交
1299
  return continueExec;
S
Shengliang Guan 已提交
1300 1301
}

S
Shengliang Guan 已提交
1302 1303
static bool mndTransPerformRedoActionStage(SMnode *pMnode, STrans *pTrans) {
  bool    continueExec = true;
1304 1305
  int32_t code = 0;

1306
  if (pTrans->exec == TRN_EXEC_SERIAL) {
1307
    code = mndTransExecuteRedoActionsSerial(pMnode, pTrans);
1308 1309 1310
  } else {
    code = mndTransExecuteRedoActions(pMnode, pTrans);
  }
S
Shengliang Guan 已提交
1311

1312
  if (mndCannotExecuteTransAction(pMnode)) return false;
1313

S
Shengliang Guan 已提交
1314
  if (code == 0) {
S
Shengliang Guan 已提交
1315
    pTrans->code = 0;
S
Shengliang Guan 已提交
1316
    pTrans->stage = TRN_STAGE_COMMIT;
1317
    mInfo("trans:%d, stage from redoAction to commit", pTrans->id);
S
Shengliang Guan 已提交
1318
    continueExec = true;
S
Shengliang Guan 已提交
1319
  } else if (code == TSDB_CODE_ACTION_IN_PROGRESS) {
1320
    mInfo("trans:%d, stage keep on redoAction since %s", pTrans->id, tstrerror(code));
S
Shengliang Guan 已提交
1321
    continueExec = false;
S
Shengliang Guan 已提交
1322
  } else {
S
Shengliang Guan 已提交
1323
    pTrans->failedTimes++;
S
Shengliang Guan 已提交
1324
    pTrans->code = terrno;
S
Shengliang Guan 已提交
1325
    if (pTrans->policy == TRN_POLICY_ROLLBACK) {
1326 1327
      if (pTrans->lastAction != 0) {
        STransAction *pAction = taosArrayGet(pTrans->redoActions, pTrans->lastAction);
1328
        if (pAction->retryCode != 0 && pAction->retryCode == pAction->errCode) {
1329 1330 1331 1332 1333 1334 1335 1336 1337 1338
          if (pTrans->failedTimes < 6) {
            mError("trans:%d, stage keep on redoAction since action:%d code:0x%x not 0x%x, failedTimes:%d", pTrans->id,
                   pTrans->lastAction, pTrans->code, pAction->retryCode, pTrans->failedTimes);
            taosMsleep(1000);
            continueExec = true;
            return true;
          }
        }
      }

1339 1340
      pTrans->stage = TRN_STAGE_ROLLBACK;
      mError("trans:%d, stage from redoAction to rollback since %s", pTrans->id, terrstr());
S
Shengliang Guan 已提交
1341
      continueExec = true;
S
Shengliang Guan 已提交
1342
    } else {
S
Shengliang Guan 已提交
1343 1344
      mError("trans:%d, stage keep on redoAction since %s, failedTimes:%d", pTrans->id, terrstr(), pTrans->failedTimes);
      continueExec = false;
S
Shengliang Guan 已提交
1345 1346 1347
    }
  }

S
Shengliang Guan 已提交
1348
  return continueExec;
S
Shengliang Guan 已提交
1349 1350
}

S
Shengliang Guan 已提交
1351
static bool mndTransPerformCommitStage(SMnode *pMnode, STrans *pTrans) {
1352
  if (mndCannotExecuteTransAction(pMnode)) return false;
1353

S
Shengliang Guan 已提交
1354 1355 1356 1357 1358
  bool    continueExec = true;
  int32_t code = mndTransCommit(pMnode, pTrans);

  if (code == 0) {
    pTrans->code = 0;
1359
    pTrans->stage = TRN_STAGE_COMMIT_ACTION;
1360
    mInfo("trans:%d, stage from commit to commitAction", pTrans->id);
S
Shengliang Guan 已提交
1361 1362 1363
    continueExec = true;
  } else {
    pTrans->code = terrno;
1364 1365 1366
    pTrans->failedTimes++;
    mError("trans:%d, stage keep on commit since %s, failedTimes:%d", pTrans->id, terrstr(), pTrans->failedTimes);
    continueExec = false;
S
Shengliang Guan 已提交
1367 1368 1369
  }

  return continueExec;
S
Shengliang Guan 已提交
1370 1371
}

1372
static bool mndTransPerformCommitActionStage(SMnode *pMnode, STrans *pTrans) {
S
Shengliang Guan 已提交
1373
  bool    continueExec = true;
1374
  int32_t code = mndTransExecuteCommitActions(pMnode, pTrans);
S
Shengliang Guan 已提交
1375 1376

  if (code == 0) {
S
Shengliang Guan 已提交
1377 1378
    pTrans->code = 0;
    pTrans->stage = TRN_STAGE_FINISHED;
1379
    mInfo("trans:%d, stage from commitAction to finished", pTrans->id);
S
Shengliang Guan 已提交
1380
    continueExec = true;
S
Shengliang Guan 已提交
1381
  } else {
S
Shengliang Guan 已提交
1382 1383
    pTrans->code = terrno;
    pTrans->failedTimes++;
1384
    mError("trans:%d, stage keep on commitAction since %s, failedTimes:%d", pTrans->id, terrstr(), pTrans->failedTimes);
S
Shengliang Guan 已提交
1385 1386 1387 1388 1389 1390 1391 1392 1393 1394 1395
    continueExec = false;
  }

  return continueExec;
}

static bool mndTransPerformUndoActionStage(SMnode *pMnode, STrans *pTrans) {
  bool    continueExec = true;
  int32_t code = mndTransExecuteUndoActions(pMnode, pTrans);

  if (code == 0) {
1396
    pTrans->stage = TRN_STAGE_FINISHED;
1397
    mInfo("trans:%d, stage from undoAction to finished", pTrans->id);
S
Shengliang Guan 已提交
1398
    continueExec = true;
S
Shengliang Guan 已提交
1399
  } else if (code == TSDB_CODE_ACTION_IN_PROGRESS) {
1400
    mInfo("trans:%d, stage keep on undoAction since %s", pTrans->id, tstrerror(code));
S
Shengliang Guan 已提交
1401 1402 1403
    continueExec = false;
  } else {
    pTrans->failedTimes++;
1404
    mError("trans:%d, stage keep on undoAction since %s, failedTimes:%d", pTrans->id, terrstr(), pTrans->failedTimes);
S
Shengliang Guan 已提交
1405 1406 1407 1408 1409 1410 1411
    continueExec = false;
  }

  return continueExec;
}

static bool mndTransPerformRollbackStage(SMnode *pMnode, STrans *pTrans) {
1412
  if (mndCannotExecuteTransAction(pMnode)) return false;
1413

S
Shengliang Guan 已提交
1414 1415 1416 1417
  bool    continueExec = true;
  int32_t code = mndTransRollback(pMnode, pTrans);

  if (code == 0) {
1418
    pTrans->stage = TRN_STAGE_UNDO_ACTION;
1419
    mInfo("trans:%d, stage from rollback to undoAction", pTrans->id);
S
Shengliang Guan 已提交
1420 1421 1422
    continueExec = true;
  } else {
    pTrans->failedTimes++;
1423
    mError("trans:%d, stage keep on rollback since %s, failedTimes:%d", pTrans->id, terrstr(), pTrans->failedTimes);
S
Shengliang Guan 已提交
1424
    continueExec = false;
S
Shengliang Guan 已提交
1425 1426
  }

S
Shengliang Guan 已提交
1427 1428 1429 1430 1431 1432 1433 1434
  return continueExec;
}

static bool mndTransPerfromFinishedStage(SMnode *pMnode, STrans *pTrans) {
  bool continueExec = false;

  SSdbRaw *pRaw = mndTransActionEncode(pTrans);
  if (pRaw == NULL) {
S
Shengliang Guan 已提交
1435
    mError("trans:%d, failed to encode while finish trans since %s", pTrans->id, terrstr());
S
Shengliang Guan 已提交
1436 1437 1438 1439 1440 1441 1442 1443
  }
  sdbSetRawStatus(pRaw, SDB_STATUS_DROPPED);

  int32_t code = sdbWrite(pMnode->pSdb, pRaw);
  if (code != 0) {
    mError("trans:%d, failed to write sdb since %s", pTrans->id, terrstr());
  }

1444
  mInfo("trans:%d, execute finished, code:0x%x, failedTimes:%d", pTrans->id, pTrans->code, pTrans->failedTimes);
S
Shengliang Guan 已提交
1445
  return continueExec;
S
Shengliang Guan 已提交
1446
}
S
Shengliang Guan 已提交
1447

1448
void mndTransExecute(SMnode *pMnode, STrans *pTrans) {
S
Shengliang Guan 已提交
1449
  bool continueExec = true;
S
Shengliang Guan 已提交
1450

S
Shengliang Guan 已提交
1451
  while (continueExec) {
1452
    mInfo("trans:%d, continue to execute, stage:%s", pTrans->id, mndTransStr(pTrans->stage));
S
Shengliang Guan 已提交
1453
    pTrans->lastExecTime = taosGetTimestampMs();
S
Shengliang Guan 已提交
1454 1455
    switch (pTrans->stage) {
      case TRN_STAGE_PREPARE:
S
Shengliang Guan 已提交
1456 1457 1458 1459 1460
        continueExec = mndTransPerformPrepareStage(pMnode, pTrans);
        break;
      case TRN_STAGE_REDO_ACTION:
        continueExec = mndTransPerformRedoActionStage(pMnode, pTrans);
        break;
1461 1462 1463 1464 1465
      case TRN_STAGE_COMMIT:
        continueExec = mndTransPerformCommitStage(pMnode, pTrans);
        break;
      case TRN_STAGE_COMMIT_ACTION:
        continueExec = mndTransPerformCommitActionStage(pMnode, pTrans);
S
Shengliang Guan 已提交
1466 1467
        break;
      case TRN_STAGE_ROLLBACK:
S
Shengliang Guan 已提交
1468 1469
        continueExec = mndTransPerformRollbackStage(pMnode, pTrans);
        break;
1470 1471 1472
      case TRN_STAGE_UNDO_ACTION:
        continueExec = mndTransPerformUndoActionStage(pMnode, pTrans);
        break;
S
Shengliang Guan 已提交
1473 1474
      case TRN_STAGE_FINISHED:
        continueExec = mndTransPerfromFinishedStage(pMnode, pTrans);
S
Shengliang Guan 已提交
1475
        break;
S
Shengliang Guan 已提交
1476
      default:
S
Shengliang Guan 已提交
1477 1478
        continueExec = false;
        break;
S
Shengliang Guan 已提交
1479 1480 1481
    }
  }

1482
  mndTransSendRpcRsp(pMnode, pTrans);
S
Shengliang Guan 已提交
1483
}
S
Shengliang Guan 已提交
1484

1485
static int32_t mndProcessTransTimer(SRpcMsg *pReq) {
S
Shengliang Guan 已提交
1486
  mndTransPullup(pReq->info.node);
S
Shengliang Guan 已提交
1487 1488 1489
  return 0;
}

1490
int32_t mndKillTrans(SMnode *pMnode, STrans *pTrans) {
S
Shengliang Guan 已提交
1491 1492 1493 1494 1495 1496 1497 1498 1499 1500
  SArray *pArray = NULL;
  if (pTrans->stage == TRN_STAGE_REDO_ACTION) {
    pArray = pTrans->redoActions;
  } else if (pTrans->stage == TRN_STAGE_UNDO_ACTION) {
    pArray = pTrans->undoActions;
  } else {
    terrno = TSDB_CODE_MND_TRANS_INVALID_STAGE;
    return -1;
  }

1501
  for (int32_t i = 0; i < taosArrayGetSize(pArray); ++i) {
S
Shengliang Guan 已提交
1502
    STransAction *pAction = taosArrayGet(pArray, i);
S
Shengliang Guan 已提交
1503 1504 1505 1506 1507
    mInfo("trans:%d, %s:%d set processed for kill msg received, errCode from %s to success", pTrans->id,
          mndTransStr(pAction->stage), i, tstrerror(pAction->errCode));
    pAction->msgSent = 1;
    pAction->msgReceived = 1;
    pAction->errCode = 0;
S
Shengliang Guan 已提交
1508 1509 1510 1511 1512 1513
  }

  mndTransExecute(pMnode, pTrans);
  return 0;
}

S
Shengliang Guan 已提交
1514 1515
static int32_t mndProcessKillTransReq(SRpcMsg *pReq) {
  SMnode       *pMnode = pReq->info.node;
S
Shengliang Guan 已提交
1516
  SKillTransReq killReq = {0};
S
Shengliang Guan 已提交
1517
  int32_t       code = -1;
1518
  STrans       *pTrans = NULL;
S
Shengliang Guan 已提交
1519

S
Shengliang Guan 已提交
1520
  if (tDeserializeSKillTransReq(pReq->pCont, pReq->contLen, &killReq) != 0) {
S
Shengliang Guan 已提交
1521
    terrno = TSDB_CODE_INVALID_MSG;
S
Shengliang Guan 已提交
1522
    goto _OVER;
S
Shengliang Guan 已提交
1523 1524 1525
  }

  mInfo("trans:%d, start to kill", killReq.transId);
1526
  if (mndCheckOperPrivilege(pMnode, pReq->info.conn.user, MND_OPER_KILL_TRANS) != 0) {
S
Shengliang Guan 已提交
1527
    goto _OVER;
S
Shengliang Guan 已提交
1528 1529
  }

S
Shengliang Guan 已提交
1530
  pTrans = mndAcquireTrans(pMnode, killReq.transId);
S
Shengliang Guan 已提交
1531
  if (pTrans == NULL) {
1532
    goto _OVER;
S
Shengliang Guan 已提交
1533 1534
  }

S
Shengliang Guan 已提交
1535 1536
  code = mndKillTrans(pMnode, pTrans);

S
Shengliang Guan 已提交
1537
_OVER:
1538
  if (code != 0) {
S
Shengliang Guan 已提交
1539 1540 1541
    mError("trans:%d, failed to kill since %s", killReq.transId, terrstr());
  }

S
Shengliang Guan 已提交
1542
  mndReleaseTrans(pMnode, pTrans);
S
Shengliang Guan 已提交
1543
  return code;
S
Shengliang Guan 已提交
1544 1545
}

1546 1547
static int32_t mndCompareTransId(int32_t *pTransId1, int32_t *pTransId2) { return *pTransId1 >= *pTransId2 ? 1 : 0; }

S
Shengliang Guan 已提交
1548
void mndTransPullup(SMnode *pMnode) {
1549 1550 1551
  SSdb   *pSdb = pMnode->pSdb;
  SArray *pArray = taosArrayInit(sdbGetSize(pSdb, SDB_TRANS), sizeof(int32_t));
  if (pArray == NULL) return;
S
Shengliang Guan 已提交
1552

1553
  void *pIter = NULL;
S
Shengliang Guan 已提交
1554
  while (1) {
1555
    STrans *pTrans = NULL;
S
Shengliang Guan 已提交
1556 1557
    pIter = sdbFetch(pMnode->pSdb, SDB_TRANS, pIter, (void **)&pTrans);
    if (pIter == NULL) break;
1558 1559 1560
    taosArrayPush(pArray, &pTrans->id);
    sdbRelease(pSdb, pTrans);
  }
S
Shengliang Guan 已提交
1561

1562 1563 1564 1565 1566 1567 1568 1569 1570
  taosArraySort(pArray, (__compar_fn_t)mndCompareTransId);

  for (int32_t i = 0; i < taosArrayGetSize(pArray); ++i) {
    int32_t *pTransId = taosArrayGet(pArray, i);
    STrans  *pTrans = mndAcquireTrans(pMnode, *pTransId);
    if (pTrans != NULL) {
      mndTransExecute(pMnode, pTrans);
    }
    mndReleaseTrans(pMnode, pTrans);
S
Shengliang Guan 已提交
1571
  }
S
Shengliang Guan 已提交
1572

1573
  sdbWriteFile(pMnode->pSdb, SDB_WRITE_DELTA);
1574
  taosArrayDestroy(pArray);
1575
}
S
Shengliang Guan 已提交
1576

S
Shengliang Guan 已提交
1577 1578
static int32_t mndRetrieveTrans(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows) {
  SMnode *pMnode = pReq->info.node;
1579
  SSdb   *pSdb = pMnode->pSdb;
S
Shengliang Guan 已提交
1580 1581 1582
  int32_t numOfRows = 0;
  STrans *pTrans = NULL;
  int32_t cols = 0;
1583
  char   *pWrite;
S
Shengliang Guan 已提交
1584 1585 1586 1587 1588 1589 1590

  while (numOfRows < rows) {
    pShow->pIter = sdbFetch(pSdb, SDB_TRANS, pShow->pIter, (void **)&pTrans);
    if (pShow->pIter == NULL) break;

    cols = 0;

S
Shengliang Guan 已提交
1591 1592
    SColumnInfoData *pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
    colDataAppend(pColInfo, numOfRows, (const char *)&pTrans->id, false);
S
Shengliang Guan 已提交
1593

S
Shengliang Guan 已提交
1594 1595
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
    colDataAppend(pColInfo, numOfRows, (const char *)&pTrans->createdTime, false);
S
Shengliang Guan 已提交
1596

S
Shengliang Guan 已提交
1597
    char stage[TSDB_TRANS_STAGE_LEN + VARSTR_HEADER_SIZE] = {0};
1598
    STR_WITH_MAXSIZE_TO_VARSTR(stage, mndTransStr(pTrans->stage), pShow->pMeta->pSchemas[cols].bytes);
S
Shengliang Guan 已提交
1599 1600
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
    colDataAppend(pColInfo, numOfRows, (const char *)stage, false);
S
Shengliang Guan 已提交
1601

1602 1603 1604 1605 1606 1607 1608
    char dbname1[TSDB_DB_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
    STR_WITH_MAXSIZE_TO_VARSTR(dbname1, mndGetDbStr(pTrans->dbname1), pShow->pMeta->pSchemas[cols].bytes);
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
    colDataAppend(pColInfo, numOfRows, (const char *)dbname1, false);

    char dbname2[TSDB_DB_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
    STR_WITH_MAXSIZE_TO_VARSTR(dbname2, mndGetDbStr(pTrans->dbname2), pShow->pMeta->pSchemas[cols].bytes);
S
Shengliang Guan 已提交
1609
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
1610
    colDataAppend(pColInfo, numOfRows, (const char *)dbname2, false);
S
Shengliang Guan 已提交
1611

1612 1613 1614
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
    colDataAppend(pColInfo, numOfRows, (const char *)&pTrans->failedTimes, false);

S
Shengliang Guan 已提交
1615 1616
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
    colDataAppend(pColInfo, numOfRows, (const char *)&pTrans->lastExecTime, false);
S
Shengliang Guan 已提交
1617

1618 1619 1620 1621 1622 1623 1624 1625 1626 1627
    char    lastInfo[TSDB_TRANS_ERROR_LEN + VARSTR_HEADER_SIZE] = {0};
    char    detail[TSDB_TRANS_ERROR_LEN] = {0};
    int32_t len = snprintf(detail, sizeof(detail), "action:%d code:0x%x(%s) ", pTrans->lastAction,
                           pTrans->lastErrorNo & 0xFFFF, tstrerror(pTrans->lastErrorNo));
    SEpSet  epset = pTrans->lastEpset;
    if (epset.numOfEps > 0) {
      len += snprintf(detail + len, sizeof(detail) - len, "msgType:%s numOfEps:%d inUse:%d ",
                      TMSG_INFO(pTrans->lastMsgType), epset.numOfEps, epset.inUse);
      for (int32_t i = 0; i < pTrans->lastEpset.numOfEps; ++i) {
        len += snprintf(detail + len, sizeof(detail) - len, "ep:%d-%s:%u ", i, epset.eps[i].fqdn, epset.eps[i].port);
1628 1629
      }
    }
1630
    STR_WITH_MAXSIZE_TO_VARSTR(lastInfo, detail, pShow->pMeta->pSchemas[cols].bytes);
S
Shengliang Guan 已提交
1631
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
1632
    colDataAppend(pColInfo, numOfRows, (const char *)lastInfo, false);
S
Shengliang Guan 已提交
1633 1634 1635 1636 1637

    numOfRows++;
    sdbRelease(pSdb, pTrans);
  }

1638
  pShow->numOfRows += numOfRows;
S
Shengliang Guan 已提交
1639 1640 1641 1642 1643 1644 1645
  return numOfRows;
}

static void mndCancelGetNextTrans(SMnode *pMnode, void *pIter) {
  SSdb *pSdb = pMnode->pSdb;
  sdbCancelFetch(pSdb, pIter);
}