taosdata / TDengine / #3562

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

push

travis-ci

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

enh: seperate tsdb async tasks to different thread pools

21498 of 109421 branches covered (19.65%)

Branch coverage included in aggregate %.

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

39441 existing lines in 157 files now uncovered.

35007 of 102566 relevant lines covered (34.13%)

53922.97 hits per line

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

0.52
/source/dnode/vnode/src/tsdb/tsdbSnapInfo.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 "tsdb.h"










17



#include "tsdbFS2.h"










18












19



#define TSDB_SNAP_MSG_VER 1










20












21



// fset partition











UNCOV

22



static int32_t tsdbFSetPartCmprFn(STsdbFSetPartition* x, STsdbFSetPartition* y) {




×








UNCOV

23



  if (x->fid < y->fid) return -1;





×








UNCOV

24



  if (x->fid > y->fid) return 1;





×







25



  return 0;




×







26



}










27













UNCOV

28



static int32_t tVersionRangeCmprFn(SVersionRange* x, SVersionRange* y) {




×








UNCOV

29



  if (x->minVer < y->minVer) return -1;





×








UNCOV

30



  if (x->minVer > y->minVer) return 1;





×







31



  if (x->maxVer < y->maxVer) return -1;





×







32



  if (x->maxVer > y->maxVer) return 1;





×







33



  return 0;




×







34



}










35












36



static int32_t tsdbTFileSetRangeCmprFn(STFileSetRange* x, STFileSetRange* y) {




×







37



  if (x->fid < y->fid) return -1;





×







38



  if (x->fid > y->fid) return 1;





×







39



  return 0;




×







40



}










41













UNCOV

42



STsdbFSetPartition* tsdbFSetPartitionCreate() {




×








UNCOV

43



  STsdbFSetPartition* pSP = taosMemoryCalloc(1, sizeof(STsdbFSetPartition));





×








UNCOV

44



  if (pSP == NULL) {





×







45



    return NULL;




×







46



  }











UNCOV

47



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





×








UNCOV

48



    TARRAY2_INIT(&pSP->verRanges[i]);




×







49



  }











UNCOV

50



  return pSP;




×







51



}










52













UNCOV

53



void tsdbFSetPartitionClear(STsdbFSetPartition** ppSP) {




×








UNCOV

54



  if (ppSP == NULL || ppSP[0] == NULL) {





×







55



    return;




×







56



  }











UNCOV

57



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





×








UNCOV

58



    TARRAY2_DESTROY(&ppSP[0]->verRanges[i], NULL);





×







59



  }











UNCOV

60



  taosMemoryFree(ppSP[0]);





×








UNCOV

61



  ppSP[0] = NULL;




×







62



}










63













UNCOV

64



static int32_t tsdbFTypeToFRangeType(tsdb_ftype_t ftype) {




×








UNCOV

65



  switch (ftype) {





×








UNCOV

66



    case TSDB_FTYPE_HEAD:




×








UNCOV

67



      return TSDB_FSET_RANGE_TYP_HEAD;




×








UNCOV

68



    case TSDB_FTYPE_DATA:




×








UNCOV

69



      return TSDB_FSET_RANGE_TYP_DATA;




×








UNCOV

70



    case TSDB_FTYPE_SMA:




×








UNCOV

71



      return TSDB_FSET_RANGE_TYP_SMA;




×








UNCOV

72



    case TSDB_FTYPE_TOMB:




×








UNCOV

73



      return TSDB_FSET_RANGE_TYP_TOMB;




×







74



    case TSDB_FTYPE_STT:




×







75



      return TSDB_FSET_RANGE_TYP_STT;




×







76



  }










77



  return TSDB_FSET_RANGE_TYP_MAX;




×







78



}










79













UNCOV

80



