#4822

Committed 27 Oct 2025 05:42AM UTC coverage: 59.732% (+1.0%) from 58.728%

Build # #4822

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

Merge pull request #33377 from taosdata/fix/main/rename-udf-path

fix: update UDF example links to correct file paths

Run Details

121214 of 258518 branches covered (46.89%)

Branch coverage included in aggregate %.

193636 of 268583 relevant lines covered (72.1%)

4002399.5 hits per line

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

59.52

/source/dnode/vnode/src/vnd/vnodeCommit.c

/*
 * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
 *
 * This program is free software: you can use, redistribute, and/or modify
 * it under the terms of the GNU Affero General Public License, version 3
 * or later ("AGPL"), as published by the Free Software Foundation.
 *
 * This program is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
 */

#include "meta.h"
#include "sync.h"
#include "vnd.h"
#include "vnodeInt.h"

extern int32_t tsdbPreCommit(STsdb *pTsdb);
extern int32_t tsdbCommitBegin(STsdb *pTsdb, SCommitInfo *pInfo);
extern int32_t tsdbCommitCommit(STsdb *pTsdb);
extern int32_t tsdbCommitAbort(STsdb *pTsdb);

#define VND_INFO_FNAME_TMP "vnode_tmp.json"

static int vnodeEncodeInfo(const SVnodeInfo *pInfo, char **ppData);
static int vnodeCommitImpl(SCommitInfo *pInfo);

#define WAIT_TIME_MILI_SEC 10  // miliseconds

static int32_t vnodeTryRecycleBufPool(SVnode *pVnode) {
  int32_t code = 0;

  if (pVnode->onRecycle == NULL) {
    if (pVnode->recycleHead == NULL) {
      vDebug("vgId:%d, no recyclable buffer pool", TD_VID(pVnode));
      goto _exit;
    } else {
      vDebug("vgId:%d, buffer pool %p of id %d on recycle queue, try to recycle", TD_VID(pVnode), pVnode->recycleHead,
             pVnode->recycleHead->id);

      pVnode->onRecycle = pVnode->recycleHead;
      if (pVnode->recycleHead == pVnode->recycleTail) {
        pVnode->recycleHead = pVnode->recycleTail = NULL;
      } else {
        pVnode->recycleHead = pVnode->recycleHead->recycleNext;
        pVnode->recycleHead->recyclePrev = NULL;
      }
      pVnode->onRecycle->recycleNext = pVnode->onRecycle->recyclePrev = NULL;
    }
  }

  code = vnodeBufPoolRecycle(pVnode->onRecycle);
  if (code) goto _exit;

_exit:
  if (code) {
    vError("vgId:%d, %s failed since %s", TD_VID(pVnode), __func__, tstrerror(code));
  }
  return code;
}
static int32_t vnodeGetBufPoolToUse(SVnode *pVnode) {
  int32_t code = 0;
  int32_t lino = 0;

  (void)taosThreadMutexLock(&pVnode->mutex);

  int32_t nTry = 0;
  for (;;) {
    ++nTry;

    if (pVnode->freeList) {
      vDebug("vgId:%d, allocate free buffer pool on %d try, pPool:%p id:%d", TD_VID(pVnode), nTry, pVnode->freeList,
             pVnode->freeList->id);

      pVnode->inUse = pVnode->freeList;
      pVnode->inUse->nRef = 1;
      pVnode->freeList = pVnode->inUse->freeNext;
      pVnode->inUse->freeNext = NULL;
      break;
    } else {
      vDebug("vgId:%d, no free buffer pool on %d try, try to recycle...", TD_VID(pVnode), nTry);

      code = vnodeTryRecycleBufPool(pVnode);
      TSDB_CHECK_CODE(code, lino, _exit);

      if (pVnode->freeList == NULL) {
        vDebug("vgId:%d, no free buffer pool on %d try, wait %d ms...", TD_VID(pVnode), nTry, WAIT_TIME_MILI_SEC);

        struct timeval  tv;
        struct timespec ts;
        if (taosGetTimeOfDay(&tv) != 0) {
          continue;
        }
        ts.tv_nsec = tv.tv_usec * 1000 + WAIT_TIME_MILI_SEC * 1000000;
        if (ts.tv_nsec > 999999999l) {
          ts.tv_sec = tv.tv_sec + 1;
          ts.tv_nsec -= 1000000000l;
        } else {
          ts.tv_sec = tv.tv_sec;
        }

        code = taosThreadCondTimedWait(&pVnode->poolNotEmpty, &pVnode->mutex, &ts);
        // ignore timeout error and retry
        if (code == TSDB_CODE_TIMEOUT_ERROR) {
          code = TSDB_CODE_SUCCESS;
        }
        TSDB_CHECK_CODE(code, lino, _exit);
      }
    }
  }

_exit:
  (void)taosThreadMutexUnlock(&pVnode->mutex);
  if (code) {
    vError("vgId:%d, %s failed at line %d since %s", TD_VID(pVnode), __func__, lino, tstrerror(code));
  }
  return code;
}
int vnodeBegin(SVnode *pVnode) {
  int32_t code = 0;
  int32_t lino = 0;

  // alloc buffer pool
  code = vnodeGetBufPoolToUse(pVnode);
  TSDB_CHECK_CODE(code, lino, _exit);

  // begin meta
  code = metaBegin(pVnode->pMeta, META_BEGIN_HEAP_BUFFERPOOL);
  TSDB_CHECK_CODE(code, lino, _exit);

  // begin tsdb
  code = tsdbBegin(pVnode->pTsdb);
  TSDB_CHECK_CODE(code, lino, _exit);

_exit:
  if (code) {
    terrno = code;
    vError("vgId:%d, %s failed at line %d since %s", TD_VID(pVnode), __func__, lino, tstrerror(code));
  }
  return code;
}

int vnodeShouldCommit(SVnode *pVnode, bool atExit) {
  bool diskAvail = osDataSpaceAvailable();
  bool needCommit = false;

  (void)taosThreadMutexLock(&pVnode->mutex);
  if (pVnode->inUse && diskAvail) {
    needCommit = (pVnode->inUse->size > pVnode->inUse->node.size) ||
                 (atExit && (pVnode->inUse->size > 0 || pVnode->pMeta->changed ||
                             pVnode->state.applied - pVnode->state.committed > 4096));
  }
  vTrace("vgId:%d, should commit:%d, disk available:%d, buffer size:%" PRId64 ", node size:%" PRId64
         ", meta changed:%d"
         ", state:[%" PRId64 ",%" PRId64 "]",
         TD_VID(pVnode), needCommit, diskAvail, pVnode->inUse ? pVnode->inUse->size : 0,
         pVnode->inUse ? pVnode->inUse->node.size : 0, pVnode->pMeta->changed, pVnode->state.applied,
         pVnode->state.committed);
  (void)taosThreadMutexUnlock(&pVnode->mutex);
  return needCommit;
}

