taosdata / TDengine / #3562

coverage: 26.655% (-32.2%) from 58.812%
#3562

push

travis-ci

web-flow 
Merge pull request #29229 from taosdata/enh/TS-5749-3.0

enh: seperate tsdb async tasks to different thread pools

21498 of 109421 branches covered (19.65%)

Branch coverage included in aggregate %.

66 of 96 new or added lines in 7 files covered. (68.75%)

39441 existing lines in 157 files now uncovered.

35007 of 102566 relevant lines covered (34.13%)

53922.97 hits per line

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

3.08
/source/common/src/msg/streamMsg.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 "streamMsg.h"










17



#include "os.h"










18



#include "tcommon.h"










19












20



typedef struct STaskId {










21



  int64_t streamId;










22



  int64_t taskId;










23



} STaskId;










24












25



typedef struct STaskCkptInfo {










26



  int64_t latestId;          // saved checkpoint id










27



  int64_t latestVer;         // saved checkpoint ver










28



  int64_t latestTime;        // latest checkpoint time










29



  int64_t latestSize;        // latest checkpoint size










30



  int8_t  remoteBackup;      // latest checkpoint backup done










31



  int64_t activeId;          // current active checkpoint id










32



  int32_t activeTransId;     // checkpoint trans id










33



  int8_t  failed;            // denote if the checkpoint is failed or not










34



  int8_t  consensusChkptId;  // required the consensus-checkpointId










35



  int64_t consensusTs;       //










36



} STaskCkptInfo;










37












38



typedef struct STaskStatusEntry {










39



  STaskId       id;










40



  int32_t       status;










41



  int32_t       statusLastDuration;  // to record the last duration of current status










42



  int64_t       stage;










43



  int32_t       nodeId;










44



  SVersionRange verRange;      // start/end version in WAL, only valid for source task










45



  int64_t       processedVer;  // only valid for source task










46



  double        inputQUsed;    // in MiB










47



  double        inputRate;










48



  double        procsThroughput;   // duration between one element put into input queue and being processed.










49



  double        procsTotal;        // duration between one element put into input queue and being processed.










50



  double        outputThroughput;  // the size of dispatched result blocks in bytes










51



  double        outputTotal;       // the size of dispatched result blocks in bytes










52



  double        sinkQuota;         // existed quota size for sink task










53



  double        sinkDataSize;      // sink to dst data size










54



  int64_t       startTime;










55



  int64_t       startCheckpointId;










56



  int64_t       startCheckpointVer;










57



  int64_t       hTaskId;










58



  STaskCkptInfo checkpointInfo;










59



} STaskStatusEntry;










60













UNCOV

61



int32_t tEncodeStreamEpInfo(SEncoder* pEncoder, const SStreamUpstreamEpInfo* pInfo) {




×








UNCOV

62



  TAOS_CHECK_RETURN(tEncodeI32(pEncoder, pInfo->taskId));





×








UNCOV

63



  TAOS_CHECK_RETURN(tEncodeI32(pEncoder, pInfo->nodeId));





×








UNCOV

64



  TAOS_CHECK_RETURN(tEncodeI32(pEncoder, pInfo->childId));





×








UNCOV

65



  TAOS_CHECK_RETURN(tEncodeSEpSet(pEncoder, &pInfo->epSet));





×








UNCOV

66



  TAOS_CHECK_RETURN(tEncodeI64(pEncoder, pInfo->stage));





×








UNCOV

67



  return 0;




×







68



}










69













UNCOV

70



int32_t tDecodeStreamEpInfo(SDecoder* pDecoder, SStreamUpstreamEpInfo* pInfo) {




×








UNCOV

71



  TAOS_CHECK_RETURN(tDecodeI32(pDecoder, &pInfo->taskId));





×








UNCOV

72



  TAOS_CHECK_RETURN(tDecodeI32(pDecoder, &pInfo->nodeId));





×








UNCOV

73



  TAOS_CHECK_RETURN(tDecodeI32(pDecoder, &pInfo->childId));





×








UNCOV

74



  TAOS_CHECK_RETURN(tDecodeSEpSet(pDecoder, &pInfo->epSet));





×








UNCOV

75



  TAOS_CHECK_RETURN(tDecodeI64(pDecoder, &pInfo->stage));





×








UNCOV

76



  return 0;




×







77



}










78













UNCOV

79



int32_t tEncodeStreamCheckpointSourceReq(SEncoder* pEncoder, const SStreamCheckpointSourceReq* pReq) {




×








UNCOV

80



  TAOS_CHECK_RETURN(tStartEncode(pEncoder));





×








UNCOV

81



  TAOS_CHECK_RETURN(tEncodeI64(pEncoder, pReq->streamId));





×








UNCOV

82



  TAOS_CHECK_RETURN(tEncodeI64(pEncoder, pReq->checkpointId));





×








UNCOV

83



  TAOS_CHECK_RETURN(tEncodeI32(pEncoder, pReq->taskId));





×








UNCOV

84



  TAOS_CHECK_RETURN(tEncodeI32(pEncoder, pReq->nodeId));





×








UNCOV

85



  TAOS_CHECK_RETURN(tEncodeSEpSet(pEncoder, &pReq->mgmtEps));





×








UNCOV

86



  TAOS_CHECK_RETURN(tEncodeI32(pEncoder, pReq->mnodeId));





×








UNCOV

87



  TAOS_CHECK_RETURN(tEncodeI64(pEncoder, pReq->expireTime));





×








UNCOV

88



  TAOS_CHECK_RETURN(tEncodeI32(pEncoder, pReq->transId));





×








UNCOV

89



  TAOS_CHECK_RETURN(tEncodeI8(pEncoder, pReq->mndTrigger));





×








UNCOV

90



  tEndEncode(pEncoder);




×








UNCOV

91



  return pEncoder->pos;




×







92



}










93













UNCOV

94



