Skip to content

Commit f02dc98

Browse files
slachiewiczclaude
slachiewicz
and
claude
committed
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 <noreply@anthropic.com>
1 parent 50bbda1 commit f02dc98

5 files changed

Lines changed: 347 additions & 47 deletions

File tree

‎lib/cpp/Makefile.am‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -190,6 +190,7 @@ include_transport_HEADERS = \
190190
src/thrift/transport/THeaderTransport.h \
191191
src/thrift/transport/TSimpleFileTransport.h \
192192
src/thrift/transport/TServerSocket.h \
193+
src/thrift/transport/TServerSocketErrors.h \
193194
src/thrift/transport/TSSLServerSocket.h \
194195
src/thrift/transport/TServerTransport.h \
195196
src/thrift/transport/TNonblockingServerTransport.h \

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

Lines changed: 83 additions & 47 deletions
Original file line numberDiff line numberDiff line change
@@ -19,10 +19,13 @@
1919

2020
#include <thrift/thrift-config.h>
2121

22+
#include <cerrno>
23+
#include <chrono>
2224
#include <cstring>
2325
#include <memory>
2426
#include <stdexcept>
2527
#include <sys/types.h>
28+
#include <thread>
2629
#ifdef HAVE_SYS_SOCKET_H
2730
#include <sys/socket.h>
2831
#endif
@@ -52,6 +55,7 @@
5255

