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/vnode/src/tq/tq.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



#include "tq.h"










17



#include "osDef.h"










18



#include "taoserror.h"










19



#include "tqCommon.h"










20



#include "tstream.h"










21



#include "vnd.h"










22












23



// 0: not init










24



// 1: already inited










25



// 2: wait to be inited or cleanup










26



static int32_t tqInitialize(STQ* pTq);










27













UNCOV

28



static FORCE_INLINE bool tqIsHandleExec(STqHandle* pHandle) { return pHandle != NULL ? TMQ_HANDLE_STATUS_EXEC == pHandle->status : true; }





×








UNCOV

29



static FORCE_INLINE void tqSetHandleExec(STqHandle* pHandle) { if (pHandle != NULL) pHandle->status = TMQ_HANDLE_STATUS_EXEC; }




×








UNCOV

30



static FORCE_INLINE void tqSetHandleIdle(STqHandle* pHandle) { if (pHandle != NULL) pHandle->status = TMQ_HANDLE_STATUS_IDLE; }




×







31













UNCOV

32



void tqDestroyTqHandle(void* data) {




×








UNCOV

33



  if (data == NULL) return;





×








UNCOV

34



  STqHandle* pData = (STqHandle*)data;




×








UNCOV

35



  qDestroyTask(pData->execHandle.task);




×







36













UNCOV

37



  if (pData->execHandle.subType == TOPIC_SUB_TYPE__COLUMN) {





×








UNCOV

38



    taosMemoryFreeClear(pData->execHandle.execCol.qmsg);





×








UNCOV

39



  } else if (pData->execHandle.subType == TOPIC_SUB_TYPE__DB) {





×








UNCOV

40



    tqReaderClose(pData->execHandle.pTqReader);




×








UNCOV

41



    walCloseReader(pData->pWalReader);




×








UNCOV

42



    taosHashCleanup(pData->execHandle.execDb.pFilterOutTbUid);




×








UNCOV

43



  } else if (pData->execHandle.subType == TOPIC_SUB_TYPE__TABLE) {





×








UNCOV

44



    walCloseReader(pData->pWalReader);




×








UNCOV

45



    tqReaderClose(pData->execHandle.pTqReader);




×








UNCOV

46



    taosMemoryFreeClear(pData->execHandle.execTb.qmsg);





×








UNCOV

47



    nodesDestroyNode(pData->execHandle.execTb.node);




×







48



  }











UNCOV

49



  if (pData->msg != NULL) {





×







50



    rpcFreeCont(pData->msg->pCont);




×







51



    taosMemoryFree(pData->msg);





×







52



    pData->msg = NULL;




×







53



  }











UNCOV

54



  if (pData->block != NULL) {





×







55



    blockDataDestroy(pData->block);




×







56



  }











UNCOV

57



  if (pData->pRef) {





×








UNCOV

58



    walCloseRef(pData->pRef->pWal, pData->pRef->refId);




×







59



  }











UNCOV

60



  taosHashCleanup(pData->tableCreateTimeHash);




×







61



}










62













UNCOV

63



static bool tqOffsetEqual(const STqOffset* pLeft, const STqOffset* pRight) {




×








UNCOV

64



  if (pLeft == NULL || pRight == NULL) {





×







65



    return false;




×







66



  }











UNCOV

67



  return pLeft->val.type == TMQ_OFFSET__LOG && pRight->val.type == TMQ_OFFSET__LOG &&





×








UNCOV

68



         pLeft->val.version == pRight->val.version;





×







69



}










70













UNCOV

71



int32_t tqOpen(const char* path, SVnode* pVnode) {




×








UNCOV

72



  if (path == NULL || pVnode == NULL) {





×







73



    return TSDB_CODE_INVALID_PARA;




×







74



  }











UNCOV

75



  STQ* pTq = taosMemoryCalloc(1, sizeof(STQ));





×








UNCOV

76



  if (pTq == NULL) {





×







77



    return terrno;




×







78



  }










79













UNCOV

80



  pVnode->pTq = pTq;




×








UNCOV

81



  pTq->pVnode = pVnode;




×







82













UNCOV

83



  pTq->path = taosStrdup(path);





×








UNCOV

84



  if (pTq->path == NULL) {





×







85



    return terrno;




×







86



  }










87













UNCOV

88



  pTq->pHandle = taosHashInit(64, MurmurHash3_32, true, HASH_ENTRY_LOCK);




×








UNCOV

89



  if (pTq->pHandle == NULL) {





×







90



    return terrno;




×







91



  }











UNCOV

92



  taosHashSetFreeFp(pTq->pHandle, tqDestroyTqHandle);




×







93













UNCOV

94



  taosInitRWLatch(&pTq->lock);




×







95













UNCOV

96



  pTq->pPushMgr = taosHashInit(64, MurmurHash3_32, false, HASH_NO_LOCK);




×








UNCOV

97



  if (pTq->pPushMgr == NULL) {





×







98



    return terrno;




×







99



  }










100













UNCOV

101



  pTq->pCheckInfo = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_ENTRY_LOCK);




×








UNCOV

102



  if (pTq->pCheckInfo == NULL) {





×







103



    return terrno;




×







104



  }











UNCOV

105



  taosHashSetFreeFp(pTq->pCheckInfo, (FDelete)tDeleteSTqCheckInfo);




×







106













UNCOV

107



  pTq->pOffset = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_VARCHAR), true, HASH_ENTRY_LOCK);




×








UNCOV

108



  if (pTq->pOffset == NULL) {





×







109



    return terrno;




×







110



  }











UNCOV

111



  taosHashSetFreeFp(pTq->pOffset, (FDelete)tDeleteSTqOffset);




×







112













UNCOV

113



  return tqInitialize(pTq);




×







114



}










115













UNCOV

116



int32_t tqInitialize(STQ* pTq) {




×








UNCOV

117



  if (pTq == NULL) {





×







118



    return TSDB_CODE_INVALID_PARA;




×







119



  }











UNCOV

120



  int32_t vgId = TD_VID(pTq->pVnode);




×








UNCOV

121



  int32_t code = streamMetaOpen(pTq->path, pTq, tqBuildStreamTask, tqExpandStreamTask, vgId, -1,




×







122



                                tqStartTaskCompleteCallback, &pTq->pStreamMeta);











UNCOV

123



  if (code != TSDB_CODE_SUCCESS) {





×







124



    return code;




×







125



  }










126













UNCOV

127



  streamMetaLoadAllTasks(pTq->pStreamMeta);




×








UNCOV

128



  return tqMetaOpen(pTq);




×







129



}










130













UNCOV

131



void tqClose(STQ* pTq) {




×








UNCOV

132



  qDebug("start to close tq");





×








UNCOV

133



  if (pTq == NULL) {





×







134



    return;




×







135



  }










136













UNCOV

137



  int32_t vgId = 0;




×








UNCOV

138



  if (pTq->pVnode != NULL) {





×








UNCOV

139



    vgId = TD_VID(pTq->pVnode);




×








UNCOV

140



  } else if (pTq->pStreamMeta != NULL) {





×







141



    vgId = pTq->pStreamMeta->vgId;




×







142



  }










143












144



  // close the stream meta firstly











UNCOV

145



  streamMetaClose(pTq->pStreamMeta);




×







146













UNCOV

147



  void* pIter = taosHashIterate(pTq->pPushMgr, NULL);




×








UNCOV

148



  while (pIter) {





×








UNCOV

149



    STqHandle* pHandle = *(STqHandle**)pIter;




×








UNCOV

150



    if (pHandle->msg != NULL) {





×








UNCOV

151



      tqPushEmptyDataRsp(pHandle, vgId);




×








UNCOV

152



      rpcFreeCont(pHandle->msg->pCont);




×








UNCOV

153



      taosMemoryFree(pHandle->msg);





×








UNCOV

154



      pHandle->msg = NULL;




×







155



    }











UNCOV

156



    pIter = taosHashIterate(pTq->pPushMgr, pIter);




×







157



  }










158













UNCOV

159



  taosHashCleanup(pTq->pHandle);




×








UNCOV

160



  taosHashCleanup(pTq->pPushMgr);




×








UNCOV

161



  taosHashCleanup(pTq->pCheckInfo);




×








UNCOV

162



  taosHashCleanup(pTq->pOffset);




×








UNCOV

163



  taosMemoryFree(pTq->path);





×








UNCOV

164



  tqMetaClose(pTq);




×








UNCOV

165



  qDebug("vgId:%d end to close tq", vgId);





×







166












167



#if 0










168



  streamMetaFreeTQDuringScanWalError(pTq);










169



#endif










170













UNCOV

171



  taosMemoryFree(pTq);





×







172



}










173













UNCOV

174



void tqNotifyClose(STQ* pTq) {




×








UNCOV

175



  if (pTq == NULL) {





×







176



    return;




×







177



  }










178













UNCOV

179



  if (pTq->pStreamMeta != NULL) {





×








UNCOV

180



    streamMetaNotifyClose(pTq->pStreamMeta);




×







181



  }










182



}










183













UNCOV

184



void tqPushEmptyDataRsp(STqHandle* pHandle, int32_t vgId) {




×








UNCOV

185



  if (pHandle == NULL) {





×







186



    return;




×







187



  }











UNCOV

188



  int32_t    code = 0;




×








UNCOV

189



  SMqPollReq req = {0};




×








UNCOV

190



  code = tDeserializeSMqPollReq(pHandle->msg->pCont, pHandle->msg->contLen, &req);




×








UNCOV

191



  if (code < 0) {





×







192



    tqError("tDeserializeSMqPollReq %d failed, code:%d", pHandle->msg->contLen, code);





×







193



    return;




×







194



  }










195













UNCOV

196



  SMqDataRsp dataRsp = {0};




×








UNCOV

197



  code = tqInitDataRsp(&dataRsp, req.reqOffset);




×








UNCOV

198



  if (code != 0) {





×







199



    tqError("tqInitDataRsp failed, code:%d", code);





×







200



    return;




×







201



  }











UNCOV

202



  dataRsp.blockNum = 0;




×








UNCOV

203



  char buf[TSDB_OFFSET_LEN] = {0};




×








UNCOV

204



  (void)tFormatOffset(buf, TSDB_OFFSET_LEN, &dataRsp.reqOffset);




×








UNCOV

205



  tqInfo("tqPushEmptyDataRsp to consumer:0x%" PRIx64 " vgId:%d, offset:%s,QID:0x%" PRIx64, req.consumerId, vgId, buf,





×







206



         req.reqId);










207













UNCOV

208



  code = tqSendDataRsp(pHandle, pHandle->msg, &req, &dataRsp, TMQ_MSG_TYPE__POLL_DATA_RSP, vgId);




×








UNCOV

209



  if (code != 0) {





×







210



    tqError("tqSendDataRsp failed, code:%d", code);





×







211



  }











UNCOV

212



  tDeleteMqDataRsp(&dataRsp);




×







213



}










214













