From 03d3563ef95efc6ab8076d66ebda48545c6089c4 Mon Sep 17 00:00:00 2001 From: Sadie Powell Date: Wed, 15 May 2019 16:22:24 +0100 Subject: Replace socketengine_{pthread,win32} with C++11 threads. --- src/configreader.cpp | 8 +- src/inspircd.cpp | 10 +- src/modules/extra/m_ldap.cpp | 22 ++-- src/modules/extra/m_mysql.cpp | 24 ++--- src/server.cpp | 2 +- src/thread.cpp | 63 +++++++++++ src/threadengine.cpp | 30 ------ src/threadengines/threadengine_pthread.cpp | 168 ----------------------------- src/threadengines/threadengine_win32.cpp | 132 ----------------------- src/threadsocket.cpp | 139 ++++++++++++++++++++++++ 10 files changed, 233 insertions(+), 365 deletions(-) create mode 100644 src/thread.cpp delete mode 100644 src/threadengine.cpp delete mode 100644 src/threadengines/threadengine_pthread.cpp delete mode 100644 src/threadengines/threadengine_win32.cpp create mode 100644 src/threadsocket.cpp (limited to 'src') diff --git a/src/configreader.cpp b/src/configreader.cpp index b844be382..666d7a492 100644 --- a/src/configreader.cpp +++ b/src/configreader.cpp @@ -661,18 +661,18 @@ std::string ServerConfig::Escape(const std::string& str) return escaped; } -void ConfigReaderThread::Run() +void ConfigReaderThread::OnStart() { Config->Read(); done = true; } -void ConfigReaderThread::Finish() +void ConfigReaderThread::OnStop() { ServerConfig* old = ServerInstance->Config; ServerInstance->Logs.Log("CONFIG", LOG_DEBUG, "Switching to new configuration..."); ServerInstance->Config = this->Config; - Config->Apply(old, TheUserUID); + Config->Apply(old, UUID); if (Config->valid) { @@ -685,7 +685,7 @@ void ConfigReaderThread::Finish() ServerInstance->Users.RehashCloneCounts(); ServerInstance->XLines->CheckELines(); ServerInstance->XLines->ApplyLines(); - User* user = ServerInstance->FindNick(TheUserUID); + User* user = ServerInstance->FindUUID(UUID); ConfigStatus status(user); const ModuleManager::ModuleMap& mods = ServerInstance->Modules.GetModules(); diff --git a/src/inspircd.cpp b/src/inspircd.cpp index 23e543dab..3468ca7f3 100644 --- a/src/inspircd.cpp +++ b/src/inspircd.cpp @@ -588,13 +588,9 @@ void InspIRCd::Run() if (this->ConfigThread && this->ConfigThread->IsDone()) { /* Rehash has completed */ - this->Logs.Log("CONFIG", LOG_DEBUG, "Detected ConfigThread exiting, tidying up..."); - - this->ConfigThread->Finish(); - - ConfigThread->join(); - delete ConfigThread; - ConfigThread = NULL; + this->Logs.Log("CONFIG", LOG_DEBUG, "New configuration has been read, applying..."); + ConfigThread->Stop(); + DeleteZero(ConfigThread); } UpdateTime(); diff --git a/src/modules/extra/m_ldap.cpp b/src/modules/extra/m_ldap.cpp index d6cad357b..fd5761cf6 100644 --- a/src/modules/extra/m_ldap.cpp +++ b/src/modules/extra/m_ldap.cpp @@ -251,7 +251,7 @@ class LDAPService : public LDAPProvider, public SocketThread public: typedef std::vector query_queue; query_queue queries, results; - Mutex process_mutex; /* held when processing requests not in either queue */ + std::mutex process_mutex; /* held when processing requests not in either queue */ LDAPService(Module* c, ConfigTag* tag) : LDAPProvider(c, "LDAP/" + tag->getString("id")) @@ -431,7 +431,7 @@ class LDAPService : public LDAPProvider, public SocketThread void SendRequests() { - process_mutex.Lock(); + process_mutex.lock(); query_queue q; this->LockQueue(); @@ -440,7 +440,7 @@ class LDAPService : public LDAPProvider, public SocketThread if (q.empty()) { - process_mutex.Unlock(); + process_mutex.unlock(); return; } @@ -472,13 +472,13 @@ class LDAPService : public LDAPProvider, public SocketThread this->NotifyParent(); - process_mutex.Unlock(); + process_mutex.unlock(); } public: - void Run() override + void OnStart() override { - while (!this->GetExitFlag()) + while (!this->IsStopping()) { this->LockQueue(); if (this->queries.empty()) @@ -545,7 +545,7 @@ class ModuleLDAP : public Module conns[id] = conn; ServerInstance->Modules.AddService(*conn); - ServerInstance->Threads.Start(conn); + conn->Start(); } else { @@ -558,7 +558,7 @@ class ModuleLDAP : public Module { LDAPService* conn = i->second; ServerInstance->Modules.DelService(*conn); - conn->join(); + conn->Stop(); conn->OnNotify(); delete conn; } @@ -572,7 +572,7 @@ class ModuleLDAP : public Module { LDAPService* s = it->second; - s->process_mutex.Lock(); + s->process_mutex.lock(); s->LockQueue(); for (unsigned int i = s->queries.size(); i > 0; --i) @@ -600,7 +600,7 @@ class ModuleLDAP : public Module } s->UnlockQueue(); - s->process_mutex.Unlock(); + s->process_mutex.unlock(); } } @@ -609,7 +609,7 @@ class ModuleLDAP : public Module for (ServiceMap::iterator i = LDAPServices.begin(); i != LDAPServices.end(); ++i) { LDAPService* conn = i->second; - conn->join(); + conn->Stop(); conn->OnNotify(); delete conn; } diff --git a/src/modules/extra/m_mysql.cpp b/src/modules/extra/m_mysql.cpp index bffc63b1a..20f9f23cd 100644 --- a/src/modules/extra/m_mysql.cpp +++ b/src/modules/extra/m_mysql.cpp @@ -119,7 +119,7 @@ class DispatcherThread : public SocketThread public: DispatcherThread(ModuleSQL* CreatorModule) : Parent(CreatorModule) { } ~DispatcherThread() { } - void Run() override; + void OnStart() override; void OnNotify() override; }; @@ -241,7 +241,7 @@ class SQLConnection : public SQL::Provider public: reference config; MYSQL *connection; - Mutex lock; + std::mutex lock; // This constructor creates an SQLConnection object with the given credentials, but does not connect yet. SQLConnection(Module* p, ConfigTag* tag) : SQL::Provider(p, "SQL/" + tag->getString("id")), @@ -397,14 +397,14 @@ ModuleSQL::ModuleSQL() void ModuleSQL::init() { Dispatcher = new DispatcherThread(this); - ServerInstance->Threads.Start(Dispatcher); + Dispatcher->Start(); } ModuleSQL::~ModuleSQL() { if (Dispatcher) { - Dispatcher->join(); + Dispatcher->Stop(); Dispatcher->OnNotify(); delete Dispatcher; } @@ -444,8 +444,8 @@ void ModuleSQL::ReadConfig(ConfigStatus& status) { ServerInstance->Modules.DelService(*i->second); // it might be running a query on this database. Wait for that to complete - i->second->lock.Lock(); - i->second->lock.Unlock(); + i->second->lock.lock(); + i->second->lock.unlock(); // now remove all active queries to this DB for (size_t j = qq.size(); j > 0; j--) { @@ -478,8 +478,8 @@ void ModuleSQL::OnUnloadModule(Module* mod) { // need to wait until the query is done // (the result will be discarded) - qq[i].c->lock.Lock(); - qq[i].c->lock.Unlock(); + qq[i].c->lock.lock(); + qq[i].c->lock.unlock(); } qq[i].q->OnError(err); delete qq[i].q; @@ -496,18 +496,18 @@ Version ModuleSQL::GetVersion() return Version("Provides MySQL support", VF_VENDOR); } -void DispatcherThread::Run() +void DispatcherThread::OnStart() { this->LockQueue(); - while (!this->GetExitFlag()) + while (!this->IsStopping()) { if (!Parent->qq.empty()) { QQueueItem i = Parent->qq.front(); - i.c->lock.Lock(); + i.c->lock.lock(); this->UnlockQueue(); MySQLresult* res = i.c->DoBlockingQuery(i.query); - i.c->lock.Unlock(); + i.c->lock.unlock(); /* * At this point, the main thread could be working on: diff --git a/src/server.cpp b/src/server.cpp index cd8212844..43c553833 100644 --- a/src/server.cpp +++ b/src/server.cpp @@ -57,7 +57,7 @@ void InspIRCd::Rehash(const std::string& uuid) if (!ServerInstance->ConfigThread) { ServerInstance->ConfigThread = new ConfigReaderThread(uuid); - ServerInstance->Threads.Start(ServerInstance->ConfigThread); + ServerInstance->ConfigThread->Start(); } } diff --git a/src/thread.cpp b/src/thread.cpp new file mode 100644 index 000000000..b4ac76ea2 --- /dev/null +++ b/src/thread.cpp @@ -0,0 +1,63 @@ +/* + * InspIRCd -- Internet Relay Chat Daemon + * + * Copyright (C) 2019 Peter Powell + * + * This file is part of InspIRCd. InspIRCd is free software: you can + * redistribute it and/or modify it under the terms of the GNU General Public + * License as published by the Free Software Foundation, version 2. + * + * This program is distributed in the hope that it will be useful, but WITHOUT + * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS + * FOR A PARTICULAR PURPOSE. See the GNU General Public License for more + * details. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see . + */ + + +#include "inspircd.h" + +bool Thread::Start() +{ + try + { + if (thread) + return false; + + thread = new std::thread(Thread::StartInternal, this); + return true; + } + catch (const std::system_error& err) + { + throw CoreException("Unable to start new thread: " + std::string(err.what())); + } +} + +void Thread::StartInternal(Thread* thread) +{ +#ifndef _WIN32 + // C++ does not have an API for this so we still need to use pthreads. + sigset_t set; + sigemptyset(&set); + sigaddset(&set, SIGPIPE); + pthread_sigmask(SIG_BLOCK, &set, nullptr); +#endif + + thread->OnStart(); +} + +bool Thread::Stop() +{ + if (!thread) + return false; + + OnStop(); + stopping = true; + thread->join(); + + delete thread; + thread = nullptr; + return true; +} diff --git a/src/threadengine.cpp b/src/threadengine.cpp deleted file mode 100644 index f757aa56c..000000000 --- a/src/threadengine.cpp +++ /dev/null @@ -1,30 +0,0 @@ -/* - * InspIRCd -- Internet Relay Chat Daemon - * - * Copyright (C) 2008 Craig Edwards - * - * This file is part of InspIRCd. InspIRCd is free software: you can - * redistribute it and/or modify it under the terms of the GNU General Public - * License as published by the Free Software Foundation, version 2. - * - * This program is distributed in the hope that it will be useful, but WITHOUT - * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS - * FOR A PARTICULAR PURPOSE. See the GNU General Public License for more - * details. - * - * You should have received a copy of the GNU General Public License - * along with this program. If not, see . - */ - - -#include "inspircd.h" - -void Thread::SetExitFlag() -{ - ExitFlag = true; -} - -void Thread::join() -{ - ServerInstance->Threads.Stop(this); -} diff --git a/src/threadengines/threadengine_pthread.cpp b/src/threadengines/threadengine_pthread.cpp deleted file mode 100644 index c2190c879..000000000 --- a/src/threadengines/threadengine_pthread.cpp +++ /dev/null @@ -1,168 +0,0 @@ -/* - * InspIRCd -- Internet Relay Chat Daemon - * - * Copyright (C) 2009-2010 Daniel De Graaf - * Copyright (C) 2008 Craig Edwards - * - * This file is part of InspIRCd. InspIRCd is free software: you can - * redistribute it and/or modify it under the terms of the GNU General Public - * License as published by the Free Software Foundation, version 2. - * - * This program is distributed in the hope that it will be useful, but WITHOUT - * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS - * FOR A PARTICULAR PURPOSE. See the GNU General Public License for more - * details. - * - * You should have received a copy of the GNU General Public License - * along with this program. If not, see . - */ - - -#include "inspircd.h" -#include "threadengines/threadengine_pthread.h" -#include -#include - -static void* entry_point(void* parameter) -{ - /* Recommended by nenolod, signal safety on a per-thread basis */ - sigset_t set; - sigemptyset(&set); - sigaddset(&set, SIGPIPE); - pthread_sigmask(SIG_BLOCK, &set, NULL); - - Thread* pt = static_cast(parameter); - pt->Run(); - return parameter; -} - - -void ThreadEngine::Start(Thread* thread) -{ - if (pthread_create(&thread->state.pthread_id, NULL, entry_point, thread) != 0) - throw CoreException("Unable to create new thread: " + std::string(strerror(errno))); -} - -void ThreadEngine::Stop(Thread* thread) -{ - thread->SetExitFlag(); - pthread_join(thread->state.pthread_id, NULL); -} - -#ifdef HAS_EVENTFD -#include - -class ThreadSignalSocket : public EventHandler -{ - SocketThread* parent; - public: - ThreadSignalSocket(SocketThread* p, int newfd) : parent(p) - { - SetFd(newfd); - SocketEngine::AddFd(this, FD_WANT_FAST_READ | FD_WANT_NO_WRITE); - } - - ~ThreadSignalSocket() - { - SocketEngine::Close(this); - } - - void Notify() - { - eventfd_write(fd, 1); - } - - void OnEventHandlerRead() override - { - eventfd_t dummy; - eventfd_read(fd, &dummy); - parent->OnNotify(); - } - - void OnEventHandlerWrite() override - { - ServerInstance->GlobalCulls.AddItem(this); - } - - void OnEventHandlerError(int errcode) override - { - ThreadSignalSocket::OnEventHandlerWrite(); - } -}; - -SocketThread::SocketThread() -{ - signal.sock = NULL; - int fd = eventfd(0, EFD_NONBLOCK); - if (fd < 0) - throw CoreException("Could not create pipe " + std::string(strerror(errno))); - signal.sock = new ThreadSignalSocket(this, fd); -} -#else - -class ThreadSignalSocket : public EventHandler -{ - SocketThread* parent; - int send_fd; - public: - ThreadSignalSocket(SocketThread* p, int recvfd, int sendfd) : - parent(p), send_fd(sendfd) - { - SetFd(recvfd); - SocketEngine::NonBlocking(fd); - SocketEngine::AddFd(this, FD_WANT_FAST_READ | FD_WANT_NO_WRITE); - } - - ~ThreadSignalSocket() - { - close(send_fd); - SocketEngine::Close(this); - } - - void Notify() - { - static const char dummy = '*'; - write(send_fd, &dummy, 1); - } - - void OnEventHandlerRead() override - { - char dummy[128]; - read(fd, dummy, 128); - parent->OnNotify(); - } - - void OnEventHandlerWrite() override - { - ServerInstance->GlobalCulls.AddItem(this); - } - - void OnEventHandlerError(int errcode) override - { - ThreadSignalSocket::OnEventHandlerWrite(); - } -}; - -SocketThread::SocketThread() -{ - signal.sock = NULL; - int fds[2]; - if (pipe(fds)) - throw CoreException("Could not create pipe " + std::string(strerror(errno))); - signal.sock = new ThreadSignalSocket(this, fds[0], fds[1]); -} -#endif - -void SocketThread::NotifyParent() -{ - signal.sock->Notify(); -} - -SocketThread::~SocketThread() -{ - if (signal.sock) - { - signal.sock->cull(); - delete signal.sock; - } -} diff --git a/src/threadengines/threadengine_win32.cpp b/src/threadengines/threadengine_win32.cpp deleted file mode 100644 index f2a4fdc85..000000000 --- a/src/threadengines/threadengine_win32.cpp +++ /dev/null @@ -1,132 +0,0 @@ -/* - * InspIRCd -- Internet Relay Chat Daemon - * - * Copyright (C) 2009 Daniel De Graaf - * Copyright (C) 2008 Craig Edwards - * - * This file is part of InspIRCd. InspIRCd is free software: you can - * redistribute it and/or modify it under the terms of the GNU General Public - * License as published by the Free Software Foundation, version 2. - * - * This program is distributed in the hope that it will be useful, but WITHOUT - * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS - * FOR A PARTICULAR PURPOSE. See the GNU General Public License for more - * details. - * - * You should have received a copy of the GNU General Public License - * along with this program. If not, see . - */ - - -#include "inspircd.h" -#include "threadengines/threadengine_win32.h" - -void ThreadEngine::Start(Thread* thread) -{ - thread->state.handle = CreateThread(NULL, 0, ThreadEngine::Entry, thread, 0, NULL); - - if (thread->state.handle == NULL) - { - DWORD lasterr = GetLastError(); - std::string err = "Unable to create new thread: " + ConvToStr(lasterr); - SetLastError(ERROR_SUCCESS); - throw CoreException(err); - } -} - -DWORD WINAPI ThreadEngine::Entry(void* parameter) -{ - Thread* pt = static_cast(parameter); - pt->Run(); - return 0; -} - -void ThreadEngine::Stop(Thread* thread) -{ - thread->SetExitFlag(); - HANDLE handle = thread->state.handle; - WaitForSingleObject(handle,INFINITE); - CloseHandle(handle); -} - -class ThreadSignalSocket : public BufferedSocket -{ - SocketThread* parent; - public: - ThreadSignalSocket(SocketThread* t, int newfd) - : BufferedSocket(newfd), parent(t) - { - } - - void OnDataReady() - { - recvq.clear(); - parent->OnNotify(); - } - - void OnError(BufferedSocketError) - { - ServerInstance->GlobalCulls.AddItem(this); - } -}; - -static bool BindAndListen(int sockfd, int port, const char* addr) -{ - irc::sockets::sockaddrs servaddr; - if (!irc::sockets::aptosa(addr, port, servaddr)) - return false; - - if (SocketEngine::Bind(sockfd, servaddr) != 0) - return false; - - if (SocketEngine::Listen(sockfd, ServerInstance->Config->MaxConn) != 0) - { - ServerInstance->Logs.Log("SOCKET", LOG_DEFAULT, "ERROR in listen(): %s", strerror(errno)); - return false; - } - - return true; -} - -SocketThread::SocketThread() -{ - int listenFD = socket(AF_INET, SOCK_STREAM, 0); - if (listenFD == -1) - throw CoreException("Could not create ITC pipe"); - int connFD = socket(AF_INET, SOCK_STREAM, 0); - if (connFD == -1) - throw CoreException("Could not create ITC pipe"); - - if (!BindAndListen(listenFD, 0, "127.0.0.1")) - throw CoreException("Could not create ITC pipe"); - SocketEngine::NonBlocking(connFD); - - struct sockaddr_in addr; - socklen_t sz = sizeof(addr); - getsockname(listenFD, reinterpret_cast(&addr), &sz); - connect(connFD, reinterpret_cast(&addr), sz); - SocketEngine::Blocking(listenFD); - int nfd = accept(listenFD, reinterpret_cast(&addr), &sz); - if (nfd < 0) - throw CoreException("Could not create ITC pipe"); - new ThreadSignalSocket(this, nfd); - closesocket(listenFD); - - SocketEngine::Blocking(connFD); - this->signal.connFD = connFD; -} - -void SocketThread::NotifyParent() -{ - char dummy = '*'; - send(signal.connFD, &dummy, 1, 0); -} - -SocketThread::~SocketThread() -{ - if (signal.connFD >= 0) - { - shutdown(signal.connFD, 2); - closesocket(signal.connFD); - } -} diff --git a/src/threadsocket.cpp b/src/threadsocket.cpp new file mode 100644 index 000000000..c7539f578 --- /dev/null +++ b/src/threadsocket.cpp @@ -0,0 +1,139 @@ +/* + * InspIRCd -- Internet Relay Chat Daemon + * + * Copyright (C) 2009-2010 Daniel De Graaf + * Copyright (C) 2008 Craig Edwards + * + * This file is part of InspIRCd. InspIRCd is free software: you can + * redistribute it and/or modify it under the terms of the GNU General Public + * License as published by the Free Software Foundation, version 2. + * + * This program is distributed in the hope that it will be useful, but WITHOUT + * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS + * FOR A PARTICULAR PURPOSE. See the GNU General Public License for more + * details. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see . + */ + + +#include "inspircd.h" + +#ifdef HAS_EVENTFD +#include + +class ThreadSignalSocket : public EventHandler +{ + SocketThread* parent; + public: + ThreadSignalSocket(SocketThread* p, int newfd) : parent(p) + { + SetFd(newfd); + SocketEngine::AddFd(this, FD_WANT_FAST_READ | FD_WANT_NO_WRITE); + } + + ~ThreadSignalSocket() + { + SocketEngine::Close(this); + } + + void Notify() + { + eventfd_write(fd, 1); + } + + void OnEventHandlerRead() override + { + eventfd_t dummy; + eventfd_read(fd, &dummy); + parent->OnNotify(); + } + + void OnEventHandlerWrite() override + { + ServerInstance->GlobalCulls.AddItem(this); + } + + void OnEventHandlerError(int errcode) override + { + ThreadSignalSocket::OnEventHandlerWrite(); + } +}; + +SocketThread::SocketThread() +{ + socket = NULL; + int fd = eventfd(0, EFD_NONBLOCK); + if (fd < 0) + throw CoreException("Could not create pipe " + std::string(strerror(errno))); + socket = new ThreadSignalSocket(this, fd); +} +#else + +class ThreadSignalSocket : public EventHandler +{ + SocketThread* parent; + int send_fd; + public: + ThreadSignalSocket(SocketThread* p, int recvfd, int sendfd) : + parent(p), send_fd(sendfd) + { + SetFd(recvfd); + SocketEngine::NonBlocking(fd); + SocketEngine::AddFd(this, FD_WANT_FAST_READ | FD_WANT_NO_WRITE); + } + + ~ThreadSignalSocket() + { + close(send_fd); + SocketEngine::Close(this); + } + + void Notify() + { + static const char dummy = '*'; + write(send_fd, &dummy, 1); + } + + void OnEventHandlerRead() override + { + char dummy[128]; + read(fd, dummy, 128); + parent->OnNotify(); + } + + void OnEventHandlerWrite() override + { + ServerInstance->GlobalCulls.AddItem(this); + } + + void OnEventHandlerError(int errcode) override + { + ThreadSignalSocket::OnEventHandlerWrite(); + } +}; + +SocketThread::SocketThread() +{ + socket = NULL; + int fds[2]; + if (pipe(fds)) + throw CoreException("Could not create pipe " + std::string(strerror(errno))); + socket = new ThreadSignalSocket(this, fds[0], fds[1]); +} +#endif + +void SocketThread::NotifyParent() +{ + socket->Notify(); +} + +SocketThread::~SocketThread() +{ + if (socket) + { + socket->cull(); + delete socket; + } +} -- cgit v1.3.1-10-gc9f91