client: always erase decoded connections from corresponding set

Connector uses a special helper for decoding responses of ready-to-decode
connections and then it erases them from `m_ReadyToDecode` set by himself.
That's not a good design and it can lead to various bugs (there is already
a bug, actually) because it's easy to forgot to erase a decoded connection.
So let's move the erasure right to the `connectionDecodeResponses` helper.
Note that the helper must be turned into a method of `Connector` since it
needs to use its fields.

The above-mentioned bug happens because we decode responses in
`Connector::wait` right before the main loop but we forgot to remove the
connection from `m_ReadyToDecode` set there.

Closes #124

Author: drewdzzz

src/Client/Connector.hpp

@@ -89,8 +89,23 @@ class Connector
 	std::set<Connection<BUFFER, NetProvider>> m_ReadyToSend;
 	void close(Connection<BUFFER, NetProvider> &conn);
 	void close(ConnectionImpl<BUFFER, NetProvider> &conn);
+
+private:
+	/**
+	 * A helper to decode responses of a connection.
+	 * Can be called when the connection is not ready to decode - it's just no-op.
+	 * Returns -1 in the case of any error, 0 on success.
+	 */
+	int connectionDecodeResponses(Connection<BUFFER, NetProvider> &conn,
+				      Response<BUFFER> *result);
+
 private:
 	NetProvider m_NetProvider;
+	/**
+	 * Set of connections that are ready to decode.
+	 * Shouldn't be modified directly - is managed by methods `readyToDecode`
+	 * and `connectionDecodeResponses`.
+	 */
 	std::set<Connection<BUFFER, NetProvider>> m_ReadyToDecode;
 };
 
@@ -157,21 +172,35 @@ Connector<BUFFER, NetProvider>::close(ConnectionImpl<BUFFER, NetProvider> &conn)
 
 template<class BUFFER, class NetProvider>
 int
-connectionDecodeResponses(Connection<BUFFER, NetProvider> &conn,
-			  Response<BUFFER> *result)
+Connector<BUFFER, NetProvider>::connectionDecodeResponses(Connection<BUFFER, NetProvider> &conn,
+							  Response<BUFFER> *result)
 {
+	if (!hasDataToDecode(conn))
+		return 0;
+
+	/* Ready to decode connection must be in the corresponding set. */
+	assert(m_ReadyToDecode.find(conn) != m_ReadyToDecode.end());
+
+	int rc = 0;
 	while (hasDataToDecode(conn)) {
-		DecodeStatus rc = processResponse(conn,  result);
-		if (rc == DECODE_ERR)
-			return -1;
+		DecodeStatus status = processResponse(conn, result);
+		if (status == DECODE_ERR) {
+			rc = -1;
+			break;
+		}
 		//In case we've received only a part of response
 		//we should wait until the rest arrives - otherwise
 		//we can't properly decode response. */
-		if (rc == DECODE_NEEDMORE)
-			return 0;
-		assert(rc == DECODE_SUCC);
+		if (status == DECODE_NEEDMORE) {
+			rc = 0;
+			break;
+		}
+		assert(status == DECODE_SUCC);
 	}
-	return 0;
+	/* A connection that has no data to decode must not be left in the set. */
+	if (!hasDataToDecode(conn))
+		m_ReadyToDecode.erase(conn);
+	return rc;
 }
 
 template<class BUFFER, class NetProvider>
@@ -191,17 +220,8 @@ Connector<BUFFER, NetProvider>::wait(Connection<BUFFER, NetProvider> &conn,
 				      strerror(errno), errno);
 			return -1;
 		}
-		if (hasDataToDecode(conn)) {
-			assert(m_ReadyToDecode.find(conn) != m_ReadyToDecode.end());
-			if (connectionDecodeResponses(conn, result) != 0)
-				return -1;
-			/*
-			 * In case we've handled whole data in input buffer -
-			 * mark connection as completed.
-			 */
-			if (!hasDataToDecode(conn))
-				m_ReadyToDecode.erase(conn);
-		}
+		if (connectionDecodeResponses(conn, result) != 0)
+			return -1;
 		if (timer.isExpired())
 			break;
 	}
@@ -233,13 +253,8 @@ Connector<BUFFER, NetProvider>::waitAll(Connection<BUFFER, NetProvider> &conn,
 				      strerror(errno), errno);
 			return -1;
 		}
