taosdata / TDengine / #3621

coverage: 2.037% (-61.5%) from 63.573%
#3621

push

travis-ci

web-flow 
Merge pull request #29874 from taosdata/merge/mainto3.0

merge: from main to 3.0 branch

4357 of 287032 branches covered (1.52%)

Branch coverage included in aggregate %.

0 of 174 new or added lines in 18 files covered. (0.0%)

213359 existing lines in 469 files now uncovered.

7260 of 283369 relevant lines covered (2.56%)

23737.72 hits per line

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

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









16



#define _DEFAULT_SOURCE










17



#include "mndAcct.h"










18



#include "mndAnode.h"










19



#include "mndArbGroup.h"










20



#include "mndCluster.h"










21



#include "mndCompact.h"










22



#include "mndCompactDetail.h"










23



#include "mndConfig.h"










24



#include "mndConsumer.h"










25



#include "mndDb.h"










26



#include "mndDnode.h"










27



#include "mndFunc.h"










28



#include "mndGrant.h"










29



#include "mndIndex.h"










30



#include "mndInfoSchema.h"










31



#include "mndMnode.h"










32



#include "mndPerfSchema.h"










33



#include "mndPrivilege.h"










34



#include "mndProfile.h"










35



#include "mndQnode.h"










36



#include "mndQuery.h"










37



#include "mndShow.h"










38



#include "mndSma.h"










39



#include "mndSnode.h"










40



#include "mndStb.h"










41



#include "mndStream.h"










42



#include "mndSubscribe.h"










43



#include "mndSync.h"










44



#include "mndTelem.h"










45



#include "mndTopic.h"










46



#include "mndTrans.h"










47



#include "mndUser.h"










48



#include "mndVgroup.h"










49



#include "mndView.h"










50













UNCOV

51



static inline int32_t mndAcquireRpc(SMnode *pMnode) {




×








UNCOV

52



  int32_t code = 0;




×








UNCOV

53



  (void)taosThreadRwlockRdlock(&pMnode->lock);




×








UNCOV

54



  if (pMnode->stopped) {





×







55



    code = TSDB_CODE_APP_IS_STOPPING;




×








UNCOV

56



  } else if (!mndIsLeader(pMnode)) {





×








UNCOV

57



    code = 1;




×







58



  } else {










59



#if 1











UNCOV

60



    (void)atomic_add_fetch_32(&pMnode->rpcRef, 1);




×







61



#else










62



    int32_t ref = atomic_add_fetch_32(&pMnode->rpcRef, 1);










63



    mTrace("mnode rpc is acquired, ref:%d", ref);










64



#endif










65



  }











UNCOV

66



  (void)taosThreadRwlockUnlock(&pMnode->lock);




×








UNCOV

67



  TAOS_RETURN(code);




×







68



}










69













UNCOV

70



static inline void mndReleaseRpc(SMnode *pMnode) {




×








UNCOV

71



  (void)taosThreadRwlockRdlock(&pMnode->lock);




×







72



#if 1











UNCOV

73



  (void)atomic_sub_fetch_32(&pMnode->rpcRef, 1);




×







74



#else










75



  int32_t ref = atomic_sub_fetch_32(&pMnode->rpcRef, 1);










76



  mTrace("mnode rpc is released, ref:%d", ref);










77



#endif











UNCOV

78



  (void)taosThreadRwlockUnlock(&pMnode->lock);




×








UNCOV

79



}




×







80













UNCOV

81



static void *mndBuildTimerMsg(int32_t *pContLen) {




×








UNCOV

82



  terrno = 0;




×








UNCOV

83



  SMTimerReq timerReq = {0};




×







84













UNCOV

85



  int32_t contLen = tSerializeSMTimerMsg(NULL, 0, &timerReq);




×








UNCOV

86



  if (contLen <= 0) return NULL;





×








UNCOV

87



  void *pReq = rpcMallocCont(contLen);




×








UNCOV

88



  if (pReq == NULL) return NULL;





×







89













UNCOV

90



  if (tSerializeSMTimerMsg(pReq, contLen, &timerReq) < 0) {





×







91



    mError("failed to serialize timer msg since %s", terrstr());





×







92



  }











UNCOV

93



  *pContLen = contLen;




×








UNCOV

94



  return pReq;




×







95



}










96













UNCOV

97



static void mndPullupTrans(SMnode *pMnode) {




×








UNCOV

98



  mTrace("pullup trans msg");





×








UNCOV

99



  int32_t contLen = 0;




×








UNCOV

100



  void   *pReq = mndBuildTimerMsg(&contLen);




×








UNCOV

101



  if (pReq != NULL) {





×








UNCOV

102



    SRpcMsg rpcMsg = {.msgType = TDMT_MND_TRANS_TIMER, .pCont = pReq, .contLen = contLen};




×







103



    // TODO check return value











UNCOV

104



    if (tmsgPutToQueue(&pMnode->msgCb, WRITE_QUEUE, &rpcMsg) < 0) {





×








UNCOV

105



      mError("failed to put into write-queue since %s, line:%d", terrstr(), __LINE__);





×







106



    }










107



  }











UNCOV

108



}




×







109













UNCOV

110



static void mndPullupCompacts(SMnode *pMnode) {




×








UNCOV

111



  mTrace("pullup compact timer msg");





×








UNCOV

112



  int32_t contLen = 0;




×








UNCOV

113



  void   *pReq = mndBuildTimerMsg(&contLen);




×








UNCOV

114



  if (pReq != NULL) {





×








UNCOV

115



    SRpcMsg rpcMsg = {.msgType = TDMT_MND_COMPACT_TIMER, .pCont = pReq, .contLen = contLen};




×







116



    // TODO check return value











UNCOV

117



    if (tmsgPutToQueue(&pMnode->msgCb, WRITE_QUEUE, &rpcMsg) < 0) {





×







118



      mError("failed to put into write-queue since %s, line:%d", terrstr(), __LINE__);





×







119



    }










120



  }











UNCOV

121



}




×







122













UNCOV

123



static void mndPullupTtl(SMnode *pMnode) {




×








UNCOV

124



  mTrace("pullup ttl");





×








UNCOV

125



  int32_t contLen = 0;




×








UNCOV

126



  void   *pReq = mndBuildTimerMsg(&contLen);




×








UNCOV

127



  SRpcMsg rpcMsg = {.msgType = TDMT_MND_TTL_TIMER, .pCont = pReq, .contLen = contLen};




×







128



  // TODO check return value











UNCOV

129



  if (tmsgPutToQueue(&pMnode->msgCb, WRITE_QUEUE, &rpcMsg) < 0) {





×







130



    mError("failed to put into write-queue since %s, line:%d", terrstr(), __LINE__);





×







131



  }











UNCOV

132



}




×







133













UNCOV

134



static void mndPullupTrimDb(SMnode *pMnode) {




×








UNCOV

135



  mTrace("pullup s3migrate");





×








UNCOV

136



  int32_t contLen = 0;




×








UNCOV

137



  void   *pReq = mndBuildTimerMsg(&contLen);




×








UNCOV

138



  SRpcMsg rpcMsg = {.msgType = TDMT_MND_TRIM_DB_TIMER, .pCont = pReq, .contLen = contLen};




×







139



  // TODO check return value











UNCOV

140



  if (tmsgPutToQueue(&pMnode->msgCb, WRITE_QUEUE, &rpcMsg) < 0) {





×







141



    mError("failed to put into write-queue since %s, line:%d", terrstr(), __LINE__);





×







142



  }











UNCOV

143



}




×







144












145



static void mndPullupS3MigrateDb(SMnode *pMnode) {




×







146



  mTrace("pullup trim");





×







147



  int32_t contLen = 0;




×







148



  void   *pReq = mndBuildTimerMsg(&contLen);




×







149



  // TODO check return value










150



  SRpcMsg rpcMsg = {.msgType = TDMT_MND_S3MIGRATE_DB_TIMER, .pCont = pReq, .contLen = contLen};




×







151



  if (tmsgPutToQueue(&pMnode->msgCb, WRITE_QUEUE, &rpcMsg) < 0) {





×







152



    mError("failed to put into write-queue since %s, line:%d", terrstr(), __LINE__);





×







153



  }










154



}




×







155













UNCOV

156



static int32_t mndPullupArbHeartbeat(SMnode *pMnode) {




×








UNCOV

157



  mTrace("pullup arb hb");





×








UNCOV

158



  int32_t contLen = 0;




×








UNCOV

159



  void   *pReq = mndBuildTimerMsg(&contLen);




×








UNCOV

160



  SRpcMsg rpcMsg = {.msgType = TDMT_MND_ARB_HEARTBEAT_TIMER, .pCont = pReq, .contLen = contLen, .info.noResp = 1};




×








UNCOV

161



  return tmsgPutToQueue(&pMnode->msgCb, ARB_QUEUE, &rpcMsg);




×







162



}










163













UNCOV

164



static int32_t mndPullupArbCheckSync(SMnode *pMnode) {




×








UNCOV

165



  mTrace("pullup arb sync");





×








UNCOV

166



  int32_t contLen = 0;




×








UNCOV

167



  void   *pReq = mndBuildTimerMsg(&contLen);




×








UNCOV

168



  SRpcMsg rpcMsg = {.msgType = TDMT_MND_ARB_CHECK_SYNC_TIMER, .pCont = pReq, .contLen = contLen, .info.noResp = 1};




×








UNCOV

169



  return tmsgPutToQueue(&pMnode->msgCb, ARB_QUEUE, &rpcMsg);




×







170



}










