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

fix(tsdb) : fix memory leak.

This commit is contained in:
Haojun Liao 2023-06-21 16:53:47 +08:00
parent 84bd2fb7a1
commit e76ae4ffc5
1 changed files with 97 additions and 75 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

172
source/dnode/vnode/src/tsdb/tsdbRead2.c
View File

@ -967,14 +967,53 @@ static void cleanupTableScanInfo(SReaderStatus* pStatus) {
doCleanupTableScanInfo(*px); doCleanupTableScanInfo(*px);
} }
}
// pStatus->mapDataCleaned = true; typedef struct SBrinRecordIter {
SArray* pBrinBlockList;
SBrinBlk* pCurrentBlk;
int32_t blockIndex;
int32_t recordIndex;
SDataFileReader* pReader;
SBrinBlock block;
SBrinRecord record;
} SBrinRecordIter;
void initBrinRecordIter(SBrinRecordIter* pIter, SDataFileReader* pReader, SArray* pList) {
memset(&pIter->block, 0, sizeof(SBrinBlock));
memset(&pIter->record, 0, sizeof(SBrinRecord));
pIter->blockIndex = -1;
pIter->recordIndex = -1;
pIter->pReader = pReader;
pIter->pBrinBlockList = pList;
}
SBrinRecord* getNextBrinRecord(SBrinRecordIter* pIter) {
if (pIter->blockIndex == -1 || (pIter->recordIndex + 1) >= TARRAY2_SIZE(pIter->block.numRow)) {
pIter->blockIndex += 1;
if (pIter->blockIndex >= taosArrayGetSize(pIter->pBrinBlockList)) {
return NULL;
}
pIter->pCurrentBlk = taosArrayGet(pIter->pBrinBlockList, pIter->blockIndex);
tBrinBlockClear(&pIter->block);
tsdbDataFileReadBrinBlock(pIter->pReader, pIter->pCurrentBlk, &pIter->block);
pIter->recordIndex = -1;
}
pIter->recordIndex += 1;
tBrinBlockGet(&pIter->block, pIter->recordIndex, &pIter->record);
return &pIter->record;
}
void clearBrinBlockIter(SBrinRecordIter* pIter) {
tBrinBlockDestroy(&pIter->block);
} }
static int32_t doLoadFileBlock(STsdbReader* pReader, SArray* pIndexList, SBlockNumber* pBlockNum, SArray* pTableScanInfoList) { static int32_t doLoadFileBlock(STsdbReader* pReader, SArray* pIndexList, SBlockNumber* pBlockNum, SArray* pTableScanInfoList) {
size_t sizeInDisk = 0; size_t sizeInDisk = 0;
size_t numOfBlocks = taosArrayGetSize(pIndexList);
int64_t st = taosGetTimestampUs(); int64_t st = taosGetTimestampUs();
cleanupTableScanInfo(&pReader->status); cleanupTableScanInfo(&pReader->status);
@ -982,26 +1021,40 @@ static int32_t doLoadFileBlock(STsdbReader* pReader, SArray* pIndexList, SBlockN
int32_t i = 0, k = 0; int32_t i = 0, k = 0;
int32_t numOfTables = tSimpleHashGetSize(pReader->status.pTableMap); int32_t numOfTables = tSimpleHashGetSize(pReader->status.pTableMap);
while(i < numOfBlocks && k < numOfTables) { int32_t step = ASCENDING_TRAVERSE(pReader->order) ? 1 : -1;
SBrinBlk* pBlk = taosArrayGet(pIndexList, i); STimeWindow w = pReader->window;
SBrinRecordIter iter = {0};
initBrinRecordIter(&iter, pReader->pFileReader, pIndexList);
SBrinRecord* pRecord = NULL;
while (k < numOfTables) {
pRecord = getNextBrinRecord(&iter);
uint64_t uid = pReader->status.uidList.tableUidList[k]; uint64_t uid = pReader->status.uidList.tableUidList[k];
if (pRecord == NULL || pRecord->suid > pReader->suid) {
break;
}
SBrinBlock block = {0}; if (pRecord->suid < pReader->suid) {
tsdbDataFileReadBrinBlock(pReader->pFileReader, pBlk, &block);
// tMapDataReset(&pScanInfo->mapData);
// tsdbReadDataBlk(pReader->pFileReader, pBlockIdx, &pScanInfo->mapData);
// taosArrayEnsureCap(pScanInfo->pBlockList, pScanInfo->mapData.nItem);
// todo set the correct size
sizeInDisk += 0;//pScanInfo->mapData.nData;
int32_t step = ASCENDING_TRAVERSE(pReader->order) ? 1 : -1;
STimeWindow w = pReader->window;
if (isEmptyQueryTimeWindow(&w)) {
continue; continue;
} }
ASSERT(pRecord->suid == pReader->suid);
if (pRecord->uid < uid) {
continue;
}
while (pRecord->uid > uid && k < numOfTables) {
k += 1;
}
if (k >= numOfTables) {
break;
}
ASSERT(pRecord->suid == pReader->suid && uid == pRecord->uid);
STableBlockScanInfo* pScanInfo = getTableBlockScanInfo(pReader->status.pTableMap, uid, pReader->idStr); STableBlockScanInfo* pScanInfo = getTableBlockScanInfo(pReader->status.pTableMap, uid, pReader->idStr);
if (ASCENDING_TRAVERSE(pReader->order)) { if (ASCENDING_TRAVERSE(pReader->order)) {
w.skey = pScanInfo->lastKey + step; w.skey = pScanInfo->lastKey + step;
@ -1009,69 +1062,39 @@ static int32_t doLoadFileBlock(STsdbReader* pReader, SArray* pIndexList, SBlockN
w.ekey = pScanInfo->lastKey + step; w.ekey = pScanInfo->lastKey + step;
} }
SBrinRecord record = {0}; if (isEmptyQueryTimeWindow(&w)) {
for (int32_t j = 0; j < TARRAY2_SIZE(block.numRow); ++j) { k += 1;
tBrinBlockGet(&block, j, &record); continue;
if (record.suid < pReader->suid) {
continue;
}
if (record.suid > pReader->suid) {
break;
}
{
while (record.uid > uid && (k + 1) < numOfTables) {
k += 1;
uid = pReader->status.uidList.tableUidList[k];
pScanInfo = getTableBlockScanInfo(pReader->status.pTableMap, uid, pReader->idStr);
if (ASCENDING_TRAVERSE(pReader->order)) {
w.skey = pScanInfo->lastKey + step;
} else {
w.ekey = pScanInfo->lastKey + step;
}
}
if (k >= numOfTables) {
break;
}
if (record.uid < uid) {
continue;
}
}
ASSERT(record.suid == pReader->suid);
// 1. time range check
if (record.firstKey > w.ekey || record.lastKey < w.skey) {
continue;
}
// 2. version range check
if (record.firstKeyVer > pReader->verRange.maxVer || record.lastKeyVer < pReader->verRange.minVer) {
continue;
}
// SBlockIndex bIndex = {.ordinalIndex = j, .inFileOffset = record.blockOffset};
// bIndex.window = (STimeWindow){.skey = record.firstKey, .ekey = record.lastKey};
void* p1 = taosArrayPush(pScanInfo->pBlockList, &record);
if (p1 == NULL) {
return TSDB_CODE_OUT_OF_MEMORY;
}
pBlockNum->numOfBlocks += 1;
} }
i += 1; // 1. time range check
if (pRecord->firstKey > w.ekey || pRecord->lastKey < w.skey) {
continue;
}
STableBlockScanInfo** p = taosArrayGetLast(pTableScanInfoList); // 2. version range check
if ((p == NULL || (*p)->uid != uid) && taosArrayGetSize(pScanInfo->pBlockList) > 0) { if (pRecord->firstKeyVer > pReader->verRange.maxVer || pRecord->lastKeyVer < pReader->verRange.minVer) {
continue;
}
void* p1 = taosArrayPush(pScanInfo->pBlockList, pRecord);
if (p1 == NULL) {
return TSDB_CODE_OUT_OF_MEMORY;
}
pBlockNum->numOfBlocks += 1;
if (taosArrayGetSize(pTableScanInfoList) == 0) {
taosArrayPush(pTableScanInfoList, &pScanInfo); taosArrayPush(pTableScanInfoList, &pScanInfo);
} else {
STableBlockScanInfo** p = taosArrayGetLast(pTableScanInfoList);
if ((*p)->uid != uid) {
taosArrayPush(pTableScanInfoList, &pScanInfo);
}
} }
} }
clearBrinBlockIter(&iter);
pBlockNum->numOfLastFiles = pReader->status.pCurrentFileset->lvlArr->size; pBlockNum->numOfLastFiles = pReader->status.pCurrentFileset->lvlArr->size;
int32_t total = pBlockNum->numOfLastFiles + pBlockNum->numOfBlocks; int32_t total = pBlockNum->numOfLastFiles + pBlockNum->numOfBlocks;
@ -1336,7 +1359,6 @@ static int32_t copyBlockDataToSDataBlock(STsdbReader* pReader) {
SBlockData* pBlockData = &pStatus->fileBlockData; SBlockData* pBlockData = &pStatus->fileBlockData;
SFileDataBlockInfo* pBlockInfo = getCurrentBlockInfo(pBlockIter); SFileDataBlockInfo* pBlockInfo = getCurrentBlockInfo(pBlockIter);
// SDataBlk* pBlock = getCurrentBlock(pBlockIter);
SSDataBlock* pResBlock = pReader->resBlockInfo.pResBlock; SSDataBlock* pResBlock = pReader->resBlockInfo.pResBlock;
int32_t numOfOutputCols = pSupInfo->numOfCols; int32_t numOfOutputCols = pSupInfo->numOfCols;
int32_t code = TSDB_CODE_SUCCESS; int32_t code = TSDB_CODE_SUCCESS;