taosdata / TDengine / #3798

coverage: 9.424% (-20.9%) from 30.372%
#3798

push

travis-ci

happyguoxy 
test:add test cases

21549 of 307601 branches covered (7.01%)

Branch coverage included in aggregate %.

36084 of 303967 relevant lines covered (11.87%)

58620.7 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












62



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




×







63



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





×







64



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





×







65



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





×







66



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





×







67



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





×







68



  return 0;




×







69



}










70












71



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




×







72



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





×







73



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





×







74



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





×







75



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





×







76



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





×







77



  return 0;




×







78



}










79












80



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




×







81



  TAOS_CHECK_RETURN(tStartEncode(pEncoder));





×







82



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





×







83



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





×







84



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





×







85



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





×







86



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





×







87



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





×







88



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





×







89



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





×







90



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





×







91



  tEndEncode(pEncoder);




×







92



  return pEncoder->pos;




×







93



}










94












95



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




×







96



  TAOS_CHECK_RETURN(tStartDecode(pDecoder));





×







97



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





×







98



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





×







99



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





×







100



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





×







101



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





×







102



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





×







103



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





×







104



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





×







105



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





×







106



  tEndDecode(pDecoder);




×







107



  return 0;




×







108



}










109












110



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




×







111



  TAOS_CHECK_RETURN(tStartEncode(pEncoder));





×







112



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





×







113



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





×







114



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





×







115



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





×







116



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





×







117



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





×







118



  tEndEncode(pEncoder);




×







119



  return pEncoder->pos;




×







120



}










121












122



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




×







123



  int32_t code = 0;




×







124



  int32_t lino;










125












126



  TAOS_CHECK_EXIT(tStartEncode(pEncoder));





×







127



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





×







128



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





×







129












130



  int32_t size = taosArrayGetSize(pMsg->pNodeList);





×







131



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, size));





×







132












133



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





×







134



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




×







135



    if (pInfo == NULL) {





×







136



      TAOS_CHECK_EXIT(terrno);





×







137



    }










138












139



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





×







140



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





×







141



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





×







142



  }










143












144



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










145



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





×







146












147



  int32_t numOfTasks = taosArrayGetSize(pMsg->pTaskList);





×







148



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, numOfTasks));





×







149












150



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





×







151



    int32_t* pId = taosArrayGet(pMsg->pTaskList, i);




×







152



    if (pId == NULL) {





×







153



      TAOS_CHECK_EXIT(terrno);





×







154



    }










155



    TAOS_CHECK_EXIT(tEncodeI32(pEncoder, *(int32_t*)pId));





×







156



  }










157












158



  tEndEncode(pEncoder);




×







159



_exit:




×







160



  if (code) {





×







161



    return code;




×







162



  } else {










163



    return pEncoder->pos;




×







164



  }










165



}










166












167



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




×







168



  int32_t code = 0;




×







169



  int32_t lino;










170












171



  TAOS_CHECK_EXIT(tStartDecode(pDecoder));





×







172



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





×







173



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





×







174












175



  int32_t size = 0;




×







176



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &size));





×







177












178



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




×







179



  TSDB_CHECK_NULL(pMsg->pNodeList, code, lino, _exit, terrno);





×







180












181



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





×







182



    SNodeUpdateInfo info = {0};




×







183



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





×







184



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





×







185



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





×







186












187



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





×







188



      TAOS_CHECK_EXIT(terrno);





×







189



    }










190



  }










191












192



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





×







193












194



  // number of tasks










195



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &size));





×







196



  pMsg->pTaskList = taosArrayInit(size, sizeof(int32_t));




×







197



  if (pMsg->pTaskList == NULL) {





×







198



    TAOS_CHECK_EXIT(terrno);





×







199



  }










200












201



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





×







202



    int32_t id = 0;




×







203



    TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &id));





×







204



    if (taosArrayPush(pMsg->pTaskList, &id) == NULL) {





×







205



      TAOS_CHECK_EXIT(terrno);





×







206



    }










207



  }










208












209



  tEndDecode(pDecoder);




×







210



_exit:




×







211



  return code;




×







212



}










213












214



