From e7887c1f013248274574ab8e3167f742ccb3d69b Mon Sep 17 00:00:00 2001 From: Adam Date: Tue, 26 Apr 2011 19:13:51 -0400 Subject: Unmodularized the socket engine because its causing problems and really is unnecessary --- src/CMakeLists.txt | 16 ++- src/dns.cpp | 5 +- src/init.cpp | 7 +- src/main.cpp | 19 ++-- src/socketengines/pipeengine_eventfd.cpp | 67 +++++++++++ src/socketengines/pipeengine_pipe.cpp | 70 ++++++++++++ src/socketengines/pipeengine_win32.cpp | 92 +++++++++++++++ src/socketengines/socketengine_epoll.cpp | 154 +++++++++++++++++++++++++ src/socketengines/socketengine_eventfd.cpp | 67 ----------- src/socketengines/socketengine_pipe.cpp | 70 ------------ src/socketengines/socketengine_poll.cpp | 176 +++++++++++++++++++++++++++++ src/socketengines/socketengine_select.cpp | 105 +++++++++++++++++ src/socketengines/socketengine_win32.cpp | 92 --------------- src/sockets.cpp | 34 +----- src/win32/anope_windows.h | 2 + src/win32/windows.cpp | 13 +++ 16 files changed, 716 insertions(+), 273 deletions(-) create mode 100644 src/socketengines/pipeengine_eventfd.cpp create mode 100644 src/socketengines/pipeengine_pipe.cpp create mode 100644 src/socketengines/pipeengine_win32.cpp create mode 100644 src/socketengines/socketengine_epoll.cpp delete mode 100644 src/socketengines/socketengine_eventfd.cpp delete mode 100644 src/socketengines/socketengine_pipe.cpp create mode 100644 src/socketengines/socketengine_poll.cpp create mode 100644 src/socketengines/socketengine_select.cpp delete mode 100644 src/socketengines/socketengine_win32.cpp (limited to 'src') diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt index f606beff6..47d2e8efb 100644 --- a/src/CMakeLists.txt +++ b/src/CMakeLists.txt @@ -5,19 +5,29 @@ file(GLOB SRC_SRCS RELATIVE ${CMAKE_CURRENT_SOURCE_DIR} "*.cpp") if(WIN32) append_to_list(SRC_SRCS win32/windows.cpp) append_to_list(SRC_SRCS threadengines/threadengine_win32.cpp) - append_to_list(SRC_SRCS socketengines/socketengine_win32.cpp) + append_to_list(SRC_SRCS socketengines/pipeengine_win32.cpp) # If not using Windows, add the pthread threading engine to the list else(WIN32) append_to_list(SRC_SRCS threadengines/threadengine_pthread.cpp) # If we have eventfd, use it if(HAVE_EVENTFD AND HAVE_SYS_EVENTFD_H) - append_to_list(SRC_SRCS socketengines/socketengine_eventfd.cpp) + append_to_list(SRC_SRCS socketengines/pipeengine_eventfd.cpp) # Else fall back to pipe else(HAVE_EVENTFD AND HAVE_SYS_EVENTFD_H) - append_to_list(SRC_SRCS socketengines/socketengine_pipe.cpp) + append_to_list(SRC_SRCS socketengines/pipeengine_pipe.cpp) endif(HAVE_EVENTFD AND HAVE_SYS_EVENTFD_H) endif(WIN32) +if(HAVE_EPOLL) + append_to_list(SRC_SRCS socketengines/socketengine_epoll.cpp) +else(HAVE_EPOLL) + if(HAVE_POLL) + append_to_list(SRC_SRCS socketengines/socketengine_poll.cpp) + else(HAVE_POLL) + append_to_list(SRC_SRCS socketengines/socketengine_select.cpp) + endif(HAVE_POLL) +endif(HAVE_EPOLL) + sort_list(SRC_SRCS) # Set all the files to use C++ as well as set their compile flags (use the module-specific compile flags, though) diff --git a/src/dns.cpp b/src/dns.cpp index 45b3db6bd..7475545c0 100644 --- a/src/dns.cpp +++ b/src/dns.cpp @@ -69,7 +69,7 @@ void DNSRequest::Process() DNSEngine->requests[this->id] = this; DNSEngine->packets.push_back(p); - SocketEngine->MarkWritable(DNSEngine->sock); + SocketEngine::MarkWritable(DNSEngine->sock); this->timeout = new DNSRequestTimeout(this, Config->DNSTimeout); } @@ -512,7 +512,8 @@ bool DNSSocket::ProcessWrite() delete r; DNSEngine->packets.erase(DNSEngine->packets.begin() + i - 1); } - SocketEngine->ClearWritable(this); + + SocketEngine::ClearWritable(this); return cont; } diff --git a/src/init.cpp b/src/init.cpp index c508b75ea..039067191 100644 --- a/src/init.cpp +++ b/src/init.cpp @@ -331,6 +331,9 @@ void Init(int ac, char **av) throw FatalException("Configuration file failed to validate"); } + /* Initialize the socket engine */ + SocketEngine::Init(); + /* Add IRCD Protocol Module; exit if there are errors */ if (protocol_module_init()) throw FatalException("Unable to load protocol module"); @@ -341,10 +344,6 @@ void Init(int ac, char **av) /* Add Encryption Modules */ ModuleManager::LoadModuleList(Config->EncModuleList); - /* Load the socket engine */ - if (ModuleManager::LoadModule(Config->SocketEngine, NULL) || !SocketEngine) - throw FatalException("Unable to load socket engine " + Config->SocketEngine); - /* Add Database Modules */ ModuleManager::LoadModuleList(Config->DBModuleList); diff --git a/src/main.cpp b/src/main.cpp index 53e348e69..7ec3cfe0e 100644 --- a/src/main.cpp +++ b/src/main.cpp @@ -100,6 +100,7 @@ UplinkSocket::UplinkSocket(bool ipv6) : ConnectionSocket(ipv6) UplinkSocket::~UplinkSocket() { + SocketEngine::Process(); UplinkSock = NULL; } @@ -148,10 +149,14 @@ void do_restart_services() if (!bi->GetUID().empty()) UserListByUID.erase(bi->GetUID()); } + + FOREACH_MOD(I_OnRestart, OnRestart()); + + ModuleManager::UnloadAll(); ircdproto->SendSquit(Config->ServerName, quitmsg); - SocketEngine->Process(); delete UplinkSock; - ModuleManager::UnloadAll(); + SocketEngine::Shutdown(); + chdir(binary_dir.c_str()); my_av[0] = const_cast(("./" + services_bin).c_str()); execve(services_bin.c_str(), my_av, my_envp); @@ -160,8 +165,6 @@ void do_restart_services() throw FatalException("Restart failed"); } - FOREACH_MOD(I_OnRestart, OnRestart()); - exit(1); } @@ -200,10 +203,12 @@ static void services_shutdown() delete u; } } - SocketEngine->Process(); - delete UplinkSock; FOREACH_MOD(I_OnShutdown, OnShutdown()); ModuleManager::UnloadAll(); + ircdproto->SendSquit(Config->ServerName, quitmsg); + delete UplinkSock; + SocketEngine::Shutdown(); + /* just in case they weren't all removed at least run once */ ModuleManager::CleanupRuntimeDirectory(); } @@ -458,7 +463,7 @@ int main(int ac, char **av, char **envp) ModeManager::ProcessModes(); /* Process the socket engine */ - SocketEngine->Process(); + SocketEngine::Process(); } if (quitting) diff --git a/src/socketengines/pipeengine_eventfd.cpp b/src/socketengines/pipeengine_eventfd.cpp new file mode 100644 index 000000000..713c73d34 --- /dev/null +++ b/src/socketengines/pipeengine_eventfd.cpp @@ -0,0 +1,67 @@ +#include "services.h" +#include + +class PipeIO : public SocketIO +{ + public: + /** Receive something from the buffer + * @param s The socket + * @param buf The buf to read to + * @param sz How much to read + * @return Number of bytes received + */ + int Recv(Socket *s, char *buf, size_t sz) const + { + static eventfd_t dummy; + return !eventfd_read(s->GetFD(), &dummy); + } + + /** Write something to the socket + * @param s The socket + * @param buf What to write + * @return Number of bytes written + */ + int Send(Socket *s, const Anope::string &buf) const + { + return !eventfd_write(s->GetFD(), 1); + } +} pipeSocketIO; + +Pipe::Pipe() : BufferedSocket() +{ + this->IO = &pipeSocketIO; + this->Sock = eventfd(0, EFD_NONBLOCK); + if (this->Sock < 0) + throw CoreException(Anope::string("Could not create pipe: ") + Anope::LastError()); + + this->IPv6 = false; + + SocketEngine::AddSocket(this); +} + +bool Pipe::ProcessRead() +{ + this->IO->Recv(this, NULL, 0); + return this->Read(""); +} + +bool Pipe::Read(const Anope::string &) +{ + this->OnNotify(); + return true; +} + +void Pipe::Notify() +{ + /* Note we send this immediatly. If use use Socket::Write and if this functions is called + * from a thread, only epoll is able to pick up the change to this sockets want flags immediately + * Other engines time out then pick up and write the change then read it back, which + * is too slow for most things. + */ + this->IO->Send(this, ""); +} + +void Pipe::OnNotify() +{ +} + diff --git a/src/socketengines/pipeengine_pipe.cpp b/src/socketengines/pipeengine_pipe.cpp new file mode 100644 index 000000000..9d5b47154 --- /dev/null +++ b/src/socketengines/pipeengine_pipe.cpp @@ -0,0 +1,70 @@ +#include "services.h" + +class PipeIO : public SocketIO +{ + public: + /** Receive something from the buffer + * @param s The socket + * @param buf The buf to read to + * @param sz How much to read + * @return Number of bytes received + */ + int Recv(Socket *s, char *buf, size_t sz) const + { + static char dummy[512]; + while (read(s->GetFD(), &dummy, 512) == 512); + return 0; + } + + /** Write something to the socket + * @param s The socket + * @param buf What to write + * @return Number of bytes written + */ + int Send(Socket *s, const Anope::string &buf) const + { + static const char dummy = '*'; + Pipe *pipe = debug_cast(s); + return write(pipe->WritePipe, &dummy, 1); + } +} pipeSocketIO; + +Pipe::Pipe() : BufferedSocket() +{ + int fds[2]; + if (pipe(fds)) + throw CoreException(Anope::string("Could not create pipe: ") + Anope::LastError()); + int flags = fcntl(fds[0], F_GETFL, 0); + fcntl(fds[0], F_SETFL, flags | O_NONBLOCK); + flags = fcntl(fds[1], F_GETFL, 0); + fcntl(fds[1], F_SETFL, flags | O_NONBLOCK); + + this->IO = &pipeSocketIO; + this->Sock = fds[0]; + this->WritePipe = fds[1]; + this->IPv6 = false; + + SocketEngine->AddSocket(this); +} + +bool Pipe::ProcessRead() +{ + this->IO->Recv(this, NULL, 0); + return this->Read(""); +} + +bool Pipe::Read(const Anope::string &) +{ + this->OnNotify(); + return true; +} + +void Pipe::Notify() +{ + this->IO->Send(this, ""); +} + +void Pipe::OnNotify() +{ +} + diff --git a/src/socketengines/pipeengine_win32.cpp b/src/socketengines/pipeengine_win32.cpp new file mode 100644 index 000000000..41ec37a4c --- /dev/null +++ b/src/socketengines/pipeengine_win32.cpp @@ -0,0 +1,92 @@ +#include "services.h" + +static ClientSocket *newsocket = NULL; + +class LSocket : public ListenSocket +{ + public: + LSocket(const Anope::string &host, int port) : ListenSocket(host, port, false) { } + + ClientSocket *OnAccept(int fd, const sockaddrs &addr) + { + newsocket = new ClientSocket(this, fd, addr); + return newsocket; + } +}; + +class PipeIO : public SocketIO +{ + public: + /** Receive something from the buffer + * @param s The socket + * @param buf The buf to read to + * @param sz How much to read + * @return Number of bytes received + */ + int Recv(Socket *s, char *buf, size_t sz) const + { + static char dummy[512]; + return recv(s->GetFD(), dummy, 512, 0); + } + + /** Write something to the socket + * @param s The socket + * @param buf What to write + * @return Number of bytes written + */ + int Send(Socket *s, const Anope::string &buf) const + { + static const char dummy = '*'; + Pipe *pipe = debug_cast(s); + return send(pipe->WritePipe, &dummy, 1, 0); + } +} pipeSocketIO; + +Pipe::Pipe() : BufferedSocket() +{ + LSocket lfs("127.0.0.1", 0); + + int cfd = socket(AF_INET, SOCK_STREAM, 0); + if (cfd == -1) + throw CoreException("Error accepting new socket for Pipe"); + + sockaddr_in addr; + socklen_t sz = sizeof(addr); + getsockname(lfs.GetFD(), reinterpret_cast(&addr), &sz); + + if (connect(cfd, reinterpret_cast(&addr), sz)) + throw CoreException("Error accepting new socket for Pipe"); + lfs.ProcessRead(); + if (!newsocket) + throw CoreException("Error accepting new socket for Pipe"); + + this->IO = &pipeSocketIO; + this->Sock = cfd; + this->WritePipe = newsocket->GetFD(); + this->IPv6 = false; + + SocketEngine::AddSocket(this); + newsocket = NULL; +} + +bool Pipe::ProcessRead() +{ + this->IO->Recv(this, NULL, 0); + return this->Read(""); +} + +bool Pipe::Read(const Anope::string &) +{ + this->OnNotify(); + return true; +} + +void Pipe::Notify() +{ + this->IO->Send(this, ""); +} + +void Pipe::OnNotify() +{ +} + diff --git a/src/socketengines/socketengine_epoll.cpp b/src/socketengines/socketengine_epoll.cpp new file mode 100644 index 000000000..0ced8e71e --- /dev/null +++ b/src/socketengines/socketengine_epoll.cpp @@ -0,0 +1,154 @@ +#include "module.h" +#include +#include + +static long max; +static int EngineHandle; +static epoll_event *events; + +void SocketEngine::Init() +{ + max = ulimit(4, 0); + + if (max <= 0) + { + Log() << "Can't determine maximum number of open sockets"; + throw CoreException("Can't determine maximum number of open sockets"); + } + + EngineHandle = epoll_create(max / 4); + + if (EngineHandle == -1) + { + Log() << "Could not initialize epoll socket engine: " << Anope::LastError(); + throw CoreException(Anope::string("Could not initialize epoll socket engine: ") + Anope::LastError()); + } + + events = new epoll_event[max]; + memset(events, 0, sizeof(epoll_event) * max); +} + +void SocketEngine::Shutdown() +{ + for (std::map::const_iterator it = Sockets.begin(), it_end = Sockets.end(); it != it_end; ++it) + delete it->second; + Sockets.clear(); + + delete [] events; +} + +void SocketEngine::AddSocket(Socket *s) +{ + epoll_event ev; + + memset(&ev, 0, sizeof(ev)); + + ev.events = EPOLLIN; + ev.data.fd = s->GetFD(); + + if (epoll_ctl(EngineHandle, EPOLL_CTL_ADD, ev.data.fd, &ev) == -1) + { + Log() << "Unable to add fd " << ev.data.fd << " to socketengine epoll: " << Anope::LastError(); + return; + } + + Sockets.insert(std::make_pair(ev.data.fd, s)); +} + +void SocketEngine::DelSocket(Socket *s) +{ + epoll_event ev; + + memset(&ev, 0, sizeof(ev)); + + ev.data.fd = s->GetFD(); + + if (epoll_ctl(EngineHandle, EPOLL_CTL_DEL, ev.data.fd, &ev) == -1) + { + Log() << "Unable to delete fd " << ev.data.fd << " from socketengine epoll: " << Anope::LastError(); + return; + } + + Sockets.erase(ev.data.fd); +} + +void SocketEngine::MarkWritable(Socket *s) +{ + if (s->HasFlag(SF_WRITABLE)) + return; + + epoll_event ev; + + memset(&ev, 0, sizeof(ev)); + + ev.events = EPOLLIN | EPOLLOUT; + ev.data.fd = s->GetFD(); + + if (epoll_ctl(EngineHandle, EPOLL_CTL_MOD, ev.data.fd, &ev) == -1) + Log() << "Unable to mark fd " << ev.data.fd << " as writable in socketengine epoll: " << Anope::LastError(); + else + s->SetFlag(SF_WRITABLE); +} + +void SocketEngine::ClearWritable(Socket *s) +{ + if (!s->HasFlag(SF_WRITABLE)) + return; + + epoll_event ev; + + memset(&ev, 0, sizeof(ev)); + + ev.events = EPOLLIN; + ev.data.fd = s->GetFD(); + + if (epoll_ctl(EngineHandle, EPOLL_CTL_MOD, ev.data.fd, &ev) == -1) + Log() << "Unable to mark fd " << ev.data.fd << " as unwritable in socketengine epoll: " << Anope::LastError(); + else + s->UnsetFlag(SF_WRITABLE); +} + +void SocketEngine::Process() +{ + int total = epoll_wait(EngineHandle, events, max - 1, Config->ReadTimeout * 1000); + Anope::CurTime = time(NULL); + + /* EINTR can be given if the read timeout expires */ + if (total == -1) + { + if (errno != EINTR) + Log() << "SockEngine::Process(): error: " << Anope::LastError(); + return; + } + + for (int i = 0; i < total; ++i) + { + epoll_event *ev = &events[i]; + Socket *s = Sockets[ev->data.fd]; + + if (s->HasFlag(SF_DEAD)) + continue; + if (ev->events & (EPOLLHUP | EPOLLERR)) + { + s->ProcessError(); + s->SetFlag(SF_DEAD); + continue; + } + + if ((ev->events & EPOLLIN) && !s->ProcessRead()) + s->SetFlag(SF_DEAD); + + if ((ev->events & EPOLLOUT) && !s->ProcessWrite()) + s->SetFlag(SF_DEAD); + } + + for (int i = 0; i < total; ++i) + { + epoll_event *ev = &events[i]; + Socket *s = Sockets[ev->data.fd]; + + if (s->HasFlag(SF_DEAD)) + delete s; + } +} + diff --git a/src/socketengines/socketengine_eventfd.cpp b/src/socketengines/socketengine_eventfd.cpp deleted file mode 100644 index 9e9ec677b..000000000 --- a/src/socketengines/socketengine_eventfd.cpp +++ /dev/null @@ -1,67 +0,0 @@ -#include "services.h" -#include - -class PipeIO : public SocketIO -{ - public: - /** Receive something from the buffer - * @param s The socket - * @param buf The buf to read to - * @param sz How much to read - * @return Number of bytes received - */ - int Recv(Socket *s, char *buf, size_t sz) const - { - static eventfd_t dummy; - return !eventfd_read(s->GetFD(), &dummy); - } - - /** Write something to the socket - * @param s The socket - * @param buf What to write - * @return Number of bytes written - */ - int Send(Socket *s, const Anope::string &buf) const - { - return !eventfd_write(s->GetFD(), 1); - } -} pipeSocketIO; - -Pipe::Pipe() : BufferedSocket() -{ - this->IO = &pipeSocketIO; - this->Sock = eventfd(0, EFD_NONBLOCK); - if (this->Sock < 0) - throw CoreException(Anope::string("Could not create pipe: ") + Anope::LastError()); - - this->IPv6 = false; - - SocketEngine->AddSocket(this); -} - -bool Pipe::ProcessRead() -{ - this->IO->Recv(this, NULL, 0); - return this->Read(""); -} - -bool Pipe::Read(const Anope::string &) -{ - this->OnNotify(); - return true; -} - -void Pipe::Notify() -{ - /* Note we send this immediatly. If use use Socket::Write and if this functions is called - * from a thread, only epoll is able to pick up the change to this sockets want flags immediately - * Other engines time out then pick up and write the change then read it back, which - * is too slow for most things. - */ - this->IO->Send(this, ""); -} - -void Pipe::OnNotify() -{ -} - diff --git a/src/socketengines/socketengine_pipe.cpp b/src/socketengines/socketengine_pipe.cpp deleted file mode 100644 index 9d5b47154..000000000 --- a/src/socketengines/socketengine_pipe.cpp +++ /dev/null @@ -1,70 +0,0 @@ -#include "services.h" - -class PipeIO : public SocketIO -{ - public: - /** Receive something from the buffer - * @param s The socket - * @param buf The buf to read to - * @param sz How much to read - * @return Number of bytes received - */ - int Recv(Socket *s, char *buf, size_t sz) const - { - static char dummy[512]; - while (read(s->GetFD(), &dummy, 512) == 512); - return 0; - } - - /** Write something to the socket - * @param s The socket - * @param buf What to write - * @return Number of bytes written - */ - int Send(Socket *s, const Anope::string &buf) const - { - static const char dummy = '*'; - Pipe *pipe = debug_cast(s); - return write(pipe->WritePipe, &dummy, 1); - } -} pipeSocketIO; - -Pipe::Pipe() : BufferedSocket() -{ - int fds[2]; - if (pipe(fds)) - throw CoreException(Anope::string("Could not create pipe: ") + Anope::LastError()); - int flags = fcntl(fds[0], F_GETFL, 0); - fcntl(fds[0], F_SETFL, flags | O_NONBLOCK); - flags = fcntl(fds[1], F_GETFL, 0); - fcntl(fds[1], F_SETFL, flags | O_NONBLOCK); - - this->IO = &pipeSocketIO; - this->Sock = fds[0]; - this->WritePipe = fds[1]; - this->IPv6 = false; - - SocketEngine->AddSocket(this); -} - -bool Pipe::ProcessRead() -{ - this->IO->Recv(this, NULL, 0); - return this->Read(""); -} - -bool Pipe::Read(const Anope::string &) -{ - this->OnNotify(); - return true; -} - -void Pipe::Notify() -{ - this->IO->Send(this, ""); -} - -void Pipe::OnNotify() -{ -} - diff --git a/src/socketengines/socketengine_poll.cpp b/src/socketengines/socketengine_poll.cpp new file mode 100644 index 000000000..b8eef91b8 --- /dev/null +++ b/src/socketengines/socketengine_poll.cpp @@ -0,0 +1,176 @@ +#include "module.h" + +#ifndef _WIN32 +# include +# include +# include +# ifndef POLLRDHUP +# define POLLRDHUP 0 +# endif +#else +# define poll WSAPoll +# define POLLRDHUP POLLHUP +#endif + +static long max; +static pollfd *events; +static int SocketCount; +static std::map socket_positions; + +void SocketEngine::Init() +{ + SocketCount = 0; +#ifndef _WIN32 + max = ulimit(4, 0); +#else + max = 1024; +#endif + + if (max <= 0) + { + Log() << "Can't determine maximum number of open sockets"; + throw CoreException("Can't determine maximum number of open sockets"); + } + + events = new pollfd[max]; +} + +void SocketEngine::Shutdown() +{ + for (std::map::const_iterator it = Sockets.begin(), it_end = Sockets.end(); it != it_end; ++it) + delete it->second; + Sockets.clear(); + + delete [] events; +} + +void SocketEngine::AddSocket(Socket *s) +{ + if (SocketCount == max) + { + Log() << "Unable to add fd " << s->GetFD() << " to socketengine poll, engine is full"; + return; + } + + pollfd *ev = &events[SocketCount]; + ev->fd = s->GetFD(); + ev->events = POLLIN; + ev->revents = 0; + + Sockets.insert(std::make_pair(ev->fd, s)); + socket_positions.insert(std::make_pair(ev->fd, SocketCount)); + + ++SocketCount; +} + +void SocketEngine::DelSocket(Socket *s) +{ + std::map::iterator pos = socket_positions.find(s->GetFD()); + if (pos == socket_positions.end()) + { + Log() << "Unable to delete unknown fd " << s->GetFD() << " from socketengine poll"; + return; + } + + if (pos->second != SocketCount - 1) + { + pollfd *ev = &events[pos->second], + *last_ev = &events[SocketCount - 1]; + + ev->fd = last_ev->fd; + ev->events = last_ev->events; + ev->revents = last_ev->revents; + + socket_positions[ev->fd] = pos->second; + } + + Sockets.erase(s->GetFD()); + socket_positions.erase(pos); + + --SocketCount; +} + +void SocketEngine::MarkWritable(Socket *s) +{ + if (s->HasFlag(SF_WRITABLE)) + return; + + std::map::iterator pos = socket_positions.find(s->GetFD()); + if (pos == socket_positions.end()) + { + Log() << "Unable to mark unknown fd " << s->GetFD() << " as writable"; + return; + } + + pollfd *ev = &events[pos->second]; + ev->events |= POLLOUT; + + s->SetFlag(SF_WRITABLE); +} + +void SocketEngine::ClearWritable(Socket *s) +{ + if (!s->HasFlag(SF_WRITABLE)) + return; + + std::map::iterator pos = socket_positions.find(s->GetFD()); + if (pos == socket_positions.end()) + { + Log() << "Unable to mark unknown fd " << s->GetFD() << " as writable"; + return; + } + + pollfd *ev = &events[pos->second]; + ev->events &= ~POLLOUT; + + s->UnsetFlag(SF_WRITABLE); +} + +void SocketEngine::Process() +{ + int total = poll(events, SocketCount, Config->ReadTimeout * 1000); + Anope::CurTime = time(NULL); + + /* EINTR can be given if the read timeout expires */ + if (total == -1) + { + if (errno != EINTR) + Log() << "SockEngine::Process(): error: " << Anope::LastError(); + return; + } + + for (int i = 0, processed = 0; i < SocketCount && processed != total; ++i) + { + pollfd *ev = &events[i]; + + if (ev->revents != 0) + ++processed; + + Socket *s = Sockets[ev->fd]; + + if (s->HasFlag(SF_DEAD)) + continue; + if (ev->revents & (POLLERR | POLLRDHUP)) + { + s->ProcessError(); + s->SetFlag(SF_DEAD); + continue; + } + + if ((ev->revents & POLLIN) && !s->ProcessRead()) + s->SetFlag(SF_DEAD); + + if ((ev->revents & POLLOUT) && !s->ProcessWrite()) + s->SetFlag(SF_DEAD); + } + + for (int i = 0; i < SocketCount; ++i) + { + pollfd *ev = &events[i]; + Socket *s = Sockets[ev->fd]; + + if (s->HasFlag(SF_DEAD)) + delete s; + } +} + diff --git a/src/socketengines/socketengine_select.cpp b/src/socketengines/socketengine_select.cpp new file mode 100644 index 000000000..43b9b17cc --- /dev/null +++ b/src/socketengines/socketengine_select.cpp @@ -0,0 +1,105 @@ +#include "module.h" + +static int MaxFD; +static fd_set ReadFDs; +static fd_set WriteFDs; + +void SocketEngine::Init() +{ + MaxFD = 0; + FD_ZERO(&ReadFDs); + FD_ZERO(&WriteFDs); +} + +void SocketEngine::Shutdown() +{ + for (std::map::const_iterator it = Sockets.begin(), it_end = Sockets.end(); it != it_end; ++it) + delete it->second; + Sockets.clear(); + + MaxFD = 0; + FD_ZERO(&ReadFDs); + FD_ZERO(&WriteFDs); +} + +void SocketEngine::AddSocket(Socket *s) +{ + if (s->GetFD() > MaxFD) + MaxFD = s->GetFD(); + FD_SET(s->GetFD(), &ReadFDs); + Sockets.insert(std::make_pair(s->GetFD(), s)); +} + +void SocketEngine::DelSocket(Socket *s) +{ + if (s->GetFD() == MaxFD) + --MaxFD; + FD_CLR(s->GetFD(), &ReadFDs); + FD_CLR(s->GetFD(), &WriteFDs); + Sockets.erase(s->GetFD()); +} + +void SocketEngine::MarkWritable(Socket *s) +{ + if (s->HasFlag(SF_WRITABLE)) + return; + FD_SET(s->GetFD(), &WriteFDs); + s->SetFlag(SF_WRITABLE); +} + +void SocketEngine::ClearWritable(Socket *s) +{ + if (!s->HasFlag(SF_WRITABLE)) + return; + FD_CLR(s->GetFD(), &WriteFDs); + s->UnsetFlag(SF_WRITABLE); +} + +void SocketEngine::Process() +{ + fd_set rfdset = ReadFDs, wfdset = WriteFDs, efdset = ReadFDs; + timeval tval; + tval.tv_sec = Config->ReadTimeout; + tval.tv_usec = 0; + + int sresult = select(MaxFD + 1, &rfdset, &wfdset, &efdset, &tval); + Anope::CurTime = time(NULL); + + if (sresult == -1) + { + Log() << "SockEngine::Process(): error: " << Anope::LastError(); + } + else if (sresult) + { + int processed = 0; + for (std::map::const_iterator it = Sockets.begin(), it_end = Sockets.end(); it != it_end && processed != sresult; ++it) + { + Socket *s = it->second; + + if (FD_ISSET(s->GetFD(), &efdset) || FD_ISSET(s->GetFD(), &rfdset) || FD_ISSET(s->GetFD(), &wfdset)) + ++processed; + if (s->HasFlag(SF_DEAD)) + continue; + if (FD_ISSET(s->GetFD(), &efdset)) + { + s->ProcessError(); + s->SetFlag(SF_DEAD); + continue; + } + if (FD_ISSET(s->GetFD(), &rfdset) && !s->ProcessRead()) + s->SetFlag(SF_DEAD); + if (FD_ISSET(s->GetFD(), &wfdset) && !s->ProcessWrite()) + s->SetFlag(SF_DEAD); + } + + for (std::map::iterator it = Sockets.begin(), it_end = Sockets.end(); it != it_end; ) + { + Socket *s = it->second; + ++it; + + if (s->HasFlag(SF_DEAD)) + delete s; + } + } +} + diff --git a/src/socketengines/socketengine_win32.cpp b/src/socketengines/socketengine_win32.cpp deleted file mode 100644 index 78d4bf92d..000000000 --- a/src/socketengines/socketengine_win32.cpp +++ /dev/null @@ -1,92 +0,0 @@ -#include "services.h" - -static ClientSocket *newsocket = NULL; - -class LSocket : public ListenSocket -{ - public: - LSocket(const Anope::string &host, int port) : ListenSocket(host, port, false) { } - - ClientSocket *OnAccept(int fd, const sockaddrs &addr) - { - newsocket = new ClientSocket(this, fd, addr); - return newsocket; - } -}; - -class PipeIO : public SocketIO -{ - public: - /** Receive something from the buffer - * @param s The socket - * @param buf The buf to read to - * @param sz How much to read - * @return Number of bytes received - */ - int Recv(Socket *s, char *buf, size_t sz) const - { - static char dummy[512]; - return recv(s->GetFD(), dummy, 512, 0); - } - - /** Write something to the socket - * @param s The socket - * @param buf What to write - * @return Number of bytes written - */ - int Send(Socket *s, const Anope::string &buf) const - { - static const char dummy = '*'; - Pipe *pipe = debug_cast(s); - return send(pipe->WritePipe, &dummy, 1, 0); - } -} pipeSocketIO; - -Pipe::Pipe() : BufferedSocket() -{ - LSocket lfs("127.0.0.1", 0); - - int cfd = socket(AF_INET, SOCK_STREAM, 0); - if (cfd == -1) - throw CoreException("Error accepting new socket for Pipe"); - - sockaddr_in addr; - socklen_t sz = sizeof(addr); - getsockname(lfs.GetFD(), reinterpret_cast(&addr), &sz); - - if (connect(cfd, reinterpret_cast(&addr), sz)) - throw CoreException("Error accepting new socket for Pipe"); - lfs.ProcessRead(); - if (!newsocket) - throw CoreException("Error accepting new socket for Pipe"); - - this->IO = &pipeSocketIO; - this->Sock = cfd; - this->WritePipe = newsocket->GetFD(); - this->IPv6 = false; - - SocketEngine->AddSocket(this); - newsocket = NULL; -} - -bool Pipe::ProcessRead() -{ - this->IO->Recv(this, NULL, 0); - return this->Read(""); -} - -bool Pipe::Read(const Anope::string &) -{ - this->OnNotify(); - return true; -} - -void Pipe::Notify() -{ - this->IO->Send(this, ""); -} - -void Pipe::OnNotify() -{ -} - diff --git a/src/sockets.cpp b/src/sockets.cpp index cdf713f81..ce2e8ef11 100644 --- a/src/sockets.cpp +++ b/src/sockets.cpp @@ -1,6 +1,7 @@ #include "services.h" -SocketEngineBase *SocketEngine = NULL; +std::map SocketEngine::Sockets; + int32 TotalRead = 0; int32 TotalWritten = 0; @@ -255,28 +256,6 @@ bool cidr::match(sockaddrs &other) return true; } -/** Default constructor - */ -SocketEngineBase::SocketEngineBase() -{ -#ifdef _WIN32 - if (WSAStartup(MAKEWORD(2, 0), &wsa)) - throw FatalException("Failed to initialize WinSock library"); -#endif -} - -/** Default destructor - */ -SocketEngineBase::~SocketEngineBase() -{ - for (std::map::const_iterator it = this->Sockets.begin(), it_end = this->Sockets.end(); it != it_end; ++it) - delete it->second; - this->Sockets.clear(); -#ifdef _WIN32 - WSACleanup(); -#endif -} - /** Receive something from the buffer * @param s The socket * @param buf The buf to read to @@ -367,15 +346,14 @@ Socket::Socket(int sock, bool ipv6, int type) : Flags(SocketFlagS this->Sock = socket(this->IPv6 ? AF_INET6 : AF_INET, type, 0); else this->Sock = sock; - SocketEngine->AddSocket(this); + SocketEngine::AddSocket(this); } /** Default destructor */ Socket::~Socket() { - if (SocketEngine) - SocketEngine->DelSocket(this); + SocketEngine::DelSocket(this); CloseSocket(this->Sock); this->IO->Destroy(); } @@ -526,7 +504,7 @@ bool BufferedSocket::ProcessWrite() return false; this->WriteBuffer = this->WriteBuffer.substr(count); if (this->WriteBuffer.empty()) - SocketEngine->ClearWritable(this); + SocketEngine::ClearWritable(this); return true; } @@ -565,7 +543,7 @@ void BufferedSocket::Write(const char *message, ...) void BufferedSocket::Write(const Anope::string &message) { WriteBuffer.append(message.str() + "\r\n"); - SocketEngine->MarkWritable(this); + SocketEngine::MarkWritable(this); } /** Get the length of the read buffer diff --git a/src/win32/anope_windows.h b/src/win32/anope_windows.h index 97ad76080..6d16e5554 100644 --- a/src/win32/anope_windows.h +++ b/src/win32/anope_windows.h @@ -42,6 +42,8 @@ /* VS2008 hates having this define before its own */ #define vsnprintf _vsnprintf +extern CoreExport void OnStartup(); +extern CoreExport void OnShutdown(); extern CoreExport USHORT WindowsGetLanguage(const char *lang); extern CoreExport int inet_pton(int af, const char *src, void *dst); extern CoreExport const char *inet_ntop(int af, const void *src, char *dst, size_t size); diff --git a/src/win32/windows.cpp b/src/win32/windows.cpp index 3876c62bf..265b5601b 100644 --- a/src/win32/windows.cpp +++ b/src/win32/windows.cpp @@ -34,6 +34,19 @@ WindowsLanguage WindowsLanguages[] = { {NULL, 0} }; +WSADATA SocketEngine::wsa; + +void OnStartup() +{ + if (WSAStartup(MAKEWORD(2, 0), &wsa)) + throw FatalException("Failed to initialize WinSock library"); +} + +void OnShutdown() +{ + WSACleanup(); +} + USHORT WindowsGetLanguage(const char *lang) { for (int i = 0; WindowsLanguages[i].languageName; ++i) -- cgit