You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@thrift.apache.org by "slachiewicz (via GitHub)" <gi...@apache.org> on 2026/10/11 14:33:01 UTC
Re: [PR] THRIFT-5779: Keep TServerSocket accepting after aborted connections and descriptor exhaustion [thrift]
slachiewicz commented on code in PR #3977:
URL: https://github.com/apache/thrift/pull/3977#discussion_r4241943507
##########
lib/cpp/src/thrift/transport/TServerSocket.cpp:
##########
@@ -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;
+ }
Review Comment:
Logging both classes is intended: as the description says, both are logged at most once a minute, so a listener that keeps failing still shows up. `EAGAIN`/`EINTR` fall under the same limit.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@thrift.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org