From a72c0c2bd2c4a8eeb02598391f04e8eee25ff2bb Mon Sep 17 00:00:00 2001 From: Haojun Liao Date: Mon, 2 Sep 2024 16:49:47 +0800 Subject: [PATCH 1/4] fix(tsdb): init merge if get the initial pschema failed. --- source/dnode/vnode/src/tsdb/tsdbRead2.c | 101 ++++++++++-------------- 1 file changed, 40 insertions(+), 61 deletions(-) diff --git a/source/dnode/vnode/src/tsdb/tsdbRead2.c b/source/dnode/vnode/src/tsdb/tsdbRead2.c index b7bfd045d1..cecc52c51c 100644 --- a/source/dnode/vnode/src/tsdb/tsdbRead2.c +++ b/source/dnode/vnode/src/tsdb/tsdbRead2.c @@ -1734,14 +1734,35 @@ static bool isCleanFileDataBlock(STsdbReader* pReader, SFileDataBlockInfo* pBloc return isCleanFileBlock; } +static int32_t initRowMergeIfNeeded(STsdbReader* pReader, int64_t uid) { + SRowMerger* pMerger = &pReader->status.merger; + int32_t code = 0; + + if (pMerger->pArray == NULL) { + STSchema* ps = getTableSchemaImpl(pReader, uid); + if (ps == NULL) { + return terrno; + } + + code = tsdbRowMergerInit(pMerger, ps); + } + + return code; +} + static int32_t buildDataBlockFromBuf(STsdbReader* pReader, STableBlockScanInfo* pBlockScanInfo, int64_t endKey) { if (!(pBlockScanInfo->iiter.hasVal || pBlockScanInfo->iter.hasVal)) { return TSDB_CODE_SUCCESS; } + int32_t code = initRowMergeIfNeeded(pReader, pBlockScanInfo->uid); + if (code != 0) { + return code; + } + int64_t st = taosGetTimestampUs(); SSDataBlock* pBlock = pReader->resBlockInfo.pResBlock; - int32_t code = buildDataBlockFromBufImpl(pBlockScanInfo, endKey, pReader->resBlockInfo.capacity, pReader); + code = buildDataBlockFromBufImpl(pBlockScanInfo, endKey, pReader->resBlockInfo.capacity, pReader); double el = (taosGetTimestampUs() - st) / 1000.0; updateComposedBlockInfo(pReader, el, pBlockScanInfo); @@ -1943,19 +1964,9 @@ static int32_t doMergeBufAndFileRows(STsdbReader* pReader, STableBlockScanInfo* TSDBROW fRow = tsdbRowFromBlockData(pBlockData, pDumpInfo->rowIndex); // merge is not initialized yet, due to the fact that the pReader->info.pSchema is not initialized - if (pMerger->pArray == NULL) { - if (pReader->info.pSchema != NULL) { - tsdbError("tsdb failed at %s:%d", __func__, __LINE__); - return TSDB_CODE_INTERNAL_ERROR; - } - STSchema* ps = getTableSchemaImpl(pReader, pBlockScanInfo->uid); - if (ps == NULL) { - return terrno; - } - int32_t code = tsdbRowMergerInit(pMerger, ps); - if (code != TSDB_CODE_SUCCESS) { - return code; - } + int32_t code = initRowMergeIfNeeded(pReader, pBlockScanInfo->uid); + if (code != 0) { + return code; } SRowKey minKey = k; @@ -1983,7 +1994,7 @@ static int32_t doMergeBufAndFileRows(STsdbReader* pReader, STableBlockScanInfo* // file block ---> stt block -----> mem if (pkCompEx(&minKey, pfKey) == 0) { - int32_t code = tsdbRowMergerAdd(pMerger, &fRow, NULL); + code = tsdbRowMergerAdd(pMerger, &fRow, NULL); if (code != TSDB_CODE_SUCCESS) { return code; } @@ -1996,7 +2007,7 @@ static int32_t doMergeBufAndFileRows(STsdbReader* pReader, STableBlockScanInfo* if (pkCompEx(&minKey, pSttKey) == 0) { TSDBROW* fRow1 = tMergeTreeGetRow(&pSttBlockReader->mergeTree); - int32_t code = tsdbRowMergerAdd(pMerger, fRow1, NULL); + code = tsdbRowMergerAdd(pMerger, fRow1, NULL); if (code != TSDB_CODE_SUCCESS) { return code; } @@ -2007,7 +2018,7 @@ static int32_t doMergeBufAndFileRows(STsdbReader* pReader, STableBlockScanInfo* } if (pkCompEx(&minKey, &k) == 0) { - int32_t code = tsdbRowMergerAdd(pMerger, pRow, pSchema); + code = tsdbRowMergerAdd(pMerger, pRow, pSchema); if (code != TSDB_CODE_SUCCESS) { return code; } @@ -2018,7 +2029,7 @@ static int32_t doMergeBufAndFileRows(STsdbReader* pReader, STableBlockScanInfo* } } - int32_t code = tsdbRowMergerGetRow(pMerger, &pTSRow); + code = tsdbRowMergerGetRow(pMerger, &pTSRow); if (code != TSDB_CODE_SUCCESS) { return code; } @@ -2039,19 +2050,9 @@ static int32_t mergeFileBlockAndSttBlock(STsdbReader* pReader, SSttBlockReader* int32_t pkSrcSlot = pReader->suppInfo.pkSrcSlot; // merge is not initialized yet, due to the fact that the pReader->info.pSchema is not initialized - if (pMerger->pArray == NULL) { - if (pReader->info.pSchema) { - tsdbError("tsdb failed at %s %d", __func__, __LINE__); - return TSDB_CODE_INTERNAL_ERROR; - } - STSchema* ps = getTableSchemaImpl(pReader, pBlockScanInfo->uid); - if (ps == NULL) { - return terrno; - } - code = tsdbRowMergerInit(pMerger, ps); - if (code != TSDB_CODE_SUCCESS) { - return code; - } + code = initRowMergeIfNeeded(pReader, pBlockScanInfo->uid); + if (code != 0) { + return code; } bool dataInDataFile = hasDataInFileBlock(pBlockData, pDumpInfo); @@ -2175,20 +2176,9 @@ static int32_t doMergeMultiLevelRows(STsdbReader* pReader, STableBlockScanInfo* } // merge is not initialized yet, due to the fact that the pReader->info.pSchema is not initialized - if (pMerger->pArray == NULL) { - if (pReader->info.pSchema != NULL) { - tsdbError("tsdb read failed at: %s:%d", __func__, __LINE__); - return TSDB_CODE_INTERNAL_ERROR; - } - STSchema* ps = getTableSchemaImpl(pReader, pBlockScanInfo->uid); - if (ps == NULL) { - return terrno; - } - - code = tsdbRowMergerInit(pMerger, ps); - if (code != TSDB_CODE_SUCCESS) { - return code; - } + code = initRowMergeIfNeeded(pReader, pBlockScanInfo->uid); + if (code != 0) { + return code; } SRowKey minKey = k; @@ -2579,20 +2569,9 @@ int32_t mergeRowsInFileBlocks(SBlockData* pBlockData, STableBlockScanInfo* pBloc } // merge is not initialized yet, due to the fact that the pReader->info.pSchema is not initialized - if (pMerger->pArray == NULL) { - if (pReader->info.pSchema != NULL) { - tsdbError("tsdb reader failed at: %s:%d", __func__, __LINE__); - return TSDB_CODE_INTERNAL_ERROR; - } - STSchema* ps = getTableSchemaImpl(pReader, pBlockScanInfo->uid); - if (ps == NULL) { - return terrno; - } - - code = tsdbRowMergerInit(pMerger, ps); - if (code != TSDB_CODE_SUCCESS) { - return code; - } + code = initRowMergeIfNeeded(pReader, pBlockScanInfo->uid); + if (code != 0) { + return code; } tRowKeyAssign(&pBlockScanInfo->lastProcKey, pKey); @@ -4770,13 +4749,13 @@ int32_t tsdbReaderOpen2(void* pVnode, SQueryTableDataCond* pCond, void* pTableLi if (pCond->suid != 0) { pReader->info.pSchema = metaGetTbTSchema(pReader->pTsdb->pVnode->pMeta, pReader->info.suid, -1, 1); if (pReader->info.pSchema == NULL) { - tsdbError("failed to get table schema, suid:%" PRIu64 ", ver:-1, %s", pReader->info.suid, pReader->idStr); + tsdbWarn("failed to get table schema, suid:%" PRIu64 ", ver:-1, %s", pReader->info.suid, pReader->idStr); } } else if (numOfTables > 0) { STableKeyInfo* pKey = pTableList; pReader->info.pSchema = metaGetTbTSchema(pReader->pTsdb->pVnode->pMeta, pKey->uid, -1, 1); if (pReader->info.pSchema == NULL) { - tsdbError("failed to get table schema, uid:%" PRIu64 ", ver:-1, %s", pKey->uid, pReader->idStr); + tsdbWarn("failed to get table schema, uid:%" PRIu64 ", ver:-1, %s", pKey->uid, pReader->idStr); } } From 277528996eb5ebf0e4c217d021d4998196bfce5a Mon Sep 17 00:00:00 2001 From: Haojun Liao Date: Thu, 5 Sep 2024 16:04:36 +0800 Subject: [PATCH 2/4] fix(stream): handle continuous retrieve during checkpoint procedure. --- include/libs/stream/tstream.h | 3 +- source/dnode/mnode/impl/src/mndStreamUtil.c | 2 +- source/dnode/vnode/src/tq/tq.c | 2 +- source/dnode/vnode/src/tqCommon/tqCommon.c | 22 +++++-- source/libs/executor/src/executor.c | 2 +- source/libs/executor/src/querytask.c | 2 +- source/libs/stream/src/streamCheckStatus.c | 7 +- source/libs/stream/src/streamCheckpoint.c | 31 +++++---- source/libs/stream/src/streamDispatch.c | 71 +++++++++++++-------- source/libs/stream/src/streamExec.c | 47 ++++++++------ source/libs/stream/src/streamMeta.c | 11 +--- source/libs/stream/src/streamQueue.c | 4 +- source/libs/stream/src/streamTask.c | 5 +- 13 files changed, 117 insertions(+), 92 deletions(-) diff --git a/include/libs/stream/tstream.h b/include/libs/stream/tstream.h index 20f91106a5..31b9f62346 100644 --- a/include/libs/stream/tstream.h +++ b/include/libs/stream/tstream.h @@ -705,7 +705,7 @@ int32_t streamTaskSetActiveCheckpointInfo(SStreamTask* pTask, int64_t activeChec void streamTaskSetFailedChkptInfo(SStreamTask* pTask, int32_t transId, int64_t checkpointId); bool streamTaskAlreadySendTrigger(SStreamTask* pTask, int32_t downstreamNodeId); void streamTaskGetTriggerRecvStatus(SStreamTask* pTask, int32_t* pRecved, int32_t* pTotal); -void streamTaskInitTriggerDispatchInfo(SStreamTask* pTask); +int32_t streamTaskInitTriggerDispatchInfo(SStreamTask* pTask); void streamTaskSetTriggerDispatchConfirmed(SStreamTask* pTask, int32_t vgId); int32_t streamTaskSendCheckpointTriggerMsg(SStreamTask* pTask, int32_t dstTaskId, int32_t downstreamNodeId, SRpcHandleInfo* pInfo, int32_t code); @@ -810,6 +810,7 @@ int32_t streamTaskBuildCheckpointSourceRsp(SStreamCheckpointSourceReq* pReq, SRp int32_t streamSendChkptReportMsg(SStreamTask* pTask, SCheckpointInfo* pCheckpointInfo, int8_t dropRelHTask); int32_t streamTaskUpdateTaskCheckpointInfo(SStreamTask* pTask, bool restored, SVUpdateCheckpointInfoReq* pReq); int32_t streamTaskCreateActiveChkptInfo(SActiveCheckpointInfo** pRes); +void streamTaskSetCheckpointFailed(SStreamTask* pTask); // stream task state machine, and event handling int32_t streamCreateStateMachine(SStreamTask* pTask); diff --git a/source/dnode/mnode/impl/src/mndStreamUtil.c b/source/dnode/mnode/impl/src/mndStreamUtil.c index ef91ccef34..d75713fd28 100644 --- a/source/dnode/mnode/impl/src/mndStreamUtil.c +++ b/source/dnode/mnode/impl/src/mndStreamUtil.c @@ -251,7 +251,7 @@ void mndKillTransImpl(SMnode *pMnode, int32_t transId, const char *pDbName) { int32_t code = mndKillTrans(pMnode, pTrans); mndReleaseTrans(pMnode, pTrans); if (code) { - mError("failed to kill trans:%d", pTrans->id); + mError("failed to kill transId:%d, code:%s", pTrans->id, tstrerror(code)); } } else { mError("failed to acquire trans in Db:%s, transId:%d", pDbName, transId); diff --git a/source/dnode/vnode/src/tq/tq.c b/source/dnode/vnode/src/tq/tq.c index 07daab4459..de295f2611 100644 --- a/source/dnode/vnode/src/tq/tq.c +++ b/source/dnode/vnode/src/tq/tq.c @@ -1129,7 +1129,7 @@ int32_t tqProcessTaskCheckPointSourceReq(STQ* pTq, SRpcMsg* pMsg, SRpcMsg* pRsp) SStreamTask* pTask = NULL; code = streamMetaAcquireTask(pMeta, req.streamId, req.taskId, &pTask); - if (pTask == NULL) { + if (pTask == NULL || code != 0) { tqError("vgId:%d failed to find s-task:0x%x, ignore checkpoint msg. checkpointId:%" PRId64 " transId:%d it may have been destroyed", vgId, req.taskId, req.checkpointId, req.transId); diff --git a/source/dnode/vnode/src/tqCommon/tqCommon.c b/source/dnode/vnode/src/tqCommon/tqCommon.c index 44b3f75289..68f43d637b 100644 --- a/source/dnode/vnode/src/tqCommon/tqCommon.c +++ b/source/dnode/vnode/src/tqCommon/tqCommon.c @@ -410,7 +410,7 @@ int32_t tqStreamTaskProcessRetrieveReq(SStreamMeta* pMeta, SRpcMsg* pMsg) { tDecoderClear(&decoder); if (code) { - tqError("vgId:%d failed to decode retrieve msg, quit handling it", pMeta->vgId); + tqError("vgId:%d failed to decode retrieve msg, discard it", pMeta->vgId); return code; } @@ -420,9 +420,16 @@ int32_t tqStreamTaskProcessRetrieveReq(SStreamMeta* pMeta, SRpcMsg* pMsg) { tqError("vgId:%d process retrieve req, failed to acquire task:0x%x, it may have been dropped already", pMeta->vgId, req.dstTaskId); tCleanupStreamRetrieveReq(&req); - return -1; + return code; } + // enqueue + tqDebug("s-task:%s (vgId:%d level:%d) recv retrieve req from task:0x%x(vgId:%d),QID:0x%" PRIx64, pTask->id.idStr, + pTask->pMeta->vgId, pTask->info.taskLevel, req.srcTaskId, req.srcNodeId, req.reqId); + + // if task is in ck status, set current ck failed + streamTaskSetCheckpointFailed(pTask); + if (pTask->info.taskLevel == TASK_LEVEL__SOURCE) { code = streamProcessRetrieveReq(pTask, &req); } else { @@ -431,14 +438,19 @@ int32_t tqStreamTaskProcessRetrieveReq(SStreamMeta* pMeta, SRpcMsg* pMsg) { code = streamTaskBroadcastRetrieveReq(pTask, &req); } - SRpcMsg rsp = {.info = pMsg->info, .code = 0}; - streamTaskSendRetrieveRsp(&req, &rsp); + if (code != TSDB_CODE_SUCCESS) { // return error not send rsp manually + tqError("s-task:0x%x vgId:%d failed to process retrieve request from 0x%x, code:%s", req.dstTaskId, req.dstNodeId, + req.srcTaskId, tstrerror(code)); + } else { // send rsp manually only on success. + SRpcMsg rsp = {.info = pMsg->info, .code = 0}; + streamTaskSendRetrieveRsp(&req, &rsp); + } streamMetaReleaseTask(pMeta, pTask); tCleanupStreamRetrieveReq(&req); // always return success, to disable the auto rsp - return TSDB_CODE_SUCCESS; + return code; } int32_t tqStreamTaskProcessCheckReq(SStreamMeta* pMeta, SRpcMsg* pMsg) { diff --git a/source/libs/executor/src/executor.c b/source/libs/executor/src/executor.c index c13104fc07..6620e2b934 100644 --- a/source/libs/executor/src/executor.c +++ b/source/libs/executor/src/executor.c @@ -601,7 +601,7 @@ int32_t qCreateExecTask(SReadHandle* readHandle, int32_t vgId, uint64_t taskId, SExecTaskInfo** pTask = (SExecTaskInfo**)pTaskInfo; (void)taosThreadOnce(&initPoolOnce, initRefPool); - qDebug("start to create task, TID:0x%" PRIx64 "QID:0x%" PRIx64 ", vgId:%d", taskId, pSubplan->id.queryId, vgId); + qDebug("start to create task, TID:0x%" PRIx64 " QID:0x%" PRIx64 ", vgId:%d", taskId, pSubplan->id.queryId, vgId); int32_t code = createExecTaskInfo(pSubplan, pTask, readHandle, taskId, vgId, sql, model); if (code != TSDB_CODE_SUCCESS || NULL == *pTask) { diff --git a/source/libs/executor/src/querytask.c b/source/libs/executor/src/querytask.c index b050143ac0..fc6c9f2861 100644 --- a/source/libs/executor/src/querytask.c +++ b/source/libs/executor/src/querytask.c @@ -287,7 +287,7 @@ void buildTaskId(uint64_t taskId, uint64_t queryId, char* dst) { memcpy(p, "TID:0x", offset); offset += tintToHex(taskId, &p[offset]); - memcpy(&p[offset], "QID:0x", 7); + memcpy(&p[offset], " QID:0x", 7); offset += 7; offset += tintToHex(queryId, &p[offset]); diff --git a/source/libs/stream/src/streamCheckStatus.c b/source/libs/stream/src/streamCheckStatus.c index 91196f31e0..41124d8543 100644 --- a/source/libs/stream/src/streamCheckStatus.c +++ b/source/libs/stream/src/streamCheckStatus.c @@ -65,12 +65,7 @@ int32_t streamTaskCheckStatus(SStreamTask* pTask, int32_t upstreamTaskId, int32_ ", prev:%" PRId64, id, upstreamTaskId, vgId, stage, pInfo->stage); // record the checkpoint failure id and sent to mnode - streamMutexLock(&pTask->lock); - ETaskStatus status = streamTaskGetStatus(pTask).state; - if (status == TASK_STATUS__CK) { - streamTaskSetFailedCheckpointId(pTask); - } - streamMutexUnlock(&pTask->lock); + streamTaskSetCheckpointFailed(pTask); } if (pInfo->stage != stage) { diff --git a/source/libs/stream/src/streamCheckpoint.c b/source/libs/stream/src/streamCheckpoint.c index d0bf24bd03..916aee4e6e 100644 --- a/source/libs/stream/src/streamCheckpoint.c +++ b/source/libs/stream/src/streamCheckpoint.c @@ -673,6 +673,15 @@ void streamTaskSetFailedCheckpointId(SStreamTask* pTask) { } } +void streamTaskSetCheckpointFailed(SStreamTask* pTask) { + streamMutexLock(&pTask->lock); + ETaskStatus status = streamTaskGetStatus(pTask).state; + if (status == TASK_STATUS__CK) { + streamTaskSetFailedCheckpointId(pTask); + } + streamMutexUnlock(&pTask->lock); +} + static int32_t getCheckpointDataMeta(const char* id, const char* path, SArray* list) { int32_t code = 0; int32_t cap = strlen(path) + 64; @@ -1111,26 +1120,20 @@ void streamTaskGetTriggerRecvStatus(SStreamTask* pTask, int32_t* pRecved, int32_ // record the dispatch checkpoint trigger info in the list // memory insufficient may cause the stream computing stopped -void streamTaskInitTriggerDispatchInfo(SStreamTask* pTask) { +int32_t streamTaskInitTriggerDispatchInfo(SStreamTask* pTask) { SActiveCheckpointInfo* pInfo = pTask->chkInfo.pActiveInfo; + int64_t now = taosGetTimestampMs(); - int64_t now = taosGetTimestampMs(); streamMutexLock(&pInfo->lock); - - // outputQ should be empty here - if (streamQueueGetNumOfUnAccessedItems(pTask->outputq.queue) > 0) { - stFatal("s-task:%s items are still in outputQ, failed to init trigger dispatch info", pTask->id.idStr); - return; - } - pInfo->dispatchTrigger = true; if (pTask->outputInfo.type == TASK_OUTPUT__FIXED_DISPATCH) { STaskDispatcherFixed* pDispatch = &pTask->outputInfo.fixedDispatcher; STaskTriggerSendInfo p = {.sendTs = now, .recved = false, .nodeId = pDispatch->nodeId, .taskId = pDispatch->taskId}; void* px = taosArrayPush(pInfo->pDispatchTriggerList, &p); - if (px == NULL) { - // pause the stream task, if memory not enough + if (px == NULL) { // pause the stream task, if memory not enough + streamMutexUnlock(&pInfo->lock); + return terrno; } } else { for (int32_t i = 0; i < streamTaskGetNumOfDownstream(pTask); ++i) { @@ -1141,13 +1144,15 @@ void streamTaskInitTriggerDispatchInfo(SStreamTask* pTask) { STaskTriggerSendInfo p = {.sendTs = now, .recved = false, .nodeId = pVgInfo->vgId, .taskId = pVgInfo->taskId}; void* px = taosArrayPush(pInfo->pDispatchTriggerList, &p); - if (px == NULL) { - // pause the stream task, if memory not enough + if (px == NULL) { // pause the stream task, if memory not enough + streamMutexUnlock(&pInfo->lock); + return terrno; } } } streamMutexUnlock(&pInfo->lock); + return 0; } int32_t streamTaskGetNumOfConfirmed(SActiveCheckpointInfo* pInfo) { diff --git a/source/libs/stream/src/streamDispatch.c b/source/libs/stream/src/streamDispatch.c index 0bc090cdfe..36b35bd3f6 100644 --- a/source/libs/stream/src/streamDispatch.c +++ b/source/libs/stream/src/streamDispatch.c @@ -726,8 +726,11 @@ int32_t streamSearchAndAddBlock(SStreamTask* pTask, SStreamDispatchReq* pReqs, S } int32_t streamDispatchStreamBlock(SStreamTask* pTask) { - const char* id = pTask->id.idStr; - int32_t numOfElems = streamQueueGetNumOfItems(pTask->outputq.queue); + const char* id = pTask->id.idStr; + int32_t code = 0; + SStreamDataBlock* pBlock = NULL; + + int32_t numOfElems = streamQueueGetNumOfItems(pTask->outputq.queue); if (numOfElems > 0) { double size = SIZE_IN_MiB(taosQueueMemorySize(pTask->outputq.queue->pQueue)); int32_t numOfUnAccessed = streamQueueGetNumOfUnAccessedItems(pTask->outputq.queue); @@ -749,41 +752,57 @@ int32_t streamDispatchStreamBlock(SStreamTask* pTask) { return 0; } + ASSERT(pTask->msgInfo.pData == NULL); + if (pTask->msgInfo.pData != NULL) { stFatal("s-task:%s not rsp data:%p exist, should not dispatch msg now", id, pTask->msgInfo.pData); } else { stDebug("s-task:%s start to dispatch msg, set output status:%d", id, pTask->outputq.status); } - SStreamDataBlock* pBlock = NULL; - streamQueueNextItem(pTask->outputq.queue, (SStreamQueueItem**)&pBlock); - if (pBlock == NULL) { - atomic_store_8(&pTask->outputq.status, TASK_OUTPUT_STATUS__NORMAL); - stDebug("s-task:%s not dispatch since no elems in outputQ, output status:%d", id, pTask->outputq.status); - return 0; - } + while (1) { + streamQueueNextItem(pTask->outputq.queue, (SStreamQueueItem**)&pBlock); + if (pBlock == NULL) { + atomic_store_8(&pTask->outputq.status, TASK_OUTPUT_STATUS__NORMAL); + stDebug("s-task:%s not dispatch since no elems in outputQ, output status:%d", id, pTask->outputq.status); + return 0; + } - int32_t type = pBlock->type; - if (!(type == STREAM_INPUT__DATA_BLOCK || type == STREAM_INPUT__CHECKPOINT_TRIGGER || - type == STREAM_INPUT__TRANS_STATE)) { - stError("s-task:%s invalid dispatch block type:%d", id, type); - return TSDB_CODE_INTERNAL_ERROR; - } + int32_t type = pBlock->type; + if (!(type == STREAM_INPUT__DATA_BLOCK || type == STREAM_INPUT__CHECKPOINT_TRIGGER || + type == STREAM_INPUT__TRANS_STATE)) { + stError("s-task:%s invalid dispatch block type:%d", id, type); + return TSDB_CODE_INTERNAL_ERROR; + } - pTask->execInfo.dispatch += 1; + pTask->execInfo.dispatch += 1; - streamMutexLock(&pTask->msgInfo.lock); - initDispatchInfo(&pTask->msgInfo, pTask->execInfo.dispatch); - streamMutexUnlock(&pTask->msgInfo.lock); + streamMutexLock(&pTask->msgInfo.lock); + initDispatchInfo(&pTask->msgInfo, pTask->execInfo.dispatch); + streamMutexUnlock(&pTask->msgInfo.lock); - int32_t code = doBuildDispatchMsg(pTask, pBlock); - if (code == 0) { - destroyStreamDataBlock(pBlock); - } else { // todo handle build dispatch msg failed - } + code = doBuildDispatchMsg(pTask, pBlock); + if (code == 0) { + destroyStreamDataBlock(pBlock); + } else { // todo handle build dispatch msg failed + } - if (type == STREAM_INPUT__CHECKPOINT_TRIGGER) { - streamTaskInitTriggerDispatchInfo(pTask); + if (type == STREAM_INPUT__CHECKPOINT_TRIGGER) { + // outputQ should be empty here, otherwise, set the checkpoint failed due to the retrieve req happens + if (streamQueueGetNumOfUnAccessedItems(pTask->outputq.queue) > 0) { + stError("s-task:%s items are still in outputQ due to downstream retrieve, failed to init trigger dispatch", + pTask->id.idStr); + streamTaskSetCheckpointFailed(pTask); + clearBufferedDispatchMsg(pTask); + continue; + } + + code = streamTaskInitTriggerDispatchInfo(pTask); + if (code != TSDB_CODE_SUCCESS) { // todo handle error + } + } + + break; } code = sendDispatchMsg(pTask, pTask->msgInfo.pData); diff --git a/source/libs/stream/src/streamExec.c b/source/libs/stream/src/streamExec.c index fca0bf403f..0ac37fd2b9 100644 --- a/source/libs/stream/src/streamExec.c +++ b/source/libs/stream/src/streamExec.c @@ -98,14 +98,13 @@ static int32_t doDumpResult(SStreamTask* pTask, SStreamQueueItem* pItem, SArray* void streamTaskExecImpl(SStreamTask* pTask, SStreamQueueItem* pItem, int64_t* totalSize, int32_t* totalBlocks) { int32_t code = TSDB_CODE_SUCCESS; void* pExecutor = pTask->exec.pExecutor; - - *totalBlocks = 0; - *totalSize = 0; - int32_t size = 0; int32_t numOfBlocks = 0; SArray* pRes = NULL; + *totalBlocks = 0; + *totalSize = 0; + while (1) { if (pRes == NULL) { pRes = taosArrayInit(4, sizeof(SSDataBlock)); @@ -131,7 +130,8 @@ void streamTaskExecImpl(SStreamTask* pTask, SStreamQueueItem* pItem, int64_t* to if (pItem->type == STREAM_INPUT__DATA_RETRIEVE) { SSDataBlock block = {0}; const SStreamDataBlock* pRetrieveBlock = (const SStreamDataBlock*)pItem; - int32_t num = taosArrayGetSize(pRetrieveBlock->blocks); + + int32_t num = taosArrayGetSize(pRetrieveBlock->blocks); if (num != 1) { stError("s-task:%s invalid retrieve block number:%d, ignore", pTask->id.idStr, num); continue; @@ -596,12 +596,32 @@ void streamProcessTransstateBlock(SStreamTask* pTask, SStreamDataBlock* pBlock) // static void streamTaskSetIdleInfo(SStreamTask* pTask, int32_t idleTime) { pTask->status.schedIdleTime = idleTime; } static void setLastExecTs(SStreamTask* pTask, int64_t ts) { pTask->status.lastExecTs = ts; } +static void doRecordThroughput(STaskExecStatisInfo* pInfo, int64_t totalBlocks, int64_t totalSize, int64_t blockSize, + double st, const char* id) { + double el = (taosGetTimestampMs() - st) / 1000.0; + + stDebug("s-task:%s batch of input blocks exec end, elapsed time:%.2fs, result size:%.2fMiB, numOfBlocks:%d", id, el, + SIZE_IN_MiB(totalSize), totalBlocks); + + pInfo->outputDataBlocks += totalBlocks; + pInfo->outputDataSize += totalSize; + if (fabs(el - 0.0) <= DBL_EPSILON) { + pInfo->procsThroughput = 0; + pInfo->outputThroughput = 0; + } else { + pInfo->outputThroughput = (totalSize / el); + pInfo->procsThroughput = (blockSize / el); + } +} + static void doStreamTaskExecImpl(SStreamTask* pTask, SStreamQueueItem* pBlock, int32_t num) { const char* id = pTask->id.idStr; int32_t blockSize = 0; int64_t st = taosGetTimestampMs(); SCheckpointInfo* pInfo = &pTask->chkInfo; int64_t ver = pInfo->processedVer; + int64_t totalSize = 0; + int32_t totalBlocks = 0; stDebug("s-task:%s start to process batch blocks, num:%d, type:%s", id, num, streamQueueItemGetTypeStr(pBlock->type)); @@ -611,23 +631,8 @@ static void doStreamTaskExecImpl(SStreamTask* pTask, SStreamQueueItem* pBlock, i return; } - int64_t totalSize = 0; - int32_t totalBlocks = 0; streamTaskExecImpl(pTask, pBlock, &totalSize, &totalBlocks); - - double el = (taosGetTimestampMs() - st) / 1000.0; - stDebug("s-task:%s batch of input blocks exec end, elapsed time:%.2fs, result size:%.2fMiB, numOfBlocks:%d", id, el, - SIZE_IN_MiB(totalSize), totalBlocks); - - pTask->execInfo.outputDataBlocks += totalBlocks; - pTask->execInfo.outputDataSize += totalSize; - if (fabs(el - 0.0) <= DBL_EPSILON) { - pTask->execInfo.procsThroughput = 0; - pTask->execInfo.outputThroughput = 0; - } else { - pTask->execInfo.outputThroughput = (totalSize / el); - pTask->execInfo.procsThroughput = (blockSize / el); - } + doRecordThroughput(&pTask->execInfo, totalBlocks, totalSize, blockSize, st, pTask->id.idStr); // update the currentVer if processing the submit blocks. if (!(pInfo->checkpointVer <= pInfo->nextProcessVer && ver >= pInfo->checkpointVer)) { diff --git a/source/libs/stream/src/streamMeta.c b/source/libs/stream/src/streamMeta.c index 084eb7b827..81061cd06d 100644 --- a/source/libs/stream/src/streamMeta.c +++ b/source/libs/stream/src/streamMeta.c @@ -1254,16 +1254,7 @@ int32_t streamMetaSendMsgBeforeCloseTasks(SStreamMeta* pMeta, SArray** pList) { continue; } - streamMutexLock(&pTask->lock); - - SStreamTaskState pState = streamTaskGetStatus(pTask); - if (pState.state == TASK_STATUS__CK) { - streamTaskSetFailedCheckpointId(pTask); - } else { - stDebug("s-task:%s status:%s not reset the checkpoint", pTask->id.idStr, pState.name); - } - - streamMutexUnlock(&pTask->lock); + streamTaskSetCheckpointFailed(pTask); streamMetaReleaseTask(pMeta, pTask); } diff --git a/source/libs/stream/src/streamQueue.c b/source/libs/stream/src/streamQueue.c index eca728b2d5..3765d0f9d3 100644 --- a/source/libs/stream/src/streamQueue.c +++ b/source/libs/stream/src/streamQueue.c @@ -287,7 +287,7 @@ int32_t streamTaskPutDataIntoInputQ(SStreamTask* pTask, SStreamQueueItem* pItem) "s-task:%s inputQ is full, capacity(size:%d num:%dMiB), current(blocks:%d, size:%.2fMiB) stop to push data", pTask->id.idStr, STREAM_TASK_QUEUE_CAPACITY, STREAM_TASK_QUEUE_CAPACITY_IN_SIZE, total, size); streamDataSubmitDestroy(px); - return -1; + return TSDB_CODE_OUT_OF_MEMORY; } int32_t msgLen = px->submit.msgLen; @@ -312,7 +312,7 @@ int32_t streamTaskPutDataIntoInputQ(SStreamTask* pTask, SStreamQueueItem* pItem) stTrace("s-task:%s input queue is full, capacity:%d size:%d MiB, current(blocks:%d, size:%.2fMiB) abort", pTask->id.idStr, STREAM_TASK_QUEUE_CAPACITY, STREAM_TASK_QUEUE_CAPACITY_IN_SIZE, total, size); streamFreeQitem(pItem); - return -1; + return TSDB_CODE_OUT_OF_MEMORY; } int32_t code = taosWriteQitem(pQueue, pItem); diff --git a/source/libs/stream/src/streamTask.c b/source/libs/stream/src/streamTask.c index 0791784656..2acfd64b2a 100644 --- a/source/libs/stream/src/streamTask.c +++ b/source/libs/stream/src/streamTask.c @@ -1098,15 +1098,12 @@ static int32_t streamTaskEnqueueRetrieve(SStreamTask* pTask, SStreamRetrieveReq* return terrno = code; } - // enqueue - stDebug("s-task:%s (vgId:%d level:%d) recv retrieve req from task:0x%x(vgId:%d),QID:0x%" PRIx64, pTask->id.idStr, - pTask->pMeta->vgId, pTask->info.taskLevel, pReq->srcTaskId, pReq->srcNodeId, pReq->reqId); - pData->type = STREAM_INPUT__DATA_RETRIEVE; pData->srcVgId = 0; code = streamRetrieveReqToData(pReq, pData, pTask->id.idStr); if (code != TSDB_CODE_SUCCESS) { + stError("s-task:%s failed to convert retrieve-data to block, code:%s", pTask->id.idStr, tstrerror(code)); taosFreeQitem(pData); return code; } From df0f71a31c8db9aca3fc4cc991c8f7313af38a03 Mon Sep 17 00:00:00 2001 From: Haojun Liao Date: Thu, 5 Sep 2024 16:12:39 +0800 Subject: [PATCH 3/4] refactor: remove assert. --- source/libs/stream/src/streamDispatch.c | 2 -- 1 file changed, 2 deletions(-) diff --git a/source/libs/stream/src/streamDispatch.c b/source/libs/stream/src/streamDispatch.c index 36b35bd3f6..ad1866807a 100644 --- a/source/libs/stream/src/streamDispatch.c +++ b/source/libs/stream/src/streamDispatch.c @@ -752,8 +752,6 @@ int32_t streamDispatchStreamBlock(SStreamTask* pTask) { return 0; } - ASSERT(pTask->msgInfo.pData == NULL); - if (pTask->msgInfo.pData != NULL) { stFatal("s-task:%s not rsp data:%p exist, should not dispatch msg now", id, pTask->msgInfo.pData); } else { From 6b945732319bc88128b2214e4bbfb80b3088907e Mon Sep 17 00:00:00 2001 From: Haojun Liao Date: Thu, 5 Sep 2024 17:17:26 +0800 Subject: [PATCH 4/4] fix(stream): fix syntax error. --- source/libs/stream/src/streamExec.c | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/source/libs/stream/src/streamExec.c b/source/libs/stream/src/streamExec.c index 0ac37fd2b9..5bb9c993de 100644 --- a/source/libs/stream/src/streamExec.c +++ b/source/libs/stream/src/streamExec.c @@ -600,8 +600,8 @@ static void doRecordThroughput(STaskExecStatisInfo* pInfo, int64_t totalBlocks, double st, const char* id) { double el = (taosGetTimestampMs() - st) / 1000.0; - stDebug("s-task:%s batch of input blocks exec end, elapsed time:%.2fs, result size:%.2fMiB, numOfBlocks:%d", id, el, - SIZE_IN_MiB(totalSize), totalBlocks); + stDebug("s-task:%s batch of input blocks exec end, elapsed time:%.2fs, result size:%.2fMiB, numOfBlocks:%" PRId64, id, + el, SIZE_IN_MiB(totalSize), totalBlocks); pInfo->outputDataBlocks += totalBlocks; pInfo->outputDataSize += totalSize;