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/mgmt/mgmt_vnode/src/vmWorker.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 "vmInt.h"










18



#include "vnodeInt.h"










19













UNCOV

20



static inline void vmSendRsp(SRpcMsg *pMsg, int32_t code) {




×








UNCOV

21



  if (pMsg->info.handle == NULL) return;





×








UNCOV

22



  SRpcMsg rsp = {




×







23



      .code = code,











UNCOV

24



      .pCont = pMsg->info.rsp,




×








UNCOV

25



      .contLen = pMsg->info.rspLen,




×







26



      .info = pMsg->info,










27



  };











UNCOV

28



  tmsgSendRsp(&rsp);




×







29



}










30













UNCOV

31



static void vmProcessMultiMgmtQueue(SQueueInfo *pInfo, SRpcMsg *pMsg) {




×








UNCOV

32



  SVnodeMgmt     *pMgmt = pInfo->ahandle;




×








UNCOV

33



  int32_t         code = -1;




×








UNCOV

34



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




×







35













UNCOV

36



  dGTrace("msg:%p, get from vnode-multi-mgmt queue", pMsg);





×








UNCOV

37



  switch (pMsg->msgType) {





×








UNCOV

38



    case TDMT_DND_CREATE_VNODE:




×








UNCOV

39



      code = vmProcessCreateVnodeReq(pMgmt, pMsg);




×








UNCOV

40



      break;




×







41



  }










42













UNCOV

43



  if (IsReq(pMsg)) {





×








UNCOV

44



    if (code != 0) {





×







45



      if (terrno != 0) code = terrno;





×







46



      dGError("msg:%p, failed to process since %s, type:%s", pMsg, tstrerror(code), TMSG_INFO(pMsg->msgType));





×







47



    }











UNCOV

48



    vmSendRsp(pMsg, code);




×







49



  }










50













UNCOV

51



  dGTrace("msg:%p, is freed, code:0x%x", pMsg, code);





×








UNCOV

52



  rpcFreeCont(pMsg->pCont);




×








UNCOV

53



  taosFreeQitem(pMsg);




×








UNCOV

54



}




×







55













UNCOV

56



static void vmProcessMgmtQueue(SQueueInfo *pInfo, SRpcMsg *pMsg) {




×








UNCOV

57



  SVnodeMgmt     *pMgmt = pInfo->ahandle;




×








UNCOV

58



  int32_t         code = -1;




×








UNCOV

59



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




×







60













UNCOV

61



  dGTrace("msg:%p, get from vnode-mgmt queue", pMsg);





×








UNCOV

62



  switch (pMsg->msgType) {





×







63



    case TDMT_DND_CREATE_VNODE:




×







64



      code = vmProcessCreateVnodeReq(pMgmt, pMsg);




×







65



      break;




×








UNCOV

66



    case TDMT_DND_DROP_VNODE:




×








UNCOV

67



      code = vmProcessDropVnodeReq(pMgmt, pMsg);




×








UNCOV

68



      break;




×








UNCOV

69



    case TDMT_VND_ALTER_REPLICA:




×








UNCOV

70



      code = vmProcessAlterVnodeReplicaReq(pMgmt, pMsg);




×








UNCOV

71



      break;




×








UNCOV

72



    case TDMT_VND_DISABLE_WRITE:




×








UNCOV

73



      code = vmProcessDisableVnodeWriteReq(pMgmt, pMsg);




×








UNCOV

74



      break;




×








UNCOV

75



    case TDMT_VND_ALTER_HASHRANGE:




×








UNCOV

76



      code = vmProcessAlterHashRangeReq(pMgmt, pMsg);




×








UNCOV

77



      break;




×








UNCOV

78



    case TDMT_DND_ALTER_VNODE_TYPE:




×








UNCOV

79



      code = vmProcessAlterVnodeTypeReq(pMgmt, pMsg);




×








UNCOV

80



      break;




×







81



    case TDMT_DND_CHECK_VNODE_LEARNER_CATCHUP:




×







82



      code = vmProcessCheckLearnCatchupReq(pMgmt, pMsg);




×







83



      break;




×








UNCOV

84



    case TDMT_VND_ARB_HEARTBEAT:




×








UNCOV

85



      code = vmProcessArbHeartBeatReq(pMgmt, pMsg);




×








UNCOV

86



      break;




×







87



    default:




×







88



      terrno = TSDB_CODE_MSG_NOT_PROCESSED;




×







89



      dGError("msg:%p, not processed in vnode-mgmt queue", pMsg);





×







90



  }










91













UNCOV

92



  if (IsReq(pMsg)) {





×








UNCOV

93



    if (code != 0) {





×








UNCOV

94



      if (terrno != 0) code = terrno;





×








UNCOV

95



      dGError("msg:%p, failed to process since %s, type:%s", pMsg, tstrerror(code), TMSG_INFO(pMsg->msgType));





×







96



    }











UNCOV

97



    vmSendRsp(pMsg, code);




×







98



  }










99













UNCOV

100



  dGTrace("msg:%p, is freed, code:0x%x", pMsg, code);





×








UNCOV

101



  rpcFreeCont(pMsg->pCont);




×








UNCOV

102



  taosFreeQitem(pMsg);




×








UNCOV

103



}




