Thrift now a TLP - INFRA-3116
git-svn-id: https://svn.apache.org/repos/asf/thrift/branches/0.1.x@1028168 13f79535-47bb-0310-9956-ffa450edef68
diff --git a/lib/cpp/src/transport/TSocket.cpp b/lib/cpp/src/transport/TSocket.cpp
new file mode 100644
index 0000000..3395dab
--- /dev/null
+++ b/lib/cpp/src/transport/TSocket.cpp
@@ -0,0 +1,589 @@
+/*
+ * 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 <config.h>
+#include <cstring>
+#include <sstream>
+#include <sys/socket.h>
+#include <sys/poll.h>
+#include <sys/types.h>
+#include <arpa/inet.h>
+#include <netinet/in.h>
+#include <netinet/tcp.h>
+#include <netdb.h>
+#include <unistd.h>
+#include <errno.h>
+#include <fcntl.h>
+
+#include "concurrency/Monitor.h"
+#include "TSocket.h"
+#include "TTransportException.h"
+
+namespace apache { namespace thrift { namespace transport {
+
+using namespace std;
+
+// Global var to track total socket sys calls
+uint32_t g_socket_syscalls = 0;
+
+/**
+ * TSocket implementation.
+ *
+ */
+
+TSocket::TSocket(string host, int port) :
+ host_(host),
+ port_(port),
+ socket_(-1),
+ connTimeout_(0),
+ sendTimeout_(0),
+ recvTimeout_(0),
+ lingerOn_(1),
+ lingerVal_(0),
+ noDelay_(1),
+ maxRecvRetries_(5) {
+ recvTimeval_.tv_sec = (int)(recvTimeout_/1000);
+ recvTimeval_.tv_usec = (int)((recvTimeout_%1000)*1000);
+}
+
+TSocket::TSocket() :
+ host_(""),
+ port_(0),
+ socket_(-1),
+ connTimeout_(0),
+ sendTimeout_(0),
+ recvTimeout_(0),
+ lingerOn_(1),
+ lingerVal_(0),
+ noDelay_(1),
+ maxRecvRetries_(5) {
+ recvTimeval_.tv_sec = (int)(recvTimeout_/1000);
+ recvTimeval_.tv_usec = (int)((recvTimeout_%1000)*1000);
+}
+
+TSocket::TSocket(int socket) :
+ host_(""),
+ port_(0),
+ socket_(socket),
+ connTimeout_(0),
+ sendTimeout_(0),
+ recvTimeout_(0),
+ lingerOn_(1),
+ lingerVal_(0),
+ noDelay_(1),
+ maxRecvRetries_(5) {
+ recvTimeval_.tv_sec = (int)(recvTimeout_/1000);
+ recvTimeval_.tv_usec = (int)((recvTimeout_%1000)*1000);
+}
+
+TSocket::~TSocket() {
+ close();
+}
+
+bool TSocket::isOpen() {
+ return (socket_ >= 0);
+}
+
+bool TSocket::peek() {
+ if (!isOpen()) {
+ return false;
+ }
+ uint8_t buf;
+ int r = recv(socket_, &buf, 1, MSG_PEEK);
+ if (r == -1) {
+ int errno_copy = errno;
+ #ifdef __FreeBSD__
+ /* shigin:
+ * freebsd returns -1 and ECONNRESET if socket was closed by
+ * the other side
+ */
+ if (errno_copy == ECONNRESET)
+ {
+ close();
+ return false;
+ }
+ #endif
+ GlobalOutput.perror("TSocket::peek() recv() " + getSocketInfo(), errno_copy);
+ throw TTransportException(TTransportException::UNKNOWN, "recv()", errno_copy);
+ }
+ return (r > 0);
+}
+
+void TSocket::openConnection(struct addrinfo *res) {
+ if (isOpen()) {
+ throw TTransportException(TTransportException::ALREADY_OPEN);
+ }
+
+ socket_ = socket(res->ai_family, res->ai_socktype, res->ai_protocol);
+ if (socket_ == -1) {
+ int errno_copy = errno;
+ GlobalOutput.perror("TSocket::open() socket() " + getSocketInfo(), errno_copy);
+ throw TTransportException(TTransportException::NOT_OPEN, "socket()", errno_copy);
+ }
+
+ // Send timeout
+ if (sendTimeout_ > 0) {
+ setSendTimeout(sendTimeout_);
+ }
+
+ // Recv timeout
+ if (recvTimeout_ > 0) {
+ setRecvTimeout(recvTimeout_);
+ }
+
+ // Linger
+ setLinger(lingerOn_, lingerVal_);
+
+ // No delay
+ setNoDelay(noDelay_);
+
+ // Set the socket to be non blocking for connect if a timeout exists
+ int flags = fcntl(socket_, F_GETFL, 0);
+ if (connTimeout_ > 0) {
+ if (-1 == fcntl(socket_, F_SETFL, flags | O_NONBLOCK)) {
+ int errno_copy = errno;
+ GlobalOutput.perror("TSocket::open() fcntl() " + getSocketInfo(), errno_copy);
+ throw TTransportException(TTransportException::NOT_OPEN, "fcntl() failed", errno_copy);
+ }
+ } else {
+ if (-1 == fcntl(socket_, F_SETFL, flags & ~O_NONBLOCK)) {
+ int errno_copy = errno;
+ GlobalOutput.perror("TSocket::open() fcntl " + getSocketInfo(), errno_copy);
+ throw TTransportException(TTransportException::NOT_OPEN, "fcntl() failed", errno_copy);
+ }
+ }
+
+ // Connect the socket
+ int ret = connect(socket_, res->ai_addr, res->ai_addrlen);
+
+ // success case
+ if (ret == 0) {
+ goto done;
+ }
+
+ if (errno != EINPROGRESS) {
+ int errno_copy = errno;
+ GlobalOutput.perror("TSocket::open() connect() " + getSocketInfo(), errno_copy);
+ throw TTransportException(TTransportException::NOT_OPEN, "connect() failed", errno_copy);
+ }
+
+
+ struct pollfd fds[1];
+ std::memset(fds, 0 , sizeof(fds));
+ fds[0].fd = socket_;
+ fds[0].events = POLLOUT;
+ ret = poll(fds, 1, connTimeout_);
+
+ if (ret > 0) {
+ // Ensure the socket is connected and that there are no errors set
+ int val;
+ socklen_t lon;
+ lon = sizeof(int);
+ int ret2 = getsockopt(socket_, SOL_SOCKET, SO_ERROR, (void *)&val, &lon);
+ if (ret2 == -1) {
+ int errno_copy = errno;
+ GlobalOutput.perror("TSocket::open() getsockopt() " + getSocketInfo(), errno_copy);
+ throw TTransportException(TTransportException::NOT_OPEN, "getsockopt()", errno_copy);
+ }
+ // no errors on socket, go to town
+ if (val == 0) {
+ goto done;
+ }
+ GlobalOutput.perror("TSocket::open() error on socket (after poll) " + getSocketInfo(), val);
+ throw TTransportException(TTransportException::NOT_OPEN, "socket open() error", val);
+ } else if (ret == 0) {
+ // socket timed out
+ string errStr = "TSocket::open() timed out " + getSocketInfo();
+ GlobalOutput(errStr.c_str());
+ throw TTransportException(TTransportException::NOT_OPEN, "open() timed out");
+ } else {
+ // error on poll()
+ int errno_copy = errno;
+ GlobalOutput.perror("TSocket::open() poll() " + getSocketInfo(), errno_copy);
+ throw TTransportException(TTransportException::NOT_OPEN, "poll() failed", errno_copy);
+ }
+
+ done:
+ // Set socket back to normal mode (blocking)
+ fcntl(socket_, F_SETFL, flags);
+}
+
+void TSocket::open() {
+ if (isOpen()) {
+ throw TTransportException(TTransportException::ALREADY_OPEN);
+ }
+
+ // Validate port number
+ if (port_ < 0 || port_ > 65536) {
+ throw TTransportException(TTransportException::NOT_OPEN, "Specified port is invalid");
+ }
+
+ struct addrinfo hints, *res, *res0;
+ res = NULL;
+ res0 = NULL;
+ int error;
+ char port[sizeof("65536")];
+ std::memset(&hints, 0, sizeof(hints));
+ hints.ai_family = PF_UNSPEC;
+ hints.ai_socktype = SOCK_STREAM;
+ hints.ai_flags = AI_PASSIVE | AI_ADDRCONFIG;
+ sprintf(port, "%d", port_);
+
+ error = getaddrinfo(host_.c_str(), port, &hints, &res0);
+
+ if (error) {
+ string errStr = "TSocket::open() getaddrinfo() " + getSocketInfo() + string(gai_strerror(error));
+ GlobalOutput(errStr.c_str());
+ close();
+ throw TTransportException(TTransportException::NOT_OPEN, "Could not resolve host for client socket.");
+ }
+
+ // Cycle through all the returned addresses until one
+ // connects or push the exception up.
+ for (res = res0; res; res = res->ai_next) {
+ try {
+ openConnection(res);
+ break;
+ } catch (TTransportException& ttx) {
+ if (res->ai_next) {
+ close();
+ } else {
+ close();
+ freeaddrinfo(res0); // cleanup on failure
+ throw;
+ }
+ }
+ }
+
+ // Free address structure memory
+ freeaddrinfo(res0);
+}
+
+void TSocket::close() {
+ if (socket_ >= 0) {
+ shutdown(socket_, SHUT_RDWR);
+ ::close(socket_);
+ }
+ socket_ = -1;
+}
+
+uint32_t TSocket::read(uint8_t* buf, uint32_t len) {
+ if (socket_ < 0) {
+ throw TTransportException(TTransportException::NOT_OPEN, "Called read on non-open socket");
+ }
+
+ int32_t retries = 0;
+
+ // EAGAIN can be signalled both when a timeout has occurred and when
+ // the system is out of resources (an awesome undocumented feature).
+ // The following is an approximation of the time interval under which
+ // EAGAIN is taken to indicate an out of resources error.
+ uint32_t eagainThresholdMicros = 0;
+ if (recvTimeout_) {
+ // if a readTimeout is specified along with a max number of recv retries, then
+ // the threshold will ensure that the read timeout is not exceeded even in the
+ // case of resource errors
+ eagainThresholdMicros = (recvTimeout_*1000)/ ((maxRecvRetries_>0) ? maxRecvRetries_ : 2);
+ }
+
+ try_again:
+ // Read from the socket
+ struct timeval begin;
+ gettimeofday(&begin, NULL);
+ int got = recv(socket_, buf, len, 0);
+ int errno_copy = errno; //gettimeofday can change errno
+ struct timeval end;
+ gettimeofday(&end, NULL);
+ uint32_t readElapsedMicros = (((end.tv_sec - begin.tv_sec) * 1000 * 1000)
+ + (((uint64_t)(end.tv_usec - begin.tv_usec))));
+ ++g_socket_syscalls;
+
+ // Check for error on read
+ if (got < 0) {
+ if (errno_copy == EAGAIN) {
+ // check if this is the lack of resources or timeout case
+ if (!eagainThresholdMicros || (readElapsedMicros < eagainThresholdMicros)) {
+ if (retries++ < maxRecvRetries_) {
+ usleep(50);
+ goto try_again;
+ } else {
+ throw TTransportException(TTransportException::TIMED_OUT,
+ "EAGAIN (unavailable resources)");
+ }
+ } else {
+ // infer that timeout has been hit
+ throw TTransportException(TTransportException::TIMED_OUT,
+ "EAGAIN (timed out)");
+ }
+ }
+
+ // If interrupted, try again
+ if (errno_copy == EINTR && retries++ < maxRecvRetries_) {
+ goto try_again;
+ }
+
+ // Now it's not a try again case, but a real probblez
+ GlobalOutput.perror("TSocket::read() recv() " + getSocketInfo(), errno_copy);
+
+ // If we disconnect with no linger time
+ if (errno_copy == ECONNRESET) {
+ #ifdef __FreeBSD__
+ /* shigin: freebsd doesn't follow POSIX semantic of recv and fails with
+ * ECONNRESET if peer performed shutdown
+ */
+ close();
+ return 0;
+ #else
+ throw TTransportException(TTransportException::NOT_OPEN, "ECONNRESET");
+ #endif
+ }
+
+ // This ish isn't open
+ if (errno_copy == ENOTCONN) {
+ throw TTransportException(TTransportException::NOT_OPEN, "ENOTCONN");
+ }
+
+ // Timed out!
+ if (errno_copy == ETIMEDOUT) {
+ throw TTransportException(TTransportException::TIMED_OUT, "ETIMEDOUT");
+ }
+
+ // Some other error, whatevz
+ throw TTransportException(TTransportException::UNKNOWN, "Unknown", errno_copy);
+ }
+
+ // The remote host has closed the socket
+ if (got == 0) {
+ close();
+ return 0;
+ }
+
+ // Pack data into string
+ return got;
+}
+
+void TSocket::write(const uint8_t* buf, uint32_t len) {
+ if (socket_ < 0) {
+ throw TTransportException(TTransportException::NOT_OPEN, "Called write on non-open socket");
+ }
+
+ uint32_t sent = 0;
+
+ while (sent < len) {
+
+ int flags = 0;
+ #ifdef MSG_NOSIGNAL
+ // Note the use of MSG_NOSIGNAL to suppress SIGPIPE errors, instead we
+ // check for the EPIPE return condition and close the socket in that case
+ flags |= MSG_NOSIGNAL;
+ #endif // ifdef MSG_NOSIGNAL
+
+ int b = send(socket_, buf + sent, len - sent, flags);
+ ++g_socket_syscalls;
+
+ // Fail on a send error
+ if (b < 0) {
+ int errno_copy = errno;
+ GlobalOutput.perror("TSocket::write() send() " + getSocketInfo(), errno_copy);
+
+ if (errno == EPIPE || errno == ECONNRESET || errno == ENOTCONN) {
+ close();
+ throw TTransportException(TTransportException::NOT_OPEN, "write() send()", errno_copy);
+ }
+
+ throw TTransportException(TTransportException::UNKNOWN, "write() send()", errno_copy);
+ }
+
+ // Fail on blocked send
+ if (b == 0) {
+ throw TTransportException(TTransportException::NOT_OPEN, "Socket send returned 0.");
+ }
+ sent += b;
+ }
+}
+
+std::string TSocket::getHost() {
+ return host_;
+}
+
+int TSocket::getPort() {
+ return port_;
+}
+
+void TSocket::setHost(string host) {
+ host_ = host;
+}
+
+void TSocket::setPort(int port) {
+ port_ = port;
+}
+
+void TSocket::setLinger(bool on, int linger) {
+ lingerOn_ = on;
+ lingerVal_ = linger;
+ if (socket_ < 0) {
+ return;
+ }
+
+ struct linger l = {(lingerOn_ ? 1 : 0), lingerVal_};
+ int ret = setsockopt(socket_, SOL_SOCKET, SO_LINGER, &l, sizeof(l));
+ if (ret == -1) {
+ int errno_copy = errno; // Copy errno because we're allocating memory.
+ GlobalOutput.perror("TSocket::setLinger() setsockopt() " + getSocketInfo(), errno_copy);
+ }
+}
+
+void TSocket::setNoDelay(bool noDelay) {
+ noDelay_ = noDelay;
+ if (socket_ < 0) {
+ return;
+ }
+
+ // Set socket to NODELAY
+ int v = noDelay_ ? 1 : 0;
+ int ret = setsockopt(socket_, IPPROTO_TCP, TCP_NODELAY, &v, sizeof(v));
+ if (ret == -1) {
+ int errno_copy = errno; // Copy errno because we're allocating memory.
+ GlobalOutput.perror("TSocket::setNoDelay() setsockopt() " + getSocketInfo(), errno_copy);
+ }
+}
+
+void TSocket::setConnTimeout(int ms) {
+ connTimeout_ = ms;
+}
+
+void TSocket::setRecvTimeout(int ms) {
+ if (ms < 0) {
+ char errBuf[512];
+ sprintf(errBuf, "TSocket::setRecvTimeout with negative input: %d\n", ms);
+ GlobalOutput(errBuf);
+ return;
+ }
+ recvTimeout_ = ms;
+
+ if (socket_ < 0) {
+ return;
+ }
+
+ recvTimeval_.tv_sec = (int)(recvTimeout_/1000);
+ recvTimeval_.tv_usec = (int)((recvTimeout_%1000)*1000);
+
+ // Copy because poll may modify
+ struct timeval r = recvTimeval_;
+ int ret = setsockopt(socket_, SOL_SOCKET, SO_RCVTIMEO, &r, sizeof(r));
+ if (ret == -1) {
+ int errno_copy = errno; // Copy errno because we're allocating memory.
+ GlobalOutput.perror("TSocket::setRecvTimeout() setsockopt() " + getSocketInfo(), errno_copy);
+ }
+}
+
+void TSocket::setSendTimeout(int ms) {
+ if (ms < 0) {
+ char errBuf[512];
+ sprintf(errBuf, "TSocket::setSendTimeout with negative input: %d\n", ms);
+ GlobalOutput(errBuf);
+ return;
+ }
+ sendTimeout_ = ms;
+
+ if (socket_ < 0) {
+ return;
+ }
+
+ struct timeval s = {(int)(sendTimeout_/1000),
+ (int)((sendTimeout_%1000)*1000)};
+ int ret = setsockopt(socket_, SOL_SOCKET, SO_SNDTIMEO, &s, sizeof(s));
+ if (ret == -1) {
+ int errno_copy = errno; // Copy errno because we're allocating memory.
+ GlobalOutput.perror("TSocket::setSendTimeout() setsockopt() " + getSocketInfo(), errno_copy);
+ }
+}
+
+void TSocket::setMaxRecvRetries(int maxRecvRetries) {
+ maxRecvRetries_ = maxRecvRetries;
+}
+
+string TSocket::getSocketInfo() {
+ std::ostringstream oss;
+ oss << "<Host: " << host_ << " Port: " << port_ << ">";
+ return oss.str();
+}
+
+std::string TSocket::getPeerHost() {
+ if (peerHost_.empty()) {
+ struct sockaddr_storage addr;
+ socklen_t addrLen = sizeof(addr);
+
+ if (socket_ < 0) {
+ return host_;
+ }
+
+ int rv = getpeername(socket_, (sockaddr*) &addr, &addrLen);
+
+ if (rv != 0) {
+ return peerHost_;
+ }
+
+ char clienthost[NI_MAXHOST];
+ char clientservice[NI_MAXSERV];
+
+ getnameinfo((sockaddr*) &addr, addrLen,
+ clienthost, sizeof(clienthost),
+ clientservice, sizeof(clientservice), 0);
+
+ peerHost_ = clienthost;
+ }
+ return peerHost_;
+}
+
+std::string TSocket::getPeerAddress() {
+ if (peerAddress_.empty()) {
+ struct sockaddr_storage addr;
+ socklen_t addrLen = sizeof(addr);
+
+ if (socket_ < 0) {
+ return peerAddress_;
+ }
+
+ int rv = getpeername(socket_, (sockaddr*) &addr, &addrLen);
+
+ if (rv != 0) {
+ return peerAddress_;
+ }
+
+ char clienthost[NI_MAXHOST];
+ char clientservice[NI_MAXSERV];
+
+ getnameinfo((sockaddr*) &addr, addrLen,
+ clienthost, sizeof(clienthost),
+ clientservice, sizeof(clientservice),
+ NI_NUMERICHOST|NI_NUMERICSERV);
+
+ peerAddress_ = clienthost;
+ peerPort_ = std::atoi(clientservice);
+ }
+ return peerAddress_;
+}
+
+int TSocket::getPeerPort() {
+ getPeerAddress();
+ return peerPort_;
+}
+
+}}} // apache::thrift::transport