UNCOV

215



int32_t tqSendDataRsp(STqHandle* pHandle, const SRpcMsg* pMsg, const SMqPollReq* pReq, SMqDataRsp* pRsp, int32_t type,




×







216



                      int32_t vgId) {











UNCOV

217



  if (pHandle == NULL || pMsg == NULL || pReq == NULL || pRsp == NULL) {





×







218



    return TSDB_CODE_INVALID_PARA;




×







219



  }











UNCOV

220



  int64_t sver = 0, ever = 0;




×








UNCOV

221



  walReaderValidVersionRange(pHandle->execHandle.pTqReader->pWalReader, &sver, &ever);




×







222













UNCOV

223



  char buf1[TSDB_OFFSET_LEN] = {0};




×








UNCOV

224



  char buf2[TSDB_OFFSET_LEN] = {0};




×








UNCOV

225



  (void)tFormatOffset(buf1, TSDB_OFFSET_LEN, &(pRsp->reqOffset));




×








UNCOV

226



  (void)tFormatOffset(buf2, TSDB_OFFSET_LEN, &(pRsp->rspOffset));




×







227













UNCOV

228



  tqDebug("tmq poll vgId:%d consumer:0x%" PRIx64 " (epoch %d) start to send rsp, block num:%d, req:%s, rsp:%s,QID:0x%" PRIx64,





×







229



          vgId, pReq->consumerId, pReq->epoch, pRsp->blockNum, buf1, buf2, pReq->reqId);










230













UNCOV

231



  return tqDoSendDataRsp(&pMsg->info, pRsp, pReq->epoch, pReq->consumerId, type, sver, ever);




×







232



}










233













UNCOV

234



int32_t tqProcessOffsetCommitReq(STQ* pTq, int64_t sversion, char* msg, int32_t msgLen) {




×








UNCOV

235



  if (pTq == NULL) {





×







236



    return TSDB_CODE_INVALID_PARA;




×







237



  }











UNCOV

238



  SMqVgOffset vgOffset = {0};




×








UNCOV

239



  int32_t     vgId = TD_VID(pTq->pVnode);




×







240













UNCOV

241



  int32_t  code = 0;




×







242



  SDecoder decoder;











UNCOV

243



  tDecoderInit(&decoder, (uint8_t*)msg, msgLen);




×








UNCOV

244



  if (tDecodeMqVgOffset(&decoder, &vgOffset) < 0) {





×







245



    code = TSDB_CODE_INVALID_MSG;




×







246



    goto end;




×







247



  }










248













UNCOV

249



  tDecoderClear(&decoder);




×







250













UNCOV

251



  STqOffset* pOffset = &vgOffset.offset;




×







252













UNCOV

253



  if (pOffset->val.type == TMQ_OFFSET__SNAPSHOT_DATA || pOffset->val.type == TMQ_OFFSET__SNAPSHOT_META) {





×








UNCOV

254



    tqDebug("receive offset commit msg to %s on vgId:%d, offset(type:snapshot) uid:%" PRId64 ", ts:%" PRId64,





×







255



            pOffset->subKey, vgId, pOffset->val.uid, pOffset->val.ts);











UNCOV

256



  } else if (pOffset->val.type == TMQ_OFFSET__LOG) {





×








UNCOV

257



    tqDebug("receive offset commit msg to %s on vgId:%d, offset(type:log) version:%" PRId64, pOffset->subKey, vgId,





×







258



            pOffset->val.version);










259



  } else {










260



    tqError("invalid commit offset type:%d", pOffset->val.type);





×







261



    code = TSDB_CODE_INVALID_MSG;




×







262



    goto end;




×







263



  }










264













UNCOV

265



  STqOffset* pSavedOffset = NULL;




×








UNCOV

266



  code = tqMetaGetOffset(pTq, pOffset->subKey, &pSavedOffset);




×








UNCOV

267



  if (code == 0 && tqOffsetEqual(pOffset, pSavedOffset)) {





×








UNCOV

268



    tqInfo("not update the offset, vgId:%d sub:%s since committed:%" PRId64 " less than/equal to existed:%" PRId64,





×







269



           vgId, pOffset->subKey, pOffset->val.version, pSavedOffset->val.version);











UNCOV

270



    goto end;  // no need to update the offset value




×







271



  }










272












273



  // save the new offset value











UNCOV

274



  if (taosHashPut(pTq->pOffset, pOffset->subKey, strlen(pOffset->subKey), pOffset, sizeof(STqOffset))) {





×







275



    terrno = TSDB_CODE_OUT_OF_MEMORY;




×







276



    return -1;




×







277



  }










278













UNCOV

279



  if (tqMetaSaveInfo(pTq, pTq->pOffsetStore, pOffset->subKey, strlen(pOffset->subKey), msg,





×







280



                     msgLen >= sizeof(vgOffset.consumerId) ? msgLen - sizeof(vgOffset.consumerId) : 0) < 0) {










281



    terrno = TSDB_CODE_OUT_OF_MEMORY;




×







282



    return -1;




×







283



  }










284













UNCOV

285



  return 0;




×








UNCOV

286



end:




×








UNCOV

287



  tOffsetDestroy(&vgOffset.offset.val);




×








UNCOV

288



  return code;




×







289



}










290













UNCOV

291



int32_t tqProcessSeekReq(STQ* pTq, SRpcMsg* pMsg) {




×








UNCOV

292



  if (pTq == NULL || pMsg == NULL) {





×







293



    return TSDB_CODE_INVALID_PARA;




×







294



  }











UNCOV

295



  SMqSeekReq req = {0};




×








UNCOV

296



  int32_t    vgId = TD_VID(pTq->pVnode);




×








UNCOV

297



  SRpcMsg    rsp = {.info = pMsg->info};




×








UNCOV

298



  int        code = 0;




×







299













UNCOV

300



  if (tDeserializeSMqSeekReq(pMsg->pCont, pMsg->contLen, &req) < 0) {





×







301



    code = TSDB_CODE_OUT_OF_MEMORY;




×







302



    goto end;




×







303



  }










304













UNCOV

305



  tqDebug("tmq seek: consumer:0x%" PRIx64 " vgId:%d, subkey %s", req.consumerId, vgId, req.subKey);





×








UNCOV

306



  taosWLockLatch(&pTq->lock);




×







307













UNCOV

308



  STqHandle* pHandle = taosHashGet(pTq->pHandle, req.subKey, strlen(req.subKey));




×








UNCOV

309



  if (pHandle == NULL) {





×







310



    tqWarn("tmq seek: consumer:0x%" PRIx64 " vgId:%d subkey %s not found", req.consumerId, vgId, req.subKey);





×







311



    code = 0;




×







312



    taosWUnLockLatch(&pTq->lock);




×







313



    goto end;




×







314



  }










315












316



  // 2. check consumer-vg assignment status











UNCOV

317



  if (pHandle->consumerId != req.consumerId) {





×







318



    tqError("ERROR tmq seek: consumer:0x%" PRIx64 " vgId:%d, subkey %s, mismatch for saved handle consumer:0x%" PRIx64,





×







319



            req.consumerId, vgId, req.subKey, pHandle->consumerId);










320



    taosWUnLockLatch(&pTq->lock);




×







321



    code = TSDB_CODE_TMQ_CONSUMER_MISMATCH;




×







322



    goto end;




×







323



  }










324












325



  // if consumer register to push manager, push empty to consumer to change vg status from TMQ_VG_STATUS__WAIT to










326



  // TMQ_VG_STATUS__IDLE, otherwise poll data failed after seek.











UNCOV

327



  tqUnregisterPushHandle(pTq, pHandle);




×








UNCOV

328



  taosWUnLockLatch(&pTq->lock);




×







329













UNCOV

330



end:




×








UNCOV

331



  rsp.code = code;




×








UNCOV

332



  tmsgSendRsp(&rsp);




×








UNCOV

333



  return 0;




×







334



}










335













UNCOV

336



int32_t tqCheckColModifiable(STQ* pTq, int64_t tbUid, int32_t colId) {




×








UNCOV

337



  if (pTq == NULL) {





×







338



    return TSDB_CODE_INVALID_PARA;




×







339



  }











UNCOV

340



  void* pIter = NULL;




×







341













UNCOV

342



  while (1) {




×








UNCOV

343



    pIter = taosHashIterate(pTq->pCheckInfo, pIter);




×








UNCOV

344



    if (pIter == NULL) {





×








UNCOV

345



      break;




×







346



    }










347













UNCOV

348



    STqCheckInfo* pCheck = (STqCheckInfo*)pIter;




×







349













UNCOV

350



    if (pCheck->ntbUid == tbUid) {





×








UNCOV

351



      int32_t sz = taosArrayGetSize(pCheck->colIdList);




×








UNCOV

352



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





×








UNCOV

353



        int16_t* pForbidColId = taosArrayGet(pCheck->colIdList, i);




×








UNCOV

354



        if (pForbidColId == NULL) {





×







355



          continue;




×







356



        }










357













UNCOV

358



        if ((*pForbidColId) == colId) {





×








UNCOV

359



          taosHashCancelIterate(pTq->pCheckInfo, pIter);




×








UNCOV

360



          return -1;




×







361



        }










362



      }










363



    }










364



  }










365













UNCOV

366



  return 0;




×







367



}










368













UNCOV

369



int32_t tqProcessPollPush(STQ* pTq, SRpcMsg* pMsg) {




×








UNCOV

370



  if (pTq == NULL) {





×







371



    return TSDB_CODE_INVALID_PARA;




×







372



  }











UNCOV

373



  int32_t vgId = TD_VID(pTq->pVnode);




×








UNCOV

374



  taosWLockLatch(&pTq->lock);




×








UNCOV

375



  if (taosHashGetSize(pTq->pPushMgr) > 0) {





×








UNCOV

376



    void* pIter = taosHashIterate(pTq->pPushMgr, NULL);




×







377













UNCOV

378



    while (pIter) {





×








UNCOV

379



      STqHandle* pHandle = *(STqHandle**)pIter;




×








UNCOV

380



      tqDebug("vgId:%d start set submit for pHandle:%p, consumer:0x%" PRIx64, vgId, pHandle, pHandle->consumerId);





×







381













UNCOV

382



      if (pHandle->msg == NULL) {





×







383



        tqError("pHandle->msg should not be null");





×







384



        taosHashCancelIterate(pTq->pPushMgr, pIter);




×







385



        break;




×







386



      } else {











UNCOV

387



        SRpcMsg msg = {.msgType = TDMT_VND_TMQ_CONSUME,




×








UNCOV

388



                       .pCont = pHandle->msg->pCont,




×








UNCOV

389



                       .contLen = pHandle->msg->contLen,




×








UNCOV

390



                       .info = pHandle->msg->info};




×








UNCOV

391



        if (tmsgPutToQueue(&pTq->pVnode->msgCb, QUERY_QUEUE, &msg) != 0){





×







392



          tqError("vgId:%d tmsgPutToQueue failed, consumer:0x%" PRIx64, vgId, pHandle->consumerId);





×







393



        }











UNCOV

394



        taosMemoryFree(pHandle->msg);





×








UNCOV

395



        pHandle->msg = NULL;




×







396



      }










397













UNCOV

398



      pIter = taosHashIterate(pTq->pPushMgr, pIter);




×







399



    }










400













UNCOV

401



    taosHashClear(pTq->pPushMgr);




×







402



  }











UNCOV

403



  taosWUnLockLatch(&pTq->lock);




×








UNCOV

404



  return 0;




×







405



}










