vmHandle.c 17.8 KB
Newer Older
S
shm  
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
shm  
Shengliang Guan 已提交
17
#include "vmInt.h"
S
shm  
Shengliang Guan 已提交
18

S
Shengliang Guan 已提交
19
void vmGetVnodeLoads(SVnodeMgmt *pMgmt, SMonVloadInfo *pInfo) {
S
Shengliang Guan 已提交
20 21 22
  pInfo->pVloads = taosArrayInit(pMgmt->state.totalVnodes, sizeof(SVnodeLoad));
  if (pInfo->pVloads == NULL) return;

23
  taosThreadRwlockRdlock(&pMgmt->lock);
S
Shengliang Guan 已提交
24 25 26 27 28 29 30 31 32 33 34 35 36

  void *pIter = taosHashIterate(pMgmt->hash, NULL);
  while (pIter) {
    SVnodeObj **ppVnode = pIter;
    if (ppVnode == NULL || *ppVnode == NULL) continue;

    SVnodeObj *pVnode = *ppVnode;
    SVnodeLoad vload = {0};
    vnodeGetLoad(pVnode->pImpl, &vload);
    taosArrayPush(pInfo->pVloads, &vload);
    pIter = taosHashIterate(pMgmt->hash, pIter);
  }

37
  taosThreadRwlockUnlock(&pMgmt->lock);
S
Shengliang Guan 已提交
38 39
}

S
Shengliang Guan 已提交
40
void vmGetMonitorInfo(SVnodeMgmt *pMgmt, SMonVmInfo *pInfo) {
S
Shengliang Guan 已提交
41
  SMonVloadInfo vloads = {0};
S
Shengliang 已提交
42
  vmGetVnodeLoads(pMgmt, &vloads);
S
Shengliang Guan 已提交
43 44 45

  SArray *pVloads = vloads.pVloads;
  if (pVloads == NULL) return;
S
Shengliang Guan 已提交
46 47 48 49 50 51 52 53 54

  int32_t totalVnodes = 0;
  int32_t masterNum = 0;
  int64_t numOfSelectReqs = 0;
  int64_t numOfInsertReqs = 0;
  int64_t numOfInsertSuccessReqs = 0;
  int64_t numOfBatchInsertReqs = 0;
  int64_t numOfBatchInsertSuccessReqs = 0;

S
Shengliang Guan 已提交
55 56
  for (int32_t i = 0; i < taosArrayGetSize(pVloads); ++i) {
    SVnodeLoad *pLoad = taosArrayGet(pVloads, i);
S
Shengliang Guan 已提交
57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72
    numOfSelectReqs += pLoad->numOfSelectReqs;
    numOfInsertReqs += pLoad->numOfInsertReqs;
    numOfInsertSuccessReqs += pLoad->numOfInsertSuccessReqs;
    numOfBatchInsertReqs += pLoad->numOfBatchInsertReqs;
    numOfBatchInsertSuccessReqs += pLoad->numOfBatchInsertSuccessReqs;
    if (pLoad->syncState == TAOS_SYNC_STATE_LEADER) masterNum++;
    totalVnodes++;
  }

  pInfo->vstat.totalVnodes = totalVnodes;
  pInfo->vstat.masterNum = masterNum;
  pInfo->vstat.numOfSelectReqs = numOfSelectReqs - pMgmt->state.numOfSelectReqs;
  pInfo->vstat.numOfInsertReqs = numOfInsertReqs - pMgmt->state.numOfInsertReqs;
  pInfo->vstat.numOfInsertSuccessReqs = numOfInsertSuccessReqs - pMgmt->state.numOfInsertSuccessReqs;
  pInfo->vstat.numOfBatchInsertReqs = numOfBatchInsertReqs - pMgmt->state.numOfBatchInsertReqs;
  pInfo->vstat.numOfBatchInsertSuccessReqs = numOfBatchInsertSuccessReqs - pMgmt->state.numOfBatchInsertSuccessReqs;
S
Shengliang Guan 已提交
73 74 75 76 77 78 79
  pMgmt->state.totalVnodes = totalVnodes;
  pMgmt->state.masterNum = masterNum;
  pMgmt->state.numOfSelectReqs = numOfSelectReqs;
  pMgmt->state.numOfInsertReqs = numOfInsertReqs;
  pMgmt->state.numOfInsertSuccessReqs = numOfInsertSuccessReqs;
  pMgmt->state.numOfBatchInsertReqs = numOfBatchInsertReqs;
  pMgmt->state.numOfBatchInsertSuccessReqs = numOfBatchInsertSuccessReqs;
S
Shengliang Guan 已提交
80

81
  tfsGetMonitorInfo(pMgmt->pTfs, &pInfo->tfs);
S
Shengliang Guan 已提交
82
  taosArrayDestroy(pVloads);
83 84
}

