client: add connection bookkeeping to Connector for waitAny robustness

Currently, all connection bookkeeping is done opaquely through the network
providers which, in turn, also do this bookkeeping opaquely using system
interfaces (e.g., libev, epoll). Because of this, we cannot handle cases
when waitAny is called and there are no connections (gh-51) or when a
connection has ready responses (gh-132). In order to improve
`waitAny` robustness, we need to add connection bookkeeping to Connector.

We should move the timer start to the beginning of the waiting loop, since
the preceding checking overhead should not be accounted for the waiting
time.

Closes #51
Needed for #132

Author: CuriousGeorgiy

src/Client/Connection.hpp

@@ -84,6 +84,7 @@ struct ConnectionImpl
 	const typename NetProvider::Stream_t &get_strm() const { return strm; }
 
 	void setError(const std::string &msg, int errno_ = 0);
+	bool hasError() const;
 
 	BUFFER &getInBuf();
 	BUFFER &getOutBuf();
@@ -154,6 +155,13 @@ ConnectionImpl<BUFFER, NetProvider>::setError(const std::string &msg, int errno_
 	error.emplace(msg, errno_);
 }
 
+template <class BUFFER, class NetProvider>
+bool
+ConnectionImpl<BUFFER, NetProvider>::hasError() const
+{
+	return error.has_value();
+}
+
 template <class BUFFER, class NetProvider>
 BUFFER &
 ConnectionImpl<BUFFER, NetProvider>::getInBuf()
@@ -466,7 +474,7 @@ template<class BUFFER, class NetProvider>
 bool
 Connection<BUFFER, NetProvider>::hasError() const
 {
-	return impl->error.has_value();
+	return impl->hasError();
 }
 
 template<class BUFFER, class NetProvider>

src/Client/Connector.hpp

@@ -129,6 +129,10 @@ class Connector
 	 * and `connectionDecodeResponses`.
 	 */
 	std::set<ConnectionImpl<BUFFER, NetProvider> *> m_ReadyToDecode;
+	/**
+	 * Set of active connections owned by connector.
+	 */
+	std::set<ConnectionImpl<BUFFER, NetProvider> *> m_Connections;
 };
 
 template<class BUFFER, class NetProvider>
@@ -159,6 +163,7 @@ Connector<BUFFER, NetProvider>::connect(Connection<BUFFER, NetProvider> &conn,
 		conn.getImpl()->prepare_auth(opts.user, opts.passwd);
 	}
 	TNT_LOG_DEBUG("Connection to ", opts.address, ':', opts.service, " has been established");
+	m_Connections.insert(conn.getImpl());
 	return 0;
 }
 
@@ -190,6 +195,7 @@ Connector<BUFFER, NetProvider>::close(ConnectionImpl<BUFFER, NetProvider> *conn)
 		m_NetProvider.close(conn->get_strm());
 		m_ReadyToSend.erase(conn);
 		m_ReadyToDecode.erase(conn);
+		m_Connections.erase(conn);
 	}
 }
 
@@ -353,9 +359,24 @@ template<class BUFFER, class NetProvider>
 std::optional<Connection<BUFFER, NetProvider>>
 Connector<BUFFER, NetProvider>::waitAny(int timeout)
 {
+	if (m_Connections.empty()) {
+		TNT_LOG_DEBUG("waitAny() called on connector without connections");
+		return std::nullopt;
+	}
 	Timer timer{timeout};
 	timer.start();
 	while (m_ReadyToDecode.empty()) {
+		bool has_alive_conn = false;
+		for (auto *conn : m_Connections) {
+			if (!conn->hasError()) {
+				has_alive_conn = true;
+				break;
+			}
+		}
+		if (!has_alive_conn) {
+			TNT_LOG_ERROR("All connections have an error");
+			return std::nullopt;
+		}
 		if (m_NetProvider.wait(timer.timeLeft()) != 0) {
 			TNT_LOG_ERROR("Failed to poll connections: ", strerror(errno));
 			return std::nullopt;

test/ClientTest.cpp

@@ -1118,11 +1118,9 @@ test_dead_connection_wait(void)
 	fail_if(client.waitCount(conn, 1) == 0);
 	fail_if(conn.futureIsReady(f));
 
-	/* FIXME(gh-51) */
-#if 0
+	TEST_CASE("waitAny() correctly handles case when all connections have an error (gh-51");
 	fail_if(client.waitAny() != std::nullopt);
 	fail_if(conn.futureIsReady(f));
-#endif
 }
 
 /**
@@ -1451,6 +1449,9 @@ test_wait(Connector<BUFFER, NetProvider> &client)
 #endif /* __linux__ */
 
 	client.close(conn);
+
+	TEST_CASE("waitAny() correctly handles case when there are no connections (gh-51");
+	fail_if(client.waitAny() != std::nullopt);
 }
 
 int main()