This is an automated email from the ASF dual-hosted git repository.
cmcfarlen pushed a commit to branch 10.1.x
in repository https://gitbox.apache.org/repos/asf/trafficserver.git
The following commit(s) were added to refs/heads/10.1.x by this push:
new 56b31e0847 Avoid JSONRPC client write spins (#13329)
56b31e0847 is described below
commit 56b31e0847019141bb680d4316e8be459bc5fde7
Author: Brian Neradt <[email protected]>
AuthorDate: Mon Jun 29 11:06:21 2026 -0500
Avoid JSONRPC client write spins (#13329)
JSONRPC clients use nonblocking Unix sockets, so a full peer receive
buffer can make write return EAGAIN. The old retry loop treated that
like EINTR and immediately retried forever, which could hang the
jsonrpc server tests during large request and restart coverage.
This waits for socket write readiness before retrying transient
backpressure and reports ETIMEDOUT when the readiness wait expires,
while preserving readiness poll errors. This also adds a regression
test that fills a socketpair and verifies that the helper returns with
that timeout signal instead of spinning.
(cherry picked from commit 18d7a9df2739ec6371203b4078d968c01c3ae6cc)
---
src/mgmt/rpc/server/unit_tests/test_rpcserver.cc | 73 ++++++++++++++++++++++++
src/shared/rpc/IPCSocketClient.cc | 17 +++++-
2 files changed, 89 insertions(+), 1 deletion(-)
diff --git a/src/mgmt/rpc/server/unit_tests/test_rpcserver.cc
b/src/mgmt/rpc/server/unit_tests/test_rpcserver.cc
index e406be2907..176d98036b 100644
--- a/src/mgmt/rpc/server/unit_tests/test_rpcserver.cc
+++ b/src/mgmt/rpc/server/unit_tests/test_rpcserver.cc
@@ -24,6 +24,9 @@
#include <sys/types.h>
#include <sys/socket.h>
#include <sys/un.h>
+#include <sys/wait.h>
+#include <fcntl.h>
+#include <signal.h>
#include <stdio.h>
#include <unistd.h>
@@ -332,6 +335,10 @@ private:
}
};
+struct TestableIPCSocketClient : shared::rpc::IPCSocketClient {
+ using shared::rpc::IPCSocketClient::_safe_write;
+};
+
// helper function to send a request and update the promise when the response
is done.
// This is to be used in a multithread test.
void
@@ -341,6 +348,72 @@ send_request(std::string json, std::promise<std::string> p)
auto resp = rpc_client.query(json);
p.set_value(resp);
}
+
+TEST_CASE("IPCSocketClient write returns when the peer stops reading",
"[socket][client]")
+{
+ int fds[2];
+ REQUIRE(::socketpair(AF_UNIX, SOCK_STREAM, 0, fds) == 0);
+
+ int const flags = ::fcntl(fds[0], F_GETFL, 0);
+ REQUIRE(flags >= 0);
+ REQUIRE(::fcntl(fds[0], F_SETFL, flags | O_NONBLOCK) == 0);
+
+ std::vector<char> fill(4096, 'x');
+ while (true) {
+ ssize_t const ret = ::write(fds[0], fill.data(), fill.size());
+ if (ret < 0) {
+ REQUIRE((errno == EAGAIN || errno == EWOULDBLOCK));
+ break;
+ }
+ REQUIRE(ret > 0);
+ }
+
+ TestableIPCSocketClient rpc_client;
+ pid_t const pid = ::fork();
+ REQUIRE(pid >= 0);
+ if (pid == 0) {
+ ::close(fds[1]);
+ char const byte = 'x';
+ auto const ret = rpc_client._safe_write(fds[0], &byte, 1);
+ auto const err = errno;
+ ::close(fds[0]);
+ _exit(ret == -1 && err == ETIMEDOUT ? 0 : 1);
+ }
+
+ int status = 0;
+ bool child_exited = false;
+ auto const child_deadline = std::chrono::steady_clock::now() +
std::chrono::seconds(10);
+ while (std::chrono::steady_clock::now() < child_deadline) {
+ auto const wait_ret = ::waitpid(pid, &status, WNOHANG);
+ if (wait_ret == pid) {
+ child_exited = true;
+ break;
+ }
+ REQUIRE(wait_ret == 0);
+ std::this_thread::sleep_for(std::chrono::milliseconds(100));
+ }
+
+ if (!child_exited) {
+ auto const wait_ret = ::waitpid(pid, &status, WNOHANG);
+ if (wait_ret == pid) {
+ child_exited = true;
+ } else {
+ REQUIRE(wait_ret == 0);
+ }
+ }
+
+ if (!child_exited) {
+ ::kill(pid, SIGKILL);
+ REQUIRE(::waitpid(pid, &status, 0) == pid);
+ FAIL("_safe_write did not return when the nonblocking socket stayed
unwritable");
+ }
+
+ ::close(fds[0]);
+ ::close(fds[1]);
+
+ REQUIRE(WIFEXITED(status));
+ REQUIRE(WEXITSTATUS(status) == 0);
+}
} // namespace
TEST_CASE("Sending 'concurrent' requests to the rpc server.", "[thread]")
{
diff --git a/src/shared/rpc/IPCSocketClient.cc
b/src/shared/rpc/IPCSocketClient.cc
index 45a67061b5..9a066c9f4a 100644
--- a/src/shared/rpc/IPCSocketClient.cc
+++ b/src/shared/rpc/IPCSocketClient.cc
@@ -89,13 +89,28 @@ IPCSocketClient::connect(std::chrono::milliseconds wait_ms,
int attempts)
std::int64_t
IPCSocketClient::_safe_write(int fd, const char *buffer, int len)
{
+ constexpr int WRITE_READY_TIMEOUT_MS = 1000;
+
std::int64_t written{0};
while (written < len) {
const ssize_t ret = ::write(fd, buffer + written, len - written);
if (ret == -1) {
- if (errno == EAGAIN || errno == EINTR) {
+ if (errno == EINTR) {
continue;
}
+ if (errno == EAGAIN || errno == EWOULDBLOCK) {
+ auto const ready = write_ready(fd, WRITE_READY_TIMEOUT_MS);
+ if (ready == 1) {
+ continue;
+ }
+ if (ready == 0) {
+ errno = ETIMEDOUT;
+ }
+ }
+ return -1;
+ }
+ if (ret == 0) {
+ errno = EIO;
return -1;
}
written += ret;
