You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@brpc.apache.org by "thweetkomputer (via GitHub)" <gi...@apache.org> on 2026/08/28 04:24:22 UTC
[PR] feat: limit Redis server connections (brpc)
thweetkomputer opened a new pull request, #3498:
URL: https://github.com/apache/brpc/pull/3498
### What problem does this PR solve?
Issue Number: N/A
Problem Summary:
`RedisService` has no server-side connection limit. `max_concurrency` limits requests only after protocol parsing, so idle clients can consume all available connection resources. On an SSL listener, accepting excess clients into brpc also allows them to enter the comparatively expensive TLS handshake before an application can reject them.
### What is changed and the side effects?
Changed:
- Add `ServerOptions.redis_max_connections` (`0` keeps the existing unlimited behavior).
- Reserve connection slots atomically in `Acceptor`, before creating a brpc `Socket`, so idle clients count and concurrent accepts cannot exceed the limit.
- Require a non-zero limit to use a dedicated Redis public listener: `redis_service` set, `enabled_protocols="redis"`, builtin services disabled, and no RPC or other protocol services on that `Server`.
- Return `-ERR max number of clients reached` to excess plaintext clients. If SSL is configured, close the accepted fd immediately without creating a brpc `Socket` or starting TLS.
- Keep internal acceptors and other `Server` instances unlimited, and expose the cumulative rejection count as `ServerStatistics.rejected_redis_connection_count`.
- Document the option in the English and Chinese server guides and add regression tests for listener isolation, plaintext rejection, independent RPC availability, slot recovery, and pre-TLS rejection.
Side effects:
- Performance effects: accepted connections now perform one relaxed atomic increment/decrement; limited listeners use a relaxed CAS for admission. There is no request hot-path cost.
- Breaking backward compatibility: the default behavior remains unlimited and existing source configuration is unchanged. The new public fields change the in-memory layouts of `ServerOptions` and `ServerStatistics`, so applications using a prebuilt shared brpc library must rebuild with the updated headers and library.
---
### Check List:
Tests / Checks:
- [x] `cmake --build build --target brpc_server_unittest --parallel 4`
- [x] The three new Redis connection-limit tests pass.
- [x] `ServerTest.close_idle_connections` passes.
- [ ] The complete `brpc_server_unittest` is not clean in this local environment: the existing timing-sensitive overload assertions in `ServerTest.http_error_code` and `ServerTest.max_concurrency` observe a successful third RPC instead of overload rejection. The focused tests above pass; this PR does not change request concurrency code.
- [x] `git diff --check upstream/master...HEAD`
Code / Documentation:
- [x] The code follows the repository style.
- [x] Public API behavior and the accept/TLS invariants are documented.
- [x] The final diff was reviewed against the merge base.
Reviewer focus:
- Whether requiring a dedicated Redis listener is the desired public contract. Protocol identification happens after TLS, so enforcing this invariant is what lets the acceptor reject excess TLS connections before authentication without affecting RPC listeners.
- The relaxed atomic slot lifecycle: reserve immediately after `accept`, release on `Socket::Create` failure or `BeforeRecycle`.
Rollback:
Set `redis_max_connections` to `0` to retain unlimited admission, or revert this commit to remove the API and accounting fields.
--
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: dev-unsubscribe@brpc.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@brpc.apache.org
For additional commands, e-mail: dev-help@brpc.apache.org
Re: [PR] feat: limit server connections per public listener (brpc)
Posted by "thweetkomputer (via GitHub)" <gi...@apache.org>.
thweetkomputer commented on code in PR #3498:
URL: https://github.com/apache/brpc/pull/3498#discussion_r4001595791
##########
src/brpc/acceptor.cpp:
##########
@@ -52,6 +57,14 @@ Acceptor::~Acceptor() {
int Acceptor::StartAccept(int listened_fd, int idle_timeout_sec,
const std::shared_ptr<SocketSSLContext>& ssl_ctx,
bool force_ssl) {
+ return StartAccept(
+ listened_fd, idle_timeout_sec, ssl_ctx, force_ssl, 0);
Review Comment:
Done in 6e3f9a6b: the forwarding `StartAccept()` call is now on a single line.
##########
src/brpc/acceptor.cpp:
##########
@@ -275,7 +317,15 @@ void Acceptor::OnNewConnectionsUntilEAGAIN(Socket* acception) {
acception->SetFailed(EINVAL, "Impossible! acception->user() MUST be Acceptor");
return;
}
-
+
+ if (!am->TryAcquireConnectionSlot()) {
+ am->_rejected_connection_count.fetch_add(
+ 1, butil::memory_order_relaxed);
+ // in_fd closes the connection before Socket::Create(), protocol
+ // parsing or TLS authentication, without a protocol-specific reply.
+ continue;
+ }
Review Comment:
Added `LOG_EVERY_SECOND(WARNING)` in 6e3f9a6b alongside the rejection counter increment. The warning includes the listening endpoint and explains that the `max_connections` limit was reached. All nine focused connection-limit/idle-connection tests pass, and the rate-limited warning was observed during the rejection tests.
##########
test/brpc_server_unittest.cpp:
##########
@@ -1426,6 +1431,378 @@ TEST_F(ServerTest, close_idle_connections) {
ASSERT_EQ(0ul, stat.connection_count);
}
+static testing::AssertionResult WaitForServerConnections(
+ const brpc::Server& server, size_t expected) {
+ brpc::ServerStatistics stat;
+ const int64_t deadline = butil::gettimeofday_us() + 1000000;
+ do {
+ server.GetStat(&stat);
+ if (stat.connection_count == expected) {
+ return testing::AssertionSuccess();
+ }
+ usleep(1000);
+ } while (butil::gettimeofday_us() < deadline);
+ return testing::AssertionFailure()
+ << "Expected " << expected << " connections, got "
+ << stat.connection_count;
+}
+
+static void ExpectConnectionClosed(int fd) {
+ const struct timeval timeout = {1, 0};
+ ASSERT_EQ(0, setsockopt(fd, SOL_SOCKET, SO_RCVTIMEO,
+ &timeout, sizeof(timeout)));
+ char response;
+ ssize_t nr;
+ do {
+ nr = recv(fd, &response, sizeof(response), 0);
+ } while (nr < 0 && errno == EINTR);
+ // Fresh idle clients have sent no data: rejection must close the fd
+ // without sending a protocol-specific error or waiting for a request.
+ EXPECT_EQ(0, nr);
+}
+
+class ConnectionLimitPingHandler : public brpc::RedisCommandHandler {
+public:
+ brpc::RedisCommandHandlerResult Run(
+ brpc::RedisConnContext*, const std::vector<butil::StringPiece>&,
+ brpc::RedisReply* output, bool) override {
+ output->SetStatus("PONG");
+ return brpc::REDIS_CMD_HANDLED;
+ }
+};
+
+TEST_F(ServerTest, connection_limit_with_mixed_protocols) {
+ ConnectionLimitPingHandler ping_handler;
+ EchoServiceImpl echo_service;
+ brpc::Server server;
+ ASSERT_EQ(0, server.AddService(
+ &echo_service, brpc::SERVER_DOESNT_OWN_SERVICE));
+ brpc::ServerOptions opt;
+ opt.redis_service = new brpc::RedisService;
+ opt.redis_service->AddCommandHandler("ping", &ping_handler);
+ opt.max_connections = 3;
+ ASSERT_EQ(0, server.Start("127.0.0.1:0", &opt));
+
+ // Each protocol uses a separate persistent connection on the same port.
+ brpc::ChannelOptions copt;
+ copt.connection_type = "single";
+ copt.timeout_ms = 1000;
+ copt.max_retry = 0;
+ copt.connection_group = "connection_limit_rpc";
+ brpc::Channel rpc_channel;
+ ASSERT_EQ(0, rpc_channel.Init(server.listen_address(), &copt));
+ brpc::Controller rpc_cntl;
+ test::EchoRequest req;
+ test::EchoResponse res;
+ req.set_message(EXP_REQUEST);
+ test::EchoService_Stub stub(&rpc_channel);
+ stub.Echo(&rpc_cntl, &req, &res, nullptr);
+ ASSERT_FALSE(rpc_cntl.Failed()) << rpc_cntl.ErrorText();
+
+ copt.protocol = "redis";
+ copt.connection_group = "connection_limit_redis";
+ brpc::Channel redis_channel;
+ ASSERT_EQ(0, redis_channel.Init(server.listen_address(), &copt));
+ brpc::RedisRequest redis_req;
+ brpc::RedisResponse redis_res;
+ brpc::Controller redis_cntl;
+ ASSERT_TRUE(redis_req.AddCommand("ping"));
+ redis_channel.CallMethod(
+ nullptr, &redis_cntl, &redis_req, &redis_res, nullptr);
+ ASSERT_FALSE(redis_cntl.Failed()) << redis_cntl.ErrorText();
+ ASSERT_EQ(1, redis_res.reply_size());
+ ASSERT_STREQ("PONG", redis_res.reply(0).c_str());
+
+ copt.protocol = "http";
+ copt.connection_type = "pooled";
+ copt.connection_group = "connection_limit_http";
+ brpc::Channel http_channel;
+ ASSERT_EQ(0, http_channel.Init(server.listen_address(), &copt));
+ brpc::Controller http_cntl;
+ http_cntl.http_request().uri() = "/status";
+ http_channel.CallMethod(nullptr, &http_cntl, nullptr, nullptr, nullptr);
+ ASSERT_FALSE(http_cntl.Failed()) << http_cntl.ErrorText();
+ ASSERT_TRUE(WaitForServerConnections(server, 3));
+
+ butil::fd_guard rejected_client(
+ tcp_connect(server.listen_address(), nullptr));
+ ASSERT_GE(rejected_client, 0);
+ ExpectConnectionClosed(rejected_client);
+ brpc::ServerStatistics stat;
+ server.GetStat(&stat);
+ EXPECT_EQ(3ul, stat.connection_count);
+ EXPECT_EQ(1ul, stat.rejected_connection_count);
+
+ // Requests on admitted sockets still work when the listener is full.
+ rpc_cntl.Reset();
+ stub.Echo(&rpc_cntl, &req, &res, nullptr);
+ ASSERT_FALSE(rpc_cntl.Failed()) << rpc_cntl.ErrorText();
+}
+
+TEST_F(ServerTest, connection_limit_runtime_updates) {
+ brpc::Server server;
+ EXPECT_EQ(-1, server.SetMaxConnections(1));
+ ASSERT_EQ(0, server.Start("127.0.0.1:0", nullptr));
+ const butil::EndPoint ep = server.listen_address();
+ butil::fd_guard first_client(tcp_connect(ep, nullptr));
+ butil::fd_guard second_client(tcp_connect(ep, nullptr));
+ ASSERT_GE(first_client, 0);
+ ASSERT_GE(second_client, 0);
+ ASSERT_TRUE(WaitForServerConnections(server, 2));
+
+ // Enabling a limit must count clients admitted while it was unlimited.
+ ASSERT_EQ(0, server.SetMaxConnections(1));
+ butil::fd_guard rejected_client(tcp_connect(ep, nullptr));
+ ASSERT_GE(rejected_client, 0);
+ ExpectConnectionClosed(rejected_client);
+ ASSERT_TRUE(WaitForServerConnections(server, 2));
+
+ // A separate Server remains reachable while this public 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));
+ SendSleepRPC(rpc_server.listen_address(), 0, true);
+
+ ASSERT_EQ(0, server.SetMaxConnections(3));
+ butil::fd_guard third_client(tcp_connect(ep, nullptr));
+ ASSERT_GE(third_client, 0);
+ ASSERT_TRUE(WaitForServerConnections(server, 3));
+
+ // Lowering the limit leaves established clients connected, but rejects
+ // new clients until the active count drops strictly below the limit.
+ ASSERT_EQ(0, server.SetMaxConnections(1));
+ ASSERT_TRUE(WaitForServerConnections(server, 3));
+ butil::fd_guard lowered_limit_client(tcp_connect(ep, nullptr));
+ ASSERT_GE(lowered_limit_client, 0);
+ ExpectConnectionClosed(lowered_limit_client);
+ first_client.reset(-1);
+ second_client.reset(-1);
+ ASSERT_TRUE(WaitForServerConnections(server, 1));
+ butil::fd_guard at_limit_client(tcp_connect(ep, nullptr));
+ ASSERT_GE(at_limit_client, 0);
+ ExpectConnectionClosed(at_limit_client);
+
+ third_client.reset(-1);
+ ASSERT_TRUE(WaitForServerConnections(server, 0));
+ butil::fd_guard recovered_client(tcp_connect(ep, nullptr));
+ ASSERT_GE(recovered_client, 0);
+ ASSERT_TRUE(WaitForServerConnections(server, 1));
+
+ ASSERT_EQ(0, server.SetMaxConnections(0));
+ first_client.reset(tcp_connect(ep, nullptr));
+ second_client.reset(tcp_connect(ep, nullptr));
+ ASSERT_GE(first_client, 0);
+ ASSERT_GE(second_client, 0);
+ ASSERT_TRUE(WaitForServerConnections(server, 3));
+ brpc::ServerStatistics stat;
+ server.GetStat(&stat);
+ EXPECT_EQ(3ul, stat.rejected_connection_count);
+ EXPECT_EQ(0ul, server.options().max_connections);
+
+ ASSERT_EQ(0, server.Stop(0));
+ EXPECT_EQ(-1, server.SetMaxConnections(1));
+ ASSERT_EQ(0, server.Join());
+ ASSERT_TRUE(WaitForServerConnections(server, 0));
+}
+
+TEST_F(ServerTest, connection_limit_keeps_internal_listener_available) {
+ brpc::Server server;
+ brpc::ServerOptions opt;
+ opt.max_connections = 1;
+ opt.internal_port = 8614;
+ ASSERT_EQ(0, server.Start("127.0.0.1:0", &opt));
Review Comment:
Updated in 6e3f9a6b to reuse `StartWithInternalPort()`. I moved the existing port-selection helpers above their first use without changing their implementation. Verified that this test passes with `127.0.0.1:8614` already occupied; all nine focused connection-limit/idle-connection tests also pass.
--
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: dev-unsubscribe@brpc.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@brpc.apache.org
For additional commands, e-mail: dev-help@brpc.apache.org
Re: [PR] feat: limit Redis server connections (brpc)
Posted by "thweetkomputer (via GitHub)" <gi...@apache.org>.
thweetkomputer commented on code in PR #3498:
URL: https://github.com/apache/brpc/pull/3498#discussion_r3975102703
##########
src/brpc/acceptor.cpp:
##########
@@ -200,9 +214,61 @@ 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) {
+ _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;
+ }
Review Comment:
This fd is owned by the caller's `butil::fd_guard in_fd(accept(...))` in `OnNewConnectionsUntilEAGAIN()`. `RejectRedisConnection(int fd)` only borrows the descriptor. After it returns, the caller executes `continue`, which destroys `in_fd`; `fd_guard::~fd_guard()` calls `::close()` for every non-negative fd. This covers both the SSL early return and the plaintext best-effort send path.
Ownership is released only after a successful `Socket::Create()`, which the rejection path never reaches. Adding a raw `close(fd)` here would therefore cause a double close and could close a descriptor reused by another thread.
Verified on f6ed17a5: rebuilt `brpc_server_unittest` and ran the dedicated-listener, plaintext/dynamic-limit, pre-TLS rejection, and idle-connection tests; all four passed. In particular, `reject_redis_connection_before_tls_handshake` asserts that the rejected client receives EOF without sending a ClientHello. No fd-lifecycle change is needed.
##########
src/brpc/acceptor.cpp:
##########
@@ -275,7 +341,12 @@ void Acceptor::OnNewConnectionsUntilEAGAIN(Socket* acception) {
acception->SetFailed(EINVAL, "Impossible! acception->user() MUST be Acceptor");
return;
}
-
+
+ if (!am->TryAcquireRedisConnectionSlot()) {
+ am->RejectRedisConnection(in_fd);
+ continue;
+ }
Review Comment:
`in_fd` is a `butil::fd_guard`, not a raw int. It is declared inside the `while` loop, so this `continue` runs its destructor, which closes the accepted fd. The guard remains the owner until `in_fd.release()` after successful `Socket::Create()`; that ownership transfer is not reached on rejection.
Thus the existing rejection path already closes the connection. Adding `close(in_fd)` before `continue`, or closing it inside `RejectRedisConnection()`, would double-close the descriptor. The existing pre-TLS rejection test also verifies peer-visible EOF and passed locally on f6ed17a5.
--
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: dev-unsubscribe@brpc.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@brpc.apache.org
For additional commands, e-mail: dev-help@brpc.apache.org
Re: [PR] feat: limit server connections per public listener (brpc)
Posted by "wwbmmm (via GitHub)" <gi...@apache.org>.
wwbmmm commented on PR #3498:
URL: https://github.com/apache/brpc/pull/3498#issuecomment-5652136428
LGTM
--
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: dev-unsubscribe@brpc.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@brpc.apache.org
For additional commands, e-mail: dev-help@brpc.apache.org
Re: [PR] feat: limit server connections per public listener (brpc)
Posted by "chenBright (via GitHub)" <gi...@apache.org>.
chenBright merged PR #3498:
URL: https://github.com/apache/brpc/pull/3498
--
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: dev-unsubscribe@brpc.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@brpc.apache.org
For additional commands, e-mail: dev-help@brpc.apache.org
Re: [PR] feat: limit Redis server connections (brpc)
Posted by "thweetkomputer (via GitHub)" <gi...@apache.org>.
thweetkomputer commented on code in PR #3498:
URL: https://github.com/apache/brpc/pull/3498#discussion_r3975133612
##########
test/brpc_server_unittest.cpp:
##########
@@ -1424,6 +1425,159 @@ 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_GT(first_client, 0);
Review Comment:
Fixed in 2c7f28478c03874a95f492d837afdb4abab9a9be: changed all six fd-validity assertions added by this PR to `ASSERT_GE(fd, 0)`, so fd 0 is accepted. Rebuilt `brpc_server_unittest` and reran the dedicated-listener, plaintext/dynamic-limit, pre-TLS rejection, and idle-connection tests; all four passed.
##########
src/brpc/server.cpp:
##########
@@ -870,6 +893,18 @@ 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=redis, "
+ "no RPC or builtin services)";
Review Comment:
Fixed in 2c7f28478c03874a95f492d837afdb4abab9a9be: expanded both startup and runtime setter validation messages to state that `redis_service` must be set, `enabled_protocols` must be exactly "redis", builtin services must be disabled, no RPC services may be registered, and all other protocol service pointers must be null. The configuration checks themselves are unchanged. Rebuilt `brpc_server_unittest`; all four focused tests passed, including the startup and runtime rejection paths.
--
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: dev-unsubscribe@brpc.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@brpc.apache.org
For additional commands, e-mail: dev-help@brpc.apache.org
Re: [PR] feat: limit Redis server connections (brpc)
Posted by "thweetkomputer (via GitHub)" <gi...@apache.org>.
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: dev-unsubscribe@brpc.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@brpc.apache.org
For additional commands, e-mail: dev-help@brpc.apache.org
Re: [PR] feat: limit Redis server connections (brpc)
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #3498:
URL: https://github.com/apache/brpc/pull/3498#discussion_r3885902102
##########
src/brpc/acceptor.cpp:
##########
@@ -200,9 +214,53 @@ 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) {
+ _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";
+ ssize_t nwritten;
+ do {
+ nwritten = send(fd,
+ response,
+ sizeof(response) - 1,
+ MSG_DONTWAIT | MSG_NOSIGNAL);
+ } while (nwritten < 0 && errno == EINTR);
Review Comment:
RejectRedisConnection() only retries send() on EINTR and ignores partial writes/EAGAIN, so an over-limit plaintext client may receive a truncated error (or EOF) even though the API/doc promise a full `-ERR ...` line. Consider handling partial writes and bailing out cleanly on EAGAIN/EWOULDBLOCK to maximize the chance the client sees the full response without blocking the accept loop.
--
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: dev-unsubscribe@brpc.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@brpc.apache.org
For additional commands, e-mail: dev-help@brpc.apache.org
Re: [PR] feat: limit server connections per public listener (brpc)
Posted by "chenBright (via GitHub)" <gi...@apache.org>.
chenBright commented on code in PR #3498:
URL: https://github.com/apache/brpc/pull/3498#discussion_r3999940207
##########
src/brpc/acceptor.cpp:
##########
@@ -52,6 +57,14 @@ Acceptor::~Acceptor() {
int Acceptor::StartAccept(int listened_fd, int idle_timeout_sec,
const std::shared_ptr<SocketSSLContext>& ssl_ctx,
bool force_ssl) {
+ return StartAccept(
+ listened_fd, idle_timeout_sec, ssl_ctx, force_ssl, 0);
Review Comment:
Please write it in a single line.
##########
src/brpc/acceptor.cpp:
##########
@@ -275,7 +317,15 @@ void Acceptor::OnNewConnectionsUntilEAGAIN(Socket* acception) {
acception->SetFailed(EINVAL, "Impossible! acception->user() MUST be Acceptor");
return;
}
-
+
+ if (!am->TryAcquireConnectionSlot()) {
+ am->_rejected_connection_count.fetch_add(
+ 1, butil::memory_order_relaxed);
+ // in_fd closes the connection before Socket::Create(), protocol
+ // parsing or TLS authentication, without a protocol-specific reply.
+ continue;
+ }
Review Comment:
The rejection branch closes the fd and continues with only the `_rejected_connection_count` updated. Under a burst of rejected connections there is no log trace to help operators understand why connections are being dropped. Consider adding a rate limited log such as LOG_EVERY_SECOND(WARNING) such alongside the counter increment.
##########
test/brpc_server_unittest.cpp:
##########
@@ -1426,6 +1431,378 @@ TEST_F(ServerTest, close_idle_connections) {
ASSERT_EQ(0ul, stat.connection_count);
}
+static testing::AssertionResult WaitForServerConnections(
+ const brpc::Server& server, size_t expected) {
+ brpc::ServerStatistics stat;
+ const int64_t deadline = butil::gettimeofday_us() + 1000000;
+ do {
+ server.GetStat(&stat);
+ if (stat.connection_count == expected) {
+ return testing::AssertionSuccess();
+ }
+ usleep(1000);
+ } while (butil::gettimeofday_us() < deadline);
+ return testing::AssertionFailure()
+ << "Expected " << expected << " connections, got "
+ << stat.connection_count;
+}
+
+static void ExpectConnectionClosed(int fd) {
+ const struct timeval timeout = {1, 0};
+ ASSERT_EQ(0, setsockopt(fd, SOL_SOCKET, SO_RCVTIMEO,
+ &timeout, sizeof(timeout)));
+ char response;
+ ssize_t nr;
+ do {
+ nr = recv(fd, &response, sizeof(response), 0);
+ } while (nr < 0 && errno == EINTR);
+ // Fresh idle clients have sent no data: rejection must close the fd
+ // without sending a protocol-specific error or waiting for a request.
+ EXPECT_EQ(0, nr);
+}
+
+class ConnectionLimitPingHandler : public brpc::RedisCommandHandler {
+public:
+ brpc::RedisCommandHandlerResult Run(
+ brpc::RedisConnContext*, const std::vector<butil::StringPiece>&,
+ brpc::RedisReply* output, bool) override {
+ output->SetStatus("PONG");
+ return brpc::REDIS_CMD_HANDLED;
+ }
+};
+
+TEST_F(ServerTest, connection_limit_with_mixed_protocols) {
+ ConnectionLimitPingHandler ping_handler;
+ EchoServiceImpl echo_service;
+ brpc::Server server;
+ ASSERT_EQ(0, server.AddService(
+ &echo_service, brpc::SERVER_DOESNT_OWN_SERVICE));
+ brpc::ServerOptions opt;
+ opt.redis_service = new brpc::RedisService;
+ opt.redis_service->AddCommandHandler("ping", &ping_handler);
+ opt.max_connections = 3;
+ ASSERT_EQ(0, server.Start("127.0.0.1:0", &opt));
+
+ // Each protocol uses a separate persistent connection on the same port.
+ brpc::ChannelOptions copt;
+ copt.connection_type = "single";
+ copt.timeout_ms = 1000;
+ copt.max_retry = 0;
+ copt.connection_group = "connection_limit_rpc";
+ brpc::Channel rpc_channel;
+ ASSERT_EQ(0, rpc_channel.Init(server.listen_address(), &copt));
+ brpc::Controller rpc_cntl;
+ test::EchoRequest req;
+ test::EchoResponse res;
+ req.set_message(EXP_REQUEST);
+ test::EchoService_Stub stub(&rpc_channel);
+ stub.Echo(&rpc_cntl, &req, &res, nullptr);
+ ASSERT_FALSE(rpc_cntl.Failed()) << rpc_cntl.ErrorText();
+
+ copt.protocol = "redis";
+ copt.connection_group = "connection_limit_redis";
+ brpc::Channel redis_channel;
+ ASSERT_EQ(0, redis_channel.Init(server.listen_address(), &copt));
+ brpc::RedisRequest redis_req;
+ brpc::RedisResponse redis_res;
+ brpc::Controller redis_cntl;
+ ASSERT_TRUE(redis_req.AddCommand("ping"));
+ redis_channel.CallMethod(
+ nullptr, &redis_cntl, &redis_req, &redis_res, nullptr);
+ ASSERT_FALSE(redis_cntl.Failed()) << redis_cntl.ErrorText();
+ ASSERT_EQ(1, redis_res.reply_size());
+ ASSERT_STREQ("PONG", redis_res.reply(0).c_str());
+
+ copt.protocol = "http";
+ copt.connection_type = "pooled";
+ copt.connection_group = "connection_limit_http";
+ brpc::Channel http_channel;
+ ASSERT_EQ(0, http_channel.Init(server.listen_address(), &copt));
+ brpc::Controller http_cntl;
+ http_cntl.http_request().uri() = "/status";
+ http_channel.CallMethod(nullptr, &http_cntl, nullptr, nullptr, nullptr);
+ ASSERT_FALSE(http_cntl.Failed()) << http_cntl.ErrorText();
+ ASSERT_TRUE(WaitForServerConnections(server, 3));
+
+ butil::fd_guard rejected_client(
+ tcp_connect(server.listen_address(), nullptr));
+ ASSERT_GE(rejected_client, 0);
+ ExpectConnectionClosed(rejected_client);
+ brpc::ServerStatistics stat;
+ server.GetStat(&stat);
+ EXPECT_EQ(3ul, stat.connection_count);
+ EXPECT_EQ(1ul, stat.rejected_connection_count);
+
+ // Requests on admitted sockets still work when the listener is full.
+ rpc_cntl.Reset();
+ stub.Echo(&rpc_cntl, &req, &res, nullptr);
+ ASSERT_FALSE(rpc_cntl.Failed()) << rpc_cntl.ErrorText();
+}
+
+TEST_F(ServerTest, connection_limit_runtime_updates) {
+ brpc::Server server;
+ EXPECT_EQ(-1, server.SetMaxConnections(1));
+ ASSERT_EQ(0, server.Start("127.0.0.1:0", nullptr));
+ const butil::EndPoint ep = server.listen_address();
+ butil::fd_guard first_client(tcp_connect(ep, nullptr));
+ butil::fd_guard second_client(tcp_connect(ep, nullptr));
+ ASSERT_GE(first_client, 0);
+ ASSERT_GE(second_client, 0);
+ ASSERT_TRUE(WaitForServerConnections(server, 2));
+
+ // Enabling a limit must count clients admitted while it was unlimited.
+ ASSERT_EQ(0, server.SetMaxConnections(1));
+ butil::fd_guard rejected_client(tcp_connect(ep, nullptr));
+ ASSERT_GE(rejected_client, 0);
+ ExpectConnectionClosed(rejected_client);
+ ASSERT_TRUE(WaitForServerConnections(server, 2));
+
+ // A separate Server remains reachable while this public 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));
+ SendSleepRPC(rpc_server.listen_address(), 0, true);
+
+ ASSERT_EQ(0, server.SetMaxConnections(3));
+ butil::fd_guard third_client(tcp_connect(ep, nullptr));
+ ASSERT_GE(third_client, 0);
+ ASSERT_TRUE(WaitForServerConnections(server, 3));
+
+ // Lowering the limit leaves established clients connected, but rejects
+ // new clients until the active count drops strictly below the limit.
+ ASSERT_EQ(0, server.SetMaxConnections(1));
+ ASSERT_TRUE(WaitForServerConnections(server, 3));
+ butil::fd_guard lowered_limit_client(tcp_connect(ep, nullptr));
+ ASSERT_GE(lowered_limit_client, 0);
+ ExpectConnectionClosed(lowered_limit_client);
+ first_client.reset(-1);
+ second_client.reset(-1);
+ ASSERT_TRUE(WaitForServerConnections(server, 1));
+ butil::fd_guard at_limit_client(tcp_connect(ep, nullptr));
+ ASSERT_GE(at_limit_client, 0);
+ ExpectConnectionClosed(at_limit_client);
+
+ third_client.reset(-1);
+ ASSERT_TRUE(WaitForServerConnections(server, 0));
+ butil::fd_guard recovered_client(tcp_connect(ep, nullptr));
+ ASSERT_GE(recovered_client, 0);
+ ASSERT_TRUE(WaitForServerConnections(server, 1));
+
+ ASSERT_EQ(0, server.SetMaxConnections(0));
+ first_client.reset(tcp_connect(ep, nullptr));
+ second_client.reset(tcp_connect(ep, nullptr));
+ ASSERT_GE(first_client, 0);
+ ASSERT_GE(second_client, 0);
+ ASSERT_TRUE(WaitForServerConnections(server, 3));
+ brpc::ServerStatistics stat;
+ server.GetStat(&stat);
+ EXPECT_EQ(3ul, stat.rejected_connection_count);
+ EXPECT_EQ(0ul, server.options().max_connections);
+
+ ASSERT_EQ(0, server.Stop(0));
+ EXPECT_EQ(-1, server.SetMaxConnections(1));
+ ASSERT_EQ(0, server.Join());
+ ASSERT_TRUE(WaitForServerConnections(server, 0));
+}
+
+TEST_F(ServerTest, connection_limit_keeps_internal_listener_available) {
+ brpc::Server server;
+ brpc::ServerOptions opt;
+ opt.max_connections = 1;
+ opt.internal_port = 8614;
+ ASSERT_EQ(0, server.Start("127.0.0.1:0", &opt));
Review Comment:
UT hardcodes `opt.internal_port = 8614`, while every other test in this PR binds to 127.0.0.1:0. Since internal_port cannot be 0, a fixed port can make this test flaky when the port is already taken, for example under parallel runs or on a busy CI host.
Please use `StartWithInternalPort()` to automatically pick unused port.
https://github.com/apache/brpc/blob/2a75bbc3878c14505b92e580e6d0c7f9bcf3e686/test/brpc_server_unittest.cpp#L1759-L1771
--
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: dev-unsubscribe@brpc.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@brpc.apache.org
For additional commands, e-mail: dev-help@brpc.apache.org
Re: [PR] feat: limit Redis server connections (brpc)
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
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: dev-unsubscribe@brpc.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@brpc.apache.org
For additional commands, e-mail: dev-help@brpc.apache.org
Re: [PR] feat: limit Redis server connections (brpc)
Posted by "thweetkomputer (via GitHub)" <gi...@apache.org>.
thweetkomputer commented on code in PR #3498:
URL: https://github.com/apache/brpc/pull/3498#discussion_r3975316236
##########
src/brpc/acceptor.cpp:
##########
@@ -200,9 +214,61 @@ 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) {
+ _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;
+ }
Review Comment:
Agreed that the ownership contract should be explicit here. Added comments in f05f8f61 at both `RejectRedisConnection()` and its call site: the function borrows the fd; the caller retains it in `butil::fd_guard`, and `continue` destroys the guard and closes the connection on both SSL and plaintext paths. No additional close is needed inside the helper. The plaintext test now also checks EOF, complementing the existing pre-TLS EOF test. Rebuilt `brpc_server_unittest`; all four focused tests passed.
##########
src/brpc/acceptor.cpp:
##########
@@ -200,9 +214,61 @@ 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) {
+ _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";
+ 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);
+ if (nwritten > 0) {
+ offset += nwritten;
+ } else if (nwritten < 0 && errno == EINTR) {
+ continue;
+ } else {
+ break;
+ }
+ }
Review Comment:
Best-effort delivery is intentional: waiting for writability in the accept loop would stall admission for other clients. In f05f8f61, both server guides now explicitly describe the nonblocking attempt, subsequent close, and the possibility of a partial or missing response under backpressure or socket errors. Added the same rationale beside the send loop. The loopback test still requires the complete error because these are fresh connections with no induced send backpressure; weakening that assertion to accept an empty response would miss a broken normal send path. This test assumption is now documented. No blocking retry was introduced. All four focused tests passed.
##########
test/brpc_server_unittest.cpp:
##########
@@ -1424,6 +1425,159 @@ 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)));
+ char response[64];
+ const ssize_t nr = recv(rejected_client, response, sizeof(response), 0);
+ const std::string expected = "-ERR max number of clients reached\r\n";
+ ASSERT_EQ(expected.size(), (size_t)nr);
+ EXPECT_EQ(expected, std::string(response, (size_t)nr));
Review Comment:
Fixed in f05f8f61: both plaintext rejection checks now accumulate reads until the expected response has been collected, EOF occurs, or a read fails/times out. EINTR is retried and the existing receive timeout remains in effect. The test deliberately uses a seven-byte buffer so the full response must span multiple recv calls, then verifies EOF after the error. Rebuilt `brpc_server_unittest`; all 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: dev-unsubscribe@brpc.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@brpc.apache.org
For additional commands, e-mail: dev-help@brpc.apache.org
Re: [PR] feat: limit server connections per public listener (brpc)
Posted by "thweetkomputer (via GitHub)" <gi...@apache.org>.
thweetkomputer commented on PR #3498:
URL: https://github.com/apache/brpc/pull/3498#issuecomment-5644724335
@wwbmmm Thanks for the suggestion. Updated in 702cba2a: `ServerOptions.max_connections` and `Server::SetMaxConnections()` now apply per public listener across all protocols. Services sharing a port share the limit; separate listeners have independent limits, even when they share the same service object. The internal listener remains unlimited.
Excess connections are closed before protocol parsing or TLS. Documentation and tests have been updated, and all 53 tests across the Server, input messenger, and SSL suites passed locally.
--
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: dev-unsubscribe@brpc.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@brpc.apache.org
For additional commands, e-mail: dev-help@brpc.apache.org
Re: [PR] feat: limit Redis server connections (brpc)
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #3498:
URL: https://github.com/apache/brpc/pull/3498#discussion_r3975231898
##########
src/brpc/acceptor.cpp:
##########
@@ -200,9 +214,61 @@ 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) {
+ _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;
+ }
Review Comment:
RejectRedisConnection() does not close/shutdown `fd` (and in the SSL case returns immediately). If `fd` lifetime is intentionally managed by an outer `fd_guard` in the accept loop, consider documenting that expectation here explicitly; otherwise, strongly consider closing the socket inside RejectRedisConnection() to make the pre-TLS rejection invariant self-contained and harder to regress.
##########
src/brpc/acceptor.cpp:
##########
@@ -200,9 +214,61 @@ 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) {
+ _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";
+ 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);
+ if (nwritten > 0) {
+ offset += nwritten;
+ } else if (nwritten < 0 && errno == EINTR) {
+ continue;
+ } else {
+ break;
+ }
+ }
Review Comment:
The send() loop breaks on EAGAIN/EWOULDBLOCK (and any non-EINTR error), which can result in clients not receiving the documented plaintext error response under load/backpressure. If best-effort delivery is acceptable, the docs/tests should tolerate partial/no response; otherwise, consider a bounded retry strategy (e.g., poll/select with a short deadline) or using an existing utility that handles partial writes robustly without blocking indefinitely.
##########
test/brpc_server_unittest.cpp:
##########
@@ -1424,6 +1425,159 @@ 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)));
+ char response[64];
+ const ssize_t nr = recv(rejected_client, response, sizeof(response), 0);
+ const std::string expected = "-ERR max number of clients reached\r\n";
+ ASSERT_EQ(expected.size(), (size_t)nr);
+ EXPECT_EQ(expected, std::string(response, (size_t)nr));
Review Comment:
This test assumes a single recv() returns the full error line, but TCP does not preserve message boundaries; recv() may return a partial payload even when the server writes it in one call. To avoid flaky failures, read in a loop until either the expected bytes are collected, EOF is hit, or the timeout expires (and then compare the accumulated buffer to `expected`).
--
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: dev-unsubscribe@brpc.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@brpc.apache.org
For additional commands, e-mail: dev-help@brpc.apache.org
Re: [PR] feat: limit Redis server connections (brpc)
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #3498:
URL: https://github.com/apache/brpc/pull/3498#discussion_r3958351666
##########
src/brpc/acceptor.cpp:
##########
@@ -200,9 +214,61 @@ 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) {
+ _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;
+ }
Review Comment:
For SSL-enabled listeners, this returns without closing `fd`, which contradicts the PR contract (“close the accepted fd immediately”) and will cause the connection to remain open until the peer closes (and leak the server fd in the meantime). Even in the plaintext branch, the function sends a response but never closes the connection afterward. Ensure `fd` is always closed (for SSL: close immediately; for plaintext: send best-effort then close).
##########
src/brpc/server.cpp:
##########
@@ -870,6 +893,18 @@ 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=redis, "
+ "no RPC or builtin services)";
Review Comment:
The validation in `is_redis_only_public_listener(...)` checks additional constraints (e.g., `enabled_protocols` must be exactly `"redis"` and several other `*_service` pointers must be `nullptr`), but the error message doesn’t reflect the full set of reasons this can fail. Consider expanding the message (or pointing to the docs/option name) so users can resolve configuration errors without reading code.
##########
src/brpc/acceptor.cpp:
##########
@@ -275,7 +341,12 @@ void Acceptor::OnNewConnectionsUntilEAGAIN(Socket* acception) {
acception->SetFailed(EINVAL, "Impossible! acception->user() MUST be Acceptor");
return;
}
-
+
+ if (!am->TryAcquireRedisConnectionSlot()) {
+ am->RejectRedisConnection(in_fd);
+ continue;
+ }
Review Comment:
`in_fd` is not closed on the rejection path. Since no `Socket` is created, nothing else will own/close this fd, which can leak file descriptors and leave rejected clients connected indefinitely (also defeating the purpose of “reject excess clients early”). Close the fd before `continue` (or make `RejectRedisConnection` always close it).
##########
test/brpc_server_unittest.cpp:
##########
@@ -1424,6 +1425,159 @@ 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_GT(first_client, 0);
Review Comment:
`tcp_connect` appears to return `-1` on failure; a valid fd can be `0` in some test environments (e.g., if stdin was closed). Using `ASSERT_GT(fd, 0)` can spuriously fail; prefer `ASSERT_GE(fd, 0)` (and similarly for the other `ASSERT_GT(..., 0)` added in these tests).
--
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: dev-unsubscribe@brpc.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@brpc.apache.org
For additional commands, e-mail: dev-help@brpc.apache.org
Re: [PR] feat: limit Redis server connections (brpc)
Posted by "wwbmmm (via GitHub)" <gi...@apache.org>.
wwbmmm commented on PR #3498:
URL: https://github.com/apache/brpc/pull/3498#issuecomment-5643470856
Sorry for the copilot spam above. This was caused by my automatic agent.
I think this feature can be more general, for example "limit server connections" , not specific to the redis protocol.
As your document says:
> Set `ServerOptions.redis_max_connections` to limit simultaneous connections on a Redis-only public listener. The default value is 0, which disables the limit. A non-zero value requires `redis_service` to be set, `enabled_protocols` to be exactly `"redis"`, builtin services to be disabled, and no RPC or other protocol services to share the Server.
Since you limit the `enabled_protocol` to "redis", the redis connection count is equal to the whole server connection count, right? So we just need to limit the server connection count.
--
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: dev-unsubscribe@brpc.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@brpc.apache.org
For additional commands, e-mail: dev-help@brpc.apache.org