diff options
Diffstat (limited to 'lib/cpp/src/thrift/transport/TServerSocket.cpp')
-rw-r--r--[-rwxr-xr-x] | lib/cpp/src/thrift/transport/TServerSocket.cpp | 275 |
1 files changed, 153 insertions, 122 deletions
diff --git a/lib/cpp/src/thrift/transport/TServerSocket.cpp b/lib/cpp/src/thrift/transport/TServerSocket.cpp index 0ce1bce60..e228dabf6 100755..100644 --- a/lib/cpp/src/thrift/transport/TServerSocket.cpp +++ b/lib/cpp/src/thrift/transport/TServerSocket.cpp @@ -52,76 +52,78 @@ #endif #ifndef SOCKOPT_CAST_T -# ifndef _WIN32 -# define SOCKOPT_CAST_T void -# else -# define SOCKOPT_CAST_T char -# endif // _WIN32 +#ifndef _WIN32 +#define SOCKOPT_CAST_T void +#else +#define SOCKOPT_CAST_T char +#endif // _WIN32 #endif -template<class T> +template <class T> inline const SOCKOPT_CAST_T* const_cast_sockopt(const T* v) { - return reinterpret_cast<const SOCKOPT_CAST_T*>(v); + return reinterpret_cast<const SOCKOPT_CAST_T*>(v); } -template<class T> +template <class T> inline SOCKOPT_CAST_T* cast_sockopt(T* v) { - return reinterpret_cast<SOCKOPT_CAST_T*>(v); + return reinterpret_cast<SOCKOPT_CAST_T*>(v); } -namespace apache { namespace thrift { namespace transport { +namespace apache { +namespace thrift { +namespace transport { using namespace std; using boost::shared_ptr; -TServerSocket::TServerSocket(int port) : - port_(port), - serverSocket_(THRIFT_INVALID_SOCKET), - acceptBacklog_(DEFAULT_BACKLOG), - sendTimeout_(0), - recvTimeout_(0), - accTimeout_(-1), - retryLimit_(0), - retryDelay_(0), - tcpSendBuffer_(0), - tcpRecvBuffer_(0), - keepAlive_(false), - intSock1_(THRIFT_INVALID_SOCKET), - intSock2_(THRIFT_INVALID_SOCKET) -{} - -TServerSocket::TServerSocket(int port, int sendTimeout, int recvTimeout) : - port_(port), - serverSocket_(THRIFT_INVALID_SOCKET), - acceptBacklog_(DEFAULT_BACKLOG), - sendTimeout_(sendTimeout), - recvTimeout_(recvTimeout), - accTimeout_(-1), - retryLimit_(0), - retryDelay_(0), - tcpSendBuffer_(0), - tcpRecvBuffer_(0), - keepAlive_(false), - intSock1_(THRIFT_INVALID_SOCKET), - intSock2_(THRIFT_INVALID_SOCKET) -{} - -TServerSocket::TServerSocket(string path) : - port_(0), - path_(path), - serverSocket_(THRIFT_INVALID_SOCKET), - acceptBacklog_(DEFAULT_BACKLOG), - sendTimeout_(0), - recvTimeout_(0), - accTimeout_(-1), - retryLimit_(0), - retryDelay_(0), - tcpSendBuffer_(0), - tcpRecvBuffer_(0), - keepAlive_(false), - intSock1_(THRIFT_INVALID_SOCKET), - intSock2_(THRIFT_INVALID_SOCKET) -{} +TServerSocket::TServerSocket(int port) + : port_(port), + serverSocket_(THRIFT_INVALID_SOCKET), + acceptBacklog_(DEFAULT_BACKLOG), + sendTimeout_(0), + recvTimeout_(0), + accTimeout_(-1), + retryLimit_(0), + retryDelay_(0), + tcpSendBuffer_(0), + tcpRecvBuffer_(0), + keepAlive_(false), + intSock1_(THRIFT_INVALID_SOCKET), + intSock2_(THRIFT_INVALID_SOCKET) { +} + +TServerSocket::TServerSocket(int port, int sendTimeout, int recvTimeout) + : port_(port), + serverSocket_(THRIFT_INVALID_SOCKET), + acceptBacklog_(DEFAULT_BACKLOG), + sendTimeout_(sendTimeout), + recvTimeout_(recvTimeout), + accTimeout_(-1), + retryLimit_(0), + retryDelay_(0), + tcpSendBuffer_(0), + tcpRecvBuffer_(0), + keepAlive_(false), + intSock1_(THRIFT_INVALID_SOCKET), + intSock2_(THRIFT_INVALID_SOCKET) { +} + +TServerSocket::TServerSocket(string path) + : port_(0), + path_(path), + serverSocket_(THRIFT_INVALID_SOCKET), + acceptBacklog_(DEFAULT_BACKLOG), + sendTimeout_(0), + recvTimeout_(0), + accTimeout_(-1), + retryLimit_(0), + retryDelay_(0), + tcpSendBuffer_(0), + tcpRecvBuffer_(0), + keepAlive_(false), + intSock1_(THRIFT_INVALID_SOCKET), + intSock2_(THRIFT_INVALID_SOCKET) { +} TServerSocket::~TServerSocket() { close(); @@ -161,7 +163,7 @@ void TServerSocket::setTcpRecvBuffer(int tcpRecvBuffer) { void TServerSocket::listen() { #ifdef _WIN32 - TWinsockSingleton::create(); + TWinsockSingleton::create(); #endif // _WIN32 THRIFT_SOCKET sv[2]; if (-1 == THRIFT_SOCKETPAIR(AF_LOCAL, SOCK_STREAM, 0, sv)) { @@ -187,7 +189,8 @@ void TServerSocket::listen() { if (error) { GlobalOutput.printf("getaddrinfo %d: %s", error, THRIFT_GAI_STRERROR(error)); close(); - throw TTransportException(TTransportException::NOT_OPEN, "Could not resolve host for server socket."); + throw TTransportException(TTransportException::NOT_OPEN, + "Could not resolve host for server socket."); } // Pick the ipv6 address first since ipv4 addresses can be mapped @@ -197,7 +200,7 @@ void TServerSocket::listen() { break; } - if (! path_.empty()) { + if (!path_.empty()) { serverSocket_ = socket(PF_UNIX, SOCK_STREAM, IPPROTO_IP); } else { serverSocket_ = socket(res->ai_family, res->ai_socktype, res->ai_protocol); @@ -207,72 +210,93 @@ void TServerSocket::listen() { int errno_copy = THRIFT_GET_SOCKET_ERROR; GlobalOutput.perror("TServerSocket::listen() socket() ", errno_copy); close(); - throw TTransportException(TTransportException::NOT_OPEN, "Could not create server socket.", errno_copy); + throw TTransportException(TTransportException::NOT_OPEN, + "Could not create server socket.", + errno_copy); } // Set THRIFT_NO_SOCKET_CACHING to prevent 2MSL delay on accept int one = 1; - if (-1 == setsockopt(serverSocket_, SOL_SOCKET, THRIFT_NO_SOCKET_CACHING, - cast_sockopt(&one), sizeof(one))) { - //ignore errors coming out of this setsockopt on Windows. This is because - //SO_EXCLUSIVEADDRUSE requires admin privileges on WinXP, but we don't - //want to force servers to be an admin. + if (-1 == setsockopt(serverSocket_, + SOL_SOCKET, + THRIFT_NO_SOCKET_CACHING, + cast_sockopt(&one), + sizeof(one))) { +// ignore errors coming out of this setsockopt on Windows. This is because +// SO_EXCLUSIVEADDRUSE requires admin privileges on WinXP, but we don't +// want to force servers to be an admin. #ifndef _WIN32 int errno_copy = THRIFT_GET_SOCKET_ERROR; - GlobalOutput.perror("TServerSocket::listen() setsockopt() THRIFT_NO_SOCKET_CACHING ", errno_copy); + GlobalOutput.perror("TServerSocket::listen() setsockopt() THRIFT_NO_SOCKET_CACHING ", + errno_copy); close(); - throw TTransportException(TTransportException::NOT_OPEN, "Could not set THRIFT_NO_SOCKET_CACHING", errno_copy); + throw TTransportException(TTransportException::NOT_OPEN, + "Could not set THRIFT_NO_SOCKET_CACHING", + errno_copy); #endif } // Set TCP buffer sizes if (tcpSendBuffer_ > 0) { - if (-1 == setsockopt(serverSocket_, SOL_SOCKET, SO_SNDBUF, - cast_sockopt(&tcpSendBuffer_), sizeof(tcpSendBuffer_))) { + if (-1 == setsockopt(serverSocket_, + SOL_SOCKET, + SO_SNDBUF, + cast_sockopt(&tcpSendBuffer_), + sizeof(tcpSendBuffer_))) { int errno_copy = THRIFT_GET_SOCKET_ERROR; GlobalOutput.perror("TServerSocket::listen() setsockopt() SO_SNDBUF ", errno_copy); close(); - throw TTransportException(TTransportException::NOT_OPEN, "Could not set SO_SNDBUF", errno_copy); + throw TTransportException(TTransportException::NOT_OPEN, + "Could not set SO_SNDBUF", + errno_copy); } } if (tcpRecvBuffer_ > 0) { - if (-1 == setsockopt(serverSocket_, SOL_SOCKET, SO_RCVBUF, - cast_sockopt(&tcpRecvBuffer_), sizeof(tcpRecvBuffer_))) { + if (-1 == setsockopt(serverSocket_, + SOL_SOCKET, + SO_RCVBUF, + cast_sockopt(&tcpRecvBuffer_), + sizeof(tcpRecvBuffer_))) { int errno_copy = THRIFT_GET_SOCKET_ERROR; GlobalOutput.perror("TServerSocket::listen() setsockopt() SO_RCVBUF ", errno_copy); close(); - throw TTransportException(TTransportException::NOT_OPEN, "Could not set SO_RCVBUF", errno_copy); + throw TTransportException(TTransportException::NOT_OPEN, + "Could not set SO_RCVBUF", + errno_copy); } } - // Defer accept - #ifdef TCP_DEFER_ACCEPT +// Defer accept +#ifdef TCP_DEFER_ACCEPT if (path_.empty()) { - if (-1 == setsockopt(serverSocket_, IPPROTO_TCP, TCP_DEFER_ACCEPT, - &one, sizeof(one))) { + if (-1 == setsockopt(serverSocket_, IPPROTO_TCP, TCP_DEFER_ACCEPT, &one, sizeof(one))) { int errno_copy = THRIFT_GET_SOCKET_ERROR; GlobalOutput.perror("TServerSocket::listen() setsockopt() TCP_DEFER_ACCEPT ", errno_copy); close(); - throw TTransportException(TTransportException::NOT_OPEN, "Could not set TCP_DEFER_ACCEPT", errno_copy); + throw TTransportException(TTransportException::NOT_OPEN, + "Could not set TCP_DEFER_ACCEPT", + errno_copy); } } - #endif // #ifdef TCP_DEFER_ACCEPT +#endif // #ifdef TCP_DEFER_ACCEPT - #ifdef IPV6_V6ONLY +#ifdef IPV6_V6ONLY if (res->ai_family == AF_INET6 && path_.empty()) { int zero = 0; - if (-1 == setsockopt(serverSocket_, IPPROTO_IPV6, IPV6_V6ONLY, - cast_sockopt(&zero), sizeof(zero))) { + if (-1 == setsockopt(serverSocket_, + IPPROTO_IPV6, + IPV6_V6ONLY, + cast_sockopt(&zero), + sizeof(zero))) { GlobalOutput.perror("TServerSocket::listen() IPV6_V6ONLY ", THRIFT_GET_SOCKET_ERROR); } } - #endif // #ifdef IPV6_V6ONLY +#endif // #ifdef IPV6_V6ONLY // Turn linger off, don't want to block on calls to close struct linger ling = {0, 0}; - if (-1 == setsockopt(serverSocket_, SOL_SOCKET, SO_LINGER, - cast_sockopt(&ling), sizeof(ling))) { + if (-1 == setsockopt(serverSocket_, SOL_SOCKET, SO_LINGER, cast_sockopt(&ling), sizeof(ling))) { int errno_copy = THRIFT_GET_SOCKET_ERROR; GlobalOutput.perror("TServerSocket::listen() setsockopt() SO_LINGER ", errno_copy); close(); @@ -282,12 +306,14 @@ void TServerSocket::listen() { // Unix Sockets do not need that if (path_.empty()) { // TCP Nodelay, speed over bandwidth - if (-1 == setsockopt(serverSocket_, IPPROTO_TCP, TCP_NODELAY, - cast_sockopt(&one), sizeof(one))) { + if (-1 + == setsockopt(serverSocket_, IPPROTO_TCP, TCP_NODELAY, cast_sockopt(&one), sizeof(one))) { int errno_copy = THRIFT_GET_SOCKET_ERROR; GlobalOutput.perror("TServerSocket::listen() setsockopt() TCP_NODELAY ", errno_copy); close(); - throw TTransportException(TTransportException::NOT_OPEN, "Could not set TCP_NODELAY", errno_copy); + throw TTransportException(TTransportException::NOT_OPEN, + "Could not set TCP_NODELAY", + errno_copy); } } @@ -310,7 +336,7 @@ void TServerSocket::listen() { // always seem to work. The client can configure the retry variables. int retries = 0; - if (! path_.empty()) { + if (!path_.empty()) { #ifndef _WIN32 @@ -328,14 +354,15 @@ void TServerSocket::listen() { socklen_t structlen = static_cast<socklen_t>(sizeof(address)); do { - if (0 == ::bind(serverSocket_, (struct sockaddr *) &address, structlen)) { + if (0 == ::bind(serverSocket_, (struct sockaddr*)&address, structlen)) { break; } // use short circuit evaluation here to only sleep if we need to } while ((retries++ < retryLimit_) && (THRIFT_SLEEP_SEC(retryDelay_) == 0)); #else GlobalOutput.perror("TSocket::open() Unix Domain socket path not supported on windows", -99); - throw TTransportException(TTransportException::NOT_OPEN, " Unix Domain socket path not supported"); + throw TTransportException(TTransportException::NOT_OPEN, + " Unix Domain socket path not supported"); #endif } else { do { @@ -356,16 +383,12 @@ void TServerSocket::listen() { if (::getsockname(serverSocket_, &sa, &len) < 0) { int errno_copy = errno; GlobalOutput.perror("TServerSocket::getPort() getsockname() ", errno_copy); - } - else { + } else { if (sa.sa_family == AF_INET6) { - const struct sockaddr_in6* - sin = reinterpret_cast<const struct sockaddr_in6 *>(&sa); + const struct sockaddr_in6* sin = reinterpret_cast<const struct sockaddr_in6*>(&sa); port_ = ntohs(sin->sin6_port); - } - else { - const struct sockaddr_in* - sin = reinterpret_cast<const struct sockaddr_in *>(&sa); + } else { + const struct sockaddr_in* sin = reinterpret_cast<const struct sockaddr_in*>(&sa); port_ = ntohs(sin->sin_port); } } @@ -375,19 +398,20 @@ void TServerSocket::listen() { // throw an error if we failed to bind properly if (retries > retryLimit_) { char errbuf[1024]; - if (! path_.empty()) { + if (!path_.empty()) { sprintf(errbuf, "TServerSocket::listen() PATH %s", path_.c_str()); - } - else { + } else { sprintf(errbuf, "TServerSocket::listen() BIND %d", port_); } GlobalOutput(errbuf); close(); - throw TTransportException(TTransportException::NOT_OPEN, "Could not bind", + throw TTransportException(TTransportException::NOT_OPEN, + "Could not bind", THRIFT_GET_SOCKET_ERROR); } - if(listenCallback_) listenCallback_(serverSocket_); + if (listenCallback_) + listenCallback_(serverSocket_); // Call listen if (-1 == ::listen(serverSocket_, acceptBacklog_)) { @@ -401,7 +425,7 @@ void TServerSocket::listen() { } int TServerSocket::getPort() { - return port_; + return port_; } shared_ptr<TTransport> TServerSocket::acceptImpl() { @@ -415,7 +439,7 @@ shared_ptr<TTransport> TServerSocket::acceptImpl() { int numEintrs = 0; while (true) { - std::memset(fds, 0 , sizeof(fds)); + std::memset(fds, 0, sizeof(fds)); fds[0].fd = serverSocket_; fds[0].events = THRIFT_POLLIN; if (intSock2_ != THRIFT_INVALID_SOCKET) { @@ -440,11 +464,11 @@ shared_ptr<TTransport> TServerSocket::acceptImpl() { throw TTransportException(TTransportException::UNKNOWN, "Unknown", errno_copy); } else if (ret > 0) { // Check for an interrupt signal - if (intSock2_ != THRIFT_INVALID_SOCKET - && (fds[1].revents & THRIFT_POLLIN)) { + if (intSock2_ != THRIFT_INVALID_SOCKET && (fds[1].revents & THRIFT_POLLIN)) { int8_t buf; if (-1 == recv(intSock2_, cast_sockopt(&buf), sizeof(int8_t), 0)) { - GlobalOutput.perror("TServerSocket::acceptImpl() recv() interrupt ", THRIFT_GET_SOCKET_ERROR); + GlobalOutput.perror("TServerSocket::acceptImpl() recv() interrupt ", + THRIFT_GET_SOCKET_ERROR); } throw TTransportException(TTransportException::INTERRUPTED); } @@ -461,9 +485,8 @@ shared_ptr<TTransport> TServerSocket::acceptImpl() { struct sockaddr_storage clientAddress; int size = sizeof(clientAddress); - THRIFT_SOCKET clientSocket = ::accept(serverSocket_, - (struct sockaddr *) &clientAddress, - (socklen_t *) &size); + THRIFT_SOCKET clientSocket + = ::accept(serverSocket_, (struct sockaddr*)&clientAddress, (socklen_t*)&size); if (clientSocket == -1) { int errno_copy = THRIFT_GET_SOCKET_ERROR; @@ -476,13 +499,19 @@ shared_ptr<TTransport> TServerSocket::acceptImpl() { if (flags == -1) { int errno_copy = THRIFT_GET_SOCKET_ERROR; GlobalOutput.perror("TServerSocket::acceptImpl() THRIFT_FCNTL() THRIFT_F_GETFL ", errno_copy); - throw TTransportException(TTransportException::UNKNOWN, "THRIFT_FCNTL(THRIFT_F_GETFL)", errno_copy); + throw TTransportException(TTransportException::UNKNOWN, + "THRIFT_FCNTL(THRIFT_F_GETFL)", + errno_copy); } if (-1 == THRIFT_FCNTL(clientSocket, THRIFT_F_SETFL, flags & ~THRIFT_O_NONBLOCK)) { int errno_copy = THRIFT_GET_SOCKET_ERROR; - GlobalOutput.perror("TServerSocket::acceptImpl() THRIFT_FCNTL() THRIFT_F_SETFL ~THRIFT_O_NONBLOCK ", errno_copy); - throw TTransportException(TTransportException::UNKNOWN, "THRIFT_FCNTL(THRIFT_F_SETFL)", errno_copy); + GlobalOutput + .perror("TServerSocket::acceptImpl() THRIFT_FCNTL() THRIFT_F_SETFL ~THRIFT_O_NONBLOCK ", + errno_copy); + throw TTransportException(TTransportException::UNKNOWN, + "THRIFT_FCNTL(THRIFT_F_SETFL)", + errno_copy); } shared_ptr<TSocket> client = createSocket(clientSocket); @@ -495,9 +524,10 @@ shared_ptr<TTransport> TServerSocket::acceptImpl() { if (keepAlive_) { client->setKeepAlive(keepAlive_); } - client->setCachedAddress((sockaddr*) &clientAddress, size); + client->setCachedAddress((sockaddr*)&clientAddress, size); - if(acceptCallback_) acceptCallback_(clientSocket); + if (acceptCallback_) + acceptCallback_(clientSocket); return client; } @@ -521,7 +551,7 @@ void TServerSocket::close() { ::THRIFT_CLOSESOCKET(serverSocket_); } if (intSock1_ != THRIFT_INVALID_SOCKET) { - ::THRIFT_CLOSESOCKET(intSock1_); + ::THRIFT_CLOSESOCKET(intSock1_); } if (intSock2_ != THRIFT_INVALID_SOCKET) { ::THRIFT_CLOSESOCKET(intSock2_); @@ -530,5 +560,6 @@ void TServerSocket::close() { intSock1_ = THRIFT_INVALID_SOCKET; intSock2_ = THRIFT_INVALID_SOCKET; } - -}}} // apache::thrift::transport +} +} +} // apache::thrift::transport |