5356
#include <thrift/transport/PlatformSocket.h>
5457
#include <thrift/transport/TServerSocket.h>
58+
#include <thrift/transport/TServerSocketErrors.h>
5559
#include <thrift/transport/TSocket.h>
5660
#include <thrift/transport/TSocketUtils.h>
5761
#include <thrift/transport/SocketCommon.h>
@@ -648,62 +652,94 @@ shared_ptr<TTransport> TServerSocket::acceptImpl() {
648652
struct THRIFT_POLLFD fds[2];
649653

650654
int maxEintrs = 5;
651-
int numEintrs = 0;
652655

653-
while (true) {
654-
std::memset(fds, 0, sizeof(fds));
655-
fds[0].fd = serverSocket_;
656-
fds[0].events = THRIFT_POLLIN;
657-
if (interruptSockReader_ != THRIFT_INVALID_SOCKET) {
658-
fds[1].fd = interruptSockReader_;
659-
fds[1].events = THRIFT_POLLIN;
660-
}
661-
/*
662-
TODO: if THRIFT_EINTR is received, we'll restart the timeout.
663-
To be accurate, we need to fix this in the future.
664-
*/
665-
int ret = THRIFT_POLL(fds, 2, accTimeout_);
666-
667-
if (ret < 0) {
668-
// error cases
669-
if (THRIFT_GET_SOCKET_ERROR == THRIFT_EINTR && (numEintrs++ < maxEintrs)) {
670-
// THRIFT_EINTR needs to be handled manually and we can tolerate
671-
// a certain number
672-
continue;
656+
struct sockaddr_storage clientAddress;
657+
int size = 0;
658+
THRIFT_SOCKET clientSocket = THRIFT_INVALID_SOCKET;
659+
int backoffMs = 0;
660+
661+
while (clientSocket == THRIFT_INVALID_SOCKET) {
662+
int numEintrs = 0;
663+
while (true) {
664+
std::memset(fds, 0, sizeof(fds));
665+
fds[0].fd = serverSocket_;
666+
fds[0].events = THRIFT_POLLIN;
667+
if (interruptSockReader_ != THRIFT_INVALID_SOCKET) {
668+
fds[1].fd = interruptSockReader_;
669+
fds[1].events = THRIFT_POLLIN;
673670
}
674-
int errno_copy = THRIFT_GET_SOCKET_ERROR;
675-
TOutput::instance().perror("TServerSocket::acceptImpl() THRIFT_POLL() ", errno_copy);
676-
throw TTransportException(TTransportException::UNKNOWN, "Unknown", errno_copy);
677-
} else if (ret > 0) {
678-
// Check for an interrupt signal
679-
if (interruptSockReader_ != THRIFT_INVALID_SOCKET && (fds[1].revents & THRIFT_POLLIN)) {
680-
int8_t buf;
681-
if (-1 == recv(interruptSockReader_, cast_sockopt(&buf), sizeof(int8_t), 0)) {
682-
TOutput::instance().perror("TServerSocket::acceptImpl() recv() interrupt ",
683-
THRIFT_GET_SOCKET_ERROR);
671+
/*
672+
TODO: if THRIFT_EINTR is received, we'll restart the timeout.
673+
To be accurate, we need to fix this in the future.
674+
*/
675+
int ret = THRIFT_POLL(fds, 2, accTimeout_);
676+
677+
if (ret < 0) {
678+
// error cases
679+
if (THRIFT_GET_SOCKET_ERROR == THRIFT_EINTR && (numEintrs++ < maxEintrs)) {
680+
// THRIFT_EINTR needs to be handled manually and we can tolerate
681+
// a certain number
682+
continue;
683+
}
684+
int errno_copy = THRIFT_GET_SOCKET_ERROR;
685+
TOutput::instance().perror("TServerSocket::acceptImpl() THRIFT_POLL() ", errno_copy);
686+
throw TTransportException(TTransportException::UNKNOWN, "Unknown", errno_copy);
687+
} else if (ret > 0) {
688+
// Check for an interrupt signal
689+
if (interruptSockReader_ != THRIFT_INVALID_SOCKET && (fds[1].revents & THRIFT_POLLIN)) {
690+
int8_t buf;
691+
if (-1 == recv(interruptSockReader_, cast_sockopt(&buf), sizeof(int8_t), 0)) {
692+
TOutput::instance().perror("TServerSocket::acceptImpl() recv() interrupt ",
693+
THRIFT_GET_SOCKET_ERROR);
694+
}
695+
throw TTransportException(TTransportException::INTERRUPTED);
684696
}
685-
throw TTransportException(TTransportException::INTERRUPTED);
686-
}
687697

688-
// Check for the actual server socket being ready
689-
if (fds[0].revents & THRIFT_POLLIN) {
690-
break;
698+
// Check for the actual server socket being ready
699+
if (fds[0].revents & THRIFT_POLLIN) {
700+
break;
701+
}
702+
} else {
703+
TOutput::instance()("TServerSocket::acceptImpl() THRIFT_POLL 0");
704+
throw TTransportException(TTransportException::UNKNOWN);
691705
}
692-
} else {
693-
TOutput::instance()("TServerSocket::acceptImpl() THRIFT_POLL 0");
694-
throw TTransportException(TTransportException::UNKNOWN);
695706
}
696-
}
697707

698-
struct sockaddr_storage clientAddress;
699-
int size = sizeof(clientAddress);
700-
THRIFT_SOCKET clientSocket
701-
= ::accept(serverSocket_, (struct sockaddr*)&clientAddress, (socklen_t*)&size);
708+
size = sizeof(clientAddress);
709+
clientSocket = ::accept(serverSocket_, (struct sockaddr*)&clientAddress, (socklen_t*)&size);
710+
if (clientSocket != THRIFT_INVALID_SOCKET) {
711+
break;
712+
}
702713

703-
if (clientSocket == THRIFT_INVALID_SOCKET) {
704714
int errno_copy = THRIFT_GET_SOCKET_ERROR;
705-
TOutput::instance().perror("TServerSocket::acceptImpl() ::accept() ", errno_copy);
706-
throw TTransportException(TTransportException::UNKNOWN, "accept()", errno_copy);
715+
bool connectionError = detail::isConnectionError(errno_copy);
716+
if (!connectionError && !detail::isResourceExhaustion(errno_copy)) {
717+
TOutput::instance().perror("TServerSocket::acceptImpl() ::accept() ", errno_copy);
718+
throw TTransportException(TTransportException::UNKNOWN, "accept()", errno_copy);
719+
}
720+
721+
// Rate-limited: a peer can trigger connection errors at will.
722+
std::chrono::steady_clock::time_point now = std::chrono::steady_clock::now();
723+
if (now >= nextAcceptErrorLog_) {
724+
nextAcceptErrorLog_ = now + std::chrono::minutes(1);
725+
TOutput::instance().perror("TServerSocket::acceptImpl() ::accept() retrying ", errno_copy);
726+
}
727+
if (connectionError) {
728+
continue;
729+
}
730+
731+
// Out of descriptors or memory: back off, stay interruptible. An interrupt is left for the
732+
// poll above. Each retry re-polls with the full accept timeout.
733+
backoffMs = detail::nextBackoffMs(backoffMs);
734+
if (interruptSockReader_ != THRIFT_INVALID_SOCKET) {
735+
struct THRIFT_POLLFD interruptFd;
736+
std::memset(&interruptFd, 0, sizeof(interruptFd));
737+
interruptFd.fd = interruptSockReader_;
738+
interruptFd.events = THRIFT_POLLIN;
739+
THRIFT_POLL(&interruptFd, 1, backoffMs);
740+
} else {
741+
std::this_thread::sleep_for(std::chrono::milliseconds(backoffMs));
742+
}
707743
}
708744

709745
// Make sure client socket is blocking

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

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@
2020
#ifndef _THRIFT_TRANSPORT_TSERVERSOCKET_H_
2121
#define _THRIFT_TRANSPORT_TSERVERSOCKET_H_ 1
2222

23+
#include <chrono>
2324
#include <functional>
2425

2526
#include <thrift/concurrency/Mutex.h>
@@ -179,6 +180,8 @@ class TServerSocket : public TServerTransport {
179180
int tcpRecvBuffer_;
180181
bool keepAlive_;
181182
bool listening_;
183+
// Rate limit for the accept() retry log message.
184+
std::chrono::steady_clock::time_point nextAcceptErrorLog_;
182185

183186
concurrency::Mutex rwMutex_; // thread-safe interrupt
184187
THRIFT_SOCKET interruptSockWriter_; // is notified on interrupt()

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

Lines changed: 96 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,96 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
20+
#ifndef _THRIFT_TRANSPORT_TSERVERSOCKETERRORS_H_
21+
#define _THRIFT_TRANSPORT_TSERVERSOCKETERRORS_H_ 1
22+
23+
#include <algorithm>
24+
#include <cerrno>
25+
26+
#include <thrift/transport/PlatformSocket.h>
27+
28+
namespace apache {
29+
namespace thrift {
30+
namespace transport {
31+
namespace detail {
32+
33+
// accept() error classes for TServerSocket. Installed, but no compatibility promise.
34+
35+
// Failure of this connection only; the listening socket is fine. Linux: see accept(2) NOTES.
36+
inline bool isConnectionError(int err) {
37+
switch (err) {
38+
case THRIFT_EINTR:
39+
case THRIFT_EAGAIN:
40+
#if THRIFT_EWOULDBLOCK != THRIFT_EAGAIN
41+
case THRIFT_EWOULDBLOCK:
42+
#endif
43+
#ifdef _WIN32
44+
case WSAECONNRESET:
45+
case WSAECONNABORTED:
46+
#else
47+
case ECONNABORTED:
48+
#ifdef EPROTO
49+
case EPROTO:
50+
#endif
51+
#ifdef __linux__
52+
case EPERM:
53+
case ENETDOWN:
54+
case ENETUNREACH:
55+
case EHOSTUNREACH:
56+
case EHOSTDOWN:
57+
case ENONET:
58+
case ENOPROTOOPT:
59+
case EOPNOTSUPP:
60+
#endif
61+
#endif
62+
return true;
63+
default:
64+
return false;
65+
}
66+
}
67+
68+
// Out of descriptors or memory; transient, so retry after a backoff.
69+
inline bool isResourceExhaustion(int err) {
70+
switch (err) {
71+
#ifdef _WIN32
72+
case WSAEMFILE:
73+
case WSAENOBUFS:
74+
#else
75+
case EMFILE:
76+
case ENFILE:
77+
case ENOBUFS:
78+
case ENOMEM:
79+
#endif
80+
return true;
81+
default:
82+
return false;
83+
}
84+
}
85+
86+
// Next wait: 5 ms, doubling to 1 s. Pass 0 for the first.
87+
inline int nextBackoffMs(int previousMs) {
88+
return (previousMs == 0) ? 5 : (std::min)(previousMs * 2, 1000);
89+
}
90+
91+
} // namespace detail
92+
} // namespace transport
93+
} // namespace thrift
94+
} // namespace apache
95+
96+
#endif // #ifndef _THRIFT_TRANSPORT_TSERVERSOCKETERRORS_H_

0 commit comments

Comments
 (0)