×







104













UNCOV

105



static void vmProcessQueryQueue(SQueueInfo *pInfo, SRpcMsg *pMsg) {




×








UNCOV

106



  SVnodeObj      *pVnode = pInfo->ahandle;




×








UNCOV

107



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




×







108













UNCOV

109



  dGTrace("vgId:%d, msg:%p get from vnode-query queue", pVnode->vgId, pMsg);





×








UNCOV

110



  int32_t code = vnodeProcessQueryMsg(pVnode->pImpl, pMsg, pInfo);




×








UNCOV

111



  if (code != 0) {





×








UNCOV

112



    if (terrno != 0) code = terrno;





×








UNCOV

113



    dGError("vgId:%d, msg:%p failed to query since %s", pVnode->vgId, pMsg, tstrerror(code));





×








UNCOV

114



    vmSendRsp(pMsg, code);




×







115



  }










116













UNCOV

117



  dGTrace("vgId:%d, msg:%p is freed, code:0x%x", pVnode->vgId, pMsg, code);





×








UNCOV

118



  rpcFreeCont(pMsg->pCont);




×








UNCOV

119



  taosFreeQitem(pMsg);




×








UNCOV

120



}




×







121













UNCOV

122



static void vmProcessStreamQueue(SQueueInfo *pInfo, SRpcMsg *pMsg) {




×








UNCOV

123



  SVnodeObj      *pVnode = pInfo->ahandle;




×








UNCOV

124



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




×







125













UNCOV

126



  dGTrace("vgId:%d, msg:%p get from vnode-stream queue", pVnode->vgId, pMsg);





×








UNCOV

127



  int32_t code = vnodeProcessStreamMsg(pVnode->pImpl, pMsg, pInfo);




×








UNCOV

128



  if (code != 0) {





×








UNCOV

129



    terrno = code;




×








UNCOV

130



    dGError("vgId:%d, msg:%p failed to process stream msg %s since %s", pVnode->vgId, pMsg, TMSG_INFO(pMsg->msgType),





×







131



            tstrerror(code));











UNCOV

132



    vmSendRsp(pMsg, code);




×







133



  }










134













UNCOV

135



  dGTrace("vgId:%d, msg:%p is freed, code:0x%x", pVnode->vgId, pMsg, code);





×








UNCOV

136



  rpcFreeCont(pMsg->pCont);




×








UNCOV

137



  taosFreeQitem(pMsg);




×








UNCOV

138



}




×







139













UNCOV

140



static void vmProcessStreamCtrlQueue(SQueueInfo *pInfo, STaosQall* pQall, int32_t numOfItems) {




×








UNCOV

141



  SVnodeObj *pVnode = pInfo->ahandle;




×








UNCOV

142



  void      *pItem = NULL;




×








UNCOV

143



  int32_t    code = 0;




×







144













UNCOV

145



  while (1) {




×








UNCOV

146



    if (taosGetQitem(pQall, &pItem) == 0) {





×








UNCOV

147



      break;




×







148



    }










149













UNCOV

150



    SRpcMsg        *pMsg = pItem;




×








UNCOV

151



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




×







152













NEW


153



    dGTrace("vgId:%d, msg:%p get from vnode-stream-ctrl queue", pVnode->vgId, pMsg);





×








UNCOV

154



    code = vnodeProcessStreamCtrlMsg(pVnode->pImpl, pMsg, pInfo);




×








UNCOV

155



    if (code != 0) {





×








UNCOV

156



      terrno = code;




×








UNCOV

157



      dGError("vgId:%d, msg:%p failed to process stream ctrl msg %s since %s", pVnode->vgId, pMsg,





×







158



              TMSG_INFO(pMsg->msgType), tstrerror(code));











UNCOV

159



      vmSendRsp(pMsg, code);




×







160



    }










161













UNCOV

162



    dGTrace("vgId:%d, msg:%p is freed, code:0x%x", pVnode->vgId, pMsg, code);





×








UNCOV

163



    rpcFreeCont(pMsg->pCont);




×








UNCOV

164



    taosFreeQitem(pMsg);




×







165



  }











UNCOV

166



}




×







167













NEW


168



