diff --git a/include/common/tglobal.h b/include/common/tglobal.h index 4644c38ec4..df46cc689b 100644 --- a/include/common/tglobal.h +++ b/include/common/tglobal.h @@ -193,7 +193,8 @@ extern int64_t tsWalFsyncDataSizeLimit; // internal extern int32_t tsTransPullupInterval; extern int32_t tsMqRebalanceInterval; -extern int32_t tsStreamCheckpointTickInterval; +extern int32_t tsStreamCheckpointInterval; +extern float tsSinkDataRate; extern int32_t tsStreamNodeCheckInterval; extern int32_t tsTtlUnit; extern int32_t tsTtlPushIntervalSec; @@ -202,9 +203,6 @@ extern int32_t tsTrimVDbIntervalSec; extern int32_t tsGrantHBInterval; extern int32_t tsUptimeInterval; -extern int32_t tsRpcRetryLimit; -extern int32_t tsRpcRetryInterval; - extern bool tsDisableStream; extern int64_t tsStreamBufferSize; extern bool tsFilterScalarMode; diff --git a/include/libs/stream/tstream.h b/include/libs/stream/tstream.h index 629efa00b3..6cbe6ad7da 100644 --- a/include/libs/stream/tstream.h +++ b/include/libs/stream/tstream.h @@ -714,7 +714,6 @@ int32_t streamTaskScanHistoryDataComplete(SStreamTask* pTask); int32_t streamStartScanHistoryAsync(SStreamTask* pTask, int8_t igUntreated); bool streamHistoryTaskSetVerRangeStep2(SStreamTask* pTask, int64_t latestVer); int32_t streamQueueGetNumOfItems(const SStreamQueue* pQueue); -int32_t streamQueueGetAvailableSpace(const SStreamQueue* pQueue, int32_t* availNum, double* availSize); // common int32_t streamRestoreParam(SStreamTask* pTask); diff --git a/include/util/tconfig.h b/include/util/tconfig.h index aaad467737..34798f3816 100644 --- a/include/util/tconfig.h +++ b/include/util/tconfig.h @@ -65,7 +65,6 @@ typedef struct SConfigItem { union { bool bval; float fval; - double dval; int32_t i32; int64_t i64; char *str; @@ -104,7 +103,6 @@ int32_t cfgAddBool(SConfig *pCfg, const char *name, bool defaultVal, int8_t scop int32_t cfgAddInt32(SConfig *pCfg, const char *name, int32_t defaultVal, int64_t minval, int64_t maxval, int8_t scope); int32_t cfgAddInt64(SConfig *pCfg, const char *name, int64_t defaultVal, int64_t minval, int64_t maxval, int8_t scope); int32_t cfgAddFloat(SConfig *pCfg, const char *name, float defaultVal, float minval, float maxval, int8_t scope); -int32_t cfgAddDouble(SConfig *pCfg, const char *name, double defaultVal, double minval, double maxval, int8_t scope); int32_t cfgAddString(SConfig *pCfg, const char *name, const char *defaultVal, int8_t scope); int32_t cfgAddDir(SConfig *pCfg, const char *name, const char *defaultVal, int8_t scope); int32_t cfgAddLocale(SConfig *pCfg, const char *name, const char *defaultVal, int8_t scope); diff --git a/source/common/src/tglobal.c b/source/common/src/tglobal.c index 3d7b38161a..568808405e 100644 --- a/source/common/src/tglobal.c +++ b/source/common/src/tglobal.c @@ -244,7 +244,8 @@ int32_t tsTtlBatchDropNum = 10000; // number of tables dropped per batch // internal int32_t tsTransPullupInterval = 2; int32_t tsMqRebalanceInterval = 2; -int32_t tsStreamCheckpointTickInterval = 300; +int32_t tsStreamCheckpointInterval = 300; +float tsSinkDataRate = 2.0; int32_t tsStreamNodeCheckInterval = 30; int32_t tsTtlUnit = 86400; int32_t tsTtlPushIntervalSec = 10; @@ -269,8 +270,6 @@ int8_t tsS3Enabled = false; int32_t tsS3BlockSize = 4096; // number of tsdb pages int32_t tsS3BlockCacheSize = 16; // number of blocks -int32_t tsCheckpointInterval = 300; - #ifndef _STORAGE int32_t taosSetTfsCfg(SConfig *pCfg) { SConfigItem *pItem = cfgGetItem(pCfg, "dataDir"); @@ -490,11 +489,11 @@ static int32_t taosAddSystemCfg(SConfig *pCfg) { if (cfgAddBool(pCfg, "enableCoreFile", 1, CFG_SCOPE_BOTH) != 0) return -1; if (cfgAddFloat(pCfg, "numOfCores", tsNumOfCores, 1, 100000, CFG_SCOPE_BOTH) != 0) return -1; - if (cfgAddBool(pCfg, "SSE42", tsSSE42Enable, CFG_SCOPE_BOTH) != 0) return -1; - if (cfgAddBool(pCfg, "AVX", tsAVXEnable, CFG_SCOPE_BOTH) != 0) return -1; - if (cfgAddBool(pCfg, "AVX2", tsAVX2Enable, CFG_SCOPE_BOTH) != 0) return -1; - if (cfgAddBool(pCfg, "FMA", tsFMAEnable, CFG_SCOPE_BOTH) != 0) return -1; - if (cfgAddBool(pCfg, "SIMD-builtins", tsSIMDBuiltins, CFG_SCOPE_BOTH) != 0) return -1; + if (cfgAddBool(pCfg, "ssd42", tsSSE42Enable, CFG_SCOPE_BOTH) != 0) return -1; + if (cfgAddBool(pCfg, "avx", tsAVXEnable, CFG_SCOPE_BOTH) != 0) return -1; + if (cfgAddBool(pCfg, "avx2", tsAVX2Enable, CFG_SCOPE_BOTH) != 0) return -1; + if (cfgAddBool(pCfg, "fma", tsFMAEnable, CFG_SCOPE_BOTH) != 0) return -1; + if (cfgAddBool(pCfg, "simdEnable", tsSIMDBuiltins, CFG_SCOPE_BOTH) != 0) return -1; if (cfgAddBool(pCfg, "tagFilterCache", tsTagFilterCache, CFG_SCOPE_BOTH) != 0) return -1; if (cfgAddInt64(pCfg, "openMax", tsOpenMax, 0, INT64_MAX, CFG_SCOPE_BOTH) != 0) return -1; @@ -650,18 +649,18 @@ static int32_t taosAddServerCfg(SConfig *pCfg) { if (cfgAddBool(pCfg, "disableStream", tsDisableStream, CFG_SCOPE_SERVER) != 0) return -1; if (cfgAddInt64(pCfg, "streamBufferSize", tsStreamBufferSize, 0, INT64_MAX, CFG_SCOPE_SERVER) != 0) return -1; - if (cfgAddInt64(pCfg, "checkpointInterval", tsStreamCheckpointTickInterval, 60, 1200, CFG_SCOPE_SERVER) != 0) return -1; + if (cfgAddInt64(pCfg, "checkpointInterval", tsStreamCheckpointInterval, 60, 1200, CFG_SCOPE_SERVER) != 0) return -1; + if (cfgAddFloat(pCfg, "streamSinkDataRate", tsSinkDataRate, 0.1, 5, CFG_SCOPE_SERVER) != 0) return -1; - if (cfgAddInt32(pCfg, "cacheLazyLoadThreshold", tsCacheLazyLoadThreshold, 0, 100000, CFG_SCOPE_SERVER) != 0) - return -1; + if (cfgAddInt32(pCfg, "cacheLazyLoadThreshold", tsCacheLazyLoadThreshold, 0, 100000, CFG_SCOPE_SERVER) != 0) return -1; - if (cfgAddString(pCfg, "LossyColumns", tsLossyColumns, CFG_SCOPE_SERVER) != 0) return -1; - if (cfgAddFloat(pCfg, "FPrecision", tsFPrecision, 0.0f, 100000.0f, CFG_SCOPE_SERVER) != 0) return -1; - if (cfgAddDouble(pCfg, "DPrecision", tsDPrecision, 0.0f, 1000000.0f, CFG_SCOPE_SERVER) != 0) return -1; - if (cfgAddInt32(pCfg, "MaxRange", tsMaxRange, 0, 65536, CFG_SCOPE_SERVER) != 0) return -1; - if (cfgAddInt32(pCfg, "CurRange", tsCurRange, 0, 65536, CFG_SCOPE_SERVER) != 0) return -1; - if (cfgAddBool(pCfg, "IfAdtFse", tsIfAdtFse, CFG_SCOPE_SERVER) != 0) return -1; - if (cfgAddString(pCfg, "Compressor", tsCompressor, CFG_SCOPE_SERVER) != 0) return -1; + if (cfgAddString(pCfg, "lossyColumns", tsLossyColumns, CFG_SCOPE_SERVER) != 0) return -1; + if (cfgAddFloat(pCfg, "fPrecision", tsFPrecision, 0.0f, 100000.0f, CFG_SCOPE_SERVER) != 0) return -1; + if (cfgAddFloat(pCfg, "dPrecision", tsDPrecision, 0.0f, 1000000.0f, CFG_SCOPE_SERVER) != 0) return -1; + if (cfgAddInt32(pCfg, "maxRange", tsMaxRange, 0, 65536, CFG_SCOPE_SERVER) != 0) return -1; + if (cfgAddInt32(pCfg, "curRange", tsCurRange, 0, 65536, CFG_SCOPE_SERVER) != 0) return -1; + if (cfgAddBool(pCfg, "ifAdtFse", tsIfAdtFse, CFG_SCOPE_SERVER) != 0) return -1; + if (cfgAddString(pCfg, "compressor", tsCompressor, CFG_SCOPE_SERVER) != 0) return -1; if (cfgAddBool(pCfg, "filterScalarMode", tsFilterScalarMode, CFG_SCOPE_SERVER) != 0) return -1; if (cfgAddInt32(pCfg, "maxStreamBackendCache", tsMaxStreamBackendCache, 16, 1024, CFG_SCOPE_SERVER) != 0) return -1; @@ -1022,7 +1021,7 @@ static int32_t taosSetServerCfg(SConfig *pCfg) { tsNumOfSnodeWriteThreads = cfgGetItem(pCfg, "numOfSnodeUniqueThreads")->i32; tsRpcQueueMemoryAllowed = cfgGetItem(pCfg, "rpcQueueMemoryAllowed")->i64; - tsSIMDBuiltins = (bool)cfgGetItem(pCfg, "SIMD-builtins")->bval; + tsSIMDBuiltins = (bool)cfgGetItem(pCfg, "simdEnable")->bval; tsTagFilterCache = (bool)cfgGetItem(pCfg, "tagFilterCache")->bval; tsEnableMonitor = cfgGetItem(pCfg, "monitor")->bval; @@ -1080,17 +1079,18 @@ static int32_t taosSetServerCfg(SConfig *pCfg) { tsCacheLazyLoadThreshold = cfgGetItem(pCfg, "cacheLazyLoadThreshold")->i32; - tstrncpy(tsLossyColumns, cfgGetItem(pCfg, "LossyColumns")->str, sizeof(tsLossyColumns)); - tsFPrecision = cfgGetItem(pCfg, "FPrecision")->fval; - tsDPrecision = cfgGetItem(pCfg, "DPrecision")->dval; - tsMaxRange = cfgGetItem(pCfg, "MaxRange")->i32; - tsCurRange = cfgGetItem(pCfg, "CurRange")->i32; - tsIfAdtFse = cfgGetItem(pCfg, "IfAdtFse")->bval; - tstrncpy(tsCompressor, cfgGetItem(pCfg, "Compressor")->str, sizeof(tsCompressor)); - + tstrncpy(tsLossyColumns, cfgGetItem(pCfg, "lossyColumns")->str, sizeof(tsLossyColumns)); + tsFPrecision = cfgGetItem(pCfg, "fPrecision")->fval; + tsDPrecision = cfgGetItem(pCfg, "dPrecision")->fval; + tsMaxRange = cfgGetItem(pCfg, "maxRange")->i32; + tsCurRange = cfgGetItem(pCfg, "curRange")->i32; + tsIfAdtFse = cfgGetItem(pCfg, "ifAdtFse")->bval; + tstrncpy(tsCompressor, cfgGetItem(pCfg, "compressor")->str, sizeof(tsCompressor)); tsDisableStream = cfgGetItem(pCfg, "disableStream")->bval; tsStreamBufferSize = cfgGetItem(pCfg, "streamBufferSize")->i64; + tsStreamCheckpointInterval = cfgGetItem(pCfg, "checkpointInterval")->i32; + tsSinkDataRate = cfgGetItem(pCfg, "streamSinkDataRate")->fval; tsFilterScalarMode = cfgGetItem(pCfg, "filterScalarMode")->bval; tsMaxStreamBackendCache = cfgGetItem(pCfg, "maxStreamBackendCache")->i32; diff --git a/source/dnode/mnode/impl/src/mndMain.c b/source/dnode/mnode/impl/src/mndMain.c index 1c87cde78a..08ad6cc909 100644 --- a/source/dnode/mnode/impl/src/mndMain.c +++ b/source/dnode/mnode/impl/src/mndMain.c @@ -281,7 +281,7 @@ static void *mndThreadFp(void *param) { mndCalMqRebalance(pMnode); } - if (sec % tsStreamCheckpointTickInterval == 0) { + if (sec % tsStreamCheckpointInterval == 0) { mndStreamCheckpointTick(pMnode, sec); } diff --git a/source/dnode/mnode/impl/src/mndStream.c b/source/dnode/mnode/impl/src/mndStream.c index dff0b004ce..f23d596449 100644 --- a/source/dnode/mnode/impl/src/mndStream.c +++ b/source/dnode/mnode/impl/src/mndStream.c @@ -937,7 +937,7 @@ static int32_t mndBuildStreamCheckpointSourceReq2(void **pBuf, int32_t *pLen, in } // static int32_t mndProcessStreamCheckpointTrans(SMnode *pMnode, SStreamObj *pStream, int64_t checkpointId) { // int64_t timestampMs = taosGetTimestampMs(); -// if (timestampMs - pStream->checkpointFreq < tsStreamCheckpointTickInterval * 1000) { +// if (timestampMs - pStream->checkpointFreq < tsStreamCheckpointInterval * 1000) { // return -1; // } diff --git a/source/dnode/vnode/src/tq/tq.c b/source/dnode/vnode/src/tq/tq.c index 8f3661dffa..1c90812d95 100644 --- a/source/dnode/vnode/src/tq/tq.c +++ b/source/dnode/vnode/src/tq/tq.c @@ -1196,7 +1196,11 @@ int32_t tqProcessTaskScanHistory(STQ* pTq, SRpcMsg* pMsg) { pStreamTask->status.keepTaskStatus = status; pStreamTask->status.taskStatus = TASK_STATUS__HALT; + // wal scan not start yet, reset it to be the start position nextProcessedVer = walReaderGetCurrentVer(pStreamTask->exec.pWalReader); + if (nextProcessedVer == -1) { + nextProcessedVer = pStreamTask->dataRange.range.maxVer + 1; + } tqDebug("s-task:%s level:%d nextProcessedVer:%" PRId64 ", sched-status:%d is halt by fill-history task:%s", pStreamTask->id.idStr, pStreamTask->info.taskLevel, nextProcessedVer, pStreamTask->status.schedStatus, diff --git a/source/libs/stream/inc/streamInt.h b/source/libs/stream/inc/streamInt.h index 4cd8319a07..d98fa2f436 100644 --- a/source/libs/stream/inc/streamInt.h +++ b/source/libs/stream/inc/streamInt.h @@ -122,7 +122,7 @@ 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, int32_t quotaRate); +int32_t streamTaskInitTokenBucket(STokenBucket* pBucket, int32_t numCap, int32_t numRate, float quotaRate); STaskId streamTaskExtractKey(const SStreamTask* pTask); void streamTaskInitForLaunchHTask(SHistoryTaskInfo* pInfo); void streamTaskSetRetryInfoForLaunch(SHistoryTaskInfo* pInfo); diff --git a/source/libs/stream/src/streamExec.c b/source/libs/stream/src/streamExec.c index 12b51e6c93..c49c647906 100644 --- a/source/libs/stream/src/streamExec.c +++ b/source/libs/stream/src/streamExec.c @@ -309,7 +309,9 @@ int32_t streamDoTransferStateToStreamTask(SStreamTask* pTask) { pStreamTask->id.idStr); } - ASSERT(pStreamTask->hTaskInfo.id.taskId == pTask->id.taskId && pTask->status.appendTranstateBlock == true); + ASSERT(((pStreamTask->status.taskStatus == TASK_STATUS__STOP) || + (pStreamTask->hTaskInfo.id.taskId == pTask->id.taskId)) && + pTask->status.appendTranstateBlock == true); STimeWindow* pTimeWindow = &pStreamTask->dataRange.window; diff --git a/source/libs/stream/src/streamQueue.c b/source/libs/stream/src/streamQueue.c index ae285046ef..e396ac77b4 100644 --- a/source/libs/stream/src/streamQueue.c +++ b/source/libs/stream/src/streamQueue.c @@ -18,7 +18,7 @@ #define MAX_STREAM_EXEC_BATCH_NUM 32 #define MAX_SMOOTH_BURST_RATIO 5 // 5 sec #define WAIT_FOR_DURATION 40 -#define SINK_TASK_IDLE_DURATION 200 // 200 ms +#define OUTPUT_QUEUE_FULL_WAIT_DURATION 500 // 500 ms // todo refactor: // read data from input queue @@ -119,14 +119,6 @@ int32_t streamQueueGetNumOfItems(const SStreamQueue* pQueue) { return numOfItems1 + numOfItems2; } -int32_t streamQueueGetAvailableSpace(const SStreamQueue* pQueue, int32_t* availNum, double* availSize) { - int32_t num = streamQueueGetNumOfItems(pQueue); - *availNum = STREAM_TASK_QUEUE_CAPACITY - num; - - *availSize = STREAM_TASK_QUEUE_CAPACITY_IN_SIZE - taosQueueMemorySize(pQueue->pQueue); - return 0; -} - // todo: fix it: data in Qall is not included here int32_t streamQueueGetItemSize(const SStreamQueue* pQueue) { return taosQueueMemorySize(pQueue->pQueue); @@ -362,9 +354,10 @@ int32_t streamTaskPutDataIntoOutputQ(SStreamTask* pTask, SStreamDataBlock* pBloc int32_t total = streamQueueGetNumOfItems(pTask->outputq.queue); double size = SIZE_IN_MiB(taosQueueMemorySize(pQueue)); // let's wait for there are enough space to hold this result pBlock - stDebug("s-task:%s outputQ is full, wait for 500ms and retry, outputQ items:%d, size:%.2fMiB", pTask->id.idStr, - total, size); - taosMsleep(500); + stDebug("s-task:%s outputQ is full, wait for %dms and retry, outputQ items:%d, size:%.2fMiB", pTask->id.idStr, + OUTPUT_QUEUE_FULL_WAIT_DURATION, total, size); + + taosMsleep(OUTPUT_QUEUE_FULL_WAIT_DURATION); } int32_t code = taosWriteQitem(pQueue, pBlock); @@ -381,7 +374,7 @@ int32_t streamTaskPutDataIntoOutputQ(SStreamTask* pTask, SStreamDataBlock* pBloc return TSDB_CODE_SUCCESS; } -int32_t streamTaskInitTokenBucket(STokenBucket* pBucket, int32_t numCap, int32_t numRate, int32_t quotaRate) { +int32_t streamTaskInitTokenBucket(STokenBucket* pBucket, int32_t numCap, int32_t numRate, float quotaRate) { if (numCap < 10 || numRate < 10 || pBucket == NULL) { stError("failed to init sink task bucket, cap:%d, rate:%d", numCap, numRate); return TSDB_CODE_INVALID_PARA; diff --git a/source/libs/stream/src/streamRecover.c b/source/libs/stream/src/streamStart.c similarity index 100% rename from source/libs/stream/src/streamRecover.c rename to source/libs/stream/src/streamStart.c diff --git a/source/libs/stream/src/streamTask.c b/source/libs/stream/src/streamTask.c index 57103e5a96..fd9902406d 100644 --- a/source/libs/stream/src/streamTask.c +++ b/source/libs/stream/src/streamTask.c @@ -427,7 +427,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, 2); + streamTaskInitTokenBucket(pTask->outputInfo.pTokenBucket, 50, 50, tsSinkDataRate); TdThreadMutexAttr attr = {0}; int code = taosThreadMutexAttrInit(&attr); diff --git a/source/util/src/tconfig.c b/source/util/src/tconfig.c index 34c3222bff..08bb8ecabf 100644 --- a/source/util/src/tconfig.c +++ b/source/util/src/tconfig.c @@ -416,16 +416,6 @@ int32_t cfgAddFloat(SConfig *pCfg, const char *name, float defaultVal, float min return cfgAddItem(pCfg, &item, name); } -int32_t cfgAddDouble(SConfig *pCfg, const char *name, double defaultVal, double minval, double maxval, int8_t scope) { - if (defaultVal < minval || defaultVal > maxval) { - terrno = TSDB_CODE_OUT_OF_RANGE; - return -1; - } - - SConfigItem item = {.dtype = CFG_DTYPE_DOUBLE, .dval = defaultVal, .fmin = minval, .fmax = maxval, .scope = scope}; - return cfgAddItem(pCfg, &item, name); -} - int32_t cfgAddString(SConfig *pCfg, const char *name, const char *defaultVal, int8_t scope) { SConfigItem item = {.dtype = CFG_DTYPE_STRING, .scope = scope}; item.str = taosStrdup(defaultVal); @@ -611,8 +601,7 @@ void cfgDumpCfg(SConfig *pCfg, bool tsc, bool dump) { switch (pItem->dtype) { case CFG_DTYPE_BOOL: if (dump) { - printf("%s %s %u", src, name, pItem->bval); - printf("\n"); + printf("%s %s %u\n", src, name, pItem->bval); } else { uInfo("%s %s %u", src, name, pItem->bval); } @@ -620,25 +609,22 @@ void cfgDumpCfg(SConfig *pCfg, bool tsc, bool dump) { break; case CFG_DTYPE_INT32: if (dump) { - printf("%s %s %d", src, name, pItem->i32); - printf("\n"); + printf("%s %s %d\n", src, name, pItem->i32); } else { uInfo("%s %s %d", src, name, pItem->i32); } break; case CFG_DTYPE_INT64: if (dump) { - printf("%s %s %" PRId64, src, name, pItem->i64); - printf("\n"); + printf("%s %s %" PRId64"\n", src, name, pItem->i64); } else { uInfo("%s %s %" PRId64, src, name, pItem->i64); } break; - case CFG_DTYPE_FLOAT: case CFG_DTYPE_DOUBLE: + case CFG_DTYPE_FLOAT: if (dump) { - printf("%s %s %.2f", src, name, pItem->fval); - printf("\n"); + printf("%s %s %.2f\n", src, name, pItem->fval); } else { uInfo("%s %s %.2f", src, name, pItem->fval); } @@ -650,8 +636,7 @@ void cfgDumpCfg(SConfig *pCfg, bool tsc, bool dump) { case CFG_DTYPE_TIMEZONE: case CFG_DTYPE_NONE: if (dump) { - printf("%s %s %s", src, name, pItem->str); - printf("\n"); + printf("%s %s %s\n", src, name, pItem->str); } else { uInfo("%s %s %s", src, name, pItem->str); } @@ -660,8 +645,7 @@ void cfgDumpCfg(SConfig *pCfg, bool tsc, bool dump) { } if (dump) { - printf("================================================================="); - printf("\n"); + printf("=================================================================\n"); } else { uInfo("================================================================="); }