From 57d341706bb4e7f23f2eb39f8e40b62040ed5d3e Mon Sep 17 00:00:00 2001 From: Liu Jicong Date: Fri, 7 Jan 2022 14:57:45 +0800 Subject: [PATCH 1/5] implement client heartbeat --- include/common/tmsg.h | 70 +++++++------ source/client/inc/clientHb.h | 28 ++---- source/client/src/clientHb.c | 77 -------------- source/client/src/clientHbMgmt.c | 166 +++++++++++++++++++++++++++++++ source/client/src/clientHbMq.c | 25 +++++ source/common/src/tmsg.c | 14 ++- source/libs/wal/src/walMgmt.c | 3 +- 7 files changed, 249 insertions(+), 134 deletions(-) delete mode 100644 source/client/src/clientHb.c create mode 100644 source/client/src/clientHbMgmt.c create mode 100644 source/client/src/clientHbMq.c diff --git a/include/common/tmsg.h b/include/common/tmsg.h index 954f24ef6a..d57fec3d5a 100644 --- a/include/common/tmsg.h +++ b/include/common/tmsg.h @@ -140,6 +140,46 @@ typedef struct SKlv { void* value; } SKlv; +typedef struct SClientHbKey { + int32_t connId; + int32_t hbType; +} SClientHbKey; + +typedef struct SClientHbReq { + SClientHbKey connKey; + SHashObj* info; // hash +} SClientHbReq; + +typedef struct SClientHbBatchReq { + int64_t reqId; + SArray* reqs; // SArray +} SClientHbBatchReq; + +typedef struct SClientHbRsp { + SClientHbKey connKey; + int32_t status; + int32_t bodyLen; + void* body; +} SClientHbRsp; + +typedef struct SClientHbBatchRsp { + int64_t reqId; + int64_t rspId; + SArray* rsps; // SArray +} SClientHbBatchRsp; + +static FORCE_INLINE uint32_t hbKeyHashFunc(const char* key, uint32_t keyLen) { + return taosIntHash_64(key, keyLen); +} + +int tSerializeSClientHbReq(void** buf, const SClientHbReq* pReq); +void* tDeserializeClientHbReq(void* buf, SClientHbReq* pReq); +static FORCE_INLINE void tFreeClientHbReq(void *pReq) { + SClientHbReq* req = pReq; + taosHashCleanup(req->info); + free(pReq); +} + static FORCE_INLINE int taosEncodeSKlv(void** buf, const SKlv* pKlv) { int tlen = 0; tlen += taosEncodeFixedI32(buf, pKlv->keyLen); @@ -156,10 +196,6 @@ static FORCE_INLINE void* taosDecodeSKlv(void* buf, SKlv* pKlv) { buf = taosDecodeBinary(buf, &pKlv->value, pKlv->valueLen); return buf; } -typedef struct SClientHbKey { - int32_t connId; - int32_t hbType; -} SClientHbKey; static FORCE_INLINE int taosEncodeSClientHbKey(void** buf, const SClientHbKey* pKey) { int tlen = 0; @@ -174,32 +210,6 @@ static FORCE_INLINE void* taosDecodeSClientHbKey(void* buf, SClientHbKey* pKey) return buf; } -typedef struct SClientHbReq { - SClientHbKey connKey; - SHashObj* info; // hash -} SClientHbReq; - -typedef struct SClientHbBatchReq { - int64_t reqId; - SArray* reqs; // SArray -} SClientHbBatchReq; - -int tSerializeSClientHbReq(void** buf, const SClientHbReq* pReq); -void* tDeserializeClientHbReq(void* buf, SClientHbReq* pReq); - -typedef struct SClientHbRsp { - SClientHbKey connKey; - int32_t status; - int32_t bodyLen; - void* body; -} SClientHbRsp; - -typedef struct SClientHbBatchRsp { - int64_t reqId; - int64_t rspId; - SArray* rsps; // SArray -} SClientHbBatchRsp; - typedef struct SBuildTableMetaInput { int32_t vgId; char* dbName; diff --git a/source/client/inc/clientHb.h b/source/client/inc/clientHb.h index 73adb41308..f410c5e3ee 100644 --- a/source/client/inc/clientHb.h +++ b/source/client/inc/clientHb.h @@ -18,9 +18,11 @@ #include "thash.h" #include "tmsg.h" +#define HEARTBEAT_INTERVAL 1500 //ms + typedef enum { - mq = 0, - // type can be added here + HEARTBEAT_TYPE_MQ = 0, + // types can be added here // HEARTBEAT_TYPE_MAX } EHbType; @@ -28,26 +30,16 @@ typedef enum { typedef int32_t (*FHbRspHandle)(SClientHbRsp* pReq); typedef int32_t (*FGetConnInfo)(SClientHbKey connKey, void* param); -typedef struct SClientHbMgr { - int8_t inited; - int32_t reportInterval; // unit ms - int32_t stats; - SRWLatch lock; - SHashObj* activeInfo; // hash - SHashObj* getInfoFuncs; // hash - FHbRspHandle handle[HEARTBEAT_TYPE_MAX]; - // input queue -} SClientHbMgr; - -static SClientHbMgr clientHbMgr = {0}; - +// called by mgmt int hbMgrInit(); void hbMgrCleanUp(); -int hbHandleRsp(void* hbMsg); - +int hbHandleRsp(SClientHbBatchRsp* hbRsp); +//called by user int hbRegisterConn(SClientHbKey connKey, FGetConnInfo func); - +void hbDeregisterConn(SClientHbKey connKey); int hbAddConnInfo(SClientHbKey connKey, void* key, void* value, int32_t keyLen, int32_t valueLen); +// mq +void hbMgrInitMqHbRspHandle(); diff --git a/source/client/src/clientHb.c b/source/client/src/clientHb.c deleted file mode 100644 index 7daa1204d0..0000000000 --- a/source/client/src/clientHb.c +++ /dev/null @@ -1,77 +0,0 @@ -/* - * Copyright (c) 2019 TAOS Data, Inc. - * - * 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 . - */ - -#include "clientHb.h" - -static int32_t mqHbRspHandle(SClientHbRsp* pReq) { - return 0; -} - -uint32_t hbKeyHashFunc(const char* key, uint32_t keyLen) { - return 0; -} - -static void hbMgrInitMqHbFunc() { - clientHbMgr.handle[mq] = mqHbRspHandle; -} - -int hbMgrInit() { - //init once - int8_t old = atomic_val_compare_exchange_8(&clientHbMgr.inited, 0, 1); - if (old == 1) return 0; - - //init config - clientHbMgr.reportInterval = 1500; - - //init stat - clientHbMgr.stats = 0; - - //init lock - taosInitRWLatch(&clientHbMgr.lock); - - //init handle funcs - hbMgrInitMqHbFunc(); - - //init hash info - clientHbMgr.activeInfo = taosHashInit(64, hbKeyHashFunc, 1, HASH_ENTRY_LOCK); - //init getInfoFunc - clientHbMgr.getInfoFuncs = taosHashInit(64, hbKeyHashFunc, 1, HASH_ENTRY_LOCK); - return 0; -} - -void hbMgrCleanUp() { - int8_t old = atomic_val_compare_exchange_8(&clientHbMgr.inited, 1, 0); - if (old == 0) return; - - taosHashCleanup(clientHbMgr.activeInfo); - taosHashCleanup(clientHbMgr.getInfoFuncs); -} - -int hbRegisterConn(SClientHbKey connKey, FGetConnInfo func) { - - return 0; -} - -int hbAddConnInfo(SClientHbKey connKey, void* key, void* value, int32_t keyLen, int32_t valueLen) { - //lock - - //find req by connection id - SClientHbReq* data = taosHashGet(clientHbMgr.activeInfo, &connKey, sizeof(SClientHbKey)); - ASSERT(data != NULL); - taosHashPut(data->info, key, keyLen, value, valueLen); - - //unlock - return 0; -} diff --git a/source/client/src/clientHbMgmt.c b/source/client/src/clientHbMgmt.c new file mode 100644 index 0000000000..6ac2263039 --- /dev/null +++ b/source/client/src/clientHbMgmt.c @@ -0,0 +1,166 @@ +/* + * Copyright (c) 2019 TAOS Data, Inc. + * + * 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 . + */ + +#include "clientHb.h" + +typedef struct SClientHbMgr { + int8_t inited; + + // statistics + int32_t reportCnt; + int32_t connKeyCnt; + int64_t reportBytes; // not implemented + int64_t startTime; + // thread + pthread_t thread; + + SHashObj* activeInfo; // hash + SHashObj* getInfoFuncs; // hash + FHbRspHandle handle[HEARTBEAT_TYPE_MAX]; +} SClientHbMgr; + +static SClientHbMgr clientHbMgr = {0}; + +static int32_t hbCreateThread(); +static void hbStopThread(); + +static FORCE_INLINE void hbMgrInitHandle() { + // init all handle + hbMgrInitMqHbRspHandle(); +} + +static SClientHbBatchReq* hbGatherAllInfo() { + SClientHbBatchReq* pReq = malloc(sizeof(SClientHbBatchReq)); + if(pReq == NULL) { + terrno = TSDB_CODE_TSC_OUT_OF_MEMORY; + return NULL; + } + int32_t connKeyCnt = atomic_load_32(&clientHbMgr.connKeyCnt); + pReq->reqs = taosArrayInit(connKeyCnt, sizeof(SClientHbReq)); + + void *pIter = taosHashIterate(clientHbMgr.activeInfo, pIter); + while (pIter != NULL) { + taosArrayPush(pReq->reqs, pIter); + SClientHbReq* pOneReq = pIter; + taosHashClear(pOneReq->info); + pIter = taosHashIterate(clientHbMgr.activeInfo, pIter); + } + + return pReq; +} + +static void* hbThreadFunc(void* param) { + setThreadName("hb"); + while (1) { + atomic_add_fetch_32(&clientHbMgr.reportCnt, 1); + taosMsleep(HEARTBEAT_INTERVAL); + } + + return NULL; +} + +static int32_t hbCreateThread() { + pthread_attr_t thAttr; + pthread_attr_init(&thAttr); + pthread_attr_setdetachstate(&thAttr, PTHREAD_CREATE_JOINABLE); + + if (pthread_create(&clientHbMgr.thread, &thAttr, hbThreadFunc, NULL) != 0) { + terrno = TAOS_SYSTEM_ERROR(errno); + return -1; + } + pthread_attr_destroy(&thAttr); + return 0; +} + +int hbMgrInit() { + // init once + int8_t old = atomic_val_compare_exchange_8(&clientHbMgr.inited, 0, 1); + if (old == 1) return 0; + + // init stat + clientHbMgr.startTime = taosGetTimestampMs(); + + // init handle funcs + hbMgrInitHandle(); + + // init hash info + clientHbMgr.activeInfo = taosHashInit(64, hbKeyHashFunc, 1, HASH_ENTRY_LOCK); + clientHbMgr.activeInfo->freeFp = tFreeClientHbReq; + // init getInfoFunc + clientHbMgr.getInfoFuncs = taosHashInit(64, hbKeyHashFunc, 1, HASH_ENTRY_LOCK); + + // init backgroud thread + return 0; +} + +void hbMgrCleanUp() { + int8_t old = atomic_val_compare_exchange_8(&clientHbMgr.inited, 1, 0); + if (old == 0) return; + + taosHashCleanup(clientHbMgr.activeInfo); + taosHashCleanup(clientHbMgr.getInfoFuncs); +} + +int hbHandleRsp(SClientHbBatchRsp* hbRsp) { + int64_t reqId = hbRsp->reqId; + int64_t rspId = hbRsp->rspId; + + SArray* rsps = hbRsp->rsps; + int32_t sz = taosArrayGetSize(rsps); + for (int i = 0; i < sz; i++) { + SClientHbRsp* pRsp = taosArrayGet(rsps, i); + if (pRsp->connKey.hbType < HEARTBEAT_TYPE_MAX) { + clientHbMgr.handle[pRsp->connKey.hbType](pRsp); + } else { + // discard rsp + } + } + return 0; +} + +int hbRegisterConn(SClientHbKey connKey, FGetConnInfo func) { + // init hash in activeinfo + void* data = taosHashGet(clientHbMgr.activeInfo, &connKey, sizeof(SClientHbKey)); + if (data != NULL) { + return 0; + } + SClientHbReq hbReq; + hbReq.connKey = connKey; + hbReq.info = taosHashInit(64, hbKeyHashFunc, 1, HASH_ENTRY_LOCK); + taosHashPut(clientHbMgr.activeInfo, &connKey, sizeof(SClientHbKey), &hbReq, sizeof(SClientHbReq)); + // init hash + if (func != NULL) { + taosHashPut(clientHbMgr.getInfoFuncs, &connKey, sizeof(SClientHbKey), func, sizeof(FGetConnInfo)); + } + + atomic_add_fetch_32(&clientHbMgr.connKeyCnt, 1); + return 0; +} + +void hbDeregisterConn(SClientHbKey connKey) { + taosHashRemove(clientHbMgr.activeInfo, &connKey, sizeof(SClientHbKey)); + taosHashRemove(clientHbMgr.getInfoFuncs, &connKey, sizeof(SClientHbKey)); + atomic_sub_fetch_32(&clientHbMgr.connKeyCnt, 1); +} + +int hbAddConnInfo(SClientHbKey connKey, void* key, void* value, int32_t keyLen, int32_t valueLen) { + // find req by connection id + SClientHbReq* pReq = taosHashGet(clientHbMgr.activeInfo, &connKey, sizeof(SClientHbKey)); + ASSERT(pReq != NULL); + + taosHashPut(pReq->info, key, keyLen, value, valueLen); + + return 0; +} diff --git a/source/client/src/clientHbMq.c b/source/client/src/clientHbMq.c new file mode 100644 index 0000000000..8f2400aedd --- /dev/null +++ b/source/client/src/clientHbMq.c @@ -0,0 +1,25 @@ +/* + * Copyright (c) 2019 TAOS Data, Inc. + * + * 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 . + */ + +#include "clientHb.h" + +static int32_t mqHbRspHandle(SClientHbRsp* pReq) { + return 0; +} + +void hbMgrInitMqHbRspHandle() { + clientHbMgr.handle[HEARTBEAT_TYPE_MQ] = mqHbRspHandle; +} + diff --git a/source/common/src/tmsg.c b/source/common/src/tmsg.c index 5cbb42c1e6..2985d5dfe8 100644 --- a/source/common/src/tmsg.c +++ b/source/common/src/tmsg.c @@ -31,17 +31,15 @@ int tSerializeSClientHbReq(void **buf, const SClientHbReq *pReq) { int tlen = 0; tlen += taosEncodeSClientHbKey(buf, &pReq->connKey); - void *pIter = NULL; - void *data; SKlv klv; - data = taosHashIterate(pReq->info, pIter); - while (data != NULL) { - taosHashGetKey(data, &klv.key, (size_t *)&klv.keyLen); - klv.valueLen = taosHashGetDataLen(data); - klv.value = data; + void* pIter = taosHashIterate(pReq->info, pIter); + while (pIter != NULL) { + taosHashGetKey(pIter, &klv.key, (size_t *)&klv.keyLen); + klv.valueLen = taosHashGetDataLen(pIter); + klv.value = pIter; taosEncodeSKlv(buf, &klv); - data = taosHashIterate(pReq->info, pIter); + pIter = taosHashIterate(pReq->info, pIter); } return tlen; } diff --git a/source/libs/wal/src/walMgmt.c b/source/libs/wal/src/walMgmt.c index 189881c86d..d12acb52c6 100644 --- a/source/libs/wal/src/walMgmt.c +++ b/source/libs/wal/src/walMgmt.c @@ -232,7 +232,8 @@ static int32_t walCreateThread() { if (pthread_create(&tsWal.thread, &thAttr, walThreadFunc, NULL) != 0) { wError("failed to create wal thread since %s", strerror(errno)); - return TAOS_SYSTEM_ERROR(errno); + terrno = TAOS_SYSTEM_ERROR(errno); + return -1; } pthread_attr_destroy(&thAttr); From 29cb7c67929a91505ea84038ed53e817328f95a6 Mon Sep 17 00:00:00 2001 From: Liu Jicong Date: Fri, 7 Jan 2022 16:48:35 +0800 Subject: [PATCH 2/5] implement client heartbeat --- include/common/tmsg.h | 36 +++++++----- include/libs/planner/planner.h | 1 + include/libs/qcom/query.h | 49 ---------------- include/libs/transport/trpc.h | 49 ++++++++++++++++ include/util/thash.h | 18 ++++++ source/client/inc/clientHb.h | 7 ++- .../client/src/{clientHbMgmt.c => clientHb.c} | 56 ++++++++++++++++++- source/client/src/clientHbMq.c | 25 --------- source/common/src/tmsg.c | 25 ++++++--- source/libs/qcom/test/CMakeLists.txt | 2 +- source/libs/qcom/test/queryTest.cpp | 3 +- source/util/src/thash.c | 4 +- 12 files changed, 172 insertions(+), 103 deletions(-) rename source/client/src/{clientHbMgmt.c => clientHb.c} (77%) delete mode 100644 source/client/src/clientHbMq.c diff --git a/include/common/tmsg.h b/include/common/tmsg.h index d57fec3d5a..f173662770 100644 --- a/include/common/tmsg.h +++ b/include/common/tmsg.h @@ -133,12 +133,12 @@ typedef enum _mgmt_table { #define TSDB_COL_IS_UD_COL(f) ((f & (~(TSDB_COL_NULL))) == TSDB_COL_UDC) #define TSDB_COL_REQ_NULL(f) (((f)&TSDB_COL_NULL) != 0) -typedef struct SKlv { +typedef struct SKv { int32_t keyLen; int32_t valueLen; void* key; void* value; -} SKlv; +} SKv; typedef struct SClientHbKey { int32_t connId; @@ -174,26 +174,36 @@ static FORCE_INLINE uint32_t hbKeyHashFunc(const char* key, uint32_t keyLen) { int tSerializeSClientHbReq(void** buf, const SClientHbReq* pReq); void* tDeserializeClientHbReq(void* buf, SClientHbReq* pReq); + static FORCE_INLINE void tFreeClientHbReq(void *pReq) { - SClientHbReq* req = pReq; + SClientHbReq* req = (SClientHbReq*)pReq; taosHashCleanup(req->info); free(pReq); } -static FORCE_INLINE int taosEncodeSKlv(void** buf, const SKlv* pKlv) { +int tSerializeSClientHbBatchReq(void** buf, const SClientHbBatchReq* pReq); +void* tDeserializeClientHbBatchReq(void* buf, SClientHbBatchReq* pReq); + +static FORCE_INLINE void tFreeClientHbBatchReq(void* pReq) { + SClientHbBatchReq *req = (SClientHbBatchReq*)pReq; + taosArrayDestroyEx(req->reqs, tFreeClientHbReq); + free(pReq); +} + +static FORCE_INLINE int taosEncodeSKv(void** buf, const SKv* pKv) { int tlen = 0; - tlen += taosEncodeFixedI32(buf, pKlv->keyLen); - tlen += taosEncodeFixedI32(buf, pKlv->valueLen); - tlen += taosEncodeBinary(buf, pKlv->key, pKlv->keyLen); - tlen += taosEncodeBinary(buf, pKlv->value, pKlv->valueLen); + tlen += taosEncodeFixedI32(buf, pKv->keyLen); + tlen += taosEncodeFixedI32(buf, pKv->valueLen); + tlen += taosEncodeBinary(buf, pKv->key, pKv->keyLen); + tlen += taosEncodeBinary(buf, pKv->value, pKv->valueLen); return tlen; } -static FORCE_INLINE void* taosDecodeSKlv(void* buf, SKlv* pKlv) { - buf = taosDecodeFixedI32(buf, &pKlv->keyLen); - buf = taosDecodeFixedI32(buf, &pKlv->valueLen); - buf = taosDecodeBinary(buf, &pKlv->key, pKlv->keyLen); - buf = taosDecodeBinary(buf, &pKlv->value, pKlv->valueLen); +static FORCE_INLINE void* taosDecodeSKv(void* buf, SKv* pKv) { + buf = taosDecodeFixedI32(buf, &pKv->keyLen); + buf = taosDecodeFixedI32(buf, &pKv->valueLen); + buf = taosDecodeBinary(buf, &pKv->key, pKv->keyLen); + buf = taosDecodeBinary(buf, &pKv->value, pKv->valueLen); return buf; } diff --git a/include/libs/planner/planner.h b/include/libs/planner/planner.h index a55a4fedab..ae76b022f2 100644 --- a/include/libs/planner/planner.h +++ b/include/libs/planner/planner.h @@ -23,6 +23,7 @@ extern "C" { #include "query.h" #include "tmsg.h" #include "tarray.h" +#include "trpc.h" #define QUERY_TYPE_MERGE 1 #define QUERY_TYPE_PARTIAL 2 diff --git a/include/libs/qcom/query.h b/include/libs/qcom/query.h index 4bfb774435..17a06a941a 100644 --- a/include/libs/qcom/query.h +++ b/include/libs/qcom/query.h @@ -94,55 +94,6 @@ typedef struct STableMetaOutput { STableMeta *tbMeta; } STableMetaOutput; -typedef struct SDataBuf { - void *pData; - uint32_t len; -} SDataBuf; - -typedef int32_t (*__async_send_cb_fn_t)(void* param, const SDataBuf* pMsg, int32_t code); -typedef int32_t (*__async_exec_fn_t)(void* param); - -typedef struct SMsgSendInfo { - __async_send_cb_fn_t fp; //async callback function - void *param; - uint64_t requestId; - uint64_t requestObjRefId; - int32_t msgType; - SDataBuf msgInfo; -} SMsgSendInfo; - -typedef struct SQueryNodeAddr{ - int32_t nodeId; //vgId or qnodeId - int8_t inUse; - int8_t numOfEps; - SEpAddrMsg epAddr[TSDB_MAX_REPLICA]; -} SQueryNodeAddr; - -bool tIsValidSchema(struct SSchema* pSchema, int32_t numOfCols, int32_t numOfTags); - -int32_t initTaskQueue(); -int32_t cleanupTaskQueue(); - -/** - * - * @param execFn The asynchronously execution function - * @param execParam The parameters of the execFn - * @param code The response code during execution the execFn - * @return - */ -int32_t taosAsyncExec(__async_exec_fn_t execFn, void* execParam, int32_t* code); - -/** - * Asynchronously send message to server, after the response received, the callback will be incured. - * - * @param pTransporter - * @param epSet - * @param pTransporterId - * @param pInfo - * @return - */ -int32_t asyncSendMsgToServer(void *pTransporter, SEpSet* epSet, int64_t* pTransporterId, const SMsgSendInfo* pInfo); - const SSchema* tGetTbnameColumnSchema(); void initQueryModuleMsgHandle(); diff --git a/include/libs/transport/trpc.h b/include/libs/transport/trpc.h index b88d45fbd6..e28098dfbf 100644 --- a/include/libs/transport/trpc.h +++ b/include/libs/transport/trpc.h @@ -84,6 +84,55 @@ void rpcSendRecv(void *shandle, SEpSet *pEpSet, SRpcMsg *pReq, SRpcMsg *pRsp) int rpcReportProgress(void *pConn, char *pCont, int contLen); void rpcCancelRequest(int64_t rid); +typedef struct SDataBuf { + void *pData; + uint32_t len; +} SDataBuf; + +typedef int32_t (*__async_send_cb_fn_t)(void* param, const SDataBuf* pMsg, int32_t code); +typedef int32_t (*__async_exec_fn_t)(void* param); + +typedef struct SMsgSendInfo { + __async_send_cb_fn_t fp; //async callback function + void *param; + uint64_t requestId; + uint64_t requestObjRefId; + int32_t msgType; + SDataBuf msgInfo; +} SMsgSendInfo; + +typedef struct SQueryNodeAddr{ + int32_t nodeId; //vgId or qnodeId + int8_t inUse; + int8_t numOfEps; + SEpAddrMsg epAddr[TSDB_MAX_REPLICA]; +} SQueryNodeAddr; + +bool tIsValidSchema(struct SSchema* pSchema, int32_t numOfCols, int32_t numOfTags); + +int32_t initTaskQueue(); +int32_t cleanupTaskQueue(); + +/** + * + * @param execFn The asynchronously execution function + * @param execParam The parameters of the execFn + * @param code The response code during execution the execFn + * @return + */ +int32_t taosAsyncExec(__async_exec_fn_t execFn, void* execParam, int32_t* code); + +/** + * Asynchronously send message to server, after the response received, the callback will be incured. + * + * @param pTransporter + * @param epSet + * @param pTransporterId + * @param pInfo + * @return + */ +int32_t asyncSendMsgToServer(void *pTransporter, SEpSet* epSet, int64_t* pTransporterId, const SMsgSendInfo* pInfo); + #ifdef __cplusplus } #endif diff --git a/include/util/thash.h b/include/util/thash.h index 4558162ac5..9dc6630461 100644 --- a/include/util/thash.h +++ b/include/util/thash.h @@ -211,6 +211,24 @@ void taosHashCancelIterate(SHashObj *pHashObj, void *p); int32_t taosHashGetKey(void *data, void** key, size_t* keyLen); +/** + * Get the corresponding key information for a given data in hash table, using memcpy + * @param data + * @param dst + * @return + */ +static FORCE_INLINE int32_t taosHashCopyKey(void *data, void* dst) { + if (NULL == data || NULL == dst) { + return -1; + } + + SHashNode * node = GET_HASH_PNODE(data); + void* key = GET_HASH_NODE_KEY(node); + memcpy(dst, key, node->keyLen); + + return 0; +} + /** * Get the corresponding data length for a given data in hash table * @param data diff --git a/source/client/inc/clientHb.h b/source/client/inc/clientHb.h index f410c5e3ee..2624c8f833 100644 --- a/source/client/inc/clientHb.h +++ b/source/client/inc/clientHb.h @@ -28,10 +28,13 @@ typedef enum { } EHbType; typedef int32_t (*FHbRspHandle)(SClientHbRsp* pReq); -typedef int32_t (*FGetConnInfo)(SClientHbKey connKey, void* param); + +//TODO: embed param into function +//return type: SArray +typedef SArray* (*FGetConnInfo)(SClientHbKey connKey, void* param); // called by mgmt -int hbMgrInit(); +int hbMgrInit(void* transporter, SEpSet epSet); void hbMgrCleanUp(); int hbHandleRsp(SClientHbBatchRsp* hbRsp); diff --git a/source/client/src/clientHbMgmt.c b/source/client/src/clientHb.c similarity index 77% rename from source/client/src/clientHbMgmt.c rename to source/client/src/clientHb.c index 6ac2263039..8672b52343 100644 --- a/source/client/src/clientHbMgmt.c +++ b/source/client/src/clientHb.c @@ -14,6 +14,7 @@ */ #include "clientHb.h" +#include "trpc.h" typedef struct SClientHbMgr { int8_t inited; @@ -23,8 +24,13 @@ typedef struct SClientHbMgr { int32_t connKeyCnt; int64_t reportBytes; // not implemented int64_t startTime; - // thread + // ctl + int8_t threadStop; pthread_t thread; + SRWLatch lock; // lock is used in serialization + // connection + void* transporter; + SEpSet epSet; SHashObj* activeInfo; // hash SHashObj* getInfoFuncs; // hash @@ -36,6 +42,14 @@ static SClientHbMgr clientHbMgr = {0}; static int32_t hbCreateThread(); static void hbStopThread(); +static int32_t hbMqHbRspHandle(SClientHbRsp* pReq) { + return 0; +} + +void hbMgrInitMqHbRspHandle() { + clientHbMgr.handle[HEARTBEAT_TYPE_MQ] = hbMqHbRspHandle; +} + static FORCE_INLINE void hbMgrInitHandle() { // init all handle hbMgrInitMqHbRspHandle(); @@ -50,11 +64,22 @@ static SClientHbBatchReq* hbGatherAllInfo() { int32_t connKeyCnt = atomic_load_32(&clientHbMgr.connKeyCnt); pReq->reqs = taosArrayInit(connKeyCnt, sizeof(SClientHbReq)); - void *pIter = taosHashIterate(clientHbMgr.activeInfo, pIter); + void *pIter = taosHashIterate(clientHbMgr.activeInfo, NULL); while (pIter != NULL) { taosArrayPush(pReq->reqs, pIter); SClientHbReq* pOneReq = pIter; taosHashClear(pOneReq->info); + + pIter = taosHashIterate(clientHbMgr.activeInfo, pIter); + } + + pIter = taosHashIterate(clientHbMgr.getInfoFuncs, NULL); + while (pIter != NULL) { + FGetConnInfo getConnInfoFp = (FGetConnInfo)pIter; + SClientHbKey connKey; + taosHashCopyKey(pIter, &connKey); + getConnInfoFp(connKey, NULL); + pIter = taosHashIterate(clientHbMgr.activeInfo, pIter); } @@ -64,8 +89,23 @@ static SClientHbBatchReq* hbGatherAllInfo() { static void* hbThreadFunc(void* param) { setThreadName("hb"); while (1) { + int8_t threadStop = atomic_load_8(&clientHbMgr.threadStop); + if(threadStop) { + break; + } + + SClientHbBatchReq* pReq = hbGatherAllInfo(); + void* reqStr = NULL; + tSerializeSClientHbBatchReq(&reqStr, pReq); + SMsgSendInfo info; + + int64_t transporterId = 0; + asyncSendMsgToServer(clientHbMgr.transporter, &clientHbMgr.epSet, &transporterId, &info); + tFreeClientHbBatchReq(pReq); + atomic_add_fetch_32(&clientHbMgr.reportCnt, 1); taosMsleep(HEARTBEAT_INTERVAL); + } return NULL; @@ -84,7 +124,11 @@ static int32_t hbCreateThread() { return 0; } -int hbMgrInit() { +static void hbStopThread() { + atomic_store_8(&clientHbMgr.threadStop, 1); +} + +int hbMgrInit(void* transporter, SEpSet epSet) { // init once int8_t old = atomic_val_compare_exchange_8(&clientHbMgr.inited, 0, 1); if (old == 1) return 0; @@ -101,7 +145,13 @@ int hbMgrInit() { // init getInfoFunc clientHbMgr.getInfoFuncs = taosHashInit(64, hbKeyHashFunc, 1, HASH_ENTRY_LOCK); + //init connection info + clientHbMgr.transporter = transporter; + clientHbMgr.epSet = epSet; + // init backgroud thread + hbCreateThread(); + return 0; } diff --git a/source/client/src/clientHbMq.c b/source/client/src/clientHbMq.c deleted file mode 100644 index 8f2400aedd..0000000000 --- a/source/client/src/clientHbMq.c +++ /dev/null @@ -1,25 +0,0 @@ -/* - * Copyright (c) 2019 TAOS Data, Inc. - * - * 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 . - */ - -#include "clientHb.h" - -static int32_t mqHbRspHandle(SClientHbRsp* pReq) { - return 0; -} - -void hbMgrInitMqHbRspHandle() { - clientHbMgr.handle[HEARTBEAT_TYPE_MQ] = mqHbRspHandle; -} - diff --git a/source/common/src/tmsg.c b/source/common/src/tmsg.c index 2985d5dfe8..408ce899b1 100644 --- a/source/common/src/tmsg.c +++ b/source/common/src/tmsg.c @@ -31,13 +31,13 @@ int tSerializeSClientHbReq(void **buf, const SClientHbReq *pReq) { int tlen = 0; tlen += taosEncodeSClientHbKey(buf, &pReq->connKey); - SKlv klv; + SKv kv; void* pIter = taosHashIterate(pReq->info, pIter); while (pIter != NULL) { - taosHashGetKey(pIter, &klv.key, (size_t *)&klv.keyLen); - klv.valueLen = taosHashGetDataLen(pIter); - klv.value = pIter; - taosEncodeSKlv(buf, &klv); + taosHashGetKey(pIter, &kv.key, (size_t *)&kv.keyLen); + kv.valueLen = taosHashGetDataLen(pIter); + kv.value = pIter; + taosEncodeSKv(buf, &kv); pIter = taosHashIterate(pReq->info, pIter); } @@ -52,13 +52,22 @@ void *tDeserializeClientHbReq(void *buf, SClientHbReq *pReq) { if (pReq->info == NULL) { pReq->info = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_NO_LOCK); } - SKlv klv; - buf = taosDecodeSKlv(buf, &klv); - taosHashPut(pReq->info, klv.key, klv.keyLen, klv.value, klv.valueLen); + SKv kv; + buf = taosDecodeSKv(buf, &kv); + taosHashPut(pReq->info, kv.key, kv.keyLen, kv.value, kv.valueLen); return buf; } +int tSerializeSClientHbBatchReq(void** buf, const SClientHbBatchReq* pReq) { + int tlen = 0; + return tlen; +} + +void* tDeserializeClientHbBatchReq(void* buf, SClientHbBatchReq* pReq) { + return buf; +} + int tSerializeSVCreateTbReq(void **buf, SVCreateTbReq *pReq) { int tlen = 0; diff --git a/source/libs/qcom/test/CMakeLists.txt b/source/libs/qcom/test/CMakeLists.txt index 7adec3752a..e3a0e11a32 100644 --- a/source/libs/qcom/test/CMakeLists.txt +++ b/source/libs/qcom/test/CMakeLists.txt @@ -15,5 +15,5 @@ TARGET_INCLUDE_DIRECTORIES( TARGET_LINK_LIBRARIES( queryUtilTest - PUBLIC os util gtest qcom common + PUBLIC os util gtest qcom common transport ) diff --git a/source/libs/qcom/test/queryTest.cpp b/source/libs/qcom/test/queryTest.cpp index ddf89c6272..8fc6b7e529 100644 --- a/source/libs/qcom/test/queryTest.cpp +++ b/source/libs/qcom/test/queryTest.cpp @@ -17,6 +17,7 @@ #include #include "tmsg.h" #include "query.h" +#include "trpc.h" #pragma GCC diagnostic ignored "-Wwrite-strings" #pragma GCC diagnostic ignored "-Wunused-function" @@ -80,4 +81,4 @@ TEST(testCase, error_in_async_test) { taosAsyncExec(testPrintError, p, &code); usleep(1000); printf("Error code:%d after asynchronously exec function\n", code); -} \ No newline at end of file +} diff --git a/source/util/src/thash.c b/source/util/src/thash.c index 181661d304..6ec7072a1b 100644 --- a/source/util/src/thash.c +++ b/source/util/src/thash.c @@ -794,7 +794,9 @@ FORCE_INLINE int32_t taosHashGetKey(void *data, void** key, size_t* keyLen) { SHashNode * node = GET_HASH_PNODE(data); *key = GET_HASH_NODE_KEY(node); - *keyLen = node->keyLen; + if (keyLen) { + *keyLen = node->keyLen; + } return 0; } From 4d635478fea25359ad812b8282bd747809a125d5 Mon Sep 17 00:00:00 2001 From: Liu Jicong Date: Fri, 7 Jan 2022 18:59:17 +0800 Subject: [PATCH 3/5] fix --- source/client/src/clientHb.c | 3 ++- source/common/src/tmsg.c | 2 +- source/libs/wal/src/walWrite.c | 1 + 3 files changed, 4 insertions(+), 2 deletions(-) diff --git a/source/client/src/clientHb.c b/source/client/src/clientHb.c index 8672b52343..1d50f7574a 100644 --- a/source/client/src/clientHb.c +++ b/source/client/src/clientHb.c @@ -96,8 +96,9 @@ static void* hbThreadFunc(void* param) { SClientHbBatchReq* pReq = hbGatherAllInfo(); void* reqStr = NULL; - tSerializeSClientHbBatchReq(&reqStr, pReq); + int tlen = tSerializeSClientHbBatchReq(&reqStr, pReq); SMsgSendInfo info; + /*info.fp = hbHandleRsp;*/ int64_t transporterId = 0; asyncSendMsgToServer(clientHbMgr.transporter, &clientHbMgr.epSet, &transporterId, &info); diff --git a/source/common/src/tmsg.c b/source/common/src/tmsg.c index 408ce899b1..09523da965 100644 --- a/source/common/src/tmsg.c +++ b/source/common/src/tmsg.c @@ -37,7 +37,7 @@ int tSerializeSClientHbReq(void **buf, const SClientHbReq *pReq) { taosHashGetKey(pIter, &kv.key, (size_t *)&kv.keyLen); kv.valueLen = taosHashGetDataLen(pIter); kv.value = pIter; - taosEncodeSKv(buf, &kv); + tlen += taosEncodeSKv(buf, &kv); pIter = taosHashIterate(pReq->info, pIter); } diff --git a/source/libs/wal/src/walWrite.c b/source/libs/wal/src/walWrite.c index 39bf06cedd..5eda16b061 100644 --- a/source/libs/wal/src/walWrite.c +++ b/source/libs/wal/src/walWrite.c @@ -244,6 +244,7 @@ int walRoll(SWal *pWal) { pWal->writeIdxTfd = idxTfd; pWal->writeLogTfd = logTfd; pWal->writeCur = taosArrayGetSize(pWal->fileInfoSet) - 1; + ASSERT(pWal->writeCur >= 0); pWal->lastRollSeq = walGetSeq(); return 0; From 1e17d776f905e998ea69e13fbde9726156d39954 Mon Sep 17 00:00:00 2001 From: Liu Jicong Date: Mon, 10 Jan 2022 16:13:05 +0800 Subject: [PATCH 4/5] fix crash --- source/client/inc/clientHb.h | 48 +++++++-- source/client/src/clientHb.c | 142 ++++++++++++++------------ source/common/src/tmsg.c | 16 +-- source/dnode/mnode/impl/src/mndSync.c | 4 +- source/libs/wal/inc/walInt.h | 1 + source/libs/wal/src/walMeta.c | 1 + source/libs/wal/src/walWrite.c | 8 ++ 7 files changed, 137 insertions(+), 83 deletions(-) diff --git a/source/client/inc/clientHb.h b/source/client/inc/clientHb.h index 2624c8f833..7bc4311b29 100644 --- a/source/client/inc/clientHb.h +++ b/source/client/inc/clientHb.h @@ -18,7 +18,7 @@ #include "thash.h" #include "tmsg.h" -#define HEARTBEAT_INTERVAL 1500 //ms +#define HEARTBEAT_INTERVAL 1500 // ms typedef enum { HEARTBEAT_TYPE_MQ = 0, @@ -29,20 +29,50 @@ typedef enum { typedef int32_t (*FHbRspHandle)(SClientHbRsp* pReq); -//TODO: embed param into function -//return type: SArray +typedef struct SAppHbMgr { + // statistics + int32_t reportCnt; + int32_t connKeyCnt; + int64_t reportBytes; // not implemented + int64_t startTime; + // ctl + SRWLatch lock; // lock is used in serialization + // connection + void* transporter; + SEpSet epSet; + // info + SHashObj* activeInfo; // hash + SHashObj* getInfoFuncs; // hash +} SAppHbMgr; + +typedef struct SClientHbMgr { + int8_t inited; + // ctl + int8_t threadStop; + pthread_t thread; + pthread_mutex_t lock; // used when app init and cleanup + SArray* appHbMgrs; // SArray one for each cluster + FHbRspHandle handle[HEARTBEAT_TYPE_MAX]; +} SClientHbMgr; + +// TODO: embed param into function +// return type: SArray typedef SArray* (*FGetConnInfo)(SClientHbKey connKey, void* param); -// called by mgmt -int hbMgrInit(void* transporter, SEpSet epSet); +// global, called by mgmt +int hbMgrInit(); void hbMgrCleanUp(); int hbHandleRsp(SClientHbBatchRsp* hbRsp); -//called by user -int hbRegisterConn(SClientHbKey connKey, FGetConnInfo func); -void hbDeregisterConn(SClientHbKey connKey); +// cluster level +SAppHbMgr* appHbMgrInit(void* transporter, SEpSet epSet); +void appHbMgrCleanup(SAppHbMgr* pAppHbMgr); -int hbAddConnInfo(SClientHbKey connKey, void* key, void* value, int32_t keyLen, int32_t valueLen); +// conn level +int hbRegisterConn(SAppHbMgr* pAppHbMgr, SClientHbKey connKey, FGetConnInfo func); +void hbDeregisterConn(SAppHbMgr* pAppHbMgr, SClientHbKey connKey); + +int hbAddConnInfo(SAppHbMgr* pAppHbMgr, SClientHbKey connKey, void* key, void* value, int32_t keyLen, int32_t valueLen); // mq void hbMgrInitMqHbRspHandle(); diff --git a/source/client/src/clientHb.c b/source/client/src/clientHb.c index 1d50f7574a..9bbd62c1d9 100644 --- a/source/client/src/clientHb.c +++ b/source/client/src/clientHb.c @@ -16,27 +16,6 @@ #include "clientHb.h" #include "trpc.h" -typedef struct SClientHbMgr { - int8_t inited; - - // statistics - int32_t reportCnt; - int32_t connKeyCnt; - int64_t reportBytes; // not implemented - int64_t startTime; - // ctl - int8_t threadStop; - pthread_t thread; - SRWLatch lock; // lock is used in serialization - // connection - void* transporter; - SEpSet epSet; - - SHashObj* activeInfo; // hash - SHashObj* getInfoFuncs; // hash - FHbRspHandle handle[HEARTBEAT_TYPE_MAX]; -} SClientHbMgr; - static SClientHbMgr clientHbMgr = {0}; static int32_t hbCreateThread(); @@ -55,32 +34,32 @@ static FORCE_INLINE void hbMgrInitHandle() { hbMgrInitMqHbRspHandle(); } -static SClientHbBatchReq* hbGatherAllInfo() { +SClientHbBatchReq* hbGatherAllInfo(SAppHbMgr *pAppHbMgr) { SClientHbBatchReq* pReq = malloc(sizeof(SClientHbBatchReq)); - if(pReq == NULL) { + if (pReq == NULL) { terrno = TSDB_CODE_TSC_OUT_OF_MEMORY; return NULL; } - int32_t connKeyCnt = atomic_load_32(&clientHbMgr.connKeyCnt); + int32_t connKeyCnt = atomic_load_32(&pAppHbMgr->connKeyCnt); pReq->reqs = taosArrayInit(connKeyCnt, sizeof(SClientHbReq)); - void *pIter = taosHashIterate(clientHbMgr.activeInfo, NULL); + void *pIter = taosHashIterate(pAppHbMgr->activeInfo, NULL); while (pIter != NULL) { taosArrayPush(pReq->reqs, pIter); SClientHbReq* pOneReq = pIter; taosHashClear(pOneReq->info); - pIter = taosHashIterate(clientHbMgr.activeInfo, pIter); + pIter = taosHashIterate(pAppHbMgr->activeInfo, pIter); } - pIter = taosHashIterate(clientHbMgr.getInfoFuncs, NULL); + pIter = taosHashIterate(pAppHbMgr->getInfoFuncs, NULL); while (pIter != NULL) { FGetConnInfo getConnInfoFp = (FGetConnInfo)pIter; SClientHbKey connKey; taosHashCopyKey(pIter, &connKey); getConnInfoFp(connKey, NULL); - pIter = taosHashIterate(clientHbMgr.activeInfo, pIter); + pIter = taosHashIterate(pAppHbMgr->activeInfo, pIter); } return pReq; @@ -93,22 +72,24 @@ static void* hbThreadFunc(void* param) { if(threadStop) { break; } - - SClientHbBatchReq* pReq = hbGatherAllInfo(); - void* reqStr = NULL; - int tlen = tSerializeSClientHbBatchReq(&reqStr, pReq); - SMsgSendInfo info; - /*info.fp = hbHandleRsp;*/ - int64_t transporterId = 0; - asyncSendMsgToServer(clientHbMgr.transporter, &clientHbMgr.epSet, &transporterId, &info); - tFreeClientHbBatchReq(pReq); + int sz = taosArrayGetSize(clientHbMgr.appHbMgrs); + for(int i = 0; i < sz; i++) { + SAppHbMgr* pAppHbMgr = taosArrayGet(clientHbMgr.appHbMgrs, i); + SClientHbBatchReq* pReq = hbGatherAllInfo(pAppHbMgr); + void* reqStr = NULL; + int tlen = tSerializeSClientHbBatchReq(&reqStr, pReq); + SMsgSendInfo info; + /*info.fp = hbHandleRsp;*/ - atomic_add_fetch_32(&clientHbMgr.reportCnt, 1); - taosMsleep(HEARTBEAT_INTERVAL); + int64_t transporterId = 0; + asyncSendMsgToServer(pAppHbMgr->transporter, &pAppHbMgr->epSet, &transporterId, &info); + tFreeClientHbBatchReq(pReq); + atomic_add_fetch_32(&pAppHbMgr->reportCnt, 1); + taosMsleep(HEARTBEAT_INTERVAL); + } } - return NULL; } @@ -129,27 +110,55 @@ static void hbStopThread() { atomic_store_8(&clientHbMgr.threadStop, 1); } -int hbMgrInit(void* transporter, SEpSet epSet) { +SAppHbMgr* appHbMgrInit(void* transporter, SEpSet epSet) { + SAppHbMgr* pAppHbMgr = malloc(sizeof(SAppHbMgr)); + if (pAppHbMgr == NULL) { + terrno = TSDB_CODE_OUT_OF_MEMORY; + return NULL; + } + // init stat + pAppHbMgr->startTime = taosGetTimestampMs(); + + // init connection info + pAppHbMgr->transporter = transporter; + pAppHbMgr->epSet = epSet; + + // init hash info + pAppHbMgr->activeInfo = taosHashInit(64, hbKeyHashFunc, 1, HASH_ENTRY_LOCK); + pAppHbMgr->activeInfo->freeFp = tFreeClientHbReq; + // init getInfoFunc + pAppHbMgr->getInfoFuncs = taosHashInit(64, hbKeyHashFunc, 1, HASH_ENTRY_LOCK); + + taosArrayPush(clientHbMgr.appHbMgrs, &pAppHbMgr); + return pAppHbMgr; +} + +void appHbMgrCleanup(SAppHbMgr* pAppHbMgr) { + pthread_mutex_lock(&clientHbMgr.lock); + + int sz = taosArrayGetSize(clientHbMgr.appHbMgrs); + for (int i = 0; i < sz; i++) { + SAppHbMgr* pTarget = taosArrayGet(clientHbMgr.appHbMgrs, i); + if (pAppHbMgr == pTarget) { + taosHashCleanup(pTarget->activeInfo); + taosHashCleanup(pTarget->getInfoFuncs); + } + } + + pthread_mutex_unlock(&clientHbMgr.lock); +} + +int hbMgrInit() { // init once int8_t old = atomic_val_compare_exchange_8(&clientHbMgr.inited, 0, 1); if (old == 1) return 0; - // init stat - clientHbMgr.startTime = taosGetTimestampMs(); + clientHbMgr.appHbMgrs = taosArrayInit(0, sizeof(void*)); + pthread_mutex_init(&clientHbMgr.lock, NULL); // init handle funcs hbMgrInitHandle(); - // init hash info - clientHbMgr.activeInfo = taosHashInit(64, hbKeyHashFunc, 1, HASH_ENTRY_LOCK); - clientHbMgr.activeInfo->freeFp = tFreeClientHbReq; - // init getInfoFunc - clientHbMgr.getInfoFuncs = taosHashInit(64, hbKeyHashFunc, 1, HASH_ENTRY_LOCK); - - //init connection info - clientHbMgr.transporter = transporter; - clientHbMgr.epSet = epSet; - // init backgroud thread hbCreateThread(); @@ -157,11 +166,12 @@ int hbMgrInit(void* transporter, SEpSet epSet) { } void hbMgrCleanUp() { + // destroy all appHbMgr int8_t old = atomic_val_compare_exchange_8(&clientHbMgr.inited, 1, 0); if (old == 0) return; - taosHashCleanup(clientHbMgr.activeInfo); - taosHashCleanup(clientHbMgr.getInfoFuncs); + taosArrayDestroy(clientHbMgr.appHbMgrs); + } int hbHandleRsp(SClientHbBatchRsp* hbRsp) { @@ -181,34 +191,34 @@ int hbHandleRsp(SClientHbBatchRsp* hbRsp) { return 0; } -int hbRegisterConn(SClientHbKey connKey, FGetConnInfo func) { +int hbRegisterConn(SAppHbMgr* pAppHbMgr, SClientHbKey connKey, FGetConnInfo func) { // init hash in activeinfo - void* data = taosHashGet(clientHbMgr.activeInfo, &connKey, sizeof(SClientHbKey)); + void* data = taosHashGet(pAppHbMgr->activeInfo, &connKey, sizeof(SClientHbKey)); if (data != NULL) { return 0; } SClientHbReq hbReq; hbReq.connKey = connKey; hbReq.info = taosHashInit(64, hbKeyHashFunc, 1, HASH_ENTRY_LOCK); - taosHashPut(clientHbMgr.activeInfo, &connKey, sizeof(SClientHbKey), &hbReq, sizeof(SClientHbReq)); + taosHashPut(pAppHbMgr->activeInfo, &connKey, sizeof(SClientHbKey), &hbReq, sizeof(SClientHbReq)); // init hash if (func != NULL) { - taosHashPut(clientHbMgr.getInfoFuncs, &connKey, sizeof(SClientHbKey), func, sizeof(FGetConnInfo)); + taosHashPut(pAppHbMgr->getInfoFuncs, &connKey, sizeof(SClientHbKey), func, sizeof(FGetConnInfo)); } - atomic_add_fetch_32(&clientHbMgr.connKeyCnt, 1); + atomic_add_fetch_32(&pAppHbMgr->connKeyCnt, 1); return 0; } -void hbDeregisterConn(SClientHbKey connKey) { - taosHashRemove(clientHbMgr.activeInfo, &connKey, sizeof(SClientHbKey)); - taosHashRemove(clientHbMgr.getInfoFuncs, &connKey, sizeof(SClientHbKey)); - atomic_sub_fetch_32(&clientHbMgr.connKeyCnt, 1); +void hbDeregisterConn(SAppHbMgr* pAppHbMgr, SClientHbKey connKey) { + taosHashRemove(pAppHbMgr->activeInfo, &connKey, sizeof(SClientHbKey)); + taosHashRemove(pAppHbMgr->getInfoFuncs, &connKey, sizeof(SClientHbKey)); + atomic_sub_fetch_32(&pAppHbMgr->connKeyCnt, 1); } -int hbAddConnInfo(SClientHbKey connKey, void* key, void* value, int32_t keyLen, int32_t valueLen) { +int hbAddConnInfo(SAppHbMgr *pAppHbMgr, SClientHbKey connKey, void* key, void* value, int32_t keyLen, int32_t valueLen) { // find req by connection id - SClientHbReq* pReq = taosHashGet(clientHbMgr.activeInfo, &connKey, sizeof(SClientHbKey)); + SClientHbReq* pReq = taosHashGet(pAppHbMgr->activeInfo, &connKey, sizeof(SClientHbKey)); ASSERT(pReq != NULL); taosHashPut(pReq->info, key, keyLen, value, valueLen); diff --git a/source/common/src/tmsg.c b/source/common/src/tmsg.c index 09523da965..b94bd6f715 100644 --- a/source/common/src/tmsg.c +++ b/source/common/src/tmsg.c @@ -31,7 +31,9 @@ int tSerializeSClientHbReq(void **buf, const SClientHbReq *pReq) { int tlen = 0; tlen += taosEncodeSClientHbKey(buf, &pReq->connKey); - SKv kv; + int kvNum = taosHashGetSize(pReq->info); + tlen += taosEncodeFixedI32(buf, kvNum); + SKv kv; void* pIter = taosHashIterate(pReq->info, pIter); while (pIter != NULL) { taosHashGetKey(pIter, &kv.key, (size_t *)&kv.keyLen); @@ -49,12 +51,14 @@ void *tDeserializeClientHbReq(void *buf, SClientHbReq *pReq) { buf = taosDecodeSClientHbKey(buf, &pReq->connKey); // TODO: error handling - if (pReq->info == NULL) { - pReq->info = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_NO_LOCK); + int kvNum; + taosDecodeFixedI32(buf, &kvNum); + pReq->info = taosHashInit(kvNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_NO_LOCK); + for(int i = 0; i < kvNum; i++) { + SKv kv; + buf = taosDecodeSKv(buf, &kv); + taosHashPut(pReq->info, kv.key, kv.keyLen, kv.value, kv.valueLen); } - SKv kv; - buf = taosDecodeSKv(buf, &kv); - taosHashPut(pReq->info, kv.key, kv.keyLen, kv.value, kv.valueLen); return buf; } diff --git a/source/dnode/mnode/impl/src/mndSync.c b/source/dnode/mnode/impl/src/mndSync.c index 47d0ce4105..e55da73f62 100644 --- a/source/dnode/mnode/impl/src/mndSync.c +++ b/source/dnode/mnode/impl/src/mndSync.c @@ -95,11 +95,11 @@ static int32_t mndRestoreWal(SMnode *pMnode) { if (sdbWriteFile(pSdb) != 0) { goto WAL_RESTORE_OVER; } - } if (walEndSnapshot(pWal) < 0) { goto WAL_RESTORE_OVER; } + } code = 0; @@ -181,4 +181,4 @@ int32_t mndSyncPropose(SMnode *pMnode, SSdbRaw *pRaw) { bool mndIsMaster(SMnode *pMnode) { SSyncMgmt *pMgmt = &pMnode->syncMgmt; return pMgmt->state == TAOS_SYNC_STATE_LEADER; -} \ No newline at end of file +} diff --git a/source/libs/wal/inc/walInt.h b/source/libs/wal/inc/walInt.h index 871c95193f..7631593dd8 100644 --- a/source/libs/wal/inc/walInt.h +++ b/source/libs/wal/inc/walInt.h @@ -131,6 +131,7 @@ int walMetaDeserialize(SWal* pWal, const char* bytes); // seek section int walChangeFile(SWal* pWal, int64_t ver); +int walChangeFileToLast(SWal* pWal); // seek section end int64_t walGetSeq(); diff --git a/source/libs/wal/src/walMeta.c b/source/libs/wal/src/walMeta.c index 6164ec2baf..270a26bf80 100644 --- a/source/libs/wal/src/walMeta.c +++ b/source/libs/wal/src/walMeta.c @@ -184,6 +184,7 @@ int walMetaDeserialize(SWal* pWal, const char* bytes) { } taosArraySetSize(pArray, sz); pWal->fileInfoSet = pArray; + pWal->writeCur = sz - 1; cJSON_Delete(pRoot); return 0; } diff --git a/source/libs/wal/src/walWrite.c b/source/libs/wal/src/walWrite.c index 5eda16b061..975f232e3d 100644 --- a/source/libs/wal/src/walWrite.c +++ b/source/libs/wal/src/walWrite.c @@ -254,6 +254,7 @@ static int walWriteIndex(SWal *pWal, int64_t ver, int64_t offset) { SWalIdxEntry entry = {.ver = ver, .offset = offset}; int size = tfWrite(pWal->writeIdxTfd, &entry, sizeof(SWalIdxEntry)); if (size != sizeof(SWalIdxEntry)) { + terrno = TAOS_SYSTEM_ERROR(errno); // TODO truncate return -1; } @@ -287,7 +288,13 @@ int64_t walWrite(SWal *pWal, int64_t index, uint8_t msgType, const void *body, i } /*if (!tfValid(pWal->writeLogTfd)) return -1;*/ + ASSERT(pWal->writeCur >= 0); + pthread_mutex_lock(&pWal->mutex); + if (pWal->writeIdxTfd == -1 || pWal->writeLogTfd == -1) { + walChangeFileToLast(pWal); + } + pWal->writeHead.head.version = index; int64_t offset = walGetCurFileOffset(pWal); @@ -309,6 +316,7 @@ int64_t walWrite(SWal *pWal, int64_t index, uint8_t msgType, const void *body, i wError("vgId:%d, file:%" PRId64 ".log, failed to write since %s", pWal->cfg.vgId, walGetLastFileFirstVer(pWal), strerror(errno)); } + code = walWriteIndex(pWal, index, offset); if (code != 0) { // TODO From 3d7b737bbc4b7c9c245cd723bfc6f5f0b2d64771 Mon Sep 17 00:00:00 2001 From: Liu Jicong Date: Mon, 10 Jan 2022 16:23:35 +0800 Subject: [PATCH 5/5] fix compile error --- source/dnode/mnode/impl/src/mndSync.c | 13 +++++++------ source/libs/planner/CMakeLists.txt | 3 ++- 2 files changed, 9 insertions(+), 7 deletions(-) diff --git a/source/dnode/mnode/impl/src/mndSync.c b/source/dnode/mnode/impl/src/mndSync.c index 581b57ea65..591367c519 100644 --- a/source/dnode/mnode/impl/src/mndSync.c +++ b/source/dnode/mnode/impl/src/mndSync.c @@ -92,13 +92,14 @@ static int32_t mndRestoreWal(SMnode *pMnode) { goto WAL_RESTORE_OVER; } - if (walBeginSnapshot(pWal, sdbVer) < 0) { - goto WAL_RESTORE_OVER; - } + if (walBeginSnapshot(pWal, sdbVer) < 0) { + goto WAL_RESTORE_OVER; + } + + if (walEndSnapshot(pWal) < 0) { + goto WAL_RESTORE_OVER; + } - if (walEndSnapshot(pWal) < 0) { - goto WAL_RESTORE_OVER; - } } code = 0; diff --git a/source/libs/planner/CMakeLists.txt b/source/libs/planner/CMakeLists.txt index 6234dbe0ac..8d8c148fde 100644 --- a/source/libs/planner/CMakeLists.txt +++ b/source/libs/planner/CMakeLists.txt @@ -8,7 +8,8 @@ target_include_directories( target_link_libraries( planner - PRIVATE os util catalog cjson parser transport function qcom + PRIVATE os util catalog cjson parser function qcom + PUBLIC transport ) if(${BUILD_TEST})