Merge branch 'master+sendq'

6 files changed, 280 insertions, 94 deletions

diff --git a/include/inspsocket.h b/include/inspsocket.h
index 221b92cc6..53eca2e91 100644
--- a/include/inspsocket.h
+++ b/include/inspsocket.h
@@ -103,14 +103,119 @@ 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<Element> 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;
 
-	/** Private send queue. Note that individual strings may be shared
+	/** Send queue of the socket
 	 */
-	std::deque<std::string> 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;
 
@@ -126,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();
@@ -169,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/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<ssl_cert> 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 7ddd77495..89c3a71a9 100644
--- a/src/inspsocket.cpp
+++ b/src/inspsocket.cpp
@@ -201,55 +201,12 @@ void StreamSocket::DoWrite()
 
 	if (GetIOHook())
 	{
-		{
-			while (error.empty() && !sendq.empty())
-			{
-				if (sendq.size() > 1 && sendq[0].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);
-				}
-				std::string& front = sendq.front();
-				int itemlen = front.length();
-
-				{
-					int rv = GetIOHook()->OnStreamSocketWrite(this, front);
-					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
+		int rv = GetIOHook()->OnStreamSocketWrite(this);
+		if (rv < 0)
+			SetError("Write Error"); // will not overwrite a better error message
 
-						// Since it is possible that a partial write took place, adjust sendq_len
-						sendq_len = sendq_len - itemlen + front.length();
-						return;
-					}
-					else
-					{
-						SetError("Write Error"); // will not overwrite a better error message
-						return;
-					}
-				}
-			}
-		}
+		// rv == 0 means the socket has blocked. Stop trying to send data.
+		// IOHook has requested unblock notification from the socketengine.
 	}
 	else
 	{
@@ -258,7 +215,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 +230,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<char*>(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<char*>(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 +255,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 +267,7 @@ void StreamSocket::DoWrite()
 					else
 					{
 						// stopped in the middle of this string
-						front.erase(0, rv);
+						sendq.erase_front(rv);
 						rv = 0;
 					}
 				}
@@ -356,7 +313,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..df676a252 100644
--- a/src/modules/extra/m_ssl_gnutls.cpp
+++ b/src/modules/extra/m_ssl_gnutls.cpp
@@ -92,6 +92,10 @@ typedef unsigned int inspircd_gnutls_session_init_flags_t;
 typedef gnutls_connection_end_t inspircd_gnutls_session_init_flags_t;
 #endif
 
+#if INSPIRCD_GNUTLS_HAS_VERSION(3, 1, 9)
+#define INSPIRCD_GNUTLS_HAS_CORK
+#endif
+
 class RandGen : public HandlerBase2<void, char*, size_t>
 {
  public:
@@ -531,14 +535,20 @@ namespace GnuTLS
 		 */
 		Priority priority;
 
+		/** Rough max size of records to send
+		 */
+		const unsigned int outrecsize;
+
 		Profile(const std::string& profilename, const std::string& certstr, const std::string& keystr,
 				std::auto_ptr<DHParams>& DH, unsigned int mindh, const std::string& hashstr,
-				const std::string& priostr, std::auto_ptr<X509CertList>& CA, std::auto_ptr<X509CRL>& CRL)
+				const std::string& priostr, std::auto_ptr<X509CertList>& CA, std::auto_ptr<X509CRL>& CRL,
+				unsigned int recsize)
 			: name(profilename)
 			, x509cred(certstr, keystr)
 			, min_dh_bits(mindh)
 			, hash(hashstr)
 			, priority(priostr)
+			, outrecsize(recsize)
 		{
 			x509cred.SetDH(DH);
 			x509cred.SetCA(CA, CRL);
@@ -587,7 +597,13 @@ namespace GnuTLS
 					crl.reset(new X509CRL(ReadFile(filename)));
 			}
 
-			return new Profile(profilename, certstr, keystr, dh, mindh, hashstr, priostr, ca, crl);
+#ifdef INSPIRCD_GNUTLS_HAS_CORK
+			// If cork support is available outrecsize represents the (rough) max amount of data we give GnuTLS while corked
+			unsigned int outrecsize = tag->getInt("outrecsize", 2048, 512);
+#else
+			unsigned int outrecsize = tag->getInt("outrecsize", 2048, 512, 16384);
+#endif
+			return new Profile(profilename, certstr, keystr, dh, mindh, hashstr, priostr, ca, crl, outrecsize);
 		}
 
 		/** Set up the given session with the settings in this profile
@@ -605,6 +621,7 @@ namespace GnuTLS
 		const std::string& GetName() const { return name; }
 		X509Credentials& GetX509Credentials() { return x509cred; }
 		gnutls_digest_algorithm_t GetHash() const { return hash.get(); }
+		unsigned int GetOutgoingRecordSize() const { return outrecsize; }
 	};
 }
 
@@ -614,6 +631,9 @@ class GnuTLSIOHook : public SSLIOHook
 	gnutls_session_t sess;
 	issl_status status;
 	reference<GnuTLS::Profile> profile;
+#ifdef INSPIRCD_GNUTLS_HAS_CORK
+	size_t gbuffersize;
+#endif
 
 	void CloseSession()
 	{
@@ -793,6 +813,43 @@ info_done_dealloc:
 		return -1;
 	}
 
+#ifdef INSPIRCD_GNUTLS_HAS_CORK
+	int FlushBuffer(StreamSocket* sock)
+	{
+		// If GnuTLS has some data buffered, write it
+		if (gbuffersize)
+			return HandleWriteRet(sock, gnutls_record_uncork(this->sess, 0));
+		return 1;
+	}
+#endif
+
+	int HandleWriteRet(StreamSocket* sock, int ret)
+	{
+		if (ret > 0)
+		{
+#ifdef INSPIRCD_GNUTLS_HAS_CORK
+			gbuffersize -= ret;
+			if (gbuffersize)
+			{
+				SocketEngine::ChangeEventMask(sock, FD_WANT_SINGLE_WRITE);
+				return 0;
+			}
+#endif
+			return ret;
+		}
+		else if (ret == GNUTLS_E_AGAIN || ret == GNUTLS_E_INTERRUPTED || ret == 0)
+		{
+			SocketEngine::ChangeEventMask(sock, FD_WANT_SINGLE_WRITE);
+			return 0;
+		}
+		else // (ret < 0)
+		{
+			sock->SetError(gnutls_strerror(ret));
+			CloseSession();
+			return -1;
+		}
+	}
+
 	static const char* UnknownIfNULL(const char* str)
 	{
 		return str ? str : "UNKNOWN";
@@ -913,6 +970,9 @@ info_done_dealloc:
 		, sess(NULL)
 		, status(ISSL_NONE)
 		, profile(sslprofile)
+#ifdef INSPIRCD_GNUTLS_HAS_CORK
+		, gbuffersize(0)
+#endif
 	{
 		gnutls_init(&sess, flags);
 		gnutls_transport_set_ptr(sess, reinterpret_cast<gnutls_transport_ptr_t>(sock));
@@ -968,7 +1028,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,34 +1036,60 @@ info_done_dealloc:
 			return prepret;
 
 		// Session is ready for transferring application data
-		int ret = 0;
+		StreamSocket::SendQueue& sendq = user->GetSendQ();
 
+#ifdef INSPIRCD_GNUTLS_HAS_CORK
+		while (true)
 		{
-			ret = gnutls_record_send(this->sess, sendq.data(), sendq.length());
-
-			if (ret == (int)sendq.length())
-			{
-				SocketEngine::ChangeEventMask(user, FD_WANT_NO_WRITE);
-				return 1;
-			}
-			else if (ret > 0)
+			// If there is something in the GnuTLS buffer try to send() it
+			int ret = FlushBuffer(user);
+			if (ret <= 0)
+				return ret; // Couldn't flush entire buffer, retry later (or close on error)
+
+			// GnuTLS buffer is empty, if the sendq is empty as well then break to set FD_WANT_NO_WRITE
+			if (sendq.empty())
+				break;
+
+			// GnuTLS buffer is empty but sendq is not, begin sending data from the sendq
+			gnutls_record_cork(this->sess);
+			while ((!sendq.empty()) && (gbuffersize < profile->GetOutgoingRecordSize()))
 			{
-				sendq.erase(0, ret);
-				SocketEngine::ChangeEventMask(user, FD_WANT_SINGLE_WRITE);
-				return 0;
+				const StreamSocket::SendQueue::Element& elem = sendq.front();
+				gbuffersize += elem.length();
+				ret = gnutls_record_send(this->sess, elem.data(), elem.length());
+				if (ret < 0)
+				{
+					CloseSession();
+					return -1;
+				}
+				sendq.pop_front();
 			}
-			else if (ret == GNUTLS_E_AGAIN || ret == GNUTLS_E_INTERRUPTED || ret == 0)
+		}
+#else
+		int ret = 0;
+
+		while (!sendq.empty())
+		{
+			FlattenSendQueue(sendq, profile->GetOutgoingRecordSize());
+			const StreamSocket::SendQueue::Element& buffer = sendq.front();
+			ret = HandleWriteRet(user, gnutls_record_send(this->sess, buffer.data(), buffer.length()));
+
+			if (ret <= 0)
+				return ret;
+			else if (ret < (int)buffer.length())
 			{
+				sendq.erase_front(ret);
 				SocketEngine::ChangeEventMask(user, FD_WANT_SINGLE_WRITE);
 				return 0;
 			}
-			else // (ret < 0)
-			{
-				user->SetError(gnutls_strerror(ret));
-				CloseSession();
-				return -1;
-			}
+
+			// Wrote entire record, continue sending
+			sendq.pop_front();
 		}
+#endif
+
+		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 c8a035fac..b037347f1 100644
--- a/src/modules/extra/m_ssl_openssl.cpp
+++ b/src/modules/extra/m_ssl_openssl.cpp
@@ -238,6 +238,10 @@ namespace OpenSSL
 		 */
 		const bool allowrenego;
 
+		/** Rough max size of records to send
+		 */
+		const unsigned int outrecsize;
+
 		static int error_callback(const char* str, size_t len, void* u)
 		{
 			Profile* profile = reinterpret_cast<Profile*>(u);
@@ -278,6 +282,7 @@ namespace OpenSSL
 			, ctx(SSL_CTX_new(SSLv23_server_method()))
 			, clictx(SSL_CTX_new(SSLv23_client_method()))
 			, allowrenego(tag->getBool("renegotiation", true))
+			, outrecsize(tag->getInt("outrecsize", 2048, 512, 16384))
 		{
 			if ((!ctx.SetDH(dh)) || (!clictx.SetDH(dh)))
 				throw Exception("Couldn't set DH parameters");
@@ -337,6 +342,7 @@ namespace OpenSSL
 		SSL* CreateClientSession() { return clictx.CreateClientSession(); }
 		const EVP_MD* GetDigest() { return digest; }
 		bool AllowRenegotiation() const { return allowrenego; }
+		unsigned int GetOutgoingRecordSize() const { return outrecsize; }
 	};
 }
 
@@ -601,7 +607,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 +617,12 @@ class OpenSSLIOHook : public SSLIOHook
 		data_to_write = true;
 
 		// 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());
 
 #ifdef INSPIRCD_OPENSSL_ENABLE_RENEGO_DETECTION
@@ -622,13 +632,12 @@ 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)
 			{
-				buffer.erase(0, ret);
+				sendq.erase_front(ret);
 				SocketEngine::ChangeEventMask(user, FD_WANT_SINGLE_WRITE);
 				return 0;
 			}
@@ -658,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)