void tDestroyNodeUpdateMsg(SStreamTaskNodeUpdateMsg* pMsg) {




×







215



  taosArrayDestroy(pMsg->pNodeList);




×







216



  taosArrayDestroy(pMsg->pTaskList);




×







217



  pMsg->pNodeList = NULL;




×







218



  pMsg->pTaskList = NULL;




×







219



}




×







220












221



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




×







222



  int32_t code = 0;




×







223



  int32_t lino;










224












225



  TAOS_CHECK_EXIT(tStartEncode(pEncoder));





×







226



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





×







227



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





×







228



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





×







229



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





×







230



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





×







231



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





×







232



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





×







233



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





×







234



  tEndEncode(pEncoder);




×







235












236



_exit:




×







237



  if (code) {





×







238



    return code;




×







239



  } else {










240



    return pEncoder->pos;




×







241



  }










242



}










243












244



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




×







245



  int32_t code = 0;




×







246



  int32_t lino;










247












248



  TAOS_CHECK_EXIT(tStartDecode(pDecoder));





×







249



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





×







250



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





×







251



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





×







252



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





×







253



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





×







254



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





×







255



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





×







256



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





×







257



  tEndDecode(pDecoder);




×







258












259



_exit:




×







260



  return code;




×







261



}










262












263



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




×







264



  int32_t code = 0;




×







265



  int32_t lino;










266












267



  TAOS_CHECK_EXIT(tStartEncode(pEncoder));





×







268



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





×







269



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





×







270



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





×







271



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





×







272



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





×







273



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





×







274



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





×







275



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





×







276



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





×







277



  tEndEncode(pEncoder);




×







278












279



_exit:




×







280



  if (code) {





×







281



    return code;




×







282



  } else {










283



    return pEncoder->pos;




×







284



  }










285



}










286












287



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




×







288



  int32_t code = 0;




×







289



  int32_t lino;










290












291



  TAOS_CHECK_EXIT(tStartDecode(pDecoder));





×







292



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





×







293



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





×







294



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





×







295



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





×







296



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





×







297



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





×







298



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





×







299



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





×







300



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





×







301



  tEndDecode(pDecoder);




×







302












303



_exit:




×







304



  return code;




×







305



}










306












307



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




×







308



  int32_t code = 0;




×







309



  int32_t lino;










310












311



  TAOS_CHECK_EXIT(tStartEncode(pEncoder));





×







312



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





×







313



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





×







314



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





×







315



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





×







316



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





×







317



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





×







318



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





×







319



  tEndEncode(pEncoder);




×







320












321



_exit:




×







322



  if (code) {





×







323



    return code;




×







324



  } else {










325



    return pEncoder->pos;




×







326



  }










327



}










328












329



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




×







330



  int32_t code = 0;




×







331



  int32_t lino;










332












333



  TAOS_CHECK_EXIT(tStartDecode(pDecoder));





×







334



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





×







335



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





×







336



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





×







337



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





×







338



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





×







339



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





×







340



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





×







341



  tEndDecode(pDecoder);




×







342












343



_exit:




×







344



  return code;




×







345



}










346












347



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




×







348



  int32_t code = 0;




×







349



  int32_t lino;










350












351



  TAOS_CHECK_EXIT(tStartEncode(pEncoder));





×







352



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





×







353



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





×







354



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





×







355



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





×







356



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





×







357



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





×







358



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





×







359



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





×







360



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





×







361



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





×







362



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





×







363



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





×







364



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





×







365












366



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





×







367



    uError("invalid dispatch req msg");





×







368



    TAOS_CHECK_EXIT(TSDB_CODE_INVALID_MSG);





×







369



  }










370












371



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





×







372



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




×







373



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




×







374



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





×







375



      TAOS_CHECK_EXIT(terrno);





×







376



    }










377












378



    TAOS_CHECK_EXIT(tEncodeI32(pEncoder, *pLen));





×







379



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





×







380



  }










381



  tEndEncode(pEncoder);




×







382



_exit:




×







383



  if (code) {





×







384



    return code;




×







385



  } else {










386



    return pEncoder->pos;




×







387



  }










388



}










389












390



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




×







391



  int32_t code = 0;




×







392



  int32_t lino;










