detect and negotiate mtu

src/brpc/rdma/rdma_endpoint.cpp

@@ -35,6 +35,7 @@
 #include "brpc/rdma/rdma_handshake_constants.h"
 
 DECLARE_int32(task_group_ntags);
+DECLARE_bool(rdma_mtu_negotiation);
 
 namespace brpc {
 namespace rdma {
@@ -156,6 +157,7 @@ void RdmaEndpoint::Reset() {
     _state.store(UNINIT, butil::memory_order_relaxed);
     _handshake_version = 0;
     _outgoing_ece.reset();
+    _outgoing_mtu.reset();
     _resource = NULL;
     _send_cq_events = 0;
     _recv_cq_events = 0;
@@ -1141,6 +1143,27 @@ int RdmaEndpoint::AllocateResources() {
 }
 
 int RdmaEndpoint::BringUpQp(const ParsedHello& remote, bool is_server) {
+    // MTU negotiation. This is pure integer arithmetic and does NOT depend on
+    // real hardware, so it runs even in UT mode (where the actual QP bring-up
+    // below is skipped via g_skip_rdma_init). Keeping it here ensures the
+    // server-side _outgoing_mtu is populated so the server hello advertises
+    // the negotiated MTU, which unit tests rely on.
+    //
+    //   Server: remote.path_mtu is the client's active MTU; we compute
+    //           min(local_active_mtu, client_mtu) and advertise it back.
+    //   Client: remote.path_mtu is already the server's negotiated MTU; we
+    //           just apply it to the QP.
+    uint32_t negotiated_mtu = IBV_MTU_1024;
+    if (FLAGS_rdma_mtu_negotiation && remote.path_mtu.has_value()) {
+        if (is_server) {
+            negotiated_mtu = std::min(GetRdmaActiveMtu(), *remote.path_mtu);
+            // Store the negotiated MTU for the server hello reply.
+            _outgoing_mtu = negotiated_mtu;
+        } else {
+            negotiated_mtu = *remote.path_mtu;
+        }
+    }
+
     if (BAIDU_UNLIKELY(g_skip_rdma_init)) {
         // For UT
         return 0;
@@ -1188,7 +1211,10 @@ 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
+    // negotiated_mtu was computed at the top of this function: the server's
+    // min(local_active_mtu, client_mtu), or the client's echo of the server's
+    // advertised value. Falls back to IBV_MTU_1024 when no MTU was advertised.
+    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

@@ -281,6 +281,14 @@ 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. Populated only when
+    // FLAGS_rdma_mtu_negotiation is on:
+    //   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

@@ -57,6 +57,13 @@ DEFINE_bool(rdma_ece, false, "Enable end-to-end ECE (Enhanced Connection Establi
                              "to no-ECE when the peer, the local libibverbs, or set_ece "
                              "does not support it. Acts as a kill switch (default off).");
 
+DEFINE_bool(rdma_mtu_negotiation, true,
+            "Enable RDMA path-MTU negotiation in the v3 handshake. When on, "
+            "the two ends pick min(local_active_mtu, peer_active_mtu) instead "
+            "of the legacy fixed IBV_MTU_1024. Backward compatible: peers that "
+            "do not advertise an MTU (v2, or older v3) fall back to "
+            "IBV_MTU_1024. Acts as a kill switch (default on).");
+
 DECLARE_bool(rdma_trace_verbose);
 
 namespace v2_wire {
@@ -318,6 +325,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 (FLAGS_rdma_mtu_negotiation && ep->_outgoing_mtu.has_value()) {
+        msg->set_mtu(*ep->_outgoing_mtu);
+    }
 }
 
 int ReadAndParseV3Hello(RdmaEndpoint* ep, RdmaHello* out) {
@@ -380,6 +396,22 @@ void TranslateHello(const RdmaHello& msg, ParsedHello* out) {
         ece.comp_mask = msg.ece().comp_mask();
         out->ece = ece;
     }
+    if (FLAGS_rdma_mtu_negotiation && msg.has_mtu()) {
+        const uint32_t advertised = msg.mtu();
+        // ibv_mtu only defines 256/512/1024/2048/4096. Reject any other value
+        // so a buggy/malicious peer cannot push an out-of-range enum through
+        // static_cast<ibv_mtu>, which would make ibv_modify_qp fail (EINVAL)
+        // and force a silent TCP fallback.
+        if (advertised == IBV_MTU_256 || advertised == IBV_MTU_512 ||
+            advertised == IBV_MTU_1024 || advertised == IBV_MTU_2048 ||
+            advertised == IBV_MTU_4096) {
+            out->path_mtu = advertised;
+        } else {
+            LOG(WARNING) << "Peer advertised an invalid RDMA MTU ("
+                         << advertised << "), ignoring it; falling back to "
+                         << "IBV_MTU_1024";
+        }
+    }
 }
 
 }  // namespace v3_wire
@@ -400,6 +432,16 @@ int RdmaHandshakeClientV3::SendLocalHello() {
         }
     }
 
+    // Advertise local active MTU in the client hello. GetRdmaActiveMtu() is a
+    // pure getter (always returns a valid enum value, defaulting to
+    // IBV_MTU_1024), so it is safe to call even in UT mode (no real RDMA
+    // hardware) and must NOT be gated on g_skip_rdma_init.
+    // Governed by FLAGS_rdma_mtu_negotiation: when off, the field is omitted
+    // and the peer falls back to IBV_MTU_1024.
+    if (FLAGS_rdma_mtu_negotiation) {
+        _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). Toggled by FLAGS_rdma_mtu_negotiation.
+    // Optional: carries the sender's active MTU (IBV_MTU_256 .. IBV_MTU_4096).
+    // Absent on v2 peers, older v3 peers, or when negotiation is disabled;
+    // 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 uint8_t g_port_num = 1;
 
@@ -458,6 +459,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;
@@ -467,6 +469,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;
@@ -516,6 +519,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();
@@ -696,6 +700,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();