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: [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]