int32_t tDecodeStreamCheckpointSourceReq(SDecoder* pDecoder, SStreamCheckpointSourceReq* pReq) {




×








UNCOV

95



  TAOS_CHECK_RETURN(tStartDecode(pDecoder));





×








UNCOV

96



  TAOS_CHECK_RETURN(tDecodeI64(pDecoder, &pReq->streamId));





×








UNCOV

97



  TAOS_CHECK_RETURN(tDecodeI64(pDecoder, &pReq->checkpointId));





×








UNCOV

98



  TAOS_CHECK_RETURN(tDecodeI32(pDecoder, &pReq->taskId));





×








UNCOV

99



  TAOS_CHECK_RETURN(tDecodeI32(pDecoder, &pReq->nodeId));





×








UNCOV

100



  TAOS_CHECK_RETURN(tDecodeSEpSet(pDecoder, &pReq->mgmtEps));





×








UNCOV

101



  TAOS_CHECK_RETURN(tDecodeI32(pDecoder, &pReq->mnodeId));





×








UNCOV

102



  TAOS_CHECK_RETURN(tDecodeI64(pDecoder, &pReq->expireTime));





×








UNCOV

103



  TAOS_CHECK_RETURN(tDecodeI32(pDecoder, &pReq->transId));





×








UNCOV

104



  TAOS_CHECK_RETURN(tDecodeI8(pDecoder, &pReq->mndTrigger));





×








UNCOV

105



  tEndDecode(pDecoder);




×








UNCOV

106



  return 0;




×







107



}










108













UNCOV

109



int32_t tEncodeStreamCheckpointSourceRsp(SEncoder* pEncoder, const SStreamCheckpointSourceRsp* pRsp) {




×








UNCOV

110



  TAOS_CHECK_RETURN(tStartEncode(pEncoder));





×








UNCOV

111



  TAOS_CHECK_RETURN(tEncodeI64(pEncoder, pRsp->streamId));





×








UNCOV

112



  TAOS_CHECK_RETURN(tEncodeI64(pEncoder, pRsp->checkpointId));





×








UNCOV

113



  TAOS_CHECK_RETURN(tEncodeI32(pEncoder, pRsp->taskId));





×








UNCOV

114



  TAOS_CHECK_RETURN(tEncodeI32(pEncoder, pRsp->nodeId));





×








UNCOV

115



  TAOS_CHECK_RETURN(tEncodeI64(pEncoder, pRsp->expireTime));





×








UNCOV

116



  TAOS_CHECK_RETURN(tEncodeI8(pEncoder, pRsp->success));





×








UNCOV

117



  tEndEncode(pEncoder);




×








UNCOV

118



  return pEncoder->pos;




×







119



}










120













UNCOV

121



int32_t tEncodeStreamTaskUpdateMsg(SEncoder* pEncoder, const SStreamTaskNodeUpdateMsg* pMsg) {




×








UNCOV

122



  int32_t code = 0;




×







123



  int32_t lino;










124













UNCOV

125



  TAOS_CHECK_EXIT(tStartEncode(pEncoder));





×








UNCOV

126



  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pMsg->streamId));





×








UNCOV

127



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pMsg->taskId));





×







128













UNCOV

129



  int32_t size = taosArrayGetSize(pMsg->pNodeList);





×








UNCOV

130



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, size));





×







131













UNCOV

132



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





×








UNCOV

133



    SNodeUpdateInfo* pInfo = taosArrayGet(pMsg->pNodeList, i);




×








UNCOV

134



    if (pInfo == NULL) {





×







135



      TAOS_CHECK_EXIT(terrno);





×







136



    }










137













UNCOV

138



    TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pInfo->nodeId));





×








UNCOV

139



    TAOS_CHECK_EXIT(tEncodeSEpSet(pEncoder, &pInfo->prevEp));





×








UNCOV

140



    TAOS_CHECK_EXIT(tEncodeSEpSet(pEncoder, &pInfo->newEp));





×







141



  }










142












143



  // todo this new attribute will be result in being incompatible with previous version











UNCOV

144



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pMsg->transId));





×








UNCOV

145



  tEndEncode(pEncoder);




×








UNCOV

146



_exit:




×








UNCOV

147



  if (code) {





×







148



    return code;




×







149



  } else {











UNCOV

150



    return pEncoder->pos;




×







151



  }










152



}










153













UNCOV

154



int32_t tDecodeStreamTaskUpdateMsg(SDecoder* pDecoder, SStreamTaskNodeUpdateMsg* pMsg) {




×








UNCOV

155



  int32_t code = 0;




×







156



  int32_t lino;










157













UNCOV

158



  TAOS_CHECK_EXIT(tStartDecode(pDecoder));





×








UNCOV

159



  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pMsg->streamId));





×








UNCOV

160



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pMsg->taskId));





×







161













UNCOV

162



  int32_t size = 0;




×








UNCOV

163



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &size));





×








UNCOV

164



  pMsg->pNodeList = taosArrayInit(size, sizeof(SNodeUpdateInfo));




×








UNCOV

165



  if (pMsg->pNodeList == NULL) {





×







166



    TAOS_CHECK_EXIT(terrno);





×







167



  }











UNCOV

168



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





×








UNCOV

169



    SNodeUpdateInfo info = {0};




×








UNCOV

170



    TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &info.nodeId));





×








UNCOV

171



    TAOS_CHECK_EXIT(tDecodeSEpSet(pDecoder, &info.prevEp));





×








UNCOV

172



    TAOS_CHECK_EXIT(tDecodeSEpSet(pDecoder, &info.newEp));





×







173













UNCOV

174



    if (taosArrayPush(pMsg->pNodeList, &info) == NULL) {





×







175



      TAOS_CHECK_EXIT(terrno);





×







176



    }










177



  }










178













UNCOV

179



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pMsg->transId));





×







180













UNCOV

181



  tEndDecode(pDecoder);




×








UNCOV

182



_exit:




×








UNCOV

183



  return code;




