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/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



  STaskNotifyEventStat notifyEventStat;










60



} STaskStatusEntry;










61













UNCOV

62



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




×








UNCOV

63



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





×








UNCOV

64



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





×








UNCOV

65



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





×








UNCOV

66



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





×








UNCOV

67



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





×








UNCOV

68



  return 0;




×







69



}










70













UNCOV

71



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




×








UNCOV

72



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





×








UNCOV

73



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





×








UNCOV

74



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





×








UNCOV

75



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





×








UNCOV

76



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





×








UNCOV

77



  return 0;




×







78



}










79













UNCOV

80



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




×








UNCOV

81



  TAOS_CHECK_RETURN(tStartEncode(pEncoder));





×








UNCOV

82



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





×








UNCOV

83



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





×








UNCOV

84



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





×








UNCOV

85



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





×








UNCOV

86



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





×








UNCOV

87



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





×








UNCOV

88



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





×








UNCOV

89



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





×








UNCOV

90



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





×








UNCOV

91



  tEndEncode(pEncoder);




×








UNCOV

92



  return pEncoder->pos;




×







93



}










94













UNCOV

95



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




×








UNCOV

96



  TAOS_CHECK_RETURN(tStartDecode(pDecoder));





×








UNCOV

97



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





×








UNCOV

98



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





×








UNCOV

99



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





×








UNCOV

100



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





×








UNCOV

101



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





×








UNCOV

102



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





×








UNCOV

103



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





×








UNCOV

104



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





×








UNCOV

105



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





×








UNCOV

106



  tEndDecode(pDecoder);




×








UNCOV

107



  return 0;




×







108



}










109













UNCOV

110



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




×








UNCOV

111



  TAOS_CHECK_RETURN(tStartEncode(pEncoder));





×








UNCOV

112



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





×








UNCOV

113



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





×








UNCOV

114



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





×








UNCOV

115



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





×








UNCOV

116



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





×








UNCOV

117



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





×








UNCOV

118



  tEndEncode(pEncoder);




×








UNCOV

119



  return pEncoder->pos;




×







120



}










121













UNCOV

122



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




×








UNCOV

123



  int32_t code = 0;




×







124



  int32_t lino;










125













UNCOV

126



  TAOS_CHECK_EXIT(tStartEncode(pEncoder));





×








UNCOV

127



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





×








UNCOV

128



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





×







129













UNCOV

130



  int32_t size = taosArrayGetSize(pMsg->pNodeList);





×








UNCOV

131



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, size));





×







132













UNCOV

133



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





×








UNCOV

134



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




×








UNCOV

135



    if (pInfo == NULL) {





×







136



      TAOS_CHECK_EXIT(terrno);





×







137



    }










138













UNCOV

139



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





×








UNCOV

140



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





×








UNCOV

141



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





×







142



  }










143












144



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











UNCOV

145



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





×








UNCOV

146



  tEndEncode(pEncoder);




×








UNCOV

147



_exit:




×








UNCOV

148



  if (code) {





×







149



    return code;




×







150



  } else {











UNCOV

151



    return pEncoder->pos;




×







152



  }










153



}










154













UNCOV

155



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




×








UNCOV

156



  int32_t code = 0;




×







157



  int32_t lino;










158













UNCOV

159



  TAOS_CHECK_EXIT(tStartDecode(pDecoder));





×








UNCOV

160



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





×








UNCOV

161



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





×







162













UNCOV

163



  int32_t size = 0;




×








UNCOV

164



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &size));





×








UNCOV

165



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




×








UNCOV

166



  if (pMsg->pNodeList == NULL) {





×







167



    TAOS_CHECK_EXIT(terrno);





×







168



  }











UNCOV

169



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





×








UNCOV

170



    SNodeUpdateInfo info = {0};




×








UNCOV

171



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





×








UNCOV

172



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





×








UNCOV

173



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





×







174













UNCOV

175



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





×







176



      TAOS_CHECK_EXIT(terrno);





×







177



    }










178



  }










179













UNCOV

180



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





×







181













UNCOV

182



  tEndDecode(pDecoder);




×








UNCOV

183



_exit:




×








UNCOV

184



  return code;




×







185



}










186













UNCOV

187



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