-		if (hasDataToDecode(conn)) {
-			assert(m_ReadyToDecode.find(conn) != m_ReadyToDecode.end());
-			if (connectionDecodeResponses(conn, static_cast<Response<BUFFER>*>(nullptr)) != 0)
-				return -1;
-			if (!hasDataToDecode(conn))
-				m_ReadyToDecode.erase(conn);
-		}
+		if (connectionDecodeResponses(conn, static_cast<Response<BUFFER>*>(nullptr)) != 0)
+			return -1;
 		bool finish = true;
 		for (size_t i = last_not_ready; i < futures.size(); ++i) {
 			if (!conn.futureIsReady(futures[i])) {
@@ -280,8 +295,6 @@ Connector<BUFFER, NetProvider>::waitAny(int timeout)
 	assert(hasDataToDecode(conn));
 	if (connectionDecodeResponses(conn, static_cast<Response<BUFFER>*>(nullptr)) != 0)
 		return std::nullopt;
-	if (!hasDataToDecode(conn))
-		m_ReadyToDecode.erase(conn);
 	return conn;
 }
 
@@ -299,13 +312,8 @@ Connector<BUFFER, NetProvider>::waitCount(Connection<BUFFER, NetProvider> &conn,
 				      strerror(errno), errno);
 			return -1;
 		}
-		if (hasDataToDecode(conn)) {
-			assert(m_ReadyToDecode.find(conn) != m_ReadyToDecode.end());
-			if (connectionDecodeResponses(conn, static_cast<Response<BUFFER>*>(nullptr)) != 0)
-				return -1;
-			if (!hasDataToDecode(conn))
-				m_ReadyToDecode.erase(conn);
-		}
+		if (connectionDecodeResponses(conn, static_cast<Response<BUFFER>*>(nullptr)) != 0)
+			return -1;
 		if ((conn.getFutureCount() - ready_futures) >= future_count)
 			return 0;
 		if (timer.isExpired())

test/ClientTest.cpp

@@ -1239,6 +1239,47 @@ test_wait(Connector<BUFFER, NetProvider> &client)
 	response = conn.getResponse(f);
 	fail_unless(response.has_value());
 
+	TEST_CASE("waitAny after several waits (gh-124)");
+	Connection<Buf_t, NetProvider> conn1(client);
+	Connection<Buf_t, NetProvider> conn2(client);
+	Connection<Buf_t, NetProvider> conn3(client);
+	rc = test_connect(client, conn1, localhost, port);
+	fail_unless(rc == 0);
+	rc = test_connect(client, conn2, localhost, port);
+	fail_unless(rc == 0);
+	rc = test_connect(client, conn3, localhost, port);
+	fail_unless(rc == 0);
+	rid_t f1 = conn1.ping();
+	rid_t f2 = conn2.ping();
+	rid_t f3 = conn3.ping();
+
+	/* Wait for all connections. */
+	fail_unless(client.wait(conn1, f1, WAIT_TIMEOUT) == 0);
+	fail_unless(conn1.futureIsReady(f1));
+	fail_unless(conn1.getResponse(f1).header.code == 0);
+
+	fail_unless(client.wait(conn2, f2, WAIT_TIMEOUT) == 0);
+	fail_unless(conn2.futureIsReady(f2));
+	fail_unless(conn2.getResponse(f2).header.code == 0);
+
+	fail_unless(client.wait(conn3, f3, WAIT_TIMEOUT) == 0);
+	fail_unless(conn3.futureIsReady(f3));
+	fail_unless(conn3.getResponse(f3).header.code == 0);
+
+	/*
+	 * Wait any - we shouldn't get any of the connections here since we've
+	 * received all the responses.
+	 * Note that the connector used to crash here (gh-124) because some of the
+	 * connnections still could appear in `m_ReadyToDecode` set.
+	 */
+	std::optional<Connection<Buf_t, NetProvider>> conn_opt = client.waitAny(WAIT_TIMEOUT);
+	fail_if(conn_opt.has_value());
+
+	/* Close all connections used only by the case. */
+	client.close(conn1);
+	client.close(conn2);
+	client.close(conn3);
+
 	client.close(conn);
 }