From c4a27569076a7d5ba418e3c4acec88a013105205 Mon Sep 17 00:00:00 2001 From: Haojun Liao Date: Mon, 17 Jun 2024 23:13:54 +0800 Subject: [PATCH] fix(stream): not revise the stream id for fill-history task. --- include/libs/stream/streamState.h | 3 ++- source/dnode/snode/src/snode.c | 6 ----- source/dnode/vnode/src/sma/smaRollup.c | 2 +- source/dnode/vnode/src/tqCommon/tqCommon.c | 31 ++++++---------------- source/libs/stream/src/streamMeta.c | 24 +++++------------ source/libs/stream/src/streamState.c | 7 ++--- 6 files changed, 21 insertions(+), 52 deletions(-) diff --git a/include/libs/stream/streamState.h b/include/libs/stream/streamState.h index ae5a733ae9..1333257dfb 100644 --- a/include/libs/stream/streamState.h +++ b/include/libs/stream/streamState.h @@ -29,7 +29,8 @@ extern "C" { #include "storageapi.h" -SStreamState* streamStateOpen(const char* path, void* pTask, bool specPath, int32_t szPage, int32_t pages); +SStreamState* streamStateOpen(const char* path, void* pTask, int64_t streamId, int32_t taskId, bool specPath, + int32_t szPage, int32_t pages); void streamStateClose(SStreamState* pState, bool remove); int32_t streamStateBegin(SStreamState* pState); int32_t streamStateCommit(SStreamState* pState); diff --git a/source/dnode/snode/src/snode.c b/source/dnode/snode/src/snode.c index d322eb2977..481033508b 100644 --- a/source/dnode/snode/src/snode.c +++ b/source/dnode/snode/src/snode.c @@ -25,12 +25,6 @@ #define sndDebug(...) do { if (sndDebugFlag & DEBUG_DEBUG) { taosPrintLog("SND ", DEBUG_DEBUG, sndDebugFlag, __VA_ARGS__);}} while (0) // clang-format on -static void restoreStreamTaskId(SStreamTask *pTask, STaskId *pId) { - ASSERT(pTask->info.fillHistory); - pTask->id.taskId = pId->taskId; - pTask->id.streamId = pId->streamId; -} - int32_t sndExpandTask(SSnode *pSnode, SStreamTask *pTask, int64_t nextProcessVer) { ASSERT(pTask->info.taskLevel == TASK_LEVEL__AGG && taosArrayGetSize(pTask->upstreamInfo.pList) != 0); int32_t code = streamTaskInit(pTask, pSnode->pMeta, &pSnode->msgCb, nextProcessVer); diff --git a/source/dnode/vnode/src/sma/smaRollup.c b/source/dnode/vnode/src/sma/smaRollup.c index 3cc7c6ec66..ac5463e492 100644 --- a/source/dnode/vnode/src/sma/smaRollup.c +++ b/source/dnode/vnode/src/sma/smaRollup.c @@ -299,7 +299,7 @@ static int32_t tdSetRSmaInfoItemParams(SSma *pSma, SRSmaParam *param, SRSmaStat tdRSmaTaskInit(pStreamTask->pMeta, pItem, &pStreamTask->id); pStreamTask->status.pSM = streamCreateStateMachine(pStreamTask); pStreamTask->chkInfo.pActiveInfo = streamTaskCreateActiveChkptInfo(); - pStreamState = streamStateOpen(taskInfDir, pStreamTask, true, -1, -1); + pStreamState = streamStateOpen(taskInfDir, pStreamTask, pStreamTask->id.streamId, pStreamTask->id.taskId, true, -1, -1); if (!pStreamState) { terrno = TSDB_CODE_RSMA_STREAM_STATE_OPEN; return TSDB_CODE_FAILED; diff --git a/source/dnode/vnode/src/tqCommon/tqCommon.c b/source/dnode/vnode/src/tqCommon/tqCommon.c index d9bc8b74d2..963503c135 100644 --- a/source/dnode/vnode/src/tqCommon/tqCommon.c +++ b/source/dnode/vnode/src/tqCommon/tqCommon.c @@ -30,37 +30,26 @@ typedef struct SMStreamCheckpointReadyRspMsg { static int32_t doProcessDummyRspMsg(SStreamMeta* pMeta, SRpcMsg* pMsg); -static STaskId replaceStreamTaskId(SStreamTask* pTask) { - ASSERT(pTask->info.fillHistory); - STaskId id = {.streamId = pTask->id.streamId, .taskId = pTask->id.taskId}; - - pTask->id.streamId = pTask->streamTaskId.streamId; - pTask->id.taskId = pTask->streamTaskId.taskId; - - return id; -} - -static void restoreStreamTaskId(SStreamTask* pTask, STaskId* pId) { - ASSERT(pTask->info.fillHistory); - pTask->id.taskId = pId->taskId; - pTask->id.streamId = pId->streamId; -} - int32_t tqExpandStreamTask(SStreamTask* pTask) { SStreamMeta* pMeta = pTask->pMeta; int32_t vgId = pMeta->vgId; - STaskId taskId = {0}; int64_t st = taosGetTimestampMs(); + int64_t streamId = 0; + int32_t taskId = 0; tqDebug("s-task:%s vgId:%d start to expand stream task", pTask->id.idStr, vgId); if (pTask->info.fillHistory) { - taskId = replaceStreamTaskId(pTask); + streamId = pTask->streamTaskId.streamId; + taskId = pTask->streamTaskId.taskId; + } else { + streamId = pTask->id.streamId; + taskId = pTask->id.taskId; } // sink task does not need the pState if (pTask->info.taskLevel != TASK_LEVEL__SINK) { - pTask->pState = streamStateOpen(pMeta->path, pTask, false, -1, -1); + pTask->pState = streamStateOpen(pMeta->path, pTask, false, streamId, taskId, -1, -1); if (pTask->pState == NULL) { tqError("s-task:%s (vgId:%d) failed to open state for task, expand task failed", pTask->id.idStr, vgId); return -1; @@ -69,10 +58,6 @@ int32_t tqExpandStreamTask(SStreamTask* pTask) { } } - if (pTask->info.fillHistory) { - restoreStreamTaskId(pTask, &taskId); - } - SReadHandle handle = { .checkpointId = pTask->chkInfo.checkpointId, .pStateBackend = pTask->pState, diff --git a/source/libs/stream/src/streamMeta.c b/source/libs/stream/src/streamMeta.c index 05c414078d..e8800c3370 100644 --- a/source/libs/stream/src/streamMeta.c +++ b/source/libs/stream/src/streamMeta.c @@ -608,6 +608,7 @@ int32_t streamMetaRegisterTask(SStreamMeta* pMeta, int64_t ver, SStreamTask* pTa } taosArrayPush(pMeta->pTaskList, &pTask->id); + taosHashPut(pMeta->pTasksMap, &id, sizeof(id), &pTask, POINTER_BYTES); if (streamMetaSaveTask(pMeta, pTask) < 0) { return -1; @@ -617,7 +618,6 @@ int32_t streamMetaRegisterTask(SStreamMeta* pMeta, int64_t ver, SStreamTask* pTa return -1; } - taosHashPut(pMeta->pTasksMap, &id, sizeof(id), &pTask, POINTER_BYTES); if (pTask->info.fillHistory == 0) { atomic_add_fetch_32(&pMeta->numOfStreamTasks, 1); } @@ -672,21 +672,16 @@ void streamMetaReleaseTask(SStreamMeta* UNUSED_PARAM(pMeta), SStreamTask* pTask) } } -static void doRemoveIdFromList(SStreamMeta* pMeta, int32_t num, SStreamTaskId* id) { +static void doRemoveIdFromList(SArray* pTaskList, int32_t num, SStreamTaskId* id) { bool remove = false; for (int32_t i = 0; i < num; ++i) { - SStreamTaskId* pTaskId = taosArrayGet(pMeta->pTaskList, i); + SStreamTaskId* pTaskId = taosArrayGet(pTaskList, i); if (pTaskId->streamId == id->streamId && pTaskId->taskId == id->taskId) { - stDebug("vgId:%d remove streamId:0x%" PRIx64 " taskId:0x%x succ", pMeta->vgId, id->streamId, id->taskId); - taosArrayRemove(pMeta->pTaskList, i); + taosArrayRemove(pTaskList, i); remove = true; break; - } else { - stDebug("vgId:%d remove streamId:0x%" PRIx64 " taskId:0x%x, entry:0x%" PRIx64 "-0x%x", pMeta->vgId, id->streamId, - id->taskId, pTaskId->streamId, pTaskId->taskId); } } - ASSERT(remove); } @@ -750,26 +745,19 @@ int32_t streamMetaUnregisterTask(SStreamMeta* pMeta, int64_t streamId, int32_t t ppTask = (SStreamTask**)taosHashGet(pMeta->pTasksMap, &id, sizeof(id)); if (ppTask) { pTask = *ppTask; - SStreamTaskId pxId = pTask->id; - ASSERT((pxId.taskId == id.taskId) && (pxId.streamId == id.streamId)); - // it is an fill-history task, remove the related stream task's id that points to it if (pTask->info.fillHistory == 0) { atomic_sub_fetch_32(&pMeta->numOfStreamTasks, 1); } - ASSERT(taosHashGetSize(pMeta->pTasksMap) == taosArrayGetSize(pMeta->pTaskList)); - taosHashRemove(pMeta->pTasksMap, &id, sizeof(id)); - doRemoveIdFromList(pMeta, (int32_t)taosArrayGetSize(pMeta->pTaskList), &pTask->id); - - ASSERT(taosHashGetSize(pMeta->pTasksMap) == taosArrayGetSize(pMeta->pTaskList)); + doRemoveIdFromList(pMeta->pTaskList, (int32_t)taosArrayGetSize(pMeta->pTaskList), &pTask->id); streamMetaRemoveTask(pMeta, &id); + ASSERT(taosHashGetSize(pMeta->pTasksMap) == taosArrayGetSize(pMeta->pTaskList)); streamMetaWUnLock(pMeta); ASSERT(pTask->status.timerActive == 0); - if (pTask->info.delaySchedParam != 0 && pTask->info.fillHistory == 0) { stDebug("s-task:%s stop schedTimer, and (before) desc ref:%d", pTask->id.idStr, pTask->refCnt); taosTmrStop(pTask->schedInfo.pDelayTimer); diff --git a/source/libs/stream/src/streamState.c b/source/libs/stream/src/streamState.c index 47324bd8c9..8919bd48ac 100644 --- a/source/libs/stream/src/streamState.c +++ b/source/libs/stream/src/streamState.c @@ -98,7 +98,8 @@ int stateKeyCmpr(const void* pKey1, int kLen1, const void* pKey2, int kLen2) { return winKeyCmprImpl(&pWin1->key, &pWin2->key); } -SStreamState* streamStateOpen(const char* path, void* pTask, bool specPath, int32_t szPage, int32_t pages) { +SStreamState* streamStateOpen(const char* path, void* pTask, int64_t streamId, int32_t taskId, bool specPath, + int32_t szPage, int32_t pages) { SStreamState* pState = taosMemoryCalloc(1, sizeof(SStreamState)); stDebug("open stream state %p, %s", pState, path); if (pState == NULL) { @@ -114,8 +115,8 @@ SStreamState* streamStateOpen(const char* path, void* pTask, bool specPath, int3 } SStreamTask* pStreamTask = pTask; - pState->taskId = pStreamTask->id.taskId; - pState->streamId = pStreamTask->id.streamId; + pState->streamId = streamId; + pState->taskId = taskId; sprintf(pState->pTdbState->idstr, "0x%" PRIx64 "-0x%x", pState->streamId, pState->taskId); streamTaskSetDb(pStreamTask->pMeta, pTask, pState->pTdbState->idstr);