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

Merge pull request #20015 from taosdata/fix/fill_history_state 

fix: save fill history state
This commit is contained in:
Haojun Liao 2023-02-17 11:46:57 +08:00 committed by GitHub
parent ab8397b273 2a4708d82a
commit 71844eeed1
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 27 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -616,6 +616,7 @@ typedef struct SStreamMeta {
SStreamMeta* streamMetaOpen(const char* path, void* ahandle, FTaskExpand expandFunc, int32_t vgId); SStreamMeta* streamMetaOpen(const char* path, void* ahandle, FTaskExpand expandFunc, int32_t vgId);
void streamMetaClose(SStreamMeta* streamMeta); void streamMetaClose(SStreamMeta* streamMeta);
int32_t streamMetaSaveTask(SStreamMeta* pMeta, SStreamTask* pTask);
int32_t streamMetaAddTask(SStreamMeta* pMeta, int64_t ver, SStreamTask* pTask); int32_t streamMetaAddTask(SStreamMeta* pMeta, int64_t ver, SStreamTask* pTask);
int32_t streamMetaAddSerializedTask(SStreamMeta* pMeta, int64_t startVer, char* msg, int32_t msgLen); int32_t streamMetaAddSerializedTask(SStreamMeta* pMeta, int64_t startVer, char* msg, int32_t msgLen);
SStreamTask* streamMetaGetTask(SStreamMeta* pMeta, int32_t taskId); SStreamTask* streamMetaGetTask(SStreamMeta* pMeta, int32_t taskId);
@ -627,7 +628,7 @@ void streamMetaRemoveTask(SStreamMeta* pMeta, int32_t taskId);
int32_t streamMetaBegin(SStreamMeta* pMeta); int32_t streamMetaBegin(SStreamMeta* pMeta);
int32_t streamMetaCommit(SStreamMeta* pMeta); int32_t streamMetaCommit(SStreamMeta* pMeta);
int32_t streamMetaRollBack(SStreamMeta* pMeta); int32_t streamMetaRollBack(SStreamMeta* pMeta);
int32_t streamLoadTasks(SStreamMeta* pMeta); int32_t streamLoadTasks(SStreamMeta* pMeta, int64_t ver);
// checkpoint // checkpoint
int32_t streamProcessCheckpointSourceReq(SStreamMeta* pMeta, SStreamTask* pTask, SStreamCheckpointSourceReq* pReq); int32_t streamProcessCheckpointSourceReq(SStreamMeta* pMeta, SStreamTask* pTask, SStreamCheckpointSourceReq* pReq);

5
source/dnode/vnode/src/tq/tq.c
View File

@ -106,7 +106,7 @@ STQ* tqOpen(const char* path, SVnode* pVnode) {
return NULL; return NULL;
} }
if (streamLoadTasks(pTq->pStreamMeta) < 0) { if (streamLoadTasks(pTq->pStreamMeta, walGetCommittedVer(pVnode->pWal)) < 0) {
return NULL; return NULL;
} }
@ -1196,6 +1196,9 @@ int32_t tqProcessTaskRecover2Req(STQ* pTq, int64_t version, char* msg, int32_t m
return -1; return -1;
} }
atomic_store_8(&pTask->fillHistory, 0);
streamMetaSaveTask(pTq->pStreamMeta, pTask);
streamMetaReleaseTask(pTq->pStreamMeta, pTask); streamMetaReleaseTask(pTq->pStreamMeta, pTask);
return 0; return 0;

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

@ -129,13 +129,8 @@ FAIL:
} }
#endif #endif
#if 1 int32_t streamMetaSaveTask(SStreamMeta* pMeta, SStreamTask* pTask) {
int32_t streamMetaAddTask(SStreamMeta* pMeta, int64_t ver, SStreamTask* pTask) {
void* buf = NULL; void* buf = NULL;
if (pMeta->expandFunc(pMeta->ahandle, pTask, ver) < 0) {
return -1;
}
int32_t len; int32_t len;
int32_t code; int32_t code;
tEncodeSize(tEncodeSStreamTask, pTask, len, code); tEncodeSize(tEncodeSStreamTask, pTask, len, code);
@ -153,11 +148,23 @@ int32_t streamMetaAddTask(SStreamMeta* pMeta, int64_t ver, SStreamTask* pTask) {
tEncoderClear(&encoder); tEncoderClear(&encoder);
if (tdbTbUpsert(pMeta->pTaskDb, &pTask->taskId, sizeof(int32_t), buf, len, pMeta->txn) < 0) { if (tdbTbUpsert(pMeta->pTaskDb, &pTask->taskId, sizeof(int32_t), buf, len, pMeta->txn) < 0) {
ASSERT(0);
return -1; return -1;
} }
taosMemoryFree(buf); taosMemoryFree(buf);
return 0;
}
#if 1
int32_t streamMetaAddTask(SStreamMeta* pMeta, int64_t ver, SStreamTask* pTask) {
if (pMeta->expandFunc(pMeta->ahandle, pTask, ver) < 0) {
return -1;
}
if (streamMetaSaveTask(pMeta, pTask) < 0) {
return -1;
}
taosHashPut(pMeta->pTasks, &pTask->taskId, sizeof(int32_t), &pTask, sizeof(void*)); taosHashPut(pMeta->pTasks, &pTask->taskId, sizeof(int32_t), &pTask, sizeof(void*));
return 0; return 0;
@ -255,7 +262,7 @@ int32_t streamMetaAbort(SStreamMeta* pMeta) {
return 0; return 0;
} }
int32_t streamLoadTasks(SStreamMeta* pMeta) { int32_t streamLoadTasks(SStreamMeta* pMeta, int64_t ver) {
TBC* pCur = NULL; TBC* pCur = NULL;
if (tdbTbcOpen(pMeta->pTaskDb, &pCur, NULL) < 0) { if (tdbTbcOpen(pMeta->pTaskDb, &pCur, NULL) < 0) {
return -1; return -1;
@ -294,7 +301,11 @@ int32_t streamLoadTasks(SStreamMeta* pMeta) {
tdbTbcClose(pCur); tdbTbcClose(pCur);
return -1; return -1;
} }
pTask->taskStatus = TASK_STATUS__NORMAL; /*pTask->taskStatus = TASK_STATUS__NORMAL;*/
if (pTask->fillHistory) {
pTask->taskStatus = TASK_STATUS__WAIT_DOWNSTREAM;
streamTaskCheckDownstream(pTask, ver);
}
} }
tdbFree(pKey); tdbFree(pKey);