Fix TServerSocket accept() to not do 200ms polling
Summary: Better to signal to unix file in select
Reviewed By: marc
git-svn-id: https://svn.apache.org/repos/asf/incubator/thrift/trunk@665056 13f79535-47bb-0310-9956-ffa450edef68
diff --git a/lib/cpp/src/transport/TServerSocket.cpp b/lib/cpp/src/transport/TServerSocket.cpp
index 2393134..3b11e9d 100644
--- a/lib/cpp/src/transport/TServerSocket.cpp
+++ b/lib/cpp/src/transport/TServerSocket.cpp
@@ -25,7 +25,8 @@
acceptBacklog_(1024),
sendTimeout_(0),
recvTimeout_(0),
- interrupt_(false) {}
+ intSock1_(-1),
+ intSock2_(-1) {}
TServerSocket::TServerSocket(int port, int sendTimeout, int recvTimeout) :
port_(port),
@@ -33,7 +34,8 @@
acceptBacklog_(1024),
sendTimeout_(sendTimeout),
recvTimeout_(recvTimeout),
- interrupt_(false) {}
+ intSock1_(-1),
+ intSock2_(-1) {}
TServerSocket::~TServerSocket() {
close();
@@ -48,6 +50,16 @@
}
void TServerSocket::listen() {
+ int sv[2];
+ if (-1 == socketpair(AF_LOCAL, SOCK_STREAM, 0, sv)) {
+ perror("TServerSocket::init()");
+ intSock1_ = -1;
+ intSock2_ = -1;
+ } else {
+ intSock1_ = sv[0];
+ intSock2_ = sv[1];
+ }
+
serverSocket_ = socket(AF_INET, SOCK_STREAM, 0);
if (serverSocket_ == -1) {
perror("TServerSocket::listen() socket");
@@ -96,6 +108,7 @@
if (flags == -1) {
throw TTransportException(TTransportException::NOT_OPEN, "fcntl() failed");
}
+
if (-1 == fcntl(serverSocket_, F_SETFL, flags | O_NONBLOCK)) {
throw TTransportException(TTransportException::NOT_OPEN, "fcntl() failed");
}
@@ -129,34 +142,35 @@
throw TTransportException(TTransportException::NOT_OPEN, "TServerSocket not listening");
}
- // 200ms timeout on accept
- struct timeval c = {0, 200000};
fd_set fds;
while (true) {
FD_ZERO(&fds);
FD_SET(serverSocket_, &fds);
- int ret = select(serverSocket_+1, &fds, NULL, NULL, &c);
-
- // Check for interrupt case
- if (ret == 0 && interrupt_) {
- interrupt_ = false;
- throw TTransportException(TTransportException::INTERRUPTED);
+ if (intSock2_ >= 0) {
+ FD_SET(intSock2_, &fds);
}
+ int ret = select(serverSocket_+1, &fds, NULL, NULL, NULL);
- // Reset interrupt flag no matter what
- interrupt_ = false;
-
- if (ret > 0) {
- // Cool, ready to accept
- break;
- } else if (ret == 0) {
- // Timed out... keep going
- continue;
- } else {
- // Bogus, select messed up
- perror("TServerSocket::select() negret");
+ if (ret < 0) {
+ perror("TServerSocket::acceptImpl() select -1");
throw TTransportException(TTransportException::UNKNOWN);
+ } else if (ret > 0) {
+ // Check for an interrupt signal
+ if (intSock2_ >= 0 && FD_ISSET(intSock2_, &fds)) {
+ int8_t buf;
+ if (-1 == recv(intSock2_, &buf, sizeof(int8_t), 0)) {
+ perror("TServerSocket::acceptImpl() interrupt receive");
+ }
+ throw TTransportException(TTransportException::INTERRUPTED);
+ }
+ // Check for the actual server socket being ready
+ if (FD_ISSET(serverSocket_, &fds)) {
+ break;
+ }
+ } else {
+ perror("TServerSocket::acceptImpl() select 0");
+ throw TTransportException(TTransportException::UNKNOWN);
}
}
@@ -193,12 +207,29 @@
return client;
}
+void TServerSocket::interrupt() {
+ if (intSock1_ >= 0) {
+ int8_t byte = 0;
+ if (-1 == send(intSock1_, &byte, sizeof(int8_t), 0)) {
+ perror("TServerSocket::interrupt()");
+ }
+ }
+}
+
void TServerSocket::close() {
if (serverSocket_ >= 0) {
shutdown(serverSocket_, SHUT_RDWR);
::close(serverSocket_);
}
+ if (intSock1_ >= 0) {
+ ::close(intSock1_);
+ }
+ if (intSock2_ >= 0) {
+ ::close(intSock2_);
+ }
serverSocket_ = -1;
+ intSock1_ = -1;
+ intSock2_ = -1;
}
}}} // facebook::thrift::transport
diff --git a/lib/cpp/src/transport/TServerSocket.h b/lib/cpp/src/transport/TServerSocket.h
index 969c00b..edd2510 100644
--- a/lib/cpp/src/transport/TServerSocket.h
+++ b/lib/cpp/src/transport/TServerSocket.h
@@ -33,9 +33,7 @@
void listen();
void close();
- void interrupt() {
- interrupt_ = true;
- }
+ void interrupt();
protected:
boost::shared_ptr<TTransport> acceptImpl();
@@ -46,7 +44,9 @@
int acceptBacklog_;
int sendTimeout_;
int recvTimeout_;
- volatile bool interrupt_;
+
+ int intSock1_;
+ int intSock2_;
};
}}} // facebook::thrift::transport