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

Merge remote-tracking branch 'origin/enh/triggerCheckPoint2' into enh/triggerCheckPoint2

This commit is contained in:
Haojun Liao 2023-08-11 10:37:40 +08:00
parent 6a9826cf6e 32983a11e1
commit 1e636a2ef9
2 changed files with 19 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -220,20 +220,29 @@ int32_t vnodeSnapRead(SVSnapReader *pReader, uint8_t **ppData, uint32_t *nData)
} }
// STREAM ============ // STREAM ============
vInfo("stream task start");
if (!pReader->streamTaskDone) { if (!pReader->streamTaskDone) {
if (pReader->pStreamTaskReader == NULL) { if (pReader->pStreamTaskReader == NULL) {
vInfo("stream task start 1");
code = streamTaskSnapReaderOpen(pReader->pVnode->pTq, pReader->sver, pReader->sver, &pReader->pStreamTaskReader); code = streamTaskSnapReaderOpen(pReader->pVnode->pTq, pReader->sver, pReader->sver, &pReader->pStreamTaskReader);
if (code) goto _err; if (code) {
vInfo("stream task start err");
goto _err;
}
} }
code = streamTaskSnapRead(pReader->pStreamTaskReader, ppData); code = streamTaskSnapRead(pReader->pStreamTaskReader, ppData);
vInfo("stream task start 2");
if (code) { if (code) {
vInfo("stream task start 3");
goto _err; goto _err;
} else { } else {
if (*ppData) { if (*ppData) {
goto _exit; goto _exit;
vInfo("stream task start 4");
} else { } else {
pReader->streamTaskDone = 1; pReader->streamTaskDone = 1;
code = streamTaskSnapReaderClose(pReader->pStreamTaskReader); code = streamTaskSnapReaderClose(pReader->pStreamTaskReader);
vInfo("stream task start 5");
if (code) goto _err; if (code) goto _err;
pReader->pStreamTaskReader = NULL; pReader->pStreamTaskReader = NULL;
} }

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

@ -43,7 +43,7 @@ SStreamMeta* streamMetaOpen(const char* path, void* ahandle, FTaskExpand expandF
SStreamMeta* pMeta = taosMemoryCalloc(1, sizeof(SStreamMeta)); SStreamMeta* pMeta = taosMemoryCalloc(1, sizeof(SStreamMeta));
if (pMeta == NULL) { if (pMeta == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY; terrno = TSDB_CODE_OUT_OF_MEMORY;
qError("vgId:%d failed to prepare stream meta, alloc size:%"PRIzu", out of memory", vgId, sizeof(SStreamMeta)); qError("vgId:%d failed to prepare stream meta, alloc size:%" PRIzu ", out of memory", vgId, sizeof(SStreamMeta));
return NULL; return NULL;
} }
@ -122,7 +122,7 @@ SStreamMeta* streamMetaOpen(const char* path, void* ahandle, FTaskExpand expandF
taosInitRWLatch(&pMeta->lock); taosInitRWLatch(&pMeta->lock);
taosThreadMutexInit(&pMeta->backendMutex, NULL); taosThreadMutexInit(&pMeta->backendMutex, NULL);
qInfo("vgId:%d open stream meta successfully, latest checkpoint:%"PRId64, vgId, chkpId); qInfo("vgId:%d open stream meta successfully, latest checkpoint:%" PRId64, vgId, chkpId);
return pMeta; return pMeta;
_err: _err:
@ -211,7 +211,7 @@ int32_t streamMetaRegisterTask(SStreamMeta* pMeta, int64_t ver, SStreamTask* pTa
*pAdded = false; *pAdded = false;
int64_t keys[2] = {pTask->id.streamId, pTask->id.taskId}; int64_t keys[2] = {pTask->id.streamId, pTask->id.taskId};
void* p = taosHashGet(pMeta->pTasks, keys, sizeof(keys)); void* p = taosHashGet(pMeta->pTasks, keys, sizeof(keys));
if (p == NULL) { if (p == NULL) {
if (pMeta->expandFunc(pMeta->ahandle, pTask, ver) < 0) { if (pMeta->expandFunc(pMeta->ahandle, pTask, ver) < 0) {
tFreeStreamTask(pTask); tFreeStreamTask(pTask);
@ -247,7 +247,7 @@ int32_t streamMetaGetNumOfTasks(SStreamMeta* pMeta) {
SStreamTask* streamMetaAcquireTask(SStreamMeta* pMeta, int64_t streamId, int32_t taskId) { SStreamTask* streamMetaAcquireTask(SStreamMeta* pMeta, int64_t streamId, int32_t taskId) {
taosRLockLatch(&pMeta->lock); taosRLockLatch(&pMeta->lock);
int64_t keys[2] = {streamId, taskId}; int64_t keys[2] = {streamId, taskId};
SStreamTask** ppTask = (SStreamTask**)taosHashGet(pMeta->pTasks, keys, sizeof(keys)); SStreamTask** ppTask = (SStreamTask**)taosHashGet(pMeta->pTasks, keys, sizeof(keys));
if (ppTask != NULL) { if (ppTask != NULL) {
if (!streamTaskShouldStop(&(*ppTask)->status)) { if (!streamTaskShouldStop(&(*ppTask)->status)) {
@ -291,7 +291,7 @@ int32_t streamMetaUnregisterTask(SStreamMeta* pMeta, int64_t streamId, int32_t t
// pre-delete operation // pre-delete operation
taosWLockLatch(&pMeta->lock); taosWLockLatch(&pMeta->lock);
int64_t keys[2] = {streamId, taskId}; int64_t keys[2] = {streamId, taskId};
SStreamTask** ppTask = (SStreamTask**)taosHashGet(pMeta->pTasks, keys, sizeof(keys)); SStreamTask** ppTask = (SStreamTask**)taosHashGet(pMeta->pTasks, keys, sizeof(keys));
if (ppTask) { if (ppTask) {
pTask = *ppTask; pTask = *ppTask;
@ -390,6 +390,9 @@ int64_t streamGetLatestCheckpointId(SStreamMeta* pMeta) {
tdbTbcMoveToFirst(pCur); tdbTbcMoveToFirst(pCur);
while (tdbTbcNext(pCur, &pKey, &kLen, &pVal, &vLen) == 0) { while (tdbTbcNext(pCur, &pKey, &kLen, &pVal, &vLen) == 0) {
if (pVal != NULL && vLen != 0) {
break;
}
SCheckpointInfo info; SCheckpointInfo info;
tDecoderInit(&decoder, (uint8_t*)pVal, vLen); tDecoderInit(&decoder, (uint8_t*)pVal, vLen);
if (tDecodeStreamTaskChkInfo(&decoder, &info) < 0) { if (tDecodeStreamTaskChkInfo(&decoder, &info) < 0) {
@ -462,7 +465,7 @@ int32_t streamLoadTasks(SStreamMeta* pMeta) {
// do duplicate task check. // do duplicate task check.
int64_t keys[2] = {pTask->id.streamId, pTask->id.taskId}; int64_t keys[2] = {pTask->id.streamId, pTask->id.taskId};
void* p = taosHashGet(pMeta->pTasks, keys, sizeof(keys)); void* p = taosHashGet(pMeta->pTasks, keys, sizeof(keys));
if (p == NULL) { if (p == NULL) {
if (pMeta->expandFunc(pMeta->ahandle, pTask, pTask->chkInfo.checkpointVer) < 0) { if (pMeta->expandFunc(pMeta->ahandle, pTask, pTask->chkInfo.checkpointVer) < 0) {
tdbFree(pKey); tdbFree(pKey);