From c9aa59cbb9c1d37b591dc6a562e62c1314673e20 Mon Sep 17 00:00:00 2001 From: Haojun Liao Date: Thu, 8 Jun 2023 18:59:25 +0800 Subject: [PATCH] refactor: do some internal refactor. --- include/libs/executor/storageapi.h | 2 +- include/libs/stream/tstreamFileState.h | 3 +- source/libs/executor/src/timewindowoperator.c | 8 ++-- source/libs/stream/src/tstreamFileState.c | 39 ++++++++++++++----- 4 files changed, 37 insertions(+), 15 deletions(-) diff --git a/include/libs/executor/storageapi.h b/include/libs/executor/storageapi.h index e9340a33c3..7f3039829a 100644 --- a/include/libs/executor/storageapi.h +++ b/include/libs/executor/storageapi.h @@ -367,7 +367,7 @@ typedef struct SStateStore { SStreamStateCur* (*streamStateSessionSeekKeyCurrentNext)(SStreamState* pState, const SSessionKey* key); struct SStreamFileState* (*streamFileStateInit)(int64_t memSize, uint32_t keySize, uint32_t rowSize, - uint32_t selectRowSize, GetTsFun fp, void* pFile, TSKEY delMark); + uint32_t selectRowSize, GetTsFun fp, void* pFile, TSKEY delMark, const char*id); void (*streamFileStateDestroy)(struct SStreamFileState* pFileState); void (*streamFileStateClear)(struct SStreamFileState* pFileState); diff --git a/include/libs/stream/tstreamFileState.h b/include/libs/stream/tstreamFileState.h index 0dbacf6c9f..8496b0ea62 100644 --- a/include/libs/stream/tstreamFileState.h +++ b/include/libs/stream/tstreamFileState.h @@ -28,11 +28,10 @@ extern "C" { #endif typedef struct SStreamFileState SStreamFileState; - typedef SList SStreamSnapshot; SStreamFileState* streamFileStateInit(int64_t memSize, uint32_t keySize, uint32_t rowSize, uint32_t selectRowSize, - GetTsFun fp, void* pFile, TSKEY delMark); + GetTsFun fp, void* pFile, TSKEY delMark, const char* id); void streamFileStateDestroy(SStreamFileState* pFileState); void streamFileStateClear(SStreamFileState* pFileState); bool needClearDiskBuff(SStreamFileState* pFileState); diff --git a/source/libs/executor/src/timewindowoperator.c b/source/libs/executor/src/timewindowoperator.c index 3a83472079..4f6e7a62dc 100644 --- a/source/libs/executor/src/timewindowoperator.c +++ b/source/libs/executor/src/timewindowoperator.c @@ -2785,7 +2785,7 @@ SOperatorInfo* createStreamFinalIntervalOperatorInfo(SOperatorInfo* downstream, pInfo->pUpdatedMap = NULL; int32_t funResSize= getMaxFunResSize(&pOperator->exprSupp, numOfCols); pInfo->pState->pFileState = pAPI->stateStore.streamFileStateInit(tsStreamBufferSize, sizeof(SWinKey), pInfo->aggSup.resultRowSize, funResSize, - compareTs, pInfo->pState, pInfo->twAggSup.deleteMark); + compareTs, pInfo->pState, pInfo->twAggSup.deleteMark, GET_TASKID(pTaskInfo)); pInfo->dataVersion = 0; pInfo->statestore = pTaskInfo->storageAPI.stateStore; pInfo->recvGetAll = false; @@ -4974,8 +4974,10 @@ SOperatorInfo* createStreamIntervalOperatorInfo(SOperatorInfo* downstream, SPhys pInfo->pUpdated = NULL; pInfo->pUpdatedMap = NULL; int32_t funResSize= getMaxFunResSize(pSup, numOfCols); - pInfo->pState->pFileState = pTaskInfo->storageAPI.stateStore.streamFileStateInit(tsStreamBufferSize, sizeof(SWinKey), pInfo->aggSup.resultRowSize, funResSize, - compareTs, pInfo->pState, pInfo->twAggSup.deleteMark); + + pInfo->pState->pFileState = pTaskInfo->storageAPI.stateStore.streamFileStateInit( + tsStreamBufferSize, sizeof(SWinKey), pInfo->aggSup.resultRowSize, funResSize, compareTs, pInfo->pState, + pInfo->twAggSup.deleteMark, GET_TASKID(pTaskInfo)); setOperatorInfo(pOperator, "StreamIntervalOperator", QUERY_NODE_PHYSICAL_PLAN_STREAM_INTERVAL, true, OP_NOT_OPENED, pInfo, pTaskInfo); diff --git a/source/libs/stream/src/tstreamFileState.c b/source/libs/stream/src/tstreamFileState.c index 87a0aa94fd..77440b968b 100644 --- a/source/libs/stream/src/tstreamFileState.c +++ b/source/libs/stream/src/tstreamFileState.c @@ -43,12 +43,13 @@ struct SStreamFileState { uint64_t maxRowCount; uint64_t curRowCount; GetTsFun getTs; + char* id; }; typedef SRowBuffPos SRowBuffInfo; SStreamFileState* streamFileStateInit(int64_t memSize, uint32_t keySize, uint32_t rowSize, uint32_t selectRowSize, - GetTsFun fp, void* pFile, TSKEY delMark) { + GetTsFun fp, void* pFile, TSKEY delMark, const char* idstr) { if (memSize <= 0) { memSize = DEFAULT_MAX_STREAM_BUFFER_SIZE; } @@ -70,6 +71,7 @@ SStreamFileState* streamFileStateInit(int64_t memSize, uint32_t keySize, uint32_ if (!pFileState->usedBuffs || !pFileState->freeBuffs || !pFileState->rowBuffMap) { goto _error; } + pFileState->keyLen = keySize; pFileState->rowSize = rowSize; pFileState->selectivityRowSize = selectRowSize; @@ -81,6 +83,8 @@ SStreamFileState* streamFileStateInit(int64_t memSize, uint32_t keySize, uint32_ pFileState->deleteMark = delMark; pFileState->flushMark = INT64_MIN; pFileState->maxTs = INT64_MIN; + pFileState->id = taosStrdup(idstr); + recoverSnapshot(pFileState); return pFileState; @@ -124,6 +128,8 @@ void streamFileStateDestroy(SStreamFileState* pFileState) { if (!pFileState) { return; } + + taosMemoryFree(pFileState->id); tdListFreeP(pFileState->usedBuffs, destroyRowBuffAllPosPtr); tdListFreeP(pFileState->freeBuffs, destroyRowBuff); tSimpleHashCleanup(pFileState->rowBuffMap); @@ -177,7 +183,8 @@ void popUsedBuffs(SStreamFileState* pFileState, SStreamSnapshot* pFlushList, uin i++; } } - qInfo("do stream state flush %d rows to disck. is used: %d", listNEles(pFlushList), used); + + qInfo("stream state flush %d rows to disk. is used:%d", listNEles(pFlushList), used); } int32_t flushRowBuff(SStreamFileState* pFileState) { @@ -185,13 +192,17 @@ int32_t flushRowBuff(SStreamFileState* pFileState) { if (!pFlushList) { return TSDB_CODE_OUT_OF_MEMORY; } + uint64_t num = (uint64_t)(pFileState->curRowCount * FLUSH_RATIO); num = TMAX(num, FLUSH_NUM); popUsedBuffs(pFileState, pFlushList, num, false); + if (isListEmpty(pFlushList)) { popUsedBuffs(pFileState, pFlushList, num, true); } + flushSnapshot(pFileState, pFlushList, false); + SListIter fIter = {0}; tdListInitIter(pFlushList, &fIter, TD_LIST_FORWARD); SListNode* pNode = NULL; @@ -201,6 +212,7 @@ int32_t flushRowBuff(SStreamFileState* pFileState) { tdListAppend(pFileState->freeBuffs, &pPos->pRowBuff); pPos->pRowBuff = NULL; } + tdListFreeP(pFlushList, destroyRowBuffPosPtr); return TSDB_CODE_SUCCESS; } @@ -269,13 +281,13 @@ int32_t getRowBuff(SStreamFileState* pFileState, void* pKey, int32_t keyLen, voi TSKEY ts = pFileState->getTs(pKey); if (ts > pFileState->maxTs - pFileState->deleteMark && ts < pFileState->flushMark) { int32_t len = 0; - void* pVal = NULL; - int32_t code = streamStateGet_rocksdb(pFileState->pFileStore, pKey, &pVal, &len); + void* p = NULL; + int32_t code = streamStateGet_rocksdb(pFileState->pFileStore, pKey, &p, &len); qDebug("===stream===get %" PRId64 " from disc, res %d", ts, code); if (code == TSDB_CODE_SUCCESS) { - memcpy(pNewPos->pRowBuff, pVal, len); + memcpy(pNewPos->pRowBuff, p, len); } - taosMemoryFree(pVal); + taosMemoryFree(p); } tSimpleHashPut(pFileState->rowBuffMap, pKey, keyLen, &pNewPos, POINTER_BYTES); @@ -348,12 +360,16 @@ int32_t flushSnapshot(SStreamFileState* pFileState, SStreamSnapshot* pSnapshot, tdListInitIter(pSnapshot, &iter, TD_LIST_FORWARD); const int32_t BATCH_LIMIT = 256; - SListNode* pNode = NULL; + + int64_t st = taosGetTimestampMs(); + int32_t numOfElems = listNEles(pSnapshot); + SListNode* pNode = NULL; void* batch = streamStateCreateBatch(); while ((pNode = tdListNext(&iter)) != NULL && code == TSDB_CODE_SUCCESS) { SRowBuffPos* pPos = *(SRowBuffPos**)pNode->data; ASSERT(pPos->pRowBuff && pFileState->rowSize > 0); + if (streamStateGetBatchSize(batch) >= BATCH_LIMIT) { code = streamStatePutBatch_rocksdb(pFileState->pFileStore, batch); streamStateClearBatch(batch); @@ -361,7 +377,8 @@ int32_t flushSnapshot(SStreamFileState* pFileState, SStreamSnapshot* pSnapshot, SStateKey sKey = {.key = *((SWinKey*)pPos->pKey), .opNum = ((SStreamState*)pFileState->pFileStore)->number}; code = streamStatePutBatch(pFileState->pFileStore, "state", batch, &sKey, pPos->pRowBuff, pFileState->rowSize, 0); - qDebug("===stream===put %" PRId64 " to disc, code:%d, size:%d", sKey.key.ts, code, pFileState->rowSize); + // todo handle failure +// qDebug("===stream===put %" PRId64 " to disc, code:%d, size:%d", sKey.key.ts, code, pFileState->rowSize); } if (streamStateGetBatchSize(batch) > 0) { @@ -370,6 +387,10 @@ int32_t flushSnapshot(SStreamFileState* pFileState, SStreamSnapshot* pSnapshot, streamStateClearBatch(batch); + int64_t elapsed = taosGetTimestampMs() - st; + qDebug("%s flush to disk in batch model completed, rows:%d, batch size:%d, elapsed time:%"PRId64"ms", pFileState->id, numOfElems, + BATCH_LIMIT, elapsed); + if (flushState) { const char* taskKey = "streamFileState"; { @@ -391,8 +412,8 @@ int32_t flushSnapshot(SStreamFileState* pFileState, SStreamSnapshot* pSnapshot, } streamStatePutBatch_rocksdb(pFileState->pFileStore, batch); } - streamStateDestroyBatch(batch); + streamStateDestroyBatch(batch); return code; }