393












394



  TAOS_CHECK_EXIT(tStartDecode(pDecoder));





×







395



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





×







396



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





×







397



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





×







398



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





×







399



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





×







400



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





×







401



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





×







402



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





×







403



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





×







404



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





×







405



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





×







406



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





×







407



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





×







408












409



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





×







410



    TAOS_CHECK_EXIT(terrno);





×







411



  }










412



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





×







413



    TAOS_CHECK_EXIT(terrno);





×







414



  }










415



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





×







416



    int32_t  len1;










417



    uint64_t len2;










418



    void*    data;










419



    TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &len1));





×







420



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





×







421












422



    if (len1 != len2) {





×







423



      TAOS_CHECK_EXIT(TSDB_CODE_INVALID_MSG);





×







424



    }










425












426



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





×







427



      TAOS_CHECK_EXIT(terrno);





×







428



    }










429












430



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





×







431



      TAOS_CHECK_EXIT(terrno);





×







432



    }










433



  }










434












435



  tEndDecode(pDecoder);




×







436



_exit:




×







437



  return code;




×







438



}










439












440



void tCleanupStreamDispatchReq(SStreamDispatchReq* pReq) {




×







441



  taosArrayDestroyP(pReq->data, NULL);




×







442



  taosArrayDestroy(pReq->dataLen);




×







443



}




×







444












445



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




×







446



  int32_t code = 0;




×







447



  int32_t lino;










448












449



  TAOS_CHECK_EXIT(tStartEncode(pEncoder));





×







450



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





×







451



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





×







452



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





×







453



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





×







454



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





×







455



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





×







456



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





×







457



  tEndEncode(pEncoder);




×







458












459



_exit:




×







460



  if (code) {





×







461



    return code;




×







462



  } else {










463



    return pEncoder->pos;




×







464



  }










465



}










466












467



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




×







468



  int32_t code = 0;




×







469



  int32_t lino;










470












471



  TAOS_CHECK_EXIT(tStartDecode(pDecoder));





×







472



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





×







473



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





×







474



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





×







475



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





×







476



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





×







477



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





×







478



  uint64_t len = 0;




×







479



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





×







480



  pReq->retrieveLen = (int32_t)len;




×







481



  tEndDecode(pDecoder);




×







482












483



_exit:




×







484



  return code;




×







485



}










486












487



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





×







488












489



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




×







490



  int32_t code = 0;




×







491



  int32_t lino;










492












493



  TAOS_CHECK_EXIT(tStartEncode(pEncoder));





×







494



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





×







495



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





×







496



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





×







497



  tEndEncode(pEncoder);




×







498












499



_exit:




×







500



  return code;




×







501



}










502












503



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




×







504



  int32_t code = 0;




×







505



  int32_t lino;










506












507



  TAOS_CHECK_EXIT(tStartDecode(pDecoder));





×







508



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





×







509



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





×







510



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





×







511



  tEndDecode(pDecoder);




×







512












513



_exit:




×







514



  return code;




×







515



}










516












517



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




×







518



  int32_t code = 0;




×







519



  int32_t lino;










520












521



  TAOS_CHECK_EXIT(tStartEncode(pEncoder));





×







522



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





×







523



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





×







524












525



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





×







526



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




×







527



    if (ps == NULL) {





×







528



      TAOS_CHECK_EXIT(terrno);





×







529



    }










530












531



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





×







532



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





×







533



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





×







534



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





×







535



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





×







536



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





×







537



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





×







538



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





×







539



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





×







540



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





×







541



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





×







542



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





×







543



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





×







544



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





×







545



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





×







546



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





×







547



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





×







548



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





×







549



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





×







550



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





×







551



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





×







552



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





×







553



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





×







554



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





×







555



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





×







556



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





×







557



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





×







558



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





×







559



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





×







560



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





×







561



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





×







562



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





×







563



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





×







564



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





×







565



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





×







566



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





×







567



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





×







568



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





×







569



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





×







570



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





×







571



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





×







572



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





×







573



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





×







574



  }










575












576



  int32_t numOfVgs = taosArrayGetSize(pReq->pUpdateNodes);





×







577



  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, numOfVgs));





×







578