static void vmProcessStreamLongExecQueue(SQueueInfo *pInfo, SRpcMsg *pMsg) {




×








NEW


169



  SVnodeObj      *pVnode = pInfo->ahandle;




×








NEW


170



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




×








NEW


171



  int32_t         code = 0;




×







172













NEW


173



  dGTrace("vgId:%d, msg:%p get from vnode-stream long-exec queue", pVnode->vgId, pMsg);





×







174













NEW


175



  code = vnodeProcessStreamLongExecMsg(pVnode->pImpl, pMsg, pInfo);




×








NEW


176



  if (code != 0) {





×








NEW


177



    terrno = code;




×








NEW


178



    dGError("vgId:%d, msg:%p failed to process stream msg %s since %s", pVnode->vgId, pMsg, TMSG_INFO(pMsg->msgType),





×







179



            tstrerror(code));











NEW


180



    vmSendRsp(pMsg, code);




×







181



  }










182













NEW


183



  dGTrace("vgId:%d, msg:%p is freed, code:0x%x", pVnode->vgId, pMsg, code);





×








NEW


184



  rpcFreeCont(pMsg->pCont);




×








NEW


185



  taosFreeQitem(pMsg);




×








NEW


186



}




×







187













UNCOV

188



static void vmProcessFetchQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs) {




×








UNCOV

189



  SVnodeObj *pVnode = pInfo->ahandle;




×








UNCOV

190



  SRpcMsg   *pMsg = NULL;




×







191













UNCOV

192



  for (int32_t i = 0; i < numOfMsgs; ++i) {





×








UNCOV

193



    if (taosGetQitem(qall, (void **)&pMsg) == 0) continue;





×








UNCOV

194



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




×








UNCOV

195



    dGTrace("vgId:%d, msg:%p get from vnode-fetch queue", pVnode->vgId, pMsg);





×







196













UNCOV

197



    terrno = 0;




×








UNCOV

198



    int32_t code = vnodeProcessFetchMsg(pVnode->pImpl, pMsg, pInfo);




×








UNCOV

199



    if (code != 0) {





×








UNCOV

200



      if (code == -1 && terrno != 0) {





×







201



        code = terrno;




×







202



      }










203













UNCOV

204



      if (code == TSDB_CODE_WAL_LOG_NOT_EXIST) {





×







205



        dGDebug("vnodeProcessFetchMsg vgId:%d, msg:%p failed to fetch since %s", pVnode->vgId, pMsg, terrstr());





×







206



      } else {











UNCOV

207



        dGError("vnodeProcessFetchMsg vgId:%d, msg:%p failed to fetch since %s", pVnode->vgId, pMsg, terrstr());





×







208



      }










209













UNCOV

210



      vmSendRsp(pMsg, code);




×







211



    }










212













UNCOV

213



    dGTrace("vnodeProcessFetchMsg vgId:%d, msg:%p is freed, code:0x%x", pVnode->vgId, pMsg, code);





×








UNCOV

214



    rpcFreeCont(pMsg->pCont);




×








UNCOV

215



    taosFreeQitem(pMsg);




×







216



  }











UNCOV

217



}




×







218













UNCOV

219



static void vmProcessSyncQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs) {




×








UNCOV

220



  SVnodeObj *pVnode = pInfo->ahandle;




×








UNCOV

221



  SRpcMsg   *pMsg = NULL;




×







222













UNCOV

223



  for (int32_t i = 0; i < numOfMsgs; ++i) {





×








UNCOV

224



    if (taosGetQitem(qall, (void **)&pMsg) == 0) continue;





×








UNCOV

225



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




×








UNCOV

226



    dGTrace("vgId:%d, msg:%p get from vnode-sync queue", pVnode->vgId, pMsg);





×







227













UNCOV

228



    int32_t code = vnodeProcessSyncMsg(pVnode->pImpl, pMsg, NULL);  // no response here




×








UNCOV

229



    dGTrace("vgId:%d, msg:%p is freed, code:0x%x", pVnode->vgId, pMsg, code);





×








UNCOV

230



    rpcFreeCont(pMsg->pCont);




×








UNCOV

231



    taosFreeQitem(pMsg);




×







232



  }











UNCOV

233



}




×







234












235



static void vmSendResponse(SRpcMsg *pMsg) {




×







236



  if (pMsg->info.handle) {





×







237



    SRpcMsg rsp = {.info = pMsg->info, .code = terrno};




×







238



    if (rpcSendResponse(&rsp) != 0) {





×







239



      dError("failed to send response since %s", terrstr());





×







240



    }










241



  }










242



}




×







243













UNCOV

244



static bool vmDataSpaceSufficient(SVnodeObj *pVnode) {




×








UNCOV

245



  STfs *pTfs = pVnode->pImpl->pTfs;




×








UNCOV

246



  if (pTfs) {





×








UNCOV

247



    return tfsDiskSpaceSufficient(pTfs, 0, pVnode->diskPrimary);




×







248



  } else {










249



    return osDataSpaceSufficient();




×







250



  }










251



}










252













UNCOV

253



static int32_t vmAcquireVnodeWrapper(SVnodeMgmt *pMgt, int32_t vgId, SVnodeObj **pNode) {




×








UNCOV

254



  *pNode = vmAcquireVnode(pMgt, vgId);




×








UNCOV

255



  if (*pNode == NULL) {





×








UNCOV

256



    return terrno;




×







257



  }











UNCOV

258



  return 0;




×







259



}