int vnodeSaveInfo(const char *dir, const SVnodeInfo *pInfo) {
  int32_t   code = 0;
  int32_t   lino;
  char      fname[TSDB_FILENAME_LEN];
  TdFilePtr pFile = NULL;
  char     *data = NULL;

  snprintf(fname, TSDB_FILENAME_LEN, "%s%s%s", dir, TD_DIRSEP, VND_INFO_FNAME_TMP);

  code = vnodeEncodeInfo(pInfo, &data);
  TSDB_CHECK_CODE(code, lino, _exit);

  // save info to a vnode_tmp.json
  pFile = taosOpenFile(fname, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC | TD_FILE_WRITE_THROUGH);
  if (pFile == NULL) {
    TSDB_CHECK_CODE(code = terrno, lino, _exit);
  }

  if (taosWriteFile(pFile, data, strlen(data)) < 0) {
    TSDB_CHECK_CODE(code = terrno, lino, _exit);
  }

  if (taosFsyncFile(pFile) < 0) {
    TSDB_CHECK_CODE(code = terrno, lino, _exit);
  }

_exit:
  if (code) {
    vError("vgId:%d %s failed at %s:%d since %s", pInfo->config.vgId, __func__, __FILE__, lino, tstrerror(code));
  } else {
    vInfo("vgId:%d, vnode info is saved, fname:%s replica:%d selfIndex:%d changeVersion:%d", pInfo->config.vgId, fname,
          pInfo->config.syncCfg.replicaNum, pInfo->config.syncCfg.myIndex, pInfo->config.syncCfg.changeVersion);
  }
  if (taosCloseFile(&pFile) != 0) {
    vError("vgId:%d, failed to close file", pInfo->config.vgId);
  }
  taosMemoryFree(data);
  TAOS_RETURN(code);
}

int vnodeCommitInfo(const char *dir) {
  char fname[TSDB_FILENAME_LEN];
  char tfname[TSDB_FILENAME_LEN];

  snprintf(fname, TSDB_FILENAME_LEN, "%s%s%s", dir, TD_DIRSEP, VND_INFO_FNAME);
  snprintf(tfname, TSDB_FILENAME_LEN, "%s%s%s", dir, TD_DIRSEP, VND_INFO_FNAME_TMP);

  int32_t code = taosRenameFile(tfname, fname);
  if (code < 0) {
    return code;
  }

  vInfo("vnode info is committed, dir:%s", dir);
  return 0;
}

int vnodeLoadInfo(const char *dir, SVnodeInfo *pInfo) {
  int32_t   code = 0;
  int32_t   lino;
  char      fname[TSDB_FILENAME_LEN];
  TdFilePtr pFile = NULL;
  char     *pData = NULL;
  int64_t   size;

  snprintf(fname, TSDB_FILENAME_LEN, "%s%s%s", dir, TD_DIRSEP, VND_INFO_FNAME);

  // read info
  pFile = taosOpenFile(fname, TD_FILE_READ);
  if (pFile == NULL) {
    TSDB_CHECK_CODE(code = terrno, lino, _exit);
  }

  code = taosFStatFile(pFile, &size, NULL);
  TSDB_CHECK_CODE(code, lino, _exit);

  pData = taosMemoryMalloc(size + 1);
  if (pData == NULL) {
    TSDB_CHECK_CODE(code = terrno, lino, _exit);
  }

  if (taosReadFile(pFile, pData, size) < 0) {
    TSDB_CHECK_CODE(code = terrno, lino, _exit);
  }

  pData[size] = '\0';

  // decode info
  code = vnodeDecodeInfo(pData, pInfo);
  TSDB_CHECK_CODE(code, lino, _exit);

  pInfo->config.walCfg.committed = pInfo->state.committed;
_exit:
  if (code) {
    vError("vgId:%d %s failed at %s:%d since %s, file:%s", pInfo->config.vgId, __func__, __FILE__, lino,
           tstrerror(code), fname);
  }
  taosMemoryFree(pData);
  if (taosCloseFile(&pFile) != 0) {
    vError("vgId:%d, failed to close file", pInfo->config.vgId);
  }
  return code;
}

