Skip to content

T
TDengine

taosdata / TDengine
1 年多 前同步成功

通知 1185
Star 22016
Fork 4786
T
TDengine
提交 418be575 编写于 作者: dengyihao's avatar dengyihao
浏览文件
操作

enh: refactor debug info

上级 0671daa7
展开全部 隐藏空白更改
内联 并排
Showing with 137 addition and 123 deletion
source/libs/transport/src/transComm.c
浏览文件 @ 418be575
......@@ -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) {
......
source/libs/transport/src/transSvr.c
浏览文件 @ 418be575
......@@ -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);
}
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或