Add business QoS configuration for RDMA and URMA

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功能。

docs/cn/urma.md

@@ -193,3 +193,63 @@ UrmaTransport 推荐用于**大包和跨节点**高吞吐路径，而 UBRing
   `ContextInitOrDie` 阶段拒绝。
 - 硬件数据通路需要受支持的 UMDK provider 和 `liburma`。
 - 当前实现面向 Linux。
+
+## 业务 SL 与优先级
+
+`ChannelOptions.urma_qos` 控制请求发送方向，
+`ServerOptions.urma_qos` 控制响应发送方向：
+
+```cpp
+brpc::ChannelOptions client_options;
+client_options.socket_mode = brpc::SOCKET_MODE_URMA;
+client_options.urma_qos.service_level = 6;
+
+brpc::ServerOptions server_options;
+server_options.socket_mode = brpc::SOCKET_MODE_URMA;
+server_options.urma_qos.service_level = 7;
+```
+
+当前使用 RM＋CTP。CTP 的 SL 由设备 `priority_info[priority].SL` 决定，
+业务 SL 与 priority 数字可以不同。`service_level` 默认为 -1，表示不指定 SL；
+显式范围为 0–255。指定 SL 后，选择首个同时匹配该 SL 且包含 CTP 能力的 priority。
+`priority` 可单独指定，范围为 0–15；同时指定 SL 和 priority 时，两者必须匹配设备映射。
+找不到 CTP 映射或配置冲突时，Channel 初始化和 Server 启动报错，列出支持的 priority:SL。
+例如设备只有 priority 6→SL 6、priority 7→SL 7 支持 CTP 时，SL 10 会被拒绝。
+
+两项均为 -1 时沿用原自动选择：首个 CTP priority，未报告时使用兼容值 6。
+设备全局映射与调度待遇由平台配置；brpc 使用初始化时读取的映射，平台修改后重启进程加载。
+本接口按当前 RM＋CTP 路径校验。
+
+不同 SL/priority 配置使用独立连接，相同配置共享连接。配置继承到服务发现、连接池、
+短连接和服务端 accept 路径，重建时重新应用。默认预建池只用于匹配其 priority 的连接，
+其他 priority 按需创建 Jetty。连接调试信息显示请求 SL/priority、Jetty priority 和映射 SL。
+请求和响应分别配置各自的发送策略，v2/v3 握手格式沿用现有协议。
+
+`example/urma_performance` 的 client/server 支持 `--urma_sl=6` 和 `--urma_priority=6`。
+示例值须在设备上有匹配的 CTP 映射，数字大小不代表调度优先关系。
+
+### 实机验证
+
+先使用同一对设备验证原生 RM、CTP、共享 JFR 数据通路。设置本机 `URMA_DEVICE`，
+客户端另设 `SERVER_IP`，先启动服务端，再启动客户端：
+
+```bash
+# 服务端
+urma_perftest send_lat --mgmt_proto tcp -d "$URMA_DEVICE" --ctp \
+    -p 0 -j true -I 0 -s 64 -n 1000 -P 8003
+# 客户端
+urma_perftest send_lat --mgmt_proto tcp -d "$URMA_DEVICE" --ctp \
+    -p 0 -j true -I 0 -s 64 -n 1000 -P 8003 -S "$SERVER_IP"
+```
+
+两端再将 `-I 0` 改为 `-I 64` 做 inline 对照。原生测试通过后，用
+默认 SL/priority 完成 brpc 单流基线，再逐个测试设备支持的显式 SL，
+请求、响应方向分别改变配置，并验证不支持的 SL 和冲突配置在初始化时被拒绝。
+连接调试信息应为 `state=ESTABLISHED`，并显示预期的 `urma_effective_priority` 与
+`urma_effective_sl`；TCP 回落的吞吐数据不计入 URMA QoS 结果。
+最后启动竞争流量，结合设备 SL 映射、队列计数器、吞吐和 P99 验证调度效果。
+
+记录两端内核、UMDK、provider 和固件版本。`status=4` 是本地内存访问错误；
+原生测试出现该错误时先定位数据通路。brpc 完成错误日志包含收发方向、请求 SL/priority、
+Jetty priority 和映射 SL，初始化日志包含 `default_priority`、`default_sl`。mock 测试验证配置传递和
+Jetty 属性，报文标记与调度效果由实机测试验证。

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.

