THRIFT-5779: Keep TServerSocket accepting after aborted connections and descriptor exhaustion

lib/cpp/Makefile.am

@@ -190,6 +190,7 @@ include_transport_HEADERS = \
                          src/thrift/transport/THeaderTransport.h \
                          src/thrift/transport/TSimpleFileTransport.h \
                          src/thrift/transport/TServerSocket.h \
+                         src/thrift/transport/TServerSocketErrors.h \
                          src/thrift/transport/TSSLServerSocket.h \
                          src/thrift/transport/TServerTransport.h \
                          src/thrift/transport/TNonblockingServerTransport.h \

lib/cpp/src/thrift/transport/TServerSocket.cpp

@@ -19,10 +19,13 @@
 
 #include <thrift/thrift-config.h>
 
+#include <cerrno>
+#include <chrono>
 #include <cstring>
 #include <memory>
 #include <stdexcept>
 #include <sys/types.h>
+#include <thread>
 #ifdef HAVE_SYS_SOCKET_H
 #include <sys/socket.h>
 #endif
@@ -52,6 +55,7 @@
 
 #include <thrift/transport/PlatformSocket.h>
 #include <thrift/transport/TServerSocket.h>
+#include <thrift/transport/TServerSocketErrors.h>
 #include <thrift/transport/TSocket.h>
 #include <thrift/transport/TSocketUtils.h>
 #include <thrift/transport/SocketCommon.h>
