diff --git a/include/libs/transport/trpc.h b/include/libs/transport/trpc.h index 467f0a9ff0..655c903c0b 100644 --- a/include/libs/transport/trpc.h +++ b/include/libs/transport/trpc.h @@ -47,8 +47,6 @@ typedef struct SRpcHandleInfo { int8_t persistHandle; // persist handle or not int8_t hasEpSet; - STraceId traceId; - // app info void *ahandle; // app handle set by client void *wrapper; // wrapper handle @@ -58,7 +56,8 @@ typedef struct SRpcHandleInfo { void *rsp; int32_t rspLen; - // conn info + STraceId traceId; + SRpcConnInfo conn; } SRpcHandleInfo; diff --git a/include/util/tdef.h b/include/util/tdef.h index f4e3385651..6fbec4bc67 100644 --- a/include/util/tdef.h +++ b/include/util/tdef.h @@ -396,7 +396,7 @@ typedef enum ELogicConditionType { #ifdef WINDOWS #define TSDB_MAX_RPC_THREADS 4 // windows pipe only support 4 connections. #else -#define TSDB_MAX_RPC_THREADS 5 +#define TSDB_MAX_RPC_THREADS 10 #endif #define TSDB_QUERY_TYPE_NON_TYPE 0x00u // none type diff --git a/source/libs/transport/inc/transComm.h b/source/libs/transport/inc/transComm.h index a9326ca429..04b58da570 100644 --- a/source/libs/transport/inc/transComm.h +++ b/source/libs/transport/inc/transComm.h @@ -128,7 +128,7 @@ typedef struct { int8_t retryCnt; int8_t retryLimit; - // bool setMaxRetry; + STransCtx appCtx; // STransMsg* pRsp; // for synchronous API tsem_t* pSem; // for synchronous API @@ -195,17 +195,7 @@ typedef enum { ConnNormal, ConnAcquire, ConnRelease, ConnBroken, ConnInPool } Co #define transLabel(trans) ((STrans*)trans)->label -// int rpcAuthenticateMsg(void* pMsg, int msgLen, void* pAuth, void* pKey); -// void rpcBuildAuthHead(void* pMsg, int msgLen, void* pAuth, void* pKey); -//// int32_t rpcCompressRpcMsg(char* pCont, int32_t contLen); -// -// int transAuthenticateMsg(void* pMsg, int msgLen, void* pAuth, void* pKey); -// void transBuildAuthHead(void* pMsg, int msgLen, void* pAuth, void* pKey); -// bool transCompressMsg(char* msg, int32_t len, int32_t* flen); -// bool transDecompressMsg(char* msg, int32_t len, int32_t* flen); - void transFreeMsg(void* msg); - // typedef struct SConnBuffer { char* buf; @@ -322,8 +312,8 @@ void* transCtxDumpBrokenlinkVal(STransCtx* ctx, int32_t* msgType); // request list typedef struct STransReq { - queue q; - void* data; + queue q; + uv_write_t wreq; } STransReq; void transReqQueueInit(queue* q); diff --git a/source/libs/transport/src/transCli.c b/source/libs/transport/src/transCli.c index 31d567a2ed..47d47d3dbe 100644 --- a/source/libs/transport/src/transCli.c +++ b/source/libs/transport/src/transCli.c @@ -16,7 +16,8 @@ #include "transComm.h" typedef struct SConnList { - queue conn; + queue conn; + int32_t size; } SConnList; typedef struct SCliConn { @@ -339,8 +340,8 @@ void cliHandleResp(SCliConn* conn) { tDebug("%s conn %p stop timer", CONN_GET_INST_LABEL(conn), conn); uv_timer_stop(conn->timer); } - conn->timer->data = NULL; taosArrayPush(pThrd->timerList, &conn->timer); + conn->timer->data = NULL; conn->timer = NULL; } @@ -510,7 +511,7 @@ static SCliConn* getConnFromPool(void* pool, char* ip, uint32_t port) { SHashObj* pPool = pool; SConnList* plist = taosHashGet(pPool, key, strlen(key)); if (plist == NULL) { - SConnList list; + SConnList list = {0}; taosHashPut(pPool, key, strlen(key), (void*)&list, sizeof(list)); plist = taosHashGet(pPool, key, strlen(key)); QUEUE_INIT(&plist->conn); @@ -519,15 +520,18 @@ static SCliConn* getConnFromPool(void* pool, char* ip, uint32_t port) { if (QUEUE_IS_EMPTY(&plist->conn)) { return NULL; } + + plist->size -= 1; queue* h = QUEUE_HEAD(&plist->conn); SCliConn* conn = QUEUE_DATA(h, SCliConn, q); conn->status = ConnNormal; QUEUE_REMOVE(&conn->q); QUEUE_INIT(&conn->q); - transDQCancel(((SCliThrd*)conn->hostThrd)->timeoutQueue, conn->task); - conn->task = NULL; - + if (conn->task != NULL) { + transDQCancel(((SCliThrd*)conn->hostThrd)->timeoutQueue, conn->task); + conn->task = NULL; + } return conn; } static void addConnToPool(void* pool, SCliConn* conn) { @@ -539,6 +543,13 @@ static void addConnToPool(void* pool, SCliConn* conn) { allocConnRef(conn, true); + if (conn->timer != NULL) { + uv_timer_stop(conn->timer); + taosArrayPush(thrd->timerList, &conn->timer); + conn->timer->data = NULL; + conn->timer = NULL; + } + STrans* pTransInst = thrd->pTransInst; cliReleaseUnfinishedMsg(conn); transQueueClear(&conn->cliMsgs); @@ -556,13 +567,17 @@ static void addConnToPool(void* pool, SCliConn* conn) { assert(conn->list != NULL); QUEUE_INIT(&conn->q); QUEUE_PUSH(&conn->list->conn, &conn->q); + conn->list->size += 1; + conn->task = NULL; assert(!QUEUE_IS_EMPTY(&conn->list->conn)); - STaskArg* arg = taosMemoryCalloc(1, sizeof(STaskArg)); - arg->param1 = conn; - arg->param2 = thrd; - conn->task = transDQSched(thrd->timeoutQueue, doCloseIdleConn, arg, CONN_PERSIST_TIME(pTransInst->idleTime)); + if (conn->list->size >= 50) { + STaskArg* arg = taosMemoryCalloc(1, sizeof(STaskArg)); + arg->param1 = conn; + arg->param2 = thrd; + conn->task = transDQSched(thrd->timeoutQueue, doCloseIdleConn, arg, CONN_PERSIST_TIME(pTransInst->idleTime)); + } } static int32_t allocConnRef(SCliConn* conn, bool update) { if (update) { @@ -1374,7 +1389,7 @@ int transReleaseCliHandle(void* handle) { } STransMsg tmsg = {.info.handle = handle}; - TRACE_SET_MSGID(&tmsg.info.traceId, tGenIdPI64()); + // TRACE_SET_MSGID(&tmsg.info.traceId, tGenIdPI64()); SCliMsg* cmsg = taosMemoryCalloc(1, sizeof(SCliMsg)); cmsg->msg = tmsg; @@ -1415,7 +1430,6 @@ int transSendRequest(void* shandle, const SEpSet* pEpSet, STransMsg* pReq, STran if (ctx != NULL) { pCtx->appCtx = *ctx; } - assert(pTransInst->connType == TAOS_CONN_CLIENT); SCliMsg* cliMsg = taosMemoryCalloc(1, sizeof(SCliMsg)); cliMsg->ctx = pCtx; diff --git a/source/libs/transport/src/transComm.c b/source/libs/transport/src/transComm.c index ed694e120a..b568163e23 100644 --- a/source/libs/transport/src/transComm.c +++ b/source/libs/transport/src/transComm.c @@ -23,33 +23,6 @@ static TdThreadOnce transModuleInit = PTHREAD_ONCE_INIT; static int32_t refMgt; static int32_t instMgt; -int transAuthenticateMsg(void* pMsg, int msgLen, void* pAuth, void* pKey) { - T_MD5_CTX context; - int ret = -1; - - tMD5Init(&context); - tMD5Update(&context, (uint8_t*)pKey, TSDB_PASSWORD_LEN); - tMD5Update(&context, (uint8_t*)pMsg, msgLen); - tMD5Update(&context, (uint8_t*)pKey, TSDB_PASSWORD_LEN); - tMD5Final(&context); - - if (memcmp(context.digest, pAuth, sizeof(context.digest)) == 0) ret = 0; - - return ret; -} - -void transBuildAuthHead(void* pMsg, int msgLen, void* pAuth, void* pKey) { - T_MD5_CTX context; - - tMD5Init(&context); - tMD5Update(&context, (uint8_t*)pKey, TSDB_PASSWORD_LEN); - tMD5Update(&context, (uint8_t*)pMsg, msgLen); - tMD5Update(&context, (uint8_t*)pKey, TSDB_PASSWORD_LEN); - tMD5Final(&context); - - memcpy(pAuth, context.digest, sizeof(context.digest)); -} - bool transCompressMsg(char* msg, int32_t len, int32_t* flen) { return false; // SRpcHead* pHead = rpcHeadFromCont(pCont); @@ -176,7 +149,6 @@ int transAllocBuffer(SConnBuffer* connBuf, uv_buf_t* uvBuf) { * info--->| */ SConnBuffer* p = connBuf; - uvBuf->base = p->buf + p->len; if (p->left == -1) { uvBuf->len = p->cap - p->len; @@ -184,7 +156,8 @@ int transAllocBuffer(SConnBuffer* connBuf, uv_buf_t* uvBuf) { if (p->left < p->cap - p->len) { uvBuf->len = p->left; } else { - p->buf = taosMemoryRealloc(p->buf, p->left + p->len); + p->cap = p->left + p->len; + p->buf = taosMemoryRealloc(p->buf, p->cap); uvBuf->base = p->buf + p->len; uvBuf->len = p->left; } @@ -266,14 +239,9 @@ int transAsyncSend(SAsyncPool* pool, queue* q) { uv_async_t* async = &(pool->asyncs[idx]); SAsyncItem* item = async->data; - int64_t st = taosGetTimestampUs(); taosThreadMutexLock(&item->mtx); QUEUE_PUSH(&item->qmsg, q); taosThreadMutexUnlock(&item->mtx); - int64_t el = taosGetTimestampUs() - st; - if (el > 50) { - // tInfo("lock and unlock cost:%d", (int)el); - } return uv_async_send(async); } @@ -349,30 +317,21 @@ void transReqQueueInit(queue* q) { QUEUE_INIT(q); } void* transReqQueuePush(queue* q) { - uv_write_t* req = taosMemoryCalloc(1, sizeof(uv_write_t)); - STransReq* wreq = taosMemoryCalloc(1, sizeof(STransReq)); - wreq->data = req; - req->data = wreq; - QUEUE_PUSH(q, &wreq->q); - return req; + STransReq* req = taosMemoryCalloc(1, sizeof(STransReq)); + req->wreq.data = req; + QUEUE_PUSH(q, &req->q); + return &req->wreq; } void* transReqQueueRemove(void* arg) { void* ret = NULL; - uv_write_t* req = arg; - STransReq* wreq = req && req->data ? req->data : NULL; + uv_write_t* wreq = arg; - assert(wreq->data == req); - if (wreq == NULL || wreq->data == NULL) { - taosMemoryFree(wreq->data); - taosMemoryFree(wreq); - return req; - } + STransReq* req = wreq ? wreq->data : NULL; + if (req == NULL) return NULL; + QUEUE_REMOVE(&req->q); - QUEUE_REMOVE(&wreq->q); - - ret = req && req->handle ? req->handle->data : NULL; - taosMemoryFree(wreq->data); - taosMemoryFree(wreq); + ret = wreq && wreq->handle ? wreq->handle->data : NULL; + taosMemoryFree(req); return ret; } @@ -381,7 +340,6 @@ void transReqQueueClear(queue* q) { queue* h = QUEUE_HEAD(q); QUEUE_REMOVE(h); STransReq* req = QUEUE_DATA(h, STransReq, q); - taosMemoryFree(req->data); taosMemoryFree(req); } } diff --git a/source/libs/transport/src/transSvr.c b/source/libs/transport/src/transSvr.c index 8b27d95e52..14b8b35478 100644 --- a/source/libs/transport/src/transSvr.c +++ b/source/libs/transport/src/transSvr.c @@ -75,7 +75,6 @@ typedef struct SWorkThrd { SAsyncPool* asyncPool; uv_prepare_t* prepare; queue msg; - TdThreadMutex msgMtx; queue conn; void* pTransInst; @@ -499,6 +498,7 @@ void uvWorkerAsyncCb(uv_async_t* handle) { tError("unexcept occurred, continue"); continue; } + // release handle to rpc init if (msg->type == Quit) { (*transAsyncHandle[msg->type])(msg, pThrd); @@ -743,7 +743,6 @@ static bool addHandleToWorkloop(SWorkThrd* pThrd, char* pipeName) { pThrd->pipe->data = pThrd; QUEUE_INIT(&pThrd->msg); - taosThreadMutexInit(&pThrd->msgMtx, NULL); pThrd->prepare = taosMemoryCalloc(1, sizeof(uv_prepare_t)); uv_prepare_init(pThrd->loop, pThrd->prepare); diff --git a/source/libs/transport/test/svrBench.c b/source/libs/transport/test/svrBench.c index 224f527385..6eb80c8504 100644 --- a/source/libs/transport/test/svrBench.c +++ b/source/libs/transport/test/svrBench.c @@ -75,15 +75,14 @@ void processShellMsg() { void *handle = pRpcMsg->info.handle; taosFreeQitem(pRpcMsg); - - { - SRpcMsg nRpcMsg = {0}; - nRpcMsg.pCont = rpcMallocCont(msgSize); - nRpcMsg.contLen = msgSize; - nRpcMsg.info.handle = handle; - nRpcMsg.code = TSDB_CODE_CTG_NOT_READY; - rpcSendResponse(&nRpcMsg); - } + //{ + // SRpcMsg nRpcMsg = {0}; + // nRpcMsg.pCont = rpcMallocCont(msgSize); + // nRpcMsg.contLen = msgSize; + // nRpcMsg.info.handle = handle; + // nRpcMsg.code = TSDB_CODE_CTG_NOT_READY; + // rpcSendResponse(&nRpcMsg); + //} } taosUpdateItemSize(qinfo.queue, numOfMsgs); diff --git a/source/util/src/tref.c b/source/util/src/tref.c index 9cd849b9be..c984ef3f34 100644 --- a/source/util/src/tref.c +++ b/source/util/src/tref.c @@ -44,11 +44,11 @@ typedef struct { void (*fp)(void *); } SRefSet; -static SRefSet tsRefSetList[TSDB_REF_OBJECTS]; +static SRefSet tsRefSetList[TSDB_REF_OBJECTS]; static TdThreadOnce tsRefModuleInit = PTHREAD_ONCE_INIT; static TdThreadMutex tsRefMutex; -static int32_t tsRefSetNum = 0; -static int32_t tsNextId = 0; +static int32_t tsRefSetNum = 0; +static int32_t tsNextId = 0; static void taosInitRefModule(void); static void taosLockList(int64_t *lockedBy);