static int32_t vnodePrepareCommit(SVnode *pVnode, SCommitInfo *pInfo) {
  int32_t code = 0;
  int32_t lino = 0;
  char    dir[TSDB_FILENAME_LEN] = {0};
  int64_t lastCommitted = pInfo->info.state.committed;

  // wait last commit task
  vnodeAWait(&pVnode->commitTask);

  code = syncNodeGetConfig(pVnode->sync, &pVnode->config.syncCfg);
  TSDB_CHECK_CODE(code, lino, _exit);

  pVnode->state.commitTerm = pVnode->state.applyTerm;

  pInfo->info.config = pVnode->config;
  pInfo->info.state.committed = pVnode->state.applied;
  pInfo->info.state.commitTerm = pVnode->state.applyTerm;
  pInfo->info.state.commitID = ++pVnode->state.commitID;
  pInfo->pVnode = pVnode;
  pInfo->txn = metaGetTxn(pVnode->pMeta);

  // save info
  vnodeGetPrimaryPath(pVnode, false, dir, TSDB_FILENAME_LEN);

  vDebug("vgId:%d, save config while prepare commit", TD_VID(pVnode));
  code = vnodeSaveInfo(dir, &pInfo->info);
  TSDB_CHECK_CODE(code, lino, _exit);

  code = tsdbPreCommit(pVnode->pTsdb);
  TSDB_CHECK_CODE(code, lino, _exit);

  code = metaPrepareAsyncCommit(pVnode->pMeta);
  TSDB_CHECK_CODE(code, lino, _exit);

  (void)taosThreadMutexLock(&pVnode->mutex);
  pVnode->onCommit = pVnode->inUse;
  pVnode->inUse = NULL;
  (void)taosThreadMutexUnlock(&pVnode->mutex);

_exit:
  if (code) {
    vError("vgId:%d, %s failed at line %d since %s, commit id:%" PRId64, TD_VID(pVnode), __func__, lino,
           tstrerror(code), pVnode->state.commitID);
  } else {
    vDebug("vgId:%d, %s done, commit id:%" PRId64, TD_VID(pVnode), __func__, pInfo->info.state.commitID);
  }

  return code;
}
static void vnodeReturnBufPool(SVnode *pVnode) {
  (void)taosThreadMutexLock(&pVnode->mutex);

  SVBufPool *pPool = pVnode->onCommit;
  int32_t    nRef = atomic_sub_fetch_32(&pPool->nRef, 1);

  pVnode->onCommit = NULL;
  if (nRef == 0) {
    vnodeBufPoolAddToFreeList(pPool);
  } else if (nRef > 0) {
    vDebug("vgId:%d, buffer pool %p of id %d is added to recycle queue", TD_VID(pVnode), pPool, pPool->id);

    if (pVnode->recycleTail == NULL) {
      pPool->recyclePrev = pPool->recycleNext = NULL;
      pVnode->recycleHead = pVnode->recycleTail = pPool;
    } else {
      pPool->recyclePrev = pVnode->recycleTail;
      pPool->recycleNext = NULL;
      pVnode->recycleTail->recycleNext = pPool;
      pVnode->recycleTail = pPool;
    }
  } else {
    vError("vgId:%d, buffer pool %p of id %d nRef:%d", TD_VID(pVnode), pPool, pPool->id, nRef);
  }

  (void)taosThreadMutexUnlock(&pVnode->mutex);
}
static int32_t vnodeCommit(void *arg) {
  int32_t code = 0;

  SCommitInfo *pInfo = (SCommitInfo *)arg;
  SVnode      *pVnode = pInfo->pVnode;

  // commit
  METRICS_TIMING_BLOCK(pVnode->writeMetrics.commit_time, METRIC_LEVEL_HIGH, {
    if ((code = vnodeCommitImpl(pInfo))) {
      vFatal("vgId:%d, failed to commit vnode since %s", TD_VID(pVnode), terrstr());
      taosMsleep(100);
      exit(EXIT_FAILURE);
      goto _exit;
    }
  });

  METRICS_UPDATE(pVnode->writeMetrics.commit_count, METRIC_LEVEL_HIGH, 1);

  vnodeReturnBufPool(pVnode);

_exit:
  taosMemoryFree(arg);
  return code;
}

static void vnodeCommitCancel(void *arg) { taosMemoryFree(arg); }

int vnodeAsyncCommit(SVnode *pVnode) {
  int32_t code = 0;
  int32_t lino = 0;

  SCommitInfo *pInfo = (SCommitInfo *)taosMemoryCalloc(1, sizeof(*pInfo));
  if (NULL == pInfo) {
    TSDB_CHECK_CODE(code = terrno, lino, _exit);
  }
  code = vnodePrepareCommit(pVnode, pInfo);
  TSDB_CHECK_CODE(code, lino, _exit);

  // schedule the task
  code = vnodeAsync(COMMIT_TASK_ASYNC, EVA_PRIORITY_HIGH, vnodeCommit, vnodeCommitCancel, pInfo, &pVnode->commitTask);
  TSDB_CHECK_CODE(code, lino, _exit);

_exit:
  if (code) {
    taosMemoryFree(pInfo);
    vError("vgId:%d %s failed at line %d since %s" PRId64, TD_VID(pVnode), __func__, lino, tstrerror(code));
  } else {
    vInfo("vgId:%d, vnode async commit done, commitId:%" PRId64 " term:%" PRId64 " applied:%" PRId64, TD_VID(pVnode),
          pVnode->state.commitID, pVnode->state.applyTerm, pVnode->state.applied);
  }
  return code;
}

int32_t vnodeSyncCommit(SVnode *pVnode) {
  int32_t lino;
  int32_t code = vnodeAsyncCommit(pVnode);
  TSDB_CHECK_CODE(code, lino, _exit);
  vnodeAWait(&pVnode->commitTask);

_exit:
  if (code) {
    vError("vgId:%d, %s failed at line %d since %s", TD_VID(pVnode), __func__, lino, tstrerror(code));
  } else {
    vInfo("vgId:%d, sync commit end", TD_VID(pVnode));
  }

  return code;
}

static int vnodeCommitImpl(SCommitInfo *pInfo) {
  int32_t code = 0;
  int32_t lino = 0;

  char    dir[TSDB_FILENAME_LEN] = {0};
  SVnode *pVnode = pInfo->pVnode;

  vInfo("vgId:%d, start to commit, commitId:%" PRId64 " version:%" PRId64 " term: %" PRId64, TD_VID(pVnode),
        pInfo->info.state.commitID, pInfo->info.state.committed, pInfo->info.state.commitTerm);

  // persist wal before starting
  if ((code = walPersist(pVnode->pWal)) < 0) {
    vError("vgId:%d, failed to persist wal since %s", TD_VID(pVnode), tstrerror(code));
    return code;
  }

  vnodeGetPrimaryPath(pVnode, false, dir, TSDB_FILENAME_LEN);

  code = syncBeginSnapshot(pVnode->sync, pInfo->info.state.committed);
  TSDB_CHECK_CODE(code, lino, _exit);

  code = tsdbCommitBegin(pVnode->pTsdb, pInfo);
  TSDB_CHECK_CODE(code, lino, _exit);

  if (!TSDB_CACHE_NO(pVnode->config)) {
    METRICS_TIMING_BLOCK(pVnode->writeMetrics.last_cache_commit_time, METRIC_LEVEL_HIGH,
                         { code = tsdbCacheCommit(pVnode->pTsdb); });
    METRICS_UPDATE(pVnode->writeMetrics.last_cache_commit_count, METRIC_LEVEL_HIGH, 1);
    TSDB_CHECK_CODE(code, lino, _exit);
  }

  // blob storage engine commit
  code = bseCommit(pVnode->pBse);
  // commit info
  code = vnodeCommitInfo(dir);
  TSDB_CHECK_CODE(code, lino, _exit);

  code = tsdbCommitCommit(pVnode->pTsdb);
  TSDB_CHECK_CODE(code, lino, _exit);

  code = metaFinishCommit(pVnode->pMeta, pInfo->txn);
  TSDB_CHECK_CODE(code, lino, _exit);

  pVnode->state.committed = pInfo->info.state.committed;

  code = syncEndSnapshot(pVnode->sync);
  TSDB_CHECK_CODE(code, lino, _exit);

  code = tqCommitOffset(pVnode->pTq);
  TSDB_CHECK_CODE(code, lino, _exit);
  
_exit:
  if (code) {
    vError("vgId:%d, %s failed at line %d since %s", TD_VID(pVnode), __func__, lino, tstrerror(code));
  } else {
    vInfo("vgId:%d, commit end", TD_VID(pVnode));
  }
  return code;
}

