From 0858cdd53cd1ec01c4539e9c36ef7dd9fab4aa16 Mon Sep 17 00:00:00 2001 From: Attila Molnar Date: Sat, 6 Jun 2015 14:20:22 +0200 Subject: Add class StreamSocket::SendQueue --- include/inspsocket.h | 106 +++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 106 insertions(+) (limited to 'include') diff --git a/include/inspsocket.h b/include/inspsocket.h index 221b92cc6..43bd3e3ab 100644 --- a/include/inspsocket.h +++ b/include/inspsocket.h @@ -103,6 +103,112 @@ class CoreExport SocketTimeout : public Timer */ class CoreExport StreamSocket : public EventHandler { + public: + /** Socket send queue + */ + class SendQueue + { + public: + /** One element of the queue, a continuous buffer + */ + typedef std::string Element; + + /** Sequence container of buffers in the queue + */ + typedef std::deque Container; + + /** Container iterator + */ + typedef Container::const_iterator const_iterator; + + SendQueue() : nbytes(0) { } + + /** Return whether the queue is empty + * @return True if the queue is empty, false otherwise + */ + bool empty() const { return (nbytes == 0); } + + /** Get the number of individual buffers in the queue + * @return Number of individual buffers in the queue + */ + Container::size_type size() const { return data.size(); } + + /** Get the number of queued bytes + * @return Size in bytes of the data in the queue + */ + size_t bytes() const { return nbytes; } + + /** Get the first buffer of the queue + * @return A reference to the first buffer in the queue + */ + const Element& front() const { return data.front(); } + + /** Get an iterator to the first buffer in the queue. + * The returned iterator cannot be used to make modifications to the queue, + * for that purpose the member functions push_*(), pop_front(), erase_front() and clear() can be used. + * @return Iterator referring to the first buffer in the queue, or end() if there are no elements. + */ + const_iterator begin() const { return data.begin(); } + + /** Get an iterator to the (theoretical) buffer one past the end of the queue. + * @return Iterator referring to one element past the end of the container + */ + const_iterator end() const { return data.end(); } + + /** Remove the first buffer in the queue + */ + void pop_front() + { + nbytes -= data.front().length(); + data.pop_front(); + } + + /** Remove bytes from the beginning of the first buffer + * @param n Number of bytes to remove + */ + void erase_front(Element::size_type n) + { + nbytes -= n; + data.front().erase(0, n); + } + + /** Insert a new buffer at the beginning of the queue + * @param newdata Data to add + */ + void push_front(const Element& newdata) + { + data.push_front(newdata); + nbytes += newdata.length(); + } + + /** Insert a new buffer at the end of the queue + * @param newdata Data to add + */ + void push_back(const Element& newdata) + { + data.push_back(newdata); + nbytes += newdata.length(); + } + + /** Clear the queue + */ + void clear() + { + data.clear(); + nbytes = 0; + } + + private: + /** Private send queue. Note that individual strings may be shared. + */ + Container data; + + /** Length, in bytes, of the sendq + */ + size_t nbytes; + }; + + private: /** The IOHook that handles raw I/O for this socket, or NULL */ IOHook* iohook; -- cgit v1.2.3 From 1f0485039a276ad1c2fa3d53d284e3a87940ec77 Mon Sep 17 00:00:00 2001 From: Attila Molnar Date: Sat, 6 Jun 2015 14:31:05 +0200 Subject: Convert all code to use StreamSocket::SendQueue Let OnStreamSocketWrite see the entire sendq instead of one element at a time --- include/inspsocket.h | 13 +++++++------ include/iohook.h | 3 +-- src/inspsocket.cpp | 31 ++++++++++++------------------- src/modules/extra/m_ssl_gnutls.cpp | 10 ++++++---- src/modules/extra/m_ssl_openssl.cpp | 6 ++++-- 5 files changed, 30 insertions(+), 33 deletions(-) (limited to 'include') diff --git a/include/inspsocket.h b/include/inspsocket.h index 43bd3e3ab..53eca2e91 100644 --- a/include/inspsocket.h +++ b/include/inspsocket.h @@ -212,11 +212,10 @@ class CoreExport StreamSocket : public EventHandler /** The IOHook that handles raw I/O for this socket, or NULL */ IOHook* iohook; - /** Private send queue. Note that individual strings may be shared + /** Send queue of the socket */ - std::deque sendq; - /** Length, in bytes, of the sendq */ - size_t sendq_len; + SendQueue sendq; + /** Error - if nonempty, the socket is dead, and this is the reason. */ std::string error; @@ -232,7 +231,7 @@ class CoreExport StreamSocket : public EventHandler protected: std::string recvq; public: - StreamSocket() : iohook(NULL), sendq_len(0) {} + StreamSocket() : iohook(NULL) { } IOHook* GetIOHook() const; void AddIOHook(IOHook* hook); void DelIOHook(); @@ -275,7 +274,9 @@ class CoreExport StreamSocket : public EventHandler */ bool GetNextLine(std::string& line, char delim = '\n'); /** Useful for implementing sendq exceeded */ - inline size_t getSendQSize() const { return sendq_len; } + size_t getSendQSize() const { return sendq.size(); } + + SendQueue& GetSendQ() { return sendq; } /** * Close the socket, remove from socket engine, etc diff --git a/include/iohook.h b/include/iohook.h index ce7ca2a1b..cf27fcb0c 100644 --- a/include/iohook.h +++ b/include/iohook.h @@ -66,11 +66,10 @@ class IOHook : public classbase * Called when a hooked stream has data to write, or when the socket * engine returns it as writable * @param sock The socket in question - * @param sendq Data to send to the socket * @return 1 if the sendq has been completely emptied, 0 if there is * still data to send, and -1 if there was an error */ - virtual int OnStreamSocketWrite(StreamSocket* sock, std::string& sendq) = 0; + virtual int OnStreamSocketWrite(StreamSocket* sock) = 0; /** Called immediately before any socket is closed. When this event is called, shutdown() * has not yet been called on the socket. diff --git a/src/inspsocket.cpp b/src/inspsocket.cpp index 7ddd77495..b8f8949dd 100644 --- a/src/inspsocket.cpp +++ b/src/inspsocket.cpp @@ -204,7 +204,7 @@ void StreamSocket::DoWrite() { while (error.empty() && !sendq.empty()) { - if (sendq.size() > 1 && sendq[0].length() < 1024) + if (sendq.size() > 1 && sendq.front().length() < 1024) { // Avoid multiple repeated SSL encryption invocations // This adds a single copy of the queue, but avoids @@ -222,24 +222,18 @@ void StreamSocket::DoWrite() } sendq.push_front(tmp); } - std::string& front = sendq.front(); - int itemlen = front.length(); { - int rv = GetIOHook()->OnStreamSocketWrite(this, front); + int rv = GetIOHook()->OnStreamSocketWrite(this); if (rv > 0) { // consumed the entire string, and is ready for more - sendq_len -= itemlen; sendq.pop_front(); } else if (rv == 0) { // socket has blocked. Stop trying to send data. // IOHook has requested unblock notification from the socketengine - - // Since it is possible that a partial write took place, adjust sendq_len - sendq_len = sendq_len - itemlen + front.length(); return; } else @@ -258,7 +252,7 @@ void StreamSocket::DoWrite() return; // start out optimistic - we won't need to write any more int eventChange = FD_WANT_EDGE_WRITE; - while (error.empty() && sendq_len && eventChange == FD_WANT_EDGE_WRITE) + while (error.empty() && !sendq.empty() && eventChange == FD_WANT_EDGE_WRITE) { // Prepare a writev() call to write all buffers efficiently int bufcount = sendq.size(); @@ -273,20 +267,21 @@ void StreamSocket::DoWrite() int rv; { SocketEngine::IOVector iovecs[MYIOV_MAX]; - for (int i = 0; i < bufcount; i++) + size_t j = 0; + for (SendQueue::const_iterator i = sendq.begin(), end = i+bufcount; i != end; ++i, j++) { - iovecs[i].iov_base = const_cast(sendq[i].data()); - iovecs[i].iov_len = sendq[i].length(); - rv_max += sendq[i].length(); + const SendQueue::Element& elem = *i; + iovecs[j].iov_base = const_cast(elem.data()); + iovecs[j].iov_len = elem.length(); + rv_max += elem.length(); } rv = SocketEngine::WriteV(this, iovecs, bufcount); } - if (rv == (int)sendq_len) + if (rv == (int)sendq.bytes()) { // 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) @@ -297,10 +292,9 @@ void StreamSocket::DoWrite() // it's going to block now eventChange = FD_WANT_FAST_WRITE | FD_WRITE_WILL_BLOCK; } - sendq_len -= rv; while (rv > 0 && !sendq.empty()) { - std::string& front = sendq.front(); + const SendQueue::Element& front = sendq.front(); if (front.length() <= (size_t)rv) { // this string got fully written out @@ -310,7 +304,7 @@ void StreamSocket::DoWrite() else { // stopped in the middle of this string - front.erase(0, rv); + sendq.erase_front(rv); rv = 0; } } @@ -356,7 +350,6 @@ void StreamSocket::WriteData(const std::string &data) /* Append the data to the back of the queue ready for writing */ sendq.push_back(data); - sendq_len += data.length(); SocketEngine::ChangeEventMask(this, FD_ADD_TRIAL_WRITE); } diff --git a/src/modules/extra/m_ssl_gnutls.cpp b/src/modules/extra/m_ssl_gnutls.cpp index d33403aba..e142ead11 100644 --- a/src/modules/extra/m_ssl_gnutls.cpp +++ b/src/modules/extra/m_ssl_gnutls.cpp @@ -968,7 +968,7 @@ info_done_dealloc: } } - int OnStreamSocketWrite(StreamSocket* user, std::string& sendq) CXX11_OVERRIDE + int OnStreamSocketWrite(StreamSocket* user) CXX11_OVERRIDE { // Finish handshake if needed int prepret = PrepareIO(user); @@ -976,19 +976,21 @@ info_done_dealloc: return prepret; // Session is ready for transferring application data + StreamSocket::SendQueue& sendq = user->GetSendQ(); int ret = 0; { - ret = gnutls_record_send(this->sess, sendq.data(), sendq.length()); + const StreamSocket::SendQueue::Element& buffer = sendq.front(); + ret = gnutls_record_send(this->sess, buffer.data(), buffer.length()); - if (ret == (int)sendq.length()) + if (ret == (int)buffer.length()) { SocketEngine::ChangeEventMask(user, FD_WANT_NO_WRITE); return 1; } else if (ret > 0) { - sendq.erase(0, ret); + sendq.erase_front(ret); SocketEngine::ChangeEventMask(user, FD_WANT_SINGLE_WRITE); return 0; } diff --git a/src/modules/extra/m_ssl_openssl.cpp b/src/modules/extra/m_ssl_openssl.cpp index c8a035fac..c2a71eeca 100644 --- a/src/modules/extra/m_ssl_openssl.cpp +++ b/src/modules/extra/m_ssl_openssl.cpp @@ -601,7 +601,7 @@ class OpenSSLIOHook : public SSLIOHook } } - int OnStreamSocketWrite(StreamSocket* user, std::string& buffer) CXX11_OVERRIDE + int OnStreamSocketWrite(StreamSocket* user) CXX11_OVERRIDE { // Finish handshake if needed int prepret = PrepareIO(user); @@ -611,8 +611,10 @@ class OpenSSLIOHook : public SSLIOHook data_to_write = true; // Session is ready for transferring application data + StreamSocket::SendQueue& sendq = user->GetSendQ(); { ERR_clear_error(); + const StreamSocket::SendQueue::Element& buffer = sendq.front(); int ret = SSL_write(sess, buffer.data(), buffer.size()); #ifdef INSPIRCD_OPENSSL_ENABLE_RENEGO_DETECTION @@ -628,7 +630,7 @@ class OpenSSLIOHook : public SSLIOHook } else if (ret > 0) { - buffer.erase(0, ret); + sendq.erase_front(ret); SocketEngine::ChangeEventMask(user, FD_WANT_SINGLE_WRITE); return 0; } -- cgit v1.2.3 From d0556a2a598e207ab468b7ea4543e427205ef903 Mon Sep 17 00:00:00 2001 From: Attila Molnar Date: Sat, 6 Jun 2015 14:42:59 +0200 Subject: Call OnStreamSocketWrite() once per write event Do sendq flattening in SSL modules, move code for it into class SSLIOHook from core --- include/modules/ssl.h | 25 +++++++++++++++++++++++++ src/inspsocket.cpp | 21 --------------------- src/modules/extra/m_ssl_gnutls.cpp | 9 +++++++-- src/modules/extra/m_ssl_openssl.cpp | 11 ++++++++--- 4 files changed, 40 insertions(+), 26 deletions(-) (limited to 'include') diff --git a/include/modules/ssl.h b/include/modules/ssl.h index 0f58e0b7b..67bfc7b2e 100644 --- a/include/modules/ssl.h +++ b/include/modules/ssl.h @@ -138,6 +138,31 @@ class SSLIOHook : public IOHook */ reference certificate; + /** Reduce elements in a send queue by appending later elements to the first element until there are no more + * elements to append or a desired length is reached + * @param sendq SendQ to work on + * @param targetsize Target size of the front element + */ + static void FlattenSendQueue(StreamSocket::SendQueue& sendq, size_t targetsize) + { + if ((sendq.size() <= 1) || (sendq.front().length() >= targetsize)) + return; + + // Avoid multiple repeated SSL encryption invocations + // This adds a single copy of the queue, but avoids + // much more overhead in terms of system calls invoked + // by an IOHook. + std::string tmp; + tmp.reserve(std::min(targetsize, sendq.bytes())+1); + do + { + tmp.append(sendq.front()); + sendq.pop_front(); + } + while (!sendq.empty() && tmp.length() < targetsize); + sendq.push_front(tmp); + } + public: SSLIOHook(IOHookProvider* hookprov) : IOHook(hookprov) diff --git a/src/inspsocket.cpp b/src/inspsocket.cpp index b8f8949dd..436cbb6bb 100644 --- a/src/inspsocket.cpp +++ b/src/inspsocket.cpp @@ -202,33 +202,12 @@ void StreamSocket::DoWrite() if (GetIOHook()) { { - while (error.empty() && !sendq.empty()) { - if (sendq.size() > 1 && sendq.front().length() < 1024) - { - // Avoid multiple repeated SSL encryption invocations - // This adds a single copy of the queue, but avoids - // much more overhead in terms of system calls invoked - // by the IOHook. - // - // The length limit of 1024 is to prevent merging strings - // more than once when writes begin to block. - std::string tmp; - tmp.reserve(1280); - while (!sendq.empty() && tmp.length() < 1024) - { - tmp.append(sendq.front()); - sendq.pop_front(); - } - sendq.push_front(tmp); - } - { int rv = GetIOHook()->OnStreamSocketWrite(this); if (rv > 0) { // consumed the entire string, and is ready for more - sendq.pop_front(); } else if (rv == 0) { diff --git a/src/modules/extra/m_ssl_gnutls.cpp b/src/modules/extra/m_ssl_gnutls.cpp index f5e52b4e1..935b0ee77 100644 --- a/src/modules/extra/m_ssl_gnutls.cpp +++ b/src/modules/extra/m_ssl_gnutls.cpp @@ -987,14 +987,16 @@ info_done_dealloc: StreamSocket::SendQueue& sendq = user->GetSendQ(); int ret = 0; + while (!sendq.empty()) { + FlattenSendQueue(sendq, profile->GetOutgoingRecordSize()); const StreamSocket::SendQueue::Element& buffer = sendq.front(); ret = gnutls_record_send(this->sess, buffer.data(), buffer.length()); if (ret == (int)buffer.length()) { - SocketEngine::ChangeEventMask(user, FD_WANT_NO_WRITE); - return 1; + // Wrote entire record, continue sending + sendq.pop_front(); } else if (ret > 0) { @@ -1014,6 +1016,9 @@ info_done_dealloc: return -1; } } + + SocketEngine::ChangeEventMask(user, FD_WANT_NO_WRITE); + return 1; } void TellCiphersAndFingerprint(LocalUser* user) diff --git a/src/modules/extra/m_ssl_openssl.cpp b/src/modules/extra/m_ssl_openssl.cpp index f4a661154..b037347f1 100644 --- a/src/modules/extra/m_ssl_openssl.cpp +++ b/src/modules/extra/m_ssl_openssl.cpp @@ -618,8 +618,10 @@ class OpenSSLIOHook : public SSLIOHook // Session is ready for transferring application data StreamSocket::SendQueue& sendq = user->GetSendQ(); + while (!sendq.empty()) { ERR_clear_error(); + FlattenSendQueue(sendq, profile->GetOutgoingRecordSize()); const StreamSocket::SendQueue::Element& buffer = sendq.front(); int ret = SSL_write(sess, buffer.data(), buffer.size()); @@ -630,9 +632,8 @@ class OpenSSLIOHook : public SSLIOHook if (ret == (int)buffer.length()) { - data_to_write = false; - SocketEngine::ChangeEventMask(user, FD_WANT_POLL_READ | FD_WANT_NO_WRITE); - return 1; + // Wrote entire record, continue sending + sendq.pop_front(); } else if (ret > 0) { @@ -666,6 +667,10 @@ class OpenSSLIOHook : public SSLIOHook } } } + + data_to_write = false; + SocketEngine::ChangeEventMask(user, FD_WANT_POLL_READ | FD_WANT_NO_WRITE); + return 1; } void TellCiphersAndFingerprint(LocalUser* user) -- cgit v1.2.3