This is an automated email from the ASF dual-hosted git repository.
chenBright pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/brpc.git
The following commit(s) were added to refs/heads/master by this push:
new 5cc723fa feat: limit server connections per public listener (#3498)
5cc723fa is described below
commit 5cc723faed9112cab7cf94809c53f59886dfe3d5
Author: Jerry Zhao <[email protected]>
AuthorDate: Mon Sep 14 10:34:22 2026 +0800
feat: limit server connections per public listener (#3498)
* feat: limit Redis server connections
* feat: update Redis connection limit at runtime
* fix: handle partial Redis rejection writes
* Fix Redis connection limit review nits
Accept fd 0 in the Redis connection tests and describe every
dedicated-listener requirement in startup and runtime errors.
Validation: brpc_server_unittest rebuilt; dedicated-listener,
plaintext/dynamic-limit, pre-TLS rejection, and idle-connection tests passed
(4/4). Diff whitespace check passed. Full suite not run.
* Clarify Redis rejection ownership and response delivery
Document that rejection borrows the fd from the accept loop guard, which
closes it on continue. Describe nonblocking best-effort error delivery in both
server guides. Accumulate short TCP reads in rejection tests and verify
plaintext EOF.
Validation: rebuilt brpc_server_unittest; dedicated-listener,
plaintext/dynamic-limit, pre-TLS rejection and idle-connection tests passed
(4/4). Merge-base diff whitespace check passed. Full suite not run. Production
behavior is unchanged.
* Improve Redis rejection portability and regression coverage
Guard MSG_NOSIGNAL and use SO_NOSIGPIPE when available; omit the optional
error if per-socket signal suppression is unavailable. Share listener
validation diagnostics, use RAII based on actual Server ownership in the
invalid-configuration test, and verify new admission after slot recovery.
Validation: cmake --build build --target brpc_server_unittest --parallel 4
succeeded. Dedicated-listener, plaintext/dynamic-limit/slot-recovery, pre-TLS
rejection, and idle-connection tests passed (4/4). Merge-base diff check
passed. Full suite and macOS fallback were not run.
* Generalize connection limits to public listeners
* Address connection limit review feedback
---
docs/cn/server.md | 10 +
docs/en/server.md | 10 +
src/brpc/acceptor.cpp | 94 +++++++--
src/brpc/acceptor.h | 24 +++
src/brpc/server.cpp | 19 +-
src/brpc/server.h | 21 ++
test/brpc_server_unittest.cpp | 442 ++++++++++++++++++++++++++++++++++++++----
7 files changed, 569 insertions(+), 51 deletions(-)
diff --git a/docs/cn/server.md b/docs/cn/server.md
index 565b7bb3..efb6e9f2 100644
--- a/docs/cn/server.md
+++ b/docs/cn/server.md
@@ -378,6 +378,16 @@ Server.set_version(...)可以为server设置一个名称+版本，可通过/vers
| ------------------------- | ----- | ----------------------------------------
| ------------------- |
| log_idle_connection_close | false | Print log when an idle connection is
closed | src/brpc/socket.cpp |
+## 限制服务端连接数
+
+设置`ServerOptions.max_connections`可以限制Server公共监听端口上的并发连接数。默认值为0，表示不限制。该端口上的所有协议和服务共享同一个上限，包括RPC、HTTP、Redis和内置服务，无需配置专用协议。
+
+Acceptor会在创建brpc
Socket前预留连接名额，因此空闲连接和尚未完成TLS握手的连接也计入上限。超限连接会在协议解析或TLS认证前被直接关闭，不发送协议专用的错误响应。Socket回收时释放连接名额。
+
+通过`ServerOptions.internal_port`配置的内部监听端口不受限制，也不占用公共端口的连接名额，因此公共端口满载时，内部端口上的内置服务仍可访问。同一个业务Service注册到多个Server、监听多个端口时，各公共监听端口分别计数和限流；即使共享同一个Service对象，也不会合并连接数。`ServerStatistics.connection_count`包含公共和内部端口的连接总数，因此可能超过`max_connections`；`ServerStatistics.rejected_connection_count`记录公共端口因超限而拒绝的累计连接数。
+
+运行中的Server可以调用`Server::SetMaxConnections()`原子更新上限，也可以在以无限制值启动后动态开启限制。调高上限会影响后续连接准入；调低上限不会断开已有连接，公共端口的活跃连接数降到新上限以下后才会重新接受新连接。设置为0会关闭限制。Server未运行时该方法返回-1；该方法不会修改`options().max_connections`记录的启动值。每次`Start()`都使用该次传入选项中的上限。
+
## pid_file
如果设置了此字段，Server启动时会创建一个同名文件，内容为进程号。默认为空。
diff --git a/docs/en/server.md b/docs/en/server.md
index f13193a8..30c7a867 100644
--- a/docs/en/server.md
+++ b/docs/en/server.md
@@ -375,6 +375,16 @@ If
[-log_idle_connection_close](http://brpc.baidu.com:8765/flags/log_idle_connec
| ------------------------- | ----- | ----------------------------------------
| ------------------- |
| log_idle_connection_close | false | Print log when an idle connection is
closed | src/brpc/socket.cpp |
+## Limit server connections
+
+Set `ServerOptions.max_connections` to limit simultaneous connections on the
Server's public listener. The default value is 0 (unlimited). All protocols and
services on that listener share the same limit, including RPC, HTTP, Redis, and
builtin services. No dedicated protocol configuration is required.
+
+The acceptor reserves a slot before creating a brpc Socket, so idle
connections and connections awaiting TLS handshakes count toward the limit.
Connections over the limit are closed immediately without a protocol-specific
response, before protocol parsing or TLS authentication. Connection slots are
released when sockets are recycled.
+
+The internal listener configured by `ServerOptions.internal_port` is unlimited
and does not consume public connection slots, so builtin services on that port
remain available when the public listener is full. A business service
registered with multiple Server instances on different ports has an independent
limit on each public listener, even when those Servers share the same service
object. `ServerStatistics.connection_count` includes both public and internal
connections and may therefor [...]
+
+Call `Server::SetMaxConnections()` to atomically update the limit on a running
Server, including one started without a limit. Raising the limit affects
subsequent admission checks. Lowering it does not close existing connections;
new connections are accepted again after the active public connection count
falls below the limit. Set the limit to 0 to disable it. The setter returns -1
if the Server is not running and leaves `options().max_connections` at its
startup value. Each `Start()` us [...]
+
## pid_file
If this field is non-empty, Server creates a file named so at start-up, with
pid as the content. Empty by default.
diff --git a/src/brpc/acceptor.cpp b/src/brpc/acceptor.cpp
index 8333a171..99280037 100644
--- a/src/brpc/acceptor.cpp
+++ b/src/brpc/acceptor.cpp
@@ -16,7 +16,9 @@
// under the License.
+#include <errno.h>
#include <inttypes.h>
+#include <sys/socket.h>
#include <gflags/gflags.h>
#include "butil/fd_guard.h" // fd_guard
#include "butil/fd_utility.h" // make_close_on_exec
@@ -38,6 +40,9 @@ Acceptor::Acceptor(bthread_keytable_pool_t* pool)
, _listened_fd(-1)
, _acception_id(0)
, _empty_cond(&_map_mutex)
+ , _connection_count(0)
+ , _rejected_connection_count(0)
+ , _max_connections(0)
, _force_ssl(false)
, _ssl_ctx(nullptr)
, _socket_mode(SOCKET_MODE_TCP)
@@ -52,6 +57,13 @@ 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);
+}
+
+int Acceptor::StartAccept(int listened_fd, int idle_timeout_sec,
+ const std::shared_ptr<SocketSSLContext>& ssl_ctx,
+ bool force_ssl,
+ size_t max_connections) {
if (listened_fd < 0) {
LOG(FATAL) << "Invalid listened_fd=" << listened_fd;
return -1;
@@ -87,6 +99,7 @@ int Acceptor::StartAccept(int listened_fd, int
idle_timeout_sec,
_idle_timeout_sec = idle_timeout_sec;
_force_ssl = force_ssl;
_ssl_ctx = ssl_ctx;
+ SetMaxConnections(max_connections);
// Creation of _acception_id is inside lock so that OnNewConnections
// (which may run immediately) should see sane fields set below.
@@ -200,9 +213,37 @@ 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::RejectedConnectionCount() const {
+ return _rejected_connection_count.load(butil::memory_order_relaxed);
+}
+
+bool Acceptor::TryAcquireConnectionSlot() {
+ size_t count = _connection_count.load(butil::memory_order_relaxed);
+ do {
+ const size_t max_connections =
+ _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::ReleaseConnectionSlot() {
+ const size_t previous =
+ _connection_count.fetch_sub(1, butil::memory_order_relaxed);
+ CHECK_GT(previous, 0u);
+}
+
+void Acceptor::SetMaxConnections(size_t max_connections) {
+ // The limit controls only future numeric admission decisions and does not
+ // publish socket state, so a relaxed store is sufficient.
+ _max_connections.store(
+ max_connections, butil::memory_order_relaxed);
}
void Acceptor::ListConnections(std::vector<SocketId>* conn_list,
@@ -275,7 +316,18 @@ 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);
+ LOG_EVERY_SECOND(WARNING)
+ << "Reject connection on " << acception->local_side()
+ << ": max_connections limit reached";
+ // in_fd closes the connection before Socket::Create(), protocol
+ // parsing or TLS authentication, without a protocol-specific
reply.
+ continue;
+ }
+
SocketId socket_id;
SocketOptions options;
options.keytable_pool = am->_keytable_pool;
@@ -288,21 +340,20 @@ void Acceptor::OnNewConnectionsUntilEAGAIN(Socket*
acception) {
options.socket_mode = am->_socket_mode;
options.bthread_tag = am->_bthread_tag;
if (Socket::Create(options, &socket_id) != 0) {
+ am->ReleaseConnectionSlot();
LOG(ERROR) << "Fail to create Socket";
continue;
}
in_fd.release(); // transfer ownership to socket_id
- // There's a funny race condition here. After Socket::Create, messages
- // from the socket are already handled and a RPC is possibly done
- // before the socket is added into _socket_map below. This is found in
- // ChannelTest.skip_parallel in test/brpc_channel_unittest.cpp (running
- // on machines with few cores) where the _messenger.ConnectionCount()
- // may surprisingly be 0 even if the RPC is already done.
+ // Socket::Create() may start processing messages immediately. The
+ // connection is already counted, but this loop owns its slot until
+ // the socket is pinned and inserted into _socket_map below.
SocketUniquePtr sock;
if (Socket::AddressFailedAsWell(socket_id, &sock) >= 0) {
bool is_running = true;
+ bool registered = false;
{
BAIDU_SCOPED_LOCK(am->_map_mutex);
is_running = (am->status() == RUNNING);
@@ -311,7 +362,13 @@ void Acceptor::OnNewConnectionsUntilEAGAIN(Socket*
acception) {
// running or not. Otherwise, `Acceptor::BeforeRecycle'
// may be called (inside Socket::BeforeRecycled) after
`Acceptor'
// has been destroyed
- am->_socket_map.insert(socket_id, ConnectStatistics());
+ registered = (am->_socket_map.insert(
+ socket_id, ConnectStatistics()) != nullptr);
+ }
+ if (!registered) {
+ am->ReleaseConnectionSlot();
+ sock->SetFailed(ENOMEM, "Fail to register accepted Socket");
+ continue;
}
if (!is_running) {
LOG(WARNING) << "Acceptor on fd=" << acception->fd()
@@ -321,8 +378,11 @@ void Acceptor::OnNewConnectionsUntilEAGAIN(Socket*
acception) {
sock->description().c_str());
return;
}
- } // else: The socket has already been destroyed, Don't add its id
- // into _socket_map
+ } else {
+ // The socket was recycled before insertion. BeforeRecycle() did
+ // not release its slot, which is still owned by this accept loop.
+ am->ReleaseConnectionSlot();
+ }
}
}
@@ -346,9 +406,11 @@ void Acceptor::BeforeRecycle(Socket* sock) {
_empty_cond.Broadcast();
return;
}
- // If a Socket could not be addressed shortly after its creation, it
- // was not added into `_socket_map'.
- _socket_map.erase(sock->id());
+ // Socket::Create() can fail or the socket can be recycled before it is
+ // inserted into the map. In those cases the accept loop releases the slot.
+ if (_socket_map.erase(sock->id()) != 0) {
+ ReleaseConnectionSlot();
+ }
if (_socket_map.empty()) {
_empty_cond.Broadcast();
}
diff --git a/src/brpc/acceptor.h b/src/brpc/acceptor.h
index 2f138b30..01a2b26a 100644
--- a/src/brpc/acceptor.h
+++ b/src/brpc/acceptor.h
@@ -19,6 +19,7 @@
#define BRPC_ACCEPTOR_H
#include "bthread/bthread.h" // bthread_t
+#include "butil/atomicops.h" // butil::atomic
#include "butil/synchronization/condition_variable.h"
#include "butil/containers/flat_map.h"
#include "brpc/input_messenger.h"
@@ -58,6 +59,12 @@ public:
int StartAccept(int listened_fd, int idle_timeout_sec,
const std::shared_ptr<SocketSSLContext>& ssl_ctx,
bool force_ssl);
+ // Limit simultaneous connections on this listener. 0 means unlimited.
+ // Excess connections are closed before protocol parsing or TLS.
+ int StartAccept(int listened_fd, int idle_timeout_sec,
+ const std::shared_ptr<SocketSSLContext>& ssl_ctx,
+ bool force_ssl,
+ size_t max_connections);
// [thread-safe] Stop accepting connections.
// `closewait_ms' is not used anymore.
@@ -72,6 +79,10 @@ public:
// Get number of existing connections.
size_t ConnectionCount() const;
+ // Get the cumulative number of connections rejected by this listener's
+ // connection limit.
+ size_t RejectedConnectionCount() const;
+
// Clear `conn_list' and append all connections into it.
void ListConnections(std::vector<SocketId>* conn_list);
@@ -93,6 +104,10 @@ private:
// Remove the accepted socket `sock' from inside
void BeforeRecycle(Socket* sock) override;
+ bool TryAcquireConnectionSlot();
+ void ReleaseConnectionSlot();
+ void SetMaxConnections(size_t max_connections);
+
bthread_keytable_pool_t* _keytable_pool; // owned by Server
Status _status;
int _idle_timeout_sec;
@@ -108,6 +123,15 @@ private:
// The map containing all the accepted sockets
SocketMap _socket_map;
+ // A slot is reserved before Socket::Create(), closing the race where a
+ // socket starts processing before it is inserted into _socket_map. Until
+ // insertion, the accept loop owns the slot; afterwards BeforeRecycle()
+ // releases it. These atomics publish no socket state, so relaxed memory
+ // ordering is sufficient.
+ butil::atomic<size_t> _connection_count;
+ butil::atomic<size_t> _rejected_connection_count;
+ butil::atomic<size_t> _max_connections;
+
bool _force_ssl;
std::shared_ptr<SocketSSLContext> _ssl_ctx;
diff --git a/src/brpc/server.cpp b/src/brpc/server.cpp
index baf8b8fa..5a29ae23 100644
--- a/src/brpc/server.cpp
+++ b/src/brpc/server.cpp
@@ -138,6 +138,7 @@ ServerOptions::ServerOptions()
, server_owns_interceptor(false)
, num_threads(8)
, max_concurrency(0)
+ , max_connections(0)
, session_local_data_factory(nullptr)
, reserved_session_local_data(0)
, thread_local_data_factory(nullptr)
@@ -1167,7 +1168,8 @@ int Server::StartInternal(const butil::EndPoint& endpoint,
// Pass ownership of `sockfd' to `_am'
if (_am->StartAccept(sockfd, _options.idle_timeout_sec,
_default_ssl_ctx,
- _options.force_ssl) != 0) {
+ _options.force_ssl,
+ _options.max_connections) != 0) {
LOG(ERROR) << "Fail to start acceptor";
return -1;
}
@@ -1209,7 +1211,8 @@ int Server::StartInternal(const butil::EndPoint& endpoint,
// Pass ownership of `sockfd' to `_internal_am'
if (_internal_am->StartAccept(sockfd, _options.idle_timeout_sec,
_default_ssl_ctx,
- false) != 0) {
+ false,
+ 0) != 0) {
LOG(ERROR) << "Fail to start internal_acceptor";
return -1;
}
@@ -1791,8 +1794,11 @@ google::protobuf::Service* Server::FindServiceByName(
void Server::GetStat(ServerStatistics* stat) const {
stat->connection_count = 0;
+ stat->rejected_connection_count = 0;
if (_am) {
stat->connection_count += _am->ConnectionCount();
+ stat->rejected_connection_count +=
+ _am->RejectedConnectionCount();
}
if (_internal_am) {
stat->connection_count += _internal_am->ConnectionCount();
@@ -1801,6 +1807,15 @@ void Server::GetStat(ServerStatistics* stat) const {
stat->builtin_service_count = builtin_service_count();
}
+int Server::SetMaxConnections(size_t max_connections) {
+ if (!IsRunning() || _am == nullptr) {
+ LOG(WARNING) << "SetMaxConnections requires a running Server";
+ return -1;
+ }
+ _am->SetMaxConnections(max_connections);
+ return 0;
+}
+
void Server::ListServices(std::vector<google::protobuf::Service*> *services) {
if (!services) {
return;
diff --git a/src/brpc/server.h b/src/brpc/server.h
index 6e7d2b2b..ccf125f9 100644
--- a/src/brpc/server.h
+++ b/src/brpc/server.h
@@ -132,6 +132,17 @@ struct ServerOptions {
// Default: 0 (unlimited)
int max_concurrency;
+ // Maximum number of simultaneous connections on the public listener,
+ // shared by all protocols and services, including builtin services.
+ // Idle connections and connections awaiting TLS handshakes count too.
+ // Excess connections are closed before protocol parsing or TLS, without
+ // a response. The internal listener has a separate, unlimited count.
+ // Other Server instances have independent limits, even if they share
+ // the same service object on different ports.
+ // Use Server::SetMaxConnections() to update the limit at runtime.
+ // Default: 0 (unlimited)
+ size_t max_connections;
+
// Default value of method-level max concurrencies,
// Overridable by Server.MaxConcurrencyOf().
AdaptiveMaxConcurrency method_max_concurrency;
@@ -309,7 +320,10 @@ private:
// This struct is originally designed to contain basic statistics of the
// server. But bvar contains more stats and is more convenient.
struct ServerStatistics {
+ // Total connections on the public and internal listeners.
size_t connection_count;
+ // Cumulative connections rejected by the public listener's limit.
+ size_t rejected_connection_count;
int user_service_count;
int builtin_service_count;
};
@@ -546,6 +560,13 @@ public:
// Get statistics of this server
void GetStat(ServerStatistics* stat) const;
+ // Atomically update the connection limit of the running public listener.
+ // Existing connections are not closed when the limit is lowered.
+ // Set to 0 to disable the limit. Does not change
options().max_connections,
+ // which records the startup setting.
+ // Returns 0 on success, -1 if this Server is not running.
+ int SetMaxConnections(size_t max_connections);
+
// Get the options passed to Start().
const ServerOptions& options() const { return _options; }
diff --git a/test/brpc_server_unittest.cpp b/test/brpc_server_unittest.cpp
index e94d8508..898e3e79 100644
--- a/test/brpc_server_unittest.cpp
+++ b/test/brpc_server_unittest.cpp
@@ -19,9 +19,12 @@
// Date: Sun Jul 13 15:04:18 CST 2014
+#include <errno.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <fstream>
+#include <limits>
+#include <memory>
#include <gtest/gtest.h>
#include <google/protobuf/descriptor.h>
#include "butil/time.h"
@@ -49,8 +52,10 @@
#include "brpc/builtin/bad_method_service.h"
#include "brpc/server.h"
#include "brpc/nshead_service.h"
+#include "brpc/acceptor.h"
#include "brpc/restful.h"
#include "brpc/channel.h"
+#include "brpc/redis.h"
#include "brpc/socket_map.h"
#include "brpc/controller.h"
#include "brpc/compress.h"
@@ -1426,6 +1431,410 @@ TEST_F(ServerTest, close_idle_connections) {
ASSERT_EQ(0ul, stat.connection_count);
}
+// Returns a port nothing is listening on, or -1. `ServerOptions.internal_port`
+// has to be an explicit number, Server::Start() rejects 0 because it stands
+// for an ephemeral port, so ask the system for a free one rather than hardcode
+// a port that another test may be listening on.
+int PickUnusedPort() {
+ butil::fd_guard sockfd(butil::tcp_listen(butil::EndPoint(butil::IP_ANY,
0)));
+ if (sockfd < 0) {
+ return -1;
+ }
+ butil::EndPoint point;
+ if (butil::get_local_side(sockfd, &point) != 0) {
+ return -1;
+ }
+ return point.port;
+}
+
+// Starts `server` on an ephemeral port and fills `options->internal_port` with
+// another one. Both are released before Start() binds them and something else
+// may take one in between, hence the retries. Returns 0 on success.
+int StartWithInternalPort(brpc::Server* server, brpc::ServerOptions* options) {
+ for (int i = 0; i < 10; ++i) {
+ int internal_port = PickUnusedPort();
+ if (internal_port < 0) {
+ continue;
+ }
+ options->internal_port = internal_port;
+ if (0 == server->Start("127.0.0.1:0", options)) {
+ return 0;
+ }
+ }
+ return -1;
+}
+
+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;
+ ASSERT_EQ(0, StartWithInternalPort(&server, &opt));
+ butil::fd_guard public_client(
+ tcp_connect(server.listen_address(), nullptr));
+ ASSERT_GE(public_client, 0);
+ ASSERT_TRUE(WaitForServerConnections(server, 1));
+
+ butil::EndPoint internal_ep = server.listen_address();
+ internal_ep.port = opt.internal_port;
+ butil::fd_guard first_internal(tcp_connect(internal_ep, nullptr));
+ butil::fd_guard second_internal(tcp_connect(internal_ep, nullptr));
+ ASSERT_GE(first_internal, 0);
+ ASSERT_GE(second_internal, 0);
+ // GetStat includes both listeners; only the public count is limited.
+ ASSERT_TRUE(WaitForServerConnections(server, 3));
+ brpc::ChannelOptions copt;
+ copt.protocol = "http";
+ copt.connection_type = "short";
+ copt.timeout_ms = 1000;
+ copt.max_retry = 0;
+ brpc::Channel channel;
+ ASSERT_EQ(0, channel.Init(internal_ep, &copt));
+ brpc::Controller cntl;
+ cntl.http_request().uri() = "/status";
+ channel.CallMethod(nullptr, &cntl, nullptr, nullptr, nullptr);
+ ASSERT_FALSE(cntl.Failed()) << 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);
+
+ // Internal connections must not prevent public connection slot recovery.
+ public_client.reset(-1);
+ ASSERT_TRUE(WaitForServerConnections(server, 2));
+ public_client.reset(tcp_connect(server.listen_address(), nullptr));
+ ASSERT_GE(public_client, 0);
+ ASSERT_TRUE(WaitForServerConnections(server, 3));
+}
+
+TEST_F(ServerTest, connection_limit_is_per_listener) {
+ // One business service is shared by two Servers listening on different
+ // ports. Each port must have its own limit, count and runtime updates.
+ EchoServiceImpl shared_service;
+ brpc::Server servers[2];
+ for (size_t i = 0; i < 2; ++i) {
+ ASSERT_EQ(0, servers[i].AddService(
+ &shared_service, brpc::SERVER_DOESNT_OWN_SERVICE));
+ brpc::ServerOptions opt;
+ opt.max_connections = i + 1;
+ ASSERT_EQ(0, servers[i].Start("127.0.0.1:0", &opt));
+ }
+ butil::fd_guard first_client(
+ tcp_connect(servers[0].listen_address(), nullptr));
+ ASSERT_GE(first_client, 0);
+ ASSERT_TRUE(WaitForServerConnections(servers[0], 1));
+ butil::fd_guard second_port_clients[2];
+ for (auto& client : second_port_clients) {
+ client.reset(tcp_connect(servers[1].listen_address(), nullptr));
+ ASSERT_GE(client, 0);
+ }
+ ASSERT_TRUE(WaitForServerConnections(servers[1], 2));
+
+ for (auto& server : servers) {
+ 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(1ul, stat.rejected_connection_count);
+ }
+
+ ASSERT_EQ(0, servers[0].SetMaxConnections(2));
+ butil::fd_guard extra_client(
+ tcp_connect(servers[0].listen_address(), nullptr));
+ ASSERT_GE(extra_client, 0);
+ ASSERT_TRUE(WaitForServerConnections(servers[0], 2));
+ butil::fd_guard still_rejected(
+ tcp_connect(servers[1].listen_address(), nullptr));
+ ASSERT_GE(still_rejected, 0);
+ ExpectConnectionClosed(still_rejected);
+ brpc::ServerStatistics stat;
+ servers[1].GetStat(&stat);
+ EXPECT_EQ(2ul, stat.connection_count);
+ EXPECT_EQ(2ul, stat.rejected_connection_count);
+}
+
+TEST_F(ServerTest, connection_limit_restarts_with_startup_setting) {
+ brpc::Server server;
+ brpc::ServerOptions opt;
+ opt.max_connections = 1;
+ ASSERT_EQ(0, server.Start("127.0.0.1:0", &opt));
+ ASSERT_EQ(0, server.SetMaxConnections(0));
+ EXPECT_EQ(1ul, server.options().max_connections);
+ butil::fd_guard clients[2];
+ for (auto& client : clients) {
+ client.reset(tcp_connect(server.listen_address(), nullptr));
+ ASSERT_GE(client, 0);
+ }
+ ASSERT_TRUE(WaitForServerConnections(server, 2));
+ ASSERT_EQ(0, server.Stop(0));
+ ASSERT_EQ(0, server.Join());
+ ASSERT_TRUE(WaitForServerConnections(server, 0));
+
+ ASSERT_EQ(0, server.Start("127.0.0.1:0", &opt));
+ clients[0].reset(tcp_connect(server.listen_address(), nullptr));
+ ASSERT_GE(clients[0], 0);
+ ASSERT_TRUE(WaitForServerConnections(server, 1));
+ clients[1].reset(tcp_connect(server.listen_address(), nullptr));
+ ASSERT_GE(clients[1], 0);
+ ExpectConnectionClosed(clients[1]);
+}
+
+TEST_F(ServerTest, connection_limit_rejects_before_tls_handshake) {
+ brpc::Server server;
+ brpc::ServerOptions opt;
+ opt.max_connections = 1;
+ opt.force_ssl = true;
+ brpc::CertInfo& cert = opt.mutable_ssl_options()->default_cert;
+ cert.certificate = "cert1.crt";
+ cert.private_key = "cert1.key";
+ ASSERT_EQ(0, server.Start("127.0.0.1:0", &opt));
+
+ const butil::EndPoint ep = server.listen_address();
+ // An idle TCP socket consumes a slot without initiating a TLS handshake.
+ butil::fd_guard first_client(tcp_connect(ep, nullptr));
+ ASSERT_GE(first_client, 0);
+ ASSERT_TRUE(WaitForServerConnections(server, 1));
+ butil::fd_guard rejected_client(tcp_connect(ep, nullptr));
+ ASSERT_GE(rejected_client, 0);
+ // EOF without a ClientHello verifies rejection before TLS authentication.
+ ExpectConnectionClosed(rejected_client);
+ brpc::ServerStatistics stat;
+ server.GetStat(&stat);
+ EXPECT_EQ(1ul, stat.connection_count);
+ EXPECT_EQ(1ul, stat.rejected_connection_count);
+}
+
+TEST_F(ServerTest, connection_limit_socket_creation_failure) {
+ brpc::Server server;
+ brpc::ServerOptions opt;
+ opt.max_connections = 1;
+ ASSERT_EQ(0, server.Start("127.0.0.1:0", &opt));
+ brpc::Acceptor* acceptor = server._am;
+ ASSERT_TRUE(acceptor->TryAcquireConnectionSlot());
+
+ brpc::SocketOptions socket_opt;
+ socket_opt.user = acceptor;
+ // Avoid STREAM_FAKE_FD, which is intentionally accepted without an OS fd.
+ socket_opt.fd = std::numeric_limits<int>::max() - 1;
+ brpc::SocketId id;
+ ASSERT_NE(0, brpc::Socket::Create(socket_opt, &id));
+ // Socket::Create can call BeforeRecycle on failure. Since the socket was
+ // never inserted into the map, the accept loop must still own the slot.
+ ASSERT_EQ(1ul, acceptor->ConnectionCount());
+ acceptor->ReleaseConnectionSlot();
+ butil::fd_guard client(tcp_connect(server.listen_address(), nullptr));
+ ASSERT_GE(client, 0);
+ ASSERT_TRUE(WaitForServerConnections(server, 1));
+}
+
+TEST_F(ServerTest, connection_limit_socket_recycled_before_registration) {
+ brpc::Server server;
+ brpc::ServerOptions opt;
+ opt.max_connections = 1;
+ ASSERT_EQ(0, server.Start("127.0.0.1:0", &opt));
+ brpc::Acceptor* acceptor = server._am;
+ ASSERT_TRUE(acceptor->TryAcquireConnectionSlot());
+
+ brpc::SocketOptions socket_opt;
+ socket_opt.user = acceptor;
+ brpc::SocketId id;
+ ASSERT_EQ(0, brpc::Socket::Create(socket_opt, &id));
+ // Exercise a socket recycled after successful creation, before the accept
+ // loop can address it and insert it into the connection map.
+ ASSERT_EQ(0, brpc::Socket::SetFailed(id));
+ brpc::SocketUniquePtr socket;
+ ASSERT_EQ(-1, brpc::Socket::AddressFailedAsWell(id, &socket));
+ ASSERT_EQ(1ul, acceptor->ConnectionCount());
+ acceptor->ReleaseConnectionSlot();
+
+ butil::fd_guard client(tcp_connect(server.listen_address(), nullptr));
+ ASSERT_GE(client, 0);
+ ASSERT_TRUE(WaitForServerConnections(server, 1));
+}
+
TEST_F(ServerTest, logoff_and_multiple_start) {
butil::Timer timer;
butil::EndPoint ep;
@@ -1737,39 +2146,6 @@ void CallEchoByHttp(const butil::EndPoint& ep,
brpc::Controller* cntl) {
chan.CallMethod(nullptr, cntl, &req, &res, nullptr);
}
-// Returns a port nothing is listening on, or -1. `ServerOptions.internal_port`
-// has to be an explicit number, Server::Start() rejects 0 because it stands
-// for an ephemeral port, so ask the system for a free one rather than hardcode
-// a port that another test may be listening on.
-int PickUnusedPort() {
- butil::fd_guard sockfd(butil::tcp_listen(butil::EndPoint(butil::IP_ANY,
0)));
- if (sockfd < 0) {
- return -1;
- }
- butil::EndPoint point;
- if (butil::get_local_side(sockfd, &point) != 0) {
- return -1;
- }
- return point.port;
-}
-
-// Starts `server` on an ephemeral port and fills `options->internal_port` with
-// another one. Both are released before Start() binds them and something else
-// may take one in between, hence the retries. Returns 0 on success.
-int StartWithInternalPort(brpc::Server* server, brpc::ServerOptions* options) {
- for (int i = 0; i < 10; ++i) {
- int internal_port = PickUnusedPort();
- if (internal_port < 0) {
- continue;
- }
- options->internal_port = internal_port;
- if (0 == server->Start("127.0.0.1:0", options)) {
- return 0;
- }
- }
- return -1;
-}
-
TEST_F(ServerTest, ordinary_services_are_not_served_on_internal_port) {
const struct {
brpc::ProtocolType protocol;
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
