Merge pull request #15176 from taosdata/fix/avoidRetryMemleak

fix: avoid retry memleak

source/libs/transport/inc/transComm.h

@@ -392,7 +392,7 @@ typedef struct SDelayQueue {
 } SDelayQueue;
 
 int  transDQCreate(uv_loop_t* loop, SDelayQueue** queue);
-void transDQDestroy(SDelayQueue* queue);
+void transDQDestroy(SDelayQueue* queue, void (*freeFunc)(void* arg));
 int  transDQSched(SDelayQueue* queue, void (*func)(void* arg), void* arg, uint64_t timeoutMs);
 
 bool transEpSetIsEqual(SEpSet* a, SEpSet* b);

source/libs/transport/src/transCli.c

@@ -140,7 +140,7 @@ static void destroyUserdata(STransMsg* userdata);
 
 static int cliRBChoseIdx(STrans* pTransInst);
 
-static void destroyCmsg(SCliMsg* cmsg);
+static void destroyCmsg(void* cmsg);
 static void transDestroyConnCtx(STransConnCtx* ctx);
 // thread obj
 static SCliThrd* createThrdObj();
@@ -198,6 +198,7 @@ static void cliReleaseUnfinishedMsg(SCliConn* conn) {
       }                                                                                                           \
       destroyCmsg(pMsg);                                                                                          \
       cliReleaseUnfinishedMsg(conn);                                                                              \
+      transQueueClear(&conn->cliMsgs);                                                                            \
       addConnToPool(((SCliThrd*)conn->hostThrd)->pool, conn);                                                     \
       return;                                                                                                     \
     }                                                                                                             \
@@ -545,6 +546,7 @@ static void addConnToPool(void* pool, SCliConn* conn) {
 
   STrans* pTransInst = thrd->pTransInst;
   conn->expireTime = taosGetTimestampMs() + CONN_PERSIST_TIME(pTransInst->idleTime);
+  cliReleaseUnfinishedMsg(conn);
   transQueueClear(&conn->cliMsgs);
   transCtxCleanup(&conn->ctx);
   conn->status = ConnInPool;
@@ -645,6 +647,7 @@ static void cliDestroy(uv_handle_t* handle) {
   conn->stream->data = NULL;
   taosMemoryFree(conn->stream);
   transCtxCleanup(&conn->ctx);
+  cliReleaseUnfinishedMsg(conn);
   transQueueDestroy(&conn->cliMsgs);
   tTrace("%s conn %p destroy successfully", CONN_GET_INST_LABEL(conn), conn);
   transReqQueueClear(&conn->wreqQueue);
@@ -962,7 +965,8 @@ static void destroyUserdata(STransMsg* userdata) {
   transFreeMsg(userdata->pCont);
   userdata->pCont = NULL;
 }
-static void destroyCmsg(SCliMsg* pMsg) {
+static void destroyCmsg(void* arg) {
+  SCliMsg* pMsg = arg;
   if (pMsg == NULL) {
     return;
   }
@@ -1001,7 +1005,7 @@ static void destroyThrdObj(SCliThrd* pThrd) {
   TRANS_DESTROY_ASYNC_POOL_MSG(pThrd->asyncPool, SCliMsg, destroyCmsg);
   transDestroyAsyncPool(pThrd->asyncPool);
 
-  transDQDestroy(pThrd->delayQueue);
+  transDQDestroy(pThrd->delayQueue, destroyCmsg);
   taosMemoryFree(pThrd->loop);
   taosMemoryFree(pThrd);
 }

source/libs/transport/src/transComm.c

@@ -456,7 +456,7 @@ int transDQCreate(uv_loop_t* loop, SDelayQueue** queue) {
   return 0;
 }
 
-void transDQDestroy(SDelayQueue* queue) {
+void transDQDestroy(SDelayQueue* queue, void (*freeFunc)(void* arg)) {
   taosMemoryFree(queue->timer);
 
   while (heapSize(queue->heap) > 0) {
@@ -467,6 +467,11 @@ void transDQDestroy(SDelayQueue* queue) {
     heapRemove(queue->heap, minNode);
 
     SDelayTask* task = container_of(minNode, SDelayTask, node);
+
+    STaskArg* arg = task->arg;
+    freeFunc(arg->param1);
+    taosMemoryFree(arg);
+
     taosMemoryFree(task);
   }
   heapDestroy(queue->heap);

tests/script/tsim/valgrind/checkError3.sim

@@ -90,7 +90,7 @@ $null=
 
 system_content sh/checkValgrind.sh -n dnode1 
 print cmd return result ----> [ $system_content ]
-if $system_content > 0 then
+if $system_content > 2 then
   return -1
 endi