Copilot commented on code in PR #3498:
URL: https://github.com/apache/brpc/pull/3498#discussion_r3975447898
##########
src/brpc/acceptor.cpp:
##########
@@ -200,9 +214,66 @@ void Acceptor::Join() {
}
size_t Acceptor::ConnectionCount() const {
- // Notice that _socket_map may be modified concurrently. This actually
- // assumes that size() is safe to call concurrently.
- return _socket_map.size();
+ return _connection_count.load(butil::memory_order_relaxed);
+}
+
+size_t Acceptor::RejectedRedisConnectionCount() const {
+ return _rejected_redis_connection_count.load(butil::memory_order_relaxed);
+}
+
+bool Acceptor::TryAcquireRedisConnectionSlot() {
+ size_t count = _connection_count.load(butil::memory_order_relaxed);
+ do {
+ const size_t max_connections =
+ _redis_max_connections.load(butil::memory_order_relaxed);
+ if (max_connections != 0 && count >= max_connections) {
+ return false;
+ }
+ } while (!_connection_count.compare_exchange_weak(
+ count, count + 1, butil::memory_order_relaxed));
+ return true;
+}
+
+void Acceptor::SetRedisMaxConnections(size_t max_connections) {
+ // The limit controls only future numeric admission decisions and does not
+ // publish socket state, so a relaxed store is sufficient.
+ _redis_max_connections.store(
+ max_connections, butil::memory_order_relaxed);
+}
+
+void Acceptor::RejectRedisConnection(int fd) {
+ // Borrowed fd: OnNewConnectionsUntilEAGAIN() retains ownership in its
+ // fd_guard. After this returns, the caller's continue destroys the guard
+ // and closes fd on both the SSL and plaintext paths. Do not close it here.
+ _rejected_redis_connection_count.fetch_add(
+ 1, butil::memory_order_relaxed);
+
+ // Reject SSL-capable listeners before doing any TLS work. Plaintext here
+ // would violate the TLS record protocol and could trigger an expensive
+ // handshake in a higher layer.
+ if (_ssl_ctx) {
+ return;
+ }
+
+ static const char response[] =
+ "-ERR max number of clients reached\r\n";
+ // Delivery is best-effort: handle short writes, but never wait for a slow
+ // peer to become writable and stall admission for other connections.
+ const size_t response_size = sizeof(response) - 1;
+ size_t offset = 0;
+ while (offset < response_size) {
+ const ssize_t nwritten = send(fd,
+ response + offset,
+ response_size - offset,
+ MSG_DONTWAIT | MSG_NOSIGNAL);
Review Comment:
`MSG_NOSIGNAL` is not available on all platforms/toolchains. If this
repository targets non-Linux builds, this may fail to compile or behave
differently. Consider guarding `MSG_NOSIGNAL` behind `#ifdef MSG_NOSIGNAL`
(fall back to `0`), or use the platform-specific socket option (e.g.,
`SO_NOSIGPIPE` where applicable) to avoid SIGPIPE.
##########
test/brpc_server_unittest.cpp:
##########
@@ -1424,6 +1426,174 @@ TEST_F(ServerTest, close_idle_connections) {
ASSERT_EQ(0ul, stat.connection_count);
}
+TEST_F(ServerTest, redis_connection_limit_requires_dedicated_listener) {
+ brpc::Server server;
+ EchoServiceImpl echo_service;
+ ASSERT_EQ(0, server.AddService(
+ &echo_service, brpc::SERVER_DOESNT_OWN_SERVICE));
+
+ brpc::ServerOptions opt;
+ opt.redis_service = new brpc::RedisService;
+ opt.redis_max_connections = 1;
+ opt.enabled_protocols = "redis";
+ opt.has_builtin_services = false;
+ const int rc = server.Start("127.0.0.1:0", &opt);
+ if (rc != 0) {
+ delete opt.redis_service;
+ opt.redis_service = nullptr;
+ }
+ EXPECT_EQ(-1, rc);
+}
+
+TEST_F(ServerTest, reject_redis_connections_over_limit) {
+ brpc::Server server;
+ brpc::ServerOptions opt;
+ opt.redis_service = new brpc::RedisService;
+ opt.redis_max_connections = 0;
+ opt.enabled_protocols = "redis";
+ opt.has_builtin_services = false;
+ ASSERT_EQ(0, server.Start("127.0.0.1:0", &opt));
+ ASSERT_EQ(0, server.SetRedisMaxConnections(1));
+
+ const butil::EndPoint ep = server.listen_address();
+ butil::fd_guard first_client(tcp_connect(ep, nullptr));
+ ASSERT_GE(first_client, 0);
+
+ brpc::ServerStatistics stat;
+ for (int retry = 0; retry < 100; ++retry) {
+ server.GetStat(&stat);
+ if (stat.connection_count == 1) {
+ break;
+ }
+ usleep(1000);
+ }
+ ASSERT_EQ(1ul, stat.connection_count);
+
+ // The Redis limit belongs to this acceptor, not to the process. A separate
+ // RPC Server must remain reachable while the Redis listener is full.
+ EchoServiceImpl rpc_service;
+ brpc::Server rpc_server;
+ ASSERT_EQ(0, rpc_server.AddService(
+ &rpc_service, brpc::SERVER_DOESNT_OWN_SERVICE));
+ ASSERT_EQ(0, rpc_server.Start("127.0.0.1:0", nullptr));
+ EXPECT_EQ(-1, rpc_server.SetRedisMaxConnections(1));
+ SendSleepRPC(rpc_server.listen_address(), 0, true);
+
+ butil::fd_guard rejected_client(tcp_connect(ep, nullptr));
+ ASSERT_GE(rejected_client, 0);
+ struct timeval timeout = {1, 0};
+ ASSERT_EQ(0, setsockopt(rejected_client, SOL_SOCKET, SO_RCVTIMEO,
+ &timeout, sizeof(timeout)));
+ const std::string expected = "-ERR max number of clients reached\r\n";
+ const auto expect_rejection = [&expected](int fd) {
+ // These fresh loopback sockets have no induced send backpressure, so
+ // require the full response here despite best-effort delivery in
+ // production. Small reads exercise accumulation without relying on
+ // TCP preserving the server's write boundaries.
+ std::string response;
+ char chunk[7];
+ while (response.size() < expected.size()) {
+ const ssize_t nr = recv(fd, chunk, sizeof(chunk), 0);
+ if (nr > 0) {
+ response.append(chunk, static_cast<size_t>(nr));
+ } else if (nr < 0 && errno == EINTR) {
+ continue;
+ } else {
+ break;
+ }
+ }
+ EXPECT_EQ(expected, response);
+ // The outer fd_guard must also close plaintext rejected connections.
+ EXPECT_EQ(0, recv(fd, chunk, sizeof(chunk), 0));
+ };
Review Comment:
The production behavior explicitly allows partial/undelivered responses
(nonblocking best-effort send), but the test requires receiving the full error
string and then immediate EOF. This can become flaky under load (e.g., if
`send(..., MSG_DONTWAIT ...)` returns `EAGAIN`, or if close manifests as
`ECONNRESET` rather than EOF on some stacks). Consider relaxing the assertion
to accept a prefix (or “either full response or empty”) and/or tolerate `recv`
returning `-1` with `ECONNRESET` on the final read.
##########
src/brpc/server.cpp:
##########
@@ -870,6 +893,19 @@ int Server::StartInternal(const butil::EndPoint& endpoint,
const ServerOptions default_opt;
const ServerOptions& real_opt = opt ? *opt : default_opt;
+ // Admission happens before protocol parsing (and, importantly, before a
+ // TLS handshake), so it is only safe on a listener dedicated to Redis.
+ // Reject ambiguous configurations instead of accidentally limiting RPCs
+ // sharing the public port.
+ if (real_opt.redis_max_connections != 0 &&
+ !is_redis_only_public_listener(real_opt, service_count())) {
+ LOG(ERROR) << "redis_max_connections requires a Redis-only public "
+ "listener (redis_service set, enabled_protocols exactly "
+ "\"redis\", has_builtin_services=false, no registered
RPC "
+ "services, and all other protocol service pointers
null)";
+ return -1;
Review Comment:
The same long “requires a Redis-only public listener …” message is
duplicated (also in `SetRedisMaxConnections`). Duplicated contract strings tend
to drift over time; consider factoring this into a shared helper/constant (or a
small function that formats the constraint) so future edits stay consistent.
##########
test/brpc_server_unittest.cpp:
##########
@@ -1424,6 +1426,174 @@ TEST_F(ServerTest, close_idle_connections) {
ASSERT_EQ(0ul, stat.connection_count);
}
+TEST_F(ServerTest, redis_connection_limit_requires_dedicated_listener) {
+ brpc::Server server;
+ EchoServiceImpl echo_service;
+ ASSERT_EQ(0, server.AddService(
+ &echo_service, brpc::SERVER_DOESNT_OWN_SERVICE));
+
+ brpc::ServerOptions opt;
+ opt.redis_service = new brpc::RedisService;
+ opt.redis_max_connections = 1;
+ opt.enabled_protocols = "redis";
+ opt.has_builtin_services = false;
+ const int rc = server.Start("127.0.0.1:0", &opt);
+ if (rc != 0) {
+ delete opt.redis_service;
+ opt.redis_service = nullptr;
+ }
+ EXPECT_EQ(-1, rc);
+}
+
+TEST_F(ServerTest, reject_redis_connections_over_limit) {
+ brpc::Server server;
+ brpc::ServerOptions opt;
+ opt.redis_service = new brpc::RedisService;
+ opt.redis_max_connections = 0;
+ opt.enabled_protocols = "redis";
+ opt.has_builtin_services = false;
+ ASSERT_EQ(0, server.Start("127.0.0.1:0", &opt));
+ ASSERT_EQ(0, server.SetRedisMaxConnections(1));
+
+ const butil::EndPoint ep = server.listen_address();
+ butil::fd_guard first_client(tcp_connect(ep, nullptr));
+ ASSERT_GE(first_client, 0);
+
+ brpc::ServerStatistics stat;
+ for (int retry = 0; retry < 100; ++retry) {
+ server.GetStat(&stat);
+ if (stat.connection_count == 1) {
+ break;
+ }
+ usleep(1000);
+ }
+ ASSERT_EQ(1ul, stat.connection_count);
+
+ // The Redis limit belongs to this acceptor, not to the process. A separate
+ // RPC Server must remain reachable while the Redis listener is full.
+ EchoServiceImpl rpc_service;
+ brpc::Server rpc_server;
+ ASSERT_EQ(0, rpc_server.AddService(
+ &rpc_service, brpc::SERVER_DOESNT_OWN_SERVICE));
+ ASSERT_EQ(0, rpc_server.Start("127.0.0.1:0", nullptr));
+ EXPECT_EQ(-1, rpc_server.SetRedisMaxConnections(1));
+ SendSleepRPC(rpc_server.listen_address(), 0, true);
+
+ butil::fd_guard rejected_client(tcp_connect(ep, nullptr));
+ ASSERT_GE(rejected_client, 0);
+ struct timeval timeout = {1, 0};
+ ASSERT_EQ(0, setsockopt(rejected_client, SOL_SOCKET, SO_RCVTIMEO,
+ &timeout, sizeof(timeout)));
+ const std::string expected = "-ERR max number of clients reached\r\n";
+ const auto expect_rejection = [&expected](int fd) {
+ // These fresh loopback sockets have no induced send backpressure, so
+ // require the full response here despite best-effort delivery in
+ // production. Small reads exercise accumulation without relying on
+ // TCP preserving the server's write boundaries.
+ std::string response;
+ char chunk[7];
+ while (response.size() < expected.size()) {
+ const ssize_t nr = recv(fd, chunk, sizeof(chunk), 0);
+ if (nr > 0) {
+ response.append(chunk, static_cast<size_t>(nr));
+ } else if (nr < 0 && errno == EINTR) {
+ continue;
+ } else {
+ break;
+ }
+ }
+ EXPECT_EQ(expected, response);
+ // The outer fd_guard must also close plaintext rejected connections.
+ EXPECT_EQ(0, recv(fd, chunk, sizeof(chunk), 0));
+ };
+ expect_rejection(rejected_client);
+
+ server.GetStat(&stat);
+ EXPECT_EQ(1ul, stat.connection_count);
+ EXPECT_EQ(1ul, stat.rejected_redis_connection_count);
+
+ ASSERT_EQ(0, server.SetRedisMaxConnections(2));
+ butil::fd_guard second_client(tcp_connect(ep, nullptr));
+ ASSERT_GE(second_client, 0);
+ for (int retry = 0; retry < 100; ++retry) {
+ server.GetStat(&stat);
+ if (stat.connection_count == 2) {
+ break;
+ }
+ usleep(1000);
+ }
+ ASSERT_EQ(2ul, stat.connection_count);
+
+ // Lowering the limit only gates future accepts; it does not disconnect
+ // the two clients that are already established.
+ ASSERT_EQ(0, server.SetRedisMaxConnections(1));
+ server.GetStat(&stat);
+ EXPECT_EQ(2ul, stat.connection_count);
+
+ butil::fd_guard lowered_limit_client(tcp_connect(ep, nullptr));
+ ASSERT_GE(lowered_limit_client, 0);
+ ASSERT_EQ(0, setsockopt(lowered_limit_client, SOL_SOCKET, SO_RCVTIMEO,
+ &timeout, sizeof(timeout)));
+ expect_rejection(lowered_limit_client);
+
+ server.GetStat(&stat);
+ EXPECT_EQ(2ul, stat.connection_count);
+ EXPECT_EQ(2ul, stat.rejected_redis_connection_count);
+
+ first_client.reset(-1);
+ second_client.reset(-1);
+ for (int retry = 0; retry < 100; ++retry) {
+ server.GetStat(&stat);
+ if (stat.connection_count == 0) {
+ break;
+ }
+ usleep(1000);
+ }
+ EXPECT_EQ(0ul, stat.connection_count);
Review Comment:
This test verifies that the count eventually drops to 0 after closing
clients, but it doesn’t verify “slot recovery” by successfully admitting a new
connection after the active count falls back under the (lowered) limit. Adding
a final connect + `connection_count` assertion here would directly test the
“accepted again after active count falls below the limit” behavior described in
the PR.
##########
test/brpc_server_unittest.cpp:
##########
@@ -1424,6 +1426,174 @@ TEST_F(ServerTest, close_idle_connections) {
ASSERT_EQ(0ul, stat.connection_count);
}
+TEST_F(ServerTest, redis_connection_limit_requires_dedicated_listener) {
+ brpc::Server server;
+ EchoServiceImpl echo_service;
+ ASSERT_EQ(0, server.AddService(
+ &echo_service, brpc::SERVER_DOESNT_OWN_SERVICE));
+
+ brpc::ServerOptions opt;
+ opt.redis_service = new brpc::RedisService;
+ opt.redis_max_connections = 1;
+ opt.enabled_protocols = "redis";
+ opt.has_builtin_services = false;
+ const int rc = server.Start("127.0.0.1:0", &opt);
+ if (rc != 0) {
+ delete opt.redis_service;
+ opt.redis_service = nullptr;
+ }
Review Comment:
This manual cleanup is easy to bypass if the test changes (e.g., an
`ASSERT_*` is added between allocation and `Start`, or an early return is
introduced). Using RAII (e.g., a smart pointer that relinquishes ownership only
upon successful `Start`) would make the test safer and reduce future
maintenance risk.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
