From f92172c7b5ab0ae4c89dd034871400bd78e41755 Mon Sep 17 00:00:00 2001
From: jacktengg <tengjp@outlook.com>
Date: Tue, 17 May 2022 20:02:36 +0800
Subject: [PATCH] [Improvement] make BE can exit normally and ASAN memory leak
 checking work properly

---
 be/src/runtime/minidump.cpp                   |  2 +-
 be/src/util/priority_thread_pool.hpp          | 19 +++++++++----------
 .../priority_work_stealing_thread_pool.hpp    | 19 +++++--------------
 3 files changed, 15 insertions(+), 25 deletions(-)

diff --git a/be/src/runtime/minidump.cpp b/be/src/runtime/minidump.cpp
index cf1f0611f66c33..1c66771403a347 100644
--- a/be/src/runtime/minidump.cpp
+++ b/be/src/runtime/minidump.cpp
@@ -104,7 +104,7 @@ bool Minidump::_minidump_cb(const google_breakpad::MinidumpDescriptor& descripto
 }
 
 void Minidump::stop() {
-    if (_stop) {
+    if (config::disable_minidump || _stop) {
         return;
     }
     _stop = true;
diff --git a/be/src/util/priority_thread_pool.hpp b/be/src/util/priority_thread_pool.hpp
index 20d90b3f41d3a9..616cb2475f2b84 100644
--- a/be/src/util/priority_thread_pool.hpp
+++ b/be/src/util/priority_thread_pool.hpp
@@ -52,7 +52,6 @@ class PriorityThreadPool {
     //  -- queue_size: the maximum size of the queue on which work items are offered. If the
     //     queue exceeds this size, subsequent calls to Offer will block until there is
     //     capacity available.
-    //  -- work_function: the function to run every time an item is consumed from the queue
     PriorityThreadPool(uint32_t num_threads, uint32_t queue_size)
             : _work_queue(queue_size), _shutdown(false) {
         for (int i = 0; i < num_threads; ++i) {
@@ -118,6 +117,15 @@ class PriorityThreadPool {
 protected:
     virtual bool is_shutdown() { return _shutdown; }
 
+    // Collection of worker threads that process work from the queue.
+    ThreadGroup _threads;
+
+    // Guards _empty_cv
+    std::mutex _lock;
+
+    // Signalled when the queue becomes empty
+    std::condition_variable _empty_cv;
+
 private:
     // Driver method for each thread in the pool. Continues to read work from the queue
     // until the pool is shutdown.
@@ -137,17 +145,8 @@ class PriorityThreadPool {
     // FIFO order.
     BlockingPriorityQueue<Task> _work_queue;
 
-    // Collection of worker threads that process work from the queue.
-    ThreadGroup _threads;
-
-    // Guards _empty_cv
-    std::mutex _lock;
-
     // Set to true when threads should stop doing work and terminate.
     std::atomic<bool> _shutdown;
-
-    // Signalled when the queue becomes empty
-    std::condition_variable _empty_cv;
 };
 
 } // namespace doris
diff --git a/be/src/util/priority_work_stealing_thread_pool.hpp b/be/src/util/priority_work_stealing_thread_pool.hpp
index c2f717a18fd08c..937012722f4fe8 100644
--- a/be/src/util/priority_work_stealing_thread_pool.hpp
+++ b/be/src/util/priority_work_stealing_thread_pool.hpp
@@ -35,7 +35,6 @@ class PriorityWorkStealingThreadPool : public PriorityThreadPool {
     //  -- queue_size: the maximum size of the queue on which work items are offered. If the
     //     queue exceeds this size, subsequent calls to Offer will block until there is
     //     capacity available.
-    //  -- work_function: the function to run every time an item is consumed from the queue
     PriorityWorkStealingThreadPool(uint32_t num_threads, uint32_t num_queues, uint32_t queue_size)
             : PriorityThreadPool(0, 0) {
         DCHECK_GT(num_queues, 0);
@@ -50,6 +49,11 @@ class PriorityWorkStealingThreadPool : public PriorityThreadPool {
         }
     }
 
+    virtual ~PriorityWorkStealingThreadPool() {
+        shutdown();
+        join();
+    }
+
     // Blocking operation that puts a work item on the queue. If the queue is full, blocks
     // until there is capacity available.
     //
@@ -79,10 +83,6 @@ class PriorityWorkStealingThreadPool : public PriorityThreadPool {
         }
     }
 
-    // Blocks until all threads are finished. shutdown does not need to have been called,
-    // since it may be called on a separate thread.
-    void join() override { _threads.join_all(); }
-
     uint32_t get_queue_size() const override {
         uint32_t size = 0;
         for (auto work_queue : _work_queues) {
@@ -141,15 +141,6 @@ class PriorityWorkStealingThreadPool : public PriorityThreadPool {
     // Queue on which work items are held until a thread is available to process them in
     // FIFO order.
     std::vector<std::shared_ptr<BlockingPriorityQueue<Task>>> _work_queues;
-
-    // Collection of worker threads that process work from the queues.
-    ThreadGroup _threads;
-
-    // Guards _empty_cv
-    std::mutex _lock;
-
-    // Signalled when the queue becomes empty
-    std::condition_variable _empty_cv;
 };
 
 } // namespace doris
\ No newline at end of file