mnode.c 15.7 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 18
#include "mndAcct.h"
#include "mndAuth.h"
S
Shengliang Guan 已提交
19
#include "mndBnode.h"
S
Shengliang Guan 已提交
20
#include "mndCluster.h"
L
Liu Jicong 已提交
21
#include "mndConsumer.h"
S
Shengliang Guan 已提交
22 23 24 25 26
#include "mndDb.h"
#include "mndDnode.h"
#include "mndFunc.h"
#include "mndMnode.h"
#include "mndProfile.h"
S
Shengliang Guan 已提交
27
#include "mndQnode.h"
S
Shengliang Guan 已提交
28
#include "mndShow.h"
S
Shengliang Guan 已提交
29
#include "mndSnode.h"
S
Shengliang Guan 已提交
30
#include "mndStb.h"
L
Liu Jicong 已提交
31
#include "mndSubscribe.h"
S
Shengliang Guan 已提交
32 33
#include "mndSync.h"
#include "mndTelem.h"
L
Liu Jicong 已提交
34
#include "mndTopic.h"
S
Shengliang Guan 已提交
35 36 37
#include "mndTrans.h"
#include "mndUser.h"
#include "mndVgroup.h"
S
Shengliang Guan 已提交
38

S
Shengliang Guan 已提交
39 40 41 42
int32_t mndSendReqToDnode(SMnode *pMnode, SEpSet *pEpSet, SRpcMsg *pMsg) {
  if (pMnode == NULL || pMnode->sendReqToDnodeFp == NULL) {
    terrno = TSDB_CODE_MND_NOT_READY;
    return -1;
43
  }
S
Shengliang Guan 已提交
44 45

  return (*pMnode->sendReqToDnodeFp)(pMnode->pDnode, pEpSet, pMsg);
S
Shengliang Guan 已提交
46
}
S
Shengliang Guan 已提交
47

S
Shengliang Guan 已提交
48 49 50 51
int32_t mndSendReqToMnode(SMnode *pMnode, SRpcMsg *pMsg) {
  if (pMnode == NULL || pMnode->sendReqToDnodeFp == NULL) {
    terrno = TSDB_CODE_MND_NOT_READY;
    return -1;
52
  }
S
Shengliang Guan 已提交
53 54

  return (*pMnode->sendReqToMnodeFp)(pMnode->pDnode, pMsg);
S
Shengliang Guan 已提交
55
}
S
Shengliang Guan 已提交
56

S
Shengliang Guan 已提交
57 58 59
void mndSendRedirectRsp(SMnode *pMnode, SRpcMsg *pMsg) {
  if (pMnode != NULL && pMnode->sendRedirectRspFp != NULL) {
    (*pMnode->sendRedirectRspFp)(pMnode->pDnode, pMsg);
60
  }
S
Shengliang Guan 已提交
61
}
S
Shengliang Guan 已提交
62

S
Shengliang Guan 已提交
63 64 65 66 67 68 69 70 71 72 73 74 75
static void *mndBuildTimerMsg(int32_t *pContLen) {
  SMTimerReq timerReq = {0};

  int32_t contLen = tSerializeSMTimerMsg(NULL, 0, &timerReq);
  if (contLen <= 0) return NULL;
  void *pReq = rpcMallocCont(contLen);
  if (pReq == NULL) return NULL;

  tSerializeSMTimerMsg(pReq, contLen, &timerReq);
  *pContLen = contLen;
  return pReq;
}

S
Shengliang Guan 已提交
76 77 78
static void mndTransReExecute(void *param, void *tmrId) {
  SMnode *pMnode = param;
  if (mndIsMaster(pMnode)) {
S
Shengliang Guan 已提交
79 80 81
    int32_t contLen = 0;
    void   *pReq = mndBuildTimerMsg(&contLen);
    SRpcMsg rpcMsg = {.msgType = TDMT_MND_TRANS, .pCont = pReq, .contLen = contLen};
S
Shengliang Guan 已提交
82
    pMnode->putReqToMWriteQFp(pMnode->pDnode, &rpcMsg);
S
Shengliang Guan 已提交
83 84 85 86 87
  }

  taosTmrReset(mndTransReExecute, 3000, pMnode, pMnode->timer, &pMnode->transTimer);
}

