github-actions[bot] commented on code in PR #66985:
URL: https://github.com/apache/doris/pull/66985#discussion_r3840339112
##########
be/src/exec/scan/scanner_context.cpp:
##########
@@ -755,13 +885,6 @@ std::shared_ptr<ScanTask>
ScannerContext::_pull_next_scan_task(
}
if (!_pending_tasks.empty()) {
- // Do not submit more pending scanners after the shared LIMIT is
exhausted while
- // completed or in-flight tasks can still make progress. If neither
exists, allow pending
- // scanners to be submitted so they can report EOS and wake the
pipeline task.
- if (_is_shared_scan_limit_exhausted() &&
Review Comment:
[P1] Keep this guard until progressing TaskExecutor work drains
Removing this guard can strand a first-schedule split on the still-supported
TaskExecutor path. For example, with shared LIMIT exhausted,
`min_scanners_concurrency >= 2`, one scanner still in flight, and pending
scanners, `_get_margin()` remains positive and now pulls another scanner even
if the executor is at capacity. `submit_scan_task()` increments the Context
in-flight count, then `enqueue_splits()` polls the runner into the task handle;
if `_do_submit()` rejects it, `_start_split()` only logs that Status and
`enqueue_splits()` still reports success. The runner was never put on the
executor queue and has no retry, so the Context can wait forever with a leaked
in-flight slot. This guard previously kept it pending until the existing
scanner completed, woke the operator, and released capacity before the liveness
admission. Please retain equivalent progressing-work suppression for
TaskExecutor and add this capacity-full shared-LIMIT interleaving to the tests.
##########
be/src/exec/scan/simplified_scan_scheduler.cpp:
##########
@@ -34,7 +38,108 @@ 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();
+ }
+ 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. Completion or operator consumption will retry
scheduling when state changes.
+ 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.
+ if (_is_stop) {
+ // Shutdown must surface: progressing tasks may never complete on a
stopped pool.
+ return Status::InternalError<false>("scanner pool {} is shutdown.",
_sched_name);
+ }
+ Status 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);
Review Comment:
[P2] Measure Context wait from actual publication to dequeue
`submit_func()` can publish and timestamp this runnable, then synchronously
spend hundreds of milliseconds creating an additional worker. This line starts
`ScannerContextWaitWorkerTime` only after that call returns, so when existing
workers are occupied it omits the runnable's real queued interval and can
report nearly zero as the new worker starts. Conversely, if a worker dequeues
early, `_run_context()` cannot clear the timer until it acquires the
caller-held transfer lock, so the counter can include post-dequeue lock wait.
That does not match the source/header claim of actual thread-pool queue
latency. Please capture a pre-submit candidate timestamp and commit it only
when submission is accepted (or reuse the pool enqueue timestamp), stop it at
dequeue, and cover slow additional-worker creation.
--
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]