static int32_t tsdbTFileSetToFSetPartition(STFileSet* fset, STsdbFSetPartition** ppSP) {




×








UNCOV

81



  STsdbFSetPartition* p = tsdbFSetPartitionCreate();




×








UNCOV

82



  if (p == NULL) {





×







83



    return terrno;




×







84



  }










85













UNCOV

86



  p->fid = fset->fid;




×







87













UNCOV

88



  int32_t code = 0;




×








UNCOV

89



  int32_t typ = 0;




×








UNCOV

90



  int32_t corrupt = false;




×








UNCOV

91



  int32_t count = 0;




×








UNCOV

92



  for (int32_t ftype = TSDB_FTYPE_MIN; ftype < TSDB_FTYPE_MAX; ++ftype) {





×








UNCOV

93



    if (fset->farr[ftype] == NULL) continue;





×








UNCOV

94



    typ = tsdbFTypeToFRangeType(ftype);




×








UNCOV

95



    STFile* f = fset->farr[ftype]->f;




×








UNCOV

96



    if (f->maxVer > fset->maxVerValid) {





×







97



      corrupt = true;




×







98



      tsdbError("skip incomplete data file: fid:%d, maxVerValid:%" PRId64 ", minVer:%" PRId64 ", maxVer:%" PRId64





×







99



                ", ftype: %d",










100



                fset->fid, fset->maxVerValid, f->minVer, f->maxVer, ftype);










101



      continue;




×







102



    }











UNCOV

103



    count++;




×








UNCOV

104



    SVersionRange vr = {.minVer = f->minVer, .maxVer = f->maxVer};




×








UNCOV

105



    code = TARRAY2_SORT_INSERT(&p->verRanges[typ], vr, tVersionRangeCmprFn);




×








UNCOV

106



    if (code) {





×







107



      tsdbFSetPartitionClear(&p);




×







108



      return code;




×







109



    }










110



  }










111













UNCOV

112



  typ = TSDB_FSET_RANGE_TYP_STT;




×







113



  const SSttLvl* lvl;











UNCOV

114



  TARRAY2_FOREACH(fset->lvlArr, lvl) {





×







115



    STFileObj* fobj;











UNCOV

116



    TARRAY2_FOREACH(lvl->fobjArr, fobj) {





×








UNCOV

117



      STFile* f = fobj->f;




×








UNCOV

118



      if (f->maxVer > fset->maxVerValid) {





×







119



        corrupt = true;




×







120



        tsdbError("skip incomplete stt file.fid:%d, maxVerValid:%" PRId64 ", minVer:%" PRId64 ", maxVer:%" PRId64





×







121



                  ", ftype: %d",










122



                  fset->fid, fset->maxVerValid, f->minVer, f->maxVer, typ);










123



        continue;




×







124



      }











UNCOV

125



      count++;




×








UNCOV

126



      SVersionRange vr = {.minVer = f->minVer, .maxVer = f->maxVer};




×








UNCOV

127



      code = TARRAY2_SORT_INSERT(&p->verRanges[typ], vr, tVersionRangeCmprFn);




×








UNCOV

128



      if (code) {





×







129



        tsdbFSetPartitionClear(&p);




×







130



        return code;




×







131



      }










132



    }










133



  }











UNCOV

134



  if (corrupt && count == 0) {





×







135



    SVersionRange vr = {.minVer = VERSION_MIN, .maxVer = fset->maxVerValid};




×







136



    code = TARRAY2_SORT_INSERT(&p->verRanges[typ], vr, tVersionRangeCmprFn);




×







137



    if (code) {





×







138



      tsdbFSetPartitionClear(&p);




×







139



      return code;




×







140



    }










141



  }











UNCOV

142



  ppSP[0] = p;




×








UNCOV

143



  return 0;




×







144



}










145












146



// fset partition list











UNCOV

147



STsdbFSetPartList* tsdbFSetPartListCreate() {




×








UNCOV

148



  STsdbFSetPartList* pList = taosMemoryCalloc(1, sizeof(STsdbFSetPartList));





×








UNCOV

149



  if (pList == NULL) {





×







150



    return NULL;




×







151



  }











UNCOV

152



  TARRAY2_INIT(pList);




×








UNCOV

153



  return pList;




×







154



}










155













UNCOV

156



void tsdbFSetPartListDestroy(STsdbFSetPartList** ppList) {




×








UNCOV

157



  if (ppList == NULL || ppList[0] == NULL) return;





×







158













UNCOV

159



  TARRAY2_DESTROY(ppList[0], tsdbFSetPartitionClear);





×








UNCOV

160



  taosMemoryFree(ppList[0]);





×








UNCOV

161



  ppList[0] = NULL;




×







162



}










163













UNCOV

164



