Merge branch 'master' into better-union-all-try2

dbms/src/Client/MultiplexedConnections.h

@@ -3,7 +3,6 @@
 #include <Common/Throttler.h>
 #include <Client/Connection.h>
 #include <Client/ConnectionPoolWithFailover.h>
-#include <Poco/ScopedLock.h>
 #include <mutex>
 
 namespace DB

dbms/src/Common/CounterInFile.h

@@ -9,7 +9,6 @@
 #include <Poco/File.h>
 #include <Poco/Exception.h>
 #include <mutex>
-#include <Poco/ScopedLock.h>
 
 #include <Common/Exception.h>
 #include <IO/ReadBufferFromFileDescriptor.h>

dbms/src/Common/Stopwatch.h

@@ -1,14 +1,25 @@
 #pragma once
 
 #include <time.h>
-#include <mutex>
-#include <Poco/ScopedLock.h>
+#include <atomic>
 #include <common/Types.h>
 
 #ifdef __APPLE__
 #include <common/apple_rt.h>
 #endif
 
+
+namespace StopWatchDetail
+{
+    inline UInt64 nanoseconds(clockid_t clock_type)
+    {
+        struct timespec ts;
+        clock_gettime(clock_type, &ts);
+        return ts.tv_sec * 1000000000ULL + ts.tv_nsec;
+    }
+}
+
+
 /** Differs from Poco::Stopwatch only by using 'clock_gettime' instead of 'gettimeofday',
   *  returns nanoseconds instead of microseconds, and also by other minor differencies.
   */
@@ -18,77 +29,64 @@ public:
     /** CLOCK_MONOTONIC works relatively efficient (~15 million calls/sec) and doesn't lead to syscall.
       * Pass CLOCK_MONOTONIC_COARSE, if you need better performance with acceptable cost of several milliseconds of inaccuracy.
       */
-    Stopwatch(clockid_t clock_type_ = CLOCK_MONOTONIC) : clock_type(clock_type_) { restart(); }
+    Stopwatch(clockid_t clock_type_ = CLOCK_MONOTONIC) : clock_type(clock_type_) { start(); }
 
-    void start()                        { setStart(); is_running = true; }
-    void stop()                         { updateElapsed(); is_running = false; }
-    void restart()                      { elapsed_ns = 0; start(); }
-    UInt64 elapsed() const              { updateElapsed(); return elapsed_ns; }
-    UInt64 elapsedMilliseconds() const  { updateElapsed(); return elapsed_ns / 1000000UL; }
-    double elapsedSeconds() const       { updateElapsed(); return static_cast<double>(elapsed_ns) / 1000000000ULL; }
+    void start()                        { start_ns = nanoseconds(); is_running = true; }
+    void stop()                         { stop_ns = nanoseconds(); is_running = false; }
+    void restart()                      { start(); }
+    UInt64 elapsed() const              { return is_running ? nanoseconds() - start_ns : stop_ns - start_ns; }
+    UInt64 elapsedMilliseconds() const  { return elapsed() / 1000000UL; }
+    double elapsedSeconds() const       { return static_cast<double>(elapsed()) / 1000000000ULL; }
 
 private:
-    mutable UInt64 start_ns;
-    mutable UInt64 elapsed_ns;
+    UInt64 start_ns;
+    UInt64 stop_ns;
     clockid_t clock_type;
     bool is_running;
 
-    void setStart()
-    {
-        struct timespec ts;
-        clock_gettime(clock_type, &ts);
-        start_ns = ts.tv_sec * 1000000000ULL + ts.tv_nsec;
-    }
-
-    void updateElapsed() const
-    {
-        if (is_running)
-        {
-            struct timespec ts;
-            clock_gettime(clock_type, &ts);
-            UInt64 current_ns = ts.tv_sec * 1000000000ULL + ts.tv_nsec;
-            elapsed_ns += current_ns - start_ns;
-            start_ns = current_ns;
-        }
-    }
+    UInt64 nanoseconds() const { return StopWatchDetail::nanoseconds(clock_type); }
 };
 
 
