github-actions[bot] commented on code in PR #67070:
URL: https://github.com/apache/doris/pull/67070#discussion_r3842878988
##########
be/src/exec/scan/scanner_context.cpp:
##########
@@ -562,6 +565,90 @@ bool ScannerContext::_is_shared_scan_limit_exhausted()
const {
return limit >= 0 && _shared_scan_limit->load(std::memory_order_acquire)
<= 0;
}
+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_block == nullptr);
+ DORIS_CHECK(!scan_task->is_eos());
+ // The state transition documents that this is an admission queue, not a
completed-result queue.
+ scan_task->set_state(ScanTask::State::PENDING);
+ _pending_tasks.push(std::move(scan_task));
+}
+
+bool ScannerContext::can_admit_scan_task(const std::unique_lock<std::mutex>&
transfer_lock) const {
+ DORIS_CHECK(transfer_lock.owns_lock());
+ if (done() || _pending_tasks.empty()) {
Review Comment:
[P1] Restore the terminal admission gates
A Context callback can already be queued when either the shared LIMIT
becomes exhausted or a scanner publishes a fatal `_process_status`, but this
predicate checks neither condition. In the LIMIT case it can admit unnecessary
work and a saturated successor submission makes `get_block_from_queue()` return
`TOO_MANY_TASKS` before draining the valid LIMIT result. In the error case the
same path can overwrite the original scan/schema error with that incidental
pool failure. Please refuse admission once `_process_status` is non-OK and
mirror `_pull_next_scan_task()`'s shared-LIMIT/progress gate, retaining the
one-task escape hatch only when pending work is this Context's sole possible
wakeup; add deterministic coverage for both stale-callback interleavings.
##########
be/src/exec/scan/scanner_context.cpp:
##########
@@ -562,6 +565,90 @@ bool ScannerContext::_is_shared_scan_limit_exhausted()
const {
return limit >= 0 && _shared_scan_limit->load(std::memory_order_acquire)
<= 0;
}
+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_block == nullptr);
+ DORIS_CHECK(!scan_task->is_eos());
+ // The state transition documents that this is an admission queue, not a
completed-result queue.
+ scan_task->set_state(ScanTask::State::PENDING);
+ _pending_tasks.push(std::move(scan_task));
+}
+
+bool ScannerContext::can_admit_scan_task(const std::unique_lock<std::mutex>&
transfer_lock) const {
+ DORIS_CHECK(transfer_lock.owns_lock());
+ if (done() || _pending_tasks.empty()) {
+ return false;
+ }
+
+ int32_t effective_max_concurrency = _max_scan_concurrency;
+ if (_enable_adaptive_scanners) {
+ effective_max_concurrency = _adaptive_processor->expected_scanners > 0
+ ?
_adaptive_processor->expected_scanners
+ : _max_scan_concurrency;
+ }
+ if (low_memory_mode()) {
+ effective_max_concurrency = std::min(effective_max_concurrency,
low_memory_mode_scanners());
+ }
+
+ // Completed blocks still occupy a concurrency slot until the operator
consumes them. Counting
+ // both collections prevents a fast producer from exceeding the
per-Context scanner limit.
+ const int32_t current_concurrency =
+ cast_set<int32_t>(_completed_tasks.size()) + _in_flight_tasks_num;
+ // Keep one task progressing even if an adaptive limit temporarily reaches
zero. Otherwise no
+ // worker can publish a result and wake the operator to make another
scheduling decision.
+ return current_concurrency == 0 || current_concurrency <
effective_max_concurrency;
+}
+
+std::shared_ptr<ScanTask> ScannerContext::try_get_next_scan_task(
+ const std::unique_lock<std::mutex>& transfer_lock) {
+ if (_enable_adaptive_scanners) {
+ // Refresh expected_scanners and feed current block estimates back to
the memory limiter,
+ // exactly as _get_margin() does on the TaskExecutor path.
can_admit_scan_task() only reads
+ // the cached value, so ThreadPool admission would otherwise never
apply adaptive limits.
+ static_cast<void>(_available_pickup_scanner_count());
+ }
+ if (!can_admit_scan_task(transfer_lock)) {
+ return nullptr;
+ }
+
+ // Pop and mark in-flight 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_tasks.top();
+ _pending_tasks.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 completed block waited for
the operator.
+ if (auto scanner_delegate = scan_task->scanner.lock()) {
+ scanner_delegate->_scanner->start_wait_worker_timer();
Review Comment:
[P2] Keep Context queue delay in `ScannerWorkerWaitTime`
This timer is reset only after the Context runnable has already waited in
and been dequeued from the ThreadPool, and `execute_scan_task()` calls
`resume()` almost immediately afterward. Under saturation, a Context can wait
seconds for a scan worker while the query's `ScannerWorkerWaitTime` records
only the tiny post-dequeue interval (the generic pool metric is not
query/scanner scoped). Please carry the Context enqueue-to-start delay into the
scanner selected at admission, while still excluding the time its previous
completed block waited for the operator, and cover that delay in a test.
##########
be/src/exec/scan/scanner_context.cpp:
##########
@@ -562,6 +565,90 @@ bool ScannerContext::_is_shared_scan_limit_exhausted()
const {
return limit >= 0 && _shared_scan_limit->load(std::memory_order_acquire)
<= 0;
}
+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_block == nullptr);
+ DORIS_CHECK(!scan_task->is_eos());
+ // The state transition documents that this is an admission queue, not a
completed-result queue.
+ scan_task->set_state(ScanTask::State::PENDING);
+ _pending_tasks.push(std::move(scan_task));
+}
+
+bool ScannerContext::can_admit_scan_task(const std::unique_lock<std::mutex>&
transfer_lock) const {
+ DORIS_CHECK(transfer_lock.owns_lock());
+ if (done() || _pending_tasks.empty()) {
+ return false;
+ }
+
+ int32_t effective_max_concurrency = _max_scan_concurrency;
+ if (_enable_adaptive_scanners) {
+ effective_max_concurrency = _adaptive_processor->expected_scanners > 0
+ ?
_adaptive_processor->expected_scanners
+ : _max_scan_concurrency;
+ }
+ if (low_memory_mode()) {
+ effective_max_concurrency = std::min(effective_max_concurrency,
low_memory_mode_scanners());
Review Comment:
[P2] Apply the low-memory cap to running scanners
Folding the low-memory limit into `effective_max_concurrency` makes
completed blocks consume the four-scanner allowance. For example, with
`_max_scan_concurrency > 4`, four completed blocks and zero in-flight tasks
cause this path to admit nothing until the operator consumes a block. The
established TaskExecutor path instead applies `low_memory_mode_scanners() -
_in_flight_tasks_num` separately from the ordinary completed-plus-in-flight
ceiling, so up to four scanners can continue running under memory pressure.
Please preserve that running-scanner cap here and add a parity test with
completed, pending, and in-flight tasks.
--
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]