int32_t tsdbFSetPartListToRangeDiff(STsdbFSetPartList* pList, TFileSetRangeArray** ppRanges) {




×








UNCOV

165



  int32_t code = 0;




×







166













UNCOV

167



  TFileSetRangeArray* pDiff = taosMemoryCalloc(1, sizeof(TFileSetRangeArray));





×








UNCOV

168



  if (pDiff == NULL) {





×







169



    code = terrno;




×







170



    goto _err;




×







171



  }











UNCOV

172



  TARRAY2_INIT(pDiff);




×







173












174



  STsdbFSetPartition* part;











UNCOV

175



  TARRAY2_FOREACH(pList, part) {





×







176



    STFileSetRange* r = taosMemoryCalloc(1, sizeof(STFileSetRange));





×







177



    if (r == NULL) {





×







178



      code = terrno;




×







179



      goto _err;




×







180



    }










181



    int64_t maxVerValid = -1;




×







182



    int32_t typMax = TSDB_FSET_RANGE_TYP_MAX;




×







183



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





×







184



      SVerRangeList* iList = &part->verRanges[i];




×







185



      SVersionRange  vr = {0};




×







186



      TARRAY2_FOREACH(iList, vr) {





×







187



        if (vr.maxVer < vr.minVer) {





×







188



          continue;




×







189



        }










190



        maxVerValid = TMAX(maxVerValid, vr.maxVer);




×







191



      }










192



    }










193



    r->fid = part->fid;




×







194



    r->sver = maxVerValid + 1;




×







195



    r->ever = VERSION_MAX;




×







196



    tsdbDebug("range diff fid:%" PRId64 ", sver:%" PRId64 ", ever:%" PRId64, part->fid, r->sver, r->ever);





×







197



    code = TARRAY2_SORT_INSERT(pDiff, r, tsdbTFileSetRangeCmprFn);




×







198



    if (code) {





×







199



      taosMemoryFree(r);





×







200



      goto _err;




×







201



    }










202



  }











UNCOV

203



  ppRanges[0] = pDiff;




×







204













UNCOV

205



  tsdbInfo("pDiff size:%d", TARRAY2_SIZE(pDiff));





×








UNCOV

206



  return 0;




×







207












208



_err:




×







209



  if (pDiff) {





×







210



    tsdbTFileSetRangeArrayDestroy(&pDiff);




×







211



  }










212



  return code;




×







213



}










214












215



// serialization











UNCOV

216



int32_t tTsdbFSetPartListDataLenCalc(STsdbFSetPartList* pList) {




×








UNCOV

217



  int32_t hdrLen = sizeof(int32_t);




×








UNCOV

218



  int32_t datLen = 0;




×







219













UNCOV

220



  int8_t  msgVer = 1;




×








UNCOV

221



  int32_t len = TARRAY2_SIZE(pList);




×








UNCOV

222



  hdrLen += sizeof(msgVer);




×








UNCOV

223



  hdrLen += sizeof(len);




×








UNCOV

224



  datLen += hdrLen;




×







225













UNCOV

226



  for (int32_t u = 0; u < len; u++) {





×








UNCOV

227



    STsdbFSetPartition* p = TARRAY2_GET(pList, u);




×








UNCOV

228



    int32_t             typMax = TSDB_FSET_RANGE_TYP_MAX;




×








UNCOV

229



    int32_t             uItem = 0;




×








UNCOV

230



    uItem += sizeof(STsdbFSetPartition);




×








UNCOV

231



    uItem += sizeof(typMax);




×







232













UNCOV

233



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





×








UNCOV

234



      int32_t iLen = TARRAY2_SIZE(&p->verRanges[i]);




×








UNCOV

235



      int32_t jItem = 0;




×








UNCOV

236



      jItem += sizeof(SVersionRange);




×








UNCOV

237



      jItem += sizeof(int64_t);




×








UNCOV

238



      uItem += sizeof(iLen) + jItem * iLen;




×







239



    }











UNCOV

240



    datLen += uItem;




×







241



  }











UNCOV

242



  return datLen;




×







243



}










244













UNCOV

245