171













UNCOV

172



static void mndCalMqRebalance(SMnode *pMnode) {




×








UNCOV

173



  int32_t contLen = 0;




×








UNCOV

174



  void   *pReq = mndBuildTimerMsg(&contLen);




×








UNCOV

175



  if (pReq != NULL) {





×








UNCOV

176



    SRpcMsg rpcMsg = {.msgType = TDMT_MND_TMQ_TIMER, .pCont = pReq, .contLen = contLen};




×








UNCOV

177



    if (tmsgPutToQueue(&pMnode->msgCb, WRITE_QUEUE, &rpcMsg) < 0) {





×








UNCOV

178



      mError("failed to put into write-queue since %s, line:%d", terrstr(), __LINE__);





×







179



    }










180



  }











UNCOV

181



}




×







182













UNCOV

183



static void mndStreamCheckpointTimer(SMnode *pMnode) {




×








UNCOV

184



  SMStreamDoCheckpointMsg *pMsg = rpcMallocCont(sizeof(SMStreamDoCheckpointMsg));




×








UNCOV

185



  if (pMsg != NULL) {





×








UNCOV

186



    int32_t size = sizeof(SMStreamDoCheckpointMsg);




×








UNCOV

187



    SRpcMsg rpcMsg = {.msgType = TDMT_MND_STREAM_BEGIN_CHECKPOINT, .pCont = pMsg, .contLen = size};




×







188



    // TODO check return value











UNCOV

189



    if (tmsgPutToQueue(&pMnode->msgCb, WRITE_QUEUE, &rpcMsg) < 0) {





×







190



      mError("failed to put into write-queue since %s, line:%d", terrstr(), __LINE__);





×







191



    }










192



  }











UNCOV

193



}




×







194













UNCOV

195



static void mndStreamCheckNode(SMnode *pMnode) {




×








UNCOV

196



  int32_t contLen = 0;




×








UNCOV

197



  void   *pReq = mndBuildTimerMsg(&contLen);




×








UNCOV

198



  if (pReq != NULL) {





×








UNCOV

199



    SRpcMsg rpcMsg = {.msgType = TDMT_MND_NODECHECK_TIMER, .pCont = pReq, .contLen = contLen};




×







200



    // TODO check return value











UNCOV

201



    if (tmsgPutToQueue(&pMnode->msgCb, READ_QUEUE, &rpcMsg) < 0) {





×







202



      mError("failed to put into read-queue since %s, line:%d", terrstr(), __LINE__);





×







203



    }










204



  }











UNCOV

205



}




×







206













UNCOV

207



static void mndStreamConsensusChkpt(SMnode *pMnode) {




×








UNCOV

208



  int32_t contLen = 0;




×








UNCOV

209



  void   *pReq = mndBuildTimerMsg(&contLen);




×








UNCOV

210



  if (pReq != NULL) {





×








UNCOV

211



    SRpcMsg rpcMsg = {.msgType = TDMT_MND_STREAM_CONSEN_TIMER, .pCont = pReq, .contLen = contLen};




×







212



    // TODO check return value











UNCOV

213



    if (tmsgPutToQueue(&pMnode->msgCb, WRITE_QUEUE, &rpcMsg) < 0) {





×







214



      mError("failed to put into write-queue since %s, line:%d", terrstr(), __LINE__);





×







215



    }










216



  }











UNCOV

217



}




×







218













UNCOV

219



static void mndPullupTelem(SMnode *pMnode) {




×








UNCOV

220



  mTrace("pullup telem msg");





×








UNCOV

221



  int32_t contLen = 0;




×








UNCOV

222



  void   *pReq = mndBuildTimerMsg(&contLen);




×








UNCOV

223



  if (pReq != NULL) {





×








UNCOV

224



    SRpcMsg rpcMsg = {.msgType = TDMT_MND_TELEM_TIMER, .pCont = pReq, .contLen = contLen};




×







225



    // TODO check return value











UNCOV

226



    if (tmsgPutToQueue(&pMnode->msgCb, READ_QUEUE, &rpcMsg) < 0) {





×







227



      mError("failed to put into read-queue since %s, line:%d", terrstr(), __LINE__);





×







228



    }










229



  }











UNCOV

230



}




×







231













UNCOV

232



static void mndPullupGrant(SMnode *pMnode) {




×








UNCOV

233



  mTrace("pullup grant msg");





×








UNCOV

234



  int32_t contLen = 0;




×








UNCOV

235



  void   *pReq = mndBuildTimerMsg(&contLen);




×








UNCOV

236



  if (pReq != NULL) {





×








UNCOV

237



    SRpcMsg rpcMsg = {.msgType = TDMT_MND_GRANT_HB_TIMER,




×







238



                      .pCont = pReq,










239



                      .contLen = contLen,










240



                      .info.notFreeAhandle = 1,










241



                      .info.ahandle = 0};










242



    // TODO check return value











UNCOV

243



    if (tmsgPutToQueue(&pMnode->msgCb, WRITE_QUEUE, &rpcMsg) < 0) {





×







244



      mError("failed to put into write-queue since %s, line:%d", terrstr(), __LINE__);





×







245



    }










246



  }











UNCOV

247



}




×







248













UNCOV

249



static void mndIncreaseUpTime(SMnode *pMnode) {




×








UNCOV

250



  mTrace("increate uptime");





×








UNCOV

251



  int32_t contLen = 0;




×








UNCOV

252



  void   *pReq = mndBuildTimerMsg(&contLen);




×








UNCOV

253



  if (pReq != NULL) {





×








UNCOV

254



    SRpcMsg rpcMsg = {.msgType = TDMT_MND_UPTIME_TIMER,




×







255



                      .pCont = pReq,










256



                      .contLen = contLen,










257



                      .info.notFreeAhandle = 1,










258



                      .info.ahandle = 0};










259



    // TODO check return value











UNCOV

260



    if (tmsgPutToQueue(&pMnode->msgCb, WRITE_QUEUE, &rpcMsg) < 0) {





×







261



      mError("failed to put into write-queue since %s, line:%d", terrstr(), __LINE__);





×







262



    }










263



  }











UNCOV

264



}




×







265













UNCOV

266



static void mndSetVgroupOffline(SMnode *pMnode, int32_t dnodeId, int64_t curMs) {




×








UNCOV

267



  SSdb *pSdb = pMnode->pSdb;




×







268













UNCOV

269



  void *pIter = NULL;




×








UNCOV

270



  while (1) {




×








UNCOV

271



    SVgObj *pVgroup = NULL;




×








UNCOV

272



    pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);




×








UNCOV

273



    if (pIter == NULL) break;





×







274













UNCOV

275



    bool stateChanged = false;




×








UNCOV

276



    for (int32_t vg = 0; vg < pVgroup->replica; ++vg) {





×








UNCOV

277



      SVnodeGid *pGid = &pVgroup->vnodeGid[vg];




×








UNCOV

278



      if (pGid->dnodeId == dnodeId) {





×








UNCOV

279



        if (pGid->syncState != TAOS_SYNC_STATE_OFFLINE) {





×








UNCOV

280



          mInfo(





×







281



              "vgId:%d, state changed by offline check, old state:%s restored:%d canRead:%d new state:error restored:0 "










282



              "canRead:0",










283



              pVgroup->vgId, syncStr(pGid->syncState), pGid->syncRestore, pGid->syncCanRead);











UNCOV

284



          pGid->syncState = TAOS_SYNC_STATE_OFFLINE;




×








UNCOV

285



          pGid->syncRestore = 0;




×








UNCOV

286



          pGid->syncCanRead = 0;




×








UNCOV

287



          pGid->startTimeMs = 0;




×








UNCOV

288



          stateChanged = true;




×







289



        }











UNCOV

290



        break;




×







291



      }










292



    }










293













UNCOV

294



    if (stateChanged) {





×








UNCOV

295



      SDbObj *pDb = mndAcquireDb(pMnode, pVgroup->dbName);




×








UNCOV

296



      if (pDb != NULL && pDb->stateTs != curMs) {





×








UNCOV

297



        mInfo("db:%s, stateTs changed by offline check, old newTs:%" PRId64 " newTs:%" PRId64, pDb->name, pDb->stateTs,





×







298



              curMs);











UNCOV

299



        pDb->stateTs = curMs;




×







300



      }











UNCOV

301



      mndReleaseDb(pMnode, pDb);




×







302



    }










303













UNCOV

304



    sdbRelease(pSdb, pVgroup);




×







305



  }











UNCOV

306



}




×







307













UNCOV

308



static void mndCheckDnodeOffline(SMnode *pMnode) {




×








UNCOV

309



  mTrace("check dnode offline");





×








UNCOV

310



  if (mndAcquireRpc(pMnode) != 0) return;





×







311













UNCOV

312



  SSdb   *pSdb = pMnode->pSdb;




×








UNCOV

313



  int64_t curMs = taosGetTimestampMs();




×







314













UNCOV

315



  void *pIter = NULL;




×








UNCOV

316



  while (1) {




×








UNCOV

317



    SDnodeObj *pDnode = NULL;




×








UNCOV

318



    pIter = sdbFetch(pSdb, SDB_DNODE, pIter, (void **)&pDnode);




×








UNCOV

319



    if (pIter == NULL) break;





×







320













UNCOV

321



    bool online = mndIsDnodeOnline(pDnode, curMs);




×








UNCOV

322



    if (!online) {





×








UNCOV

323



      mInfo("dnode:%d, in offline state", pDnode->id);





×








UNCOV

324



      mndSetVgroupOffline(pMnode, pDnode->id, curMs);




×







325



    }










326













UNCOV

327



    sdbRelease(pSdb, pDnode);




×







328



  }










329













UNCOV

330



  mndReleaseRpc(pMnode);




×







331



}