UNCOV

260



static int32_t vmPutMsgToQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg, EQueueType qtype) {




×








UNCOV

261



  int32_t         code = 0;




×








UNCOV

262



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




×








UNCOV

263



  if (pMsg->contLen < sizeof(SMsgHead)) {





×







264



    dGError("invalid rpc msg with no msg head at pCont. pMsg:%p, type:%s, contLen:%d", pMsg, TMSG_INFO(pMsg->msgType),





×







265



            pMsg->contLen);










266



    return TSDB_CODE_INVALID_MSG;




×







267



  }










268













UNCOV

269



  SMsgHead *pHead = pMsg->pCont;




×







270













UNCOV

271



  pHead->contLen = ntohl(pHead->contLen);




×








UNCOV

272



  pHead->vgId = ntohl(pHead->vgId);




×







273













UNCOV

274



  SVnodeObj *pVnode = NULL;




×








UNCOV

275



  code = vmAcquireVnodeWrapper(pMgmt, pHead->vgId, &pVnode);




×








UNCOV

276



  if (code != 0) {





×








UNCOV

277



    dGDebug("vgId:%d, msg:%p failed to put into vnode queue since %s, type:%s qtype:%d contLen:%d", pHead->vgId, pMsg,





×







278



            tstrerror(code), TMSG_INFO(pMsg->msgType), qtype, pHead->contLen);











UNCOV

279



    return code;




×







280



  }










281













UNCOV

282



  switch (qtype) {





×








UNCOV

283



    case QUERY_QUEUE:




×








UNCOV

284



      code = vnodePreprocessQueryMsg(pVnode->pImpl, pMsg);




×








UNCOV

285



      if (code) {





×








UNCOV

286



        dError("vgId:%d, msg:%p preprocess query msg failed since %s", pVnode->vgId, pMsg, tstrerror(code));





×







287



      } else {











UNCOV

288



        dGTrace("vgId:%d, msg:%p put into vnode-query queue", pVnode->vgId, pMsg);





×








UNCOV

289



        code = taosWriteQitem(pVnode->pQueryQ, pMsg);




×







290



      }











UNCOV

291



      break;




×








UNCOV

292



    case STREAM_QUEUE:




×








UNCOV

293



      dGTrace("vgId:%d, msg:%p put into vnode-stream queue", pVnode->vgId, pMsg);





×








UNCOV

294



      code = taosWriteQitem(pVnode->pStreamQ, pMsg);




×








UNCOV

295



      break;




×








UNCOV

296



    case STREAM_CTRL_QUEUE:




×








NEW


297



      dGTrace("vgId:%d, msg:%p put into vnode-stream-ctrl queue", pVnode->vgId, pMsg);





×








UNCOV

298



      code = taosWriteQitem(pVnode->pStreamCtrlQ, pMsg);




×








UNCOV

299



      break;




×








NEW


300



    case STREAM_LONG_EXEC_QUEUE:




×








NEW


301



      dGTrace("vgId:%d, msg:%p put into vnode-stream-long-exec queue", pVnode->vgId, pMsg);





×








NEW


302



      code = taosWriteQitem(pVnode->pStreamLongExecQ, pMsg);




×








NEW


303



      break;




×








UNCOV

304



    case FETCH_QUEUE:




×








UNCOV

305



      dGTrace("vgId:%d, msg:%p put into vnode-fetch queue", pVnode->vgId, pMsg);





×








UNCOV

306



      code = taosWriteQitem(pVnode->pFetchQ, pMsg);




×








UNCOV

307



      break;




×








UNCOV

308



    case WRITE_QUEUE:




×








UNCOV

309



      if (!vmDataSpaceSufficient(pVnode)) {





×







310



        code = TSDB_CODE_NO_ENOUGH_DISKSPACE;




×







311



        dError("vgId:%d, msg:%p put into vnode-write queue failed since %s", pVnode->vgId, pMsg, tstrerror(code));





×







312



        break;




×







313



      }











UNCOV

314



      if (pMsg->msgType == TDMT_VND_SUBMIT && (grantCheck(TSDB_GRANT_STORAGE) != TSDB_CODE_SUCCESS)) {





×







315



        code = TSDB_CODE_VND_NO_WRITE_AUTH;




×







316



        dDebug("vgId:%d, msg:%p put into vnode-write queue failed since %s", pVnode->vgId, pMsg, tstrerror(code));





×







317



        break;




×







318



      }











UNCOV

319



      if (pMsg->msgType != TDMT_VND_ALTER_CONFIRM && pVnode->disable) {





×








UNCOV

320



        dDebug("vgId:%d, msg:%p put into vnode-write queue failed since its disable", pVnode->vgId, pMsg);





×








UNCOV

321



        code = TSDB_CODE_VND_STOPPED;




×








UNCOV

322



        break;




×







323



      }











UNCOV

324



      dGTrace("vgId:%d, msg:%p put into vnode-write queue", pVnode->vgId, pMsg);





×








UNCOV

325



      code = taosWriteQitem(pVnode->pWriteW.queue, pMsg);




×








UNCOV

326



      break;




×








UNCOV

327



    case SYNC_QUEUE:




×








UNCOV

328



      dGTrace("vgId:%d, msg:%p put into vnode-sync queue", pVnode->vgId, pMsg);





×








UNCOV

329



      code = taosWriteQitem(pVnode->pSyncW.queue, pMsg);




×








UNCOV

330



      break;




×








UNCOV

331



    case SYNC_RD_QUEUE:




×








UNCOV

332



      dGTrace("vgId:%d, msg:%p put into vnode-sync-rd queue", pVnode->vgId, pMsg);





×








UNCOV

333



      code = taosWriteQitem(pVnode->pSyncRdW.queue, pMsg);




×








UNCOV

334



      break;




×








UNCOV

335



    case APPLY_QUEUE:




×








UNCOV

336



      dGTrace("vgId:%d, msg:%p put into vnode-apply queue", pVnode->vgId, pMsg);





×








UNCOV

337



      code = taosWriteQitem(pVnode->pApplyW.queue, pMsg);




×








UNCOV

338



      break;




×








UNCOV

339



    default:




×








UNCOV

340



      code = TSDB_CODE_INVALID_MSG;




×








UNCOV

341



      break;




×







342



  }










343













UNCOV

344



  vmReleaseVnode(pMgmt, pVnode);




×








UNCOV

345



  return code;




×







346



}










