github-actions[bot] commented on code in PR #66838:
URL: https://github.com/apache/doris/pull/66838#discussion_r3795726675
##########
be/src/exec/scan/scanner_context.cpp:
##########
@@ -178,6 +180,36 @@ int ScannerContext::_available_pickup_scanner_count() {
return scanners;
}
+static Status init_task_executor(ScannerScheduler* scanner_scheduler,
RuntimeState* state,
+ const std::string& ctx_id,
+ std::weak_ptr<TaskExecutor>* task_executor,
+ std::shared_ptr<TaskHandle>* task_handle,
+ int32_t* max_scan_concurrency) {
+ if (auto* task_executor_scheduler =
+
dynamic_cast<TaskExecutorSimplifiedScanScheduler*>(scanner_scheduler)) {
+ std::shared_ptr<TaskExecutor> executor =
task_executor_scheduler->task_executor();
+ *task_executor = executor;
+ TaskId task_id(fmt::format("{}-{}", print_id(state->query_id()),
ctx_id));
+ int initial_task_concurrency =
+ config::task_executor_initial_max_concurrency_per_task > 0
+ ?
config::task_executor_initial_max_concurrency_per_task
+ : std::max(48, CpuInfo::num_cores() * 2);
+ if (config::task_executor_max_concurrency_per_task > 0) {
+ initial_task_concurrency = std::min(initial_task_concurrency,
+
config::task_executor_max_concurrency_per_task);
+ }
+ *task_handle = DORIS_TRY(executor->create_task(
+ task_id, []() { return 0.0; }, initial_task_concurrency,
+ std::chrono::milliseconds(100), std::nullopt));
+ // ScannerContext counts every submitted task as in-flight, while
TaskExecutor keeps
+ // submissions beyond this limit in a private per-handle queue. Do not
create that hidden
+ // backlog: a non-EOS scanner may be parked by ScannerContext while it
still owns a task
+ // handle slot, so queued tasks cannot provide a wake-up if every
admitted scanner parks.
Review Comment:
The assignment immediately below turns
`task_executor_initial_max_concurrency_per_task` into a lifetime cap for the
context. That setting is documented as an initial target, and
`SplitConcurrencyController` plus
`test_min_concurrency_per_task_when_target_concurrency_increases` explicitly
support growing it. With `initial=2`, `max=16`, and a file scan eligible for 16
scanners, the handle target can grow but `_pull_next_scan_task()` can never
submit a third occupied slot after this assignment, so the scan stays pinned at
two scanners. The new test uses `initial=max=2`, which hides this regression.
Please preserve the eventual local/configured maximum while coordinating
submissions with the handle's current admission capacity (or otherwise fixing
the hidden-slot accounting), and add an `initial < max` test that proves both
liveness and scale-up.
--
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]
