xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Merge pull request #24685 from taosdata/fix/3_liaohj 

fix(stream): limit the available threads for stream processing.
This commit is contained in:
Haojun Liao 2024-01-31 11:13:15 +08:00 committed by GitHub
parent e0583cb343 2716119c92
commit 91c9abfcba
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 3 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
source/common/src/tglobal.c
View File

@ -58,7 +58,7 @@ int32_t tsNumOfMnodeQueryThreads = 4;
int32_t tsNumOfMnodeFetchThreads = 1; int32_t tsNumOfMnodeFetchThreads = 1;
int32_t tsNumOfMnodeReadThreads = 1; int32_t tsNumOfMnodeReadThreads = 1;
int32_t tsNumOfVnodeQueryThreads = 4; int32_t tsNumOfVnodeQueryThreads = 4;
float tsRatioOfVnodeStreamThreads = 1.0; float tsRatioOfVnodeStreamThreads = 0.5F;
int32_t tsNumOfVnodeFetchThreads = 4; int32_t tsNumOfVnodeFetchThreads = 4;
int32_t tsNumOfVnodeRsmaThreads = 2; int32_t tsNumOfVnodeRsmaThreads = 2;
int32_t tsNumOfQnodeQueryThreads = 4; int32_t tsNumOfQnodeQueryThreads = 4;
@ -622,7 +622,7 @@ static int32_t taosAddServerCfg(SConfig *pCfg) {
0) 0)
return -1; return -1;
if (cfgAddFloat(pCfg, "ratioOfVnodeStreamThreads", tsRatioOfVnodeStreamThreads, 0.01, 10, CFG_SCOPE_SERVER, if (cfgAddFloat(pCfg, "ratioOfVnodeStreamThreads", tsRatioOfVnodeStreamThreads, 0.01, 4, CFG_SCOPE_SERVER,
CFG_DYN_NONE) != 0) CFG_DYN_NONE) != 0)
return -1; return -1;

2
source/util/src/tworker.c
View File

@ -221,7 +221,7 @@ STaosQueue *tAutoQWorkerAllocQueue(SAutoQWorkerPool *pool, void *ahandle, FItem
int32_t queueNum = taosGetQueueNumber(pool->qset); int32_t queueNum = taosGetQueueNumber(pool->qset);
int32_t curWorkerNum = taosArrayGetSize(pool->workers); int32_t curWorkerNum = taosArrayGetSize(pool->workers);
int32_t dstWorkerNum = ceilf(queueNum * pool->ratio); int32_t dstWorkerNum = ceilf(queueNum * pool->ratio);
if (dstWorkerNum < 1) dstWorkerNum = 1; if (dstWorkerNum < 2) dstWorkerNum = 2;
// spawn a thread to process queue // spawn a thread to process queue
while (curWorkerNum < dstWorkerNum) { while (curWorkerNum < dstWorkerNum) {