xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

fix: global data sink manager issue

This commit is contained in:
dapan1121 2023-08-22 18:29:25 +08:00
parent b5bd8f7c23
commit fe4e45647e
5 changed files with 22 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
source/libs/executor/src/dataDeleter.c
View File

@ -224,6 +224,8 @@ static int32_t destroyDataSinker(SDataSinkHandle* pHandle) {
} }
taosCloseQueue(pDeleter->pDataBlocks); taosCloseQueue(pDeleter->pDataBlocks);
taosThreadMutexDestroy(&pDeleter->mutex); taosThreadMutexDestroy(&pDeleter->mutex);
taosMemoryFree(pDeleter->pManager);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }

1
source/libs/executor/src/dataDispatcher.c
View File

@ -226,6 +226,7 @@ static int32_t destroyDataSinker(SDataSinkHandle* pHandle) {
} }
taosCloseQueue(pDispatcher->pDataBlocks); taosCloseQueue(pDispatcher->pDataBlocks);
taosThreadMutexDestroy(&pDispatcher->mutex); taosThreadMutexDestroy(&pDispatcher->mutex);
taosMemoryFree(pDispatcher->pManager);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }

2
source/libs/executor/src/dataInserter.c
View File

@ -395,6 +395,8 @@ static int32_t destroyDataSinker(SDataSinkHandle* pHandle) {
taosMemoryFree(pInserter->pParam); taosMemoryFree(pInserter->pParam);
taosHashCleanup(pInserter->pCols); taosHashCleanup(pInserter->pCols);
taosThreadMutexDestroy(&pInserter->mutex); taosThreadMutexDestroy(&pInserter->mutex);
taosMemoryFree(pInserter->pManager);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }

22
source/libs/executor/src/dataSinkMgt.c
View File

@ -18,12 +18,17 @@
#include "planner.h" #include "planner.h"
#include "tarray.h" #include "tarray.h"
static SDataSinkManager gDataSinkManager = {0};
SDataSinkStat gDataSinkStat = {0}; SDataSinkStat gDataSinkStat = {0};
int32_t dsDataSinkMgtInit(SDataSinkMgtCfg* cfg, SStorageAPI* pAPI) { int32_t dsDataSinkMgtInit(SDataSinkMgtCfg* cfg, SStorageAPI* pAPI, void** ppSinkManager) {
gDataSinkManager.cfg = *cfg; SDataSinkManager* pSinkManager = taosMemoryMalloc(sizeof(SDataSinkManager));
gDataSinkManager.pAPI = pAPI; if (NULL == pSinkManager) {
return TSDB_CODE_OUT_OF_MEMORY;
}
pSinkManager->cfg = *cfg;
pSinkManager->pAPI = pAPI;
*ppSinkManager = pSinkManager;
return 0; // to avoid compiler eror return 0; // to avoid compiler eror
} }
@ -33,15 +38,16 @@ int32_t dsDataSinkGetCacheSize(SDataSinkStat* pStat) {
return 0; return 0;
} }
int32_t dsCreateDataSinker(const SDataSinkNode* pDataSink, DataSinkHandle* pHandle, void* pParam, const char* id) { int32_t dsCreateDataSinker(void* pSinkManager, const SDataSinkNode* pDataSink, DataSinkHandle* pHandle, void* pParam, const char* id) {
SDataSinkManager* pManager = pSinkManager;
switch ((int)nodeType(pDataSink)) { switch ((int)nodeType(pDataSink)) {
case QUERY_NODE_PHYSICAL_PLAN_DISPATCH: case QUERY_NODE_PHYSICAL_PLAN_DISPATCH:
return createDataDispatcher(&gDataSinkManager, pDataSink, pHandle); return createDataDispatcher(pManager, pDataSink, pHandle);
case QUERY_NODE_PHYSICAL_PLAN_DELETE: { case QUERY_NODE_PHYSICAL_PLAN_DELETE: {
return createDataDeleter(&gDataSinkManager, pDataSink, pHandle, pParam); return createDataDeleter(pManager, pDataSink, pHandle, pParam);
} }
case QUERY_NODE_PHYSICAL_PLAN_QUERY_INSERT: { case QUERY_NODE_PHYSICAL_PLAN_QUERY_INSERT: {
return createDataInserter(&gDataSinkManager, pDataSink, pHandle, pParam); return createDataInserter(pManager, pDataSink, pHandle, pParam);
} }
} }

5
source/libs/executor/src/executor.c
View File

@ -512,7 +512,8 @@ int32_t qCreateExecTask(SReadHandle* readHandle, int32_t vgId, uint64_t taskId,
} }
SDataSinkMgtCfg cfg = {.maxDataBlockNum = 500, .maxDataBlockNumPerQuery = 50}; SDataSinkMgtCfg cfg = {.maxDataBlockNum = 500, .maxDataBlockNumPerQuery = 50};
code = dsDataSinkMgtInit(&cfg, &(*pTask)->storageAPI); void* pSinkManager = NULL;
code = dsDataSinkMgtInit(&cfg, &(*pTask)->storageAPI, &pSinkManager);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
qError("failed to dsDataSinkMgtInit, code:%s, %s", tstrerror(code), (*pTask)->id.str); qError("failed to dsDataSinkMgtInit, code:%s, %s", tstrerror(code), (*pTask)->id.str);
goto _error; goto _error;
@ -527,7 +528,7 @@ int32_t qCreateExecTask(SReadHandle* readHandle, int32_t vgId, uint64_t taskId,
} }
// pSinkParam has been freed during create sinker. // pSinkParam has been freed during create sinker.
code = dsCreateDataSinker(pSubplan->pDataSink, handle, pSinkParam, (*pTask)->id.str); code = dsCreateDataSinker(pSinkManager, pSubplan->pDataSink, handle, pSinkParam, (*pTask)->id.str);
} }
qDebug("subplan task create completed, TID:0x%" PRIx64 " QID:0x%" PRIx64, taskId, pSubplan->id.queryId); qDebug("subplan task create completed, TID:0x%" PRIx64 " QID:0x%" PRIx64, taskId, pSubplan->id.queryId);