diff --git a/include/libs/wal/wal.h b/include/libs/wal/wal.h index 8b9fc84334eae4c1f9a8ba38837a05a0599f52be..a1ae1e429dd5dbb18f6521b263576e7096482327 100644 --- a/include/libs/wal/wal.h +++ b/include/libs/wal/wal.h @@ -33,16 +33,16 @@ extern "C" { #define wTrace(...) { if (wDebugFlag & DEBUG_TRACE) { taosPrintLog("WAL ", DEBUG_TRACE, wDebugFlag, __VA_ARGS__); }} // clang-format on -#define WAL_PROTO_VER 0 -#define WAL_NOSUFFIX_LEN 20 -#define WAL_SUFFIX_AT (WAL_NOSUFFIX_LEN + 1) -#define WAL_LOG_SUFFIX "log" -#define WAL_INDEX_SUFFIX "idx" -#define WAL_REFRESH_MS 1000 -#define WAL_PATH_LEN (TSDB_FILENAME_LEN + 12) -#define WAL_FILE_LEN (WAL_PATH_LEN + 32) -#define WAL_MAGIC 0xFAFBFCFDF4F3F2F1ULL -#define WAL_SCAN_BUF_SIZE (1024 * 1024 * 3) +#define WAL_PROTO_VER 0 +#define WAL_NOSUFFIX_LEN 20 +#define WAL_SUFFIX_AT (WAL_NOSUFFIX_LEN + 1) +#define WAL_LOG_SUFFIX "log" +#define WAL_INDEX_SUFFIX "idx" +#define WAL_REFRESH_MS 1000 +#define WAL_PATH_LEN (TSDB_FILENAME_LEN + 12) +#define WAL_FILE_LEN (WAL_PATH_LEN + 32) +#define WAL_MAGIC 0xFAFBFCFDF4F3F2F1ULL +#define WAL_SCAN_BUF_SIZE (1024 * 1024 * 3) typedef enum { TAOS_WAL_WRITE = 1, diff --git a/source/client/src/clientMain.c b/source/client/src/clientMain.c index 581cd636e841d3a5245061c7d2b85614086be821..87f5e5fa40ff06efb0983689767f3b4f582be55b 100644 --- a/source/client/src/clientMain.c +++ b/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); diff --git a/source/client/src/clientTmq.c b/source/client/src/clientTmq.c index db717a4e4e0bbbc6f4f714255800bf54184c7108..4352ec69d328baed90f4f5c6606e1e1f23c9fe9c 100644 --- a/source/client/src/clientTmq.c +++ b/source/client/src/clientTmq.c @@ -814,24 +814,55 @@ 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; + } } } @@ -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;*/ } diff --git a/source/dnode/vnode/src/tq/tq.c b/source/dnode/vnode/src/tq/tq.c index 191fc1b49c274faf95d7b531076d53921449408a..389c8013f9ca92dbc5d24443898aa4c28a5fe479 100644 --- a/source/dnode/vnode/src/tq/tq.c +++ b/source/dnode/vnode/src/tq/tq.c @@ -725,9 +725,15 @@ int32_t tqProcessDeleteSubReq(STQ* pTq, int64_t version, char* msg, int32_t msgL } taosWUnLockLatch(&pTq->pushLock); - 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); + 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); @@ -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; } diff --git a/source/libs/wal/src/walMgmt.c b/source/libs/wal/src/walMgmt.c index f53088fac642d4059164afe0de64fe59bb8db769..702d05f57696c55e88b94d2d335674bd8f262c49 100644 --- a/source/libs/wal/src/walMgmt.c +++ b/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); diff --git a/source/libs/wal/src/walRef.c b/source/libs/wal/src/walRef.c index eae5d9f1a7f7e1b4624270e2e74f91581b1e9541..e86111109ce4fc9768be8b662670ae711f935e39 100644 --- a/source/libs/wal/src/walRef.c +++ b/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;