×







184



}










185













UNCOV

186



int32_t tEncodeStreamTaskCheckReq(SEncoder* pEncoder, const SStreamTaskCheckReq* pReq) {




×








UNCOV

187



  int32_t code = 0;




×







188



  int32_t lino;










189













UNCOV

190



  TAOS_CHECK_EXIT(tStartEncode(pEncoder));





×








UNCOV

191



  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pReq->reqId));





×








UNCOV

192



  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pReq->streamId));





×








UNCOV

193



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->upstreamNodeId));





×








UNCOV

194



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->upstreamTaskId));





×








UNCOV

195



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->downstreamNodeId));





×








UNCOV

196



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->downstreamTaskId));





×








UNCOV

197



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->childId));





×








UNCOV

198



  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pReq->stage));





×








UNCOV

199



  tEndEncode(pEncoder);




×







200













UNCOV

201



_exit:




×








UNCOV

202



  if (code) {





×







203



    return code;




×







204



  } else {











UNCOV

205



    return pEncoder->pos;




×







206



  }










207



}










208













UNCOV

209



int32_t tDecodeStreamTaskCheckReq(SDecoder* pDecoder, SStreamTaskCheckReq* pReq) {




×








UNCOV

210



  int32_t code = 0;




×







211



  int32_t lino;










212













UNCOV

213



  TAOS_CHECK_EXIT(tStartDecode(pDecoder));





×








UNCOV

214



  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pReq->reqId));





×








UNCOV

215



  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pReq->streamId));





×








UNCOV

216



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pReq->upstreamNodeId));





×








UNCOV

217



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pReq->upstreamTaskId));





×








UNCOV

218



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pReq->downstreamNodeId));





×








UNCOV

219



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pReq->downstreamTaskId));





×








UNCOV

220



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pReq->childId));





×








UNCOV

221



  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pReq->stage));





×








UNCOV

222



  tEndDecode(pDecoder);




×







223













UNCOV

224



_exit:




×








UNCOV

225



  return code;




×







226



}










227













UNCOV

228



int32_t tEncodeStreamTaskCheckRsp(SEncoder* pEncoder, const SStreamTaskCheckRsp* pRsp) {




×








UNCOV

229



  int32_t code = 0;




×







230



  int32_t lino;










231













UNCOV

232



  TAOS_CHECK_EXIT(tStartEncode(pEncoder));





×








UNCOV

233



  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pRsp->reqId));





×








UNCOV

234



  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pRsp->streamId));





×








UNCOV

235



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pRsp->upstreamNodeId));





×








UNCOV

236



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pRsp->upstreamTaskId));





×








UNCOV

237



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pRsp->downstreamNodeId));





×








UNCOV

238



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pRsp->downstreamTaskId));





×








UNCOV

239



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pRsp->childId));





×








UNCOV

240



  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pRsp->oldStage));





×








UNCOV

241



  TAOS_CHECK_EXIT(tEncodeI8(pEncoder, pRsp->status));





×








UNCOV

242



  tEndEncode(pEncoder);




×







243













UNCOV

244



_exit:




×








UNCOV

245



  if (code) {





×







246



    return code;




×







247



  } else {











UNCOV

248



    return pEncoder->pos;




×







249



  }










250



}










251













UNCOV

252



int32_t tDecodeStreamTaskCheckRsp(SDecoder* pDecoder, SStreamTaskCheckRsp* pRsp) {




×








UNCOV

253



  int32_t code = 0;




×







254



  int32_t lino;










255













UNCOV

256



  TAOS_CHECK_EXIT(tStartDecode(pDecoder));





×








UNCOV

257



  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pRsp->reqId));





×








UNCOV

258



  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pRsp->streamId));





×








UNCOV

259



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pRsp->upstreamNodeId));





×








UNCOV

260



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pRsp->upstreamTaskId));





×








UNCOV

261



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pRsp->downstreamNodeId));





×








UNCOV

262



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pRsp->downstreamTaskId));





×








UNCOV

263



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pRsp->childId));





×








UNCOV

264



  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pRsp->oldStage));





×








UNCOV

265



  TAOS_CHECK_EXIT(tDecodeI8(pDecoder, &pRsp->status));





×








UNCOV

266



  tEndDecode(pDecoder);




×







267













UNCOV

268



_exit:




×








UNCOV

269



  return code;




×







270



}










271













UNCOV

272



int32_t tEncodeStreamCheckpointReadyMsg(SEncoder* pEncoder, const SStreamCheckpointReadyMsg* pReq) {




×








UNCOV

273



  int32_t code = 0;




×







274



  int32_t lino;










275













UNCOV

276



  TAOS_CHECK_EXIT(tStartEncode(pEncoder));





×








UNCOV

277



  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pReq->streamId));





×








UNCOV

278



  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pReq->checkpointId));





×








UNCOV

279



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->downstreamTaskId));





×








UNCOV

280



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->downstreamNodeId));





×








UNCOV

281



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->upstreamTaskId));





×








UNCOV

282



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->upstreamNodeId));





×








UNCOV

283



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->childId));





×








UNCOV

284



  tEndEncode(pEncoder);




×







285













UNCOV

286



_exit:




×








UNCOV

287



  if (code) {





×







288



    return code;




×







289



  } else {











UNCOV

290



    return pEncoder->pos;




×







291



  }










292



}










293













UNCOV

294



int32_t tDecodeStreamCheckpointReadyMsg(SDecoder* pDecoder, SStreamCheckpointReadyMsg* pRsp) {




×








UNCOV

295



  int32_t code = 0;




×







296



  int32_t lino;










297













UNCOV

298



  TAOS_CHECK_EXIT(tStartDecode(pDecoder));





×








UNCOV

299



  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pRsp->streamId));





×








UNCOV

300



  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pRsp->checkpointId));





×








UNCOV

301



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pRsp->downstreamTaskId));





×








UNCOV

302



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pRsp->downstreamNodeId));