332













UNCOV

333



static bool mnodeIsNotLeader(SMnode *pMnode) {




×








UNCOV

334



  terrno = 0;




×








UNCOV

335



  (void)taosThreadRwlockRdlock(&pMnode->lock);




×








UNCOV

336



  SSyncState state = syncGetState(pMnode->syncMgmt.sync);




×








UNCOV

337



  if (terrno != 0) {





×







338



    (void)taosThreadRwlockUnlock(&pMnode->lock);




×







339



    return true;




×







340



  }










341













UNCOV

342



  if (state.state != TAOS_SYNC_STATE_LEADER) {





×








UNCOV

343



    (void)taosThreadRwlockUnlock(&pMnode->lock);




×








UNCOV

344



    terrno = TSDB_CODE_SYN_NOT_LEADER;




×








UNCOV

345



    return true;




×







346



  }











UNCOV

347



  if (!state.restored || !pMnode->restored) {





×








UNCOV

348



    (void)taosThreadRwlockUnlock(&pMnode->lock);




×








UNCOV

349



    terrno = TSDB_CODE_SYN_RESTORING;




×








UNCOV

350



    return true;




×







351



  }











UNCOV

352



  (void)taosThreadRwlockUnlock(&pMnode->lock);




×








UNCOV

353



  return false;




×







354



}










355













UNCOV

356



static int32_t minCronTime() {




×








UNCOV

357



  int32_t min = INT32_MAX;




×








UNCOV

358



  min = TMIN(min, tsTtlPushIntervalSec);




×








UNCOV

359



  min = TMIN(min, tsTrimVDbIntervalSec);




×








UNCOV

360



  min = TMIN(min, tsS3MigrateIntervalSec);




×








UNCOV

361



  min = TMIN(min, tsTransPullupInterval);




×








UNCOV

362



  min = TMIN(min, tsCompactPullupInterval);




×








UNCOV

363



  min = TMIN(min, tsMqRebalanceInterval);




×








UNCOV

364



  min = TMIN(min, tsStreamCheckpointInterval);




×








UNCOV

365



  min = TMIN(min, tsStreamNodeCheckInterval);




×








UNCOV

366



  min = TMIN(min, tsArbHeartBeatIntervalSec);




×








UNCOV

367



  min = TMIN(min, tsArbCheckSyncIntervalSec);




×







368













UNCOV

369



  int64_t telemInt = TMIN(60, (tsTelemInterval - 1));




×








UNCOV

370



  min = TMIN(min, telemInt);




×








UNCOV

371



  min = TMIN(min, tsGrantHBInterval);




×








UNCOV

372



  min = TMIN(min, tsUptimeInterval);




×







373













UNCOV

374



  return min <= 1 ? 2 : min;




×







375



}











UNCOV

376



void mndDoTimerPullupTask(SMnode *pMnode, int64_t sec) {




×








UNCOV

377



  int32_t code = 0;




×








UNCOV

378



  if (sec % tsTtlPushIntervalSec == 0) {





×








UNCOV

379



    mndPullupTtl(pMnode);




×







380



  }










381













UNCOV

382



  if (sec % tsTrimVDbIntervalSec == 0) {





×








UNCOV

383



    mndPullupTrimDb(pMnode);




×







384



  }










385













UNCOV

386



  if (tsS3MigrateEnabled && sec % tsS3MigrateIntervalSec == 0) {





×







387



    mndPullupS3MigrateDb(pMnode);




×







388



  }










389













UNCOV

390



  if (sec % tsTransPullupInterval == 0) {





×








UNCOV

391



    mndPullupTrans(pMnode);




×







392



  }










393













UNCOV

394



  if (sec % tsCompactPullupInterval == 0) {





×








UNCOV

395



    mndPullupCompacts(pMnode);




×







396



  }










397













UNCOV

398



  if (sec % tsMqRebalanceInterval == 0) {





×








UNCOV

399



    mndCalMqRebalance(pMnode);




×







400



  }










401













UNCOV

402



  if (sec % 30 == 0) {  // send the checkpoint info every 30 sec





×








UNCOV

403



    mndStreamCheckpointTimer(pMnode);




×







404



  }










405













UNCOV

406



  if (sec % tsStreamNodeCheckInterval == 0) {





×








UNCOV

407



    mndStreamCheckNode(pMnode);




×







408



  }










409













UNCOV

410



  if (sec % 5 == 0) {





×








UNCOV

411



    mndStreamConsensusChkpt(pMnode);




×







412



  }










413













UNCOV

414



  if (sec % tsTelemInterval == (TMIN(86400, (tsTelemInterval - 1)))) {





×








UNCOV

415



    mndPullupTelem(pMnode);




×







416



  }










417













UNCOV

418



  if (sec % tsGrantHBInterval == 0) {





×








UNCOV

419



    mndPullupGrant(pMnode);




×







420



  }










421













UNCOV

422



  if (sec % tsUptimeInterval == 0) {





×








UNCOV

423



    mndIncreaseUpTime(pMnode);




×







424



  }










425













UNCOV

426



  if (sec % (tsArbHeartBeatIntervalSec) == 0) {





×








UNCOV

427



    if ((code = mndPullupArbHeartbeat(pMnode)) != 0) {





×







428



      mError("failed to pullup arb heartbeat, since:%s", tstrerror(code));





×







429



    }










430



  }










431













UNCOV

432



  if (sec % (tsArbCheckSyncIntervalSec) == 0) {





×








UNCOV

433



    if ((code = mndPullupArbCheckSync(pMnode)) != 0) {





×







434



      mError("failed to pullup arb check sync, since:%s", tstrerror(code));





×







435



    }










436



  }











UNCOV

437



}




×








UNCOV

438



void mndDoTimerCheckTask(SMnode *pMnode, int64_t sec) {




×








UNCOV

439



  if (sec % (tsStatusInterval * 5) == 0) {





×








UNCOV

440



    mndCheckDnodeOffline(pMnode);




×







441



  }











UNCOV

442



  if (sec % (MNODE_TIMEOUT_SEC / 2) == 0) {





×








UNCOV

443



    mndSyncCheckTimeout(pMnode);




×







444



  }











UNCOV

445



}




×







446













UNCOV

447



static void *mndThreadFp(void *param) {




×








UNCOV

448



  SMnode *pMnode = param;




×








UNCOV

449



  int64_t lastTime = 0;




×








UNCOV

450



  setThreadName("mnode-timer");




×







451













UNCOV

452



  while (1) {




×








UNCOV

453



    lastTime++;




×








UNCOV

454



    taosMsleep(100);




×








UNCOV

455



    if (mndGetStop(pMnode)) break;





×








UNCOV

456



    if (lastTime % 10 != 0) continue;





×







457













UNCOV

458



    int64_t sec = lastTime / 10;




×








UNCOV

459



    mndDoTimerCheckTask(pMnode, sec);




×







460













UNCOV

461



    int64_t minCron = minCronTime();




×








UNCOV

462



    if (sec % minCron == 0 && mnodeIsNotLeader(pMnode)) {





×







463



      // not leader, do nothing











UNCOV

464



      mTrace("timer not process since mnode is not leader, reason: %s", tstrerror(terrno));





×








UNCOV

465



      terrno = 0;




×








UNCOV

466



      continue;




×







467



    }











UNCOV

468



    mndDoTimerPullupTask(pMnode, sec);




×







469



  }










470













UNCOV

471



  return NULL;




×







472



}










473













UNCOV

474



static int32_t mndInitTimer(SMnode *pMnode) {




×








UNCOV

475



  int32_t      code = 0;




×







476



  TdThreadAttr thAttr;











UNCOV

477



  (void)taosThreadAttrInit(&thAttr);




×








UNCOV

478



  (void)taosThreadAttrSetDetachState(&thAttr, PTHREAD_CREATE_JOINABLE);




×








UNCOV

479



  if ((code = taosThreadCreate(&pMnode->thread, &thAttr, mndThreadFp, pMnode)) != 0) {





×







480



    mError("failed to create timer thread since %s", tstrerror(code));





×







481



    TAOS_RETURN(code);




×







482



  }










483













UNCOV

484



  (void)taosThreadAttrDestroy(&thAttr);




×








UNCOV

485



  tmsgReportStartup("mnode-timer", "initialized");




×








UNCOV

486



  TAOS_RETURN(code);




×







487



}










488













UNCOV

489



static void mndCleanupTimer(SMnode *pMnode) {




×








UNCOV

490



  if (taosCheckPthreadValid(pMnode->thread)) {





×








UNCOV

491



    (void)taosThreadJoin(pMnode->thread, NULL);




×








UNCOV

492



    taosThreadClear(&pMnode->thread);




×







493



  }











UNCOV

494



}




×







495













UNCOV

496



