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

support reopen stream state

This commit is contained in:
yihaoDeng 2023-08-07 08:51:01 +00:00
parent 7ef4df8752
commit eeb97351e8
6 changed files with 60 additions and 38 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
include/libs/stream/tstream.h
View File

@ -612,6 +612,7 @@ void streamMetaReleaseTask(SStreamMeta* pMeta, SStreamTask* pTask);
void streamMetaRemoveTask(SStreamMeta* pMeta, int32_t taskId);
int32_t streamStateRebuild(SStreamMeta* pMeta, char* path, int64_t chkpId);
int32_t streamStateReopen(SStreamMeta *pMeta, int64_t chkpId);
int32_t streamMetaBegin(SStreamMeta* pMeta);
int32_t streamMetaCommit(SStreamMeta* pMeta);

3
source/dnode/vnode/src/inc/vnodeInt.h
View File

@ -331,6 +331,9 @@ int32_t streamStateSnapWriterClose(SStreamStateWriter* pWriter, int8_t rollback)
int32_t streamStateSnapWrite(SStreamStateWriter* pWriter, uint8_t* pData, uint32_t nData);
int32_t streamStateRebuildFromSnap(SStreamStateWriter* pWriter, char* path, int64_t chkpId);
int32_t streamStateLoadTasks(SStreamStateWriter* pWriter);
// SStreamTaskReader ======================================
// SStreamStateWriter =====================================
// SStreamStateReader =====================================

11
source/dnode/vnode/src/tq/tqStreamStateSnap.c
View File

@ -163,7 +163,16 @@ int32_t streamStateSnapWriterClose(SStreamStateWriter* pWriter, int8_t rollback)
return code;
}
int32_t streamStateRebuildFromSnap(SStreamStateWriter* pWriter, char* path, int64_t chkpId) {
return streamStateRebuild(pWriter->pTq->pStreamMeta, path, chkpId);
return streamStateReopen(pWriter->pTq->pStreamMeta, chkpId);
}
int32_t streamStateLoadTasksImpl(SStreamMeta* pMeta, int64_t ver) {
// impl later
return streamLoadTasks(pMeta, ver);
}
int32_t streamStateLoadTasks(SStreamStateWriter* pWriter) {
SWal* pWal = pWriter->pTq->pVnode->pWal;
return streamStateLoadTasksImpl(pWriter->pTq->pStreamMeta, walGetCommittedVer(pWal));
}
int32_t streamStateSnapWrite(SStreamStateWriter* pWriter, uint8_t* pData, uint32_t nData) {

6
source/dnode/vnode/src/vnd/vnodeSnapshot.c
View File

@ -409,6 +409,12 @@ int32_t vnodeSnapWriterClose(SVSnapWriter *pWriter, int8_t rollback, SSnapshot *
if (pWriter->pStreamStateWriter) {
code = streamStateSnapWriterClose(pWriter->pStreamStateWriter, rollback);
if (code) goto _exit;
code = streamStateRebuildFromSnap(pWriter->pStreamStateWriter, NULL, 0);
if (code) goto _exit;
code = streamStateLoadTasks(pWriter->pStreamStateWriter);
if (code) goto _exit;
}
if (pWriter->pRsmaSnapWriter) {

8
source/libs/stream/src/streamBackendRocksdb.c
View File

@ -732,8 +732,8 @@ int stateSessionKeyDBComp(void* state, const char* aBuf, size_t aLen, const char
return stateSessionKeyCmpr(&w1, sizeof(w1), &w2, sizeof(w2));
}
int stateSessionKeyEncode(void* ses, char* buf) {
SStateSessionKey* sess = ses;
int stateSessionKeyEncode(void* k, char* buf) {
SStateSessionKey* sess = k;
int len = 0;
len += taosEncodeFixedI64((void**)&buf, sess->key.win.skey);
len += taosEncodeFixedI64((void**)&buf, sess->key.win.ekey);
@ -741,8 +741,8 @@ int stateSessionKeyEncode(void* ses, char* buf) {
len += taosEncodeFixedI64((void**)&buf, sess->opNum);
return len;
}
int stateSessionKeyDecode(void* ses, char* buf) {
SStateSessionKey* sess = ses;
int stateSessionKeyDecode(void* k, char* buf) {
SStateSessionKey* sess = k;
int len = 0;
char* p = buf;

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

@ -128,48 +128,51 @@ _err:
return NULL;
}
void streamMetaReopen(SStreamMeta** ppMeta) {
SStreamMeta* pMeta = *ppMeta;
void streamMetaReopen(SStreamMeta* pMeta, int64_t chkpId) {
// stop all running tasking and reopen later
void* pIter = NULL;
while (1) {
pIter = taosHashIterate(pMeta->pTasks, pIter);
if (pIter == NULL) {
break;
}
SStreamMeta* pNewMeta = taosMemoryCalloc(1, sizeof(SStreamMeta));
pNewMeta->path = taosStrdup(pMeta->path);
pNewMeta->vgId = pMeta->vgId;
pNewMeta->walScanCounter = 0;
pNewMeta->ahandle = pMeta->ahandle;
pNewMeta->expandFunc = pMeta->expandFunc;
SStreamTask* pTask = *(SStreamTask**)pIter;
if (pTask->schedTimer) {
taosTmrStop(pTask->schedTimer);
pTask->schedTimer = NULL;
}
*ppMeta = pNewMeta;
if (pTask->launchTaskTimer) {
taosTmrStop(pTask->launchTaskTimer);
pTask->launchTaskTimer = NULL;
}
streamMetaClose(pMeta);
tFreeStreamTask(pTask);
}
// tdbAbort(pMeta->db, pMeta->txn);
// tdbTbClose(pMeta->pTaskDb);
// tdbTbClose(pMeta->pCheckpointDb);
// tdbClose(pMeta->db);
// close stream backend
streamBackendCleanup(pMeta->streamBackend);
taosRemoveRef(streamBackendId, pMeta->streamBackendRid);
pMeta->streamBackendRid = -1;
pMeta->streamBackend = NULL;
// void* pIter = NULL;
// while (1) {
// pIter = taosHashIterate(pMeta->pTasks, pIter);
// if (pIter == NULL) {
// break;
// }
pMeta->streamBackend = streamBackendInit(pMeta->path, chkpId);
pMeta->streamBackendRid = taosAddRef(streamBackendId, pMeta->streamBackend);
// SStreamTask* pTask = *(SStreamTask**)pIter;
// if (pTask->schedTimer) {
// taosTmrStop(pTask->schedTimer);
// pTask->schedTimer = NULL;
// }
taosHashClear(pMeta->pTasks);
// if (pTask->launchTaskTimer) {
// taosTmrStop(pTask->launchTaskTimer);
// pTask->launchTaskTimer = NULL;
// }
taosArrayClear(pMeta->pTaskList);
// tFreeStreamTask(pTask);
// }
taosHashClear(pMeta->pTaskBackendUnique);
// taosHashClear(pMeta->pTasks);
// taosRemoveRef(streamBackendId, pMeta->streamBackendRid);
taosArrayClear(pMeta->checkpointSaved);
taosArrayClear(pMeta->checkpointInUse);
// if (streamLoadTasks(pMeta,int64_t ver))
return;
}
void streamMetaClose(SStreamMeta* pMeta) {
tdbAbort(pMeta->db, pMeta->txn);