×








UNCOV

303



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pRsp->upstreamTaskId));





×








UNCOV

304



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pRsp->upstreamNodeId));





×








UNCOV

305



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pRsp->childId));





×








UNCOV

306



  tEndDecode(pDecoder);




×







307













UNCOV

308



_exit:




×








UNCOV

309



  return code;




×







310



}










311













UNCOV

312



int32_t tEncodeStreamDispatchReq(SEncoder* pEncoder, const SStreamDispatchReq* pReq) {




×








UNCOV

313



  int32_t code = 0;




×







314



  int32_t lino;










315













UNCOV

316



  TAOS_CHECK_EXIT(tStartEncode(pEncoder));





×








UNCOV

317



  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pReq->stage));





×








UNCOV

318



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->msgId));





×








UNCOV

319



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->srcVgId));





×








UNCOV

320



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->type));





×








UNCOV

321



  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pReq->streamId));





×








UNCOV

322



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->taskId));





×








UNCOV

323



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->type));





×








UNCOV

324



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->upstreamTaskId));





×








UNCOV

325



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->upstreamChildId));





×








UNCOV

326



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->upstreamNodeId));





×








UNCOV

327



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->upstreamRelTaskId));





×








UNCOV

328



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->blockNum));





×








UNCOV

329



  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pReq->totalLen));





×







330













UNCOV

331



  if (taosArrayGetSize(pReq->data) != pReq->blockNum || taosArrayGetSize(pReq->dataLen) != pReq->blockNum) {





×







332



    uError("invalid dispatch req msg");





×








UNCOV

333



    TAOS_CHECK_EXIT(TSDB_CODE_INVALID_MSG);





×







334



  }










335













UNCOV

336



  for (int32_t i = 0; i < pReq->blockNum; i++) {





×








UNCOV

337



    int32_t* pLen = taosArrayGet(pReq->dataLen, i);




×








UNCOV

338



    void*    data = taosArrayGetP(pReq->data, i);




×








UNCOV

339



    if (data == NULL || pLen == NULL) {





×








UNCOV

340



      TAOS_CHECK_EXIT(terrno);





×







341



    }










342













UNCOV

343



    TAOS_CHECK_EXIT(tEncodeI32(pEncoder, *pLen));





×








UNCOV

344



    TAOS_CHECK_EXIT(tEncodeBinary(pEncoder, data, *pLen));





×







345



  }











UNCOV

346



  tEndEncode(pEncoder);




×








UNCOV

347



_exit:




×








UNCOV

348



  if (code) {





×







349



    return code;




×







350



  } else {











UNCOV

351



    return pEncoder->pos;




×







352



  }










353



}










354













UNCOV

355



int32_t tDecodeStreamDispatchReq(SDecoder* pDecoder, SStreamDispatchReq* pReq) {




×








UNCOV

356



  int32_t code = 0;




×







357



  int32_t lino;










358













UNCOV

359



  TAOS_CHECK_EXIT(tStartDecode(pDecoder));





×








UNCOV

360



  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pReq->stage));





×








UNCOV

361



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pReq->msgId));





×








UNCOV

362



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pReq->srcVgId));





×








UNCOV

363



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pReq->type));





×








UNCOV

364



  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pReq->streamId));





×








UNCOV

365



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pReq->taskId));





×








UNCOV

366



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pReq->type));





×








UNCOV

367



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pReq->upstreamTaskId));





×








UNCOV

368



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pReq->upstreamChildId));





×








UNCOV

369



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pReq->upstreamNodeId));





×








UNCOV

370



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pReq->upstreamRelTaskId));





×








UNCOV

371



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pReq->blockNum));





×








UNCOV

372



  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pReq->totalLen));





×







373













UNCOV

374



  if ((pReq->data = taosArrayInit(pReq->blockNum, sizeof(void*))) == NULL) {





×







375



    TAOS_CHECK_EXIT(terrno);





×







376



  }











UNCOV

377



  if ((pReq->dataLen = taosArrayInit(pReq->blockNum, sizeof(int32_t))) == NULL) {





×








UNCOV

378



    TAOS_CHECK_EXIT(terrno);





×







379



  }











UNCOV

380



  for (int32_t i = 0; i < pReq->blockNum; i++) {





×







381



    int32_t  len1;










382



    uint64_t len2;










383



    void*    data;











UNCOV

384



    TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &len1));





×








UNCOV

385



    TAOS_CHECK_EXIT(tDecodeBinaryAlloc(pDecoder, &data, &len2));





×







386













UNCOV

387



    if (len1 != len2) {





×







388



      TAOS_CHECK_EXIT(TSDB_CODE_INVALID_MSG);





×







389



    }










390













UNCOV

391



    if (taosArrayPush(pReq->dataLen, &len1) == NULL) {





×







392



      TAOS_CHECK_EXIT(terrno);





×







393



    }










394













UNCOV

395



    if (taosArrayPush(pReq->data, &data) == NULL) {





×







396



      TAOS_CHECK_EXIT(terrno);





×







397



    }










398



  }










399













UNCOV

400



  tEndDecode(pDecoder);




×








UNCOV

401



_exit:




×








UNCOV

402



  return code;




×







403



}










404













UNCOV

405



void tCleanupStreamDispatchReq(SStreamDispatchReq* pReq) {




×








UNCOV

406



  taosArrayDestroyP(pReq->data, NULL);




×








UNCOV

407



  taosArrayDestroy(pReq->dataLen);




×








UNCOV

408



}




×







409













UNCOV

410



int32_t tEncodeStreamRetrieveReq(SEncoder* pEncoder, const SStreamRetrieveReq* pReq) {




×








UNCOV

411



  int32_t code = 0;




×







412



  int32_t lino;










413













UNCOV

414



  TAOS_CHECK_EXIT(tStartEncode(pEncoder));





×








UNCOV

415



  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pReq->streamId));





×








UNCOV

416



  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pReq->reqId));





