#3542

Committed 27 Nov 2024 02:52AM UTC coverage: 60.819% (+0.04%) from 60.776%

Build # #3542

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

Merge pull request #28931 from taosdata/enh/jdbc-demo-3.0

update jdbc demo, and version history

Run Details

120305 of 252779 branches covered (47.59%)

Branch coverage included in aggregate %.

201010 of 275538 relevant lines covered (72.95%)

19989893.51 hits per line

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

66.27

/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);
        if (code && code != TSDB_CODE_TIMEOUT_ERROR) {
          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);

  // begin sma
  if (VND_IS_RSMA(pVnode)) {
    code = smaBegin(pVnode->pSma);
    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);
  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) {
    if (pFile) {
      vError("vgId:%d %s failed at %s:%d since %s", pInfo->config.vgId, __func__, __FILE__, lino, tstrerror(code));
    }
  }
  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
  vnodeGetPrimaryDir(pVnode->path, pVnode->diskPrimary, pVnode->pTfs, 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);

  code = smaPrepareAsyncCommit(pVnode->pSma);
  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
  if ((code = vnodeCommitImpl(pInfo))) {
    vFatal("vgId:%d, failed to commit vnode since %s", TD_VID(pVnode), terrstr());
    taosMsleep(100);
    exit(EXIT_FAILURE);
    goto _exit;
  }

  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);
  }

  // prepare to commit
  code = vnodePrepareCommit(pVnode, pInfo);
  TSDB_CHECK_CODE(code, lino, _exit);

  // schedule the task
  code =
      vnodeAsync(&pVnode->commitChannel, 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;
  }

  vnodeGetPrimaryDir(pVnode->path, pVnode->diskPrimary, pVnode->pTfs, 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)) {
    code = tsdbCacheCommit(pVnode->pTsdb);
    TSDB_CHECK_CODE(code, lino, _exit);
  }

  if (VND_IS_RSMA(pVnode)) {
    code = smaCommit(pVnode->pSma, pInfo);
    TSDB_CHECK_CODE(code, lino, _exit);
  }

  // commit info
  code = vnodeCommitInfo(dir);
  TSDB_CHECK_CODE(code, lino, _exit);

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

  if (VND_IS_RSMA(pVnode)) {
    code = smaFinishCommit(pVnode->pSma);
    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;

  if (smaPostCommit(pVnode->pSma) < 0) {
    vError("vgId:%d, failed to post-commit sma since %s", TD_VID(pVnode), tstrerror(terrno));
    return -1;
  }

  code = syncEndSnapshot(pVnode->sync);
  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;

  vnodeGetPrimaryDir(pVnode->path, pVnode->diskPrimary, pVnode->pTfs, 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;

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

taosdata / TDengine / #3542

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