client: replace all internal Connection usage with ConnectionImpl

`Connection`s model shared pointers on `ConnectionImpl` and are intended
for external usage by library consumers. Using them internally in the
Connector can cause bugs. For instance, we internally store `Connection`
objects in the LibevNetProvider, which adds an additional reference to them
and prevents them from being deleted, even when all user objects are dead.
Also we leak connections in `close`, since we do not erase them from the
`m_ReadyToSend` and `m_ReadyToDecode` sets.

To fix this, let's internally use `ConnectionImpl` to model weak pointers.
We rely on the fact that the lifetime of these weak pointers is tied to the
lifetime of the shared user objects.

The ideal solution would be to remove the `Connection`<-`ConnectionImpl`
constructor, but we still need to return new Connection objects in methods
like waitAny, so let's just enforce as a rule that we should not use
`Connection` objects internally.

While we are here, let's also fix all the formating issues that the linter
is reporting for the refactored code.

Closes #140
Closes #147

Author: CuriousGeorgiy

src/Client/Connection.hpp

@@ -79,8 +79,21 @@ struct ConnectionImpl
 	void ref();
 	void unref();
 
+	typename NetProvider::Stream_t &get_strm() { return strm; }
+	const typename NetProvider::Stream_t &get_strm() const { return strm; }
+
+	void setError(const std::string &msg, int errno_ = 0);
+
+	BUFFER &getInBuf();
+	BUFFER &getOutBuf();
+
+	void prepare_auth(std::string_view user, std::string_view passwd);
+	void commit_auth(std::string_view user, std::string_view passwd);
+
 	Connector<BUFFER, NetProvider> &connector;
 	BUFFER inBuf;
+	static constexpr size_t GC_STEP_CNT = 100;
+	size_t gc_step = 0;
 	BUFFER outBuf;
 	RequestEncoder<BUFFER> enc;
 	ResponseDecoder<BUFFER> dec;
@@ -112,7 +125,7 @@ ConnectionImpl<BUFFER, NetProvider>::~ConnectionImpl()
 {
 	assert(refs == 0);
 	if (!strm.has_status(SS_DEAD)) {
-		connector.close(*this);
+		connector.close(this);
 	}
 }
 
@@ -133,6 +146,42 @@ ConnectionImpl<BUFFER, NetProvider>::unref()
 		delete this;
 }
 
+template <class BUFFER, class NetProvider>
+void
+ConnectionImpl<BUFFER, NetProvider>::setError(const std::string &msg, int errno_)
+{
+	error.emplace(msg, errno_);
+}
+
+template <class BUFFER, class NetProvider>
+BUFFER &
+ConnectionImpl<BUFFER, NetProvider>::getInBuf()
+{
+	return inBuf;
+}
+
+template <class BUFFER, class NetProvider>
+BUFFER &
+ConnectionImpl<BUFFER, NetProvider>::getOutBuf()
+{
+	return outBuf;
+}
+
+template <class BUFFER, class NetProvider>
+void
+ConnectionImpl<BUFFER, NetProvider>::prepare_auth(std::string_view user, std::string_view passwd)
+{
+	enc.encodeAuth(user, passwd, greeting);
+}
+
+template <class BUFFER, class NetProvider>
+void
+ConnectionImpl<BUFFER, NetProvider>::commit_auth(std::string_view user, std::string_view passwd)
+{
+	enc.reencodeAuth(user, passwd, greeting);
+	connector.readyToSend(this);
+}
+
 /** Each connection is supposed to be bound to a single socket. */
 template<class BUFFER, class NetProvider>
 class Connection
@@ -212,44 +261,11 @@ class Connection
 	BUFFER& getInBuf();
 	BUFFER& getOutBuf();
 
-	template<class B, class N>
-	friend
-	void hasSentBytes(Connection<B, N> &conn, size_t bytes);
-
-	template<class B, class N>
-	friend
-	void hasNotRecvBytes(Connection<B, N> &conn, size_t bytes);
-
-	template<class B, class N>
-	friend
-	bool hasDataToSend(Connection<B, N> &conn);
-
-	template<class B, class N>
-	friend
-	bool hasDataToDecode(Connection<B, N> &conn);
-
-	template<class B, class N>
-	friend
-	enum DecodeStatus processResponse(Connection<B, N> &conn, int req_sync, Response<B> *result);
-
-	template<class B, class N>
-	friend
-	void inputBufGC(Connection<B, N> &conn);
-
-	template<class B, class N>
-	friend
-	int decodeGreeting(Connection<B, N> &conn);
-
-	rid_t prepare_auth(std::string_view user,
-			   std::string_view passwd);
-
-	rid_t commit_auth(std::string_view user,
-			  std::string_view passwd);
+	void prepare_auth(std::string_view user, std::string_view passwd);
+	void commit_auth(std::string_view user, std::string_view passwd);
 
 private:
 	ConnectionImpl<BUFFER, NetProvider> *impl;
