From 3527860c75b03b10b78b8843889b357621a4e5fe Mon Sep 17 00:00:00 2001 From: Haojun Liao Date: Thu, 23 Nov 2023 10:00:34 +0800 Subject: [PATCH 1/8] refactor: do some internal refactor. --- contrib/CMakeLists.txt | 86 ------------------------------------------ 1 file changed, 86 deletions(-) diff --git a/contrib/CMakeLists.txt b/contrib/CMakeLists.txt index c5715bd53f..b786f2df25 100644 --- a/contrib/CMakeLists.txt +++ b/contrib/CMakeLists.txt @@ -109,11 +109,6 @@ cat("${TD_SUPPORT_DIR}/zlib_CMakeLists.txt.in" ${CONTRIB_TMP_FILE}) # cJson cat("${TD_SUPPORT_DIR}/cjson_CMakeLists.txt.in" ${CONTRIB_TMP_FILE}) -# leveldb -if(${BUILD_WITH_LEVELDB}) - cat("${TD_SUPPORT_DIR}/leveldb_CMakeLists.txt.in" ${CONTRIB_TMP_FILE}) -endif(${BUILD_WITH_LEVELDB}) - if (${BUILD_CONTRIB}) if(${BUILD_WITH_ROCKSDB}) cat("${TD_SUPPORT_DIR}/rocksdb_CMakeLists.txt.in" ${CONTRIB_TMP_FILE}) @@ -132,28 +127,11 @@ else() endif() endif() -# canonical-raft -if(${BUILD_WITH_CRAFT}) - cat("${TD_SUPPORT_DIR}/craft_CMakeLists.txt.in" ${CONTRIB_TMP_FILE}) - SET(BUILD_WITH_UV ON CACHE BOOL "craft need libuv" FORCE) -endif(${BUILD_WITH_CRAFT}) - -# traft -if(${BUILD_WITH_TRAFT}) - cat("${TD_SUPPORT_DIR}/traft_CMakeLists.txt.in" ${CONTRIB_TMP_FILE}) - SET(BUILD_WITH_UV ON CACHE BOOL "traft need libuv" FORCE) -endif(${BUILD_WITH_TRAFT}) - #libuv if(${BUILD_WITH_UV}) cat("${TD_SUPPORT_DIR}/libuv_CMakeLists.txt.in" ${CONTRIB_TMP_FILE}) endif(${BUILD_WITH_UV}) -# bdb -if(${BUILD_WITH_BDB}) - cat("${TD_SUPPORT_DIR}/bdb_CMakeLists.txt.in" ${CONTRIB_TMP_FILE}) -endif(${BUILD_WITH_BDB}) - # sqlite if(${BUILD_WITH_SQLITE}) cat("${TD_SUPPORT_DIR}/sqlite_CMakeLists.txt.in" ${CONTRIB_TMP_FILE}) @@ -178,17 +156,6 @@ elseif(${BUILD_WITH_COS}) endif() -# lucene -if(${BUILD_WITH_LUCENE}) - cat("${TD_SUPPORT_DIR}/lucene_CMakeLists.txt.in" ${CONTRIB_TMP_FILE}) - add_definitions(-DUSE_LUCENE) -endif(${BUILD_WITH_LUCENE}) - -# NuRaft -if(${BUILD_WITH_NURAFT}) - cat("${TD_SUPPORT_DIR}/nuraft_CMakeLists.txt.in" ${CONTRIB_TMP_FILE}) -endif(${BUILD_WITH_NURAFT}) - # crashdump if(${BUILD_CRASHDUMP}) cat("${TD_SUPPORT_DIR}/crashdump_CMakeLists.txt.in" ${CONTRIB_TMP_FILE}) @@ -437,23 +404,6 @@ elseif(${BUILD_WITH_COS}) endif() -# lucene -# To support build on ubuntu: sudo apt-get install libboost-all-dev -if(${BUILD_WITH_LUCENE}) - option(ENABLE_TEST "Enable the tests" OFF) - add_subdirectory(lucene EXCLUDE_FROM_ALL) - target_include_directories( - lucene++ - PUBLIC $ - ) - -endif(${BUILD_WITH_LUCENE}) - -# NuRaft -if(${BUILD_WITH_NURAFT}) - add_subdirectory(nuraft EXCLUDE_FROM_ALL) -endif(${BUILD_WITH_NURAFT}) - # pthread if(${BUILD_PTHREAD}) set(CMAKE_BUILD_TYPE debug) @@ -524,30 +474,6 @@ if(${BUILD_WCWIDTH}) SET_TARGET_PROPERTIES(wcwidth PROPERTIES OUTPUT_NAME wcwidth) endif(${BUILD_WCWIDTH}) -# CRAFT -if(${BUILD_WITH_CRAFT}) - add_library(craft STATIC IMPORTED GLOBAL) - set_target_properties(craft PROPERTIES - IMPORTED_LOCATION "${CMAKE_CURRENT_SOURCE_DIR}/craft/.libs/libraft.a" - INTERFACE_INCLUDE_DIRECTORIES "${CMAKE_CURRENT_SOURCE_DIR}/craft/include" - ) - # target_link_libraries(craft - # INTERFACE pthread - # ) -endif(${BUILD_WITH_CRAFT}) - -# TRAFT -if(${BUILD_WITH_TRAFT}) - add_library(traft STATIC IMPORTED GLOBAL) - set_target_properties(traft PROPERTIES - IMPORTED_LOCATION "${CMAKE_CURRENT_SOURCE_DIR}/traft/.libs/libraft.a" - INTERFACE_INCLUDE_DIRECTORIES "${CMAKE_CURRENT_SOURCE_DIR}/traft/include" - ) - # target_link_libraries(craft - # INTERFACE pthread - # ) -endif(${BUILD_WITH_TRAFT}) - # LIBUV if(${BUILD_WITH_UV}) if (TD_WINDOWS) @@ -559,18 +485,6 @@ if(${BUILD_WITH_UV}) add_subdirectory(libuv EXCLUDE_FROM_ALL) endif(${BUILD_WITH_UV}) -# BDB -if(${BUILD_WITH_BDB}) - add_library(bdb STATIC IMPORTED GLOBAL) - set_target_properties(bdb PROPERTIES - IMPORTED_LOCATION "${CMAKE_CURRENT_SOURCE_DIR}/bdb/libdb.a" - INTERFACE_INCLUDE_DIRECTORIES "${CMAKE_CURRENT_SOURCE_DIR}/bdb" - ) - target_link_libraries(bdb - INTERFACE pthread - ) -endif(${BUILD_WITH_BDB}) - # SQLite # see https://stackoverflow.com/questions/8774593/cmake-link-to-external-library#comment58570736_10550334 if(${BUILD_WITH_SQLITE}) From 9814109175400b8812d72517d5e734f986bf9dfc Mon Sep 17 00:00:00 2001 From: Haojun Liao Date: Thu, 23 Nov 2023 10:00:50 +0800 Subject: [PATCH 2/8] refactor: do some internal refactor. --- cmake/bdb_CMakeLists.txt.in | 13 ------------- cmake/leveldb_CMakeLists.txt.in | 13 ------------- cmake/lucene_CMakeLists.txt.in | 12 ------------ 3 files changed, 38 deletions(-) delete mode 100644 cmake/bdb_CMakeLists.txt.in delete mode 100644 cmake/leveldb_CMakeLists.txt.in delete mode 100644 cmake/lucene_CMakeLists.txt.in diff --git a/cmake/bdb_CMakeLists.txt.in b/cmake/bdb_CMakeLists.txt.in deleted file mode 100644 index dd21020105..0000000000 --- a/cmake/bdb_CMakeLists.txt.in +++ /dev/null @@ -1,13 +0,0 @@ - -# bdb -ExternalProject_Add(bdb - GIT_REPOSITORY https://github.com/berkeleydb/libdb.git - GIT_TAG v5.3.28 - SOURCE_DIR "${TD_CONTRIB_DIR}/bdb" - BINARY_DIR "${TD_CONTRIB_DIR}/bdb" - #BUILD_IN_SOURCE TRUE - CONFIGURE_COMMAND COMMAND ./dist/configure --enable-debug - BUILD_COMMAND "$(MAKE)" - INSTALL_COMMAND "" - TEST_COMMAND "" -) diff --git a/cmake/leveldb_CMakeLists.txt.in b/cmake/leveldb_CMakeLists.txt.in deleted file mode 100644 index 6878711bc1..0000000000 --- a/cmake/leveldb_CMakeLists.txt.in +++ /dev/null @@ -1,13 +0,0 @@ - -# leveldb -ExternalProject_Add(leveldb - GIT_REPOSITORY https://github.com/taosdata-contrib/leveldb.git - GIT_TAG master - SOURCE_DIR "${TD_CONTRIB_DIR}/leveldb" - BINARY_DIR "" - #BUILD_IN_SOURCE TRUE - CONFIGURE_COMMAND "" - BUILD_COMMAND "" - INSTALL_COMMAND "" - TEST_COMMAND "" -) \ No newline at end of file diff --git a/cmake/lucene_CMakeLists.txt.in b/cmake/lucene_CMakeLists.txt.in deleted file mode 100644 index 9fd7471705..0000000000 --- a/cmake/lucene_CMakeLists.txt.in +++ /dev/null @@ -1,12 +0,0 @@ - -# lucene -ExternalProject_Add(lucene - GIT_REPOSITORY https://github.com/yihaoDeng/LucenePlusPlus.git - SOURCE_DIR "${TD_CONTRIB_DIR}/lucene" - BINARY_DIR "" - #BUILD_IN_SOURCE TRUE - CONFIGURE_COMMAND "" - BUILD_COMMAND "" - INSTALL_COMMAND "" - TEST_COMMAND "" -) From 281f25fdcc71ad5fd8705890b9e96163b6569e68 Mon Sep 17 00:00:00 2001 From: Haojun Liao Date: Thu, 23 Nov 2023 10:01:42 +0800 Subject: [PATCH 3/8] refactor: do some internal refactor. --- cmake/nuraft_CMakeLists.txt.in | 12 ------------ 1 file changed, 12 deletions(-) delete mode 100644 cmake/nuraft_CMakeLists.txt.in diff --git a/cmake/nuraft_CMakeLists.txt.in b/cmake/nuraft_CMakeLists.txt.in deleted file mode 100644 index 593c6fed26..0000000000 --- a/cmake/nuraft_CMakeLists.txt.in +++ /dev/null @@ -1,12 +0,0 @@ - -# NuRaft -ExternalProject_Add(NuRaft - GIT_REPOSITORY https://github.com/eBay/NuRaft.git - GIT_TAG v1.3.0 - SOURCE_DIR "${TD_CONTRIB_DIR}/nuraft" - BINARY_DIR "${TD_CONTRIB_DIR}/nuraft" - CONFIGURE_COMMAND "./prepare.sh" - BUILD_COMMAND "" - INSTALL_COMMAND "" - TEST_COMMAND "" - ) \ No newline at end of file From 1a1319908d82286ec1c65d2bdc122329ec0ca750 Mon Sep 17 00:00:00 2001 From: Haojun Liao Date: Thu, 23 Nov 2023 14:25:52 +0800 Subject: [PATCH 4/8] refactor: do some internal refactor. --- source/libs/stream/inc/streamInt.h | 20 +++++++++----------- source/libs/stream/src/streamData.c | 2 +- source/libs/stream/src/streamQueue.c | 2 +- source/libs/stream/src/streamStart.c | 4 ++-- source/libs/stream/src/streamTaskSm.c | 6 +++--- 5 files changed, 16 insertions(+), 18 deletions(-) diff --git a/source/libs/stream/inc/streamInt.h b/source/libs/stream/inc/streamInt.h index 6dd1e5c1c3..b54f19e4ce 100644 --- a/source/libs/stream/inc/streamInt.h +++ b/source/libs/stream/inc/streamInt.h @@ -113,25 +113,23 @@ int32_t streamAddCheckpointReadyMsg(SStreamTask* pTask, int32_t srcTaskId, int32 int32_t streamTaskSendCheckpointReadyMsg(SStreamTask* pTask); int32_t streamTaskSendCheckpointSourceRsp(SStreamTask* pTask); int32_t streamTaskGetNumOfDownstream(const SStreamTask* pTask); +int32_t streamTaskInitTokenBucket(STokenBucket* pBucket, int32_t numCap, int32_t numRate, float quotaRate, const char*); +STaskId streamTaskExtractKey(const SStreamTask* pTask); +void streamTaskInitForLaunchHTask(SHistoryTaskInfo* pInfo); +void streamTaskSetRetryInfoForLaunch(SHistoryTaskInfo* pInfo); +int32_t streamTaskBuildScanhistoryRspMsg(SStreamTask* pTask, SStreamScanHistoryFinishReq* pReq, void** pBuffer, int32_t* pLen); +int32_t streamTaskFillHistoryFinished(SStreamTask* pTask); int32_t streamTaskGetDataFromInputQ(SStreamTask* pTask, SStreamQueueItem** pInput, int32_t* numOfBlocks, int32_t* blockSize); int32_t streamQueueItemGetSize(const SStreamQueueItem* pItem); void streamQueueItemIncSize(const SStreamQueueItem* pItem, int32_t size); const char* streamQueueItemGetTypeStr(int32_t type); +SStreamQueueItem* streamQueueMergeQueueItem(SStreamQueueItem* dst, SStreamQueueItem* pElem); -SStreamQueueItem* streamMergeQueueItem(SStreamQueueItem* dst, SStreamQueueItem* pElem); - -int32_t streamTaskBuildScanhistoryRspMsg(SStreamTask* pTask, SStreamScanHistoryFinishReq* pReq, void** pBuffer, int32_t* pLen); int32_t streamAddEndScanHistoryMsg(SStreamTask* pTask, SRpcHandleInfo* pRpcInfo, SStreamScanHistoryFinishReq* pReq); int32_t streamNotifyUpstreamContinue(SStreamTask* pTask); -int32_t streamTaskFillHistoryFinished(SStreamTask* pTask); int32_t streamTransferStateToStreamTask(SStreamTask* pTask); -int32_t streamTaskInitTokenBucket(STokenBucket* pBucket, int32_t numCap, int32_t numRate, float quotaRate, const char*); -STaskId streamTaskExtractKey(const SStreamTask* pTask); -void streamTaskInitForLaunchHTask(SHistoryTaskInfo* pInfo); -void streamTaskSetRetryInfoForLaunch(SHistoryTaskInfo* pInfo); - SStreamQueue* streamQueueOpen(int64_t cap); void streamQueueClose(SStreamQueue* pQueue, int32_t taskId); void streamQueueProcessSuccess(SStreamQueue* queue); @@ -152,8 +150,8 @@ int downloadCheckpoint(char* id, char* path); int deleteCheckpoint(char* id); int deleteCheckpointFile(char* id, char* name); -int32_t onNormalTaskReady(SStreamTask* pTask); -int32_t onScanhistoryTaskReady(SStreamTask* pTask); +int32_t streamTaskOnNormalTaskReady(SStreamTask* pTask); +int32_t streamTaskOnScanhistoryTaskReady(SStreamTask* pTask); #ifdef __cplusplus } diff --git a/source/libs/stream/src/streamData.c b/source/libs/stream/src/streamData.c index f6ec6e9fdb..bcda85e7a7 100644 --- a/source/libs/stream/src/streamData.c +++ b/source/libs/stream/src/streamData.c @@ -158,7 +158,7 @@ int32_t streamMergeSubmit(SStreamMergedSubmit* pMerged, SStreamDataSubmit* pSubm } // todo handle memory error -SStreamQueueItem* streamMergeQueueItem(SStreamQueueItem* dst, SStreamQueueItem* pElem) { +SStreamQueueItem* streamQueueMergeQueueItem(SStreamQueueItem* dst, SStreamQueueItem* pElem) { terrno = 0; if (dst->type == STREAM_INPUT__DATA_BLOCK && pElem->type == STREAM_INPUT__DATA_BLOCK) { diff --git a/source/libs/stream/src/streamQueue.c b/source/libs/stream/src/streamQueue.c index d19dfc13bf..d1610362f9 100644 --- a/source/libs/stream/src/streamQueue.c +++ b/source/libs/stream/src/streamQueue.c @@ -221,7 +221,7 @@ int32_t streamTaskGetDataFromInputQ(SStreamTask* pTask, SStreamQueueItem** pInpu *pInput = qItem; } else { // merge current block failed, let's handle the already merged blocks. - void* newRet = streamMergeQueueItem(*pInput, qItem); + void* newRet = streamQueueMergeQueueItem(*pInput, qItem); if (newRet == NULL) { if (terrno != 0) { stError("s-task:%s failed to merge blocks from inputQ, numOfBlocks:%d, code:%s", id, *numOfBlocks, diff --git a/source/libs/stream/src/streamStart.c b/source/libs/stream/src/streamStart.c index 9b27281915..745d030e15 100644 --- a/source/libs/stream/src/streamStart.c +++ b/source/libs/stream/src/streamStart.c @@ -322,7 +322,7 @@ int32_t streamTaskCheckStatus(SStreamTask* pTask, int32_t upstreamTaskId, int32_ } } -int32_t onNormalTaskReady(SStreamTask* pTask) { +int32_t streamTaskOnNormalTaskReady(SStreamTask* pTask) { const char* id = pTask->id.idStr; streamTaskSetReady(pTask); @@ -347,7 +347,7 @@ int32_t onNormalTaskReady(SStreamTask* pTask) { return TSDB_CODE_SUCCESS; } -int32_t onScanhistoryTaskReady(SStreamTask* pTask) { +int32_t streamTaskOnScanhistoryTaskReady(SStreamTask* pTask) { const char* id = pTask->id.idStr; // set the state to be ready diff --git a/source/libs/stream/src/streamTaskSm.c b/source/libs/stream/src/streamTaskSm.c index 1c951e1452..cac3766893 100644 --- a/source/libs/stream/src/streamTaskSm.c +++ b/source/libs/stream/src/streamTaskSm.c @@ -457,11 +457,11 @@ void doInitStateTransferTable(void) { streamTaskSMTrans = taosArrayInit(8, sizeof(STaskStateTrans)); // initialization event handle - STaskStateTrans trans = createStateTransform(TASK_STATUS__UNINIT, TASK_STATUS__READY, TASK_EVENT_INIT, streamTaskInitStatus, onNormalTaskReady, false, false); + STaskStateTrans trans = createStateTransform(TASK_STATUS__UNINIT, TASK_STATUS__READY, TASK_EVENT_INIT, streamTaskInitStatus, streamTaskOnNormalTaskReady, false, false); taosArrayPush(streamTaskSMTrans, &trans); - trans = createStateTransform(TASK_STATUS__UNINIT, TASK_STATUS__SCAN_HISTORY, TASK_EVENT_INIT_SCANHIST, streamTaskInitStatus, onScanhistoryTaskReady, false, false); + trans = createStateTransform(TASK_STATUS__UNINIT, TASK_STATUS__SCAN_HISTORY, TASK_EVENT_INIT_SCANHIST, streamTaskInitStatus, streamTaskOnScanhistoryTaskReady, false, false); taosArrayPush(streamTaskSMTrans, &trans); - trans = createStateTransform(TASK_STATUS__UNINIT, TASK_STATUS__STREAM_SCAN_HISTORY, TASK_EVENT_INIT_STREAM_SCANHIST, streamTaskInitStatus, onScanhistoryTaskReady, false, false); + trans = createStateTransform(TASK_STATUS__UNINIT, TASK_STATUS__STREAM_SCAN_HISTORY, TASK_EVENT_INIT_STREAM_SCANHIST, streamTaskInitStatus, streamTaskOnScanhistoryTaskReady, false, false); taosArrayPush(streamTaskSMTrans, &trans); // scan-history related event From 0a90c33b4d1e30607852b2ee6d3b8a20d739939f Mon Sep 17 00:00:00 2001 From: Haojun Liao Date: Tue, 21 Nov 2023 18:22:38 +0800 Subject: [PATCH 5/8] fix(stream): drop the related fill-history task & set the task to be ready, if the drop fill-history task msg is missing due to the vnode restarting. --- include/common/tmsg.h | 2 +- include/common/tmsgdef.h | 1 + include/libs/stream/tstream.h | 20 +- source/dnode/mgmt/mgmt_snode/src/smHandle.c | 1 + source/dnode/mgmt/mgmt_vnode/src/vmHandle.c | 1 + source/dnode/mnode/impl/src/mndStream.c | 361 +++++++++++++------- source/dnode/vnode/src/inc/vnodeInt.h | 1 + source/dnode/vnode/src/tq/tq.c | 31 ++ source/dnode/vnode/src/vnd/vnodeSvr.c | 5 + source/libs/stream/src/streamExec.c | 2 +- 10 files changed, 295 insertions(+), 130 deletions(-) diff --git a/include/common/tmsg.h b/include/common/tmsg.h index e17a72992c..73d1ab2473 100644 --- a/include/common/tmsg.h +++ b/include/common/tmsg.h @@ -3271,7 +3271,7 @@ typedef struct { SMsgHead head; int64_t streamId; int32_t taskId; -} SVPauseStreamTaskReq, SVResetStreamTaskReq; +} SVPauseStreamTaskReq, SVResetStreamTaskReq, SVDropHTaskReq; typedef struct { int8_t reserved; diff --git a/include/common/tmsgdef.h b/include/common/tmsgdef.h index 4eb8328caa..2d1148a209 100644 --- a/include/common/tmsgdef.h +++ b/include/common/tmsgdef.h @@ -271,6 +271,7 @@ enum { // WARN: new msg should be appended to segment tail TD_DEF_MSG_TYPE(TDMT_STREAM_TASK_PAUSE, "stream-task-pause", NULL, NULL) TD_DEF_MSG_TYPE(TDMT_STREAM_TASK_RESUME, "stream-task-resume", NULL, NULL) TD_DEF_MSG_TYPE(TDMT_STREAM_TASK_STOP, "stream-task-stop", NULL, NULL) + TD_DEF_MSG_TYPE(TDMT_STREAM_HTASK_DROP, "stream-htask-drop", NULL, NULL) TD_DEF_MSG_TYPE(TDMT_STREAM_MAX_MSG, "stream-max", NULL, NULL) TD_NEW_MSG_SEG(TDMT_MON_MSG) diff --git a/include/libs/stream/tstream.h b/include/libs/stream/tstream.h index ad2e9514d1..58439d142a 100644 --- a/include/libs/stream/tstream.h +++ b/include/libs/stream/tstream.h @@ -655,17 +655,19 @@ int32_t tDecodeStreamCheckpointReadyMsg(SDecoder* pDecoder, SStreamCheckpointRea typedef struct STaskStatusEntry { STaskId id; int32_t status; - int64_t stage; + int32_t statusLastDuration; // to record the last duration of current status + int32_t stage; int32_t nodeId; - int64_t verStart; // start version in WAL, only valid for source task - int64_t verEnd; // end version in WAL, only valid for source task - int64_t processedVer; // only valid for source task - int64_t activeCheckpointId; // current active checkpoint id - bool checkpointFailed; // denote if the checkpoint is failed or not - double inputQUsed; // in MiB + int64_t verStart; // start version in WAL, only valid for source task + int64_t verEnd; // end version in WAL, only valid for source task + int64_t processedVer; // only valid for source task + int32_t relatedHTask; // has related fill-history task + int64_t activeCheckpointId; // current active checkpoint id + bool checkpointFailed; // denote if the checkpoint is failed or not + double inputQUsed; // in MiB double inputRate; - double sinkQuota; // existed quota size for sink task - double sinkDataSize; // sink to dest data size + double sinkQuota; // existed quota size for sink task + double sinkDataSize; // sink to dst data size } STaskStatusEntry; typedef struct SStreamHbMsg { diff --git a/source/dnode/mgmt/mgmt_snode/src/smHandle.c b/source/dnode/mgmt/mgmt_snode/src/smHandle.c index cd81b9873f..6de29f8513 100644 --- a/source/dnode/mgmt/mgmt_snode/src/smHandle.c +++ b/source/dnode/mgmt/mgmt_snode/src/smHandle.c @@ -84,6 +84,7 @@ SArray *smGetMsgHandles() { if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_PAUSE, smPutNodeMsgToMgmtQueue, 1) == NULL) goto _OVER; if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_RESUME, smPutNodeMsgToMgmtQueue, 1) == NULL) goto _OVER; if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_STOP, smPutNodeMsgToMgmtQueue, 1) == NULL) goto _OVER; + if (dmSetMgmtHandle(pArray, TDMT_STREAM_HTASK_DROP, smPutNodeMsgToMgmtQueue, 1) == NULL) goto _OVER; if (dmSetMgmtHandle(pArray, TDMT_VND_STREAM_TASK_CHECK, smPutNodeMsgToStreamQueue, 1) == NULL) goto _OVER; if (dmSetMgmtHandle(pArray, TDMT_VND_STREAM_TASK_CHECK_RSP, smPutNodeMsgToStreamQueue, 1) == NULL) goto _OVER; if (dmSetMgmtHandle(pArray, TDMT_VND_STREAM_SCAN_HISTORY_FINISH, smPutNodeMsgToStreamQueue, 1) == NULL) goto _OVER; diff --git a/source/dnode/mgmt/mgmt_vnode/src/vmHandle.c b/source/dnode/mgmt/mgmt_vnode/src/vmHandle.c index 670eb48bd2..770a0ac6f7 100644 --- a/source/dnode/mgmt/mgmt_vnode/src/vmHandle.c +++ b/source/dnode/mgmt/mgmt_vnode/src/vmHandle.c @@ -833,6 +833,7 @@ SArray *vmGetMsgHandles() { if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_PAUSE, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER; if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_RESUME, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER; if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_STOP, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER; + if (dmSetMgmtHandle(pArray, TDMT_STREAM_HTASK_DROP, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER; if (dmSetMgmtHandle(pArray, TDMT_VND_STREAM_CHECK_POINT_SOURCE, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER; if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_CHECKPOINT_READY, vmPutMsgToStreamQueue, 0) == NULL) goto _OVER; if (dmSetMgmtHandle(pArray, TDMT_VND_STREAM_TASK_UPDATE, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER; diff --git a/source/dnode/mnode/impl/src/mndStream.c b/source/dnode/mnode/impl/src/mndStream.c index d2f0a13038..924e6475f1 100644 --- a/source/dnode/mnode/impl/src/mndStream.c +++ b/source/dnode/mnode/impl/src/mndStream.c @@ -78,6 +78,8 @@ static int32_t mndProcessNodeCheckReq(SRpcMsg *pMsg); static SArray *extractNodeListFromStream(SMnode *pMnode); static SArray *mndTakeVgroupSnapshot(SMnode *pMnode, bool *allReady); +static SStreamObj *mndGetStreamObj(SMnode *pMnode, int64_t streamId); + static SVgroupChangeInfo mndFindChangedNodeInfo(SMnode *pMnode, const SArray *pPrevNodeList, const SArray *pNodeList); static STrans *doCreateTrans(SMnode *pMnode, SStreamObj *pStream, SRpcMsg *pReq, const char *name, const char* pMsg); @@ -1570,6 +1572,123 @@ static void mndCancelGetNextStream(SMnode *pMnode, void *pIter) { sdbCancelFetch(pSdb, pIter); } +static void setTaskAttrInResBlock(SStreamObj* pStream, SStreamTask* pTask, SSDataBlock* pBlock, int32_t numOfRows) { + SColumnInfoData *pColInfo; + int32_t cols = 0; + + // stream name + char streamName[TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE] = {0}; + STR_WITH_MAXSIZE_TO_VARSTR(streamName, mndGetDbStr(pStream->name), sizeof(streamName)); + + pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); + colDataSetVal(pColInfo, numOfRows, (const char *)streamName, false); + + // task id + pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); + + char idstr[128] = {0}; + int32_t len = tintToHex(pTask->id.taskId, &idstr[4]); + idstr[2] = '0'; + idstr[3] = 'x'; + varDataSetLen(idstr, len + 2); + colDataSetVal(pColInfo, numOfRows, idstr, false); + + // node type + char nodeType[20 + VARSTR_HEADER_SIZE] = {0}; + varDataSetLen(nodeType, 5); + pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); + if (pTask->info.nodeId > 0) { + memcpy(varDataVal(nodeType), "vnode", 5); + } else { + memcpy(varDataVal(nodeType), "snode", 5); + } + colDataSetVal(pColInfo, numOfRows, nodeType, false); + + // node id + pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); + int64_t nodeId = TMAX(pTask->info.nodeId, 0); + colDataSetVal(pColInfo, numOfRows, (const char *)&nodeId, false); + + // level + char level[20 + VARSTR_HEADER_SIZE] = {0}; + if (pTask->info.taskLevel == TASK_LEVEL__SOURCE) { + memcpy(varDataVal(level), "source", 6); + varDataSetLen(level, 6); + } else if (pTask->info.taskLevel == TASK_LEVEL__AGG) { + memcpy(varDataVal(level), "agg", 3); + varDataSetLen(level, 3); + } else if (pTask->info.taskLevel == TASK_LEVEL__SINK) { + memcpy(varDataVal(level), "sink", 4); + varDataSetLen(level, 4); + } + + pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); + colDataSetVal(pColInfo, numOfRows, (const char *)level, false); + + // status + char status[20 + VARSTR_HEADER_SIZE] = {0}; + STaskId id = {.streamId = pTask->id.streamId, .taskId = pTask->id.taskId}; + + STaskStatusEntry *pe = taosHashGet(execInfo.pTaskMap, &id, sizeof(id)); + if (pe == NULL) { + return; + } + + const char *pStatus = streamTaskGetStatusStr(pe->status); + STR_TO_VARSTR(status, pStatus); + + // status + pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); + colDataSetVal(pColInfo, numOfRows, (const char *)status, false); + + // stage + pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); + colDataSetVal(pColInfo, numOfRows, (const char *)&pe->stage, false); + + // input queue + char vbuf[30] = {0}; + char buf[25] = {0}; + const char *queueInfoStr = "%4.2fMiB (%5.2f%)"; + sprintf(buf, queueInfoStr, pe->inputQUsed, pe->inputRate); + STR_TO_VARSTR(vbuf, buf); + + pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); + colDataSetVal(pColInfo, numOfRows, (const char *)vbuf, false); + + // output queue + // sprintf(buf, queueInfoStr, pe->outputQUsed, pe->outputRate); + // STR_TO_VARSTR(vbuf, buf); + + // pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); + // colDataSetVal(pColInfo, numOfRows, (const char*)vbuf, false); + + if (pTask->info.taskLevel == TASK_LEVEL__SINK) { + const char *sinkStr = "%.2fMiB"; + sprintf(buf, sinkStr, pe->sinkDataSize); + } else if (pTask->info.taskLevel == TASK_LEVEL__SOURCE) { + // offset info + const char *offsetStr = "%" PRId64 " [%" PRId64 ", %" PRId64 "]"; + sprintf(buf, offsetStr, pe->processedVer, pe->verStart, pe->verEnd); + } + + STR_TO_VARSTR(vbuf, buf); + + pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); + colDataSetVal(pColInfo, numOfRows, (const char *)vbuf, false); +} + +static int32_t getNumOfTasks(SArray* pTaskList) { + int32_t numOfLevels = taosArrayGetSize(pTaskList); + + int32_t count = 0; + for (int32_t i = 0; i < numOfLevels; i++) { + SArray *pLevel = taosArrayGetP(pTaskList, i); + count += taosArrayGetSize(pLevel); + } + + return count; +} + static int32_t mndRetrieveStreamTask(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rowsCapacity) { SMnode * pMnode = pReq->info.node; SSdb * pSdb = pMnode->pSdb; @@ -1585,137 +1704,25 @@ static int32_t mndRetrieveStreamTask(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock // lock taosRLockLatch(&pStream->lock); - // count task num - int32_t sz = taosArrayGetSize(pStream->tasks); - - int32_t count = 0; - for (int32_t i = 0; i < sz; i++) { - SArray *pLevel = taosArrayGetP(pStream->tasks, i); - count += taosArrayGetSize(pLevel); - } - + int32_t count = getNumOfTasks(pStream->tasks); if (numOfRows + count > rowsCapacity) { blockDataEnsureCapacity(pBlock, numOfRows + count); } // add row for each task - for (int32_t i = 0; i < sz; i++) { + for (int32_t i = 0; i < taosArrayGetSize(pStream->tasks); i++) { SArray *pLevel = taosArrayGetP(pStream->tasks, i); - int32_t levelCnt = taosArrayGetSize(pLevel); - for (int32_t j = 0; j < levelCnt; j++) { + int32_t numOfLevels = taosArrayGetSize(pLevel); + for (int32_t j = 0; j < numOfLevels; j++) { SStreamTask *pTask = taosArrayGetP(pLevel, j); - - SColumnInfoData *pColInfo; - int32_t cols = 0; - - // stream name - char streamName[TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE] = {0}; - STR_WITH_MAXSIZE_TO_VARSTR(streamName, mndGetDbStr(pStream->name), sizeof(streamName)); - - pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); - colDataSetVal(pColInfo, numOfRows, (const char *)streamName, false); - - // task id - pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); - - char idstr[128] = {0}; - int32_t len = tintToHex(pTask->id.taskId, &idstr[4]); - idstr[2] = '0'; - idstr[3] = 'x'; - varDataSetLen(idstr, len + 2); - colDataSetVal(pColInfo, numOfRows, idstr, false); - - // node type - char nodeType[20 + VARSTR_HEADER_SIZE] = {0}; - varDataSetLen(nodeType, 5); - pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); - if (pTask->info.nodeId > 0) { - memcpy(varDataVal(nodeType), "vnode", 5); - } else { - memcpy(varDataVal(nodeType), "snode", 5); - } - colDataSetVal(pColInfo, numOfRows, nodeType, false); - - // node id - pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); - int64_t nodeId = TMAX(pTask->info.nodeId, 0); - colDataSetVal(pColInfo, numOfRows, (const char *)&nodeId, false); - - // level - char level[20 + VARSTR_HEADER_SIZE] = {0}; - if (pTask->info.taskLevel == TASK_LEVEL__SOURCE) { - memcpy(varDataVal(level), "source", 6); - varDataSetLen(level, 6); - } else if (pTask->info.taskLevel == TASK_LEVEL__AGG) { - memcpy(varDataVal(level), "agg", 3); - varDataSetLen(level, 3); - } else if (pTask->info.taskLevel == TASK_LEVEL__SINK) { - memcpy(varDataVal(level), "sink", 4); - varDataSetLen(level, 4); - } - - pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); - colDataSetVal(pColInfo, numOfRows, (const char *)level, false); - - // status - char status[20 + VARSTR_HEADER_SIZE] = {0}; - - STaskId id = {.streamId = pTask->id.streamId, .taskId = pTask->id.taskId}; - STaskStatusEntry *pe = taosHashGet(execInfo.pTaskMap, &id, sizeof(id)); - if (pe == NULL) { - continue; - } - - const char *pStatus = streamTaskGetStatusStr(pe->status); - STR_TO_VARSTR(status, pStatus); - - // status - pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); - colDataSetVal(pColInfo, numOfRows, (const char *)status, false); - - // stage - pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); - colDataSetVal(pColInfo, numOfRows, (const char *)&pe->stage, false); - - // input queue - char vbuf[30] = {0}; - char buf[25] = {0}; - const char *queueInfoStr = "%4.2fMiB (%5.2f%)"; - sprintf(buf, queueInfoStr, pe->inputQUsed, pe->inputRate); - STR_TO_VARSTR(vbuf, buf); - - pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); - colDataSetVal(pColInfo, numOfRows, (const char *)vbuf, false); - - // output queue - // sprintf(buf, queueInfoStr, pe->outputQUsed, pe->outputRate); - // STR_TO_VARSTR(vbuf, buf); - - // pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); - // colDataSetVal(pColInfo, numOfRows, (const char*)vbuf, false); - - if (pTask->info.taskLevel == TASK_LEVEL__SINK) { - const char *sinkStr = "%.2fMiB"; - sprintf(buf, sinkStr, pe->sinkDataSize); - } else if (pTask->info.taskLevel == TASK_LEVEL__SOURCE) { - // offset info - const char *offsetStr = "%" PRId64 " [%" PRId64 ", %" PRId64 "]"; - sprintf(buf, offsetStr, pe->processedVer, pe->verStart, pe->verEnd); - } - - STR_TO_VARSTR(vbuf, buf); - - pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); - colDataSetVal(pColInfo, numOfRows, (const char *)vbuf, false); - + setTaskAttrInResBlock(pStream, pTask, pBlock, numOfRows); numOfRows++; } } // unlock taosRUnLockLatch(&pStream->lock); - sdbRelease(pSdb, pStream); } @@ -2729,7 +2736,7 @@ int32_t killActiveCheckpointTrans(SMnode *pMnode, const char* pDBName, size_t le return TSDB_CODE_SUCCESS; } -int32_t mndResetStatusFromCheckpoint(SMnode *pMnode, int32_t transId) { +static int32_t mndResetStatusFromCheckpoint(SMnode *pMnode) { STrans* pTrans = mndAcquireTrans(pMnode, transId); if (pTrans != NULL) { mInfo("kill checkpoint transId:%d to reset task status", transId); @@ -2765,6 +2772,88 @@ int32_t mndResetStatusFromCheckpoint(SMnode *pMnode, int32_t transId) { return 0; } +static SStreamTask* mndGetStreamTask(STaskId* pId, SStreamObj* pStream) { + for (int32_t i = 0; i < taosArrayGetSize(pStream->tasks); i++) { + SArray *pLevel = taosArrayGetP(pStream->tasks, i); + + int32_t numOfLevels = taosArrayGetSize(pLevel); + for (int32_t j = 0; j < numOfLevels; j++) { + SStreamTask *pTask = taosArrayGetP(pLevel, j); + if (pTask->id.taskId == pId->taskId) { + return pTask; + } + } + } + + return NULL; +} + +static bool needDropRelatedFillhistoryTask(STaskStatusEntry *pTaskEntry, SStreamExecInfo *pExecNode) { + if (pTaskEntry->status == TASK_STATUS__STREAM_SCAN_HISTORY && pTaskEntry->statusLastDuration >= 10) { + if (fabs(pTaskEntry->inputQUsed) <= DBL_EPSILON) { + int32_t numOfReady = 0; + int32_t numOfTotal = 0; + for (int32_t k = 0; k < taosArrayGetSize(pExecNode->pTaskList); ++k) { + STaskId *pId = taosArrayGet(pExecNode->pTaskList, k); + if (pTaskEntry->id.streamId == pId->streamId) { + numOfTotal++; + + if (pTaskEntry->id.taskId != pId->taskId && pTaskEntry->status == TASK_STATUS__READY) { + numOfReady++; + } + } + } + + if (numOfReady > 0) { + mDebug("stream:0x%" PRIx64 + " %d tasks are ready, %d tasks in stream-scan-history for more than 50s, drop related fill-history task", + pTaskEntry->id.streamId, numOfReady, numOfTotal - numOfReady); + return true; + } else { + return false; + } + } + } + + return false; +} + +// currently only handle the sink task +// 1. sink task, drop related fill-history task msg is missing +// 2. other tasks are in ready state for at least 3 * hb_interval +static int32_t mndDropRelatedFillhistoryTask(SMnode *pMnode, STaskStatusEntry *pTaskEntry, SStreamObj *pStream) { + SStreamTask *pTask = mndGetStreamTask(&pTaskEntry->id, pStream); + if (pTask == NULL) { + mError("failed to get the stream task:0x%x, may have been dropped", (int32_t) pTaskEntry->id.taskId); + return -1; + } + + SVDropHTaskReq *pReq = rpcMallocCont(sizeof(SVDropHTaskReq)); + if (pReq == NULL) { + mError("failed to malloc in drop related fill-history task, size:%" PRIzu ", code:%s", sizeof(SVDropHTaskReq), + tstrerror(TSDB_CODE_OUT_OF_MEMORY)); + terrno = TSDB_CODE_OUT_OF_MEMORY; + return -1; + } + + pReq->head.vgId = htonl(pTask->info.nodeId); + pReq->taskId = pTask->id.taskId; + pReq->streamId = pTask->id.streamId; + + SRpcMsg msg = {.info.noResp = 1}; + + initRpcMsg(&msg, TDMT_STREAM_HTASK_DROP, pReq, sizeof(SVDropHTaskReq)); + + mDebug("build and send drop related fill-history task for task:0x%x", pTask->id.taskId); + + SVgObj *pVgObj = mndAcquireVgroup(pMnode, pTask->info.nodeId); + SEpSet epset = mndGetVgroupEpset(pMnode, pVgObj); + mndReleaseVgroup(pMnode, pVgObj); + + tmsgSendReq(&epset, &msg); + return TSDB_CODE_SUCCESS; +} + int32_t setNodeEpsetExpiredFlag(const SArray *pNodeList) { int32_t num = taosArrayGetSize(pNodeList); mInfo("set node expired for %d nodes", num); @@ -2865,9 +2954,28 @@ int32_t mndProcessStreamHb(SRpcMsg *pReq) { } } - pTaskEntry->status = p->status; + if (p->status == pTaskEntry->status) { + pTaskEntry->statusLastDuration++; + } else { + pTaskEntry->status = p->status; + pTaskEntry->statusLastDuration = 0; + } + if (p->status != TASK_STATUS__READY) { mDebug("received s-task:0x%" PRIx64 " not in ready status:%s", p->id.taskId, streamTaskGetStatusStr(p->status)); + + if (p->status == TASK_STATUS__STREAM_SCAN_HISTORY) { + bool drop = needDropRelatedFillhistoryTask(pTaskEntry, &execInfo); + if(drop) { + SStreamObj *pStreamObj = mndGetStreamObj(pMnode, pTaskEntry->id.streamId); + if (pStreamObj == NULL) { + mError("failed to acquire the streamObj:0x%" PRIx64 " it may have been dropped", pStreamObj->uid); + } else { + mndDropRelatedFillhistoryTask(pMnode, pTaskEntry, pStreamObj); + mndReleaseStream(pMnode, pStreamObj); + } + } + } } } @@ -2894,3 +3002,18 @@ int32_t mndProcessStreamHb(SRpcMsg *pReq) { taosArrayDestroy(req.pUpdateNodes); return TSDB_CODE_SUCCESS; } + +SStreamObj *mndGetStreamObj(SMnode *pMnode, int64_t streamId) { + void *pIter = NULL; + SSdb *pSdb = pMnode->pSdb; + SStreamObj *pStream = NULL; + + while ((pIter = sdbFetch(pSdb, SDB_STREAM, pIter, (void **)&pStream)) != NULL) { + if (pStream->uid == streamId) { + sdbCancelFetch(pSdb, pIter); + return pStream; + } + } + + return NULL; +} \ No newline at end of file diff --git a/source/dnode/vnode/src/inc/vnodeInt.h b/source/dnode/vnode/src/inc/vnodeInt.h index df1720d4a7..16379db053 100644 --- a/source/dnode/vnode/src/inc/vnodeInt.h +++ b/source/dnode/vnode/src/inc/vnodeInt.h @@ -232,6 +232,7 @@ int32_t tqProcessTaskCheckPointSourceReq(STQ* pTq, SRpcMsg* pMsg, SRpcMsg* pRsp) int32_t tqProcessTaskCheckpointReadyMsg(STQ* pTq, SRpcMsg* pMsg); int32_t tqProcessTaskUpdateReq(STQ* pTq, SRpcMsg* pMsg); int32_t tqProcessTaskResetReq(STQ* pTq, SRpcMsg* pMsg); +int32_t tqProcessTaskDropHTask(STQ* pTq, SRpcMsg* pMsg); int32_t tqStartStreamTaskAsync(STQ* pTq, bool restart); int32_t tqRestartStreamTasks(STQ* pTq); diff --git a/source/dnode/vnode/src/tq/tq.c b/source/dnode/vnode/src/tq/tq.c index 7be2a357d3..9d16402ee6 100644 --- a/source/dnode/vnode/src/tq/tq.c +++ b/source/dnode/vnode/src/tq/tq.c @@ -2014,3 +2014,34 @@ int32_t tqProcessTaskResetReq(STQ* pTq, SRpcMsg* pMsg) { streamMetaReleaseTask(pMeta, pTask); return TSDB_CODE_SUCCESS; } + +int32_t tqProcessTaskDropHTask(STQ* pTq, SRpcMsg* pMsg) { + SVDropHTaskReq* pReq = (SVDropHTaskReq*) pMsg->pCont; + + SStreamMeta* pMeta = pTq->pStreamMeta; + SStreamTask* pTask = streamMetaAcquireTask(pMeta, pReq->streamId, pReq->taskId); + if (pTask == NULL) { + tqError("vgId:%d process drop fill-history task req, failed to acquire task:0x%x, it may have been dropped already", + pMeta->vgId, pReq->taskId); + return TSDB_CODE_SUCCESS; + } + + tqDebug("s-task:%s receive drop fill-history msg from mnode", pTask->id.idStr); + if (pTask->hTaskInfo.id.taskId == 0) { + tqError("vgId:%d s-task:%s not have related fill-history task", pMeta->vgId, pTask->id.idStr); + streamMetaReleaseTask(pMeta, pTask); + return TSDB_CODE_SUCCESS; + } + + ETaskStatus status = streamTaskGetStatus(pTask, NULL); + ASSERT(status == TASK_STATUS__STREAM_SCAN_HISTORY); + + streamTaskHandleEvent(pTask->status.pSM, TASK_EVENT_SCANHIST_DONE); + + SStreamTaskId id = {.streamId = pTask->hTaskInfo.id.streamId, .taskId = pTask->hTaskInfo.id.taskId}; + streamBuildAndSendDropTaskMsg(pTask->pMsgCb, pMeta->vgId, &id); + + streamMetaReleaseTask(pMeta, pTask); + return TSDB_CODE_SUCCESS; +} + diff --git a/source/dnode/vnode/src/vnd/vnodeSvr.c b/source/dnode/vnode/src/vnd/vnodeSvr.c index 33b4114009..8cbca403e3 100644 --- a/source/dnode/vnode/src/vnd/vnodeSvr.c +++ b/source/dnode/vnode/src/vnd/vnodeSvr.c @@ -595,6 +595,11 @@ int32_t vnodeProcessWriteMsg(SVnode *pVnode, SRpcMsg *pMsg, int64_t ver, SRpcMsg tqProcessTaskResetReq(pVnode->pTq, pMsg); } } break; + case TDMT_STREAM_HTASK_DROP: { + if (pVnode->restored && vnodeIsLeader(pVnode)) { + tqProcessTaskDropHTask(pVnode->pTq, pMsg); + } + } break; case TDMT_VND_ALTER_CONFIRM: needCommit = pVnode->config.hashChange; if (vnodeProcessAlterConfirmReq(pVnode, ver, pReq, len, pRsp) < 0) { diff --git a/source/libs/stream/src/streamExec.c b/source/libs/stream/src/streamExec.c index 43875319b7..8b14846414 100644 --- a/source/libs/stream/src/streamExec.c +++ b/source/libs/stream/src/streamExec.c @@ -401,7 +401,7 @@ int32_t streamDoTransferStateToStreamTask(SStreamTask* pTask) { // 5. save to disk pStreamTask->status.taskStatus = streamTaskGetStatus(pStreamTask, NULL); - // 6. pause allowed. + // 6. add empty delete block if ((pStreamTask->info.taskLevel == TASK_LEVEL__SOURCE) && taosQueueEmpty(pStreamTask->inputq.queue->pQueue)) { SStreamRefDataBlock* pItem = taosAllocateQitem(sizeof(SStreamRefDataBlock), DEF_QITEM, 0); From de4cc2009dc142dce6048272190c7a0aad2d3062 Mon Sep 17 00:00:00 2001 From: Haojun Liao Date: Thu, 23 Nov 2023 16:04:06 +0800 Subject: [PATCH 6/8] fix(stream): fix error. --- include/libs/stream/tstream.h | 2 +- source/dnode/mnode/impl/src/mndStream.c | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/include/libs/stream/tstream.h b/include/libs/stream/tstream.h index 58439d142a..cf9fc1d826 100644 --- a/include/libs/stream/tstream.h +++ b/include/libs/stream/tstream.h @@ -656,7 +656,7 @@ typedef struct STaskStatusEntry { STaskId id; int32_t status; int32_t statusLastDuration; // to record the last duration of current status - int32_t stage; + int64_t stage; int32_t nodeId; int64_t verStart; // start version in WAL, only valid for source task int64_t verEnd; // end version in WAL, only valid for source task diff --git a/source/dnode/mnode/impl/src/mndStream.c b/source/dnode/mnode/impl/src/mndStream.c index 924e6475f1..aa07172d2a 100644 --- a/source/dnode/mnode/impl/src/mndStream.c +++ b/source/dnode/mnode/impl/src/mndStream.c @@ -2736,7 +2736,7 @@ int32_t killActiveCheckpointTrans(SMnode *pMnode, const char* pDBName, size_t le return TSDB_CODE_SUCCESS; } -static int32_t mndResetStatusFromCheckpoint(SMnode *pMnode) { +static int32_t mndResetStatusFromCheckpoint(SMnode *pMnode, int64_t transId) { STrans* pTrans = mndAcquireTrans(pMnode, transId); if (pTrans != NULL) { mInfo("kill checkpoint transId:%d to reset task status", transId); From 810eb8e0e388b12db67b02cef175d6fdd35add6b Mon Sep 17 00:00:00 2001 From: Haojun Liao Date: Thu, 23 Nov 2023 16:25:12 +0800 Subject: [PATCH 7/8] fix(stream): fix syntax error. --- source/dnode/mnode/impl/src/mndStream.c | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/source/dnode/mnode/impl/src/mndStream.c b/source/dnode/mnode/impl/src/mndStream.c index aa07172d2a..230e54f5ce 100644 --- a/source/dnode/mnode/impl/src/mndStream.c +++ b/source/dnode/mnode/impl/src/mndStream.c @@ -2736,7 +2736,7 @@ int32_t killActiveCheckpointTrans(SMnode *pMnode, const char* pDBName, size_t le return TSDB_CODE_SUCCESS; } -static int32_t mndResetStatusFromCheckpoint(SMnode *pMnode, int64_t transId) { +static int32_t mndResetStatusFromCheckpoint(SMnode *pMnode, int32_t transId) { STrans* pTrans = mndAcquireTrans(pMnode, transId); if (pTrans != NULL) { mInfo("kill checkpoint transId:%d to reset task status", transId); From 2e922fefc6ef5f5b3287bf7441861546f4cb2c22 Mon Sep 17 00:00:00 2001 From: Haojun Liao Date: Thu, 23 Nov 2023 18:04:09 +0800 Subject: [PATCH 8/8] fix(stream): reduce the sink throughput. and fix the bug in status check. --- source/dnode/mnode/impl/src/mndStream.c | 7 +++++-- source/libs/stream/src/streamTask.c | 2 +- 2 files changed, 6 insertions(+), 3 deletions(-) diff --git a/source/dnode/mnode/impl/src/mndStream.c b/source/dnode/mnode/impl/src/mndStream.c index 230e54f5ce..db013017e3 100644 --- a/source/dnode/mnode/impl/src/mndStream.c +++ b/source/dnode/mnode/impl/src/mndStream.c @@ -2798,8 +2798,11 @@ static bool needDropRelatedFillhistoryTask(STaskStatusEntry *pTaskEntry, SStream if (pTaskEntry->id.streamId == pId->streamId) { numOfTotal++; - if (pTaskEntry->id.taskId != pId->taskId && pTaskEntry->status == TASK_STATUS__READY) { - numOfReady++; + if (pTaskEntry->id.taskId != pId->taskId) { + STaskStatusEntry *pEntry = taosHashGet(execInfo.pTaskMap, pId, sizeof(*pId)); + if (pEntry->status == TASK_STATUS__READY) { + numOfReady++; + } } } } diff --git a/source/libs/stream/src/streamTask.c b/source/libs/stream/src/streamTask.c index c93d658adb..beaab0a415 100644 --- a/source/libs/stream/src/streamTask.c +++ b/source/libs/stream/src/streamTask.c @@ -451,7 +451,7 @@ int32_t streamTaskInit(SStreamTask* pTask, SStreamMeta* pMeta, SMsgCb* pMsgCb, i // 2MiB per second for sink task // 50 times sink operator per second - streamTaskInitTokenBucket(pTask->outputInfo.pTokenBucket, 50, 50, tsSinkDataRate, pTask->id.idStr); + streamTaskInitTokenBucket(pTask->outputInfo.pTokenBucket, 35, 35, tsSinkDataRate, pTask->id.idStr); TdThreadMutexAttr attr = {0}; int code = taosThreadMutexAttrInit(&attr);