diff --git a/source/libs/transport/src/transCli.c b/source/libs/transport/src/transCli.c index 9f35965431..2bf5fa11ac 100644 --- a/source/libs/transport/src/transCli.c +++ b/source/libs/transport/src/transCli.c @@ -90,6 +90,8 @@ typedef struct SCliConn { int64_t refId; int32_t seq; int32_t shareCnt; + + int8_t registered; } SCliConn; typedef struct SCliReq { @@ -196,11 +198,11 @@ static int32_t allocConnRef(SCliConn* conn, bool update); static int cliNotifyCb(SCliConn* pConn, SCliReq* pReq, STransMsg* pResp); void cliResetConnTimer(SCliConn* conn); -static void cliDestroyConn(SCliConn* pConn, bool clear /*clear tcp handle or not*/); -static void cliDestroy(uv_handle_t* handle); -static void cliSend(SCliConn* pConn); -static void cliSendBatch(SCliConn* pConn); -static void cliDestroyConnMsgs(SCliConn* conn, bool destroy); +static void cliDestroyConn(SCliConn* pConn, bool clear /*clear tcp handle or not*/); +static void cliDestroy(uv_handle_t* handle); +static int32_t cliSend(SCliConn* pConn); +static void cliSendBatch(SCliConn* pConn); +static void cliDestroyConnMsgs(SCliConn* conn, bool destroy); static void doFreeTimeoutMsg(void* param); static int32_t cliPreCheckSessionLimitForMsg(SCliThrd* pThrd, char* addr, SCliReq** pReq); @@ -324,6 +326,17 @@ int32_t transHeapDelete(SHeap* heap, SCliConn* p); } \ } while (0) +static int32_t cliConnFindToSendMsg(SCliConn* pConn, SCliReq** pReq) { + int32_t code = 0; + for (int32_t i = 0; i < transQueueSize(&pConn->reqs); i++) { + SCliReq* p = transQueueGet(&pConn->reqs, i); + if (p->sent == 0) { + *pReq = p; + return 0; + } + } + return TSDB_CODE_OUT_OF_RANGE; +} #define CONN_SET_PERSIST_BY_APP(conn) \ do { \ if (conn->status == ConnNormal) { \ @@ -377,7 +390,7 @@ bool cliMaySendCachedMsg(SCliConn* conn) { if (!transQueueEmpty(&conn->reqs)) { SCliReq* pCliMsg = NULL; CONN_GET_NEXT_SENDMSG(conn); - cliSend(conn); + (void)cliSend(conn); return true; } return false; @@ -404,7 +417,7 @@ bool cliConnSendSeqMsg(int64_t refId, SCliConn* conn) { (void)transQueuePush(&conn->reqs, t); tDebug("pop from conn %p, refId: %" PRId64 "", conn, refId); (void)transReleaseExHandle(transGetRefMgt(), refId); - cliSend(conn); + (void)cliSend(conn); return true; } taosWUnLockLatch(&exh->latch); @@ -1037,7 +1050,7 @@ static void addConnToPool(void* pool, SCliConn* conn) { (void)transQueuePush(&conn->reqs, pReq); conn->status = ConnNormal; - cliSend(conn); + (void)cliSend(conn); return; } @@ -1195,6 +1208,7 @@ static int32_t cliCreateConn2(SCliThrd* pThrd, SCliReq* pReq, SCliConn** ppConn) if (code != TSDB_CODE_RPC_ASYNC_IN_PROCESS) { cliRmReqFromConn(pConn, NULL); cliDestroyConn2(pConn); + delConnFromHeapCache(pThrd->connHeapCache, pConn); return code; } else { } @@ -1202,7 +1216,15 @@ static int32_t cliCreateConn2(SCliThrd* pThrd, SCliReq* pReq, SCliConn** ppConn) } // not any ref, -static int32_t cliDestroyConn2(SCliConn* conn) { return 0; } +static int32_t cliDestroyConn2(SCliConn* conn) { + if (conn->registered == 0) { + taosMemoryFree(conn->dstAddr); + taosMemoryFree(conn); + } else { + cliDestroyConn(conn, true); + } + return 0; +} static int32_t cliCreateConn(SCliThrd* pThrd, const SCliReq* pReq, SCliConn** pCliConn) { char addr[TSDB_FQDN_LEN + 64] = {0}; CONN_CONSTRUCT_HASH_KEY(addr, EPSET_GET_INUSE_IP(&pReq->ctx->epSet), EPSET_GET_INUSE_PORT(&pReq->ctx->epSet)); @@ -1559,37 +1581,33 @@ _exception: return; } - -int32_t cliSend2(SCliConn* pConn) { - -} +// int32_t cliSend2(SCliConn* pConn) {} int32_t cliSendReq(SCliConn* pConn, SCliReq* pCliMsg) { int32_t code = 0; transQueuePush(&pConn->reqs, pCliMsg); - cliSend(pConn); - - return 0; + code = cliSend(pConn); + return code; } -void cliSend(SCliConn* pConn) { +int32_t cliSend(SCliConn* pConn) { SCliThrd* pThrd = pConn->hostThrd; STrans* pInst = pThrd->pInst; + SCliReq* pCliReq = NULL; + int32_t code = cliConnFindToSendMsg(pConn, &pCliReq); - if (transQueueEmpty(&pConn->reqs)) { - tError("%s conn %p not msg to send", pInst->label, pConn); - cliHandleExcept(pConn, -1); - return; + if (code != 0) { + return code; } - SCliReq* pCliMsg = NULL; - CONN_GET_NEXT_SENDMSG(pConn); - pCliMsg->sent = 1; + SReqCtx* pCtx = pCliReq->ctx; - SReqCtx* pCtx = pCliMsg->ctx; - - STransMsg* pReq = (STransMsg*)(&pCliMsg->msg); + STransMsg* pReq = (STransMsg*)(&pCliReq->msg); if (pReq->pCont == 0) { pReq->pCont = (void*)rpcMallocCont(0); + if (pReq->pCont == NULL) { + return TSDB_CODE_OUT_OF_MEMORY; + } + tDebug("malloc memory: %p", pReq->pCont); pReq->contLen = 0; } @@ -1603,7 +1621,7 @@ void cliSend(SCliConn* pConn) { pHead->persist = REQUEST_PERSIS_HANDLE(pReq) ? 1 : 0; pHead->msgType = pReq->msgType; pHead->msgLen = (int32_t)htonl((uint32_t)msgLen); - pHead->release = REQUEST_RELEASE_HANDLE(pCliMsg) ? 1 : 0; + pHead->release = REQUEST_RELEASE_HANDLE(pCliReq) ? 1 : 0; memcpy(pHead->user, pInst->user, strlen(pInst->user)); pHead->traceId = pReq->info.traceId; pHead->magicNum = htonl(TRANS_MAGIC_NUM); @@ -1636,18 +1654,19 @@ void cliSend(SCliConn* pConn) { tGError("%s conn %p failed to send msg:%s, errmsg:%s", CONN_GET_INST_LABEL(pConn), pConn, TMSG_INFO(pReq->msgType), tstrerror(TSDB_CODE_OUT_OF_MEMORY)); cliHandleExcept(pConn, -1); - return; + return TSDB_CODE_OUT_OF_MEMORY; } + pCliReq->sent = 1; int status = uv_write(req, (uv_stream_t*)pConn->stream, &wb, 1, cliSendCb); if (status != 0) { tGError("%s conn %p failed to send msg:%s, errmsg:%s", CONN_GET_INST_LABEL(pConn), pConn, TMSG_INFO(pReq->msgType), uv_err_name(status)); cliHandleExcept(pConn, -1); + return TSDB_CODE_THIRDPARTY_ERROR; } - return; -_RETURN: - return; + + return TSDB_CODE_RPC_ASYNC_IN_PROCESS; } static void cliDestroyBatch(SCliBatch* pBatch) { @@ -1720,7 +1739,7 @@ static int32_t cliDoConn(SCliThrd* pThrd, SCliConn* conn, char* ip, int port) { tError("%s conn %p failed to start timer, reason:%s", transLabel(pInst), conn, uv_err_name(ret)); TAOS_CHECK_GOTO(TSDB_CODE_THIRDPARTY_ERROR, &lino, _exception2); } - + conn->registered = 1; return TSDB_CODE_RPC_ASYNC_IN_PROCESS; _exception1: @@ -1889,7 +1908,7 @@ void cliConnCb(uv_connect_t* req, int status) { return cliSendBatch_shareConn(pConn); } - return cliSend(pConn); + (void)cliSend(pConn); } static void doNotifyCb(SCliReq* pReq, SCliThrd* pThrd, int32_t code) { @@ -1950,7 +1969,7 @@ static void cliHandleRelease(SCliThrd* pThrd, SCliReq* pReq) { if (!transQueuePush(&conn->reqs, pReq)) { return; } - cliSend(conn); + (void)cliSend(conn); } else { tError("%s conn %p already released", CONN_GET_INST_LABEL(conn), conn); destroyReq(pReq); @@ -2211,7 +2230,7 @@ void cliHandleReq__noShareConn(SCliThrd* pThrd, SCliReq* pReq) { if (code == TSDB_CODE_RPC_MAX_SESSIONS) { TAOS_CHECK_GOTO(code, &lino, _exception); } else if (code == TSDB_CODE_RPC_ASYNC_IN_PROCESS) { - // do nothing, notifyCb + // do nothing, notiy return; } else { code = cliSendReq(pConn, pReq); @@ -3526,6 +3545,7 @@ _RETURN1: pReq->pCont = NULL; return code; } + int32_t transCreateSyncMsg(STransMsg* pTransMsg, int64_t* refId) { int32_t code = 0; tsem2_t* sem = taosMemoryCalloc(1, sizeof(tsem2_t)); @@ -3562,6 +3582,7 @@ _EXIT: taosMemoryFree(pSyncMsg); return code; } + int32_t transSendRecvWithTimeout(void* pInstRef, SEpSet* pEpSet, STransMsg* pReq, STransMsg* pRsp, int8_t* epUpdated, int32_t timeoutMs) { int32_t code = 0;