Merge pull request #13090 from taosdata/fix/rpcMemoryLeak

fix: avoid mem leak when taosd quit

source/libs/transport/inc/transComm.h

@@ -217,6 +217,22 @@ SAsyncPool* transCreateAsyncPool(uv_loop_t* loop, int sz, void* arg, AsyncCB cb)
 void        transDestroyAsyncPool(SAsyncPool* pool);
 int         transSendAsync(SAsyncPool* pool, queue* mq);
 
+#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);                                      \
+        }                                                     \
+      }                                                       \
+    }                                                         \
+  } while (0)
 int  transInitBuffer(SConnBuffer* buf);
 int  transClearBuffer(SConnBuffer* buf);
 int  transDestroyBuffer(SConnBuffer* buf);

source/libs/transport/src/transCli.c

@@ -892,6 +892,7 @@ static void destroyThrdObj(SCliThrdObj* pThrd) {
   taosThreadJoin(pThrd->thread, NULL);
   CLI_RELEASE_UV(pThrd->loop);
   taosThreadMutexDestroy(&pThrd->msgMtx);
+  TRANS_DESTROY_ASYNC_POOL_MSG(pThrd->asyncPool, SCliMsg, destroyCmsg);
   transDestroyAsyncPool(pThrd->asyncPool);
 
   transDQDestroy(pThrd->delayQueue);

source/libs/transport/src/transComm.c

@@ -190,6 +190,7 @@ SAsyncPool* transCreateAsyncPool(uv_loop_t* loop, int sz, void* arg, AsyncCB cb)
   }
   return pool;
 }
+
 void transDestroyAsyncPool(SAsyncPool* pool) {
   for (int i = 0; i < pool->nAsync; i++) {
     uv_async_t* async = &(pool->asyncs[i]);

source/libs/transport/src/transSrv.c

@@ -1036,6 +1036,7 @@ void destroyWorkThrd(SWorkThrdObj* pThrd) {
   }
   taosThreadJoin(pThrd->thread, NULL);
   SRV_RELEASE_UV(pThrd->loop);
+  TRANS_DESTROY_ASYNC_POOL_MSG(pThrd->asyncPool, SSrvMsg, destroySmsg);
   transDestroyAsyncPool(pThrd->asyncPool);
   taosMemoryFree(pThrd->loop);
   taosMemoryFree(pThrd);