From 939cb0ba987f927f1ad900d45f98ba6d8e03e9d4 Mon Sep 17 00:00:00 2001 From: danieldg Date: Sat, 26 Sep 2009 01:43:09 +0000 Subject: [PATCH] Clean up SocketEngine interface to allow edge-triggered I/O and sockets that do not force readability. git-svn-id: http://svn.inspircd.org/repository/trunk/inspircd@11760 e03df62e-2008-0410-955e-edbf42e46eb7 --- include/socketengine.h | 175 +++++++++++--- include/socketengines/socketengine_epoll.h | 7 +- include/socketengines/socketengine_iocp.h | 7 +- include/socketengines/socketengine_kqueue.h | 5 +- include/socketengines/socketengine_poll.h | 6 +- include/socketengines/socketengine_ports.h | 5 +- include/socketengines/socketengine_select.h | 18 +- src/dns.cpp | 2 +- src/inspircd.cpp | 1 + src/inspsocket.cpp | 147 +++++++---- src/listensocket.cpp | 2 +- src/modules/extra/m_ssl_gnutls.cpp | 17 +- src/modules/extra/m_ssl_openssl.cpp | 254 ++++++-------------- src/modules/m_httpd_stats.cpp | 2 +- src/modules/m_ident.cpp | 8 +- src/socketengine.cpp | 64 +++-- src/socketengines/socketengine_epoll.cpp | 128 +++++----- src/socketengines/socketengine_iocp.cpp | 12 +- src/socketengines/socketengine_kqueue.cpp | 96 ++++---- src/socketengines/socketengine_poll.cpp | 126 ++++------ src/socketengines/socketengine_ports.cpp | 85 ++++--- src/socketengines/socketengine_select.cpp | 77 +++--- src/usermanager.cpp | 2 +- 23 files changed, 625 insertions(+), 621 deletions(-) diff --git a/include/socketengine.h b/include/socketengine.h index 4e87f7494..9c45bcf21 100644 --- a/include/socketengine.h +++ b/include/socketengine.h @@ -37,6 +37,97 @@ enum EventType EVENT_ERROR = 2 }; +/** + * Event mask for SocketEngine events + */ +enum EventMask +{ + /** Do not test this socket for readability + */ + FD_WANT_NO_READ = 0x1, + /** Give a read event at all times when reads will not block. + */ + FD_WANT_POLL_READ = 0x2, + /** Give a read event when there is new data to read. + * + * An event MUST be sent if there is new data to be read, and the most + * recent read/recv() on this FD returned EAGAIN. An event MAY be sent + * at any time there is data to be read on the socket. + */ + FD_WANT_FAST_READ = 0x4, + /** Give an optional read event when reads begin to unblock + * + * This state is useful if you want to leave data in the OS receive + * queue but not get continuous event notifications about it, because + * it may not require a system call to transition from FD_WANT_FAST_READ + */ + FD_WANT_EDGE_READ = 0x8, + + /** Mask for all read events */ + FD_WANT_READ_MASK = 0x0F, + + /** Do not test this socket for writeability + */ + FD_WANT_NO_WRITE = 0x10, + /** Give a write event at all times when writes will not block. + * + * You probably don't need to use this state; try your write first, and + * then use FD_WANT_FAST_WRITE. + */ + FD_WANT_POLL_WRITE = 0x20, + /** Give a write event when writes don't block any more + * + * An event MUST be sent if writes will not block, and the most recent + * write/send() on this FD returned EAGAIN, or connect() returned + * EINPROGRESS. An event MAY be sent at any time that writes will not + * block. + * + * Before calling HandleEvent, a socket engine MAY change the state of + * the FD back to FD_WANT_EDGE_WRITE if it is simpler (for example, if a + * one-shot notification was registered). If further writes are needed, + * it is the responsibility of the event handler to change the state to + * one that will generate the required notifications + */ + FD_WANT_FAST_WRITE = 0x40, + /** Give an optional write event on edge-triggered write unblock. + * + * This state is useful to avoid system calls when moving to/from + * FD_WANT_FAST_WRITE when writing data to a mostly-unblocked socket. + */ + FD_WANT_EDGE_WRITE = 0x80, + + /** Mask for all write events */ + FD_WANT_WRITE_MASK = 0xF0, + + /** Add a trial read. During the next DispatchEvents invocation, this + * will call HandleEvent with EVENT_READ unless reads are known to be + * blocking. Useful for edge-triggered reads; does nothing if + * FD_READ_WILL_BLOCK has been set on this EventHandler. + */ + FD_ADD_TRIAL_READ = 0x100, + /** Assert that reads are known to block. This cancels FD_ADD_TRIAL_READ. + */ + FD_READ_WILL_BLOCK = 0x200, + + /** Add a trial write. During the next DispatchEvents invocation, this + * will call HandleEvent with EVENT_WRITE unless writes are known to be + * blocking. + * + * This could be used to group several writes together into a single + * send() syscall, or to ensure that writes are blocking when attempting + * to use FD_WANT_FAST_WRITE. + */ + FD_ADD_TRIAL_WRITE = 0x1000, + /** Assert that writes are known to block. This cancels FD_ADD_TRIAL_WRITE. + */ + FD_WRITE_WILL_BLOCK = 0x2000, + + /** Mask for trial read/write items */ + FD_TRIAL_NOTE_MASK = 0x1100, + /** Mask for read/write blocking notifications */ + FD_BLOCK_NOTE_MASK = 0x2200 +}; + class InspIRCd; class Module; @@ -60,13 +151,15 @@ class Module; */ class CoreExport EventHandler : public Extensible { + private: + /** Private state maintained by socket engine */ + int event_mask; protected: /** File descriptor. - * All events which can be handled - * must have a file descriptor. - * This allows you to add events for - * sockets, fifo's, pipes, and various - * other forms of IPC. + * All events which can be handled must have a file descriptor. This + * allows you to add events for sockets, fifo's, pipes, and various + * other forms of IPC. Do not change this while the object is + * registered with the SocketEngine */ int fd; public: @@ -75,10 +168,11 @@ class CoreExport EventHandler : public Extensible */ inline int GetFd() const { return fd; } + inline int GetEventMask() const { return event_mask; } + /** Set a new file desciptor - * @param FD The new file descriptor. Do not - * call this method without first deleting the - * object from the SocketEngine if you have + * @param FD The new file descriptor. Do not call this method without + * first deleting the object from the SocketEngine if you have * added it to a SocketEngine instance. */ void SetFd(int FD); @@ -99,6 +193,8 @@ class CoreExport EventHandler : public Extensible * and EVENT_WRITE for write events. */ virtual void HandleEvent(EventType et, int errornum = 0) = 0; + + friend class SocketEngine; }; /** Provides basic file-descriptor-based I/O support. @@ -121,16 +217,16 @@ class CoreExport EventHandler : public Extensible */ class CoreExport SocketEngine { -protected: - /** Handle to socket engine, where needed. - */ - int EngineHandle; + protected: /** Current number of descriptors in the engine */ int CurrentSetSize; /** Reference table, contains all current handlers */ EventHandler** ref; + /** List of handlers that want a trial read/write + */ + std::set trials; int MAX_DESCRIPTORS; @@ -139,6 +235,9 @@ protected: time_t lastempty; void UpdateStats(size_t len_in, size_t len_out); + + virtual void OnSetEvent(EventHandler* eh, int old_mask, int new_mask) = 0; + void SetEventMask(EventHandler* eh, int value); public: double TotalEvents; @@ -167,9 +266,9 @@ public: * must provide an object derived from EventHandler which implements * HandleEvent(). * @param eh An event handling object to add - * @param writeFirst Wait for a write event instead of a read + * @param event_mask The initial event mask for the object */ - virtual bool AddFd(EventHandler* eh, bool writeFirst = false) = 0; + virtual bool AddFd(EventHandler* eh, int event_mask) = 0; /** If you call this function and pass it an * event handler, that event handler will @@ -179,22 +278,20 @@ public: * an eventhandler in the writeable state, * as this will consume large amounts of * CPU time. - * @param eh An event handler which wants to - * receive the next writeability event. + * @param eh The event handler to change + * @param event_mask The changes to make to the wait state */ - virtual void WantWrite(EventHandler* eh) = 0; + void ChangeEventMask(EventHandler* eh, int event_mask); - /** Returns the maximum number of file descriptors - * you may store in the socket engine at any one time. + /** Returns the highest file descriptor you may store in the socket engine * @return The maximum fd value */ - virtual int GetMaxFds(); + inline int GetMaxFds() const { return MAX_DESCRIPTORS; } - /** Returns the number of file descriptor slots - * which are available for storing fds. - * @return The number of remaining fd's + /** Returns the number of file descriptors being queried + * @return The set size */ - virtual int GetRemainingFds(); + inline int GetUsedFds() const { return CurrentSetSize; } /** Delete an event handler from the engine. * This function call deletes an EventHandler @@ -229,27 +326,29 @@ public: */ virtual EventHandler* GetRef(int fd); - /** Waits for events and dispatches them to handlers. - * Please note that this doesnt wait long, only - * a couple of milliseconds. It returns the number of - * events which occured during this call. - * This method will dispatch events to their handlers - * by calling their EventHandler::HandleEvent() - * methods with the neccessary EventType value. + /** Waits for events and dispatches them to handlers. Please note that + * this doesn't wait long, only a couple of milliseconds. It returns the + * number of events which occurred during this call. This method will + * dispatch events to their handlers by calling their + * EventHandler::HandleEvent() methods with the necessary EventType + * value. * @return The number of events which have occured. */ - virtual int DispatchEvents(); + virtual int DispatchEvents() = 0; + + /** Dispatch trial reads and writes. This causes the actual socket I/O + * to happen when writes have been pre-buffered. + */ + virtual void DispatchTrialWrites(); - /** Returns the socket engines name. - * This returns the name of the engine for use - * in /VERSION responses. + /** Returns the socket engines name. This returns the name of the + * engine for use in /VERSION responses. * @return The socket engine name */ virtual std::string GetName() = 0; - /** Returns true if the file descriptors in the - * given event handler are within sensible ranges - * which can be handled by the socket engine. + /** Returns true if the file descriptors in the given event handler are + * within sensible ranges which can be handled by the socket engine. */ virtual bool BoundsCheckFd(EventHandler* eh); diff --git a/include/socketengines/socketengine_epoll.h b/include/socketengines/socketengine_epoll.h index 3e08179ae..08a9b1bcc 100644 --- a/include/socketengines/socketengine_epoll.h +++ b/include/socketengines/socketengine_epoll.h @@ -31,6 +31,7 @@ private: /** These are used by epoll() to hold socket events */ struct epoll_event* events; + int EngineHandle; public: /** Create a new EPollEngine */ @@ -38,13 +39,11 @@ public: /** Delete an EPollEngine */ virtual ~EPollEngine(); - virtual bool AddFd(EventHandler* eh, bool writeFirst = false); - virtual int GetMaxFds(); - virtual int GetRemainingFds(); + virtual bool AddFd(EventHandler* eh, int event_mask); + virtual void OnSetEvent(EventHandler* eh, int old_mask, int new_mask); virtual bool DelFd(EventHandler* eh, bool force = false); virtual int DispatchEvents(); virtual std::string GetName(); - virtual void WantWrite(EventHandler* eh); }; /** Creates a SocketEngine diff --git a/include/socketengines/socketengine_iocp.h b/include/socketengines/socketengine_iocp.h index dd34227d3..0f4f6bdf5 100644 --- a/include/socketengines/socketengine_iocp.h +++ b/include/socketengines/socketengine_iocp.h @@ -131,7 +131,7 @@ public: * @param eh EventHandler to add * @return True if success, false if no room */ - bool AddFd(EventHandler* eh, bool writeFirst = false); + bool AddFd(EventHandler* eh, int event_mask); /** Gets the maximum number of file descriptors that this engine can handle. * @return The number of file descriptors @@ -166,10 +166,7 @@ public: */ std::string GetName(); - /** Queues a Write event on the specified event handler. - * @param eh EventHandler that needs data sent on - */ - void WantWrite(EventHandler* eh); + void OnSetEvent(EventHandler* eh, int old_mask, int new_mask); /** Posts a completion event on the specified socket. * @param eh EventHandler for message diff --git a/include/socketengines/socketengine_kqueue.h b/include/socketengines/socketengine_kqueue.h index a09e93553..80993e2d4 100644 --- a/include/socketengines/socketengine_kqueue.h +++ b/include/socketengines/socketengine_kqueue.h @@ -42,9 +42,8 @@ public: /** Delete a KQueueEngine */ virtual ~KQueueEngine(); - virtual bool AddFd(EventHandler* eh, bool writeFirst = false); - virtual int GetMaxFds(); - virtual int GetRemainingFds(); + bool AddFd(EventHandler* eh, int event_mask); + void OnSetEvent(EventHandler* eh, int old_mask, int new_mask); virtual bool DelFd(EventHandler* eh, bool force = false); virtual int DispatchEvents(); virtual std::string GetName(); diff --git a/include/socketengines/socketengine_poll.h b/include/socketengines/socketengine_poll.h index 725ad225c..a99ce915c 100644 --- a/include/socketengines/socketengine_poll.h +++ b/include/socketengines/socketengine_poll.h @@ -53,14 +53,12 @@ public: /** Delete a PollEngine */ virtual ~PollEngine(); - virtual bool AddFd(EventHandler* eh, bool writeFirst = false); + virtual bool AddFd(EventHandler* eh, int event_mask); + virtual void OnSetEvent(EventHandler* eh, int old_mask, int new_mask); virtual EventHandler* GetRef(int fd); - virtual int GetMaxFds(); - virtual int GetRemainingFds(); virtual bool DelFd(EventHandler* eh, bool force = false); virtual int DispatchEvents(); virtual std::string GetName(); - virtual void WantWrite(EventHandler* eh); }; /** Creates a SocketEngine diff --git a/include/socketengines/socketengine_ports.h b/include/socketengines/socketengine_ports.h index 03df456f2..be8c4bad2 100644 --- a/include/socketengines/socketengine_ports.h +++ b/include/socketengines/socketengine_ports.h @@ -42,9 +42,8 @@ public: /** Delete a PortsEngine */ virtual ~PortsEngine(); - virtual bool AddFd(EventHandler* eh, bool writeFirst = false); - virtual int GetMaxFds(); - virtual int GetRemainingFds(); + virtual bool AddFd(EventHandler* eh, int event_mask); + void OnSetEvent(EventHandler* eh, int old_event, int new_event); virtual bool DelFd(EventHandler* eh, bool force = false); virtual int DispatchEvents(); virtual std::string GetName(); diff --git a/include/socketengines/socketengine_select.h b/include/socketengines/socketengine_select.h index a1993f19f..5a4aee86b 100644 --- a/include/socketengines/socketengine_select.h +++ b/include/socketengines/socketengine_select.h @@ -28,32 +28,18 @@ */ class SelectEngine : public SocketEngine { -private: - /** Because select() does not track an fd list for us between calls, we have one of our own - */ - std::set fds; - /** List of writeable ones (WantWrite()) - */ - std::vector writeable; - /** The read set and write set, populated before each call to select(). - */ - fd_set wfdset, rfdset, errfdset; - public: /** Create a new SelectEngine - * @param Instance The creator of this object */ SelectEngine(); /** Delete a SelectEngine */ virtual ~SelectEngine(); - virtual bool AddFd(EventHandler* eh, bool writeFirst = false); - virtual int GetMaxFds(); - virtual int GetRemainingFds(); + virtual bool AddFd(EventHandler* eh, int event_mask); virtual bool DelFd(EventHandler* eh, bool force = false); + void OnSetEvent(EventHandler* eh, int, int); virtual int DispatchEvents(); virtual std::string GetName(); - virtual void WantWrite(EventHandler* eh); }; /** Creates a SocketEngine diff --git a/src/dns.cpp b/src/dns.cpp index 94a01e64c..3356700ef 100644 --- a/src/dns.cpp +++ b/src/dns.cpp @@ -369,7 +369,7 @@ void DNS::Rehash() /* Hook the descriptor into the socket engine */ if (ServerInstance && ServerInstance->SE) { - if (!ServerInstance->SE->AddFd(this)) + if (!ServerInstance->SE->AddFd(this, FD_WANT_POLL_READ | FD_WANT_NO_WRITE)) { ServerInstance->Logs->Log("RESOLVER",DEFAULT,"Internal error starting DNS - hostnames will NOT resolve."); ServerInstance->SE->Shutdown(this, 2); diff --git a/src/inspircd.cpp b/src/inspircd.cpp index f28324d16..0fc3535e2 100644 --- a/src/inspircd.cpp +++ b/src/inspircd.cpp @@ -823,6 +823,7 @@ int InspIRCd::Run() * This will cause any read or write events to be * dispatched to their handlers. */ + this->SE->DispatchTrialWrites(); this->SE->DispatchEvents(); /* if any users were quit, take them out */ diff --git a/src/inspsocket.cpp b/src/inspsocket.cpp index 907acea67..6348d7982 100644 --- a/src/inspsocket.cpp +++ b/src/inspsocket.cpp @@ -28,7 +28,7 @@ BufferedSocket::BufferedSocket(int newfd) this->fd = newfd; this->state = I_CONNECTED; if (fd > -1) - ServerInstance->SE->AddFd(this); + ServerInstance->SE->AddFd(this, FD_WANT_FAST_READ | FD_WANT_EDGE_WRITE); } void BufferedSocket::DoConnect(const std::string &ipaddr, int aport, unsigned long maxtime, const std::string &connectbindip) @@ -97,7 +97,7 @@ BufferedSocketError BufferedSocket::BeginConnect(const irc::sockets::sockaddrs& this->state = I_CONNECTING; - if (!ServerInstance->SE->AddFd(this, true)) + if (!ServerInstance->SE->AddFd(this, FD_WANT_NO_READ | FD_WANT_POLL_WRITE)) return I_ERR_NOMOREFDS; this->Timeout = new SocketTimeout(this->GetFd(), this, timeout, ServerInstance->Time()); @@ -178,18 +178,35 @@ void StreamSocket::DoRead() { char* ReadBuffer = ServerInstance->GetReadBuffer(); int n = recv(fd, ReadBuffer, ServerInstance->Config->NetBufferSize, 0); - if (n > 0) + if (n == ServerInstance->Config->NetBufferSize) { + ServerInstance->SE->ChangeEventMask(this, FD_WANT_FAST_READ | FD_ADD_TRIAL_READ); + recvq.append(ReadBuffer, n); + OnDataReady(); + } + else if (n > 0) + { + ServerInstance->SE->ChangeEventMask(this, FD_WANT_FAST_READ); recvq.append(ReadBuffer, n); OnDataReady(); } else if (n == 0) { error = "Connection closed"; + ServerInstance->SE->ChangeEventMask(this, FD_WANT_NO_READ | FD_WANT_NO_WRITE); + } + else if (errno == EAGAIN) + { + ServerInstance->SE->ChangeEventMask(this, FD_WANT_FAST_READ | FD_READ_WILL_BLOCK); } - else if (errno != EAGAIN && errno != EINTR) + else if (errno == EINTR) + { + ServerInstance->SE->ChangeEventMask(this, FD_WANT_FAST_READ | FD_ADD_TRIAL_READ); + } + else { error = strerror(errno); + ServerInstance->SE->ChangeEventMask(this, FD_WANT_NO_READ | FD_WANT_NO_WRITE); } } } @@ -255,58 +272,90 @@ void StreamSocket::DoWrite() } else { - // Prepare a writev() call to write all buffers efficiently - int bufcount = sendq.size(); + bool again = true; + while (again) + { + again = false; + + // Prepare a writev() call to write all buffers efficiently + int bufcount = sendq.size(); - // cap the number of buffers at IOV_MAX - if (bufcount > IOV_MAX) - bufcount = IOV_MAX; + // cap the number of buffers at IOV_MAX + if (bufcount > IOV_MAX) + { + bufcount = IOV_MAX; + again = true; + } - iovec* iovecs = new iovec[bufcount]; - for(int i=0; i < bufcount; i++) - { - iovecs[i].iov_base = const_cast(sendq[i].data()); - iovecs[i].iov_len = sendq[i].length(); - } - int rv = writev(fd, iovecs, bufcount); - delete[] iovecs; - if (rv == (int)sendq_len) - { - // it's our lucky day, everything got written out. Fast cleanup. - sendq_len = 0; - sendq.clear(); - } - else if (rv > 0) - { - // Partial write. Clean out strings from the sendq - sendq_len -= rv; - while (rv > 0 && !sendq.empty()) + iovec* iovecs = new iovec[bufcount]; + for(int i=0; i < bufcount; i++) { - std::string& front = sendq.front(); - if (front.length() < (size_t)rv) - { - // this string got fully written out - rv -= front.length(); - sendq.pop_front(); - } - else + iovecs[i].iov_base = const_cast(sendq[i].data()); + iovecs[i].iov_len = sendq[i].length(); + } + int rv = writev(fd, iovecs, bufcount); + delete[] iovecs; + + if (rv == (int)sendq_len) + { + // it's our lucky day, everything got written out. Fast cleanup. + // This won't ever happen if the number of buffers got capped. + sendq_len = 0; + sendq.clear(); + } + else if (rv > 0) + { + // Partial write. Clean out strings from the sendq + sendq_len -= rv; + while (rv > 0 && !sendq.empty()) { - // stopped in the middle of this string - front = front.substr(rv); - rv = 0; + std::string& front = sendq.front(); + if (front.length() < (size_t)rv) + { + // this string got fully written out + rv -= front.length(); + sendq.pop_front(); + } + else + { + // stopped in the middle of this string + front = front.substr(rv); + rv = 0; + } } } + else if (rv == 0) + { + error = "Connection closed"; + } + else if (errno == EAGAIN) + { + again = false; + } + else if (errno == EINTR) + { + again = true; + } + else + { + error = strerror(errno); + } } - else if (rv == 0) + if (!error.empty()) { - error = "Connection closed"; + // error - kill all events + ServerInstance->SE->ChangeEventMask(this, FD_WANT_NO_READ | FD_WANT_NO_WRITE); } - else if (errno != EAGAIN && errno != EINTR) + else if (sendq_len) { - error = strerror(errno); + // writes have blocked, we can use FAST_WRITE to find when they unblock + ServerInstance->SE->ChangeEventMask(this, FD_WANT_FAST_WRITE | FD_WRITE_WILL_BLOCK); + } + else + { + // writes are done, we can use EDGE_WRITE to stop asking for write + ServerInstance->SE->ChangeEventMask(this, FD_WANT_EDGE_WRITE); } - if (sendq_len && error.empty()) - ServerInstance->SE->WantWrite(this); } } @@ -318,18 +367,12 @@ void StreamSocket::WriteData(const std::string &data) data.c_str()); return; } - bool newWrite = sendq.empty() && !data.empty(); /* Append the data to the back of the queue ready for writing */ sendq.push_back(data); sendq_len += data.length(); - if (newWrite) - { - // TODO perhaps we should try writing first, before asking SE about writes? - // DoWrite(); - ServerInstance->SE->WantWrite(this); - } + ServerInstance->SE->ChangeEventMask(this, FD_ADD_TRIAL_WRITE); } void SocketTimeout::Tick(time_t) diff --git a/src/listensocket.cpp b/src/listensocket.cpp index 823cb9eca..dfd2f11f7 100644 --- a/src/listensocket.cpp +++ b/src/listensocket.cpp @@ -57,7 +57,7 @@ ListenSocketBase::ListenSocketBase(InspIRCd* Instance, int port, const std::stri else { Instance->SE->NonBlocking(this->fd); - Instance->SE->AddFd(this); + Instance->SE->AddFd(this, FD_WANT_POLL_READ | FD_WANT_NO_WRITE); } } } diff --git a/src/modules/extra/m_ssl_gnutls.cpp b/src/modules/extra/m_ssl_gnutls.cpp index a7175005f..e72666062 100644 --- a/src/modules/extra/m_ssl_gnutls.cpp +++ b/src/modules/extra/m_ssl_gnutls.cpp @@ -439,7 +439,7 @@ class ModuleSSLGnuTLS : public Module } else if (session->status == ISSL_HANDSHAKING_WRITE) { - MakePollWrite(user); + ServerInstance->SE->ChangeEventMask(user, FD_WANT_NO_READ | FD_WANT_POLL_WRITE); return 0; } @@ -515,17 +515,18 @@ class ModuleSSLGnuTLS : public Module if (ret == (int)sendq.length()) { + ServerInstance->SE->ChangeEventMask(user, FD_WANT_NO_WRITE); return 1; } else if (ret > 0) { sendq = sendq.substr(ret); - MakePollWrite(user); + ServerInstance->SE->ChangeEventMask(user, FD_WANT_POLL_WRITE); return 0; } else if (ret == GNUTLS_E_AGAIN || ret == GNUTLS_E_INTERRUPTED) { - MakePollWrite(user); + ServerInstance->SE->ChangeEventMask(user, FD_WANT_POLL_WRITE); return 0; } else if (ret == 0) @@ -559,12 +560,13 @@ class ModuleSSLGnuTLS : public Module { // gnutls_handshake() wants to read() again. session->status = ISSL_HANDSHAKING_READ; + ServerInstance->SE->ChangeEventMask(user, FD_WANT_POLL_READ | FD_WANT_NO_WRITE); } else { // gnutls_handshake() wants to write() again. session->status = ISSL_HANDSHAKING_WRITE; - MakePollWrite(user); + ServerInstance->SE->ChangeEventMask(user, FD_WANT_NO_READ | FD_WANT_POLL_WRITE); } } else @@ -583,7 +585,7 @@ class ModuleSSLGnuTLS : public Module VerifyCertificate(session,user); // Finish writing, if any left - MakePollWrite(user); + ServerInstance->SE->ChangeEventMask(user, FD_WANT_POLL_READ | FD_WANT_NO_WRITE | FD_ADD_TRIAL_WRITE); return true; } @@ -605,11 +607,6 @@ class ModuleSSLGnuTLS : public Module } } - void MakePollWrite(EventHandler* eh) - { - ServerInstance->SE->WantWrite(eh); - } - void CloseSession(issl_session* session) { if(session->sess) diff --git a/src/modules/extra/m_ssl_openssl.cpp b/src/modules/extra/m_ssl_openssl.cpp index 8a2737f0e..e72bdc816 100644 --- a/src/modules/extra/m_ssl_openssl.cpp +++ b/src/modules/extra/m_ssl_openssl.cpp @@ -36,7 +36,6 @@ enum issl_status { ISSL_NONE, ISSL_HANDSHAKING, ISSL_OPEN }; -enum issl_io_status { ISSL_WRITE, ISSL_READ }; static bool SelfSigned = false; @@ -54,20 +53,15 @@ class issl_session : public classbase public: SSL* sess; issl_status status; - issl_io_status rstat; - issl_io_status wstat; - unsigned int inbufoffset; - char* inbuf; // Buffer OpenSSL reads into. - std::string outbuf; int fd; bool outbound; + bool data_to_write; issl_session() { outbound = false; - rstat = ISSL_READ; - wstat = ISSL_WRITE; + data_to_write = false; } }; @@ -106,10 +100,7 @@ class ModuleSSLOpenSSL : public Module public: - InspIRCd* PublicInstance; - ModuleSSLOpenSSL(InspIRCd* Me) - : Module(Me), PublicInstance(Me) { ServerInstance->Modules->PublishInterface("BufferedSocketHook", this); @@ -137,7 +128,7 @@ class ModuleSSLOpenSSL : public Module // Needs the flag as it ignores a plain /rehash OnModuleRehash(NULL,"ssl"); Implementation eventlist[] = { - I_On005Numeric, I_OnBufferFlushed, I_OnRequest, I_OnRehash, I_OnModuleRehash, I_OnPostConnect, + I_On005Numeric, I_OnRequest, I_OnRehash, I_OnModuleRehash, I_OnPostConnect, I_OnHookIO }; ServerInstance->Modules->Attach(eventlist, this, sizeof(eventlist)/sizeof(Implementation)); } @@ -350,8 +341,6 @@ class ModuleSSLOpenSSL : public Module issl_session* session = &sessions[fd]; session->fd = fd; - session->inbuf = new char[inbufsize]; - session->inbufoffset = 0; session->sess = SSL_new(ctx); session->status = ISSL_NONE; session->outbound = false; @@ -378,8 +367,6 @@ class ModuleSSLOpenSSL : public Module issl_session* session = &sessions[fd]; session->fd = fd; - session->inbuf = new char[inbufsize]; - session->inbufoffset = 0; session->sess = SSL_new(clictx); session->status = ISSL_NONE; session->outbound = true; @@ -423,19 +410,12 @@ class ModuleSSLOpenSSL : public Module if (session->status == ISSL_HANDSHAKING) { - if (session->rstat == ISSL_READ || session->wstat == ISSL_READ) - { - // The handshake isn't finished and it wants to read, try to finish it. - if (!Handshake(user, session)) - { - // Couldn't resume handshake. - if (session->status == ISSL_NONE) - return -1; - return 0; - } - } - else + // The handshake isn't finished and it wants to read, try to finish it. + if (!Handshake(user, session)) { + // Couldn't resume handshake. + if (session->status == ISSL_NONE) + return -1; return 0; } } @@ -444,26 +424,40 @@ class ModuleSSLOpenSSL : public Module if (session->status == ISSL_OPEN) { - if (session->wstat == ISSL_READ) + char* buffer = ServerInstance->GetReadBuffer(); + size_t bufsiz = ServerInstance->Config->NetBufferSize; + int ret = SSL_read(session->sess, buffer, bufsiz); + + if (ret > 0) { - if(DoWrite(user, session) == 0) - return 0; + recvq.append(buffer, ret); + return 1; } - - if (session->rstat == ISSL_READ) + else if (ret == 0) + { + // Client closed connection. + CloseSession(session); + return -1; + } + else if (ret < 0) { - int ret = DoRead(user, session); + int err = SSL_get_error(session->sess, ret); - if (ret > 0) + if (err == SSL_ERROR_WANT_READ) { - recvq.append(session->inbuf, session->inbufoffset); - session->inbufoffset = 0; - return 1; + ServerInstance->SE->ChangeEventMask(user, FD_WANT_POLL_READ); + return 0; } - else if (errno == EAGAIN || errno == EINTR) + else if (err == SSL_ERROR_WANT_WRITE) + { + ServerInstance->SE->ChangeEventMask(user, FD_WANT_NO_READ | FD_WANT_POLL_WRITE); return 0; + } else + { + CloseSession(session); return -1; + } } } @@ -473,9 +467,6 @@ class ModuleSSLOpenSSL : public Module int OnStreamSocketWrite(StreamSocket* user, std::string& buffer) { int fd = user->GetFd(); - /* Are there any possibilities of an out of range fd? Hope not, but lets be paranoid */ - if ((fd < 0) || (fd > ServerInstance->SE->GetMaxFds() - 1)) - return -1; issl_session* session = &sessions[fd]; @@ -485,136 +476,61 @@ class ModuleSSLOpenSSL : public Module return -1; } + session->data_to_write = true; + if (session->status == ISSL_HANDSHAKING) { - // The handshake isn't finished, try to finish it. - if (session->rstat == ISSL_WRITE || session->wstat == ISSL_WRITE) + if (!Handshake(user, session)) { - if (!Handshake(user, session)) - { - // Couldn't resume handshake. - if (session->status == ISSL_NONE) - return -1; - return 0; - } + // Couldn't resume handshake. + if (session->status == ISSL_NONE) + return -1; + return 0; } } - int rv = 0; - - // don't pull items into the output buffer until they are - // unlikely to block; this allows sendq exceeded to continue - // to work for SSL users. - // TODO better signaling for I/O requests so this isn't needed - if (session->outbuf.empty()) - { - session->outbuf = buffer; - rv = 1; - } - if (session->status == ISSL_OPEN) { - if (session->rstat == ISSL_WRITE) + int ret = SSL_write(session->sess, buffer.data(), buffer.size()); + if (ret == (int)buffer.length()) { - DoRead(user, session); + session->data_to_write = false; + ServerInstance->SE->ChangeEventMask(user, FD_WANT_POLL_READ | FD_WANT_NO_WRITE); + return 1; } - - if (session->wstat == ISSL_WRITE) + else if (ret > 0) { - DoWrite(user, session); - } - } - - if (rv == 0 || !session->outbuf.empty()) - ServerInstance->SE->WantWrite(user); - - return rv; - } - - int DoWrite(StreamSocket* user, issl_session* session) - { - if (!session->outbuf.size()) - return -1; - - int ret = SSL_write(session->sess, session->outbuf.data(), session->outbuf.size()); - - if (ret == 0) - { - CloseSession(session); - return 0; - } - else if (ret < 0) - { - int err = SSL_get_error(session->sess, ret); - - if (err == SSL_ERROR_WANT_WRITE) - { - session->wstat = ISSL_WRITE; - ServerInstance->SE->WantWrite(user); - return -1; - } - else if (err == SSL_ERROR_WANT_READ) - { - session->wstat = ISSL_READ; - return -1; - } - else - { - CloseSession(session); + buffer = buffer.substr(ret); + ServerInstance->SE->ChangeEventMask(user, FD_WANT_POLL_WRITE); return 0; } - } - else - { - session->outbuf = session->outbuf.substr(ret); - return ret; - } - } - - int DoRead(StreamSocket* user, issl_session* session) - { - // Is this right? Not sure if the unencrypted data is garaunteed to be the same length. - // Read into the inbuffer, offset from the beginning by the amount of data we have that insp hasn't taken yet. - - int ret = SSL_read(session->sess, session->inbuf + session->inbufoffset, inbufsize - session->inbufoffset); - - if (ret == 0) - { - // Client closed connection. - CloseSession(session); - return 0; - } - else if (ret < 0) - { - int err = SSL_get_error(session->sess, ret); - - if (err == SSL_ERROR_WANT_READ) + else if (ret == 0) { - session->rstat = ISSL_READ; - return -1; - } - else if (err == SSL_ERROR_WANT_WRITE) - { - session->rstat = ISSL_WRITE; - ServerInstance->SE->WantWrite(user); + CloseSession(session); return -1; } - else + else if (ret < 0) { - CloseSession(session); - return 0; - } - } - else - { - // Read successfully 'ret' bytes into inbuf + inbufoffset - // There are 'ret' + 'inbufoffset' bytes of data in 'inbuf' - // 'buffer' is 'count' long - - session->inbufoffset += ret; + int err = SSL_get_error(session->sess, ret); - return ret; + if (err == SSL_ERROR_WANT_WRITE) + { + ServerInstance->SE->ChangeEventMask(user, FD_WANT_POLL_WRITE); + return 0; + } + else if (err == SSL_ERROR_WANT_READ) + { + ServerInstance->SE->ChangeEventMask(user, FD_WANT_POLL_READ | FD_WANT_NO_WRITE); + return 0; + } + else + { + CloseSession(session); + return -1; + } + } } + return 0; } bool Handshake(EventHandler* user, issl_session* session) @@ -632,15 +548,14 @@ class ModuleSSLOpenSSL : public Module if (err == SSL_ERROR_WANT_READ) { - session->rstat = ISSL_READ; + ServerInstance->SE->ChangeEventMask(user, FD_WANT_POLL_READ | FD_WANT_NO_WRITE); session->status = ISSL_HANDSHAKING; return true; } else if (err == SSL_ERROR_WANT_WRITE) { - session->wstat = ISSL_WRITE; + ServerInstance->SE->ChangeEventMask(user, FD_WANT_NO_READ | FD_WANT_POLL_WRITE); session->status = ISSL_HANDSHAKING; - ServerInstance->SE->WantWrite(user); return true; } else @@ -653,13 +568,11 @@ class ModuleSSLOpenSSL : public Module else if (ret > 0) { // Handshake complete. - // This will do for setting the ssl flag...it could be done earlier if it's needed. But this seems neater. - EventHandler *u = ServerInstance->SE->GetRef(session->fd); - VerifyCertificate(session, u); + VerifyCertificate(session, user); session->status = ISSL_OPEN; - ServerInstance->SE->WantWrite(user); + ServerInstance->SE->ChangeEventMask(user, FD_WANT_POLL_READ | FD_WANT_NO_WRITE | FD_ADD_TRIAL_WRITE); return true; } @@ -672,17 +585,6 @@ class ModuleSSLOpenSSL : public Module return true; } - void OnBufferFlushed(User* user) - { - if (user->GetIOHook() == this) - { - std::string dummy; - issl_session* session = &sessions[user->GetFd()]; - if (session && session->outbuf.size()) - OnStreamSocketWrite(user, dummy); - } - } - void CloseSession(issl_session* session) { if (session->sess) @@ -691,13 +593,6 @@ class ModuleSSLOpenSSL : public Module SSL_free(session->sess); } - if (session->inbuf) - { - delete[] session->inbuf; - } - - session->outbuf.clear(); - session->inbuf = NULL; session->sess = NULL; session->status = ISSL_NONE; errno = EIO; @@ -771,8 +666,7 @@ class ModuleSSLOpenSSL : public Module static int error_callback(const char *str, size_t len, void *u) { - ModuleSSLOpenSSL* mssl = (ModuleSSLOpenSSL*)u; - mssl->PublicInstance->Logs->Log("m_ssl_openssl",DEFAULT, "SSL error: " + std::string(str, len - 1)); + ServerInstance->Logs->Log("m_ssl_openssl",DEFAULT, "SSL error: " + std::string(str, len - 1)); // // XXX: Remove this line, it causes valgrind warnings... diff --git a/src/modules/m_httpd_stats.cpp b/src/modules/m_httpd_stats.cpp index 884793e1c..01d77d806 100644 --- a/src/modules/m_httpd_stats.cpp +++ b/src/modules/m_httpd_stats.cpp @@ -87,7 +87,7 @@ class ModuleHttpStats : public Module data << "" << ServerInstance->Users->clientlist->size() << ""; data << "" << ServerInstance->chanlist->size() << ""; data << "" << ServerInstance->Users->all_opers.size() << ""; - data << "" << (ServerInstance->SE->GetMaxFds() - ServerInstance->SE->GetRemainingFds()) << "" << ServerInstance->SE->GetMaxFds() << "" << ServerInstance->SE->GetName() << ""; + data << "" << (ServerInstance->SE->GetUsedFds()) << "" << ServerInstance->SE->GetMaxFds() << "" << ServerInstance->SE->GetName() << ""; time_t current_time = 0; current_time = ServerInstance->Time(); diff --git a/src/modules/m_ident.cpp b/src/modules/m_ident.cpp index 973c93f72..66bd8835a 100644 --- a/src/modules/m_ident.cpp +++ b/src/modules/m_ident.cpp @@ -127,21 +127,17 @@ class IdentRequestSocket : public EventHandler } /* Add fd to socket engine */ - if (!ServerInstance->SE->AddFd(this)) + if (!ServerInstance->SE->AddFd(this, FD_WANT_NO_READ | FD_WANT_POLL_WRITE)) { this->Close(); throw ModuleException("out of fds"); } - - /* Important: We set WantWrite immediately after connect() - * because a successful connection will trigger a writability event - */ - ServerInstance->SE->WantWrite(this); } virtual void OnConnected() { ServerInstance->Logs->Log("m_ident",DEBUG,"OnConnected()"); + ServerInstance->SE->ChangeEventMask(this, FD_WANT_POLL_READ | FD_WANT_NO_WRITE); char req[32]; diff --git a/src/socketengine.cpp b/src/socketengine.cpp index c0ae3f278..49624481f 100644 --- a/src/socketengine.cpp +++ b/src/socketengine.cpp @@ -62,40 +62,74 @@ SocketEngine::~SocketEngine() { } -bool SocketEngine::HasFd(int fd) +void SocketEngine::SetEventMask(EventHandler* eh, int mask) { - if ((fd < 0) || (fd > MAX_DESCRIPTORS)) - return false; - return ref[fd]; + eh->event_mask = mask; } -EventHandler* SocketEngine::GetRef(int fd) +void SocketEngine::ChangeEventMask(EventHandler* eh, int change) { - if ((fd < 0) || (fd > MAX_DESCRIPTORS)) - return 0; - return ref[fd]; + int old_m = eh->event_mask; + int new_m = old_m; + + // if we are changing read/write type, remove the previously set bit + if (change & FD_WANT_READ_MASK) + new_m &= ~FD_WANT_READ_MASK; + if (change & FD_WANT_WRITE_MASK) + new_m &= ~FD_WANT_WRITE_MASK; + + // if adding a trial read/write, insert it into the set + if (change & FD_TRIAL_NOTE_MASK && !(old_m & FD_TRIAL_NOTE_MASK)) + trials.insert(eh->GetFd()); + + new_m |= change; + if (new_m == old_m) + return; + + eh->event_mask = new_m; + OnSetEvent(eh, old_m, new_m); } -int SocketEngine::GetMaxFds() +void SocketEngine::DispatchTrialWrites() { - return 0; + std::vector working_list; + working_list.reserve(trials.size()); + working_list.assign(trials.begin(), trials.end()); + trials.clear(); + for(unsigned int i=0; i < working_list.size(); i++) + { + int fd = working_list[i]; + EventHandler* eh = GetRef(fd); + if (!eh) + continue; + int mask = eh->event_mask; + eh->event_mask &= ~(FD_ADD_TRIAL_READ | FD_ADD_TRIAL_WRITE); + if ((mask & (FD_ADD_TRIAL_READ | FD_READ_WILL_BLOCK)) == FD_ADD_TRIAL_READ) + eh->HandleEvent(EVENT_READ, 0); + if ((mask & (FD_ADD_TRIAL_WRITE | FD_WRITE_WILL_BLOCK)) == FD_ADD_TRIAL_WRITE) + eh->HandleEvent(EVENT_WRITE, 0); + } } -int SocketEngine::GetRemainingFds() +bool SocketEngine::HasFd(int fd) { - return 0; + if ((fd < 0) || (fd > GetMaxFds())) + return false; + return ref[fd]; } -int SocketEngine::DispatchEvents() +EventHandler* SocketEngine::GetRef(int fd) { - return 0; + if ((fd < 0) || (fd > GetMaxFds())) + return 0; + return ref[fd]; } bool SocketEngine::BoundsCheckFd(EventHandler* eh) { if (!eh) return false; - if ((eh->GetFd() < 0) || (eh->GetFd() > MAX_DESCRIPTORS)) + if ((eh->GetFd() < 0) || (eh->GetFd() > GetMaxFds())) return false; return true; } diff --git a/src/socketengines/socketengine_epoll.cpp b/src/socketengines/socketengine_epoll.cpp index 7fed6f250..672ff4a7b 100644 --- a/src/socketengines/socketengine_epoll.cpp +++ b/src/socketengines/socketengine_epoll.cpp @@ -18,7 +18,18 @@ EPollEngine::EPollEngine() { - MAX_DESCRIPTORS = 0; + int max = ulimit(4, 0); + if (max > 0) + { + MAX_DESCRIPTORS = max; + } + else + { + ServerInstance->Logs->Log("SOCKET", DEFAULT, "ERROR: Can't determine maximum number of open sockets!"); + printf("ERROR: Can't determine maximum number of open sockets!\n"); + ServerInstance->Exit(EXIT_STATUS_SOCKETENGINE); + } + // This is not a maximum, just a hint at the eventual number of sockets that may be polled. EngineHandle = epoll_create(GetMaxFds() / 4); @@ -26,11 +37,10 @@ EPollEngine::EPollEngine() { ServerInstance->Logs->Log("SOCKET",DEFAULT, "ERROR: Could not initialize socket engine: %s", strerror(errno)); ServerInstance->Logs->Log("SOCKET",DEFAULT, "ERROR: Your kernel probably does not have the proper features. This is a fatal error, exiting now."); - printf("ERROR: Could not initialize socket engine: %s\n", strerror(errno)); + printf("ERROR: Could not initialize epoll socket engine: %s\n", strerror(errno)); printf("ERROR: Your kernel probably does not have the proper features. This is a fatal error, exiting now.\n"); ServerInstance->Exit(EXIT_STATUS_SOCKETENGINE); } - CurrentSetSize = 0; ref = new EventHandler* [GetMaxFds()]; events = new struct epoll_event[GetMaxFds()]; @@ -45,18 +55,35 @@ EPollEngine::~EPollEngine() delete[] events; } -bool EPollEngine::AddFd(EventHandler* eh, bool writeFirst) +static int mask_to_epoll(int event_mask) { - int fd = eh->GetFd(); - if ((fd < 0) || (fd > GetMaxFds() - 1)) + int rv = 0; + if (event_mask & (FD_WANT_POLL_READ | FD_WANT_POLL_WRITE)) { - ServerInstance->Logs->Log("SOCKET",DEBUG,"AddFd out of range: (fd: %d, max: %d)", fd, GetMaxFds()); - return false; + // we need to use standard polling on this FD + if (event_mask & (FD_WANT_POLL_READ | FD_WANT_FAST_READ)) + rv |= EPOLLIN; + if (event_mask & (FD_WANT_POLL_WRITE | FD_WANT_FAST_WRITE)) + rv |= EPOLLOUT; } + else + { + // we can use edge-triggered polling on this FD + rv = EPOLLET; + if (event_mask & (FD_WANT_FAST_READ | FD_WANT_EDGE_READ)) + rv |= EPOLLIN; + if (event_mask & (FD_WANT_FAST_WRITE | FD_WANT_EDGE_WRITE)) + rv |= EPOLLOUT; + } + return rv; +} - if (GetRemainingFds() <= 1) +bool EPollEngine::AddFd(EventHandler* eh, int event_mask) +{ + int fd = eh->GetFd(); + if ((fd < 0) || (fd > GetMaxFds() - 1)) { - ServerInstance->Logs->Log("SOCKET",DEBUG,"No remaining FDs cannot add fd: %d", fd); + ServerInstance->Logs->Log("SOCKET",DEBUG,"AddFd out of range: (fd: %d, max: %d)", fd, GetMaxFds()); return false; } @@ -68,7 +95,7 @@ bool EPollEngine::AddFd(EventHandler* eh, bool writeFirst) struct epoll_event ev; memset(&ev,0,sizeof(ev)); - ev.events = writeFirst ? EPOLLOUT : EPOLLIN; + ev.events = mask_to_epoll(event_mask); ev.data.fd = fd; int i = epoll_ctl(EngineHandle, EPOLL_CTL_ADD, fd, &ev); if (i < 0) @@ -80,20 +107,24 @@ bool EPollEngine::AddFd(EventHandler* eh, bool writeFirst) ServerInstance->Logs->Log("SOCKET",DEBUG,"New file descriptor: %d", fd); ref[fd] = eh; + SocketEngine::SetEventMask(eh, event_mask); CurrentSetSize++; return true; } -void EPollEngine::WantWrite(EventHandler* eh) +void EPollEngine::OnSetEvent(EventHandler* eh, int old_mask, int new_mask) { - /** Use oneshot so that the system removes the writeable - * status for us and saves us a call. - */ - struct epoll_event ev; - memset(&ev,0,sizeof(ev)); - ev.events = EPOLLIN | EPOLLOUT; - ev.data.fd = eh->GetFd(); - epoll_ctl(EngineHandle, EPOLL_CTL_MOD, eh->GetFd(), &ev); + int old_events = mask_to_epoll(old_mask); + int new_events = mask_to_epoll(new_mask); + if (old_events != new_events) + { + // ok, we actually have something to tell the kernel about + struct epoll_event ev; + memset(&ev,0,sizeof(ev)); + ev.events = new_events; + ev.data.fd = eh->GetFd(); + epoll_ctl(EngineHandle, EPOLL_CTL_MOD, eh->GetFd(), &ev); + } } bool EPollEngine::DelFd(EventHandler* eh, bool force) @@ -117,37 +148,12 @@ bool EPollEngine::DelFd(EventHandler* eh, bool force) } ref[fd] = NULL; - CurrentSetSize--; ServerInstance->Logs->Log("SOCKET",DEBUG,"Remove file descriptor: %d", fd); + CurrentSetSize--; return true; } -int EPollEngine::GetMaxFds() -{ - if (MAX_DESCRIPTORS) - return MAX_DESCRIPTORS; - - int max = ulimit(4, 0); - if (max > 0) - { - MAX_DESCRIPTORS = max; - return max; - } - else - { - ServerInstance->Logs->Log("SOCKET", DEFAULT, "ERROR: Can't determine maximum number of open sockets!"); - printf("ERROR: Can't determine maximum number of open sockets!\n"); - ServerInstance->Exit(EXIT_STATUS_SOCKETENGINE); - } - return 0; -} - -int EPollEngine::GetRemainingFds() -{ - return GetMaxFds() - CurrentSetSize; -} - int EPollEngine::DispatchEvents() { socklen_t codesize = sizeof(int); @@ -158,11 +164,13 @@ int EPollEngine::DispatchEvents() for (int j = 0; j < i; j++) { + EventHandler* eh = ref[events[j].data.fd]; + if (!eh) + continue; if (events[j].events & EPOLLHUP) { ErrorEvents++; - if (ref[events[j].data.fd]) - ref[events[j].data.fd]->HandleEvent(EVENT_ERROR, 0); + eh->HandleEvent(EVENT_ERROR, 0); continue; } if (events[j].events & EPOLLERR) @@ -171,26 +179,20 @@ int EPollEngine::DispatchEvents() /* Get error number */ if (getsockopt(events[j].data.fd, SOL_SOCKET, SO_ERROR, &errcode, &codesize) < 0) errcode = errno; - if (ref[events[j].data.fd]) - ref[events[j].data.fd]->HandleEvent(EVENT_ERROR, errcode); + eh->HandleEvent(EVENT_ERROR, errcode); continue; } - if (events[j].events & EPOLLOUT) + if (events[j].events & EPOLLIN) { - WriteEvents++; - struct epoll_event ev; - memset(&ev,0,sizeof(ev)); - ev.events = EPOLLIN; - ev.data.fd = events[j].data.fd; - epoll_ctl(EngineHandle, EPOLL_CTL_MOD, events[j].data.fd, &ev); - if (ref[events[j].data.fd]) - ref[events[j].data.fd]->HandleEvent(EVENT_WRITE); + ReadEvents++; + SetEventMask(eh, eh->GetEventMask() & ~FD_READ_WILL_BLOCK); + eh->HandleEvent(EVENT_READ); } - else + if (events[j].events & EPOLLOUT) { - ReadEvents++; - if (ref[events[j].data.fd]) - ref[events[j].data.fd]->HandleEvent(EVENT_READ); + WriteEvents++; + SetEventMask(eh, eh->GetEventMask() & ~FD_WRITE_WILL_BLOCK); + eh->HandleEvent(EVENT_WRITE); } } diff --git a/src/socketengines/socketengine_iocp.cpp b/src/socketengines/socketengine_iocp.cpp index 3c3181909..e09fb4d0a 100644 --- a/src/socketengines/socketengine_iocp.cpp +++ b/src/socketengines/socketengine_iocp.cpp @@ -33,7 +33,6 @@ IOCPEngine::IOCPEngine() /* Null variables out. */ CurrentSetSize = 0; - EngineHandle = 0; MAX_DESCRIPTORS = 10240; ref = new EventHandler* [10240]; memset(ref, 0, sizeof(EventHandler*) * MAX_DESCRIPTORS); @@ -47,7 +46,7 @@ IOCPEngine::~IOCPEngine() delete[] ref; } -bool IOCPEngine::AddFd(EventHandler* eh, bool writeFirst) +bool IOCPEngine::AddFd(EventHandler* eh, int event_mask) { /* Does it at least look valid? */ if (!eh) @@ -92,7 +91,7 @@ bool IOCPEngine::AddFd(EventHandler* eh, bool writeFirst) ServerInstance->Logs->Log("SOCKET",DEBUG, "New fake fd: %u, real fd: %u, address 0x%p", *fake_fd, eh->GetFd(), eh); /* post a write event if there is data to be written */ - if(writeFirst) + if (event_mask & (FD_WANT_POLL_WRITE | FD_WANT_FAST_WRITE)) WantWrite(eh); /* we're all good =) */ @@ -107,6 +106,7 @@ bool IOCPEngine::AddFd(EventHandler* eh, bool writeFirst) } ++CurrentSetSize; + SocketEngine::SetEventMask(eh, event_mask); ref[*fake_fd] = eh; return true; @@ -171,7 +171,7 @@ bool IOCPEngine::DelFd(EventHandler* eh, bool force /* = false */) return true; } -void IOCPEngine::WantWrite(EventHandler* eh) +void IOCPEngine::OnSetEvent(EventHandler* eh, int old_mask, int new_mask) { if (!eh) return; @@ -183,7 +183,7 @@ void IOCPEngine::WantWrite(EventHandler* eh) return; /* Post event - write begin */ - if(!eh->GetExt("windows_writeevent", m_writeEvent)) + if((new_mask & (FD_WANT_POLL_WRITE | FD_WANT_FAST_WRITE)) && !eh->GetExt("windows_writeevent", m_writeEvent)) { ULONG_PTR completion_key = (ULONG_PTR)*fake_fd; Overlapped * ov = new Overlapped(SOCKET_IO_EVENT_WRITE_READY, 0); @@ -315,6 +315,7 @@ int IOCPEngine::DispatchEvents() { WriteEvents++; eh->Shrink("windows_writeevent"); + SetEventMask(eh, eh->GetEventMask() & ~FD_WRITE_WILL_BLOCK); eh->HandleEvent(EVENT_WRITE, 0); } break; @@ -322,6 +323,7 @@ int IOCPEngine::DispatchEvents() case SOCKET_IO_EVENT_READ_READY: { ReadEvents++; + SetEventMask(eh, eh->GetEventMask() & ~FD_READ_WILL_BLOCK); if(ov->m_params) { // if we had params, it means we are a udp socket with a udp_overlap pointer in this long. diff --git a/src/socketengines/socketengine_kqueue.cpp b/src/socketengines/socketengine_kqueue.cpp index cbe3e959d..c9734e85d 100644 --- a/src/socketengines/socketengine_kqueue.cpp +++ b/src/socketengines/socketengine_kqueue.cpp @@ -54,16 +54,13 @@ KQueueEngine::~KQueueEngine() delete[] ke_list; } -bool KQueueEngine::AddFd(EventHandler* eh, bool writeFirst) +bool KQueueEngine::AddFd(EventHandler* eh, int event_mask) { int fd = eh->GetFd(); if ((fd < 0) || (fd > GetMaxFds() - 1)) return false; - if (GetRemainingFds() <= 1) - return false; - if (ref[fd]) return false; @@ -79,12 +76,13 @@ bool KQueueEngine::AddFd(EventHandler* eh, bool writeFirst) return false; } - if (writeFirst) { + if (event_mask & (FD_WANT_POLL_WRITE | FD_WANT_FAST_WRITE)) { // ...and sometimes want to write WantWrite(eh); } ref[fd] = eh; + SocketEngine::SetEventMask(eh, event_mask); CurrentSetSize++; ServerInstance->Logs->Log("SOCKET",DEBUG,"New file descriptor: %d", fd); @@ -126,38 +124,41 @@ bool KQueueEngine::DelFd(EventHandler* eh, bool force) return true; } -void KQueueEngine::WantWrite(EventHandler* eh) +void KQueueEngine::OnSetEvent(EventHandler* eh, int old_mask, int new_mask) { - struct kevent ke; - // EV_ONESHOT since we only ever want one write event - EV_SET(&ke, eh->GetFd(), EVFILT_WRITE, EV_ADD | EV_ONESHOT, 0, 0, NULL); - int i = kevent(EngineHandle, &ke, 1, 0, 0, NULL); - if (i < 0) { - ServerInstance->Logs->Log("SOCKET",DEFAULT,"Failed to mark for writing: %d %s", - eh->GetFd(), strerror(errno)); + if ((new_mask & FD_WANT_POLL_WRITE) && !(old_mask & FD_WANT_POLL_WRITE)) + { + // new poll-style write + struct kevent ke; + EV_SET(&ke, eh->GetFd(), EVFILT_WRITE, EV_ADD, 0, 0, NULL); + int i = kevent(EngineHandle, &ke, 1, 0, 0, NULL); + if (i < 0) { + ServerInstance->Logs->Log("SOCKET",DEFAULT,"Failed to mark for writing: %d %s", + eh->GetFd(), strerror(errno)); + } } -} - -int KQueueEngine::GetMaxFds() -{ - if (!MAX_DESCRIPTORS) + else if ((old_mask & FD_WANT_POLL_WRITE) && !(new_mask & FD_WANT_POLL_WRITE)) { - int mib[2], maxfiles; - size_t len; - - mib[0] = CTL_KERN; - mib[1] = KERN_MAXFILES; - len = sizeof(maxfiles); - sysctl(mib, 2, &maxfiles, &len, NULL, 0); - MAX_DESCRIPTORS = maxfiles; - return maxfiles; + // removing poll-style write + struct kevent ke; + EV_SET(&ke, eh->GetFd(), EVFILT_WRITE, EV_DELETE, 0, 0, NULL); + int i = kevent(EngineHandle, &ke, 1, 0, 0, NULL); + if (i < 0) { + ServerInstance->Logs->Log("SOCKET",DEFAULT,"Failed to mark for writing: %d %s", + eh->GetFd(), strerror(errno)); + } + } + if ((new_mask & FD_WANT_EDGE_WRITE) && !(old_mask & FD_WANT_EDGE_WRITE)) + { + // new one-shot write + struct kevent ke; + EV_SET(&ke, eh->GetFd(), EVFILT_WRITE, EV_ADD | EV_ONESHOT, 0, 0, NULL); + int i = kevent(EngineHandle, &ke, 1, 0, 0, NULL); + if (i < 0) { + ServerInstance->Logs->Log("SOCKET",DEFAULT,"Failed to mark for writing: %d %s", + eh->GetFd(), strerror(errno)); + } } - return MAX_DESCRIPTORS; -} - -int KQueueEngine::GetRemainingFds() -{ - return GetMaxFds() - CurrentSetSize; } int KQueueEngine::DispatchEvents() @@ -171,34 +172,31 @@ int KQueueEngine::DispatchEvents() for (int j = 0; j < i; j++) { + EventHandler* eh = ref[ke_list[j].ident]; + if (!eh) + continue; if (ke_list[j].flags & EV_EOF) { - /* We love you kqueue, oh yes we do *sings*! - * kqueue gives us the error number directly in the EOF state! - * Unlike smelly epoll and select, where we have to getsockopt - * to get the error, this saves us time and cpu cycles. Go BSD! - */ ErrorEvents++; - if (ref[ke_list[j].ident]) - ref[ke_list[j].ident]->HandleEvent(EVENT_ERROR, ke_list[j].fflags); + eh->HandleEvent(EVENT_ERROR, ke_list[j].fflags); continue; } if (ke_list[j].filter == EVFILT_WRITE) { - /* We only ever add write events with EV_ONESHOT, which - * means they are automatically removed once such a - * event fires, so nothing to do here. - */ - WriteEvents++; - if (ref[ke_list[j].ident]) - ref[ke_list[j].ident]->HandleEvent(EVENT_WRITE); + /* When mask is FD_WANT_FAST_WRITE, we set a one-shot + * write, so we need to clear that bit to detect when it + * set again. + */ + const int bits_to_clr = FD_WANT_FAST_WRITE | FD_WRITE_WILL_BLOCK; + SetEventMask(eh, eh->GetEventMask() & ~bits_to_clr); + eh->HandleEvent(EVENT_WRITE); } if (ke_list[j].filter == EVFILT_READ) { ReadEvents++; - if (ref[ke_list[j].ident]) - ref[ke_list[j].ident]->HandleEvent(EVENT_READ); + SetEventMask(eh, eh->GetEventMask() & ~FD_READ_WILL_BLOCK); + eh->HandleEvent(EVENT_READ); } } diff --git a/src/socketengines/socketengine_poll.cpp b/src/socketengines/socketengine_poll.cpp index 6d5ddb9f5..6f50e2798 100644 --- a/src/socketengines/socketengine_poll.cpp +++ b/src/socketengines/socketengine_poll.cpp @@ -21,9 +21,28 @@ PollEngine::PollEngine() { - // Poll requires no special setup (which is nice). CurrentSetSize = 0; - MAX_DESCRIPTORS = 0; +#ifndef __FreeBSD__ + int max = ulimit(4, 0); + if (max > 0) + { + MAX_DESCRIPTORS = max; + } + else + { + ServerInstance->Logs->Log("SOCKET", DEFAULT, "ERROR: Can't determine maximum number of open sockets: %s", strerror(errno)); + printf("ERROR: Can't determine maximum number of open sockets: %s\n", strerror(errno)); + ServerInstance->Exit(EXIT_STATUS_SOCKETENGINE); + } +#else + int mib[2]; + size_t len; + + mib[0] = CTL_KERN; + mib[1] = KERN_MAXFILES; + len = sizeof(MAX_DESCRIPTORS); + sysctl(mib, 2, &MAX_DESCRIPTORS, &len, NULL, 0); +#endif ref = new EventHandler* [GetMaxFds()]; events = new struct pollfd[GetMaxFds()]; @@ -39,7 +58,17 @@ PollEngine::~PollEngine() delete[] events; } -bool PollEngine::AddFd(EventHandler* eh, bool writeFirst) +static int mask_to_poll(int event_mask) +{ + int rv = 0; + if (event_mask & (FD_WANT_POLL_READ | FD_WANT_FAST_READ)) + rv |= POLLIN; + if (event_mask & (FD_WANT_POLL_WRITE | FD_WANT_FAST_WRITE)) + rv |= POLLOUT; + return rv; +} + +bool PollEngine::AddFd(EventHandler* eh, int event_mask) { int fd = eh->GetFd(); if ((fd < 0) || (fd > GetMaxFds() - 1)) @@ -48,12 +77,6 @@ bool PollEngine::AddFd(EventHandler* eh, bool writeFirst) return false; } - if (GetRemainingFds() <= 1) - { - ServerInstance->Logs->Log("SOCKET",DEBUG,"No remaining FDs cannot add fd: %d", fd); - return false; - } - if (fd_mappings.find(fd) != fd_mappings.end()) { ServerInstance->Logs->Log("SOCKET",DEBUG,"Attempt to add duplicate fd: %d", fd); @@ -65,16 +88,10 @@ bool PollEngine::AddFd(EventHandler* eh, bool writeFirst) fd_mappings[fd] = index; ref[index] = eh; events[index].fd = fd; - if (writeFirst) - { - events[index].events = POLLOUT; - } - else - { - events[index].events = POLLIN; - } + events[index].events = mask_to_poll(event_mask); ServerInstance->Logs->Log("SOCKET", DEBUG,"New file descriptor: %d (%d; index %d)", fd, events[fd].events, index); + SocketEngine::SetEventMask(eh, event_mask); CurrentSetSize++; return true; } @@ -87,16 +104,16 @@ EventHandler* PollEngine::GetRef(int fd) return ref[it->second]; } -void PollEngine::WantWrite(EventHandler* eh) +void PollEngine::OnSetEvent(EventHandler* eh, int old_mask, int new_mask) { std::map::iterator it = fd_mappings.find(eh->GetFd()); if (it == fd_mappings.end()) { - ServerInstance->Logs->Log("SOCKET",DEBUG,"WantWrite() on unknown fd: %d", eh->GetFd()); + ServerInstance->Logs->Log("SOCKET",DEBUG,"SetEvents() on unknown fd: %d", eh->GetFd()); return; } - events[it->second].events = POLLIN | POLLOUT; + events[it->second].events = mask_to_poll(new_mask); } bool PollEngine::DelFd(EventHandler* eh, bool force) @@ -147,48 +164,6 @@ bool PollEngine::DelFd(EventHandler* eh, bool force) return true; } -int PollEngine::GetMaxFds() -{ -#ifndef __FreeBSD__ - if (MAX_DESCRIPTORS) - return MAX_DESCRIPTORS; - - int max = ulimit(4, 0); - if (max > 0) - { - MAX_DESCRIPTORS = max; - return max; - } - else - { - MAX_DESCRIPTORS = 0; - ServerInstance->Logs->Log("SOCKET", DEFAULT, "ERROR: Can't determine maximum number of open sockets: %s", strerror(errno)); - printf("ERROR: Can't determine maximum number of open sockets: %s\n", strerror(errno)); - ServerInstance->Exit(EXIT_STATUS_SOCKETENGINE); - } - return 0; -#else - if (!MAX_DESCRIPTORS) - { - int mib[2], maxfiles; - size_t len; - - mib[0] = CTL_KERN; - mib[1] = KERN_MAXFILES; - len = sizeof(maxfiles); - sysctl(mib, 2, &maxfiles, &len, NULL, 0); - MAX_DESCRIPTORS = maxfiles; - return maxfiles; - } - return MAX_DESCRIPTORS; -#endif -} - -int PollEngine::GetRemainingFds() -{ - return MAX_DESCRIPTORS - CurrentSetSize; -} - int PollEngine::DispatchEvents() { int i = poll(events, CurrentSetSize, 1000); @@ -203,11 +178,13 @@ int PollEngine::DispatchEvents() { if (events[index].revents) processed++; + EventHandler* eh = ref[index]; + if (!eh) + continue; if (events[index].revents & POLLHUP) { - if (ref[index]) - ref[index]->HandleEvent(EVENT_ERROR, 0); + eh->HandleEvent(EVENT_ERROR, 0); continue; } @@ -219,25 +196,20 @@ int PollEngine::DispatchEvents() // Get error number if (getsockopt(fd, SOL_SOCKET, SO_ERROR, &errcode, &codesize) < 0) errcode = errno; - if (ref[index]) - ref[index]->HandleEvent(EVENT_ERROR, errcode); + eh->HandleEvent(EVENT_ERROR, errcode); continue; } - if (events[index].revents & POLLOUT) + if (events[index].revents & POLLIN) { - // Switch to wanting read again - // event handlers have to request to write again if they need it - events[index].events = POLLIN; - - if (ref[index]) - ref[index]->HandleEvent(EVENT_WRITE); + SetEventMask(eh, eh->GetEventMask() & ~FD_READ_WILL_BLOCK); + eh->HandleEvent(EVENT_READ); } - - if (events[index].revents & POLLIN) + + if (events[index].revents & POLLOUT) { - if (ref[index]) - ref[index]->HandleEvent(EVENT_READ); + SetEventMask(eh, eh->GetEventMask() & ~FD_WRITE_WILL_BLOCK); + eh->HandleEvent(EVENT_WRITE); } } } diff --git a/src/socketengines/socketengine_ports.cpp b/src/socketengines/socketengine_ports.cpp index eb08839d0..a99806fc4 100644 --- a/src/socketengines/socketengine_ports.cpp +++ b/src/socketengines/socketengine_ports.cpp @@ -19,7 +19,18 @@ PortsEngine::PortsEngine() { - MAX_DESCRIPTORS = 0; + int max = ulimit(4, 0); + if (max > 0) + { + MAX_DESCRIPTORS = max; + return max; + } + else + { + ServerInstance->Logs->Log("SOCKET", DEFAULT, "ERROR: Can't determine maximum number of open sockets!"); + printf("ERROR: Can't determine maximum number of open sockets!\n"); + ServerInstance->Exit(EXIT_STATUS_SOCKETENGINE); + } EngineHandle = port_create(); if (EngineHandle == -1) @@ -44,29 +55,38 @@ PortsEngine::~PortsEngine() delete[] events; } -bool PortsEngine::AddFd(EventHandler* eh, bool writeFirst) +static int mask_to_events(int event_mask) +{ + int rv = 0; + if (event_mask & (FD_WANT_POLL_READ | FD_WANT_FAST_READ)) + rv |= POLLRDNORM; + if (event_mask & (FD_WANT_POLL_WRITE | FD_WANT_FAST_WRITE)) + rv |= POLLWRNORM; + return rv; +} + +bool PortsEngine::AddFd(EventHandler* eh, int event_mask) { int fd = eh->GetFd(); if ((fd < 0) || (fd > GetMaxFds() - 1)) return false; - if (GetRemainingFds() <= 1) - return false; - if (ref[fd]) return false; ref[fd] = eh; - port_associate(EngineHandle, PORT_SOURCE_FD, fd, writeFirst ? POLLWRNORM : POLLRDNORM, eh); + SocketEngine::SetEventMask(eh, event_mask); + port_associate(EngineHandle, PORT_SOURCE_FD, fd, mask_to_events(event_mask), eh); ServerInstance->Logs->Log("SOCKET",DEBUG,"New file descriptor: %d", fd); CurrentSetSize++; return true; } -void PortsEngine::WantWrite(EventHandler* eh) +void PortsEngine::WantWrite(EventHandler* eh, int old_mask, int new_mask) { - port_associate(EngineHandle, PORT_SOURCE_FD, eh->GetFd(), POLLRDNORM | POLLWRNORM, eh); + if (mask_to_events(new_mask) != mask_to_events(old_mask)) + port_associate(EngineHandle, PORT_SOURCE_FD, eh->GetFd(), mask_to_events(new_mask), eh); } bool PortsEngine::DelFd(EventHandler* eh, bool force) @@ -84,31 +104,6 @@ bool PortsEngine::DelFd(EventHandler* eh, bool force) return true; } -int PortsEngine::GetMaxFds() -{ - if (MAX_DESCRIPTORS) - return MAX_DESCRIPTORS; - - int max = ulimit(4, 0); - if (max > 0) - { - MAX_DESCRIPTORS = max; - return max; - } - else - { - ServerInstance->Logs->Log("SOCKET", DEFAULT, "ERROR: Can't determine maximum number of open sockets!"); - printf("ERROR: Can't determine maximum number of open sockets!\n"); - ServerInstance->Exit(EXIT_STATUS_SOCKETENGINE); - } -#include -} - -int PortsEngine::GetRemainingFds() -{ - return GetMaxFds() - CurrentSetSize; -} - int PortsEngine::DispatchEvents() { struct timespec poll_time; @@ -132,15 +127,27 @@ int PortsEngine::DispatchEvents() case PORT_SOURCE_FD: { int fd = this->events[i].portev_object; - if (ref[fd]) + EventHandler* eh = ref[fd]; + if (eh) { - // reinsert port for next time around - port_associate(EngineHandle, PORT_SOURCE_FD, fd, POLLRDNORM, ref[fd]); - if ((this->events[i].portev_events & POLLRDNORM)) + int mask = eh->GetEventMask(); + if (events[i].portev_events & POLLWRNORM) + mask &= ~(FD_WRITE_WILL_BLOCK | FD_WANT_FAST_WRITE); + if (events[i].portev_events & POLLRDNORM) + mask &= ~FD_READ_WILL_BLOCK; + // reinsert port for next time around, pretending to be one-shot for writes + SetEventMask(ev, mask); + port_associate(EngineHandle, PORT_SOURCE_FD, fd, mask_to_events(mask), eh); + if (events[i].portev_events & POLLRDNORM) + { ReadEvents++; - else + eh->HandleEvent(EVENT_READ); + } + if (events[i].portev_events & POLLWRNORM) + { WriteEvents++; - ref[fd]->HandleEvent((this->events[i].portev_events & POLLRDNORM) ? EVENT_READ : EVENT_WRITE); + eh->HandleEvent(EVENT_WRITE); + } } } default: diff --git a/src/socketengines/socketengine_select.cpp b/src/socketengines/socketengine_select.cpp index 7f6a4e283..f089fd698 100644 --- a/src/socketengines/socketengine_select.cpp +++ b/src/socketengines/socketengine_select.cpp @@ -21,10 +21,8 @@ SelectEngine::SelectEngine() { MAX_DESCRIPTORS = FD_SETSIZE; - EngineHandle = 0; CurrentSetSize = 0; - writeable.assign(GetMaxFds(), false); ref = new EventHandler* [GetMaxFds()]; memset(ref, 0, GetMaxFds() * sizeof(EventHandler*)); } @@ -34,33 +32,23 @@ SelectEngine::~SelectEngine() delete[] ref; } -bool SelectEngine::AddFd(EventHandler* eh, bool writeFirst) +bool SelectEngine::AddFd(EventHandler* eh, int) { int fd = eh->GetFd(); if ((fd < 0) || (fd > GetMaxFds() - 1)) return false; - if (GetRemainingFds() <= 1) - return false; - if (ref[fd]) return false; - fds.insert(fd); ref[fd] = eh; + SocketEngine::SetEventMask(eh, event_mask); CurrentSetSize++; - writeable[eh->GetFd()] = writeFirst; - ServerInstance->Logs->Log("SOCKET",DEBUG,"New file descriptor: %d", fd); return true; } -void SelectEngine::WantWrite(EventHandler* eh) -{ - writeable[eh->GetFd()] = true; -} - bool SelectEngine::DelFd(EventHandler* eh, bool force) { int fd = eh->GetFd(); @@ -68,10 +56,6 @@ bool SelectEngine::DelFd(EventHandler* eh, bool force) if ((fd < 0) || (fd > GetMaxFds() - 1)) return false; - std::set::iterator t = fds.find(fd); - if (t != fds.end()) - fds.erase(t); - CurrentSetSize--; ref[fd] = NULL; @@ -79,14 +63,9 @@ bool SelectEngine::DelFd(EventHandler* eh, bool force) return true; } -int SelectEngine::GetMaxFds() +void SelectEngine::OnSetEvent(EventHandler* eh, int old_mask, int new_mask) { - return FD_SETSIZE; -} - -int SelectEngine::GetRemainingFds() -{ - return GetMaxFds() - CurrentSetSize; + // deal with it later } int SelectEngine::DispatchEvents() @@ -96,24 +75,26 @@ int SelectEngine::DispatchEvents() socklen_t codesize = sizeof(int); int errcode = 0; + fd_set wfdset, rfdset, errfdset; FD_ZERO(&wfdset); FD_ZERO(&rfdset); FD_ZERO(&errfdset); - /* Populate the select FD set (this is why select sucks compared to epoll, kqueue, IOCP) */ - for (std::set::iterator a = fds.begin(); a != fds.end(); a++) + /* Populate the select FD sets (this is why select sucks compared to epoll, kqueue, IOCP) */ + for (int i = 0; i < FD_SETSIZE; i++) { - /* Explicitly one-time writeable */ - if (writeable[*a]) - FD_SET (*a, &wfdset); - else - FD_SET (*a, &rfdset); - - /* All sockets must receive error notifications regardless */ - FD_SET (*a, &errfdset); + EventHandler* eh = ref[i]; + if (!eh) + continue; + int state = eh->GetEventMask(); + if (state & (FD_WANT_POLL_READ | FD_WANT_FAST_READ)) + FD_SET (i, &rfdset); + if (state & (FD_WANT_POLL_WRITE | FD_WANT_FAST_WRITE)) + FD_SET (i, &wfdset); + FD_SET (i, &errfdset); } - /* One second waits */ + /* One second wait */ tval.tv_sec = 1; tval.tv_usec = 0; @@ -123,16 +104,15 @@ int SelectEngine::DispatchEvents() if (sresult < 1) return 0; - std::vector copy(fds.begin(), fds.end()); - for (std::vector::iterator a = copy.begin(); a != copy.end(); a++) + for (int i = 0; i < FD_SETSIZE; i++) { - EventHandler* ev = ref[*a]; + EventHandler* ev = ref[i]; if (ev) { - if (FD_ISSET (ev->GetFd(), &errfdset)) + if (FD_ISSET (i, &errfdset)) { ErrorEvents++; - if (getsockopt(ev->GetFd(), SOL_SOCKET, SO_ERROR, (char*)&errcode, &codesize) < 0) + if (getsockopt(i, SOL_SOCKET, SO_ERROR, (char*)&errcode, &codesize) < 0) errcode = errno; ev->HandleEvent(EVENT_ERROR, errcode); @@ -145,16 +125,17 @@ int SelectEngine::DispatchEvents() * If an error event occurs above it is not worth processing the * read and write states even if set. */ - if (FD_ISSET (ev->GetFd(), &wfdset)) + if (FD_ISSET (i, &rfdset)) { - WriteEvents++; - writeable[ev->GetFd()] = false; - ev->HandleEvent(EVENT_WRITE); + ReadEvents++; + SetEventMask(eh, eh->GetEventMask() & ~FD_READ_WILL_BLOCK); + ev->HandleEvent(EVENT_READ); } - if (FD_ISSET (ev->GetFd(), &rfdset)) + if (FD_ISSET (i, &wfdset)) { - ReadEvents++; - ev->HandleEvent(EVENT_READ); + WriteEvents++; + SetEventMask(eh, eh->GetEventMask() & ~FD_WRITE_WILL_BLOCK); + ev->HandleEvent(EVENT_WRITE); } } } diff --git a/src/usermanager.cpp b/src/usermanager.cpp index 15196ac69..fe6d280c0 100644 --- a/src/usermanager.cpp +++ b/src/usermanager.cpp @@ -141,7 +141,7 @@ void UserManager::AddUser(InspIRCd* Instance, int socket, ClientListenSocket* vi } } - if (!Instance->SE->AddFd(New)) + if (!Instance->SE->AddFd(New, FD_WANT_FAST_READ | FD_WANT_EDGE_WRITE)) { Instance->Logs->Log("USERS", DEBUG,"Internal error on new connection"); this->QuitUser(New, "Internal error handling connection"); -- 2.39.2