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 29034570a699ea6c86387d0e2aaba33a7ff1f160 Author: Hu Shenggang <[email protected]> AuthorDate: Tue Jul 14 18:47:22 2026 +0800 [opt](scan) Schedule scanners up to max concurrency --- be/src/exec/scan/scanner_context.cpp | 26 +++++++------------------- 1 file changed, 7 insertions(+), 19 deletions(-) diff --git a/be/src/exec/scan/scanner_context.cpp b/be/src/exec/scan/scanner_context.cpp index f7376fd2946..2115f119425 100644 --- a/be/src/exec/scan/scanner_context.cpp +++ b/be/src/exec/scan/scanner_context.cpp @@ -451,33 +451,21 @@ 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) { - // 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); + int32_t margin = _max_scan_concurrency - _num_scheduled_scanners; - // 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) { + if (margin <= 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_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); + VLOG_DEBUG << fmt::format("[{}|{}] schedule scan task, margin: {} = {} - {}", + print_id(_query_id), ctx_id, margin, _max_scan_concurrency, + _num_scheduled_scanners); return margin; } @@ -526,8 +514,8 @@ Status ScannerContext::schedule_scan_task(std::shared_ptr<ScanTask> current_scan while (margin-- > 0) { std::shared_ptr<ScanTask> task_to_run; - const int32_t current_concurrency = cast_set<int32_t>( - _tasks_queue.size() + _num_scheduled_scanners + tasks_to_submit.size()); + const auto current_concurrency = + cast_set<int32_t>(_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]
