From 83950025e51612093c3398d1150b57d07e8600a6 Mon Sep 17 00:00:00 2001 From: Haojun Liao Date: Wed, 31 Jan 2024 08:58:17 +0800 Subject: [PATCH] fix(stream): limit the available threads for stream processing. --- source/common/src/tglobal.c | 2 +- source/util/src/tworker.c | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/source/common/src/tglobal.c b/source/common/src/tglobal.c index a7d80fe5db..3b965553af 100644 --- a/source/common/src/tglobal.c +++ b/source/common/src/tglobal.c @@ -58,7 +58,7 @@ int32_t tsNumOfMnodeQueryThreads = 4; int32_t tsNumOfMnodeFetchThreads = 1; int32_t tsNumOfMnodeReadThreads = 1; int32_t tsNumOfVnodeQueryThreads = 4; -float tsRatioOfVnodeStreamThreads = 1.0; +float tsRatioOfVnodeStreamThreads = 0.5; int32_t tsNumOfVnodeFetchThreads = 4; int32_t tsNumOfVnodeRsmaThreads = 2; int32_t tsNumOfQnodeQueryThreads = 4; diff --git a/source/util/src/tworker.c b/source/util/src/tworker.c index 57dc60e539..28f7fd4783 100644 --- a/source/util/src/tworker.c +++ b/source/util/src/tworker.c @@ -221,7 +221,7 @@ STaosQueue *tAutoQWorkerAllocQueue(SAutoQWorkerPool *pool, void *ahandle, FItem int32_t queueNum = taosGetQueueNumber(pool->qset); int32_t curWorkerNum = taosArrayGetSize(pool->workers); int32_t dstWorkerNum = ceilf(queueNum * pool->ratio); - if (dstWorkerNum < 1) dstWorkerNum = 1; + if (dstWorkerNum < 2) dstWorkerNum = 2; // spawn a thread to process queue while (curWorkerNum < dstWorkerNum) {