github-actions[bot] commented on code in PR #66985:
URL: https://github.com/apache/doris/pull/66985#discussion_r3820398280
##########
be/src/exec/scan/scanner_context.cpp:
##########
@@ -559,10 +573,89 @@ void ScannerContext::_set_scanner_done() {
_dependency->set_always_ready();
}
-bool ScannerContext::_is_shared_scan_limit_exhausted() const {
+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);
+ if (queued) {
+ // A Context is deduplicated while queued, so this timestamp covers
exactly one submitted
+ // runnable rather than the wait time of any particular scanner it may
later choose.
+ DORIS_CHECK(_context_wait_worker_start_ns == 0);
+ _context_wait_worker_start_ns = MonotonicNanos();
+ } else {
+ // A worker clears the state immediately after dequeueing the
runnable. Record the elapsed
+ // time here so queue-state changes and profiling cannot diverge.
Failed submissions never
+ // set this state, and therefore never enter this branch.
+ DORIS_CHECK(_context_wait_worker_start_ns != 0);
+#ifndef BE_TEST
+ DORIS_CHECK(_context_wait_worker_timer != nullptr);
+ COUNTER_UPDATE(_context_wait_worker_timer,
+ MonotonicNanos() - _context_wait_worker_start_ns);
+#endif
+ _context_wait_worker_start_ns = 0;
+ }
+ _is_context_queued = queued;
+}
+
+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));
+}
+
+std::shared_ptr<ScanTask> ScannerContext::try_get_next_scan_task(
+ const std::unique_lock<std::mutex>& transfer_lock) {
+ DORIS_CHECK(transfer_lock.owns_lock());
+ if (done() || _pending_tasks.empty()) {
Review Comment:
[P1] Recompute adaptive concurrency on the ThreadPool path
`expected_scanners` starts at 0 and is assigned only by
`_available_pickup_scanner_count()`, which is reached from the TaskExecutor
`_get_margin()` path. The new default ThreadPool path calls
`try_get_next_scan_task()` directly, so this branch always substitutes
`_max_scan_concurrency` and never feeds updated block estimates back through
`MemLimiter`/`MemShareArbitrator`. Because `enable_adaptive_scan` defaults to
true and file scans default to 16 scanners per context, a context whose memory
budget permits one scanner can still admit all 16, defeating the adaptive OOM
protection. Please compute/refresh the adaptive limit during ThreadPool
admission and cover it with a real ThreadPool adaptive test.
##########
be/src/exec/scan/simplified_scan_scheduler.cpp:
##########
@@ -34,7 +36,73 @@ 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 and distort
Context queue latency.
+ return Status::OK();
+ }
+
+ // transfer_lock prevents another producer from submitting concurrently.
The worker callback
+ // also waits for this lock, so it cannot run between successful
submission and marking queued.
+ Status status;
+ if (_is_stop) {
+ status = Status::InternalError<false>("scanner pool {} is shutdown.",
_sched_name);
+ } else {
Review Comment:
[P2] Preserve the scanner queue-rejection error contract
This direct `submit_func()` path bypasses `ScannerScheduler::submit()`,
which intentionally converts pool submission failures to `TOO_MANY_TASKS` and
adds the scanner storage type. With both scheduler defaults changed to
ThreadPool, ordinary local and remote scan saturation now surfaces raw
`SERVICE_UNAVAILABLE` instead, so error handling and diagnostics depend on the
selected scheduler. Please normalize capacity rejection consistently here
(while keeping true shutdown distinct if intended) and add a saturated-pool
parity test.
##########
be/src/exec/scan/simplified_scan_scheduler.cpp:
##########
@@ -34,7 +36,73 @@ 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);
+ }
Review Comment:
[P1] Submit a context only when it can admit work
The last EOS is consumed before `get_block_from_queue()` performs its
terminal check, and that path calls `schedule_scan_task(ctx, nullptr)`. Here,
with `_pending_tasks` empty and `_is_context_queued` false, we still call
`submit_func()`. If the workload-group scan pool is at capacity,
`ThreadPool::do_submit()` returns `SERVICE_UNAVAILABLE`, which propagates from
`get_block_from_queue()` and fails a query whose scan has already completed.
The same missing can-admit predicate also creates a no-op callback after each
admission that fills the Context concurrency cap. Please submit only when
pending work can currently be admitted under `transfer_lock`, and test final
EOS with a rejecting/full pool plus concurrency-one dispatch counts.
##########
be/src/exec/scan/simplified_scan_scheduler.cpp:
##########
@@ -34,7 +36,73 @@ 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 and distort
Context queue latency.
+ return Status::OK();
+ }
+
+ // transfer_lock prevents another producer from submitting concurrently.
The worker callback
+ // also waits for this lock, so it cannot run between successful
submission and marking queued.
+ Status status;
+ if (_is_stop) {
+ status = Status::InternalError<false>("scanner pool {} is shutdown.",
_sched_name);
+ } else {
+ status = _scan_thread_pool->submit_func([this, scanner_ctx] {
_run_context(scanner_ctx); });
+ }
+ if (status.ok()) {
+ // Start the Context wait interval only after submission succeeds.
This excludes failed
+ // submit_func() calls, which never waited for a worker and must not
affect the profile.
+ scanner_ctx->set_context_queued(true, transfer_lock);
+ } else {
+ // No worker can dequeue a rejected runnable. The Context remains
unqueued, so a later
+ // scheduling attempt can submit it again without clearing state or
accounting queue time.
+ LOG(WARNING) << fmt::format("Failed to submit scanner context {},
reason: {}",
+ scanner_ctx->debug_string(),
status.to_string());
+ }
+ return status;
+}
+
+void
ThreadPoolSimplifiedScanScheduler::_run_context(std::shared_ptr<ScannerContext>
scanner_ctx) {
+ std::shared_ptr<ScanTask> scan_task;
+ {
+ std::unique_lock<std::mutex>
transfer_lock(scanner_ctx->transfer_lock());
+ // The worker has dequeued the Context. Clearing the marker also
charges its queue latency:
+ // the interval from successful submit_func() to worker start, not
scanner execution time.
+ scanner_ctx->set_context_queued(false, transfer_lock);
+
+ auto task_execution_lock = scanner_ctx->task_exec_ctx();
+ if (task_execution_lock == nullptr) {
+ return;
+ }
+
+ // Admission checks completed results, active tasks, adaptive limits,
and shared LIMIT while
+ // holding transfer_lock. A null task means the Context is currently
not allowed to run one.
+ scan_task = scanner_ctx->try_get_next_scan_task(transfer_lock);
+ if (scan_task == nullptr) {
+ return;
+ }
+
+ // Queue the next Context runnable before executing this task.
Example: with a concurrency
+ // limit of two, the next worker may admit scanner B while this worker
scans scanner A.
+ // Releasing transfer_lock only after resubmission keeps the admission
decision atomic.
+ Status resubmit_status = schedule_scan_task(scanner_ctx, nullptr,
transfer_lock);
+ if (!resubmit_status.ok()) {
+ LOG(WARNING) << fmt::format("Failed to resubmit scanner context
{}, reason: {}",
+ scanner_ctx->ctx_id,
resubmit_status.to_string());
+ }
+ }
+ // The scan runs without transfer_lock so the operator and other Context
workers can continue
+ // consuming results and admitting work. Completion reacquires the lock
before publishing.
Review Comment:
[P2] Exclude consumer backpressure from scanner worker-wait time
On this path no code resets the scanner's wait watch at admission. The prior
attempt's `pause()` starts that watch when its block is produced, and the next
`resume()` adds everything since then, including time the completed block
waited for operator consumption. The aggregate scanner worker-wait counter and
`PerScannerWaitTime` therefore report downstream/consumer delay as thread-pool
queue delay; the new Context timer does not stop those existing counters.
Please reset or suppress the legacy scanner timer at the ThreadPool admission
boundary and test with a deliberately blocked consumer.
##########
be/src/exec/scan/simplified_scan_scheduler.cpp:
##########
@@ -34,7 +36,73 @@ 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 and distort
Context queue latency.
+ return Status::OK();
+ }
+
+ // transfer_lock prevents another producer from submitting concurrently.
The worker callback
+ // also waits for this lock, so it cannot run between successful
submission and marking queued.
+ Status status;
+ if (_is_stop) {
+ status = Status::InternalError<false>("scanner pool {} is shutdown.",
_sched_name);
+ } else {
+ status = _scan_thread_pool->submit_func([this, scanner_ctx] {
_run_context(scanner_ctx); });
+ }
+ if (status.ok()) {
+ // Start the Context wait interval only after submission succeeds.
This excludes failed
+ // submit_func() calls, which never waited for a worker and must not
affect the profile.
+ scanner_ctx->set_context_queued(true, transfer_lock);
+ } else {
+ // No worker can dequeue a rejected runnable. The Context remains
unqueued, so a later
+ // scheduling attempt can submit it again without clearing state or
accounting queue time.
+ LOG(WARNING) << fmt::format("Failed to submit scanner context {},
reason: {}",
+ scanner_ctx->debug_string(),
status.to_string());
+ }
+ return status;
+}
+
+void
ThreadPoolSimplifiedScanScheduler::_run_context(std::shared_ptr<ScannerContext>
scanner_ctx) {
+ std::shared_ptr<ScanTask> scan_task;
+ {
+ std::unique_lock<std::mutex>
transfer_lock(scanner_ctx->transfer_lock());
+ // The worker has dequeued the Context. Clearing the marker also
charges its queue latency:
+ // the interval from successful submit_func() to worker start, not
scanner execution time.
+ scanner_ctx->set_context_queued(false, transfer_lock);
+
+ auto task_execution_lock = scanner_ctx->task_exec_ctx();
+ if (task_execution_lock == nullptr) {
+ return;
+ }
+
+ // Admission checks completed results, active tasks, adaptive limits,
and shared LIMIT while
+ // holding transfer_lock. A null task means the Context is currently
not allowed to run one.
+ scan_task = scanner_ctx->try_get_next_scan_task(transfer_lock);
+ if (scan_task == nullptr) {
+ return;
+ }
+
+ // Queue the next Context runnable before executing this task.
Example: with a concurrency
+ // limit of two, the next worker may admit scanner B while this worker
scans scanner A.
+ // Releasing transfer_lock only after resubmission keeps the admission
decision atomic.
Review Comment:
[P1] Attach and catch the callback-side resubmission scope
Pinning `task_exec_ctx()` does not attach this worker to the query. After
admission increments `_in_flight_tasks_num`, this call allocates the next
`FunctionRunnable` while the thread is still on the Orphan tracker; the first
`SCOPED_ATTACH_TASK` and `RETURN_IF_CATCH_EXCEPTION` are only reached later
inside `_scanner_scan()`. With orphan checking enabled this violates the
callback-entry memory invariant, and an allocation/tracker exception escapes
`ThreadPool::dispatch_thread()` (which has no catch), terminating the process
without publishing the admitted task or releasing its slot. Please attach the
pre-execution callback scope and convert any failure after admission into a
completed error task, with fault-injected coverage.
##########
be/src/exec/scan/simplified_scan_scheduler.cpp:
##########
@@ -34,7 +36,73 @@ 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 and distort
Context queue latency.
+ return Status::OK();
+ }
+
+ // transfer_lock prevents another producer from submitting concurrently.
The worker callback
+ // also waits for this lock, so it cannot run between successful
submission and marking queued.
+ Status status;
+ if (_is_stop) {
+ status = Status::InternalError<false>("scanner pool {} is shutdown.",
_sched_name);
+ } else {
+ status = _scan_thread_pool->submit_func([this, scanner_ctx] {
_run_context(scanner_ctx); });
+ }
+ if (status.ok()) {
+ // Start the Context wait interval only after submission succeeds.
This excludes failed
+ // submit_func() calls, which never waited for a worker and must not
affect the profile.
+ scanner_ctx->set_context_queued(true, transfer_lock);
+ } else {
Review Comment:
[P1] Handle submissions that fail after retaining the runnable
A non-OK `submit_func()` does not always mean that no worker can ever
dequeue this callback. `ThreadPool::do_submit()` first appends the task, then
creates a needed worker; if creation fails while the pool has zero workers, it
returns that error without removing the queued task. This branch therefore
leaves `_is_context_queued` false, but a later successful worker creation can
run the stale callback, whose `set_context_queued(false)` hits the
queue-state/start-time `DORIS_CHECK`s and crashes the BE. Please make
failed-after-enqueue submission transactional (or expose an accepted/queued
result that lets this code mark the callback) and add a deterministic
worker-creation failure/recovery 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]
