From c3702a9b3e9492fb3798fe5e71ccd7a3bc8a2cd0 Mon Sep 17 00:00:00 2001 From: Sylwester Lachiewicz Date: Mon, 28 Sep 2026 21:52:59 +0200 Subject: [PATCH] THRIFT-5779: Keep accepting connections after a transient accept() failure Client: cpp When accept() failed with EMFILE, TServerFramework::serve() stopped for good. acceptImpl() now waits out running out of descriptors or memory, backing off from 5 ms to 1 s and staying interruptible, and goes back to poll() when the error belongs to the one connection. Linux keeps that connection queued, so retrying without the wait would spin. Neither case is logged per attempt, because a peer can cause both. Co-Authored-By: Claude Opus 5.5 --- lib/cpp/Makefile.am | 1 + .../src/thrift/transport/TServerSocket.cpp | 130 +++++++++----- lib/cpp/src/thrift/transport/TServerSocket.h | 3 + .../thrift/transport/TServerSocketErrors.h | 96 ++++++++++ lib/cpp/test/TServerSocketTest.cpp | 164 ++++++++++++++++++ 5 files changed, 347 insertions(+), 47 deletions(-) create mode 100644 lib/cpp/src/thrift/transport/TServerSocketErrors.h diff --git a/lib/cpp/Makefile.am b/lib/cpp/Makefile.am index dcbcf2c5db2..e419bc18150 100644 --- a/lib/cpp/Makefile.am +++ b/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 \ diff --git a/lib/cpp/src/thrift/transport/TServerSocket.cpp b/lib/cpp/src/thrift/transport/TServerSocket.cpp index 625672bd76d..f0ad58db402 100644 --- a/lib/cpp/src/thrift/transport/TServerSocket.cpp +++ b/lib/cpp/src/thrift/transport/TServerSocket.cpp @@ -19,10 +19,13 @@ #include +#include +#include #include #include #include #include +#include #ifdef HAVE_SYS_SOCKET_H #include #endif @@ -52,6 +55,7 @@ #include #include +#include #include #include #include @@ -648,62 +652,94 @@ shared_ptr 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 diff --git a/lib/cpp/src/thrift/transport/TServerSocket.h b/lib/cpp/src/thrift/transport/TServerSocket.h index 4d43ccf5ade..4bd1efa77c9 100644 --- a/lib/cpp/src/thrift/transport/TServerSocket.h +++ b/lib/cpp/src/thrift/transport/TServerSocket.h @@ -20,6 +20,7 @@ #ifndef _THRIFT_TRANSPORT_TSERVERSOCKET_H_ #define _THRIFT_TRANSPORT_TSERVERSOCKET_H_ 1 +#include #include #include @@ -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() diff --git a/lib/cpp/src/thrift/transport/TServerSocketErrors.h b/lib/cpp/src/thrift/transport/TServerSocketErrors.h new file mode 100644 index 00000000000..f7f10e72baf --- /dev/null +++ b/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 +#include + +#include + +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_ diff --git a/lib/cpp/test/TServerSocketTest.cpp b/lib/cpp/test/TServerSocketTest.cpp index 0860242bc25..3be4549860c 100644 --- a/lib/cpp/test/TServerSocketTest.cpp +++ b/lib/cpp/test/TServerSocketTest.cpp @@ -17,17 +17,32 @@ * under the License. */ +#include + #include #include #include +#include +#include #include +#include +#include #include "TTransportCheckThrow.h" #include +#include +#if !defined(_WIN32) && defined(HAVE_SYS_RESOURCE_H) +#include +#include +#include +#endif using apache::thrift::transport::TServerSocket; using apache::thrift::transport::TSocket; using apache::thrift::transport::TTransport; using apache::thrift::transport::TTransportException; +using apache::thrift::transport::detail::isConnectionError; +using apache::thrift::transport::detail::isResourceExhaustion; +using apache::thrift::transport::detail::nextBackoffMs; using std::shared_ptr; BOOST_AUTO_TEST_SUITE(TServerSocketTest) @@ -70,4 +85,153 @@ BOOST_AUTO_TEST_CASE(test_get_port) { BOOST_CHECK_EQUAL(888, sock1.getPort()); } +BOOST_AUTO_TEST_CASE(test_accept_error_classes) { + // accept() goes back to poll() on these: only the one connection is lost. + BOOST_CHECK(isConnectionError(THRIFT_EINTR)); + BOOST_CHECK(isConnectionError(THRIFT_EAGAIN)); + BOOST_CHECK(isConnectionError(THRIFT_EWOULDBLOCK)); +#ifdef _WIN32 + BOOST_CHECK(isConnectionError(WSAECONNRESET)); + BOOST_CHECK(isConnectionError(WSAECONNABORTED)); + BOOST_CHECK(isResourceExhaustion(WSAEMFILE)); + BOOST_CHECK(isResourceExhaustion(WSAENOBUFS)); + BOOST_CHECK(!isConnectionError(WSAENOTSOCK)); + BOOST_CHECK(!isResourceExhaustion(WSAENOTSOCK)); +#else + BOOST_CHECK(isConnectionError(ECONNABORTED)); +#ifdef EPROTO + BOOST_CHECK(isConnectionError(EPROTO)); +#endif +#ifdef __linux__ + BOOST_CHECK(isConnectionError(EPERM)); + BOOST_CHECK(isConnectionError(ENETDOWN)); + BOOST_CHECK(isConnectionError(ENETUNREACH)); + BOOST_CHECK(isConnectionError(EHOSTUNREACH)); + BOOST_CHECK(isConnectionError(EHOSTDOWN)); + BOOST_CHECK(isConnectionError(ENONET)); + BOOST_CHECK(isConnectionError(ENOPROTOOPT)); + BOOST_CHECK(isConnectionError(EOPNOTSUPP)); +#endif + // accept() backs off and retries on these. + BOOST_CHECK(isResourceExhaustion(EMFILE)); + BOOST_CHECK(isResourceExhaustion(ENFILE)); + BOOST_CHECK(isResourceExhaustion(ENOBUFS)); + BOOST_CHECK(isResourceExhaustion(ENOMEM)); + // accept() throws on anything else, such as a listening socket that is no longer valid. + BOOST_CHECK(!isConnectionError(EBADF)); + BOOST_CHECK(!isResourceExhaustion(EBADF)); + BOOST_CHECK(!isConnectionError(EINVAL)); + BOOST_CHECK(!isResourceExhaustion(EINVAL)); +#endif +} + +BOOST_AUTO_TEST_CASE(test_accept_backoff_doubles_up_to_one_second) { + const int expected[] = {5, 10, 20, 40, 80, 160, 320, 640, 1000, 1000, 1000}; + int backoffMs = 0; + for (int step : expected) { + backoffMs = nextBackoffMs(backoffMs); + BOOST_CHECK_EQUAL(step, backoffMs); + } +} + +#if !defined(_WIN32) && defined(HAVE_SYS_RESOURCE_H) +// Sends a byte: TCP_DEFER_ACCEPT (Linux) holds accept() until data arrives. +void connectAndSend(TSocket& socket) { + socket.open(); + uint8_t byte = 0; + socket.write(&byte, 1); + socket.flush(); +} + +// Opens descriptors until EMFILE. Process-wide: tests must run serially. +class DescriptorExhaustion { +public: + DescriptorExhaustion() { + BOOST_REQUIRE_EQUAL(0, getrlimit(RLIMIT_NOFILE, &saved_)); + struct rlimit low = saved_; + if (low.rlim_cur == RLIM_INFINITY || low.rlim_cur > 256) { + low.rlim_cur = 256; + } + BOOST_REQUIRE_EQUAL(0, setrlimit(RLIMIT_NOFILE, &low)); + int fd = ::open("/dev/null", O_RDONLY); + if (fd >= 0) { + int base = fd; + do { + fds_.push_back(fd); + } while ((fd = dup(base)) >= 0); + } + exhausted_ = (errno == EMFILE); + } + bool exhausted() const { return exhausted_; } + ~DescriptorExhaustion() { release(); } + void release() { + for (int fd : fds_) { + ::close(fd); + } + fds_.clear(); + BOOST_WARN_EQUAL(0, setrlimit(RLIMIT_NOFILE, &saved_)); + } + +private: + struct rlimit saved_; + std::vector fds_; + bool exhausted_; +}; + +BOOST_AUTO_TEST_CASE(test_accept_waits_out_descriptor_exhaustion) { + TServerSocket server("localhost", 0); + server.setAcceptTimeout(5000); // fail rather than hang if accept() never returns + server.listen(); + TSocket client("localhost", server.getPort()); + connectAndSend(client); + + // Linux keeps the connection queued after EMFILE, macOS drops it; a second client covers both. + TSocket lateClient("localhost", server.getPort()); + shared_ptr accepted; + { + DescriptorExhaustion exhaustion; + BOOST_REQUIRE(exhaustion.exhausted()); + std::thread releaser([&exhaustion, &lateClient] { + std::this_thread::sleep_for(std::chrono::milliseconds(200)); + exhaustion.release(); + connectAndSend(lateClient); + }); + try { + accepted = server.accept(); + } catch (TTransportException& ex) { + BOOST_ERROR("accept() gave up while descriptors were exhausted: " << ex.what()); + } + releaser.join(); + } + BOOST_CHECK(accepted); + if (accepted) { + accepted->close(); + } + lateClient.close(); + client.close(); + server.close(); +} + +BOOST_AUTO_TEST_CASE(test_interrupt_ends_accept_while_descriptors_are_exhausted) { + TServerSocket server("localhost", 0); + server.setAcceptTimeout(5000); // fail rather than hang if accept() never returns + server.listen(); + TSocket client("localhost", server.getPort()); + connectAndSend(client); + + { + DescriptorExhaustion exhaustion; + BOOST_REQUIRE(exhaustion.exhausted()); + std::thread interrupter([&server] { + std::this_thread::sleep_for(std::chrono::milliseconds(200)); + server.interrupt(); + }); + TTRANSPORT_CHECK_THROW(server.accept(), TTransportException::INTERRUPTED); + interrupter.join(); + } + client.close(); + server.close(); +} +#endif + BOOST_AUTO_TEST_SUITE_END()