347













UNCOV

348



int32_t vmPutMsgToSyncRdQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) { return vmPutMsgToQueue(pMgmt, pMsg, SYNC_RD_QUEUE); }




×







349













UNCOV

350



int32_t vmPutMsgToSyncQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) { return vmPutMsgToQueue(pMgmt, pMsg, SYNC_QUEUE); }




×







351













UNCOV

352



int32_t vmPutMsgToWriteQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) { return vmPutMsgToQueue(pMgmt, pMsg, WRITE_QUEUE); }




×







353













UNCOV

354



int32_t vmPutMsgToQueryQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) { return vmPutMsgToQueue(pMgmt, pMsg, QUERY_QUEUE); }




×







355













UNCOV

356



int32_t vmPutMsgToFetchQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) { return vmPutMsgToQueue(pMgmt, pMsg, FETCH_QUEUE); }




×







357













UNCOV

358



int32_t vmPutMsgToStreamQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) { return vmPutMsgToQueue(pMgmt, pMsg, STREAM_QUEUE); }




×







359













UNCOV

360



int32_t vmPutMsgToStreamCtrlQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) { return vmPutMsgToQueue(pMgmt, pMsg, STREAM_CTRL_QUEUE); }




×







361













NEW


362



int32_t vmPutMsgToStreamLongExecQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) { return vmPutMsgToQueue(pMgmt, pMsg, STREAM_LONG_EXEC_QUEUE); }




×







363













UNCOV

364



int32_t vmPutMsgToMultiMgmtQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {




×








UNCOV

365



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




×








UNCOV

366



  dGTrace("msg:%p, put into vnode-multi-mgmt queue", pMsg);





×








UNCOV

367



  return taosWriteQitem(pMgmt->mgmtMultiWorker.queue, pMsg);




×







368



}










369













UNCOV

370



int32_t vmPutMsgToMgmtQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {




×








UNCOV

371



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




×








UNCOV

372



  dGTrace("msg:%p, put into vnode-mgmt queue", pMsg);





×








UNCOV

373



  return taosWriteQitem(pMgmt->mgmtWorker.queue, pMsg);




×







374



}










375













UNCOV

376



int32_t vmPutRpcMsgToQueue(SVnodeMgmt *pMgmt, EQueueType qtype, SRpcMsg *pRpc) {




×







377



  int32_t code;











UNCOV

378



  if (pRpc->contLen < sizeof(SMsgHead)) {





×







379



    dError("invalid rpc msg with no msg head at pCont. pRpc:%p, type:%s, len:%d", pRpc, TMSG_INFO(pRpc->msgType),





×







380



           pRpc->contLen);










381



    rpcFreeCont(pRpc->pCont);




×







382



    pRpc->pCont = NULL;




×







383



    return TSDB_CODE_INVALID_MSG;




×







384



  }










385













UNCOV

386



  EQItype  itype = APPLY_QUEUE == qtype ? APPLY_QITEM : RPC_QITEM;





×







387



  SRpcMsg *pMsg;











UNCOV

388



  code = taosAllocateQitem(sizeof(SRpcMsg), itype, pRpc->contLen, (void **)&pMsg);




×








UNCOV

389



  if (code) {





×







390



    rpcFreeCont(pRpc->pCont);




×







391



    pRpc->pCont = NULL;




×







392



    return code;




×







393



  }










394













UNCOV

395



  SMsgHead *pHead = pRpc->pCont;




×








UNCOV

396



  dTrace("vgId:%d, msg:%p is created, type:%s len:%d", pHead->vgId, pMsg, TMSG_INFO(pRpc->msgType), pRpc->contLen);





×







397













UNCOV

398



  pHead->contLen = htonl(pHead->contLen);




×








UNCOV

399



  pHead->vgId = htonl(pHead->vgId);




×








UNCOV

400



  memcpy(pMsg, pRpc, sizeof(SRpcMsg));




×








UNCOV

401



  pRpc->pCont = NULL;




×







402













UNCOV

403



  code = vmPutMsgToQueue(pMgmt, pMsg, qtype);




×








UNCOV

404



  if (code != 0) {





×








UNCOV

405



    dTrace("msg:%p, is freed", pMsg);





×








UNCOV

406



    rpcFreeCont(pMsg->pCont);




×








UNCOV

407



    taosFreeQitem(pMsg);




×







408



  }










409













UNCOV

410



  return code;




×







411



}










