Merge pull request #30000 from ClickHouse/profile-snapshot

Make read of Counters snapshot non-atomic

src/Common/ProfileEvents.cpp

@@ -301,11 +301,15 @@ void Counters::reset()
     resetCounters();
 }
 
-Counters Counters::getPartiallyAtomicSnapshot() const
+Counters::Snapshot::Snapshot()
+    : counters_holder(new Count[num_counters] {})
+{}
+
+Counters::Snapshot Counters::getPartiallyAtomicSnapshot() const
 {
-    Counters res(VariableContext::Snapshot, nullptr);
+    Snapshot res;
     for (Event i = 0; i < num_counters; ++i)
-        res.counters[i].store(counters[i].load(std::memory_order_relaxed), std::memory_order_relaxed);
+        res.counters_holder[i] = counters[i].load(std::memory_order_relaxed);
     return res;
 }
 

src/Common/ProfileEvents.h

@@ -59,8 +59,23 @@ namespace ProfileEvents
             } while (current != nullptr);
         }
 
+        struct Snapshot
+        {
+            Snapshot();
+
+            const Count & operator[] (Event event) const
+            {
+                return counters_holder[event];
+            }
+
+        private:
+            std::unique_ptr<Count[]> counters_holder;
+
+            friend class Counters;
+        };
+
         /// Every single value is fetched atomically, but not all values as a whole.
-        Counters getPartiallyAtomicSnapshot() const;
+        Snapshot getPartiallyAtomicSnapshot() const;
 
         /// Reset all counters to zero and reset parent.
         void reset();

src/Common/VariableContext.h

@@ -8,5 +8,4 @@ enum class VariableContext
     User,           /// Group of processes
     Process,        /// For example, a query or a merge
     Thread,         /// A thread of a process
-    Snapshot        /// Does not belong to anybody
 };

src/Interpreters/ProcessList.cpp

@@ -408,7 +408,7 @@ QueryStatusInfo QueryStatus::getInfo(bool get_thread_list, bool get_profile_even
         }
 
         if (get_profile_events)
-            res.profile_counters = std::make_shared<ProfileEvents::Counters>(thread_group->performance_counters.getPartiallyAtomicSnapshot());
+            res.profile_counters = std::make_shared<ProfileEvents::Counters::Snapshot>(thread_group->performance_counters.getPartiallyAtomicSnapshot());
     }
 
     if (get_settings && getContext())
@@ -446,7 +446,7 @@ ProcessListForUserInfo ProcessListForUser::getInfo(bool get_profile_events) cons
     res.peak_memory_usage = user_memory_tracker.getPeak();
 
     if (get_profile_events)
-        res.profile_counters = std::make_shared<ProfileEvents::Counters>(user_performance_counters.getPartiallyAtomicSnapshot());
+        res.profile_counters = std::make_shared<ProfileEvents::Counters::Snapshot>(user_performance_counters.getPartiallyAtomicSnapshot());
 
     return res;
 }

src/Interpreters/ProcessList.h

@@ -66,7 +66,7 @@ struct QueryStatusInfo
 
     /// Optional fields, filled by query
     std::vector<UInt64> thread_ids;
-    std::shared_ptr<ProfileEvents::Counters> profile_counters;
+    std::shared_ptr<ProfileEvents::Counters::Snapshot> profile_counters;
     std::shared_ptr<Settings> query_settings;
     std::string current_database;
 };
@@ -186,7 +186,7 @@ struct ProcessListForUserInfo
     Int64 peak_memory_usage;
 
     // Optional field, filled by request.
-    std::shared_ptr<ProfileEvents::Counters> profile_counters;
+    std::shared_ptr<ProfileEvents::Counters::Snapshot> profile_counters;
 };
 
 

src/Interpreters/ProfileEventsExt.cpp

