This is an automated email from the ASF dual-hosted git repository. mrhhsg pushed a commit to branch opt_perf_4.1 in repository https://gitbox.apache.org/repos/asf/doris.git
commit 4834528b3fb0d95d64c76cec99cd56d86c5a4fd5 Author: Hu Shenggang <[email protected]> AuthorDate: Fri Jul 17 16:30:05 2026 +0800 Revert "[opt](scan) Schedule scanners up to max concurrency" This reverts commit 29034570a699ea6c86387d0e2aaba33a7ff1f160. --- be/src/exec/scan/scanner_context.cpp | 26 +++++++++++++++++++------- 1 file changed, 19 insertions(+), 7 deletions(-) diff --git a/be/src/exec/scan/scanner_context.cpp b/be/src/exec/scan/scanner_context.cpp index 2115f119425..f7376fd2946 100644 --- a/be/src/exec/scan/scanner_context.cpp +++ b/be/src/exec/scan/scanner_context.cpp @@ -451,21 +451,33 @@ void ScannerContext::update_peak_running_scanner(int num) { int32_t ScannerContext::_get_margin(std::unique_lock<std::mutex>& transfer_lock, std::unique_lock<std::shared_mutex>& scheduler_lock) { - int32_t margin = _max_scan_concurrency - _num_scheduled_scanners; + // margin_1 is used to ensure each scan operator could have at least _min_scan_concurrency scan tasks. + int32_t margin_1 = _min_scan_concurrency - + (cast_set<int32_t>(_tasks_queue.size()) + _num_scheduled_scanners); - if (margin <= 0) { + // margin_2 is used to ensure the scan scheduler could have at least _min_scan_concurrency_of_scan_scheduler scan tasks. + int32_t margin_2 = + _min_scan_concurrency_of_scan_scheduler - + (_scanner_scheduler->get_active_threads() + _scanner_scheduler->get_queue_size()); + + if (margin_1 <= 0 && margin_2 <= 0) { return 0; } + int32_t margin = std::max(margin_1, margin_2); + if (low_memory_mode()) { // In low memory mode, we will limit the number of running scanners to `low_memory_mode_scanners()`. // So that we will not submit too many scan tasks to scheduler. margin = std::min(low_memory_mode_scanners() - _num_scheduled_scanners, margin); } - VLOG_DEBUG << fmt::format("[{}|{}] schedule scan task, margin: {} = {} - {}", - print_id(_query_id), ctx_id, margin, _max_scan_concurrency, - _num_scheduled_scanners); + VLOG_DEBUG << fmt::format( + "[{}|{}] schedule scan task, margin_1: {} = {} - ({} + {}), margin_2: {} = {} - " + "({} + {}), margin: {}", + print_id(_query_id), ctx_id, margin_1, _min_scan_concurrency, _tasks_queue.size(), + _num_scheduled_scanners, margin_2, _min_scan_concurrency_of_scan_scheduler, + _scanner_scheduler->get_active_threads(), _scanner_scheduler->get_queue_size(), margin); return margin; } @@ -514,8 +526,8 @@ Status ScannerContext::schedule_scan_task(std::shared_ptr<ScanTask> current_scan while (margin-- > 0) { std::shared_ptr<ScanTask> task_to_run; - const auto current_concurrency = - cast_set<int32_t>(_num_scheduled_scanners + tasks_to_submit.size()); + const int32_t current_concurrency = cast_set<int32_t>( + _tasks_queue.size() + _num_scheduled_scanners + tasks_to_submit.size()); VLOG_DEBUG << fmt::format("{} currenct concurrency: {} = {} + {} + {}", ctx_id, current_concurrency, _tasks_queue.size(), _num_scheduled_scanners, tasks_to_submit.size()); --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