-	static constexpr size_t GC_STEP_CNT = 100;
-	size_t gc_step = 0;
 
 	template <class T>
 	rid_t insert(const T &tuple, uint32_t space_id);
@@ -442,7 +458,7 @@ template<class BUFFER, class NetProvider>
 void
 Connection<BUFFER, NetProvider>::setError(const std::string &msg, int errno_)
 {
-	impl->error.emplace(msg, errno_);
+	impl->setError(msg, errno_);
 }
 
 template<class BUFFER, class NetProvider>
@@ -471,73 +487,79 @@ template<class BUFFER, class NetProvider>
 BUFFER&
 Connection<BUFFER, NetProvider>::getInBuf()
 {
-	return impl->inBuf;
+	return impl->getInBuf();
 }
 
 template<class BUFFER, class NetProvider>
 BUFFER&
 Connection<BUFFER, NetProvider>::getOutBuf()
 {
-	return impl->outBuf;
+	return impl->getOutBuf();
 }
 
-template<class BUFFER, class NetProvider>
+template <class BUFFER, class NetProvider>
 void
-hasSentBytes(Connection<BUFFER, NetProvider> &conn, size_t bytes)
+hasSentBytes(ConnectionImpl<BUFFER, NetProvider> *conn, size_t bytes)
 {
 	//dropBack()/dropFront() interfaces require number of bytes be greater
 	//than zero so let's check it first.
 	if (bytes > 0)
-		conn.impl->outBuf.dropFront(bytes);
+		conn->getOutBuf().dropFront(bytes);
 }
 
-template<class BUFFER, class NetProvider>
+template <class BUFFER, class NetProvider>
 void
-hasNotRecvBytes(Connection<BUFFER, NetProvider> &conn, size_t bytes)
+hasNotRecvBytes(ConnectionImpl<BUFFER, NetProvider> *conn, size_t bytes)
 {
 	if (bytes > 0)
-		conn.impl->inBuf.dropBack(bytes);
+		conn->getInBuf().dropBack(bytes);
 }
 
-template<class BUFFER, class NetProvider>
+template <class BUFFER, class NetProvider>
 bool
-hasDataToSend(Connection<BUFFER, NetProvider> &conn)
+hasDataToSend(ConnectionImpl<BUFFER, NetProvider> *conn)
 {
 	//We drop content of input buffer once it has been sent. So to detect
 	//if there's any data to send it's enough to check buffer's emptiness.
-	return !conn.impl->outBuf.empty();
+	return !conn->getOutBuf().empty();
 }
 
-template<class BUFFER, class NetProvider>
+template <class BUFFER, class NetProvider>
 bool
 hasDataToDecode(Connection<BUFFER, NetProvider> &conn)
 {
-	assert(conn.impl->endDecoded < conn.impl->inBuf.end() ||
-	       conn.impl->endDecoded == conn.impl->inBuf.end());
-	return conn.impl->endDecoded != conn.impl->inBuf.end();
+	return hasDataToDecode(conn.getImpl());
 }
 
-template<class BUFFER, class NetProvider>
+template <class BUFFER, class NetProvider>
+bool
+hasDataToDecode(ConnectionImpl<BUFFER, NetProvider> *conn)
+{
+	assert(conn->endDecoded < conn->getInBuf().end() || conn->endDecoded == conn->getInBuf().end());
+	return conn->endDecoded != conn->getInBuf().end();
+}
+
+template <class BUFFER, class NetProvider>
 static void