static int32_t mndCreateDir(SMnode *pMnode, const char *path) {




×








UNCOV

497



  int32_t code = 0;




×








UNCOV

498



  pMnode->path = taosStrdup(path);





×








UNCOV

499



  if (pMnode->path == NULL) {





×







500



    code = terrno;




×







501



    TAOS_RETURN(code);




×







502



  }










503













UNCOV

504



  if (taosMkDir(pMnode->path) != 0) {





×







505



    code = terrno;




×







506



    TAOS_RETURN(code);




×







507



  }










508













UNCOV

509



  TAOS_RETURN(code);




×







510



}










511













UNCOV

512



static int32_t mndInitWal(SMnode *pMnode) {




×








UNCOV

513



  int32_t code = 0;




×








UNCOV

514



  char    path[PATH_MAX + 20] = {0};




×








UNCOV

515



  (void)snprintf(path, sizeof(path), "%s%swal", pMnode->path, TD_DIRSEP);




×








UNCOV

516



  SWalCfg cfg = {.vgId = 1,




×







517



                 .fsyncPeriod = 0,










518



                 .rollPeriod = -1,










519



                 .segSize = -1,










520



                 .committed = -1,










521



                 .retentionPeriod = 0,










522



                 .retentionSize = 0,










523



                 .level = TAOS_WAL_FSYNC,










524



                 .encryptAlgorithm = 0,










525



                 .encryptKey = {0}};










526












527



#if defined(TD_ENTERPRISE)











UNCOV

528



  if (tsiEncryptAlgorithm == DND_CA_SM4 && (tsiEncryptScope & DND_CS_MNODE_WAL) == DND_CS_MNODE_WAL) {





×







529



    cfg.encryptAlgorithm = (tsiEncryptScope & DND_CS_MNODE_WAL) ? tsiEncryptAlgorithm : 0;





×







530



    if (tsEncryptKey[0] == '\0') {





×







531



      code = TSDB_CODE_DNODE_INVALID_ENCRYPTKEY;




×







532



      TAOS_RETURN(code);




×







533



    } else {










534



      tstrncpy(cfg.encryptKey, tsEncryptKey, ENCRYPT_KEY_LEN + 1);




×







535



    }










536



  }










537



#endif










538













UNCOV

539



  pMnode->pWal = walOpen(path, &cfg);




×








UNCOV

540



  if (pMnode->pWal == NULL) {





×







541



    code = TSDB_CODE_MND_RETURN_VALUE_NULL;




×







542



    if (terrno != 0) code = terrno;





×







543



    mError("failed to open wal since %s. wal:%s", tstrerror(code), path);





×







544



    TAOS_RETURN(code);




×







545



  }










546













UNCOV

547



  TAOS_RETURN(code);




×







548



}










549













UNCOV

550



static void mndCloseWal(SMnode *pMnode) {




×








UNCOV

551



  if (pMnode->pWal != NULL) {





×








UNCOV

552



    walClose(pMnode->pWal);




×








UNCOV

553



    pMnode->pWal = NULL;




×







554



  }











UNCOV

555



}




×







556













UNCOV

557



static int32_t mndInitSdb(SMnode *pMnode) {




×








UNCOV

558



  int32_t code = 0;




×








UNCOV

559



  SSdbOpt opt = {0};




×








UNCOV

560



  opt.path = pMnode->path;




×








UNCOV

561



  opt.pMnode = pMnode;




×








UNCOV

562



  opt.pWal = pMnode->pWal;




×







563













UNCOV

564



  pMnode->pSdb = sdbInit(&opt);




×








UNCOV

565



  if (pMnode->pSdb == NULL) {





×







566



    code = TSDB_CODE_MND_RETURN_VALUE_NULL;




×







567



    if (terrno != 0) code = terrno;





×







568



    TAOS_RETURN(code);




×







569



  }










570













UNCOV

571



  TAOS_RETURN(code);




×







572



}










573













UNCOV

574



static int32_t mndOpenSdb(SMnode *pMnode) {




×








UNCOV

575



  int32_t code = 0;




×








UNCOV

576



  if (!pMnode->deploy) {





×








UNCOV

577



    code = sdbReadFile(pMnode->pSdb);




×







578



  }










579













UNCOV

580



  mInfo("vgId:1, mnode sdb is opened, with applied index:%" PRId64, pMnode->pSdb->commitIndex);





×







581













UNCOV

582



  atomic_store_64(&pMnode->applied, pMnode->pSdb->commitIndex);




×








UNCOV

583



  return code;




×







584



}










585













UNCOV

586



static void mndCleanupSdb(SMnode *pMnode) {




×








UNCOV

587



  if (pMnode->pSdb) {





×








UNCOV

588



    sdbCleanup(pMnode->pSdb);




×








UNCOV

589



    pMnode->pSdb = NULL;




×







590



  }











UNCOV

591



}




×







592













UNCOV

593



static int32_t mndAllocStep(SMnode *pMnode, char *name, MndInitFp initFp, MndCleanupFp cleanupFp) {




×








UNCOV

594



  SMnodeStep step = {0};




×








UNCOV

595



  step.name = name;




×








UNCOV

596



  step.initFp = initFp;




×








UNCOV

597



  step.cleanupFp = cleanupFp;




×








UNCOV

598



  if (taosArrayPush(pMnode->pSteps, &step) == NULL) {





×







599



    TAOS_RETURN(terrno);




×







600



  }










601













UNCOV

602



  TAOS_RETURN(0);




×







603



}










604













UNCOV

605



static int32_t mndInitSteps(SMnode *pMnode) {




×








UNCOV

606



  TAOS_CHECK_RETURN(mndAllocStep(pMnode, "mnode-wal", mndInitWal, mndCloseWal));





×








UNCOV

607



  TAOS_CHECK_RETURN(mndAllocStep(pMnode, "mnode-sdb", mndInitSdb, mndCleanupSdb));





×








UNCOV

608



  TAOS_CHECK_RETURN(mndAllocStep(pMnode, "mnode-trans", mndInitTrans, mndCleanupTrans));





×








UNCOV

609



  TAOS_CHECK_RETURN(mndAllocStep(pMnode, "mnode-cluster", mndInitCluster, mndCleanupCluster));





×








UNCOV

610



  TAOS_CHECK_RETURN(mndAllocStep(pMnode, "mnode-mnode", mndInitMnode, mndCleanupMnode));





×








UNCOV

611



  TAOS_CHECK_RETURN(mndAllocStep(pMnode, "mnode-qnode", mndInitQnode, mndCleanupQnode));





×








UNCOV

612



  TAOS_CHECK_RETURN(mndAllocStep(pMnode, "mnode-snode", mndInitSnode, mndCleanupSnode));





×








UNCOV

613



  TAOS_CHECK_RETURN(mndAllocStep(pMnode, "mnode-anode", mndInitAnode, mndCleanupAnode));





×








UNCOV

614



  TAOS_CHECK_RETURN(mndAllocStep(pMnode, "mnode-arbgroup", mndInitArbGroup, mndCleanupArbGroup));





×








UNCOV

615



  TAOS_CHECK_RETURN(mndAllocStep(pMnode, "mnode-config", mndInitConfig, NULL));





×








UNCOV

616



  TAOS_CHECK_RETURN(mndAllocStep(pMnode, "mnode-dnode", mndInitDnode, mndCleanupDnode));





×








UNCOV

617



  TAOS_CHECK_RETURN(mndAllocStep(pMnode, "mnode-user", mndInitUser, mndCleanupUser));





×








UNCOV

618



  TAOS_CHECK_RETURN(mndAllocStep(pMnode, "mnode-grant", mndInitGrant, mndCleanupGrant));





×








UNCOV

619



  TAOS_CHECK_RETURN(mndAllocStep(pMnode, "mnode-privilege", mndInitPrivilege, mndCleanupPrivilege));





×








UNCOV

620



  TAOS_CHECK_RETURN(mndAllocStep(pMnode, "mnode-acct", mndInitAcct, mndCleanupAcct));





×








UNCOV

621



  TAOS_CHECK_RETURN(mndAllocStep(pMnode, "mnode-stream", mndInitStream, mndCleanupStream));





×








UNCOV

622



  TAOS_CHECK_RETURN(mndAllocStep(pMnode, "mnode-topic", mndInitTopic, mndCleanupTopic));





×








UNCOV

623



  TAOS_CHECK_RETURN(mndAllocStep(pMnode, "mnode-consumer", mndInitConsumer, mndCleanupConsumer));





×








UNCOV

624



  TAOS_CHECK_RETURN(mndAllocStep(pMnode, "mnode-subscribe", mndInitSubscribe, mndCleanupSubscribe));





×








UNCOV

625



  TAOS_CHECK_RETURN(mndAllocStep(pMnode, "mnode-vgroup", mndInitVgroup, mndCleanupVgroup));





×








UNCOV

626



  TAOS_CHECK_RETURN(mndAllocStep(pMnode, "mnode-stb", mndInitStb, mndCleanupStb));





×








UNCOV

627



  TAOS_CHECK_RETURN(mndAllocStep(pMnode, "mnode-sma", mndInitSma, mndCleanupSma));





×








UNCOV

628



  TAOS_CHECK_RETURN(mndAllocStep(pMnode, "mnode-idx", mndInitIdx, mndCleanupIdx));





×








UNCOV

629



  TAOS_CHECK_RETURN(mndAllocStep(pMnode, "mnode-infos", mndInitInfos, mndCleanupInfos));





×








UNCOV

630



  TAOS_CHECK_RETURN(mndAllocStep(pMnode, "mnode-perfs", mndInitPerfs, mndCleanupPerfs));





×








UNCOV

631



  TAOS_CHECK_RETURN(mndAllocStep(pMnode, "mnode-db", mndInitDb, mndCleanupDb));





×








UNCOV

632



  TAOS_CHECK_RETURN(mndAllocStep(pMnode, "mnode-func", mndInitFunc, mndCleanupFunc));





×








UNCOV

633



  TAOS_CHECK_RETURN(mndAllocStep(pMnode, "mnode-view", mndInitView, mndCleanupView));





×








UNCOV

634



  TAOS_CHECK_RETURN(mndAllocStep(pMnode, "mnode-compact", mndInitCompact, mndCleanupCompact));





×








UNCOV

635



  TAOS_CHECK_RETURN(mndAllocStep(pMnode, "mnode-compact-detail", mndInitCompactDetail, mndCleanupCompactDetail));





×








UNCOV

636



  TAOS_CHECK_RETURN(mndAllocStep(pMnode, "mnode-sdb", mndOpenSdb, NULL));





×








UNCOV

637



  TAOS_CHECK_RETURN(mndAllocStep(pMnode, "mnode-profile", mndInitProfile, mndCleanupProfile));





×








UNCOV

638



  TAOS_CHECK_RETURN(mndAllocStep(pMnode, "mnode-show", mndInitShow, mndCleanupShow));





×








UNCOV

639



  TAOS_CHECK_RETURN(mndAllocStep(pMnode, "mnode-query", mndInitQuery, mndCleanupQuery));





×








UNCOV

640



  TAOS_CHECK_RETURN(mndAllocStep(pMnode, "mnode-sync", mndInitSync, mndCleanupSync));





×








UNCOV

641



  TAOS_CHECK_RETURN(mndAllocStep(pMnode, "mnode-telem", mndInitTelem, mndCleanupTelem));





×








UNCOV

642



  return 0;




×







643



}










