github-actions[bot] commented on code in PR #67066:
URL: https://github.com/apache/doris/pull/67066#discussion_r3842692189
##########
be/src/common/config.cpp:
##########
@@ -333,9 +333,9 @@ DEFINE_Int32(task_executor_max_concurrency_per_task, "-1");
DEFINE_Int32(task_executor_initial_max_concurrency_per_task, "-1");
// Enable task executor in internal table scan.
-DEFINE_Bool(enable_task_executor_in_internal_table, "false");
+DEFINE_Bool(enable_task_executor_in_internal_table, "true");
Review Comment:
[P1] Do not default to TaskExecutor while submit failures can strand splits
— Initial scanner admission increments the Context's in-flight count, but
`_start_split()` only logs `_do_submit()` failures and `enqueue_splits()` still
returns success. Capacity rejection leaves the split marked running with no
executor queue owner; a failed first worker creation leaves it queued in an
executor that can have zero workers because `init()` ignores minimum-worker
creation failures. In either case nothing releases the slot or wakes the
dependency, so the scan can hang. Please propagate/roll back pre-publication
rejection and guarantee progress for retained work, with deterministic
ScannerContext coverage, before enabling this path by default.
##########
be/src/common/config.cpp:
##########
@@ -333,9 +333,9 @@ DEFINE_Int32(task_executor_max_concurrency_per_task, "-1");
DEFINE_Int32(task_executor_initial_max_concurrency_per_task, "-1");
// Enable task executor in internal table scan.
-DEFINE_Bool(enable_task_executor_in_internal_table, "false");
+DEFINE_Bool(enable_task_executor_in_internal_table, "true");
Review Comment:
[P1] Fix the TaskExecutor shutdown wakeup before enabling it by default — A
scanner submission can reserve the only pending worker and enter
`_create_thread()` outside the lock while concurrent scheduler `stop()` clears
the queue and waits for live+pending to become zero. If creation fails,
`_do_submit()` decrements the final pending count and returns without notifying
`_no_threads_cond`; no dispatch worker exists to send the other notification,
so WorkloadGroup shutdown can wait forever. Please notify this transition and
add deterministic stop/create-failure coverage before switching the defaults.
##########
be/src/util/threadpool.cpp:
##########
@@ -454,12 +454,10 @@ Status ThreadPool::do_submit(std::shared_ptr<Runnable> r,
ThreadPoolToken* token
// We assume that each current inactive thread will grab one item from the
// queue. If it seems like we'll need another thread, we create one.
//
- // Rather than creating an additional thread here, while holding the lock,
- // we defer it to down below. This is because thread creation can be rather
- // slow (hundreds of milliseconds in some cases) and we'd like to allow the
- // existing threads to continue to process tasks while we do so. The first
- // thread is an exception: it must be created before publishing the task so
- // a failed submission cannot leave a runnable in a pool with no workers.
+ // Rather than creating the thread here, while holding the lock, we defer
Review Comment:
[P1] Keep failed submissions from retaining the runnable — With a
zero-worker pool, this now publishes the task before `create_thread()`. If
`pthread_create` fails, the branch below returns non-OK without removing the
token/queue entry, so a later successful submit can execute a callback its
caller already treated as rejected. That is unsafe for generic callers: for
example, `S3FileWriter::close()` performs the close and fulfills its promise
synchronously after a failed submit, while other callers capture stack/object
state by reference. Please create the first worker before publication (or fully
roll back the task) so a non-OK submit guarantees the runnable cannot run, and
retain the deleted regression coverage.
##########
be/src/util/threadpool.cpp:
##########
@@ -582,12 +563,7 @@ void ThreadPool::dispatch_thread() {
break;
}
- // A runtime shrink may leave more live + pending threads than
_max_threads. Excess threads
- // retire here, but the last live worker must stay while tasks are
queued: a pending
- // replacement may still fail to start, and a pool with queued tasks
and no worker can only
- // recover on the next submit. The excess thread retires once the
queue drains.
- if (_num_threads + _num_threads_pending_start > _max_threads &&
- (_num_threads > 1 || _queue.empty())) {
+ if (_num_threads + _num_threads_pending_start > _max_threads) {
Review Comment:
[P1] Keep the last live worker while work is queued — A submit can publish
task B and count one pending replacement while the current worker is active. If
the pool is concurrently shrunk from 2 to 1, this condition retires that sole
live worker because `1 live + 1 pending > 1`; if the replacement then fails to
start, B is stranded with no worker and waits cannot complete. Please preserve
the last live worker while `_queue` is nonempty until another worker has
actually started, and retain the deleted shrink/failure regression test.
##########
be/src/common/config.cpp:
##########
@@ -333,9 +333,9 @@ DEFINE_Int32(task_executor_max_concurrency_per_task, "-1");
DEFINE_Int32(task_executor_initial_max_concurrency_per_task, "-1");
// Enable task executor in internal table scan.
-DEFINE_Bool(enable_task_executor_in_internal_table, "false");
+DEFINE_Bool(enable_task_executor_in_internal_table, "true");
// Enable task executor in external table scan.
-DEFINE_Bool(enable_task_executor_in_external_table, "false");
+DEFINE_Bool(enable_task_executor_in_external_table, "true");
Review Comment:
[P1] Preserve a live TaskExecutor worker across runtime shrink — External
WorkloadGroup schedulers can shrink from max 2 to 1 while one worker is active,
task B is queued, and a replacement is only pending. This exit condition then
retires the sole live worker; if the replacement's thread creation fails, B
remains queued with no worker and the scanner dependency cannot complete.
Please keep the last live worker while queued work relies only on a fallible
pending replacement, and cover resize plus injected creation failure before
making this executor the default.
##########
be/src/common/config.cpp:
##########
@@ -333,9 +333,9 @@ DEFINE_Int32(task_executor_max_concurrency_per_task, "-1");
DEFINE_Int32(task_executor_initial_max_concurrency_per_task, "-1");
// Enable task executor in internal table scan.
-DEFINE_Bool(enable_task_executor_in_internal_table, "false");
+DEFINE_Bool(enable_task_executor_in_internal_table, "true");
Review Comment:
[P1] Honor the configured per-task maximum on the new default path —
Production constructs TimeSharing with `enable_concurrency_control=false`, so
direct scanner `enqueue_splits()` polls against the handle's initial target (at
least 48) and never consults `task_executor_max_concurrency_per_task`; that
executor-wide cap is checked only on a later entrant path. Setting the
documented maximum to 1 can therefore still promote the Context's normal 4
internal or 16 external splits. Please enforce the maximum even when adaptive
control is disabled and cover the production constructor mode before enabling
it by default.
##########
be/src/common/config.cpp:
##########
@@ -333,9 +333,9 @@ DEFINE_Int32(task_executor_max_concurrency_per_task, "-1");
DEFINE_Int32(task_executor_initial_max_concurrency_per_task, "-1");
// Enable task executor in internal table scan.
-DEFINE_Bool(enable_task_executor_in_internal_table, "false");
+DEFINE_Bool(enable_task_executor_in_internal_table, "true");
// Enable task executor in external table scan.
-DEFINE_Bool(enable_task_executor_in_external_table, "false");
+DEFINE_Bool(enable_task_executor_in_external_table, "true");
Review Comment:
[P1] Preserve the low-memory occupancy cap on the new default path —
TaskExecutor's low-memory clamp subtracts only in-flight tasks, although
completed blocks still occupy scanner memory and concurrency slots. After four
tasks complete, consuming one can leave three completed and admit four more
when the scheduler margin is positive, producing seven occupied tasks despite
the documented cap of four; the ThreadPool path correctly compares completed
plus in-flight. Please cap total occupied concurrency and add this
completed-block state to TaskExecutor coverage before enabling it by default
for external scans.
##########
be/src/util/threadpool.cpp:
##########
@@ -750,9 +717,6 @@ Status ThreadPool::set_min_threads(int min_threads) {
Status ThreadPool::set_max_threads(int max_threads) {
std::lock_guard<std::mutex> l(_lock);
- if (max_threads <= 0) {
Review Comment:
[P2] Reject a zero runtime maximum — When `min_threads` is zero, removing
this check lets `set_max_threads(0)` succeed even though the builder requires a
positive maximum. Existing workers then retire; queued work can trip the
last-worker `CHECK`, and an idle pool can later accept a task without ever
satisfying the condition that creates a worker. Please retain the
positive-maximum invariant and its deleted regression test.
##########
be/src/exec/scan/scanner_context.cpp:
##########
@@ -885,6 +840,13 @@ 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] Apply the exhausted-LIMIT guard to the current scanner too — After a
peer exhausts the shared LIMIT, consuming scanner A's non-EOS block can leave
scanner B completed or in flight, so the Context is not terminal yet. This
function returns A before reaching this new guard; with a positive minimum and
a saturated default TaskExecutor, the unnecessary re-enqueue is rejected and
`TOO_MANY_TASKS` replaces an already-complete LIMIT result even though B
guarantees progress. Please check the exhausted-LIMIT/progress condition before
selecting `current_scan_task` and cover this consumed-block/re-enqueue case.
##########
be/src/exec/scan/scanner_scheduler.cpp:
##########
@@ -185,10 +185,6 @@ void
ScannerScheduler::_scanner_scan(std::shared_ptr<ScannerContext> ctx,
ASSIGN_STATUS_IF_CATCH_EXCEPTION(
RuntimeState* state = ctx->state(); DCHECK(nullptr != state);
- // Do not suppress admission when shared LIMIT is exhausted. A
queued scanner still
- // needs to complete as EOS so push_completed_scan_task() releases
its in-flight slot
- // and the pipeline can observe completion instead of waiting
indefinitely.
- if (ctx->is_shared_scan_limit_exhausted()) { eos = true; }
Review Comment:
[P1] Check shared LIMIT before preparing or opening an admitted scanner —
Scanner A can satisfy the shared LIMIT while scanner B is already admitted but
has not started. Without this guard, B reaches `prepare()`/`open()` before
`Scanner::get_block()` notices the exhausted counter, so an otherwise unused
JDBC/tablet/file split can publish an open error and fail the query before A's
valid final block is consumed. Please retain the pre-open EOS check (while
still publishing B to release its in-flight slot) and cover a second scanner
whose open would fail after a peer produces the final row.
##########
be/src/exec/scan/simplified_scan_scheduler.cpp:
##########
@@ -58,43 +58,32 @@ Status
ThreadPoolSimplifiedScanScheduler::schedule_scan_task(
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 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()) {
- // 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);
- return Status::OK();
- }
- // 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());
- if (scanner_ctx->has_progressing_task(transfer_lock) ||
- scanner_ctx->is_shared_scan_limit_exhausted()) {
- // Someone will retry: a progressing task completes, the operator
consumes its result and
- // reschedules; after shared LIMIT is exhausted,
get_block_from_queue() finishes the
- // Context on its next call. Pool saturation must not fail a query
that still progresses.
return Status::OK();
}
- // Nothing will retry this Context. Surface the failure, normalized like
- // ScannerScheduler::submit() so both schedulers report saturation as
TOO_MANY_TASKS.
- return Status::TooManyTasks("Failed to submit scanner context {} to
scanner pool, reason: {}",
- scanner_ctx->ctx_id, status.msg());
+ 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);
Review Comment:
[P2] Do not abort when an admitted scanner guarantees retry progress — This
path also handles ordinary queue-capacity rejection, which occurs before
ThreadPool retains the callback. `_run_context()` has already admitted scanner
A and is only pre-queuing the next Context; executing A would publish a
result/EOS, wake the operator, and provide a safe retry point. Marking the
Context terminal instead turns transient backpressure into a query failure.
Please preserve retry-on-progress for unambiguous pre-enqueue rejection while
fixing the separate generic post-publication failure contract.
##########
be/src/util/threadpool.cpp:
##########
@@ -537,16 +521,13 @@ Status ThreadPool::do_submit(std::shared_ptr<Runnable> r,
ThreadPoolToken* token
l.lock();
_num_threads_pending_start--;
if (_num_threads + _num_threads_pending_start == 0) {
- _no_threads_cond.notify_all();
- }
- if (_pool_status.ok()) {
- // A published task either already ran or still has a live or
pending worker.
- // The last-worker guard in dispatch_thread() keeps this true;
a violation is a
- // pool logic error.
- DORIS_CHECK(_queue.empty() || _num_threads +
_num_threads_pending_start > 0);
- LOG(WARNING) << "Thread pool " << _name
- << " failed to create thread: " <<
status.to_string();
+ // If we have no threads, we can't do any work.
+ return status;
Review Comment:
[P1] Wake shutdown when the final worker creation fails — `shutdown()` can
clear the queues and wait on `_no_threads_cond` while this submitter is outside
the lock creating the only pending worker. If creation fails, this branch
decrements live+pending to zero and returns without notifying, and there is no
dispatch thread left to send the other notification. The concurrent shutdown
can therefore wait forever. Please notify `_no_threads_cond` whenever this
failure transition reaches zero and add deterministic concurrent-shutdown
coverage.
--
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]