bool vnodeShouldRollback(SVnode *pVnode) {
  char    tFName[TSDB_FILENAME_LEN] = {0};
  int32_t offset = 0;

  vnodeGetPrimaryPath(pVnode, false, tFName, TSDB_FILENAME_LEN);
  offset = strlen(tFName);
  snprintf(tFName + offset, TSDB_FILENAME_LEN - offset - 1, "%s%s", TD_DIRSEP, VND_INFO_FNAME_TMP);

  return taosCheckExistFile(tFName);
}

void vnodeRollback(SVnode *pVnode) {
  char    tFName[TSDB_FILENAME_LEN] = {0};
  int32_t offset = 0;

  vnodeGetPrimaryPath(pVnode, false, tFName, TSDB_FILENAME_LEN);
  offset = strlen(tFName);
  snprintf(tFName + offset, TSDB_FILENAME_LEN - offset - 1, "%s%s", TD_DIRSEP, VND_INFO_FNAME_TMP);

  if (taosRemoveFile(tFName) != 0) {
    vError("vgId:%d, failed to remove file %s since %s", TD_VID(pVnode), tFName, tstrerror(terrno));
  }
}

static int vnodeEncodeState(const void *pObj, SJson *pJson) {
  const SVState *pState = (SVState *)pObj;

  TAOS_CHECK_RETURN(tjsonAddIntegerToObject(pJson, "commit version", pState->committed));
  TAOS_CHECK_RETURN(tjsonAddIntegerToObject(pJson, "commit ID", pState->commitID));
  TAOS_CHECK_RETURN(tjsonAddIntegerToObject(pJson, "commit term", pState->commitTerm));

  return 0;
}

static int vnodeDecodeState(const SJson *pJson, void *pObj) {
  SVState *pState = (SVState *)pObj;

  int32_t code;
  tjsonGetNumberValue(pJson, "commit version", pState->committed, code);
  if (code) return code;
  tjsonGetNumberValue(pJson, "commit ID", pState->commitID, code);
  if (code) return code;
  tjsonGetNumberValue(pJson, "commit term", pState->commitTerm, code);
  if (code) return code;

  return 0;
}

static int vnodeEncodeInfo(const SVnodeInfo *pInfo, char **ppData) {
  int32_t code = 0;
  int32_t lino;
  SJson  *pJson = NULL;
  char   *pData = NULL;

  pJson = tjsonCreateObject();
  if (pJson == NULL) {
    TSDB_CHECK_CODE(code = terrno, lino, _exit);
  }

  code = tjsonAddObject(pJson, "config", vnodeEncodeConfig, (void *)&pInfo->config);
  TSDB_CHECK_CODE(code, lino, _exit);

  code = tjsonAddObject(pJson, "state", vnodeEncodeState, (void *)&pInfo->state);
  TSDB_CHECK_CODE(code, lino, _exit);

  pData = tjsonToString(pJson);
  if (pData == NULL) {
    TSDB_CHECK_CODE(code = terrno, lino, _exit);
  }

  tjsonDelete(pJson);

_exit:
  if (code) {
    tjsonDelete(pJson);
    *ppData = NULL;
  } else {
    *ppData = pData;
  }
  return code;
}