static int32_t tSerializeTsdbFSetPartList(void* buf, int32_t bufLen, STsdbFSetPartList* pList, int32_t* encodeSize) {




×








UNCOV

246



  SEncoder encoder = {0};




×








UNCOV

247



  int8_t   reserved8 = 0;




×








UNCOV

248



  int16_t  reserved16 = 0;




×








UNCOV

249



  int64_t  reserved64 = 0;




×








UNCOV

250



  int8_t   msgVer = TSDB_SNAP_MSG_VER;




×








UNCOV

251



  int32_t  len = TARRAY2_SIZE(pList);




×








UNCOV

252



  int32_t  code = 0;




×







253













UNCOV

254



  tEncoderInit(&encoder, buf, bufLen);




×








UNCOV

255



  if ((code = tStartEncode(&encoder))) goto _exit;





×








UNCOV

256



  if ((code = tEncodeI8(&encoder, msgVer))) goto _exit;





×








UNCOV

257



  if ((code = tEncodeI32(&encoder, len))) goto _exit;





×







258













UNCOV

259



  for (int32_t u = 0; u < len; u++) {





×








UNCOV

260



    STsdbFSetPartition* p = TARRAY2_GET(pList, u);




×








UNCOV

261



    if ((code = tEncodeI64(&encoder, p->fid))) goto _exit;





×








UNCOV

262



    if ((code = tEncodeI8(&encoder, p->stat))) goto _exit;





×








UNCOV

263



    if ((code = tEncodeI8(&encoder, reserved8))) goto _exit;





×








UNCOV

264



    if ((code = tEncodeI16(&encoder, reserved16))) goto _exit;





×







265













UNCOV

266



    int32_t typMax = TSDB_FSET_RANGE_TYP_MAX;





×








UNCOV

267



    if ((code = tEncodeI32(&encoder, typMax))) goto _exit;





×







268













UNCOV

269



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





×








UNCOV

270



      SVerRangeList* iList = &p->verRanges[i];




×








UNCOV

271



      int32_t        iLen = TARRAY2_SIZE(iList);





×







272













UNCOV

273



      if ((code = tEncodeI32(&encoder, iLen))) goto _exit;





×








UNCOV

274



      for (int32_t j = 0; j < iLen; j++) {





×








UNCOV

275



        SVersionRange r = TARRAY2_GET(iList, j);




×








UNCOV

276



        if ((code = tEncodeI64(&encoder, r.minVer))) goto _exit;





×








UNCOV

277



        if ((code = tEncodeI64(&encoder, r.maxVer))) goto _exit;





×








UNCOV

278



        if ((code = tEncodeI64(&encoder, reserved64))) goto _exit;





×







279



      }










280



    }










281



  }










282













UNCOV

283



  tEndEncode(&encoder);




×







284













UNCOV

285



  if (encodeSize) {





×








UNCOV

286



    encodeSize[0] = encoder.pos;




×







287



  }










288












289



_exit:




×








UNCOV

290



  tEncoderClear(&encoder);




×








UNCOV

291



  return code;




×







292



}










293













UNCOV

294



int32_t tDeserializeTsdbFSetPartList(void* buf, int32_t bufLen, STsdbFSetPartList* pList) {




×








UNCOV

295



  SDecoder decoder = {0};




×








UNCOV

296



  int8_t   reserved8 = 0;




×








UNCOV

297



  int16_t  reserved16 = 0;




×








UNCOV

298



  int64_t  reserved64 = 0;




×








UNCOV

299



  int32_t  code = 0;




×







300













UNCOV

301



  STsdbFSetPartition* p = NULL;




×







302













UNCOV

303



  tDecoderInit(&decoder, buf, bufLen);




×








UNCOV

304



  int8_t  msgVer = 0;




×








UNCOV

305



  int32_t len = 0;




×








UNCOV

306



  if ((code = tStartDecode(&decoder))) goto _err;





×








UNCOV

307



  if ((code = tDecodeI8(&decoder, &msgVer))) goto _err;





×








UNCOV

308



  if (msgVer != TSDB_SNAP_MSG_VER) {





×







309



    code = TSDB_CODE_INVALID_MSG;




×







310



    goto _err;




×







311



  }











UNCOV

312



  if ((code = tDecodeI32(&decoder, &len))) goto _err;





×







313













UNCOV

314



  for (int32_t u = 0; u < len; u++) {





×







315



    p = tsdbFSetPartitionCreate();




×







316



    if (p == NULL) {





×







317



      code = terrno;




×







318



      goto _err;




×







319



    }










320












321



    if ((code = tDecodeI64(&decoder, &p->fid))) goto _err;





×







322



    if ((code = tDecodeI8(&decoder, &p->stat))) goto _err;





×







323



    if ((code = tDecodeI8(&decoder, &reserved8))) goto _err;





×







324



    if ((code = tDecodeI16(&decoder, &reserved16))) goto _err;





×







325












326



    int32_t typMax = 0;




×







327



    if ((code = tDecodeI32(&decoder, &typMax))) goto _err;





×







328












329



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





×







330



      SVerRangeList* iList = &p->verRanges[i];




×







331



      int32_t        iLen = 0;




×







332



      if ((code = tDecodeI32(&decoder, &iLen))) goto _err;





×







333



      for (int32_t j = 0; j < iLen; j++) {





×







334



        SVersionRange r = {0};




×







335



        if ((code = tDecodeI64(&decoder, &r.minVer))) goto _err;





×







336



        if ((code = tDecodeI64(&decoder, &r.maxVer))) goto _err;





×







337



        if ((code = tDecodeI64(&decoder, &reserved64))) goto _err;





×







338



        if ((code = TARRAY2_APPEND(iList, r))) goto _err;





×







339



      }










340



    }










341



    if ((code = TARRAY2_APPEND(pList, p))) goto _err;





×







342



    p = NULL;




×







343



  }










