This is an automated email from the ASF dual-hosted git repository. yiguolei pushed a commit to branch scanner_refactor in repository https://gitbox.apache.org/repos/asf/doris.git
commit 2f36dcedcd623275405af9c17f08b203142a1af6 Author: yiguolei <[email protected]> AuthorDate: Thu Aug 20 09:49:17 2026 +0800 [improvement](be) Refactor thread-pool scan scheduling Issue Number: None Related PR: None Problem Summary: Thread-pool scan scheduling submitted individual scanner tasks, allowing one ScannerContext to occupy multiple queue nodes and scattering admission decisions between scheduling and execution. Queue ScannerContext runnables instead, deduplicate each context while queued, atomically admit tasks under the context transfer lock, and execute admitted tasks on the current worker after resubmission. None - Test: Unit Test (added coverage; execution blocked because no Java runtime is installed) - Header hygiene: build-support/check-build-hygiene.sh - Unit Test: ./run-be-ut.sh --run --filter=ScannerContextTest.test_init (blocked: Java runtime unavailable) - Formatting: build-support/clang-format.sh (blocked: Homebrew llvm@16 keg unavailable) - Behavior changed: Yes (ThreadPoolSimplifiedScanScheduler uses context FIFO scheduling) - Does this need documentation: No [fix](be) Complete scanners after shared limit exhaustion Issue Number: None Related PR: None Problem Summary: Checking the shared scan limit while admitting pending tasks can prevent queued scanners from running to EOS and releasing their in-flight slots. Move the check into scanner execution so each admitted scanner completes and the pipeline can observe completion. None - Test: No need to test (scheduler control-flow change; header hygiene and whitespace checks passed) - Behavior changed: Yes (shared-limit exhaustion is handled during scanner execution) - Does this need documentation: No f f --- be/src/common/config.cpp | 4 +- be/src/exec/scan/scanner_context.cpp | 112 ++++++++++++++++++++++--- be/src/exec/scan/scanner_context.h | 70 +++++++++++++--- be/src/exec/scan/scanner_scheduler.cpp | 34 +++++--- be/src/exec/scan/scanner_scheduler.h | 9 +- be/src/exec/scan/simplified_scan_scheduler.cpp | 72 +++++++++++++++- be/test/exec/scan/scanner_context_test.cpp | 36 +++++++- 7 files changed, 291 insertions(+), 46 deletions(-) diff --git a/be/src/common/config.cpp b/be/src/common/config.cpp index a6b86212f3c..b9ac4d2d103 100644 --- a/be/src/common/config.cpp +++ b/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, "true"); +DEFINE_Bool(enable_task_executor_in_internal_table, "false"); // Enable task executor in external table scan. -DEFINE_Bool(enable_task_executor_in_external_table, "true"); +DEFINE_Bool(enable_task_executor_in_external_table, "false"); // number of scanner thread pool size for olap table // and the min thread num of remote scanner thread pool diff --git a/be/src/exec/scan/scanner_context.cpp b/be/src/exec/scan/scanner_context.cpp index 7fbf6ed951c..877caa9f576 100644 --- a/be/src/exec/scan/scanner_context.cpp +++ b/be/src/exec/scan/scanner_context.cpp @@ -184,6 +184,9 @@ Status ScannerContext::init() { _scanner_profile = _local_state->_scanner_profile; _newly_create_free_blocks_num = _local_state->_newly_create_free_blocks_num; _scanner_memory_used_counter = _local_state->_memory_used_counter; + // ThreadPool scheduling queues a Context rather than a Scanner. Its queue delay is therefore + // meaningful only as Context-level latency; per-scanner delays depend on arbitrary selection. + _context_wait_worker_timer = ADD_TIMER(_scanner_profile, "ScannerContextWaitWorkerTime"); // 3. get thread token if (!_state->get_query_ctx()) { @@ -328,9 +331,9 @@ void ScannerContext::return_free_block(BlockUPtr block) { Status ScannerContext::submit_scan_task(std::shared_ptr<ScanTask> scan_task, std::unique_lock<std::mutex>& /*transfer_lock*/) { // increase _num_finished_scanners no matter the scan_task is submitted successfully or not. - // since if submit failed, it will be added back by ScannerContext::push_back_scan_task + // since if submit failed, it will be added back by ScannerContext::push_completed_scan_task // and _num_finished_scanners will be reduced. - // if submit succeed, it will be also added back by ScannerContext::push_back_scan_task + // if submit succeed, it will be also added back by ScannerContext::push_completed_scan_task // see ScannerScheduler::_scanner_scan. _in_flight_tasks_num++; return _scanner_scheduler->submit(shared_from_this(), scan_task); @@ -340,7 +343,7 @@ void ScannerContext::clear_free_blocks() { clear_blocks(_free_blocks); } -void ScannerContext::push_back_scan_task(std::shared_ptr<ScanTask> scan_task) { +void ScannerContext::push_completed_scan_task(std::shared_ptr<ScanTask> scan_task) { if (scan_task->status_ok()) { if (scan_task->cached_block && scan_task->cached_block->rows() > 0) { Status st = validate_block_schema(scan_task->cached_block.get()); @@ -350,6 +353,8 @@ void ScannerContext::push_back_scan_task(std::shared_ptr<ScanTask> scan_task) { } } + // Publishing the result and releasing its in-flight slot must be atomic. Otherwise a worker + // could observe an available slot before the operator can observe this completed task. std::lock_guard<std::mutex> l(_transfer_lock); if (!scan_task->status_ok()) { _process_status = scan_task->get_status(); @@ -410,15 +415,24 @@ Status ScannerContext::get_block_from_queue(RuntimeState* state, Block* block, b _num_finished_scanners++; RETURN_IF_ERROR(_scanner_scheduler->schedule_scan_task(shared_from_this(), nullptr, l)); } else { - scan_task->set_state(ScanTask::State::IN_FLIGHT); + // A completed non-EOS attempt is still non-terminal. This covers a task whose block + // was just consumed above as well as one that produced no block. The scheduler returns + // it to PENDING (ThreadPool) or re-admits it (TaskExecutor) for another scan. RETURN_IF_ERROR( _scanner_scheduler->schedule_scan_task(shared_from_this(), scan_task, l)); } } + // A scanner can make shared LIMIT exhausted while it is still producing the final block: + // 1. The remaining limit is 5 and an in-flight scanner reads 100 rows. + // 2. The scanner decrements the shared limit below zero, then publishes its 100-row block. If not check + // _in_flight_tasks_num == 0 here, the operator will find the limit has reached, but actuall it + // do not get the cached block yet. + // 3. The operator consumes the block and applies the final limit of 5 rows. + // Therefore, wait for every worker to publish its block or EOS before completing this Context. if (_completed_tasks.empty() && (_num_finished_scanners == _all_scanners.size() || - (_is_shared_scan_limit_exhausted() && _in_flight_tasks_num == 0))) { + (is_shared_scan_limit_exhausted() && _in_flight_tasks_num == 0))) { _set_scanner_done(); _is_finished = true; } @@ -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()) { + return nullptr; + } + + int32_t effective_max_concurrency = _max_scan_concurrency; + if (_enable_adaptive_scanners) { + effective_max_concurrency = _adaptive_processor->expected_scanners > 0 + ? _adaptive_processor->expected_scanners + : _max_scan_concurrency; + } + if (low_memory_mode()) { + effective_max_concurrency = std::min(effective_max_concurrency, low_memory_mode_scanners()); + } + + // Completed blocks still occupy a concurrency slot until the operator consumes them. Counting + // both collections prevents a fast producer from exceeding the per-Context scanner limit. + const int32_t current_concurrency = + cast_set<int32_t>(_completed_tasks.size()) + _in_flight_tasks_num; + // Keep at least one task progressing whenever a pending scanner exists: + // 1. An adaptive or low-memory limit can temporarily reduce effective concurrency to zero. + // 2. If there are no completed or in-flight tasks, no worker can publish a block or EOS. + // 3. Admit one scanner in that case so the query can make progress and cannot stall. + const bool has_progressing_task = current_concurrency > 0; + if (has_progressing_task && current_concurrency >= effective_max_concurrency) { + return nullptr; + } + + // Pop and mark in-flight while holding the same lock used by completion and consumption. + // Thus concurrent Context workers cannot admit the same task or both pass the limit check. + auto scan_task = _pending_tasks.top(); + _pending_tasks.pop(); + scan_task->set_state(ScanTask::State::IN_FLIGHT); + ++_in_flight_tasks_num; + return scan_task; +} + void ScannerContext::update_peak_running_scanner(int num) { #ifndef BE_TEST _local_state->_peak_running_scanner->add(num); @@ -755,13 +848,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() && - (_in_flight_tasks_num != 0 || !_completed_tasks.empty())) { - return nullptr; - } std::shared_ptr<ScanTask> next_scan_task; next_scan_task = _pending_tasks.top(); _pending_tasks.pop(); diff --git a/be/src/exec/scan/scanner_context.h b/be/src/exec/scan/scanner_context.h index b5730897cdb..62f91591b98 100644 --- a/be/src/exec/scan/scanner_context.h +++ b/be/src/exec/scan/scanner_context.h @@ -129,7 +129,11 @@ public: void set_state(State state) { switch (state) { case State::PENDING: - DCHECK(_state == State::PENDING || _state == State::IN_FLIGHT) << (int)_state; + // A task returns to PENDING after the operator consumes its non-EOS cached block. + // For example, one scanner may produce several blocks, so COMPLETED is not terminal. + DCHECK(_state == State::PENDING || _state == State::IN_FLIGHT || + _state == State::COMPLETED) + << (int)_state; DCHECK(cached_block == nullptr); break; case State::IN_FLIGHT: @@ -208,12 +212,18 @@ public: // set the `eos` to `ScanTask::eos` if there is no more data in current scanner Status submit_scan_task(std::shared_ptr<ScanTask> scan_task, std::unique_lock<std::mutex>&); - // Push back a scan task. - void push_back_scan_task(std::shared_ptr<ScanTask> scan_task); + // Publish a task whose current scan attempt has completed. The operator consumes its cached + // block and returns a non-EOS task to PENDING for its next scan attempt. + void push_completed_scan_task(std::shared_ptr<ScanTask> scan_task); // Return true if this ScannerContext need no more process bool done() const { return _is_finished || _should_stop; } + // This is checked by ScannerScheduler::_scanner_scan(), rather than task admission, so an + // already queued scanner can finish as EOS and release its in-flight slot after shared LIMIT + // is reached. The limit is atomic and can therefore be read without _transfer_lock. + bool is_shared_scan_limit_exhausted() const; + std::string debug_string(); std::shared_ptr<TaskHandle> task_handle() const { return _task_handle; } @@ -254,6 +264,30 @@ public: std::unique_lock<std::mutex>& transfer_lock, std::unique_lock<std::shared_mutex>& scheduler_lock); + // Context scheduling and operator consumption share this lock so queue-state changes and task + // admission form one atomic decision. For example, two worker callbacks cannot both admit the + // last available concurrency slot. + std::mutex& transfer_lock() { return _transfer_lock; } + + // One Context runnable represents many pending scanners in the ThreadPool scheduler. Keeping + // this separate from scanner execution prevents duplicate Context runnables from accumulating. + bool is_context_queued(const std::unique_lock<std::mutex>& transfer_lock) const; + // Transition the Context runnable's queue state and maintain its wait-time interval. Setting + // true records enqueue time; clearing false charges that interval to the Context profile. + // The scheduler sets true only after submit succeeds, so failed submissions have no interval. + // Example: a queue-full submit leaves the state false and contributes no worker-wait time. + void set_context_queued(bool queued, const std::unique_lock<std::mutex>& transfer_lock); + + // Return a scanner to the admission queue after its block is consumed. It may not own a cached + // block and may not be EOS: EOS scanners are terminal and must not run again. + void push_pending_scan_task(std::shared_ptr<ScanTask> scan_task, + const std::unique_lock<std::mutex>& transfer_lock); + + // Atomically check whether this context can start another scan task, move one task from + // pending to in-flight, and return it. The caller must hold _transfer_lock. + std::shared_ptr<ScanTask> try_get_next_scan_task( + const std::unique_lock<std::mutex>& transfer_lock); + protected: /// Four criteria to determine whether to increase the parallelism of the scanners /// 1. It ran for at least `SCALE_UP_DURATION` ms after last scale up @@ -261,7 +295,6 @@ protected: /// 3. `_free_blocks_memory_usage` < `_max_bytes_in_queue`, remains enough memory to scale up /// 4. At most scale up `MAX_SCALE_UP_RATIO` times to `_max_thread_num` void _set_scanner_done(); - bool _is_shared_scan_limit_exhausted() const; RuntimeState* _state = nullptr; ScanLocalStateBase* _local_state = nullptr; @@ -295,26 +328,37 @@ protected: // current_concurrency = _completed_tasks.size() + _in_flight_tasks_num // // Lifecycle of a ScanTask: - // _pending_tasks --(submit_scan_task)--> [thread pool] --(push_back_scan_task)--> + // _pending_tasks --(submit_scan_task)--> [thread pool] --(push_completed_scan_task)--> // _completed_tasks --(get_block_from_queue)--> operator // After consumption: non-EOS task goes back to _pending_tasks; EOS increments // _num_finished_scanners. // Completed scan tasks whose cached_block is ready for the operator to consume. - // Protected by _transfer_lock. Written by push_back_scan_task() (scanner thread), + // Protected by _transfer_lock. Written by push_completed_scan_task() (scanner thread), // read/popped by get_block_from_queue() (operator thread). std::list<std::shared_ptr<ScanTask>> _completed_tasks; - // Scanners waiting to be submitted to the scheduler thread pool. Stored as a stack - // (LIFO) so that recently-used scanners are re-scheduled first, which is more likely - // to be cache-friendly. Protected by _transfer_lock. Populated in the constructor - // and by schedule_scan_task() when the concurrency limit is reached; drained by - // _pull_next_scan_task() during scheduling. + // Scanners waiting to be admitted for execution. Stored as a stack (LIFO) so that + // recently-used scanners are re-scheduled first, which is more likely to be cache-friendly. + // Protected by _transfer_lock. Populated in the constructor and when an operator returns a + // non-EOS task; drained by try_get_next_scan_task() or the TaskExecutor scheduler. std::stack<std::shared_ptr<ScanTask>> _pending_tasks; + // True only while one runnable for this context is waiting in the thread pool. It does not + // describe scanners currently executing on worker threads. This deduplicates Context + // submission: N pending scanners still create exactly one runnable. Protected by _transfer_lock. + bool _is_context_queued = false; + + // Start time for one queued Context runnable. This is accounted to the scan operator profile + // when the thread-pool worker dequeues the runnable, so the profile reports actual thread-pool + // queue latency even though a worker may select any pending scanner. Protected by _transfer_lock. + int64_t _context_wait_worker_start_ns = 0; + RuntimeProfile::Counter* _context_wait_worker_timer = nullptr; + // Number of scan tasks currently submitted to the scanner scheduler thread pool - // (i.e. in-flight). Incremented by submit_scan_task() before submission and - // decremented by push_back_scan_task() when the thread pool returns the task. + // (i.e. in-flight). Incremented before a task is submitted or directly admitted for + // thread-pool execution, and decremented by push_completed_scan_task() when the worker + // returns it. // Declared atomic so it can be read without _transfer_lock in non-critical paths, // but must be read under _transfer_lock whenever combined with _completed_tasks.size() // to form a consistent concurrency snapshot. diff --git a/be/src/exec/scan/scanner_scheduler.cpp b/be/src/exec/scan/scanner_scheduler.cpp index 230a54048a2..e23603ff923 100644 --- a/be/src/exec/scan/scanner_scheduler.cpp +++ b/be/src/exec/scan/scanner_scheduler.cpp @@ -75,17 +75,7 @@ Status ScannerScheduler::submit(std::shared_ptr<ScannerContext> ctx, TabletStorageType type = scanner_delegate->_scanner->get_storage_type(); auto sumbit_task = [&]() { auto work_func = [scanner_ref = scan_task, ctx]() { - auto status = [&] { - RETURN_IF_CATCH_EXCEPTION(_scanner_scan(ctx, scanner_ref)); - return Status::OK(); - }(); - - if (!status.ok()) { - scanner_ref->set_status(status); - ctx->push_back_scan_task(scanner_ref); - return true; - } - return scanner_ref->is_eos(); + return execute_scan_task(ctx, scanner_ref); }; SimplifiedScanTask simple_scan_task = {work_func, ctx, scan_task}; return this->submit_scan_task(simple_scan_task); @@ -104,6 +94,22 @@ Status ScannerScheduler::submit(std::shared_ptr<ScannerContext> ctx, return Status::OK(); } +bool ScannerScheduler::execute_scan_task(const std::shared_ptr<ScannerContext>& ctx, + const std::shared_ptr<ScanTask>& scan_task) { + // Both schedulers admit tasks differently, but exceptions must always become a completed task + // so the operator observes the error and releases the task's in-flight concurrency slot. + auto status = [&] { + RETURN_IF_CATCH_EXCEPTION(_scanner_scan(ctx, scan_task)); + return Status::OK(); + }(); + if (!status.ok()) { + scan_task->set_status(status); + ctx->push_completed_scan_task(scan_task); + return true; + } + return scan_task->is_eos(); +} + void handle_reserve_memory_failure(RuntimeState* state, std::shared_ptr<ScannerContext> ctx, const Status& st, size_t reserve_size) { ctx->clear_free_blocks(); @@ -179,6 +185,10 @@ 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; } // scanner->open may alloc plenty amount of memory(read blocks of data), // so better to also check low memory and clear free blocks here. if (ctx->low_memory_mode()) { ctx->clear_free_blocks(); } @@ -308,7 +318,7 @@ void ScannerScheduler::_scanner_scan(std::shared_ptr<ScannerContext> ctx, "{}, eos: {}, status: {}", ctx->ctx_id, ctx->num_scheduled_scanners(), eos, status.to_string()); - ctx->push_back_scan_task(scan_task); + ctx->push_completed_scan_task(scan_task); } // NOLINTEND(readability-function-cognitive-complexity,readability-function-size) diff --git a/be/src/exec/scan/scanner_scheduler.h b/be/src/exec/scan/scanner_scheduler.h index fa5387d2736..273954274d4 100644 --- a/be/src/exec/scan/scanner_scheduler.h +++ b/be/src/exec/scan/scanner_scheduler.h @@ -138,7 +138,11 @@ public: protected: int _min_active_scan_threads; -private: + // Execute one admitted task for both scheduler implementations. The return value is consumed + // by TaskExecutor to distinguish terminal EOS/error tasks from scanners that remain runnable. + static bool execute_scan_task(const std::shared_ptr<ScannerContext>& ctx, + const std::shared_ptr<ScanTask>& scan_task); + static void _scanner_scan(std::shared_ptr<ScannerContext> ctx, std::shared_ptr<ScanTask> scan_task); @@ -240,12 +244,13 @@ public: std::unique_lock<std::mutex>& transfer_lock) override; private: + void _run_context(std::shared_ptr<ScannerContext> scanner_ctx); + std::unique_ptr<ThreadPool> _scan_thread_pool; std::atomic<bool> _is_stop; std::weak_ptr<CgroupCpuCtl> _cgroup_cpu_ctl; std::string _sched_name; std::string _workload_group; - std::shared_mutex _lock; }; class TaskExecutorSimplifiedScanScheduler final : public ScannerScheduler { diff --git a/be/src/exec/scan/simplified_scan_scheduler.cpp b/be/src/exec/scan/simplified_scan_scheduler.cpp index 275461ff1dd..aeb8583ef18 100644 --- a/be/src/exec/scan/simplified_scan_scheduler.cpp +++ b/be/src/exec/scan/simplified_scan_scheduler.cpp @@ -17,6 +17,8 @@ #include <memory> +#include "common/exception.h" +#include "common/logging.h" #include "exec/scan/scanner_context.h" #include "exec/scan/scanner_scheduler.h" @@ -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. + execute_scan_task(scanner_ctx, scan_task); } } // namespace doris diff --git a/be/test/exec/scan/scanner_context_test.cpp b/be/test/exec/scan/scanner_context_test.cpp index 0e2aa0a9bde..1928062abbd 100644 --- a/be/test/exec/scan/scanner_context_test.cpp +++ b/be/test/exec/scan/scanner_context_test.cpp @@ -171,7 +171,10 @@ TEST_F(ScannerContextTest, test_init) { state->set_query_options(query_options); std::unique_ptr<MockSimplifiedScanScheduler> scheduler = std::make_unique<MockSimplifiedScanScheduler>(cgroup_cpu_ctl); + // init() is invoked twice below, and each invocation performs one initial scheduling attempt. + // Keep this expectation explicit so changing bootstrap scheduling updates this test too. EXPECT_CALL(*scheduler, schedule_scan_task(testing::_, testing::_, testing::_)) + .Times(2) .WillRepeatedly(testing::Return(Status::OK())); scanner_context->_scanner_scheduler = scheduler.get(); @@ -458,7 +461,7 @@ TEST_F(ScannerContextTest, test_max_column_reader_num) { ASSERT_EQ(scanner_context->_max_scan_concurrency, 1); } -TEST_F(ScannerContextTest, test_push_back_scan_task) { +TEST_F(ScannerContextTest, test_push_completed_scan_task) { const int parallel_tasks = 1; auto scan_operator = std::make_unique<OlapScanOperatorX>(obj_pool.get(), tnode, 0, *descs, parallel_tasks, TQueryCacheParam {}); @@ -491,7 +494,7 @@ TEST_F(ScannerContextTest, test_push_back_scan_task) { for (int i = 0; i < 5; ++i) { auto scan_task = std::make_shared<ScanTask>(std::make_shared<ScannerDelegate>(scanner)); - scanner_context->push_back_scan_task(scan_task); + scanner_context->push_completed_scan_task(scan_task); ASSERT_EQ(scanner_context->_in_flight_tasks_num, 10 - i); } } @@ -669,6 +672,35 @@ TEST_F(ScannerContextTest, pull_next_scan_task) { pull_scan_task = scanner_context->_pull_next_scan_task( nullptr, scanner_context->_max_scan_concurrency - 1); EXPECT_NE(pull_scan_task, nullptr); + + std::unique_lock<std::mutex> context_transfer_lock(scanner_context->transfer_lock()); + scanner_context->_pending_tasks = std::stack<std::shared_ptr<ScanTask>>(); + scanner_context->_completed_tasks.clear(); + scanner_context->_in_flight_tasks_num = 0; + // Even if the effective limit is temporarily zero, one pending task must run so it can publish + // a block or EOS and prevent the Context from stalling. + scanner_context->_max_scan_concurrency = 0; + + auto completed_task = std::make_shared<ScanTask>(std::make_shared<ScannerDelegate>(scanner)); + completed_task->set_state(ScanTask::State::IN_FLIGHT); + completed_task->cached_block = Block::create_unique(); + completed_task->set_state(ScanTask::State::COMPLETED); + completed_task->cached_block.reset(); + // A consumed non-EOS result must be eligible for another Context admission. This also covers + // the COMPLETED -> PENDING transition used by ThreadPool scheduling. + scanner_context->push_pending_scan_task(completed_task, context_transfer_lock); + + EXPECT_FALSE(scanner_context->is_context_queued(context_transfer_lock)); + scanner_context->set_context_queued(true, context_transfer_lock); + EXPECT_TRUE(scanner_context->is_context_queued(context_transfer_lock)); + scanner_context->set_context_queued(false, context_transfer_lock); + + // The Context can admit exactly one scanner at its configured concurrency limit. + auto admitted_task = scanner_context->try_get_next_scan_task(context_transfer_lock); + EXPECT_EQ(admitted_task, completed_task); + EXPECT_EQ(admitted_task->_state, ScanTask::State::IN_FLIGHT); + EXPECT_EQ(scanner_context->_in_flight_tasks_num, 1); + EXPECT_EQ(scanner_context->try_get_next_scan_task(context_transfer_lock), nullptr); } TEST_F(ScannerContextTest, schedule_scan_task) { --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