406













UNCOV

407



int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) {




×








UNCOV

408



  if (pTq == NULL || pMsg == NULL) {





×







409



    return TSDB_CODE_INVALID_PARA;




×







410



  }











UNCOV

411



  SMqPollReq req = {0};




×








UNCOV

412



  int        code = tDeserializeSMqPollReq(pMsg->pCont, pMsg->contLen, &req);




×








UNCOV

413



  if (code < 0) {





×







414



    tqError("tDeserializeSMqPollReq %d failed", pMsg->contLen);





×







415



    terrno = TSDB_CODE_INVALID_MSG;




×







416



    goto END;




×







417



  }











UNCOV

418



  if (req.rawData == 1){





×








UNCOV

419



    req.uidHash = taosHashInit(8, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, HASH_NO_LOCK);




×








UNCOV

420



    if (req.uidHash == NULL) {





×







421



      tqError("tq poll rawData taosHashInit failed");





×







422



      code = terrno;




×







423



      goto END;




×







424



    }










425



  }











UNCOV

426



  int64_t      consumerId = req.consumerId;




×








UNCOV

427



  int32_t      reqEpoch = req.epoch;




×








UNCOV

428



  STqOffsetVal reqOffset = req.reqOffset;




×








UNCOV

429



  int32_t      vgId = TD_VID(pTq->pVnode);




×








UNCOV

430



  STqHandle*   pHandle = NULL;




×







431













UNCOV

432



  while (1) {




×








UNCOV

433



    taosWLockLatch(&pTq->lock);




×







434



    // 1. find handle











UNCOV

435



    code = tqMetaGetHandle(pTq, req.subKey, &pHandle);




×








UNCOV

436



    if (code != TDB_CODE_SUCCESS) {





×








UNCOV

437



      tqError("tmq poll: consumer:0x%" PRIx64 " vgId:%d subkey %s not found", consumerId, vgId, req.subKey);





×








UNCOV

438



      terrno = TSDB_CODE_INVALID_MSG;




×








UNCOV

439



      taosWUnLockLatch(&pTq->lock);




×








UNCOV

440



      return -1;




×







441



    }










442












443



    // 2. check rebalance status











UNCOV

444



    if (pHandle->consumerId != consumerId) {





×








UNCOV

445



      tqError("ERROR tmq poll: consumer:0x%" PRIx64





×







446



              " vgId:%d, subkey %s, mismatch for saved handle consumer:0x%" PRIx64,










447



              consumerId, TD_VID(pTq->pVnode), req.subKey, pHandle->consumerId);











UNCOV

448



      terrno = TSDB_CODE_TMQ_CONSUMER_MISMATCH;




×








UNCOV

449



      taosWUnLockLatch(&pTq->lock);




×








UNCOV

450



      code = -1;




×








UNCOV

451



      goto END;




×







452



    }










453













UNCOV

454



    bool exec = tqIsHandleExec(pHandle);





×








UNCOV

455



    if (!exec) {





×








UNCOV

456



      tqSetHandleExec(pHandle);





×







457



      //      qSetTaskCode(pHandle->execHandle.task, TDB_CODE_SUCCESS);











UNCOV

458



      tqDebug("tmq poll: consumer:0x%" PRIx64 " vgId:%d, topic:%s, set handle exec, pHandle:%p", consumerId, vgId,





×







459



              req.subKey, pHandle);











UNCOV

460



      taosWUnLockLatch(&pTq->lock);




×








UNCOV

461



      break;




×







462



    }










463



    taosWUnLockLatch(&pTq->lock);




×







464













UNCOV

465



    tqDebug("tmq poll: consumer:0x%" PRIx64





×







466



            " vgId:%d, topic:%s, subscription is executing, wait for 10ms and retry, pHandle:%p",










467



            consumerId, vgId, req.subKey, pHandle);











UNCOV

468



    taosMsleep(10);




×







469



  }










470












471



  // 3. update the epoch value











UNCOV

472



  if (pHandle->epoch < reqEpoch) {





×








UNCOV

473



    tqDebug("tmq poll: consumer:0x%" PRIx64 " epoch update from %d to %d by poll req", consumerId, pHandle->epoch,





×







474



            reqEpoch);











UNCOV

475



    pHandle->epoch = reqEpoch;




×







476



  }










477













UNCOV

478



  char buf[TSDB_OFFSET_LEN] = {0};




×








UNCOV

479



  (void)tFormatOffset(buf, TSDB_OFFSET_LEN, &reqOffset);




×








UNCOV

480



  tqDebug("tmq poll: consumer:0x%" PRIx64 " (epoch %d), subkey %s, recv poll req vgId:%d, req:%s,QID:0x%" PRIx64,





×







481



          consumerId, req.epoch, pHandle->subKey, vgId, buf, req.reqId);










482













UNCOV

483



  code = tqExtractDataForMq(pTq, pHandle, &req, pMsg);




×








UNCOV

484



  tqSetHandleIdle(pHandle);





×







485













UNCOV

486



  tqDebug("tmq poll: consumer:0x%" PRIx64 " vgId:%d, topic:%s, set handle idle, pHandle:%p", consumerId, vgId,





×







487



          req.subKey, pHandle);










488












489



END:




×








UNCOV

490



  tDestroySMqPollReq(&req);




×








UNCOV

491



  return code;




×







492



}










493













UNCOV

494



int32_t tqProcessVgCommittedInfoReq(STQ* pTq, SRpcMsg* pMsg) {




×








UNCOV

495



  if (pTq == NULL || pMsg == NULL) {





×







496



    return TSDB_CODE_INVALID_PARA;




×







497



  }











UNCOV

498



  void*   data = POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead));




×








UNCOV

499



  int32_t len = pMsg->contLen - sizeof(SMsgHead);




×







500













UNCOV

501



  SMqVgOffset vgOffset = {0};




×







502












503



  SDecoder decoder;











UNCOV

504



  tDecoderInit(&decoder, (uint8_t*)data, len);




×








UNCOV

505



  if (tDecodeMqVgOffset(&decoder, &vgOffset) < 0) {





×







506



    terrno = TSDB_CODE_OUT_OF_MEMORY;




×







507



    return terrno;




×







508



  }










509













UNCOV

510



  tDecoderClear(&decoder);




×







511













UNCOV

512



  STqOffset* pSavedOffset = NULL;




×








UNCOV

513



  int32_t    code = tqMetaGetOffset(pTq, vgOffset.offset.subKey, &pSavedOffset);




×








UNCOV

514



  if (code != 0) {





×








UNCOV

515



    return TSDB_CODE_TMQ_NO_COMMITTED;




×







516



  }











UNCOV

517



  vgOffset.offset = *pSavedOffset;




×







518













UNCOV

519



  tEncodeSize(tEncodeMqVgOffset, &vgOffset, len, code);





×








UNCOV

520



  if (code < 0) {





×







521



    return TAOS_GET_TERRNO(TSDB_CODE_INVALID_PARA);





×







522



  }










523













UNCOV

524



  void* buf = rpcMallocCont(len);




×








UNCOV

525



  if (buf == NULL) {





×







526



    return terrno;




×







527



  }











UNCOV

528



  SEncoder encoder = {0};




×








UNCOV

529



  tEncoderInit(&encoder, buf, len);




×








UNCOV

530



  code = tEncodeMqVgOffset(&encoder, &vgOffset);




×








UNCOV

531



  tEncoderClear(&encoder);




×








UNCOV

532



  if (code < 0) {





×







533



    rpcFreeCont(buf);




×







534



    return TAOS_GET_TERRNO(TSDB_CODE_INVALID_PARA);





×







535



  }










536













UNCOV

537



  SRpcMsg rsp = {.info = pMsg->info, .pCont = buf, .contLen = len, .code = 0};




×







538













UNCOV

539



  tmsgSendRsp(&rsp);




×








UNCOV

540



  return 0;




×







541



}










542













UNCOV

543



