Merge pull request #18854 from taosdata/feature/stream

fix: mem leak

source/client/src/clientMain.c

@@ -186,10 +186,10 @@ void taos_free_result(TAOS_RES *res) {
     destroyRequest(pRequest);
   } else if (TD_RES_TMQ_METADATA(res)) {
     SMqTaosxRspObj *pRsp = (SMqTaosxRspObj *)res;
-    if (pRsp->rsp.blockData) taosArrayDestroyP(pRsp->rsp.blockData, taosMemoryFree);
-    if (pRsp->rsp.blockDataLen) taosArrayDestroy(pRsp->rsp.blockDataLen);
-    if (pRsp->rsp.withTbName) taosArrayDestroyP(pRsp->rsp.blockTbName, taosMemoryFree);
-    if (pRsp->rsp.withSchema) taosArrayDestroyP(pRsp->rsp.blockSchema, (FDelete)tDeleteSSchemaWrapper);
+    taosArrayDestroyP(pRsp->rsp.blockData, taosMemoryFree);
+    taosArrayDestroy(pRsp->rsp.blockDataLen);
+    taosArrayDestroyP(pRsp->rsp.blockTbName, taosMemoryFree);
+    taosArrayDestroyP(pRsp->rsp.blockSchema, (FDelete)tDeleteSSchemaWrapper);
     // taosx
     taosArrayDestroy(pRsp->rsp.createTableLen);
     taosArrayDestroyP(pRsp->rsp.createTableReq, taosMemoryFree);
@@ -199,10 +199,10 @@ void taos_free_result(TAOS_RES *res) {
     taosMemoryFree(pRsp);
   } else if (TD_RES_TMQ(res)) {
     SMqRspObj *pRsp = (SMqRspObj *)res;
-    if (pRsp->rsp.blockData) taosArrayDestroyP(pRsp->rsp.blockData, taosMemoryFree);
-    if (pRsp->rsp.blockDataLen) taosArrayDestroy(pRsp->rsp.blockDataLen);
-    if (pRsp->rsp.withTbName) taosArrayDestroyP(pRsp->rsp.blockTbName, taosMemoryFree);
-    if (pRsp->rsp.withSchema) taosArrayDestroyP(pRsp->rsp.blockSchema, (FDelete)tDeleteSSchemaWrapper);
+    taosArrayDestroyP(pRsp->rsp.blockData, taosMemoryFree);
+    taosArrayDestroy(pRsp->rsp.blockDataLen);
+    taosArrayDestroyP(pRsp->rsp.blockTbName, taosMemoryFree);
+    taosArrayDestroyP(pRsp->rsp.blockSchema, (FDelete)tDeleteSSchemaWrapper);
     pRsp->resInfo.pRspMsg = NULL;
     doFreeReqResultInfo(&pRsp->resInfo);
     taosMemoryFree(pRsp);

source/client/src/clientTmq.c

@@ -814,25 +814,56 @@ int32_t tmqHandleAllDelayedTask(tmq_t* tmq) {
   return 0;
 }
 
+static void tmqFreeRspWrapper(SMqRspWrapper* rspWrapper) {
+  if (rspWrapper->tmqRspType == TMQ_MSG_TYPE__END_RSP) {
+    // do nothing
+  } else if (rspWrapper->tmqRspType == TMQ_MSG_TYPE__EP_RSP) {
+    SMqAskEpRspWrapper* pEpRspWrapper = (SMqAskEpRspWrapper*)rspWrapper;
+    tDeleteSMqAskEpRsp(&pEpRspWrapper->msg);
+  } else if (rspWrapper->tmqRspType == TMQ_MSG_TYPE__POLL_RSP) {
+    SMqPollRspWrapper* pRsp = (SMqPollRspWrapper*)rspWrapper;
+    taosArrayDestroyP(pRsp->dataRsp.blockData, taosMemoryFree);
+    taosArrayDestroy(pRsp->dataRsp.blockDataLen);
+    taosArrayDestroyP(pRsp->dataRsp.blockTbName, taosMemoryFree);
+    taosArrayDestroyP(pRsp->dataRsp.blockSchema, (FDelete)tDeleteSSchemaWrapper);
+  } else if (rspWrapper->tmqRspType == TMQ_MSG_TYPE__POLL_META_RSP) {
+    SMqPollRspWrapper* pRsp = (SMqPollRspWrapper*)rspWrapper;
+    taosMemoryFree(pRsp->metaRsp.metaRsp);
+  } else if (rspWrapper->tmqRspType == TMQ_MSG_TYPE__TAOSX_RSP) {
+    SMqPollRspWrapper* pRsp = (SMqPollRspWrapper*)rspWrapper;
+    taosArrayDestroyP(pRsp->taosxRsp.blockData, taosMemoryFree);
+    taosArrayDestroy(pRsp->taosxRsp.blockDataLen);
+    taosArrayDestroyP(pRsp->taosxRsp.blockTbName, taosMemoryFree);
+    taosArrayDestroyP(pRsp->taosxRsp.blockSchema, (FDelete)tDeleteSSchemaWrapper);
+    // taosx
+    taosArrayDestroy(pRsp->taosxRsp.createTableLen);
+    taosArrayDestroyP(pRsp->taosxRsp.createTableReq, taosMemoryFree);
+  }
+}
+
 void tmqClearUnhandleMsg(tmq_t* tmq) {
-  SMqRspWrapper* msg = NULL;
+  SMqRspWrapper* rspWrapper = NULL;
   while (1) {
-    taosGetQitem(tmq->qall, (void**)&msg);
-    if (msg)
-      taosFreeQitem(msg);
-    else
+    taosGetQitem(tmq->qall, (void**)&rspWrapper);
+    if (rspWrapper) {
+      tmqFreeRspWrapper(rspWrapper);
+      taosFreeQitem(rspWrapper);
+    } else {
       break;
     }
+  }
 
-  msg = NULL;
+  rspWrapper = NULL;
   taosReadAllQitems(tmq->mqueue, tmq->qall);
   while (1) {
-    taosGetQitem(tmq->qall, (void**)&msg);
-    if (msg)
-      taosFreeQitem(msg);
-    else
+    taosGetQitem(tmq->qall, (void**)&rspWrapper);
+    if (rspWrapper) {
+      tmqFreeRspWrapper(rspWrapper);
+      taosFreeQitem(rspWrapper);
+    } else {
       break;
     }
+  }
 }
 
 int32_t tmqSubscribeCb(void* param, SDataBuf* pMsg, int32_t code) {
@@ -1644,6 +1675,7 @@ int32_t tmqHandleNoPollRsp(tmq_t* tmq, SMqRspWrapper* rspWrapper, bool* pReset)
       tDeleteSMqAskEpRsp(rspMsg);
       *pReset = true;
     } else {
+      tmqFreeRspWrapper(rspWrapper);
       *pReset = false;
     }
   } else {
@@ -1695,6 +1727,7 @@ void* tmqHandleAllRsp(tmq_t* tmq, int64_t timeout, bool pollIfReset) {
       } else {
         tscDebug("msg discard since epoch mismatch: msg epoch %d, consumer epoch %d",
                  pollRspWrapper->dataRsp.head.epoch, consumerEpoch);
+        tmqFreeRspWrapper(rspWrapper);
         taosFreeQitem(pollRspWrapper);
       }
     } else if (rspWrapper->tmqRspType == TMQ_MSG_TYPE__POLL_META_RSP) {
@@ -1713,6 +1746,7 @@ void* tmqHandleAllRsp(tmq_t* tmq, int64_t timeout, bool pollIfReset) {
       } else {
         tscDebug("msg discard since epoch mismatch: msg epoch %d, consumer epoch %d",
                  pollRspWrapper->metaRsp.head.epoch, consumerEpoch);
+        tmqFreeRspWrapper(rspWrapper);
         taosFreeQitem(pollRspWrapper);
       }
     } else if (rspWrapper->tmqRspType == TMQ_MSG_TYPE__TAOSX_RSP) {
@@ -1743,6 +1777,7 @@ void* tmqHandleAllRsp(tmq_t* tmq, int64_t timeout, bool pollIfReset) {
       } else {
         tscDebug("msg discard since epoch mismatch: msg epoch %d, consumer epoch %d",
                  pollRspWrapper->taosxRsp.head.epoch, consumerEpoch);
+        tmqFreeRspWrapper(rspWrapper);
         taosFreeQitem(pollRspWrapper);
       }
     } else {
@@ -1794,7 +1829,7 @@ TAOS_RES* tmq_consumer_poll(tmq_t* tmq, int64_t timeout) {
   while (1) {
     tmqHandleAllDelayedTask(tmq);
     if (tmqPollImpl(tmq, timeout) < 0) {
-      tscDebug("return since poll err");
+      tscDebug("consumer:%" PRId64 " return since poll err", tmq->consumerId);
       /*return NULL;*/
     }
 

source/dnode/vnode/src/tq/tq.c

@@ -725,10 +725,16 @@ int32_t tqProcessDeleteSubReq(STQ* pTq, int64_t version, char* msg, int32_t msgL
   }
   taosWUnLockLatch(&pTq->pushLock);
 
+  STqHandle* pHandle = taosHashGet(pTq->pHandle, pReq->subKey, strlen(pReq->subKey));
+  if (pHandle) {
+    if (pHandle->pRef) {
+      walCloseRef(pTq->pVnode->pWal, pHandle->pRef->refId);
+    }
     code = taosHashRemove(pTq->pHandle, pReq->subKey, strlen(pReq->subKey));
     if (code != 0) {
       tqError("cannot process tq delete req %s, since no such handle", pReq->subKey);
     }
+  }
 
   code = tqOffsetDelete(pTq->pOffsetStore, pReq->subKey);
   if (code != 0) {
@@ -736,7 +742,7 @@ int32_t tqProcessDeleteSubReq(STQ* pTq, int64_t version, char* msg, int32_t msgL
   }
 
   if (tqMetaDeleteHandle(pTq, pReq->subKey) < 0) {
-    ASSERT(0);
+    tqError("cannot process tq delete req %s, since no such offset in tdb", pReq->subKey);
   }
   return 0;
 }

source/libs/wal/src/walMgmt.c

@@ -223,6 +223,7 @@ void walClose(SWal *pWal) {
     taosMemoryFree(pRef);
   }
   taosHashCleanup(pWal->pRefHash);
+  pWal->pRefHash = NULL;
   taosThreadMutexUnlock(&pWal->mutex);
 
   taosRemoveRef(tsWal.refSetId, pWal->refId);

source/libs/wal/src/walRef.c

@@ -32,15 +32,18 @@ SWalRef *walOpenRef(SWal *pWal) {
   return pRef;
 }
 
-#if 1
 void walCloseRef(SWal *pWal, int64_t refId) {
   SWalRef **ppRef = taosHashGet(pWal->pRefHash, &refId, sizeof(int64_t));
   if (ppRef == NULL) return;
   SWalRef *pRef = *ppRef;
+  if (pRef) {
+    wDebug("vgId:%d, wal close ref %" PRId64 ", refId %" PRId64, pWal->cfg.vgId, pRef->refVer, pRef->refId);
+  } else {
+    wDebug("vgId:%d, wal close ref null, refId %" PRId64, pWal->cfg.vgId, refId);
+  }
   taosHashRemove(pWal->pRefHash, &refId, sizeof(int64_t));
   taosMemoryFree(pRef);
 }
-#endif
 
 int32_t walRefVer(SWalRef *pRef, int64_t ver) {
   SWal *pWal = pRef->pWal;