#3386

Committed 26 Mar 2024 09:11AM UTC coverage: 65.961% (+45.8%) from 20.168%

Build # #3386

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

Merge pull request #25150 from taosdata/fix/rpc_portMainTo30

Fix/rpc port main to30

Run Details

91651 of 173149 branches covered (52.93%)

Branch coverage included in aggregate %.

17 of 20 new or added lines in 1 file covered. (85.0%)

739 existing lines in 23 files now uncovered.

163129 of 213107 relevant lines covered (76.55%)

2676225.01 hits per line

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

55.6

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

  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;
        taosGetTimeOfDay(&tv);
        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;
        }

        int32_t rc = taosThreadCondTimedWait(&pVnode->poolNotEmpty, &pVnode->mutex, &ts);
        if (rc && rc != ETIMEDOUT) {
          code = TAOS_SYSTEM_ERROR(rc);
          TSDB_CHECK_CODE(code, lino, _exit);
        }
      }
    }
  }

_exit:
  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
  if (metaBegin(pVnode->pMeta, META_BEGIN_HEAP_BUFFERPOOL) < 0) {
    code = terrno;
    TSDB_CHECK_CODE(code, lino, _exit);
  }

  // begin tsdb
  if (tsdbBegin(pVnode->pTsdb) < 0) {
    code = terrno;
    TSDB_CHECK_CODE(code, lino, _exit);
  }

  // begin sma
  if (VND_IS_RSMA(pVnode) && smaBegin(pVnode->pSma) < 0) {
    code = terrno;
    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;

  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));
  }
  taosThreadMutexUnlock(&pVnode->mutex);
  return needCommit;
}

int vnodeSaveInfo(const char *dir, const SVnodeInfo *pInfo) {
  char      fname[TSDB_FILENAME_LEN];
  TdFilePtr pFile;
  char     *data;

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

  // encode info
  data = NULL;

  if (vnodeEncodeInfo(pInfo, &data) < 0) {
    vError("failed to encode json info.");
    return -1;
  }

  // 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) {
    vError("failed to open info file:%s for write:%s", fname, terrstr());
    terrno = TAOS_SYSTEM_ERROR(errno);
    return -1;
  }

  if (taosWriteFile(pFile, data, strlen(data)) < 0) {
    vError("failed to write info file:%s error:%s", fname, terrstr());
    terrno = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

  if (taosFsyncFile(pFile) < 0) {
    vError("failed to fsync info file:%s error:%s", fname, terrstr());
    terrno = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

  taosCloseFile(&pFile);

  // free info binary
  taosMemoryFree(data);

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

  return 0;

_err:
  taosCloseFile(&pFile);
  taosMemoryFree(data);
  return -1;
}

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

  if (taosRenameFile(tfname, fname) < 0) {
    terrno = TAOS_SYSTEM_ERROR(errno);
    return -1;
  }

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

int vnodeLoadInfo(const char *dir, SVnodeInfo *pInfo) {
  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) {
    terrno = TAOS_SYSTEM_ERROR(errno);
    return -1;
  }

  if (taosFStatFile(pFile, &size, NULL) < 0) {
    terrno = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

  pData = taosMemoryMalloc(size + 1);
  if (pData == NULL) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    goto _err;
  }

  if (taosReadFile(pFile, pData, size) < 0) {
    terrno = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

  pData[size] = '\0';

  taosCloseFile(&pFile);

  // decode info
  if (vnodeDecodeInfo(pData, pInfo) < 0) {
    taosMemoryFree(pData);
    return -1;
  }

  taosMemoryFree(pData);

  return 0;

_err:
  taosCloseFile(&pFile);
  taosMemoryFree(pData);
  return -1;
}

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(vnodeAsyncHandle[0], pVnode->commitTask);

  if (syncNodeGetConfig(pVnode->sync, &pVnode->config.syncCfg) != 0) goto _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));
  if (vnodeSaveInfo(dir, &pInfo->info) < 0) {
    code = terrno;
    TSDB_CHECK_CODE(code, lino, _exit);
  }

  tsdbPreCommit(pVnode->pTsdb);

  metaPrepareAsyncCommit(pVnode->pMeta);

  code = smaPrepareAsyncCommit(pVnode->pSma);
  if (code) goto _exit;

  taosThreadMutexLock(&pVnode->mutex);
  ASSERT(pVnode->onCommit == NULL);
  pVnode->onCommit = pVnode->inUse;
  pVnode->inUse = NULL;
  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) {
  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 {
    ASSERT(0);
  }

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

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

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

  vnodeReturnBufPool(pVnode);

_exit:
  return code;
}

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

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

  SCommitInfo *pInfo = (SCommitInfo *)taosMemoryCalloc(1, sizeof(*pInfo));
  if (NULL == pInfo) {
    code = TSDB_CODE_OUT_OF_MEMORY;
    goto _exit;
  }

  // prepare to commit
  code = vnodePrepareCommit(pVnode, pInfo);
  if (TSDB_CODE_SUCCESS != code) {
    goto _exit;
  }

  // schedule the task
  code = vnodeAsyncC(vnodeAsyncHandle[0], pVnode->commitChannel, EVA_PRIORITY_HIGH, vnodeCommitTask,
                     vnodeCompleteCommit, pInfo, &pVnode->commitTask);