×








UNCOV

417



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->dstNodeId));





×








UNCOV

418



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->dstTaskId));





×








UNCOV

419



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->srcNodeId));





×








UNCOV

420



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->srcTaskId));





×








UNCOV

421



  TAOS_CHECK_EXIT(tEncodeBinary(pEncoder, (const uint8_t*)pReq->pRetrieve, pReq->retrieveLen));





×








UNCOV

422



  tEndEncode(pEncoder);




×







423













UNCOV

424



_exit:




×








UNCOV

425



  if (code) {





×







426



    return code;




×







427



  } else {











UNCOV

428



    return pEncoder->pos;




×







429



  }










430



}










431













UNCOV

432



int32_t tDecodeStreamRetrieveReq(SDecoder* pDecoder, SStreamRetrieveReq* pReq) {




×








UNCOV

433



  int32_t code = 0;




×







434



  int32_t lino;










435













UNCOV

436



  TAOS_CHECK_EXIT(tStartDecode(pDecoder));





×








UNCOV

437



  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pReq->streamId));





×








UNCOV

438



  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pReq->reqId));





×








UNCOV

439



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pReq->dstNodeId));





×








UNCOV

440



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pReq->dstTaskId));





×








UNCOV

441



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pReq->srcNodeId));





×








UNCOV

442



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pReq->srcTaskId));





×








UNCOV

443



  uint64_t len = 0;




×








UNCOV

444



  TAOS_CHECK_EXIT(tDecodeBinaryAlloc(pDecoder, (void**)&pReq->pRetrieve, &len));





×








UNCOV

445



  pReq->retrieveLen = (int32_t)len;




×








UNCOV

446



  tEndDecode(pDecoder);




×







447













UNCOV

448



_exit:




×








UNCOV

449



  return code;




×







450



}










451













UNCOV

452



void tCleanupStreamRetrieveReq(SStreamRetrieveReq* pReq) { taosMemoryFree(pReq->pRetrieve); }





×







453













UNCOV

454



int32_t tEncodeStreamTaskCheckpointReq(SEncoder* pEncoder, const SStreamTaskCheckpointReq* pReq) {




×








UNCOV

455



  int32_t code = 0;




×







456



  int32_t lino;










457













UNCOV

458



  TAOS_CHECK_EXIT(tStartEncode(pEncoder));





×








UNCOV

459



  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pReq->streamId));





×








UNCOV

460



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->taskId));





×








UNCOV

461



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->nodeId));





×








UNCOV

462



  tEndEncode(pEncoder);




×







463













UNCOV

464



_exit:




×








UNCOV

465



  return code;




×







466



}










467













UNCOV

468



int32_t tDecodeStreamTaskCheckpointReq(SDecoder* pDecoder, SStreamTaskCheckpointReq* pReq) {




×








UNCOV

469



  int32_t code = 0;




×







470



  int32_t lino;










471













UNCOV

472



  TAOS_CHECK_EXIT(tStartDecode(pDecoder));





×








UNCOV

473



  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pReq->streamId));





×








UNCOV

474



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pReq->taskId));





×








UNCOV

475



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pReq->nodeId));





×








UNCOV

476



  tEndDecode(pDecoder);




×







477













UNCOV

478



_exit:




×








UNCOV

479



  return code;




×







480



}










481













UNCOV

482



int32_t tEncodeStreamHbMsg(SEncoder* pEncoder, const SStreamHbMsg* pReq) {




×








UNCOV

483



  int32_t code = 0;




×







484



  int32_t lino;










485













UNCOV

486



  TAOS_CHECK_EXIT(tStartEncode(pEncoder));





×








UNCOV

487



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->vgId));





×








UNCOV

488



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->numOfTasks));





×







489













UNCOV

490



  for (int32_t i = 0; i < pReq->numOfTasks; ++i) {





×








UNCOV

491



    STaskStatusEntry* ps = taosArrayGet(pReq->pTaskStatus, i);




×








UNCOV

492



    if (ps == NULL) {





×







493



      TAOS_CHECK_EXIT(terrno);





×







494



    }










495













UNCOV

496



    TAOS_CHECK_EXIT(tEncodeI64(pEncoder, ps->id.streamId));





×








UNCOV

497



    TAOS_CHECK_EXIT(tEncodeI32(pEncoder, ps->id.taskId));





×








UNCOV

498



    TAOS_CHECK_EXIT(tEncodeI32(pEncoder, ps->status));





×








UNCOV

499



    TAOS_CHECK_EXIT(tEncodeI64(pEncoder, ps->stage));





×








UNCOV

500



    TAOS_CHECK_EXIT(tEncodeI32(pEncoder, ps->nodeId));





×








UNCOV

501



    TAOS_CHECK_EXIT(tEncodeDouble(pEncoder, ps->inputQUsed));





×








UNCOV

502



    TAOS_CHECK_EXIT(tEncodeDouble(pEncoder, ps->inputRate));





×








UNCOV

503



    TAOS_CHECK_EXIT(tEncodeDouble(pEncoder, ps->procsTotal));





×








UNCOV

504



    TAOS_CHECK_EXIT(tEncodeDouble(pEncoder, ps->procsThroughput));





×








UNCOV

505



    TAOS_CHECK_EXIT(tEncodeDouble(pEncoder, ps->outputTotal));





×








UNCOV

506



    TAOS_CHECK_EXIT(tEncodeDouble(pEncoder, ps->outputThroughput));





×








UNCOV

507



    TAOS_CHECK_EXIT(tEncodeDouble(pEncoder, ps->sinkQuota));





×








UNCOV

508



    TAOS_CHECK_EXIT(tEncodeDouble(pEncoder, ps->sinkDataSize));





×








UNCOV

509



    TAOS_CHECK_EXIT(tEncodeI64(pEncoder, ps->processedVer));





×








UNCOV

510



    TAOS_CHECK_EXIT(tEncodeI64(pEncoder, ps->verRange.minVer));





×








UNCOV

511



    TAOS_CHECK_EXIT(tEncodeI64(pEncoder, ps->verRange.maxVer));





×








UNCOV

512



    TAOS_CHECK_EXIT(tEncodeI64(pEncoder, ps->checkpointInfo.activeId));





×








UNCOV

513



    TAOS_CHECK_EXIT(tEncodeI8(pEncoder, ps->checkpointInfo.failed));





×








UNCOV

514



    TAOS_CHECK_EXIT(tEncodeI32(pEncoder, ps->checkpointInfo.activeTransId));





×








UNCOV

515



    TAOS_CHECK_EXIT(tEncodeI64(pEncoder, ps->checkpointInfo.latestId));





×








UNCOV

516



    TAOS_CHECK_EXIT(tEncodeI64(pEncoder, ps->checkpointInfo.latestVer));





×








UNCOV

517



    TAOS_CHECK_EXIT(tEncodeI64(pEncoder, ps->checkpointInfo.latestTime));





×








UNCOV

518



    TAOS_CHECK_EXIT(tEncodeI64(pEncoder, ps->checkpointInfo.latestSize));





×








UNCOV

519



    TAOS_CHECK_EXIT(tEncodeI8(pEncoder, ps->checkpointInfo.remoteBackup));





×








UNCOV

520



    TAOS_CHECK_EXIT(tEncodeI8(pEncoder, ps->checkpointInfo.consensusChkptId));





×








UNCOV

521



    TAOS_CHECK_EXIT(tEncodeI64(pEncoder, ps->checkpointInfo.consensusTs));





×








UNCOV

522



    TAOS_CHECK_EXIT(tEncodeI64(pEncoder, ps->startTime));





×








UNCOV

523



    TAOS_CHECK_EXIT(tEncodeI64(pEncoder, ps->startCheckpointId));





×








UNCOV

524



    TAOS_CHECK_EXIT(tEncodeI64(pEncoder, ps->startCheckpointVer));





×








UNCOV

525



    TAOS_CHECK_EXIT(tEncodeI64(pEncoder, ps->hTaskId));





×







526



  }










