diff --git a/include/libs/transport/thttp.h b/include/libs/transport/thttp.h index 9a6aee418794b61abd88e132b42964c56c69451c..f6f1f7f02771b6a3cefbd85864d01e9b46b39b29 100644 --- a/include/libs/transport/thttp.h +++ b/include/libs/transport/thttp.h @@ -17,6 +17,7 @@ #define _TD_UTIL_HTTP_H_ #include "os.h" +#include "tref.h" #ifdef __cplusplus extern "C" { @@ -24,7 +25,8 @@ extern "C" { typedef enum { HTTP_GZIP, HTTP_FLAT } EHttpCompFlag; -int32_t taosSendHttpReport(const char* server, const char* uri, uint16_t port, char* pCont, int32_t contLen, EHttpCompFlag flag); +int32_t taosSendHttpReport(const char* server, const char* uri, uint16_t port, char* pCont, int32_t contLen, + EHttpCompFlag flag); #ifdef __cplusplus } diff --git a/source/dnode/mgmt/node_util/inc/dmUtil.h b/source/dnode/mgmt/node_util/inc/dmUtil.h index c2f403dfbb4e827cb137a26056e8d7cec41fcf0a..55ee6d697337456d113ba5eb8c993fe923042f48 100644 --- a/source/dnode/mgmt/node_util/inc/dmUtil.h +++ b/source/dnode/mgmt/node_util/inc/dmUtil.h @@ -53,12 +53,12 @@ extern "C" { #define dDebug(...) { if (dDebugFlag & DEBUG_DEBUG) { taosPrintLog("DND ", DEBUG_DEBUG, dDebugFlag, __VA_ARGS__); }} #define dTrace(...) { if (dDebugFlag & DEBUG_TRACE) { taosPrintLog("DND ", DEBUG_TRACE, dDebugFlag, __VA_ARGS__); }} -#define dGFatal(param, ...) { char buf[40] = {0}; TRACE_TO_STR(trace, buf); dFatal(param ", gtid:%s", __VA_ARGS__, buf);} -#define dGError(param, ...) { char buf[40] = {0}; TRACE_TO_STR(trace, buf); dError(param ", gtid:%s", __VA_ARGS__, buf);} -#define dGWarn(param, ...) { char buf[40] = {0}; TRACE_TO_STR(trace, buf); dWarn (param ", gtid:%s", __VA_ARGS__, buf);} -#define dGInfo(param, ...) { char buf[40] = {0}; TRACE_TO_STR(trace, buf); dInfo (param ", gtid:%s", __VA_ARGS__, buf);} -#define dGDebug(param, ...) { char buf[40] = {0}; TRACE_TO_STR(trace, buf); dDebug(param ", gtid:%s", __VA_ARGS__, buf);} -#define dGTrace(param, ...) { char buf[40] = {0}; TRACE_TO_STR(trace, buf); dTrace(param ", gtid:%s", __VA_ARGS__, buf);} +#define dGFatal(param, ...) {if (dDebugFlag & DEBUG_FATAL) { char buf[40] = {0}; TRACE_TO_STR(trace, buf); dFatal(param ", gtid:%s", __VA_ARGS__, buf);}} +#define dGError(param, ...) {if (dDebugFlag & DEBUG_ERROR) { char buf[40] = {0}; TRACE_TO_STR(trace, buf); dError(param ", gtid:%s", __VA_ARGS__, buf);}} +#define dGWarn(param, ...) {if (dDebugFlag & DEBUG_WARN) { char buf[40] = {0}; TRACE_TO_STR(trace, buf); dWarn(param ", gtid:%s", __VA_ARGS__, buf);}} +#define dGInfo(param, ...) {if (dDebugFlag & DEBUG_INFO) { char buf[40] = {0}; TRACE_TO_STR(trace, buf); dInfo(param ", gtid:%s", __VA_ARGS__, buf);}} +#define dGDebug(param, ...) {if (dDebugFlag & DEBUG_DEBUG) { char buf[40] = {0}; TRACE_TO_STR(trace, buf); dDebug(param ", gtid:%s", __VA_ARGS__, buf);}} +#define dGTrace(param, ...) {if (dDebugFlag & DEBUG_TRACE) { char buf[40] = {0}; TRACE_TO_STR(trace, buf); dTrace(param ", gtid:%s", __VA_ARGS__, buf);}} // clang-format on diff --git a/source/dnode/mnode/impl/inc/mndInt.h b/source/dnode/mnode/impl/inc/mndInt.h index d851578a744f740cf0e211ee6928b2ec3f0dd705..006c74ef3de223a34aaa483346b28102bfdfe54a 100644 --- a/source/dnode/mnode/impl/inc/mndInt.h +++ b/source/dnode/mnode/impl/inc/mndInt.h @@ -41,12 +41,12 @@ extern "C" { #define mDebug(...) { if (mDebugFlag & DEBUG_DEBUG) { taosPrintLog("MND ", DEBUG_DEBUG, mDebugFlag, __VA_ARGS__); }} #define mTrace(...) { if (mDebugFlag & DEBUG_TRACE) { taosPrintLog("MND ", DEBUG_TRACE, mDebugFlag, __VA_ARGS__); }} -#define mGFatal(param, ...) { char buf[40] = {0}; TRACE_TO_STR(trace, buf); mFatal(param ", gtid:%s", __VA_ARGS__, buf);} -#define mGError(param, ...) { char buf[40] = {0}; TRACE_TO_STR(trace, buf); mError(param ", gtid:%s", __VA_ARGS__, buf);} -#define mGWarn(param, ...) { char buf[40] = {0}; TRACE_TO_STR(trace, buf); mWarn (param ", gtid:%s", __VA_ARGS__, buf);} -#define mGInfo(param, ...) { char buf[40] = {0}; TRACE_TO_STR(trace, buf); mInfo (param ", gtid:%s", __VA_ARGS__, buf);} -#define mGDebug(param, ...) { char buf[40] = {0}; TRACE_TO_STR(trace, buf); mDebug(param ", gtid:%s", __VA_ARGS__, buf);} -#define mGTrace(param, ...) { char buf[40] = {0}; TRACE_TO_STR(trace, buf); mTrace(param ", gtid:%s", __VA_ARGS__, buf);} +#define mGFatal(param, ...) { if (mDebugFlag & DEBUG_FATAL){ char buf[40] = {0}; TRACE_TO_STR(trace, buf); mFatal(param ", gtid:%s", __VA_ARGS__, buf);}} +#define mGError(param, ...) { if (mDebugFlag & DEBUG_ERROR){ char buf[40] = {0}; TRACE_TO_STR(trace, buf); mError(param ", gtid:%s", __VA_ARGS__, buf);}} +#define mGWarn(param, ...) { if (mDebugFlag & DEBUG_WARN){ char buf[40] = {0}; TRACE_TO_STR(trace, buf); mWarn (param ", gtid:%s", __VA_ARGS__, buf);}} +#define mGInfo(param, ...) { if (mDebugFlag & DEBUG_INFO){ char buf[40] = {0}; TRACE_TO_STR(trace, buf); mInfo (param ", gtid:%s", __VA_ARGS__, buf);}} +#define mGDebug(param, ...) { if (mDebugFlag & DEBUG_DEBUG){ char buf[40] = {0}; TRACE_TO_STR(trace, buf); mDebug(param ", gtid:%s", __VA_ARGS__, buf);}} +#define mGTrace(param, ...) { if (mDebugFlag & DEBUG_TRACE){ char buf[40] = {0}; TRACE_TO_STR(trace, buf); mTrace(param ", gtid:%s", __VA_ARGS__, buf);}} // clang-format on #define SYSTABLE_SCH_TABLE_NAME_LEN ((TSDB_TABLE_NAME_LEN - 1) + VARSTR_HEADER_SIZE) @@ -80,7 +80,7 @@ typedef struct { typedef struct { TdThreadMutex lock; - char email[TSDB_FQDN_LEN]; + char email[TSDB_FQDN_LEN]; } STelemMgmt; typedef struct { diff --git a/source/dnode/vnode/src/meta/metaTable.c b/source/dnode/vnode/src/meta/metaTable.c index 94aa464354ca4755fa2f8ca876bd4b6feee3b1f6..3325f4055ce190ac119d7fe6eb274954a8b24d2a 100644 --- a/source/dnode/vnode/src/meta/metaTable.c +++ b/source/dnode/vnode/src/meta/metaTable.c @@ -1921,10 +1921,10 @@ int metaCreateTagIdxKey(tb_uid_t suid, int32_t cid, const void *pTagData, int32_ // refactor if (IS_VAR_DATA_TYPE(type)) { memcpy((*ppTagIdxKey)->data, (uint16_t *)&nTagData, VARSTR_HEADER_SIZE); - memcpy((*ppTagIdxKey)->data + VARSTR_HEADER_SIZE, pTagData, nTagData); + if (pTagData != NULL) memcpy((*ppTagIdxKey)->data + VARSTR_HEADER_SIZE, pTagData, nTagData); *(tb_uid_t *)((*ppTagIdxKey)->data + VARSTR_HEADER_SIZE + nTagData) = uid; } else { - memcpy((*ppTagIdxKey)->data, pTagData, nTagData); + if (pTagData != NULL) memcpy((*ppTagIdxKey)->data, pTagData, nTagData); *(tb_uid_t *)((*ppTagIdxKey)->data + nTagData) = uid; } diff --git a/source/libs/index/src/indexFstFile.c b/source/libs/index/src/indexFstFile.c index 5a9c8dfe3d3906bd452111bf885e62b65638f6e5..40c50ed9cb97fd0ea64eb8d50bcf04801a24a89b 100644 --- a/source/libs/index/src/indexFstFile.c +++ b/source/libs/index/src/indexFstFile.c @@ -18,6 +18,7 @@ #include "indexInt.h" #include "indexUtil.h" #include "os.h" +#include "osDef.h" #include "tutil.h" static int32_t kBlockSize = 4096; @@ -172,7 +173,8 @@ static FORCE_INLINE int idxFileCtxDoFlush(IFileCtx* ctx) { int32_t nw = taosWriteFile(ctx->file.pFile, ctx->file.wBuf, ctx->file.wBufOffset); ctx->file.wBufOffset = 0; } - taosFsyncFile(ctx->file.pFile); + int ret = taosFsyncFile(ctx->file.pFile); + UNUSED(ret); } else { // do nothing } @@ -180,11 +182,11 @@ static FORCE_INLINE int idxFileCtxDoFlush(IFileCtx* ctx) { } IFileCtx* idxFileCtxCreate(WriterType type, const char* path, bool readOnly, int32_t capacity) { + int code = 0; IFileCtx* ctx = taosMemoryCalloc(1, sizeof(IFileCtx)); if (ctx == NULL) { return NULL; } - ctx->type = type; if (ctx->type == TFILE) { // ugly code, refactor later @@ -192,15 +194,21 @@ IFileCtx* idxFileCtxCreate(WriterType type, const char* path, bool readOnly, int memcpy(ctx->file.buf, path, strlen(path)); if (readOnly == false) { ctx->file.pFile = taosOpenFile(path, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_APPEND); - taosFtruncateFile(ctx->file.pFile, 0); - taosStatFile(path, &ctx->file.size, NULL); + + code = taosFtruncateFile(ctx->file.pFile, 0); + UNUSED(code); + + code = taosStatFile(path, &ctx->file.size, NULL); + UNUSED(code); ctx->file.wBufOffset = 0; ctx->file.wBufCap = kBlockSize * 4; ctx->file.wBuf = taosMemoryCalloc(1, ctx->file.wBufCap); } else { ctx->file.pFile = taosOpenFile(path, TD_FILE_READ); - taosFStatFile(ctx->file.pFile, &ctx->file.size, NULL); + code = taosFStatFile(ctx->file.pFile, &ctx->file.size, NULL); + UNUSED(code); + ctx->file.wBufOffset = 0; #ifdef USE_MMAP diff --git a/source/libs/transport/src/thttp.c b/source/libs/transport/src/thttp.c index bab86948fae5e718c39d7090690f8833c80f30c1..04b546b36a48bd1df071c7b077d82c57c2ddea8a 100644 --- a/source/libs/transport/src/thttp.c +++ b/source/libs/transport/src/thttp.c @@ -26,6 +26,8 @@ #define HTTP_RECV_BUF_SIZE 1024 +static int32_t httpRefMgt = 0; +static int64_t httpRef = -1; typedef struct SHttpModule { uv_loop_t* loop; SAsyncPool* asyncPool; @@ -41,7 +43,6 @@ typedef struct SHttpMsg { int32_t len; EHttpCompFlag flag; int8_t quit; - SHttpModule* http; } SHttpMsg; @@ -57,7 +58,6 @@ typedef struct SHttpClient { } SHttpClient; static TdThreadOnce transHttpInit = PTHREAD_ONCE_INIT; -static SHttpModule* thttp = NULL; static void transHttpEnvInit(); static void httpHandleReq(SHttpMsg* msg); @@ -280,26 +280,29 @@ static void clientConnCb(uv_connect_t* req, int32_t status) { } int32_t httpSendQuit() { + SHttpModule* http = taosAcquireRef(httpRefMgt, httpRef); + if (http == NULL) return 0; + SHttpMsg* msg = taosMemoryCalloc(1, sizeof(SHttpMsg)); msg->quit = 1; - SHttpModule* load = atomic_load_ptr(&thttp); - if (load == NULL) { - httpDestroyMsg(msg); - tError("http-report already released"); - return -1; - } else { - msg->http = load; - } - transAsyncSend(load->asyncPool, &(msg->q)); + transAsyncSend(http->asyncPool, &(msg->q)); + taosReleaseRef(httpRefMgt, httpRef); return 0; } static int32_t taosSendHttpReportImpl(const char* server, const char* uri, uint16_t port, char* pCont, int32_t contLen, EHttpCompFlag flag) { + SHttpModule* load = taosAcquireRef(httpRefMgt, httpRef); + if (load == NULL) { + tError("http-report already released"); + return -1; + } + SHttpMsg* msg = taosMemoryMalloc(sizeof(SHttpMsg)); + msg->server = taosStrdup(server); - msg->uri = taosStrdup(uri); + msg->uri = taosStrdup(uri); msg->port = port; msg->cont = taosMemoryMalloc(contLen); memcpy(msg->cont, pCont, contLen); @@ -307,15 +310,9 @@ static int32_t taosSendHttpReportImpl(const char* server, const char* uri, uint1 msg->flag = flag; msg->quit = 0; - SHttpModule* load = atomic_load_ptr(&thttp); - if (load == NULL) { - httpDestroyMsg(msg); - tError("http-report already released"); - return -1; - } - - msg->http = load; - return transAsyncSend(load->asyncPool, &(msg->q)); + int ret = transAsyncSend(load->asyncPool, &(msg->q)); + taosReleaseRef(httpRefMgt, httpRef); + return ret; } static void httpDestroyClientCb(uv_handle_t* handle) { @@ -335,13 +332,19 @@ static void httpWalkCb(uv_handle_t* handle, void* arg) { return; } static void httpHandleQuit(SHttpMsg* msg) { - SHttpModule* http = msg->http; taosMemoryFree(msg); + SHttpModule* http = taosAcquireRef(httpRefMgt, httpRef); + if (http == NULL) return; + uv_walk(http->loop, httpWalkCb, NULL); + taosReleaseRef(httpRefMgt, httpRef); } static void httpHandleReq(SHttpMsg* msg) { - SHttpModule* http = msg->http; + SHttpModule* http = taosAcquireRef(httpRefMgt, httpRef); + if (http == NULL) { + goto END; + } struct sockaddr_in dest = {0}; if (taosBuildDstAddr(msg->server, msg->port, &dest) < 0) { @@ -391,6 +394,7 @@ static void httpHandleReq(SHttpMsg* msg) { int ret = uv_tcp_open((uv_tcp_t*)&cli->tcp, fd); if (ret != 0) { tError("http-report failed to open socket, reason:%s, dst:%s:%d", uv_strerror(ret), cli->addr, cli->port); + taosReleaseRef(httpRefMgt, httpRef); destroyHttpClient(cli); return; } @@ -401,21 +405,26 @@ static void httpHandleReq(SHttpMsg* msg) { cli->port); destroyHttpClient(cli); } + taosReleaseRef(httpRefMgt, httpRef); return; END: tError("http-report failed to report, reason: %s, addr: %s:%d", terrstr(), msg->server, msg->port); httpDestroyMsg(msg); + taosReleaseRef(httpRefMgt, httpRef); } -int32_t taosSendHttpReport(const char* server, const char* uri, uint16_t port, char* pCont, int32_t contLen, EHttpCompFlag flag) { +int32_t taosSendHttpReport(const char* server, const char* uri, uint16_t port, char* pCont, int32_t contLen, + EHttpCompFlag flag) { taosThreadOnce(&transHttpInit, transHttpEnvInit); return taosSendHttpReportImpl(server, uri, port, pCont, contLen, flag); } +static void transHttpDestroyHandle(void* handle) { taosMemoryFree(handle); } static void transHttpEnvInit() { - SHttpModule* http = taosMemoryMalloc(sizeof(SHttpModule)); + httpRefMgt = taosOpenRef(1, transHttpDestroyHandle); + SHttpModule* http = taosMemoryMalloc(sizeof(SHttpModule)); http->loop = taosMemoryMalloc(sizeof(uv_loop_t)); uv_loop_init(http->loop); @@ -426,21 +435,22 @@ static void transHttpEnvInit() { http = NULL; return; } - + int err = taosThreadCreate(&http->thread, NULL, httpThread, (void*)http); if (err != 0) { taosMemoryFree(http->loop); taosMemoryFree(http); http = NULL; } - atomic_store_ptr(&thttp, http); + httpRef = taosAddRef(httpRefMgt, http); } void transHttpEnvDestroy() { - SHttpModule* load = atomic_load_ptr(&thttp); - if (load == NULL) { + // remove http + if (httpRef == -1) { return; } + SHttpModule* load = taosAcquireRef(httpRefMgt, httpRef); httpSendQuit(); taosThreadJoin(load->thread, NULL); @@ -448,7 +458,7 @@ void transHttpEnvDestroy() { transAsyncPoolDestroy(load->asyncPool); uv_loop_close(load->loop); taosMemoryFree(load->loop); - taosMemoryFree(load); - atomic_store_ptr(&thttp, NULL); + taosReleaseRef(httpRefMgt, httpRef); + taosRemoveRef(httpRefMgt, httpRef); } diff --git a/source/libs/transport/src/transCli.c b/source/libs/transport/src/transCli.c index e88308c4320080b6faa9822a26bc74e600d7934b..8dc7df0264867c5ea4cfe42cdd9c3e64a3672a63 100644 --- a/source/libs/transport/src/transCli.c +++ b/source/libs/transport/src/transCli.c @@ -12,7 +12,9 @@ * along with this program. If not, see . */ +// #include "osMemory.h" #include "transComm.h" +#include "tutil.h" typedef struct SConnList { queue conns; @@ -65,15 +67,13 @@ typedef struct SCliConn { SCliBatch* pBatch; - int64_t refId; - char* ip; - SDelayTask* task; - // debug and log info - char src[32]; - char dst[32]; + char* ip; + char src[32]; + char dst[32]; + int64_t refId; } SCliConn; typedef struct SCliMsg { @@ -132,6 +132,7 @@ typedef struct { int32_t threshold; int64_t interval; } SFailFastItem; + // conn pool // add expire timeout and capacity limit static void* createConnPool(int size); @@ -224,9 +225,13 @@ static void cliWalkCb(uv_handle_t* handle, void* arg); } while (0); // snprintf may cause performance problem -#define CONN_CONSTRUCT_HASH_KEY(key, ip, port) \ - do { \ - snprintf(key, sizeof(key), "%s:%d", ip, (int)port); \ +#define CONN_CONSTRUCT_HASH_KEY(key, ip, port) \ + do { \ + char* t = key; \ + int16_t len = strlen(ip); \ + if (ip != NULL) memcpy(t, ip, len); \ + t[len] = ':'; \ + titoa(port, 10, &t[len + 1]); \ } while (0) #define CONN_PERSIST_TIME(para) ((para) <= 90000 ? 90000 : (para)) @@ -330,12 +335,8 @@ bool cliMaySendCachedMsg(SCliConn* conn) { if (!transQueueEmpty(&conn->cliMsgs)) { SCliMsg* pCliMsg = NULL; CONN_GET_NEXT_SENDMSG(conn); - if (pCliMsg == NULL) - return false; - else { - cliSend(conn); - return true; - } + cliSend(conn); + return true; } return false; _RETURN: @@ -359,6 +360,7 @@ void cliHandleResp(SCliConn* conn) { int32_t msgLen = transDumpFromBuffer(&conn->readBuf, (char**)&pHead); if (msgLen <= 0) { + taosMemoryFree(pHead); tDebug("%s conn %p recv invalid packet ", CONN_GET_INST_LABEL(conn), conn); return; } @@ -1727,17 +1729,23 @@ void* transInitClient(uint32_t ip, uint32_t port, char* label, int numOfThreads, for (int i = 0; i < cli->numOfThreads; i++) { SCliThrd* pThrd = createThrdObj(shandle); if (pThrd == NULL) { - return NULL; + goto _err; } int err = taosThreadCreate(&pThrd->thread, NULL, cliWorkThread, (void*)(pThrd)); - if (err == 0) { + if (err != 0) { + goto _err; + } else { tDebug("success to create tranport-cli thread:%d", i); } cli->pThreadObj[i] = pThrd; } - return cli; + +_err: + taosMemoryFree(cli->pThreadObj); + taosMemoryFree(cli); + return NULL; } static FORCE_INLINE void destroyUserdata(STransMsg* userdata) {