S
Shengliang Guan 已提交
85
int32_t vmProcessGetMonitorInfoReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
86
  SMonVmInfo vmInfo = {0};
S
Shengliang 已提交
87
  vmGetMonitorInfo(pMgmt, &vmInfo);
S
Shengliang Guan 已提交
88
  dmGetMonitorSystemInfo(&vmInfo.sys);
89
  monGetLogs(&vmInfo.log);
90 91 92 93 94 95 96 97 98 99 100 101 102 103

  int32_t rspLen = tSerializeSMonVmInfo(NULL, 0, &vmInfo);
  if (rspLen < 0) {
    terrno = TSDB_CODE_INVALID_MSG;
    return -1;
  }

  void *pRsp = rpcMallocCont(rspLen);
  if (pRsp == NULL) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return -1;
  }

  tSerializeSMonVmInfo(pRsp, rspLen, &vmInfo);
S
Shengliang Guan 已提交
104 105
  pMsg->info.rsp = pRsp;
  pMsg->info.rspLen = rspLen;
106 107 108 109
  tFreeSMonVmInfo(&vmInfo);
  return 0;
}

S
Shengliang Guan 已提交
110
int32_t vmProcessGetLoadsReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
111
  SMonVloadInfo vloads = {0};
S
Shengliang 已提交
112
  vmGetVnodeLoads(pMgmt, &vloads);
113 114 115 116 117 118 119 120 121 122 123 124 125 126

  int32_t rspLen = tSerializeSMonVloadInfo(NULL, 0, &vloads);
  if (rspLen < 0) {
    terrno = TSDB_CODE_INVALID_MSG;
    return -1;
  }

  void *pRsp = rpcMallocCont(rspLen);
  if (pRsp == NULL) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return -1;
  }

  tSerializeSMonVloadInfo(pRsp, rspLen, &vloads);
S
Shengliang Guan 已提交
127 128
  pMsg->info.rsp = pRsp;
  pMsg->info.rspLen = rspLen;
129 130 131 132
  tFreeSMonVloadInfo(&vloads);
  return 0;
}