644













UNCOV

645



static void mndCleanupSteps(SMnode *pMnode, int32_t pos) {




×








UNCOV

646



  if (pMnode->pSteps == NULL) return;





×







647













UNCOV

648



  if (pos == -1) {





×








UNCOV

649



    pos = taosArrayGetSize(pMnode->pSteps) - 1;




×







650



  }










651













UNCOV

652



  for (int32_t s = pos; s >= 0; s--) {





×








UNCOV

653



    SMnodeStep *pStep = taosArrayGet(pMnode->pSteps, s);




×








UNCOV

654



    mInfo("%s will cleanup", pStep->name);





×








UNCOV

655



    if (pStep->cleanupFp != NULL) {





×








UNCOV

656



      (*pStep->cleanupFp)(pMnode);




×







657



    }










658



  }










659













UNCOV

660



  taosArrayClear(pMnode->pSteps);




×








UNCOV

661



  taosArrayDestroy(pMnode->pSteps);




×








UNCOV

662



  pMnode->pSteps = NULL;




×







663



}










664













UNCOV

665



static int32_t mndExecSteps(SMnode *pMnode) {




×








UNCOV

666



  int32_t code = 0;




×








UNCOV

667



  int32_t size = taosArrayGetSize(pMnode->pSteps);




×








UNCOV

668



  for (int32_t pos = 0; pos < size; pos++) {





×








UNCOV

669



    SMnodeStep *pStep = taosArrayGet(pMnode->pSteps, pos);




×








UNCOV

670



    if (pStep->initFp == NULL) continue;





×







671













UNCOV

672



    if ((code = (*pStep->initFp)(pMnode)) != 0) {





×







673



      mError("%s exec failed since %s, start to cleanup", pStep->name, tstrerror(code));





×







674



      mndCleanupSteps(pMnode, pos);




×







675



      TAOS_RETURN(code);




×







676



    } else {











UNCOV

677



      mInfo("%s is initialized", pStep->name);





×








UNCOV

678



      tmsgReportStartup(pStep->name, "initialized");




×







679



    }










680



  }










681













UNCOV

682



  pMnode->clusterId = mndGetClusterId(pMnode);




×








UNCOV

683



  TAOS_RETURN(0);




×







684



}










685













UNCOV

686



static void mndSetOptions(SMnode *pMnode, const SMnodeOpt *pOption) {




×








UNCOV

687



  pMnode->msgCb = pOption->msgCb;




×








UNCOV

688



  pMnode->selfDnodeId = pOption->dnodeId;




×








UNCOV

689



  pMnode->syncMgmt.selfIndex = pOption->selfIndex;




×








UNCOV

690



  pMnode->syncMgmt.numOfReplicas = pOption->numOfReplicas;




×








UNCOV

691



  pMnode->syncMgmt.numOfTotalReplicas = pOption->numOfTotalReplicas;




×








UNCOV

692



  pMnode->syncMgmt.lastIndex = pOption->lastIndex;




×








UNCOV

693



  (void)memcpy(pMnode->syncMgmt.replicas, pOption->replicas, sizeof(pOption->replicas));




×








UNCOV

694



  (void)memcpy(pMnode->syncMgmt.nodeRoles, pOption->nodeRoles, sizeof(pOption->nodeRoles));




×








UNCOV

695



}




×







696













UNCOV

697



SMnode *mndOpen(const char *path, const SMnodeOpt *pOption) {




×








UNCOV

698



  terrno = 0;




×








UNCOV

699



  mInfo("start to open mnode in %s", path);





×







700













UNCOV

701



  SMnode *pMnode = taosMemoryCalloc(1, sizeof(SMnode));





×








UNCOV

702



  if (pMnode == NULL) {





×







703



    terrno = TSDB_CODE_OUT_OF_MEMORY;




×







704



    mError("failed to open mnode since %s", terrstr());





×







705



    return NULL;




×







706



  }











UNCOV

707



  (void)memset(pMnode, 0, sizeof(SMnode));




×







708













UNCOV

709



  int32_t code = taosThreadRwlockInit(&pMnode->lock, NULL);




×








UNCOV

710



  if (code != 0) {





×







711



    taosMemoryFree(pMnode);





×







712



    mError("failed to open mnode lock since %s", tstrerror(code));





×







713



    return NULL;




×







714



  }










715













UNCOV

716



  char timestr[24] = "1970-01-01 00:00:00.00";




×








UNCOV

717



  code = taosParseTime(timestr, &pMnode->checkTime, (int32_t)strlen(timestr), TSDB_TIME_PRECISION_MILLI, NULL);




×








UNCOV

718



  if (code < 0) {





×







719



    mError("failed to parse time since %s", tstrerror(code));





×







720



    (void)taosThreadRwlockDestroy(&pMnode->lock);




×







721



    taosMemoryFree(pMnode);





×







722



    return NULL;




×







723



  }











UNCOV

724



  mndSetOptions(pMnode, pOption);




×







725













UNCOV

726



  pMnode->deploy = pOption->deploy;




×








UNCOV

727



  pMnode->pSteps = taosArrayInit(24, sizeof(SMnodeStep));




×








UNCOV

728



  if (pMnode->pSteps == NULL) {





×







729



    taosMemoryFree(pMnode);





×







730



    terrno = TSDB_CODE_OUT_OF_MEMORY;




×







731



    mError("failed to open mnode since %s", terrstr());





×







732



    return NULL;




×







733



  }










734













UNCOV

735



  code = mndCreateDir(pMnode, path);




×








UNCOV

736



  if (code != 0) {





×







737



    code = terrno;




×







738



    mError("failed to open mnode since %s", tstrerror(code));





×







739



    mndClose(pMnode);




×







740



    terrno = code;




×







741



    return NULL;




×







742



  }










743













UNCOV

744



  code = mndInitSteps(pMnode);




×








UNCOV

745



  if (code != 0) {





×







746



    code = terrno;




×







747



    mError("failed to open mnode since %s", tstrerror(code));





×







748



    mndClose(pMnode);




×







749



    terrno = code;




×







750



    return NULL;




×







751



  }










752













UNCOV

753



  code = mndExecSteps(pMnode);




×








UNCOV

754



  if (code != 0) {





×







755



    code = terrno;




×







756



    mError("failed to open mnode since %s", tstrerror(code));





×







757



    mndClose(pMnode);




×







758



    terrno = code;




×







759



    return NULL;




×







760



  }










761













UNCOV

762



  mInfo("mnode open successfully");





×








UNCOV

763



  return pMnode;




×







764



}










765













UNCOV

766



void mndPreClose(SMnode *pMnode) {




×








UNCOV

767



  if (pMnode != NULL) {





×








UNCOV

768



    int32_t code = 0;




×







769



    // TODO check return value











UNCOV

770



    code = syncLeaderTransfer(pMnode->syncMgmt.sync);




×








UNCOV

771



    if (code < 0) {





×








UNCOV

772



      mError("failed to transfer leader since %s", tstrerror(code));





×







773



    }











UNCOV

774



    syncPreStop(pMnode->syncMgmt.sync);




×








UNCOV

775



    code = sdbWriteFile(pMnode->pSdb, 0);




×








UNCOV

776



    if (code < 0) {





×







777



      mError("failed to write sdb since %s", tstrerror(code));





×







778



    }










779



  }











UNCOV

780



}




×







781













UNCOV

782



