diff --git a/source/libs/transport/inc/transComm.h b/source/libs/transport/inc/transComm.h index f29ff3769b099be21365f221d977ad682471248f..5bb6349d9b88b515656ccc42d1da1a74787df649 100644 --- a/source/libs/transport/inc/transComm.h +++ b/source/libs/transport/inc/transComm.h @@ -317,6 +317,11 @@ typedef struct STransReq { void* data; } STransReq; +void transReqQueueInit(queue* q); +void* transReqQueuePushReq(queue* q); +void* transReqQueueRemove(void* arg); +void transReqQueueClear(queue* q); + // queue sending msgs typedef struct { SArray* q; diff --git a/source/libs/transport/src/transCli.c b/source/libs/transport/src/transCli.c index f5110f247181bc57b982bf9de37167078d1a8dd1..5d087d57693bf13e51f2ece38dd1abf0b79b9c97 100644 --- a/source/libs/transport/src/transCli.c +++ b/source/libs/transport/src/transCli.c @@ -19,7 +19,7 @@ typedef struct SCliConn { T_REF_DECLARE() uv_connect_t connReq; uv_stream_t* stream; - uv_write_t writeReq; + queue wreqQueue; void* hostThrd; @@ -586,9 +586,10 @@ static SCliConn* cliCreateConn(SCliThrd* pThrd) { uv_tcp_init(pThrd->loop, (uv_tcp_t*)(conn->stream)); conn->stream->data = conn; - conn->writeReq.data = conn; conn->connReq.data = conn; + transReqQueueInit(&conn->wreqQueue); + transQueueInit(&conn->cliMsgs, NULL); QUEUE_INIT(&conn->conn); conn->hostThrd = pThrd; @@ -627,6 +628,8 @@ static void cliDestroy(uv_handle_t* handle) { transCtxCleanup(&conn->ctx); transQueueDestroy(&conn->cliMsgs); tTrace("%s conn %p destroy successfully", CONN_GET_INST_LABEL(conn), conn); + transReqQueueClear(&conn->wreqQueue); + transDestroyBuffer(&conn->readBuf); taosMemoryFree(conn); } @@ -649,11 +652,8 @@ static bool cliHandleNoResp(SCliConn* conn) { return res; } static void cliSendCb(uv_write_t* req, int status) { - SCliConn* pConn = req && req->handle ? req->handle->data : NULL; - taosMemoryFree(req); - if (pConn == NULL) { - return; - } + SCliConn* pConn = transReqQueueRemove(req); + if (pConn == NULL) return; if (status == 0) { tTrace("%s conn %p data already was written out", CONN_GET_INST_LABEL(pConn), pConn); @@ -711,7 +711,7 @@ void cliSend(SCliConn* pConn) { CONN_SET_PERSIST_BY_APP(pConn); } - uv_write_t* req = taosMemoryCalloc(1, sizeof(uv_write_t)); + uv_write_t* req = transReqQueuePushReq(&pConn->wreqQueue); uv_write(req, (uv_stream_t*)pConn->stream, &wb, 1, cliSendCb); return; _RETURN: diff --git a/source/libs/transport/src/transComm.c b/source/libs/transport/src/transComm.c index 812123441c274d7ff11af70ca0a06b236fd27124..3ab15ad804c943482648e43398e5fe2ea224067e 100644 --- a/source/libs/transport/src/transComm.c +++ b/source/libs/transport/src/transComm.c @@ -293,6 +293,48 @@ void* transCtxDumpBrokenlinkVal(STransCtx* ctx, int32_t* msgType) { return ret; } +void transReqQueueInit(queue* q) { + // init req queue + QUEUE_INIT(q); +} +void* transReqQueuePushReq(queue* q) { + uv_write_t* req = taosMemoryCalloc(1, sizeof(uv_write_t)); + STransReq* wreq = taosMemoryCalloc(1, sizeof(STransReq)); + wreq->data = req; + req->data = wreq; + QUEUE_PUSH(q, &wreq->q); + return req; +} +void* transReqQueueRemove(void* arg) { + void* ret = NULL; + uv_write_t* req = arg; + STransReq* wreq = req && req->data ? req->data : NULL; + + assert(wreq->data == req); + if (wreq == NULL || wreq->data == NULL) { + taosMemoryFree(wreq->data); + taosMemoryFree(wreq); + return req; + } + + QUEUE_REMOVE(&wreq->q); + + ret = req && req->handle ? req->handle->data : NULL; + taosMemoryFree(wreq->data); + taosMemoryFree(wreq); + + return ret; +} +void transReqQueueClear(queue* q) { + while (!QUEUE_IS_EMPTY(q)) { + queue* h = QUEUE_HEAD(q); + QUEUE_REMOVE(h); + STransReq* req = QUEUE_DATA(h, STransReq, q); + taosMemoryFree(req->data); + taosMemoryFree(req); + } +} + void transQueueInit(STransQueue* queue, void (*freeFunc)(const void* arg)) { queue->q = taosArrayInit(2, sizeof(void*)); queue->freeFunc = (void (*)(const void*))freeFunc; diff --git a/source/libs/transport/src/transSvr.c b/source/libs/transport/src/transSvr.c index fb11465d4a081c4ae7ac322a913c28eb557a3515..fbf6c0df7644b3669570cb1c02832ff4b7bd316b 100644 --- a/source/libs/transport/src/transSvr.c +++ b/source/libs/transport/src/transSvr.c @@ -331,14 +331,7 @@ void uvOnTimeoutCb(uv_timer_t* handle) { } void uvOnSendCb(uv_write_t* req, int status) { - STransReq* wreq = req && req->data ? req->data : NULL; - SSvrConn* conn = req && req->handle ? req->handle->data : NULL; - if (wreq != NULL && conn != NULL) { - QUEUE_REMOVE(&wreq->q); - taosMemoryFree(wreq->data); - taosMemoryFree(wreq); - } - + SSvrConn* conn = transReqQueueRemove(req); if (conn == NULL) return; if (status == 0) { @@ -442,12 +435,7 @@ static void uvStartSendRespInternal(SSvrMsg* smsg) { transRefSrvHandle(pConn); - uv_write_t* req = taosMemoryCalloc(1, sizeof(uv_write_t)); - STransReq* wreq = taosMemoryCalloc(1, sizeof(STransReq)); - wreq->data = req; - req->data = wreq; - QUEUE_PUSH(&pConn->wreqQueue, &wreq->q); - + uv_write_t* req = transReqQueuePushReq(&pConn->wreqQueue); uv_write(req, (uv_stream_t*)pConn->pTcp, &wb, 1, uvOnSendCb); } static void uvStartSendResp(SSvrMsg* smsg) { @@ -757,7 +745,7 @@ static SSvrConn* createConn(void* hThrd) { SSvrConn* pConn = (SSvrConn*)taosMemoryCalloc(1, sizeof(SSvrConn)); - QUEUE_INIT(&pConn->wreqQueue); + transReqQueueInit(&pConn->wreqQueue); QUEUE_INIT(&pConn->queue); QUEUE_PUSH(&pThrd->conn, &pConn->queue); @@ -834,13 +822,7 @@ static void uvDestroyConn(uv_handle_t* handle) { destroySmsg(msg); } - while (!QUEUE_IS_EMPTY(&conn->wreqQueue)) { - queue* h = QUEUE_HEAD(&conn->wreqQueue); - QUEUE_REMOVE(h); - STransReq* req = QUEUE_DATA(h, STransReq, q); - taosMemoryFree(req->data); - taosMemoryFree(req); - } + transReqQueueClear(&conn->wreqQueue); transQueueDestroy(&conn->srvMsgs); QUEUE_REMOVE(&conn->queue);