344













UNCOV

345



  tEndDecode(&decoder);




×








UNCOV

346



  tDecoderClear(&decoder);




×








UNCOV

347



  return 0;




×







348












349



_err:




×







350



  if (p) {





×







351



    tsdbFSetPartitionClear(&p);




×







352



  }










353



  tDecoderClear(&decoder);




×







354



  return code;




×







355



}










356












357



// fs state











UNCOV

358



static STsdbFSetPartList* tsdbSnapGetFSetPartList(STFileSystem* fs) {




×








UNCOV

359



  STsdbFSetPartList* pList = tsdbFSetPartListCreate();




×








UNCOV

360



  if (pList == NULL) {





×







361



    return NULL;




×







362



  }










363













UNCOV

364



  int32_t code = 0;




×








UNCOV

365



  (void)taosThreadMutexLock(&fs->tsdb->mutex);




×







366



  STFileSet* fset;











UNCOV

367



  TARRAY2_FOREACH(fs->fSetArr, fset) {





×








UNCOV

368



    STsdbFSetPartition* pItem = NULL;




×








UNCOV

369



    code = tsdbTFileSetToFSetPartition(fset, &pItem);




×








UNCOV

370



    if (code) {





×







371



      terrno = code;




×







372



      break;




×







373



    }











UNCOV

374



    code = TARRAY2_SORT_INSERT(pList, pItem, tsdbFSetPartCmprFn);




×








UNCOV

375



    if (code) {





×







376



      terrno = code;




×







377



      break;




×







378



    }










379



  }











UNCOV

380



  (void)taosThreadMutexUnlock(&fs->tsdb->mutex);




×







381













UNCOV

382



  if (code) {





×







383



    TARRAY2_DESTROY(pList, tsdbFSetPartitionClear);





×







384



    taosMemoryFree(pList);





×







385



    pList = NULL;




×







386



  }











UNCOV

387



  return pList;




×







388



}










389












390



ETsdbFsState tsdbSnapGetFsState(SVnode* pVnode) {




18,316✔





391



  if (!VND_IS_RSMA(pVnode)) {





18,316!





392



    return pVnode->pTsdb->pFS->fsstate;




18,316✔





393



  }











UNCOV

394



  for (int32_t lvl = 0; lvl < TSDB_RETENTION_MAX; ++lvl) {





×








UNCOV

395



    STsdb* pTsdb = SMA_RSMA_GET_TSDB(pVnode, lvl);





×








UNCOV

396



    if (pTsdb && pTsdb->pFS->fsstate != TSDB_FS_STATE_NORMAL) {





×







397



      return TSDB_FS_STATE_INCOMPLETE;




×







398



    }










399



  }











UNCOV

400



  return TSDB_FS_STATE_NORMAL;




×







401



}










402












403



// description










404



typedef struct STsdbPartitionInfo {










405



  int32_t            vgId;










406



  int32_t            tsdbMaxCnt;










407



  int32_t            subTyps[TSDB_RETENTION_MAX];










408



  STsdbFSetPartList* pLists[TSDB_RETENTION_MAX];










409



} STsdbPartitionInfo;










410













UNCOV

411



static int32_t tsdbPartitionInfoInit(SVnode* pVnode, STsdbPartitionInfo* pInfo) {




×








UNCOV

412



  int32_t subTyps[TSDB_RETENTION_MAX] = {SNAP_DATA_TSDB, SNAP_DATA_RSMA1, SNAP_DATA_RSMA2};




×








UNCOV

413



  pInfo->vgId = TD_VID(pVnode);




×








UNCOV

414



  pInfo->tsdbMaxCnt = (!VND_IS_RSMA(pVnode) ? 1 : TSDB_RETENTION_MAX);





×







415












416



  if (!(sizeof(pInfo->subTyps) == sizeof(subTyps))) {










417



    return TSDB_CODE_INVALID_PARA;










418



  }











UNCOV

419



  memcpy(pInfo->subTyps, (char*)subTyps, sizeof(subTyps));




×







420












421



  // fset partition list











UNCOV

422



  memset(pInfo->pLists, 0, sizeof(pInfo->pLists[0]) * TSDB_RETENTION_MAX);




×








UNCOV

423



  for (int32_t j = 0; j < pInfo->tsdbMaxCnt; ++j) {





×








UNCOV

424



    STsdb* pTsdb = SMA_RSMA_GET_TSDB(pVnode, j);





×








UNCOV

425



    pInfo->pLists[j] = tsdbSnapGetFSetPartList(pTsdb->pFS);




×








UNCOV

426



    if (pInfo->pLists[j] == NULL) {





×







427



      return terrno;




×







428



    }










429



  }











UNCOV

430



  return 0;




×







431



}