void mndClose(SMnode *pMnode) {




×








UNCOV

783



  if (pMnode != NULL) {





×








UNCOV

784



    mInfo("start to close mnode");





×








UNCOV

785



    mndCleanupSteps(pMnode, -1);




×








UNCOV

786



    taosMemoryFreeClear(pMnode->path);





×








UNCOV

787



    taosMemoryFreeClear(pMnode);





×








UNCOV

788



    mInfo("mnode is closed");





×







789



  }











UNCOV

790



}




×







791













UNCOV

792



int32_t mndStart(SMnode *pMnode) {




×








UNCOV

793



  mndSyncStart(pMnode);




×








UNCOV

794



  if (pMnode->deploy) {





×








UNCOV

795



    if (sdbDeploy(pMnode->pSdb) != 0) {





×







796



      mError("failed to deploy sdb while start mnode");





×







797



      return -1;




×







798



    }











UNCOV

799



    mndSetRestored(pMnode, true);




×







800



  }











UNCOV

801



  grantReset(pMnode, TSDB_GRANT_ALL, 0);




×







802













UNCOV

803



  return mndInitTimer(pMnode);




×







804



}










805













UNCOV

806



int32_t mndIsCatchUp(SMnode *pMnode) {




×








UNCOV

807



  int64_t rid = pMnode->syncMgmt.sync;




×








UNCOV

808



  return syncIsCatchUp(rid);




×







809



}










810













UNCOV

811



ESyncRole mndGetRole(SMnode *pMnode) {




×








UNCOV

812



  int64_t rid = pMnode->syncMgmt.sync;




×








UNCOV

813



  return syncGetRole(rid);




×







814



}










815













UNCOV

816



int64_t mndGetTerm(SMnode *pMnode) {




×








UNCOV

817



  int64_t rid = pMnode->syncMgmt.sync;




×








UNCOV

818



  return syncGetTerm(rid);




×







819



}










820













UNCOV

821



int32_t mndGetArbToken(SMnode *pMnode, char *outToken) { return syncGetArbToken(pMnode->syncMgmt.sync, outToken); }




×







822













UNCOV

823



void mndStop(SMnode *pMnode) {




×








UNCOV

824



  mndSetStop(pMnode);




×








UNCOV

825



  mndSyncStop(pMnode);




×








UNCOV

826



  mndCleanupTimer(pMnode);




×








UNCOV

827



}




×







828













UNCOV

829



int32_t mndProcessSyncMsg(SRpcMsg *pMsg) {




×








UNCOV

830



  SMnode    *pMnode = pMsg->info.node;




×








UNCOV

831



  SSyncMgmt *pMgmt = &pMnode->syncMgmt;




×







832













UNCOV

833



  const STraceId *trace = &pMsg->info.traceId;




×








UNCOV

834



  mGTrace("vgId:1, sync msg:%p will be processed, type:%s", pMsg, TMSG_INFO(pMsg->msgType));





×







835













UNCOV

836



  int32_t code = syncProcessMsg(pMgmt->sync, pMsg);




×








UNCOV

837



  if (code != 0) {





×








UNCOV

838



    mGError("vgId:1, failed to process sync msg:%p type:%s, reason: %s, code:0x%x", pMsg, TMSG_INFO(pMsg->msgType),





×







839



            tstrerror(code), code);










840



  }










841













UNCOV

842



  return code;




×







843



}










844













UNCOV

845



static int32_t mndCheckMnodeState(SRpcMsg *pMsg) {




×








UNCOV

846



  int32_t code = 0;




×








UNCOV

847



  if (!IsReq(pMsg)) TAOS_RETURN(code);





×








UNCOV

848



  if (pMsg->msgType == TDMT_SCH_QUERY || pMsg->msgType == TDMT_SCH_MERGE_QUERY ||





×








UNCOV

849



      pMsg->msgType == TDMT_SCH_QUERY_CONTINUE || pMsg->msgType == TDMT_SCH_QUERY_HEARTBEAT ||





×








UNCOV

850



      pMsg->msgType == TDMT_SCH_FETCH || pMsg->msgType == TDMT_SCH_MERGE_FETCH || pMsg->msgType == TDMT_SCH_DROP_TASK ||





×








UNCOV

851



      pMsg->msgType == TDMT_SCH_TASK_NOTIFY) {





×








UNCOV

852



    TAOS_RETURN(code);




×







853



  }










854













UNCOV

855



  SMnode *pMnode = pMsg->info.node;




×








UNCOV

856



  (void)taosThreadRwlockRdlock(&pMnode->lock);




×








UNCOV

857



  if (pMnode->stopped) {





×








UNCOV

858



    (void)taosThreadRwlockUnlock(&pMnode->lock);




×








UNCOV

859



    code = TSDB_CODE_APP_IS_STOPPING;




×








UNCOV

860



    TAOS_RETURN(code);




×







861



  }










862













UNCOV

863



  terrno = 0;




×








UNCOV

864



  SSyncState state = syncGetState(pMnode->syncMgmt.sync);




×








UNCOV

865



  if (terrno != 0) {





×







866



    (void)taosThreadRwlockUnlock(&pMnode->lock);




×







867



    code = terrno;




×







868



    TAOS_RETURN(code);




×







869



  }










870













UNCOV

871



  if (state.state != TAOS_SYNC_STATE_LEADER) {





×








UNCOV

872



    (void)taosThreadRwlockUnlock(&pMnode->lock);




×








UNCOV

873



    code = TSDB_CODE_SYN_NOT_LEADER;




×








UNCOV

874



    goto _OVER;




×







875



  }










876













UNCOV

877



  if (!state.restored || !pMnode->restored) {





×








UNCOV

878



    (void)taosThreadRwlockUnlock(&pMnode->lock);




×








UNCOV

879



    code = TSDB_CODE_SYN_RESTORING;




×








UNCOV

880



    goto _OVER;




×







881



  }










882












883



#if 1











UNCOV

884



  (void)atomic_add_fetch_32(&pMnode->rpcRef, 1);




×







885



#else










886



  int32_t ref = atomic_add_fetch_32(&pMnode->rpcRef, 1);










887



  mTrace("mnode rpc is acquired, ref:%d", ref);










888



#endif










889













UNCOV

890



  (void)taosThreadRwlockUnlock(&pMnode->lock);




×








UNCOV

891



  TAOS_RETURN(code);




×







892













UNCOV

893



_OVER:




×








UNCOV

894



  if (pMsg->msgType == TDMT_MND_TMQ_TIMER || pMsg->msgType == TDMT_MND_TELEM_TIMER ||





×








UNCOV

895



      pMsg->msgType == TDMT_MND_TRANS_TIMER || pMsg->msgType == TDMT_MND_TTL_TIMER ||





×








UNCOV

896



      pMsg->msgType == TDMT_MND_TRIM_DB_TIMER || pMsg->msgType == TDMT_MND_UPTIME_TIMER ||





×








UNCOV

897



      pMsg->msgType == TDMT_MND_COMPACT_TIMER || pMsg->msgType == TDMT_MND_NODECHECK_TIMER ||





×








UNCOV

898



      pMsg->msgType == TDMT_MND_GRANT_HB_TIMER || pMsg->msgType == TDMT_MND_STREAM_REQ_CHKPT ||





×








UNCOV

899



      pMsg->msgType == TDMT_MND_S3MIGRATE_DB_TIMER || pMsg->msgType == TDMT_MND_ARB_HEARTBEAT_TIMER ||





×








UNCOV

900



      pMsg->msgType == TDMT_MND_ARB_CHECK_SYNC_TIMER) {





×








UNCOV

901



    mTrace("timer not process since mnode restored:%d stopped:%d, sync restored:%d role:%s ", pMnode->restored,





×







902



           pMnode->stopped, state.restored, syncStr(state.state));











UNCOV

903



    TAOS_RETURN(code);




×







904



  }










905













UNCOV

906



  const STraceId *trace = &pMsg->info.traceId;




×








UNCOV

907



  SEpSet          epSet = {0};




×








UNCOV

908



  mndGetMnodeEpSet(pMnode, &epSet);




×







909













UNCOV

910



  mGDebug(





×







911



      "msg:%p, type:%s failed to process since %s, mnode restored:%d stopped:%d, sync restored:%d "










912



      "role:%s, redirect numOfEps:%d inUse:%d, type:%s",










913



      pMsg, TMSG_INFO(pMsg->msgType), tstrerror(code), pMnode->restored, pMnode->stopped, state.restored,










914



      syncStr(state.state), epSet.numOfEps, epSet.inUse, TMSG_INFO(pMsg->msgType));










915













UNCOV

916



  if (epSet.numOfEps <= 0) return -1;





×







917













UNCOV

918



  for (int32_t i = 0; i < epSet.numOfEps; ++i) {





×








UNCOV

919



    mDebug("mnode index:%d, ep:%s:%u", i, epSet.eps[i].fqdn, epSet.eps[i].port);





×







920



  }










921













UNCOV

922



  int32_t contLen = tSerializeSEpSet(NULL, 0, &epSet);




×








UNCOV

923



  pMsg->info.rsp = rpcMallocCont(contLen);




×








UNCOV

924



  if (pMsg->info.rsp != NULL) {





×








UNCOV

925



    if (tSerializeSEpSet(pMsg->info.rsp, contLen, &epSet) < 0) {





×







926



      mError("failed to serialize ep set");





×







927



    }











UNCOV

928



    pMsg->info.hasEpSet = 1;




×








UNCOV

929



    pMsg->info.rspLen = contLen;




×







930



  }










931













UNCOV

932



  TAOS_RETURN(code);




×







933



}










