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

refactor: wait for a while when no quota available.

This commit is contained in:
Haojun Liao 2023-11-07 17:49:49 +08:00
parent 06d02b862e
commit 43b14ba8b4
1 changed files with 3 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
source/libs/stream/src/streamQueue.c
View File

@ -160,7 +160,7 @@ int32_t streamTaskGetDataFromInputQ(SStreamTask* pTask, SStreamQueueItem** pInpu
// no available token in bucket for sink task, let's wait for a little bit // no available token in bucket for sink task, let's wait for a little bit
if (taskLevel == TASK_LEVEL__SINK && (!streamTaskExtractAvailableToken(pTask->outputInfo.pTokenBucket, pTask->id.idStr))) { if (taskLevel == TASK_LEVEL__SINK && (!streamTaskExtractAvailableToken(pTask->outputInfo.pTokenBucket, pTask->id.idStr))) {
stDebug("s-task:%s no available token in bucket for sink data, wait for 10ms", id); stDebug("s-task:%s no available token in bucket for sink data, wait for 10ms", id);
// taosMsleep(10); taosMsleep(10);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
@ -414,9 +414,8 @@ static void fillTokenBucket(STokenBucket* pBucket, const char* id) {
} }
if (incNum > 0 || incSize > 0) { if (incNum > 0 || incSize > 0) {
stDebug("new token and capacity available, current token:%d inc:%d, current quota:%.2fMiB inc:%.3fMiB, ts:%" PRId64 stTrace("token/quota available, token:%d inc:%d, quota:%.2fMiB inc:%.3fMiB, ts:%" PRId64 " idle:%" PRId64 "ms, %s",
" idle for %.2f Sec, %s", pBucket->numOfToken, incNum, pBucket->quotaRemain, incSize, now, delta, id);
pBucket->numOfToken, incNum, pBucket->quotaRemain, incSize, now, delta / 1000.0, id);
} }
} }