412













UNCOV

413



int32_t vmGetQueueSize(SVnodeMgmt *pMgmt, int32_t vgId, EQueueType qtype) {




×








UNCOV

414



  int32_t    size = -1;




×








UNCOV

415



  SVnodeObj *pVnode = vmAcquireVnode(pMgmt, vgId);




×








UNCOV

416



  if (pVnode != NULL) {





×








UNCOV

417



    switch (qtype) {





×







418



      case WRITE_QUEUE:




×







419



        size = taosQueueItemSize(pVnode->pWriteW.queue);




×







420



        break;




×







421



      case SYNC_QUEUE:




×







422



        size = taosQueueItemSize(pVnode->pSyncW.queue);




×







423



        break;




×








UNCOV

424



      case APPLY_QUEUE:




×








UNCOV

425



        size = taosQueueItemSize(pVnode->pApplyW.queue);




×








UNCOV

426



        break;




×







427



      case QUERY_QUEUE:




×







428



        size = taosQueueItemSize(pVnode->pQueryQ);




×







429



        break;




×







430



      case FETCH_QUEUE:




×







431



        size = taosQueueItemSize(pVnode->pFetchQ);




×







432



        break;




×







433



      case STREAM_QUEUE:




×







434



        size = taosQueueItemSize(pVnode->pStreamQ);




×







435



        break;




×







436



      case STREAM_CTRL_QUEUE:




×







437



        size = taosQueueItemSize(pVnode->pStreamCtrlQ);




×








NEW


438



        break;




×








NEW


439



      case STREAM_LONG_EXEC_QUEUE:




×








NEW


440



        size = taosQueueItemSize(pVnode->pStreamLongExecQ);




×








NEW


441



        break;




×







442



      default:




×







443



        break;




×







444



    }










445



  }











UNCOV

446



  if (pVnode) vmReleaseVnode(pMgmt, pVnode);





×








UNCOV

447



  if (size < 0) {





×








UNCOV

448



    dTrace("vgId:%d, can't get size from queue since %s, qtype:%d", vgId, terrstr(), qtype);





×








UNCOV

449



    size = 0;




×







450



  }











UNCOV

451



  return size;




×







452



}










453













UNCOV

454



