added priority queue

contrib/boost

@@ -1 +1 @@
-Subproject commit 4b98e2befd3f3265b0db0acb5d20c4812ef8d88e
+Subproject commit 66d17f060c4867aeea99fa2a20cfdae89ae2a2ec

contrib/boost-cmake/CMakeLists.txt

@@ -15,11 +15,12 @@ if (NOT USE_INTERNAL_BOOST_LIBRARY)
         coroutine
         graph
         circular_buffer
+        heap
     )
 
     if(Boost_INCLUDE_DIR AND Boost_FILESYSTEM_LIBRARY AND
         Boost_PROGRAM_OPTIONS_LIBRARY AND Boost_REGEX_LIBRARY AND Boost_SYSTEM_LIBRARY AND Boost_CONTEXT_LIBRARY AND
-        Boost_COROUTINE_LIBRARY AND Boost_GRAPH_LIBRARY AND Boost_CIRCULAR_BUFFER_LIBRARY)
+        Boost_COROUTINE_LIBRARY AND Boost_GRAPH_LIBRARY AND Boost_CIRCULAR_BUFFER_LIBRARY AND Boost_HEAP_LIBRARY)
 
         set(EXTERNAL_BOOST_FOUND 1)
 
@@ -242,9 +243,14 @@ if (NOT EXTERNAL_BOOST_FOUND)
     target_include_directories (_boost_graph PRIVATE ${LIBRARY_DIR})
     target_link_libraries(_boost_graph PRIVATE _boost_regex)
 
-
+    # circular buffer
     add_library(_boost_circular_buffer INTERFACE)
     add_library(boost::circular_buffer ALIAS _boost_circular_buffer)
     target_include_directories(_boost_circular_buffer SYSTEM BEFORE INTERFACE ${LIBRARY_DIR})
 
+    # heap
+    add_library(_boost_heap INTERFACE)
+    add_library(boost::heap ALIAS _boost_heap)
+    target_include_directories(_boost_heap SYSTEM BEFORE INTERFACE ${LIBRARY_DIR})
+
 endif ()

src/CMakeLists.txt

@@ -297,7 +297,6 @@ target_link_libraries(clickhouse_common_io
         PUBLIC
             boost::program_options
             boost::system
-            boost::circular_buffer
             ${CITYHASH_LIBRARIES}
             ${ZLIB_LIBRARIES}
             pcg_random
@@ -337,6 +336,7 @@ dbms_target_link_libraries (
         boost::filesystem
         boost::program_options
         boost::circular_buffer
+        boost::heap
         clickhouse_common_config
         clickhouse_common_zookeeper
         clickhouse_dictionaries_embedded

src/Common/PriorityQueue.h

@@ -1,53 +0,0 @@
-#pragma once
-
-#include <queue>
-
-namespace DB
-{
-
-
-template <class T, class Comparator = std::less<T>>
-class PriorityQueue
-{
-public:
-
-    T pop()
-    {
-        assert(!buffer.empty());
-        std::pop_heap(buffer.begin(), buffer.end(), comparator);
-        auto element = std::move(buffer.back());
-        buffer.pop_back();
-        return element;
-    }
-
-    void push(T element)
-    {
-        buffer.push_back(std::move(element));
-        std::push_heap(buffer.begin(), buffer.end(), comparator);
-    }
-
-    template< class... Args >
-    void emplace(Args &&... args)
-    {
-        buffer.emplace_back(std::forward<Args>(args)...);
-        std::push_heap(buffer.begin(), buffer.end(), comparator);
-    }
-
-    bool empty() { return buffer.empty(); }
-    size_t size() { return buffer.size(); }
-    void reserve(size_t count) { buffer.reserve(count); }
-    void resize(size_t count)
-    {
-        buffer.resize(count);
-        std::make_heap(buffer.begin(), buffer.end(), comparator);
-    }
-
-private:
-
-    Comparator comparator;
-    std::vector<T> buffer;
-
-
-};
-
-}

src/Common/ThreadPool.cpp

@@ -123,7 +123,6 @@ ReturnType ThreadPoolImpl<Thread>::scheduleImpl(Job job, int priority, std::opti
         /// Check if there are enough threads to process job.
         if (threads.size() < std::min(max_threads, scheduled_jobs + 1))
         {
-            ALLOW_ALLOCATIONS_IN_SCOPE;
             try
             {
                 threads.emplace_front();
@@ -249,9 +248,16 @@ void ThreadPoolImpl<Thread>::worker(typename std::list<Thread>::iterator thread_
             need_shutdown = shutdown;
 
             if (!jobs.empty())
-                job = std::move(jobs.pop().job);
+            {
+                job = std::move(jobs.top().job);
+                jobs.pop();
+            }
             else
-                return; /// shutdown is true, simply finish the thread.
+            {
+                /// shutdown is true, simply finish the thread.
+                return;
+            }
+
         }
 
         if (!need_shutdown)

src/Common/ThreadPool.h

@@ -9,6 +9,8 @@
 #include <list>
 #include <optional>
 
+#include <boost/heap/priority_queue.hpp>
+
 #include <Poco/Event.h>
 #include <Common/ThreadStatus.h>
 #include <Common/PriorityQueue.h>
@@ -104,7 +106,7 @@ private:
         }
     };
 
-    DB::PriorityQueue<JobWithPriority> jobs;
+    boost::heap::priority_queue<JobWithPriority> jobs;
     std::list<Thread> threads;
     std::exception_ptr first_exception;
 

src/Common/tests/gtest_priority_queue.cpp

@@ -1,25 +0,0 @@
-#include <gtest/gtest.h>
-
-#include <random>
-
-#include <Common/PriorityQueue.h>
-
-using namespace DB;
-
-TEST(PriorityQueue, Simple)
-{
-    PriorityQueue<int> my;
-    std::priority_queue<int> original;
-
-    for (int i = 0; i < 1000; ++i)
-    {
-        my.push(i);
-        original.emplace(i);
-    }
-
-    for (int i = 0; i < 1000; ++i)
-    {
-        ASSERT_EQ(my.pop(), original.top());
-        original.pop();
-    }
-}

src/Storages/MergeTree/MergeMutateExecutor.cpp

@@ -207,8 +207,11 @@ void MergeTreeBackgroundExecutor::schedulerThreadFunction()
         ItemPtr item = std::move(pending.front());
         pending.pop_front();
 
+        bool res = false;
+        {
+            ALLOW_ALLOCATIONS_IN_SCOPE;
             /// Execute a piece of task
-        bool res = pool.trySchedule([this, item]
+            res = pool.trySchedule([this, item]
             {
                 routine(item);
                 /// When storage shutdowns it will wait until all related background tasks
@@ -216,6 +219,7 @@ void MergeTreeBackgroundExecutor::schedulerThreadFunction()
                 /// and this will cause segfault.
                 item->is_done.set();
             });
+        }
 
         if (!res)
         {