L
Liu Jicong 已提交
88 89
static void mndCalMqRebalance(void *param, void *tmrId) {
  SMnode *pMnode = param;
L
Liu Jicong 已提交
90
  if (mndIsMaster(pMnode)) {
S
Shengliang Guan 已提交
91 92 93
    int32_t contLen = 0;
    void   *pReq = mndBuildTimerMsg(&contLen);
    SRpcMsg rpcMsg = {.msgType = TDMT_MND_MQ_TIMER, .pCont = pReq, .contLen = contLen};
L
Liu Jicong 已提交
94
    pMnode->putReqToMReadQFp(pMnode->pDnode, &rpcMsg);
L
Liu Jicong 已提交
95 96
  }

L
Liu Jicong 已提交
97
  taosTmrReset(mndCalMqRebalance, 3000, pMnode, pMnode->timer, &pMnode->mqTimer);
L
Liu Jicong 已提交
98 99
}

100 101 102
static int32_t mndInitTimer(SMnode *pMnode) {
  if (pMnode->timer == NULL) {
    pMnode->timer = taosTmrInit(5000, 200, 3600000, "MND");
S
Shengliang Guan 已提交
103 104
  }

105
  if (pMnode->timer == NULL) {
S
Shengliang Guan 已提交
106
    terrno = TSDB_CODE_OUT_OF_MEMORY;
S
Shengliang Guan 已提交
107 108 109
    return -1;
  }

S
Shengliang Guan 已提交
110
  if (taosTmrReset(mndTransReExecute, 6000, pMnode, pMnode->timer, &pMnode->transTimer)) {
S
Shengliang Guan 已提交
111 112 113 114
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return -1;
  }

L
Liu Jicong 已提交
115 116 117 118 119
  if (taosTmrReset(mndCalMqRebalance, 3000, pMnode, pMnode->timer, &pMnode->mqTimer)) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return -1;
  }

S
Shengliang Guan 已提交
120 121 122
  return 0;
}

123 124
static void mndCleanupTimer(SMnode *pMnode) {
  if (pMnode->timer != NULL) {
S
Shengliang Guan 已提交
125 126
    taosTmrStop(pMnode->transTimer);
    pMnode->transTimer = NULL;
L
Liu Jicong 已提交
127 128
    taosTmrStop(pMnode->mqTimer);
    pMnode->mqTimer = NULL;
129 130
    taosTmrCleanUp(pMnode->timer);
    pMnode->timer = NULL;
S
Shengliang Guan 已提交
131 132 133
  }
}

S
Shengliang Guan 已提交
134
static int32_t mndCreateDir(SMnode *pMnode, const char *path) {
135 136 137
  pMnode->path = strdup(path);
  if (pMnode->path == NULL) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
S
Shengliang Guan 已提交
138
    return -1;
139 140 141 142
  }

  if (taosMkDir(pMnode->path) != 0) {
    terrno = TAOS_SYSTEM_ERROR(errno);
S
Shengliang Guan 已提交
143
    return -1;
144
  }
145 146

  return 0;
147
}
S
Shengliang Guan 已提交
148

149 150 151
static int32_t mndInitSdb(SMnode *pMnode) {
  SSdbOpt opt = {0};
  opt.path = pMnode->path;
S
Shengliang Guan 已提交
152
  opt.pMnode = pMnode;
S
Shengliang Guan 已提交
153

S
Shengliang Guan 已提交
154
  pMnode->pSdb = sdbInit(&opt);
155
  if (pMnode->pSdb == NULL) {
S
Shengliang Guan 已提交
156 157 158 159 160 161
    return -1;
  }

  return 0;
}

162
static int32_t mndDeploySdb(SMnode *pMnode) { return sdbDeploy(pMnode->pSdb); }
S
Shengliang Guan 已提交
163
static int32_t mndReadSdb(SMnode *pMnode) { return sdbReadFile(pMnode->pSdb); }
164 165 166

static void mndCleanupSdb(SMnode *pMnode) {
  if (pMnode->pSdb) {
S
Shengliang Guan 已提交
167
    sdbCleanup(pMnode->pSdb);
168 169 170 171
    pMnode->pSdb = NULL;
  }
}

172 173 174 175 176
static int32_t mndAllocStep(SMnode *pMnode, char *name, MndInitFp initFp, MndCleanupFp cleanupFp) {
  SMnodeStep step = {0};
  step.name = name;
  step.initFp = initFp;
  step.cleanupFp = cleanupFp;
S
Shengliang Guan 已提交
177
  if (taosArrayPush(pMnode->pSteps, &step) == NULL) {
178 179 180 181
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return -1;
  }

S
Shengliang Guan 已提交
182 183 184
  return 0;
}

