github-actions[bot] commented on code in PR #68571:
URL: https://github.com/apache/doris/pull/68571#discussion_r4121225259
##########
be/src/exec/scan/simplified_scan_scheduler.cpp:
##########
@@ -34,7 +39,112 @@ Status
TaskExecutorSimplifiedScanScheduler::schedule_scan_task(
Status ThreadPoolSimplifiedScanScheduler::schedule_scan_task(
std::shared_ptr<ScannerContext> scanner_ctx, std::shared_ptr<ScanTask>
current_scan_task,
std::unique_lock<std::mutex>& transfer_lock) {
- std::unique_lock<std::shared_mutex> wl(_lock);
- return scanner_ctx->schedule_scan_task(current_scan_task, transfer_lock,
wl);
+ // Unlike TaskExecutor, ThreadPool queues a Context runnable. It later
admits one pending task
+ // under transfer_lock. This bounds queue entries to one per Context even
when many scanners
+ // become runnable together.
+ DORIS_CHECK(transfer_lock.owns_lock());
+ if (current_scan_task != nullptr) {
+ // The operator has consumed all blocks of a non-EOS scanner, making
it eligible for
+ // another scan attempt. Queue the scanner first; the Context runnable
chooses it later.
+ scanner_ctx->push_pending_scan_task(std::move(current_scan_task),
transfer_lock);
+ }
+ const bool context_queued = scanner_ctx->is_context_queued(transfer_lock);
+ if (context_queued && !_is_stop) {
+ // A queued runnable will see all pending scanners added before it
obtains transfer_lock.
+ // Submitting another runnable would only duplicate work. Once the
pool is stopped, its
+ // shutdown drops queued runnables without running them, so the marker
may never be
+ // cleared; fail below instead of waiting for that runnable forever.
+ return Status::OK();
+ }
+ if (!context_queued && !scanner_ctx->can_admit_scan_task(transfer_lock,
false)) {
Review Comment:
[P1] Keep the scheduler-wide budget check and successor submission atomic
across Contexts. Each Context now holds only its own transfer lock. With three
pool threads, zero queue capacity, and two active Context callbacks each with
another scanner pending, both can observe `active + queued == 2 <
min_active_scan_threads == 3` and try to submit a successor. The first consumes
the last slot; the second gets `SERVICE_UNAVAILABLE` and fails its query. This
race remains even without discounting the current worker. The removed `_lock`
previously covered `_get_margin()` through submission, letting the second
Context defer instead of fail.
##########
be/src/exec/scan/scanner_context.cpp:
##########
@@ -448,17 +452,119 @@ std::string ScannerContext::debug_string() {
return fmt::format(
"id: {}, total scanners: {}, pending tasks: {},"
" _should_stop: {}, _is_finished: {}, free blocks: {},"
- " limit: {}, _num_running_scanners: {}, _max_thread_num: {},"
+ " limit: {}, _num_running_scanners: {}, _is_context_queued: {},"
+ " _num_finished_scanners: {}, _max_thread_num: {},"
" _max_bytes_in_queue: {}, query_id: {}",
ctx_id, _all_scanners.size(), _tasks_queue.size(), _should_stop,
_is_finished,
- _free_blocks.size_approx(), limit, _num_scheduled_scanners,
_max_scan_concurrency,
- _max_bytes_in_queue, print_id(_query_id));
+ _free_blocks.size_approx(), limit, _num_scheduled_scanners,
_is_context_queued,
+ _num_finished_scanners, _max_scan_concurrency, _max_bytes_in_queue,
+ print_id(_query_id));
}
void ScannerContext::_set_scanner_done() {
_dependency->set_always_ready();
}
+bool ScannerContext::is_context_queued(const std::unique_lock<std::mutex>&
transfer_lock) const {
+ DORIS_CHECK(transfer_lock.owns_lock());
+ return _is_context_queued;
+}
+
+void ScannerContext::set_context_queued(bool queued,
+ const std::unique_lock<std::mutex>&
transfer_lock) {
+ DORIS_CHECK(transfer_lock.owns_lock());
+ DORIS_CHECK(_is_context_queued != queued);
+ _is_context_queued = queued;
+}
+
+void ScannerContext::set_context_failure(const Status& failure,
+ const std::unique_lock<std::mutex>&
transfer_lock) {
+ DORIS_CHECK(transfer_lock.owns_lock());
+ DORIS_CHECK(!failure.ok());
+ _process_status = failure;
+ _is_finished = true;
+ _set_scanner_done();
+}
+
+void ScannerContext::push_pending_scan_task(std::shared_ptr<ScanTask>
scan_task,
+ const
std::unique_lock<std::mutex>& transfer_lock) {
+ DORIS_CHECK(transfer_lock.owns_lock());
+ DORIS_CHECK(scan_task != nullptr);
+ DORIS_CHECK(scan_task->cached_blocks.empty());
+ DORIS_CHECK(!scan_task->is_eos());
+ _pending_scanners.push(std::move(scan_task));
+}
+
+bool ScannerContext::can_admit_scan_task(const std::unique_lock<std::mutex>&
transfer_lock,
+ bool admitting_on_worker) const {
+ DORIS_CHECK(transfer_lock.owns_lock());
+ if (done() || _pending_scanners.empty()) {
+ return false;
+ }
+
+ // Blocks waiting in _tasks_queue still occupy a concurrency slot until
the operator consumes
+ // them. Counting both prevents a fast producer from exceeding the
per-Context scanner limit.
+ const int32_t current_concurrency =
+ cast_set<int32_t>(_tasks_queue.size()) + _num_scheduled_scanners;
+ // Keep one task progressing whatever the limits are. Otherwise no worker
can publish a result
+ // and wake the operator to make another scheduling decision.
+ if (current_concurrency == 0) {
+ return true;
+ }
+ // The per-Context ceiling applied by _pull_next_scan_task().
+ if (current_concurrency >= _max_scan_concurrency) {
+ return false;
+ }
+ // In low memory mode _get_margin() limits the number of running scanners.
+ if (low_memory_mode() && _num_scheduled_scanners >=
low_memory_mode_scanners()) {
+ return false;
+ }
+ // Mirror the scheduler-wide budget of _get_margin(): while the pool has
slack a Context may
+ // ramp to its maximum. Once it has none, a Context is held at its target
concurrency, which is
+ // its minimum unless the operator is starving. Both counters are read
here under
+ // _transfer_lock exactly as the TaskExecutor path reads them. A worker
admitting the task it
+ // runs itself is already counted as active; like the task _get_margin()
is about to submit, it
+ // must not count against the budget, otherwise the pool would stop one
slot short of it.
+ const int32_t busy_scan_slots = _scanner_scheduler->get_active_threads() +
+ _scanner_scheduler->get_queue_size() -
+ (admitting_on_worker ? 1 : 0);
+ if (busy_scan_slots < _min_scan_concurrency_of_scan_scheduler) {
+ return true;
+ }
+ const int32_t target_scan_concurrency =
+ _scan_starving && _tasks_queue.empty() ? _max_scan_concurrency :
_min_scan_concurrency;
+ return current_concurrency < target_scan_concurrency;
+}
+
+std::shared_ptr<ScanTask> ScannerContext::try_get_next_scan_task(
+ const std::unique_lock<std::mutex>& transfer_lock, int64_t
context_queue_wait_ns) {
+ if (!can_admit_scan_task(transfer_lock, true)) {
+ VLOG_DEBUG << fmt::format(
+ "[{}|{}] refuse admission, pending: {}, task queue: {},
scheduled: {}, done: {}",
+ print_id(_query_id), ctx_id, _pending_scanners.size(),
_tasks_queue.size(),
+ _num_scheduled_scanners, done());
+ return nullptr;
+ }
+
+ // Pop and count as scheduled while holding the same lock used by
completion and consumption.
+ // Thus concurrent Context workers cannot admit the same task or both pass
the limit check.
+ auto scan_task = _pending_scanners.top();
+ _pending_scanners.pop();
+ // ThreadPool admission bypasses ScannerScheduler::submit(); restart the
per-scanner wait
+ // timer here so it measures admission-to-execution instead of everything
since the previous
+ // attempt paused, which would include time the cached blocks waited for
the operator. The
+ // Context runnable waited for a worker on behalf of this scanner, so
credit that wait too.
+ if (auto scanner_delegate = scan_task->scanner.lock()) {
+ scanner_delegate->_scanner->start_wait_worker_timer();
+
scanner_delegate->_scanner->add_wait_worker_time(context_queue_wait_ns);
Review Comment:
[P3] Credit wait time from the selected scanner's own pending interval. With
one worker, positive queue capacity, two scanners, and per-Context minimum
concurrency 2, a successor Context runnable is queued before scanner A runs. A
can finish and be LIFO-requeued by the operator before that successor starts;
it then selects A here and credits the entire runnable queue age, including A's
own execution time. This inflates `ScannerWorkerWaitTime` and
`PerScannerWaitTime`. A per-scanner pending timestamp would preserve the
metric's meaning.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