S
shm  
Shengliang Guan 已提交
133
static void vmGenerateVnodeCfg(SCreateVnodeReq *pCreate, SVnodeCfg *pCfg) {
H
Hongze Cheng 已提交
134 135
  memcpy(pCfg, &vnodeCfgDefault, sizeof(SVnodeCfg));

S
shm  
Shengliang Guan 已提交
136
  pCfg->vgId = pCreate->vgId;
S
Shengliang Guan 已提交
137 138
  tstrncpy(pCfg->dbname, pCreate->db, sizeof(pCfg->dbname));
  pCfg->dbId = pCreate->dbUid;
H
Hongze Cheng 已提交
139 140
  pCfg->szPage = pCreate->pageSize * 1024;
  pCfg->szCache = pCreate->pages;
141 142
  pCfg->cacheLast = pCreate->cacheLast;
  pCfg->cacheLastSize = pCreate->cacheLastSize;
H
Hongze Cheng 已提交
143
  pCfg->szBuf = (uint64_t)pCreate->buffer * 1024 * 1024;
S
shm  
Shengliang Guan 已提交
144
  pCfg->isWeak = true;
C
Cary Xu 已提交
145
  pCfg->isTsma = pCreate->isTsma;
C
Cary Xu 已提交
146
  pCfg->tsdbCfg.compression = pCreate->compression;
H
Hongze Cheng 已提交
147
  pCfg->tsdbCfg.precision = pCreate->precision;
148 149 150 151
  pCfg->tsdbCfg.days = pCreate->daysPerFile;
  pCfg->tsdbCfg.keep0 = pCreate->daysToKeep0;
  pCfg->tsdbCfg.keep1 = pCreate->daysToKeep1;
  pCfg->tsdbCfg.keep2 = pCreate->daysToKeep2;
H
Hongze Cheng 已提交
152 153
  pCfg->tsdbCfg.minRows = pCreate->minRows;
  pCfg->tsdbCfg.maxRows = pCreate->maxRows;
C
Cary Xu 已提交
154
  for (size_t i = 0; i < taosArrayGetSize(pCreate->pRetensions); ++i) {
C
Cary Xu 已提交
155 156 157 158 159
    SRetention *pRetention = &pCfg->tsdbCfg.retentions[i];
    memcpy(pRetention, taosArrayGet(pCreate->pRetensions, i), sizeof(SRetention));
    if (i == 0) {
      if ((pRetention->freq > 0 && pRetention->keep > 0)) pCfg->isRsma = 1;
    }
C
Cary Xu 已提交
160
  }
C
Cary Xu 已提交
161

S
shm  
Shengliang Guan 已提交
162
  pCfg->walCfg.vgId = pCreate->vgId;
163
  pCfg->walCfg.fsyncPeriod = pCreate->walFsyncPeriod;
S
Shengliang Guan 已提交
164
  pCfg->walCfg.retentionPeriod = pCreate->walRetentionPeriod;
L
Liu Jicong 已提交
165 166
  pCfg->walCfg.rollPeriod = pCreate->walRollPeriod;
  pCfg->walCfg.retentionSize = pCreate->walRetentionSize;
S
Shengliang Guan 已提交
167 168 169
  pCfg->walCfg.segSize = pCreate->walSegmentSize;
  pCfg->walCfg.level = pCreate->walLevel;

170
  pCfg->sstTrigger = pCreate->sstTrigger;
S
shm  
Shengliang Guan 已提交
171 172 173
  pCfg->hashBegin = pCreate->hashBegin;
  pCfg->hashEnd = pCreate->hashEnd;
  pCfg->hashMethod = pCreate->hashMethod;
174 175
  pCfg->hashPrefix = pCreate->hashPrefix;
  pCfg->hashSuffix = pCreate->hashSuffix;
M
Minghao Li 已提交
176

177
  pCfg->standby = pCfg->standby;
M
Minghao Li 已提交
178 179
  pCfg->syncCfg.myIndex = pCreate->selfIndex;
  pCfg->syncCfg.replicaNum = pCreate->replica;
S
Shengliang Guan 已提交
180
  memset(&pCfg->syncCfg.nodeInfo, 0, sizeof(pCfg->syncCfg.nodeInfo));
M
Minghao Li 已提交
181
  for (int i = 0; i < pCreate->replica; ++i) {
182 183 184
    SNodeInfo *pNode = &pCfg->syncCfg.nodeInfo[i];
    pNode->nodePort = pCreate->replicas[i].port;
    tstrncpy(pNode->nodeFqdn, pCreate->replicas[i].fqdn, sizeof(pNode->nodeFqdn));
M
Minghao Li 已提交
185
  }
S
shm  
Shengliang Guan 已提交
186
}
S
shm  
Shengliang Guan 已提交
187

S
Shengliang 已提交
188
static void vmGenerateWrapperCfg(SVnodeMgmt *pMgmt, SCreateVnodeReq *pCreate, SWrapperCfg *pCfg) {
S
shm  
Shengliang Guan 已提交
189 190
  pCfg->vgId = pCreate->vgId;
  pCfg->vgVersion = pCreate->vgVersion;
S
Shengliang Guan 已提交
191 192
  pCfg->dropped = 0;
  snprintf(pCfg->path, sizeof(pCfg->path), "%s%svnode%d", pMgmt->path, TD_DIRSEP, pCreate->vgId);
S
shm  
Shengliang Guan 已提交
193 194
}

C
Cary Xu 已提交
195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212
static int32_t vmTsmaAdjustDays(SVnodeCfg *pCfg, SCreateVnodeReq *pReq) {
  if (pReq->isTsma) {
    SMsgHead *smaMsg = pReq->pTsma;
    uint32_t  contLen = (uint32_t)(htonl(smaMsg->contLen) - sizeof(SMsgHead));
    return smaGetTSmaDays(pCfg, POINTER_SHIFT(smaMsg, sizeof(SMsgHead)), contLen, &pCfg->tsdbCfg.days);
  }
  return 0;
}

