This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new bc38b2fdfb [improvement](new-scan) graceful quit scanner scheduler
(#12715)
bc38b2fdfb is described below
commit bc38b2fdfb2435f92f732a94f457c1881da3ba77
Author: Mingyu Chen <[email protected]>
AuthorDate: Mon Sep 19 08:39:08 2022 +0800
[improvement](new-scan) graceful quit scanner scheduler (#12715)
---
be/src/vec/exec/scan/scanner_scheduler.cpp | 29 ++++++++++++++++++++++++-----
be/src/vec/exec/scan/scanner_scheduler.h | 14 +++-----------
2 files changed, 27 insertions(+), 16 deletions(-)
diff --git a/be/src/vec/exec/scan/scanner_scheduler.cpp
b/be/src/vec/exec/scan/scanner_scheduler.cpp
index 7108143967..9787ecae5d 100644
--- a/be/src/vec/exec/scan/scanner_scheduler.cpp
+++ b/be/src/vec/exec/scan/scanner_scheduler.cpp
@@ -32,11 +32,28 @@ namespace doris::vectorized {
ScannerScheduler::ScannerScheduler() {}
ScannerScheduler::~ScannerScheduler() {
+ if (!_is_init) {
+ return;
+ }
+
+ for (int i = 0; i < QUEUE_NUM; i++) {
+ _pending_queues[i]->shutdown();
+ }
+
_is_closed = true;
+
_scheduler_pool->shutdown();
_local_scan_thread_pool->shutdown();
_remote_scan_thread_pool->shutdown();
- // TODO: safely delete all objects and graceful exit
+
+ _scheduler_pool->wait();
+ _local_scan_thread_pool->join();
+ _remote_scan_thread_pool->join();
+
+ for (int i = 0; i < QUEUE_NUM; i++) {
+ delete _pending_queues[i];
+ }
+ delete[] _pending_queues;
}
Status ScannerScheduler::init(ExecEnv* env) {
@@ -53,14 +70,16 @@ Status ScannerScheduler::init(ExecEnv* env) {
}
// 2. local scan thread pool
- _local_scan_thread_pool = new PriorityWorkStealingThreadPool(
+ _local_scan_thread_pool.reset(new PriorityWorkStealingThreadPool(
config::doris_scanner_thread_pool_thread_num,
env->store_paths().size(),
- config::doris_scanner_thread_pool_queue_size);
+ config::doris_scanner_thread_pool_queue_size));
// 3. remote scan thread pool
- _remote_scan_thread_pool = new
PriorityThreadPool(config::doris_scanner_thread_pool_thread_num,
-
config::doris_scanner_thread_pool_queue_size);
+ _remote_scan_thread_pool.reset(
+ new
PriorityThreadPool(config::doris_scanner_thread_pool_thread_num,
+
config::doris_scanner_thread_pool_queue_size));
+ _is_init = true;
return Status::OK();
}
diff --git a/be/src/vec/exec/scan/scanner_scheduler.h
b/be/src/vec/exec/scan/scanner_scheduler.h
index a72fd5021e..f8c1a8f3df 100644
--- a/be/src/vec/exec/scan/scanner_scheduler.h
+++ b/be/src/vec/exec/scan/scanner_scheduler.h
@@ -40,13 +40,6 @@ namespace doris::vectorized {
// Each Scanner will act as a producer, read a group of blocks and put
them into
// the corresponding block queue.
// The corresponding ScanNode will act as a consumer to consume blocks
from the block queue.
-
-using ContextMap = phmap::parallel_flat_hash_map<
- std::string, std::shared_ptr<ScannerContext>,
phmap::priv::hash_default_hash<std::string>,
- phmap::priv::hash_default_eq<std::string>,
- std::allocator<std::pair<const std::string,
std::shared_ptr<ScannerContext>>>, 12,
- std::mutex>;
-
class Env;
class ScannerScheduler {
public:
@@ -82,13 +75,12 @@ private:
// execution thread pool
// _local_scan_thread_pool is for local scan task(typically, olap scanner)
// _remote_scan_thread_pool is for remote scan task(cold data on s3, hdfs,
etc.)
- PriorityThreadPool* _local_scan_thread_pool;
- PriorityThreadPool* _remote_scan_thread_pool;
+ std::unique_ptr<PriorityThreadPool> _local_scan_thread_pool;
+ std::unique_ptr<PriorityThreadPool> _remote_scan_thread_pool;
// true is the scheduler is closed.
std::atomic_bool _is_closed = {false};
-
- ContextMap _context_map;
+ bool _is_init = false;
};
} // namespace doris::vectorized
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
