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 <thread>

 #include <ace/SOCK_Connector.h>
 #include <boost/interprocess/mapped_region.hpp>

 #include <geode/ExceptionTypes.hpp>
 #include <geode/internal/chrono/duration.hpp>

 #include "util/Log.hpp"

 namespace apache {
 namespace geode {
 namespace client {

 const size_t TcpConn::kChunkSize = TcpConn::getDefaultChunkSize();

 void TcpConn::clearNagle(ACE_HANDLE sock) {
   int32_t val = 1;

   if (0 != ACE_OS::setsockopt(sock, IPPROTO_TCP, 1,
                               reinterpret_cast<const char*>(&val),
                               sizeof(val))) {
     int32_t lastError = ACE_OS::last_error();
     LOGERROR("Failed to set TCP_NODELAY on socket. Errno: %d: %s", lastError,
              ACE_OS::strerror(lastError));
   }
 }

 int32_t TcpConn::maxSize(ACE_HANDLE sock, int32_t flag, int32_t size) {
   int32_t val = 0;

   int32_t inc = 32120;
   val = size - (3 * inc);
   if (val < 0) val = 0;
   if (size == 0) size = maxBuffSizePool_;
   int32_t red = 0;
   int32_t lastRed = -1;
   while (lastRed != red) {
     lastRed = red;
     val += inc;
     if (0 != ACE_OS::setsockopt(sock, SOL_SOCKET, flag,
                                 reinterpret_cast<const char*>(&val),
                                 sizeof(val))) {
       int32_t lastError = ACE_OS::last_error();
       LOGERROR("Failed to set socket options. Errno: %d : %s ", lastError,
                ACE_OS::strerror(lastError));
     }
     int plen = sizeof(val);
     if (0 != ACE_OS::getsockopt(sock, SOL_SOCKET, flag,
                                 reinterpret_cast<char*>(&val), &plen)) {
       int32_t lastError = ACE_OS::last_error();
       LOGERROR(
           "Failed to get buffer size for flag %d on socket. Errno: %d : %s",
           flag, lastError, ACE_OS::strerror(lastError));
     }
 #ifdef _LINUX
     val /= 2;
 #endif
     if ((val >= maxBuffSizePool_) || (val >= size)) continue;
     red = val;
   }
   return val;
 }

 void TcpConn::createSocket(ACE_HANDLE sock) {
   LOGDEBUG("Creating plain socket stream");
   stream_ = std::unique_ptr<ACE_SOCK_Stream>(new ACE_SOCK_Stream(sock));
 }

 void TcpConn::init() {
 #ifdef WITH_IPV6
   ACE_HANDLE sock = ACE_OS::socket(inetAddress_.get_type(), SOCK_STREAM, 0);
 #else
   ACE_HANDLE sock = ACE_OS::socket(AF_INET, SOCK_STREAM, 0);
 #endif
   if (sock == ACE_INVALID_HANDLE) {
     int32_t lastError = ACE_OS::last_error();
     LOGERROR("Failed to create socket. Errno: %d: %s", lastError,
              ACE_OS::strerror(lastError));
     char msg[256];
     std::snprintf(msg, 256, "TcpConn::connect failed with errno: %d: %s",
                   lastError, ACE_OS::strerror(lastError));
     throw GeodeIOException(msg);
   }

   clearNagle(sock);

   int32_t readSize = 0;
   int32_t writeSize = 0;
   int32_t originalReadSize = readSize;
   readSize = maxSize(sock, SO_SNDBUF, readSize);
   if (originalReadSize != readSize) {
     // This should get logged once at startup and again only if it changes
     LOGFINEST("Using socket send buffer size of %d.", readSize);
   }
   int32_t originalWriteSize = writeSize;
   writeSize = maxSize(sock, SO_RCVBUF, writeSize);
   if (originalWriteSize != writeSize) {
     // This should get logged once at startup and again only if it changes
     LOGFINEST("Using socket receive buffer size of %d.", writeSize);
   }

   createSocket(sock);

   connect();
 }

 TcpConn::TcpConn(const std::string& address,
                  std::chrono::microseconds waitSeconds, int32_t maxBuffSizePool)
     : stream_(nullptr),
       maxBuffSizePool_(maxBuffSizePool),
       inetAddress_(address.c_str()),
       endpoint_(address),
       timeout_(waitSeconds) {}

 TcpConn::TcpConn(const std::string& hostname, uint16_t port,
                  std::chrono::microseconds waitSeconds, int32_t maxBuffSizePool)
     : stream_(nullptr),
       maxBuffSizePool_(maxBuffSizePool),
       inetAddress_(port, hostname.c_str()),
       endpoint_(hostname + ":" + std::to_string(port)),
       timeout_(waitSeconds) {}

 void TcpConn::connect() {
   using apache::geode::internal::chrono::duration::to_string;

   ACE_OS::signal(SIGPIPE, SIG_IGN);  // Ignore broken pipe

   LOGFINER("Connecting plain socket stream to " + endpoint_ + " waiting " +
            to_string(timeout_));

   const ACE_Time_Value aceTimeout(timeout_);
   const auto timeout =
       (timeout_ > std::chrono::microseconds::zero()) ? &aceTimeout : nullptr;
   if (ACE_SOCK_Connector{}.connect(*stream_, inetAddress_, timeout) == -1) {
     const auto lastError = ACE_OS::last_error();
     if (lastError == ETIME || lastError == ETIMEDOUT) {
       throw TimeoutException(
           "TcpConn::connect Attempt to connect timed out after " +
           to_string(timeout_) + ".");
     }
     close();
     throw GeodeIOException("TcpConn::connect failed with errno: " +
                            ACE_errno_to_string(lastError));
   }

   if (stream_->enable(ACE_NONBLOCK)) {
     LOGINFO("TcpConn::NONBLOCK: " + ACE_errno_to_string(ACE_OS::last_error()));
   }
 }

 void TcpConn::close() {
   if (stream_) {
     stream_->close();
     stream_ = nullptr;
   }
 }

 size_t TcpConn::receive(char* buff, size_t len,
                         std::chrono::microseconds waitSeconds) {
   return socketOp(SOCK_READ, buff, len, waitSeconds);
 }

 size_t TcpConn::send(const char* buff, size_t len,
                      std::chrono::microseconds waitSeconds) {
   return socketOp(SOCK_WRITE, const_cast<char*>(buff), len, waitSeconds);
 }

 size_t TcpConn::socketOp(TcpConn::SockOp op, char* buff, size_t len,
                          std::chrono::microseconds waitDuration) {
   {
     ACE_Time_Value waitTime(waitDuration);
     auto endTime = std::chrono::steady_clock::now() + waitDuration;
     size_t readLen = 0;
     ssize_t retVal;
     bool errnoSet = false;

     size_t totalsend = 0;
     while (len > 0 && waitTime > ACE_Time_Value::zero) {
       size_t sendlen;
       if (len > kChunkSize) {
         sendlen = kChunkSize;
         len -= kChunkSize;
       } else {
         sendlen = len;
         len = 0;
       }
       do {
         retVal = doOperation(op, buff, sendlen, waitTime, readLen);
         sendlen -= readLen;
         totalsend += readLen;
         if (retVal < 0) {
           int32_t lastError = ACE_OS::last_error();
           if (lastError == EAGAIN) {
             std::this_thread::sleep_for(std::chrono::microseconds(100));
           } else {
             errnoSet = true;
             break;
           }
         } else if (retVal == 0 && readLen == 0) {
           ACE_OS::last_error(EPIPE);
           errnoSet = true;
           break;
         }

         buff += readLen;
         if (sendlen == 0) break;
         waitTime = endTime - std::chrono::steady_clock::now();
         if (waitTime <= ACE_Time_Value::zero) break;
       } while (sendlen > 0);
       if (errnoSet) break;
     }

     if (len > 0 && !errnoSet) {
       ACE_OS::last_error(ETIME);
     }

     return totalsend;
   }
 }
 ssize_t TcpConn::doOperation(const TcpConn::SockOp& op, void* buff,
                              size_t sendlen, ACE_Time_Value& waitTime,
                              size_t& readLen) const {
   if (op == SOCK_READ) {
     return stream_->recv_n(buff, sendlen, &waitTime, &readLen);
   } else {
     return stream_->send_n(buff, sendlen, &waitTime, &readLen);
   }
 }

 //  Return the local port for this TCP connection.
 uint16_t TcpConn::getPort() {
   ACE_INET_Addr localAddr;
   stream_->get_local_addr(localAddr);
   return localAddr.get_port_number();
 }

 size_t TcpConn::getDefaultChunkSize() {
   //
   auto pageSize = boost::interprocess::mapped_region::get_page_size();
   if (pageSize > 16000000) {
     return 16000000;
   } else if (pageSize > 0) {
     return pageSize + (16000000 / pageSize) * pageSize;
   }

   return 16000000;
 }

 }  // 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 <thread>

	#include <ace/SOCK_Connector.h>
	#include <boost/interprocess/mapped_region.hpp>

	#include <geode/ExceptionTypes.hpp>
	#include <geode/internal/chrono/duration.hpp>

	#include "util/Log.hpp"

	namespace apache {
	namespace geode {
	namespace client {

	const size_t TcpConn::kChunkSize = TcpConn::getDefaultChunkSize();

	void TcpConn::clearNagle(ACE_HANDLE sock) {
	int32_t val = 1;

	if (0 != ACE_OS::setsockopt(sock, IPPROTO_TCP, 1,
	reinterpret_cast<const char*>(&val),
	sizeof(val))) {
	int32_t lastError = ACE_OS::last_error();
	LOGERROR("Failed to set TCP_NODELAY on socket. Errno: %d: %s", lastError,
	ACE_OS::strerror(lastError));
	}
	}

	int32_t TcpConn::maxSize(ACE_HANDLE sock, int32_t flag, int32_t size) {
	int32_t val = 0;

	int32_t inc = 32120;
	val = size - (3 * inc);
	if (val < 0) val = 0;
	if (size == 0) size = maxBuffSizePool_;
	int32_t red = 0;
	int32_t lastRed = -1;
	while (lastRed != red) {
	lastRed = red;
	val += inc;
	if (0 != ACE_OS::setsockopt(sock, SOL_SOCKET, flag,
	reinterpret_cast<const char*>(&val),
	sizeof(val))) {
	int32_t lastError = ACE_OS::last_error();
	LOGERROR("Failed to set socket options. Errno: %d : %s ", lastError,
	ACE_OS::strerror(lastError));
	}
	int plen = sizeof(val);
	if (0 != ACE_OS::getsockopt(sock, SOL_SOCKET, flag,
	reinterpret_cast<char*>(&val), &plen)) {
	int32_t lastError = ACE_OS::last_error();
	LOGERROR(
	"Failed to get buffer size for flag %d on socket. Errno: %d : %s",
	flag, lastError, ACE_OS::strerror(lastError));
	}
	#ifdef _LINUX
	val /= 2;
	#endif
	if ((val >= maxBuffSizePool_) \|\| (val >= size)) continue;
	red = val;
	}
	return val;
	}

	void TcpConn::createSocket(ACE_HANDLE sock) {
	LOGDEBUG("Creating plain socket stream");
	stream_ = std::unique_ptr<ACE_SOCK_Stream>(new ACE_SOCK_Stream(sock));
	}

	void TcpConn::init() {
	#ifdef WITH_IPV6
	ACE_HANDLE sock = ACE_OS::socket(inetAddress_.get_type(), SOCK_STREAM, 0);
	#else
	ACE_HANDLE sock = ACE_OS::socket(AF_INET, SOCK_STREAM, 0);
	#endif
	if (sock == ACE_INVALID_HANDLE) {
	int32_t lastError = ACE_OS::last_error();
	LOGERROR("Failed to create socket. Errno: %d: %s", lastError,
	ACE_OS::strerror(lastError));
	char msg[256];
	std::snprintf(msg, 256, "TcpConn::connect failed with errno: %d: %s",
	lastError, ACE_OS::strerror(lastError));
	throw GeodeIOException(msg);
	}

	clearNagle(sock);

	int32_t readSize = 0;
	int32_t writeSize = 0;
	int32_t originalReadSize = readSize;
	readSize = maxSize(sock, SO_SNDBUF, readSize);
	if (originalReadSize != readSize) {
	// This should get logged once at startup and again only if it changes
	LOGFINEST("Using socket send buffer size of %d.", readSize);
	}
	int32_t originalWriteSize = writeSize;
	writeSize = maxSize(sock, SO_RCVBUF, writeSize);
	if (originalWriteSize != writeSize) {
	// This should get logged once at startup and again only if it changes
	LOGFINEST("Using socket receive buffer size of %d.", writeSize);
	}

	createSocket(sock);

	connect();
	}

	TcpConn::TcpConn(const std::string& address,
	std::chrono::microseconds waitSeconds, int32_t maxBuffSizePool)
	: stream_(nullptr),
	maxBuffSizePool_(maxBuffSizePool),
	inetAddress_(address.c_str()),
	endpoint_(address),
	timeout_(waitSeconds) {}

	TcpConn::TcpConn(const std::string& hostname, uint16_t port,
	std::chrono::microseconds waitSeconds, int32_t maxBuffSizePool)
	: stream_(nullptr),
	maxBuffSizePool_(maxBuffSizePool),
	inetAddress_(port, hostname.c_str()),
	endpoint_(hostname + ":" + std::to_string(port)),
	timeout_(waitSeconds) {}

	void TcpConn::connect() {
	using apache::geode::internal::chrono::duration::to_string;

	ACE_OS::signal(SIGPIPE, SIG_IGN); // Ignore broken pipe

	LOGFINER("Connecting plain socket stream to " + endpoint_ + " waiting " +
	to_string(timeout_));

	const ACE_Time_Value aceTimeout(timeout_);
	const auto timeout =
	(timeout_ > std::chrono::microseconds::zero()) ? &aceTimeout : nullptr;
	if (ACE_SOCK_Connector{}.connect(*stream_, inetAddress_, timeout) == -1) {
	const auto lastError = ACE_OS::last_error();
	if (lastError == ETIME \|\| lastError == ETIMEDOUT) {
	throw TimeoutException(
	"TcpConn::connect Attempt to connect timed out after " +
	to_string(timeout_) + ".");
	}
	close();
	throw GeodeIOException("TcpConn::connect failed with errno: " +
	ACE_errno_to_string(lastError));
	}

	if (stream_->enable(ACE_NONBLOCK)) {
	LOGINFO("TcpConn::NONBLOCK: " + ACE_errno_to_string(ACE_OS::last_error()));
	}
	}

	void TcpConn::close() {
	if (stream_) {
	stream_->close();
	stream_ = nullptr;
	}
	}

	size_t TcpConn::receive(char* buff, size_t len,
	std::chrono::microseconds waitSeconds) {
	return socketOp(SOCK_READ, buff, len, waitSeconds);
	}

	size_t TcpConn::send(const char* buff, size_t len,
	std::chrono::microseconds waitSeconds) {
	return socketOp(SOCK_WRITE, const_cast<char*>(buff), len, waitSeconds);
	}

	size_t TcpConn::socketOp(TcpConn::SockOp op, char* buff, size_t len,
	std::chrono::microseconds waitDuration) {
	{
	ACE_Time_Value waitTime(waitDuration);
	auto endTime = std::chrono::steady_clock::now() + waitDuration;
	size_t readLen = 0;
	ssize_t retVal;
	bool errnoSet = false;

	size_t totalsend = 0;
	while (len > 0 && waitTime > ACE_Time_Value::zero) {
	size_t sendlen;
	if (len > kChunkSize) {
	sendlen = kChunkSize;
	len -= kChunkSize;
	} else {
	sendlen = len;
	len = 0;
	}
	do {
	retVal = doOperation(op, buff, sendlen, waitTime, readLen);
	sendlen -= readLen;
	totalsend += readLen;
	if (retVal < 0) {
	int32_t lastError = ACE_OS::last_error();
	if (lastError == EAGAIN) {
	std::this_thread::sleep_for(std::chrono::microseconds(100));
	} else {
	errnoSet = true;
	break;
	}
	} else if (retVal == 0 && readLen == 0) {
	ACE_OS::last_error(EPIPE);
	errnoSet = true;
	break;
	}

	buff += readLen;
	if (sendlen == 0) break;
	waitTime = endTime - std::chrono::steady_clock::now();
	if (waitTime <= ACE_Time_Value::zero) break;
	} while (sendlen > 0);
	if (errnoSet) break;
	}

	if (len > 0 && !errnoSet) {
	ACE_OS::last_error(ETIME);
	}

	return totalsend;
	}
	}
	ssize_t TcpConn::doOperation(const TcpConn::SockOp& op, void* buff,
	size_t sendlen, ACE_Time_Value& waitTime,
	size_t& readLen) const {
	if (op == SOCK_READ) {
	return stream_->recv_n(buff, sendlen, &waitTime, &readLen);
	} else {
	return stream_->send_n(buff, sendlen, &waitTime, &readLen);
	}
	}

	// Return the local port for this TCP connection.
	uint16_t TcpConn::getPort() {
	ACE_INET_Addr localAddr;
	stream_->get_local_addr(localAddr);
	return localAddr.get_port_number();
	}

	size_t TcpConn::getDefaultChunkSize() {
	//
	auto pageSize = boost::interprocess::mapped_region::get_page_size();
	if (pageSize > 16000000) {
	return 16000000;
	} else if (pageSize > 0) {
	return pageSize + (16000000 / pageSize) * pageSize;
	}

	return 16000000;
	}

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