185
static int32_t mndInitSteps(SMnode *pMnode) {
S
Shengliang Guan 已提交
186 187 188 189
  if (mndAllocStep(pMnode, "mnode-sdb", mndInitSdb, mndCleanupSdb) != 0) return -1;
  if (mndAllocStep(pMnode, "mnode-trans", mndInitTrans, mndCleanupTrans) != 0) return -1;
  if (mndAllocStep(pMnode, "mnode-cluster", mndInitCluster, mndCleanupCluster) != 0) return -1;
  if (mndAllocStep(pMnode, "mnode-mnode", mndInitMnode, mndCleanupMnode) != 0) return -1;
S
Shengliang Guan 已提交
190 191 192
  if (mndAllocStep(pMnode, "mnode-qnode", mndInitQnode, mndCleanupQnode) != 0) return -1;
  if (mndAllocStep(pMnode, "mnode-qnode", mndInitSnode, mndCleanupSnode) != 0) return -1;
  if (mndAllocStep(pMnode, "mnode-qnode", mndInitBnode, mndCleanupBnode) != 0) return -1;
S
Shengliang Guan 已提交
193
  if (mndAllocStep(pMnode, "mnode-dnode", mndInitDnode, mndCleanupDnode) != 0) return -1;
S
Shengliang Guan 已提交
194
  if (mndAllocStep(pMnode, "mnode-user", mndInitUser, mndCleanupUser) != 0) return -1;
S
Shengliang Guan 已提交
195 196 197
  if (mndAllocStep(pMnode, "mnode-auth", mndInitAuth, mndCleanupAuth) != 0) return -1;
  if (mndAllocStep(pMnode, "mnode-acct", mndInitAcct, mndCleanupAcct) != 0) return -1;
  if (mndAllocStep(pMnode, "mnode-topic", mndInitTopic, mndCleanupTopic) != 0) return -1;
L
Liu Jicong 已提交
198 199
  if (mndAllocStep(pMnode, "mnode-consumer", mndInitConsumer, mndCleanupConsumer) != 0) return -1;
  if (mndAllocStep(pMnode, "mnode-subscribe", mndInitSubscribe, mndCleanupSubscribe) != 0) return -1;
S
Shengliang Guan 已提交
200
  if (mndAllocStep(pMnode, "mnode-vgroup", mndInitVgroup, mndCleanupVgroup) != 0) return -1;
S
Shengliang Guan 已提交
201
  if (mndAllocStep(pMnode, "mnode-stb", mndInitStb, mndCleanupStb) != 0) return -1;
S
Shengliang Guan 已提交
202
  if (mndAllocStep(pMnode, "mnode-db", mndInitDb, mndCleanupDb) != 0) return -1;
S
Shengliang Guan 已提交
203
  if (mndAllocStep(pMnode, "mnode-func", mndInitFunc, mndCleanupFunc) != 0) return -1;
204
  if (pMnode->clusterId <= 0) {
S
Shengliang Guan 已提交
205 206 207 208 209 210 211 212 213 214
    if (mndAllocStep(pMnode, "mnode-sdb-deploy", mndDeploySdb, NULL) != 0) return -1;
  } else {
    if (mndAllocStep(pMnode, "mnode-sdb-read", mndReadSdb, NULL) != 0) return -1;
  }
  if (mndAllocStep(pMnode, "mnode-timer", mndInitTimer, NULL) != 0) return -1;
  if (mndAllocStep(pMnode, "mnode-profile", mndInitProfile, mndCleanupProfile) != 0) return -1;
  if (mndAllocStep(pMnode, "mnode-show", mndInitShow, mndCleanupShow) != 0) return -1;
  if (mndAllocStep(pMnode, "mnode-sync", mndInitSync, mndCleanupSync) != 0) return -1;
  if (mndAllocStep(pMnode, "mnode-telem", mndInitTelem, mndCleanupTelem) != 0) return -1;
  if (mndAllocStep(pMnode, "mnode-timer", NULL, mndCleanupTimer) != 0) return -1;
S
Shengliang Guan 已提交
215 216 217 218

  return 0;
}

