Better AsynchronousBlockInputStream in sync mode. [#CLICKHOUSE-2910]

dbms/src/Common/ThreadStatus.cpp

@@ -280,6 +280,7 @@ void ThreadStatus::attachQuery(
         if (auto current_query_context = getQueryContext())
         {
             log_to_query_thread_log = current_query_context->getSettingsRef().log_query_threads.value != 0;
+            log_profile_events = current_query_context->getSettingsRef().log_profile_events.value != 0;
 
             if (!getGlobalContext())
                 global_context = &current_query_context->getGlobalContext();
@@ -365,6 +366,7 @@ void ThreadStatus::logToQueryThreadLog(QueryThreadLog & thread_log)
     elem.written_bytes = progress_out.bytes.load(std::memory_order_relaxed);
     elem.memory_usage = std::max(0, memory_tracker.getPeak());
 
+    elem.thread_name = getThreadName();
     elem.thread_number = poco_thread_number;
     elem.os_thread_id = os_thread_id;
 
@@ -380,14 +382,11 @@ void ThreadStatus::logToQueryThreadLog(QueryThreadLog & thread_log)
         elem.client_info = query->getClientInfo();
     }
 
-    if (auto current_context = getQueryContext())
+    if (log_profile_events)
     {
-        if (current_context->getSettingsRef().log_profile_events)
-        {
-            /// 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);
-        }
+        /// 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);
     }
 
     thread_log.add(elem);

dbms/src/Common/ThreadStatus.h

@@ -92,7 +92,8 @@ protected:
     /// Is set once
     std::atomic<Context *> global_context{nullptr};
 
-    bool log_to_query_thread_log = false;
+    bool log_to_query_thread_log = true;
+    bool log_profile_events = true;
 
     Poco::Logger * log = nullptr;
 

dbms/src/Common/setThreadName.cpp

@@ -6,10 +6,19 @@
 #else
 #include <sys/prctl.h>
 #endif
+#include <pthread.h>
 
 #include <Common/Exception.h>
 #include <Common/setThreadName.h>
 
+namespace DB
+{
+namespace ErrorCodes
+{
+    extern const int PTHREAD_ERROR;
+}
+}
+
 
 void setThreadName(const char * name)
 {
@@ -22,5 +31,21 @@ void setThreadName(const char * name)
 #else
     if (0 != prctl(PR_SET_NAME, name, 0, 0, 0))
 #endif
-        DB::throwFromErrno("Cannot set thread name with prctl(PR_SET_NAME...)");
+        DB::throwFromErrno("Cannot set thread name with prctl(PR_SET_NAME, ...)");
+}
+
+std::string getThreadName()
+{
+    std::string name(16, '\0');
+
+#if defined(__FreeBSD__) || defined(__APPLE__)
+    if (pthread_get_name_np(pthread_self(), name.data(), name.size());
+        throw DB::Exception("Cannot get thread name with pthread_get_name_np()", DB::ErrorCodes::PTHREAD_ERROR);
+#else
+    if (0 != prctl(PR_GET_NAME, name.data(), 0, 0, 0))
+#endif
+        DB::throwFromErrno("Cannot get thread name with prctl(PR_GET_NAME)");
+
+    name.resize(strlen(name.data()));
+    return name;
 }

dbms/src/Common/setThreadName.h

@@ -1,7 +1,10 @@
 #pragma once
+#include <string>
 
 /** Sets the thread name (maximum length is 15 bytes),
   *  which will be visible in ps, gdb, /proc,
   *  for convenience of observation and debugging.
   */
 void setThreadName(const char * name);
+
+std::string getThreadName();

dbms/src/DataStreams/AsynchronousBlockInputStream.h

@@ -7,7 +7,6 @@
 #include <Common/CurrentMetrics.h>
 #include <common/ThreadPool.h>
 #include <Common/MemoryTracker.h>
-#include <Common/CurrentThread.h>
 #include <Poco/Ext/ThreadNumber.h>
 
 
@@ -93,66 +92,12 @@ protected:
     Block block;
     std::exception_ptr exception;
 
+    Block readImpl() override;
 
-    Block readImpl() override
-    {
-        /// If there were no calculations yet, calculate the first block synchronously
-        if (!started)
-        {
-            calculate();
-            started = true;
-        }
-        else    /// If the calculations are already in progress - wait for the result
-            pool.wait();
-
-        if (exception)
-            std::rethrow_exception(exception);
-
-        Block res = block;
-        if (!res)
-            return res;
-
-        /// Start the next block calculation
-        block.clear();
-        next();
-
-        return res;
-    }
-
-
-    void next()
-    {
-        ready.reset();
-        pool.schedule([this, main_thread=CurrentThread::get()] (){
-            CurrentThread::attachQueryFromSiblingThreadIfDetached(main_thread);
-            calculate();
-        });
-    }
-
+    void next();
 
     /// Calculations that can be performed in a separate thread
-    void calculate()
-    {
-        CurrentMetrics::Increment metric_increment{CurrentMetrics::QueryThread};
-
-        try
-        {
-            if (first)
-            {
-                first = false;
-                setThreadName("AsyncBlockInput");
-                children.back()->readPrefix();
-            }
-
-            block = children.back()->read();
-        }
-        catch (...)
-        {
-            exception = std::current_exception();
-        }
-
-        ready.set();
-    }
+    void calculate();
 };
 
 }

dbms/src/Interpreters/QueryThreadLog.cpp

@@ -32,6 +32,7 @@ Block QueryThreadLogElement::createBlock()
         {std::make_shared<DataTypeUInt64>(),        "written_bytes"},
         {std::make_shared<DataTypeUInt64>(),        "memory_usage"},
 
+        {std::make_shared<DataTypeString>(),        "thread_name"},
         {std::make_shared<DataTypeUInt32>(),        "thread_number"},
         {std::make_shared<DataTypeInt32>(),         "os_thread_id"},
         {std::make_shared<DataTypeUInt32>(),        "master_thread_number"},
@@ -81,6 +82,7 @@ void QueryThreadLogElement::appendToBlock(Block & block) const
 
     columns[i++]->insert(UInt64(memory_usage));
 
+    columns[i++]->insertData(thread_name.data(), thread_name.size());
     columns[i++]->insert(UInt64(thread_number));
     columns[i++]->insert(Int64(os_thread_id));
     columns[i++]->insert(UInt64(master_thread_number));

dbms/src/Interpreters/QueryThreadLog.h

@@ -24,6 +24,7 @@ struct QueryThreadLogElement
 
     UInt64 memory_usage{};
 
+    String thread_name;
     UInt32 thread_number{};
     Int32 os_thread_id{};
     UInt32 master_thread_number{};