×








UNCOV

188



  int32_t code = 0;




×







189



  int32_t lino;










190













UNCOV

191



  TAOS_CHECK_EXIT(tStartEncode(pEncoder));





×








UNCOV

192



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





×








UNCOV

193



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





×








UNCOV

194



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





×








UNCOV

195



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





×








UNCOV

196



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





×








UNCOV

197



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





×








UNCOV

198



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





×








UNCOV

199



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





×








UNCOV

200



  tEndEncode(pEncoder);




×







201













UNCOV

202



_exit:




×








UNCOV

203



  if (code) {





×







204



    return code;




×







205



  } else {











UNCOV

206



    return pEncoder->pos;




×







207



  }










208



}










209













UNCOV

210



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




×








UNCOV

211



  int32_t code = 0;




×







212



  int32_t lino;










213













UNCOV

214



  TAOS_CHECK_EXIT(tStartDecode(pDecoder));





×








UNCOV

215



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





×








UNCOV

216



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





×








UNCOV

217



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





×








UNCOV

218



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





×








UNCOV

219



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





×








UNCOV

220



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





×








UNCOV

221



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





×








UNCOV

222



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





×








UNCOV

223



  tEndDecode(pDecoder);




×







224













UNCOV

225



_exit:




×








UNCOV

226



  return code;




×







227



}










228













UNCOV

229



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




×








UNCOV

230



  int32_t code = 0;




×







231



  int32_t lino;










232













UNCOV

233



  TAOS_CHECK_EXIT(tStartEncode(pEncoder));





×








UNCOV

234



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





×








UNCOV

235



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





×








UNCOV

236



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





×








UNCOV

237



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





×








UNCOV

238



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





×








UNCOV

239



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





×








UNCOV

240



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





×








UNCOV

241



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





×








UNCOV

242



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





×








UNCOV

243



  tEndEncode(pEncoder);




×







244













UNCOV

245



_exit:




×








UNCOV

246



  if (code) {





×







247



    return code;




×







248



  } else {











UNCOV

249



    return pEncoder->pos;




×







250



  }










251



}










252













UNCOV

253



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




×








UNCOV

254



  int32_t code = 0;




×







255



  int32_t lino;










256













UNCOV

257



  TAOS_CHECK_EXIT(tStartDecode(pDecoder));





×








UNCOV

258



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





×








UNCOV

259



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





×








UNCOV

260



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





×








UNCOV

261



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





×








UNCOV

262



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





×








UNCOV

263



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





×








UNCOV

264



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





×








UNCOV

265



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





×








UNCOV

266



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





×








UNCOV

267



  tEndDecode(pDecoder);




×







268













UNCOV

269



_exit:




×








UNCOV

270



  return code;




×







271



}










272













UNCOV

273



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




×








UNCOV

274



  int32_t code = 0;




×







275



  int32_t lino;










276













UNCOV

277



  TAOS_CHECK_EXIT(tStartEncode(pEncoder));





×








UNCOV

278



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





×








UNCOV

279



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





×








UNCOV

280



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





×








UNCOV

281



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





×








UNCOV

282



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





×








UNCOV

283



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





×








UNCOV

284



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





×








UNCOV

285



  tEndEncode(pEncoder);




×







286













UNCOV

287



_exit:




×








UNCOV

288



  if (code) {





×







289



    return code;




×







290



  } else {











UNCOV

291



    return pEncoder->pos;




×







292



  }










293



}










294













UNCOV

295



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




×








UNCOV

296



  int32_t code = 0;




×







297



  int32_t lino;










298













UNCOV

299



  TAOS_CHECK_EXIT(tStartDecode(pDecoder));





×








UNCOV

300



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





×








UNCOV

301



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





×








UNCOV

302



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





×








UNCOV

303



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





×








UNCOV

304



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





×








UNCOV

305



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





×








UNCOV

306



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





×








UNCOV

307



  tEndDecode(pDecoder);




×







308













UNCOV

309



_exit:




×








UNCOV

310



  return code;




×







311



}










312













UNCOV

313



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




×








UNCOV

314



  int32_t code = 0;




×







315



  int32_t lino;










316













UNCOV

317



  TAOS_CHECK_EXIT(tStartEncode(pEncoder));