int32_t tqProcessVgWalInfoReq(STQ* pTq, SRpcMsg* pMsg) {




×








UNCOV

544



  if (pTq == NULL || pMsg == NULL) {





×







545



    return TSDB_CODE_INVALID_PARA;




×







546



  }











UNCOV

547



  int32_t    code = 0;




×








UNCOV

548



  SMqPollReq req = {0};




×








UNCOV

549



  if (tDeserializeSMqPollReq(pMsg->pCont, pMsg->contLen, &req) < 0) {





×







550



    tqError("tDeserializeSMqPollReq %d failed", pMsg->contLen);





×







551



    return TSDB_CODE_INVALID_MSG;




×







552



  }










553













UNCOV

554



  int64_t      consumerId = req.consumerId;




×








UNCOV

555



  STqOffsetVal reqOffset = req.reqOffset;




×








UNCOV

556



  int32_t      vgId = TD_VID(pTq->pVnode);




×







557












558



  // 1. find handle











UNCOV

559



  taosRLockLatch(&pTq->lock);




×








UNCOV

560



  STqHandle* pHandle = taosHashGet(pTq->pHandle, req.subKey, strlen(req.subKey));




×








UNCOV

561



  if (pHandle == NULL) {





×







562



    tqError("consumer:0x%" PRIx64 " vgId:%d subkey:%s not found", consumerId, vgId, req.subKey);





×







563



    taosRUnLockLatch(&pTq->lock);




×







564



    return TSDB_CODE_INVALID_MSG;




×







565



  }










566












567



  // 2. check rebalance status











UNCOV

568



  if (pHandle->consumerId != consumerId) {





×







569



    tqDebug("ERROR consumer:0x%" PRIx64 " vgId:%d, subkey %s, mismatch for saved handle consumer:0x%" PRIx64,





×







570



            consumerId, vgId, req.subKey, pHandle->consumerId);










571



    taosRUnLockLatch(&pTq->lock);




×







572



    return TSDB_CODE_TMQ_CONSUMER_MISMATCH;




×







573



  }










574













UNCOV

575



  int64_t sver = 0, ever = 0;




×








UNCOV

576



  walReaderValidVersionRange(pHandle->execHandle.pTqReader->pWalReader, &sver, &ever);




×








UNCOV

577



  taosRUnLockLatch(&pTq->lock);




×







578













UNCOV

579



  SMqDataRsp dataRsp = {0};




×








UNCOV

580



  code = tqInitDataRsp(&dataRsp, req.reqOffset);




×








UNCOV

581



  if (code != 0) {





×







582



    return code;




×







583



  }










584













UNCOV

585



  if (req.useSnapshot == true) {





×







586



    tqError("consumer:0x%" PRIx64 " vgId:%d subkey:%s snapshot not support wal info", consumerId, vgId, req.subKey);





×







587



    code = TSDB_CODE_INVALID_PARA;




×







588



    goto END;




×







589



  }










590













UNCOV

591



  dataRsp.rspOffset.type = TMQ_OFFSET__LOG;




×







592













UNCOV

593



  if (reqOffset.type == TMQ_OFFSET__LOG) {





×








UNCOV

594



    dataRsp.rspOffset.version = reqOffset.version;




×








UNCOV

595



  } else if (reqOffset.type < 0) {





×








UNCOV

596



    STqOffset* pOffset = NULL;




×








UNCOV

597



    code = tqMetaGetOffset(pTq, req.subKey, &pOffset);




×








UNCOV

598



    if (code == 0) {





×








UNCOV

599



      if (pOffset->val.type != TMQ_OFFSET__LOG) {





×







600



        tqError("consumer:0x%" PRIx64 " vgId:%d subkey:%s, no valid wal info", consumerId, vgId, req.subKey);





×







601



        code = TSDB_CODE_INVALID_PARA;




×







602



        goto END;




×







603



      }










604













UNCOV

605



      dataRsp.rspOffset.version = pOffset->val.version;




×








UNCOV

606



      tqInfo("consumer:0x%" PRIx64 " vgId:%d subkey:%s get assignment from store:%" PRId64, consumerId, vgId,





×







607



             req.subKey, dataRsp.rspOffset.version);










608



    } else {











UNCOV

609



      if (reqOffset.type == TMQ_OFFSET__RESET_EARLIEST) {





×








UNCOV

610



        dataRsp.rspOffset.version = sver;  // not consume yet, set the earliest position




×








UNCOV

611



      } else if (reqOffset.type == TMQ_OFFSET__RESET_LATEST) {





×








UNCOV

612



        dataRsp.rspOffset.version = ever;




×







613



      }











UNCOV

614



      tqInfo("consumer:0x%" PRIx64 " vgId:%d subkey:%s get assignment from init:%" PRId64, consumerId, vgId, req.subKey,





×







615



             dataRsp.rspOffset.version);










616



    }










617



  } else {










618



    tqError("consumer:0x%" PRIx64 " vgId:%d subkey:%s invalid offset type:%d", consumerId, vgId, req.subKey,





×







619



            reqOffset.type);










620



    code = TSDB_CODE_INVALID_PARA;




×







621



    goto END;




×







622



  }










623













UNCOV

624



  code = tqDoSendDataRsp(&pMsg->info, &dataRsp, req.epoch, req.consumerId, TMQ_MSG_TYPE__WALINFO_RSP, sver, ever);




×







625













UNCOV

626



END:




×








UNCOV

627



  tDeleteMqDataRsp(&dataRsp);




×








UNCOV

628



  return code;




×







629



}










630













UNCOV

631



int32_t tqProcessDeleteSubReq(STQ* pTq, int64_t sversion, char* msg, int32_t msgLen) {




×








UNCOV

632



  if (pTq == NULL || msg == NULL) {





×







633



    return TSDB_CODE_INVALID_PARA;




×







634



  }











UNCOV

635



  SMqVDeleteReq* pReq = (SMqVDeleteReq*)msg;




×








UNCOV

636



  int32_t        vgId = TD_VID(pTq->pVnode);




×







637













UNCOV

638



  tqInfo("vgId:%d, tq process delete sub req %s", vgId, pReq->subKey);





×








UNCOV

639



  int32_t code = 0;




×







640













UNCOV

641



  STqHandle* pHandle = taosHashGet(pTq->pHandle, pReq->subKey, strlen(pReq->subKey));




×








UNCOV

642



  if (pHandle) {





×







643



    while (1) {




×








UNCOV

644



      taosWLockLatch(&pTq->lock);




×








UNCOV

645



      bool exec = tqIsHandleExec(pHandle);




×







646













UNCOV

647



      if (exec) {





×







648



        tqInfo("vgId:%d, topic:%s, subscription is executing, delete wait for 10ms and retry, pHandle:%p", vgId,





×







649



               pHandle->subKey, pHandle);










650



        taosWUnLockLatch(&pTq->lock);




×







651



        taosMsleep(10);




×







652



        continue;




×







653



      }











UNCOV

654



      tqUnregisterPushHandle(pTq, pHandle);




×








UNCOV

655



      code = taosHashRemove(pTq->pHandle, pReq->subKey, strlen(pReq->subKey));




×








UNCOV

656



      if (code != 0) {





×







657



        tqError("cannot process tq delete req %s, since no such handle", pReq->subKey);





×







658



      }











UNCOV

659



      taosWUnLockLatch(&pTq->lock);




×








UNCOV

660



      break;




×







661



    }










662



  }










663













UNCOV

664



  taosWLockLatch(&pTq->lock);




×








UNCOV

665



  if (taosHashRemove(pTq->pOffset, pReq->subKey, strlen(pReq->subKey)) != 0) {





×








UNCOV

666



    tqError("cannot process tq delete req %s, since no such offset in hash", pReq->subKey);





×







667



  }











UNCOV

668



  if (tqMetaDeleteInfo(pTq, pTq->pOffsetStore, pReq->subKey, strlen(pReq->subKey)) != 0) {





×








UNCOV

669



    tqError("cannot process tq delete req %s, since no such offset in tdb", pReq->subKey);





×







670



  }










671













UNCOV

672



  if (tqMetaDeleteInfo(pTq, pTq->pExecStore, pReq->subKey, strlen(pReq->subKey)) < 0) {





×







673



    tqError("cannot process tq delete req %s, since no such offset in tdb", pReq->subKey);





×







674



  }











UNCOV

675



  taosWUnLockLatch(&pTq->lock);




×







676













UNCOV

677



  return 0;




×







678



}










679













UNCOV

680



int32_t tqProcessAddCheckInfoReq(STQ* pTq, int64_t sversion, char* msg, int32_t msgLen) {




×








UNCOV

681



  if (pTq == NULL || msg == NULL) {





×







682



    return TSDB_CODE_INVALID_PARA;




×







683



  }











UNCOV

684



  STqCheckInfo info = {0};




×








UNCOV

685



  int32_t      code = tqMetaDecodeCheckInfo(&info, msg, msgLen >= 0 ? msgLen : 0);




×








UNCOV

686



  if (code != 0) {





×







687



    return code;




×







688



  }










689













UNCOV

690



  code = taosHashPut(pTq->pCheckInfo, info.topic, strlen(info.topic), &info, sizeof(STqCheckInfo));




×








UNCOV

691



  if (code != 0) {





×







692



    tDeleteSTqCheckInfo(&info);




×







693



    return code;




×







694



  }










695













UNCOV

696



  return tqMetaSaveInfo(pTq, pTq->pCheckStore, info.topic, strlen(info.topic), msg, msgLen >= 0 ? msgLen : 0);




×







697



}










698













UNCOV

699



int32_t tqProcessDelCheckInfoReq(STQ* pTq, int64_t sversion, char* msg, int32_t msgLen) {




×








UNCOV

700



  if (pTq == NULL || msg == NULL) {





×







701



    return TSDB_CODE_INVALID_PARA;




×







702



  }











UNCOV

703



  if (taosHashRemove(pTq->pCheckInfo, msg, strlen(msg)) < 0) {





×








UNCOV

704



    return TSDB_CODE_TSC_INTERNAL_ERROR;




×







705



  }











UNCOV

706



  return tqMetaDeleteInfo(pTq, pTq->pCheckStore, msg, strlen(msg));




×







707



}










708













UNCOV

709



int32_t tqProcessSubscribeReq(STQ* pTq, int64_t sversion, char* msg, int32_t msgLen) {




×








UNCOV

710



  if (pTq == NULL || msg == NULL) {





×







711



    return TSDB_CODE_INVALID_PARA;




×







712



  }











UNCOV

713



  int         ret = 0;




×








UNCOV

714



  SMqRebVgReq req = {0};




×








UNCOV

715



  SDecoder    dc = {0};




×







716













UNCOV

717



  tDecoderInit(&dc, (uint8_t*)msg, msgLen);




×








UNCOV

718



  ret = tDecodeSMqRebVgReq(&dc, &req);




×








UNCOV

719



  if (ret < 0) {





×







720



    goto end;




×







721



  }










722













UNCOV

723



  tqInfo("vgId:%d, tq process sub req:%s, Id:0x%" PRIx64 " -> Id:0x%" PRIx64, pTq->pVnode->config.vgId, req.subKey,





×







724



         req.oldConsumerId, req.newConsumerId);










725













UNCOV

726



  taosRLockLatch(&pTq->lock);




×








UNCOV

727



  STqHandle* pHandle = NULL;




×








UNCOV

728



  int32_t code = tqMetaGetHandle(pTq, req.subKey, &pHandle);




×








UNCOV

729



  if (code != 0){





×








UNCOV

730



    tqInfo("vgId:%d, tq process sub req:%s, no such handle, create new one", pTq->pVnode->config.vgId, req.subKey);





×







731



  }











UNCOV

732



  taosRUnLockLatch(&pTq->lock);




×








UNCOV

733



  if (pHandle == NULL) {





×








UNCOV

734



    if (req.oldConsumerId != -1) {





×








UNCOV

735



      tqError("vgId:%d, build new consumer handle %s for consumer:0x%" PRIx64 ", but old consumerId:0x%" PRIx64,





×







736



              req.vgId, req.subKey, req.newConsumerId, req.oldConsumerId);










737



    }











UNCOV

738



    if (req.newConsumerId == -1) {





×








UNCOV

739



      tqError("vgId:%d, tq invalid rebalance request, new consumerId %" PRId64 "", req.vgId, req.newConsumerId);





×








UNCOV

740



      ret = TSDB_CODE_INVALID_PARA;




×








UNCOV

741



      goto end;




×







742



    }











UNCOV

743



    STqHandle handle = {0};




×








UNCOV

744



    ret = tqMetaCreateHandle(pTq, &req, &handle);




×








UNCOV

745



    if (ret < 0) {





×







746



      tqDestroyTqHandle(&handle);




×







747



      goto end;




×







748



    }











UNCOV

749



    taosWLockLatch(&pTq->lock);




×








UNCOV

750



    ret = tqMetaSaveHandle(pTq, req.subKey, &handle);




×








UNCOV

751



    taosWUnLockLatch(&pTq->lock);




×







752



  } else {











UNCOV

753



    while (1) {




×








UNCOV

754



      taosWLockLatch(&pTq->lock);




×








UNCOV

755



      bool exec = tqIsHandleExec(pHandle);





×








UNCOV

756



      if (exec) {





×








UNCOV

757



        tqInfo("vgId:%d, topic:%s, subscription is executing, sub wait for 10ms and retry, pHandle:%p",





×







758



               pTq->pVnode->config.vgId, pHandle->subKey, pHandle);











UNCOV

759



        taosWUnLockLatch(&pTq->lock);




×








UNCOV

760



        taosMsleep(10);




×








UNCOV

761



        continue;




×







762



      }











UNCOV

763



      if (pHandle->consumerId == req.newConsumerId) {  // do nothing





×








UNCOV

764



        tqInfo("vgId:%d no switch consumer:0x%" PRIx64 " remains, because redo wal log", req.vgId, req.newConsumerId);





×







765



      } else {











UNCOV

766



        tqInfo("vgId:%d switch consumer from Id:0x%" PRIx64 " to Id:0x%" PRIx64, req.vgId, pHandle->consumerId,





×







767



               req.newConsumerId);










768













UNCOV

769



        atomic_store_64(&pHandle->consumerId, req.newConsumerId);




×








UNCOV

770



        atomic_store_32(&pHandle->epoch, 0);




×








UNCOV

771



        tqUnregisterPushHandle(pTq, pHandle);




×








UNCOV

772



        ret = tqMetaSaveHandle(pTq, req.subKey, pHandle);




×







773



      }











UNCOV

774



      taosWUnLockLatch(&pTq->lock);




×








UNCOV

775



      break;




×







776



    }










777



  }










778













UNCOV

779



end:




×








UNCOV

780



  tDecoderClear(&dc);




×








UNCOV

781



  return ret;




×







782



}