@@ -648,62 +652,94 @@ shared_ptr<TTransport> TServerSocket::acceptImpl() {
   struct THRIFT_POLLFD fds[2];
 
   int maxEintrs = 5;
-  int numEintrs = 0;
 
-  while (true) {
-    std::memset(fds, 0, sizeof(fds));
-    fds[0].fd = serverSocket_;
-    fds[0].events = THRIFT_POLLIN;
-    if (interruptSockReader_ != THRIFT_INVALID_SOCKET) {
-      fds[1].fd = interruptSockReader_;
-      fds[1].events = THRIFT_POLLIN;
-    }
-    /*
-      TODO: if THRIFT_EINTR is received, we'll restart the timeout.
-      To be accurate, we need to fix this in the future.
-     */
-    int ret = THRIFT_POLL(fds, 2, accTimeout_);
-
-    if (ret < 0) {
-      // error cases
-      if (THRIFT_GET_SOCKET_ERROR == THRIFT_EINTR && (numEintrs++ < maxEintrs)) {
-        // THRIFT_EINTR needs to be handled manually and we can tolerate
-        // a certain number
-        continue;
+  struct sockaddr_storage clientAddress;
+  int size = 0;
+  THRIFT_SOCKET clientSocket = THRIFT_INVALID_SOCKET;
+  int backoffMs = 0;
+
+  while (clientSocket == THRIFT_INVALID_SOCKET) {
+    int numEintrs = 0;
+    while (true) {
+      std::memset(fds, 0, sizeof(fds));
+      fds[0].fd = serverSocket_;
+      fds[0].events = THRIFT_POLLIN;
+      if (interruptSockReader_ != THRIFT_INVALID_SOCKET) {
+        fds[1].fd = interruptSockReader_;
+        fds[1].events = THRIFT_POLLIN;
       }
-      int errno_copy = THRIFT_GET_SOCKET_ERROR;
-      TOutput::instance().perror("TServerSocket::acceptImpl() THRIFT_POLL() ", errno_copy);
-      throw TTransportException(TTransportException::UNKNOWN, "Unknown", errno_copy);
-    } else if (ret > 0) {
-      // Check for an interrupt signal
-      if (interruptSockReader_ != THRIFT_INVALID_SOCKET && (fds[1].revents & THRIFT_POLLIN)) {
-        int8_t buf;
-        if (-1 == recv(interruptSockReader_, cast_sockopt(&buf), sizeof(int8_t), 0)) {
-          TOutput::instance().perror("TServerSocket::acceptImpl() recv() interrupt ",
-                              THRIFT_GET_SOCKET_ERROR);
+      /*
+        TODO: if THRIFT_EINTR is received, we'll restart the timeout.
+        To be accurate, we need to fix this in the future.
+       */
+      int ret = THRIFT_POLL(fds, 2, accTimeout_);
+
+      if (ret < 0) {
+        // error cases
+        if (THRIFT_GET_SOCKET_ERROR == THRIFT_EINTR && (numEintrs++ < maxEintrs)) {
+          // THRIFT_EINTR needs to be handled manually and we can tolerate
+          // a certain number
+          continue;
+        }
+        int errno_copy = THRIFT_GET_SOCKET_ERROR;
+        TOutput::instance().perror("TServerSocket::acceptImpl() THRIFT_POLL() ", errno_copy);
+        throw TTransportException(TTransportException::UNKNOWN, "Unknown", errno_copy);
+      } else if (ret > 0) {
+        // Check for an interrupt signal
+        if (interruptSockReader_ != THRIFT_INVALID_SOCKET && (fds[1].revents & THRIFT_POLLIN)) {
+          int8_t buf;
+          if (-1 == recv(interruptSockReader_, cast_sockopt(&buf), sizeof(int8_t), 0)) {
+            TOutput::instance().perror("TServerSocket::acceptImpl() recv() interrupt ",
+                                       THRIFT_GET_SOCKET_ERROR);
+          }
+          throw TTransportException(TTransportException::INTERRUPTED);
         }
-        throw TTransportException(TTransportException::INTERRUPTED);
-      }
 
-      // Check for the actual server socket being ready
-      if (fds[0].revents & THRIFT_POLLIN) {
-        break;
+        // Check for the actual server socket being ready
+        if (fds[0].revents & THRIFT_POLLIN) {
+          break;
+        }
+      } else {
+        TOutput::instance()("TServerSocket::acceptImpl() THRIFT_POLL 0");
+        throw TTransportException(TTransportException::UNKNOWN);
       }
-    } else {
-      TOutput::instance()("TServerSocket::acceptImpl() THRIFT_POLL 0");
-      throw TTransportException(TTransportException::UNKNOWN);
     }
-  }
 
-  struct sockaddr_storage clientAddress;
-  int size = sizeof(clientAddress);
-  THRIFT_SOCKET clientSocket
-      = ::accept(serverSocket_, (struct sockaddr*)&clientAddress, (socklen_t*)&size);
+    size = sizeof(clientAddress);
+    clientSocket = ::accept(serverSocket_, (struct sockaddr*)&clientAddress, (socklen_t*)&size);
+    if (clientSocket != THRIFT_INVALID_SOCKET) {
+      break;
+    }
 
-  if (clientSocket == THRIFT_INVALID_SOCKET) {
     int errno_copy = THRIFT_GET_SOCKET_ERROR;
-    TOutput::instance().perror("TServerSocket::acceptImpl() ::accept() ", errno_copy);
-    throw TTransportException(TTransportException::UNKNOWN, "accept()", errno_copy);
+    bool connectionError = detail::isConnectionError(errno_copy);
+    if (!connectionError && !detail::isResourceExhaustion(errno_copy)) {
+      TOutput::instance().perror("TServerSocket::acceptImpl() ::accept() ", errno_copy);
+      throw TTransportException(TTransportException::UNKNOWN, "accept()", errno_copy);
+    }
+
+    // Rate-limited: a peer can trigger connection errors at will.
+    std::chrono::steady_clock::time_point now = std::chrono::steady_clock::now();
+    if (now >= nextAcceptErrorLog_) {
+      nextAcceptErrorLog_ = now + std::chrono::minutes(1);
+      TOutput::instance().perror("TServerSocket::acceptImpl() ::accept() retrying ", errno_copy);
+    }
+    if (connectionError) {
+      continue;
+    }
+
+    // Out of descriptors or memory: back off, stay interruptible. An interrupt is left for the
+    // poll above. Each retry re-polls with the full accept timeout.
+    backoffMs = detail::nextBackoffMs(backoffMs);
+    if (interruptSockReader_ != THRIFT_INVALID_SOCKET) {
+      struct THRIFT_POLLFD interruptFd;
+      std::memset(&interruptFd, 0, sizeof(interruptFd));
+      interruptFd.fd = interruptSockReader_;
+      interruptFd.events = THRIFT_POLLIN;
+      THRIFT_POLL(&interruptFd, 1, backoffMs);
+    } else {
+      std::this_thread::sleep_for(std::chrono::milliseconds(backoffMs));
+    }
   }
 
   // Make sure client socket is blocking

lib/cpp/src/thrift/transport/TServerSocket.h

@@ -20,6 +20,7 @@
 #ifndef _THRIFT_TRANSPORT_TSERVERSOCKET_H_
 #define _THRIFT_TRANSPORT_TSERVERSOCKET_H_ 1
 
+#include <chrono>
 #include <functional>
 
 #include <thrift/concurrency/Mutex.h>
@@ -179,6 +180,8 @@ class TServerSocket : public TServerTransport {
   int tcpRecvBuffer_;
   bool keepAlive_;
   bool listening_;
+  // Rate limit for the accept() retry log message.
+  std::chrono::steady_clock::time_point nextAcceptErrorLog_;
 
   concurrency::Mutex rwMutex_;                                 // thread-safe interrupt
   THRIFT_SOCKET interruptSockWriter_;                          // is notified on interrupt()

lib/cpp/src/thrift/transport/TServerSocketErrors.h

@@ -0,0 +1,96 @@
+/*
+ * 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.
+ */
+
+#ifndef _THRIFT_TRANSPORT_TSERVERSOCKETERRORS_H_
+#define _THRIFT_TRANSPORT_TSERVERSOCKETERRORS_H_ 1
+
+#include <algorithm>
+#include <cerrno>
+
+#include <thrift/transport/PlatformSocket.h>
+
+namespace apache {
+namespace thrift {
+namespace transport {
+namespace detail {
+
+// accept() error classes for TServerSocket. Installed, but no compatibility promise.
+
+// Failure of this connection only; the listening socket is fine. Linux: see accept(2) NOTES.
+inline bool isConnectionError(int err) {
+  switch (err) {
+  case THRIFT_EINTR:
+  case THRIFT_EAGAIN:
+#if THRIFT_EWOULDBLOCK != THRIFT_EAGAIN
+  case THRIFT_EWOULDBLOCK:
+#endif
+#ifdef _WIN32
+  case WSAECONNRESET:
+  case WSAECONNABORTED:
+#else
+  case ECONNABORTED:
+#ifdef EPROTO
+  case EPROTO:
+#endif
+#ifdef __linux__
+  case EPERM:
+  case ENETDOWN:
+  case ENETUNREACH:
+  case EHOSTUNREACH:
+  case EHOSTDOWN:
+  case ENONET:
+  case ENOPROTOOPT:
+  case EOPNOTSUPP:
+#endif
+#endif
+    return true;
+  default:
+    return false;
+  }
+}
+
+// Out of descriptors or memory; transient, so retry after a backoff.
+inline bool isResourceExhaustion(int err) {
+  switch (err) {
+#ifdef _WIN32
+  case WSAEMFILE:
+  case WSAENOBUFS:
+#else
+  case EMFILE:
+  case ENFILE:
+  case ENOBUFS:
+  case ENOMEM:
+#endif
+    return true;
+  default:
+    return false;
+  }
+}
+
+// Next wait: 5 ms, doubling to 1 s. Pass 0 for the first.
+inline int nextBackoffMs(int previousMs) {
+  return (previousMs == 0) ? 5 : (std::min)(previousMs * 2, 1000);
+}
+
+} // namespace detail
+} // namespace transport
+} // namespace thrift
+} // namespace apache
+
+#endif // #ifndef _THRIFT_TRANSPORT_TSERVERSOCKETERRORS_H_