×








UNCOV

318



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





×








UNCOV

319



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





×








UNCOV

320



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





×








UNCOV

321



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





×








UNCOV

322



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





×








UNCOV

323



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





×








UNCOV

324



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





×








UNCOV

325



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





×








UNCOV

326



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





×








UNCOV

327



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





×








UNCOV

328



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





×








UNCOV

329



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





×








UNCOV

330



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





×







331













UNCOV

332



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





×







333



    uError("invalid dispatch req msg");





×







334



    TAOS_CHECK_EXIT(TSDB_CODE_INVALID_MSG);





×







335



  }










336













UNCOV

337



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





×








UNCOV

338



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




×








UNCOV

339



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




×








UNCOV

340



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





×








UNCOV

341



      TAOS_CHECK_EXIT(terrno);





×







342



    }










343













UNCOV

344



    TAOS_CHECK_EXIT(tEncodeI32(pEncoder, *pLen));





×








UNCOV

345



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





×







346



  }











UNCOV

347



  tEndEncode(pEncoder);




×








UNCOV

348



_exit:




×








UNCOV

349



  if (code) {





×







350



    return code;




×







351



  } else {











UNCOV

352



    return pEncoder->pos;




×







353



  }










354



}










355













UNCOV

356



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




×








UNCOV

357



  int32_t code = 0;




×







358



  int32_t lino;










359













UNCOV

360



  TAOS_CHECK_EXIT(tStartDecode(pDecoder));





×








UNCOV

361



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





×








UNCOV

362



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





×








UNCOV

363



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





×








UNCOV

364



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





×








UNCOV

365



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





×








UNCOV

366



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





×








UNCOV

367



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





×








UNCOV

368



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





×








UNCOV

369



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





×








UNCOV

370



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





×








UNCOV

371



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





×








UNCOV

372



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





×








UNCOV

373



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





×







374













UNCOV

375



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





×







376



    TAOS_CHECK_EXIT(terrno);





×







377



  }











UNCOV

378



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





×







379



    TAOS_CHECK_EXIT(terrno);





×







380



  }











UNCOV

381



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





×







382



    int32_t  len1;










383



    uint64_t len2;










384



    void*    data;











UNCOV

385



    TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &len1));





×








UNCOV

386



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





×







387













UNCOV

388



    if (len1 != len2) {





×







389



      TAOS_CHECK_EXIT(TSDB_CODE_INVALID_MSG);





×







390



    }










391













UNCOV

392



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





×







393



      TAOS_CHECK_EXIT(terrno);





×







394



    }










395













UNCOV

396



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





×







397



      TAOS_CHECK_EXIT(terrno);





×







398



    }










399



  }










400













UNCOV

401



  tEndDecode(pDecoder);




×








UNCOV

402



_exit:




×








UNCOV

403



  return code;




×







404



}










405













UNCOV

406



void tCleanupStreamDispatchReq(SStreamDispatchReq* pReq) {




×








UNCOV

407



  taosArrayDestroyP(pReq->data, NULL);




×








UNCOV

408



  taosArrayDestroy(pReq->dataLen);




×








UNCOV

409



}




×







410













UNCOV

411



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




×








UNCOV

412



  int32_t code = 0;




×







413



  int32_t lino;










414













UNCOV

415



  TAOS_CHECK_EXIT(tStartEncode(pEncoder));





×








UNCOV

416



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





×








UNCOV

417



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





×








UNCOV

418



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





×








UNCOV

419



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





×








UNCOV

420



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





×








UNCOV

421



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





×








UNCOV

422



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





×








UNCOV

423



  tEndEncode(pEncoder);




×







424













UNCOV

425



_exit:




×








UNCOV

426



  if (code) {





×







427



    return code;




×







428



  } else {











UNCOV

429



    return pEncoder->pos;




×







430



  }










431



}










432













UNCOV

433



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




×








UNCOV

434



  int32_t code = 0;




×







435



  int32_t lino;










436













UNCOV

437



  TAOS_CHECK_EXIT(tStartDecode(pDecoder));





×








UNCOV

438



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





×








UNCOV

439



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





×








UNCOV

440



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





×








UNCOV

441



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





×








UNCOV

442



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





×








