detect and negotiate mtu

src/brpc/rdma/rdma_endpoint.cpp

@@ -17,6 +17,7 @@
 
 #if BRPC_WITH_RDMA
 
+#include <algorithm>                            // std::min
 #include <gflags/gflags.h>
 #include "butil/fd_utility.h"
 #include "butil/logging.h"                   // CHECK, LOG
@@ -1401,6 +1402,15 @@ int RdmaEndpoint::StartCqEvents() {
 }
 
 int RdmaEndpoint::BringUpQp(const ParsedHello& remote, bool is_server) {
+    // MTU negotiation (server side): compute min(local, client) and store it
+    // so FillLocalRdmaHello can advertise the negotiated value in the server
+    // hello reply. This is done before the g_skip_rdma_init early-return so
+    // that UT (which skips real QP bring-up) still sets _outgoing_mtu.
+    if (is_server && remote.path_mtu.has_value()) {
+        uint32_t local_mtu = GetRdmaActiveMtu();
+        _outgoing_mtu = std::min(local_mtu, *remote.path_mtu);
+    }
+
     if (BAIDU_UNLIKELY(g_skip_rdma_init)) {
         // For UT
         return 0;
@@ -1448,7 +1458,20 @@ int RdmaEndpoint::BringUpQp(const ParsedHello& remote, bool is_server) {
     }
 
     attr.qp_state = IBV_QPS_RTR;
-    attr.path_mtu = IBV_MTU_1024;  // TODO: support more mtu in future
+    // MTU negotiation: use the peer-advertised MTU if available, otherwise
+    // fall back to the legacy default (IBV_MTU_1024).
+    //   Server side: _outgoing_mtu was already computed at function entry;
+    //     reuse it for the QP path_mtu attribute.
+    //   Client side: remote.path_mtu is already the server's negotiated
+    //     min(local, client) and we use it as-is.
+    uint32_t negotiated_mtu =
+        std::min(static_cast<uint32_t>(IBV_MTU_1024), GetRdmaActiveMtu());
+    if (is_server && _outgoing_mtu.has_value()) {
+        negotiated_mtu = *_outgoing_mtu;
+    } else if (!is_server && remote.path_mtu.has_value()) {
+        negotiated_mtu = *remote.path_mtu;
+    }
+    attr.path_mtu = static_cast<ibv_mtu>(negotiated_mtu);
     attr.ah_attr.grh.dgid = remote.gid;
     attr.ah_attr.grh.flow_label = 0;
     attr.ah_attr.grh.sgid_index = GetRdmaGidIndex();

src/brpc/rdma/rdma_endpoint.h

@@ -314,6 +314,13 @@ friend int v3_wire::WriteV3Hello(RdmaEndpoint*, const RdmaHello&);
     //           QP reached RTS (filled in BringUpQp).
     butil::optional<ibv_ece> _outgoing_ece;
 
+    // MTU payload to advertise in the next local hello:
+    //   Client: the locally queried active MTU (filled
+    //           before C_HELLO_SEND);
+    //   Server: the negotiated MTU = min(local_active_mtu, client_mtu)
+    //           (filled in BringUpQp).
+    butil::optional<uint32_t> _outgoing_mtu;
+
     // rdma resource
     RdmaResource* _resource;
 

src/brpc/rdma/rdma_handshake.cpp

@@ -286,6 +286,12 @@ bool ValidRdmaHello(const RdmaHello& msg) {
     if (msg.qp_num() == 0 && !g_skip_rdma_init) {
         return false;
     }
+    // Validate MTU: must be a valid ibv_mtu enum value (IBV_MTU_256..IBV_MTU_4096)
+    // to prevent invalid values from being passed to ibv_modify_qp.
+    if (msg.has_mtu() &&
+        (msg.mtu() < IBV_MTU_256 || msg.mtu() > IBV_MTU_4096)) {
+        return false;
+    }
     return true;
 }
 
@@ -318,6 +324,15 @@ void FillLocalRdmaHello(const RdmaEndpoint* ep, RdmaHello* msg) {
         ece->set_options(ep->_outgoing_ece->options);
         ece->set_comp_mask(ep->_outgoing_ece->comp_mask);
     }
+
+    // Advertise MTU if the endpoint has a value to advertise.
+    // Client side: queried local active MTU (filled before C_HELLO_SEND).
+    // Server side: negotiated MTU = min(local_mtu, client_mtu)
+    //   (filled in BringUpQp).
+    // nullopt -> omit the field (peer falls back to IBV_MTU_1024).
+    if (ep->_outgoing_mtu.has_value()) {
+        msg->set_mtu(*ep->_outgoing_mtu);
+    }
 }
 
 int ReadAndParseV3Hello(RdmaEndpoint* ep, RdmaHello* out) {
@@ -380,6 +395,9 @@ void TranslateHello(const RdmaHello& msg, ParsedHello* out) {
         ece.comp_mask = msg.ece().comp_mask();
         out->ece = ece;
     }
+    if (msg.has_mtu()) {
+        out->path_mtu = msg.mtu();
+    }
 }
 
 }  // namespace v3_wire
