From 771912e58c4c5bffacd4317bd4a03f70a86c7393 Mon Sep 17 00:00:00 2001 From: yihaoDeng Date: Thu, 5 Sep 2024 12:44:50 +0800 Subject: [PATCH 1/6] opt parameter --- include/util/tdef.h | 2 +- source/libs/transport/src/transSvr.c | 11 +++++++++-- 2 files changed, 10 insertions(+), 3 deletions(-) diff --git a/include/util/tdef.h b/include/util/tdef.h index a750074953..46a0d01457 100644 --- a/include/util/tdef.h +++ b/include/util/tdef.h @@ -499,7 +499,7 @@ typedef enum ELogicConditionType { #ifdef WINDOWS #define TSDB_MAX_RPC_THREADS 4 // windows pipe only support 4 connections. #else -#define TSDB_MAX_RPC_THREADS 10 +#define TSDB_MAX_RPC_THREADS 50 #endif #define TSDB_QUERY_TYPE_NON_TYPE 0x00u // none type diff --git a/source/libs/transport/src/transSvr.c b/source/libs/transport/src/transSvr.c index 53a7dee7be..e25851fa23 100644 --- a/source/libs/transport/src/transSvr.c +++ b/source/libs/transport/src/transSvr.c @@ -373,6 +373,7 @@ static bool uvHandleReq(SSvrConn* pConn) { STrans* pTransInst = pConn->pTransInst; SWorkThrd* pThrd = pConn->hostThrd; + int8_t acquire = 0; STransMsgHead* pHead = NULL; int8_t resetBuf = pConn->status == ConnAcquire ? 0 : 1; @@ -459,7 +460,13 @@ static bool uvHandleReq(SSvrConn* pConn) { // 2. once send out data, cli conn released to conn pool immediately // 3. not mixed with persist transMsg.info.ahandle = (void*)pHead->ahandle; - transMsg.info.handle = (void*)transAcquireExHandle(transGetRefMgt(), pConn->refId); + + if (pHead->noResp == 1) { + transMsg.info.handle = NULL; + } else { + transMsg.info.handle = (void*)transAcquireExHandle(transGetRefMgt(), pConn->refId); + acquire = 1; + } transMsg.info.refId = pConn->refId; transMsg.info.traceId = pHead->traceId; transMsg.info.cliVer = htonl(pHead->compatibilityVer); @@ -483,7 +490,7 @@ static bool uvHandleReq(SSvrConn* pConn) { pConnInfo->clientPort = pConn->port; tstrncpy(pConnInfo->user, pConn->user, sizeof(pConnInfo->user)); - (void)transReleaseExHandle(transGetRefMgt(), pConn->refId); + if (acquire) transReleaseExHandle(transGetRefMgt(), pConn->refId); (*pTransInst->cfp)(pTransInst->parent, &transMsg, NULL); return true; From 96f46edffccc8edf04e0d92401ac05dbb3d47324 Mon Sep 17 00:00:00 2001 From: yihaoDeng Date: Thu, 5 Sep 2024 12:52:19 +0800 Subject: [PATCH 2/6] opt parameter --- source/libs/transport/src/transSvr.c | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/source/libs/transport/src/transSvr.c b/source/libs/transport/src/transSvr.c index e25851fa23..ad3e132de4 100644 --- a/source/libs/transport/src/transSvr.c +++ b/source/libs/transport/src/transSvr.c @@ -475,10 +475,10 @@ static bool uvHandleReq(SSvrConn* pConn) { tGTrace("%s handle %p conn:%p translated to app, refId:%" PRIu64, transLabel(pTransInst), transMsg.info.handle, pConn, pConn->refId); - if (transMsg.info.handle == NULL) { - tError("%s handle %p conn:%p handle failed to init" PRIu64, transLabel(pTransInst), transMsg.info.handle, pConn); - return false; - } + // if (transMsg.info.handle == NULL) { + // tError("%s handle %p conn:%p handle failed to init" PRIu64, transLabel(pTransInst), transMsg.info.handle, pConn); + // return false; + // } if (pHead->noResp == 1) { transMsg.info.refId = -1; From 59946626d9a1c26ea3b73278d53ce9f87fa9f553 Mon Sep 17 00:00:00 2001 From: yihaoDeng Date: Thu, 5 Sep 2024 13:25:41 +0800 Subject: [PATCH 3/6] opt parameter --- source/libs/transport/src/transComm.c | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/source/libs/transport/src/transComm.c b/source/libs/transport/src/transComm.c index 1758ca65cc..1329489be6 100644 --- a/source/libs/transport/src/transComm.c +++ b/source/libs/transport/src/transComm.c @@ -20,6 +20,7 @@ static TdThreadOnce transModuleInit = PTHREAD_ONCE_INIT; static int32_t refMgt; +static int32_t svrRefMgt; static int32_t instMgt; static int32_t transSyncMsgMgt; @@ -704,12 +705,14 @@ bool transEpSetIsEqual2(SEpSet* a, SEpSet* b) { static void transInitEnv() { refMgt = transOpenRefMgt(50000, transDestroyExHandle); + svrRefMgt = transOpenRefMgt(50000, transDestroyExHandle); instMgt = taosOpenRef(50, rpcCloseImpl); transSyncMsgMgt = taosOpenRef(50, transDestroySyncMsg); (void)uv_os_setenv("UV_TCP_SINGLE_ACCEPT", "1"); } static void transDestroyEnv() { transCloseRefMgt(refMgt); + transCloseRefMgt(svrRefMgt); transCloseRefMgt(instMgt); transCloseRefMgt(transSyncMsgMgt); } @@ -724,6 +727,7 @@ int32_t transInit() { } int32_t transGetRefMgt() { return refMgt; } +int32_t transGetSvrRefMgt() { return svrRefMgt; } int32_t transGetInstMgt() { return instMgt; } int32_t transGetSyncMsgMgt() { return transSyncMsgMgt; } From ccea816fb9e4cd68c785697d082ffc7f54e387de Mon Sep 17 00:00:00 2001 From: yihaoDeng Date: Thu, 5 Sep 2024 13:25:56 +0800 Subject: [PATCH 4/6] opt transport --- source/libs/transport/inc/transComm.h | 1 + source/libs/transport/src/transSvr.c | 50 +++++++++++++-------------- 2 files changed, 26 insertions(+), 25 deletions(-) diff --git a/source/libs/transport/inc/transComm.h b/source/libs/transport/inc/transComm.h index 84f0ffc8cb..8664305208 100644 --- a/source/libs/transport/inc/transComm.h +++ b/source/libs/transport/inc/transComm.h @@ -443,6 +443,7 @@ int32_t transReleaseExHandle(int32_t refMgt, int64_t refId); void transDestroyExHandle(void* handle); int32_t transGetRefMgt(); +int32_t transGetSvrRefMgt(); int32_t transGetInstMgt(); int32_t transGetSyncMsgMgt(); diff --git a/source/libs/transport/src/transSvr.c b/source/libs/transport/src/transSvr.c index ad3e132de4..11aa468b19 100644 --- a/source/libs/transport/src/transSvr.c +++ b/source/libs/transport/src/transSvr.c @@ -464,7 +464,7 @@ static bool uvHandleReq(SSvrConn* pConn) { if (pHead->noResp == 1) { transMsg.info.handle = NULL; } else { - transMsg.info.handle = (void*)transAcquireExHandle(transGetRefMgt(), pConn->refId); + transMsg.info.handle = (void*)transAcquireExHandle(transGetSvrRefMgt(), pConn->refId); acquire = 1; } transMsg.info.refId = pConn->refId; @@ -490,7 +490,7 @@ static bool uvHandleReq(SSvrConn* pConn) { pConnInfo->clientPort = pConn->port; tstrncpy(pConnInfo->user, pConn->user, sizeof(pConnInfo->user)); - if (acquire) transReleaseExHandle(transGetRefMgt(), pConn->refId); + if (acquire) transReleaseExHandle(transGetSvrRefMgt(), pConn->refId); (*pTransInst->cfp)(pTransInst->parent, &transMsg, NULL); return true; @@ -777,15 +777,15 @@ void uvWorkerAsyncCb(uv_async_t* handle) { SExHandle* exh1 = transMsg.info.handle; int64_t refId = transMsg.info.refId; - SExHandle* exh2 = transAcquireExHandle(transGetRefMgt(), refId); + SExHandle* exh2 = transAcquireExHandle(transGetSvrRefMgt(), refId); if (exh2 == NULL || exh1 != exh2) { tTrace("handle except msg %p, ignore it", exh1); - (void)transReleaseExHandle(transGetRefMgt(), refId); + (void)transReleaseExHandle(transGetSvrRefMgt(), refId); destroySmsg(msg); continue; } msg->pConn = exh1->handle; - (void)transReleaseExHandle(transGetRefMgt(), refId); + (void)transReleaseExHandle(transGetSvrRefMgt(), refId); (*transAsyncHandle[msg->type])(msg, pThrd); } } @@ -881,15 +881,15 @@ static void uvPrepareCb(uv_prepare_t* handle) { SExHandle* exh1 = transMsg.info.handle; int64_t refId = transMsg.info.refId; - SExHandle* exh2 = transAcquireExHandle(transGetRefMgt(), refId); + SExHandle* exh2 = transAcquireExHandle(transGetSvrRefMgt(), refId); if (exh2 == NULL || exh1 != exh2) { tTrace("handle except msg %p, ignore it", exh1); - (void)transReleaseExHandle(transGetRefMgt(), refId); + (void)transReleaseExHandle(transGetSvrRefMgt(), refId); destroySmsg(msg); continue; } msg->pConn = exh1->handle; - (void)transReleaseExHandle(transGetRefMgt(), refId); + (void)transReleaseExHandle(transGetSvrRefMgt(), refId); (*transAsyncHandle[msg->type])(msg, pThrd); } } @@ -1222,14 +1222,14 @@ static FORCE_INLINE SSvrConn* createConn(void* hThrd) { exh->handle = pConn; exh->pThrd = pThrd; - exh->refId = transAddExHandle(transGetRefMgt(), exh); + exh->refId = transAddExHandle(transGetSvrRefMgt(), exh); if (exh->refId < 0) { TAOS_CHECK_GOTO(TSDB_CODE_REF_INVALID_ID, NULL, _end); } QUEUE_INIT(&exh->q); - SExHandle* pSelf = transAcquireExHandle(transGetRefMgt(), exh->refId); + SExHandle* pSelf = transAcquireExHandle(transGetSvrRefMgt(), exh->refId); if (pSelf != exh) { TAOS_CHECK_GOTO(TSDB_CODE_REF_INVALID_ID, NULL, _end); } @@ -1291,8 +1291,8 @@ static FORCE_INLINE void destroyConnRegArg(SSvrConn* conn) { } static int32_t reallocConnRef(SSvrConn* conn) { if (conn->refId > 0) { - (void)transReleaseExHandle(transGetRefMgt(), conn->refId); - (void)transRemoveExHandle(transGetRefMgt(), conn->refId); + (void)transReleaseExHandle(transGetSvrRefMgt(), conn->refId); + (void)transRemoveExHandle(transGetSvrRefMgt(), conn->refId); } // avoid app continue to send msg on invalid handle SExHandle* exh = taosMemoryMalloc(sizeof(SExHandle)); @@ -1302,14 +1302,14 @@ static int32_t reallocConnRef(SSvrConn* conn) { exh->handle = conn; exh->pThrd = conn->hostThrd; - exh->refId = transAddExHandle(transGetRefMgt(), exh); + exh->refId = transAddExHandle(transGetSvrRefMgt(), exh); if (exh->refId < 0) { taosMemoryFree(exh); return TSDB_CODE_REF_INVALID_ID; } QUEUE_INIT(&exh->q); - SExHandle* pSelf = transAcquireExHandle(transGetRefMgt(), exh->refId); + SExHandle* pSelf = transAcquireExHandle(transGetSvrRefMgt(), exh->refId); if (pSelf != exh) { tError("conn %p failed to acquire handle", conn); taosMemoryFree(exh); @@ -1328,8 +1328,8 @@ static void uvDestroyConn(uv_handle_t* handle) { } SWorkThrd* thrd = conn->hostThrd; - (void)transReleaseExHandle(transGetRefMgt(), conn->refId); - (void)transRemoveExHandle(transGetRefMgt(), conn->refId); + (void)transReleaseExHandle(transGetSvrRefMgt(), conn->refId); + (void)transRemoveExHandle(transGetSvrRefMgt(), conn->refId); STrans* pTransInst = thrd->pTransInst; tDebug("%s conn %p destroy", transLabel(pTransInst), conn); @@ -1759,15 +1759,15 @@ int32_t transReleaseSrvHandle(void* handle) { tDebug("%s conn %p start to release", transLabel(pThrd->pTransInst), exh->handle); if ((code = transAsyncSend(pThrd->asyncPool, &m->q)) != 0) { destroySmsg(m); - (void)transReleaseExHandle(transGetRefMgt(), refId); + (void)transReleaseExHandle(transGetSvrRefMgt(), refId); return code; } - (void)transReleaseExHandle(transGetRefMgt(), refId); + (void)transReleaseExHandle(transGetSvrRefMgt(), refId); return 0; _return1: tDebug("handle %p failed to send to release handle", exh); - (void)transReleaseExHandle(transGetRefMgt(), refId); + (void)transReleaseExHandle(transGetSvrRefMgt(), refId); return code; _return2: tDebug("handle %p failed to send to release handle", exh); @@ -1810,17 +1810,17 @@ int32_t transSendResponse(const STransMsg* msg) { tGDebug("conn %p start to send resp (1/2)", exh->handle); if ((code = transAsyncSend(pThrd->asyncPool, &m->q)) != 0) { destroySmsg(m); - (void)transReleaseExHandle(transGetRefMgt(), refId); + (void)transReleaseExHandle(transGetSvrRefMgt(), refId); return code; } - (void)transReleaseExHandle(transGetRefMgt(), refId); + (void)transReleaseExHandle(transGetSvrRefMgt(), refId); return 0; _return1: tDebug("handle %p failed to send resp", exh); rpcFreeCont(msg->pCont); - (void)transReleaseExHandle(transGetRefMgt(), refId); + (void)transReleaseExHandle(transGetSvrRefMgt(), refId); return code; _return2: tDebug("handle %p failed to send resp", exh); @@ -1855,17 +1855,17 @@ int32_t transRegisterMsg(const STransMsg* msg) { tDebug("%s conn %p start to register brokenlink callback", transLabel(pTransInst), exh->handle); if ((code = transAsyncSend(pThrd->asyncPool, &m->q)) != 0) { destroySmsg(m); - (void)transReleaseExHandle(transGetRefMgt(), refId); + (void)transReleaseExHandle(transGetSvrRefMgt(), refId); return code; } - (void)transReleaseExHandle(transGetRefMgt(), refId); + (void)transReleaseExHandle(transGetSvrRefMgt(), refId); return 0; _return1: tDebug("handle %p failed to register brokenlink", exh); rpcFreeCont(msg->pCont); - (void)transReleaseExHandle(transGetRefMgt(), refId); + (void)transReleaseExHandle(transGetSvrRefMgt(), refId); return code; _return2: tDebug("handle %p failed to register brokenlink", exh); From 667da97c3a10f964a04148185a0a53578a5e6239 Mon Sep 17 00:00:00 2001 From: yihaoDeng Date: Thu, 5 Sep 2024 13:48:17 +0800 Subject: [PATCH 5/6] opt transport --- source/libs/transport/inc/transComm.h | 26 +++++++++++++------------- 1 file changed, 13 insertions(+), 13 deletions(-) diff --git a/source/libs/transport/inc/transComm.h b/source/libs/transport/inc/transComm.h index 8664305208..4bf9acfac4 100644 --- a/source/libs/transport/inc/transComm.h +++ b/source/libs/transport/inc/transComm.h @@ -272,19 +272,19 @@ bool transAsyncPoolIsEmpty(SAsyncPool* pool); } \ } while (0) -#define ASYNC_CHECK_HANDLE(exh1, id) \ - do { \ - if (id > 0) { \ - SExHandle* exh2 = transAcquireExHandle(transGetRefMgt(), id); \ - if (exh2 == NULL || id != exh2->refId) { \ - tDebug("ref:%" PRId64 " already released" PRIu64, id); \ - code = terrno; \ - goto _return1; \ - } \ - } else { \ - tWarn("invalid handle to release"); \ - goto _return2; \ - } \ +#define ASYNC_CHECK_HANDLE(exh1, id) \ + do { \ + if (id > 0) { \ + SExHandle* exh2 = transAcquireExHandle(transGetSvrRefMgt(), id); \ + if (exh2 == NULL || id != exh2->refId) { \ + tDebug("ref:%" PRId64 " already released", id); \ + code = terrno; \ + goto _return1; \ + } \ + } else { \ + tWarn("invalid handle to release"); \ + goto _return2; \ + } \ } while (0) int32_t transInitBuffer(SConnBuffer* buf); From 27d22b79b907775524c90307aad9efe3a893ba36 Mon Sep 17 00:00:00 2001 From: yihaoDeng Date: Thu, 5 Sep 2024 13:52:08 +0800 Subject: [PATCH 6/6] opt transport --- source/libs/transport/inc/transComm.h | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/source/libs/transport/inc/transComm.h b/source/libs/transport/inc/transComm.h index 4bf9acfac4..95ea6944e3 100644 --- a/source/libs/transport/inc/transComm.h +++ b/source/libs/transport/inc/transComm.h @@ -282,7 +282,7 @@ bool transAsyncPoolIsEmpty(SAsyncPool* pool); goto _return1; \ } \ } else { \ - tWarn("invalid handle to release"); \ + tDebug("invalid handle to release"); \ goto _return2; \ } \ } while (0)