docs/en/urma.md

@@ -207,3 +207,71 @@ 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 SL and priority
+
+Set `ChannelOptions.urma_qos` for outgoing requests and
+`ServerOptions.urma_qos` for outgoing responses:
+
+```cpp
+brpc::ChannelOptions client_options;
+client_options.socket_mode = brpc::SOCKET_MODE_URMA;
+client_options.urma_qos.service_level = 6;
+
+brpc::ServerOptions server_options;
+server_options.socket_mode = brpc::SOCKET_MODE_URMA;
+server_options.urma_qos.service_level = 7;
+```
+
+The current path uses RM and CTP. CTP derives SL from
+`priority_info[priority].SL`; SL and priority need not have equal numeric values.
+`service_level` defaults to -1 (unspecified), with explicit values in [0,255].
+An explicit SL selects the first priority with both that SL and CTP capability.
+`priority` can also be set explicitly in [0,15]. When both are specified, the
+device mapping must satisfy both. Channel initialization and Server startup
+reject unsupported or conflicting requests and report supported priority:SL pairs.
+For example, if only priority 6/SL 6 and priority 7/SL 7 support CTP, SL 10 is rejected.
+
+With both fields at -1, automatic selection retains the first CTP priority and
+the existing compatibility value 6 when none is reported. The platform configures
+device mappings and scheduling; brpc uses the initialization snapshot. Restart
+processes after changing platform mappings. Validation applies to the RM/CTP path.
+
+Different SL/priority configurations use separate connections; 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 the
+requested SL/priority, Jetty priority and mapped SL. Each peer configures its outgoing direction;
+v2/v3 handshake formats remain the same.
+
+The client and server in `example/urma_performance` accept `--urma_sl=6` and
+`--urma_priority=6`. Example values require matching CTP mappings on the selected
+device; their numeric order does not specify scheduling precedence.
+
+### Hardware validation
+
+First test the same device pair with native RM, CTP and shared JFR. Set the local
+`URMA_DEVICE` on each host and `SERVER_IP` on the client. Start the server first:
+
+```bash
+# Server
+urma_perftest send_lat --mgmt_proto tcp -d "$URMA_DEVICE" --ctp \
+    -p 0 -j true -I 0 -s 64 -n 1000 -P 8003
+# Client
+urma_perftest send_lat --mgmt_proto tcp -d "$URMA_DEVICE" --ctp \
+    -p 0 -j true -I 0 -s 64 -n 1000 -P 8003 -S "$SERVER_IP"
+```
+
+Repeat with `-I 64` on both hosts for the inline comparison. Once native tests
+pass, establish a single-flow brpc baseline with default SL/priority, then test
+supported explicit SLs, changing the request and response directions separately.
+Verify that unsupported SLs and conflicting settings fail initialization.
+Connection debug output must show `state=ESTABLISHED` and the expected
+`urma_effective_priority` and `urma_effective_sl`. Exclude TCP fallback measurements from URMA QoS results.
+Then compare device SL mappings, queue counters, throughput and P99 under contention.
+
+Record kernel, UMDK, provider and firmware versions on both hosts. Status 4 means
+a local memory access error; resolve native data-path failures before measuring
+QoS. brpc completion errors log the direction, requested SL/priority, Jetty priority
+and mapped SL; initialization logs `default_priority` and `default_sl`. Mock tests check configuration propagation
+and Jetty attributes; hardware tests check packet markings and scheduling effects.

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();

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<uint64_t> 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;

example/urma_performance/client.cpp

@@ -44,6 +44,8 @@ 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]");
+DEFINE_int32(urma_sl, -1, "Outbound URMA SL: -1 unspecified, or [0,255]");
 
 bvar::LatencyRecorder g_latency("client");
 bvar::Adder<int64_t> g_error_count("client_error_count");