@@ -400,6 +418,14 @@ int RdmaHandshakeClientV3::SendLocalHello() {
         }
     }
 
+    // Query local active MTU so it can be advertised in the client hello.
+    // Best-effort: any failure just means we won't advertise MTU
+    // (the peer falls back to IBV_MTU_1024).
+    // In UT mode (g_skip_rdma_init=true), GetRdmaActiveMtu() returns the
+    // default IBV_MTU_1024 (g_active_mtu is never overwritten by OpenDevice),
+    // so the client hello still includes a valid MTU.
+    _ep->_outgoing_mtu = GetRdmaActiveMtu();
+
     RdmaHello local_msg{};
     v3_wire::FillLocalRdmaHello(_ep, &local_msg);
     return v3_wire::WriteV3Hello(_ep, local_msg);

src/brpc/rdma/rdma_handshake.h

@@ -53,6 +53,14 @@ struct ParsedHello {
     //   - on the server side: the client's queried ECE capabilities;
     //   - on the client side: the server's reduced/negotiated ECE.
     butil::optional<ibv_ece> ece;
+
+    // MTU negotiation, v3 handshake only.
+    // nullopt means the peer did not advertise an MTU (v2 peer or older v3
+    // peer that predates MTU negotiation). When engaged:
+    //   - on the server side: the client's active MTU;
+    //   - on the client side: the server's negotiated MTU
+    //     (= min(local_active_mtu, client_mtu)).
+    butil::optional<uint32_t> path_mtu;
 };
 
 // Result of reading/parsing a peer's hello (see ReceiveAndParseRemoteHello).

src/brpc/rdma/rdma_handshake.proto

@@ -55,6 +55,16 @@ message RdmaHello {
     //   Server hello: carries the REDUCED/negotiated ECE
     //                   queried after the QP reached RTS.
     optional RdmaEce ece = 7;
+
+    // MTU negotiation (v3 only).
+    // Optional: carries the sender's active MTU (IBV_MTU_256 .. IBV_MTU_4096).
+    // Absent on v2 peers and on older v3 peers that predate MTU negotiation;
+    // the receiver then falls back to IBV_MTU_1024.
+    //
+    // Semantics differ by sender role:
+    //   Client hello: the locally queried active MTU of the RDMA port.
+    //   Server hello: the negotiated MTU = min(local_active_mtu, client_mtu).
+    optional uint32 mtu = 8;
 }
 
 // Mirrors struct ibv_ece { uint32 vendor_id; uint32 options; uint32 comp_mask; }.

src/brpc/rdma/rdma_helper.cpp

@@ -83,6 +83,7 @@ static int g_gid_tbl_len = 0;
 static uint8_t g_gid_index = 0;
 static ibv_gid g_gid;
 static uint16_t g_lid;
+static uint32_t g_active_mtu = IBV_MTU_1024;
 static int g_max_sge = 0;
 static uint32_t g_vendor_id = 0;
 static uint8_t g_port_num = 1;
@@ -459,6 +460,7 @@ static ibv_context* OpenDevice(int num_total, int* num_available_devices) {
                 ret_context = context.release();
                 g_gid_tbl_len = attr.gid_tbl_len;
                 g_lid = attr.lid;
+                g_active_mtu = attr.active_mtu;
             } else {
                 LOG(INFO) << "Device name not match: " << context->device->name
                           << " vs " << FLAGS_rdma_device;
@@ -468,6 +470,7 @@ static ibv_context* OpenDevice(int num_total, int* num_available_devices) {
             ret_context = context.release();
             g_gid_tbl_len = attr.gid_tbl_len;
             g_lid = attr.lid;
+            g_active_mtu = attr.active_mtu;
         }
     }
     return ret_context;
@@ -517,6 +520,7 @@ static void GlobalRdmaInitializeOrDieImpl() {
         LOG(INFO) << "RDMA device: " << g_context->device->name;
     }
     LOG(INFO) << "RDMA LID: " << g_lid;
+    LOG(INFO) << "RDMA Active MTU: " << g_active_mtu;
     if (!FindRdmaGid(g_context)) {
         LOG(ERROR) << "Fail to find available RDMA GID";
         ExitWithError();
@@ -687,6 +691,10 @@ uint16_t GetRdmaLid() {
     return g_lid;
 }
 
+uint32_t GetRdmaActiveMtu() {
+    return g_active_mtu;
+}
+
 uint8_t GetRdmaGidIndex() {
     return g_gid_index;
 }

src/brpc/rdma/rdma_helper.h

@@ -68,6 +68,9 @@ ibv_gid GetRdmaGid();
 // Return Global LID
 uint16_t GetRdmaLid();
 
+// Return active MTU of the RDMA port (IBV_MTU_256 .. IBV_MTU_4096).
+uint32_t GetRdmaActiveMtu();
+
 // Return suggested comp vector for CQ
 int GetRdmaCompVector();