_exit:
  if (code) {
    if (NULL != pInfo) {
      taosMemoryFree(pInfo);
    }
    vError("vgId:%d, %s failed since %s, commit id:%" PRId64, TD_VID(pVnode), __func__, tstrerror(code),
           pVnode->state.commitID);
  } 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;
}

int vnodeSyncCommit(SVnode *pVnode) {
  vnodeAsyncCommit(pVnode);
  vnodeAWait(vnodeAsyncHandle[0], pVnode->commitTask);
  return 0;
}

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 (walPersist(pVnode->pWal) < 0) {
    vError("vgId:%d, failed to persist wal since %s", TD_VID(pVnode), terrstr());
    return -1;
  }

  vnodeGetPrimaryDir(pVnode->path, pVnode->diskPrimary, pVnode->pTfs, dir, TSDB_FILENAME_LEN);

  syncBeginSnapshot(pVnode->sync, pInfo->info.state.committed);

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

  if (tqCommit(pVnode->pTq) < 0) {
    code = TSDB_CODE_FAILED;
    TSDB_CHECK_CODE(code, lino, _exit);
  }

  // commit info
  if (vnodeCommitInfo(dir) < 0) {
    code = terrno;
    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);
  }

  if (metaFinishCommit(pVnode->pMeta, pInfo->txn) < 0) {
    code = terrno;
    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;
  }

  syncEndSnapshot(pVnode->sync);

_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 0;
}

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

  (void)taosRemoveFile(tFName);
}

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

  if (tjsonAddIntegerToObject(pJson, "commit version", pState->committed) < 0) return -1;
  if (tjsonAddIntegerToObject(pJson, "commit ID", pState->commitID) < 0) return -1;
  if (tjsonAddIntegerToObject(pJson, "commit term", pState->commitTerm) < 0) return -1;

  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 < 0) return -1;
  tjsonGetNumberValue(pJson, "commit ID", pState->commitID, code);
  if (code < 0) return -1;
  tjsonGetNumberValue(pJson, "commit term", pState->commitTerm, code);
  if (code < 0) return -1;

  return 0;
}

static int vnodeEncodeInfo(const SVnodeInfo *pInfo, char **ppData) {
  SJson *pJson;
  char  *pData;

  *ppData = NULL;

  pJson = tjsonCreateObject();
  if (pJson == NULL) {
    return -1;
  }

  if (tjsonAddObject(pJson, "config", vnodeEncodeConfig, (void *)&pInfo->config) < 0) {
    goto _err;
  }

  if (tjsonAddObject(pJson, "state", vnodeEncodeState, (void *)&pInfo->state) < 0) {
    goto _err;
  }

  pData = tjsonToString(pJson);
  if (pData == NULL) {
    goto _err;
  }

  tjsonDelete(pJson);

  *ppData = pData;
  return 0;

_err:
  tjsonDelete(pJson);
  return -1;
}

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

  pJson = tjsonParse(pData);
  if (pJson == NULL) {
    return -1;
  }

  if (tjsonToObject(pJson, "config", vnodeDecodeConfig, (void *)&pInfo->config) < 0) {
    goto _err;
  }

  if (tjsonToObject(pJson, "state", vnodeDecodeState, (void *)&pInfo->state) < 0) {
    goto _err;
  }

  tjsonDelete(pJson);

  return 0;

_err:
  tjsonDelete(pJson);
  return -1;
}

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

taosdata / TDengine / #3386

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