219
static void mndCleanupSteps(SMnode *pMnode, int32_t pos) {
S
Shengliang Guan 已提交
220 221
  if (pMnode->pSteps == NULL) return;

222
  if (pos == -1) {
223
    pos = taosArrayGetSize(pMnode->pSteps) - 1;
S
Shengliang Guan 已提交
224 225
  }

226
  for (int32_t s = pos; s >= 0; s--) {
227
    SMnodeStep *pStep = taosArrayGet(pMnode->pSteps, s);
S
Shengliang Guan 已提交
228
    mDebug("%s will cleanup", pStep->name);
229 230 231
    if (pStep->cleanupFp != NULL) {
      (*pStep->cleanupFp)(pMnode);
    }
S
Shengliang Guan 已提交
232 233
  }

S
Shengliang Guan 已提交
234
  taosArrayClear(pMnode->pSteps);
S
Shengliang Guan 已提交
235
  taosArrayDestroy(pMnode->pSteps);
S
Shengliang Guan 已提交
236
  pMnode->pSteps = NULL;
237
}
S
Shengliang Guan 已提交
238

239
static int32_t mndExecSteps(SMnode *pMnode) {
S
Shengliang Guan 已提交
240
  int32_t size = taosArrayGetSize(pMnode->pSteps);
241
  for (int32_t pos = 0; pos < size; pos++) {
S
Shengliang Guan 已提交
242
    SMnodeStep *pStep = taosArrayGet(pMnode->pSteps, pos);
243
    if (pStep->initFp == NULL) continue;
S
Shengliang Guan 已提交
244

S
Shengliang Guan 已提交
245
    if ((*pStep->initFp)(pMnode) != 0) {
S
Shengliang Guan 已提交
246
      int32_t code = terrno;
S
Shengliang Guan 已提交
247
      mError("%s exec failed since %s, start to cleanup", pStep->name, terrstr());
248
      mndCleanupSteps(pMnode, pos);
S
Shengliang Guan 已提交
249
      terrno = code;
S
Shengliang Guan 已提交
250
      return -1;
S
Shengliang Guan 已提交
251
    } else {
S
Shengliang Guan 已提交
252
      mDebug("%s is initialized", pStep->name);
S
Shengliang Guan 已提交
253 254
    }
  }
S
Shengliang Guan 已提交
255 256

  return 0;
257
}
S
Shengliang Guan 已提交
258

259 260 261 262 263 264 265
static int32_t mndSetOptions(SMnode *pMnode, const SMnodeOpt *pOption) {
  pMnode->dnodeId = pOption->dnodeId;
  pMnode->clusterId = pOption->clusterId;
  pMnode->replica = pOption->replica;
  pMnode->selfIndex = pOption->selfIndex;
  memcpy(&pMnode->replicas, pOption->replicas, sizeof(SReplica) * TSDB_MAX_REPLICA);
  pMnode->pDnode = pOption->pDnode;
S
Shengliang Guan 已提交
266
  pMnode->putReqToMWriteQFp = pOption->putReqToMWriteQFp;
L
Liu Jicong 已提交
267
  pMnode->putReqToMReadQFp = pOption->putReqToMReadQFp;
S
Shengliang Guan 已提交
268 269 270
  pMnode->sendReqToDnodeFp = pOption->sendReqToDnodeFp;
  pMnode->sendReqToMnodeFp = pOption->sendReqToMnodeFp;
  pMnode->sendRedirectRspFp = pOption->sendRedirectRspFp;
S
Shengliang Guan 已提交
271 272 273 274 275 276 277 278 279
  pMnode->cfg.sver = pOption->cfg.sver;
  pMnode->cfg.enableTelem = pOption->cfg.enableTelem;
  pMnode->cfg.statusInterval = pOption->cfg.statusInterval;
  pMnode->cfg.shellActivityTimer = pOption->cfg.shellActivityTimer;
  pMnode->cfg.timezone = strdup(pOption->cfg.timezone);
  pMnode->cfg.locale = strdup(pOption->cfg.locale);
  pMnode->cfg.charset = strdup(pOption->cfg.charset);
  pMnode->cfg.gitinfo = strdup(pOption->cfg.gitinfo);
  pMnode->cfg.buildinfo = strdup(pOption->cfg.buildinfo);
280

S
Shengliang Guan 已提交
281 282 283
  if (pMnode->sendReqToDnodeFp == NULL || pMnode->sendReqToMnodeFp == NULL || pMnode->sendRedirectRspFp == NULL ||
      pMnode->putReqToMWriteQFp == NULL || pMnode->dnodeId < 0 || pMnode->clusterId < 0 ||
      pMnode->cfg.statusInterval < 1) {
S
Shengliang Guan 已提交
284
    terrno = TSDB_CODE_MND_INVALID_OPTIONS;
S
Shengliang Guan 已提交
285 286 287
    return -1;
  }

S
Shengliang Guan 已提交
288
  if (pMnode->cfg.timezone == NULL || pMnode->cfg.locale == NULL || pMnode->cfg.charset == NULL) {
S
Shengliang Guan 已提交
289
    terrno = TSDB_CODE_OUT_OF_MEMORY;
S
Shengliang Guan 已提交
290
    return -1;
291 292 293
  }

  return 0;
L
Liu Jicong 已提交
294
}
295

