src/threadengines/threadengine_pthread.cpp

   1 /*
   2  * InspIRCd -- Internet Relay Chat Daemon
   3  *
   4  *   Copyright (C) 2013-2015 Attila Molnar <attilamolnar@hush.com>
   5  *   Copyright (C) 2012 Robby <robby@chatbelgie.be>
   6  *   Copyright (C) 2009-2010 Daniel De Graaf <danieldg@inspircd.org>
   7  *   Copyright (C) 2008, 2010 Craig Edwards <brain@inspircd.org>
   8  *
   9  * This file is part of InspIRCd.  InspIRCd is free software: you can
  10  * redistribute it and/or modify it under the terms of the GNU General Public
  11  * License as published by the Free Software Foundation, version 2.
  12  *
  13  * This program is distributed in the hope that it will be useful, but WITHOUT
  14  * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
  15  * FOR A PARTICULAR PURPOSE.  See the GNU General Public License for more
  16  * details.
  17  *
  18  * You should have received a copy of the GNU General Public License
  19  * along with this program.  If not, see <http://www.gnu.org/licenses/>.
  20  */
  21
  22
  23 #include "inspircd.h"
  24 #include "threadengines/threadengine_pthread.h"
  25 #include <pthread.h>
  26 #include <fcntl.h>
  27
  28 static void* entry_point(void* parameter)
  29 {
  30         /* Recommended by nenolod, signal safety on a per-thread basis */
  31         sigset_t set;
  32         sigemptyset(&set);
  33         sigaddset(&set, SIGPIPE);
  34         pthread_sigmask(SIG_BLOCK, &set, NULL);
  35
  36         Thread* pt = static_cast<Thread*>(parameter);
  37         pt->Run();
  38         return parameter;
  39 }
  40
  41
  42 void ThreadEngine::Start(Thread* thread)
  43 {
  44         if (pthread_create(&thread->state.pthread_id, NULL, entry_point, thread) != 0)
  45                 throw CoreException("Unable to create new thread: " + std::string(strerror(errno)));
  46 }
  47
  48 void ThreadEngine::Stop(Thread* thread)
  49 {
  50         thread->SetExitFlag();
  51         pthread_join(thread->state.pthread_id, NULL);
  52 }
  53
  54 #ifdef HAS_EVENTFD
  55 #include <sys/eventfd.h>
  56
  57 class ThreadSignalSocket : public EventHandler
  58 {
  59         SocketThread* parent;
  60  public:
  61         ThreadSignalSocket(SocketThread* p, int newfd) : parent(p)
  62         {
  63                 SetFd(newfd);
  64                 SocketEngine::AddFd(this, FD_WANT_FAST_READ | FD_WANT_NO_WRITE);
  65         }
  66
  67         ~ThreadSignalSocket()
  68         {
  69                 SocketEngine::Close(this);
  70         }
  71
  72         void Notify()
  73         {
  74                 eventfd_write(fd, 1);
  75         }
  76
  77         void OnEventHandlerRead() CXX11_OVERRIDE
  78         {
  79                 eventfd_t dummy;
  80                 eventfd_read(fd, &dummy);
  81                 parent->OnNotify();
  82         }
  83
  84         void OnEventHandlerWrite() CXX11_OVERRIDE
  85         {
  86                 ServerInstance->GlobalCulls.AddItem(this);
  87         }
  88
  89         void OnEventHandlerError(int errcode) CXX11_OVERRIDE
  90         {
  91                 ThreadSignalSocket::OnEventHandlerWrite();
  92         }
  93 };
  94
  95 SocketThread::SocketThread()
  96 {
  97         signal.sock = NULL;
  98         int fd = eventfd(0, EFD_NONBLOCK);
  99         if (fd < 0)
 100                 throw CoreException("Could not create pipe " + std::string(strerror(errno)));
 101         signal.sock = new ThreadSignalSocket(this, fd);
 102 }
 103 #else
 104
 105 class ThreadSignalSocket : public EventHandler
 106 {
 107         SocketThread* parent;
 108         int send_fd;
 109  public:
 110         ThreadSignalSocket(SocketThread* p, int recvfd, int sendfd) :
 111                 parent(p), send_fd(sendfd)
 112         {
 113                 SetFd(recvfd);
 114                 SocketEngine::NonBlocking(fd);
 115                 SocketEngine::AddFd(this, FD_WANT_FAST_READ | FD_WANT_NO_WRITE);
 116         }
 117
 118         ~ThreadSignalSocket()
 119         {
 120                 close(send_fd);
 121                 SocketEngine::Close(this);
 122         }
 123
 124         void Notify()
 125         {
 126                 static const char dummy = '*';
 127                 write(send_fd, &dummy, 1);
 128         }
 129
 130         void OnEventHandlerRead() CXX11_OVERRIDE
 131         {
 132                 char dummy[128];
 133                 read(fd, dummy, 128);
 134                 parent->OnNotify();
 135         }
 136
 137         void OnEventHandlerWrite() CXX11_OVERRIDE
 138         {
 139                 ServerInstance->GlobalCulls.AddItem(this);
 140         }
 141
 142         void OnEventHandlerError(int errcode) CXX11_OVERRIDE
 143         {
 144                 ThreadSignalSocket::OnEventHandlerWrite();
 145         }
 146 };
 147
 148 SocketThread::SocketThread()
 149 {
 150         signal.sock = NULL;
 151         int fds[2];
 152         if (pipe(fds))
 153                 throw CoreException("Could not create pipe " + std::string(strerror(errno)));
 154         signal.sock = new ThreadSignalSocket(this, fds[0], fds[1]);
 155 }
 156 #endif
 157
 158 void SocketThread::NotifyParent()
 159 {
 160         signal.sock->Notify();
 161 }
 162
 163 SocketThread::~SocketThread()
 164 {
 165         if (signal.sock)
 166         {
 167                 signal.sock->cull();
 168                 delete signal.sock;
 169         }
 170 }