783













UNCOV

784



static void freePtr(void* ptr) { taosMemoryFree(*(void**)ptr); }





×







785













UNCOV

786



int32_t tqBuildStreamTask(void* pTqObj, SStreamTask* pTask, int64_t nextProcessVer) {




×








UNCOV

787



  STQ*             pTq = (STQ*)pTqObj;




×








UNCOV

788



  int32_t          vgId = TD_VID(pTq->pVnode);




×








UNCOV

789



  SCheckpointInfo* pChkInfo = NULL;




×







790













UNCOV

791



  tqDebug("s-task:0x%x start to build task", pTask->id.taskId);





×







792













UNCOV

793



  int32_t code = streamTaskInit(pTask, pTq->pStreamMeta, &pTq->pVnode->msgCb, nextProcessVer);




×








UNCOV

794



  if (code != TSDB_CODE_SUCCESS) {





×







795



    return code;




×







796



  }










797













UNCOV

798



  pTask->pBackend = NULL;




×







799












800



  // sink











UNCOV

801



  STaskOutputInfo* pOutputInfo = &pTask->outputInfo;




×








UNCOV

802



  if (pOutputInfo->type == TASK_OUTPUT__SMA) {





×








UNCOV

803



    pOutputInfo->smaSink.vnode = pTq->pVnode;




×








UNCOV

804



    pOutputInfo->smaSink.smaSink = smaHandleRes;




×








UNCOV

805



  } else if (pOutputInfo->type == TASK_OUTPUT__TABLE) {





×








UNCOV

806



    pOutputInfo->tbSink.vnode = pTq->pVnode;




×








UNCOV

807



    pOutputInfo->tbSink.tbSinkFunc = tqSinkDataIntoDstTable;




×







808













UNCOV

809



    int32_t   ver1 = 1;




×








UNCOV

810



    SMetaInfo info = {0};




×








UNCOV

811



    code = metaGetInfo(pTq->pVnode->pMeta, pOutputInfo->tbSink.stbUid, &info, NULL);




×








UNCOV

812



    if (code == TSDB_CODE_SUCCESS) {





×








UNCOV

813



      ver1 = info.skmVer;




×







814



    }










815













UNCOV

816



    SSchemaWrapper* pschemaWrapper = pOutputInfo->tbSink.pSchemaWrapper;




×








UNCOV

817



    pOutputInfo->tbSink.pTSchema = tBuildTSchema(pschemaWrapper->pSchema, pschemaWrapper->nCols, ver1);




×








UNCOV

818



    if (pOutputInfo->tbSink.pTSchema == NULL) {





×







819



      return terrno;




×







820



    }










821













UNCOV

822



    pOutputInfo->tbSink.pTbInfo = tSimpleHashInit(10240, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT));




×








UNCOV

823



    if (pOutputInfo->tbSink.pTbInfo == NULL) {





×







824



      tqError("vgId:%d failed init sink tableInfo, code:%s", vgId, tstrerror(terrno));





×







825



      return terrno;




×







826



    }










827













UNCOV

828



    tSimpleHashSetFreeFp(pOutputInfo->tbSink.pTbInfo, freePtr);




×







829



  }










830













UNCOV

831



  if (pTask->info.taskLevel == TASK_LEVEL__SOURCE) {





×








UNCOV

832



    bool scanDropCtb = pTask->subtableWithoutMd5 ? true : false;




×








UNCOV

833



    SWalFilterCond cond = {.deleteMsg = 1, .scanDropCtb = scanDropCtb};  // delete msg also extract from wal files




×








UNCOV

834



    pTask->exec.pWalReader = walOpenReader(pTq->pVnode->pWal, &cond, pTask->id.taskId);




×








UNCOV

835



    if (pTask->exec.pWalReader == NULL) {





×







836



      tqError("vgId:%d failed init wal reader, code:%s", vgId, tstrerror(terrno));





×







837



      return terrno;




×







838



    }










839



  }










840













UNCOV

841



  streamTaskResetUpstreamStageInfo(pTask);




×







842













UNCOV

843



  pChkInfo = &pTask->chkInfo;




×








UNCOV

844



  tqSetRestoreVersionInfo(pTask);




×







845













UNCOV

846



  char*       p = streamTaskGetStatus(pTask).name;




×








UNCOV

847



  const char* pNext = streamTaskGetStatusStr(pTask->status.taskStatus);




×







848













UNCOV

849



  if (pTask->info.fillHistory) {





×








UNCOV

850



    tqInfo("vgId:%d build stream task, s-task:%s, %p checkpointId:%" PRId64 " checkpointVer:%" PRId64





×







851



           " nextProcessVer:%" PRId64










852



           " child id:%d, level:%d, cur-status:%s, next-status:%s fill-history:%d, related stream task:0x%x "










853



           "delaySched:%" PRId64 " ms, inputVer:%" PRId64,










854



           vgId, pTask->id.idStr, pTask, pChkInfo->checkpointId, pChkInfo->checkpointVer, pChkInfo->nextProcessVer,










855



           pTask->info.selfChildId, pTask->info.taskLevel, p, pNext, pTask->info.fillHistory,










856



           (int32_t)pTask->streamTaskId.taskId, pTask->info.delaySchedParam, nextProcessVer);










857



  } else {











UNCOV

858



    tqInfo("vgId:%d build stream task, s-task:%s, %p checkpointId:%" PRId64 " checkpointVer:%" PRId64





×







859



           " nextProcessVer:%" PRId64










860



           " child id:%d, level:%d, cur-status:%s next-status:%s fill-history:%d, related fill-task:0x%x "










861



           "delaySched:%" PRId64 " ms, inputVer:%" PRId64,










862



           vgId, pTask->id.idStr, pTask, pChkInfo->checkpointId, pChkInfo->checkpointVer, pChkInfo->nextProcessVer,










863



           pTask->info.selfChildId, pTask->info.taskLevel, p, pNext, pTask->info.fillHistory,










864



           (int32_t)pTask->hTaskInfo.id.taskId, pTask->info.delaySchedParam, nextProcessVer);










865













UNCOV

866



    if (pChkInfo->checkpointVer > pChkInfo->nextProcessVer) {





×







867



      tqError("vgId:%d build stream task, s-task:%s, checkpointVer:%" PRId64 " > nextProcessVer:%" PRId64, vgId,





×







868



              pTask->id.idStr, pChkInfo->checkpointVer, pChkInfo->nextProcessVer);










869



      return TSDB_CODE_STREAM_INTERNAL_ERROR;




×







870



    }










871



  }










872













UNCOV

873



  return 0;




×







874



}










875













UNCOV

876



int32_t tqProcessTaskCheckReq(STQ* pTq, SRpcMsg* pMsg) { return tqStreamTaskProcessCheckReq(pTq->pStreamMeta, pMsg); }




×







877













UNCOV

878



int32_t tqProcessTaskCheckRsp(STQ* pTq, SRpcMsg* pMsg) {




×








UNCOV

879



  return tqStreamTaskProcessCheckRsp(pTq->pStreamMeta, pMsg, vnodeIsRoleLeader(pTq->pVnode));




×







880



}










881













UNCOV

882



int32_t tqProcessTaskDeployReq(STQ* pTq, int64_t sversion, char* msg, int32_t msgLen) {




×








UNCOV

883



  return tqStreamTaskProcessDeployReq(pTq->pStreamMeta, &pTq->pVnode->msgCb, sversion, msg, msgLen,




×








UNCOV

884



                                      vnodeIsRoleLeader(pTq->pVnode), pTq->pVnode->restored);




×







885



}










886













UNCOV

887



static void doStartFillhistoryStep2(SStreamTask* pTask, SStreamTask* pStreamTask, STQ* pTq) {




×








UNCOV

888



  const char*    id = pTask->id.idStr;




×








UNCOV

889



  int64_t        nextProcessedVer = pStreamTask->hTaskInfo.haltVer;




×








UNCOV

890



  SVersionRange* pStep2Range = &pTask->step2Range;




×








UNCOV

891



  int32_t        vgId = pTask->pMeta->vgId;




×







892












893



  // if it's an source task, extract the last version in wal.











UNCOV

894



  bool done = streamHistoryTaskSetVerRangeStep2(pTask, nextProcessedVer);




×








UNCOV

895



  pTask->execInfo.step2Start = taosGetTimestampMs();




×







896













UNCOV

897



  if (done) {





×








UNCOV

898



    qDebug("s-task:%s scan wal(step 2) verRange:%" PRId64 "-%" PRId64 " ended, elapsed time:%.2fs", id,





×







899



           pStep2Range->minVer, pStep2Range->maxVer, 0.0);











UNCOV

900



    int32_t code = streamTaskPutTranstateIntoInputQ(pTask);  // todo: msg lost.




×








UNCOV

901



    if (code) {





×







902



      qError("s-task:%s failed put trans-state into inputQ, code:%s", id, tstrerror(code));





×







903



    }











UNCOV

904



    (void)streamExecTask(pTask);  // exec directly




×







905



  } else {











UNCOV

906



    STimeWindow* pWindow = &pTask->dataRange.window;




×








UNCOV

907



    tqDebug("s-task:%s level:%d verRange:%" PRId64 "-%" PRId64 " window:%" PRId64 "-%" PRId64





×







908



            ", do secondary scan-history from WAL after halt the related stream task:%s",










909



            id, pTask->info.taskLevel, pStep2Range->minVer, pStep2Range->maxVer, pWindow->skey, pWindow->ekey,










910



            pStreamTask->id.idStr);











UNCOV

911



    if (pTask->status.schedStatus != TASK_SCHED_STATUS__WAITING) {





×







912



      tqError("s-task:%s level:%d unexpected sched-status:%d", id, pTask->info.taskLevel, pTask->status.schedStatus);





×







913



    }










914













UNCOV

915



    int32_t code = streamSetParamForStreamScannerStep2(pTask, pStep2Range, pWindow);




×








UNCOV

916



    if (code) {





×







917



      tqError("s-task:%s level:%d failed to set step2 param", id, pTask->info.taskLevel);





×







918



    }










919













UNCOV

920



    int64_t dstVer = pStep2Range->minVer;




×








UNCOV

921



    pTask->chkInfo.nextProcessVer = dstVer;




×







922













UNCOV

923



    walReaderSetSkipToVersion(pTask->exec.pWalReader, dstVer);




×








UNCOV

924



    tqDebug("s-task:%s wal reader start scan WAL verRange:%" PRId64 "-%" PRId64 ", set sched-status:%d", id, dstVer,





×







925



            pStep2Range->maxVer, TASK_SCHED_STATUS__INACTIVE);










926













UNCOV

927



    int8_t status = streamTaskSetSchedStatusInactive(pTask);




×







928












929



    // now the fill-history task starts to scan data from wal files.











UNCOV

930



    code = streamTaskHandleEvent(pTask->status.pSM, TASK_EVENT_SCANHIST_DONE);




×








UNCOV

931



    if (code == TSDB_CODE_SUCCESS) {





×








UNCOV

932



      code = tqScanWalAsync(pTq, false);




×








UNCOV

933



      if (code) {





×







934



        tqError("vgId:%d failed to start scan wal file, code:%s", vgId, tstrerror(code));





×







935



      }










936



    }










937



  }











UNCOV

938



}




