Use ThreadFromGlobalPool in Poco::ThreadPool

base/poco/Foundation/CMakeLists.txt

@@ -168,6 +168,9 @@ set (SRCS
 add_library (_poco_foundation ${SRCS})
 add_library (Poco::Foundation ALIAS _poco_foundation)
 
+target_link_libraries (_poco_foundation PRIVATE clickhouse_common_io)
+target_compile_definitions(_poco_foundation PUBLIC USE_CLICKHOUSE_THREADS=1)
+
 # TODO: remove these warning exclusions
 target_compile_options (_poco_foundation
     PRIVATE

base/poco/Foundation/src/ThreadPool.cpp

@@ -20,12 +20,15 @@
 #include "Poco/ErrorHandler.h"
 #include <sstream>
 #include <ctime>
+#if USE_CLICKHOUSE_THREADS
+#include <Common/ThreadPool.h>
+#endif
 
 
 namespace Poco {
 
 
-class PooledThread: public Runnable
+class PooledThread : public Runnable
 {
 public:
 	PooledThread(const std::string& name, int stackSize = POCO_THREAD_STACK_SIZE);
@@ -46,7 +49,11 @@ private:
 	volatile std::time_t _idleTime;
 	Runnable*            _pTarget;
 	std::string          _name;
+#if USE_CLICKHOUSE_THREADS
+	ThreadFromGlobalPool _thread;
+#else
 	Thread               _thread;
+#endif
 	Event                _targetReady;
 	Event                _targetCompleted;
 	Event                _started;
@@ -54,16 +61,20 @@ private:
 };
 
 
-PooledThread::PooledThread(const std::string& name, int stackSize): 
-	_idle(true), 
-	_idleTime(0), 
-	_pTarget(0), 
-	_name(name), 
+PooledThread::PooledThread(const std::string& name, [[maybe_unused]] int stackSize):
+	_idle(true),
+	_idleTime(0),
+	_pTarget(0),
+	_name(name),
+#if !USE_CLICKHOUSE_THREADS
 	_thread(name),
+#endif
 	_targetCompleted(false)
 {
 	poco_assert_dbg (stackSize >= 0);
+#if !USE_CLICKHOUSE_THREADS
 	_thread.setStackSize(stackSize);
+#endif
 	_idleTime = std::time(NULL);
 }
 
@@ -75,24 +86,32 @@ PooledThread::~PooledThread()
 
 void PooledThread::start()
 {
+#if USE_CLICKHOUSE_THREADS
+	_thread = ThreadFromGlobalPool([this]() { run(); });
+#else
 	_thread.start(*this);
+#endif
 	_started.wait();
 }
 
 
-void PooledThread::start(Thread::Priority priority, Runnable& target)
+void PooledThread::start([[maybe_unused]] Thread::Priority priority, Runnable& target)
 {
 	FastMutex::ScopedLock lock(_mutex);
-	
+
 	poco_assert (_pTarget == 0);
 
 	_pTarget = &target;
+
+#if !USE_CLICKHOUSE_THREADS
 	_thread.setPriority(priority);
+#endif
+
 	_targetReady.set();
 }
 
 
-void PooledThread::start(Thread::Priority priority, Runnable& target, const std::string& name)
+void PooledThread::start([[maybe_unused]] Thread::Priority priority, Runnable& target, const std::string& name)
 {
 	FastMutex::ScopedLock lock(_mutex);
 
@@ -107,9 +126,12 @@ void PooledThread::start(Thread::Priority priority, Runnable& target, const std:
 		fullName.append(_name);
 		fullName.append(")");
 	}
+
+#if !USE_CLICKHOUSE_THREADS
 	_thread.setName(fullName);
 	_thread.setPriority(priority);
-	
+#endif
+
 	poco_assert (_pTarget == 0);
 
 	_pTarget = &target;
@@ -145,7 +167,7 @@ void PooledThread::join()
 void PooledThread::activate()
 {
 	FastMutex::ScopedLock lock(_mutex);
-	
+
 	poco_assert (_idle);
 	_idle = false;
 	_targetCompleted.reset();
@@ -154,21 +176,30 @@ void PooledThread::activate()
 
 void PooledThread::release()
 {
-	const long JOIN_TIMEOUT = 10000;
-	
 	_mutex.lock();
 	_pTarget = 0;
 	_mutex.unlock();
+
 	// In case of a statically allocated thread pool (such
 	// as the default thread pool), Windows may have already
 	// terminated the thread before we got here.
+#if USE_CLICKHOUSE_THREADS
+	if (_thread.joinable())
+#else
 	if (_thread.isRunning())
+#endif
 		_targetReady.set();
 
+#if USE_CLICKHOUSE_THREADS
+	if (_thread.joinable())
+		_thread.join();
+#else
+	const long JOIN_TIMEOUT = 10000;
 	if (_thread.tryJoin(JOIN_TIMEOUT))
 	{
 		delete this;
 	}
+#endif
 }
 
 
@@ -205,8 +236,10 @@ void PooledThread::run()
 			_idle     = true;
 			_targetCompleted.set();
 			ThreadLocalStorage::clear();
+#if !USE_CLICKHOUSE_THREADS
 			_thread.setName(_name);
 			_thread.setPriority(Thread::PRIO_NORMAL);
+#endif
 		}
 		else
 		{
@@ -220,9 +253,9 @@ void PooledThread::run()
 ThreadPool::ThreadPool(int minCapacity,
 	int maxCapacity,
 	int idleTime,
-	int stackSize): 
-	_minCapacity(minCapacity), 
-	_maxCapacity(maxCapacity), 
+	int stackSize):
+	_minCapacity(minCapacity),
+	_maxCapacity(maxCapacity),
 	_idleTime(idleTime),
 	_serial(0),
 	_age(0),
@@ -245,8 +278,8 @@ ThreadPool::ThreadPool(const std::string& name,
 	int idleTime,
 	int stackSize):
 	_name(name),
-	_minCapacity(minCapacity), 
-	_maxCapacity(maxCapacity), 
+	_minCapacity(minCapacity),
+	_maxCapacity(maxCapacity),
 	_idleTime(idleTime),
 	_serial(0),
 	_age(0),
@@ -393,15 +426,15 @@ void ThreadPool::housekeep()
 	ThreadVec activeThreads;
 	idleThreads.reserve(_threads.size());
 	activeThreads.reserve(_threads.size());
-	
+
 	for (ThreadVec::iterator it = _threads.begin(); it != _threads.end(); ++it)
 	{
 		if ((*it)->idle())
 		{
 			if ((*it)->idleTime() < _idleTime)
 				idleThreads.push_back(*it);
-			else 
-				expiredThreads.push_back(*it);	
+			else
+				expiredThreads.push_back(*it);
 		}
 		else activeThreads.push_back(*it);
 	}
@@ -481,7 +514,7 @@ public:
 	ThreadPool* pool()
 	{
 		FastMutex::ScopedLock lock(_mutex);
-		
+
 		if (!_pPool)
 		{
 			_pPool = new ThreadPool("default");
@@ -490,7 +523,7 @@ public:
 		}
 		return _pPool;
 	}
-	
+
 private:
 	ThreadPool* _pPool;
 	FastMutex   _mutex;

programs/server/Server.cpp

@@ -721,11 +721,6 @@ try
     CurrentMetrics::set(CurrentMetrics::Revision, ClickHouseRevision::getVersionRevision());
     CurrentMetrics::set(CurrentMetrics::VersionInteger, ClickHouseRevision::getVersionInteger());
 
-    Poco::ThreadPool server_pool(3, server_settings.max_connections);
-    std::mutex servers_lock;
-    std::vector<ProtocolServerAdapter> servers;
-    std::vector<ProtocolServerAdapter> servers_to_start_before_tables;
-
     /** Context contains all that query execution is dependent:
       *  settings, available functions, data types, aggregate functions, databases, ...
       */
@@ -823,6 +818,11 @@ try
             total_memory_tracker.setSampleMaxAllocationSize(server_settings.total_memory_profiler_sample_max_allocation_size);
     }
 
+    Poco::ThreadPool server_pool(3, server_settings.max_connections);
+    std::mutex servers_lock;
+    std::vector<ProtocolServerAdapter> servers;
+    std::vector<ProtocolServerAdapter> servers_to_start_before_tables;
+
     /// Wait for all threads to avoid possible use-after-free (for example logging objects can be already destroyed).
     SCOPE_EXIT({
         Stopwatch watch;

src/Server/KeeperTCPHandler.cpp

@@ -309,7 +309,6 @@ Poco::Timespan KeeperTCPHandler::receiveHandshake(int32_t handshake_length, bool
 void KeeperTCPHandler::runImpl()
 {
     setThreadName("KeeperHandler");
-    ThreadStatus thread_status;
 
     socket().setReceiveTimeout(receive_timeout);
     socket().setSendTimeout(send_timeout);

src/Server/TCPHandler.cpp

@@ -246,7 +246,6 @@ TCPHandler::~TCPHandler()
 void TCPHandler::runImpl()
 {
     setThreadName("TCPHandler");
-    ThreadStatus thread_status;
 
     extractConnectionSettingsFromContext(server.context());