static int32_t vmTsmaProcessCreate(SVnode *pVnode, SCreateVnodeReq *pReq) {
  if (pReq->isTsma) {
    SMsgHead *smaMsg = pReq->pTsma;
    uint32_t  contLen = (uint32_t)(htonl(smaMsg->contLen) - sizeof(SMsgHead));
    return vnodeProcessCreateTSma(pVnode, POINTER_SHIFT(smaMsg, sizeof(SMsgHead)), contLen);
  }
  return 0;
}

S
Shengliang Guan 已提交
213
int32_t vmProcessCreateVnodeReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
S
shm  
Shengliang Guan 已提交
214
  SCreateVnodeReq createReq = {0};
215 216
  SVnodeCfg       vnodeCfg = {0};
  SWrapperCfg     wrapperCfg = {0};
S
Shengliang Guan 已提交
217 218
  int32_t         code = -1;
  char            path[TSDB_FILENAME_LEN] = {0};
H
Hongze Cheng 已提交
219

S
Shengliang Guan 已提交
220
  if (tDeserializeSCreateVnodeReq(pMsg->pCont, pMsg->contLen, &createReq) != 0) {
S
shm  
Shengliang Guan 已提交
221 222 223 224
    terrno = TSDB_CODE_INVALID_MSG;
    return -1;
  }

225
  dInfo("vgId:%d, start to create vnode, tsma:%d standby:%d cacheLast:%d cacheLastSize:%d sstTrigger:%d",
226 227
         createReq.vgId, createReq.isTsma, createReq.standby, createReq.cacheLast, createReq.cacheLastSize,
         createReq.sstTrigger);
228 229
  dInfo("vgId:%d, hashMethod:%d begin:%u end:%u prefix:%d surfix:%d", createReq.vgId, createReq.hashMethod,
         createReq.hashBegin, createReq.hashEnd, createReq.hashPrefix, createReq.hashSuffix);
S
shm  
Shengliang Guan 已提交
230 231
  vmGenerateVnodeCfg(&createReq, &vnodeCfg);

C
Cary Xu 已提交
232 233 234 235
  if (vmTsmaAdjustDays(&vnodeCfg, &createReq) < 0) {
    dError("vgId:%d, failed to adjust tsma days since %s", createReq.vgId, terrstr());
    code = terrno;
    goto _OVER;
C
Cary Xu 已提交
236 237
  }

S
shm  
Shengliang Guan 已提交
238 239 240 241 242
  vmGenerateWrapperCfg(pMgmt, &createReq, &wrapperCfg);

  SVnodeObj *pVnode = vmAcquireVnode(pMgmt, createReq.vgId);
  if (pVnode != NULL) {
    dDebug("vgId:%d, already exist", createReq.vgId);
S
Shengliang Guan 已提交
243
    tFreeSCreateVnodeReq(&createReq);
S
shm  
Shengliang Guan 已提交
244
    vmReleaseVnode(pMgmt, pVnode);
245
    terrno = TSDB_CODE_NODE_ALREADY_DEPLOYED;
C
Cary Xu 已提交
246
    code = terrno;
247
    return 0;
S
shm  
Shengliang Guan 已提交
248 249
  }

H
Hongze Cheng 已提交
250 251 252 253
  snprintf(path, TSDB_FILENAME_LEN, "vnode%svnode%d", TD_DIRSEP, vnodeCfg.vgId);
  if (vnodeCreate(path, &vnodeCfg, pMgmt->pTfs) < 0) {
    tFreeSCreateVnodeReq(&createReq);
    dError("vgId:%d, failed to create vnode since %s", createReq.vgId, terrstr());
C
Cary Xu 已提交
254 255
    code = terrno;
    goto _OVER;
H
Hongze Cheng 已提交
256 257
  }

S
Shengliang 已提交
258
  SVnode *pImpl = vnodeOpen(path, pMgmt->pTfs, pMgmt->msgCb);