UNCOV

443



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





×








UNCOV

444



  uint64_t len = 0;




×








UNCOV

445



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





×








UNCOV

446



  pReq->retrieveLen = (int32_t)len;




×








UNCOV

447



  tEndDecode(pDecoder);




×







448













UNCOV

449



_exit:




×








UNCOV

450



  return code;




×







451



}










452













UNCOV

453



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





×







454













UNCOV

455



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




×








UNCOV

456



  int32_t code = 0;




×







457



  int32_t lino;










458













UNCOV

459



  TAOS_CHECK_EXIT(tStartEncode(pEncoder));





×








UNCOV

460



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





×








UNCOV

461



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





×








UNCOV

462



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





×








UNCOV

463



  tEndEncode(pEncoder);




×







464













UNCOV

465



_exit:




×








UNCOV

466



  return code;




×







467



}










468













UNCOV

469



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




×








UNCOV

470



  int32_t code = 0;




×







471



  int32_t lino;










472













UNCOV

473



  TAOS_CHECK_EXIT(tStartDecode(pDecoder));





×








UNCOV

474



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





×








UNCOV

475



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





×








UNCOV

476



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





×








UNCOV

477



  tEndDecode(pDecoder);




×







478













UNCOV

479



_exit:




×








UNCOV

480



  return code;




×







481



}










482













UNCOV

483



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




×








UNCOV

484



  int32_t code = 0;




×







485



  int32_t lino;










486













UNCOV

487



  TAOS_CHECK_EXIT(tStartEncode(pEncoder));





×








UNCOV

488



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





×








UNCOV

489



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





×







490













UNCOV

491



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





×








UNCOV

492



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




×








UNCOV

493



    if (ps == NULL) {





×







494



      TAOS_CHECK_EXIT(terrno);





×







495



    }










496













UNCOV

497



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





×








UNCOV

498



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





×








UNCOV

499



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





×








UNCOV

500



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





×








UNCOV

501



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





×








UNCOV

502



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





×








UNCOV

503



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





×








UNCOV

504



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





×








UNCOV

505



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





×








UNCOV

506



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





×








UNCOV

507



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





×








UNCOV

508



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





×








UNCOV

509



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





×








UNCOV

510



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





×








UNCOV

511



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





×








UNCOV

512



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





×








UNCOV

513



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





×








UNCOV

514



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





×








UNCOV

515



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





×








UNCOV

516



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





×








UNCOV

517



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





×








UNCOV

518



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





×








UNCOV

519



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





×








UNCOV

520



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





×








UNCOV

521



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





×








UNCOV

522



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





×








UNCOV

523



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





×








UNCOV

524



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





×








UNCOV

525



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





×








UNCOV

526



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





×








UNCOV

527



    TAOS_CHECK_EXIT(tEncodeI64(pEncoder, ps->notifyEventStat.notifyEventAddTimes));





×








UNCOV

528



    TAOS_CHECK_EXIT(tEncodeI64(pEncoder, ps->notifyEventStat.notifyEventAddElems));





×








UNCOV

529



    TAOS_CHECK_EXIT(tEncodeDouble(pEncoder, ps->notifyEventStat.notifyEventAddCostSec));





×








UNCOV

530



    TAOS_CHECK_EXIT(tEncodeI64(pEncoder, ps->notifyEventStat.notifyEventPushTimes));





×








UNCOV

531



    TAOS_CHECK_EXIT(tEncodeI64(pEncoder, ps->notifyEventStat.notifyEventPushElems));





×








UNCOV

532



    TAOS_CHECK_EXIT(tEncodeDouble(pEncoder, ps->notifyEventStat.notifyEventPushCostSec));





×








UNCOV

533



    TAOS_CHECK_EXIT(tEncodeI64(pEncoder, ps->notifyEventStat.notifyEventPackTimes));





×








UNCOV

534



    TAOS_CHECK_EXIT(tEncodeI64(pEncoder, ps->notifyEventStat.notifyEventPackElems));





×








UNCOV

535



    TAOS_CHECK_EXIT(tEncodeDouble(pEncoder, ps->notifyEventStat.notifyEventPackCostSec));





×








UNCOV

