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

fix mem leak while taosd quit

This commit is contained in:
yihaoDeng 2024-02-29 18:40:21 +08:00
parent 593a10d81d
commit a0c77f5baa
4 changed files with 40 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
source/dnode/mgmt/node_mgmt/src/dmTransport.c
View File

@ -345,6 +345,7 @@ int32_t dmInitClient(SDnode *pDnode) {
rpcInit.parent = pDnode; rpcInit.parent = pDnode;
rpcInit.rfp = rpcRfp; rpcInit.rfp = rpcRfp;
rpcInit.compressSize = tsCompressMsgSize; rpcInit.compressSize = tsCompressMsgSize;
rpcInit.dfp = destroyAhandle;
rpcInit.retryMinInterval = tsRedirectPeriod; rpcInit.retryMinInterval = tsRedirectPeriod;
rpcInit.retryStepFactor = tsRedirectFactor; rpcInit.retryStepFactor = tsRedirectFactor;

4
source/libs/transport/inc/transComm.h
View File

@ -256,7 +256,7 @@ void transAsyncPoolDestroy(SAsyncPool* pool);
int transAsyncSend(SAsyncPool* pool, queue* mq); int transAsyncSend(SAsyncPool* pool, queue* mq);
bool transAsyncPoolIsEmpty(SAsyncPool* pool); bool transAsyncPoolIsEmpty(SAsyncPool* pool);
#define TRANS_DESTROY_ASYNC_POOL_MSG(pool, msgType, freeFunc) \ #define TRANS_DESTROY_ASYNC_POOL_MSG(pool, msgType, freeFunc, param) \
do { \ do { \
for (int i = 0; i < pool->nAsync; i++) { \ for (int i = 0; i < pool->nAsync; i++) { \
uv_async_t* async = &(pool->asyncs[i]); \ uv_async_t* async = &(pool->asyncs[i]); \
@ -267,7 +267,7 @@ bool transAsyncPoolIsEmpty(SAsyncPool* pool);
QUEUE_REMOVE(h); \ QUEUE_REMOVE(h); \
msgType* msg = QUEUE_DATA(h, msgType, q); \ msgType* msg = QUEUE_DATA(h, msgType, q); \
if (msg != NULL) { \ if (msg != NULL) { \
freeFunc(msg); \ freeFunc(msg, param); \
} \ } \
} \ } \
} \ } \

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

@ -219,6 +219,8 @@ static void (*cliAsyncHandle[])(SCliMsg* pMsg, SCliThrd* pThrd) = {cliHandleReq,
/// NULL,cliHandleUpdate}; /// NULL,cliHandleUpdate};
static FORCE_INLINE void destroyCmsg(void* cmsg); 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 void destroyCmsgAndAhandle(void* cmsg);
static FORCE_INLINE int cliRBChoseIdx(STrans* pTransInst); static FORCE_INLINE int cliRBChoseIdx(STrans* pTransInst);
static FORCE_INLINE void transDestroyConnCtx(STransConnCtx* ctx); static FORCE_INLINE void transDestroyConnCtx(STransConnCtx* ctx);
@ -1963,7 +1965,19 @@ static FORCE_INLINE void destroyCmsg(void* arg) {
transFreeMsg(pMsg->msg.pCont); transFreeMsg(pMsg->msg.pCont);
taosMemoryFree(pMsg); 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) { static FORCE_INLINE void destroyCmsgAndAhandle(void* param) {
if (param == NULL) return; if (param == NULL) return;
@ -2057,7 +2071,7 @@ static void destroyThrdObj(SCliThrd* pThrd) {
taosThreadJoin(pThrd->thread, NULL); taosThreadJoin(pThrd->thread, NULL);
CLI_RELEASE_UV(pThrd->loop); CLI_RELEASE_UV(pThrd->loop);
taosThreadMutexDestroy(&pThrd->msgMtx); 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); transAsyncPoolDestroy(pThrd->asyncPool);
transDQDestroy(pThrd->delayQueue, destroyCmsgAndAhandle); transDQDestroy(pThrd->delayQueue, destroyCmsgAndAhandle);

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

@ -671,6 +671,7 @@ static FORCE_INLINE void destroySmsg(SSvrMsg* smsg) {
transFreeMsg(smsg->msg.pCont); transFreeMsg(smsg->msg.pCont);
taosMemoryFree(smsg); taosMemoryFree(smsg);
} }
static FORCE_INLINE void destroySmsgWrapper(void* smsg, void* param) { destroySmsg((SSvrMsg*)smsg); }
static void destroyAllConn(SWorkThrd* pThrd) { static void destroyAllConn(SWorkThrd* pThrd) {
tTrace("thread %p destroy all conn ", pThrd); tTrace("thread %p destroy all conn ", pThrd);
while (!QUEUE_IS_EMPTY(&pThrd->conn)) { while (!QUEUE_IS_EMPTY(&pThrd->conn)) {
@ -1394,7 +1395,7 @@ void destroyWorkThrd(SWorkThrd* pThrd) {
} }
taosThreadJoin(pThrd->thread, NULL); taosThreadJoin(pThrd->thread, NULL);
SRV_RELEASE_UV(pThrd->loop); 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); transAsyncPoolDestroy(pThrd->asyncPool);
uvWhiteListDestroy(pThrd->pWhiteList); uvWhiteListDestroy(pThrd->pWhiteList);