thweetkomputer commented on code in PR #3498:
URL: https://github.com/apache/brpc/pull/3498#discussion_r3975489502
##########
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:
Addressed in ebd39a58. `MSG_NOSIGNAL` is now conditional; SDKs without it
use `SO_NOSIGPIPE` when available. If that socket option fails, or neither
mechanism exists, we omit the optional error and let the outer fd_guard close
the connection. This preserves nonblocking rejection without depending on the
application's process-wide SIGPIPE disposition. The Linux build and all four
focused tests passed. The macOS fallback was not built or run in this
environment.
##########
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:
Keeping the full-response assertion for this normal-path regression test.
Each connection is fresh and local, with a 35-byte server response, no prior
server writes, no induced backpressure, and no client request bytes that could
remain unread when the server closes. The EOF read also has SO_RCVTIMEO set; it
does not require closure to be observable immediately. These conditions
deliberately exercise normal error delivery and orderly closure. Accepting an
empty response would also let a regression that removes the send entirely pass.
Backpressure/error behavior remains documented as best-effort; that contract
does not require weakening the normal-path assertion. All four focused tests
passed again locally. No cross-platform reset failure was reproduced here, so I
have not added blanket ECONNRESET tolerance.
##########
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:
Addressed in ebd39a58. After both existing clients close and the active
count reaches zero, the test opens a new idle connection under the lowered
limit, waits for connection_count == 1, and verifies the rejection count stays
at 2. This directly verifies renewed admission using the recovered slot. The
updated test and the other three focused tests passed.
##########
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:
Addressed in ebd39a58. Startup and runtime validation now use the same
`kRedisOnlyPublicListenerRequirements` character array beside the predicate.
The operation-specific prefix and log severity remain at each call site. This
introduces no non-trivial static destructor. The build and four focused tests
passed.
##########
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:
Addressed in ebd39a58 with a std::unique_ptr. One ownership detail matters
here: Start() can copy the options and take ownership before failing later, so
releasing only on successful Start() would risk double deletion. The test
checks whether server.options().redis_service matches the guarded pointer and
relinquishes ownership when the Server has actually adopted it. Early
validation failures leave ownership with the smart pointer. This is documented
beside the check. The build and four focused tests passed.
--
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]
