xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

remove void

This commit is contained in:
yihaoDeng 2024-09-19 14:50:08 +08:00
parent e1c099dd3d
commit 2e9c7c522f
4 changed files with 179 additions and 181 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

44
source/libs/transport/src/thttp.c
View File

@ -213,7 +213,7 @@ static FORCE_INLINE int32_t taosBuildDstAddr(const char* server, uint16_t port,
static void* httpThread(void* arg) {
SHttpModule* http = (SHttpModule*)arg;
setThreadName("http-cli-send-thread");
(void)uv_run(http->loop, UV_RUN_DEFAULT);
TAOS_UNUSED(uv_run(http->loop, UV_RUN_DEFAULT));
return NULL;
}
@ -324,7 +324,7 @@ static void httpAsyncCb(uv_async_t* handle) {
static int32_t BATCH_SIZE = 20;
int32_t count = 0;
(void)taosThreadMutexLock(&item->mtx);
TAOS_UNUSED(taosThreadMutexLock(&item->mtx));
httpMayDiscardMsg(http, item);
while (!QUEUE_IS_EMPTY(&item->qmsg) && count++ < BATCH_SIZE) {
@ -332,7 +332,7 @@ static void httpAsyncCb(uv_async_t* handle) {
QUEUE_REMOVE(h);
QUEUE_PUSH(&wq, h);
}
(void)taosThreadMutexUnlock(&item->mtx);
TAOS_UNUSED(taosThreadMutexUnlock(&item->mtx));
httpTrace(&wq);
@ -365,7 +365,7 @@ static FORCE_INLINE void clientCloseCb(uv_handle_t* handle) {
SHttpModule* http = taosAcquireRef(httpRefMgt, cli->chanId);
if (http != NULL) {
http->connNum -= 1;
(void)taosReleaseRef(httpRefMgt, chanId);
TAOS_UNUSED(taosReleaseRef(httpRefMgt, chanId));
}
destroyHttpClient(cli);
@ -425,7 +425,7 @@ static void clientConnCb(uv_connect_t* req, int32_t status) {
if (!uv_is_closing((uv_handle_t*)&cli->tcp)) {
uv_close((uv_handle_t*)&cli->tcp, clientCloseCb);
}
(void)taosReleaseRef(httpRefMgt, chanId);
TAOS_UNUSED(taosReleaseRef(httpRefMgt, chanId));
return;
}
http->connNum += 1;
@ -440,7 +440,7 @@ static void clientConnCb(uv_connect_t* req, int32_t status) {
uv_close((uv_handle_t*)&cli->tcp, clientCloseCb);
}
}
(void)taosReleaseRef(httpRefMgt, chanId);
TAOS_UNUSED(taosReleaseRef(httpRefMgt, chanId));
}
int32_t httpSendQuit(SHttpModule* http, int64_t chanId) {
@ -486,7 +486,7 @@ static void httpHandleQuit(SHttpMsg* msg) {
SHttpModule* http = taosAcquireRef(httpRefMgt, chanId);
if (http == NULL) return;
uv_walk(http->loop, httpWalkCb, NULL);
(void)taosReleaseRef(httpRefMgt, chanId);
TAOS_UNUSED(taosReleaseRef(httpRefMgt, chanId));
}
static bool httpFailFastShoudIgnoreMsg(SHashObj* pTable, char* server, int16_t port) {
@ -508,11 +508,11 @@ static bool httpFailFastShoudIgnoreMsg(SHashObj* pTable, char* server, int16_t p
}
static void httpFailFastMayUpdate(SHashObj* pTable, char* server, int16_t port, int8_t succ) {
int32_t code = 0;
char buf[256] = {0};
char buf[256] = {0};
sprintf(buf, "%s:%d", server, port);
if (succ) {
(void)taosHashRemove(pTable, buf, strlen(buf));
TAOS_UNUSED(taosHashRemove(pTable, buf, strlen(buf)));
} else {
int32_t st = taosGetTimestampSec();
if ((code = taosHashPut(pTable, buf, strlen(buf), &st, sizeof(st))) != 0) {
@ -599,7 +599,7 @@ static void httpHandleReq(SHttpMsg* msg) {
tError("http-report failed to alloc read buf, dst:%s:%d, chanId:%" PRId64 ", seq:%" PRId64 ",reason:%s", cli->addr,
cli->port, chanId, cli->seq, tstrerror(TSDB_CODE_OUT_OF_MEMORY));
destroyHttpClient(cli);
(void)taosReleaseRef(httpRefMgt, chanId);
TAOS_UNUSED(taosReleaseRef(httpRefMgt, chanId));
return;
}
@ -608,7 +608,7 @@ static void httpHandleReq(SHttpMsg* msg) {
tError("http-report failed to init socket handle, dst:%s:%d, chanId:%" PRId64 ", seq:%" PRId64 ", reason:%s",
cli->addr, cli->port, chanId, cli->seq, uv_strerror(err));
destroyHttpClient(cli);
(void)taosReleaseRef(httpRefMgt, chanId);
TAOS_UNUSED(taosReleaseRef(httpRefMgt, chanId));
return;
}
@ -618,7 +618,7 @@ static void httpHandleReq(SHttpMsg* msg) {
tError("http-report failed to open socket, dst:%s:%d, chanId:%" PRId64 ", seq:%" PRId64 ", reason:%s", cli->addr,
cli->port, chanId, cli->seq, tstrerror(terrno));
destroyHttpClient(cli);
(void)taosReleaseRef(httpRefMgt, chanId);
TAOS_UNUSED(taosReleaseRef(httpRefMgt, chanId));
return;
}
@ -628,7 +628,7 @@ static void httpHandleReq(SHttpMsg* msg) {
cli->port, chanId, cli->seq, uv_strerror(ret));
destroyHttpClient(cli);
(void)taosReleaseRef(httpRefMgt, chanId);
TAOS_UNUSED(taosReleaseRef(httpRefMgt, chanId));
return;
}
@ -639,7 +639,7 @@ static void httpHandleReq(SHttpMsg* msg) {
httpFailFastMayUpdate(http->connStatusTable, cli->addr, cli->port, 0);
destroyHttpClient(cli);
}
(void)taosReleaseRef(httpRefMgt, chanId);
TAOS_UNUSED(taosReleaseRef(httpRefMgt, chanId));
return;
END:
@ -649,7 +649,7 @@ END:
}
httpDestroyMsg(msg);
taosMemoryFree(header);
(void)taosReleaseRef(httpRefMgt, chanId);
TAOS_UNUSED(taosReleaseRef(httpRefMgt, chanId));
}
static void httpModuleDestroy(SHttpModule* http) {
@ -660,7 +660,7 @@ static void httpModuleDestroy(SHttpModule* http) {
transAsyncPoolDestroy(http->asyncPool);
}
if (http->loop) {
(void)uv_loop_close(http->loop);
TAOS_UNUSED(uv_loop_close(http->loop));
taosMemoryFree(http->loop);
}
@ -720,7 +720,7 @@ int32_t taosSendHttpReportByChan(const char* server, const char* uri, uint16_t p
int32_t taosSendHttpReport(const char* server, const char* uri, uint16_t port, char* pCont, int32_t contLen,
EHttpCompFlag flag) {
(void)taosThreadOnce(&transHttpInit, transHttpEnvInit);
TAOS_UNUSED(taosThreadOnce(&transHttpInit, transHttpEnvInit));
return taosSendHttpReportImplByChan(server, uri, port, pCont, contLen, flag, httpDefaultChanId);
}
@ -791,7 +791,7 @@ _ERROR:
return code;
}
int64_t taosInitHttpChan() {
(void)taosThreadOnce(&transHttpInit, transHttpEnvInit);
TAOS_UNUSED(taosThreadOnce(&transHttpInit, transHttpEnvInit));
return transInitHttpChanImpl();
}
@ -810,14 +810,14 @@ void taosDestroyHttpChan(int64_t chanId) {
ret = httpSendQuit(load, chanId);
if (ret != 0) {
tDebug("http-report already destroyed, chanId %" PRId64 ",reason:%s", chanId, tstrerror(ret));
(void)taosReleaseRef(httpRefMgt, chanId);
TAOS_UNUSED(taosReleaseRef(httpRefMgt, chanId));
return;
}
(void)taosThreadJoin(load->thread, NULL);
TAOS_UNUSED(taosThreadJoin(load->thread, NULL));
httpModuleDestroy(load);
(void)taosReleaseRef(httpRefMgt, chanId);
(void)taosRemoveRef(httpRefMgt, chanId);
TAOS_UNUSED(taosReleaseRef(httpRefMgt, chanId));
TAOS_UNUSED(taosRemoveRef(httpRefMgt, chanId));
}

205
source/libs/transport/src/transCli.c
View File

@ -233,11 +233,11 @@ static void destroyThrdObj(SCliThrd* pThrd);
int32_t cliSendQuit(SCliThrd* thrd);
static void cliWalkCb(uv_handle_t* handle, void* arg);
#define CLI_RELEASE_UV(loop) \
do { \
(void)uv_walk(loop, cliWalkCb, NULL); \
(void)uv_run(loop, UV_RUN_DEFAULT); \
(void)uv_loop_close(loop); \
#define CLI_RELEASE_UV(loop) \
do { \
TAOS_UNUSED(uv_walk(loop, cliWalkCb, NULL)); \
TAOS_UNUSED(uv_run(loop, UV_RUN_DEFAULT)); \
TAOS_UNUSED(uv_loop_close(loop)); \
} while (0);
// snprintf may cause performance problem
@ -247,7 +247,7 @@ static void cliWalkCb(uv_handle_t* handle, void* arg);
int16_t len = strlen(ip); \
if (ip != NULL) memcpy(t, ip, len); \
t[len] = ':'; \
(void)titoa(port, 10, &t[len + 1]); \
TAOS_UNUSED(titoa(port, 10, &t[len + 1])); \
} while (0)
#define CONN_PERSIST_TIME(para) ((para) <= 90000 ? 90000 : (para))
@ -336,7 +336,7 @@ void cliResetTimer(SCliThrd* pThrd, SCliConn* conn) {
if (conn->timer) {
if (uv_is_active((uv_handle_t*)conn->timer)) {
tDebug("%s conn %p stop timer", CONN_GET_INST_LABEL(conn), conn);
(void)uv_timer_stop(conn->timer);
TAOS_UNUSED(uv_timer_stop(conn->timer));
}
if (taosArrayPush(pThrd->timerList, &conn->timer) == NULL) {
tError("failed to push timer %p to list, reason:%s", conn->timer, tstrerror(TSDB_CODE_OUT_OF_MEMORY));
@ -375,15 +375,15 @@ bool cliConnSendSeqMsg(int64_t refId, SCliConn* conn) {
taosWUnLockLatch(&exh->latch);
SCliMsg* t = QUEUE_DATA(h, SCliMsg, seqq);
transCtxMerge(&conn->ctx, &t->ctx->appCtx);
(void)transQueuePush(&conn->cliMsgs, t);
TAOS_UNUSED(transQueuePush(&conn->cliMsgs, t));
tDebug("pop from conn %p, refId: %" PRId64 "", conn, refId);
(void)transReleaseExHandle(transGetRefMgt(), refId);
TAOS_UNUSED(transReleaseExHandle(transGetRefMgt(), refId));
cliSend(conn);
return true;
}
taosWUnLockLatch(&exh->latch);
tDebug("empty conn %p, refId: %" PRId64 "", conn, refId);
(void)transReleaseExHandle(transGetRefMgt(), refId);
TAOS_UNUSED(transReleaseExHandle(transGetRefMgt(), refId));
return false;
}
@ -496,7 +496,7 @@ void cliHandleResp(SCliConn* conn) {
return addConnToPool(pThrd->pool, conn);
}
(void)uv_read_start((uv_stream_t*)conn->stream, cliAllocRecvBufferCb, cliRecvCb);
TAOS_UNUSED(uv_read_start((uv_stream_t*)conn->stream, cliAllocRecvBufferCb, cliRecvCb));
}
static void cliDestroyMsgInExhandle(int64_t refId) {
if (refId == 0) return;
@ -510,7 +510,7 @@ static void cliDestroyMsgInExhandle(int64_t refId) {
destroyCmsg(t);
}
taosWUnLockLatch(&exh->latch);
(void)transReleaseExHandle(transGetRefMgt(), refId);
TAOS_UNUSED(transReleaseExHandle(transGetRefMgt(), refId));
}
}
@ -598,7 +598,7 @@ void cliConnTimeout(uv_timer_t* handle) {
tTrace("%s conn %p conn timeout, ref:%d", CONN_GET_INST_LABEL(conn), conn, T_REF_VAL_GET(conn));
(void)uv_timer_stop(handle);
TAOS_UNUSED(uv_timer_stop(handle));
handle->data = NULL;
cliResetTimer(pThrd, conn);
@ -610,7 +610,7 @@ void cliReadTimeoutCb(uv_timer_t* handle) {
// set up timeout cb
SCliConn* conn = handle->data;
tTrace("%s conn %p timeout, ref:%d", CONN_GET_INST_LABEL(conn), conn, T_REF_VAL_GET(conn));
(void)uv_read_stop(conn->stream);
TAOS_UNUSED(uv_read_stop(conn->stream));
cliHandleExceptImpl(conn, TSDB_CODE_RPC_TIMEOUT);
}
@ -847,7 +847,7 @@ static void addConnToPool(void* pool, SCliConn* conn) {
pMsg->ctx->task = NULL;
transCtxMerge(&conn->ctx, &pMsg->ctx->appCtx);
(void)transQueuePush(&conn->cliMsgs, pMsg);
TAOS_UNUSED(transQueuePush(&conn->cliMsgs, pMsg));
conn->status = ConnNormal;
cliSend(conn);
@ -871,8 +871,8 @@ static void addConnToPool(void* pool, SCliConn* conn) {
}
static int32_t allocConnRef(SCliConn* conn, bool update) {
if (update) {
(void)transReleaseExHandle(transGetRefMgt(), conn->refId);
(void)transRemoveExHandle(transGetRefMgt(), conn->refId);
TAOS_UNUSED(transReleaseExHandle(transGetRefMgt(), conn->refId));
TAOS_UNUSED((transGetRefMgt(), conn->refId));
conn->refId = -1;
}
@ -907,8 +907,8 @@ static int32_t allocConnRef(SCliConn* conn, bool update) {
static int32_t specifyConnRef(SCliConn* conn, bool update, int64_t handle) {
if (update) {
(void)transReleaseExHandle(transGetRefMgt(), conn->refId);
(void)transRemoveExHandle(transGetRefMgt(), conn->refId);
TAOS_UNUSED(transReleaseExHandle(transGetRefMgt(), conn->refId));
TAOS_UNUSED(transRemoveExHandle(transGetRefMgt(), conn->refId));
conn->refId = -1;
}
SExHandle* exh = transAcquireExHandle(transGetRefMgt(), handle);
@ -924,7 +924,7 @@ static int32_t specifyConnRef(SCliConn* conn, bool update, int64_t handle) {
tDebug("conn %p specified by %" PRId64 "", conn, handle);
(void)transReleaseExHandle(transGetRefMgt(), handle);
TAOS_UNUSED(transReleaseExHandle(transGetRefMgt(), handle));
return 0;
}
@ -1004,7 +1004,7 @@ static int32_t cliCreateConn(SCliThrd* pThrd, SCliConn** pCliConn) {
}
tDebug("no available timer, create a timer %p", timer);
(void)uv_timer_init(pThrd->loop, timer);
TAOS_UNUSED(uv_timer_init(pThrd->loop, timer));
}
timer->data = conn;
@ -1022,7 +1022,7 @@ static int32_t cliCreateConn(SCliThrd* pThrd, SCliConn** pCliConn) {
conn->broken = false;
transRefCliHandle(conn);
(void)atomic_add_fetch_32(&pThrd->connCount, 1);
TAOS_UNUSED(atomic_add_fetch_32(&pThrd->connCount, 1));
TAOS_CHECK_GOTO(allocConnRef(conn, false), NULL, _failed);
@ -1032,7 +1032,7 @@ _failed:
if (conn) {
taosMemoryFree(conn->stream);
transReqQueueClear(&conn->wreqQueue);
(void)transDestroyBuffer(&conn->readBuf);
TAOS_UNUSED(transDestroyBuffer(&conn->readBuf));
transQueueDestroy(&conn->cliMsgs);
}
taosMemoryFree(conn);
@ -1059,8 +1059,8 @@ static void cliDestroyConn(SCliConn* conn, bool clear) {
}
conn->list = NULL;
(void)transReleaseExHandle(transGetRefMgt(), conn->refId);
(void)transRemoveExHandle(transGetRefMgt(), conn->refId);
TAOS_UNUSED(transReleaseExHandle(transGetRefMgt(), conn->refId));
TAOS_UNUSED(transRemoveExHandle(transGetRefMgt(), conn->refId));
conn->refId = -1;
if (conn->task != NULL) {
@ -1071,7 +1071,7 @@ static void cliDestroyConn(SCliConn* conn, bool clear) {
if (clear) {
if (!uv_is_closing((uv_handle_t*)conn->stream)) {
(void)uv_read_stop(conn->stream);
TAOS_UNUSED(uv_read_stop(conn->stream));
uv_close((uv_handle_t*)conn->stream, cliDestroy);
}
}
@ -1084,11 +1084,11 @@ static void cliDestroy(uv_handle_t* handle) {
SCliThrd* pThrd = conn->hostThrd;
cliResetTimer(pThrd, conn);
(void)atomic_sub_fetch_32(&pThrd->connCount, 1);
TAOS_UNUSED(atomic_sub_fetch_32(&pThrd->connCount, 1));
if (conn->refId > 0) {
(void)transReleaseExHandle(transGetRefMgt(), conn->refId);
(void)transRemoveExHandle(transGetRefMgt(), conn->refId);
TAOS_UNUSED(transReleaseExHandle(transGetRefMgt(), conn->refId));
TAOS_UNUSED(transRemoveExHandle(transGetRefMgt(), conn->refId));
}
taosMemoryFree(conn->dstAddr);
taosMemoryFree(conn->stream);
@ -1097,7 +1097,7 @@ static void cliDestroy(uv_handle_t* handle) {
tTrace("%s conn %p destroy successfully", CONN_GET_INST_LABEL(conn), conn);
transReqQueueClear(&conn->wreqQueue);
(void)transDestroyBuffer(&conn->readBuf);
TAOS_UNUSED(transDestroyBuffer(&conn->readBuf));
taosMemoryFree(conn);
}
@ -1106,7 +1106,7 @@ static bool cliHandleNoResp(SCliConn* conn) {
if (!transQueueEmpty(&conn->cliMsgs)) {
SCliMsg* pMsg = transQueueGet(&conn->cliMsgs, 0);
if (REQUEST_NO_RESP(&pMsg->msg)) {
(void)transQueuePop(&conn->cliMsgs);
TAOS_UNUSED(transQueuePop(&conn->cliMsgs));
destroyCmsg(pMsg);
res = true;
}
@ -1153,7 +1153,7 @@ static void cliSendCb(uv_write_t* req, int status) {
tTrace("%s conn %p no resp required", CONN_GET_INST_LABEL(pConn), pConn);
return;
}
(void)uv_read_start((uv_stream_t*)pConn->stream, cliAllocRecvBufferCb, cliRecvCb);
TAOS_UNUSED(uv_read_start((uv_stream_t*)pConn->stream, cliAllocRecvBufferCb, cliRecvCb));
}
void cliSendBatch(SCliConn* pConn) {
int32_t code = 0;
@ -1296,13 +1296,13 @@ void cliSend(SCliConn* pConn) {
if (timer == NULL) {
timer = taosMemoryCalloc(1, sizeof(uv_timer_t));
tDebug("no available timer, create a timer %p", timer);
(void)uv_timer_init(pThrd->loop, timer);
TAOS_UNUSED(uv_timer_init(pThrd->loop, timer));
}
timer->data = pConn;
pConn->timer = timer;
tGTrace("%s conn %p start timer for msg:%s", CONN_GET_INST_LABEL(pConn), pConn, TMSG_INFO(pMsg->msgType));
(void)uv_timer_start((uv_timer_t*)pConn->timer, cliReadTimeoutCb, TRANS_READ_TIMEOUT, 0);
TAOS_UNUSED(uv_timer_start((uv_timer_t*)pConn->timer, cliReadTimeoutCb, TRANS_READ_TIMEOUT, 0));
}
if (pHead->comp == 0 && pMsg->info.compressed == 0 && pConn->clientIp != pConn->serverIp) {
@ -1402,8 +1402,7 @@ static void cliHandleBatchReq(SCliBatch* pBatch, SCliThrd* pThrd) {
tTrace("%s conn %p try to connect to %s", pTransInst->label, conn, pList->dst);
int32_t fd = taosCreateSocketWithTimeout(TRANS_CONN_TIMEOUT * 10);
if (fd == -1) {
tError("%s conn %p failed to create socket, reason:%s", transLabel(pTransInst), conn,
tstrerror(terrno));
tError("%s conn %p failed to create socket, reason:%s", transLabel(pTransInst), conn, tstrerror(terrno));
cliHandleFastFail(conn, -1);
return;
}
@ -1428,7 +1427,7 @@ static void cliHandleBatchReq(SCliBatch* pBatch, SCliThrd* pThrd) {
cliHandleFastFail(conn, -1);
return;
}
(void)uv_timer_start(conn->timer, cliConnTimeout, TRANS_CONN_TIMEOUT, 0);
TAOS_UNUSED(uv_timer_start(conn->timer, cliConnTimeout, TRANS_CONN_TIMEOUT, 0));
return;
}
@ -1541,12 +1540,12 @@ void cliConnCb(uv_connect_t* req, int status) {
struct sockaddr peername, sockname;
int addrlen = sizeof(peername);
(void)uv_tcp_getpeername((uv_tcp_t*)pConn->stream, &peername, &addrlen);
(void)transSockInfo2Str(&peername, pConn->dst);
TAOS_UNUSED(uv_tcp_getpeername((uv_tcp_t*)pConn->stream, &peername, &addrlen));
TAOS_UNUSED(transSockInfo2Str(&peername, pConn->dst));
addrlen = sizeof(sockname);
(void)uv_tcp_getsockname((uv_tcp_t*)pConn->stream, &sockname, &addrlen);
(void)transSockInfo2Str(&sockname, pConn->src);
TAOS_UNUSED(uv_tcp_getsockname((uv_tcp_t*)pConn->stream, &sockname, &addrlen));
TAOS_UNUSED(transSockInfo2Str(&sockname, pConn->src));
struct sockaddr_in addr = *(struct sockaddr_in*)&sockname;
struct sockaddr_in saddr = *(struct sockaddr_in*)&peername;
@ -1596,8 +1595,8 @@ static void cliHandleQuit(SCliMsg* pMsg, SCliThrd* pThrd) {
tDebug("cli work thread %p start to quit", pThrd);
destroyCmsg(pMsg);
(void)destroyConnPool(pThrd);
(void)uv_walk(pThrd->loop, cliWalkCb, NULL);
TAOS_UNUSED(destroyConnPool(pThrd));
TAOS_UNUSED(uv_walk(pThrd->loop, cliWalkCb, NULL));
}
static void cliHandleRelease(SCliMsg* pMsg, SCliThrd* pThrd) {
int64_t refId = (int64_t)(pMsg->msg.info.handle);
@ -1611,8 +1610,7 @@ static void cliHandleRelease(SCliMsg* pMsg, SCliThrd* pThrd) {
taosRLockLatch(&exh->latch);
SCliConn* conn = exh->handle;
taosRUnLockLatch(&exh->latch);
(void)transReleaseExHandle(transGetRefMgt(), refId);
TAOS_UNUSED(transReleaseExHandle(transGetRefMgt(), refId));
tDebug("%s conn %p start to release to inst", CONN_GET_INST_LABEL(conn), conn);
if (T_REF_VAL_GET(conn) == 2) {
@ -1656,7 +1654,7 @@ static void cliHandleFreeById(SCliMsg* pMsg, SCliThrd* pThrd) {
TAOS_CHECK_GOTO(TSDB_CODE_REF_INVALID_ID, NULL, _exception);
} else {
destroyCmsg(pMsg);
(void)transReleaseExHandle(transGetRefMgt(), refId);
TAOS_UNUSED(transReleaseExHandle(transGetRefMgt(), refId));
while (T_REF_VAL_GET(conn) >= 1) {
transUnrefCliHandle(conn);
@ -1666,9 +1664,9 @@ static void cliHandleFreeById(SCliMsg* pMsg, SCliThrd* pThrd) {
_exception:
tDebug("already free conn %p by id %" PRId64 "", conn, refId);
(void)transReleaseExHandle(transGetRefMgt(), refId);
(void)transReleaseExHandle(transGetRefMgt(), refId);
(void)transRemoveExHandle(transGetRefMgt(), refId);
TAOS_UNUSED(transReleaseExHandle(transGetRefMgt(), refId));
TAOS_UNUSED(transReleaseExHandle(transGetRefMgt(), refId));
TAOS_UNUSED(transRemoveExHandle(transGetRefMgt(), refId));
destroyCmsg(pMsg);
}
@ -1691,7 +1689,7 @@ SCliConn* cliGetConn(SCliMsg** pMsg, SCliThrd* pThrd, bool* ignore, char* addr)
conn = getConnFromPool2(pThrd, addr, pMsg);
if (conn != NULL) specifyConnRef(conn, true, refId);
}
(void)transReleaseExHandle(transGetRefMgt(), refId);
TAOS_UNUSED(transReleaseExHandle(transGetRefMgt(), refId));
}
return conn;
};
@ -1787,7 +1785,7 @@ static void cliMayUpdateFqdnCache(SHashObj* cache, char* dst) {
if (i > 0) {
char fqdn[TSDB_FQDN_LEN + 1] = {0};
memcpy(fqdn, dst, i);
(void)cliUpdateFqdnCache(cache, fqdn);
TAOS_UNUSED(cliUpdateFqdnCache(cache, fqdn));
}
}
@ -1824,7 +1822,7 @@ void cliHandleReq(SCliMsg* pMsg, SCliThrd* pThrd) {
if (ignore == true) {
// persist conn already release by server
STransMsg resp = {0};
(void)cliBuildExceptResp(pMsg, &resp);
TAOS_UNUSED(cliBuildExceptResp(pMsg, &resp));
// refactorr later
resp.info.cliVer = pTransInst->compatibilityVer;
@ -1841,14 +1839,14 @@ void cliHandleReq(SCliMsg* pMsg, SCliThrd* pThrd) {
if (conn != NULL) {
transCtxMerge(&conn->ctx, &pMsg->ctx->appCtx);
(void)transQueuePush(&conn->cliMsgs, pMsg);
TAOS_UNUSED(transQueuePush(&conn->cliMsgs, pMsg));
cliSend(conn);
} else {
code = cliCreateConn(pThrd, &conn);
if (code != 0) {
tError("%s failed to create conn, reason:%s", pTransInst->label, tstrerror(code));
STransMsg resp = {.code = code};
(void)cliBuildExceptResp(pMsg, &resp);
TAOS_UNUSED(cliBuildExceptResp(pMsg, &resp));
resp.info.cliVer = pTransInst->compatibilityVer;
if (pMsg->type != Release) {
@ -1862,7 +1860,7 @@ void cliHandleReq(SCliMsg* pMsg, SCliThrd* pThrd) {
if (refId != 0) specifyConnRef(conn, true, refId);
transCtxMerge(&conn->ctx, &pMsg->ctx->appCtx);
(void)transQueuePush(&conn->cliMsgs, pMsg);
TAOS_UNUSED(transQueuePush(&conn->cliMsgs, pMsg));
conn->dstAddr = taosStrdup(addr);
@ -1882,8 +1880,7 @@ void cliHandleReq(SCliMsg* pMsg, SCliThrd* pThrd) {
tGTrace("%s conn %p try to connect to %s", pTransInst->label, conn, conn->dstAddr);
int32_t fd = taosCreateSocketWithTimeout(TRANS_CONN_TIMEOUT * 10);
if (fd == -1) {
tGError("%s conn %p failed to create socket, reason:%s", transLabel(pTransInst), conn,
tstrerror(terrno));
tGError("%s conn %p failed to create socket, reason:%s", transLabel(pTransInst), conn, tstrerror(terrno));
cliHandleExcept(conn, -1);
terrno = 0;
return;
@ -1911,7 +1908,7 @@ void cliHandleReq(SCliMsg* pMsg, SCliThrd* pThrd) {
cliHandleFastFail(conn, ret);
return;
}
(void)uv_timer_start(conn->timer, cliConnTimeout, TRANS_CONN_TIMEOUT, 0);
TAOS_UNUSED(uv_timer_start(conn->timer, cliConnTimeout, TRANS_CONN_TIMEOUT, 0));
}
tGTrace("%s conn %p ready", pTransInst->label, conn);
}
@ -2110,9 +2107,9 @@ static void cliAsyncCb(uv_async_t* handle) {
// batch process to avoid to lock/unlock frequently
queue wq;
(void)taosThreadMutexLock(&item->mtx);
TAOS_UNUSED(taosThreadMutexLock(&item->mtx));
QUEUE_MOVE(&item->qmsg, &wq);
(void)taosThreadMutexUnlock(&item->mtx);
TAOS_UNUSED(taosThreadMutexUnlock(&item->mtx));
int8_t supportBatch = pTransInst->supportBatch;
if (supportBatch == 0) {
@ -2162,7 +2159,7 @@ bool cliRecvReleaseReq(SCliConn* conn, STransMsgHead* pHead) {
tDebug("%s conn %p receive release request, refId:%" PRId64 ", may ignore", CONN_GET_INST_LABEL(conn), conn,
conn->refId);
(void)transClearBuffer(&conn->readBuf);
TAOS_UNUSED(transClearBuffer(&conn->readBuf));
transFreeMsg(transContFromHead((char*)pHead));
for (int i = 0; ahandle == 0 && i < transQueueSize(&conn->cliMsgs); i++) {
@ -2192,10 +2189,10 @@ static void* cliWorkThread(void* arg) {
SCliThrd* pThrd = (SCliThrd*)arg;
pThrd->pid = taosGetSelfPthreadId();
(void)strtolower(threadName, pThrd->pTransInst->label);
TAOS_UNUSED(strtolower(threadName, pThrd->pTransInst->label));
setThreadName(threadName);
(void)uv_run(pThrd->loop, UV_RUN_DEFAULT);
TAOS_UNUSED(uv_run(pThrd->loop, UV_RUN_DEFAULT));
tDebug("thread quit-thread:%08" PRId64, pThrd->pid);
return NULL;
@ -2239,7 +2236,7 @@ _err:
if (cli) {
for (int i = 0; i < cli->numOfThreads; i++) {
if (cli->pThreadObj[i]) {
(void)cliSendQuit(cli->pThreadObj[i]);
TAOS_UNUSED(cliSendQuit(cli->pThreadObj[i]));
destroyThrdObj(cli->pThreadObj[i]);
}
}
@ -2283,8 +2280,8 @@ static FORCE_INLINE void destroyCmsgAndAhandle(void* param) {
}
if (pMsg->msg.info.handle != 0) {
(void)transReleaseExHandle(transGetRefMgt(), (int64_t)pMsg->msg.info.handle);
(void)transRemoveExHandle(transGetRefMgt(), (int64_t)pMsg->msg.info.handle);
TAOS_UNUSED(transReleaseExHandle(transGetRefMgt(), (int64_t)pMsg->msg.info.handle));
TAOS_UNUSED(transRemoveExHandle(transGetRefMgt(), (int64_t)pMsg->msg.info.handle));
}
transDestroyConnCtx(pMsg->ctx);
@ -2302,7 +2299,7 @@ static int32_t createThrdObj(void* trans, SCliThrd** ppThrd) {
}
QUEUE_INIT(&pThrd->msg);
(void)taosThreadMutexInit(&pThrd->msgMtx, NULL);
TAOS_UNUSED(taosThreadMutexInit(&pThrd->msgMtx, NULL));
pThrd->loop = (uv_loop_t*)taosMemoryMalloc(sizeof(uv_loop_t));
if (pThrd->loop == NULL) {
@ -2366,7 +2363,7 @@ static int32_t createThrdObj(void* trans, SCliThrd** ppThrd) {
if (timer == NULL) {
TAOS_CHECK_GOTO(TSDB_CODE_OUT_OF_MEMORY, NULL, _end);
}
(void)uv_timer_init(pThrd->loop, timer);
TAOS_UNUSED(uv_timer_init(pThrd->loop, timer));
if (taosArrayPush(pThrd->timerList, &timer) == NULL) {
TAOS_CHECK_GOTO(TSDB_CODE_OUT_OF_MEMORY, NULL, _end);
}
@ -2380,19 +2377,19 @@ static int32_t createThrdObj(void* trans, SCliThrd** ppThrd) {
_end:
if (pThrd) {
(void)taosThreadMutexDestroy(&pThrd->msgMtx);
TAOS_UNUSED(taosThreadMutexDestroy(&pThrd->msgMtx));
(void)uv_loop_close(pThrd->loop);
TAOS_UNUSED(uv_loop_close(pThrd->loop));
taosMemoryFree(pThrd->loop);
transAsyncPoolDestroy(pThrd->asyncPool);
for (int i = 0; i < taosArrayGetSize(pThrd->timerList); i++) {
uv_timer_t* timer = taosArrayGetP(pThrd->timerList, i);
(void)uv_timer_stop(timer);
TAOS_UNUSED(uv_timer_stop(timer));
taosMemoryFree(timer);
}
taosArrayDestroy(pThrd->timerList);
(void)destroyConnPool(pThrd);
TAOS_UNUSED(destroyConnPool(pThrd));
transDQDestroy(pThrd->delayQueue, NULL);
transDQDestroy(pThrd->timeoutQueue, NULL);
transDQDestroy(pThrd->waitConnQueue, NULL);
@ -2409,9 +2406,9 @@ static void destroyThrdObj(SCliThrd* pThrd) {
return;
}
(void)taosThreadJoin(pThrd->thread, NULL);
TAOS_UNUSED(taosThreadJoin(pThrd->thread, NULL));
CLI_RELEASE_UV(pThrd->loop);
(void)taosThreadMutexDestroy(&pThrd->msgMtx);
TAOS_UNUSED(taosThreadMutexDestroy(&pThrd->msgMtx));
TRANS_DESTROY_ASYNC_POOL_MSG(pThrd->asyncPool, SCliMsg, destroyCmsgWrapper, (void*)pThrd);
transAsyncPoolDestroy(pThrd->asyncPool);
@ -2478,9 +2475,9 @@ void cliWalkCb(uv_handle_t* handle, void* arg) {
if (uv_handle_get_type(handle) == UV_TIMER) {
// do nothing
} else {
(void)uv_read_stop((uv_stream_t*)handle);
TAOS_UNUSED(uv_read_stop((uv_stream_t*)handle));
}
(void)uv_close(handle, cliDestroy);
TAOS_UNUSED(uv_close(handle, cliDestroy));
}
}
@ -2518,7 +2515,7 @@ static void cliSchedMsgToDebug(SCliMsg* pMsg, char* label) {
STransConnCtx* pCtx = pMsg->ctx;
STraceId* trace = &pMsg->msg.info.traceId;
char tbuf[512] = {0};
(void)epsetToStr(&pCtx->epSet, tbuf, tListLen(tbuf));
TAOS_UNUSED(epsetToStr(&pCtx->epSet, tbuf, tListLen(tbuf)));
tGDebug("%s retry on next node,use:%s, step: %d,timeout:%" PRId64 "", label, tbuf, pCtx->retryStep,
pCtx->retryNextInterval);
return;
@ -2533,7 +2530,7 @@ static void cliSchedMsgToNextNode(SCliMsg* pMsg, SCliThrd* pThrd) {
arg->param1 = pMsg;
arg->param2 = pThrd;
(void)transDQSched(pThrd->delayQueue, doDelayTask, arg, pCtx->retryNextInterval);
TAOS_UNUSED(transDQSched(pThrd->delayQueue, doDelayTask, arg, pCtx->retryNextInterval));
}
FORCE_INLINE bool cliTryExtractEpSet(STransMsg* pResp, SEpSet* dst) {
@ -2751,7 +2748,7 @@ int cliAppCb(SCliConn* pConn, STransMsg* pResp, SCliMsg* pMsg) {
if (hasEpSet) {
if (rpcDebugFlag & DEBUG_TRACE) {
char tbuf[512] = {0};
(void)epsetToStr(&pCtx->epSet, tbuf, tListLen(tbuf));
TAOS_UNUSED(epsetToStr(&pCtx->epSet, tbuf, tListLen(tbuf)));
tGTrace("%s conn %p extract epset from msg", CONN_GET_INST_LABEL(pConn), pConn);
}
}
@ -2763,7 +2760,7 @@ int cliAppCb(SCliConn* pConn, STransMsg* pResp, SCliMsg* pMsg) {
} else {
memcpy((char*)pCtx->pRsp, (char*)pResp, sizeof(*pResp));
}
(void)tsem_post(pCtx->pSem);
TAOS_UNUSED(tsem_post(pCtx->pSem));
pCtx->pRsp = NULL;
} else {
STransSyncMsg* pSyncMsg = taosAcquireRef(transGetSyncMsgMgt(), pCtx->syncMsgRef);
@ -2773,8 +2770,8 @@ int cliAppCb(SCliConn* pConn, STransMsg* pResp, SCliMsg* pMsg) {
pSyncMsg->hasEpSet = 1;
epsetAssign(&pSyncMsg->epSet, &pCtx->epSet);
}
(void)tsem2_post(pSyncMsg->pSem);
(void)taosReleaseRef(transGetSyncMsgMgt(), pCtx->syncMsgRef);
TAOS_UNUSED(tsem2_post(pSyncMsg->pSem));
TAOS_UNUSED(taosReleaseRef(transGetSyncMsgMgt(), pCtx->syncMsgRef));
} else {
rpcFreeCont(pResp->pCont);
}
@ -2843,7 +2840,7 @@ static FORCE_INLINE SCliThrd* transGetWorkThrdFromHandle(STrans* trans, int64_t
pThrd = exh->pThrd;
taosWUnLockLatch(&exh->latch);
(void)transReleaseExHandle(transGetRefMgt(), handle);
TAOS_UNUSED(transReleaseExHandle(transGetRefMgt(), handle));
return pThrd;
}
@ -2949,7 +2946,7 @@ int32_t transSendRequest(void* shandle, const SEpSet* pEpSet, STransMsg* pReq, S
code = transInitMsg(shandle, pEpSet, pReq, ctx, &pCliMsg);
if (code != 0) {
taosWUnLockLatch(&exh->latch);
(void)transReleaseExHandle(transGetRefMgt(), handle);
TAOS_UNUSED(transReleaseExHandle(transGetRefMgt(), handle));
TAOS_CHECK_GOTO(code, NULL, _exception);
}
@ -2957,13 +2954,13 @@ int32_t transSendRequest(void* shandle, const SEpSet* pEpSet, STransMsg* pReq, S
taosWUnLockLatch(&exh->latch);
tDebug("msg refId: %" PRId64 "", handle);
(void)transReleaseExHandle(transGetRefMgt(), handle);
(void)transReleaseExHandle(transGetInstMgt(), (int64_t)shandle);
TAOS_UNUSED(transReleaseExHandle(transGetRefMgt(), handle));
TAOS_UNUSED(transReleaseExHandle(transGetInstMgt(), (int64_t)shandle));
return 0;
} else {
exh->inited = 1;
taosWUnLockLatch(&exh->latch);
(void)transReleaseExHandle(transGetRefMgt(), handle);
TAOS_UNUSED(transReleaseExHandle(transGetRefMgt(), handle));
}
}
}
@ -2976,16 +2973,16 @@ int32_t transSendRequest(void* shandle, const SEpSet* pEpSet, STransMsg* pReq, S
EPSET_GET_INUSE_IP(pEpSet), EPSET_GET_INUSE_PORT(pEpSet), pReq->info.ahandle);
if ((code = transAsyncSend(pThrd->asyncPool, &(pCliMsg->q))) != 0) {
destroyCmsg(pCliMsg);
(void)transReleaseExHandle(transGetInstMgt(), (int64_t)shandle);
TAOS_UNUSED(transReleaseExHandle(transGetInstMgt(), (int64_t)shandle));
return (code == TSDB_CODE_RPC_ASYNC_MODULE_QUIT ? TSDB_CODE_RPC_MODULE_QUIT : code);
}
(void)transReleaseExHandle(transGetInstMgt(), (int64_t)shandle);
TAOS_UNUSED(transReleaseExHandle(transGetInstMgt(), (int64_t)shandle));
return 0;
_exception:
transFreeMsg(pReq->pCont);
pReq->pCont = NULL;
(void)transReleaseExHandle(transGetInstMgt(), (int64_t)shandle);
TAOS_UNUSED(transReleaseExHandle(transGetInstMgt(), (int64_t)shandle));
return code;
}
int32_t transSendRequestWithId(void* shandle, const SEpSet* pEpSet, STransMsg* pReq, int64_t* transpointId) {
@ -3021,16 +3018,16 @@ int32_t transSendRequestWithId(void* shandle, const SEpSet* pEpSet, STransMsg* p
EPSET_GET_INUSE_IP(pEpSet), EPSET_GET_INUSE_PORT(pEpSet), pReq->info.ahandle);
if ((code = transAsyncSend(pThrd->asyncPool, &(pCliMsg->q))) != 0) {
destroyCmsg(pCliMsg);
(void)transReleaseExHandle(transGetInstMgt(), (int64_t)shandle);
TAOS_UNUSED(transReleaseExHandle(transGetInstMgt(), (int64_t)shandle));
return (code == TSDB_CODE_RPC_ASYNC_MODULE_QUIT ? TSDB_CODE_RPC_MODULE_QUIT : code);
}
(void)transReleaseExHandle(transGetInstMgt(), (int64_t)shandle);
TAOS_UNUSED(transReleaseExHandle(transGetInstMgt(), (int64_t)shandle));
return 0;
_exception:
transFreeMsg(pReq->pCont);
pReq->pCont = NULL;
(void)transReleaseExHandle(transGetInstMgt(), (int64_t)shandle);
TAOS_UNUSED(transReleaseExHandle(transGetInstMgt(), (int64_t)shandle));
return code;
}
@ -3068,7 +3065,7 @@ int32_t transSendRecv(void* shandle, const SEpSet* pEpSet, STransMsg* pReq, STra
STransConnCtx* pCtx = taosMemoryCalloc(1, sizeof(STransConnCtx));
if (pCtx == NULL) {
(void)tsem_destroy(sem);
TAOS_UNUSED(tsem_destroy(sem));
taosMemoryFree(sem);
TAOS_CHECK_GOTO(terrno, NULL, _RETURN1);
}
@ -3082,7 +3079,7 @@ int32_t transSendRecv(void* shandle, const SEpSet* pEpSet, STransMsg* pReq, STra
SCliMsg* cliMsg = taosMemoryCalloc(1, sizeof(SCliMsg));
if (cliMsg == NULL) {
(void)tsem_destroy(sem);
TAOS_UNUSED(tsem_destroy(sem));
taosMemoryFree(sem);
taosMemoryFree(pCtx);
TAOS_CHECK_GOTO(terrno, NULL, _RETURN1);
@ -3103,18 +3100,18 @@ int32_t transSendRecv(void* shandle, const SEpSet* pEpSet, STransMsg* pReq, STra
destroyCmsg(cliMsg);
TAOS_CHECK_GOTO((code == TSDB_CODE_RPC_ASYNC_MODULE_QUIT ? TSDB_CODE_RPC_MODULE_QUIT : code), NULL, _RETURN);
}
(void)tsem_wait(sem);
TAOS_UNUSED(tsem_wait(sem));
memcpy(pRsp, pTransRsp, sizeof(STransMsg));
_RETURN:
tsem_destroy(sem);
taosMemoryFree(sem);
(void)transReleaseExHandle(transGetInstMgt(), (int64_t)shandle);
TAOS_UNUSED(transReleaseExHandle(transGetInstMgt(), (int64_t)shandle));
taosMemoryFree(pTransRsp);
return code;
_RETURN1:
(void)transReleaseExHandle(transGetInstMgt(), (int64_t)shandle);
TAOS_UNUSED(transReleaseExHandle(transGetInstMgt(), (int64_t)shandle));
taosMemoryFree(pTransRsp);
taosMemoryFree(pReq->pCont);
pReq->pCont = NULL;
@ -3151,7 +3148,7 @@ int32_t transCreateSyncMsg(STransMsg* pTransMsg, int64_t* refId) {
return 0;
_EXIT:
(void)tsem2_destroy(sem);
TAOS_UNUSED(tsem2_destroy(sem));
taosMemoryFree(sem);
taosMemoryFree(pSyncMsg);
return code;
@ -3235,15 +3232,15 @@ int32_t transSendRecvWithTimeout(void* shandle, SEpSet* pEpSet, STransMsg* pReq,
}
}
_RETURN:
(void)transReleaseExHandle(transGetInstMgt(), (int64_t)shandle);
(void)taosReleaseRef(transGetSyncMsgMgt(), ref);
(void)taosRemoveRef(transGetSyncMsgMgt(), ref);
TAOS_UNUSED(transReleaseExHandle(transGetInstMgt(), (int64_t)shandle));
TAOS_UNUSED(taosReleaseRef(transGetSyncMsgMgt(), ref));
TAOS_UNUSED(taosRemoveRef(transGetSyncMsgMgt(), ref));
return code;
_RETURN2:
transFreeMsg(pReq->pCont);
pReq->pCont = NULL;
taosMemoryFree(pTransMsg);
(void)transReleaseExHandle(transGetInstMgt(), (int64_t)shandle);
TAOS_UNUSED(transReleaseExHandle(transGetInstMgt(), (int64_t)shandle));
return code;
}
/*
@ -3295,7 +3292,7 @@ int32_t transSetDefaultAddr(void* shandle, const char* ip, const char* fqdn) {
}
}
(void)transReleaseExHandle(transGetInstMgt(), (int64_t)shandle);
TAOS_UNUSED(transReleaseExHandle(transGetInstMgt(), (int64_t)shandle));
return code;
}

24
source/libs/transport/src/transComm.c
View File

@ -272,7 +272,7 @@ int32_t transAsyncPoolCreate(uv_loop_t* loop, int sz, void* arg, AsyncCB cb, SAs
}
item->pThrd = arg;
QUEUE_INIT(&item->qmsg);
(void)taosThreadMutexInit(&item->mtx, NULL);
TAOS_UNUSED(taosThreadMutexInit(&item->mtx, NULL));
async->data = item;
err = uv_async_init(loop, async, cb);
@ -301,7 +301,7 @@ void transAsyncPoolDestroy(SAsyncPool* pool) {
SAsyncItem* item = async->data;
if (item == NULL) continue;
(void)taosThreadMutexDestroy(&item->mtx);
TAOS_UNUSED(taosThreadMutexDestroy(&item->mtx));
taosMemoryFree(item);
}
taosMemoryFree(pool->asyncs);
@ -328,9 +328,9 @@ int transAsyncSend(SAsyncPool* pool, queue* q) {
uv_async_t* async = &(pool->asyncs[idx]);
SAsyncItem* item = async->data;
(void)taosThreadMutexLock(&item->mtx);
TAOS_UNUSED(taosThreadMutexLock(&item->mtx));
QUEUE_PUSH(&item->qmsg, q);
(void)taosThreadMutexUnlock(&item->mtx);
TAOS_UNUSED(taosThreadMutexUnlock(&item->mtx));
int ret = uv_async_send(async);
if (ret != 0) {
tError("failed to send async,reason:%s", uv_err_name(ret));
@ -393,7 +393,7 @@ void* transCtxDumpVal(STransCtx* ctx, int32_t key) {
return NULL;
}
void* ret = NULL;
(void)(*cVal->clone)(cVal->val, &ret);
TAOS_UNUSED((*cVal->clone)(cVal->val, &ret));
return ret;
}
void* transCtxDumpBrokenlinkVal(STransCtx* ctx, int32_t* msgType) {
@ -401,7 +401,7 @@ void* transCtxDumpBrokenlinkVal(STransCtx* ctx, int32_t* msgType) {
if (ctx->brokenVal.clone == NULL) {
return ret;
}
(void)(*ctx->brokenVal.clone)(ctx->brokenVal.val, &ret);
TAOS_UNUSED((*ctx->brokenVal.clone)(ctx->brokenVal.val, &ret));
*msgType = ctx->brokenVal.msgType;
@ -549,7 +549,7 @@ static void transDQTimeout(uv_timer_t* timer) {
}
} while (1);
if (timeout != 0) {
(void)uv_timer_start(queue->timer, transDQTimeout, timeout, 0);
TAOS_UNUSED(uv_timer_start(queue->timer, transDQTimeout, timeout, 0));
}
}
int32_t transDQCreate(uv_loop_t* loop, SDelayQueue** queue) {
@ -614,7 +614,7 @@ void transDQDestroy(SDelayQueue* queue, void (*freeFunc)(void* arg)) {
taosMemoryFree(queue);
}
void transDQCancel(SDelayQueue* queue, SDelayTask* task) {
(void)uv_timer_stop(queue->timer);
TAOS_UNUSED(uv_timer_stop(queue->timer));
if (heapSize(queue->heap) <= 0) {
taosMemoryFree(task->arg);
@ -634,7 +634,7 @@ void transDQCancel(SDelayQueue* queue, SDelayTask* task) {
SDelayTask* task = container_of(minNode, SDelayTask, node);
uint64_t timeout = now > task->execTime ? now - task->execTime : 0;
(void)uv_timer_start(queue->timer, transDQTimeout, timeout, 0);
TAOS_UNUSED(uv_timer_start(queue->timer, transDQTimeout, timeout, 0));
}
}
@ -659,7 +659,7 @@ SDelayTask* transDQSched(SDelayQueue* queue, void (*func)(void* arg), void* arg,
tTrace("timer %p put task into delay queue, timeoutMs:%" PRIu64, queue->timer, timeoutMs);
heapInsert(queue->heap, &task->node);
(void)uv_timer_start(queue->timer, transDQTimeout, timeoutMs, 0);
TAOS_UNUSED(uv_timer_start(queue->timer, transDQTimeout, timeoutMs, 0));
return task;
}
@ -708,7 +708,7 @@ static void transInitEnv() {
svrRefMgt = transOpenRefMgt(50000, transDestroyExHandle);
instMgt = taosOpenRef(50, rpcCloseImpl);
transSyncMsgMgt = taosOpenRef(50, transDestroySyncMsg);
(void)uv_os_setenv("UV_TCP_SINGLE_ACCEPT", "1");
TAOS_UNUSED(uv_os_setenv("UV_TCP_SINGLE_ACCEPT", "1"));
}
static void transDestroyEnv() {
transCloseRefMgt(refMgt);
@ -775,7 +775,7 @@ void transDestroySyncMsg(void* msg) {
if (msg == NULL) return;
STransSyncMsg* pSyncMsg = msg;
(void)tsem2_destroy(pSyncMsg->pSem);
TAOS_UNUSED(tsem2_destroy(pSyncMsg->pSem));
taosMemoryFree(pSyncMsg->pSem);
transFreeMsg(pSyncMsg->pRsp->pCont);
taosMemoryFree(pSyncMsg->pRsp);

85
source/libs/transport/src/transSvr.c
View File

@ -188,11 +188,11 @@ static void sendQuitToWorkThrd(SWorkThrd* pThrd);
static int32_t addHandleToWorkloop(SWorkThrd* pThrd, char* pipeName);
static int32_t addHandleToAcceptloop(void* arg);
#define SRV_RELEASE_UV(loop) \
do { \
(void)uv_walk(loop, uvWalkCb, NULL); \
(void)uv_run(loop, UV_RUN_DEFAULT); \
(void)uv_loop_close(loop); \
#define SRV_RELEASE_UV(loop) \
do { \
TAOS_UNUSED(uv_walk(loop, uvWalkCb, NULL)); \
TAOS_UNUSED(uv_run(loop, UV_RUN_DEFAULT)); \
TAOS_UNUSED(uv_loop_close(loop)); \
} while (0);
#define ASYNC_ERR_JRET(thrd) \
@ -594,7 +594,7 @@ void uvOnSendCb(uv_write_t* req, int status) {
(pTransInst->cfp)(pTransInst->parent, &(conn->regArg.msg), NULL);
memset(&conn->regArg, 0, sizeof(conn->regArg));
}
(void)transQueuePop(&conn->srvMsgs);
TAOS_UNUSED(transQueuePop(&conn->srvMsgs));
taosMemoryFree(msg);
msg = (SSvrMsg*)transQueueGet(&conn->srvMsgs, 0);
@ -651,7 +651,7 @@ static int uvPrepareSendData(SSvrMsg* smsg, uv_buf_t* wb) {
// handle invalid drop_task resp, TD-20098
if (pConn->inType == TDMT_SCH_DROP_TASK && pMsg->code == TSDB_CODE_VND_INVALID_VGROUP_ID) {
(void)transQueuePop(&pConn->srvMsgs);
TAOS_UNUSED(transQueuePop(&pConn->srvMsgs));
destroySmsg(smsg);
return TSDB_CODE_INVALID_MSG;
}
@ -707,7 +707,7 @@ static FORCE_INLINE void uvStartSendRespImpl(SSvrMsg* smsg) {
transRefSrvHandle(pConn);
uv_write_t* req = transReqQueuePush(&pConn->wreqQueue);
(void)uv_write(req, (uv_stream_t*)pConn->pTcp, &wb, 1, uvOnSendCb);
TAOS_UNUSED(uv_write(req, (uv_stream_t*)pConn->pTcp, &wb, 1, uvOnSendCb));
}
static void uvStartSendResp(SSvrMsg* smsg) {
// impl
@ -759,9 +759,9 @@ void uvWorkerAsyncCb(uv_async_t* handle) {
queue wq;
// batch process to avoid to lock/unlock frequently
(void)taosThreadMutexLock(&item->mtx);
TAOS_UNUSED(taosThreadMutexLock(&item->mtx));
QUEUE_MOVE(&item->qmsg, &wq);
(void)taosThreadMutexUnlock(&item->mtx);
TAOS_UNUSED(taosThreadMutexUnlock(&item->mtx));
while (!QUEUE_IS_EMPTY(&wq)) {
queue* head = QUEUE_HEAD(&wq);
@ -784,12 +784,12 @@ void uvWorkerAsyncCb(uv_async_t* handle) {
SExHandle* exh2 = transAcquireExHandle(transGetSvrRefMgt(), refId);
if (exh2 == NULL || exh1 != exh2) {
tTrace("handle except msg %p, ignore it", exh1);
(void)transReleaseExHandle(transGetSvrRefMgt(), refId);
TAOS_UNUSED(transReleaseExHandle(transGetSvrRefMgt(), refId));
destroySmsg(msg);
continue;
}
msg->pConn = exh1->handle;
(void)transReleaseExHandle(transGetSvrRefMgt(), refId);
TAOS_UNUSED(transReleaseExHandle(transGetSvrRefMgt(), refId));
(*transAsyncHandle[msg->type])(msg, pThrd);
}
}
@ -827,7 +827,7 @@ static bool uvRecvReleaseReq(SSvrConn* pConn, STransMsgHead* pHead) {
tTrace("conn %p received release request", pConn);
STraceId traceId = pHead->traceId;
(void)transClearBuffer(&pConn->readBuf);
TAOS_UNUSED(transClearBuffer(&pConn->readBuf));
transFreeMsg(transContFromHead((char*)pHead));
if (pConn->status != ConnAcquire) {
return true;
@ -906,7 +906,8 @@ void uvOnAcceptCb(uv_stream_t* stream, int status) {
tTrace("new connection accepted by main server, dispatch to %dth worker-thread", pObj->workerIdx);
(void)uv_write2(wr, (uv_stream_t*)&(pObj->pipe[pObj->workerIdx][0]), &buf, 1, (uv_stream_t*)cli, uvOnPipeWriteCb);
TAOS_UNUSED(
uv_write2(wr, (uv_stream_t*)&(pObj->pipe[pObj->workerIdx][0]), &buf, 1, (uv_stream_t*)cli, uvOnPipeWriteCb));
} else {
if (!uv_is_closing((uv_handle_t*)cli)) {
tError("failed to accept tcp: %s", uv_err_name(err));
@ -971,7 +972,7 @@ 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;
(void)uv_fileno((const uv_handle_t*)pConn->pTcp, &fd);
TAOS_UNUSED(uv_fileno((const uv_handle_t*)pConn->pTcp, &fd));
tTrace("conn %p created, fd:%d", pConn, fd);
struct sockaddr peername, sockname;
@ -981,7 +982,7 @@ void uvOnConnectionCb(uv_stream_t* q, ssize_t nread, const uv_buf_t* buf) {
transUnrefSrvHandle(pConn);
return;
}
(void)transSockInfo2Str(&peername, pConn->dst);
TAOS_UNUSED(transSockInfo2Str(&peername, pConn->dst));
addrlen = sizeof(sockname);
if (0 != uv_tcp_getsockname(pConn->pTcp, (struct sockaddr*)&sockname, &addrlen)) {
@ -989,7 +990,7 @@ void uvOnConnectionCb(uv_stream_t* q, ssize_t nread, const uv_buf_t* buf) {
transUnrefSrvHandle(pConn);
return;
}
(void)transSockInfo2Str(&sockname, pConn->src);
TAOS_UNUSED(transSockInfo2Str(&sockname, pConn->src));
struct sockaddr_in addr = *(struct sockaddr_in*)&peername;
struct sockaddr_in saddr = *(struct sockaddr_in*)&sockname;
@ -998,7 +999,7 @@ void uvOnConnectionCb(uv_stream_t* q, ssize_t nread, const uv_buf_t* buf) {
pConn->serverIp = saddr.sin_addr.s_addr;
pConn->port = ntohs(addr.sin_port);
(void)uv_read_start((uv_stream_t*)(pConn->pTcp), uvAllocRecvBufferCb, uvOnRecvCb);
TAOS_UNUSED(uv_read_start((uv_stream_t*)(pConn->pTcp), uvAllocRecvBufferCb, uvOnRecvCb));
} else {
tDebug("failed to create new connection");
@ -1010,7 +1011,7 @@ void* transAcceptThread(void* arg) {
// opt
setThreadName("trans-accept");
SServerObj* srv = (SServerObj*)arg;
(void)uv_run(srv->loop, UV_RUN_DEFAULT);
TAOS_UNUSED(uv_run(srv->loop, UV_RUN_DEFAULT));
return NULL;
}
@ -1021,7 +1022,7 @@ void uvOnPipeConnectionCb(uv_connect_t* connect, int status) {
};
SWorkThrd* pThrd = container_of(connect, SWorkThrd, connect_req);
(void)uv_read_start((uv_stream_t*)pThrd->pipe, uvAllocConnBufferCb, uvOnConnectionCb);
TAOS_UNUSED(uv_read_start((uv_stream_t*)pThrd->pipe, uvAllocConnBufferCb, uvOnConnectionCb));
}
static int32_t addHandleToWorkloop(SWorkThrd* pThrd, char* pipeName) {
int32_t code = 0;
@ -1124,7 +1125,7 @@ static int32_t addHandleToAcceptloop(void* arg) {
void* transWorkerThread(void* arg) {
setThreadName("trans-svr-work");
SWorkThrd* pThrd = (SWorkThrd*)arg;
(void)uv_run(pThrd->loop, UV_RUN_DEFAULT);
TAOS_UNUSED(uv_run(pThrd->loop, UV_RUN_DEFAULT));
return NULL;
}
@ -1201,7 +1202,7 @@ static FORCE_INLINE SSvrConn* createConn(void* hThrd) {
_end:
if (pConn) {
transQueueDestroy(&pConn->srvMsgs);
(void)transDestroyBuffer(&pConn->readBuf);
TAOS_UNUSED(transDestroyBuffer(&pConn->readBuf));
taosMemoryFree(pConn->pTcp);
taosMemoryFree(pConn);
pConn = NULL;
@ -1230,8 +1231,8 @@ static FORCE_INLINE void destroyConnRegArg(SSvrConn* conn) {
}
static int32_t reallocConnRef(SSvrConn* conn) {
if (conn->refId > 0) {
(void)transReleaseExHandle(transGetSvrRefMgt(), conn->refId);
(void)transRemoveExHandle(transGetSvrRefMgt(), conn->refId);
TAOS_UNUSED(transReleaseExHandle(transGetSvrRefMgt(), conn->refId));
TAOS_UNUSED(transRemoveExHandle(transGetSvrRefMgt(), conn->refId));
}
// avoid app continue to send msg on invalid handle
SExHandle* exh = taosMemoryMalloc(sizeof(SExHandle));
@ -1267,8 +1268,8 @@ static void uvDestroyConn(uv_handle_t* handle) {
}
SWorkThrd* thrd = conn->hostThrd;
(void)transReleaseExHandle(transGetSvrRefMgt(), conn->refId);
(void)transRemoveExHandle(transGetSvrRefMgt(), conn->refId);
TAOS_UNUSED(transReleaseExHandle(transGetSvrRefMgt(), conn->refId));
TAOS_UNUSED(transRemoveExHandle(transGetSvrRefMgt(), conn->refId));
STrans* pTransInst = thrd->pTransInst;
tDebug("%s conn %p destroy", transLabel(pTransInst), conn);
@ -1283,7 +1284,7 @@ static void uvDestroyConn(uv_handle_t* handle) {
QUEUE_REMOVE(&conn->queue);
taosMemoryFree(conn->pTcp);
destroyConnRegArg(conn);
(void)transDestroyBuffer(&conn->readBuf);
TAOS_UNUSED(transDestroyBuffer(&conn->readBuf));
taosMemoryFree(conn);
if (thrd->quit && QUEUE_IS_EMPTY(&thrd->conn)) {
@ -1563,7 +1564,7 @@ void uvHandleRegister(SSvrMsg* msg, SWorkThrd* thrd) {
if (!transQueuePush(&conn->srvMsgs, msg)) {
return;
}
(void)transQueuePop(&conn->srvMsgs);
TAOS_UNUSED(transQueuePop(&conn->srvMsgs));
if (conn->regArg.init) {
transFreeMsg(conn->regArg.msg.pCont);
@ -1636,7 +1637,7 @@ void destroyWorkThrd(SWorkThrd* pThrd) {
}
if (pThrd->inited) {
sendQuitToWorkThrd(pThrd);
(void)taosThreadJoin(pThrd->thread, NULL);
TAOS_UNUSED(taosThreadJoin(pThrd->thread, NULL));
SRV_RELEASE_UV(pThrd->loop);
TRANS_DESTROY_ASYNC_POOL_MSG(pThrd->asyncPool, SSvrMsg, destroySmsgWrapper, NULL);
}
@ -1646,7 +1647,7 @@ void sendQuitToWorkThrd(SWorkThrd* pThrd) {
SSvrMsg* msg = taosMemoryCalloc(1, sizeof(SSvrMsg));
msg->type = Quit;
tDebug("server send quit msg to work thread");
(void)transAsyncSend(pThrd->asyncPool, &msg->q);
TAOS_UNUSED(transAsyncSend(pThrd->asyncPool, &msg->q));
}
void transCloseServer(void* arg) {
@ -1655,8 +1656,8 @@ void transCloseServer(void* arg) {
if (srv->inited) {
tDebug("send quit msg to accept thread");
(void)uv_async_send(srv->pAcceptAsync);
(void)taosThreadJoin(srv->thread, NULL);
TAOS_UNUSED(uv_async_send(srv->pAcceptAsync));
TAOS_UNUSED(taosThreadJoin(srv->thread, NULL));
SRV_RELEASE_UV(srv->loop);
for (int i = 0; i < srv->numOfThreads; i++) {
@ -1724,15 +1725,15 @@ int32_t transReleaseSrvHandle(void* handle) {
tDebug("%s conn %p start to release", transLabel(pThrd->pTransInst), exh->handle);
if ((code = transAsyncSend(pThrd->asyncPool, &m->q)) != 0) {
destroySmsg(m);
(void)transReleaseExHandle(transGetSvrRefMgt(), refId);
TAOS_UNUSED(transReleaseExHandle(transGetSvrRefMgt(), refId));
return code;
}
(void)transReleaseExHandle(transGetSvrRefMgt(), refId);
TAOS_UNUSED(transReleaseExHandle(transGetSvrRefMgt(), refId));
return 0;
_return1:
tDebug("handle %p failed to send to release handle", exh);
(void)transReleaseExHandle(transGetSvrRefMgt(), refId);
TAOS_UNUSED(transReleaseExHandle(transGetSvrRefMgt(), refId));
return code;
_return2:
tDebug("handle %p failed to send to release handle", exh);
@ -1775,17 +1776,17 @@ int32_t transSendResponse(const STransMsg* msg) {
tGDebug("conn %p start to send resp (1/2)", exh->handle);
if ((code = transAsyncSend(pThrd->asyncPool, &m->q)) != 0) {
destroySmsg(m);
(void)transReleaseExHandle(transGetSvrRefMgt(), refId);
TAOS_UNUSED(transReleaseExHandle(transGetSvrRefMgt(), refId));
return code;
}
(void)transReleaseExHandle(transGetSvrRefMgt(), refId);
TAOS_UNUSED(transReleaseExHandle(transGetSvrRefMgt(), refId));
return 0;
_return1:
tDebug("handle %p failed to send resp", exh);
rpcFreeCont(msg->pCont);
(void)transReleaseExHandle(transGetSvrRefMgt(), refId);
TAOS_UNUSED(transReleaseExHandle(transGetSvrRefMgt(), refId));
return code;
_return2:
tDebug("handle %p failed to send resp", exh);
@ -1820,17 +1821,17 @@ int32_t transRegisterMsg(const STransMsg* msg) {
tDebug("%s conn %p start to register brokenlink callback", transLabel(pTransInst), exh->handle);
if ((code = transAsyncSend(pThrd->asyncPool, &m->q)) != 0) {
destroySmsg(m);
(void)transReleaseExHandle(transGetSvrRefMgt(), refId);
TAOS_UNUSED(transReleaseExHandle(transGetSvrRefMgt(), refId));
return code;
}
(void)transReleaseExHandle(transGetSvrRefMgt(), refId);
TAOS_UNUSED(transReleaseExHandle(transGetSvrRefMgt(), refId));
return 0;
_return1:
tDebug("handle %p failed to register brokenlink", exh);
rpcFreeCont(msg->pCont);
(void)transReleaseExHandle(transGetSvrRefMgt(), refId);
TAOS_UNUSED(transReleaseExHandle(transGetSvrRefMgt(), refId));
return code;
_return2:
tDebug("handle %p failed to register brokenlink", exh);
@ -1875,7 +1876,7 @@ int32_t transSetIpWhiteList(void* thandle, void* arg, FilteFunc* func) {
break;
}
}
(void)transReleaseExHandle(transGetInstMgt(), (int64_t)thandle);
TAOS_UNUSED(transReleaseExHandle(transGetInstMgt(), (int64_t)thandle));
if (code != 0) {
tError("ip-white-list update failed since %s", tstrerror(code));