536



    TAOS_CHECK_EXIT(tEncodeI64(pEncoder, ps->notifyEventStat.notifyEventSendTimes));





×








UNCOV

537



    TAOS_CHECK_EXIT(tEncodeI64(pEncoder, ps->notifyEventStat.notifyEventSendElems));





×








UNCOV

538



    TAOS_CHECK_EXIT(tEncodeDouble(pEncoder, ps->notifyEventStat.notifyEventSendCostSec));





×








UNCOV

539



    TAOS_CHECK_EXIT(tEncodeI64(pEncoder, ps->notifyEventStat.notifyEventHoldElems));





×







540



  }










541













UNCOV

542



  int32_t numOfVgs = taosArrayGetSize(pReq->pUpdateNodes);





×








UNCOV

543



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, numOfVgs));





×







544













UNCOV

545



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





×








UNCOV

546



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




×








UNCOV

547



    if (pVgId == NULL) {





×







548



      TAOS_CHECK_EXIT(terrno);





×







549



    }










550













UNCOV

551



    TAOS_CHECK_EXIT(tEncodeI32(pEncoder, *pVgId));





×







552



  }










553













UNCOV

554



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





×








UNCOV

555



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





×








UNCOV

556



  tEndEncode(pEncoder);




×







557













UNCOV

558



_exit:




×








UNCOV

559



  if (code) {





×







560



    return code;




×







561



  } else {











UNCOV

562



    return pEncoder->pos;




×







563



  }










564



}










565













UNCOV

566



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




×








UNCOV

567



  int32_t code = 0;




×







568



  int32_t lino;










569













UNCOV

570



  TAOS_CHECK_EXIT(tStartDecode(pDecoder));





×








UNCOV

571



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





×








UNCOV

572



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





×







573













UNCOV

574



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





×







575



    TAOS_CHECK_EXIT(terrno);





×







576



  }











UNCOV

577



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





×








UNCOV

578



    int32_t          taskId = 0;




×








UNCOV

579



    STaskStatusEntry entry = {0};




×







580













UNCOV

581



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





×








UNCOV

582



    TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &taskId));





×








UNCOV

583



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





×








UNCOV

584



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





×








UNCOV

585



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





×








UNCOV

586



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





×








UNCOV

587



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





×








UNCOV

588



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





×








UNCOV

589



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





×








UNCOV

590



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





×








UNCOV

591



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





×








UNCOV

592



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





×








UNCOV

593



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





×








UNCOV

594



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





×








UNCOV

595



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





×








UNCOV

596



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





×








UNCOV

597



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





×








UNCOV

598



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





×








UNCOV

599



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





×







600













UNCOV

601



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





×








UNCOV

602



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





×








UNCOV

603



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





×








UNCOV

604



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





×








UNCOV

605



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





×








UNCOV

606



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





×








UNCOV

607



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





×








UNCOV

608



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





×








UNCOV

609



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





×








UNCOV

610



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





×








UNCOV

611



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





×







612













UNCOV

613



    TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &entry.notifyEventStat.notifyEventAddTimes));





×








UNCOV

614



    TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &entry.notifyEventStat.notifyEventAddElems));





×








UNCOV

615



    TAOS_CHECK_EXIT(tDecodeDouble(pDecoder, &entry.notifyEventStat.notifyEventAddCostSec));





×








UNCOV

616



    TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &entry.notifyEventStat.notifyEventPushTimes));





×








UNCOV

617



    TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &entry.notifyEventStat.notifyEventPushElems));





×








UNCOV

618



    TAOS_CHECK_EXIT(tDecodeDouble(pDecoder, &entry.notifyEventStat.notifyEventPushCostSec));





×








UNCOV

619



    TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &entry.notifyEventStat.notifyEventPackTimes));





×








UNCOV

620



    TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &entry.notifyEventStat.notifyEventPackElems));





×








UNCOV

621



    TAOS_CHECK_EXIT(tDecodeDouble(pDecoder, &entry.notifyEventStat.notifyEventPackCostSec));





×








UNCOV

622



    TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &entry.notifyEventStat.notifyEventSendTimes));





×








UNCOV

623



    TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &entry.notifyEventStat.notifyEventSendElems));





×








UNCOV

