client: return requested future with result argument of wait

Currently, the argument returns any decoded future - that is
inconvenient and completely unusable. Let's return only the requested
future, or nothing, if it's not ready.

Closes #112

Author: drewdzzz

src/Client/Connection.hpp

@@ -230,8 +230,7 @@ class Connection
 
 	template<class B, class N>
 	friend
-	enum DecodeStatus processResponse(Connection<B, N> &conn,
-					  Response<B> *result);
+	enum DecodeStatus processResponse(Connection<B, N> &conn, int req_sync, Response<B> *result);
 
 	template<class B, class N>
 	friend
@@ -530,8 +529,7 @@ inputBufGC(Connection<BUFFER, NetProvider> &conn)
 
 template<class BUFFER, class NetProvider>
 DecodeStatus
-processResponse(Connection<BUFFER, NetProvider> &conn,
-		Response<BUFFER> *result)
+processResponse(Connection<BUFFER, NetProvider> &conn, int req_sync, Response<BUFFER> *result)
 {
 	//Decode response. In case of success - fill in feature map
 	//and adjust end-of-decoded data pointer. Call GC if needed.
@@ -563,7 +561,7 @@ processResponse(Connection<BUFFER, NetProvider> &conn,
 	}
 	LOG_DEBUG("Header: sync=", response.header.sync, ", code=",
 		  response.header.code, ", schema=", response.header.schema_id);
-	if (result != nullptr) {
+	if (result != nullptr && response.header.sync == req_sync) {
 		*result = std::move(response);
 	} else {
 		conn.impl->futures.insert({response.header.sync,

src/Client/Connector.hpp

@@ -89,14 +89,17 @@ 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.
+	 * If `result` is not `nullptr`, it is used to return response for a request with
+	 * `req_sync` sync. If `result` is `nullptr` - `req_sync` is ignored.
 	 * Returns -1 in the case of any error, 0 on success.
 	 */
-	int connectionDecodeResponses(Connection<BUFFER, NetProvider> &conn,
-				      Response<BUFFER> *result);
+	int connectionDecodeResponses(Connection<BUFFER, NetProvider> &conn, int req_sync, Response<BUFFER> *result);
+
 private:
 	NetProvider m_NetProvider;
 	/**
@@ -170,7 +173,7 @@ Connector<BUFFER, NetProvider>::close(ConnectionImpl<BUFFER, NetProvider> &conn)
 
 template<class BUFFER, class NetProvider>
 int
-Connector<BUFFER, NetProvider>::connectionDecodeResponses(Connection<BUFFER, NetProvider> &conn,
+Connector<BUFFER, NetProvider>::connectionDecodeResponses(Connection<BUFFER, NetProvider> &conn, int req_sync,
 							  Response<BUFFER> *result)
 {
 	if (!hasDataToDecode(conn))
@@ -181,7 +184,7 @@ Connector<BUFFER, NetProvider>::connectionDecodeResponses(Connection<BUFFER, Net
 
 	int rc = 0;
 	while (hasDataToDecode(conn)) {
-		DecodeStatus status = processResponse(conn,  result);
+		DecodeStatus status = processResponse(conn, req_sync, result);
 		if (status == DECODE_ERR) {
 			rc = -1;
 			break;
@@ -211,9 +214,10 @@ Connector<BUFFER, NetProvider>::wait(Connection<BUFFER, NetProvider> &conn,
 	Timer timer{timeout};
 	timer.start();
 	static constexpr int INVALID_SYNC = -1;
+	int req_sync = static_cast<int>(future);
 	if (result != NULL)
 		result->header.sync = INVALID_SYNC;
-	if (connectionDecodeResponses(conn, result) != 0)
+	if (connectionDecodeResponses(conn, req_sync, result) != 0)
 		return -1;
 	if (result != NULL && result->header.sync != INVALID_SYNC) {
 		LOG_DEBUG("Future ", future, " is ready and decoded");
@@ -225,7 +229,7 @@ Connector<BUFFER, NetProvider>::wait(Connection<BUFFER, NetProvider> &conn,
 				      strerror(errno), errno);
 			return -1;
 		}
-		if (connectionDecodeResponses(conn, result) != 0)
+		if (connectionDecodeResponses(conn, req_sync, result) != 0)
 			return -1;
 		if (result != NULL && result->header.sync != INVALID_SYNC) {
 			LOG_DEBUG("Future ", future, " is ready and decoded");
@@ -264,7 +268,7 @@ Connector<BUFFER, NetProvider>::waitAll(Connection<BUFFER, NetProvider> &conn,
 				      strerror(errno), errno);
 			return -1;
 		}
-		if (connectionDecodeResponses(conn, nullptr) != 0)
+		if (connectionDecodeResponses(conn, 0, nullptr) != 0)
 			return -1;
 		bool finish = true;
 		for (size_t i = last_not_ready; i < futures.size(); ++i) {
@@ -304,7 +308,7 @@ Connector<BUFFER, NetProvider>::waitAny(int timeout)
 	}
 	Connection<BUFFER, NetProvider> conn = *m_ReadyToDecode.begin();
 	assert(hasDataToDecode(conn));
-	if (connectionDecodeResponses(conn, nullptr) != 0)
+	if (connectionDecodeResponses(conn, 0, nullptr) != 0)
 		return std::nullopt;
 	return conn;
 }
@@ -323,7 +327,7 @@ Connector<BUFFER, NetProvider>::waitCount(Connection<BUFFER, NetProvider> &conn,
 				      strerror(errno), errno);
 			return -1;
 		}
-		if (connectionDecodeResponses(conn, nullptr) != 0)
+		if (connectionDecodeResponses(conn, 0, nullptr) != 0)
 			return -1;
 		if ((conn.getFutureCount() - ready_futures) >= future_count)
 			return 0;

test/ClientTest.cpp

@@ -1303,6 +1303,27 @@ test_wait(Connector<BUFFER, NetProvider> &client)
 	fail_unless(result.header.sync == static_cast<int>(f));
 	fail_unless(result.header.code == 0);
 
+	TEST_CASE("wait with argument result - several requests");
+	/* Obtain in direct order. */
+	f1 = conn.ping();
+	f2 = conn.ping();
+	fail_unless(client.wait(conn, f1, WAIT_TIMEOUT, &result) == 0);
+	fail_unless(result.header.sync == static_cast<int>(f1));
+	fail_unless(result.header.code == 0);
+	fail_unless(client.wait(conn, f2, WAIT_TIMEOUT, &result) == 0);
+	fail_unless(result.header.sync == static_cast<int>(f2));
+	fail_unless(result.header.code == 0);
+
+	/* Obtain in reversed order. */
+	f1 = conn.ping();
+	f2 = conn.ping();
+	fail_unless(client.wait(conn, f2, WAIT_TIMEOUT, &result) == 0);
+	fail_unless(result.header.sync == static_cast<int>(f2));
+	fail_unless(result.header.code == 0);
+	fail_unless(client.wait(conn, f1, WAIT_TIMEOUT, &result) == 0);
+	fail_unless(result.header.sync == static_cast<int>(f1));
+	fail_unless(result.header.code == 0);
+
 	client.close(conn);
 }