diff --git a/README-CN.md b/README-CN.md index 437d671bb97a1009f5d48d2423a3d17f75bf1494..a5d239a53296b5a7d96a3e1624879386ab57bf13 100644 --- a/README-CN.md +++ b/README-CN.md @@ -60,7 +60,7 @@ sudo apt-get install -y gcc cmake build-essential git libssl-dev 为了在 Ubuntu/Debian 系统上编译 [taos-tools](https://github.com/taosdata/taos-tools) 需要安装如下软件: ```bash -sudo apt install build-essential libjansson-dev libsnappy-dev liblzma-dev libz-dev pkg-config +sudo apt install build-essential libjansson-dev libsnappy-dev liblzma-dev libz-dev zlib1g pkg-config ``` ### CentOS 7.9 @@ -85,7 +85,7 @@ sudo dnf install -y gcc gcc-c++ make cmake epel-release git openssl-devel ``` -sudo yum install -y zlib-devel xz-devel snappy-devel jansson jansson-devel pkgconfig libatomic libstdc++-static openssl-devel +sudo yum install -y zlib-devel zlib-static xz-devel snappy-devel jansson jansson-devel pkgconfig libatomic libatomic-static libstdc++-static openssl-devel ``` #### CentOS 8/Rocky Linux @@ -94,7 +94,7 @@ sudo yum install -y zlib-devel xz-devel snappy-devel jansson jansson-devel pkgco sudo yum install -y epel-release sudo yum install -y dnf-plugins-core sudo yum config-manager --set-enabled powertools -sudo yum install -y zlib-devel xz-devel snappy-devel jansson jansson-devel pkgconfig libatomic libstdc++-static openssl-devel +sudo yum install -y zlib-devel zlib-static xz-devel snappy-devel jansson jansson-devel pkgconfig libatomic libatomic-static libstdc++-static openssl-devel ``` 注意:由于 snappy 缺乏 pkg-config 支持(参考 [链接](https://github.com/google/snappy/pull/86)),会导致 cmake 提示无法发现 libsnappy,实际上工作正常。 diff --git a/README.md b/README.md index 6aec756ec76d6801bfe72e3c66bc36a325d8245e..05c1c075f0c3b3894c93d0aa2e126a9dc6340758 100644 --- a/README.md +++ b/README.md @@ -62,7 +62,7 @@ sudo apt-get install -y gcc cmake build-essential git libssl-dev To build the [taosTools](https://github.com/taosdata/taos-tools) on Ubuntu/Debian, the following packages need to be installed. ```bash -sudo apt install build-essential libjansson-dev libsnappy-dev liblzma-dev libz-dev pkg-config +sudo apt install build-essential libjansson-dev libsnappy-dev liblzma-dev libz-dev zlib1g pkg-config ``` ### CentOS 7.9 @@ -85,7 +85,7 @@ sudo dnf install -y gcc gcc-c++ make cmake epel-release git openssl-devel #### CentOS 7.9 ``` -sudo yum install -y zlib-devel xz-devel snappy-devel jansson jansson-devel pkgconfig libatomic libstdc++-static openssl-devel +sudo yum install -y zlib-devel zlib-static xz-devel snappy-devel jansson jansson-devel pkgconfig libatomic libatomic-static libstdc++-static openssl-devel ``` #### CentOS 8/Rocky Linux @@ -94,7 +94,7 @@ sudo yum install -y zlib-devel xz-devel snappy-devel jansson jansson-devel pkgco sudo yum install -y epel-release sudo yum install -y dnf-plugins-core sudo yum config-manager --set-enabled powertools -sudo yum install -y zlib-devel xz-devel snappy-devel jansson jansson-devel pkgconfig libatomic libstdc++-static openssl-devel +sudo yum install -y zlib-devel zlib-static xz-devel snappy-devel jansson jansson-devel pkgconfig libatomic libatomic-static libstdc++-static openssl-devel ``` Note: Since snappy lacks pkg-config support (refer to [link](https://github.com/google/snappy/pull/86)), it leads a cmake prompt libsnappy not found. But snappy still works well. diff --git a/cmake/taostools_CMakeLists.txt.in b/cmake/taostools_CMakeLists.txt.in index 2d7bcf6592cddb101a53c7445f83a293a2fed0db..4abd9e86a6a4167886c24123314f5c14d2c9d2e0 100644 --- a/cmake/taostools_CMakeLists.txt.in +++ b/cmake/taostools_CMakeLists.txt.in @@ -2,7 +2,7 @@ # taos-tools ExternalProject_Add(taos-tools GIT_REPOSITORY https://github.com/taosdata/taos-tools.git - GIT_TAG e62c5ea + GIT_TAG ac69142 SOURCE_DIR "${TD_SOURCE_DIR}/tools/taos-tools" BINARY_DIR "" #BUILD_IN_SOURCE TRUE diff --git a/include/common/tmsg.h b/include/common/tmsg.h index 470b7bba7e738e1a4100e94e466e723e702f2424..bb1addf1b698479ca418882a1a16c3c59e5347cc 100644 --- a/include/common/tmsg.h +++ b/include/common/tmsg.h @@ -70,6 +70,11 @@ static inline bool vnodeIsMsgBlock(tmsg_t type) { return (type == TDMT_VND_CREATE_TABLE) || (type == TDMT_VND_ALTER_TABLE) || (type == TDMT_VND_DROP_TABLE) || (type == TDMT_VND_UPDATE_TAG_VAL); } + +static inline bool syncUtilUserCommit(tmsg_t msgType) { + return msgType != TDMT_SYNC_NOOP && msgType != TDMT_SYNC_LEADER_TRANSFER; +} + /* ------------------------ OTHER DEFINITIONS ------------------------ */ // IE type #define TSDB_IE_TYPE_SEC 1 diff --git a/include/libs/nodes/nodes.h b/include/libs/nodes/nodes.h index f94e0c98dc2d4614a8d743f6a7ab07465b46b700..412054b13e3bdceec6234919040c4b99a43d6102 100644 --- a/include/libs/nodes/nodes.h +++ b/include/libs/nodes/nodes.h @@ -234,7 +234,6 @@ typedef enum ENodeType { QUERY_NODE_PHYSICAL_PLAN_SYSTABLE_SCAN, QUERY_NODE_PHYSICAL_PLAN_BLOCK_DIST_SCAN, QUERY_NODE_PHYSICAL_PLAN_LAST_ROW_SCAN, - QUERY_NODE_PHYSICAL_PLAN_TABLE_COUNT_SCAN, QUERY_NODE_PHYSICAL_PLAN_PROJECT, QUERY_NODE_PHYSICAL_PLAN_MERGE_JOIN, QUERY_NODE_PHYSICAL_PLAN_HASH_AGG, @@ -265,7 +264,8 @@ typedef enum ENodeType { QUERY_NODE_PHYSICAL_PLAN_QUERY_INSERT, QUERY_NODE_PHYSICAL_PLAN_DELETE, QUERY_NODE_PHYSICAL_SUBPLAN, - QUERY_NODE_PHYSICAL_PLAN + QUERY_NODE_PHYSICAL_PLAN, + QUERY_NODE_PHYSICAL_PLAN_TABLE_COUNT_SCAN } ENodeType; /** diff --git a/source/client/src/clientMain.c b/source/client/src/clientMain.c index 9f3c78aba2bd78632cfd8c007c4457c88445a485..f2dec7217f8adb3e1b01529687173be449c3275a 100644 --- a/source/client/src/clientMain.c +++ b/source/client/src/clientMain.c @@ -796,9 +796,10 @@ static void doAsyncQueryFromParse(SMetaData *pResultMeta, void *param, int32_t c SQuery *pQuery = pRequest->pQuery; pRequest->metric.ctgEnd = taosGetTimestampUs(); - qDebug("0x%" PRIx64 " start to continue parse, reqId:0x%" PRIx64, pRequest->self, pRequest->requestId); + qDebug("0x%" PRIx64 " start to continue parse, reqId:0x%" PRIx64 ", code:%s", pRequest->self, pRequest->requestId, tstrerror(code)); if (code == TSDB_CODE_SUCCESS) { + pWrapper->pCatalogReq->forceUpdate = false; code = qContinueParseSql(pWrapper->pParseCtx, pWrapper->pCatalogReq, pResultMeta, pQuery); } diff --git a/source/dnode/mgmt/node_mgmt/src/dmTransport.c b/source/dnode/mgmt/node_mgmt/src/dmTransport.c index 2383a0934376fbe5d744568d638d0e3931898590..99f8bd002a5944ec1f541c18ec523efab41e977e 100644 --- a/source/dnode/mgmt/node_mgmt/src/dmTransport.c +++ b/source/dnode/mgmt/node_mgmt/src/dmTransport.c @@ -301,6 +301,7 @@ int32_t dmInitServer(SDnode *pDnode) { rpcInit.connType = TAOS_CONN_SERVER; rpcInit.idleTime = tsShellActivityTimer * 1000; rpcInit.parent = pDnode; + rpcInit.compressSize = tsCompressMsgSize; pTrans->serverRpc = rpcOpen(&rpcInit); if (pTrans->serverRpc == NULL) { diff --git a/source/dnode/mnode/impl/src/mndSync.c b/source/dnode/mnode/impl/src/mndSync.c index 895d7fedc4ff2dff78ea3c81f262d04c7e4a3be3..7cba69ed50d64ec8dd5ebc06eaa0bc3438afba4a 100644 --- a/source/dnode/mnode/impl/src/mndSync.c +++ b/source/dnode/mnode/impl/src/mndSync.c @@ -119,7 +119,13 @@ int32_t mndProcessWriteMsg(const SSyncFSM *pFsm, SRpcMsg *pMsg, const SFsmCbMeta } int32_t mndSyncCommitMsg(const SSyncFSM *pFsm, SRpcMsg *pMsg, const SFsmCbMeta *pMeta) { - int32_t code = mndProcessWriteMsg(pFsm, pMsg, pMeta); + int32_t code = 0; + if (!syncUtilUserCommit(pMsg->msgType)) { + goto _out; + } + code = mndProcessWriteMsg(pFsm, pMsg, pMeta); + +_out: rpcFreeCont(pMsg->pCont); pMsg->pCont = NULL; return code; diff --git a/source/dnode/vnode/src/meta/metaCache.c b/source/dnode/vnode/src/meta/metaCache.c index 6a704d042577391ea382fc84e979f124e6250c62..0ed68dce952424d50a9966196024c9ce58067324 100644 --- a/source/dnode/vnode/src/meta/metaCache.c +++ b/source/dnode/vnode/src/meta/metaCache.c @@ -454,7 +454,7 @@ int32_t metaGetCachedTableUidList(SMeta* pMeta, tb_uid_t suid, const uint8_t* pK SListNode* pNode = NULL; while ((pNode = tdListNext(&iter)) != NULL) { - memcpy(pBuf + sizeof(suid), pNode->data, keyLen); + memcpy(&pBuf[1], pNode->data, keyLen); // check whether it is existed in LRU cache, and remove it from linked list if not. LRUHandle* pRes = taosLRUCacheLookup(pCache, pBuf, len); diff --git a/source/dnode/vnode/src/tsdb/tsdbCommit.c b/source/dnode/vnode/src/tsdb/tsdbCommit.c index 906e3b2638b44ef9a85fa7badb57b9dc3549219e..8ec59ea95949a4bccda38c7cc2fda25e2f769fb4 100644 --- a/source/dnode/vnode/src/tsdb/tsdbCommit.c +++ b/source/dnode/vnode/src/tsdb/tsdbCommit.c @@ -53,6 +53,7 @@ typedef struct { // -------------- TSKEY nextKey; // reset by each table commit int32_t commitFid; + int32_t expLevel; TSKEY minKey; TSKEY maxKey; // commit file data @@ -503,6 +504,7 @@ static int32_t tsdbCommitFileDataStart(SCommitter *pCommitter) { // memory pCommitter->commitFid = tsdbKeyFid(pCommitter->nextKey, pCommitter->minutes, pCommitter->precision); + pCommitter->expLevel = tsdbFidLevel(pCommitter->commitFid, &pCommitter->pTsdb->keepCfg, taosGetTimestampSec()); tsdbFidKeyRange(pCommitter->commitFid, pCommitter->minutes, pCommitter->precision, &pCommitter->minKey, &pCommitter->maxKey); #if 0 @@ -556,7 +558,10 @@ static int32_t tsdbCommitFileDataStart(SCommitter *pCommitter) { } } else { SDiskID did = {0}; - tfsAllocDisk(pTsdb->pVnode->pTfs, 0, &did); + if (tfsAllocDisk(pTsdb->pVnode->pTfs, pCommitter->expLevel, &did) < 0) { + code = terrno; + TSDB_CHECK_CODE(code, lino, _exit); + } tfsMkdirRecurAt(pTsdb->pVnode->pTfs, pTsdb->path, did); wSet.diskId = did; wSet.nSttF = 1; diff --git a/source/dnode/vnode/src/vnd/vnodeSvr.c b/source/dnode/vnode/src/vnd/vnodeSvr.c index d8c8a3e1b2de6f01f6901b834c76ff23c58aecbb..0fc42f3744194f7695ff4806c6260ff7082b9c5c 100644 --- a/source/dnode/vnode/src/vnd/vnodeSvr.c +++ b/source/dnode/vnode/src/vnd/vnodeSvr.c @@ -190,9 +190,13 @@ int32_t vnodeProcessWriteMsg(SVnode *pVnode, SRpcMsg *pMsg, int64_t version, SRp version); ASSERT(pVnode->state.applyTerm <= pMsg->info.conn.applyTerm); + ASSERT(pVnode->state.applied + 1 == version); + pVnode->state.applied = version; pVnode->state.applyTerm = pMsg->info.conn.applyTerm; + if (!syncUtilUserCommit(pMsg->msgType)) goto _exit; + // skip header pReq = POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead)); len = pMsg->contLen - sizeof(SMsgHead); diff --git a/source/libs/catalog/inc/catalogInt.h b/source/libs/catalog/inc/catalogInt.h index ba9990658955f29b437ae0f17c36e24b3a6e84c3..c435c46d467f26bd4818f8074a4d90a263fe0e21 100644 --- a/source/libs/catalog/inc/catalogInt.h +++ b/source/libs/catalog/inc/catalogInt.h @@ -547,6 +547,14 @@ typedef struct SCtgOperation { #define ctgDebug(param, ...) qDebug("CTG:%p " param, pCtg, __VA_ARGS__) #define ctgTrace(param, ...) qTrace("CTG:%p " param, pCtg, __VA_ARGS__) +#define ctgTaskFatal(param, ...) qFatal("QID:%" PRIx64 " CTG:%p " param, pTask->pJob->queryId, pCtg, __VA_ARGS__) +#define ctgTaskError(param, ...) qError("QID:%" PRIx64 " CTG:%p " param, pTask->pJob->queryId, pCtg, __VA_ARGS__) +#define ctgTaskWarn(param, ...) qWarn("QID:%" PRIx64 " CTG:%p " param, pTask->pJob->queryId, pCtg, __VA_ARGS__) +#define ctgTaskInfo(param, ...) qInfo("QID:%" PRIx64 " CTG:%p " param, pTask->pJob->queryId, pCtg, __VA_ARGS__) +#define ctgTaskDebug(param, ...) qDebug("QID:%" PRIx64 " CTG:%p " param, pTask->pJob->queryId, pCtg, __VA_ARGS__) +#define ctgTaskTrace(param, ...) qTrace("QID:%" PRIx64 " CTG:%p " param, pTask->pJob->queryId, pCtg, __VA_ARGS__) + + #define CTG_LOCK_DEBUG(...) \ do { \ if (gCTGDebug.lockEnable) { \ diff --git a/source/libs/catalog/src/ctgAsync.c b/source/libs/catalog/src/ctgAsync.c index 7138e0ce16a4c3543bf145251640447fd986e54d..5b01ac1fb96007234863d7c96f7aaa1e75b3c962 100644 --- a/source/libs/catalog/src/ctgAsync.c +++ b/source/libs/catalog/src/ctgAsync.c @@ -1094,6 +1094,9 @@ _return: ctgReleaseVgInfoToCache(pCtg, dbCache); } + if (code) { + ctgTaskError("Get table %d.%s.%s meta failed with error %s", pName->acctId, pName->dbname, pName->tname, tstrerror(code)); + } if (pTask->res || code) { ctgHandleTaskEnd(pTask, code); } @@ -1124,7 +1127,7 @@ int32_t ctgHandleGetTbMetasRsp(SCtgTaskReq* tReq, int32_t reqType, const SDataBu SVgroupInfo vgInfo = {0}; CTG_ERR_JRET(ctgGetVgInfoFromHashValue(pCtg, pOut->dbVgroup, pName, &vgInfo)); - ctgDebug("will refresh tbmeta, not supposed to be stb, tbName:%s, flag:%d", tNameGetTableName(pName), flag); + ctgTaskDebug("will refresh tbmeta, not supposed to be stb, tbName:%s, flag:%d", tNameGetTableName(pName), flag); *vgId = vgInfo.vgId; CTG_ERR_JRET(ctgGetTbMetaFromVnode(pCtg, pConn, pName, &vgInfo, NULL, tReq)); @@ -1144,7 +1147,7 @@ int32_t ctgHandleGetTbMetasRsp(SCtgTaskReq* tReq, int32_t reqType, const SDataBu SVgroupInfo vgInfo = {0}; CTG_ERR_JRET(ctgGetVgInfoFromHashValue(pCtg, dbCache->vgCache.vgInfo, pName, &vgInfo)); - ctgDebug("will refresh tbmeta, supposed to be stb, tbName:%s, flag:%d", tNameGetTableName(pName), flag); + ctgTaskDebug("will refresh tbmeta, supposed to be stb, tbName:%s, flag:%d", tNameGetTableName(pName), flag); *vgId = vgInfo.vgId; CTG_ERR_JRET(ctgGetTbMetaFromVnode(pCtg, pConn, pName, &vgInfo, NULL, tReq)); @@ -1162,7 +1165,7 @@ int32_t ctgHandleGetTbMetasRsp(SCtgTaskReq* tReq, int32_t reqType, const SDataBu return TSDB_CODE_SUCCESS; } - ctgError("no tbmeta got, tbName:%s", tNameGetTableName(pName)); + ctgTaskError("no tbmeta got, tbName:%s", tNameGetTableName(pName)); ctgRemoveTbMetaFromCache(pCtg, pName, false); CTG_ERR_JRET(CTG_ERR_CODE_TABLE_NOT_EXIST); @@ -1180,7 +1183,7 @@ int32_t ctgHandleGetTbMetasRsp(SCtgTaskReq* tReq, int32_t reqType, const SDataBu STableMetaOutput* pOut = (STableMetaOutput*)pMsgCtx->out; if (CTG_IS_META_NULL(pOut->metaType)) { - ctgError("no tbmeta got, tbNmae:%s", tNameGetTableName(pName)); + ctgTaskError("no tbmeta got, tbNmae:%s", tNameGetTableName(pName)); ctgRemoveTbMetaFromCache(pCtg, pName, false); CTG_ERR_JRET(CTG_ERR_CODE_TABLE_NOT_EXIST); } @@ -1190,7 +1193,7 @@ int32_t ctgHandleGetTbMetasRsp(SCtgTaskReq* tReq, int32_t reqType, const SDataBu } if (CTG_IS_META_TABLE(pOut->metaType) && TSDB_SUPER_TABLE == pOut->tbMeta->tableType) { - ctgDebug("will continue to refresh tbmeta since got stb, tbName:%s", tNameGetTableName(pName)); + ctgTaskDebug("will continue to refresh tbmeta since got stb, tbName:%s", tNameGetTableName(pName)); taosMemoryFreeClear(pOut->tbMeta); @@ -1207,11 +1210,11 @@ int32_t ctgHandleGetTbMetasRsp(SCtgTaskReq* tReq, int32_t reqType, const SDataBu STableMeta* stbMeta = NULL; (void)ctgReadTbMetaFromCache(pCtg, &stbCtx, &stbMeta); if (stbMeta && stbMeta->sversion >= pOut->tbMeta->sversion) { - ctgDebug("use cached stb meta, tbName:%s", tNameGetTableName(pName)); + ctgTaskDebug("use cached stb meta, tbName:%s", tNameGetTableName(pName)); exist = 1; taosMemoryFreeClear(stbMeta); } else { - ctgDebug("need to get/update stb meta, tbName:%s", tNameGetTableName(pName)); + ctgTaskDebug("need to get/update stb meta, tbName:%s", tNameGetTableName(pName)); taosMemoryFreeClear(pOut->tbMeta); taosMemoryFreeClear(stbMeta); } @@ -1225,7 +1228,7 @@ int32_t ctgHandleGetTbMetasRsp(SCtgTaskReq* tReq, int32_t reqType, const SDataBu break; } default: - ctgError("invalid reqType %d", reqType); + ctgTaskError("invalid reqType %d", reqType); CTG_ERR_JRET(TSDB_CODE_INVALID_MSG); } @@ -1280,6 +1283,7 @@ _return: TSWAP(pTask->res, ctx->pResList); taskDone = true; } + ctgTaskError("Get table %d.%s.%s meta failed with error %s", pName->acctId, pName->dbname, pName->tname, tstrerror(code)); } if (pTask->res && taskDone) { diff --git a/source/libs/executor/src/timesliceoperator.c b/source/libs/executor/src/timesliceoperator.c index 2374d80bbf3f15430230f95177b4efe9beba2e4c..072a96fa832075dd8323c211d75ba5ca94a65b16 100644 --- a/source/libs/executor/src/timesliceoperator.c +++ b/source/libs/executor/src/timesliceoperator.c @@ -95,6 +95,8 @@ static void doKeepLinearInfo(STimeSliceOperatorInfo* pSliceInfo, const SSDataBlo // TODO: optimize to ignore null values for linear interpolation. if (!pLinearInfo->isStartSet) { if (!colDataIsNull_s(pColInfoData, rowIndex)) { + ASSERT(IS_MATHABLE_TYPE(pColInfoData->info.type)); + pLinearInfo->start.key = *(int64_t*)colDataGetData(pTsCol, rowIndex); memcpy(pLinearInfo->start.val, colDataGetData(pColInfoData, rowIndex), pLinearInfo->bytes); } diff --git a/source/libs/parser/src/parInsertSql.c b/source/libs/parser/src/parInsertSql.c index df7a3953f219497d9b3a15377c90cc12a633f08e..59545d8fbc1c1a46da78e747abfbf5363501fdf5 100644 --- a/source/libs/parser/src/parInsertSql.c +++ b/source/libs/parser/src/parInsertSql.c @@ -44,6 +44,7 @@ typedef struct SInsertParseContext { SParsedDataColInfo tags; // for stmt bool missCache; bool usingDuplicateTable; + bool forceUpdate; } SInsertParseContext; typedef int32_t (*_row_append_fn_t)(SMsgBuf* pMsgBuf, const void* value, int32_t len, void* param); @@ -829,6 +830,11 @@ static int32_t getTableVgroup(SParseContext* pCxt, SVnodeModifOpStmt* pStmt, boo } static int32_t getTargetTableSchema(SInsertParseContext* pCxt, SVnodeModifOpStmt* pStmt) { + if (pCxt->forceUpdate) { + pCxt->missCache = true; + return TSDB_CODE_SUCCESS; + } + int32_t code = checkAuth(pCxt->pComCxt, &pStmt->targetTableName, &pCxt->missCache); if (TSDB_CODE_SUCCESS == code && !pCxt->missCache) { code = getTableMeta(pCxt, &pStmt->targetTableName, false, &pStmt->pTableMeta, &pCxt->missCache); @@ -844,6 +850,11 @@ static int32_t preParseUsingTableName(SInsertParseContext* pCxt, SVnodeModifOpSt } static int32_t getUsingTableSchema(SInsertParseContext* pCxt, SVnodeModifOpStmt* pStmt) { + if (pCxt->forceUpdate) { + pCxt->missCache = true; + return TSDB_CODE_SUCCESS; + } + int32_t code = checkAuth(pCxt->pComCxt, &pStmt->targetTableName, &pCxt->missCache); if (TSDB_CODE_SUCCESS == code && !pCxt->missCache) { code = getTableMeta(pCxt, &pStmt->usingTableName, true, &pStmt->pTableMeta, &pCxt->missCache); @@ -1909,6 +1920,7 @@ int32_t parseInsertSql(SParseContext* pCxt, SQuery** pQuery, SCatalogReq* pCatal .msg = {.buf = pCxt->pMsg, .len = pCxt->msgLen}, .missCache = false, .usingDuplicateTable = false, + .forceUpdate = (NULL != pCatalogReq ? pCatalogReq->forceUpdate : false) }; int32_t code = initInsertQuery(&context, pCatalogReq, pMetaData, pQuery); diff --git a/source/libs/sync/inc/syncUtil.h b/source/libs/sync/inc/syncUtil.h index f198f3809d7eaa307a189487e74077009f64bb4d..be14ef91a40ef6b10c76a5bcf34ca8953e00bed1 100644 --- a/source/libs/sync/inc/syncUtil.h +++ b/source/libs/sync/inc/syncUtil.h @@ -79,7 +79,6 @@ char* syncUtilPrintBin2(char* ptr, uint32_t len); void syncUtilMsgHtoN(void* msg); void syncUtilMsgNtoH(void* msg); bool syncUtilUserPreCommit(tmsg_t msgType); -bool syncUtilUserCommit(tmsg_t msgType); bool syncUtilUserRollback(tmsg_t msgType); void syncPrintNodeLog(const char* flags, ELogLevel level, int32_t dflag, SSyncNode* pNode, const char* format, ...); diff --git a/source/libs/sync/src/syncPipeline.c b/source/libs/sync/src/syncPipeline.c index 5d6905f99e8a5b7f769ff2235dea1e4e64a6c256..f8fcdbddb318e722096908fa7114b470df8abdc0 100644 --- a/source/libs/sync/src/syncPipeline.c +++ b/source/libs/sync/src/syncPipeline.c @@ -513,13 +513,8 @@ int32_t syncLogBufferCommit(SSyncLogBuffer* pBuf, SSyncNode* pNode, int64_t comm if (!syncUtilUserCommit(pEntry->originalRpcType)) { sInfo("vgId:%d, commit sync barrier. index: %" PRId64 ", term:%" PRId64 ", type: %s", vgId, pEntry->index, pEntry->term, TMSG_INFO(pEntry->originalRpcType)); - pBuf->commitIndex = index; - if (!inBuf) { - syncEntryDestroy(pEntry); - pEntry = NULL; - } - continue; } + if (syncLogFsmExecute(pNode, pFsm, role, term, pEntry) != 0) { sError("vgId:%d, failed to execute sync log entry. index:%" PRId64 ", term:%" PRId64 ", role: %d, current term: %" PRId64, diff --git a/source/libs/sync/src/syncUtil.c b/source/libs/sync/src/syncUtil.c index 7787438dfe0850929f7472d3ee1e4338b22a80f8..22ff50dbbb31f35f04cbd0fe998dc132d34e54b2 100644 --- a/source/libs/sync/src/syncUtil.c +++ b/source/libs/sync/src/syncUtil.c @@ -160,8 +160,6 @@ void syncUtilMsgNtoH(void* msg) { bool syncUtilUserPreCommit(tmsg_t msgType) { return msgType != TDMT_SYNC_NOOP && msgType != TDMT_SYNC_LEADER_TRANSFER; } -bool syncUtilUserCommit(tmsg_t msgType) { return msgType != TDMT_SYNC_NOOP && msgType != TDMT_SYNC_LEADER_TRANSFER; } - bool syncUtilUserRollback(tmsg_t msgType) { return msgType != TDMT_SYNC_NOOP && msgType != TDMT_SYNC_LEADER_TRANSFER; } void syncCfg2SimpleStr(const SSyncCfg* pCfg, char* buf, int32_t bufLen) { diff --git a/source/libs/transport/src/transSvr.c b/source/libs/transport/src/transSvr.c index 73c6dc4011f5d1e4baf9e29f81246ff4b0b1f0c5..2b1f68d5f64952cbd933e3c20c1c507c2f2be966 100644 --- a/source/libs/transport/src/transSvr.c +++ b/source/libs/transport/src/transSvr.c @@ -1195,6 +1195,8 @@ void transCloseServer(void* arg) { sendQuitToWorkThrd(srv->pThreadObj[i]); destroyWorkThrd(srv->pThreadObj[i]); } + } else { + uv_loop_close(srv->loop); } taosMemoryFree(srv->pThreadObj);