client: poll connections on non-blocking wait

In the previous commits, we made `wait(0)` non-blocking. However, it
must poll connections (otherwise, it is just pointless) - for this purpose,
we should check if the timer has expired after the first call of
`NetProvider.wait()`.

Also, `LibevNetProvider` should run event loop with a special
`EVRUN_NOWAIT` flag when the timeout is zero for non-blocking poll.

Closes #111

Author: drewdzzz

src/Client/Connector.hpp

@@ -185,8 +185,7 @@ 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()) {
+	while (!conn.hasError() && !conn.futureIsReady(future)) {
 		if (m_NetProvider.wait(timer.timeLeft()) != 0) {
 			conn.setError(std::string("Failed to poll: ") +
 				      strerror(errno), errno);
@@ -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,7 +227,7 @@ Connector<BUFFER, NetProvider>::waitAll(Connection<BUFFER, NetProvider> &conn,
 	Timer timer{timeout};
 	timer.start();
 	size_t last_not_ready = 0;
-	while (!conn.hasError() && !timer.isExpired()) {
+	while (!conn.hasError()) {
 		if (m_NetProvider.wait(timer.timeLeft()) != 0) {
 			conn.setError(std::string("Failed to poll: ") +
 				      strerror(errno), errno);
@@ -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())
+	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,7 +293,7 @@ Connector<BUFFER, NetProvider>::waitCount(Connection<BUFFER, NetProvider> &conn,
 	Timer timer{timeout};
 	timer.start();
 	size_t ready_futures = conn.getFutureCount();
-	while (!conn.hasError() && !timer.isExpired()) {
+	while (!conn.hasError()) {
 		if (m_NetProvider.wait(timer.timeLeft()) != 0) {
 			conn.setError(std::string("Failed to poll: ") +
 				      strerror(errno), errno);
@@ -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/LibevNetProvider.hpp

@@ -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;
 }

test/ClientTest.cpp

@@ -1166,13 +1166,35 @@ test_wait(Connector<BUFFER, NetProvider> &client)
 	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)));
@@ -1184,6 +1206,17 @@ test_wait(Connector<BUFFER, NetProvider> &client)
 	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)));
@@ -1195,6 +1228,17 @@ test_wait(Connector<BUFFER, NetProvider> &client)
 	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);
 }