Skip to content

branch-4.1: [opt](scan) Schedule scanners up to max concurrency #66106

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
HappenLee wants to merge 1 commit into from
Closed

branch-4.1: [opt](scan) Schedule scanners up to max concurrency #66106

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 7 additions & 19 deletions be/src/exec/scan/scanner_context.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
}
Expand Down Expand Up @@ -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());
Expand Down
Loading