diff --git a/source/libs/transport/src/transCli.c b/source/libs/transport/src/transCli.c index fbf4d6aa9f..37b02615d3 100644 --- a/source/libs/transport/src/transCli.c +++ b/source/libs/transport/src/transCli.c @@ -171,23 +171,23 @@ static void cliReleaseUnfinishedMsg(SCliConn* conn) { #define CONN_PERSIST_TIME(para) (para * 1000 * 10) #define CONN_GET_HOST_THREAD(conn) (conn ? ((SCliConn*)conn)->hostThrd : NULL) #define CONN_GET_INST_LABEL(conn) (((STrans*)(((SCliThrdObj*)(conn)->hostThrd)->pTransInst))->label) -#define CONN_SHOULD_RELEASE(conn, head) \ - do { \ - if ((head)->release == 1 && (head->msgLen) == sizeof(*head)) { \ - uint64_t ahandle = head->ahandle; \ - CONN_GET_MSGCTX_BY_AHANDLE(conn, ahandle); \ - conn->status = ConnRelease; \ - transClearBuffer(&conn->readBuf); \ - transFreeMsg(transContFromHead((char*)head)); \ - tDebug("cli conn %p receive release request, ref: %d", conn, T_REF_VAL_GET(conn)); \ - if (T_REF_VAL_GET(conn) > 1) { \ - transUnrefCliHandle(conn); \ - } \ - destroyCmsg(pMsg); \ - cliReleaseUnfinishedMsg(conn); \ - addConnToPool(((SCliThrdObj*)conn->hostThrd)->pool, conn); \ - return; \ - } \ +#define CONN_SHOULD_RELEASE(conn, head) \ + do { \ + if ((head)->release == 1 && (head->msgLen) == sizeof(*head)) { \ + uint64_t ahandle = head->ahandle; \ + CONN_GET_MSGCTX_BY_AHANDLE(conn, ahandle); \ + conn->status = ConnRelease; \ + transClearBuffer(&conn->readBuf); \ + transFreeMsg(transContFromHead((char*)head)); \ + tDebug("%s conn %p receive release request, ref: %d", CONN_GET_INST_LABEL(conn), conn, T_REF_VAL_GET(conn)); \ + if (T_REF_VAL_GET(conn) > 1) { \ + transUnrefCliHandle(conn); \ + } \ + destroyCmsg(pMsg); \ + cliReleaseUnfinishedMsg(conn); \ + addConnToPool(((SCliThrdObj*)conn->hostThrd)->pool, conn); \ + return; \ + } \ } while (0) #define CONN_GET_MSGCTX_BY_AHANDLE(conn, ahandle) \ @@ -293,27 +293,28 @@ void cliHandleResp(SCliConn* conn) { if (transMsg.info.ahandle == NULL) { transMsg.info.ahandle = transCtxDumpBrokenlinkVal(&conn->ctx, (int32_t*)&(transMsg.msgType)); } - tDebug("cli conn %p construct ahandle %p, persist: 0", conn, transMsg.info.ahandle); + tDebug("%s conn %p construct ahandle %p, persist: 0", CONN_GET_INST_LABEL(conn), conn, transMsg.info.ahandle); } else { transMsg.info.ahandle = pCtx ? pCtx->ahandle : NULL; - tDebug("cli conn %p get ahandle %p, persist: 0", conn, transMsg.info.ahandle); + tDebug("%s conn %p get ahandle %p, persist: 0", CONN_GET_INST_LABEL(conn), conn, transMsg.info.ahandle); } } else { uint64_t ahandle = (uint64_t)pHead->ahandle; CONN_GET_MSGCTX_BY_AHANDLE(conn, ahandle); if (pMsg == NULL) { transMsg.info.ahandle = transCtxDumpVal(&conn->ctx, transMsg.msgType); - tDebug("cli conn %p construct ahandle %p by %s, persist: 1", conn, transMsg.info.ahandle, - TMSG_INFO(transMsg.msgType)); + tDebug("%s conn %p construct ahandle %p by %s, persist: 1", CONN_GET_INST_LABEL(conn), conn, + transMsg.info.ahandle, TMSG_INFO(transMsg.msgType)); if (!CONN_RELEASE_BY_SERVER(conn) && transMsg.info.ahandle == NULL) { transMsg.code = TSDB_CODE_RPC_NETWORK_UNAVAIL; transMsg.info.ahandle = transCtxDumpBrokenlinkVal(&conn->ctx, (int32_t*)&(transMsg.msgType)); - tDebug("cli conn %p construct ahandle %p due brokenlink, persist: 1", conn, transMsg.info.ahandle); + tDebug("%s conn %p construct ahandle %p due brokenlink, persist: 1", CONN_GET_INST_LABEL(conn), conn, + transMsg.info.ahandle); } } else { pCtx = pMsg ? pMsg->ctx : NULL; transMsg.info.ahandle = pCtx ? pCtx->ahandle : NULL; - tDebug("cli conn %p get ahandle %p, persist: 1", conn, transMsg.info.ahandle); + tDebug("%s conn %p get ahandle %p, persist: 1", CONN_GET_INST_LABEL(conn), conn, transMsg.info.ahandle); } } // buf's mem alread translated to transMsg.pCont @@ -321,26 +322,25 @@ void cliHandleResp(SCliConn* conn) { if (!CONN_NO_PERSIST_BY_APP(conn)) { transMsg.info.handle = conn; - tDebug("%s cli conn %p ref by app", CONN_GET_INST_LABEL(conn), conn); + tDebug("%s conn %p ref by app", CONN_GET_INST_LABEL(conn), conn); } - tDebug("%s cli conn %p %s received from %s:%d, local info: %s:%d, msg size: %d", pTransInst->label, conn, + tDebug("%s conn %p %s received from %s:%d, local info: %s:%d, msg size: %d", pTransInst->label, conn, TMSG_INFO(pHead->msgType), taosInetNtoa(conn->addr.sin_addr), ntohs(conn->addr.sin_port), taosInetNtoa(conn->localAddr.sin_addr), ntohs(conn->localAddr.sin_port), transMsg.contLen); if (pCtx == NULL && CONN_NO_PERSIST_BY_APP(conn)) { - tTrace("except, server continue send while cli ignore it"); + tTrace("%s except, server continue send while cli ignore it", CONN_GET_INST_LABEL(conn)); // transUnrefCliHandle(conn); return; } if (CONN_RELEASE_BY_SERVER(conn) && transMsg.info.ahandle == NULL) { - tTrace("except, server continue send while cli ignore it"); + tTrace("%s except, server continue send while cli ignore it", CONN_GET_INST_LABEL(conn)); // transUnrefCliHandle(conn); return; } if (cliAppCb(conn, &transMsg, pMsg) != 0) { - tTrace("try to send req to next node"); return; } destroyCmsg(pMsg); @@ -363,7 +363,7 @@ void cliHandleResp(SCliConn* conn) { void cliHandleExcept(SCliConn* pConn) { if (transQueueEmpty(&pConn->cliMsgs)) { if (pConn->broken == true && CONN_NO_PERSIST_BY_APP(pConn)) { - tTrace("%s cli conn %p handle except, persist:0", CONN_GET_INST_LABEL(pConn), pConn); + tTrace("%s conn %p handle except, persist:0", CONN_GET_INST_LABEL(pConn), pConn); transUnrefCliHandle(pConn); return; } @@ -386,11 +386,11 @@ void cliHandleExcept(SCliConn* pConn) { if (pMsg == NULL && !CONN_NO_PERSIST_BY_APP(pConn)) { transMsg.info.ahandle = transCtxDumpVal(&pConn->ctx, transMsg.msgType); - tDebug("%s cli conn %p construct ahandle %p by %s", CONN_GET_INST_LABEL(pConn), pConn, transMsg.info.ahandle, + tDebug("%s conn %p construct ahandle %p by %s", CONN_GET_INST_LABEL(pConn), pConn, transMsg.info.ahandle, TMSG_INFO(transMsg.msgType)); if (transMsg.info.ahandle == NULL) { transMsg.info.ahandle = transCtxDumpBrokenlinkVal(&pConn->ctx, (int32_t*)&(transMsg.msgType)); - tDebug("%s cli conn %p construct ahandle %p due to brokenlink", CONN_GET_INST_LABEL(pConn), pConn, + tDebug("%s conn %p construct ahandle %p due to brokenlink", CONN_GET_INST_LABEL(pConn), pConn, transMsg.info.ahandle); } } else { @@ -404,11 +404,10 @@ void cliHandleExcept(SCliConn* pConn) { } } if (cliAppCb(pConn, &transMsg, pMsg) != 0) { - tTrace("try to send req to next node"); return; } destroyCmsg(pMsg); - tTrace("%s cli conn %p start to destroy", CONN_GET_INST_LABEL(pConn), pConn); + tTrace("%s conn %p start to destroy", CONN_GET_INST_LABEL(pConn), pConn); } while (!transQueueEmpty(&pConn->cliMsgs)); transUnrefCliHandle(pConn); } @@ -417,7 +416,7 @@ void cliTimeoutCb(uv_timer_t* handle) { SCliThrdObj* pThrd = handle->data; STrans* pTransInst = pThrd->pTransInst; int64_t currentTime = pThrd->nextTimeout; - tTrace("%s, cli conn timeout, try to remove expire conn from conn pool", pTransInst->label); + tTrace("%s conn timeout, try to remove expire conn from conn pool", pTransInst->label); SConnList* p = taosHashIterate((SHashObj*)pThrd->pool, NULL); while (p != NULL) { @@ -492,7 +491,7 @@ static void addConnToPool(void* pool, SCliConn* conn) { char key[128] = {0}; CONN_CONSTRUCT_HASH_KEY(key, conn->ip, conn->port); - tTrace("cli conn %p added to conn pool, read buf cap: %d", conn, conn->readBuf.cap); + tTrace("%s conn %p added to conn pool, read buf cap: %d", CONN_GET_INST_LABEL(conn), conn, conn->readBuf.cap); SConnList* plist = taosHashGet((SHashObj*)pool, key, strlen(key)); // list already create before @@ -515,10 +514,10 @@ static void cliRecvCb(uv_stream_t* handle, ssize_t nread, const uv_buf_t* buf) { if (nread > 0) { pBuf->len += nread; if (transReadComplete(pBuf)) { - tTrace("%s cli conn %p read complete", CONN_GET_INST_LABEL(conn), conn); + tTrace("%s conn %p read complete", CONN_GET_INST_LABEL(conn), conn); cliHandleResp(conn); } else { - tTrace("%s cli conn %p read partial packet, continue to read", CONN_GET_INST_LABEL(conn), conn); + tTrace("%s conn %p read partial packet, continue to read", CONN_GET_INST_LABEL(conn), conn); } return; } @@ -528,11 +527,11 @@ static void cliRecvCb(uv_stream_t* handle, ssize_t nread, const uv_buf_t* buf) { // ref http://docs.libuv.org/en/v1.x/stream.html?highlight=uv_read_start#c.uv_read_cb // nread might be 0, which does not indicate an error or EOF. This is equivalent to EAGAIN or EWOULDBLOCK under // read(2). - tTrace("%s cli conn %p read empty", CONN_GET_INST_LABEL(conn), conn); + tTrace("%s conn %p read empty", CONN_GET_INST_LABEL(conn), conn); return; } if (nread < 0) { - tError("%s cli conn %p read error: %s", CONN_GET_INST_LABEL(conn), conn, uv_err_name(nread)); + tError("%s conn %p read error: %s", CONN_GET_INST_LABEL(conn), conn, uv_err_name(nread)); conn->broken = true; cliHandleExcept(conn); } @@ -557,7 +556,7 @@ static SCliConn* cliCreateConn(SCliThrdObj* pThrd) { return conn; } static void cliDestroyConn(SCliConn* conn, bool clear) { - tTrace("%s cli conn %p remove from conn pool", CONN_GET_INST_LABEL(conn), conn); + tTrace("%s conn %p remove from conn pool", CONN_GET_INST_LABEL(conn), conn); QUEUE_REMOVE(&conn->conn); if (clear) { @@ -570,7 +569,7 @@ static void cliDestroy(uv_handle_t* handle) { taosMemoryFree(conn->stream); transCtxCleanup(&conn->ctx); transQueueDestroy(&conn->cliMsgs); - tTrace("%s cli conn %p destroy successfully", CONN_GET_INST_LABEL(conn), conn); + tTrace("%s conn %p destroy successfully", CONN_GET_INST_LABEL(conn), conn); transDestroyBuffer(&conn->readBuf); taosMemoryFree(conn); } @@ -597,14 +596,14 @@ static void cliSendCb(uv_write_t* req, int status) { SCliConn* pConn = req->data; if (status == 0) { - tTrace("%s cli conn %p data already was written out", CONN_GET_INST_LABEL(pConn), pConn); + tTrace("%s conn %p data already was written out", CONN_GET_INST_LABEL(pConn), pConn); } else { - tError("%s cli conn %p failed to write: %s", CONN_GET_INST_LABEL(pConn), pConn, uv_err_name(status)); + tError("%s conn %p failed to write: %s", CONN_GET_INST_LABEL(pConn), pConn, uv_err_name(status)); cliHandleExcept(pConn); return; } if (cliHandleNoResp(pConn) == true) { - tTrace("%s cli conn %p no resp required", CONN_GET_INST_LABEL(pConn), pConn); + tTrace("%s conn %p no resp required", CONN_GET_INST_LABEL(pConn), pConn); return; } uv_read_start((uv_stream_t*)pConn->stream, cliAllocRecvBufferCb, cliRecvCb); @@ -642,7 +641,7 @@ void cliSend(SCliConn* pConn) { memcpy(pHead->user, pTransInst->user, strlen(pTransInst->user)); uv_buf_t wb = uv_buf_init((char*)pHead, msgLen); - tDebug("%s cli conn %p %s is send to %s:%d, local info %s:%d", CONN_GET_INST_LABEL(pConn), pConn, + tDebug("%s conn %p %s is send to %s:%d, local info %s:%d", CONN_GET_INST_LABEL(pConn), pConn, TMSG_INFO(pHead->msgType), taosInetNtoa(pConn->addr.sin_addr), ntohs(pConn->addr.sin_port), taosInetNtoa(pConn->localAddr.sin_addr), ntohs(pConn->localAddr.sin_port)); @@ -662,7 +661,7 @@ void cliConnCb(uv_connect_t* req, int status) { // impl later SCliConn* pConn = req->data; if (status != 0) { - tError("%s cli conn %p failed to connect server: %s", CONN_GET_INST_LABEL(pConn), pConn, uv_strerror(status)); + tError("%s conn %p failed to connect server: %s", CONN_GET_INST_LABEL(pConn), pConn, uv_strerror(status)); cliHandleExcept(pConn); return; } @@ -672,7 +671,7 @@ void cliConnCb(uv_connect_t* req, int status) { addrlen = sizeof(pConn->localAddr); uv_tcp_getsockname((uv_tcp_t*)pConn->stream, (struct sockaddr*)&pConn->localAddr, &addrlen); - tTrace("%s cli conn %p connect to server successfully", CONN_GET_INST_LABEL(pConn), pConn); + tTrace("%s conn %p connect to server successfully", CONN_GET_INST_LABEL(pConn), pConn); assert(pConn->stream == req->handle); cliSend(pConn); @@ -691,7 +690,7 @@ static void cliHandleQuit(SCliMsg* pMsg, SCliThrdObj* pThrd) { } static void cliHandleRelease(SCliMsg* pMsg, SCliThrdObj* pThrd) { SCliConn* conn = pMsg->msg.info.handle; - tDebug("%s cli conn %p start to release to inst", CONN_GET_INST_LABEL(conn), conn); + tDebug("%s conn %p start to release to inst", CONN_GET_INST_LABEL(conn), conn); if (T_REF_VAL_GET(conn) == 2) { transUnrefCliHandle(conn); @@ -716,13 +715,13 @@ SCliConn* cliGetConn(SCliMsg* pMsg, SCliThrdObj* pThrd) { if (pMsg->msg.info.handle != NULL) { conn = (SCliConn*)(pMsg->msg.info.handle); if (conn != NULL) { - tTrace("%s cli conn %p reused", CONN_GET_INST_LABEL(conn), conn); + tTrace("%s conn %p reused", CONN_GET_INST_LABEL(conn), conn); } } else { STransConnCtx* pCtx = pMsg->ctx; conn = getConnFromPool(pThrd->pool, EPSET_GET_INUSE_IP(&pCtx->epSet), EPSET_GET_INUSE_PORT(&pCtx->epSet)); if (conn != NULL) { - tTrace("%s cli conn %p get from conn pool", CONN_GET_INST_LABEL(conn), conn); + tTrace("%s conn %p get from conn pool", CONN_GET_INST_LABEL(conn), conn); } else { tTrace("not found conn in conn pool %p", pThrd->pool); } @@ -750,6 +749,8 @@ void cliHandleReq(SCliMsg* pMsg, SCliThrdObj* pThrd) { cliMayCvtFqdnToIp(&pCtx->epSet, &pThrd->cvtAddr); + transPrintEpSet(&pCtx->epSet); + SCliConn* conn = cliGetConn(pMsg, pThrd); if (conn != NULL) { conn->hThrdIdx = pCtx->hThrdIdx; @@ -768,11 +769,11 @@ void cliHandleReq(SCliMsg* pMsg, SCliThrdObj* pThrd) { int ret = transSetConnOption((uv_tcp_t*)conn->stream); if (ret) { - tError("%s cli conn %p failed to set conn option, errmsg %s", pTransInst->label, conn, uv_err_name(ret)); + tError("%s conn %p failed to set conn option, errmsg %s", pTransInst->label, conn, uv_err_name(ret)); } int fd = taosCreateSocketWithTimeOutOpt(TRANS_CONN_TIMEOUT); if (fd == -1) { - tTrace("%s cli conn %p failed to create socket", pTransInst->label, conn); + tTrace("%s conn %p failed to create socket", pTransInst->label, conn); cliHandleExcept(conn); return; } @@ -782,10 +783,10 @@ void cliHandleReq(SCliMsg* pMsg, SCliThrdObj* pThrd) { addr.sin_family = AF_INET; addr.sin_addr.s_addr = taosGetIpv4FromFqdn(conn->ip); addr.sin_port = (uint16_t)htons((uint16_t)conn->port); - tTrace("%s cli conn %p try to connect to %s:%d", pTransInst->label, conn, conn->ip, conn->port); + tTrace("%s conn %p try to connect to %s:%d", pTransInst->label, conn, conn->ip, conn->port); ret = uv_tcp_connect(&conn->connReq, (uv_tcp_t*)(conn->stream), (const struct sockaddr*)&addr, cliConnCb); if (ret != 0) { - tTrace("%s cli conn %p failed to connect to %s:%d, reason: %s", pTransInst->label, conn, conn->ip, conn->port, + tTrace("%s conn %p failed to connect to %s:%d, reason: %s", pTransInst->label, conn, conn->ip, conn->port, uv_err_name(ret)); cliHandleExcept(conn); return; @@ -944,14 +945,13 @@ int cliAppCb(SCliConn* pConn, STransMsg* pResp, SCliMsg* pMsg) { STrans* pTransInst = pThrd->pTransInst; if (pMsg == NULL || pMsg->ctx == NULL) { - tTrace("%s cli conn %p handle resp", pTransInst->label, pConn); + tTrace("%s conn %p handle resp", pTransInst->label, pConn); pTransInst->cfp(pTransInst->parent, pResp, NULL); return 0; } STransConnCtx* pCtx = pMsg->ctx; SEpSet* pEpSet = &pCtx->epSet; - transPrintEpSet(pEpSet); if (pCtx->retryCount == 0) { pCtx->origEpSet = pCtx->epSet; @@ -964,6 +964,7 @@ int cliAppCb(SCliConn* pConn, STransMsg* pResp, SCliMsg* pMsg) { (pResp->code == TSDB_CODE_RPC_NETWORK_UNAVAIL || pResp->code == TSDB_CODE_APP_NOT_READY || pResp->code == TSDB_CODE_NODE_NOT_DEPLOYED || pResp->code == TSDB_CODE_SYN_NOT_LEADER)) { pMsg->sent = 0; + tTrace("try to send req to next node"); pMsg->st = taosGetTimestampUs(); pCtx->retryCount += 1; if (pResp->code == TSDB_CODE_RPC_NETWORK_UNAVAIL) { @@ -974,22 +975,27 @@ int cliAppCb(SCliConn* pConn, STransMsg* pResp, SCliMsg* pMsg) { arg->param1 = pMsg; arg->param2 = pThrd; transDQSched(pThrd->delayQueue, doDelayTask, arg, TRANS_RETRY_INTERVAL); - tTrace("use local epset, current in use: %d, retry count:%d, limit: %d", pEpSet->inUse, pCtx->retryCount + 1, - pEpSet->numOfEps * 3); + transPrintEpSet(pEpSet); + tTrace("%s use local epset, inUse: %d, retry count:%d, limit: %d", pTransInst->label, pEpSet->inUse, + pCtx->retryCount + 1, pEpSet->numOfEps * 3); + transUnrefCliHandle(pConn); return -1; } } else if (pCtx->retryCount < TRANS_RETRY_COUNT_LIMIT) { if (pResp->contLen == 0) { pEpSet->inUse = (++pEpSet->inUse) % pEpSet->numOfEps; - tTrace("use local epset, current in use: %d, retry count:%d, limit: %d", pEpSet->inUse, pCtx->retryCount + 1, - TRANS_RETRY_COUNT_LIMIT); + transPrintEpSet(&pCtx->epSet); + tTrace("%s use local epset, inUse: %d, retry count:%d, limit: %d", pTransInst->label, pEpSet->inUse, + pCtx->retryCount + 1, TRANS_RETRY_COUNT_LIMIT); } else { SEpSet epSet = {0}; tDeserializeSEpSet(pResp->pCont, pResp->contLen, &epSet); pCtx->epSet = epSet; - tTrace("use remote epset, current in use: %d, retry count:%d, limit: %d", pEpSet->inUse, pCtx->retryCount + 1, - TRANS_RETRY_COUNT_LIMIT); + + transPrintEpSet(&pCtx->epSet); + tTrace("%s use remote epset, inUse: %d, retry count:%d, limit: %d", pTransInst->label, pEpSet->inUse, + pCtx->retryCount + 1, TRANS_RETRY_COUNT_LIMIT); } addConnToPool(pThrd->pool, pConn); @@ -1002,16 +1008,16 @@ int cliAppCb(SCliConn* pConn, STransMsg* pResp, SCliMsg* pMsg) { } if (pCtx->pSem != NULL) { - tTrace("%s cli conn %p(sync) handle resp", pTransInst->label, pConn); + tTrace("%s conn %p(sync) handle resp", pTransInst->label, pConn); if (pCtx->pRsp == NULL) { - tTrace("%s cli conn %p(sync) failed to resp, ignore", pTransInst->label, pConn); + tTrace("%s conn %p(sync) failed to resp, ignore", pTransInst->label, pConn); } else { memcpy((char*)pCtx->pRsp, (char*)pResp, sizeof(*pResp)); } tsem_post(pCtx->pSem); pCtx->pRsp = NULL; } else { - tTrace("%s cli conn %p handle resp", pTransInst->label, pConn); + tTrace("%s conn %p handle resp", pTransInst->label, pConn); if (pResp->code != 0 || pCtx->retryCount == 0 || transEpSetIsEqual(&pCtx->epSet, &pCtx->origEpSet)) { pTransInst->cfp(pTransInst->parent, pResp, NULL); } else { @@ -1039,6 +1045,7 @@ void transRefCliHandle(void* handle) { return; } int ref = T_REF_INC((SCliConn*)handle); + tTrace("%s conn %p ref %d", CONN_GET_INST_LABEL((SCliConn*)handle), handle, ref); UNUSED(ref); } void transUnrefCliHandle(void* handle) { @@ -1046,7 +1053,7 @@ void transUnrefCliHandle(void* handle) { return; } int ref = T_REF_DEC((SCliConn*)handle); - tDebug("%s cli conn %p ref %d", CONN_GET_INST_LABEL((SCliConn*)handle), handle, ref); + tTrace("%s conn %p ref %d", CONN_GET_INST_LABEL((SCliConn*)handle), handle, ref); if (ref == 0) { cliDestroyConn((SCliConn*)handle, true); } @@ -1091,8 +1098,9 @@ void transSendRequest(void* shandle, const SEpSet* pEpSet, STransMsg* pReq, STra SCliThrdObj* thrd = ((SCliObj*)pTransInst->tcphandle)->pThreadObj[index]; - tDebug("send request at thread:%d, threadID: %08" PRId64 ", msg: %p, dst: %s:%d, app:%p", index, thrd->pid, pReq, - EPSET_GET_INUSE_IP(&pCtx->epSet), EPSET_GET_INUSE_PORT(&pCtx->epSet), pReq->info.ahandle); + tDebug("%s send request at thread:%d, threadID: %08" PRId64 ", msg: %p, dst: %s:%d, app:%p", pTransInst->label, + index, thrd->pid, pReq, EPSET_GET_INUSE_IP(&pCtx->epSet), EPSET_GET_INUSE_PORT(&pCtx->epSet), + pReq->info.ahandle); ASSERT(transSendAsync(thrd->asyncPool, &(cliMsg->q)) == 0); } @@ -1120,8 +1128,8 @@ void transSendRecv(void* shandle, const SEpSet* pEpSet, STransMsg* pReq, STransM cliMsg->type = Normal; SCliThrdObj* thrd = ((SCliObj*)pTransInst->tcphandle)->pThreadObj[index]; - tDebug("send request at thread:%d, threadID:%08" PRId64 ", msg: %p, dst: %s:%d, app:%p", index, thrd->pid, pReq, - EPSET_GET_INUSE_IP(&pCtx->epSet), EPSET_GET_INUSE_PORT(&pCtx->epSet), pReq->info.ahandle); + tDebug("%s send request at thread:%d, threadID:%08" PRId64 ", msg: %p, dst: %s:%d, app:%p", pTransInst->label, index, + thrd->pid, pReq, EPSET_GET_INUSE_IP(&pCtx->epSet), EPSET_GET_INUSE_PORT(&pCtx->epSet), pReq->info.ahandle); transSendAsync(thrd->asyncPool, &(cliMsg->q)); tsem_wait(sem); @@ -1151,7 +1159,7 @@ void transSetDefaultAddr(void* ahandle, const char* ip, const char* fqdn) { cliMsg->type = Update; SCliThrdObj* thrd = ((SCliObj*)pTransInst->tcphandle)->pThreadObj[i]; - tDebug("update epset at thread:%d, threadID:%08" PRId64 "", i, thrd->pid); + tDebug("%s update epset at thread:%d, threadID:%08" PRId64 "", pTransInst->label, i, thrd->pid); transSendAsync(thrd->asyncPool, &(cliMsg->q)); } diff --git a/source/libs/transport/src/transComm.c b/source/libs/transport/src/transComm.c index a04e8b5fca..61f791df9c 100644 --- a/source/libs/transport/src/transComm.c +++ b/source/libs/transport/src/transComm.c @@ -442,7 +442,7 @@ int transDQSched(SDelayQueue* queue, void (*func)(void* arg), void* arg, uint64_ } } - tTrace("timer %p put task into queue, timeoutMs: %" PRIu64 "", queue->timer, timeoutMs); + tTrace("timer %p put task into delay queue, timeoutMs: %" PRIu64 "", queue->timer, timeoutMs); heapInsert(queue->heap, &task->node); uv_timer_start(queue->timer, transDQTimeout, timeoutMs, 0); return 0; @@ -453,11 +453,17 @@ void transPrintEpSet(SEpSet* pEpSet) { tTrace("NULL epset"); return; } - tTrace("epset begin inUse: %d", pEpSet->inUse); + char buf[512] = {0}; + int len = snprintf(buf, sizeof(buf), "epset { "); for (int i = 0; i < pEpSet->numOfEps; i++) { - tTrace("ip: %s, port: %d", pEpSet->eps[i].fqdn, pEpSet->eps[i].port); + if (i == pEpSet->numOfEps - 1) { + len += snprintf(buf + len, sizeof(buf) - len, "%d. %s:%d ", i, pEpSet->eps[i].fqdn, pEpSet->eps[i].port); + } else { + len += snprintf(buf + len, sizeof(buf) - len, "%d. %s:%d, ", i, pEpSet->eps[i].fqdn, pEpSet->eps[i].port); + } } - tTrace("epset end"); + len += snprintf(buf + len, sizeof(buf) - len, "}"); + tTrace("%s, inUse: %d", buf, pEpSet->inUse); } bool transEpSetIsEqual(SEpSet* a, SEpSet* b) { if (a->numOfEps != b->numOfEps || a->inUse != b->inUse) { diff --git a/source/libs/transport/src/transSvr.c b/source/libs/transport/src/transSvr.c index 020435d076..f2351e3a8a 100644 --- a/source/libs/transport/src/transSvr.c +++ b/source/libs/transport/src/transSvr.c @@ -169,7 +169,7 @@ static bool addHandleToAcceptloop(void* arg); conn->status = ConnRelease; \ transClearBuffer(&conn->readBuf); \ transFreeMsg(transContFromHead((char*)head)); \ - tTrace("server conn %p received release request", conn); \ + tTrace("conn %p received release request", conn); \ \ STransMsg tmsg = {.code = 0, .info.handle = (void*)conn, .info.ahandle = NULL}; \ SSvrMsg* srvMsg = taosMemoryCalloc(1, sizeof(SSvrMsg)); \ @@ -181,7 +181,7 @@ static bool addHandleToAcceptloop(void* arg); return; \ } \ if (conn->regArg.init) { \ - tTrace("server conn %p release, notify server app", conn); \ + tTrace("conn %p release, notify server app", conn); \ STrans* pTransInst = conn->pTransInst; \ (*pTransInst->cfp)(pTransInst->parent, &(conn->regArg.msg), NULL); \ memset(&conn->regArg, 0, sizeof(conn->regArg)); \ @@ -209,25 +209,25 @@ static bool addHandleToAcceptloop(void* arg); #define ASYNC_CHECK_HANDLE(exh1, refId) \ do { \ if (refId > 0) { \ - tTrace("server handle step1"); \ + tTrace("handle step1"); \ SExHandle* exh2 = transAcquireExHandle(refMgt, refId); \ if (exh2 == NULL || refId != exh2->refId) { \ - tTrace("server handle %p except, may already freed, ignore msg, ref1: %" PRIu64 ", ref2 : %" PRIu64 "", exh1, \ + tTrace("handle %p except, may already freed, ignore msg, ref1: %" PRIu64 ", ref2 : %" PRIu64 "", exh1, \ exh2 ? exh2->refId : 0, refId); \ goto _return1; \ } \ } else if (refId == 0) { \ - tTrace("server handle step2"); \ + tTrace("handle step2"); \ SExHandle* exh2 = transAcquireExHandle(refMgt, refId); \ if (exh2 == NULL || refId != exh2->refId) { \ - tTrace("server handle %p except, may already freed, ignore msg, ref1: %" PRIu64 ", ref2 : %" PRIu64 "", exh1, \ - refId, exh2 ? exh2->refId : 0); \ + tTrace("handle %p except, may already freed, ignore msg, ref1: %" PRIu64 ", ref2 : %" PRIu64 "", exh1, refId, \ + exh2 ? exh2->refId : 0); \ goto _return1; \ } else { \ refId = exh1->refId; \ } \ } else if (refId < 0) { \ - tTrace("server handle step3"); \ + tTrace("handle step3"); \ goto _return2; \ } \ } while (0) @@ -280,16 +280,16 @@ static void uvHandleReq(SSvrConn* pConn) { if (pHead->persist == 1) { pConn->status = ConnAcquire; transRefSrvHandle(pConn); - tDebug("server conn %p acquired by server app", pConn); + tDebug("conn %p acquired by server app", pConn); } } if (pConn->status == ConnNormal && pHead->noResp == 0) { transRefSrvHandle(pConn); - tDebug("server conn %p %s received from %s:%d, local info: %s:%d, msg size: %d", pConn, TMSG_INFO(transMsg.msgType), + tDebug("conn %p %s received from %s:%d, local info: %s:%d, msg size: %d", pConn, TMSG_INFO(transMsg.msgType), taosInetNtoa(pConn->addr.sin_addr), ntohs(pConn->addr.sin_port), taosInetNtoa(pConn->localAddr.sin_addr), ntohs(pConn->localAddr.sin_port), transMsg.contLen); } else { - tDebug("server conn %p %s received from %s:%d, local info: %s:%d, msg size: %d, resp:%d ", pConn, + tDebug("conn %p %s received from %s:%d, local info: %s:%d, msg size: %d, resp:%d ", pConn, TMSG_INFO(transMsg.msgType), taosInetNtoa(pConn->addr.sin_addr), ntohs(pConn->addr.sin_port), taosInetNtoa(pConn->localAddr.sin_addr), ntohs(pConn->localAddr.sin_port), transMsg.contLen, pHead->noResp); // no ref here @@ -302,7 +302,7 @@ static void uvHandleReq(SSvrConn* pConn) { transMsg.info.handle = (void*)transAcquireExHandle(refMgt, pConn->refId); transMsg.info.refId = pConn->refId; - tTrace("server handle %p conn: %p translated to app, refId: %" PRIu64 "", transMsg.info.handle, pConn, pConn->refId); + tTrace("handle %p conn: %p translated to app, refId: %" PRIu64 "", transMsg.info.handle, pConn, pConn->refId); assert(transMsg.info.handle != NULL); if (pHead->noResp == 1) { transMsg.info.refId = -1; @@ -327,12 +327,12 @@ void uvOnRecvCb(uv_stream_t* cli, ssize_t nread, const uv_buf_t* buf) { SConnBuffer* pBuf = &conn->readBuf; if (nread > 0) { pBuf->len += nread; - tTrace("server conn %p read summary, total read: %d, current read: %d", conn, pBuf->len, (int)nread); + tTrace("conn %p total read: %d, current read: %d", conn, pBuf->len, (int)nread); if (transReadComplete(pBuf)) { - tTrace("server conn %p alread read complete packet", conn); + tTrace("conn %p alread read complete packet", conn); uvHandleReq(conn); } else { - tTrace("server %p read partial packet, continue to read", conn); + tTrace("conn %p read partial packet, continue to read", conn); } return; } @@ -340,12 +340,12 @@ void uvOnRecvCb(uv_stream_t* cli, ssize_t nread, const uv_buf_t* buf) { return; } - tError("server conn %p read error: %s", conn, uv_err_name(nread)); + tError("conn %p read error: %s", conn, uv_err_name(nread)); if (nread < 0) { conn->broken = true; if (conn->status == ConnAcquire) { if (conn->regArg.init) { - tTrace("server conn %p broken, notify server app", conn); + tTrace("conn %p broken, notify server app", conn); STrans* pTransInst = conn->pTransInst; (*pTransInst->cfp)(pTransInst->parent, &(conn->regArg.msg), NULL); memset(&conn->regArg, 0, sizeof(conn->regArg)); @@ -362,14 +362,14 @@ void uvAllocConnBufferCb(uv_handle_t* handle, size_t suggested_size, uv_buf_t* b void uvOnTimeoutCb(uv_timer_t* handle) { // opt SSvrConn* pConn = handle->data; - tError("server conn %p time out", pConn); + tError("conn %p time out", pConn); } void uvOnSendCb(uv_write_t* req, int status) { SSvrConn* conn = req->data; // transClearBuffer(&conn->readBuf); if (status == 0) { - tTrace("server conn %p data already was written on stream", conn); + tTrace("conn %p data already was written on stream", conn); if (!transQueueEmpty(&conn->srvMsgs)) { SSvrMsg* msg = transQueuePop(&conn->srvMsgs); // if (msg->type == Release && conn->status != ConnNormal) { @@ -406,7 +406,7 @@ void uvOnSendCb(uv_write_t* req, int status) { } } } else { - tError("server conn %p failed to write data, %s", conn, uv_err_name(status)); + tError("conn %p failed to write data, %s", conn, uv_err_name(status)); conn->broken = true; transUnrefSrvHandle(conn); } @@ -423,7 +423,7 @@ static void uvOnPipeWriteCb(uv_write_t* req, int status) { } static void uvPrepareSendData(SSvrMsg* smsg, uv_buf_t* wb) { - tTrace("server conn %p prepare to send resp", smsg->pConn); + tTrace("conn %p prepare to send resp", smsg->pConn); SSvrConn* pConn = smsg->pConn; STransMsg* pMsg = &smsg->msg; @@ -453,7 +453,7 @@ static void uvPrepareSendData(SSvrMsg* smsg, uv_buf_t* wb) { char* msg = (char*)pHead; int32_t len = transMsgLenFromCont(pMsg->contLen); - tDebug("server conn %p %s is sent to %s:%d, local info: %s:%d, msglen:%d", pConn, TMSG_INFO(pHead->msgType), + tDebug("conn %p %s is sent to %s:%d, local info: %s:%d, msglen:%d", pConn, TMSG_INFO(pHead->msgType), taosInetNtoa(pConn->addr.sin_addr), ntohs(pConn->addr.sin_port), taosInetNtoa(pConn->localAddr.sin_addr), ntohs(pConn->localAddr.sin_port), len); pHead->msgLen = htonl(len); @@ -544,7 +544,7 @@ void uvWorkerAsyncCb(uv_async_t* handle) { int64_t refId = transMsg.info.refId; SExHandle* exh2 = transAcquireExHandle(refMgt, refId); if (exh2 == NULL || exh1 != exh2) { - tTrace("server handle except msg %p, ignore it", exh1); + tTrace("handle except msg %p, ignore it", exh1); transReleaseExHandle(refMgt, refId); destroySmsg(msg); continue; @@ -582,18 +582,18 @@ static void uvShutDownCb(uv_shutdown_t* req, int status) { static void uvWorkDoTask(uv_work_t* req) { // doing time-consumeing task // only auth conn currently, add more func later - tTrace("server conn %p start to be processed in BG Thread", req->data); + tTrace("conn %p start to be processed in BG Thread", req->data); return; } static void uvWorkAfterTask(uv_work_t* req, int status) { if (status != 0) { - tTrace("server conn %p failed to processed ", req->data); + tTrace("conn %p failed to processed ", req->data); } // Done time-consumeing task // add more func later // this func called in main loop - tTrace("server conn %p already processed ", req->data); + tTrace("conn %p already processed ", req->data); taosMemoryFree(req); } @@ -628,7 +628,7 @@ void uvOnAcceptCb(uv_stream_t* stream, int status) { } } void uvOnConnectionCb(uv_stream_t* q, ssize_t nread, const uv_buf_t* buf) { - tTrace("server connection coming"); + tTrace("connection coming"); if (nread < 0) { if (nread != UV_EOF) { tError("read error %s", uv_err_name(nread)); @@ -677,18 +677,18 @@ void uvOnConnectionCb(uv_stream_t* q, ssize_t nread, const uv_buf_t* buf) { if (uv_accept(q, (uv_stream_t*)(pConn->pTcp)) == 0) { uv_os_fd_t fd; uv_fileno((const uv_handle_t*)pConn->pTcp, &fd); - tTrace("server conn %p created, fd: %d", pConn, fd); + tTrace("conn %p created, fd: %d", pConn, fd); int addrlen = sizeof(pConn->addr); if (0 != uv_tcp_getpeername(pConn->pTcp, (struct sockaddr*)&pConn->addr, &addrlen)) { - tError("server conn %p failed to get peer info", pConn); + tError("conn %p failed to get peer info", pConn); transUnrefSrvHandle(pConn); return; } addrlen = sizeof(pConn->localAddr); if (0 != uv_tcp_getsockname(pConn->pTcp, (struct sockaddr*)&pConn->localAddr, &addrlen)) { - tError("server conn %p failed to get local info", pConn); + tError("conn %p failed to get local info", pConn); transUnrefSrvHandle(pConn); return; } @@ -797,7 +797,7 @@ static SSvrConn* createConn(void* hThrd) { pConn->refId = exh->refId; transRefSrvHandle(pConn); - tTrace("server handle %p, conn %p created, refId: %" PRId64 "", exh, pConn, pConn->refId); + tTrace("handle %p, conn %p created, refId: %" PRId64 "", exh, pConn, pConn->refId); return pConn; } @@ -808,7 +808,7 @@ static void destroyConn(SSvrConn* conn, bool clear) { transDestroyBuffer(&conn->readBuf); if (clear) { - tTrace("server conn %p to be destroyed", conn); + tTrace("conn %p to be destroyed", conn); // uv_shutdown_t* req = taosMemoryMalloc(sizeof(uv_shutdown_t)); uv_close((uv_handle_t*)conn->pTcp, uvDestroyConn); // uv_close(conn->pTcp) @@ -844,7 +844,7 @@ static void uvDestroyConn(uv_handle_t* handle) { transReleaseExHandle(refMgt, conn->refId); transRemoveExHandle(refMgt, conn->refId); - tDebug("server conn %p destroy", conn); + tDebug("conn %p destroy", conn); // uv_timer_stop(&conn->pTimer); transQueueDestroy(&conn->srvMsgs); @@ -973,18 +973,18 @@ void uvHandleRelease(SSvrMsg* msg, SWorkThrdObj* thrd) { uvStartSendRespInternal(msg); return; } else if (conn->status == ConnRelease || conn->status == ConnNormal) { - tDebug("server conn %p already released, ignore release-msg", conn); + tDebug("conn %p already released, ignore release-msg", conn); } destroySmsg(msg); } void uvHandleResp(SSvrMsg* msg, SWorkThrdObj* thrd) { // send msg to client - tDebug("server conn %p start to send resp (2/2)", msg->pConn); + tDebug("conn %p start to send resp (2/2)", msg->pConn); uvStartSendResp(msg); } void uvHandleRegister(SSvrMsg* msg, SWorkThrdObj* thrd) { SSvrConn* conn = msg->pConn; - tDebug("server conn %p register brokenlink callback", conn); + tDebug("conn %p register brokenlink callback", conn); if (conn->status == ConnAcquire) { if (!transQueuePush(&conn->srvMsgs, msg)) { return; @@ -993,7 +993,7 @@ void uvHandleRegister(SSvrMsg* msg, SWorkThrdObj* thrd) { conn->regArg.notifyCount = 0; conn->regArg.init = 1; conn->regArg.msg = msg->msg; - tDebug("server conn %p register brokenlink callback succ", conn); + tDebug("conn %p register brokenlink callback succ", conn); if (conn->broken) { STrans* pTransInst = conn->pTransInst; @@ -1061,7 +1061,7 @@ void transRefSrvHandle(void* handle) { return; } int ref = T_REF_INC((SSvrConn*)handle); - tDebug("server conn %p ref count: %d", handle, ref); + tDebug("conn %p ref count: %d", handle, ref); } void transUnrefSrvHandle(void* handle) { @@ -1069,7 +1069,7 @@ void transUnrefSrvHandle(void* handle) { return; } int ref = T_REF_DEC((SSvrConn*)handle); - tDebug("server conn %p ref count: %d", handle, ref); + tDebug("conn %p ref count: %d", handle, ref); if (ref == 0) { destroyConn((SSvrConn*)handle, true); } @@ -1090,16 +1090,16 @@ void transReleaseSrvHandle(void* handle) { m->msg = tmsg; m->type = Release; - tTrace("server conn %p start to release", exh->handle); + tTrace("conn %p start to release", exh->handle); transSendAsync(pThrd->asyncPool, &m->q); transReleaseExHandle(refMgt, refId); return; _return1: - tTrace("server handle %p failed to send to release handle", exh); + tTrace("handle %p failed to send to release handle", exh); transReleaseExHandle(refMgt, refId); return; _return2: - tTrace("server handle %p failed to send to release handle", exh); + tTrace("handle %p failed to send to release handle", exh); return; } void transSendResponse(const STransMsg* msg) { @@ -1117,17 +1117,17 @@ void transSendResponse(const STransMsg* msg) { SSvrMsg* m = taosMemoryCalloc(1, sizeof(SSvrMsg)); m->msg = tmsg; m->type = Normal; - tDebug("server conn %p start to send resp (1/2)", exh->handle); + tDebug("conn %p start to send resp (1/2)", exh->handle); transSendAsync(pThrd->asyncPool, &m->q); transReleaseExHandle(refMgt, refId); return; _return1: - tTrace("server handle %p failed to send resp", exh); + tTrace("handle %p failed to send resp", exh); rpcFreeCont(msg->pCont); transReleaseExHandle(refMgt, refId); return; _return2: - tTrace("server handle %p failed to send resp", exh); + tTrace("handle %p failed to send resp", exh); rpcFreeCont(msg->pCont); return; } @@ -1145,18 +1145,18 @@ void transRegisterMsg(const STransMsg* msg) { SSvrMsg* m = taosMemoryCalloc(1, sizeof(SSvrMsg)); m->msg = tmsg; m->type = Register; - tTrace("server conn %p start to register brokenlink callback", exh->handle); + tTrace("conn %p start to register brokenlink callback", exh->handle); transSendAsync(pThrd->asyncPool, &m->q); transReleaseExHandle(refMgt, refId); return; _return1: - tTrace("server handle %p failed to send to register brokenlink", exh); + tTrace("handle %p failed to register brokenlink", exh); rpcFreeCont(msg->pCont); transReleaseExHandle(refMgt, refId); return; _return2: - tTrace("server handle %p failed to send to register brokenlink", exh); + tTrace("handle %p failed to register brokenlink", exh); rpcFreeCont(msg->pCont); }