@@ -88,6 +90,8 @@ 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.urma_qos.service_level = FLAGS_urma_sl;
     options.socket_mode = FLAGS_use_urma ? brpc::SOCKET_MODE_URMA
                                           : brpc::SOCKET_MODE_TCP;
     options.connect_timeout_ms = FLAGS_rpc_timeout_ms;
@@ -110,7 +114,10 @@ int main(int argc, char* argv[]) {
     stub.Test(&warmup_cntl, &warmup_req, &warmup_resp, nullptr);
     if (warmup_cntl.Failed()) {
         LOG(ERROR) << "Warm-up RPC failed after timeout_ms="
-                   << FLAGS_rpc_timeout_ms << ": "
+                   << FLAGS_rpc_timeout_ms
+                   << " use_urma=" << FLAGS_use_urma
+                   << " urma_priority=" << FLAGS_urma_priority
+                   << " urma_sl=" << FLAGS_urma_sl << ": "
                    << warmup_cntl.ErrorText();
         return -1;
     }
@@ -131,6 +138,8 @@ int main(int argc, char* argv[]) {
     }
     LOG(INFO) << "URMA performance client started (server=" << FLAGS_server
               << ", use_urma=" << FLAGS_use_urma
+              << ", urma_priority=" << FLAGS_urma_priority
+              << ", urma_sl=" << FLAGS_urma_sl
               << ", threads=" << thread_num
               << ", rpc_timeout_ms=" << FLAGS_rpc_timeout_ms << ")";
     while (!brpc::IsAskedToQuit()) {

example/urma_performance/server.cpp

@@ -30,6 +30,8 @@
 
 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]");
+DEFINE_int32(urma_sl, -1, "Outbound URMA SL: -1 unspecified, or [0,255]");
 
 butil::atomic<uint64_t> g_last_time(0);
 
@@ -73,14 +75,18 @@ int main(int argc, char* argv[]) {
     }
 
     brpc::ServerOptions options;
+    options.urma_qos.priority = FLAGS_urma_priority;
+    options.urma_qos.service_level = FLAGS_urma_sl;
     options.socket_mode = FLAGS_use_urma ? brpc::SOCKET_MODE_URMA
                                           : brpc::SOCKET_MODE_TCP;
     if (server.Start(FLAGS_port, &options) != 0) {
         LOG(ERROR) << "Fail to start server";
         return -1;
     }
     LOG(INFO) << "URMA performance server started on port " << FLAGS_port
-              << " (use_urma=" << FLAGS_use_urma << ")";
+              << " (use_urma=" << FLAGS_use_urma
+              << ", urma_priority=" << FLAGS_urma_priority
+              << ", urma_sl=" << FLAGS_urma_sl << ")";
     server.RunUntilAskedToQuit();
     return 0;
 }

src/brpc/acceptor.cpp

@@ -338,6 +338,8 @@ 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.urma_qos = am->_urma_qos;
         options.bthread_tag = am->_bthread_tag;
         if (Socket::Create(options, &socket_id) != 0) {
             am->ReleaseConnectionSlot();

src/brpc/acceptor.h

@@ -24,6 +24,8 @@
 #include "butil/containers/flat_map.h"
 #include "brpc/input_messenger.h"
 #include "brpc/socket_mode.h"
+#include "brpc/rdma/rdma_qos.h"
+#include "brpc/urma/urma_qos.h"
 
 
 namespace brpc {
@@ -137,6 +139,8 @@ 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;

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,8 +140,12 @@ 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");
+            buf.append("|priority=").append(std::to_string(opt.urma_qos.priority));
+            buf.append("|sl=").append(std::to_string(opt.urma_qos.service_level));
         }
         butil::MurmurHash3_x64_128_Update(&mm_ctx, buf.data(), buf.size());
         buf.clear();
@@ -187,6 +192,15 @@ 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;
+    }
+    if (_options.socket_mode == SOCKET_MODE_URMA && !_options.urma_qos.IsValid()) {
+        LOG(ERROR) << "URMA priority must be -1 or in [0,15], "
+                      "and SL must be -1 or 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 +421,8 @@ 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.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),
@@ -453,6 +469,8 @@ 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.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;