diff --git a/source/libs/transport/src/trans.c b/source/libs/transport/src/trans.c index 5fa57fdbc2a416c1c7cf92a2f7274953605d140e..7633820292ae2442f78c85537fb5c46b4caabea9 100644 --- a/source/libs/transport/src/trans.c +++ b/source/libs/transport/src/trans.c @@ -80,8 +80,7 @@ void* rpcOpen(const SRpcInit* pInit) { int64_t refId = transAddExHandle(transGetInstMgt(), pRpc); transAcquireExHandle(transGetInstMgt(), refId); pRpc->refId = refId; - return pRpc; - // return (void*)refId; + return (void*)refId; } void rpcClose(void* arg) { tInfo("start to close rpc"); diff --git a/source/libs/transport/src/transCli.c b/source/libs/transport/src/transCli.c index 11ba2a50d38a233985a272023ff8952c039cada1..61888db74ebfe7b47185ffa3ae63e1bb8da0b46e 100644 --- a/source/libs/transport/src/transCli.c +++ b/source/libs/transport/src/transCli.c @@ -564,7 +564,7 @@ static void addConnToPool(void* pool, SCliConn* conn) { conn->task = NULL; assert(!QUEUE_IS_EMPTY(&conn->list->conn)); - if (conn->list->size >= 10) { + if (conn->list->size >= 50) { STaskArg* arg = taosMemoryCalloc(1, sizeof(STaskArg)); arg->param1 = conn; arg->param2 = thrd; @@ -1372,7 +1372,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; @@ -1388,12 +1388,11 @@ int transReleaseCliHandle(void* handle) { } int transSendRequest(void* shandle, const SEpSet* pEpSet, STransMsg* pReq, STransCtx* ctx) { - // STrans* pTransInst = (STrans*)transAcquireExHandle(transGetInstMgt(), (int64_t)shandle); - // if (pTransInst == NULL) { - // transFreeMsg(pReq->pCont); - // return -1; - // } - STrans* pTransInst = shandle; + STrans* pTransInst = (STrans*)transAcquireExHandle(transGetInstMgt(), (int64_t)shandle); + if (pTransInst == NULL) { + transFreeMsg(pReq->pCont); + return -1; + } bool valid = false; SCliThrd* pThrd = transGetWorkThrd(pTransInst, (int64_t)pReq->info.handle, &valid); @@ -1429,23 +1428,22 @@ int transSendRequest(void* shandle, const SEpSet* pEpSet, STransMsg* pReq, STran transReleaseExHandle(transGetInstMgt(), (int64_t)shandle); return -1; } - // transReleaseExHandle(transGetInstMgt(), (int64_t)shandle); + transReleaseExHandle(transGetInstMgt(), (int64_t)shandle); return 0; } int transSendRecv(void* shandle, const SEpSet* pEpSet, STransMsg* pReq, STransMsg* pRsp) { - STrans* pTransInst = shandle; - // STrans* pTransInst = (STrans*)transAcquireExHandle(transGetInstMgt(), (int64_t)shandle); - // if (pTransInst == NULL) { - // transFreeMsg(pReq->pCont); - // return -1; - // } + STrans* pTransInst = (STrans*)transAcquireExHandle(transGetInstMgt(), (int64_t)shandle); + if (pTransInst == NULL) { + transFreeMsg(pReq->pCont); + return -1; + } bool valid = false; SCliThrd* pThrd = transGetWorkThrd(pTransInst, (int64_t)pReq->info.handle, &valid); if (pThrd == NULL && valid == false) { transFreeMsg(pReq->pCont); - // transReleaseExHandle(transGetInstMgt(), (int64_t)shandle); + transReleaseExHandle(transGetInstMgt(), (int64_t)shandle); return -1; } @@ -1483,7 +1481,7 @@ int transSendRecv(void* shandle, const SEpSet* pEpSet, STransMsg* pReq, STransMs _RETURN: tsem_destroy(sem); taosMemoryFree(sem); - // transReleaseExHandle(transGetInstMgt(), (int64_t)shandle); + transReleaseExHandle(transGetInstMgt(), (int64_t)shandle); return ret; } /* diff --git a/source/libs/transport/src/transComm.c b/source/libs/transport/src/transComm.c index ca405fa5364b1572de5215905a0a1f53423a7a78..f0b1885d66734899ea2d8d12c27ae0018e64a5c9 100644 --- a/source/libs/transport/src/transComm.c +++ b/source/libs/transport/src/transComm.c @@ -156,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; }