Copilot commented on code in PR #3977:
URL: https://github.com/apache/thrift/pull/3977#discussion_r4152171990
##########
lib/cpp/test/TServerSocketTest.cpp:
##########
@@ -20,9 +20,18 @@
#include <boost/test/unit_test.hpp>
#include <thrift/transport/TSocket.h>
#include <thrift/transport/TServerSocket.h>
+#include <chrono>
#include <memory>
+#include <thread>
+#include <vector>
#include "TTransportCheckThrow.h"
#include <iostream>
+#ifndef _WIN32
+#include <cerrno>
+#include <fcntl.h>
+#include <sys/resource.h>
Review Comment:
This test is enabled on every non-Windows target, but `sys/resource.h` is an
optional, feature-detected header in this project
(`build/cmake/config.h.in:103`), and the production use is guarded by
`HAVE_SYS_RESOURCE_H` (`lib/cpp/src/thrift/server/TServer.cpp:25`). On a
non-Windows target without that header, the test suite now fails to compile.
Please guard both this include and the descriptor-exhaustion tests with
`HAVE_SYS_RESOURCE_H` (and include `thrift-config.h` explicitly if needed).
##########
lib/cpp/src/thrift/transport/TServerSocket.cpp:
##########
@@ -94,6 +98,64 @@ void destroyer_of_fine_sockets(THRIFT_SOCKET* ssock) {
delete ssock;
}
+namespace {
+
+// accept() failed because of the one connection it tried to take: the peer
went away before it
+// was accepted or, on Linux, a firewall rule refused it or a network error
arrived that accept(2)
+// says to treat like EAGAIN. The listening socket is fine, so the next
connection can be accepted.
+bool isConnectionError(int err) {
+ switch (err) {
+ case THRIFT_EINTR:
+ case THRIFT_EAGAIN:
+#if THRIFT_EWOULDBLOCK != THRIFT_EAGAIN
Review Comment:
Neither added test exercises this new connection-error path; both force only
`EMFILE`. A regression that throws again for `ECONNABORTED`/`EPROTO` or the
Linux-specific cases would therefore pass the suite. Please add a test seam
around `accept()` (or another deterministic platform-specific test) that
injects a representative connection error and verifies that acceptance
continues.
##########
lib/cpp/src/thrift/transport/TServerSocket.cpp:
##########
@@ -650,60 +712,94 @@ shared_ptr<TTransport> TServerSocket::acceptImpl() {
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) {
+ 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);
+ if (isConnectionError(errno_copy)) {
+ // Only that connection is lost; wait for the next one. Not logged,
because a peer can
+ // cause this as often as it likes.
+ continue;
+ }
+ if (!isResourceExhaustion(errno_copy)) {
+ TOutput::instance().perror("TServerSocket::acceptImpl() ::accept() ",
errno_copy);
+ throw TTransportException(TTransportException::UNKNOWN, "accept()",
errno_copy);
+ }
+
+ // Out of descriptors or memory: back off before trying again, but stay
interruptible. An
+ // interrupt that arrives meanwhile is left for the poll above to report.
Every retry polls
+ // again with the full accept timeout, so the whole call can take longer
than that timeout.
+ // Logged at most once a minute: connections that come and go keep
bringing it back.
+ std::chrono::steady_clock::time_point now =
std::chrono::steady_clock::now();
+ if (now >= nextExhaustionLog_) {
+ nextExhaustionLog_ = now + std::chrono::minutes(1);
+ TOutput::instance().perror("TServerSocket::acceptImpl() ::accept()
waiting to retry ",
+ errno_copy);
+ }
+ backoffMs = (backoffMs == 0) ? 5 : (std::min)(backoffMs * 2, 1000);
+ 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);
Review Comment:
The new tests prove retry and interruption, but they would also pass if
resource exhaustion were retried in a tight loop: neither observes the delay or
retry count. Since bounded exponential backoff is a central behavior of this
change, please add deterministic coverage (for example, via an injectable
wait/clock) that verifies growth and the 1-second cap without relying on
wall-clock timing.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