int32_t vmAllocQueue(SVnodeMgmt *pMgmt, SVnodeObj *pVnode) {




×








UNCOV

455



  int32_t         code = 0;




×








UNCOV

456



  SMultiWorkerCfg wcfg = {.max = 1, .name = "vnode-write", .fp = (FItems)vnodeProposeWriteMsg, .param = pVnode->pImpl};




×








UNCOV

457



  SMultiWorkerCfg scfg = {.max = 1, .name = "vnode-sync", .fp = (FItems)vmProcessSyncQueue, .param = pVnode};




×








UNCOV

458



  SMultiWorkerCfg sccfg = {.max = 1, .name = "vnode-sync-rd", .fp = (FItems)vmProcessSyncQueue, .param = pVnode};




×








UNCOV

459



  SMultiWorkerCfg acfg = {.max = 1, .name = "vnode-apply", .fp = (FItems)vnodeApplyWriteMsg, .param = pVnode->pImpl};




×








UNCOV

460



  code = tMultiWorkerInit(&pVnode->pWriteW, &wcfg);




×








UNCOV

461



  if (code) {





×







462



    return code;




×







463



  }











UNCOV

464



  code = tMultiWorkerInit(&pVnode->pSyncW, &scfg);




×








UNCOV

465



  if (code) {





×







466



    tMultiWorkerCleanup(&pVnode->pWriteW);




×







467



    return code;




×







468



  }











UNCOV

469



  code = tMultiWorkerInit(&pVnode->pSyncRdW, &sccfg);




×








UNCOV

470



  if (code) {





×







471



    tMultiWorkerCleanup(&pVnode->pWriteW);




×







472



    tMultiWorkerCleanup(&pVnode->pSyncW);




×







473



    return code;




×







474



  }











UNCOV

475



  code = tMultiWorkerInit(&pVnode->pApplyW, &acfg);




×








UNCOV

476



  if (code) {





×







477



    tMultiWorkerCleanup(&pVnode->pWriteW);




×







478



    tMultiWorkerCleanup(&pVnode->pSyncW);




×







479



    tMultiWorkerCleanup(&pVnode->pSyncRdW);




×







480



    return code;




×







481



  }










482













UNCOV

483



  pVnode->pQueryQ = tQueryAutoQWorkerAllocQueue(&pMgmt->queryPool, pVnode, (FItem)vmProcessQueryQueue);




×








UNCOV

484



  pVnode->pFetchQ = tWWorkerAllocQueue(&pMgmt->fetchPool, pVnode, (FItems)vmProcessFetchQueue);




×







485












486



  // init stream msg processing queue family











NEW


487



  pVnode->pStreamQ = tAutoQWorkerAllocQueue(&pMgmt->streamPool, pVnode, (FItem)vmProcessStreamQueue, 2);




×








UNCOV

488



  pVnode->pStreamCtrlQ = tWWorkerAllocQueue(&pMgmt->streamCtrlPool, pVnode, (FItems)vmProcessStreamCtrlQueue);




×








NEW


489



  pVnode->pStreamLongExecQ = tAutoQWorkerAllocQueue(&pMgmt->streamLongExecPool, pVnode, (FItem)vmProcessStreamLongExecQueue, 1);




×







490













UNCOV

491



  if (pVnode->pWriteW.queue == NULL || pVnode->pSyncW.queue == NULL || pVnode->pSyncRdW.queue == NULL ||





×








UNCOV

492



      pVnode->pApplyW.queue == NULL || pVnode->pQueryQ == NULL || pVnode->pStreamQ == NULL || pVnode->pFetchQ == NULL





×








NEW


493



      || pVnode->pStreamCtrlQ == NULL || pVnode->pStreamLongExecQ == NULL) {





×







494



    return TSDB_CODE_OUT_OF_MEMORY;




×







495



  }










496













UNCOV

497



  dInfo("vgId:%d, write-queue:%p is alloced, thread:%08" PRId64, pVnode->vgId, pVnode->pWriteW.queue,





×







498



        taosQueueGetThreadId(pVnode->pWriteW.queue));











UNCOV

499



  dInfo("vgId:%d, sync-queue:%p is alloced, thread:%08" PRId64, pVnode->vgId, pVnode->pSyncW.queue,





×







500



        taosQueueGetThreadId(pVnode->pSyncW.queue));











UNCOV

501



  dInfo("vgId:%d, sync-rd-queue:%p is alloced, thread:%08" PRId64, pVnode->vgId, pVnode->pSyncRdW.queue,





×







502



        taosQueueGetThreadId(pVnode->pSyncRdW.queue));











UNCOV

503



  dInfo("vgId:%d, apply-queue:%p is alloced, thread:%08" PRId64, pVnode->vgId, pVnode->pApplyW.queue,





×







504



        taosQueueGetThreadId(pVnode->pApplyW.queue));











UNCOV

505



  dInfo("vgId:%d, query-queue:%p is alloced", pVnode->vgId, pVnode->pQueryQ);





×








UNCOV

506



  dInfo("vgId:%d, fetch-queue:%p is alloced, thread:%08" PRId64, pVnode->vgId, pVnode->pFetchQ,





×







507



        taosQueueGetThreadId(pVnode->pFetchQ));











UNCOV

508



  dInfo("vgId:%d, stream-queue:%p is alloced", pVnode->vgId, pVnode->pStreamQ);





×








NEW


509



  dInfo("vgId:%d, stream-long-exec-queue:%p is alloced", pVnode->vgId, pVnode->pStreamLongExecQ);





×








UNCOV

510



  dInfo("vgId:%d, stream-ctrl-queue:%p is alloced, thread:%08" PRId64, pVnode->vgId, pVnode->pStreamCtrlQ,





×







511



        taosQueueGetThreadId(pVnode->pStreamCtrlQ));











UNCOV

512



  return 0;




×







513



}










514













UNCOV

515



void vmFreeQueue(SVnodeMgmt *pMgmt, SVnodeObj *pVnode) {




×








UNCOV

516



  tQueryAutoQWorkerFreeQueue(&pMgmt->queryPool, pVnode->pQueryQ);




×








UNCOV

517



  tAutoQWorkerFreeQueue(&pMgmt->streamPool, pVnode->pStreamQ);




×








NEW


518



  tAutoQWorkerFreeQueue(&pMgmt->streamLongExecPool, pVnode->pStreamLongExecQ);




×








UNCOV

519



  tWWorkerFreeQueue(&pMgmt->streamCtrlPool, pVnode->pStreamCtrlQ);




×








UNCOV

520



  tWWorkerFreeQueue(&pMgmt->fetchPool, pVnode->pFetchQ);




×








UNCOV

521



  pVnode->pQueryQ = NULL;




×








NEW


522



  pVnode->pFetchQ = NULL;




×







523













UNCOV

524



  pVnode->pStreamQ = NULL;




×








UNCOV

525



  pVnode->pStreamCtrlQ = NULL;




×








NEW


526



  pVnode->pStreamLongExecQ = NULL;




×







527













UNCOV

528



  dDebug("vgId:%d, queue is freed", pVnode->vgId);





×








UNCOV

529



}