-inputBufGC(Connection<BUFFER, NetProvider> &conn)
+inputBufGC(ConnectionImpl<BUFFER, NetProvider> *conn)
 {
-	if ((conn.gc_step++ % Connection<BUFFER, NetProvider>::GC_STEP_CNT) == 0) {
-		LOG_DEBUG("Flushed input buffer of the connection %p", &conn);
-		conn.impl->inBuf.flush();
+	if (conn->gc_step++ % ConnectionImpl<BUFFER, NetProvider>::GC_STEP_CNT == 0) {
+		LOG_DEBUG("Flushed input buffer of the connection %p", conn);
+		conn->getInBuf().flush();
 	}
 }
 
-template<class BUFFER, class NetProvider>
+template <class BUFFER, class NetProvider>
 DecodeStatus
-processResponse(Connection<BUFFER, NetProvider> &conn, int req_sync, Response<BUFFER> *result)
+processResponse(ConnectionImpl<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.
-	if (! conn.impl->inBuf.has(conn.impl->endDecoded, MP_RESPONSE_SIZE))
+	if (!conn->getInBuf().has(conn->endDecoded, MP_RESPONSE_SIZE))
 		return DECODE_NEEDMORE;
 
 	Response<BUFFER> response;
-	response.size = conn.impl->dec.decodeResponseSize();
+	response.size = conn->dec.decodeResponseSize();
 	if (response.size < 0) {
 		LOG_ERROR("Failed to decode response size");
 		//In case of corrupted response size all other data in the buffer
@@ -548,55 +570,53 @@ processResponse(Connection<BUFFER, NetProvider> &conn, int req_sync, Response<BU
 
 	}
 	response.size += MP_RESPONSE_SIZE;
-	if (! conn.impl->inBuf.has(conn.impl->endDecoded, response.size)) {
+	if (!conn->getInBuf().has(conn->endDecoded, response.size)) {
 		//Response was received only partially. Reset decoder position
 		//to the start of response to make this function re-entered.
-		conn.impl->dec.reset(conn.impl->endDecoded);
+		conn->dec.reset(conn->endDecoded);
 		return DECODE_NEEDMORE;
 	}
-	if (conn.impl->dec.decodeResponse(response) != 0) {
-		conn.setError("Failed to decode response, skipping bytes..");
-		conn.impl->endDecoded += response.size;
+	if (conn->dec.decodeResponse(response) != 0) {
+		conn->setError("Failed to decode response, skipping bytes..");
+		conn->endDecoded += response.size;
 		return DECODE_ERR;
 	}
 	LOG_DEBUG("Header: sync=", response.header.sync, ", code=",
 		  response.header.code, ", schema=", response.header.schema_id);
 	if (result != nullptr && response.header.sync == req_sync) {
 		*result = std::move(response);
 	} else {
-		conn.impl->futures.insert({response.header.sync,
-					   std::move(response)});
+		conn->futures.insert({response.header.sync, std::move(response)});
 	}
-	conn.impl->endDecoded += response.size;
+	conn->endDecoded += response.size;
 	inputBufGC(conn);
 	return DECODE_SUCC;
 }
 
-template<class BUFFER, class NetProvider>
+template <class BUFFER, class NetProvider>
 int
-decodeGreeting(Connection<BUFFER, NetProvider> &conn)
+decodeGreeting(ConnectionImpl<BUFFER, NetProvider> *conn)
 {
 	//TODO: that's not zero-copy, should be rewritten in that pattern.
-	assert(conn.getInBuf().has(conn.impl->endDecoded, Iproto::GREETING_SIZE));
+	assert(conn->getInBuf().has(conn->endDecoded, Iproto::GREETING_SIZE));
 	char greeting_buf[Iproto::GREETING_SIZE];
-	conn.impl->endDecoded.read({greeting_buf, sizeof(greeting_buf)});
-	conn.impl->dec.reset(conn.impl->endDecoded);
-	if (parseGreeting(std::string_view{greeting_buf, Iproto::GREETING_SIZE},
-			  conn.impl->greeting) != 0)
+	conn->endDecoded.read({greeting_buf, sizeof(greeting_buf)});
+	conn->dec.reset(conn->endDecoded);
+	if (parseGreeting(std::string_view {greeting_buf, Iproto::GREETING_SIZE}, conn->greeting) != 0)
 		return -1;
-	conn.impl->is_greeting_received = true;
-	LOG_DEBUG("Version: ", conn.impl->greeting.version_id);
+	conn->is_greeting_received = true;
+	LOG_DEBUG("Version: ", conn->greeting.version_id);
 
 #ifndef NDEBUG
 	//print salt in hex format.
 	char hex_salt[Iproto::MAX_SALT_SIZE * 2 + 1];
 	const char *hex = "0123456789abcdef";
-	for (size_t i = 0; i < conn.impl->greeting.salt_size; i++) {
-		uint8_t u = conn.impl->greeting.salt[i];
+	for (size_t i = 0; i < conn->greeting.salt_size; i++) {
+		uint8_t u = conn->greeting.salt[i];
 		hex_salt[i * 2] = hex[u / 16];
 		hex_salt[i * 2 + 1] = hex[u % 16];
 	}
-	hex_salt[conn.impl->greeting.salt_size * 2] = 0;
+	hex_salt[conn->greeting.salt_size * 2] = 0;
 	LOG_DEBUG("Salt: ", hex_salt);
 #endif
 	return 0;
@@ -717,21 +737,16 @@ Connection<BUFFER, NetProvider>::select(const T &key, uint32_t space_id,
 	return impl->enc.getSync();
 }
 
-template<class BUFFER, class NetProvider>
-rid_t
-Connection<BUFFER, NetProvider>::prepare_auth(std::string_view user,
-					      std::string_view passwd)
+template <class BUFFER, class NetProvider>
+void
+Connection<BUFFER, NetProvider>::prepare_auth(std::string_view user, std::string_view passwd)
 {
-	impl->enc.encodeAuth(user, passwd, impl->greeting);
-	return 0;
+	impl->prepare_auth(user, passwd);
 }
 
-template<class BUFFER, class NetProvider>
-rid_t
-Connection<BUFFER, NetProvider>::commit_auth(std::string_view user,
-					      std::string_view passwd)
+template <class BUFFER, class NetProvider>
+void
+Connection<BUFFER, NetProvider>::commit_auth(std::string_view user, std::string_view passwd)
 {
-	impl->enc.reencodeAuth(user, passwd, impl->greeting);;
-	impl->connector.readyToSend(*this);
-	return 0;
+	impl->commit_auth(user, passwd);
 }