S
shm  
Shengliang Guan 已提交
259
  if (pImpl == NULL) {
C
Cary Xu 已提交
260
    dError("vgId:%d, failed to open vnode since %s", createReq.vgId, terrstr());
261
    code = terrno;
S
Shengliang Guan 已提交
262
    goto _OVER;
S
shm  
Shengliang Guan 已提交
263 264
  }

S
Shengliang Guan 已提交
265
  code = vmOpenVnode(pMgmt, &wrapperCfg, pImpl);
S
shm  
Shengliang Guan 已提交
266 267
  if (code != 0) {
    dError("vgId:%d, failed to open vnode since %s", createReq.vgId, terrstr());
C
Cary Xu 已提交
268
    code = terrno;
S
Shengliang Guan 已提交
269
    goto _OVER;
S
shm  
Shengliang Guan 已提交
270 271
  }

C
Cary Xu 已提交
272 273 274 275 276
  code = vmTsmaProcessCreate(pImpl, &createReq);
  if (code != 0) {
    dError("vgId:%d, failed to create tsma since %s", createReq.vgId, terrstr());
    code = terrno;
    goto _OVER;
C
Cary Xu 已提交
277 278
  }

L
Li Minghao 已提交
279 280 281
  code = vnodeStart(pImpl);
  if (code != 0) {
    dError("vgId:%d, failed to start sync since %s", createReq.vgId, terrstr());
S
Shengliang Guan 已提交
282
    goto _OVER;
L
Li Minghao 已提交
283 284
  }

S
Shengliang Guan 已提交
285
  code = vmWriteVnodeListToFile(pMgmt);
C
Cary Xu 已提交
286 287 288 289
  if (code != 0) {
    code = terrno;
    goto _OVER;
  }
S
Shengliang Guan 已提交
290 291

_OVER:
S
shm  
Shengliang Guan 已提交
292 293
  if (code != 0) {
    vnodeClose(pImpl);
H
refact  
Hongze Cheng 已提交
294
    vnodeDestroy(path, pMgmt->pTfs);
S
Shengliang Guan 已提交
295 296
  } else {
    dInfo("vgId:%d, vnode is created", createReq.vgId);
S
shm  
Shengliang Guan 已提交
297 298
  }

S
Shengliang Guan 已提交
299 300 301
  tFreeSCreateVnodeReq(&createReq);
  terrno = code;
  return code;
S
shm  
Shengliang Guan 已提交
302 303
}

S
Shengliang Guan 已提交
304
int32_t vmProcessDropVnodeReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
S
shm  
Shengliang Guan 已提交
305
  SDropVnodeReq dropReq = {0};
S
Shengliang Guan 已提交
306
  if (tDeserializeSDropVnodeReq(pMsg->pCont, pMsg->contLen, &dropReq) != 0) {
S
shm  
Shengliang Guan 已提交
307 308 309 310 311
    terrno = TSDB_CODE_INVALID_MSG;
    return -1;
  }

  int32_t vgId = dropReq.vgId;
S
Shengliang Guan 已提交
312
  dDebug("vgId:%d, start to drop vnode", vgId);
S
shm  
Shengliang Guan 已提交
313 314 315 316

  SVnodeObj *pVnode = vmAcquireVnode(pMgmt, vgId);
  if (pVnode == NULL) {
    dDebug("vgId:%d, failed to drop since %s", vgId, terrstr());
317
    terrno = TSDB_CODE_NODE_NOT_DEPLOYED;
S
shm  
Shengliang Guan 已提交
318 319 320 321
    return -1;
  }

  pVnode->dropped = 1;
S
Shengliang Guan 已提交
322
  if (vmWriteVnodeListToFile(pMgmt) != 0) {
S
shm  
Shengliang Guan 已提交
323 324 325 326 327 328
    pVnode->dropped = 0;
    vmReleaseVnode(pMgmt, pVnode);
    return -1;
  }

  vmCloseVnode(pMgmt, pVnode);
S
Shengliang Guan 已提交
329
  vmWriteVnodeListToFile(pMgmt);
S
shm  
Shengliang Guan 已提交
330 331 332 333

  return 0;
}

