ircd::net: Develop interface for async_wait / waiting on socket.

include/ircd/net/net.h

@@ -54,6 +54,7 @@ namespace ircd::net
 #include "sock_opts.h"
 #include "open.h"
 #include "close.h"
+#include "wait.h"
 #include "read.h"
 #include "write.h"
 

include/ircd/net/socket.h

@@ -97,11 +97,15 @@ struct ircd::net::socket
 	template<class iov> size_t read_any(iov&&);  // yielding
 	template<class iov> size_t read_all(iov&&);  // yielding
 
-	// Asynchronous callback when socket ready
-	void operator()(const wait_type &, const milliseconds &timeout, ec_handler);
-	void operator()(const wait_type &, ec_handler);
+	// low level wait suite
+	void wait(const wait_opts &);
+	void wait(const wait_opts &, wait_callback_ec);
+	void wait(const wait_opts &, wait_callback_eptr);
 	bool cancel() noexcept;
 
+	// Alias to wait()
+	template<class... args> auto operator()(args&&...);
+
 	void disconnect(const close_opts &, eptr_handler);
 	void handshake(const open_opts &, eptr_handler);
 	void connect(const endpoint &, const open_opts &, eptr_handler);
@@ -133,6 +137,13 @@ class ircd::net::socket::scope_timeout
 	~scope_timeout() noexcept;
 };
 
+template<class... args>
+auto
+ircd::net::socket::operator()(args&&... a)
+{
+	return this->wait(std::forward<args>(a)...);
+}
+
 /// Yields ircd::ctx until buffers are full.
 template<class iov>
 size_t

include/ircd/net/wait.h

@@ -0,0 +1,85 @@
+/*
+ * Copyright (C) 2017 Charybdis Development Team
+ * Copyright (C) 2017 Jason Volk <jason@zemos.net>
+ *
+ * Permission to use, copy, modify, and/or distribute this software for any
+ * purpose with or without fee is hereby granted, provided that the above
+ * copyright notice and this permission notice is present in all copies.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR
+ * IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+ * WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+ * DISCLAIMED. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT,
+ * INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
+ * (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
+ * SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
+ * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT,
+ * STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING
+ * IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
+ * POSSIBILITY OF SUCH DAMAGE.
+ */
+
+#pragma once
+#define HAVE_IRCD_NET_WAIT_H
+
+namespace ircd::net
+{
+	enum class ready :uint8_t;
+	struct wait_opts extern const wait_opts_default;
+	using wait_callback_ec = std::function<void (const error_code &)>;
+	using wait_callback_eptr = std::function<void (std::exception_ptr)>;
+
+	// Asynchronous callback when ready with error_code
+	void wait(socket &, const wait_opts &, wait_callback_ec);
+
+	// Asynchronous callback when ready with exception_ptr (convenience)
+	void wait(socket &, const wait_opts &, wait_callback_eptr);
+
+	// Yields ircd::ctx for the wait condition.
+	void wait(socket &, const wait_opts & = wait_opts_default);
+
+	// Yields ircd::ctx for the wait condition. Won't throw the error_code.
+	error_code wait(nothrow_t, socket &, const wait_opts & = wait_opts_default);
+
+	// Explicit overload to return a ctx::future
+	ctx::future<void> wait(use_future_t, socket &, const wait_opts & = wait_opts_default);
+}
+
+/// Types of things to wait for the socket to have "ready"
+enum class ircd::net::ready
+:uint8_t
+{
+	ANY,       ///< Wait for anything.
+	READ,      ///< Data is available for a read().
+	WRITE,     ///< Space is free in the SNDBUF for a write().
+	ERROR,     ///< Socket has an error.
+};
+
+struct ircd::net::wait_opts
+{
+	ready type
+	{
+		ready::ANY
+	};
+
+	milliseconds timeout
+	{
+		-1
+	};
+
+	wait_opts(const ready &, const milliseconds &timeout);
+	wait_opts(const ready &);
+	wait_opts() = default;
+};
+
+inline
+ircd::net::wait_opts::wait_opts(const ready &type)
+:type{type}
+{}
+
+inline
+ircd::net::wait_opts::wait_opts(const ready &type,
+                                const milliseconds &timeout)
+:type{type}
+,timeout{timeout}
+{}

ircd/client.cc