296
SMnode *mndOpen(const char *path, const SMnodeOpt *pOption) {
S
Shengliang Guan 已提交
297 298
  mDebug("start to open mnode in %s", path);

299
  SMnode *pMnode = calloc(1, sizeof(SMnode));
S
Shengliang Guan 已提交
300 301 302 303 304 305
  if (pMnode == NULL) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    mError("failed to open mnode since %s", terrstr());
    return NULL;
  }

S
Shengliang Guan 已提交
306 307 308
  char timestr[24] = "1970-01-01 00:00:00.00";
  (void)taosParseTime(timestr, &pMnode->checkTime, (int32_t)strlen(timestr), TSDB_TIME_PRECISION_MILLI, 0);

S
Shengliang Guan 已提交
309 310 311 312 313 314 315
  pMnode->pSteps = taosArrayInit(24, sizeof(SMnodeStep));
  if (pMnode->pSteps == NULL) {
    free(pMnode);
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    mError("failed to open mnode since %s", terrstr());
    return NULL;
  }
316

S
Shengliang Guan 已提交
317
  int32_t code = mndCreateDir(pMnode, path);
318
  if (code != 0) {
S
Shengliang Guan 已提交
319 320
    code = terrno;
    mError("failed to open mnode since %s", terrstr());
321 322 323 324 325 326
    mndClose(pMnode);
    terrno = code;
    return NULL;
  }

  code = mndSetOptions(pMnode, pOption);
327
  if (code != 0) {
S
Shengliang Guan 已提交
328 329
    code = terrno;
    mError("failed to open mnode since %s", terrstr());
330 331 332 333 334 335 336
    mndClose(pMnode);
    terrno = code;
    return NULL;
  }

  code = mndInitSteps(pMnode);
  if (code != 0) {
S
Shengliang Guan 已提交
337 338
    code = terrno;
    mError("failed to open mnode since %s", terrstr());
339 340 341 342 343 344 345
    mndClose(pMnode);
    terrno = code;
    return NULL;
  }

  code = mndExecSteps(pMnode);
  if (code != 0) {
S
Shengliang Guan 已提交
346 347
    code = terrno;
    mError("failed to open mnode since %s", terrstr());
348 349 350 351
    mndClose(pMnode);
    terrno = code;
    return NULL;
  }
S
Shengliang Guan 已提交
352

S
Shengliang Guan 已提交
353
  mDebug("mnode open successfully ");
S
Shengliang Guan 已提交
354 355
  return pMnode;
}
S
Shengliang Guan 已提交
356

357
void mndClose(SMnode *pMnode) {
S
Shengliang Guan 已提交
358 359 360 361
  if (pMnode != NULL) {
    mDebug("start to close mnode");
    mndCleanupSteps(pMnode, -1);
    tfree(pMnode->path);
S
Shengliang Guan 已提交
362 363 364 365 366
    tfree(pMnode->cfg.charset);
    tfree(pMnode->cfg.locale);
    tfree(pMnode->cfg.timezone);
    tfree(pMnode->cfg.gitinfo);
    tfree(pMnode->cfg.buildinfo);
S
Shengliang Guan 已提交
367 368 369
    tfree(pMnode);
    mDebug("mnode is closed");
  }
370
}
S
Shengliang Guan 已提交
371

372
int32_t mndAlter(SMnode *pMnode, const SMnodeOpt *pOption) {
S
Shengliang Guan 已提交
373 374
  mDebug("start to alter mnode");
  mDebug("mnode is altered");
375 376
  return 0;
}
S
Shengliang Guan 已提交
377