-class StopwatchWithLock : public Stopwatch
+class AtomicStopwatch
 {
 public:
-    /** If specified amount of time has passed and timer is not locked right now, then restarts timer and returns true.
+    AtomicStopwatch(clockid_t clock_type_ = CLOCK_MONOTONIC) : clock_type(clock_type_) { restart(); }
+
+    void restart()                      { start_ns = nanoseconds(); }
+    UInt64 elapsed() const              { return nanoseconds() - start_ns; }
+    UInt64 elapsedMilliseconds() const  { return elapsed() / 1000000UL; }
+    double elapsedSeconds() const       { return static_cast<double>(elapsed()) / 1000000000ULL; }
+
+    /** If specified amount of time has passed, then restarts timer and returns true.
       * Otherwise returns false.
       * This is done atomically.
       */
-    bool lockTestAndRestart(double seconds)
+    bool compareAndRestart(double seconds)
     {
-        std::unique_lock<std::mutex> lock(mutex, std::defer_lock);
-        if (!lock.try_lock())
-            return false;
+        UInt64 threshold = seconds * 1000000000ULL;
+        UInt64 current_ns = nanoseconds();
+        UInt64 current_start_ns = start_ns;
 
-        if (elapsedSeconds() >= seconds)
+        while (true)
         {
-            restart();
-            return true;
+            if (current_ns < current_start_ns + threshold)
+                return false;
+
+            if (start_ns.compare_exchange_weak(current_start_ns, current_ns))
+                return true;
         }
-        else
-            return false;
     }
 
     struct Lock
     {
-        StopwatchWithLock * parent = nullptr;
-        std::unique_lock<std::mutex> lock;
+        AtomicStopwatch * parent = nullptr;
 
         Lock() {}
 
         operator bool() const { return parent != nullptr; }
 
-        Lock(StopwatchWithLock * parent, std::unique_lock<std::mutex> && lock)
-            : parent(parent), lock(std::move(lock))
-        {
-        }
+        Lock(AtomicStopwatch * parent) : parent(parent) {}
 
         Lock(Lock &&) = default;
 
@@ -105,21 +103,33 @@ public:
       * This is done atomically.
       *
       * Usage:
-      * if (auto lock = timer.lockTestAndRestartAfter(1))
+      * if (auto lock = timer.compareAndRestartDeferred(1))
       *        /// do some work, that must be done in one thread and not more frequently than each second.
       */
-    Lock lockTestAndRestartAfter(double seconds)
+    Lock compareAndRestartDeferred(double seconds)
     {
-        std::unique_lock<std::mutex> lock(mutex, std::defer_lock);
-        if (!lock.try_lock())
-            return {};
+        UInt64 threshold = seconds * 1000000000ULL;
+        UInt64 current_ns = nanoseconds();
+        UInt64 current_start_ns = start_ns;
 
-        if (elapsedSeconds() >= seconds)
-            return Lock(this, std::move(lock));
+        while (true)
+        {
+            if ((current_start_ns & 0x8000000000000000ULL))
+                return {};
 
-        return {};
+            if (current_ns < current_start_ns + threshold)
+                return {};
+
+            if (start_ns.compare_exchange_weak(current_start_ns, current_ns | 0x8000000000000000ULL))
+                return Lock(this);
+        }
     }
 
 private:
-    std::mutex mutex;
+    std::atomic<UInt64> start_ns;
+    std::atomic<bool> lock {false};
+    clockid_t clock_type;
+
+    /// Most significant bit is a lock. When it is set, compareAndRestartDeferred method will return false.
+    UInt64 nanoseconds() const { return StopWatchDetail::nanoseconds(clock_type) & 0x7FFFFFFFFFFFFFFFULL; }
 };

dbms/src/Common/tests/CMakeLists.txt

@@ -68,3 +68,6 @@ target_link_libraries (allocator clickhouse_common_io)
 
 add_executable (cow_columns cow_columns.cpp)
 target_link_libraries (cow_columns clickhouse_common_io)
+
+add_executable (stopwatch stopwatch.cpp)
+target_link_libraries (stopwatch clickhouse_common_io)

dbms/src/Common/tests/parallel_aggregation.cpp

@@ -167,7 +167,7 @@ void aggregate33(Map & local_map, Map & global_map, Mutex & mutex, Source::const
 
         if (inserted && local_map.size() == threshold)
         {
-            Poco::ScopedLock<Mutex> lock(mutex);
+            std::lock_guard<Mutex> lock(mutex);
             for (auto & value_type : local_map)
                 global_map[value_type.first] += value_type.second;
 

dbms/src/Common/tests/stopwatch.cpp

@@ -0,0 +1,40 @@
+#include <vector>
+#include <thread>
+#include <iostream>
+#include <Common/Stopwatch.h>
+
+
+int main(int, char **)
+{
+    static constexpr size_t num_threads = 10;
+    static constexpr size_t num_iterations = 3;
+
+    std::vector<std::thread> threads(num_threads);
+
+    AtomicStopwatch watch;
+    Stopwatch total_watch;
+
+    for (size_t i = 0; i < num_threads; ++i)
+    {
+        threads[i] = std::thread([i, &watch, &total_watch]
+        {
+            size_t iteration = 0;
+            while (iteration < num_iterations)
+            {
+                if (auto lock = watch.compareAndRestartDeferred(1))
+                {
+                    std::cerr << "Thread " << i << ": begin iteration " << iteration << ", elapsed: " << total_watch.elapsedMilliseconds() << " ms.\n";
+                    std::this_thread::sleep_for(std::chrono::milliseconds(500));
+                    std::cerr << "Thread " << i << ": end iteration " << iteration << ", elapsed: " << total_watch.elapsedMilliseconds() << " ms.\n";
+                    ++iteration;
+                }
+                std::this_thread::sleep_for(std::chrono::milliseconds(500));
+            }
+        });
+    }
+
+    for (auto & thread : threads)
+        thread.join();
+
+    return 0;
+}

dbms/src/Databases/DatabaseOrdinary.cpp

@@ -149,7 +149,7 @@ void DatabaseOrdinary::loadTables(
 
     String data_path = context.getPath() + "data/" + escapeForFileName(name) + "/";
 
-    StopwatchWithLock watch;
+    AtomicStopwatch watch;
     std::atomic<size_t> tables_processed {0};
 
     auto task_function = [&](FileNames::const_iterator begin, FileNames::const_iterator end)
@@ -160,7 +160,7 @@ void DatabaseOrdinary::loadTables(
 
             /// Messages, so that it's not boring to wait for the server to load for a long time.
             if ((++tables_processed) % PRINT_MESSAGE_EACH_N_TABLES == 0
-                || watch.lockTestAndRestart(PRINT_MESSAGE_EACH_N_SECONDS))
+                || watch.compareAndRestart(PRINT_MESSAGE_EACH_N_SECONDS))
             {
                 LOG_INFO(log, std::fixed << std::setprecision(2) << tables_processed * 100.0 / total_tables << "%");
                 watch.restart();
@@ -200,7 +200,7 @@ void DatabaseOrdinary::startupTables(ThreadPool * thread_pool)
 {
     LOG_INFO(log, "Starting up tables.");
 
-    StopwatchWithLock watch;
+    AtomicStopwatch watch;
     std::atomic<size_t> tables_processed {0};
     size_t total_tables = tables.size();
 
@@ -209,7 +209,7 @@ void DatabaseOrdinary::startupTables(ThreadPool * thread_pool)
         for (auto it = begin; it != end; ++it)
         {
             if ((++tables_processed) % PRINT_MESSAGE_EACH_N_TABLES == 0
-                || watch.lockTestAndRestart(PRINT_MESSAGE_EACH_N_SECONDS))
+                || watch.compareAndRestart(PRINT_MESSAGE_EACH_N_SECONDS))
             {
                 LOG_INFO(log, std::fixed << std::setprecision(2) << tables_processed * 100.0 / total_tables << "%");
                 watch.restart();

dbms/src/Interpreters/Context.cpp

@@ -229,7 +229,7 @@ struct ContextShared
         Databases current_databases;
 
         {
-            Poco::ScopedLock<Poco::Mutex> lock(mutex);
+            std::lock_guard lock(mutex);
             current_databases = databases;
         }
 
@@ -237,7 +237,7 @@ struct ContextShared
             database.second->shutdown();
 
         {
-            Poco::ScopedLock<Poco::Mutex> lock(mutex);
+            std::lock_guard lock(mutex);
             databases.clear();
         }
     }
@@ -1428,7 +1428,7 @@ QueryLog & Context::getQueryLog()
         String partition_by = config.getString("query_log.partition_by", "toYYYYMM(event_date)");
         size_t flush_interval_milliseconds = config.getUInt64(
                 "query_log.flush_interval_milliseconds", DEFAULT_QUERY_LOG_FLUSH_INTERVAL_MILLISECONDS);
-        
+
         String engine = "ENGINE = MergeTree PARTITION BY (" + partition_by + ") ORDER BY (event_date, event_time) SETTINGS index_granularity = 1024";
 
         system_logs->query_log = std::make_unique<QueryLog>(*global_context, database, table, engine, flush_interval_milliseconds);

dbms/src/Storages/StorageMergeTree.cpp

@@ -293,7 +293,7 @@ bool StorageMergeTree::merge(
     String * out_disable_reason)
 {
     /// Clear old parts. It does not matter to do it more frequently than each second.
-    if (auto lock = time_after_previous_cleanup.lockTestAndRestartAfter(1))
+    if (auto lock = time_after_previous_cleanup.compareAndRestartDeferred(1))
     {
         data.clearOldPartsFromFilesystem();
         data.clearOldTemporaryDirectories();

dbms/src/Storages/StorageMergeTree.h

@@ -100,7 +100,7 @@ private:
     SimpleIncrement increment{0};
 
     /// For clearOldParts, clearOldTemporaryDirectories.
-    StopwatchWithLock time_after_previous_cleanup;
+    AtomicStopwatch time_after_previous_cleanup;
 
     MergeTreeData::DataParts currently_merging;
     std::mutex currently_merging_mutex;

libs/libdaemon/src/BaseDaemon.cpp

@@ -49,7 +49,6 @@
 #include <Poco/Util/XMLConfiguration.h>
 #include <Poco/Util/MapConfiguration.h>
 #include <Poco/Util/Application.h>
-#include <Poco/ScopedLock.h>
 #include <Poco/Exception.h>
 #include <Poco/ErrorHandler.h>
 #include <Poco/NumberFormatter.h>