934













UNCOV

935



int32_t mndProcessRpcMsg(SRpcMsg *pMsg, SQueueInfo *pQueueInfo) {




×








UNCOV

936



  SMnode         *pMnode = pMsg->info.node;




×








UNCOV

937



  const STraceId *trace = &pMsg->info.traceId;




×








UNCOV

938



  int32_t         code = TSDB_CODE_SUCCESS;




×







939













UNCOV

940



  MndMsgFp    fp = pMnode->msgFp[TMSG_INDEX(pMsg->msgType)];




×








UNCOV

941



  MndMsgFpExt fpExt = NULL;




×








UNCOV

942



  if (fp == NULL) {





×








UNCOV

943



    fpExt = pMnode->msgFpExt[TMSG_INDEX(pMsg->msgType)];




×








UNCOV

944



    if (fpExt == NULL) {





×







945



      mGError("msg:%p, failed to get msg handle, app:%p type:%s", pMsg, pMsg->info.ahandle, TMSG_INFO(pMsg->msgType));





×







946



      code = TSDB_CODE_MSG_NOT_PROCESSED;




×







947



      TAOS_RETURN(code);




×







948



    }










949



  }










950













UNCOV

951



  TAOS_CHECK_RETURN(mndCheckMnodeState(pMsg));





×







952













UNCOV

953



  mGTrace("msg:%p, start to process in mnode, app:%p type:%s", pMsg, pMsg->info.ahandle, TMSG_INFO(pMsg->msgType));





×








UNCOV

954



  if (fp)





×








UNCOV

955



    code = (*fp)(pMsg);




×







956



  else











UNCOV

957



    code = (*fpExt)(pMsg, pQueueInfo);




×








UNCOV

958



  mndReleaseRpc(pMnode);




×







959













UNCOV

960



  if (code == TSDB_CODE_ACTION_IN_PROGRESS) {





×








UNCOV

961



    mGTrace("msg:%p, won't response immediately since in progress", pMsg);





×








UNCOV

962



  } else if (code == 0) {





×








UNCOV

963



    mGTrace("msg:%p, successfully processed", pMsg);





×







964



  } else {










965



    // TODO removve this wrong set code











UNCOV

966



    if (code == -1) {





×








UNCOV

967



      code = terrno;




×







968



    }











UNCOV

969



    mGError("msg:%p, failed to process since %s, app:%p type:%s", pMsg, tstrerror(code), pMsg->info.ahandle,





×







970



            TMSG_INFO(pMsg->msgType));










971



  }










972













UNCOV

973



  TAOS_RETURN(code);




×







974



}










975













UNCOV

976



void mndSetMsgHandle(SMnode *pMnode, tmsg_t msgType, MndMsgFp fp) {




×








UNCOV

977



  tmsg_t type = TMSG_INDEX(msgType);




×








UNCOV

978



  if (type < TDMT_MAX) {





×








UNCOV

979



    pMnode->msgFp[type] = fp;




×







980



  }











UNCOV

981



}




×







982













UNCOV

983



void mndSetMsgHandleExt(SMnode *pMnode, tmsg_t msgType, MndMsgFpExt fp) {




×








UNCOV

984



  tmsg_t type = TMSG_INDEX(msgType);




×








UNCOV

985



  if (type < TDMT_MAX) {





×








UNCOV

986



    pMnode->msgFpExt[type] = fp;




×







987



  }











UNCOV

988



}




×







989












990



// Note: uid 0 is reserved











UNCOV

991



int64_t mndGenerateUid(const char *name, int32_t len) {




×








UNCOV

992



  int32_t hashval = MurmurHash3_32(name, len);




×







993



  do {




×








UNCOV

994



    int64_t us = taosGetTimestampUs();




×








UNCOV

995



    int64_t x = (us & 0x000000FFFFFFFFFF) << 24;




×








UNCOV

996



    int64_t uuid = x + ((hashval & ((1ul << 16) - 1ul)) << 8) + (taosRand() & ((1ul << 8) - 1ul));




×








UNCOV

997



    if (uuid) {





×








UNCOV

998



      return llabs(uuid);




×







999



    }










1000



  } while (true);










1001



}










1002













UNCOV

1003



int32_t mndGetMonitorInfo(SMnode *pMnode, SMonClusterInfo *pClusterInfo, SMonVgroupInfo *pVgroupInfo,




×







1004



                          SMonStbInfo *pStbInfo, SMonGrantInfo *pGrantInfo) {











UNCOV

1005



  int32_t code = mndAcquireRpc(pMnode);




×








UNCOV

1006



  if (code < 0) {





×







1007



    TAOS_RETURN(code);




×








UNCOV

1008



  } else if (code == 1) {





×







1009



    TAOS_RETURN(TSDB_CODE_SUCCESS);




×







1010



  }










1011













UNCOV

1012



  SSdb   *pSdb = pMnode->pSdb;




×








UNCOV

1013



  int64_t ms = taosGetTimestampMs();




×







1014













UNCOV

1015



  pClusterInfo->dnodes = taosArrayInit(sdbGetSize(pSdb, SDB_DNODE), sizeof(SMonDnodeDesc));




×








UNCOV

1016



  pClusterInfo->mnodes = taosArrayInit(sdbGetSize(pSdb, SDB_MNODE), sizeof(SMonMnodeDesc));




×








UNCOV

1017



  pVgroupInfo->vgroups = taosArrayInit(sdbGetSize(pSdb, SDB_VGROUP), sizeof(SMonVgroupDesc));




×








UNCOV

1018



  pStbInfo->stbs = taosArrayInit(sdbGetSize(pSdb, SDB_STB), sizeof(SMonStbDesc));




×








UNCOV

1019



  if (pClusterInfo->dnodes == NULL || pClusterInfo->mnodes == NULL || pVgroupInfo->vgroups == NULL ||





×








UNCOV

1020



      pStbInfo->stbs == NULL) {





×







1021



    mndReleaseRpc(pMnode);




×







1022



    code = TSDB_CODE_MND_RETURN_VALUE_NULL;




×







1023



    if (terrno != 0) code = terrno;





×







1024



    TAOS_RETURN(code);




×







1025



  }










1026












1027



  // cluster info











UNCOV

1028



  tstrncpy(pClusterInfo->version, td_version, sizeof(pClusterInfo->version));




×








UNCOV

1029



  pClusterInfo->monitor_interval = tsMonitorInterval;




×








UNCOV

1030



  pClusterInfo->connections_total = mndGetNumOfConnections(pMnode);




×








UNCOV

1031



  pClusterInfo->dbs_total = sdbGetSize(pSdb, SDB_DB);




×








UNCOV

1032



  pClusterInfo->stbs_total = sdbGetSize(pSdb, SDB_STB);




×








UNCOV

1033



  pClusterInfo->topics_toal = sdbGetSize(pSdb, SDB_TOPIC);




×








UNCOV

1034



  pClusterInfo->streams_total = sdbGetSize(pSdb, SDB_STREAM);




×







1035













UNCOV

1036



  void *pIter = NULL;




×








UNCOV

1037



  while (1) {




×








UNCOV

1038



    SDnodeObj *pObj = NULL;




×








UNCOV

1039



    pIter = sdbFetch(pSdb, SDB_DNODE, pIter, (void **)&pObj);




×








UNCOV

1040



    if (pIter == NULL) break;





×







1041













UNCOV

1042



    SMonDnodeDesc desc = {0};




×








UNCOV

1043



    desc.dnode_id = pObj->id;




×








UNCOV

1044



    tstrncpy(desc.dnode_ep, pObj->ep, sizeof(desc.dnode_ep));




×








UNCOV

1045



    if (mndIsDnodeOnline(pObj, ms)) {





×








UNCOV

1046



      tstrncpy(desc.status, "ready", sizeof(desc.status));




×







1047



    } else {











UNCOV

1048



      tstrncpy(desc.status, "offline", sizeof(desc.status));




×







1049



    }











UNCOV

1050



    if (taosArrayPush(pClusterInfo->dnodes, &desc) == NULL) {





×







1051



      mError("failed put dnode into array, but continue at this monitor report")





×







1052



    }











UNCOV

1053



    sdbRelease(pSdb, pObj);




×







1054



  }










1055













UNCOV

1056



  pIter = NULL;




×








UNCOV

1057



  while (1) {




×








UNCOV

1058



    SMnodeObj *pObj = NULL;




×








UNCOV

1059



    pIter = sdbFetch(pSdb, SDB_MNODE, pIter, (void **)&pObj);




×








UNCOV

1060



    if (pIter == NULL) break;





×







1061













UNCOV

1062



    SMonMnodeDesc desc = {0};




×








UNCOV

1063



    desc.mnode_id = pObj->id;




×








UNCOV

1064



    tstrncpy(desc.mnode_ep, pObj->pDnode->ep, sizeof(desc.mnode_ep));




×







1065













UNCOV

1066



    if (pObj->id == pMnode->selfDnodeId) {





×








UNCOV

1067



      pClusterInfo->first_ep_dnode_id = pObj->id;




×








UNCOV

1068



      tstrncpy(pClusterInfo->first_ep, pObj->pDnode->ep, sizeof(pClusterInfo->first_ep));




×







1069



      // pClusterInfo->master_uptime = (float)mndGetClusterUpTime(pMnode) / 86400.0f;











UNCOV

1070



      pClusterInfo->master_uptime = mndGetClusterUpTime(pMnode);




×







1071



      // pClusterInfo->master_uptime = (ms - pObj->stateStartTime) / (86400000.0f);











UNCOV

1072



      tstrncpy(desc.role, syncStr(TAOS_SYNC_STATE_LEADER), sizeof(desc.role));




×








UNCOV

1073



      desc.syncState = TAOS_SYNC_STATE_LEADER;




×







1074



    } else {










1075



      tstrncpy(desc.role, syncStr(pObj->syncState), sizeof(desc.role));




×







1076



      desc.syncState = pObj->syncState;




×







1077



    }











UNCOV

1078



    if (taosArrayPush(pClusterInfo->mnodes, &desc) == NULL) {





×







1079



      mError("failed to put mnode into array, but continue at this monitor report");





×







1080



    }











UNCOV

1081



    sdbRelease(pSdb, pObj);




×







1082



  }










1083












1084



  // vgroup info











UNCOV

1085



  pIter = NULL;




×








UNCOV

1086



  while (1) {




×








UNCOV

1087



    SVgObj *pVgroup = NULL;




×








UNCOV

1088



    pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);




×








UNCOV

1089



    if (pIter == NULL) break;





×







1090













UNCOV

1091



    pClusterInfo->vgroups_total++;




×








UNCOV

1092



    pClusterInfo->tbs_total += pVgroup->numOfTables;




×







1093













UNCOV

1094



    SMonVgroupDesc desc = {0};




×








UNCOV

1095



    desc.vgroup_id = pVgroup->vgId;




×







1096













UNCOV

1097



    SName name = {0};




×








UNCOV

1098



    code = tNameFromString(&name, pVgroup->dbName, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE);




×








UNCOV

1099



    if (code < 0) {





×







1100



      mError("failed to get db name since %s", tstrerror(code));





×







1101



      sdbRelease(pSdb, pVgroup);




×







1102



      TAOS_RETURN(code);




×







1103



    }











UNCOV

1104



    (void)tNameGetDbName(&name, desc.database_name);




×







1105













UNCOV

1106



    desc.tables_num = pVgroup->numOfTables;




×








UNCOV

1107



    pGrantInfo->timeseries_used += pVgroup->numOfTimeSeries;




×








UNCOV

1108



    tstrncpy(desc.status, "unsynced", sizeof(desc.status));




×








UNCOV

1109



    for (int32_t i = 0; i < pVgroup->replica; ++i) {





×








UNCOV

1110



      SVnodeGid     *pVgid = &pVgroup->vnodeGid[i];




×








UNCOV

1111



      SMonVnodeDesc *pVnDesc = &desc.vnodes[i];




×








UNCOV

1112



      pVnDesc->dnode_id = pVgid->dnodeId;




×








UNCOV

1113



      tstrncpy(pVnDesc->vnode_role, syncStr(pVgid->syncState), sizeof(pVnDesc->vnode_role));




×








UNCOV

1114



      pVnDesc->syncState = pVgid->syncState;




×








UNCOV

1115



      if (pVgid->syncState == TAOS_SYNC_STATE_LEADER || pVgid->syncState == TAOS_SYNC_STATE_ASSIGNED_LEADER) {





×








UNCOV

1116



        tstrncpy(desc.status, "ready", sizeof(desc.status));




×








UNCOV

1117



        pClusterInfo->vgroups_alive++;




×







1118



      }











UNCOV

1119



      if (pVgid->syncState != TAOS_SYNC_STATE_ERROR && pVgid->syncState != TAOS_SYNC_STATE_OFFLINE) {





×








UNCOV

1120



        pClusterInfo->vnodes_alive++;




×







1121



      }











UNCOV

1122



      pClusterInfo->vnodes_total++;




×







1123



    }










1124













UNCOV

1125



    if (taosArrayPush(pVgroupInfo->vgroups, &desc) == NULL) {





×







1126



      mError("failed to put vgroup into array, but continue at this monitor report")





×







1127



    }











UNCOV

1128



    sdbRelease(pSdb, pVgroup);




×







1129



  }