432













UNCOV

433



static void tsdbPartitionInfoClear(STsdbPartitionInfo* pInfo) {




×








UNCOV

434



  for (int32_t j = 0; j < pInfo->tsdbMaxCnt; ++j) {





×








UNCOV

435



    if (pInfo->pLists[j] == NULL) continue;





×








UNCOV

436



    tsdbFSetPartListDestroy(&pInfo->pLists[j]);




×







437



  }











UNCOV

438



}




×







439













UNCOV

440



static int32_t tsdbPartitionInfoEstSize(STsdbPartitionInfo* pInfo) {




×








UNCOV

441



  int32_t dataLen = 0;




×








UNCOV

442



  for (int32_t j = 0; j < pInfo->tsdbMaxCnt; ++j) {





×








UNCOV

443



    dataLen += sizeof(SSyncTLV);  // subTyps[j]




×








UNCOV

444



    dataLen += tTsdbFSetPartListDataLenCalc(pInfo->pLists[j]);




×







445



  }











UNCOV

446



  return dataLen;




×







447



}










448













UNCOV

449



static int32_t tsdbPartitionInfoSerialize(STsdbPartitionInfo* pInfo, uint8_t* buf, int32_t bufLen) {




×








UNCOV

450



  int32_t tlen = 0;




×








UNCOV

451



  int32_t offset = 0;




×








UNCOV

452



  for (int32_t j = 0; j < pInfo->tsdbMaxCnt; ++j) {





×








UNCOV

453



    SSyncTLV* pSubHead = (void*)((char*)buf + offset);




×








UNCOV

454



    int32_t   valOffset = offset + sizeof(*pSubHead);




×








UNCOV

455



    int32_t   code = tSerializeTsdbFSetPartList(pSubHead->val, bufLen - valOffset, pInfo->pLists[j], &tlen);




×








UNCOV

456



    if (code) {





×







457



      tsdbError("vgId:%d, failed to serialize fset partition list of tsdb %d since %s", pInfo->vgId, j, terrstr());





×







458



      return code;




×







459



    }











UNCOV

460



    pSubHead->typ = pInfo->subTyps[j];




×








UNCOV

461



    pSubHead->len = tlen;




×








UNCOV

462



    offset += sizeof(*pSubHead) + tlen;




×







463



  }











UNCOV

464



  return offset;




×







465



}










466












467



// tsdb replication opts











UNCOV

468



static int32_t tTsdbRepOptsDataLenCalc(STsdbRepOpts* pInfo) {




×








UNCOV

469



  int32_t hdrLen = sizeof(int32_t);




×








UNCOV

470



  int32_t datLen = 0;




×







471













UNCOV

472



  int8_t  msgVer = 0;




×








UNCOV

473



  int64_t reserved64 = 0;




×








UNCOV

474



  int16_t format = 0;




×








UNCOV

475



  hdrLen += sizeof(msgVer);




×








UNCOV

476



  datLen += hdrLen;




×








UNCOV

477



  datLen += sizeof(format);




×








UNCOV

478



  datLen += sizeof(reserved64);




×








UNCOV

479



  datLen += sizeof(*pInfo);




×








UNCOV

480



  return datLen;




×







481



}










482













UNCOV

483