×







939













UNCOV

940



int32_t handleStep2Async(SStreamTask* pStreamTask, void* param) {




×








UNCOV

941



  STQ* pTq = param;




×







942













UNCOV

943



  SStreamMeta* pMeta = pStreamTask->pMeta;




×








UNCOV

944



  STaskId      hId = pStreamTask->hTaskInfo.id;




×








UNCOV

945



  SStreamTask* pTask = NULL;




×








UNCOV

946



  int32_t      code = streamMetaAcquireTask(pStreamTask->pMeta, hId.streamId, hId.taskId, &pTask);




×








UNCOV

947



  if (pTask == NULL) {





×







948



    tqWarn("s-task:0x%x failed to acquired it to exec step 2, scan wal quit", (int32_t)hId.taskId);





×







949



    return TSDB_CODE_SUCCESS;




×







950



  }










951













UNCOV

952



  doStartFillhistoryStep2(pTask, pStreamTask, pTq);




×







953













UNCOV

954



  streamMetaReleaseTask(pMeta, pTask);




×








UNCOV

955



  return TSDB_CODE_SUCCESS;




×







956



}










957












958



// this function should be executed by only one thread, so we set an sentinel to protect this function











UNCOV

959



int32_t tqProcessTaskScanHistory(STQ* pTq, SRpcMsg* pMsg) {




×








UNCOV

960



  SStreamScanHistoryReq* pReq = (SStreamScanHistoryReq*)pMsg->pCont;




×








UNCOV

961



  SStreamMeta*           pMeta = pTq->pStreamMeta;




×








UNCOV

962



  int32_t                code = TSDB_CODE_SUCCESS;




×








UNCOV

963



  SStreamTask*           pTask = NULL;




×








UNCOV

964



  SStreamTask*           pStreamTask = NULL;




×







965













UNCOV

966



  code = streamMetaAcquireTask(pMeta, pReq->streamId, pReq->taskId, &pTask);




×








UNCOV

967



  if (pTask == NULL) {





×







968



    tqError("vgId:%d failed to acquire stream task:0x%x during scan history data, task may have been destroyed",





×







969



            pMeta->vgId, pReq->taskId);










970



    return code;




×







971



  }










972












973



  // do recovery step1











UNCOV

974



  const char* id = pTask->id.idStr;




×








UNCOV

975



  char*       pStatus = streamTaskGetStatus(pTask).name;




×







976












977



  // avoid multi-thread exec










978



  while (1) {




×








UNCOV

979



    int32_t sentinel = atomic_val_compare_exchange_32(&pTask->status.inScanHistorySentinel, 0, 1);




×








UNCOV

980



    if (sentinel != 0) {





×








UNCOV

981



      tqDebug("s-task:%s already in scan-history func, wait for 100ms, and try again", id);





×








UNCOV

982



      taosMsleep(100);




×







983



    } else {











UNCOV

984



      break;




×







985



    }










986



  }










987












988



  // let's decide which step should be executed now











UNCOV

989



  if (pTask->execInfo.step1Start == 0) {





×








UNCOV

990



    int64_t ts = taosGetTimestampMs();




×








UNCOV

991



    pTask->execInfo.step1Start = ts;




×








UNCOV

992



    tqDebug("s-task:%s start scan-history stage(step 1), status:%s, step1 startTs:%" PRId64, id, pStatus, ts);





×







993



  } else {











UNCOV

994



    if (pTask->execInfo.step2Start == 0) {





×








UNCOV

995



      tqDebug("s-task:%s continue exec scan-history(step1), original step1 startTs:%" PRId64 ", already elapsed:%.2fs",





×







996



              id, pTask->execInfo.step1Start, pTask->execInfo.step1El);










997



    } else {











UNCOV

998



      tqDebug("s-task:%s already in step2, no need to scan-history data, step2 startTs:%" PRId64, id,





×







999



              pTask->execInfo.step2Start);










1000













UNCOV

1001



      atomic_store_32(&pTask->status.inScanHistorySentinel, 0);




×








UNCOV

1002



      streamMetaReleaseTask(pMeta, pTask);




×








UNCOV

1003



      return 0;




×







1004



    }










1005



  }










1006












1007



  // we have to continue retrying to successfully execute the scan history task.











UNCOV

1008



  if (!streamTaskSetSchedStatusWait(pTask)) {





×







1009



    tqError(





×







1010



        "s-task:%s failed to start scan-history in first stream time window since already started, unexpected "










1011



        "sched-status:%d",










1012



        id, pTask->status.schedStatus);










1013



    atomic_store_32(&pTask->status.inScanHistorySentinel, 0);




×







1014



    streamMetaReleaseTask(pMeta, pTask);




×







1015



    return 0;




×







1016



  }










1017













UNCOV

1018



  int64_t              st = taosGetTimestampMs();




×








UNCOV

1019



  SScanhistoryDataInfo retInfo = streamScanHistoryData(pTask, st);




×







1020













UNCOV

1021



  double el = (taosGetTimestampMs() - st) / 1000.0;




×








UNCOV

1022



  pTask->execInfo.step1El += el;




×







1023













UNCOV

1024



  if (retInfo.ret == TASK_SCANHISTORY_QUIT || retInfo.ret == TASK_SCANHISTORY_REXEC) {





×








UNCOV

1025



    int8_t status = streamTaskSetSchedStatusInactive(pTask);




×








UNCOV

1026



    atomic_store_32(&pTask->status.inScanHistorySentinel, 0);




×







1027













UNCOV

1028



    if (retInfo.ret == TASK_SCANHISTORY_REXEC) {





×








UNCOV

1029



      streamExecScanHistoryInFuture(pTask, retInfo.idleTime);




×







1030



    } else {











UNCOV

1031



      SStreamTaskState p = streamTaskGetStatus(pTask);




×








UNCOV

1032



      ETaskStatus      s = p.state;




×







1033













UNCOV

1034



      if (s == TASK_STATUS__PAUSE) {





×







1035



        tqDebug("s-task:%s is paused in the step1, elapsed time:%.2fs total:%.2fs, sched-status:%d", id, el,





×







1036



                pTask->execInfo.step1El, status);











UNCOV

1037



      } else if (s == TASK_STATUS__STOP || s == TASK_STATUS__DROPPING) {





×








UNCOV

1038



        tqDebug("s-task:%s status:%p not continue scan-history data, total elapsed time:%.2fs quit", id, p.name,





×







1039



                pTask->execInfo.step1El);










1040



      }










1041



    }










1042













UNCOV

1043



    streamMetaReleaseTask(pMeta, pTask);




×








UNCOV

1044



    return 0;




×







1045



  }










1046












1047



  // the following procedure should be executed, no matter status is stop/pause or not











UNCOV

1048



  tqDebug("s-task:%s scan-history(step 1) ended, elapsed time:%.2fs", id, pTask->execInfo.step1El);





×







1049













UNCOV

1050



  if (pTask->info.fillHistory != 1) {





×







1051



    tqError("s-task:%s fill-history is disabled, unexpected", id);





×







1052



    return TSDB_CODE_STREAM_INTERNAL_ERROR;




×







1053



  }










1054












1055



  // 1. get the related stream task











UNCOV

1056



  code = streamMetaAcquireTask(pMeta, pTask->streamTaskId.streamId, pTask->streamTaskId.taskId, &pStreamTask);




×








UNCOV

1057



  if (pStreamTask == NULL) {





×








UNCOV

1058



    tqError("failed to find s-task:0x%" PRIx64 ", it may have been destroyed, drop related fill-history task:%s",





×







1059



            pTask->streamTaskId.taskId, pTask->id.idStr);










1060













UNCOV

1061



    tqDebug("s-task:%s fill-history task set status to be dropping", id);





×








UNCOV

1062



    code = streamBuildAndSendDropTaskMsg(pTask->pMsgCb, pMeta->vgId, &pTask->id, 0);




×







1063













UNCOV

1064



    atomic_store_32(&pTask->status.inScanHistorySentinel, 0);




×








UNCOV

1065



    streamMetaReleaseTask(pMeta, pTask);




×








UNCOV

1066



    return code;




×







1067



  }










1068













UNCOV

1069



  if (pStreamTask->info.taskLevel != TASK_LEVEL__SOURCE) {





×







1070



    tqError("s-task:%s fill-history task related stream task level:%d, unexpected", id, pStreamTask->info.taskLevel);





×







1071



    return TSDB_CODE_STREAM_INTERNAL_ERROR;




×







1072



  }










1073













UNCOV

1074



  code = streamTaskHandleEventAsync(pStreamTask->status.pSM, TASK_EVENT_HALT, handleStep2Async, pTq);




×








UNCOV

1075



  streamMetaReleaseTask(pMeta, pStreamTask);




×







1076













UNCOV

1077



  atomic_store_32(&pTask->status.inScanHistorySentinel, 0);




×








UNCOV

1078



  streamMetaReleaseTask(pMeta, pTask);




×








UNCOV

1079



  return code;




×







1080



}










1081













UNCOV

1082



int32_t tqProcessTaskRunReq(STQ* pTq, SRpcMsg* pMsg) {




×








UNCOV

1083



  int32_t  code = 0;




×








UNCOV

1084



  char*    msg = POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead));




