thevar1able/ClickHouse
1
0
Fork 0
mirror of https://github.com/ClickHouse/ClickHouse.git synced 2024-09-19 16:20:50 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'use-thread-from-global-pool-in-poco-threadpool' into keeper-some-improvement

Browse Source
This commit is contained in:
Antonio Andelic 2024-06-11 09:46:58 +02:00
commit 737d7484c5
9 changed files with 73 additions and 45 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
base/poco/Foundation/CMakeLists.txt
View File

@ -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

79
base/poco/Foundation/src/ThreadPool.cpp
View File

@ -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;

27
programs/server/Server.cpp
View File

@ -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);
}
auto server_pool = std::make_unique<Poco::ThreadPool>(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;
@ -898,7 +898,8 @@ try
global_context->shutdownKeeperDispatcher();
/// Wait server pool to avoid use-after-free of destroyed context in the handlers
server_pool.joinAll();
server_pool->joinAll();
server_pool.reset();
/** Explicitly destroy Context. It is more convenient than in destructor of Server, because logger is still available.
* At this moment, no one could own shared part of Context.
@ -1629,7 +1630,7 @@ try
if (global_context->isServerCompletelyStarted())
{
std::lock_guard lock(servers_lock);
updateServers(*config, server_pool, async_metrics, servers, servers_to_start_before_tables);
updateServers(*config, *server_pool, async_metrics, servers, servers_to_start_before_tables);
}
}
@ -1726,7 +1727,7 @@ try
config_getter, global_context->getKeeperDispatcher(),
global_context->getSettingsRef().receive_timeout.totalSeconds(),
global_context->getSettingsRef().send_timeout.totalSeconds(),
false), server_pool, socket));
false), *server_pool, socket));
});
const char * secure_port_name = "keeper_server.tcp_port_secure";
@ -1748,7 +1749,7 @@ try
new KeeperTCPHandlerFactory(
config_getter, global_context->getKeeperDispatcher(),
global_context->getSettingsRef().receive_timeout.totalSeconds(),
global_context->getSettingsRef().send_timeout.totalSeconds(), true), server_pool, socket));
global_context->getSettingsRef().send_timeout.totalSeconds(), true), *server_pool, socket));
#else
UNUSED(port);
throw Exception(ErrorCodes::SUPPORT_IS_DISABLED, "SSL support for TCP protocol is disabled because Poco library was built without NetSSL support.");
@ -1780,7 +1781,7 @@ try
createKeeperHTTPControlMainHandlerFactory(
config_getter(),
global_context->getKeeperDispatcher(),
"KeeperHTTPControlHandler-factory"), server_pool, socket, http_params));
"KeeperHTTPControlHandler-factory"), *server_pool, socket, http_params));
});
}
#else
@ -1804,7 +1805,7 @@ try
config(),
interserver_listen_hosts,
listen_try,
server_pool,
*server_pool,
async_metrics,
servers_to_start_before_tables,
/* start_servers= */ false);
@ -1855,7 +1856,7 @@ try
config(),
listen_hosts,
listen_try,
server_pool,
*server_pool,
async_metrics,
servers,
/* start_servers= */ true,
@ -2027,7 +2028,7 @@ try
{
std::lock_guard lock(servers_lock);
createServers(config(), listen_hosts, listen_try, server_pool, async_metrics, servers);
createServers(config(), listen_hosts, listen_try, *server_pool, async_metrics, servers);
if (servers.empty())
throw Exception(ErrorCodes::NO_ELEMENTS_IN_CONFIG,
"No servers started (add valid listen_host and 'tcp_port' or 'http_port' "

1
src/Server/HTTPHandler.cpp
View File

@ -1060,7 +1060,6 @@ void HTTPHandler::formatExceptionForClient(int exception_code, HTTPServerRequest
void HTTPHandler::handleRequest(HTTPServerRequest & request, HTTPServerResponse & response, const ProfileEvents::Event & write_event)
{
setThreadName("HTTPHandler");
ThreadStatus thread_status;
session = std::make_unique<Session>(server.context(), ClientInfo::Interface::HTTP, request.isSecure());
SCOPE_EXIT({ session.reset(); });

2
src/Server/InterserverIOHTTPHandler.cpp
View File

@ -8,7 +8,6 @@
#include <Interpreters/InterserverIOHandler.h>
#include <Server/HTTP/HTMLForm.h>
#include <Server/HTTP/WriteBufferFromHTTPServerResponse.h>
#include <Common/ThreadStatus.h>
#include <Common/logger_useful.h>
#include <Common/setThreadName.h>
@ -81,7 +80,6 @@ void InterserverIOHTTPHandler::processQuery(HTTPServerRequest & request, HTTPSer
void InterserverIOHTTPHandler::handleRequest(HTTPServerRequest & request, HTTPServerResponse & response, const ProfileEvents::Event & write_event)
{
setThreadName("IntersrvHandler");
ThreadStatus thread_status;
/// In order to work keep-alive.
if (request.getVersion() == HTTPServerRequest::HTTP_1_1)

1
src/Server/KeeperTCPHandler.cpp
View File

@ -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);

2
src/Server/MySQLHandler.cpp
View File

@ -24,7 +24,6 @@
#include <Common/CurrentThread.h>
#include <Common/NetException.h>
#include <Common/OpenSSLHelpers.h>
#include <Common/ThreadStatus.h>
#include <Common/config_version.h>
#include <Common/logger_useful.h>
#include <Common/re2.h>
@ -199,7 +198,6 @@ MySQLHandler::~MySQLHandler() = default;
void MySQLHandler::run()
{
setThreadName("MySQLHandler");
ThreadStatus thread_status;
session = std::make_unique<Session>(server.context(), ClientInfo::Interface::MYSQL);
SCOPE_EXIT({ session.reset(); });

2
src/Server/PostgreSQLHandler.cpp
View File

@ -10,7 +10,6 @@
#include <base/scope_guard.h>
#include <pcg_random.hpp>
#include <Common/CurrentThread.h>
#include <Common/ThreadStatus.h>
#include <Common/config_version.h>
#include <Common/randomSeed.h>
#include <Common/setThreadName.h>
@ -59,7 +58,6 @@ void PostgreSQLHandler::changeIO(Poco::Net::StreamSocket & socket)
void PostgreSQLHandler::run()
{
setThreadName("PostgresHandler");
ThreadStatus thread_status;
session = std::make_unique<Session>(server.context(), ClientInfo::Interface::POSTGRESQL);
SCOPE_EXIT({ session.reset(); });

1
src/Server/TCPHandler.cpp
View File

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