int32_t tSerializeTsdbRepOpts(void* buf, int32_t bufLen, STsdbRepOpts* pOpts) {




×








UNCOV

484



  int32_t  code = 0;




×








UNCOV

485



  SEncoder encoder = {0};




×








UNCOV

486



  int64_t  reserved64 = 0;




×








UNCOV

487



  int8_t   msgVer = TSDB_SNAP_MSG_VER;




×







488













UNCOV

489



  tEncoderInit(&encoder, buf, bufLen);




×







490













UNCOV

491



  if ((code = tStartEncode(&encoder))) goto _err;





×








UNCOV

492



  if ((code = tEncodeI8(&encoder, msgVer))) goto _err;





×








UNCOV

493



  int16_t format = pOpts->format;




×








UNCOV

494



  if ((code = tEncodeI16(&encoder, format))) goto _err;





×








UNCOV

495



  if ((code = tEncodeI64(&encoder, reserved64))) goto _err;





×







496













UNCOV

497



  tEndEncode(&encoder);




×








UNCOV

498



  int32_t tlen = encoder.pos;




×








UNCOV

499



  tEncoderClear(&encoder);




×








UNCOV

500



  return tlen;




×







501












502



_err:




×







503



  tEncoderClear(&encoder);




×







504



  return code;




×







505



}










506













UNCOV

507



int32_t tDeserializeTsdbRepOpts(void* buf, int32_t bufLen, STsdbRepOpts* pOpts) {




×







508



  int32_t  code;











UNCOV

509



  SDecoder decoder = {0};




×








UNCOV

510



  int64_t  reserved64 = 0;




×








UNCOV

511



  int8_t   msgVer = 0;




×







512













UNCOV

513



  tDecoderInit(&decoder, buf, bufLen);




×







514













UNCOV

515



  if ((code = tStartDecode(&decoder))) goto _err;





×








UNCOV

516



  if ((code = tDecodeI8(&decoder, &msgVer))) goto _err;





×








UNCOV

517



  if (msgVer != TSDB_SNAP_MSG_VER) goto _err;





×








UNCOV

518



  int16_t format = 0;




×








UNCOV

519



  if ((code = tDecodeI16(&decoder, &format))) goto _err;





×








UNCOV

520



  pOpts->format = format;




×








UNCOV

521



  if ((code = tDecodeI64(&decoder, &reserved64))) goto _err;





×







522













UNCOV

523



  tEndDecode(&decoder);




×








UNCOV

524



  tDecoderClear(&decoder);




×








UNCOV

525



  return 0;




×







526












527



_err:




×







528



  tDecoderClear(&decoder);




×







529



  return code;




×







530



}










531













UNCOV

532



static int32_t tsdbRepOptsEstSize(STsdbRepOpts* pOpts) {




×








UNCOV

533



  int32_t dataLen = 0;




×








UNCOV

534



  dataLen += sizeof(SSyncTLV);




×








UNCOV

535



  dataLen += tTsdbRepOptsDataLenCalc(pOpts);




×








UNCOV

536



  return dataLen;




×







537



}










538













UNCOV

539



static int32_t tsdbRepOptsSerialize(STsdbRepOpts* pOpts, void* buf, int32_t bufLen) {




×








UNCOV

540



  SSyncTLV* pSubHead = buf;




×








UNCOV

541



  int32_t   offset = 0;




×








UNCOV

542



  int32_t   tlen = 0;




×








UNCOV

543



  if ((tlen = tSerializeTsdbRepOpts(pSubHead->val, bufLen, pOpts)) < 0) {





×







544



    return tlen;




×







545



  }











UNCOV

546



  pSubHead->typ = SNAP_DATA_RAW;




×








UNCOV

547



  pSubHead->len = tlen;




×








UNCOV

548



  offset += sizeof(*pSubHead) + tlen;




×








UNCOV

549



  return offset;




×







550



}










551












552



// snap info











UNCOV

553



static int32_t tsdbSnapPrepDealWithSnapInfo(SVnode* pVnode, SSnapshot* pSnap, STsdbRepOpts* pInfo) {




×








UNCOV

554



  if (!pSnap->data) {





×







555



    return 0;




×







556



  }











UNCOV

557



  int32_t code = 0;




×







558













UNCOV

559



  SSyncTLV* pHead = (void*)pSnap->data;




×








UNCOV

560



  int32_t   offset = 0;




×







561













UNCOV

562



  while (offset + sizeof(*pHead) < pHead->len) {





×








UNCOV

563



    SSyncTLV* pField = (void*)(pHead->val + offset);




×








UNCOV

564



    offset += sizeof(*pField) + pField->len;




×








UNCOV

565



    void*   buf = pField->val;




×








UNCOV

566



    int32_t bufLen = pField->len;




×







567













UNCOV

568



    switch (pField->typ) {





×








UNCOV

569



      case SNAP_DATA_TSDB:




×







570



      case SNAP_DATA_RSMA1:










571



      case SNAP_DATA_RSMA2: {











UNCOV

572



      } break;




×








UNCOV

573



      case SNAP_DATA_RAW: {




×








UNCOV

574



        code = tDeserializeTsdbRepOpts(buf, bufLen, pInfo);




×








UNCOV

575



        if (code < 0) {





×







576



          tsdbError("vgId:%d, failed to deserialize tsdb rep opts since %s", TD_VID(pVnode), terrstr());





×







577



          return code;




×







578



        }











UNCOV

579



      } break;




×







580



      default:




×







581



        code = TSDB_CODE_INVALID_MSG;




×







582



        tsdbError("vgId:%d, unexpected subfield type of snap info. typ:%d", TD_VID(pVnode), pField->typ);





×







583



        return code;




×







584



    }










585



  }










586













UNCOV

587



  return code;




×







588



}










