cppcache/src/TcpConn.cpp - geode-native - Git at Google

 /*
  * Licensed to the Apache Software Foundation (ASF) under one or more
  * contributor license agreements.  See the NOTICE file distributed with
  * this work for additional information regarding copyright ownership.
  * The ASF licenses this file to You under the Apache License, Version 2.0
  * (the "License"); you may not use this file except in compliance with
  * the License.  You may obtain a copy of the License at
  *
  *      http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */

 #include "TcpConn.hpp"

 #include <iomanip>
 #include <iostream>

 #include <boost/optional.hpp>
 #include <boost/system/error_code.hpp>
 #include <boost/system/system_error.hpp>

 #include "util/Log.hpp"

 namespace {
 template <int Level, int Name>
 class timeval {
  public:
   // This is not an instance of the template, but of the system provided type
   // to be written to the socket API.
 #if defined(_WINDOWS)
   using value_type = DWORD;
 #else
   using value_type = ::timeval;
 #endif

  private:
   value_type value_{};

  public:
   timeval() {}

   explicit timeval(value_type v) : value_(v) {}

   timeval &operator=(value_type v) {
     value_ = v;
     return *this;
   }

   value_type value() const { return value_; }

   template <typename Protocol>
   int level(const Protocol &) const {
     return Level;
   }

   template <typename Protocol>
   int name(const Protocol &) const {
     return Name;
   }

   template <typename Protocol>
   value_type *data(const Protocol &) {
     return &value_;
   }

   template <typename Protocol>
   const value_type *data(const Protocol &) const {
     return &value_;
   }

   template <typename Protocol>
   std::size_t size(const Protocol &) const {
     return sizeof(value_);
   }

   template <typename Protocol>
   void resize(const Protocol &, std::size_t s) {
     if (s != sizeof(value_)) {
       throw std::length_error("timeval socket option resize");
     }
   }
 };

 // Asio doesn't support these socket options directly, but every major platform
 // does. Timeout on IO socket operations are supported by the platform directly.
 // This means We can all receive without needing to use the timeout interface -
 // and more importantly, we can send while holding to per-operation time
 // constraints and without blocking indefinitely.
 //
 // The default timeout is infinite, or by setting the socket option to null,
 // which I won't provide - just don't construct a TcpConn with send and
 // receieve timeouts.
 typedef timeval<SOL_SOCKET, SO_SNDTIMEO> send_timeout;
 typedef timeval<SOL_SOCKET, SO_RCVTIMEO> receive_timeout;
 }  // namespace

 namespace apache {
 namespace geode {
 namespace client {
 TcpConn::TcpConn(const std::string ipaddr,
                  std::chrono::microseconds connect_timeout,
                  int32_t maxBuffSizePool)
     : TcpConn{
           ipaddr.substr(0, ipaddr.find(':')),
           static_cast<uint16_t>(std::stoi(ipaddr.substr(ipaddr.find(':') + 1))),
           connect_timeout, maxBuffSizePool} {}

 TcpConn::TcpConn(const std::string host, uint16_t port,
                  std::chrono::microseconds timeout, int32_t maxBuffSizePool)
     : socket_{io_context_} {
   auto beforeResolvePoint = std::chrono::system_clock::now();
   auto results = resolve(host, port, timeout);
   auto elapsedTime = std::chrono::duration<double, std::micro>(
       std::chrono::system_clock::now() - beforeResolvePoint);

   // We must connect first so we have a valid file descriptor to set options
   // on.
   auto connectTimeout = std::chrono::duration_cast<std::chrono::microseconds>(
       timeout - elapsedTime);
   connect(results, connectTimeout);

   socket_.set_option(::boost::asio::ip::tcp::no_delay{true});
   socket_.set_option(
       ::boost::asio::socket_base::send_buffer_size{maxBuffSizePool});
   socket_.set_option(
       ::boost::asio::socket_base::receive_buffer_size{maxBuffSizePool});
 }

 TcpConn::TcpConn(const std::string ipaddr,
                  std::chrono::microseconds connect_timeout,
                  int32_t maxBuffSizePool, std::chrono::microseconds send_time,
                  std::chrono::microseconds receive_time)
     : TcpConn{ipaddr, connect_timeout, maxBuffSizePool} {
 #if defined(_WINDOWS)
   socket_.set_option(::send_timeout{static_cast<DWORD>(send_time.count())});
   socket_.set_option(
       ::receive_timeout{static_cast<DWORD>(receive_time.count())});
 #else

   auto send_seconds =
       std::chrono::duration_cast<std::chrono::seconds>(send_time);
   auto send_microseconds =
       send_time % std::chrono::duration_cast<std::chrono::microseconds>(
                       std::chrono::seconds{1});
   socket_.set_option(
       ::send_timeout{{static_cast<int>(send_seconds.count()),
                       static_cast<int>(send_microseconds.count())}});

   auto receive_seconds =
       std::chrono::duration_cast<std::chrono::seconds>(receive_time);
   auto receive_microseconds =
       receive_time % std::chrono::duration_cast<std::chrono::microseconds>(
                          std::chrono::seconds{1});
   socket_.set_option(
       ::receive_timeout{{static_cast<int>(receive_seconds.count()),
                          static_cast<int>(receive_microseconds.count())}});
 #endif
 }

 TcpConn::~TcpConn() {
   std::stringstream ss;

   try {
     ss << "Disconnected " << socket_.local_endpoint() << " -> "
        << socket_.remote_endpoint();

     socket_.shutdown(boost::asio::ip::tcp::socket::shutdown_both);

   } catch (...) {
     ss = std::stringstream{};

     ss << "Closed socket " << socket_.local_endpoint();
   }

   socket_.close();

   LOGFINE(ss.str());
 }

 size_t TcpConn::receive(char *buff, const size_t len,
                         std::chrono::milliseconds timeout) {
   std::stringstream ss;
   ss << "Receiving " << len << " bytes from " << socket_.remote_endpoint()
      << " -> " << socket_.local_endpoint();
   LOGDEBUG(ss.str());
   return receive(buff, len, timeout, true);
 }

 size_t TcpConn::receive_nothrowiftimeout(char *buff, const size_t len,
                                          std::chrono::milliseconds timeout) {
   std::stringstream ss;
   ss << "Receiving an unknown number of bytes from "
      << socket_.remote_endpoint() << " -> " << socket_.local_endpoint();
   LOGDEBUG(ss.str());
   return receive(buff, len, timeout, false);
 }

 size_t TcpConn::receive(char *buff, const size_t len,
                         std::chrono::milliseconds timeout,
                         bool throwTimeoutException) {
   boost::optional<boost::system::error_code> read_result;
   std::size_t bytes_read = 0;

   auto beforeResolvePoint = std::chrono::system_clock::now();

   try {
     prepareAsyncRead(buff, len, read_result, bytes_read);
     io_context_.restart();
     io_context_.run_for(timeout);
   } catch (...) {
     LOGDEBUG("Throwing an unexpected read exception");
     throw;
   }

   if (read_result && *read_result) {
     LOGDEBUG("Throwing a read exception: %s", read_result->message().c_str());
     socket_.cancel();
     // Get the abort
     io_context_.restart();
     io_context_.run();
     throw boost::system::system_error{*read_result};
   }

   if (bytes_read == 0) {
     auto elapsedTime = std::chrono::duration<double, std::micro>(
         std::chrono::system_clock::now() - beforeResolvePoint);
     if (elapsedTime < timeout) {
       LOGDEBUG("Throwing an IO exception");
       socket_.cancel();
       // Get the abort
       io_context_.restart();
       io_context_.run();
       throw boost::system::system_error{boost::asio::error::broken_pipe};
     } else {
       LOGDEBUG("Throwing an eof exception");
       socket_.cancel();
       // Get the abort
       io_context_.restart();
       io_context_.run();
       throw boost::system::system_error{boost::asio::error::eof};
     }
   }

   if (bytes_read != len && throwTimeoutException) {
     LOGDEBUG("Throwing a read timeout exception");
     socket_.cancel();
     // Get the abort
     io_context_.restart();
     io_context_.run();
     throw boost::system::system_error{boost::asio::error::operation_aborted};
   }

   return bytes_read;
 }

 size_t TcpConn::send(const char *buff, const size_t len,
                      std::chrono::milliseconds timeout) {
   std::stringstream ss;
   ss << "Sending " << len << " bytes from " << socket_.local_endpoint()
      << " -> " << socket_.remote_endpoint();
   LOGDEBUG(ss.str());

   boost::optional<boost::system::error_code> write_result;
   std::size_t bytes_written = 0;

   try {
     prepareAsyncWrite(buff, len, write_result, bytes_written);
     io_context_.restart();
     io_context_.run_for(timeout);
   } catch (...) {
     LOGDEBUG("Throwing an unexpected write exception");
     throw;
   }

   if (write_result && *write_result) {
     LOGDEBUG("Throwing a write exception. %s", write_result->message().c_str());
     throw boost::system::system_error{*write_result};
   }

   if (bytes_written != len) {
     LOGDEBUG("Throwing a write timeout exception");
     socket_.cancel();
     // Get the abort
     io_context_.restart();
     io_context_.run();
     throw boost::system::system_error{boost::asio::error::operation_aborted};
   }

   return bytes_written;
 }

 //  Return the local port for this TCP connection.
 uint16_t TcpConn::getPort() { return socket_.local_endpoint().port(); }

 void TcpConn::connect(boost::asio::ip::tcp::resolver::results_type r,
                       std::chrono::microseconds timeout) {
   boost::optional<boost::system::error_code> connect_result;

   try {
     // We must connect first so we have a valid file descriptor to set
     // options on.
     boost::asio::async_connect(
         socket_, r,
         [&connect_result](const boost::system::error_code &ec,
                           const boost::asio::ip::tcp::endpoint) {
           connect_result = ec;
         });

     io_context_.restart();
     io_context_.run_for(timeout);
   } catch (...) {
     LOGDEBUG("Throwing an unexpected connect exception");
     throw;
   }

   if (connect_result && *connect_result) {
     LOGDEBUG("Throwing a connect exception: %s",
              connect_result->message().c_str());
     throw boost::system::system_error{*connect_result};
   }

   if (!connect_result) {
     LOGDEBUG("Throwing a connect timeout exception");
     throw boost::system::system_error{boost::asio::error::operation_aborted};
   }

   std::stringstream ss;
   ss << "Connected " << socket_.local_endpoint() << " -> "
      << socket_.remote_endpoint();
   LOGDEBUG(ss.str());
 }

 boost::asio::ip::tcp::resolver::results_type TcpConn::resolve(
     const std::string host, uint16_t port, std::chrono::microseconds timeout) {
   boost::optional<boost::system::error_code> resolve_result;
   boost::asio::ip::tcp::resolver::results_type results;

   try {
     boost::asio::ip::tcp::resolver resolver(io_context_);
     resolver.async_resolve(host, std::to_string(port),
                            [&resolve_result, &results](
                                const boost::system::error_code &ec,
                                boost::asio::ip::tcp::resolver::results_type r) {
                              if (ec) {
                                resolve_result = ec;
                              } else {
                                resolve_result = boost::system::error_code{};
                                results = r;
                              }
                            });

     io_context_.restart();
     io_context_.run_for(timeout);
   } catch (...) {
     LOGDEBUG("Throwing an unexpected resolve exception");
     throw;
   }

   if (resolve_result && *resolve_result) {
     LOGDEBUG("Throwing a resolve exception: %s",
              resolve_result->message().c_str());
     throw boost::system::system_error{*resolve_result};
   }

   if (!resolve_result) {
     LOGDEBUG("Throwing a resolve timeout exception");
     socket_.cancel();
     // Get the abort
     io_context_.restart();
     io_context_.run();
     throw boost::system::system_error{boost::asio::error::operation_aborted};
   }

   return results;
 }

 void TcpConn::prepareAsyncRead(
     char *buff, size_t len,
     boost::optional<boost::system::error_code> &read_result,
     std::size_t &bytes_read) {
   boost::asio::async_read(
       socket_, boost::asio::buffer(buff, len),
       [&read_result, &bytes_read](const boost::system::error_code &ec,
                                   const size_t n) {
         bytes_read = n;

         // EOF itself occurs when there is no data available on the socket at
         // the time of the read. It may simply imply data has yet to arrive.
         // Do nothing. Defer to timeout rather than assume a broken
         // connection.
         if (ec != boost::asio::error::eof &&
             ec != boost::asio::error::try_again) {
           read_result = ec;
           return;
         }
       });
 }

 void TcpConn::prepareAsyncWrite(
     const char *buff, size_t len,
     boost::optional<boost::system::error_code> &write_result,
     std::size_t &bytes_written) {
   boost::asio::async_write(
       socket_, boost::asio::buffer(buff, len),
       [&write_result, &bytes_written](const boost::system::error_code &ec,
                                       const size_t n) {
         bytes_written = n;

         if (ec != boost::asio::error::eof &&
             ec != boost::asio::error::try_again) {
           write_result = ec;
           return;
         }
       });
 }

 }  // namespace client
 }  // namespace geode
 }  // namespace apache
	/*
	* Licensed to the Apache Software Foundation (ASF) under one or more
	* contributor license agreements. See the NOTICE file distributed with
	* this work for additional information regarding copyright ownership.
	* The ASF licenses this file to You under the Apache License, Version 2.0
	* (the "License"); you may not use this file except in compliance with
	* the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software
	* distributed under the License is distributed on an "AS IS" BASIS,
	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	* See the License for the specific language governing permissions and
	* limitations under the License.
	*/

	#include "TcpConn.hpp"

	#include <iomanip>
	#include <iostream>

	#include <boost/optional.hpp>
	#include <boost/system/error_code.hpp>
	#include <boost/system/system_error.hpp>

	#include "util/Log.hpp"

	namespace {
	template <int Level, int Name>
	class timeval {
	public:
	// This is not an instance of the template, but of the system provided type
	// to be written to the socket API.
	#if defined(_WINDOWS)
	using value_type = DWORD;
	#else
	using value_type = ::timeval;
	#endif

	private:
	value_type value_{};

	public:
	timeval() {}

	explicit timeval(value_type v) : value_(v) {}

	timeval &operator=(value_type v) {
	value_ = v;
	return *this;
	}

	value_type value() const { return value_; }

	template <typename Protocol>
	int level(const Protocol &) const {
	return Level;
	}

	template <typename Protocol>
	int name(const Protocol &) const {
	return Name;
	}

	template <typename Protocol>
	value_type *data(const Protocol &) {
	return &value_;
	}

	template <typename Protocol>
	const value_type *data(const Protocol &) const {
	return &value_;
	}

	template <typename Protocol>
	std::size_t size(const Protocol &) const {
	return sizeof(value_);
	}

	template <typename Protocol>
	void resize(const Protocol &, std::size_t s) {
	if (s != sizeof(value_)) {
	throw std::length_error("timeval socket option resize");
	}
	}
	};

	// Asio doesn't support these socket options directly, but every major platform
	// does. Timeout on IO socket operations are supported by the platform directly.
	// This means We can all receive without needing to use the timeout interface -
	// and more importantly, we can send while holding to per-operation time
	// constraints and without blocking indefinitely.
	//
	// The default timeout is infinite, or by setting the socket option to null,
	// which I won't provide - just don't construct a TcpConn with send and
	// receieve timeouts.
	typedef timeval<SOL_SOCKET, SO_SNDTIMEO> send_timeout;
	typedef timeval<SOL_SOCKET, SO_RCVTIMEO> receive_timeout;
	} // namespace

	namespace apache {
	namespace geode {
	namespace client {
	TcpConn::TcpConn(const std::string ipaddr,
	std::chrono::microseconds connect_timeout,
	int32_t maxBuffSizePool)
	: TcpConn{
	ipaddr.substr(0, ipaddr.find(':')),
	static_cast<uint16_t>(std::stoi(ipaddr.substr(ipaddr.find(':') + 1))),
	connect_timeout, maxBuffSizePool} {}

	TcpConn::TcpConn(const std::string host, uint16_t port,
	std::chrono::microseconds timeout, int32_t maxBuffSizePool)
	: socket_{io_context_} {
	auto beforeResolvePoint = std::chrono::system_clock::now();
	auto results = resolve(host, port, timeout);
	auto elapsedTime = std::chrono::duration<double, std::micro>(
	std::chrono::system_clock::now() - beforeResolvePoint);

	// We must connect first so we have a valid file descriptor to set options
	// on.
	auto connectTimeout = std::chrono::duration_cast<std::chrono::microseconds>(
	timeout - elapsedTime);
	connect(results, connectTimeout);

	socket_.set_option(::boost::asio::ip::tcp::no_delay{true});
	socket_.set_option(
	::boost::asio::socket_base::send_buffer_size{maxBuffSizePool});
	socket_.set_option(
	::boost::asio::socket_base::receive_buffer_size{maxBuffSizePool});
	}

	TcpConn::TcpConn(const std::string ipaddr,
	std::chrono::microseconds connect_timeout,
	int32_t maxBuffSizePool, std::chrono::microseconds send_time,
	std::chrono::microseconds receive_time)
	: TcpConn{ipaddr, connect_timeout, maxBuffSizePool} {
	#if defined(_WINDOWS)
	socket_.set_option(::send_timeout{static_cast<DWORD>(send_time.count())});
	socket_.set_option(
	::receive_timeout{static_cast<DWORD>(receive_time.count())});
	#else

	auto send_seconds =
	std::chrono::duration_cast<std::chrono::seconds>(send_time);
	auto send_microseconds =
	send_time % std::chrono::duration_cast<std::chrono::microseconds>(
	std::chrono::seconds{1});
	socket_.set_option(
	::send_timeout{{static_cast<int>(send_seconds.count()),
	static_cast<int>(send_microseconds.count())}});

	auto receive_seconds =
	std::chrono::duration_cast<std::chrono::seconds>(receive_time);
	auto receive_microseconds =
	receive_time % std::chrono::duration_cast<std::chrono::microseconds>(
	std::chrono::seconds{1});
	socket_.set_option(
	::receive_timeout{{static_cast<int>(receive_seconds.count()),
	static_cast<int>(receive_microseconds.count())}});
	#endif
	}

	TcpConn::~TcpConn() {
	std::stringstream ss;

	try {
	ss << "Disconnected " << socket_.local_endpoint() << " -> "
	<< socket_.remote_endpoint();

	socket_.shutdown(boost::asio::ip::tcp::socket::shutdown_both);

	} catch (...) {
	ss = std::stringstream{};

	ss << "Closed socket " << socket_.local_endpoint();
	}

	socket_.close();

	LOGFINE(ss.str());
	}

	size_t TcpConn::receive(char *buff, const size_t len,
	std::chrono::milliseconds timeout) {
	std::stringstream ss;
	ss << "Receiving " << len << " bytes from " << socket_.remote_endpoint()
	<< " -> " << socket_.local_endpoint();
	LOGDEBUG(ss.str());
	return receive(buff, len, timeout, true);
	}

	size_t TcpConn::receive_nothrowiftimeout(char *buff, const size_t len,
	std::chrono::milliseconds timeout) {
	std::stringstream ss;
	ss << "Receiving an unknown number of bytes from "
	<< socket_.remote_endpoint() << " -> " << socket_.local_endpoint();
	LOGDEBUG(ss.str());
	return receive(buff, len, timeout, false);
	}

	size_t TcpConn::receive(char *buff, const size_t len,
	std::chrono::milliseconds timeout,
	bool throwTimeoutException) {
	boost::optional<boost::system::error_code> read_result;
	std::size_t bytes_read = 0;

	auto beforeResolvePoint = std::chrono::system_clock::now();

	try {
	prepareAsyncRead(buff, len, read_result, bytes_read);
	io_context_.restart();
	io_context_.run_for(timeout);
	} catch (...) {
	LOGDEBUG("Throwing an unexpected read exception");
	throw;
	}

	if (read_result && *read_result) {
	LOGDEBUG("Throwing a read exception: %s", read_result->message().c_str());
	socket_.cancel();
	// Get the abort
	io_context_.restart();
	io_context_.run();
	throw boost::system::system_error{*read_result};
	}

	if (bytes_read == 0) {
	auto elapsedTime = std::chrono::duration<double, std::micro>(
	std::chrono::system_clock::now() - beforeResolvePoint);
	if (elapsedTime < timeout) {
	LOGDEBUG("Throwing an IO exception");
	socket_.cancel();
	// Get the abort
	io_context_.restart();
	io_context_.run();
	throw boost::system::system_error{boost::asio::error::broken_pipe};
	} else {
	LOGDEBUG("Throwing an eof exception");
	socket_.cancel();
	// Get the abort
	io_context_.restart();
	io_context_.run();
	throw boost::system::system_error{boost::asio::error::eof};
	}
	}

	if (bytes_read != len && throwTimeoutException) {
	LOGDEBUG("Throwing a read timeout exception");
	socket_.cancel();
	// Get the abort
	io_context_.restart();
	io_context_.run();
	throw boost::system::system_error{boost::asio::error::operation_aborted};
	}

	return bytes_read;
	}

	size_t TcpConn::send(const char *buff, const size_t len,
	std::chrono::milliseconds timeout) {
	std::stringstream ss;
	ss << "Sending " << len << " bytes from " << socket_.local_endpoint()
	<< " -> " << socket_.remote_endpoint();
	LOGDEBUG(ss.str());

	boost::optional<boost::system::error_code> write_result;
	std::size_t bytes_written = 0;

	try {
	prepareAsyncWrite(buff, len, write_result, bytes_written);
	io_context_.restart();
	io_context_.run_for(timeout);
	} catch (...) {
	LOGDEBUG("Throwing an unexpected write exception");
	throw;
	}

	if (write_result && *write_result) {
	LOGDEBUG("Throwing a write exception. %s", write_result->message().c_str());
	throw boost::system::system_error{*write_result};
	}

	if (bytes_written != len) {
	LOGDEBUG("Throwing a write timeout exception");
	socket_.cancel();
	// Get the abort
	io_context_.restart();
	io_context_.run();
	throw boost::system::system_error{boost::asio::error::operation_aborted};
	}

	return bytes_written;
	}

	// Return the local port for this TCP connection.
	uint16_t TcpConn::getPort() { return socket_.local_endpoint().port(); }

	void TcpConn::connect(boost::asio::ip::tcp::resolver::results_type r,
	std::chrono::microseconds timeout) {
	boost::optional<boost::system::error_code> connect_result;

	try {
	// We must connect first so we have a valid file descriptor to set
	// options on.
	boost::asio::async_connect(
	socket_, r,
	[&connect_result](const boost::system::error_code &ec,
	const boost::asio::ip::tcp::endpoint) {
	connect_result = ec;
	});

	io_context_.restart();
	io_context_.run_for(timeout);
	} catch (...) {
	LOGDEBUG("Throwing an unexpected connect exception");
	throw;
	}

	if (connect_result && *connect_result) {
	LOGDEBUG("Throwing a connect exception: %s",
	connect_result->message().c_str());
	throw boost::system::system_error{*connect_result};
	}

	if (!connect_result) {
	LOGDEBUG("Throwing a connect timeout exception");
	throw boost::system::system_error{boost::asio::error::operation_aborted};
	}

	std::stringstream ss;
	ss << "Connected " << socket_.local_endpoint() << " -> "
	<< socket_.remote_endpoint();
	LOGDEBUG(ss.str());
	}

	boost::asio::ip::tcp::resolver::results_type TcpConn::resolve(
	const std::string host, uint16_t port, std::chrono::microseconds timeout) {
	boost::optional<boost::system::error_code> resolve_result;
	boost::asio::ip::tcp::resolver::results_type results;

	try {
	boost::asio::ip::tcp::resolver resolver(io_context_);
	resolver.async_resolve(host, std::to_string(port),
	[&resolve_result, &results](
	const boost::system::error_code &ec,
	boost::asio::ip::tcp::resolver::results_type r) {
	if (ec) {
	resolve_result = ec;
	} else {
	resolve_result = boost::system::error_code{};
	results = r;
	}
	});

	io_context_.restart();
	io_context_.run_for(timeout);
	} catch (...) {
	LOGDEBUG("Throwing an unexpected resolve exception");
	throw;
	}

	if (resolve_result && *resolve_result) {
	LOGDEBUG("Throwing a resolve exception: %s",
	resolve_result->message().c_str());
	throw boost::system::system_error{*resolve_result};
	}

	if (!resolve_result) {
	LOGDEBUG("Throwing a resolve timeout exception");
	socket_.cancel();
	// Get the abort
	io_context_.restart();
	io_context_.run();
	throw boost::system::system_error{boost::asio::error::operation_aborted};
	}

	return results;
	}

	void TcpConn::prepareAsyncRead(
	char *buff, size_t len,
	boost::optional<boost::system::error_code> &read_result,
	std::size_t &bytes_read) {
	boost::asio::async_read(
	socket_, boost::asio::buffer(buff, len),
	[&read_result, &bytes_read](const boost::system::error_code &ec,
	const size_t n) {
	bytes_read = n;

	// EOF itself occurs when there is no data available on the socket at
	// the time of the read. It may simply imply data has yet to arrive.
	// Do nothing. Defer to timeout rather than assume a broken
	// connection.
	if (ec != boost::asio::error::eof &&
	ec != boost::asio::error::try_again) {
	read_result = ec;
	return;
	}
	});
	}

	void TcpConn::prepareAsyncWrite(
	const char *buff, size_t len,
	boost::optional<boost::system::error_code> &write_result,
	std::size_t &bytes_written) {
	boost::asio::async_write(
	socket_, boost::asio::buffer(buff, len),
	[&write_result, &bytes_written](const boost::system::error_code &ec,
	const size_t n) {
	bytes_written = n;

	if (ec != boost::asio::error::eof &&
	ec != boost::asio::error::try_again) {
	write_result = ec;
	return;
	}
	});
	}

	} // namespace client
	} // namespace geode
	} // namespace apache