From 08ed3e2b713851db7db5692e05416467a9a31618 Mon Sep 17 00:00:00 2001 From: Spicy-cream <299359524+Spicy-cream@users.noreply.github.com> Date: Tue, 27 Oct 2026 03:10:44 -0700 Subject: [PATCH 1/2] Add per-channel RDMA SL and Traffic Class configuration --- docs/cn/rdma.md | 34 +++++++ docs/en/rdma.md | 18 ++++ example/rdma_performance/client.cpp | 12 +++ example/rdma_performance/server.cpp | 4 + src/brpc/acceptor.cpp | 1 + src/brpc/acceptor.h | 2 + src/brpc/channel.cpp | 11 ++- src/brpc/channel.h | 4 + src/brpc/rdma/rdma_endpoint.cpp | 6 +- src/brpc/rdma/rdma_qos.h | 36 +++++++ src/brpc/server.cpp | 6 ++ src/brpc/server.h | 4 + src/brpc/socket.cpp | 7 ++ src/brpc/socket.h | 4 + test/brpc_rdma_unittest.cpp | 143 ++++++++++++++++++++++++++++ 15 files changed, 289 insertions(+), 3 deletions(-) create mode 100644 src/brpc/rdma/rdma_qos.h diff --git a/docs/cn/rdma.md b/docs/cn/rdma.md index fd70686bb1..f000b52fb9 100644 --- a/docs/cn/rdma.md +++ b/docs/cn/rdma.md @@ -1,3 +1,37 @@ +# 业务级 QoS + +通过 `ChannelOptions.rdma_qos` 配置请求方向的 RDMA QoS： + +```cpp +brpc::ChannelOptions options; +options.socket_mode = brpc::SOCKET_MODE_RDMA; +options.rdma_qos.service_level = 3; +options.rdma_qos.traffic_class = (26 << 2) | 2; // DSCP=26, ECN=2 +``` + +`service_level` 范围为 0–15，`traffic_class` 为完整的 8 位 GRH/IP Traffic Class，范围为 0–255， +两者默认均为 0。SL 的调度待遇由网络侧 SL→VL 或优先级映射决定。 +RoCE 部署结合网卡及交换机的 PCP/DSCP 映射配置；示例中的 DSCP/ECN 值按部署策略选择。 + +同一地址下，相同 QoS 的 Channel 可共享连接，不同 SL 或 Traffic Class 使用独立连接及 QP。 +配置随单连接、连接池和短连接传递，并在 QP 重建时重新应用。 + +响应方向通过 `ServerOptions.rdma_qos` 配置： + +```cpp +brpc::ServerOptions options; +options.socket_mode = brpc::SOCKET_MODE_RDMA; +options.rdma_qos.service_level = 3; +options.rdma_qos.traffic_class = 106; +``` + +客户端和服务端分别配置各自的发送方向。设置在 Channel 初始化或 Server 启动时生效， +V2/V3 握手沿用现有格式；TCP 传输按原配置运行。 +连接调试信息提供 `rdma_sl` 和 `rdma_traffic_class`。 + +`example/rdma_performance` 的 client/server 支持 `--rdma_sl`、`--rdma_traffic_class`。 +实机验证时检查 QP 属性和实际报文标记，再比较竞争流量下的吞吐与 P99。 + # 编译 由于RDMA对驱动与硬件有要求，目前仅支持在Linux系统编译并运行RDMA功能。 diff --git a/docs/en/rdma.md b/docs/en/rdma.md index 98ac6981bc..cfac9f83f3 100644 --- a/docs/en/rdma.md +++ b/docs/en/rdma.md @@ -1,3 +1,21 @@ +# Business-level QoS + +Set `ChannelOptions.rdma_qos.service_level` (0–15) and +`ChannelOptions.rdma_qos.traffic_class` (0–255) for outbound requests. +Both default to 0. Traffic Class is the full GRH/IP byte, including DSCP and ECN. +Set `ServerOptions.rdma_qos` independently for outbound responses. + +Different SL or Traffic Class values use separate connections and QPs, while equal +settings can share connections. Settings follow pooled and short sockets and are +applied whenever the QP is brought up. V2/V3 handshake formats stay the same. +TCP transports retain their existing settings. + +The fabric defines how SL maps to VLs and scheduling priority. For RoCE, configure +the NIC and switch PCP/DSCP mappings to match the deployment. The performance +client/server accept `--rdma_sl` and `--rdma_traffic_class`; socket diagnostics show +both applied values. Validate QP attributes and packet markings on hardware before +comparing throughput and P99 under competing workloads. + # Build Since RDMA requires driver and hardware support, only the build on linux is verified. diff --git a/example/rdma_performance/client.cpp b/example/rdma_performance/client.cpp index 8c07eca236..365c61697d 100644 --- a/example/rdma_performance/client.cpp +++ b/example/rdma_performance/client.cpp @@ -42,6 +42,8 @@ DEFINE_string(connection_type, "single", "Connection type of the channel"); DEFINE_string(protocol, "baidu_std", "Protocol type."); DEFINE_string(servers, "0.0.0.0:8002+0.0.0.0:8002", "IP Address of servers"); DEFINE_bool(use_rdma, true, "Use RDMA or not"); +DEFINE_int32(rdma_sl, 0, "Outbound RDMA service level [0,15]"); +DEFINE_int32(rdma_traffic_class, 0, "Outbound RDMA traffic class byte [0,255]"); DEFINE_int32(rpc_timeout_ms, 2000, "RPC call timeout"); DEFINE_int32(test_seconds, 20, "Test running time"); DEFINE_int32(test_iterations, 0, "Test iterations"); @@ -103,6 +105,8 @@ class PerformanceTest { int Init() { brpc::ChannelOptions options; options.socket_mode = FLAGS_use_rdma? brpc::SOCKET_MODE_RDMA : brpc::SOCKET_MODE_TCP; + options.rdma_qos.service_level = FLAGS_rdma_sl; + options.rdma_qos.traffic_class = FLAGS_rdma_traffic_class; options.protocol = FLAGS_protocol; options.connection_type = FLAGS_connection_type; options.timeout_ms = FLAGS_rpc_timeout_ms; @@ -274,6 +278,14 @@ void Test(int thread_num, int attachment_size) { int main(int argc, char* argv[]) { GFLAGS_NAMESPACE::ParseCommandLineFlags(&argc, &argv, true); + brpc::RdmaQosOptions qos; + qos.service_level = FLAGS_rdma_sl; + qos.traffic_class = FLAGS_rdma_traffic_class; + if (FLAGS_use_rdma && !qos.IsValid()) { + LOG(ERROR) << "RDMA SL must be in [0,15] and traffic class in [0,255]"; + return 1; + } + // Initialize RDMA environment in advance. if (FLAGS_use_rdma) { brpc::rdma::GlobalRdmaInitializeOrDie(); diff --git a/example/rdma_performance/server.cpp b/example/rdma_performance/server.cpp index 2e93e1eec7..b444b7b02d 100644 --- a/example/rdma_performance/server.cpp +++ b/example/rdma_performance/server.cpp @@ -28,6 +28,8 @@ DEFINE_int32(port, 8002, "TCP Port of this server"); DEFINE_bool(use_rdma, true, "Use RDMA or not"); +DEFINE_int32(rdma_sl, 0, "Outbound RDMA service level [0,15]"); +DEFINE_int32(rdma_traffic_class, 0, "Outbound RDMA traffic class byte [0,255]"); butil::atomic g_last_time(0); @@ -77,6 +79,8 @@ int main(int argc, char* argv[]) { brpc::ServerOptions options; options.socket_mode = FLAGS_use_rdma? brpc::SOCKET_MODE_RDMA : brpc::SOCKET_MODE_TCP; + options.rdma_qos.service_level = FLAGS_rdma_sl; + options.rdma_qos.traffic_class = FLAGS_rdma_traffic_class; if (server.Start(FLAGS_port, &options) != 0) { LOG(ERROR) << "Fail to start EchoServer"; return -1; diff --git a/src/brpc/acceptor.cpp b/src/brpc/acceptor.cpp index 99280037d5..d19d98ee0a 100644 --- a/src/brpc/acceptor.cpp +++ b/src/brpc/acceptor.cpp @@ -338,6 +338,7 @@ void Acceptor::OnNewConnectionsUntilEAGAIN(Socket* acception) { options.force_ssl = am->_force_ssl; options.initial_ssl_ctx = am->_ssl_ctx; options.socket_mode = am->_socket_mode; + options.rdma_qos = am->_rdma_qos; options.bthread_tag = am->_bthread_tag; if (Socket::Create(options, &socket_id) != 0) { am->ReleaseConnectionSlot(); diff --git a/src/brpc/acceptor.h b/src/brpc/acceptor.h index 01a2b26a9d..2485d9f387 100644 --- a/src/brpc/acceptor.h +++ b/src/brpc/acceptor.h @@ -24,6 +24,7 @@ #include "butil/containers/flat_map.h" #include "brpc/input_messenger.h" #include "brpc/socket_mode.h" +#include "brpc/rdma/rdma_qos.h" namespace brpc { @@ -137,6 +138,7 @@ friend class Server; // Choose to use a certain socket: 0 TCP, 1 RDMA SocketMode _socket_mode; + RdmaQosOptions _rdma_qos; // Acceptor belongs to this tag bthread_tag_t _bthread_tag; diff --git a/src/brpc/channel.cpp b/src/brpc/channel.cpp index de0a1bde49..9a7811f8f3 100644 --- a/src/brpc/channel.cpp +++ b/src/brpc/channel.cpp @@ -78,7 +78,8 @@ ChannelSSLOptions* ChannelOptions::mutable_ssl_options() { } static ChannelSignature ComputeChannelSignature(const ChannelOptions& opt) { - if (opt.auth == nullptr && + if (opt.socket_mode == SOCKET_MODE_TCP && + opt.auth == nullptr && !opt.has_ssl_options() && opt.client_host.empty() && opt.device_name.empty() && @@ -139,6 +140,8 @@ static ChannelSignature ComputeChannelSignature(const ChannelOptions& opt) { } if (opt.socket_mode == SOCKET_MODE_RDMA) { buf.append("|rdma"); + buf.append("|sl=").append(std::to_string(opt.rdma_qos.service_level)); + buf.append("|tc=").append(std::to_string(opt.rdma_qos.traffic_class)); } else if (opt.socket_mode == SOCKET_MODE_URMA) { buf.append("|urma"); } @@ -187,6 +190,10 @@ int Channel::InitChannelOptions(const ChannelOptions* options) { if (options) { // Override default options if user provided one. _options = *options; } + if (_options.socket_mode == SOCKET_MODE_RDMA && !_options.rdma_qos.IsValid()) { + LOG(ERROR) << "RDMA SL must be in [0,15] and traffic class in [0,255]"; + return -1; + } const Protocol* protocol = FindProtocol(_options.protocol); if (nullptr == protocol || !protocol->support_client()) { LOG(ERROR) << "Channel does not support the protocol"; @@ -407,6 +414,7 @@ int Channel::InitSingle(const butil::EndPoint& server_addr_and_port, opt.local_side = client_endpoint; opt.initial_ssl_ctx = ssl_ctx; opt.socket_mode = _options.socket_mode; + opt.rdma_qos = _options.rdma_qos; opt.hc_option = _options.hc_option; opt.device_name = _options.device_name; if (SocketMapInsert(SocketMapKey(server_addr_and_port, sig), @@ -453,6 +461,7 @@ int Channel::Init(const char* ns_url, ns_opt.succeed_without_server = _options.succeed_without_server; ns_opt.log_succeed_without_server = _options.log_succeed_without_server; ns_opt.socket_option.socket_mode = _options.socket_mode; + ns_opt.socket_option.rdma_qos = _options.rdma_qos; ns_opt.channel_signature = ComputeChannelSignature(_options); ns_opt.socket_option.hc_option = _options.hc_option; ns_opt.socket_option.local_side = client_endpoint; diff --git a/src/brpc/channel.h b/src/brpc/channel.h index a778a1bd2a..b4bff6092b 100644 --- a/src/brpc/channel.h +++ b/src/brpc/channel.h @@ -38,6 +38,7 @@ #include "brpc/naming_service_filter.h" #include "brpc/health_check_option.h" #include "brpc/socket_mode.h" +#include "brpc/rdma/rdma_qos.h" namespace brpc { @@ -111,6 +112,9 @@ struct ChannelOptions { // Default: SOCKET_MODE_TCP SocketMode socket_mode; + // Outbound RDMA QoS for this Channel. Used when socket_mode is RDMA. + RdmaQosOptions rdma_qos; + // Turn on authentication for this channel if `auth' is not nullptr. // Note `auth' will not be deleted by channel and must remain valid when // the channel is being used. diff --git a/src/brpc/rdma/rdma_endpoint.cpp b/src/brpc/rdma/rdma_endpoint.cpp index 0dc41c9dd1..fe3b25e66f 100644 --- a/src/brpc/rdma/rdma_endpoint.cpp +++ b/src/brpc/rdma/rdma_endpoint.cpp @@ -1453,9 +1453,9 @@ int RdmaEndpoint::BringUpQp(const ParsedHello& remote, bool is_server) { attr.ah_attr.grh.flow_label = 0; attr.ah_attr.grh.sgid_index = GetRdmaGidIndex(); attr.ah_attr.grh.hop_limit = MAX_HOP_LIMIT; - attr.ah_attr.grh.traffic_class = 0; + attr.ah_attr.grh.traffic_class = _socket->rdma_qos().traffic_class; attr.ah_attr.dlid = remote.lid; - attr.ah_attr.sl = 0; + attr.ah_attr.sl = _socket->rdma_qos().service_level; attr.ah_attr.src_path_bits = 0; attr.ah_attr.static_rate = 0; attr.ah_attr.is_global = 1; @@ -1867,6 +1867,8 @@ std::string RdmaEndpoint::GetStateStr() const { void RdmaEndpoint::DebugInfo(std::ostream& os, butil::StringPiece connector) const { os << "rdma_state=ON" + << connector << "rdma_sl=" << _socket->rdma_qos().service_level + << connector << "rdma_traffic_class=" << _socket->rdma_qos().traffic_class << connector << "handshake_state=" << GetStateStr() << connector << "handshake_version=" << static_cast(_handshake_version) << connector << "rdma_sq_imm_window_size=" << _sq_imm_window_size diff --git a/src/brpc/rdma/rdma_qos.h b/src/brpc/rdma/rdma_qos.h new file mode 100644 index 0000000000..e75cdafa61 --- /dev/null +++ b/src/brpc/rdma/rdma_qos.h @@ -0,0 +1,36 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#ifndef BRPC_RDMA_QOS_H +#define BRPC_RDMA_QOS_H + +namespace brpc { + +// Outbound QP attributes. Their scheduling behavior is defined by the fabric. +struct RdmaQosOptions { + int service_level = 0; // SL, range [0, 15]. + int traffic_class = 0; // Full GRH/IP traffic class byte, range [0, 255]. + + bool IsValid() const { + return service_level >= 0 && service_level <= 15 && + traffic_class >= 0 && traffic_class <= 255; + } +}; + +} // namespace brpc + +#endif // BRPC_RDMA_QOS_H diff --git a/src/brpc/server.cpp b/src/brpc/server.cpp index 19d58ac422..96734ae8d0 100644 --- a/src/brpc/server.cpp +++ b/src/brpc/server.cpp @@ -875,6 +875,11 @@ int Server::StartInternal(const butil::EndPoint& endpoint, const ServerOptions default_opt; const ServerOptions& real_opt = opt ? *opt : default_opt; + if (real_opt.socket_mode == SOCKET_MODE_RDMA && !real_opt.rdma_qos.IsValid()) { + LOG(ERROR) << "RDMA SL must be in [0,15] and traffic class in [0,255]"; + return -1; + } + if (!real_opt.h2_settings.IsValid(true/*log_error*/)) { LOG(ERROR) << "Invalid h2_settings"; return -1; @@ -1162,6 +1167,7 @@ int Server::StartInternal(const butil::EndPoint& endpoint, _am->_socket_mode = _options.socket_mode; _am->_bthread_tag = _options.bthread_tag; } + _am->_rdma_qos = _options.rdma_qos; // Set `_status' to RUNNING before accepting connections // to prevent requests being rejected as ELOGOFF _status = RUNNING; diff --git a/src/brpc/server.h b/src/brpc/server.h index 31dc69af2f..769cee4150 100644 --- a/src/brpc/server.h +++ b/src/brpc/server.h @@ -46,6 +46,7 @@ #include "brpc/baidu_master_service.h" #include "brpc/rpc_pb_message_factory.h" #include "brpc/socket_mode.h" +#include "brpc/rdma/rdma_qos.h" namespace brpc { @@ -242,6 +243,9 @@ struct ServerOptions { // Default: SOCKET_MODE_TCP SocketMode socket_mode; + // Outbound RDMA QoS for responses on this server's accepted connections. + RdmaQosOptions rdma_qos; + // [CAUTION] This option is for implementing specialized baidu-std proxies, // most users don't need it. Don't change this option unless you fully // understand the description below. diff --git a/src/brpc/socket.cpp b/src/brpc/socket.cpp index 57cfd2296a..f560320b4b 100644 --- a/src/brpc/socket.cpp +++ b/src/brpc/socket.cpp @@ -716,6 +716,10 @@ void SetSocketBufferOptions(int fd) { // version: from version part of _versioned_nref, must be an EVEN number. // slot: designated by ResourcePool. int Socket::Create(const SocketOptions& options, SocketId* id) { + if (options.socket_mode == SOCKET_MODE_RDMA && !options.rdma_qos.IsValid()) { + errno = EINVAL; + return -1; + } return VersionedRefWithId::Create(id, options); } @@ -731,6 +735,7 @@ int Socket::OnCreated(const SocketOptions& options) { }); // start build the transport _socket_mode = options.socket_mode; + _rdma_qos = options.rdma_qos; _transport = TransportFactory::CreateTransport(options.socket_mode); CHECK(nullptr != _transport); _transport->Init(this, options); @@ -2830,6 +2835,7 @@ int Socket::GetPooledSocket(SocketUniquePtr* pooled_socket) { opt.keytable_pool = _keytable_pool; opt.app_connect = _app_connect; opt.socket_mode = _socket_mode; + opt.rdma_qos = _rdma_qos; socket_pool = new SocketPool(opt); SocketPool* expected = nullptr; if (!main_sp->socket_pool.compare_exchange_strong( @@ -2936,6 +2942,7 @@ int Socket::GetShortSocket(SocketUniquePtr* short_socket) { opt.keytable_pool = _keytable_pool; opt.app_connect = _app_connect; opt.socket_mode = _socket_mode; + opt.rdma_qos = _rdma_qos; if (get_client_side_messenger()->Create(opt, &id) != 0 || Address(id, short_socket) != 0) { return -1; diff --git a/src/brpc/socket.h b/src/brpc/socket.h index da701e9ad7..9cd26674c4 100644 --- a/src/brpc/socket.h +++ b/src/brpc/socket.h @@ -43,6 +43,7 @@ #include "brpc/versioned_ref_with_id.h" #include "brpc/health_check_option.h" #include "brpc/socket_mode.h" +#include "brpc/rdma/rdma_qos.h" #include "brpc/input_messenger_processor.h" // InputMessengerProcessor namespace brpc { @@ -309,6 +310,7 @@ struct SocketOptions { bool force_ssl{false}; std::shared_ptr initial_ssl_ctx; SocketMode socket_mode{SOCKET_MODE_TCP}; + RdmaQosOptions rdma_qos; bthread_keytable_pool_t* keytable_pool{nullptr}; SocketConnection* conn{nullptr}; std::shared_ptr app_connect; @@ -632,6 +634,7 @@ friend class TransportFactory; // The socket mode decided at creation. SocketMode socket_mode() const { return _socket_mode; } + const RdmaQosOptions& rdma_qos() const { return _rdma_qos; } // Get an UNUSED socket connecting to the same place as this socket // from the SocketPool of this socket. @@ -999,6 +1002,7 @@ friend class TransportFactory; // Should use SOCKET_MODE_RDMA or SOCKET_MODE_TCP or Other, default is SOCKET_MODE_TCP Transport SocketMode _socket_mode; + RdmaQosOptions _rdma_qos; std::unique_ptr _transport; // Pass from controller, for progressive reading. diff --git a/test/brpc_rdma_unittest.cpp b/test/brpc_rdma_unittest.cpp index 6ec54ed90b..1d3ac2c934 100644 --- a/test/brpc_rdma_unittest.cpp +++ b/test/brpc_rdma_unittest.cpp @@ -25,6 +25,7 @@ #include #include #include +#include #include #include "butil/endpoint.h" #include "butil/fd_guard.h" @@ -47,6 +48,7 @@ #include "brpc/rdma/rdma_handshake_constants.h" #include "brpc/rdma/rdma_handshake.pb.h" #include "brpc/rdma/rdma_helper.h" +#include "brpc/details/load_balancer_with_naming.h" #include "echo.pb.h" static const int PORT = 8713; @@ -3316,6 +3318,147 @@ TEST(RdmaInlineDataTest, should_post_inline) { EXPECT_FALSE(rdma::ShouldPostInline(true, 1, 0)); } +TEST_F(RdmaTest, qos_rejects_invalid_options_before_connecting) { + for (const auto& values : {std::pair(-1, 0), {16, 0}, {0, -1}, {0, 256}}) { + ChannelOptions options; + options.socket_mode = SOCKET_MODE_RDMA; + options.rdma_qos.service_level = values.first; + options.rdma_qos.traffic_class = values.second; + Channel channel; + EXPECT_EQ(-1, channel.Init(g_ep, &options)); + ServerOptions server_options; + server_options.socket_mode = SOCKET_MODE_RDMA; + server_options.rdma_qos = options.rdma_qos; + Server server; + EXPECT_EQ(-1, server.Start(PORT, &server_options)); + SocketOptions socket_options; + socket_options.socket_mode = SOCKET_MODE_RDMA; + socket_options.rdma_qos = options.rdma_qos; + SocketId sid; + EXPECT_EQ(-1, Socket::Create(socket_options, &sid)); + EXPECT_EQ(EINVAL, errno); + } +} + +TEST_F(RdmaTest, qos_isolates_channels_and_preserves_equal_config_sharing) { + ChannelOptions options; + options.socket_mode = SOCKET_MODE_RDMA; + Channel a, b, same, different_tc, default_rdma, tcp, tcp_with_qos; + ASSERT_EQ(0, default_rdma.Init(g_ep, &options)); + ASSERT_EQ(0, tcp.Init(g_ep, nullptr)); + EXPECT_NE(default_rdma._server_id, tcp._server_id); + options.rdma_qos.service_level = 3; + options.rdma_qos.traffic_class = 106; + ASSERT_EQ(0, a.Init(g_ep, &options)); + ASSERT_EQ(0, same.Init(g_ep, &options)); + EXPECT_EQ(a._server_id, same._server_id); + EXPECT_NE(a._server_id, default_rdma._server_id); + options.rdma_qos.service_level = 4; + ASSERT_EQ(0, b.Init(g_ep, &options)); + EXPECT_NE(a._server_id, b._server_id); + options.rdma_qos.service_level = 3; + options.rdma_qos.traffic_class = 104; + ASSERT_EQ(0, different_tc.Init(g_ep, &options)); + EXPECT_NE(a._server_id, different_tc._server_id); + options.socket_mode = SOCKET_MODE_TCP; + ASSERT_EQ(0, tcp_with_qos.Init(g_ep, &options)); + EXPECT_EQ(tcp._server_id, tcp_with_qos._server_id); + SocketUniquePtr socket; + ASSERT_EQ(0, Socket::Address(a._server_id, &socket)); + EXPECT_EQ(3, socket->rdma_qos().service_level); + EXPECT_EQ(106, socket->rdma_qos().traffic_class); +} + +TEST_F(RdmaTest, qos_follows_pooled_short_and_named_connections) { + ChannelOptions options; + options.socket_mode = SOCKET_MODE_RDMA; + options.rdma_qos.service_level = 7; + options.rdma_qos.traffic_class = 255; + Channel direct, named; + ASSERT_EQ(0, direct.Init(g_ep, &options)); + ASSERT_EQ(0, named.Init(_naming_url.c_str(), "rr", &options)); + SocketUniquePtr main_socket; + ASSERT_EQ(0, Socket::Address(direct._server_id, &main_socket)); + SocketUniquePtr pooled, short_socket, selected; + ASSERT_EQ(0, main_socket->GetPooledSocket(&pooled)); + ASSERT_EQ(0, main_socket->GetShortSocket(&short_socket)); + LoadBalancer::SelectIn input{}; + LoadBalancer::SelectOut output(&selected); + ASSERT_EQ(0, named._lb->SelectServer(input, &output)); + for (Socket* s : {pooled.get(), short_socket.get(), selected.get()}) { + EXPECT_EQ(7, s->rdma_qos().service_level); + EXPECT_EQ(255, s->rdma_qos().traffic_class); + } + EXPECT_EQ(main_socket->id(), selected->id()); + EXPECT_EQ(0, pooled->ReturnToPool()); + short_socket->SetFailed(); +} + +TEST_F(RdmaTest, qos_server_options_follow_accept_and_restart) { + for (int sl : {3, 7}) { + ServerOptions options; + options.socket_mode = SOCKET_MODE_RDMA; + options.rdma_qos.service_level = sl; + options.rdma_qos.traffic_class = 106; + ASSERT_EQ(0, _server.Start(PORT, &options)); + butil::fd_guard fd; + ASSERT_NO_FATAL_FAILURE(ConnectToServer(&fd)); + Socket* socket = WaitForServerSocket(); + ASSERT_TRUE(socket != nullptr); + EXPECT_EQ(sl, socket->rdma_qos().service_level); + EXPECT_EQ(106, socket->rdma_qos().traffic_class); + fd.reset(-1); + ASSERT_TRUE(WaitForServerSocketGone()); + StopServer(); + } +} + +static ibv_ah_attr g_qos_rtr_address; +static int g_qos_rtr_calls; +static int CaptureQosModify(ibv_qp*, ibv_qp_attr* attr, ibv_qp_attr_mask mask) { + if (mask & IBV_QP_AV) { + g_qos_rtr_address = attr->ah_attr; + ++g_qos_rtr_calls; + } + return 0; +} + +TEST_F(RdmaTest, qos_applies_to_rtr_on_each_qp_bringup) { + for (const auto& values : {std::pair(15, 255), {0, 0}}) { + SocketOptions options; + options.socket_mode = SOCKET_MODE_RDMA; + options.rdma_qos.service_level = values.first; + options.rdma_qos.traffic_class = values.second; + SocketId sid; + ASSERT_EQ(0, Socket::Create(options, &sid)); + SocketUniquePtr socket; + ASSERT_EQ(0, Socket::Address(sid, &socket)); + auto* ep = RdmaTransportOf(socket)->_rdma_ep; + ibv_qp qp{}; + rdma::RdmaResource resource; + resource.qp = &qp; + ep->_resource = &resource; + ep->_rq_size = 0; + rdma::ParsedHello remote{}; + const bool skip = rdma::g_skip_rdma_init; + const auto modify = rdma::IbvModifyQp; + rdma::g_skip_rdma_init = false; + rdma::IbvModifyQp = CaptureQosModify; + g_qos_rtr_calls = 0; + // Reapply attributes on both client and server bringup. + EXPECT_EQ(0, ep->BringUpQp(remote, false)); + EXPECT_EQ(0, ep->BringUpQp(remote, true)); + rdma::g_skip_rdma_init = skip; + rdma::IbvModifyQp = modify; + EXPECT_EQ(2, g_qos_rtr_calls); + EXPECT_EQ(values.first, g_qos_rtr_address.sl); + EXPECT_EQ(values.second, g_qos_rtr_address.grh.traffic_class); + ep->_resource = nullptr; + resource.qp = nullptr; + socket->SetFailed(); + } +} + #endif // if BRPC_WITH_RDMA int main(int argc, char* argv[]) { From b0414f57cbf156a5dfd91cb0140006aea2c521eb Mon Sep 17 00:00:00 2001 From: Spicy-cream <299359524+Spicy-cream@users.noreply.github.com> Date: Tue, 27 Oct 2026 20:10:32 -0700 Subject: [PATCH 2/2] Adapt business QoS configuration to URMA --- docs/cn/urma.md | 28 +++ docs/en/urma.md | 31 ++++ example/urma_performance/client.cpp | 2 + example/urma_performance/server.cpp | 2 + src/brpc/acceptor.cpp | 1 + src/brpc/acceptor.h | 2 + src/brpc/channel.cpp | 7 + src/brpc/channel.h | 4 + src/brpc/server.cpp | 6 + src/brpc/server.h | 4 + src/brpc/socket.cpp | 7 + src/brpc/socket.h | 4 + src/brpc/urma/mock_urma.cpp | 3 + src/brpc/urma/urma_endpoint.cpp | 15 +- src/brpc/urma/urma_endpoint.h | 1 + src/brpc/urma/urma_helper.cpp | 25 +++ src/brpc/urma/urma_helper.h | 6 + src/brpc/urma/urma_qos.h | 32 ++++ src/brpc/urma_transport.cpp | 10 ++ test/brpc_urma_qos_unittest.cpp | 256 ++++++++++++++++++++++++++++ 20 files changed, 443 insertions(+), 3 deletions(-) create mode 100644 src/brpc/urma/urma_qos.h create mode 100644 test/brpc_urma_qos_unittest.cpp diff --git a/docs/cn/urma.md b/docs/cn/urma.md index 62e10389a8..14105a02d1 100644 --- a/docs/cn/urma.md +++ b/docs/cn/urma.md @@ -193,3 +193,31 @@ UrmaTransport 推荐用于**大包和跨节点**高吞吐路径，而 UBRing `ContextInitOrDie` 阶段拒绝。 - 硬件数据通路需要受支持的 UMDK provider 和 `liburma`。 - 当前实现面向 Linux。 + +## 业务优先级 + +`ChannelOptions.urma_qos.priority` 控制请求发送方向， +`ServerOptions.urma_qos.priority` 控制响应发送方向： + +```cpp +brpc::ChannelOptions client_options; +client_options.socket_mode = brpc::SOCKET_MODE_URMA; +client_options.urma_qos.priority = 10; + +brpc::ServerOptions server_options; +server_options.socket_mode = brpc::SOCKET_MODE_URMA; +server_options.urma_qos.priority = 6; +``` + +默认值 -1 沿用设备自动选择，取设备能力表中首个 CTP priority；未报告时沿用兼容值 6。 +显式值范围为 0–15，并要求所选设备的 `priority_info[priority].tp_type` 包含 CTP。 +不满足范围或能力要求时，Channel 初始化和 Server 启动返回错误。 +priority 对应的 SL 和调度待遇由设备及网络配置决定。 + +不同 priority 使用独立连接和 Jetty，相同配置共享连接。配置继承到服务发现、连接池、 +短连接和服务端 accept 路径，重建时重新应用。默认预建池只用于匹配其 priority 的连接， +其他 priority 按需创建 Jetty。连接调试信息显示配置值和实际 Jetty priority。 +请求和响应分别配置各自的发送策略，v2/v3 握手格式沿用现有协议。 + +`example/urma_performance` 的 client/server 支持 `--urma_priority=10`。 +以不同 priority 启动竞争流量，对比吞吐和 P99，并结合设备 SL 映射验证调度效果。 diff --git a/docs/en/urma.md b/docs/en/urma.md index a5b80f45b9..caa5eb09b3 100644 --- a/docs/en/urma.md +++ b/docs/en/urma.md @@ -207,3 +207,34 @@ is tracked as a future enhancement. rejected at `ContextInitOrDie`. - A hardware data path requires a supported UMDK provider and `liburma`. - The current implementation targets Linux. + +## Business priority + +Set `ChannelOptions.urma_qos.priority` for outgoing requests and +`ServerOptions.urma_qos.priority` for outgoing responses: + +```cpp +brpc::ChannelOptions client_options; +client_options.socket_mode = brpc::SOCKET_MODE_URMA; +client_options.urma_qos.priority = 10; + +brpc::ServerOptions server_options; +server_options.socket_mode = brpc::SOCKET_MODE_URMA; +server_options.urma_qos.priority = 6; +``` + +The default -1 keeps automatic selection: the first CTP priority reported by the +device, with the existing compatibility value 6 when none is reported. Explicit +values are in [0,15] and require CTP in `priority_info[priority].tp_type` on the +selected device. Channel initialization and Server startup reject unsupported +values. Device and network configuration determine SL mapping and scheduling. + +Different priorities use separate connections and Jettys; equal configurations +share connections. Naming services, pooled and short connections, accepted sockets +and rebuilt Jettys inherit the configuration. The prepared pool serves matching +priorities; other priorities allocate Jettys on demand. Debug output shows both +the configured and effective priority. Each peer configures its outgoing direction; +v2/v3 handshake formats remain the same. + +The client and server in `example/urma_performance` accept `--urma_priority=10`. +Compare throughput and P99 under competing priorities using the device SL mapping. diff --git a/example/urma_performance/client.cpp b/example/urma_performance/client.cpp index 9449dbe960..a66043b2d7 100644 --- a/example/urma_performance/client.cpp +++ b/example/urma_performance/client.cpp @@ -44,6 +44,7 @@ DEFINE_int32(attachment_size, -1, "Attachment size is used (in Bytes)"); DEFINE_int32(rpc_timeout_ms, 5000, "Timeout for each RPC in milliseconds"); DEFINE_bool(echo_attachment, false, "Select whether attachment should be echo"); DEFINE_bool(use_urma, true, "Use URMA transport (true) or TCP (false)"); +DEFINE_int32(urma_priority, -1, "Outbound URMA priority: -1 automatic, or [0,15]"); bvar::LatencyRecorder g_latency("client"); bvar::Adder g_error_count("client_error_count"); @@ -88,6 +89,7 @@ static void* worker(void* arg) { int main(int argc, char* argv[]) { gflags::ParseCommandLineFlags(&argc, &argv, true); brpc::ChannelOptions options; + options.urma_qos.priority = FLAGS_urma_priority; options.socket_mode = FLAGS_use_urma ? brpc::SOCKET_MODE_URMA : brpc::SOCKET_MODE_TCP; options.connect_timeout_ms = FLAGS_rpc_timeout_ms; diff --git a/example/urma_performance/server.cpp b/example/urma_performance/server.cpp index baaff323b6..b874b6a424 100644 --- a/example/urma_performance/server.cpp +++ b/example/urma_performance/server.cpp @@ -30,6 +30,7 @@ DEFINE_int32(port, 8003, "TCP Port of this server"); DEFINE_bool(use_urma, true, "Use URMA transport (true) or TCP (false)"); +DEFINE_int32(urma_priority, -1, "Outbound URMA priority: -1 automatic, or [0,15]"); butil::atomic g_last_time(0); @@ -73,6 +74,7 @@ int main(int argc, char* argv[]) { } brpc::ServerOptions options; + options.urma_qos.priority = FLAGS_urma_priority; options.socket_mode = FLAGS_use_urma ? brpc::SOCKET_MODE_URMA : brpc::SOCKET_MODE_TCP; if (server.Start(FLAGS_port, &options) != 0) { diff --git a/src/brpc/acceptor.cpp b/src/brpc/acceptor.cpp index d19d98ee0a..9c5504f207 100644 --- a/src/brpc/acceptor.cpp +++ b/src/brpc/acceptor.cpp @@ -339,6 +339,7 @@ void Acceptor::OnNewConnectionsUntilEAGAIN(Socket* acception) { options.initial_ssl_ctx = am->_ssl_ctx; options.socket_mode = am->_socket_mode; options.rdma_qos = am->_rdma_qos; + options.urma_qos = am->_urma_qos; options.bthread_tag = am->_bthread_tag; if (Socket::Create(options, &socket_id) != 0) { am->ReleaseConnectionSlot(); diff --git a/src/brpc/acceptor.h b/src/brpc/acceptor.h index 2485d9f387..b474fdd5a0 100644 --- a/src/brpc/acceptor.h +++ b/src/brpc/acceptor.h @@ -25,6 +25,7 @@ #include "brpc/input_messenger.h" #include "brpc/socket_mode.h" #include "brpc/rdma/rdma_qos.h" +#include "brpc/urma/urma_qos.h" namespace brpc { @@ -139,6 +140,7 @@ friend class Server; // Choose to use a certain socket: 0 TCP, 1 RDMA SocketMode _socket_mode; RdmaQosOptions _rdma_qos; + UrmaQosOptions _urma_qos; // Acceptor belongs to this tag bthread_tag_t _bthread_tag; diff --git a/src/brpc/channel.cpp b/src/brpc/channel.cpp index 9a7811f8f3..ea87d58e7e 100644 --- a/src/brpc/channel.cpp +++ b/src/brpc/channel.cpp @@ -144,6 +144,7 @@ static ChannelSignature ComputeChannelSignature(const ChannelOptions& opt) { buf.append("|tc=").append(std::to_string(opt.rdma_qos.traffic_class)); } else if (opt.socket_mode == SOCKET_MODE_URMA) { buf.append("|urma"); + buf.append("|priority=").append(std::to_string(opt.urma_qos.priority)); } butil::MurmurHash3_x64_128_Update(&mm_ctx, buf.data(), buf.size()); buf.clear(); @@ -194,6 +195,10 @@ int Channel::InitChannelOptions(const ChannelOptions* options) { LOG(ERROR) << "RDMA SL must be in [0,15] and traffic class in [0,255]"; return -1; } + if (_options.socket_mode == SOCKET_MODE_URMA && !_options.urma_qos.IsValid()) { + LOG(ERROR) << "URMA priority must be -1 (automatic) or in [0,15]"; + return -1; + } const Protocol* protocol = FindProtocol(_options.protocol); if (nullptr == protocol || !protocol->support_client()) { LOG(ERROR) << "Channel does not support the protocol"; @@ -415,6 +420,7 @@ int Channel::InitSingle(const butil::EndPoint& server_addr_and_port, opt.initial_ssl_ctx = ssl_ctx; opt.socket_mode = _options.socket_mode; opt.rdma_qos = _options.rdma_qos; + opt.urma_qos = _options.urma_qos; opt.hc_option = _options.hc_option; opt.device_name = _options.device_name; if (SocketMapInsert(SocketMapKey(server_addr_and_port, sig), @@ -462,6 +468,7 @@ int Channel::Init(const char* ns_url, ns_opt.log_succeed_without_server = _options.log_succeed_without_server; ns_opt.socket_option.socket_mode = _options.socket_mode; ns_opt.socket_option.rdma_qos = _options.rdma_qos; + ns_opt.socket_option.urma_qos = _options.urma_qos; ns_opt.channel_signature = ComputeChannelSignature(_options); ns_opt.socket_option.hc_option = _options.hc_option; ns_opt.socket_option.local_side = client_endpoint; diff --git a/src/brpc/channel.h b/src/brpc/channel.h index b4bff6092b..f11bdb04d7 100644 --- a/src/brpc/channel.h +++ b/src/brpc/channel.h @@ -39,6 +39,7 @@ #include "brpc/health_check_option.h" #include "brpc/socket_mode.h" #include "brpc/rdma/rdma_qos.h" +#include "brpc/urma/urma_qos.h" namespace brpc { @@ -115,6 +116,9 @@ struct ChannelOptions { // Outbound RDMA QoS for this Channel. Used when socket_mode is RDMA. RdmaQosOptions rdma_qos; + // Outbound URMA QoS for this Channel. Used when socket_mode is URMA. + UrmaQosOptions urma_qos; + // Turn on authentication for this channel if `auth' is not nullptr. // Note `auth' will not be deleted by channel and must remain valid when // the channel is being used. diff --git a/src/brpc/server.cpp b/src/brpc/server.cpp index 96734ae8d0..4ed8bba4ae 100644 --- a/src/brpc/server.cpp +++ b/src/brpc/server.cpp @@ -880,6 +880,11 @@ int Server::StartInternal(const butil::EndPoint& endpoint, return -1; } + if (real_opt.socket_mode == SOCKET_MODE_URMA && !real_opt.urma_qos.IsValid()) { + LOG(ERROR) << "URMA priority must be -1 (automatic) or in [0,15]"; + return -1; + } + if (!real_opt.h2_settings.IsValid(true/*log_error*/)) { LOG(ERROR) << "Invalid h2_settings"; return -1; @@ -1168,6 +1173,7 @@ int Server::StartInternal(const butil::EndPoint& endpoint, _am->_bthread_tag = _options.bthread_tag; } _am->_rdma_qos = _options.rdma_qos; + _am->_urma_qos = _options.urma_qos; // Set `_status' to RUNNING before accepting connections // to prevent requests being rejected as ELOGOFF _status = RUNNING; diff --git a/src/brpc/server.h b/src/brpc/server.h index 769cee4150..83e1d6027d 100644 --- a/src/brpc/server.h +++ b/src/brpc/server.h @@ -47,6 +47,7 @@ #include "brpc/rpc_pb_message_factory.h" #include "brpc/socket_mode.h" #include "brpc/rdma/rdma_qos.h" +#include "brpc/urma/urma_qos.h" namespace brpc { @@ -246,6 +247,9 @@ struct ServerOptions { // Outbound RDMA QoS for responses on this server's accepted connections. RdmaQosOptions rdma_qos; + // Outbound URMA QoS for responses on accepted connections. + UrmaQosOptions urma_qos; + // [CAUTION] This option is for implementing specialized baidu-std proxies, // most users don't need it. Don't change this option unless you fully // understand the description below. diff --git a/src/brpc/socket.cpp b/src/brpc/socket.cpp index f560320b4b..4b49d9ea72 100644 --- a/src/brpc/socket.cpp +++ b/src/brpc/socket.cpp @@ -720,6 +720,10 @@ int Socket::Create(const SocketOptions& options, SocketId* id) { errno = EINVAL; return -1; } + if (options.socket_mode == SOCKET_MODE_URMA && !options.urma_qos.IsValid()) { + errno = EINVAL; + return -1; + } return VersionedRefWithId::Create(id, options); } @@ -736,6 +740,7 @@ int Socket::OnCreated(const SocketOptions& options) { // start build the transport _socket_mode = options.socket_mode; _rdma_qos = options.rdma_qos; + _urma_qos = options.urma_qos; _transport = TransportFactory::CreateTransport(options.socket_mode); CHECK(nullptr != _transport); _transport->Init(this, options); @@ -2836,6 +2841,7 @@ int Socket::GetPooledSocket(SocketUniquePtr* pooled_socket) { opt.app_connect = _app_connect; opt.socket_mode = _socket_mode; opt.rdma_qos = _rdma_qos; + opt.urma_qos = _urma_qos; socket_pool = new SocketPool(opt); SocketPool* expected = nullptr; if (!main_sp->socket_pool.compare_exchange_strong( @@ -2943,6 +2949,7 @@ int Socket::GetShortSocket(SocketUniquePtr* short_socket) { opt.app_connect = _app_connect; opt.socket_mode = _socket_mode; opt.rdma_qos = _rdma_qos; + opt.urma_qos = _urma_qos; if (get_client_side_messenger()->Create(opt, &id) != 0 || Address(id, short_socket) != 0) { return -1; diff --git a/src/brpc/socket.h b/src/brpc/socket.h index 9cd26674c4..8911b801a8 100644 --- a/src/brpc/socket.h +++ b/src/brpc/socket.h @@ -44,6 +44,7 @@ #include "brpc/health_check_option.h" #include "brpc/socket_mode.h" #include "brpc/rdma/rdma_qos.h" +#include "brpc/urma/urma_qos.h" #include "brpc/input_messenger_processor.h" // InputMessengerProcessor namespace brpc { @@ -311,6 +312,7 @@ struct SocketOptions { std::shared_ptr initial_ssl_ctx; SocketMode socket_mode{SOCKET_MODE_TCP}; RdmaQosOptions rdma_qos; + UrmaQosOptions urma_qos; bthread_keytable_pool_t* keytable_pool{nullptr}; SocketConnection* conn{nullptr}; std::shared_ptr app_connect; @@ -635,6 +637,7 @@ friend class TransportFactory; // The socket mode decided at creation. SocketMode socket_mode() const { return _socket_mode; } const RdmaQosOptions& rdma_qos() const { return _rdma_qos; } + const UrmaQosOptions& urma_qos() const { return _urma_qos; } // Get an UNUSED socket connecting to the same place as this socket // from the SocketPool of this socket. @@ -1003,6 +1006,7 @@ friend class TransportFactory; // Should use SOCKET_MODE_RDMA or SOCKET_MODE_TCP or Other, default is SOCKET_MODE_TCP Transport SocketMode _socket_mode; RdmaQosOptions _rdma_qos; + UrmaQosOptions _urma_qos; std::unique_ptr _transport; // Pass from controller, for progressive reading. diff --git a/src/brpc/urma/mock_urma.cpp b/src/brpc/urma/mock_urma.cpp index b0572c2e86..f81b54bf1a 100644 --- a/src/brpc/urma/mock_urma.cpp +++ b/src/brpc/urma/mock_urma.cpp @@ -240,6 +240,9 @@ urma_status_t urma_query_device(urma_device_t *device, if (!device || !attr) { return URMA_EINVAL; } + mock_device_attr.dev_cap.priority_info[3].tp_type.bs.rtp = 1; + mock_device_attr.dev_cap.priority_info[6].tp_type.bs.ctp = 1; + mock_device_attr.dev_cap.priority_info[10].tp_type.bs.ctp = 1; mock_device_attr.dev_cap.max_jfc = 1024; mock_device_attr.dev_cap.max_jetty = 1024; mock_device_attr.dev_cap.max_jfs_sge = 8; diff --git a/src/brpc/urma/urma_endpoint.cpp b/src/brpc/urma/urma_endpoint.cpp index 26c7a1b9dc..c3fc26e1af 100644 --- a/src/brpc/urma/urma_endpoint.cpp +++ b/src/brpc/urma/urma_endpoint.cpp @@ -390,6 +390,10 @@ int UrmaEndpoint::AllocateResources() { if (_resource) { return 0; } + uint8_t priority; + if (!ResolveUrmaJettyPriority(_socket->urma_qos(), &priority)) { + return -1; + } urma_context_t* ctx = GetUrmaContext(); if (!ctx) { errno = ENODEV; @@ -401,8 +405,9 @@ int UrmaEndpoint::AllocateResources() { return -1; } - // Try the prepared pool first (sized sq/rq match). - if (_sq_size <= static_cast(FLAGS_urma_sq_size) && + // Prepared Jettys carry the default priority; other priorities allocate on demand. + if (priority == GetUrmaJettyPriority() && + _sq_size <= static_cast(FLAGS_urma_sq_size) && _rq_size <= static_cast(FLAGS_urma_rq_size)) { BAIDU_SCOPED_LOCK(g_prepared_mutex); if (g_prepared_list) { @@ -451,7 +456,7 @@ int UrmaEndpoint::AllocateResources() { jetty_cfg.flag.bs.share_jfr = 1; jetty_cfg.jfs_cfg.depth = static_cast(_sq_size); jetty_cfg.jfs_cfg.trans_mode = URMA_TM_RM; - jetty_cfg.jfs_cfg.priority = GetUrmaJettyPriority(); + jetty_cfg.jfs_cfg.priority = priority; jetty_cfg.jfs_cfg.max_sge = static_cast(GetUrmaMaxSge()); jetty_cfg.jfs_cfg.rnr_retry = URMA_TYPICAL_RNR_RETRY; @@ -1560,6 +1565,10 @@ void UrmaEndpoint::DebugInfo(std::ostream& os, butil::StringPiece) const { << " remote_recv_block_size=" << _remote_recv_block_size << " sq_window=" << _sq_window_size.load(butil::memory_order_relaxed) << " remote_rq_window=" << _remote_rq_window_size.load(butil::memory_order_relaxed) + << " urma_priority=" << _socket->urma_qos().priority + << " urma_effective_priority=" + << (_resource && _resource->jetty ? + static_cast(_resource->jetty->jetty_cfg.jfs_cfg.priority) : -1) << " handshake_version=" << _handshake_version << " read_buf=" << _input_processor.read_buf().size(); } diff --git a/src/brpc/urma/urma_endpoint.h b/src/brpc/urma/urma_endpoint.h index e5aed5ac66..a237758f2c 100644 --- a/src/brpc/urma/urma_endpoint.h +++ b/src/brpc/urma/urma_endpoint.h @@ -100,6 +100,7 @@ struct UrmaResource { // windows, and the registered buffer bookkeeping. Cache-line padded to avoid // false sharing between connections. class BAIDU_CACHELINE_ALIGNMENT UrmaEndpoint : public SocketUser { + friend class UrmaQosTestPeer; friend class UrmaConnect; friend class Socket; friend class UrmaTransport; diff --git a/src/brpc/urma/urma_helper.cpp b/src/brpc/urma/urma_helper.cpp index ded662fb22..e15bef6494 100644 --- a/src/brpc/urma/urma_helper.cpp +++ b/src/brpc/urma/urma_helper.cpp @@ -687,6 +687,31 @@ int FindUrmaPriorityForTpType(const urma_device_attr_t& attr, return -1; } uint8_t GetUrmaJettyPriority() { return g_jetty_priority; } + +bool IsUrmaCtpPrioritySupported(const urma_device_attr_t& attr, int priority) { + return priority >= 0 && priority <= URMA_MAX_PRIORITY && + attr.dev_cap.priority_info[priority].tp_type.bs.ctp; +} + +bool ResolveUrmaJettyPriority(const UrmaQosOptions& qos, uint8_t* priority) { + if (!qos.IsValid()) { + errno = EINVAL; + return false; + } + if (qos.priority == -1) { + *priority = GetUrmaJettyPriority(); + return true; + } + if (!IsUrmaAvailable() || !g_context || + !IsUrmaCtpPrioritySupported(g_device_attr, qos.priority)) { + LOG(ERROR) << "URMA priority " << qos.priority + << " does not advertise CTP on the selected device"; + errno = EINVAL; + return false; + } + *priority = static_cast(qos.priority); + return true; +} int GetUrmaMaxSge() { return g_max_sge; } size_t GetUrmaRecvBlockSize() { return g_recv_block_size; } diff --git a/src/brpc/urma/urma_helper.h b/src/brpc/urma/urma_helper.h index 871f1a6354..de663e9906 100644 --- a/src/brpc/urma/urma_helper.h +++ b/src/brpc/urma/urma_helper.h @@ -29,6 +29,7 @@ #if BRPC_WITH_URMA #include "urma_api.h" +#include "brpc/urma/urma_qos.h" #include "urma_types.h" namespace brpc { @@ -83,6 +84,11 @@ int FindUrmaPriorityForTpType(const urma_device_attr_t& attr, // Return the priority selected for the CTP jettys created by brpc. uint8_t GetUrmaJettyPriority(); +// Check whether an explicit priority advertises the CTP path used by brpc. +bool IsUrmaCtpPrioritySupported(const urma_device_attr_t& attr, int priority); +// Resolve automatic selection or validate against the initialized device. +bool ResolveUrmaJettyPriority(const UrmaQosOptions& qos, uint8_t* priority); + // If the URMA environment is available. bool IsUrmaAvailable(); diff --git a/src/brpc/urma/urma_qos.h b/src/brpc/urma/urma_qos.h new file mode 100644 index 0000000000..5c8924bdee --- /dev/null +++ b/src/brpc/urma/urma_qos.h @@ -0,0 +1,32 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#ifndef BRPC_URMA_QOS_H +#define BRPC_URMA_QOS_H + +namespace brpc { + +// Outbound Jetty priority. Device CTP capabilities define valid selections. +struct UrmaQosOptions { + int priority = -1; // -1 selects the device default; explicit range [0, 15]. + + bool IsValid() const { return priority >= -1 && priority <= 15; } +}; + +} // namespace brpc + +#endif // BRPC_URMA_QOS_H diff --git a/src/brpc/urma_transport.cpp b/src/brpc/urma_transport.cpp index c5cc163ffa..a1e8f9b425 100644 --- a/src/brpc/urma_transport.cpp +++ b/src/brpc/urma_transport.cpp @@ -197,6 +197,11 @@ int UrmaTransport::ContextInitOrDie(bool server_or_not, const void* options) { return -1; } urma::GlobalUrmaInitializeOrDie(); + uint8_t priority; + if (!urma::ResolveUrmaJettyPriority( + static_cast(options)->urma_qos, &priority)) { + return -1; + } if (!urma::InitPollingModeWithTag( static_cast(options)->bthread_tag)) { return -1; @@ -207,6 +212,11 @@ int UrmaTransport::ContextInitOrDie(bool server_or_not, const void* options) { return -1; } urma::GlobalUrmaInitializeOrDie(); + uint8_t priority; + if (!urma::ResolveUrmaJettyPriority( + static_cast(options)->urma_qos, &priority)) { + return -1; + } if (!urma::InitPollingModeWithTag(bthread_self_tag())) { return -1; } diff --git a/test/brpc_urma_qos_unittest.cpp b/test/brpc_urma_qos_unittest.cpp new file mode 100644 index 0000000000..7e63bb5878 --- /dev/null +++ b/test/brpc_urma_qos_unittest.cpp @@ -0,0 +1,256 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#include +#include +#include +#include +#include + +#if BRPC_WITH_URMA +#include "brpc/acceptor.h" +#include "brpc/channel.h" +#include "brpc/controller.h" +#include "brpc/server.h" +#include "brpc/input_messenger.h" +#include "brpc/load_balancer.h" +#include "brpc/urma/urma_endpoint.h" +#include "brpc/urma/urma_helper.h" +#include "brpc/urma_transport.h" +#include "echo.pb.h" + +using namespace brpc; + +namespace brpc { +namespace urma { +DECLARE_int32(urma_client_handshake_version); +extern bool g_skip_urma_init; +DECLARE_int32(urma_buffer_count); +DECLARE_int32(urma_prepared_jetty_cnt); + +class UrmaQosTestPeer { +public: + static int Allocate(UrmaEndpoint* ep) { return ep->AllocateResources(); } + static int Priority(UrmaEndpoint* ep) { + return ep->_resource->jetty->jetty_cfg.jfs_cfg.priority; + } +}; + +} // namespace urma +} // namespace brpc + +TEST(UrmaHelperTest, explicit_priority_requires_ctp_capability) { + urma_device_attr_t attr{}; + attr.dev_cap.priority_info[0].tp_type.bs.ctp = 1; + attr.dev_cap.priority_info[15].tp_type.bs.ctp = 1; + attr.dev_cap.priority_info[15].tp_type.bs.rtp = 1; + attr.dev_cap.priority_info[3].tp_type.bs.rtp = 1; + EXPECT_TRUE(urma::IsUrmaCtpPrioritySupported(attr, 0)); + EXPECT_TRUE(urma::IsUrmaCtpPrioritySupported(attr, 15)); + for (int priority : {-2, -1, 3, 4, 16}) { + EXPECT_FALSE(urma::IsUrmaCtpPrioritySupported(attr, priority)); + } +} + +class UrmaQosTest : public ::testing::Test { +protected: + static void SetUpTestSuite() { + urma_init_attr_t attr{}; + const auto status = urma_init(&attr); + if (status != URMA_SUCCESS && status != URMA_EEXIST) { + return; + } + int count = 0; + auto** devices = urma_get_device_list(&count); + _mock = devices && count > 0 && + strcmp(devices[0]->name, "mock_urma_device") == 0; + urma_free_device_list(devices); + if (status == URMA_SUCCESS) { + urma_uninit(); + } + if (!_mock) { + return; + } + urma::g_skip_urma_init = false; + urma::FLAGS_urma_buffer_count = 1024; + urma::FLAGS_urma_prepared_jetty_cnt = 2; + urma::GlobalUrmaInitializeOrDie(); + } + void SetUp() override { + if (!_mock) { + GTEST_SKIP() << "URMA QoS integration tests use the link-time mock"; + } + ASSERT_TRUE(urma::IsUrmaAvailable()); + ASSERT_NE(nullptr, urma::GetUrmaContext()); + } + static bool _mock; + static urma::UrmaEndpoint* Endpoint(SocketUniquePtr& socket) { + return static_cast(socket->_transport.get())->GetUrmaEp(); + } +}; +bool UrmaQosTest::_mock = false; + +TEST_F(UrmaQosTest, rejects_invalid_and_unsupported_priority) { + butil::EndPoint address; + ASSERT_EQ(0, butil::str2endpoint("127.0.0.1:8123", &address)); + for (int priority : {-2, 16, 3, 9}) { + ChannelOptions co; + co.socket_mode = SOCKET_MODE_URMA; + co.urma_qos.priority = priority; + Channel channel; + EXPECT_EQ(-1, channel.Init(address, &co)); + ServerOptions so; + so.socket_mode = SOCKET_MODE_URMA; + so.urma_qos = co.urma_qos; + Server server; + EXPECT_EQ(-1, server.Start("127.0.0.1:0", &so)); + if (!co.urma_qos.IsValid()) { + SocketOptions options; + options.socket_mode = SOCKET_MODE_URMA; + options.urma_qos = co.urma_qos; + SocketId sid; + EXPECT_EQ(-1, Socket::Create(options, &sid)); + EXPECT_EQ(EINVAL, errno); + } else { + SocketOptions options; + options.socket_mode = SOCKET_MODE_URMA; + options.urma_qos = co.urma_qos; + SocketId sid; + ASSERT_EQ(0, Socket::Create(options, &sid)); + SocketUniquePtr socket; + ASSERT_EQ(0, Socket::Address(sid, &socket)); + EXPECT_EQ(-1, urma::UrmaQosTestPeer::Allocate(Endpoint(socket))); + EXPECT_EQ(EINVAL, errno); + socket->SetFailed(); + } + } +} + +TEST_F(UrmaQosTest, channels_isolate_priority_and_inherit_to_child_connections) { + ChannelOptions co; + co.socket_mode = SOCKET_MODE_URMA; + Channel automatic, low, same, high, named, tcp, tcp_with_qos; + ASSERT_EQ(0, automatic.Init("127.0.0.1:8123", &co)); + co.urma_qos.priority = 6; + ASSERT_EQ(0, low.Init("127.0.0.1:8123", &co)); + ASSERT_EQ(0, same.Init("127.0.0.1:8123", &co)); + ASSERT_EQ(0, named.Init("list://127.0.0.1:8123", "rr", &co)); + co.urma_qos.priority = 10; + ASSERT_EQ(0, high.Init("127.0.0.1:8123", &co)); + EXPECT_EQ(low._server_id, same._server_id); + EXPECT_NE(low._server_id, high._server_id); + EXPECT_NE(low._server_id, automatic._server_id); + co.socket_mode = SOCKET_MODE_TCP; + ASSERT_EQ(0, tcp_with_qos.Init("127.0.0.1:8123", &co)); + ASSERT_EQ(0, tcp.Init("127.0.0.1:8123", nullptr)); + EXPECT_EQ(tcp._server_id, tcp_with_qos._server_id); + EXPECT_NE(tcp._server_id, automatic._server_id); + SocketUniquePtr main, pooled, short_socket, selected; + ASSERT_EQ(0, Socket::Address(low._server_id, &main)); + ASSERT_EQ(0, main->GetPooledSocket(&pooled)); + ASSERT_EQ(0, main->GetShortSocket(&short_socket)); + LoadBalancer::SelectIn input{}; + LoadBalancer::SelectOut output(&selected); + ASSERT_EQ(0, named._lb->SelectServer(input, &output)); + for (auto* socket : {main.get(), pooled.get(), short_socket.get(), selected.get()}) { + EXPECT_EQ(6, socket->urma_qos().priority); + } + EXPECT_EQ(main->id(), selected->id()); + EXPECT_EQ(0, pooled->ReturnToPool()); + short_socket->SetFailed(); +} + +TEST_F(UrmaQosTest, prepared_jettys_are_used_only_for_matching_priority) { + urma::UrmaEndpoint::GlobalRelease(); + ASSERT_EQ(0, urma::UrmaEndpoint::GlobalInitialize()); + // Allocate priority 10 before consuming the two prepared default Jettys. + for (int priority : {10, -1, 6, 10}) { + SocketOptions options; + options.socket_mode = SOCKET_MODE_URMA; + options.urma_qos.priority = priority; + SocketId sid; + ASSERT_EQ(0, Socket::Create(options, &sid)); + SocketUniquePtr socket; + ASSERT_EQ(0, Socket::Address(sid, &socket)); + auto* ep = Endpoint(socket); + ASSERT_EQ(0, urma::UrmaQosTestPeer::Allocate(ep)); + EXPECT_EQ(priority == -1 ? 6 : priority, urma::UrmaQosTestPeer::Priority(ep)); + std::ostringstream debug; + ep->DebugInfo(debug); + EXPECT_NE(std::string::npos, debug.str().find("urma_effective_priority=")); + socket->SetFailed(); + } +} + +class UrmaQosEcho : public ::test::EchoService { + void Echo(google::protobuf::RpcController*, const ::test::EchoRequest* request, + ::test::EchoResponse* response, google::protobuf::Closure* done) override { + response->set_message(request->message()); + done->Run(); + } +}; + +TEST_F(UrmaQosTest, request_and_response_priorities_apply_across_restart) { + GFLAGS_NAMESPACE::FlagSaver saved_flags; + UrmaQosEcho service; + Server server; + ASSERT_EQ(0, server.AddService(&service, SERVER_DOESNT_OWN_SERVICE)); + for (int response_priority : {6, 10}) { + urma::FLAGS_urma_client_handshake_version = response_priority == 6 ? 2 : 3; + ServerOptions so; + so.socket_mode = SOCKET_MODE_URMA; + so.urma_qos.priority = response_priority; + ASSERT_EQ(0, server.Start("127.0.0.1:0", &so)); + ChannelOptions co; + co.socket_mode = SOCKET_MODE_URMA; + co.urma_qos.priority = response_priority == 6 ? 10 : 6; + co.timeout_ms = 2000; + co.max_retry = 0; + co.connection_type = "single"; + Channel channel; + ASSERT_EQ(0, channel.Init(server.listen_address(), &co)); + Controller cntl; + ::test::EchoRequest request; + ::test::EchoResponse response; + request.set_message("URMA QoS"); + ::test::EchoService_Stub(&channel).Echo(&cntl, &request, &response, nullptr); + EXPECT_FALSE(cntl.Failed()) << cntl.ErrorText(); + EXPECT_EQ(request.message(), response.message()); + SocketUniquePtr client; + ASSERT_EQ(0, Socket::Address(channel._server_id, &client)); + EXPECT_EQ(co.urma_qos.priority, urma::UrmaQosTestPeer::Priority(Endpoint(client))); + std::vector sockets; + server._am->ListConnections(&sockets); + ASSERT_EQ(1u, sockets.size()); + SocketUniquePtr accepted; + ASSERT_EQ(0, Socket::Address(sockets[0], &accepted)); + EXPECT_EQ(response_priority, accepted->urma_qos().priority); + EXPECT_EQ(response_priority, urma::UrmaQosTestPeer::Priority(Endpoint(accepted))); + accepted.reset(); + client.reset(); + server.Stop(0); + server.Join(); + } +} + +#endif // BRPC_WITH_URMA + +int main(int argc, char** argv) { + testing::InitGoogleTest(&argc, argv); + gflags::ParseCommandLineFlags(&argc, &argv, true); + return RUN_ALL_TESTS(); +}