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

optimize clear expired state

This commit is contained in:
54liuyao 2024-10-22 17:55:05 +08:00
parent 657c29b4e0
commit c17bed9807
1 changed files with 9 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
source/libs/stream/src/tstreamFileState.c
View File

@ -1220,20 +1220,22 @@ void clearExpiredState(SStreamFileState* pFileState) {
SSHashObj* pSearchBuff = pFileState->searchBuff; SSHashObj* pSearchBuff = pFileState->searchBuff;
void* pIte = NULL; void* pIte = NULL;
int32_t iter = 0; int32_t iter = 0;
while ((pIte = tSimpleHashIterate(pFileState->searchBuff, pIte, &iter)) != NULL) { while ((pIte = tSimpleHashIterate(pSearchBuff, pIte, &iter)) != NULL) {
SArray* pWinStates = *((void**)pIte); SArray* pWinStates = *((void**)pIte);
int32_t size = taosArrayGetSize(pWinStates); int32_t size = taosArrayGetSize(pWinStates);
for (int32_t i = 0; i < size - 1; i++) { for (int32_t i = 0; i < size - 1; i++) {
SWinKey* pKey = taosArrayGet(pWinStates, i); SWinKey* pKey = taosArrayGet(pWinStates, i);
int32_t code_buff = pFileState->stateBuffRemoveFn(pFileState->rowStateBuff, pKey, sizeof(SWinKey)); int32_t code_buff = pFileState->stateBuffRemoveFn(pFileState->rowStateBuff, pKey, sizeof(SWinKey));
qTrace("%s at line %d res:%d", __func__, __LINE__, code_buff); qTrace("clear expired buff, ts:%" PRId64 ". %s at line %d res:%d", pKey->ts, __func__, __LINE__, code_buff);
int32_t code_file = pFileState->stateFileRemoveFn(pFileState, pKey); if (isFlushedState(pFileState, pKey->ts, 0)) {
qTrace("%s at line %d res:%d", __func__, __LINE__, code_file); int32_t code_file = pFileState->stateFileRemoveFn(pFileState, pKey);
qTrace("clear expired file, ts:%" PRId64 ". %s at line %d res:%d", pKey->ts, __func__, __LINE__, code_file);
}
if (pFileState->hasFillCatch == false || isFlushedState(pFileState, pKey->ts, 0)) { if (pFileState->hasFillCatch == false) {
code_file = streamStateFillDel_rocksdb(pFileState->pFileStore, pKey); int32_t code_file = streamStateFillDel_rocksdb(pFileState->pFileStore, pKey);
qTrace("state fill delete.%s at line %d res %d", __func__, __LINE__, code_file); qTrace("force clear expired file, ts:%" PRId64 ". %s at line %d res %d", pKey->ts, __func__, __LINE__, code_file);
} }
} }
taosArrayRemoveBatch(pWinStates, 0, size - 1, NULL); taosArrayRemoveBatch(pWinStates, 0, size - 1, NULL);