This is an automated email from the ASF dual-hosted git repository. kxiao pushed a commit to branch branch-2.0 in repository https://gitbox.apache.org/repos/asf/doris.git
commit 54795289dcd88e7cf08a86dc344ba4c4e94bf0f1 Author: Pxl <[email protected]> AuthorDate: Tue Jul 18 11:55:21 2023 +0800 [Bug](pipeline) adjust scanner scheduler.submit and _num_scheduling_ctx maintain (#21843) adjust scanner scheduler.submit and _num_scheduling_ctx maintain --- be/src/olap/tablet_schema_cache.cpp | 1 - be/src/vec/exec/scan/pip_scanner_context.h | 2 +- be/src/vec/exec/scan/scanner_context.cpp | 38 ++++++++++++++++++++---------- be/src/vec/exec/scan/scanner_context.h | 11 ++++----- be/src/vec/exec/scan/scanner_scheduler.cpp | 6 +++-- be/src/vec/exec/scan/scanner_scheduler.h | 2 +- 6 files changed, 36 insertions(+), 24 deletions(-) diff --git a/be/src/olap/tablet_schema_cache.cpp b/be/src/olap/tablet_schema_cache.cpp index 24aa72a995..ee14358495 100644 --- a/be/src/olap/tablet_schema_cache.cpp +++ b/be/src/olap/tablet_schema_cache.cpp @@ -42,7 +42,6 @@ void TabletSchemaCache::stop() { while (!_is_stopped) { std::this_thread::sleep_for(std::chrono::seconds(1)); } - LOG(INFO) << "xxx stopped"; } /** diff --git a/be/src/vec/exec/scan/pip_scanner_context.h b/be/src/vec/exec/scan/pip_scanner_context.h index 731c3bb926..c720c22d04 100644 --- a/be/src/vec/exec/scan/pip_scanner_context.h +++ b/be/src/vec/exec/scan/pip_scanner_context.h @@ -45,7 +45,7 @@ public: { std::unique_lock l(_transfer_lock); if (state->is_cancelled()) { - _process_status = Status::Cancelled("cancelled"); + set_status_on_error(Status::Cancelled("cancelled"), false); } if (!_process_status.ok()) { diff --git a/be/src/vec/exec/scan/scanner_context.cpp b/be/src/vec/exec/scan/scanner_context.cpp index 7bd98f3071..12111e9cec 100644 --- a/be/src/vec/exec/scan/scanner_context.cpp +++ b/be/src/vec/exec/scan/scanner_context.cpp @@ -28,6 +28,7 @@ #include <utility> #include "common/config.h" +#include "common/status.h" #include "runtime/descriptors.h" #include "runtime/exec_env.h" #include "runtime/query_context.h" @@ -188,8 +189,12 @@ Status ScannerContext::get_block_from_queue(RuntimeState* state, vectorized::Blo // At this point, consumers are required to trigger new scheduling to ensure that // data can be continuously fetched. if (has_enough_space_in_blocks_queue() && _num_running_scanners == 0) { - _num_scheduling_ctx++; - _scanner_scheduler->submit(this); + auto state = _scanner_scheduler->submit(this); + if (state.ok()) { + _num_scheduling_ctx++; + } else { + set_status_on_error(state, false); + } } // Wait for block from queue if (wait) { @@ -201,11 +206,11 @@ Status ScannerContext::get_block_from_queue(RuntimeState* state, vectorized::Blo } if (state->is_cancelled()) { - _process_status = Status::Cancelled("cancelled"); + set_status_on_error(Status::Cancelled("cancelled"), false); } - if (!_process_status.ok()) { - return _process_status; + if (!status().ok()) { + return status(); } if (!_blocks_queue.empty()) { @@ -221,12 +226,16 @@ Status ScannerContext::get_block_from_queue(RuntimeState* state, vectorized::Blo return Status::OK(); } -bool ScannerContext::set_status_on_error(const Status& status) { - std::lock_guard l(_transfer_lock); +bool ScannerContext::set_status_on_error(const Status& status, bool need_lock) { + std::unique_lock l(_transfer_lock, std::defer_lock); + if (need_lock) { + l.lock(); + } if (_process_status.ok()) { _process_status = status; _status_error = true; _blocks_queue_added_cv.notify_one(); + _should_stop = true; return true; } return false; @@ -313,10 +322,12 @@ std::string ScannerContext::debug_string() { void ScannerContext::reschedule_scanner_ctx() { std::lock_guard l(_transfer_lock); - auto submit_st = _scanner_scheduler->submit(this); + auto state = _scanner_scheduler->submit(this); //todo(wb) rethinking is it better to mark current scan_context failed when submit failed many times? - if (submit_st.ok()) { + if (state.ok()) { _num_scheduling_ctx++; + } else { + set_status_on_error(state, false); } } @@ -327,10 +338,11 @@ void ScannerContext::push_back_scanner_and_reschedule(VScannerSPtr scanner) { } std::lock_guard l(_transfer_lock); if (has_enough_space_in_blocks_queue()) { - _num_scheduling_ctx++; - auto submit_st = _scanner_scheduler->submit(this); - if (!submit_st.ok()) { - _num_scheduling_ctx--; + auto state = _scanner_scheduler->submit(this); + if (state.ok()) { + _num_scheduling_ctx++; + } else { + set_status_on_error(state, false); } } diff --git a/be/src/vec/exec/scan/scanner_context.h b/be/src/vec/exec/scan/scanner_context.h index 4d1fa251fa..c93c016613 100644 --- a/be/src/vec/exec/scan/scanner_context.h +++ b/be/src/vec/exec/scan/scanner_context.h @@ -82,10 +82,12 @@ public: // to return the scanner to the list for next scheduling. void push_back_scanner_and_reschedule(VScannerSPtr scanner); - bool set_status_on_error(const Status& status); + bool set_status_on_error(const Status& status, bool need_lock = true); Status status() { - std::lock_guard l(_transfer_lock); + if (_process_status.is<ErrorCode::END_OF_FILE>()) { + return Status::OK(); + } return _process_status; } @@ -98,10 +100,7 @@ public: } // Return true if this ScannerContext need no more process - virtual bool done() { - std::unique_lock l(_transfer_lock); - return _is_finished || _should_stop || !_process_status.ok(); - } + virtual bool done() { return _is_finished || _should_stop; } // Update the running num of scanners and contexts void update_num_running(int32_t scanner_inc, int32_t sched_inc) { diff --git a/be/src/vec/exec/scan/scanner_scheduler.cpp b/be/src/vec/exec/scan/scanner_scheduler.cpp index 91bed3ee2e..da240fa644 100644 --- a/be/src/vec/exec/scan/scanner_scheduler.cpp +++ b/be/src/vec/exec/scan/scanner_scheduler.cpp @@ -50,7 +50,7 @@ namespace doris::vectorized { -ScannerScheduler::ScannerScheduler() {} +ScannerScheduler::ScannerScheduler() = default; ScannerScheduler::~ScannerScheduler() { if (!_is_init) { @@ -114,6 +114,9 @@ Status ScannerScheduler::init(ExecEnv* env) { } Status ScannerScheduler::submit(ScannerContext* ctx) { + if (ctx->done()) { + return Status::EndOfFile("ScannerContext is done"); + } if (ctx->queue_idx == -1) { ctx->queue_idx = (_queue_idx++ % QUEUE_NUM); } @@ -142,7 +145,6 @@ void ScannerScheduler::_schedule_thread(int queue_id) { // If ctx is done, no need to schedule it again. // But should notice that there may still scanners running in scanner pool. } - return; } [[maybe_unused]] static void* run_scanner_bthread(void* arg) { diff --git a/be/src/vec/exec/scan/scanner_scheduler.h b/be/src/vec/exec/scan/scanner_scheduler.h index d8f51d4ad0..630e3a3803 100644 --- a/be/src/vec/exec/scan/scanner_scheduler.h +++ b/be/src/vec/exec/scan/scanner_scheduler.h @@ -63,7 +63,7 @@ public: Status init(ExecEnv* env); - Status submit(ScannerContext* ctx); + [[nodiscard]] Status submit(ScannerContext* ctx); std::unique_ptr<ThreadPoolToken> new_limited_scan_pool_token(ThreadPool::ExecutionMode mode, int max_concurrency); --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