579



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





×







580



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




×







581



    if (pVgId == NULL) {





×







582



      TAOS_CHECK_EXIT(terrno);





×







583



    }










584












585



    TAOS_CHECK_EXIT(tEncodeI32(pEncoder, *pVgId));





×







586



  }










587












588



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





×







589



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





×







590



  tEndEncode(pEncoder);




×







591












592



_exit:




×







593



  if (code) {





×







594



    return code;




×







595



  } else {










596



    return pEncoder->pos;




×







597



  }










598



}










599












600



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




×







601



  int32_t code = 0;




×







602



  int32_t lino;










603












604



  TAOS_CHECK_EXIT(tStartDecode(pDecoder));





×







605



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





×







606



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





×







607












608



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





×







609



    TAOS_CHECK_EXIT(terrno);





×







610



  }










611



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





×







612



    int32_t          taskId = 0;




×







613



    STaskStatusEntry entry = {0};




×







614












615



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





×







616



    TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &taskId));





×







617



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





×







618



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





×







619



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





×







620



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





×







621



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





×







622



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





×







623



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





×







624



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





×







625



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





×







626



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





×







627



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





×







628



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





×







629



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





×







630



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





×







631



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





×







632



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





×







633



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





×







634












635



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





×







636



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





×







637



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





×







638



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





×







639



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





×







640



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





×







641



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





×







642



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





×







643



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





×







644



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





×







645



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





×







646












647



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





×







648



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





×







649



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





×







650



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





×







651



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





×







652



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





×







653



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





×







654



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





×







655



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





×







656



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





×







657



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





×







658



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





×







659



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





×







660












661



    entry.id.taskId = taskId;




×







662



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





×







663



      TAOS_CHECK_EXIT(terrno);





×







664



    }










665



  }










666












667



  int32_t numOfVgs = 0;




×







668



  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &numOfVgs));





×







669












670



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





×







671



    TAOS_CHECK_EXIT(terrno);





×







672



  }










673












674



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





×







675



    int32_t vgId = 0;




×







676



    TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &vgId));





×







677



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





×







678



      TAOS_CHECK_EXIT(terrno);





×







679



    }










680



  }










681












682



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





×







683



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





×







684



  tEndDecode(pDecoder);




×







685












686



_exit:




×







687



  return code;




×







688



}










689












690



void tCleanupStreamHbMsg(SStreamHbMsg* pMsg) {




×







691



  if (pMsg == NULL) {





×







692



    return;




×







693



  }










694












695



  if (pMsg->pUpdateNodes != NULL) {





×







696



    taosArrayDestroy(pMsg->pUpdateNodes);




×







697



    pMsg->pUpdateNodes = NULL;




×







698



  }










699












700



  if (pMsg->pTaskStatus != NULL) {





×







701



    taosArrayDestroy(pMsg->pTaskStatus);




×







702



    pMsg->pTaskStatus = NULL;




×







703



  }










704












705



  pMsg->msgId = -1;




×







706



  pMsg->vgId = -1;




×







707



  pMsg->numOfTasks = -1;




×







708



}










709












710



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




×







711



  int32_t code = 0;




×







712



  int32_t lino;










713












714



  TAOS_CHECK_EXIT(tStartEncode(pEncoder));





×







715



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





×







716



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





×







717



  tEndEncode(pEncoder);




×







718












719



_exit:




×







720



  return code;




×







721



}










722












723



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




×







724



  int32_t code = 0;




×







725



  int32_t lino;










726












727



  TAOS_CHECK_EXIT(tStartDecode(pDecoder));





×







728



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





×







729



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





×







730



  tEndDecode(pDecoder);




×







731












732



_exit:




×







733



  return code;




×







734



}










735












736



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




×







737



  int32_t code = 0;




×







738



  int32_t lino;










739












740



  TAOS_CHECK_EXIT(tStartEncode(pEncoder));





×







741



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





×







742



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





×







743



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





×







744



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





×







745



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





×







746



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





×







747



  tEndEncode(pEncoder);




×







748












749



_exit:




×







750



  return code;




×







751



}










752












753



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




×







754



  int32_t code = 0;




×







755



  int32_t lino;










756