624



    TAOS_CHECK_EXIT(tDecodeDouble(pDecoder, &entry.notifyEventStat.notifyEventSendCostSec));





×








UNCOV

625



    TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &entry.notifyEventStat.notifyEventHoldElems));





×







626













UNCOV

627



    entry.id.taskId = taskId;




×








UNCOV

628



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





×







629



      TAOS_CHECK_EXIT(terrno);





×







630



    }










631



  }










632













UNCOV

633



  int32_t numOfVgs = 0;




×








UNCOV

634



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &numOfVgs));





×







635













UNCOV

636



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





×







637



    TAOS_CHECK_EXIT(terrno);





×







638



  }










639













UNCOV

640



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





×








UNCOV

641



    int32_t vgId = 0;




×








UNCOV

642



    TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &vgId));





×








UNCOV

643



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





×







644



      TAOS_CHECK_EXIT(terrno);





×







645



    }










646



  }










647













UNCOV

648



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





×








UNCOV

649



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





×








UNCOV

650



  tEndDecode(pDecoder);




×







651













UNCOV

652



_exit:




×








UNCOV

653



  return code;




×







654



}










655













UNCOV

656



void tCleanupStreamHbMsg(SStreamHbMsg* pMsg) {




×








UNCOV

657



  if (pMsg == NULL) {





×







658



    return;




×







659



  }










660













UNCOV

661



  if (pMsg->pUpdateNodes != NULL) {





×








UNCOV

662



    taosArrayDestroy(pMsg->pUpdateNodes);




×








UNCOV

663



    pMsg->pUpdateNodes = NULL;




×







664



  }










665













UNCOV

666



  if (pMsg->pTaskStatus != NULL) {





×








UNCOV

667



    taosArrayDestroy(pMsg->pTaskStatus);




×








UNCOV

668



    pMsg->pTaskStatus = NULL;




×







669



  }










670













UNCOV

671



  pMsg->msgId = -1;




×








UNCOV

672



  pMsg->vgId = -1;




×








UNCOV

673



  pMsg->numOfTasks = -1;




×







674



}










675













UNCOV

676



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




×








UNCOV

677



  int32_t code = 0;




×







678



  int32_t lino;










679













UNCOV

680



  TAOS_CHECK_EXIT(tStartEncode(pEncoder));





×








UNCOV

681



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





×








UNCOV

682



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





×








UNCOV

683



  tEndEncode(pEncoder);




×







684













UNCOV

685



_exit:




×








UNCOV

686



  return code;




×







687



}










688













UNCOV

689



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




×








UNCOV

690



  int32_t code = 0;




×







691



  int32_t lino;










692













UNCOV

693



  TAOS_CHECK_EXIT(tStartDecode(pDecoder));





×








UNCOV

694



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





×








UNCOV

695



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





×








UNCOV

696



  tEndDecode(pDecoder);




×







697













UNCOV

698



_exit:




×








UNCOV

699



  return code;




×







700



}










701













UNCOV

702



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




×








UNCOV

703



  int32_t code = 0;




×







704



  int32_t lino;










705













UNCOV

706



  TAOS_CHECK_EXIT(tStartEncode(pEncoder));





×








UNCOV

707



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





×








UNCOV

708



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





×








UNCOV

709



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





×








UNCOV

710



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





×








UNCOV

711



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





×








UNCOV

712



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





×








UNCOV

713



  tEndEncode(pEncoder);




×







714













UNCOV

715



_exit:




×








UNCOV

716



  return code;




×







717



}










718













UNCOV

719



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




×








UNCOV

720



  int32_t code = 0;




×







721



  int32_t lino;










722













UNCOV

723



  TAOS_CHECK_EXIT(tStartDecode(pDecoder));





×








UNCOV

724



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





×








UNCOV

725



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





×








UNCOV

726



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





×








UNCOV

727



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





×








UNCOV

728



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





×








UNCOV

729



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





×








UNCOV

730



  tEndDecode(pDecoder);




×







731













UNCOV

732



_exit:




×








UNCOV

733



  return code;




×







734



}










735













UNCOV

736



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




×








UNCOV

737



  int32_t code = 0;




×







738



  int32_t lino;










739













UNCOV

740



  TAOS_CHECK_EXIT(tStartEncode(pEncoder));