×








UNCOV

1085



  int32_t  len = pMsg->contLen - sizeof(SMsgHead);




×







1086



  SDecoder decoder;










1087













UNCOV

1088



  SStreamTaskRunReq req = {0};




×








UNCOV

1089



  tDecoderInit(&decoder, (uint8_t*)msg, len);




×








UNCOV

1090



  if ((code = tDecodeStreamTaskRunReq(&decoder, &req)) < 0) {





×







1091



    tqError("vgId:%d failed to decode task run req, code:%s", pTq->pStreamMeta->vgId, tstrerror(code));





×







1092



    tDecoderClear(&decoder);




×







1093



    return TSDB_CODE_SUCCESS;




×







1094



  }










1095













UNCOV

1096



  tDecoderClear(&decoder);




×







1097












1098



  // extracted submit data from wal files for all tasks











UNCOV

1099



  if (req.reqType == STREAM_EXEC_T_EXTRACT_WAL_DATA) {





×








UNCOV

1100



    return tqScanWal(pTq);




×







1101



  }










1102













UNCOV

1103



  code = tqStreamTaskProcessRunReq(pTq->pStreamMeta, pMsg, vnodeIsRoleLeader(pTq->pVnode));




×








UNCOV

1104



  if (code) {





×








UNCOV

1105



    tqError("vgId:%d failed to create task run req, code:%s", TD_VID(pTq->pVnode), tstrerror(code));





×








UNCOV

1106



    return code;




×







1107



  }










1108












1109



  // let's continue scan data in the wal files











UNCOV

1110



  if (req.reqType >= 0 || req.reqType == STREAM_EXEC_T_RESUME_TASK) {





×








UNCOV

1111



    code = tqScanWalAsync(pTq, false);  // it's ok to failed




×








UNCOV

1112



    if (code) {





×








UNCOV

1113



      tqError("vgId:%d failed to start scan wal file, code:%s", pTq->pStreamMeta->vgId, tstrerror(code));





×







1114



    }










1115



  }










1116













UNCOV

1117



  return code;




×







1118



}










1119













UNCOV

1120



int32_t tqProcessTaskDispatchReq(STQ* pTq, SRpcMsg* pMsg) {




×








UNCOV

1121



  return tqStreamTaskProcessDispatchReq(pTq->pStreamMeta, pMsg);




×







1122



}










1123













UNCOV

1124



int32_t tqProcessTaskDispatchRsp(STQ* pTq, SRpcMsg* pMsg) {




×








UNCOV

1125



  return tqStreamTaskProcessDispatchRsp(pTq->pStreamMeta, pMsg);




×







1126



}










1127













UNCOV

1128



int32_t tqProcessTaskDropReq(STQ* pTq, char* msg, int32_t msgLen) {




×








UNCOV

1129



  return tqStreamTaskProcessDropReq(pTq->pStreamMeta, msg, msgLen);




×







1130



}










1131













UNCOV

1132



int32_t tqProcessTaskUpdateCheckpointReq(STQ* pTq, char* msg, int32_t msgLen) {




×








UNCOV

1133



  return tqStreamTaskProcessUpdateCheckpointReq(pTq->pStreamMeta, pTq->pVnode->restored, msg);




×







1134



}










1135













UNCOV

1136



int32_t tqProcessTaskConsenChkptIdReq(STQ* pTq, SRpcMsg* pMsg) {




×








UNCOV

1137



  return tqStreamTaskProcessConsenChkptIdReq(pTq->pStreamMeta, pMsg);




×







1138



}










1139













UNCOV

1140



int32_t tqProcessTaskPauseReq(STQ* pTq, int64_t sversion, char* msg, int32_t msgLen) {




×








UNCOV

1141



  return tqStreamTaskProcessTaskPauseReq(pTq->pStreamMeta, msg);




×







1142



}










1143













UNCOV

1144



int32_t tqProcessTaskResumeReq(STQ* pTq, int64_t sversion, char* msg, int32_t msgLen) {




×








UNCOV

1145



  return tqStreamTaskProcessTaskResumeReq(pTq, sversion, msg, true);




×







1146



}










1147













UNCOV

1148



int32_t tqProcessTaskRetrieveReq(STQ* pTq, SRpcMsg* pMsg) {




×








UNCOV

1149



  return tqStreamTaskProcessRetrieveReq(pTq->pStreamMeta, pMsg);




×







1150



}










1151













UNCOV

1152



int32_t tqProcessTaskRetrieveRsp(STQ* pTq, SRpcMsg* pMsg) { return 0; }




×







1153












1154



int32_t tqStreamProgressRetrieveReq(STQ* pTq, SRpcMsg* pMsg) {




×







1155



  char*               msgStr = pMsg->pCont;




×







1156



  char*               msgBody = POINTER_SHIFT(msgStr, sizeof(SMsgHead));




×







1157



  int32_t             msgLen = pMsg->contLen - sizeof(SMsgHead);




×







1158



  int32_t             code = 0;




×







1159



  SStreamProgressReq  req;










1160



  char*               pRspBuf = taosMemoryCalloc(1, sizeof(SMsgHead) + sizeof(SStreamProgressRsp));





×







1161



  SStreamProgressRsp* pRsp = POINTER_SHIFT(pRspBuf, sizeof(SMsgHead));




×







1162



  if (!pRspBuf) {





×







1163



    terrno = TSDB_CODE_OUT_OF_MEMORY;




×







1164



    code = -1;




×







1165



    goto _OVER;




×







1166



  }










1167












1168



  code = tDeserializeStreamProgressReq(msgBody, msgLen, &req);




×







1169



  if (code == TSDB_CODE_SUCCESS) {





×







1170



    code = tqGetStreamExecInfo(pTq->pVnode, req.streamId, &pRsp->progressDelay, &pRsp->fillHisFinished);




×







1171



  }










1172



  if (code == TSDB_CODE_SUCCESS) {





×







1173



    pRsp->fetchIdx = req.fetchIdx;




×







1174



    pRsp->subFetchIdx = req.subFetchIdx;




×







1175



    pRsp->vgId = req.vgId;




×







1176



    pRsp->streamId = req.streamId;




×







1177



    code = tSerializeStreamProgressRsp(pRsp, sizeof(SStreamProgressRsp) + sizeof(SMsgHead), pRsp);




×







1178



    if (code) {





×







1179



      goto _OVER;




×







1180



    }










1181












1182



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




×







1183



    rsp.pCont = pRspBuf;




×







1184



    pRspBuf = NULL;




×







1185



    rsp.contLen = sizeof(SMsgHead) + sizeof(SStreamProgressRsp);




×







1186



    tmsgSendRsp(&rsp);




×







1187



  }










1188












1189



_OVER:




×







1190



  if (pRspBuf) {





×







1191



    taosMemoryFree(pRspBuf);





×







1192



  }










1193



  return code;




×







1194



}










1195












1196



// always return success to mnode










1197



//todo: handle failure of build and send msg to mnode











UNCOV

1198



static void doSendChkptSourceRsp(SStreamCheckpointSourceReq* pReq, SRpcHandleInfo* pRpcInfo, int32_t code,




×







1199



                                 int32_t taskId) {











UNCOV

1200



  SRpcMsg rsp = {0};




×








UNCOV

1201



  int32_t ret = streamTaskBuildCheckpointSourceRsp(pReq, pRpcInfo, &rsp, code);




×








UNCOV

1202



  if (ret) {  // suppress the error in build checkpoint source rsp





×







1203



    tqError("s-task:0x%x failed to build checkpoint-source rsp, code:%s", taskId, tstrerror(ret));





×







1204



  }











UNCOV

1205



  tmsgSendRsp(&rsp);  // error occurs




×








UNCOV

1206



}




×







1207












1208



// no matter what kinds of error happened, make sure the mnode will receive the success execution code.











UNCOV

1209



int32_t tqProcessTaskCheckPointSourceReq(STQ* pTq, SRpcMsg* pMsg, SRpcMsg* pRsp) {




×








UNCOV

1210



  int32_t                    vgId = TD_VID(pTq->pVnode);




×








UNCOV

1211



  SStreamMeta*               pMeta = pTq->pStreamMeta;




×








UNCOV

1212



  char*                      msg = POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead));




×








UNCOV

1213



  int32_t                    len = pMsg->contLen - sizeof(SMsgHead);




×








UNCOV

1214



  int32_t                    code = 0;




×








UNCOV

1215



  SStreamCheckpointSourceReq req = {0};




×








UNCOV

1216



  SDecoder                   decoder = {0};




×








UNCOV

1217



  SStreamTask*               pTask = NULL;




×








UNCOV

1218



  int64_t                    checkpointId = 0;




×







1219












1220



  // disable auto rsp to mnode











UNCOV

1221



  pRsp->info.handle = NULL;




×







1222













UNCOV

1223



  tDecoderInit(&decoder, (uint8_t*)msg, len);




×








UNCOV

1224



  if (tDecodeStreamCheckpointSourceReq(&decoder, &req) < 0) {





×







1225



    code = TSDB_CODE_MSG_DECODE_ERROR;




×







1226



    tDecoderClear(&decoder);




×







1227



    tqError("vgId:%d failed to decode checkpoint-source msg, code:%s", vgId, tstrerror(code));





×







1228



    doSendChkptSourceRsp(&req, &pMsg->info, TSDB_CODE_SUCCESS, req.taskId);




×







1229



    return TSDB_CODE_SUCCESS;  // always return success to mnode,




×







1230



  }










1231













UNCOV

1232



  tDecoderClear(&decoder);




×







1233













UNCOV

1234



  if (!vnodeIsRoleLeader(pTq->pVnode)) {





×








UNCOV

1235



    tqDebug("vgId:%d not leader, ignore checkpoint-source msg, s-task:0x%x", vgId, req.taskId);





×








UNCOV

1236



    doSendChkptSourceRsp(&req, &pMsg->info, TSDB_CODE_SUCCESS, req.taskId);




×








UNCOV

1237



    return TSDB_CODE_SUCCESS;  // always return success to mnode




×







1238



  }










1239













UNCOV

1240



  if (!pTq->pVnode->restored) {





×








UNCOV

1241



    tqDebug("vgId:%d checkpoint-source msg received during restoring, checkpointId:%" PRId64





×







1242



            ", transId:%d s-task:0x%x ignore it",










1243



            vgId, req.checkpointId, req.transId, req.taskId);











UNCOV

1244



    doSendChkptSourceRsp(&req, &pMsg->info, TSDB_CODE_SUCCESS, req.taskId);




×








UNCOV

1245



    return TSDB_CODE_SUCCESS;  // always return success to mnode




×







1246



  }










1247













UNCOV

1248



  code = streamMetaAcquireTask(pMeta, req.streamId, req.taskId, &pTask);




×








UNCOV

