From 74d5f029697cac47d63eb43ec52765c50572f9de Mon Sep 17 00:00:00 2001 From: yihaoDeng Date: Mon, 5 Jun 2023 04:56:07 +0000 Subject: [PATCH 1/7] fix invalid free --- include/libs/function/function.h | 1 + include/libs/stream/tstream.h | 1 - source/libs/stream/inc/streamInc.h | 7 ++++-- source/libs/stream/src/streamBackendRocksdb.c | 18 +++++++++----- source/libs/stream/src/streamMeta.c | 3 +-- source/libs/stream/src/streamState.c | 24 +++++++++---------- 6 files changed, 31 insertions(+), 23 deletions(-) diff --git a/include/libs/function/function.h b/include/libs/function/function.h index e015f4182e..c92ce254a8 100644 --- a/include/libs/function/function.h +++ b/include/libs/function/function.h @@ -163,6 +163,7 @@ typedef struct { int64_t checkPointId; int32_t taskId; int64_t streamId; + int64_t streamBackendRid; } SStreamState; typedef struct SFunctionStateStore { diff --git a/include/libs/stream/tstream.h b/include/libs/stream/tstream.h index 8316e6ef50..3222a125dd 100644 --- a/include/libs/stream/tstream.h +++ b/include/libs/stream/tstream.h @@ -344,7 +344,6 @@ typedef struct SStreamMeta { SRWLatch lock; int32_t walScanCounter; void* streamBackend; - int32_t streamBackendId; int64_t streamBackendRid; SHashObj* pTaskBackendUnique; } SStreamMeta; diff --git a/source/libs/stream/inc/streamInc.h b/source/libs/stream/inc/streamInc.h index 2c1956998a..c7ee308b61 100644 --- a/source/libs/stream/inc/streamInc.h +++ b/source/libs/stream/inc/streamInc.h @@ -36,8 +36,9 @@ static SStreamGlobalEnv streamEnv; int32_t streamDispatchStreamBlock(SStreamTask* pTask); SStreamDataBlock* createStreamDataFromDispatchMsg(const SStreamDispatchReq* pReq, int32_t blockType, int32_t srcVg); -SStreamDataBlock* createStreamBlockFromResults(SStreamQueueItem* pItem, SStreamTask* pTask, int64_t resultSize, SArray* pRes); -void destroyStreamDataBlock(SStreamDataBlock* pBlock); +SStreamDataBlock* createStreamBlockFromResults(SStreamQueueItem* pItem, SStreamTask* pTask, int64_t resultSize, + SArray* pRes); +void destroyStreamDataBlock(SStreamDataBlock* pBlock); int32_t streamRetrieveReqToData(const SStreamRetrieveReq* pReq, SStreamDataBlock* pData); int32_t streamDispatchAllBlocks(SStreamTask* pTask, const SStreamDataBlock* data); @@ -53,6 +54,8 @@ int32_t streamDispatchOneRecoverFinishReq(SStreamTask* pTask, const SStreamRecov SStreamQueueItem* streamMergeQueueItem(SStreamQueueItem* dst, SStreamQueueItem* pElem); +extern int32_t streamBackendId; + #ifdef __cplusplus } #endif diff --git a/source/libs/stream/src/streamBackendRocksdb.c b/source/libs/stream/src/streamBackendRocksdb.c index b3995f020b..d190f4b43e 100644 --- a/source/libs/stream/src/streamBackendRocksdb.c +++ b/source/libs/stream/src/streamBackendRocksdb.c @@ -16,7 +16,9 @@ #include "streamBackendRocksdb.h" #include "executor.h" #include "query.h" +#include "streamInc.h" #include "tcommon.h" +#include "tref.h" typedef struct SCompactFilteFactory { void* status; @@ -79,7 +81,7 @@ const char* compareParKeyName(void* name); const char* comparePartagKeyName(void* name); void* streamBackendInit(const char* path) { - qDebug("init stream backend"); + qDebug("start to init stream backend at %s", path); SBackendHandle* pHandle = calloc(1, sizeof(SBackendHandle)); pHandle->list = tdListNew(sizeof(SCfComparator)); taosThreadMutexInit(&pHandle->mutex, NULL); @@ -129,6 +131,7 @@ void* streamBackendInit(const char* path) { if (cfs != NULL) { rocksdb_list_column_families_destroy(cfs, nCf); } + qDebug("succ to init stream backend at %s, backend:%p", path, pHandle); return (void*)pHandle; _EXIT: @@ -141,6 +144,7 @@ _EXIT: rocksdb_compactionfilterfactory_destroy(pHandle->filterFactory); tdListFree(pHandle->list); free(pHandle); + qDebug("failed to init stream backend at %s", path); return NULL; } void streamBackendCleanup(void* arg) { @@ -180,7 +184,7 @@ void streamBackendCleanup(void* arg) { taosThreadMutexDestroy(&pHandle->cfMutex); taosMemoryFree(pHandle); - + qDebug("destroy stream backend backend:%p", pHandle); return; } SListNode* streamBackendAddCompare(void* backend, void* arg) { @@ -803,7 +807,8 @@ int32_t streamStateOpenBackendCf(void* backend, char* name, char** cfs, int32_t return 0; } int streamStateOpenBackend(void* backend, SStreamState* pState) { - qInfo("start to open backend, %p 0x%" PRIx64 "-%d", pState, pState->streamId, pState->taskId); + qInfo("start to open state %p on backend %p 0x%" PRIx64 "-%d", pState, backend, pState->streamId, pState->taskId); + taosAcquireRef(streamBackendId, pState->streamBackendRid); SBackendHandle* handle = backend; sprintf(pState->pTdbState->idstr, "0x%" PRIx64 "-%d", pState->streamId, pState->taskId); @@ -866,7 +871,7 @@ int streamStateOpenBackend(void* backend, SStreamState* pState) { SCfComparator compare = {.comp = pCompare, .numOfComp = cfLen}; pState->pTdbState->pComparNode = streamBackendAddCompare(handle, &compare); // rocksdb_writeoptions_disable_WAL(pState->pTdbState->writeOpts, 1); - qInfo("succ to open backend, %p, 0x%" PRIx64 "-%d", pState, pState->streamId, pState->taskId); + qInfo("succ to open state %p on backend, %p, 0x%" PRIx64 "-%d", pState, handle, pState->streamId, pState->taskId); return 0; } @@ -882,8 +887,8 @@ void streamStateCloseBackend(SStreamState* pState, bool remove) { taosThreadMutexUnlock(&pHandle->cfMutex); char* status[] = {"close", "drop"}; - qInfo("start to %s backend, %p, 0x%" PRIx64 "-%d", status[remove == false ? 0 : 1], pState, pState->streamId, - pState->taskId); + qInfo("start to %s state %p on backend %p 0x%" PRIx64 "-%d", status[remove == false ? 0 : 1], pState, pHandle, + pState->streamId, pState->taskId); if (pState->pTdbState->rocksdb == NULL) { return; } @@ -938,6 +943,7 @@ void streamStateCloseBackend(SStreamState* pState, bool remove) { taosThreadRwlockDestroy(&pState->pTdbState->rwLock); pState->pTdbState->rocksdb = NULL; + taosReleaseRef(streamBackendId, pState->streamBackendRid); } void streamStateDestroyCompar(void* arg) { SCfComparator* comp = (SCfComparator*)arg; diff --git a/source/libs/stream/src/streamMeta.c b/source/libs/stream/src/streamMeta.c index 8c26052fdb..ed9f99cf78 100644 --- a/source/libs/stream/src/streamMeta.c +++ b/source/libs/stream/src/streamMeta.c @@ -20,7 +20,7 @@ #include "ttimer.h" static TdThreadOnce streamMetaModuleInit = PTHREAD_ONCE_INIT; -static int32_t streamBackendId = 0; +int32_t streamBackendId = 0; static void streamMetaEnvInit() { streamBackendId = taosOpenRef(20, streamBackendCleanup); } void streamMetaInit() { taosThreadOnce(&streamMetaModuleInit, streamMetaEnvInit); } @@ -79,7 +79,6 @@ SStreamMeta* streamMetaOpen(const char* path, void* ahandle, FTaskExpand expandF pMeta->vgId = vgId; pMeta->ahandle = ahandle; pMeta->expandFunc = expandFunc; - pMeta->streamBackendId = streamBackendId; memset(streamPath, 0, len); sprintf(streamPath, "%s/%s", pMeta->path, "state"); diff --git a/source/libs/stream/src/streamState.c b/source/libs/stream/src/streamState.c index 71a21ac150..0b91936f53 100644 --- a/source/libs/stream/src/streamState.c +++ b/source/libs/stream/src/streamState.c @@ -106,7 +106,7 @@ SStreamState* streamStateOpen(char* path, void* pTask, bool specPath, int32_t sz } SStreamTask* pStreamTask = pTask; - char statePath[1024]; + char statePath[1024]; if (!specPath) { sprintf(statePath, "%s/%d", path, pStreamTask->id.taskId); } else { @@ -119,10 +119,10 @@ SStreamState* streamStateOpen(char* path, void* pTask, bool specPath, int32_t sz #ifdef USE_ROCKSDB SStreamMeta* pMeta = pStreamTask->pMeta; - taosAcquireRef(pMeta->streamBackendId, pMeta->streamBackendRid); + pState->streamBackendRid = pMeta->streamBackendRid; int code = streamStateOpenBackend(pMeta->streamBackend, pState); if (code == -1) { - taosReleaseRef(pMeta->streamBackendId, pMeta->streamBackendRid); + taosReleaseRef(streamBackendId, pMeta->streamBackendRid); taosMemoryFree(pState); pState = NULL; } @@ -224,7 +224,7 @@ void streamStateClose(SStreamState* pState, bool remove) { #ifdef USE_ROCKSDB // streamStateCloseBackend(pState); streamStateDestroy(pState, remove); - taosReleaseRef(pTask->pMeta->streamBackendId, pTask->pMeta->streamBackendRid); + //taosReleaseRef(pTask->pMeta->streamBackendId, pTask->pMeta->streamBackendRid); #else tdbCommit(pState->pTdbState->db, pState->pTdbState->txn); tdbPostCommit(pState->pTdbState->db, pState->pTdbState->txn); @@ -278,10 +278,10 @@ int32_t streamStateCommit(SStreamState* pState) { int32_t streamStateFuncPut(SStreamState* pState, const SWinKey* key, const void* value, int32_t vLen) { #ifdef USE_ROCKSDB - void* pVal = NULL; - int32_t len = 0; - int32_t code = getRowBuff(pState->pFileState, (void*)key, sizeof(SWinKey), &pVal, &len); - char* buf = ((SRowBuffPos*)pVal)->pRowBuff; + void* pVal = NULL; + int32_t len = 0; + int32_t code = getRowBuff(pState->pFileState, (void*)key, sizeof(SWinKey), &pVal, &len); + char* buf = ((SRowBuffPos*)pVal)->pRowBuff; uint32_t rowSize = streamFileStateGeSelectRowSize(pState->pFileState); memcpy(buf + len - rowSize, value, vLen); return code; @@ -291,10 +291,10 @@ int32_t streamStateFuncPut(SStreamState* pState, const SWinKey* key, const void* } int32_t streamStateFuncGet(SStreamState* pState, const SWinKey* key, void** ppVal, int32_t* pVLen) { #ifdef USE_ROCKSDB - void* pVal = NULL; - int32_t len = 0; - int32_t code = getRowBuff(pState->pFileState, (void*)key, sizeof(SWinKey), (void**)(&pVal), &len); - char* buf = ((SRowBuffPos*)pVal)->pRowBuff; + void* pVal = NULL; + int32_t len = 0; + int32_t code = getRowBuff(pState->pFileState, (void*)key, sizeof(SWinKey), (void**)(&pVal), &len); + char* buf = ((SRowBuffPos*)pVal)->pRowBuff; uint32_t rowSize = streamFileStateGeSelectRowSize(pState->pFileState); *ppVal = buf + len - rowSize; return code; From 3108ecf84be54b1eea9718591072230c5f3f03ca Mon Sep 17 00:00:00 2001 From: yihaoDeng Date: Mon, 5 Jun 2023 06:38:17 +0000 Subject: [PATCH 2/7] fix invalid free --- source/libs/stream/src/streamBackendRocksdb.c | 5 +++-- source/libs/stream/src/streamState.c | 2 -- 2 files changed, 3 insertions(+), 4 deletions(-) diff --git a/source/libs/stream/src/streamBackendRocksdb.c b/source/libs/stream/src/streamBackendRocksdb.c index d190f4b43e..9f1cab56e7 100644 --- a/source/libs/stream/src/streamBackendRocksdb.c +++ b/source/libs/stream/src/streamBackendRocksdb.c @@ -82,7 +82,7 @@ const char* comparePartagKeyName(void* name); void* streamBackendInit(const char* path) { qDebug("start to init stream backend at %s", path); - SBackendHandle* pHandle = calloc(1, sizeof(SBackendHandle)); + SBackendHandle* pHandle = taosMemoryCalloc(1, sizeof(SBackendHandle)); pHandle->list = tdListNew(sizeof(SCfComparator)); taosThreadMutexInit(&pHandle->mutex, NULL); taosThreadMutexInit(&pHandle->cfMutex, NULL); @@ -121,6 +121,7 @@ void* streamBackendInit(const char* path) { if (err != NULL) { qError("failed to open rocksdb, path:%s, reason:%s", path, err); taosMemoryFreeClear(err); + goto _EXIT; } } else { /* @@ -887,7 +888,7 @@ void streamStateCloseBackend(SStreamState* pState, bool remove) { taosThreadMutexUnlock(&pHandle->cfMutex); char* status[] = {"close", "drop"}; - qInfo("start to %s state %p on backend %p 0x%" PRIx64 "-%d", status[remove == false ? 0 : 1], pState, pHandle, + qInfo("start to close %s state %p on backend %p 0x%" PRIx64 "-%d", status[remove == false ? 0 : 1], pState, pHandle, pState->streamId, pState->taskId); if (pState->pTdbState->rocksdb == NULL) { return; diff --git a/source/libs/stream/src/streamState.c b/source/libs/stream/src/streamState.c index 0b91936f53..967c7733c9 100644 --- a/source/libs/stream/src/streamState.c +++ b/source/libs/stream/src/streamState.c @@ -222,9 +222,7 @@ _err: void streamStateClose(SStreamState* pState, bool remove) { SStreamTask* pTask = pState->pTdbState->pOwner; #ifdef USE_ROCKSDB - // streamStateCloseBackend(pState); streamStateDestroy(pState, remove); - //taosReleaseRef(pTask->pMeta->streamBackendId, pTask->pMeta->streamBackendRid); #else tdbCommit(pState->pTdbState->db, pState->pTdbState->txn); tdbPostCommit(pState->pTdbState->db, pState->pTdbState->txn); From c77f7f65d2eaa9b7442362fda71658289dbb89e3 Mon Sep 17 00:00:00 2001 From: yihaoDeng Date: Mon, 5 Jun 2023 07:32:28 +0000 Subject: [PATCH 3/7] fix invalid free --- source/libs/stream/src/streamBackendRocksdb.c | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/source/libs/stream/src/streamBackendRocksdb.c b/source/libs/stream/src/streamBackendRocksdb.c index 9f1cab56e7..3335881b37 100644 --- a/source/libs/stream/src/streamBackendRocksdb.c +++ b/source/libs/stream/src/streamBackendRocksdb.c @@ -173,15 +173,16 @@ void streamBackendCleanup(void* arg) { rocksdb_env_destroy(pHandle->env); rocksdb_cache_destroy(pHandle->cache); - taosThreadMutexDestroy(&pHandle->mutex); SListNode* head = tdListPopHead(pHandle->list); while (head != NULL) { streamStateDestroyCompar(head->data); taosMemoryFree(head); head = tdListPopHead(pHandle->list); } - // rocksdb_compactionfilterfactory_destroy(pHandle->filterFactory); + tdListFree(pHandle->list); + taosThreadMutexDestroy(&pHandle->mutex); + taosThreadMutexDestroy(&pHandle->cfMutex); taosMemoryFree(pHandle); From 63c4929089a80190670b2f5d3b15bd251bf36680 Mon Sep 17 00:00:00 2001 From: yihaoDeng Date: Mon, 5 Jun 2023 08:42:00 +0000 Subject: [PATCH 4/7] fix invalid free --- source/libs/stream/src/streamBackendRocksdb.c | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/source/libs/stream/src/streamBackendRocksdb.c b/source/libs/stream/src/streamBackendRocksdb.c index 3335881b37..df045eef20 100644 --- a/source/libs/stream/src/streamBackendRocksdb.c +++ b/source/libs/stream/src/streamBackendRocksdb.c @@ -144,7 +144,7 @@ _EXIT: taosHashCleanup(pHandle->cfInst); rocksdb_compactionfilterfactory_destroy(pHandle->filterFactory); tdListFree(pHandle->list); - free(pHandle); + taosMemoryFree(pHandle); qDebug("failed to init stream backend at %s", path); return NULL; } From 4f2f441816863fdad92a7e37b6ff5d77c46dde4a Mon Sep 17 00:00:00 2001 From: yihaoDeng Date: Mon, 5 Jun 2023 09:06:50 +0000 Subject: [PATCH 5/7] fix invalid free --- include/os/osMemory.h | 25 ++++++++++++------------- 1 file changed, 12 insertions(+), 13 deletions(-) diff --git a/include/os/osMemory.h b/include/os/osMemory.h index 18cd0d9cc6..683d10e926 100644 --- a/include/os/osMemory.h +++ b/include/os/osMemory.h @@ -22,21 +22,20 @@ extern "C" { // If the error is in a third-party library, place this header file under the third-party library header file. // When you want to use this feature, you should find or add the same function in the following sectio -// #if !defined(WINDOWS) +#if !defined(WINDOWS) -// #ifndef ALLOW_FORBID_FUNC -// #define malloc MALLOC_FUNC_TAOS_FORBID -// #define calloc CALLOC_FUNC_TAOS_FORBID -// #define realloc REALLOC_FUNC_TAOS_FORBID -// #define free FREE_FUNC_TAOS_FORBID -// #ifdef strdup -// #undef strdup -// #define strdup STRDUP_FUNC_TAOS_FORBID -// #endif -// #endif // ifndef ALLOW_FORBID_FUNC -// #endif // if !defined(WINDOWS) +#ifndef ALLOW_FORBID_FUNC +#define malloc MALLOC_FUNC_TAOS_FORBID +#define calloc CALLOC_FUNC_TAOS_FORBID +#define realloc REALLOC_FUNC_TAOS_FORBID +#define free FREE_FUNC_TAOS_FORBID +#ifdef strdup +#undef strdup +#define strdup STRDUP_FUNC_TAOS_FORBID +#endif +#endif // ifndef ALLOW_FORBID_FUNC +#endif // if !defined(WINDOWS) -// // #define taosMemoryFree malloc // #define taosMemoryMalloc malloc // #define taosMemoryCalloc calloc // #define taosMemoryRealloc realloc From c6242fee0132d496ac66297e412014f2fd3cf365 Mon Sep 17 00:00:00 2001 From: yihaoDeng Date: Mon, 5 Jun 2023 10:54:02 +0000 Subject: [PATCH 6/7] fix invalid free --- source/libs/stream/src/tstreamFileState.c | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/source/libs/stream/src/tstreamFileState.c b/source/libs/stream/src/tstreamFileState.c index bc84509728..bfaeca89f6 100644 --- a/source/libs/stream/src/tstreamFileState.c +++ b/source/libs/stream/src/tstreamFileState.c @@ -419,7 +419,7 @@ int32_t deleteExpiredCheckPoint(SStreamFileState* pFileState, TSKEY mark) { if (code != 0 || len == 0 || val == NULL) { return TSDB_CODE_FAILED; } - memcpy(val, buf, len); + memcpy(buf, val, len); buf[len] = 0; maxCheckPointId = atol((char*)buf); taosMemoryFree(val); @@ -433,7 +433,7 @@ int32_t deleteExpiredCheckPoint(SStreamFileState* pFileState, TSKEY mark) { if (code != 0) { return TSDB_CODE_FAILED; } - memcpy(val, buf, len); + memcpy(buf, val, len); buf[len] = 0; taosMemoryFree(val); From 66f483e51a52b155bb4677057bc2058f28121515 Mon Sep 17 00:00:00 2001 From: yihaoDeng Date: Tue, 6 Jun 2023 02:32:06 +0000 Subject: [PATCH 7/7] fix invalid free --- source/libs/stream/src/streamMeta.c | 3 +++ 1 file changed, 3 insertions(+) diff --git a/source/libs/stream/src/streamMeta.c b/source/libs/stream/src/streamMeta.c index ed9f99cf78..5c31b1dd60 100644 --- a/source/libs/stream/src/streamMeta.c +++ b/source/libs/stream/src/streamMeta.c @@ -89,6 +89,9 @@ SStreamMeta* streamMetaOpen(const char* path, void* ahandle, FTaskExpand expandF } pMeta->streamBackend = streamBackendInit(streamPath); + if (pMeta->streamBackend == NULL) { + goto _err; + } pMeta->streamBackendRid = taosAddRef(streamBackendId, pMeta->streamBackend); taosMemoryFree(streamPath);