757



  TAOS_CHECK_EXIT(tStartDecode(pDecoder));





×







758



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





×







759



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





×







760



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





×







761



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





×







762



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





×







763



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





×







764



  tEndDecode(pDecoder);




×







765












766



_exit:




×







767



  return code;




×







768



}










769












770



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




×







771



  int32_t code = 0;




×







772



  int32_t lino;










773












774



  TAOS_CHECK_EXIT(tStartEncode(pEncoder));





×







775



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





×







776



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





×







777



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





×







778



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





×







779



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





×







780



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





×







781



  tEndEncode(pEncoder);




×







782












783



_exit:




×







784



  return code;




×







785



}










786












787



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




×







788



  int32_t code = 0;




×







789



  int32_t lino;










790












791



  TAOS_CHECK_EXIT(tStartDecode(pDecoder));





×







792



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





×







793



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





×







794



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





×







795



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





×







796



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





×







797



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





×







798



  tEndDecode(pDecoder);




×







799












800



_exit:




×







801



  return code;




×







802



}










803












804



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




×







805



  int32_t code = 0;




×







806



  int32_t lino;










807












808



  TAOS_CHECK_EXIT(tStartEncode(pEncoder));





×







809



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





×







810



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





×







811



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





×







812



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





×







813



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





×







814



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





×







815



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





×







816



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





×







817



  tEndEncode(pEncoder);




×







818












819



_exit:




×







820



  return code;




×







821



}










822












823



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




×







824



  int32_t code = 0;




×







825



  int32_t lino;










826












827



  TAOS_CHECK_EXIT(tStartDecode(pDecoder));





×







828



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





×







829



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





×







830



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





×







831



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





×







832



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





×







833



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





×







834



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





×







835



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





×







836



  tEndDecode(pDecoder);




×







837












838



_exit:




×







839



  return code;




×







840



}










841












842



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




×







843



  int32_t code = 0;




×







844



  int32_t lino;










845












846



  TAOS_CHECK_EXIT(tStartEncode(pEncoder));





×







847



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





×







848



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





×







849



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





×







850



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





×







851



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





×







852



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





×







853



  tEndEncode(pEncoder);




×







854












855



_exit:




×







856



  return code;




×







857



}










858












859



// todo: serialized term attributes.










860



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




×







861



  int32_t code = 0;




×







862



  int32_t lino;










863












864



  TAOS_CHECK_EXIT(tStartDecode(pDecoder));





×







865



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





×







866



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





×







867



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





×







868



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





×







869



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





×







870



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





×







871



  tEndDecode(pDecoder);




×







872












873



_exit:




×







874



  return code;




×







875



}










876












877



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




×







878



  int32_t code = 0;




×







879



  int32_t lino;










880












881



  TAOS_CHECK_EXIT(tStartEncode(pEncoder));





×







882



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





×







883



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





×







884



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





×







885



  tEndEncode(pEncoder);




×







886












887



_exit:




×







888



  return code;




×







889



}










890












891



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




×







892



  int32_t code = 0;




×







893



  int32_t lino;










894












895



  TAOS_CHECK_EXIT(tStartDecode(pDecoder));





×







896



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





×







897



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





×







898



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





×







899



  tEndDecode(pDecoder);




×







900












901



_exit:




×







902



  return code;




×







903



}










904












905



int32_t tEncodeStreamTaskStopReq(SEncoder* pEncoder, const SStreamTaskStopReq* pReq) {




×







906



  int32_t code = 0;




×







907



  int32_t lino;










908












909



  TAOS_CHECK_EXIT(tStartEncode(pEncoder));





×







910



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





×







911



  tEndEncode(pEncoder);




×







912












913



_exit:




×







914



  return code;




×







915



}










916












917



int32_t tDecodeStreamTaskStopReq(SDecoder* pDecoder, SStreamTaskStopReq* pReq) {




×







918



  int32_t code = 0;




×







919



  int32_t lino;










920












921



  TAOS_CHECK_EXIT(tStartDecode(pDecoder));





×







922



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





×







923



  tEndDecode(pDecoder);




×







924












925



_exit:




×







926



  return code;




×







927












928



}
























    

  • 

    •