src/threadengines/threadengine_pthread.cpp

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