@@ -235,9 +235,13 @@ ircd::async_recv_next(std::shared_ptr<client> client,
 	// This call returns immediately so we no longer block the current context and
 	// its stack while waiting for activity on idle connections between requests.
 
+	const net::wait_opts opts
+	{
+		net::ready::READ, timeout
+	};
+
 	auto &sock(*client->sock);
-	static const auto &op{sock.sd.wait_read};
-	sock(op, timeout, [client(std::move(client)), timeout](const error_code &ec)
+	sock(opts, [client(std::move(client)), timeout](const error_code &ec)
 	noexcept
 	{
 		// Right here this handler is executing on the main stack (not in any

ircd/net.cc

@@ -247,6 +247,77 @@ ircd::net::read_one(socket &socket,
 	return socket.read_one(buffers);
 }
 
+///////////////////////////////////////////////////////////////////////////////
+//
+// net/wait.h
+//
+
+ircd::net::wait_opts
+const ircd::net::wait_opts_default
+{
+};
+
+/// Wait for socket to become "ready" using a ctx::future.
+ircd::ctx::future<void>
+ircd::net::wait(use_future_t,
+                socket &socket,
+                const wait_opts &wait_opts)
+{
+	ctx::promise<void> p;
+	ctx::future<void> f{p};
+	wait(socket, wait_opts, [p(std::move(p))]
+	(std::exception_ptr eptr)
+	mutable
+	{
+		if(eptr)
+			p.set_exception(std::move(eptr));
+		else
+			p.set_value();
+	});
+
+	return f;
+}
+
+/// Wait for socket to become "ready"; yields ircd::ctx returning code.
+ircd::error_code
+ircd::net::wait(nothrow_t,
+                socket &socket,
+                const wait_opts &wait_opts)
+try
+{
+	wait(socket, wait_opts);
+	return {};
+}
+catch(const boost::system::system_error &e)
+{
+	return e.code();
+}
+
+/// Wait for socket to become "ready"; yields ircd::ctx; throws errors.
+void
+ircd::net::wait(socket &socket,
+                const wait_opts &wait_opts)
+{
+	socket.wait(wait_opts);
+}
+
+/// Wait for socket to become "ready"; callback with exception_ptr
+void
+ircd::net::wait(socket &socket,
+                const wait_opts &wait_opts,
+                wait_callback_eptr callback)
+{
+	socket.wait(wait_opts, std::move(callback));
+}
+
+void
+ircd::net::wait(socket &socket,
+                const wait_opts &wait_opts,
+                wait_callback_ec callback)
+{
+	socket.wait(wait_opts, std::move(callback));
+}
+
 ///////////////////////////////////////////////////////////////////////////////
 //
 // net/close.h
@@ -1271,15 +1342,15 @@ noexcept
 	return std::all_of(begin(ec), end(ec), good);
 }
 
-/// Asynchronous callback when the socket is ready
-///
-/// Overload for operator() without a timeout. see: operator()
-///
 void
-ircd::net::socket::operator()(const wait_type &type,
-                              ec_handler h)
+ircd::net::socket::wait(const wait_opts &opts,
+                        wait_callback_eptr callback)
 {
-	operator()(type, milliseconds(-1), std::move(h));
+	wait(opts, [callback(std::move(callback))]
+	(const error_code &ec)
+	{
+		callback(make_eptr(ec));
+	});
 }
 
 /// Asynchronous callback when the socket is ready
@@ -1288,39 +1359,70 @@ ircd::net::socket::operator()(const wait_type &type,
 /// for the operation of the specified type.
 ///
 void
-ircd::net::socket::operator()(const wait_type &type,
-                              const milliseconds &timeout,
-                              ec_handler callback)
+ircd::net::socket::wait(const wait_opts &opts,
+                        wait_callback_ec callback)
 {
 	auto handle
 	{
 		std::bind(&socket::handle, this, weak_from(*this), std::move(callback), ph::_1)
 	};
 
-	assert(connected(*this));
-	switch(type)
+	switch(opts.type)
 	{
-		case wait_type::wait_error:
-		case wait_type::wait_write:
-		//case wait_type::wait_read:
-		{
-			sd.async_wait(type, std::move(handle));
+		case net::ready::ERROR:
+			sd.async_wait(wait_type::wait_error, std::move(handle));
+			break;
+
+		case net::ready::WRITE:
+			sd.async_wait(wait_type::wait_write, std::move(handle));
 			break;
-		}
 
 		// The new async_wait() on linux triggers a bug which is only
 		// reproducible when serving a large number of assets: a ready status
 		// for the socket is not indicated when it ought to be, at random.
 		// This is fixed below by doing it the old way.
-		case wait_type::wait_read:
-		{
+		case net::ready::READ:
 			sd.async_receive(buffer::null_buffers, std::move(handle));
 			break;
-		}
+
+		default:
+			throw ircd::not_implemented{};
 	}
 
 	// Commit to timeout here in case exception was thrown earlier.
-	set_timeout(timeout);
+	set_timeout(opts.timeout);
+}
+
+/// Asynchronous callback when the socket is ready
+///
+/// Overload for operator() without a timeout. see: operator()
+///
+void
+ircd::net::socket::wait(const wait_opts &opts)
+{
+	const scope_timeout timeout
+	{
+		*this, opts.timeout
+	};
+
+	switch(opts.type)
+	{
+		case net::ready::ERROR:
+			sd.async_wait(wait_type::wait_error, yield_context{to_asio{}});
+			break;
+
+		case net::ready::WRITE:
+			sd.async_wait(wait_type::wait_write, yield_context{to_asio{}});
+			break;
+
+		// See bug comment in callback version
+		case net::ready::READ:
+			sd.async_receive(buffer::null_buffers, yield_context{to_asio{}});
+			break;
+
+		default:
+			throw ircd::not_implemented{};
+	}
 }
 
 void