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.

Do not forget that `wait(0)` must poll connections - for this purpose,
we should check if the timer has expired after the first call of
`NetProvider.wait()`. The `timer.isExpired()` call actually protected us
from calling `NetProvider.wait()` with negative timeout, and now it's
gone, so let's replace `timer.elapsed()` method with new
`timer.timeLeft()` which is more safe to use here (see #128 for details).

Closes #111
Closes #128

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);
@@ -185,9 +185,8 @@ Connector<BUFFER, NetProvider>::wait(Connection<BUFFER, NetProvider> &conn,
 	timer.start();
 	if (connectionDecodeResponses(conn, result) != 0)
 		return -1;
-	while (!conn.hasError() && !conn.futureIsReady(future) &&
-	       !timer.isExpired()) {
-		if (m_NetProvider.wait(timeout - timer.elapsed()) != 0) {
+	while (!conn.hasError() && !conn.futureIsReady(future)) {
+		if (m_NetProvider.wait(timer.timeLeft()) != 0) {
 			conn.setError(std::string("Failed to poll: ") +
 				      strerror(errno), errno);
 			return -1;
@@ -203,6 +202,8 @@ Connector<BUFFER, NetProvider>::wait(Connection<BUFFER, NetProvider> &conn,
 			if (!hasDataToDecode(conn))
 				m_ReadyToDecode.erase(conn);
 		}
+		if (timer.isExpired())
+			break;
 	}
 	if (conn.hasError()) {
 		LOG_ERROR("Connection got an error: ", conn.getError().msg);
@@ -226,8 +227,8 @@ Connector<BUFFER, NetProvider>::waitAll(Connection<BUFFER, NetProvider> &conn,
 	Timer timer{timeout};
 	timer.start();
 	size_t last_not_ready = 0;
-	while (!conn.hasError() && !timer.isExpired()) {
-		if (m_NetProvider.wait(timeout - timer.elapsed()) != 0) {
+	while (!conn.hasError()) {
+		if (m_NetProvider.wait(timer.timeLeft()) != 0) {
 			conn.setError(std::string("Failed to poll: ") +
 				      strerror(errno), errno);
 			return -1;
@@ -249,6 +250,8 @@ Connector<BUFFER, NetProvider>::waitAll(Connection<BUFFER, NetProvider> &conn,
 		}
 		if (finish)
 			return 0;
+		if (timer.isExpired())
+			break;
 	}
 	if (conn.hasError()) {
 		LOG_ERROR("Connection got an error: ", conn.getError().msg);
@@ -264,8 +267,11 @@ Connector<BUFFER, NetProvider>::waitAny(int timeout)
 {
 	Timer timer{timeout};
 	timer.start();
-	while (m_ReadyToDecode.empty() && !timer.isExpired())
-		m_NetProvider.wait(timeout - timer.elapsed());
+	while (m_ReadyToDecode.empty()) {
+		m_NetProvider.wait(timer.timeLeft());
+		if (timer.isExpired())
+			break;
+	}
 	if (m_ReadyToDecode.empty()) {
 		LOG_ERROR("wait() has been timed out! No responses are received");
 		return std::nullopt;
@@ -287,8 +293,8 @@ Connector<BUFFER, NetProvider>::waitCount(Connection<BUFFER, NetProvider> &conn,
 	Timer timer{timeout};
 	timer.start();
 	size_t ready_futures = conn.getFutureCount();
-	while (!conn.hasError() && !timer.isExpired()) {
-		if (m_NetProvider.wait(timeout - timer.elapsed()) != 0) {
+	while (!conn.hasError()) {
+		if (m_NetProvider.wait(timer.timeLeft()) != 0) {
 			conn.setError(std::string("Failed to poll: ") +
 				      strerror(errno), errno);
 			return -1;
@@ -302,6 +308,8 @@ Connector<BUFFER, NetProvider>::waitCount(Connection<BUFFER, NetProvider> &conn,
 		}
 		if ((conn.getFutureCount() - ready_futures) >= future_count)
 			return 0;
+		if (timer.isExpired())
+			break;
 	}
 	if (conn.hasError()) {
 		LOG_ERROR("Connection got an error: ", conn.getError().msg);

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);
@@ -381,6 +381,8 @@ LibevNetProvider<BUFFER, Stream>::wait(int timeout)
 
 		}
 	}
-	ev_run(m_Loop, EVRUN_ONCE);
+	/* Work in non-blocking mode when the timeout is zero. */
+	int flags = timeout == 0 ? EVRUN_NOWAIT : EVRUN_ONCE;
+	ev_run(m_Loop, flags);
 	return 0;
 }

src/Utils/Timer.hpp

@@ -29,6 +29,7 @@
  * THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
  * SUCH DAMAGE.
  */
+#include <algorithm>
 #include <chrono>
 
 class Timer {
@@ -40,21 +41,30 @@ 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();
 		std::chrono::milliseconds elapsed =
 			std::chrono::duration_cast<std::chrono::milliseconds>(end - m_Start);
 		return elapsed >= m_Timeout;
 	}
-	int elapsed() const
+	/**
+	 * The function to obtain amount of time left. Returns:
+	 * 1. `-1` if the initial timeout was `-1`.
+	 * 2. `0` if the timer has expired.
+	 * 3. Otherwise, amount of milliseconds left is returned.
+	 * NB: the function should not be used for expiration check - use `isExpired` instead.
+	 */
+	int timeLeft() const
 	{
-		if (m_Timeout == std::chrono::milliseconds{0})
-			return 0;
+		if (m_Timeout == std::chrono::milliseconds{-1})
+			return -1;
 		std::chrono::time_point<std::chrono::steady_clock> end =
 			std::chrono::steady_clock::now();
-		return std::chrono::duration_cast<std::chrono::milliseconds>(end - m_Start).count();
+		int timeLeft =
+			m_Timeout.count() - std::chrono::duration_cast<std::chrono::milliseconds>(end - m_Start).count();
+		return std::max(0, timeLeft);
 	}
 private:
 	std::chrono::milliseconds m_Timeout;

test/ClientTest.cpp

@@ -1144,6 +1144,104 @@ 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("wait(0) polls connections");
+	f = conn.call("remote_sleep", std::forward_as_tuple(SLEEP_TIME));
+	fail_unless(!conn.futureIsReady(f));
+	while (!conn.futureIsReady(f)) {
+		client.wait(conn, f, 0);
+		usleep(10 * 1000); /* 10ms */
+	}
+	fail_unless(conn.futureIsReady(f));
+	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("waitAny(0) polls connections");
+	f = conn.call("remote_sleep", std::forward_as_tuple(SLEEP_TIME));
+	fail_unless(!conn.futureIsReady(f));
+	while (!conn.futureIsReady(f)) {
+		client.waitAny(0);
+		usleep(10 * 1000); /* 10ms */
+	}
+	fail_unless(conn.futureIsReady(f));
+	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("waitAll(0) polls connections");
+	f = conn.call("remote_sleep", std::forward_as_tuple(SLEEP_TIME));
+	fail_unless(!conn.futureIsReady(f));
+	while (!conn.futureIsReady(f)) {
+		client.waitAll(conn, std::vector<rid_t>{f}, 0);
+		usleep(10 * 1000); /* 10ms */
+	}
+	fail_unless(conn.futureIsReady(f));
+	response = conn.getResponse(f);
+	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());
+
+	TEST_CASE("waitCount(0) polls connections");
+	f = conn.call("remote_sleep", std::forward_as_tuple(SLEEP_TIME));
+	fail_unless(!conn.futureIsReady(f));
+	while (!conn.futureIsReady(f)) {
+		client.waitCount(conn, 1, 0);
+		usleep(10 * 1000); /* 10ms */
+	}
+	fail_unless(conn.futureIsReady(f));
+	response = conn.getResponse(f);
+	fail_unless(response.has_value());
+
+	client.close(conn);
+}
+
 int main()
 {
 #ifdef TNTCXX_ENABLE_SSL
@@ -1193,5 +1291,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