From b38e2530191715ec0e3db9e2ff97347d5c7c349a Mon Sep 17 00:00:00 2001 From: Haojun Liao Date: Fri, 26 May 2023 22:51:58 +0800 Subject: [PATCH] fix(stream): set the skip offset when resume stream task. --- include/libs/wal/wal.h | 3 +++ source/dnode/vnode/src/tq/tq.c | 27 ++++++++++++++++----------- source/dnode/vnode/src/tq/tqRestore.c | 10 ++++++++++ source/libs/wal/src/walRead.c | 8 ++++++++ 4 files changed, 37 insertions(+), 11 deletions(-) diff --git a/include/libs/wal/wal.h b/include/libs/wal/wal.h index 35a6838b2e..a081527144 100644 --- a/include/libs/wal/wal.h +++ b/include/libs/wal/wal.h @@ -149,6 +149,7 @@ struct SWalReader { TdFilePtr pIdxFile; int64_t curFileFirstVer; int64_t curVersion; + int64_t skipToVersion; // skip data and jump to destination version, usually used by stream resume ignoring untreated data int64_t capacity; TdThreadMutex mutex; SWalFilterCond cond; @@ -200,6 +201,8 @@ int32_t walReaderSeekVer(SWalReader *pRead, int64_t ver); int32_t walNextValidMsg(SWalReader *pRead); int64_t walReaderGetCurrentVer(const SWalReader *pReader); int64_t walReaderGetValidFirstVer(const SWalReader *pReader); +int64_t walReaderGetSkipToVersion(SWalReader *pReader); +void walReaderSetSkipToVersion(SWalReader *pReader, int64_t ver); void walReaderValidVersionRange(SWalReader *pReader, int64_t *sver, int64_t *ever); void walReaderVerifyOffset(SWalReader *pWalReader, STqOffsetVal* pOffset); diff --git a/source/dnode/vnode/src/tq/tq.c b/source/dnode/vnode/src/tq/tq.c index c6077a3f8f..633b15a1fb 100644 --- a/source/dnode/vnode/src/tq/tq.c +++ b/source/dnode/vnode/src/tq/tq.c @@ -1303,19 +1303,22 @@ int32_t tqProcessTaskPauseReq(STQ* pTq, int64_t sversion, char* msg, int32_t msg int32_t tqProcessTaskResumeReq(STQ* pTq, int64_t sversion, char* msg, int32_t msgLen) { SVResumeStreamTaskReq* pReq = (SVResumeStreamTaskReq*)msg; + + int32_t vgId = pTq->pStreamMeta->vgId; SStreamTask* pTask = streamMetaAcquireTask(pTq->pStreamMeta, pReq->taskId); if (pTask) { atomic_store_8(&pTask->status.taskStatus, pTask->status.keepTaskStatus); // no lock needs to secure the access of the version - if (pReq->igUntreated && pTask->taskLevel == TASK_LEVEL__SOURCE) { // discard all the data when the stream task is suspended. - pTask->chkInfo.currentVer = sversion; - walReaderSeekVer(pTask->exec.pWalReader, sversion); - tqDebug("vgId:%d s-task:%s resume to normal from the latest version:%" PRId64 ", vnode ver:%" PRId64 ", schedStatus:%d", pTq->pStreamMeta->vgId, - pTask->id.idStr, pTask->chkInfo.currentVer, sversion, pTask->status.schedStatus); + if (pReq->igUntreated && pTask->taskLevel == TASK_LEVEL__SOURCE) { + // discard all the data when the stream task is suspended. + walReaderSetSkipToVersion(pTask->exec.pWalReader, sversion); + tqDebug("vgId:%d s-task:%s resume to exec, prev paused version:%" PRId64 ", start from vnode ver:%" PRId64 + ", schedStatus:%d", + vgId, pTask->id.idStr, pTask->chkInfo.currentVer, sversion, pTask->status.schedStatus); } else { // from the previous paused version and go on - tqDebug("vgId:%d s-task:%s resume to normal from paused ver:%" PRId64 ", vnode ver:%" PRId64 ", schedStatus:%d", pTq->pStreamMeta->vgId, - pTask->id.idStr, pTask->chkInfo.currentVer, sversion, pTask->status.schedStatus); + tqDebug("vgId:%d s-task:%s resume to exec, from paused ver:%" PRId64 ", vnode ver:%" PRId64 ", schedStatus:%d", + vgId, pTask->id.idStr, pTask->chkInfo.currentVer, sversion, pTask->status.schedStatus); } if (pTask->taskLevel == TASK_LEVEL__SOURCE && taosQueueItemSize(pTask->inputQueue->queue) == 0) { @@ -1324,6 +1327,8 @@ int32_t tqProcessTaskResumeReq(STQ* pTq, int64_t sversion, char* msg, int32_t ms streamSchedExec(pTask); } streamMetaReleaseTask(pTq->pStreamMeta, pTask); + } else { + tqError("vgId:%d failed to find the s-task:0x%x for resume stream task", vgId, pReq->taskId); } return 0; @@ -1432,7 +1437,7 @@ int32_t tqStartStreamTasks(STQ* pTq) { int32_t numOfTasks = taosArrayGetSize(pMeta->pTaskList); if (numOfTasks == 0) { tqInfo("vgId:%d no stream tasks exist", vgId); - taosWUnLockLatch(&pTq->pStreamMeta->lock); + taosWUnLockLatch(&pMeta->lock); return 0; } @@ -1440,7 +1445,7 @@ int32_t tqStartStreamTasks(STQ* pTq) { if (pMeta->walScanCounter > 1) { tqDebug("vgId:%d wal read task has been launched, remain scan times:%d", vgId, pMeta->walScanCounter); - taosWUnLockLatch(&pTq->pStreamMeta->lock); + taosWUnLockLatch(&pMeta->lock); return 0; } @@ -1448,7 +1453,7 @@ int32_t tqStartStreamTasks(STQ* pTq) { if (pRunReq == NULL) { terrno = TSDB_CODE_OUT_OF_MEMORY; tqError("vgId:%d failed to create msg to start wal scanning to launch stream tasks, code:%s", vgId, terrstr()); - taosWUnLockLatch(&pTq->pStreamMeta->lock); + taosWUnLockLatch(&pMeta->lock); return -1; } @@ -1459,7 +1464,7 @@ int32_t tqStartStreamTasks(STQ* pTq) { SRpcMsg msg = {.msgType = TDMT_STREAM_TASK_RUN, .pCont = pRunReq, .contLen = sizeof(SStreamTaskRunReq)}; tmsgPutToQueue(&pTq->pVnode->msgCb, STREAM_QUEUE, &msg); - taosWUnLockLatch(&pTq->pStreamMeta->lock); + taosWUnLockLatch(&pMeta->lock); return 0; } diff --git a/source/dnode/vnode/src/tq/tqRestore.c b/source/dnode/vnode/src/tq/tqRestore.c index 6e3036a66b..fe80f48691 100644 --- a/source/dnode/vnode/src/tq/tqRestore.c +++ b/source/dnode/vnode/src/tq/tqRestore.c @@ -87,6 +87,16 @@ static int32_t doSetOffsetForWalReader(SStreamTask *pTask, int32_t vgId) { } } + int64_t skipToVer = walReaderGetSkipToVersion(pTask->exec.pWalReader); + if (skipToVer != 0 && skipToVer > pTask->chkInfo.currentVer) { + int32_t code = walReaderSeekVer(pTask->exec.pWalReader, skipToVer); + if (code != TSDB_CODE_SUCCESS) { // no data in wal, quit + return code; + } + + tqDebug("vgId:%d s-task:%s wal reader jump to ver:%" PRId64, vgId, pTask->id.idStr, skipToVer); + } + return TSDB_CODE_SUCCESS; } diff --git a/source/libs/wal/src/walRead.c b/source/libs/wal/src/walRead.c index c29d82bcf3..0eb3d6ef09 100644 --- a/source/libs/wal/src/walRead.c +++ b/source/libs/wal/src/walRead.c @@ -108,6 +108,14 @@ int32_t walNextValidMsg(SWalReader *pReader) { int64_t walReaderGetCurrentVer(const SWalReader *pReader) { return pReader->curVersion; } int64_t walReaderGetValidFirstVer(const SWalReader *pReader) { return walGetFirstVer(pReader->pWal); } +void walReaderSetSkipToVersion(SWalReader *pReader, int64_t ver) { atomic_store_64(&pReader->skipToVersion, ver); } + +// this function is NOT multi-thread safe, and no need to be. +int64_t walReaderGetSkipToVersion(SWalReader *pReader) { + int64_t newVersion = pReader->skipToVersion; + pReader->skipToVersion = 0; + return newVersion; +} void walReaderValidVersionRange(SWalReader *pReader, int64_t *sver, int64_t *ever) { *sver = walGetFirstVer(pReader->pWal); -- GitLab