527













UNCOV

528



  int32_t numOfVgs = taosArrayGetSize(pReq->pUpdateNodes);





×








UNCOV

529



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, numOfVgs));





×







530













UNCOV

531



  for (int j = 0; j < numOfVgs; ++j) {





×








UNCOV

532



    int32_t* pVgId = taosArrayGet(pReq->pUpdateNodes, j);




×








UNCOV

533



    if (pVgId == NULL) {





×







534



      TAOS_CHECK_EXIT(terrno);





×







535



    }










536













UNCOV

537



    TAOS_CHECK_EXIT(tEncodeI32(pEncoder, *pVgId));





×







538



  }










539













UNCOV

540



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->msgId));





×








UNCOV

541



  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pReq->ts));





×








UNCOV

542



  tEndEncode(pEncoder);




×







543













UNCOV

544



_exit:




×








UNCOV

545



  if (code) {





×







546



    return code;




×







547



  } else {











UNCOV

548



    return pEncoder->pos;




×







549



  }










550



}










551













UNCOV

552



int32_t tDecodeStreamHbMsg(SDecoder* pDecoder, SStreamHbMsg* pReq) {




×








UNCOV

553



  int32_t code = 0;




×







554



  int32_t lino;










555













UNCOV

556



  TAOS_CHECK_EXIT(tStartDecode(pDecoder));





×








UNCOV

557



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pReq->vgId));





×








UNCOV

558



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pReq->numOfTasks));





×







559













UNCOV

560



  if ((pReq->pTaskStatus = taosArrayInit(pReq->numOfTasks, sizeof(STaskStatusEntry))) == NULL) {





×







561



    TAOS_CHECK_EXIT(terrno);





×







562



  }











UNCOV

563



  for (int32_t i = 0; i < pReq->numOfTasks; ++i) {





×








UNCOV

564



    int32_t          taskId = 0;




×








UNCOV

565



    STaskStatusEntry entry = {0};




×







566













UNCOV

567



    TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &entry.id.streamId));





×








UNCOV

568



    TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &taskId));





×








UNCOV

569



    TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &entry.status));





×








UNCOV

570



    TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &entry.stage));





×








UNCOV

571



    TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &entry.nodeId));





×








UNCOV

572



    TAOS_CHECK_EXIT(tDecodeDouble(pDecoder, &entry.inputQUsed));





×








UNCOV

573



    TAOS_CHECK_EXIT(tDecodeDouble(pDecoder, &entry.inputRate));





×








UNCOV

574



    TAOS_CHECK_EXIT(tDecodeDouble(pDecoder, &entry.procsTotal));





×








UNCOV

575



    TAOS_CHECK_EXIT(tDecodeDouble(pDecoder, &entry.procsThroughput));





×








UNCOV

576



    TAOS_CHECK_EXIT(tDecodeDouble(pDecoder, &entry.outputTotal));





×








UNCOV

577



    TAOS_CHECK_EXIT(tDecodeDouble(pDecoder, &entry.outputThroughput));





×








UNCOV

578



    TAOS_CHECK_EXIT(tDecodeDouble(pDecoder, &entry.sinkQuota));





×








UNCOV

579



    TAOS_CHECK_EXIT(tDecodeDouble(pDecoder, &entry.sinkDataSize));





×








UNCOV

580



    TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &entry.processedVer));





×








UNCOV

581



    TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &entry.verRange.minVer));





×








UNCOV

582



    TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &entry.verRange.maxVer));





×








UNCOV

583



    TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &entry.checkpointInfo.activeId));





×








UNCOV

584



    TAOS_CHECK_EXIT(tDecodeI8(pDecoder, &entry.checkpointInfo.failed));





×








UNCOV

585



    TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &entry.checkpointInfo.activeTransId));





×







586













UNCOV

587



    TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &entry.checkpointInfo.latestId));