1130












1131



  // stb info











UNCOV

1132



  pIter = NULL;




×








UNCOV

1133



  while (1) {




×








UNCOV

1134



    SStbObj *pStb = NULL;




×








UNCOV

1135



    pIter = sdbFetch(pSdb, SDB_STB, pIter, (void **)&pStb);




×








UNCOV

1136



    if (pIter == NULL) break;





×







1137













UNCOV

1138



    SMonStbDesc desc = {0};




×







1139













UNCOV

1140



    SName name1 = {0};




×








UNCOV

1141



    code = tNameFromString(&name1, pStb->db, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE);




×








UNCOV

1142



    if (code < 0) {





×







1143



      mError("failed to get db name since %s", tstrerror(code));





×







1144



      sdbRelease(pSdb, pStb);




×







1145



      TAOS_RETURN(code);




×







1146



    }











UNCOV

1147



    (void)tNameGetDbName(&name1, desc.database_name);




×







1148













UNCOV

1149



    SName name2 = {0};




×








UNCOV

1150



    code = tNameFromString(&name2, pStb->name, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE);




×








UNCOV

1151



    if (code < 0) {





×







1152



      mError("failed to get table name since %s", tstrerror(code));





×







1153



      sdbRelease(pSdb, pStb);




×







1154



      TAOS_RETURN(code);




×







1155



    }











UNCOV

1156



    tstrncpy(desc.stb_name, tNameGetTableName(&name2), TSDB_TABLE_NAME_LEN);




×







1157













UNCOV

1158



    if (taosArrayPush(pStbInfo->stbs, &desc) == NULL) {





×







1159



      mError("failed to put stb into array, but continue at this monitor report");





×







1160



    }











UNCOV

1161



    sdbRelease(pSdb, pStb);




×







1162



  }










1163












1164



  // grant info











UNCOV

1165



  pGrantInfo->expire_time = (pMnode->grant.expireTimeMS - ms) / 1000;




×








UNCOV

1166



  pGrantInfo->timeseries_total = pMnode->grant.timeseriesAllowed;




×








UNCOV

1167



  if (pMnode->grant.expireTimeMS == 0) {





×







1168



    pGrantInfo->expire_time = 0;




×







1169



    pGrantInfo->timeseries_total = 0;




×







1170



  }










1171













UNCOV

1172



  mndReleaseRpc(pMnode);




×








UNCOV

1173



  TAOS_RETURN(code);




×







1174



}










1175













UNCOV

1176



int32_t mndGetLoad(SMnode *pMnode, SMnodeLoad *pLoad) {




×








UNCOV

1177



  SSyncState state = syncGetState(pMnode->syncMgmt.sync);




×








UNCOV

1178



  pLoad->syncState = state.state;




×








UNCOV

1179



  pLoad->syncRestore = state.restored;




×








UNCOV

1180



  pLoad->syncTerm = state.term;




×








UNCOV

1181



  pLoad->roleTimeMs = state.roleTimeMs;




×








UNCOV

1182



  mTrace("mnode current syncState is %s, syncRestore:%d, syncTerm:%" PRId64 " ,roleTimeMs:%" PRId64,





×







1183



         syncStr(pLoad->syncState), pLoad->syncRestore, pLoad->syncTerm, pLoad->roleTimeMs);











UNCOV

1184



  return 0;




×







1185



}










1186













UNCOV

1187



int64_t mndGetRoleTimeMs(SMnode *pMnode) {




×








UNCOV

1188



  SSyncState state = syncGetState(pMnode->syncMgmt.sync);




×








UNCOV

1189



  return state.roleTimeMs;




×







1190



}










1191













UNCOV

1192



void mndSetRestored(SMnode *pMnode, bool restored) {




×








UNCOV

1193



  if (restored) {





×








UNCOV

1194



    (void)taosThreadRwlockWrlock(&pMnode->lock);




×








UNCOV

1195



    pMnode->restored = true;




×








UNCOV

1196



    (void)taosThreadRwlockUnlock(&pMnode->lock);




×








UNCOV

1197



    mInfo("mnode set restored:%d", restored);





×







1198



  } else {











UNCOV

1199



    (void)taosThreadRwlockWrlock(&pMnode->lock);




×








UNCOV

1200



    pMnode->restored = false;




×








UNCOV

1201



    (void)taosThreadRwlockUnlock(&pMnode->lock);




×








UNCOV

1202



    mInfo("mnode set restored:%d", restored);





×







1203



    while (1) {











UNCOV

1204



      if (pMnode->rpcRef <= 0) break;





×







1205



      taosMsleep(3);




×







1206



    }










1207



  }











UNCOV

1208



}




×







1209












1210



bool mndGetRestored(SMnode *pMnode) { return pMnode->restored; }




×







1211













UNCOV

1212



void mndSetStop(SMnode *pMnode) {




×








UNCOV

1213



  (void)taosThreadRwlockWrlock(&pMnode->lock);




×








UNCOV

1214



  pMnode->stopped = true;




×








UNCOV

1215



  (void)taosThreadRwlockUnlock(&pMnode->lock);




×








UNCOV

1216



  mInfo("mnode set stopped");





×








UNCOV

1217



}




×







1218













UNCOV

1219



bool mndGetStop(SMnode *pMnode) { return pMnode->stopped; }




×





















    

  • 

    •