589













UNCOV

590



int32_t tsdbSnapPrepDescription(SVnode* pVnode, SSnapshot* pSnap) {




×








UNCOV

591



  STsdbPartitionInfo  partitionInfo = {0};




×








UNCOV

592



  int                 code = 0;




×








UNCOV

593



  STsdbPartitionInfo* pInfo = &partitionInfo;




×







594













UNCOV

595



  code = tsdbPartitionInfoInit(pVnode, pInfo);




×








UNCOV

596



  if (code) {





×







597



    goto _out;




×







598



  }










599












600



  // deal with snap info for reply











UNCOV

601



  STsdbRepOpts opts = {.format = TSDB_SNAP_REP_FMT_RAW};




×








UNCOV

602



  if (pSnap->type == TDMT_SYNC_PREP_SNAPSHOT_REPLY) {





×








UNCOV

603



    STsdbRepOpts leaderOpts = {0};




×








UNCOV

604



    if ((code = tsdbSnapPrepDealWithSnapInfo(pVnode, pSnap, &leaderOpts)) < 0) {





×







605



      tsdbError("vgId:%d, failed to deal with snap info for reply since %s", TD_VID(pVnode), terrstr());





×







606



      goto _out;




×







607



    }











UNCOV

608



    opts.format = TMIN(opts.format, leaderOpts.format);




×







609



  }










610












611



  // info data realloc











UNCOV

612



  const int32_t headLen = sizeof(SSyncTLV);




×








UNCOV

613



  int32_t       bufLen = headLen;




×








UNCOV

614



  bufLen += tsdbPartitionInfoEstSize(pInfo);




×








UNCOV

615



  bufLen += tsdbRepOptsEstSize(&opts);




×








UNCOV

616



  if ((code = syncSnapInfoDataRealloc(pSnap, bufLen)) != 0) {





×







617



    tsdbError("vgId:%d, failed to realloc memory for data of snap info. bytes:%d", TD_VID(pVnode), bufLen);





×







618



    goto _out;




×







619



  }










620












621



  // serialization











UNCOV

622



  char*   buf = (void*)pSnap->data;




×








UNCOV

623



  int32_t offset = headLen;




×








UNCOV

624



  int32_t tlen = 0;




×







625













UNCOV

626



  if ((tlen = tsdbPartitionInfoSerialize(pInfo, (uint8_t*)(buf + offset), bufLen - offset)) < 0) {





×







627



    code = tlen;




×







628



    tsdbError("vgId:%d, failed to serialize tsdb partition info since %s", TD_VID(pVnode), terrstr());





×







629



    goto _out;




×







630



  }











UNCOV

631



  offset += tlen;




×







632













UNCOV

633



  if ((tlen = tsdbRepOptsSerialize(&opts, buf + offset, bufLen - offset)) < 0) {





×







634



    code = tlen;




×







635



    tsdbError("vgId:%d, failed to serialize tsdb rep opts since %s", TD_VID(pVnode), terrstr());





×







636



    goto _out;




×







637



  }











UNCOV

638



  offset += tlen;




×







639












640



  // set header of info data











UNCOV

641



  SSyncTLV* pHead = pSnap->data;




×








UNCOV

642



  pHead->typ = pSnap->type;




×








UNCOV

643



  pHead->len = offset - headLen;




×







644













UNCOV

645



  tsdbInfo("vgId:%d, tsdb snap info prepared. type:%s, val length:%d", TD_VID(pVnode), TMSG_INFO(pHead->typ),





×







646



           pHead->len);










647












648



_out:




×








UNCOV

649



  tsdbPartitionInfoClear(pInfo);




×








UNCOV

650



  return code;




×







651



}
























    

  • 

    •