github-actions[bot] commented on code in PR #67070:
URL: https://github.com/apache/doris/pull/67070#discussion_r3844985703
##########
be/src/exec/scan/scanner_context.cpp:
##########
@@ -562,6 +566,97 @@ 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
Review Comment:
[P1] Keep zero adaptive slots at the one-task escape
`MemLimiter::available_scanner_count()` can legitimately return 0 for later
non-serial instances when the node-wide scanner budget is smaller than operator
parallelism (for example, `max_count = 1` and `ins_idx = 1`).
`_available_pickup_scanner_count()` preserves that zero, but this fallback
restores `_max_scan_concurrency`. Because `_run_context()` queues its successor
before executing, callbacks can grow completed plus in-flight work from one to
the full maximum even without consumption; the TaskExecutor margin keeps only
the one progress task. This defeats the adaptive memory budget under pressure.
Treat zero as the real ceiling here and let only `current_concurrency == 0`
take the escape, with a zero-allocation parity test.
##########
be/src/exec/scan/scanner_context.cpp:
##########
@@ -541,15 +544,16 @@ std::string ScannerContext::debug_string() {
return fmt::format(
"_query_id: {}, id: {}, total scanners: {}, pending tasks: {},
completed tasks: {},"
" _should_stop: {}, _is_finished: {}, free blocks: {},"
- " limit: {}, _in_flight_tasks_num: {}, remaining_limit: {},
_num_running_scanners: {}, "
- "_max_thread_num: {},"
+ " limit: {}, _in_flight_tasks_num: {}, _is_context_queued: {},
remaining_limit: {}, "
Review Comment:
[P2] Align the new queue marker with its debug label
The placeholders after `limit` expect in-flight, queued, and remaining-limit
values, but the arguments are currently remaining limit, in-flight, and queued.
For example, remaining 100, in-flight 2, and queued true prints
`_in_flight_tasks_num: 100`, `_is_context_queued: 2`, and `remaining_limit:
true`. The new field therefore cannot diagnose the stalled Context this PR
targets. Reorder these arguments and add a distinguishable debug-string
assertion.
##########
be/src/exec/scan/simplified_scan_scheduler.cpp:
##########
@@ -34,7 +38,104 @@ 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 a non-EOS result, making this scanner
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);
+ }
+ if (scanner_ctx->is_context_queued(transfer_lock)) {
+ // A queued runnable will see all pending scanners added before it
obtains transfer_lock.
+ // Submitting another runnable would only duplicate work.
+ return Status::OK();
+ }
+ if (!scanner_ctx->can_admit_scan_task(transfer_lock)) {
+ // No runnable is needed when the Context has no pending scanner or
its concurrency slots
+ // are occupied. A completion only wakes the operator; the operator's
next consumption in
+ // get_block_from_queue(), or the successor submit in _run_context(),
retries scheduling.
+ return Status::OK();
+ }
+
+ if (_is_stop) {
+ Status failure = Status::InternalError<false>("scanner pool {} is
shutdown.", _sched_name);
+ scanner_ctx->set_context_failure(failure, transfer_lock);
+ return failure;
+ }
+
+ // ThreadPool::submit_func() may return an error after retaining the
runnable. Set the marker
+ // before submission so either outcome is safe: a retained callback clears
it, while a truly
+ // rejected callback leaves a terminal Context that no longer needs
rescheduling.
+ scanner_ctx->set_context_queued(true, transfer_lock);
+ Status status =
+ _scan_thread_pool->submit_func([this, scanner_ctx] {
_run_context(scanner_ctx); });
+ if (status.ok()) {
+ VLOG_DEBUG << "submit context runnable to scanner pool " <<
_sched_name << ", "
+ << scanner_ctx->debug_string();
+ return Status::OK();
+ }
+ Status failure =
+ Status::TooManyTasks("Failed to submit scanner context {} to
scanner pool, reason: {}",
+ scanner_ctx->ctx_id, status.msg());
+ scanner_ctx->set_context_failure(failure, transfer_lock);
+ return failure;
+}
+
+void
ThreadPoolSimplifiedScanScheduler::_run_context(std::shared_ptr<ScannerContext>
scanner_ctx) {
+ std::shared_ptr<ScanTask> scan_task;
+ Status admission_status = [&]() -> Status {
+ std::unique_lock<std::mutex>
transfer_lock(scanner_ctx->transfer_lock());
+ scanner_ctx->set_context_queued(false, transfer_lock);
+
+ auto task_execution_lock = scanner_ctx->task_exec_ctx();
+ if (task_execution_lock == nullptr) {
+ return Status::OK();
+ }
+#ifndef BE_TEST
+ // Attach before admission: allocations below (for example the
resubmitted
+ // FunctionRunnable) must charge the query rather than the orphan
tracker. Scoped to this
+ // lambda so it detaches before execute_scan_task(), whose
_scanner_scan() attaches again.
+ SCOPED_ATTACH_TASK(scanner_ctx->state());
+#endif
+ Status status = [&]() -> Status {
+ RETURN_IF_CATCH_EXCEPTION({
+ // Admission checks completed results, active tasks, and
adaptive limits while
+ // holding transfer_lock.
+ scan_task = scanner_ctx->try_get_next_scan_task(transfer_lock);
+ if (scan_task != nullptr) {
+
DBUG_EXECUTE_IF("ThreadPoolSimplifiedScanScheduler._run_context.inject_failure",
+ {
+ throw
Exception(ErrorCode::INTERNAL_ERROR,
+ "injected admission
failure");
+ });
+ // Queue the next Context runnable before executing this
task. Holding
+ // transfer_lock keeps the admission decision atomic.
+ RETURN_IF_ERROR(schedule_scan_task(scanner_ctx, nullptr,
transfer_lock));
Review Comment:
[P1] Preserve the scheduler-wide load budget
The previous ThreadPool path used `_get_margin()`: once active plus queued
work reached `_min_scan_concurrency_of_scan_scheduler`, each Context stayed at
its minimum outstanding scanners, normally one. This unconditional
pre-execution successor instead keeps admitting until every Context reaches
`_max_scan_concurrency`. With a backpressured operator, completion preserves
completed plus in-flight occupancy and each attempt force-allocates its first
block, so successive FIFO rounds can raise saturated per-Context block
occupancy from the default `1r` to `4r` across all Contexts even though the
pool has no load deficit. Keep an equivalent shared-pool budget without
reintroducing duplicate runnables, and add a saturated multi-Context test.
--
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]
