client: redesign waiting methods

Currently, call of any waiting method (`wait`, `waitAll` and so on)
with zero timeout blocks execution until all required responses are
received. That's not great because in some scenarios users want to
check if there are any responses ready without blocking execution.

Of course, one can use `wait(1)` which sleeps for only a millisecond,
but anyway it's often unexpected that `wait(0)` sleeps forever. Let's
stick to `epoll_wait` interface - `wait(0)` is non-blocking and
`wait(-1)` blocks forever. Note that before the commit, passing `-1`
would lead to an assertion failure.

Note that we used to block forever in waiting methods by default. To
maintain this behavior, the commit updates default value of timeout
since the old default (zero) means non-blocking polling of connections
now.

Also, the commit makes `wait(0)` non-blocking, but it doesn't poll
connections. We will implement that part in a separate commit for the
sake of clarity of the patch.

Part of #111

Author: drewdzzz

src/Client/Connector.hpp

@@ -75,12 +75,12 @@ class Connector
 		    const std::string& addr, unsigned port);
 
 	int wait(Connection<BUFFER, NetProvider> &conn, rid_t future,
-		 int timeout = 0, Response<BUFFER> *result = nullptr);
+		 int timeout = -1, Response<BUFFER> *result = nullptr);
 	int waitAll(Connection<BUFFER, NetProvider> &conn,
-		    const std::vector<rid_t > &futures, int timeout = 0);
+		    const std::vector<rid_t > &futures, int timeout = -1);
 	int waitCount(Connection<BUFFER, NetProvider> &conn,
-		      size_t feature_count, int timeout = 0);
-	std::optional<Connection<BUFFER, NetProvider>> waitAny(int timeout = 0);
+		      size_t feature_count, int timeout = -1);
+	std::optional<Connection<BUFFER, NetProvider>> waitAny(int timeout = -1);
 	////////////////////////////Service interfaces//////////////////////////
 	void readyToDecode(const Connection<BUFFER, NetProvider> &conn);
 	void readyToSend(const Connection<BUFFER, NetProvider> &conn);

src/Client/EpollNetProvider.hpp

@@ -258,8 +258,8 @@ template<class BUFFER, class Stream>
 int
 EpollNetProvider<BUFFER, Stream>::wait(int timeout)
 {
-	assert(timeout >= 0);
-	if (timeout == 0)
+	assert(timeout >= -1);
+	if (timeout == -1)
 		timeout = TIMEOUT_INFINITY;
 	LOG_DEBUG("Network engine wait for ", timeout, " milliseconds");
 	/* Send pending requests. */

src/Client/LibevNetProvider.hpp

@@ -363,7 +363,7 @@ template<class BUFFER, class Stream>
 int
 LibevNetProvider<BUFFER, Stream>::wait(int timeout)
 {
-	assert(timeout >= 0);
+	assert(timeout >= -1);
 	if (timeout > 0) {
 		ev_timer_init(&m_TimeoutWatcher, &timeout_cb, timeout / MILLISECONDS, 0 /* repeat */);
 		ev_timer_start(m_Loop, &m_TimeoutWatcher);

src/Utils/Timer.hpp

@@ -40,7 +40,7 @@ class Timer {
 	}
 	bool isExpired() const
 	{
-		if (m_Timeout == std::chrono::milliseconds{0})
+		if (m_Timeout == std::chrono::milliseconds{-1})
 			return false;
 		std::chrono::time_point<std::chrono::steady_clock> end =
 			std::chrono::steady_clock::now();
@@ -50,7 +50,7 @@ class Timer {
 	}
 	int elapsed() const
 	{
-		if (m_Timeout == std::chrono::milliseconds{0})
+		if (m_Timeout == std::chrono::milliseconds{-1})
 			return 0;
 		std::chrono::time_point<std::chrono::steady_clock> end =
 			std::chrono::steady_clock::now();

test/ClientTest.cpp

@@ -1144,6 +1144,60 @@ response_decoding(Connector<BUFFER, NetProvider> &client)
 	client.close(conn);
 }
 
+/** Checks all available `wait` methods of connector. */
+template <class BUFFER, class NetProvider>
+void
+test_wait(Connector<BUFFER, NetProvider> &client)
+{
+	TEST_INIT(0);
+	static constexpr double SLEEP_TIME = 0.1;
+
+	Connection<Buf_t, NetProvider> conn(client);
+	int rc = test_connect(client, conn, localhost, port);
+	fail_unless(rc == 0);
+
+	TEST_CASE("wait(0) and wait(-1)");
+	rid_t f = conn.call("remote_sleep", std::forward_as_tuple(SLEEP_TIME));
+	fail_unless(!conn.futureIsReady(f));
+	client.wait(conn, f, 0);
+	fail_unless(!conn.futureIsReady(f));
+	client.wait(conn, f, -1);
+	fail_unless(conn.futureIsReady(f));
+	std::optional<Response<Buf_t>> response = conn.getResponse(f);
+	fail_unless(response.has_value());
+
+	TEST_CASE("waitAny(0) and waitAny(-1)");
+	f = conn.call("remote_sleep", std::forward_as_tuple(SLEEP_TIME));
+	fail_unless(!client.waitAny(0).has_value());
+	fail_unless(client.waitAny(-1).has_value());
+	response = conn.getResponse(f);
+	fail_unless(response.has_value());
+
+	TEST_CASE("waitAll(0) and waitAll(-1)");
+	std::vector<rid_t> fs;
+	fs.push_back(conn.call("remote_sleep", std::forward_as_tuple(SLEEP_TIME)));
+	fs.push_back(conn.call("remote_sleep", std::forward_as_tuple(SLEEP_TIME)));
+	fail_unless(client.waitAll(conn, fs, 0) == -1);
+	fail_unless(client.waitAll(conn, fs, -1) == 0);
+	response = conn.getResponse(fs[0]);
+	fail_unless(response.has_value());
+	response = conn.getResponse(fs[1]);
+	fail_unless(response.has_value());
+
+	TEST_CASE("waitCount(0) and waitCount(-1)");
+	fs.clear();
+	fs.push_back(conn.call("remote_sleep", std::forward_as_tuple(SLEEP_TIME)));
+	fs.push_back(conn.call("remote_sleep", std::forward_as_tuple(SLEEP_TIME)));
+	fail_unless(client.waitCount(conn, 2, 0) == -1);
+	fail_unless(client.waitCount(conn, 2, -1) == 0);
+	response = conn.getResponse(fs[0]);
+	fail_unless(response.has_value());
+	response = conn.getResponse(fs[1]);
+	fail_unless(response.has_value());
+
+	client.close(conn);
+}
+
 int main()
 {
 #ifdef TNTCXX_ENABLE_SSL
@@ -1193,5 +1247,6 @@ int main()
 #endif
 	::test_dead_connection_wait();
 	response_decoding(client);
+	test_wait(client);
 	return 0;
 }

test/cfg.lua

@@ -34,6 +34,12 @@ function remote_echo(...)
     return {...}
 end
 
+function remote_sleep(timeout)
+    local fiber = require('fiber')
+    fiber.sleep(timeout)
+    return nil
+end
+
 function get_rps()
     return box.stat.net().REQUESTS.rps
 end

test/cfg_ssl.lua

@@ -32,6 +32,12 @@ function remote_echo(...)
     return {...}
 end
 
+function remote_sleep(timeout)
+    local fiber = require('fiber')
+    fiber.sleep(timeout)
+    return nil
+end
+
 function get_rps()
     return box.stat.net().REQUESTS.rps
 end