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

refactor stream code

This commit is contained in:
Yihao Deng 2024-01-08 08:38:16 +00:00
parent 1c28f53525
commit 4c5441da7e
4 changed files with 77 additions and 75 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
source/libs/stream/src/streamMeta.c
View File

@ -201,7 +201,7 @@ int32_t streamMetaCvtDbFormat(SStreamMeta* pMeta) {
void* key = taosHashGetKey(pIter, NULL); void* key = taosHashGetKey(pIter, NULL);
code = streamStateCvtDataFormat(pMeta->path, key, *(void**)pIter); code = streamStateCvtDataFormat(pMeta->path, key, *(void**)pIter);
if (code != 0) { if (code != 0) {
qError("failed to cvt data"); stError("failed to cvt data");
goto _EXIT; goto _EXIT;
} }
@ -225,11 +225,11 @@ int32_t streamMetaMayCvtDbFormat(SStreamMeta* pMeta) {
if (compatible == STREAM_STATA_COMPATIBLE) { if (compatible == STREAM_STATA_COMPATIBLE) {
return 0; return 0;
} else if (compatible == STREAM_STATA_NEED_CONVERT) { } else if (compatible == STREAM_STATA_NEED_CONVERT) {
qInfo("stream state need covert backend format"); stInfo("stream state need covert backend format");
return streamMetaCvtDbFormat(pMeta); return streamMetaCvtDbFormat(pMeta);
} else if (compatible == STREAM_STATA_NO_COMPATIBLE) { } else if (compatible == STREAM_STATA_NO_COMPATIBLE) {
qError( stError(
"stream read incompatible data, rm %s/vnode/vnode*/tq/stream if taosd cannot start, and rebuild stream " "stream read incompatible data, rm %s/vnode/vnode*/tq/stream if taosd cannot start, and rebuild stream "
"manually", "manually",
tsDataDir); tsDataDir);

30
source/libs/stream/src/streamSessionState.c
View File

@ -17,12 +17,9 @@
#include "query.h" #include "query.h"
#include "streamBackendRocksdb.h" #include "streamBackendRocksdb.h"
#include "taos.h"
#include "tcommon.h" #include "tcommon.h"
#include "thash.h"
#include "tsimplehash.h" #include "tsimplehash.h"
typedef int (*__session_compare_fn_t)(const SSessionKey* pWin, const void* pDatas, int pos); typedef int (*__session_compare_fn_t)(const SSessionKey* pWin, const void* pDatas, int pos);
int sessionStateKeyCompare(const SSessionKey* pWin1, const void* pDatas, int pos) { int sessionStateKeyCompare(const SSessionKey* pWin1, const void* pDatas, int pos) {
@ -80,7 +77,8 @@ static SRowBuffPos* addNewSessionWindow(SStreamFileState* pFileState, SArray* pW
return pNewPos; return pNewPos;
} }
static SRowBuffPos* insertNewSessionWindow(SStreamFileState* pFileState, SArray* pWinInfos, const SSessionKey* pKey, int32_t index) { static SRowBuffPos* insertNewSessionWindow(SStreamFileState* pFileState, SArray* pWinInfos, const SSessionKey* pKey,
int32_t index) {
SRowBuffPos* pNewPos = getNewRowPosForWrite(pFileState); SRowBuffPos* pNewPos = getNewRowPosForWrite(pFileState);
ASSERT(pNewPos->pRowBuff); ASSERT(pNewPos->pRowBuff);
memcpy(pNewPos->pKey, pKey, sizeof(SSessionKey)); memcpy(pNewPos->pKey, pKey, sizeof(SSessionKey));
@ -97,7 +95,8 @@ SRowBuffPos* createSessionWinBuff(SStreamFileState* pFileState, SSessionKey* pKe
return pNewPos; return pNewPos;
} }
int32_t getSessionWinResultBuff(SStreamFileState* pFileState, SSessionKey* pKey, TSKEY gap, void** pVal, int32_t* pVLen) { int32_t getSessionWinResultBuff(SStreamFileState* pFileState, SSessionKey* pKey, TSKEY gap, void** pVal,
int32_t* pVLen) {
int32_t code = TSDB_CODE_SUCCESS; int32_t code = TSDB_CODE_SUCCESS;
SSHashObj* pSessionBuff = getRowStateBuff(pFileState); SSHashObj* pSessionBuff = getRowStateBuff(pFileState);
SArray* pWinStates = NULL; SArray* pWinStates = NULL;
@ -270,7 +269,8 @@ int32_t deleteSessionWinStateBuffByPosFn(SStreamFileState* pFileState, SRowBuffP
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
int32_t allocSessioncWinBuffByNextPosition(SStreamFileState* pFileState, SStreamStateCur* pCur, const SSessionKey* pWinKey, void** ppVal, int32_t* pVLen) { int32_t allocSessioncWinBuffByNextPosition(SStreamFileState* pFileState, SStreamStateCur* pCur,
const SSessionKey* pWinKey, void** ppVal, int32_t* pVLen) {
SRowBuffPos* pNewPos = NULL; SRowBuffPos* pNewPos = NULL;
SSHashObj* pSessionBuff = getRowStateBuff(pFileState); SSHashObj* pSessionBuff = getRowStateBuff(pFileState);
void** ppBuff = tSimpleHashGet(pSessionBuff, &pWinKey->groupId, sizeof(uint64_t)); void** ppBuff = tSimpleHashGet(pSessionBuff, &pWinKey->groupId, sizeof(uint64_t));
@ -395,10 +395,12 @@ static void transformCursor(SStreamFileState* pFileState, SStreamStateCur* pCur)
pCur->pStreamFileState = pFileState; pCur->pStreamFileState = pFileState;
} }
static void checkAndTransformCursor(SStreamFileState* pFileState, const uint64_t groupId, SArray* pWinStates, SStreamStateCur** ppCur) { static void checkAndTransformCursor(SStreamFileState* pFileState, const uint64_t groupId, SArray* pWinStates,
SStreamStateCur** ppCur) {
SSessionKey key = {.groupId = groupId}; SSessionKey key = {.groupId = groupId};
int32_t code = streamStateSessionGetKVByCur_rocksdb(*ppCur, &key, NULL, NULL); int32_t code = streamStateSessionGetKVByCur_rocksdb(*ppCur, &key, NULL, NULL);
if (taosArrayGetSize(pWinStates) > 0 && (code == TSDB_CODE_FAILED || sessionStateKeyCompare(&key, pWinStates, 0) >= 0)) { if (taosArrayGetSize(pWinStates) > 0 &&
(code == TSDB_CODE_FAILED || sessionStateKeyCompare(&key, pWinStates, 0) >= 0)) {
if (!(*ppCur)) { if (!(*ppCur)) {
(*ppCur) = createStreamStateCursor(); (*ppCur) = createStreamStateCursor();
} }
@ -467,7 +469,8 @@ int32_t sessionWinStateGetKVByCur(SStreamStateCur* pCur, SSessionKey* pKey, void
} else { } else {
void* pData = NULL; void* pData = NULL;
code = streamStateSessionGetKVByCur_rocksdb(pCur, pKey, &pData, pVLen); code = streamStateSessionGetKVByCur_rocksdb(pCur, pKey, &pData, pVLen);
if (taosArrayGetSize(pWinStates) > 0 && (code == TSDB_CODE_FAILED || sessionStateKeyCompare(pKey, pWinStates, 0) >= 0)) { if (taosArrayGetSize(pWinStates) > 0 &&
(code == TSDB_CODE_FAILED || sessionStateKeyCompare(pKey, pWinStates, 0) >= 0)) {
transformCursor(pCur->pStreamFileState, pCur); transformCursor(pCur->pStreamFileState, pCur);
SRowBuffPos* pPos = taosArrayGetP(pWinStates, pCur->buffIndex); SRowBuffPos* pPos = taosArrayGetP(pWinStates, pCur->buffIndex);
if (pVal) { if (pVal) {
@ -560,7 +563,8 @@ int32_t getStateWinResultBuff(SStreamFileState* pFileState, SSessionKey* key, ch
if (code_file == TSDB_CODE_SUCCESS || isFlushedState(pFileState, endTs, 0)) { if (code_file == TSDB_CODE_SUCCESS || isFlushedState(pFileState, endTs, 0)) {
(*pVal) = createSessionWinBuff(pFileState, pWinKey, p, pVLen); (*pVal) = createSessionWinBuff(pFileState, pWinKey, p, pVLen);
code = code_file; code = code_file;
qDebug("===stream===0 get state win:%" PRId64 ",%" PRId64 " from disc, res %d", pWinKey->win.skey, pWinKey->win.ekey, code_file); qDebug("===stream===0 get state win:%" PRId64 ",%" PRId64 " from disc, res %d", pWinKey->win.skey,
pWinKey->win.ekey, code_file);
} else { } else {
(*pVal) = addNewSessionWindow(pFileState, pWinStates, key); (*pVal) = addNewSessionWindow(pFileState, pWinStates, key);
code = TSDB_CODE_FAILED; code = TSDB_CODE_FAILED;
@ -589,7 +593,8 @@ int32_t getStateWinResultBuff(SStreamFileState* pFileState, SSessionKey* key, ch
if (index + 1 < size) { if (index + 1 < size) {
pPos = taosArrayGetP(pWinStates, index + 1); pPos = taosArrayGetP(pWinStates, index + 1);
void* stateKey = (char*)(pPos->pRowBuff) + (valSize - keyDataLen); void* stateKey = (char*)(pPos->pRowBuff) + (valSize - keyDataLen);
if (inSessionWindow(pPos->pKey, startTs, gap) || (endTs != INT64_MIN && inSessionWindow(pPos->pKey, endTs, gap)) || fn(pKeyData, stateKey) == true) { if (inSessionWindow(pPos->pKey, startTs, gap) || (endTs != INT64_MIN && inSessionWindow(pPos->pKey, endTs, gap)) ||
fn(pKeyData, stateKey) == true) {
(*pVal) = pPos; (*pVal) = pPos;
SSessionKey* pDestWinKey = (SSessionKey*)pPos->pKey; SSessionKey* pDestWinKey = (SSessionKey*)pPos->pKey;
pPos->beUsed = true; pPos->beUsed = true;
@ -607,7 +612,8 @@ int32_t getStateWinResultBuff(SStreamFileState* pFileState, SSessionKey* key, ch
if (code_file == TSDB_CODE_SUCCESS || isFlushedState(pFileState, endTs, 0)) { if (code_file == TSDB_CODE_SUCCESS || isFlushedState(pFileState, endTs, 0)) {
(*pVal) = createSessionWinBuff(pFileState, pWinKey, p, pVLen); (*pVal) = createSessionWinBuff(pFileState, pWinKey, p, pVLen);
code = code_file; code = code_file;
qDebug("===stream===1 get state win:%" PRId64 ",%" PRId64 " from disc, res %d", pWinKey->win.skey, pWinKey->win.ekey, code_file); qDebug("===stream===1 get state win:%" PRId64 ",%" PRId64 " from disc, res %d", pWinKey->win.skey,
pWinKey->win.ekey, code_file);
goto _end; goto _end;
} else { } else {
taosMemoryFree(p); taosMemoryFree(p);

14
source/libs/stream/src/streamSnapshot.c
View File

@ -15,10 +15,8 @@
#include "streamSnapshot.h" #include "streamSnapshot.h"
#include "query.h" #include "query.h"
#include "rocksdb/c.h"
#include "streamBackendRocksdb.h" #include "streamBackendRocksdb.h"
#include "streamInt.h" #include "streamInt.h"
#include "tcommon.h"
enum SBackendFileType { enum SBackendFileType {
ROCKSDB_OPTIONS_TYPE = 1, ROCKSDB_OPTIONS_TYPE = 1,
@ -158,7 +156,7 @@ void snapFileDebugInfo(SBackendSnapFile2* pSnapFile) {
} }
sprintf(buf + strlen(buf) - 1, "]"); sprintf(buf + strlen(buf) - 1, "]");
qInfo("%s %" PRId64 "-%" PRId64 " get file list: %s", STREAM_STATE_TRANSFER, pSnapFile->snapInfo.streamId, stInfo("%s %" PRId64 "-%" PRId64 " get file list: %s", STREAM_STATE_TRANSFER, pSnapFile->snapInfo.streamId,
pSnapFile->snapInfo.taskId, buf); pSnapFile->snapInfo.taskId, buf);
taosMemoryFree(buf); taosMemoryFree(buf);
} }
@ -203,7 +201,7 @@ int32_t snapFileGenMeta(SBackendSnapFile2* pSnapFile) {
int32_t snapFileReadMeta(SBackendSnapFile2* pSnapFile) { int32_t snapFileReadMeta(SBackendSnapFile2* pSnapFile) {
TdDirPtr pDir = taosOpenDir(pSnapFile->path); TdDirPtr pDir = taosOpenDir(pSnapFile->path);
if (NULL == pDir) { if (NULL == pDir) {
qError("%s failed to open %s", STREAM_STATE_TRANSFER, pSnapFile->path); stError("%s failed to open %s", STREAM_STATE_TRANSFER, pSnapFile->path);
return -1; return -1;
} }
@ -397,7 +395,7 @@ _NEXT:
} }
item = taosArrayGet(pSnapFile->pFileList, pSnapFile->currFileIdx); item = taosArrayGet(pSnapFile->pFileList, pSnapFile->currFileIdx);
qDebug("%s start to read file %s, current offset:%" PRId64 ", size:%" PRId64 stDebug("%s start to read file %s, current offset:%" PRId64 ", size:%" PRId64
", file no.%d, total set:%d, current set idx: %d", ", file no.%d, total set:%d, current set idx: %d",
STREAM_STATE_TRANSFER, item->name, (int64_t)pSnapFile->offset, item->size, pSnapFile->currFileIdx, STREAM_STATE_TRANSFER, item->name, (int64_t)pSnapFile->offset, item->size, pSnapFile->currFileIdx,
(int)taosArrayGetSize(pHandle->pDbSnapSet), pHandle->currIdx); (int)taosArrayGetSize(pHandle->pDbSnapSet), pHandle->currIdx);
@ -515,7 +513,7 @@ int32_t streamSnapWriteImpl(SStreamSnapWriter* pWriter, uint8_t* pData, uint32_t
stError("%s failed to write snap, file name:%s, reason:%s", STREAM_STATE_TRANSFER, pHdr->name, tstrerror(code)); stError("%s failed to write snap, file name:%s, reason:%s", STREAM_STATE_TRANSFER, pHdr->name, tstrerror(code));
return code; return code;
} else { } else {
qInfo("succ to write data %s", pItem->name); stInfo("succ to write data %s", pItem->name);
} }
pSnapFile->offset += bytes; pSnapFile->offset += bytes;
} else { } else {
@ -538,7 +536,7 @@ int32_t streamSnapWriteImpl(SStreamSnapWriter* pWriter, uint8_t* pData, uint32_t
} }
taosPWriteFile(pSnapFile->fd, pHdr->data, pHdr->size, pSnapFile->offset); taosPWriteFile(pSnapFile->fd, pHdr->data, pHdr->size, pSnapFile->offset);
qInfo("succ to write data %s", pItem->name); stInfo("succ to write data %s", pItem->name);
pSnapFile->offset += pHdr->size; pSnapFile->offset += pHdr->size;
} }
code = 0; code = 0;
@ -563,7 +561,7 @@ int32_t streamSnapWrite(SStreamSnapWriter* pWriter, uint8_t* pData, uint32_t nDa
"checkpoint", snapInfo.chkpId); "checkpoint", snapInfo.chkpId);
if (!taosIsDir(path)) { if (!taosIsDir(path)) {
code = taosMulMkDir(path); code = taosMulMkDir(path);
qInfo("%s mkdir %s", STREAM_STATE_TRANSFER, path); stInfo("%s mkdir %s", STREAM_STATE_TRANSFER, path);
ASSERT(code == 0); ASSERT(code == 0);
} }

2
source/libs/stream/src/streamUpdate.c
View File

@ -13,8 +13,6 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>. * along with this program. If not, see <http://www.gnu.org/licenses/>.
*/ */
#include "query.h"
#include "tdatablock.h"
#include "tencode.h" #include "tencode.h"
#include "tstreamUpdate.h" #include "tstreamUpdate.h"
#include "ttime.h" #include "ttime.h"