378
void mndDestroy(const char *path) {
S
Shengliang Guan 已提交
379
  mDebug("start to destroy mnode at %s", path);
S
Shengliang Guan 已提交
380
  taosRemoveDir(path);
S
Shengliang Guan 已提交
381
  mDebug("mnode is destroyed");
382
}
S
Shengliang Guan 已提交
383

384
int32_t mndGetLoad(SMnode *pMnode, SMnodeLoad *pLoad) {
S
Shengliang Guan 已提交
385 386 387 388 389 390 391 392 393 394 395
  pLoad->numOfDnode = 0;
  pLoad->numOfMnode = 0;
  pLoad->numOfVgroup = 0;
  pLoad->numOfDatabase = 0;
  pLoad->numOfSuperTable = 0;
  pLoad->numOfChildTable = 0;
  pLoad->numOfColumn = 0;
  pLoad->totalPoints = 0;
  pLoad->totalStorage = 0;
  pLoad->compStorage = 0;

396 397
  return 0;
}
S
Shengliang Guan 已提交
398

399
SMnodeMsg *mndInitMsg(SMnode *pMnode, SRpcMsg *pRpcMsg) {
S
Shengliang Guan 已提交
400 401 402
  SMnodeMsg *pMsg = taosAllocateQitem(sizeof(SMnodeMsg));
  if (pMsg == NULL) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
S
Shengliang Guan 已提交
403
    mError("failed to create msg since %s, app:%p RPC:%p", terrstr(), pRpcMsg->ahandle, pRpcMsg->handle);
S
Shengliang Guan 已提交
404 405 406
    return NULL;
  }

L
Liu Jicong 已提交
407
  if (pRpcMsg->msgType != TDMT_MND_TRANS && pRpcMsg->msgType != TDMT_MND_MQ_TIMER && pRpcMsg->msgType != TDMT_MND_MQ_DO_REBALANCE) {
408 409 410 411 412 413 414 415
    SRpcConnInfo connInfo = {0};
    if ((pRpcMsg->msgType & 1U) && rpcGetConnInfo(pRpcMsg->handle, &connInfo) != 0) {
      taosFreeQitem(pMsg);
      terrno = TSDB_CODE_MND_NO_USER_FROM_CONN;
      mError("failed to create msg since %s, app:%p RPC:%p", terrstr(), pRpcMsg->ahandle, pRpcMsg->handle);
      return NULL;
    }
    memcpy(pMsg->user, connInfo.user, TSDB_USER_LEN);
S
Shengliang Guan 已提交
416 417
  }

S
Shengliang Guan 已提交
418
  pMsg->pMnode = pMnode;
S
Shengliang Guan 已提交
419 420 421
  pMsg->rpcMsg = *pRpcMsg;
  pMsg->createdTime = taosGetTimestampSec();

S
Shengliang Guan 已提交
422
  mTrace("msg:%p, is created, app:%p RPC:%p user:%s", pMsg, pRpcMsg->ahandle, pRpcMsg->handle, pMsg->user);
S
Shengliang Guan 已提交
423 424 425
  return pMsg;
}

426
void mndCleanupMsg(SMnodeMsg *pMsg) {
S
Shengliang Guan 已提交
427
  mTrace("msg:%p, is destroyed, app:%p RPC:%p", pMsg, pMsg->rpcMsg.ahandle, pMsg->rpcMsg.handle);
S
Shengliang Guan 已提交
428 429
  rpcFreeCont(pMsg->rpcMsg.pCont);
  pMsg->rpcMsg.pCont = NULL;
S
Shengliang Guan 已提交
430 431 432
  taosFreeQitem(pMsg);
}

433 434 435 436
void mndSendRsp(SMnodeMsg *pMsg, int32_t code) {
  SRpcMsg rpcRsp = {.handle = pMsg->rpcMsg.handle, .code = code};
  rpcSendResponse(&rpcRsp);
}
S
Shengliang Guan 已提交
437