×







530













UNCOV

531



int32_t vmStartWorker(SVnodeMgmt *pMgmt) {




×








NEW


532



  int32_t code = 0;




×







533













UNCOV

534



  SQueryAutoQWorkerPool *pQPool = &pMgmt->queryPool;




×








UNCOV

535



  pQPool->name = "vnode-query";




×








UNCOV

536



  pQPool->min = tsNumOfVnodeQueryThreads;




×








UNCOV

537



  pQPool->max = tsNumOfVnodeQueryThreads;




×








UNCOV

538



  if ((code = tQueryAutoQWorkerInit(pQPool)) != 0) return code;





×







539













UNCOV

540



  tsNumOfQueryThreads += tsNumOfVnodeQueryThreads;




×







541













UNCOV

542



  SAutoQWorkerPool *pStreamPool = &pMgmt->streamPool;




×








UNCOV

543



  pStreamPool->name = "vnode-stream";




×








UNCOV

544



  pStreamPool->ratio = tsRatioOfVnodeStreamThreads;




×








UNCOV

545



  if ((code = tAutoQWorkerInit(pStreamPool)) != 0) return code;





×







546













NEW


547



  SAutoQWorkerPool *pLongExecPool = &pMgmt->streamLongExecPool;




×








NEW


548



  pLongExecPool->name = "vnode-stream-long-exec";




×








NEW


549



  pLongExecPool->ratio = tsRatioOfVnodeStreamThreads/3;




×








NEW


550



  if ((code = tAutoQWorkerInit(pLongExecPool)) != 0) return code;





×







551













UNCOV

552



  SWWorkerPool *pStreamCtrlPool = &pMgmt->streamCtrlPool;




×








NEW


553



  pStreamCtrlPool->name = "vnode-stream-ctrl";




×








UNCOV

554



  pStreamCtrlPool->max = 1;




×








UNCOV

555



  if ((code = tWWorkerInit(pStreamCtrlPool)) != 0) return code;





×







556













UNCOV

557



  SWWorkerPool *pFPool = &pMgmt->fetchPool;




×








UNCOV

558



  pFPool->name = "vnode-fetch";




×








UNCOV

559



  pFPool->max = tsNumOfVnodeFetchThreads;




×








UNCOV

560



  if ((code = tWWorkerInit(pFPool)) != 0) return code;





×







561













UNCOV

562



  SSingleWorkerCfg mgmtCfg = {




×







563



      .min = 1, .max = 1, .name = "vnode-mgmt", .fp = (FItem)vmProcessMgmtQueue, .param = pMgmt};










564













UNCOV

565



  if ((code = tSingleWorkerInit(&pMgmt->mgmtWorker, &mgmtCfg)) != 0) return code;





×







566













UNCOV

567



  int32_t threadNum = 0;




×








UNCOV

568



  if (tsNumOfCores == 1) {





×







569



    threadNum = 2;




×







570



  } else {











UNCOV

571



    threadNum = tsNumOfCores;




×







572



  }











UNCOV

573



  SSingleWorkerCfg multiMgmtCfg = {.min = threadNum,




×







574



                                   .max = threadNum,










575



                                   .name = "vnode-multi-mgmt",










576



                                   .fp = (FItem)vmProcessMultiMgmtQueue,










577



                                   .param = pMgmt};










578













UNCOV

579



  if ((code = tSingleWorkerInit(&pMgmt->mgmtMultiWorker, &multiMgmtCfg)) != 0) return code;





×







580













UNCOV

581



  dDebug("vnode workers are initialized");





×








UNCOV

582



  return 0;




×







583



}










584













UNCOV

585



void vmStopWorker(SVnodeMgmt *pMgmt) {




×








UNCOV

586



  tQueryAutoQWorkerCleanup(&pMgmt->queryPool);




×








UNCOV

587



  tAutoQWorkerCleanup(&pMgmt->streamPool);




×








NEW


588



  tAutoQWorkerCleanup(&pMgmt->streamLongExecPool);




×








UNCOV

589



  tWWorkerCleanup(&pMgmt->streamCtrlPool);




×








UNCOV

590



  tWWorkerCleanup(&pMgmt->fetchPool);




×








UNCOV

591



  dDebug("vnode workers are closed");





×








UNCOV

592



}




×





















    

  • 

    •