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