×








UNCOV

588



    TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &entry.checkpointInfo.latestVer));





×








UNCOV

589



    TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &entry.checkpointInfo.latestTime));





×








UNCOV

590



    TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &entry.checkpointInfo.latestSize));





×








UNCOV

591



    TAOS_CHECK_EXIT(tDecodeI8(pDecoder, &entry.checkpointInfo.remoteBackup));





×








UNCOV

592



    TAOS_CHECK_EXIT(tDecodeI8(pDecoder, &entry.checkpointInfo.consensusChkptId));





×








UNCOV

593



    TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &entry.checkpointInfo.consensusTs));





×








UNCOV

594



    TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &entry.startTime));





×








UNCOV

595



    TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &entry.startCheckpointId));





×








UNCOV

596



    TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &entry.startCheckpointVer));





×








UNCOV

597



    TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &entry.hTaskId));





×







598













UNCOV

599



    entry.id.taskId = taskId;




×








UNCOV

600



    if (taosArrayPush(pReq->pTaskStatus, &entry) == NULL) {





×







601



      TAOS_CHECK_EXIT(terrno);





×







602



    }










603



  }










604













UNCOV

605



  int32_t numOfVgs = 0;




×








UNCOV

606



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &numOfVgs));





×







607













UNCOV

608



  if ((pReq->pUpdateNodes = taosArrayInit(numOfVgs, sizeof(int32_t))) == NULL) {





×







609



    TAOS_CHECK_EXIT(terrno);





×







610



  }










611













UNCOV

612



  for (int j = 0; j < numOfVgs; ++j) {





×








UNCOV

613



    int32_t vgId = 0;




×








UNCOV

614



    TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &vgId));





×








UNCOV

615



    if (taosArrayPush(pReq->pUpdateNodes, &vgId) == NULL) {





×







616



      TAOS_CHECK_EXIT(terrno);





×







617



    }










618



  }










619













UNCOV

620



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pReq->msgId));





×








UNCOV

621



  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pReq->ts));





×








UNCOV

622



  tEndDecode(pDecoder);




×







623













UNCOV

624



_exit:




×








UNCOV

625



  return code;




×







626



}










627












628



void tCleanupStreamHbMsg(SStreamHbMsg* pMsg) {




944✔





629



  if (pMsg == NULL) {





944!





630



    return;




×







631



  }










632












633



  if (pMsg->pUpdateNodes != NULL) {





944✔





634



    taosArrayDestroy(pMsg->pUpdateNodes);




430✔





635



    pMsg->pUpdateNodes = NULL;




430✔





636



  }










637












638



  if (pMsg->pTaskStatus != NULL) {





944✔





639



    taosArrayDestroy(pMsg->pTaskStatus);




430✔





640



    pMsg->pTaskStatus = NULL;




430✔





641



  }










642












643



  pMsg->msgId = -1;




944✔





644



  pMsg->vgId = -1;




944✔





645



  pMsg->numOfTasks = -1;




944✔





646



}










647













UNCOV

648



int32_t tEncodeStreamHbRsp(SEncoder* pEncoder, const SMStreamHbRspMsg* pRsp) {




×








UNCOV

649



  int32_t code = 0;




×







650



  int32_t lino;










651













UNCOV

652



  TAOS_CHECK_EXIT(tStartEncode(pEncoder));





×








UNCOV

653



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pRsp->msgId));





×








UNCOV

654



  TAOS_CHECK_EXIT(tEncodeSEpSet(pEncoder, &pRsp->mndEpset));





×








UNCOV

655



  tEndEncode(pEncoder);




×







656













UNCOV

657



_exit:




×








UNCOV

658



  return code;




×







659



}










660













UNCOV

661



int32_t tDecodeStreamHbRsp(SDecoder* pDecoder, SMStreamHbRspMsg* pRsp) {




×








UNCOV

662



  int32_t code = 0;




×







663



  int32_t lino;










664













UNCOV

665



  TAOS_CHECK_EXIT(tStartDecode(pDecoder));





×








UNCOV

666



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pRsp->msgId));





×








UNCOV

667



  TAOS_CHECK_EXIT(tDecodeSEpSet(pDecoder, &pRsp->mndEpset));





×








UNCOV

668



  tEndDecode(pDecoder);




×







669













UNCOV

670



_exit:




×








UNCOV

671



  return code;




×







672



}










673












674



int32_t tEncodeRetrieveChkptTriggerReq(SEncoder* pEncoder, const SRetrieveChkptTriggerReq* pReq) {




×







675



  int32_t code = 0;




×







676



  int32_t lino;










677












678



  TAOS_CHECK_EXIT(tStartEncode(pEncoder));





×







679



  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pReq->streamId));





×







680



  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pReq->checkpointId));





×







681



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->upstreamNodeId));





×







682



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->upstreamTaskId));





×







683



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->downstreamNodeId));





×







684



  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pReq->downstreamTaskId));





×







685



  tEndEncode(pEncoder);




×







686












687



_exit:




×







688



  return code;




×







689



}










690












691



int32_t tDecodeRetrieveChkptTriggerReq(SDecoder* pDecoder, SRetrieveChkptTriggerReq* pReq) {




×







692



  int32_t code = 0;




×







693



  int32_t lino;










694












695



  TAOS_CHECK_EXIT(tStartDecode(pDecoder));





×







696



  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pReq->streamId));





×







697



  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pReq->checkpointId));





×







698



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pReq->upstreamNodeId));





×







699



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pReq->upstreamTaskId));





×







700



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pReq->downstreamNodeId));





×







701



  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pReq->downstreamTaskId));





×







702



  tEndDecode(pDecoder);




×







703












704



_exit:




×







705



  return code;




×







706



}










707












708



int32_t tEncodeCheckpointTriggerRsp(SEncoder* pEncoder, const SCheckpointTriggerRsp* pRsp) {




×







709



  int32_t code = 0;




×







710



  int32_t lino;










711












712



  TAOS_CHECK_EXIT(tStartEncode(pEncoder));





×







713



  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pRsp->streamId));