S
Shengliang Guan 已提交
438
void mndProcessMsg(SMnodeMsg *pMsg) {
439
  SMnode *pMnode = pMsg->pMnode;
S
Shengliang Guan 已提交
440
  int32_t code = 0;
441
  tmsg_t  msgType = pMsg->rpcMsg.msgType;
S
Shengliang Guan 已提交
442
  void   *ahandle = pMsg->rpcMsg.ahandle;
S
Shengliang Guan 已提交
443
  bool    isReq = (msgType & 1U);
444

S
Shengliang Guan 已提交
445
  mTrace("msg:%p, type:%s will be processed, app:%p", pMsg, TMSG_INFO(msgType), ahandle);
S
Shengliang Guan 已提交
446

S
Shengliang Guan 已提交
447 448
  if (isReq && !mndIsMaster(pMnode)) {
    code = TSDB_CODE_APP_NOT_READY;
S
Shengliang Guan 已提交
449
    mDebug("msg:%p, failed to process since %s, app:%p", pMsg, terrstr(), ahandle);
S
Shengliang Guan 已提交
450 451 452 453 454
    goto PROCESS_RPC_END;
  }

  if (isReq && pMsg->rpcMsg.pCont == NULL) {
    code = TSDB_CODE_MND_INVALID_MSG_LEN;
S
Shengliang Guan 已提交
455
    mError("msg:%p, failed to process since %s, app:%p", pMsg, terrstr(), ahandle);
S
Shengliang Guan 已提交
456
    goto PROCESS_RPC_END;
S
Shengliang Guan 已提交
457 458
  }

459
  MndMsgFp fp = pMnode->msgFp[TMSG_INDEX(msgType)];
S
Shengliang Guan 已提交
460
  if (fp == NULL) {
S
Shengliang Guan 已提交
461
    code = TSDB_CODE_MSG_NOT_PROCESSED;
S
Shengliang Guan 已提交
462
    mError("msg:%p, failed to process since no msg handle, app:%p", pMsg, ahandle);
S
Shengliang Guan 已提交
463
    goto PROCESS_RPC_END;
S
Shengliang Guan 已提交
464 465
  }

466
  code = (*fp)(pMsg);
S
Shengliang Guan 已提交
467
  if (code == TSDB_CODE_MND_ACTION_IN_PROGRESS) {
S
Shengliang Guan 已提交
468
    mTrace("msg:%p, in progress, app:%p", pMsg, ahandle);
S
Shengliang Guan 已提交
469 470
    return;
  } else if (code != 0) {
S
Shengliang Guan 已提交
471
    code = terrno;
S
Shengliang Guan 已提交
472
    mError("msg:%p, failed to process since %s, app:%p", pMsg, terrstr(), ahandle);
S
Shengliang Guan 已提交
473
    goto PROCESS_RPC_END;
474
  } else {
S
Shengliang Guan 已提交
475
    mTrace("msg:%p, is processed, app:%p", pMsg, ahandle);
S
Shengliang Guan 已提交
476 477 478 479
  }

PROCESS_RPC_END:
  if (isReq) {
S
Shengliang Guan 已提交
480 481
    if (pMsg->rpcMsg.handle == NULL) return;

S
Shengliang Guan 已提交
482
    if (code == TSDB_CODE_APP_NOT_READY) {
S
Shengliang Guan 已提交
483
      mndSendRedirectRsp(pMnode, &pMsg->rpcMsg);
S
Shengliang Guan 已提交
484
    } else if (code != 0) {
S
Shengliang Guan 已提交
485 486
      SRpcMsg rpcRsp = {.handle = pMsg->rpcMsg.handle, .code = code};
      rpcSendResponse(&rpcRsp);
S
Shengliang Guan 已提交
487
    } else {
S
Shengliang Guan 已提交
488 489
      SRpcMsg rpcRsp = {.handle = pMsg->rpcMsg.handle, .contLen = pMsg->contLen, .pCont = pMsg->pCont};
      rpcSendResponse(&rpcRsp);
S
Shengliang Guan 已提交
490
    }
S
Shengliang Guan 已提交
491
  }
S
Shengliang Guan 已提交
492 493
}

494 495 496 497
void mndSetMsgHandle(SMnode *pMnode, tmsg_t msgType, MndMsgFp fp) {
  tmsg_t type = TMSG_INDEX(msgType);
  if (type >= 0 && type < TDMT_MAX) {
    pMnode->msgFp[type] = fp;
S
Shengliang Guan 已提交
498 499 500
  }
}

S
Shengliang Guan 已提交
501 502 503 504 505
uint64_t mndGenerateUid(char *name, int32_t len) {
  int64_t  us = taosGetTimestampUs();
  int32_t  hashval = MurmurHash3_32(name, len);
  uint64_t x = (us & 0x000000FFFFFFFFFF) << 24;
  return x + ((hashval & ((1ul << 16) - 1ul)) << 8) + (taosRand() & ((1ul << 8) - 1ul));
L
Liu Jicong 已提交
506
}