S
Shengliang Guan 已提交
334
SArray *vmGetMsgHandles() {
S
Shengliang 已提交
335
  int32_t code = -1;
S
Shengliang Guan 已提交
336
  SArray *pArray = taosArrayInit(32, sizeof(SMgmtHandle));
S
Shengliang 已提交
337 338
  if (pArray == NULL) goto _OVER;

S
Shengliang Guan 已提交
339 340
  if (dmSetMgmtHandle(pArray, TDMT_MON_VM_INFO, vmPutMsgToMonitorQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_MON_VM_LOAD, vmPutMsgToMonitorQueue, 0) == NULL) goto _OVER;
341

S
Shengliang Guan 已提交
342
  if (dmSetMgmtHandle(pArray, TDMT_VND_SUBMIT, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
D
dapan1121 已提交
343
  if (dmSetMgmtHandle(pArray, TDMT_SCH_QUERY, vmPutMsgToQueryQueue, 0) == NULL) goto _OVER;
D
dapan1121 已提交
344
  if (dmSetMgmtHandle(pArray, TDMT_SCH_MERGE_QUERY, vmPutMsgToQueryQueue, 0) == NULL) goto _OVER;
D
dapan1121 已提交
345
  if (dmSetMgmtHandle(pArray, TDMT_SCH_QUERY_CONTINUE, vmPutMsgToQueryQueue, 0) == NULL) goto _OVER;
C
Cary Xu 已提交
346
  if (dmSetMgmtHandle(pArray, TDMT_VND_FETCH_RSMA, vmPutMsgToQueryQueue, 0) == NULL) goto _OVER;
C
Cary Xu 已提交
347
  if (dmSetMgmtHandle(pArray, TDMT_VND_EXEC_RSMA, vmPutMsgToQueryQueue, 0) == NULL) goto _OVER;
D
dapan1121 已提交
348
  if (dmSetMgmtHandle(pArray, TDMT_SCH_FETCH, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER;
D
dapan1121 已提交
349
  if (dmSetMgmtHandle(pArray, TDMT_SCH_MERGE_FETCH, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER;
S
Shengliang Guan 已提交
350 351 352
  if (dmSetMgmtHandle(pArray, TDMT_VND_ALTER_TABLE, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_VND_UPDATE_TAG_VAL, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_VND_TABLE_META, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER;
D
dapan1121 已提交
353
  if (dmSetMgmtHandle(pArray, TDMT_VND_TABLE_CFG, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER;
D
dapan1121 已提交
354
  if (dmSetMgmtHandle(pArray, TDMT_VND_BATCH_META, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER;
S
Shengliang Guan 已提交
355
  if (dmSetMgmtHandle(pArray, TDMT_VND_TABLES_META, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER;
D
dapan1121 已提交
356 357
  if (dmSetMgmtHandle(pArray, TDMT_SCH_CANCEL_TASK, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_SCH_DROP_TASK, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER;
S
Shengliang Guan 已提交
358
  if (dmSetMgmtHandle(pArray, TDMT_VND_CREATE_STB, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
wmmhello's avatar
wmmhello 已提交
359
  if (dmSetMgmtHandle(pArray, TDMT_VND_DROP_TTL_TABLE, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
S
Shengliang Guan 已提交
360 361 362 363 364 365 366 367 368 369
  if (dmSetMgmtHandle(pArray, TDMT_VND_ALTER_STB, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_VND_DROP_STB, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_VND_CREATE_TABLE, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_VND_DROP_TABLE, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_VND_CREATE_SMA, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_VND_CANCEL_SMA, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_VND_DROP_SMA, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_VND_SUBMIT_RSMA, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_VND_MQ_VG_CHANGE, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_VND_MQ_VG_DELETE, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
370
  if (dmSetMgmtHandle(pArray, TDMT_VND_MQ_COMMIT_OFFSET, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
371 372
  if (dmSetMgmtHandle(pArray, TDMT_VND_ADD_CHECK_INFO, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_VND_DELETE_CHECK_INFO, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
S
Shengliang Guan 已提交
373
  if (dmSetMgmtHandle(pArray, TDMT_VND_CONSUME, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER;
374
  if (dmSetMgmtHandle(pArray, TDMT_VND_DELETE, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
375
  if (dmSetMgmtHandle(pArray, TDMT_VND_BATCH_DEL, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
H
Hongze Cheng 已提交
376
  if (dmSetMgmtHandle(pArray, TDMT_VND_COMMIT, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
D
dapan1121 已提交
377
  if (dmSetMgmtHandle(pArray, TDMT_SCH_QUERY_HEARTBEAT, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER;
378

L
Liu Jicong 已提交
379
  if (dmSetMgmtHandle(pArray, TDMT_VND_STREAM_TRIGGER, vmPutMsgToStreamQueue, 0) == NULL) goto _OVER;
L
Liu Jicong 已提交
380
  if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_DROP, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
S
Shengliang Guan 已提交
381
  if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_DEPLOY, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
L
Liu Jicong 已提交
382 383 384 385 386 387 388
  if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_RUN, vmPutMsgToStreamQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_DISPATCH, vmPutMsgToStreamQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_DISPATCH_RSP, vmPutMsgToStreamQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_RECOVER, vmPutMsgToStreamQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_RECOVER_RSP, vmPutMsgToStreamQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_STREAM_RETRIEVE, vmPutMsgToStreamQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_STREAM_RETRIEVE_RSP, vmPutMsgToStreamQueue, 0) == NULL) goto _OVER;
S
Shengliang Guan 已提交
389 390 391

  if (dmSetMgmtHandle(pArray, TDMT_VND_ALTER_REPLICA, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_VND_ALTER_CONFIG, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
392
  if (dmSetMgmtHandle(pArray, TDMT_VND_ALTER_CONFIRM, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
S
Shengliang Guan 已提交
393
  if (dmSetMgmtHandle(pArray, TDMT_VND_ALTER_HASHRANGE, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
S
Shengliang Guan 已提交
394
  if (dmSetMgmtHandle(pArray, TDMT_VND_COMPACT, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
S
Shengliang Guan 已提交
395
  if (dmSetMgmtHandle(pArray, TDMT_VND_TRIM, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
S
Shengliang Guan 已提交
396 397 398
  if (dmSetMgmtHandle(pArray, TDMT_DND_CREATE_VNODE, vmPutMsgToMgmtQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_DND_DROP_VNODE, vmPutMsgToMgmtQueue, 0) == NULL) goto _OVER;

399 400 401 402
  if (dmSetMgmtHandle(pArray, TDMT_SYNC_TIMEOUT, vmPutMsgToSyncQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_SYNC_PING, vmPutMsgToSyncQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_SYNC_PING_REPLY, vmPutMsgToSyncQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_SYNC_CLIENT_REQUEST, vmPutMsgToSyncQueue, 0) == NULL) goto _OVER;
403
  if (dmSetMgmtHandle(pArray, TDMT_SYNC_CLIENT_REQUEST_BATCH, vmPutMsgToSyncQueue, 0) == NULL) goto _OVER;
404 405 406 407
  if (dmSetMgmtHandle(pArray, TDMT_SYNC_CLIENT_REQUEST_REPLY, vmPutMsgToSyncQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_SYNC_REQUEST_VOTE, vmPutMsgToSyncQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_SYNC_REQUEST_VOTE_REPLY, vmPutMsgToSyncQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_SYNC_APPEND_ENTRIES, vmPutMsgToSyncQueue, 0) == NULL) goto _OVER;
408
  if (dmSetMgmtHandle(pArray, TDMT_SYNC_APPEND_ENTRIES_BATCH, vmPutMsgToSyncQueue, 0) == NULL) goto _OVER;
409
  if (dmSetMgmtHandle(pArray, TDMT_SYNC_APPEND_ENTRIES_REPLY, vmPutMsgToSyncQueue, 0) == NULL) goto _OVER;
410 411
  if (dmSetMgmtHandle(pArray, TDMT_SYNC_SNAPSHOT_SEND, vmPutMsgToSyncQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_SYNC_SNAPSHOT_RSP, vmPutMsgToSyncQueue, 0) == NULL) goto _OVER;
412
  if (dmSetMgmtHandle(pArray, TDMT_SYNC_SET_VNODE_STANDBY, vmPutMsgToSyncQueue, 0) == NULL) goto _OVER;
S
Shengliang 已提交
413 414 415 416 417 418 419 420 421 422

  code = 0;

_OVER:
  if (code != 0) {
    taosArrayDestroy(pArray);
    return NULL;
  } else {
    return pArray;
  }
S
shm  
Shengliang Guan 已提交
423
}