×







714



  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pRsp->checkpointId));





×







715



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pRsp->upstreamTaskId));





×







716



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pRsp->taskId));





×







717



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pRsp->transId));





×







718



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pRsp->rspCode));





×







719



  tEndEncode(pEncoder);




×







720












721



_exit:




×







722



  return code;




×







723



}










724












725



int32_t tDecodeCheckpointTriggerRsp(SDecoder* pDecoder, SCheckpointTriggerRsp* pRsp) {




×







726



  int32_t code = 0;




×







727



  int32_t lino;










728












729



  TAOS_CHECK_EXIT(tStartDecode(pDecoder));





×







730



  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pRsp->streamId));





×







731



  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pRsp->checkpointId));





×







732



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pRsp->upstreamTaskId));





×







733



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pRsp->taskId));





×







734



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pRsp->transId));





×







735



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pRsp->rspCode));





×







736



  tEndDecode(pDecoder);




×







737












738



_exit:




×







739



  return code;




×







740



}










741













UNCOV

742



int32_t tEncodeStreamTaskChkptReport(SEncoder* pEncoder, const SCheckpointReport* pReq) {




×








UNCOV

743



  int32_t code = 0;




×







744



  int32_t lino;










745













UNCOV

746



  TAOS_CHECK_EXIT(tStartEncode(pEncoder));





×








UNCOV

747



  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pReq->streamId));





×








UNCOV

748



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->taskId));





×








UNCOV

749



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->nodeId));





×








UNCOV

750



  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pReq->checkpointId));





×








UNCOV

751



  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pReq->checkpointVer));





×








UNCOV

752



  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pReq->checkpointTs));





×








UNCOV

753



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->transId));





×








UNCOV

754



  TAOS_CHECK_EXIT(tEncodeI8(pEncoder, pReq->dropHTask));





×








UNCOV

755



  tEndEncode(pEncoder);




×







756













UNCOV

757



_exit:




×








UNCOV

758



  return code;




×







759



}










760













UNCOV

761



int32_t tDecodeStreamTaskChkptReport(SDecoder* pDecoder, SCheckpointReport* pReq) {




×








UNCOV

762



  int32_t code = 0;




×







763



  int32_t lino;










764













UNCOV

765



  TAOS_CHECK_EXIT(tStartDecode(pDecoder));





×








UNCOV

766



  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pReq->streamId));





×








UNCOV

767



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pReq->taskId));





×








UNCOV

768



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pReq->nodeId));





×








UNCOV

769



  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pReq->checkpointId));





×








UNCOV

770



  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pReq->checkpointVer));





×








UNCOV

771



  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pReq->checkpointTs));





×








UNCOV

772



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pReq->transId));





×








UNCOV

773



  TAOS_CHECK_EXIT(tDecodeI8(pDecoder, &pReq->dropHTask));





×








UNCOV

774



  tEndDecode(pDecoder);




×







775













UNCOV

776



_exit:




×








UNCOV

777



  return code;




×







778



}










779













UNCOV

780



int32_t tEncodeRestoreCheckpointInfo(SEncoder* pEncoder, const SRestoreCheckpointInfo* pReq) {




×








UNCOV

781



  int32_t code = 0;




×







782



  int32_t lino;










783













UNCOV

784



  TAOS_CHECK_EXIT(tStartEncode(pEncoder));





×








UNCOV

785



  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pReq->startTs));





×








UNCOV

786



  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pReq->streamId));





×








UNCOV

787



  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pReq->checkpointId));





×








UNCOV

788



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->transId));





×








UNCOV

789



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->taskId));





×








UNCOV

790



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->nodeId));





×








UNCOV

791



  tEndEncode(pEncoder);




×







792













UNCOV

793



_exit:




×








UNCOV

794



  return code;




×







795



}










796













UNCOV

797



int32_t tDecodeRestoreCheckpointInfo(SDecoder* pDecoder, SRestoreCheckpointInfo* pReq) {




×








UNCOV

798



  int32_t code = 0;




×







799



  int32_t lino;










800













UNCOV

801



  TAOS_CHECK_EXIT(tStartDecode(pDecoder));





×








UNCOV

802



  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pReq->startTs));





×








UNCOV

803



  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pReq->streamId));





×








UNCOV

804



  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pReq->checkpointId));





×








UNCOV

805



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pReq->transId));





×








UNCOV

806



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pReq->taskId));





×








UNCOV

807



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pReq->nodeId));





×








UNCOV

808



  tEndDecode(pDecoder);




×







809













UNCOV

810



_exit:




×








UNCOV

811



  return code;




×







812



}










813












814



int32_t tEncodeStreamTaskRunReq (SEncoder* pEncoder, const SStreamTaskRunReq* pReq) {




1,844✔





815



  int32_t code = 0;




1,844✔





816



  int32_t lino;










817












818



  TAOS_CHECK_EXIT(tStartEncode(pEncoder));





1,844!





819



  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pReq->streamId));





3,688!





820



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->taskId));





3,688!





821



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pReq->reqType));





3,688!





822



  tEndEncode(pEncoder);




1,844✔





823












824



_exit:




1,844✔





825



  return code;




1,844✔





826



}










827












828



int32_t tDecodeStreamTaskRunReq(SDecoder* pDecoder, SStreamTaskRunReq* pReq) {




1,842✔





829



  int32_t code = 0;




1,842✔





830



  int32_t lino;










831












832



  TAOS_CHECK_EXIT(tStartDecode(pDecoder));





1,842!





833



  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pReq->streamId));





3,684!





834



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pReq->taskId));





3,684!





835



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pReq->reqType));





3,684!





836



  tEndDecode(pDecoder);




1,842✔





837












838



_exit:




1,842✔





839



  return code;




1,842✔





840



}























    

  • 

    •