This is an automated email from the ASF dual-hosted git repository. morningman pushed a commit to branch dev-1.0.1 in repository https://gitbox.apache.org/repos/asf/doris.git
commit 9a28b26f723f8a8aa6176c00b7b7febe9616d51d Author: jacktengg <[email protected]> AuthorDate: Wed May 18 07:40:57 2022 +0800 [Improvement][ASAN] make BE can exit normally and ASAN memory leak checking work (#9620) --- be/src/util/priority_thread_pool.hpp | 20 ++++++++++---------- be/src/util/priority_work_stealing_thread_pool.hpp | 19 +++++-------------- 2 files changed, 15 insertions(+), 24 deletions(-) diff --git a/be/src/util/priority_thread_pool.hpp b/be/src/util/priority_thread_pool.hpp index ed0fe90157..db5700137e 100644 --- a/be/src/util/priority_thread_pool.hpp +++ b/be/src/util/priority_thread_pool.hpp @@ -53,7 +53,6 @@ public: // -- 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) { @@ -117,6 +116,16 @@ public: } 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. @@ -136,17 +145,8 @@ private: // 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 3c86098034..c11ee7d198 100644 --- a/be/src/util/priority_work_stealing_thread_pool.hpp +++ b/be/src/util/priority_work_stealing_thread_pool.hpp @@ -36,7 +36,6 @@ public: // -- 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); @@ -51,6 +50,11 @@ public: } } + 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. // @@ -82,10 +86,6 @@ public: } } - // 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 @@ private: // 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 --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