1249



  if (pTask == NULL || code != 0) {





×








UNCOV

1250



    tqError("vgId:%d failed to find s-task:0x%x, ignore checkpoint msg. checkpointId:%" PRId64





×







1251



            " transId:%d it may have been destroyed",










1252



            vgId, req.taskId, req.checkpointId, req.transId);











UNCOV

1253



    doSendChkptSourceRsp(&req, &pMsg->info, TSDB_CODE_SUCCESS, req.taskId);




×







1254



    return TSDB_CODE_SUCCESS;




×







1255



  }










1256













UNCOV

1257



  if (pTask->status.downstreamReady != 1) {





×







1258



    // record the latest failed checkpoint id











UNCOV

1259



    streamTaskSetFailedChkptInfo(pTask, req.transId, req.checkpointId);




×








UNCOV

1260



    tqError("s-task:%s not ready for checkpoint, since downstream not ready, ignore this checkpointId:%" PRId64





×







1261



            ", transId:%d set it failed",










1262



            pTask->id.idStr, req.checkpointId, req.transId);










1263













UNCOV

1264



    streamMetaReleaseTask(pMeta, pTask);




×








UNCOV

1265



    doSendChkptSourceRsp(&req, &pMsg->info, TSDB_CODE_SUCCESS, req.taskId);




×








UNCOV

1266



    return TSDB_CODE_SUCCESS;  // todo retry handle error




×







1267



  }










1268












1269



  // todo save the checkpoint failed info











UNCOV

1270



  streamMutexLock(&pTask->lock);




×








UNCOV

1271



  ETaskStatus status = streamTaskGetStatus(pTask).state;




×







1272













UNCOV

1273



  if (req.mndTrigger == 1) {





×








UNCOV

1274



    if (status == TASK_STATUS__HALT || status == TASK_STATUS__PAUSE) {





×







1275



      tqError("s-task:%s not ready for checkpoint, since it is halt, ignore checkpointId:%" PRId64 ", set it failure",





×







1276



              pTask->id.idStr, req.checkpointId);










1277












1278



      streamMutexUnlock(&pTask->lock);




×







1279



      streamMetaReleaseTask(pMeta, pTask);




×







1280



      doSendChkptSourceRsp(&req, &pMsg->info, TSDB_CODE_SUCCESS, req.taskId);




×







1281



      return TSDB_CODE_SUCCESS;




×







1282



    }










1283



  } else {











UNCOV

1284



    if (status != TASK_STATUS__HALT) {





×







1285



      tqError("s-task:%s should in halt status, let's halt it directly", pTask->id.idStr);





×







1286



      //      streamTaskHandleEvent(pTask->status.pSM, TASK_EVENT_HALT);










1287



    }










1288



  }










1289












1290



  // check if the checkpoint msg already sent or not.











UNCOV

1291



  if (status == TASK_STATUS__CK) {





×







1292



    streamTaskGetActiveCheckpointInfo(pTask, NULL, &checkpointId);




×







1293












1294



    tqWarn("s-task:%s repeatly recv checkpoint-source msg checkpointId:%" PRId64





×







1295



           " transId:%d already handled, ignore msg and continue process checkpoint",










1296



           pTask->id.idStr, checkpointId, req.transId);










1297












1298



    streamMutexUnlock(&pTask->lock);




×







1299



    streamMetaReleaseTask(pMeta, pTask);




×







1300



    doSendChkptSourceRsp(&req, &pMsg->info, TSDB_CODE_SYN_PROPOSE_NOT_READY, req.taskId);




×







1301



    return TSDB_CODE_SUCCESS;




×







1302



  } else {  // checkpoint already finished, and not in checkpoint status











UNCOV

1303



    if (req.checkpointId <= pTask->chkInfo.checkpointId) {





×







1304



      tqWarn("s-task:%s repeatly recv checkpoint-source msg checkpointId:%" PRId64





×







1305



             " transId:%d already handled, return success",










1306



             pTask->id.idStr, req.checkpointId, req.transId);










1307












1308



      streamMutexUnlock(&pTask->lock);




×







1309



      streamMetaReleaseTask(pMeta, pTask);




×







1310



      doSendChkptSourceRsp(&req, &pMsg->info, TSDB_CODE_SUCCESS, req.taskId);




×







1311



      return TSDB_CODE_SUCCESS;




×







1312



    }










1313



  }










1314













UNCOV

1315



  code = streamProcessCheckpointSourceReq(pTask, &req);




×








UNCOV

1316



  streamMutexUnlock(&pTask->lock);




×







1317













UNCOV

1318



  if (code) {





×







1319



    qError("s-task:%s (vgId:%d) failed to process checkpoint-source req, code:%s", pTask->id.idStr, vgId,





×







1320



           tstrerror(code));










1321



    streamMetaReleaseTask(pMeta, pTask);




×







1322



    doSendChkptSourceRsp(&req, &pMsg->info, TSDB_CODE_SUCCESS, req.taskId);




×








UNCOV

1323



    return TSDB_CODE_SUCCESS;




×







1324



  }










1325













UNCOV

1326



  if (req.mndTrigger) {





×








UNCOV

1327



    tqInfo("s-task:%s (vgId:%d) level:%d receive checkpoint-source msg chkpt:%" PRId64 ", transId:%d, ",





×







1328



           pTask->id.idStr, vgId, pTask->info.taskLevel, req.checkpointId, req.transId);










1329



  } else {











UNCOV

1330



    const char* pPrevStatus = streamTaskGetStatusStr(streamTaskGetPrevStatus(pTask));




×








UNCOV

1331



    tqInfo("s-task:%s (vgId:%d) level:%d receive checkpoint-source msg chkpt:%" PRId64





×







1332



           ", transId:%d after transfer-state, prev status:%s",










1333



           pTask->id.idStr, vgId, pTask->info.taskLevel, req.checkpointId, req.transId, pPrevStatus);










1334



  }










1335













UNCOV

1336



  code = streamAddCheckpointSourceRspMsg(&req, &pMsg->info, pTask);




×








UNCOV

1337



  if (code != TSDB_CODE_SUCCESS) {





×







1338



    streamTaskSetCheckpointFailed(pTask);  // set the checkpoint failed




×







1339



    doSendChkptSourceRsp(&req, &pMsg->info, TSDB_CODE_SUCCESS, req.taskId);




×







1340



  }










1341













UNCOV

1342



  streamMetaReleaseTask(pMeta, pTask);




×








UNCOV

1343



  return TSDB_CODE_SUCCESS;




×







1344



}










1345












1346



// downstream task has complete the stream task checkpoint procedure, let's start the handle the rsp by execute task











UNCOV

1347



int32_t tqProcessTaskCheckpointReadyMsg(STQ* pTq, SRpcMsg* pMsg) {




×








UNCOV

1348



  int32_t vgId = TD_VID(pTq->pVnode);




×







1349













UNCOV

1350



  SStreamCheckpointReadyMsg* pReq = (SStreamCheckpointReadyMsg*)pMsg->pCont;




×








UNCOV

1351



  if (!vnodeIsRoleLeader(pTq->pVnode)) {





×







1352



    tqError("vgId:%d not leader, ignore the retrieve checkpoint-trigger msg from 0x%x", vgId,





×







1353



            (int32_t)pReq->downstreamTaskId);










1354



    return TSDB_CODE_STREAM_NOT_LEADER;




×







1355



  }










1356













UNCOV

1357



  return tqStreamTaskProcessCheckpointReadyMsg(pTq->pStreamMeta, pMsg);




×







1358



}










1359













UNCOV

1360



int32_t tqProcessTaskUpdateReq(STQ* pTq, SRpcMsg* pMsg) {




×








UNCOV

1361



  return tqStreamTaskProcessUpdateReq(pTq->pStreamMeta, &pTq->pVnode->msgCb, pMsg, pTq->pVnode->restored);




×







1362



}










1363













UNCOV

1364



int32_t tqProcessTaskResetReq(STQ* pTq, SRpcMsg* pMsg) {




×








UNCOV

1365



  return tqStreamTaskProcessTaskResetReq(pTq->pStreamMeta, pMsg->pCont);




×







1366



}










1367













UNCOV

1368



int32_t tqProcessTaskRetrieveTriggerReq(STQ* pTq, SRpcMsg* pMsg) {




×








UNCOV

1369



  int32_t vgId = TD_VID(pTq->pVnode);




×







1370













UNCOV

1371



  if (!vnodeIsRoleLeader(pTq->pVnode)) {





×







1372



    SRetrieveChkptTriggerReq req = {0};




×







1373












1374



    char*    msg = POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead));




×







1375



    int32_t  len = pMsg->contLen - sizeof(SMsgHead);




×







1376



    SDecoder decoder = {0};




×







1377












1378



    tDecoderInit(&decoder, (uint8_t*)msg, len);




×







1379



    if (tDecodeRetrieveChkptTriggerReq(&decoder, &req) < 0) {





×







1380



      tDecoderClear(&decoder);




×







1381



      tqError("vgId:%d invalid retrieve checkpoint-trigger req received", vgId);





×







1382



      return TSDB_CODE_INVALID_MSG;




×







1383



    }










1384



    tDecoderClear(&decoder);




×







1385












1386



    tqError("vgId:%d not leader, ignore the retrieve checkpoint-trigger msg from s-task:0x%" PRId64, vgId,





×







1387



            req.downstreamTaskId);










1388



    return TSDB_CODE_STREAM_NOT_LEADER;




×







1389



  }










1390













UNCOV

1391



  return tqStreamTaskProcessRetrieveTriggerReq(pTq->pStreamMeta, pMsg);




×







1392



}










1393













UNCOV

1394



int32_t tqProcessTaskRetrieveTriggerRsp(STQ* pTq, SRpcMsg* pMsg) {




×








UNCOV

1395



  return tqStreamTaskProcessRetrieveTriggerRsp(pTq->pStreamMeta, pMsg);




×







1396



}










1397












1398



// this function is needed, do not try to remove it.











UNCOV

1399



int32_t tqProcessStreamHbRsp(STQ* pTq, SRpcMsg* pMsg) { return tqStreamProcessStreamHbRsp(pTq->pStreamMeta, pMsg); }




×







1400













UNCOV

1401



int32_t tqProcessStreamReqCheckpointRsp(STQ* pTq, SRpcMsg* pMsg) {




×








UNCOV

1402



  return tqStreamProcessReqCheckpointRsp(pTq->pStreamMeta, pMsg);




×







1403



}










1404













UNCOV

1405



int32_t tqProcessTaskCheckpointReadyRsp(STQ* pTq, SRpcMsg* pMsg) {




×








UNCOV

1406



  return tqStreamProcessCheckpointReadyRsp(pTq->pStreamMeta, pMsg);




×







1407



}










1408













UNCOV

1409



int32_t tqProcessTaskChkptReportRsp(STQ* pTq, SRpcMsg* pMsg) {




×








UNCOV

1410



  return tqStreamProcessChkptReportRsp(pTq->pStreamMeta, pMsg);




×







1411



}










1412


























    

  • 

    •