int vnodeDecodeInfo(uint8_t *pData, SVnodeInfo *pInfo) {
  int32_t code = 0;
  int32_t lino;
  SJson  *pJson = NULL;

  pJson = tjsonParse(pData);
  if (pJson == NULL) {
    TSDB_CHECK_CODE(code = TSDB_CODE_INVALID_DATA_FMT, lino, _exit);
  }

  code = tjsonToObject(pJson, "config", vnodeDecodeConfig, (void *)&pInfo->config);
  TSDB_CHECK_CODE(code, lino, _exit);

  code = tjsonToObject(pJson, "state", vnodeDecodeState, (void *)&pInfo->state);
  TSDB_CHECK_CODE(code, lino, _exit);

_exit:
  tjsonDelete(pJson);
  return code;
}

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 "meta.h"
17	#include "sync.h"
18	#include "vnd.h"
19	#include "vnodeInt.h"
20
21	extern int32_t tsdbPreCommit(STsdb *pTsdb);
22	extern int32_t tsdbCommitBegin(STsdb pTsdb, SCommitInfo pInfo);
23	extern int32_t tsdbCommitCommit(STsdb *pTsdb);
24	extern int32_t tsdbCommitAbort(STsdb *pTsdb);
25
26	#define VND_INFO_FNAME_TMP "vnode_tmp.json"
27
28	static int vnodeEncodeInfo(const SVnodeInfo pInfo, char *ppData);
29	static int vnodeCommitImpl(SCommitInfo *pInfo);
30
31	#define WAIT_TIME_MILI_SEC 10 // miliseconds
32
33	static int32_t vnodeTryRecycleBufPool(SVnode *pVnode) {	×
34	int32_t code = 0;	×
35
36	if (pVnode->onRecycle == NULL) {	×
37	if (pVnode->recycleHead == NULL) {	×
38	vDebug("vgId:%d, no recyclable buffer pool", TD_VID(pVnode));	×
39	goto _exit;	×
40	} else {
41	vDebug("vgId:%d, buffer pool %p of id %d on recycle queue, try to recycle", TD_VID(pVnode), pVnode->recycleHead,	×
42	pVnode->recycleHead->id);
43
44	pVnode->onRecycle = pVnode->recycleHead;	×
45	if (pVnode->recycleHead == pVnode->recycleTail) {	×
46	pVnode->recycleHead = pVnode->recycleTail = NULL;	×
47	} else {
48	pVnode->recycleHead = pVnode->recycleHead->recycleNext;	×
49	pVnode->recycleHead->recyclePrev = NULL;	×
50	}
51	pVnode->onRecycle->recycleNext = pVnode->onRecycle->recyclePrev = NULL;	×
52	}
53	}
54
55	code = vnodeBufPoolRecycle(pVnode->onRecycle);	×
56	if (code) goto _exit;	×
57
58	_exit:	×
59	if (code) {	×
60	vError("vgId:%d, %s failed since %s", TD_VID(pVnode), __func__, tstrerror(code));	×
61	}
62	return code;	×
63	}
64	static int32_t vnodeGetBufPoolToUse(SVnode *pVnode) {	23,739✔
65	int32_t code = 0;	23,739✔
66	int32_t lino = 0;	23,739✔
67
68	(void)taosThreadMutexLock(&pVnode->mutex);	23,739✔
69
70	int32_t nTry = 0;	23,739✔
71	for (;;) {
72	++nTry;	23,739✔
73
74	if (pVnode->freeList) {	23,739!
75	vDebug("vgId:%d, allocate free buffer pool on %d try, pPool:%p id:%d", TD_VID(pVnode), nTry, pVnode->freeList,	23,739✔
76	pVnode->freeList->id);
77
78	pVnode->inUse = pVnode->freeList;	23,739✔
79	pVnode->inUse->nRef = 1;	23,739✔
80	pVnode->freeList = pVnode->inUse->freeNext;	23,739✔
81	pVnode->inUse->freeNext = NULL;	23,739✔
82	break;	23,739✔
83	} else {
84	vDebug("vgId:%d, no free buffer pool on %d try, try to recycle...", TD_VID(pVnode), nTry);	×
85
86	code = vnodeTryRecycleBufPool(pVnode);	×
87	TSDB_CHECK_CODE(code, lino, _exit);	×
88
89	if (pVnode->freeList == NULL) {	×
90	vDebug("vgId:%d, no free buffer pool on %d try, wait %d ms...", TD_VID(pVnode), nTry, WAIT_TIME_MILI_SEC);	×
91
92	struct timeval tv;
93	struct timespec ts;
94	if (taosGetTimeOfDay(&tv) != 0) {	×
95	continue;	×
96	}
97	ts.tv_nsec = tv.tv_usec * 1000 + WAIT_TIME_MILI_SEC * 1000000;	×
98	if (ts.tv_nsec > 999999999l) {	×
99	ts.tv_sec = tv.tv_sec + 1;	×
100	ts.tv_nsec -= 1000000000l;	×
101	} else {
102	ts.tv_sec = tv.tv_sec;	×
103	}
104
105	code = taosThreadCondTimedWait(&pVnode->poolNotEmpty, &pVnode->mutex, &ts);	×
106	// ignore timeout error and retry
107	if (code == TSDB_CODE_TIMEOUT_ERROR) {	×
108	code = TSDB_CODE_SUCCESS;	×
109	}
110	TSDB_CHECK_CODE(code, lino, _exit);	×
111	}
112	}
113	}
114
115	_exit:	23,739✔
116	(void)taosThreadMutexUnlock(&pVnode->mutex);	23,739✔
117	if (code) {	23,739!
118	vError("vgId:%d, %s failed at line %d since %s", TD_VID(pVnode), __func__, lino, tstrerror(code));	×
119	}
120	return code;	23,739✔
121	}
122	int vnodeBegin(SVnode *pVnode) {	23,739✔
123	int32_t code = 0;	23,739✔
124	int32_t lino = 0;	23,739✔
125
126	// alloc buffer pool
127	code = vnodeGetBufPoolToUse(pVnode);	23,739✔
128	TSDB_CHECK_CODE(code, lino, _exit);	23,739!
129
130	// begin meta
131	code = metaBegin(pVnode->pMeta, META_BEGIN_HEAP_BUFFERPOOL);	23,739✔
132	TSDB_CHECK_CODE(code, lino, _exit);	23,739!
133
134	// begin tsdb
135	code = tsdbBegin(pVnode->pTsdb);	23,739✔
136	TSDB_CHECK_CODE(code, lino, _exit);	23,738!
137
138	_exit:	23,738✔
139	if (code) {	23,738!
140	terrno = code;	×
141	vError("vgId:%d, %s failed at line %d since %s", TD_VID(pVnode), __func__, lino, tstrerror(code));	×
142	}
143	return code;	23,738✔
144	}
145
146	int vnodeShouldCommit(SVnode *pVnode, bool atExit) {	1,408,803✔
147	bool diskAvail = osDataSpaceAvailable();	1,408,803✔
148	bool needCommit = false;	1,408,804✔
149
150	(void)taosThreadMutexLock(&pVnode->mutex);	1,408,804✔
151	if (pVnode->inUse && diskAvail) {	1,408,844!
152	needCommit = (pVnode->inUse->size > pVnode->inUse->node.size) \|\|	1,416,122✔
153	(atExit && (pVnode->inUse->size > 0 \|\| pVnode->pMeta->changed \|\|	7,276✔
154	pVnode->state.applied - pVnode->state.committed > 4096));	2,953✔
155	}
156	vTrace("vgId:%d, should commit:%d, disk available:%d, buffer size:%" PRId64 ", node size:%" PRId64	1,408,844!
157	", meta changed:%d"
158	", state:[%" PRId64 ",%" PRId64 "]",
159	TD_VID(pVnode), needCommit, diskAvail, pVnode->inUse ? pVnode->inUse->size : 0,
160	pVnode->inUse ? pVnode->inUse->node.size : 0, pVnode->pMeta->changed, pVnode->state.applied,
161	pVnode->state.committed);
162	(void)taosThreadMutexUnlock(&pVnode->mutex);	1,408,844✔
163	return needCommit;	1,408,850✔
164	}
165
166	int vnodeSaveInfo(const char dir, const SVnodeInfo pInfo) {	22,689✔
167	int32_t code = 0;	22,689✔
168	int32_t lino;
169	char fname[TSDB_FILENAME_LEN];
170	TdFilePtr pFile = NULL;	22,689✔
171	char *data = NULL;	22,689✔
172
173	snprintf(fname, TSDB_FILENAME_LEN, "%s%s%s", dir, TD_DIRSEP, VND_INFO_FNAME_TMP);	22,689✔
174
175	code = vnodeEncodeInfo(pInfo, &data);	22,689✔
176	TSDB_CHECK_CODE(code, lino, _exit);	22,684!
177
178	// save info to a vnode_tmp.json
179	pFile = taosOpenFile(fname, TD_FILE_CREATE \| TD_FILE_WRITE \| TD_FILE_TRUNC \| TD_FILE_WRITE_THROUGH);	22,684✔
180	if (pFile == NULL) {	22,651!
181	TSDB_CHECK_CODE(code = terrno, lino, _exit);	×
182	}
183
184	if (taosWriteFile(pFile, data, strlen(data)) < 0) {	22,651!
185	TSDB_CHECK_CODE(code = terrno, lino, _exit);	×
186	}
187
188	if (taosFsyncFile(pFile) < 0) {	22,669!
189	TSDB_CHECK_CODE(code = terrno, lino, _exit);	×
190	}
191
192	_exit:	22,686✔
193	if (code) {	22,686!
194	vError("vgId:%d %s failed at %s:%d since %s", pInfo->config.vgId, __func__, __FILE__, lino, tstrerror(code));	×
195	} else {
196	vInfo("vgId:%d, vnode info is saved, fname:%s replica:%d selfIndex:%d changeVersion:%d", pInfo->config.vgId, fname,	22,686✔
197	pInfo->config.syncCfg.replicaNum, pInfo->config.syncCfg.myIndex, pInfo->config.syncCfg.changeVersion);
198	}
199	if (taosCloseFile(&pFile) != 0) {	22,688!
200	vError("vgId:%d, failed to close file", pInfo->config.vgId);	×
201	}
202	taosMemoryFree(data);	22,689!
203	TAOS_RETURN(code);	22,689✔
204	}
205
206	int vnodeCommitInfo(const char *dir) {	22,689✔
207	char fname[TSDB_FILENAME_LEN];
208	char tfname[TSDB_FILENAME_LEN];
209
210	snprintf(fname, TSDB_FILENAME_LEN, "%s%s%s", dir, TD_DIRSEP, VND_INFO_FNAME);	22,689✔
211	snprintf(tfname, TSDB_FILENAME_LEN, "%s%s%s", dir, TD_DIRSEP, VND_INFO_FNAME_TMP);	22,689✔
212
213	int32_t code = taosRenameFile(tfname, fname);	22,689✔
214	if (code < 0) {	22,688!
215	return code;	×
216	}
217
218	vInfo("vnode info is committed, dir:%s", dir);	22,688✔
219	return 0;	22,689✔
220	}
221
222	int vnodeLoadInfo(const char dir, SVnodeInfo pInfo) {	17,864✔
223	int32_t code = 0;	17,864✔
224	int32_t lino;
225	char fname[TSDB_FILENAME_LEN];
226	TdFilePtr pFile = NULL;	17,864✔
227	char *pData = NULL;	17,864✔
228	int64_t size;
229
230	snprintf(fname, TSDB_FILENAME_LEN, "%s%s%s", dir, TD_DIRSEP, VND_INFO_FNAME);	17,864✔
231
232	// read info
233	pFile = taosOpenFile(fname, TD_FILE_READ);	17,864✔
234	if (pFile == NULL) {	17,865✔
235	TSDB_CHECK_CODE(code = terrno, lino, _exit);	6,904!
236	}
237
238	code = taosFStatFile(pFile, &size, NULL);	10,961✔
239	TSDB_CHECK_CODE(code, lino, _exit);	10,960!
240
241	pData = taosMemoryMalloc(size + 1);	10,960!
242	if (pData == NULL) {	10,960!
243	TSDB_CHECK_CODE(code = terrno, lino, _exit);	×
244	}
245
246	if (taosReadFile(pFile, pData, size) < 0) {	10,960!
247	TSDB_CHECK_CODE(code = terrno, lino, _exit);	×
248	}
249
250	pData[size] = '\0';	10,955✔
251
252	// decode info
253	code = vnodeDecodeInfo(pData, pInfo);	10,955✔
254	TSDB_CHECK_CODE(code, lino, _exit);	10,959!
255
256	pInfo->config.walCfg.committed = pInfo->state.committed;	10,959✔
257	_exit:	17,863✔
258	if (code) {	17,863✔
259	vError("vgId:%d %s failed at %s:%d since %s, file:%s", pInfo->config.vgId, __func__, __FILE__, lino,	6,904✔
260	tstrerror(code), fname);
261	}
262	taosMemoryFree(pData);	17,863!
263	if (taosCloseFile(&pFile) != 0) {	17,864!
264	vError("vgId:%d, failed to close file", pInfo->config.vgId);	×
265	}
266	return code;	17,865✔
267	}
268
269	static int32_t vnodePrepareCommit(SVnode pVnode, SCommitInfo pInfo) {	14,240✔
270	int32_t code = 0;	14,240✔
271	int32_t lino = 0;	14,240✔
272	char dir[TSDB_FILENAME_LEN] = {0};	14,240✔
273	int64_t lastCommitted = pInfo->info.state.committed;	14,240✔
274
275	// wait last commit task
276	vnodeAWait(&pVnode->commitTask);	14,240✔
277
278	code = syncNodeGetConfig(pVnode->sync, &pVnode->config.syncCfg);	14,240✔
279	TSDB_CHECK_CODE(code, lino, _exit);	14,242!
280
281	pVnode->state.commitTerm = pVnode->state.applyTerm;	14,242✔
282
283	pInfo->info.config = pVnode->config;	14,242✔
284	pInfo->info.state.committed = pVnode->state.applied;	14,242✔
285	pInfo->info.state.commitTerm = pVnode->state.applyTerm;	14,242✔
286	pInfo->info.state.commitID = ++pVnode->state.commitID;	14,242✔
287	pInfo->pVnode = pVnode;	14,242✔
288	pInfo->txn = metaGetTxn(pVnode->pMeta);	14,242✔
289
290	// save info
291	vnodeGetPrimaryPath(pVnode, false, dir, TSDB_FILENAME_LEN);	14,241✔
292
293	vDebug("vgId:%d, save config while prepare commit", TD_VID(pVnode));	14,240✔
294	code = vnodeSaveInfo(dir, &pInfo->info);	14,242✔
295	TSDB_CHECK_CODE(code, lino, _exit);	14,242!
296
297	code = tsdbPreCommit(pVnode->pTsdb);	14,242✔
298	TSDB_CHECK_CODE(code, lino, _exit);	14,241!
299
300	code = metaPrepareAsyncCommit(pVnode->pMeta);	14,241✔
301	TSDB_CHECK_CODE(code, lino, _exit);	14,239!
302
303	(void)taosThreadMutexLock(&pVnode->mutex);	14,239✔
304	pVnode->onCommit = pVnode->inUse;	14,241✔
305	pVnode->inUse = NULL;	14,241✔
306	(void)taosThreadMutexUnlock(&pVnode->mutex);	14,241✔
307
308	_exit:	14,240✔
309	if (code) {	14,240!
310	vError("vgId:%d, %s failed at line %d since %s, commit id:%" PRId64, TD_VID(pVnode), __func__, lino,	×
311	tstrerror(code), pVnode->state.commitID);
312	} else {
313	vDebug("vgId:%d, %s done, commit id:%" PRId64, TD_VID(pVnode), __func__, pInfo->info.state.commitID);	14,240✔
314	}
315
316	return code;	14,242✔
317	}
318	static void vnodeReturnBufPool(SVnode *pVnode) {	14,242✔
319	(void)taosThreadMutexLock(&pVnode->mutex);	14,242✔
320
321	SVBufPool *pPool = pVnode->onCommit;	14,242✔
322	int32_t nRef = atomic_sub_fetch_32(&pPool->nRef, 1);	14,242✔
323
324	pVnode->onCommit = NULL;	14,242✔
325	if (nRef == 0) {	14,242✔
326	vnodeBufPoolAddToFreeList(pPool);	14,170✔
327	} else if (nRef > 0) {	72!
328	vDebug("vgId:%d, buffer pool %p of id %d is added to recycle queue", TD_VID(pVnode), pPool, pPool->id);	72✔
329
330	if (pVnode->recycleTail == NULL) {	72!
331	pPool->recyclePrev = pPool->recycleNext = NULL;	72✔
332	pVnode->recycleHead = pVnode->recycleTail = pPool;	72✔
333	} else {
334	pPool->recyclePrev = pVnode->recycleTail;	×
335	pPool->recycleNext = NULL;	×
336	pVnode->recycleTail->recycleNext = pPool;	×
337	pVnode->recycleTail = pPool;	×
338	}
339	} else {
340	vError("vgId:%d, buffer pool %p of id %d nRef:%d", TD_VID(pVnode), pPool, pPool->id, nRef);	×
341	}
342
343	(void)taosThreadMutexUnlock(&pVnode->mutex);	14,242✔
344	}	14,242✔
345	static int32_t vnodeCommit(void *arg) {	14,242✔
346	int32_t code = 0;	14,242✔
347
348	SCommitInfo pInfo = (SCommitInfo )arg;	14,242✔
349	SVnode *pVnode = pInfo->pVnode;	14,242✔
350
351	// commit
352	METRICS_TIMING_BLOCK(pVnode->writeMetrics.commit_time, METRIC_LEVEL_HIGH, {	14,242!
353	if ((code = vnodeCommitImpl(pInfo))) {
354	vFatal("vgId:%d, failed to commit vnode since %s", TD_VID(pVnode), terrstr());
355	taosMsleep(100);
356	exit(EXIT_FAILURE);
357	goto _exit;
358	}
359	});
360
361	METRICS_UPDATE(pVnode->writeMetrics.commit_count, METRIC_LEVEL_HIGH, 1);	14,242!
362
363	vnodeReturnBufPool(pVnode);	14,242✔
364
365	_exit:	14,242✔
366	taosMemoryFree(arg);	14,242!
367	return code;	14,242✔
368	}
369
370	static void vnodeCommitCancel(void *arg) { taosMemoryFree(arg); }	×
371
372	int vnodeAsyncCommit(SVnode *pVnode) {	14,242✔
373	int32_t code = 0;	14,242✔
374	int32_t lino = 0;	14,242✔
375
376	SCommitInfo pInfo = (SCommitInfo )taosMemoryCalloc(1, sizeof(*pInfo));	14,242!
377	if (NULL == pInfo) {	14,242!
378	TSDB_CHECK_CODE(code = terrno, lino, _exit);	×
379	}
380	code = vnodePrepareCommit(pVnode, pInfo);	14,242✔
381	TSDB_CHECK_CODE(code, lino, _exit);	14,242!
382
383	// schedule the task
384	code = vnodeAsync(COMMIT_TASK_ASYNC, EVA_PRIORITY_HIGH, vnodeCommit, vnodeCommitCancel, pInfo, &pVnode->commitTask);	14,242✔
385	TSDB_CHECK_CODE(code, lino, _exit);	14,242!
386
387	_exit:	14,242✔
388	if (code) {	14,242!
389	taosMemoryFree(pInfo);	×
390	vError("vgId:%d %s failed at line %d since %s" PRId64, TD_VID(pVnode), __func__, lino, tstrerror(code));	×
391	} else {
392	vInfo("vgId:%d, vnode async commit done, commitId:%" PRId64 " term:%" PRId64 " applied:%" PRId64, TD_VID(pVnode),	14,242✔
393	pVnode->state.commitID, pVnode->state.applyTerm, pVnode->state.applied);
394	}
395	return code;	14,242✔
396	}
397
398	int32_t vnodeSyncCommit(SVnode *pVnode) {	158✔
399	int32_t lino;
400	int32_t code = vnodeAsyncCommit(pVnode);	158✔
401	TSDB_CHECK_CODE(code, lino, _exit);	158!
402	vnodeAWait(&pVnode->commitTask);	158✔
403
404	_exit:	158✔
405	if (code) {	158!
406	vError("vgId:%d, %s failed at line %d since %s", TD_VID(pVnode), __func__, lino, tstrerror(code));	×
407	} else {
408	vInfo("vgId:%d, sync commit end", TD_VID(pVnode));	158!
409	}
410
411	return code;	158✔
412	}
413
414	static int vnodeCommitImpl(SCommitInfo *pInfo) {	14,242✔
415	int32_t code = 0;	14,242✔
416	int32_t lino = 0;	14,242✔
417
418	char dir[TSDB_FILENAME_LEN] = {0};	14,242✔
419	SVnode *pVnode = pInfo->pVnode;	14,242✔
420
421	vInfo("vgId:%d, start to commit, commitId:%" PRId64 " version:%" PRId64 " term: %" PRId64, TD_VID(pVnode),	14,242✔
422	pInfo->info.state.commitID, pInfo->info.state.committed, pInfo->info.state.commitTerm);
423
424	// persist wal before starting
425	if ((code = walPersist(pVnode->pWal)) < 0) {	14,245!
426	vError("vgId:%d, failed to persist wal since %s", TD_VID(pVnode), tstrerror(code));	×
427	return code;	×
428	}
429
430	vnodeGetPrimaryPath(pVnode, false, dir, TSDB_FILENAME_LEN);	14,242✔
431
432	code = syncBeginSnapshot(pVnode->sync, pInfo->info.state.committed);	14,242✔
433	TSDB_CHECK_CODE(code, lino, _exit);	14,242!
434
435	code = tsdbCommitBegin(pVnode->pTsdb, pInfo);	14,242✔
436	TSDB_CHECK_CODE(code, lino, _exit);	14,242!
437
438	if (!TSDB_CACHE_NO(pVnode->config)) {	14,242✔
439	METRICS_TIMING_BLOCK(pVnode->writeMetrics.last_cache_commit_time, METRIC_LEVEL_HIGH,	111!
440	{ code = tsdbCacheCommit(pVnode->pTsdb); });
441	METRICS_UPDATE(pVnode->writeMetrics.last_cache_commit_count, METRIC_LEVEL_HIGH, 1);	111!
442	TSDB_CHECK_CODE(code, lino, _exit);	111!
443	}
444
445	// blob storage engine commit
446	code = bseCommit(pVnode->pBse);	14,242✔
447	// commit info
448	code = vnodeCommitInfo(dir);	14,242✔
449	TSDB_CHECK_CODE(code, lino, _exit);	14,242!
450
451	code = tsdbCommitCommit(pVnode->pTsdb);	14,242✔
452	TSDB_CHECK_CODE(code, lino, _exit);	14,242!
453
454	code = metaFinishCommit(pVnode->pMeta, pInfo->txn);	14,242✔
455	TSDB_CHECK_CODE(code, lino, _exit);	14,242!
456
457	pVnode->state.committed = pInfo->info.state.committed;	14,242✔
458
459	code = syncEndSnapshot(pVnode->sync);	14,242✔
460	TSDB_CHECK_CODE(code, lino, _exit);	14,242!
461
462	code = tqCommitOffset(pVnode->pTq);	14,242✔
463	TSDB_CHECK_CODE(code, lino, _exit);	14,242!
464
465	_exit:	14,242✔
466	if (code) {	14,242!
467	vError("vgId:%d, %s failed at line %d since %s", TD_VID(pVnode), __func__, lino, tstrerror(code));	×
468	} else {
469	vInfo("vgId:%d, commit end", TD_VID(pVnode));	14,242✔
470	}
471	return code;	14,242✔
472	}
473
474	bool vnodeShouldRollback(SVnode *pVnode) {	9,501✔
475	char tFName[TSDB_FILENAME_LEN] = {0};	9,501✔
476	int32_t offset = 0;	9,501✔
477
478	vnodeGetPrimaryPath(pVnode, false, tFName, TSDB_FILENAME_LEN);	9,501✔
479	offset = strlen(tFName);	9,501✔
480	snprintf(tFName + offset, TSDB_FILENAME_LEN - offset - 1, "%s%s", TD_DIRSEP, VND_INFO_FNAME_TMP);	9,501✔
481
482	return taosCheckExistFile(tFName);	9,501✔
483	}
484
485	void vnodeRollback(SVnode *pVnode) {	×
486	char tFName[TSDB_FILENAME_LEN] = {0};	×
487	int32_t offset = 0;	×
488
489	vnodeGetPrimaryPath(pVnode, false, tFName, TSDB_FILENAME_LEN);	×
490	offset = strlen(tFName);	×
491	snprintf(tFName + offset, TSDB_FILENAME_LEN - offset - 1, "%s%s", TD_DIRSEP, VND_INFO_FNAME_TMP);	×
492
493	if (taosRemoveFile(tFName) != 0) {	×
494	vError("vgId:%d, failed to remove file %s since %s", TD_VID(pVnode), tFName, tstrerror(terrno));	×
495	}
496	}	×
497
498	static int vnodeEncodeState(const void pObj, SJson pJson) {	22,687✔
499	const SVState pState = (SVState )pObj;	22,687✔
500
501	TAOS_CHECK_RETURN(tjsonAddIntegerToObject(pJson, "commit version", pState->committed));	22,687!
502	TAOS_CHECK_RETURN(tjsonAddIntegerToObject(pJson, "commit ID", pState->commitID));	22,688!
503	TAOS_CHECK_RETURN(tjsonAddIntegerToObject(pJson, "commit term", pState->commitTerm));	22,689!
504
505	return 0;	22,689✔
506	}
507
508	static int vnodeDecodeState(const SJson pJson, void pObj) {	11,049✔
509	SVState pState = (SVState )pObj;	11,049✔
510
511	int32_t code;
512	tjsonGetNumberValue(pJson, "commit version", pState->committed, code);	11,049✔
513	if (code) return code;	11,050!
514	tjsonGetNumberValue(pJson, "commit ID", pState->commitID, code);	11,050✔
515	if (code) return code;	11,051!
516	tjsonGetNumberValue(pJson, "commit term", pState->commitTerm, code);	11,051✔
517	if (code) return code;	11,052!
518
519	return 0;	11,052✔
520	}
521
522	static int vnodeEncodeInfo(const SVnodeInfo pInfo, char *ppData) {	22,689✔
523	int32_t code = 0;	22,689✔
524	int32_t lino;
525	SJson *pJson = NULL;	22,689✔
526	char *pData = NULL;	22,689✔
527
528	pJson = tjsonCreateObject();	22,689✔
529	if (pJson == NULL) {	22,689!
530	TSDB_CHECK_CODE(code = terrno, lino, _exit);	×
531	}
532
533	code = tjsonAddObject(pJson, "config", vnodeEncodeConfig, (void *)&pInfo->config);	22,689✔
534	TSDB_CHECK_CODE(code, lino, _exit);	22,686!
535
536	code = tjsonAddObject(pJson, "state", vnodeEncodeState, (void *)&pInfo->state);	22,686✔
537	TSDB_CHECK_CODE(code, lino, _exit);	22,688!
538
539	pData = tjsonToString(pJson);	22,688✔
540	if (pData == NULL) {	22,684!
541	TSDB_CHECK_CODE(code = terrno, lino, _exit);	×
542	}
543
544	tjsonDelete(pJson);	22,684✔
545
546	_exit:	22,686✔
547	if (code) {	22,686!
548	tjsonDelete(pJson);	×
549	*ppData = NULL;	×
550	} else {
551	*ppData = pData;	22,686✔
552	}
553	return code;	22,686✔
554	}
555
556	int vnodeDecodeInfo(uint8_t pData, SVnodeInfo pInfo) {	11,046✔
557	int32_t code = 0;	11,046✔
558	int32_t lino;
559	SJson *pJson = NULL;	11,046✔
560
561	pJson = tjsonParse(pData);	11,046✔
562	if (pJson == NULL) {	11,043!
563	TSDB_CHECK_CODE(code = TSDB_CODE_INVALID_DATA_FMT, lino, _exit);	×
564	}
565
566	code = tjsonToObject(pJson, "config", vnodeDecodeConfig, (void *)&pInfo->config);	11,043✔
567	TSDB_CHECK_CODE(code, lino, _exit);	11,050!
568
569	code = tjsonToObject(pJson, "state", vnodeDecodeState, (void *)&pInfo->state);	11,050✔
570	TSDB_CHECK_CODE(code, lino, _exit);	11,050!
571
572	_exit:	11,050✔
573	tjsonDelete(pJson);	11,050✔
574	return code;	11,051✔
575	}

taosdata / TDengine / #4822

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

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