diff --git a/source/dnode/mgmt/node_mgmt/src/dmTransport.c b/source/dnode/mgmt/node_mgmt/src/dmTransport.c index 1a31f08801..c54da860dd 100644 --- a/source/dnode/mgmt/node_mgmt/src/dmTransport.c +++ b/source/dnode/mgmt/node_mgmt/src/dmTransport.c @@ -87,9 +87,9 @@ static bool dmIsForbiddenIp(int8_t forbidden, char *user, uint32_t clientIp) { } } static void dmProcessRpcMsg(SDnode *pDnode, SRpcMsg *pRpc, SEpSet *pEpSet) { - SDnodeTrans *pTrans = &pDnode->trans; + SDnodeTrans * pTrans = &pDnode->trans; int32_t code = -1; - SRpcMsg *pMsg = NULL; + SRpcMsg * pMsg = NULL; SMgmtWrapper *pWrapper = NULL; SDnodeHandle *pHandle = &pTrans->msgHandles[TMSG_INDEX(pRpc->msgType)]; @@ -256,11 +256,11 @@ int32_t dmInitMsgHandle(SDnode *pDnode) { for (EDndNodeType ntype = DNODE; ntype < NODE_END; ++ntype) { SMgmtWrapper *pWrapper = &pDnode->wrappers[ntype]; - SArray *pArray = (*pWrapper->func.getHandlesFp)(); + SArray * pArray = (*pWrapper->func.getHandlesFp)(); if (pArray == NULL) return -1; for (int32_t i = 0; i < taosArrayGetSize(pArray); ++i) { - SMgmtHandle *pMgmt = taosArrayGet(pArray, i); + SMgmtHandle * pMgmt = taosArrayGet(pArray, i); SDnodeHandle *pHandle = &pTrans->msgHandles[TMSG_INDEX(pMgmt->msgType)]; if (pMgmt->needCheckVgId) { pHandle->needCheckVgId = pMgmt->needCheckVgId; @@ -345,6 +345,7 @@ int32_t dmInitClient(SDnode *pDnode) { rpcInit.parent = pDnode; rpcInit.rfp = rpcRfp; rpcInit.compressSize = tsCompressMsgSize; + rpcInit.dfp = destroyAhandle; rpcInit.retryMinInterval = tsRedirectPeriod; rpcInit.retryStepFactor = tsRedirectFactor; diff --git a/source/libs/transport/inc/transComm.h b/source/libs/transport/inc/transComm.h index 5b18d56d70..c010e31320 100644 --- a/source/libs/transport/inc/transComm.h +++ b/source/libs/transport/inc/transComm.h @@ -256,21 +256,21 @@ void transAsyncPoolDestroy(SAsyncPool* pool); int transAsyncSend(SAsyncPool* pool, queue* mq); bool transAsyncPoolIsEmpty(SAsyncPool* pool); -#define TRANS_DESTROY_ASYNC_POOL_MSG(pool, msgType, freeFunc) \ - do { \ - for (int i = 0; i < pool->nAsync; i++) { \ - uv_async_t* async = &(pool->asyncs[i]); \ - SAsyncItem* item = async->data; \ - while (!QUEUE_IS_EMPTY(&item->qmsg)) { \ - tTrace("destroy msg in async pool "); \ - queue* h = QUEUE_HEAD(&item->qmsg); \ - QUEUE_REMOVE(h); \ - msgType* msg = QUEUE_DATA(h, msgType, q); \ - if (msg != NULL) { \ - freeFunc(msg); \ - } \ - } \ - } \ +#define TRANS_DESTROY_ASYNC_POOL_MSG(pool, msgType, freeFunc, param) \ + do { \ + for (int i = 0; i < pool->nAsync; i++) { \ + uv_async_t* async = &(pool->asyncs[i]); \ + SAsyncItem* item = async->data; \ + while (!QUEUE_IS_EMPTY(&item->qmsg)) { \ + tTrace("destroy msg in async pool "); \ + queue* h = QUEUE_HEAD(&item->qmsg); \ + QUEUE_REMOVE(h); \ + msgType* msg = QUEUE_DATA(h, msgType, q); \ + if (msg != NULL) { \ + freeFunc(msg, param); \ + } \ + } \ + } \ } while (0) #define ASYNC_CHECK_HANDLE(exh1, id) \ diff --git a/source/libs/transport/src/transCli.c b/source/libs/transport/src/transCli.c index e937d2e65e..e369bf8f7d 100644 --- a/source/libs/transport/src/transCli.c +++ b/source/libs/transport/src/transCli.c @@ -219,6 +219,8 @@ static void (*cliAsyncHandle[])(SCliMsg* pMsg, SCliThrd* pThrd) = {cliHandleReq, /// NULL,cliHandleUpdate}; static FORCE_INLINE void destroyCmsg(void* cmsg); + +static FORCE_INLINE void destroyQueuedMsg(void* arg, void* param); static FORCE_INLINE void destroyCmsgAndAhandle(void* cmsg); static FORCE_INLINE int cliRBChoseIdx(STrans* pTransInst); static FORCE_INLINE void transDestroyConnCtx(STransConnCtx* ctx); @@ -1963,7 +1965,19 @@ static FORCE_INLINE void destroyCmsg(void* arg) { transFreeMsg(pMsg->msg.pCont); taosMemoryFree(pMsg); } - +static FORCE_INLINE void destroyQueuedMsg(void* arg, void* param) { + SCliMsg* pMsg = arg; + if (pMsg == NULL) { + return; + } + if (param != NULL) { + SCliThrd* pThrd = param; + if (pThrd->destroyAhandleFp) (*pThrd->destroyAhandleFp)(pMsg->msg.info.ahandle); + } + transDestroyConnCtx(pMsg->ctx); + transFreeMsg(pMsg->msg.pCont); + taosMemoryFree(pMsg); +} static FORCE_INLINE void destroyCmsgAndAhandle(void* param) { if (param == NULL) return; @@ -2057,7 +2071,7 @@ static void destroyThrdObj(SCliThrd* pThrd) { taosThreadJoin(pThrd->thread, NULL); CLI_RELEASE_UV(pThrd->loop); taosThreadMutexDestroy(&pThrd->msgMtx); - TRANS_DESTROY_ASYNC_POOL_MSG(pThrd->asyncPool, SCliMsg, destroyCmsg); + TRANS_DESTROY_ASYNC_POOL_MSG(pThrd->asyncPool, SCliMsg, destroyQueuedMsg, (void*)pThrd); transAsyncPoolDestroy(pThrd->asyncPool); transDQDestroy(pThrd->delayQueue, destroyCmsgAndAhandle); diff --git a/source/libs/transport/src/transSvr.c b/source/libs/transport/src/transSvr.c index b324ca5f91..5a1ef31b7d 100644 --- a/source/libs/transport/src/transSvr.c +++ b/source/libs/transport/src/transSvr.c @@ -159,7 +159,7 @@ static void uvStartSendResp(SSvrMsg* msg); static void uvNotifyLinkBrokenToApp(SSvrConn* conn); -static FORCE_INLINE void destroySmsg(SSvrMsg* smsg); +static FORCE_INLINE void destroySmsg(SSvrMsg* smsg); static FORCE_INLINE SSvrConn* createConn(void* hThrd); static FORCE_INLINE void destroyConn(SSvrConn* conn, bool clear /*clear handle or not*/); static FORCE_INLINE void destroyConnRegArg(SSvrConn* conn); @@ -671,7 +671,8 @@ static FORCE_INLINE void destroySmsg(SSvrMsg* smsg) { transFreeMsg(smsg->msg.pCont); taosMemoryFree(smsg); } -static void destroyAllConn(SWorkThrd* pThrd) { +static FORCE_INLINE void destroySmsgWrapper(void* smsg, void* param) { destroySmsg((SSvrMsg*)smsg); } +static void destroyAllConn(SWorkThrd* pThrd) { tTrace("thread %p destroy all conn ", pThrd); while (!QUEUE_IS_EMPTY(&pThrd->conn)) { queue* h = QUEUE_HEAD(&pThrd->conn); @@ -1394,7 +1395,7 @@ void destroyWorkThrd(SWorkThrd* pThrd) { } taosThreadJoin(pThrd->thread, NULL); SRV_RELEASE_UV(pThrd->loop); - TRANS_DESTROY_ASYNC_POOL_MSG(pThrd->asyncPool, SSvrMsg, destroySmsg); + TRANS_DESTROY_ASYNC_POOL_MSG(pThrd->asyncPool, SSvrMsg, destroySmsgWrapper, NULL); transAsyncPoolDestroy(pThrd->asyncPool); uvWhiteListDestroy(pThrd->pWhiteList);