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

fix(stream): reduce the sink throughput. and fix the bug in status check.

This commit is contained in:
Haojun Liao 2023-11-23 18:04:09 +08:00
parent 523573844a
commit 9d735e7f8b
2 changed files with 6 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
source/dnode/mnode/impl/src/mndStream.c
View File

@ -2775,11 +2775,14 @@ static bool needDropRelatedFillhistoryTask(STaskStatusEntry *pTaskEntry, SStream
if (pTaskEntry->id.streamId == pId->streamId) { if (pTaskEntry->id.streamId == pId->streamId) {
numOfTotal++; numOfTotal++;
if (pTaskEntry->id.taskId != pId->taskId && pTaskEntry->status == TASK_STATUS__READY) { if (pTaskEntry->id.taskId != pId->taskId) {
STaskStatusEntry *pEntry = taosHashGet(execInfo.pTaskMap, pId, sizeof(*pId));
if (pEntry->status == TASK_STATUS__READY) {
numOfReady++; numOfReady++;
} }
} }
} }
}
if (numOfReady > 0) { if (numOfReady > 0) {
mDebug("stream:0x%" PRIx64 mDebug("stream:0x%" PRIx64

2
source/libs/stream/src/streamTask.c
View File

@ -448,7 +448,7 @@ int32_t streamTaskInit(SStreamTask* pTask, SStreamMeta* pMeta, SMsgCb* pMsgCb, i
// 2MiB per second for sink task // 2MiB per second for sink task
// 50 times sink operator per second // 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}; TdThreadMutexAttr attr = {0};
int code = taosThreadMutexAttrInit(&attr); int code = taosThreadMutexAttrInit(&attr);