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

refactor stream scan log

This commit is contained in:
liuyao 2023-08-15 16:29:48 +08:00
parent 37e6100204
commit 19dfcf3a59
3 changed files with 5 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
source/libs/executor/src/scanoperator.c
View File

@ -2050,7 +2050,7 @@ FETCH_NEXT_BLOCK:
default: default:
break; break;
} }
printSpecDataBlock(pBlock, getStreamOpName(pOperator->operatorType), "recv", GET_TASKID(pTaskInfo)); printDataBlock(pBlock, getStreamOpName(pOperator->operatorType), GET_TASKID(pTaskInfo));
return pBlock; return pBlock;
} else if (pInfo->blockType == STREAM_INPUT__DATA_SUBMIT) { } else if (pInfo->blockType == STREAM_INPUT__DATA_SUBMIT) {
qDebug("stream scan mode:%d, %s", pInfo->scanMode, id); qDebug("stream scan mode:%d, %s", pInfo->scanMode, id);

4
source/libs/stream/src/streamExec.c
View File

@ -415,11 +415,11 @@ static int32_t extractMsgFromInputQ(SStreamTask* pTask, SStreamQueueItem** pInpu
if (qItem == NULL) { if (qItem == NULL) {
if (pTask->info.taskLevel == TASK_LEVEL__SOURCE && (++retryTimes) < MAX_RETRY_TIMES) { if (pTask->info.taskLevel == TASK_LEVEL__SOURCE && (++retryTimes) < MAX_RETRY_TIMES) {
taosMsleep(10); taosMsleep(10);
qDebug("===stream===try again batchSize:%d, retry:%d", *numOfBlocks, retryTimes); qDebug("try again batchSize:%d, retry:%d", *numOfBlocks, retryTimes);
continue; continue;
} }
qDebug("===stream===break batchSize:%d", *numOfBlocks); qDebug("break batchSize:%d", *numOfBlocks);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }

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

@ -130,11 +130,11 @@ SStreamQueueItem* doReadMultiBlocksFromQueue(SQueueReader* pReader, const char*
if (pReader->taskLevel == TASK_LEVEL__SOURCE && numOfBlocks < MIN_STREAM_EXEC_BATCH_NUM && tryCount < pReader->waitDuration) { if (pReader->taskLevel == TASK_LEVEL__SOURCE && numOfBlocks < MIN_STREAM_EXEC_BATCH_NUM && tryCount < pReader->waitDuration) {
tryCount++; tryCount++;
taosMsleep(1); taosMsleep(1);
qDebug("===stream===try again batchSize:%d", numOfBlocks); qDebug("try again batchSize:%d", numOfBlocks);
continue; continue;
} }
qDebug("===stream===break batchSize:%d", numOfBlocks); qDebug("break batchSize:%d", numOfBlocks);
break; break;
} }