From 4f330fab1fe48f8eb925a2f57e6b9f7d71de6422 Mon Sep 17 00:00:00 2001 From: yihaoDeng Date: Tue, 15 Mar 2022 23:47:37 +0800 Subject: [PATCH] handle except --- include/libs/transport/trpc.h | 2 +- source/libs/transport/inc/transComm.h | 3 ++ source/libs/transport/src/trans.c | 13 ++++-- source/libs/transport/src/transCli.c | 63 +++++++++++++++++++++++---- source/libs/transport/src/transSrv.c | 7 ++- source/libs/transport/test/transUT.cc | 53 +++++++++++++++++++++- 6 files changed, 126 insertions(+), 15 deletions(-) diff --git a/include/libs/transport/trpc.h b/include/libs/transport/trpc.h index 231bc4af45..b5b8d6ab66 100644 --- a/include/libs/transport/trpc.h +++ b/include/libs/transport/trpc.h @@ -94,7 +94,7 @@ int rpcReportProgress(void *pConn, char *pCont, int contLen); void rpcCancelRequest(int64_t rid); // just release client conn to rpc instance, no close sock -void rpcReleaseHandle(void *handle); +void rpcReleaseHandle(void *handle, int8_t type); void rpcRefHandle(void *handle, int8_t type); void rpcUnrefHandle(void *handle, int8_t type); diff --git a/source/libs/transport/inc/transComm.h b/source/libs/transport/inc/transComm.h index 76347cdba2..8ea65b193d 100644 --- a/source/libs/transport/inc/transComm.h +++ b/source/libs/transport/inc/transComm.h @@ -252,6 +252,9 @@ void transUnrefSrvHandle(void* handle); void transRefCliHandle(void* handle); void transUnrefCliHandle(void* handle); +void transReleaseCliHandle(void* handle); +void transReleaseSrvHandle(void* handle); + void transSendRequest(void* shandle, const char* ip, uint32_t port, STransMsg* pMsg); void transSendRecv(void* shandle, const char* ip, uint32_t port, STransMsg* pMsg, STransMsg* pRsp); void transSendResponse(const STransMsg* pMsg); diff --git a/source/libs/transport/src/trans.c b/source/libs/transport/src/trans.c index 213782c5bd..2cab03f133 100644 --- a/source/libs/transport/src/trans.c +++ b/source/libs/transport/src/trans.c @@ -22,6 +22,11 @@ void* (*taosInitHandle[])(uint32_t ip, uint32_t port, char* label, int numOfThre void (*taosCloseHandle[])(void* arg) = {transCloseServer, transCloseClient}; +void (*taosRefHandle[])(void* handle) = {transRefSrvHandle, transRefCliHandle}; +void (*taosUnRefHandle[])(void* handle) = {transUnrefSrvHandle, transUnrefCliHandle}; + +void (*transReleaseHandle[])(void* handle) = {transReleaseSrvHandle, transReleaseCliHandle}; + void* rpcOpen(const SRpcInit* pInit) { SRpcInfo* pRpc = calloc(1, sizeof(SRpcInfo)); if (pRpc == NULL) { @@ -127,9 +132,6 @@ void rpcSendRecv(void* shandle, SEpSet* pEpSet, SRpcMsg* pMsg, SRpcMsg* pRsp) { void rpcSendResponse(const SRpcMsg* pMsg) { transSendResponse(pMsg); } int rpcGetConnInfo(void* thandle, SRpcConnInfo* pInfo) { return transGetConnInfo((void*)thandle, pInfo); } -void (*taosRefHandle[])(void* handle) = {transRefSrvHandle, transRefCliHandle}; -void (*taosUnRefHandle[])(void* handle) = {transUnrefSrvHandle, transUnrefCliHandle}; - void rpcRefHandle(void* handle, int8_t type) { assert(type == TAOS_CONN_SERVER || type == TAOS_CONN_CLIENT); (*taosRefHandle[type])(handle); @@ -140,6 +142,11 @@ void rpcUnrefHandle(void* handle, int8_t type) { (*taosUnRefHandle[type])(handle); } +void rpcReleaseHandle(void* handle, int8_t type) { + assert(type == TAOS_CONN_SERVER || type == TAOS_CONN_CLIENT); + (*transReleaseHandle[type])(handle); +} + int32_t rpcInit() { // impl later return 0; diff --git a/source/libs/transport/src/transCli.c b/source/libs/transport/src/transCli.c index 5c50ca2bc3..931f58097e 100644 --- a/source/libs/transport/src/transCli.c +++ b/source/libs/transport/src/transCli.c @@ -17,6 +17,11 @@ #include "transComm.h" +// Normal(default): send/recv msg +// Quit: quit rpc inst +// Release: release handle to rpc inst +typedef enum { Normal, Quit, Release } SCliMsgType; + typedef struct SCliConn { T_REF_DECLARE() uv_connect_t connReq; @@ -49,6 +54,7 @@ typedef struct SCliMsg { STransMsg msg; queue q; uint64_t st; + SCliMsgType type; } SCliMsg; typedef struct SCliThrdObj { @@ -108,6 +114,8 @@ static void cliHandleExcept(SCliConn* conn); // handle req from app static void cliHandleReq(SCliMsg* pMsg, SCliThrdObj* pThrd); static void cliHandleQuit(SCliMsg* pMsg, SCliThrdObj* pThrd); +static void cliHandleRelease(SCliMsg* pMsg, SCliThrdObj* pThrd); + static void cliSendQuit(SCliThrdObj* thrd); static void destroyUserdata(STransMsg* userdata); @@ -121,8 +129,8 @@ static void destroyThrdObj(SCliThrdObj* pThrd); #define CONN_HOST_THREAD_INDEX(conn) (conn ? ((SCliConn*)conn)->hThrdIdx : -1) #define CONN_PERSIST_TIME(para) (para * 1000 * 10) - -#define CONN_GET_INST_LABEL(conn) (((STrans*)(((SCliThrdObj*)conn->hostThrd)->pTransInst))->label) +#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_HANDLE_THREAD_QUIT(conn, thrd) \ do { \ if (thrd->quit) { \ @@ -344,6 +352,8 @@ static void addConnToPool(void* pool, SCliConn* conn) { static void cliAllocRecvBufferCb(uv_handle_t* handle, size_t suggested_size, uv_buf_t* buf) { SCliConn* conn = handle->data; SConnBuffer* pBuf = &conn->readBuf; + // avoid conn + QUEUE_REMOVE(&conn->conn); transAllocBuffer(pBuf, buf); } static void cliRecvCb(uv_stream_t* handle, ssize_t nread, const uv_buf_t* buf) { @@ -506,6 +516,21 @@ static void cliHandleQuit(SCliMsg* pMsg, SCliThrdObj* pThrd) { pThrd->quit = true; uv_stop(pThrd->loop); } +static void cliHandleRelease(SCliMsg* pMsg, SCliThrdObj* pThrd) { + SCliConn* conn = pMsg->msg.handle; + tDebug("%s cli conn %p release to inst", CONN_GET_INST_LABEL(conn), conn); + + destroyCmsg(pMsg); + conn->data = NULL; + + transDestroyBuffer(&conn->readBuf); + if (conn->persist && T_REF_VAL_GET(conn) >= 2) { + transUnrefCliHandle(conn); + addConnToPool(pThrd->pool, conn); + } else { + transUnrefCliHandle(conn); + } +} SCliConn* cliGetConn(SCliMsg* pMsg, SCliThrdObj* pThrd) { SCliConn* conn = NULL; @@ -517,7 +542,9 @@ SCliConn* cliGetConn(SCliMsg* pMsg, SCliThrdObj* pThrd) { } else { STransConnCtx* pCtx = pMsg->ctx; conn = getConnFromPool(pThrd->pool, pCtx->ip, pCtx->port); - if (conn != NULL) tTrace("%s cli conn %p get from conn pool", CONN_GET_INST_LABEL(conn), conn); + if (conn != NULL) { + tTrace("%s cli conn %p get from conn pool", CONN_GET_INST_LABEL(conn), conn); + } } return conn; } @@ -572,10 +599,13 @@ static void cliAsyncCb(uv_async_t* handle) { QUEUE_REMOVE(h); SCliMsg* pMsg = QUEUE_DATA(h, SCliMsg, q); - if (pMsg->ctx == NULL) { - cliHandleQuit(pMsg, pThrd); - } else { + + if (pMsg->type == Normal) { cliHandleReq(pMsg, pThrd); + } else if (pMsg->type == Quit) { + cliHandleQuit(pMsg, pThrd); + } else if (pMsg->type == Release) { + cliHandleRelease(pMsg, pThrd); } count++; } @@ -671,8 +701,10 @@ static void transDestroyConnCtx(STransConnCtx* ctx) { void cliSendQuit(SCliThrdObj* thrd) { // cli can stop gracefully SCliMsg* msg = calloc(1, sizeof(SCliMsg)); + msg->type = Quit; transSendAsync(thrd->asyncPool, &msg->q); } + int cliRBChoseIdx(STrans* pTransInst) { int64_t index = pTransInst->index; if (pTransInst->index++ >= pTransInst->numOfThreads) { @@ -702,10 +734,25 @@ 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); if (ref == 0) { cliDestroyConn((SCliConn*)handle, true); } } +void transReleaseCliHandle(void* handle) { + SCliThrdObj* thrd = CONN_GET_HOST_THREAD(handle); + if (thrd == NULL) { + return; + } + + STransMsg tmsg = {.handle = handle}; + + SCliMsg* cmsg = calloc(1, sizeof(SCliMsg)); + cmsg->type = Release; + cmsg->msg = tmsg; + + transSendAsync(thrd->asyncPool, &cmsg->q); +} void transSendRequest(void* shandle, const char* ip, uint32_t port, STransMsg* pMsg) { STrans* pTransInst = (STrans*)shandle; @@ -728,7 +775,7 @@ void transSendRequest(void* shandle, const char* ip, uint32_t port, STransMsg* p assert(pTransInst->connType == TAOS_CONN_CLIENT); // atomic or not - SCliMsg* cliMsg = malloc(sizeof(SCliMsg)); + SCliMsg* cliMsg = calloc(1, sizeof(SCliMsg)); cliMsg->ctx = pCtx; cliMsg->msg = *pMsg; cliMsg->st = taosGetTimestampUs(); @@ -753,7 +800,7 @@ void transSendRecv(void* shandle, const char* ip, uint32_t port, STransMsg* pReq pCtx->pRsp = pRsp; tsem_init(pCtx->pSem, 0, 0); - SCliMsg* cliMsg = malloc(sizeof(SCliMsg)); + SCliMsg* cliMsg = calloc(1, sizeof(SCliMsg)); cliMsg->ctx = pCtx; cliMsg->msg = *pReq; cliMsg->st = taosGetTimestampUs(); diff --git a/source/libs/transport/src/transSrv.c b/source/libs/transport/src/transSrv.c index a3cc6b6181..960e064b8f 100644 --- a/source/libs/transport/src/transSrv.c +++ b/source/libs/transport/src/transSrv.c @@ -641,7 +641,7 @@ static void uvDestroyConn(uv_handle_t* handle) { uv_timer_stop(&conn->pTimer); QUEUE_REMOVE(&conn->queue); free(conn->pTcp); - free(conn); + // free(conn); if (thrd->quit && QUEUE_IS_EMPTY(&thrd->conn)) { uv_loop_close(thrd->loop); @@ -786,6 +786,11 @@ void transUnrefSrvHandle(void* handle) { } // unref srv handle } + +void transReleaseSrvHandle(void* handle) { + // do nothing currently + // +} void transSendResponse(const STransMsg* pMsg) { if (pMsg->handle == NULL) { return; diff --git a/source/libs/transport/test/transUT.cc b/source/libs/transport/test/transUT.cc index b0fae2f8b4..46f6424b0b 100644 --- a/source/libs/transport/test/transUT.cc +++ b/source/libs/transport/test/transUT.cc @@ -110,6 +110,14 @@ class Client { SemWait(); *resp = this->resp; } + void SendAndRecvNoHandle(SRpcMsg *req, SRpcMsg *resp) { + if (req->handle != NULL) { + rpcReleaseHandle(req->handle, TAOS_CONN_CLIENT); + req->handle = NULL; + } + SendAndRecv(req, resp); + } + void SendWithHandle(SRpcMsg *req, SRpcMsg *resp) {} void SemWait() { tsem_wait(&this->sem); } void SemPost() { tsem_post(&this->sem); } @@ -268,6 +276,7 @@ class TransObj { } void RestartSrv() { srv->Restart(); } void cliSendAndRecv(SRpcMsg *req, SRpcMsg *resp) { cli->SendAndRecv(req, resp); } + void cliSendAndRecvNoHandle(SRpcMsg *req, SRpcMsg *resp) { cli->SendAndRecvNoHandle(req, resp); } ~TransObj() { delete cli; @@ -352,7 +361,47 @@ TEST_F(TransEnv, cliPersistHandle) { EXPECT_TRUE(resp.code != 0); } } + ////////////////// +} +TEST_F(TransEnv, cliReleaseHandle) { + tr->SetCliPersistFp(cliPersistHandle); + + SRpcMsg resp = {0}; + for (int i = 0; i < 10; i++) { + SRpcMsg req = {.handle = resp.handle}; + req.msgType = 1; + req.pCont = rpcMallocCont(10); + req.contLen = 10; + tr->cliSendAndRecvNoHandle(&req, &resp); + // if (i == 5) { + // std::cout << "stop server" << std::endl; + // tr->StopSrv(); + //} + // if (i >= 6) { + EXPECT_TRUE(resp.code == 0); + //} + } + ////////////////// +} +TEST_F(TransEnv, cliReleaseHandleExcept) { + tr->SetCliPersistFp(cliPersistHandle); + + SRpcMsg resp = {0}; + for (int i = 0; i < 10; i++) { + SRpcMsg req = {.handle = resp.handle}; + req.msgType = 1; + req.pCont = rpcMallocCont(10); + req.contLen = 10; + tr->cliSendAndRecvNoHandle(&req, &resp); + if (i == 5) { + std::cout << "stop server" << std::endl; + tr->StopSrv(); + } + if (i >= 6) { + EXPECT_TRUE(resp.code != 0); + } + } ////////////////// } TEST_F(TransEnv, srvContinueSend) { @@ -367,11 +416,11 @@ TEST_F(TransEnv, srvContinueSend) { taosMsleep(2000); } -TEST_F(TransEnv, srvPersisHandleExcept) { +TEST_F(TransEnv, srvPersistHandleExcept) { // conn breken // } -TEST_F(TransEnv, cliPersisHandleExcept) { +TEST_F(TransEnv, cliPersistHandleExcept) { // conn breken }