×








UNCOV

741



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





×








UNCOV

742



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





×








UNCOV

743



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





×








UNCOV

744



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





×








UNCOV

745



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





×








UNCOV

746



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





×








UNCOV

747



  tEndEncode(pEncoder);




×







748













UNCOV

749



_exit:




×








UNCOV

750



  return code;




×







751



}










752













UNCOV

753



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




×








UNCOV

754



  int32_t code = 0;




×







755



  int32_t lino;










756













UNCOV

757



  TAOS_CHECK_EXIT(tStartDecode(pDecoder));





×








UNCOV

758



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





×








UNCOV

759



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





×








UNCOV

760



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





×








UNCOV

761



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





×








UNCOV

762



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





×








UNCOV

763



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





×








UNCOV

764



  tEndDecode(pDecoder);




×







765













UNCOV

766



_exit:




×








UNCOV

767



  return code;




×







768



}










769













UNCOV

770



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




×








UNCOV

771



  int32_t code = 0;




×







772



  int32_t lino;










773













UNCOV

774



  TAOS_CHECK_EXIT(tStartEncode(pEncoder));





×








UNCOV

775



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





×








UNCOV

776



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





×








UNCOV

777



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





×








UNCOV

778



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





×








UNCOV

779



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





×








UNCOV

780



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





×








UNCOV

781



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





×








UNCOV

782



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





×








UNCOV

783



  tEndEncode(pEncoder);




×







784













UNCOV

785



_exit:




×








UNCOV

786



  return code;




×







787



}










788













UNCOV

789



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




×








UNCOV

790



  int32_t code = 0;




×







791



  int32_t lino;










792













UNCOV

793



  TAOS_CHECK_EXIT(tStartDecode(pDecoder));





×








UNCOV

794



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





×








UNCOV

795



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





×








UNCOV

796



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





×








UNCOV

797



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





×








UNCOV

798



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





×








UNCOV

799



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





×








UNCOV

800



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





×








UNCOV

801



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





×








UNCOV

802



  tEndDecode(pDecoder);




×







803













UNCOV

804



_exit:




×








UNCOV

805



  return code;




×







806



}










807













UNCOV

808



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




×








UNCOV

809



  int32_t code = 0;




×







810



  int32_t lino;










811













UNCOV

812



  TAOS_CHECK_EXIT(tStartEncode(pEncoder));





×








UNCOV

813



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





×








UNCOV

814



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





×








UNCOV

815



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





×








UNCOV

816



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





×








UNCOV

817



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





×








UNCOV

818



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





×








UNCOV

819



  tEndEncode(pEncoder);




×







820













UNCOV

821



_exit:




×








UNCOV

822



  return code;




×







823



}










824













UNCOV

825



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




×








UNCOV

826



  int32_t code = 0;




×







827



  int32_t lino;










828













UNCOV

829



  TAOS_CHECK_EXIT(tStartDecode(pDecoder));





×








UNCOV

830



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





×








UNCOV

831



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





×








UNCOV

832



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





×








UNCOV

833



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





×








UNCOV

834



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





×








UNCOV

835



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





×








UNCOV

836



  tEndDecode(pDecoder);




×







837













UNCOV

838



_exit:




×








UNCOV

839



  return code;




×







840



}










841













UNCOV

842



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




×








UNCOV

843



  int32_t code = 0;




×







844



  int32_t lino;










845













UNCOV

846



  TAOS_CHECK_EXIT(tStartEncode(pEncoder));





×








UNCOV

847



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





×








UNCOV

848



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





×








UNCOV

849



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





×








UNCOV

850



  tEndEncode(pEncoder);




×







851













UNCOV

852



_exit:




×








UNCOV

853



  return code;




×







854



}










855













UNCOV

856



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




×








UNCOV

857



  int32_t code = 0;




×







858



  int32_t lino;










859













UNCOV

860



  TAOS_CHECK_EXIT(tStartDecode(pDecoder));





×








UNCOV

861



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





×








UNCOV

862



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





×








UNCOV

863



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





×








UNCOV

864



  tEndDecode(pDecoder);




×







865













UNCOV

866



_exit:




×








UNCOV

867



  return code;




×







868



}
























    

  • 

    •