@@ -12,7 +12,7 @@ namespace ProfileEvents
 {
 
 /// Put implementation here to avoid extra linking dependencies for clickhouse_common_io
-void dumpToMapColumn(const Counters & counters, DB::IColumn * column, bool nonzero_only)
+void dumpToMapColumn(const Counters::Snapshot & counters, DB::IColumn * column, bool nonzero_only)
 {
     auto * column_map = column ? &typeid_cast<DB::ColumnMap &>(*column) : nullptr;
     if (!column_map)
@@ -26,7 +26,7 @@ void dumpToMapColumn(const Counters & counters, DB::IColumn * column, bool nonze
     size_t size = 0;
     for (Event event = 0; event < Counters::num_counters; ++event)
     {
-        UInt64 value = counters[event].load(std::memory_order_relaxed);
+        UInt64 value = counters[event];
 
         if (nonzero_only && 0 == value)
             continue;

src/Interpreters/ProfileEventsExt.h

@@ -7,6 +7,6 @@ namespace ProfileEvents
 {
 
 /// Dumps profile events to columns Map(String, UInt64)
-void dumpToMapColumn(const Counters & counters, DB::IColumn * column, bool nonzero_only = true);
+void dumpToMapColumn(const Counters::Snapshot & counters, DB::IColumn * column, bool nonzero_only = true);
 
 }

src/Interpreters/QueryLog.h

@@ -80,7 +80,7 @@ struct QueryLogElement
     String log_comment;
 
     std::vector<UInt64> thread_ids;
-    std::shared_ptr<ProfileEvents::Counters> profile_counters;
+    std::shared_ptr<ProfileEvents::Counters::Snapshot> profile_counters;
     std::shared_ptr<Settings> query_settings;
 
     static std::string name() { return "QueryLog"; }

src/Interpreters/QueryThreadLog.h

@@ -45,7 +45,7 @@ struct QueryThreadLogElement
 
     ClientInfo client_info;
 
-    std::shared_ptr<ProfileEvents::Counters> profile_counters;
+    std::shared_ptr<ProfileEvents::Counters::Snapshot> profile_counters;
 
     static std::string name() { return "QueryThreadLog"; }
 

src/Interpreters/QueryViewsLog.h

@@ -64,7 +64,7 @@ struct QueryViewsLogElement
     UInt64 written_rows{};
     UInt64 written_bytes{};
     Int64 peak_memory_usage{};
-    std::shared_ptr<ProfileEvents::Counters> profile_counters;
+    std::shared_ptr<ProfileEvents::Counters::Snapshot> profile_counters;
 
     ViewStatus status = ViewStatus::QUERY_START;
     Int32 exception_code{};

src/Interpreters/ThreadStatusExt.cpp

@@ -478,7 +478,7 @@ void ThreadStatus::logToQueryThreadLog(QueryThreadLog & thread_log, const String
         if (query_context_ptr->getSettingsRef().log_profile_events != 0)
         {
             /// NOTE: Here we are in the same thread, so we can make memcpy()
-            elem.profile_counters = std::make_shared<ProfileEvents::Counters>(performance_counters.getPartiallyAtomicSnapshot());
+            elem.profile_counters = std::make_shared<ProfileEvents::Counters::Snapshot>(performance_counters.getPartiallyAtomicSnapshot());
         }
     }
 
@@ -519,7 +519,7 @@ void ThreadStatus::logToQueryViewsLog(const ViewRuntimeData & vinfo)
         element.view_query = getCleanQueryAst(vinfo.query, query_context_ptr);
     element.view_target = vinfo.runtime_stats->target_name;
 
-    auto events = std::make_shared<ProfileEvents::Counters>(performance_counters.getPartiallyAtomicSnapshot());
+    auto events = std::make_shared<ProfileEvents::Counters::Snapshot>(performance_counters.getPartiallyAtomicSnapshot());
     element.read_rows = progress_in.read_rows.load(std::memory_order_relaxed);
     element.read_bytes = progress_in.read_bytes.load(std::memory_order_relaxed);
     element.written_rows = (*events)[ProfileEvents::InsertedRows];

src/Server/TCPHandler.cpp

@@ -843,7 +843,7 @@ namespace
     struct ProfileEventsSnapshot
     {
         UInt64 thread_id;
-        ProfileEvents::Counters counters;
+        ProfileEvents::Counters::Snapshot counters;
         Int64 memory_usage;
         time_t current_time;
     };
@@ -861,7 +861,7 @@ namespace
         auto & value_column = columns[VALUE_COLUMN_INDEX];
         for (ProfileEvents::Event event = 0; event < ProfileEvents::Counters::num_counters; ++event)
         {
-            UInt64 value = snapshot.counters[event].load(std::memory_order_relaxed);
+            UInt64 value = snapshot.counters[event];
 
             if (value == 0)
                 continue;