diff --git a/src/brpc/rdma/rdma_endpoint.cpp b/src/brpc/rdma/rdma_endpoint.cpp index e2ce2e0c1b..d2b76ccc83 100644 --- a/src/brpc/rdma/rdma_endpoint.cpp +++ b/src/brpc/rdma/rdma_endpoint.cpp @@ -17,6 +17,7 @@ #if BRPC_WITH_RDMA +#include // std::min #include #include "butil/fd_utility.h" #include "butil/logging.h" // CHECK, LOG @@ -1165,6 +1166,15 @@ int RdmaEndpoint::DoAllocateResources() { } 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; @@ -1212,7 +1222,19 @@ 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 = IBV_MTU_1024; + 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(negotiated_mtu); attr.ah_attr.grh.dgid = remote.gid; attr.ah_attr.grh.flow_label = 0; attr.ah_attr.grh.sgid_index = GetRdmaGidIndex(); diff --git a/src/brpc/rdma/rdma_endpoint.h b/src/brpc/rdma/rdma_endpoint.h index 03bec81408..c6596313e7 100644 --- a/src/brpc/rdma/rdma_endpoint.h +++ b/src/brpc/rdma/rdma_endpoint.h @@ -287,6 +287,13 @@ friend int v3_wire::WriteV3Hello(RdmaEndpoint*, const RdmaHello&); // QP reached RTS (filled in BringUpQp). butil::optional _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 _outgoing_mtu; + // rdma resource RdmaResource* _resource; diff --git a/src/brpc/rdma/rdma_handshake.cpp b/src/brpc/rdma/rdma_handshake.cpp index 180c2b3f0b..f2e68aface 100644 --- a/src/brpc/rdma/rdma_handshake.cpp +++ b/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); diff --git a/src/brpc/rdma/rdma_handshake.h b/src/brpc/rdma/rdma_handshake.h index 6238d424f0..d91dccf04f 100644 --- a/src/brpc/rdma/rdma_handshake.h +++ b/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 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 path_mtu; }; // Result of reading/parsing a peer's hello (see ReceiveAndParseRemoteHello). diff --git a/src/brpc/rdma/rdma_handshake.proto b/src/brpc/rdma/rdma_handshake.proto index b5627b2e9d..08491657ba 100644 --- a/src/brpc/rdma/rdma_handshake.proto +++ b/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; }. diff --git a/src/brpc/rdma/rdma_helper.cpp b/src/brpc/rdma/rdma_helper.cpp index b0e13ad72c..c598f3cad1 100644 --- a/src/brpc/rdma/rdma_helper.cpp +++ b/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; } diff --git a/src/brpc/rdma/rdma_helper.h b/src/brpc/rdma/rdma_helper.h index 052763325b..fc69a31f9e 100644 --- a/src/brpc/rdma/rdma_helper.h +++ b/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(); diff --git a/test/brpc_rdma_unittest.cpp b/test/brpc_rdma_unittest.cpp index 2ecd1f3cac..1aa0593c70 100644 --- a/test/brpc_rdma_unittest.cpp +++ b/test/brpc_rdma_unittest.cpp @@ -2026,6 +2026,275 @@ TEST_F(RdmaTest, server_alloc_resource_fail_fallback_tcp) { StopServer(); } +// Build a valid v3 hello that also carries an MTU value. +rdma::RdmaHello MakeValidV3HelloWithMtu(uint32_t mtu) { + rdma::RdmaHello msg = MakeValidV3Hello(); + msg.set_mtu(mtu); + return msg; +} + +// A client hello carrying MTU must not break the server handshake: +// the server still parses the hello and advances to S_ACK_WAIT. +TEST_F(RdmaTest, v3_server_accepts_client_hello_with_mtu) { + StartServer(); + + sockaddr_in addr; + bzero((char*)&addr, sizeof(addr)); + addr.sin_family = AF_INET; + addr.sin_port = htons(PORT); + butil::fd_guard sockfd(socket(AF_INET, SOCK_STREAM, 0)); + ASSERT_TRUE(sockfd >= 0); + ASSERT_EQ(0, connect(sockfd, (sockaddr*)&addr, sizeof(sockaddr))); + usleep(100000); + Socket* s = GetSocketFromServer(0); + ASSERT_TRUE(s != NULL); + + rdma::RdmaHello msg = MakeValidV3HelloWithMtu(IBV_MTU_4096); + std::string packet = MakeV3Packet(msg); + ASSERT_EQ((ssize_t)packet.size(), + write(sockfd, packet.data(), packet.size())); + usleep(100000); + + ASSERT_EQ(rdma::RdmaEndpoint::S_ACK_WAIT, + static_cast(s->_transport.get())->_rdma_ep->_state); + + rdma::RdmaHello reply; + ReadServerV3Reply(sockfd, &reply); + + // ACK flags=0 -> clean FALLBACK_TCP so the test ends without hardware. + uint32_t flags = butil::HostToNet32(0); + ASSERT_EQ((ssize_t)sizeof(flags), write(sockfd, &flags, sizeof(flags))); + usleep(100000); + ASSERT_EQ(rdma::RdmaEndpoint::FALLBACK_TCP, + static_cast(s->_transport.get())->_rdma_ep->_state); + + sockfd.reset(-1); + usleep(100000); + ASSERT_EQ(NULL, GetSocketFromServer(0)); + StopServer(); +} + +// The server reply must carry a negotiated MTU when the client advertised one. +// The server computes min(local, client) at the start of BringUpQp (before the +// g_skip_rdma_init early-return), so _outgoing_mtu is set even in UT mode; +// FillLocalRdmaHello then includes it in the reply. +TEST_F(RdmaTest, v3_server_reply_has_negotiated_mtu) { + StartServer(); + + sockaddr_in addr; + bzero((char*)&addr, sizeof(addr)); + addr.sin_family = AF_INET; + addr.sin_port = htons(PORT); + butil::fd_guard sockfd(socket(AF_INET, SOCK_STREAM, 0)); + ASSERT_TRUE(sockfd >= 0); + ASSERT_EQ(0, connect(sockfd, (sockaddr*)&addr, sizeof(sockaddr))); + usleep(100000); + Socket* s = GetSocketFromServer(0); + ASSERT_TRUE(s != NULL); + + // Advertise a non-default MTU so we can verify negotiation. + rdma::RdmaHello msg = MakeValidV3HelloWithMtu(IBV_MTU_4096); + std::string packet = MakeV3Packet(msg); + ASSERT_EQ((ssize_t)packet.size(), + write(sockfd, packet.data(), packet.size())); + usleep(100000); + + rdma::RdmaHello reply; + ReadServerV3Reply(sockfd, &reply); + + // The server should advertise a negotiated MTU in its reply. + EXPECT_TRUE(reply.has_mtu()); + // Negotiated MTU = min(local_active_mtu, client_mtu). Since + // g_skip_rdma_init is true in UT, OpenDevice is never called and + // g_active_mtu stays at the default IBV_MTU_1024. + // So negotiated = min(1024, 4096) = 1024. + EXPECT_EQ(IBV_MTU_1024, reply.mtu()); + + uint32_t flags = butil::HostToNet32(0); + ASSERT_EQ((ssize_t)sizeof(flags), write(sockfd, &flags, sizeof(flags))); + usleep(100000); + + sockfd.reset(-1); + usleep(100000); + StopServer(); +} + +// When the client does NOT advertise an MTU (e.g., v2 peer or older v3 peer), +// the server must NOT include an MTU in its reply (backward-compatible +// degradation to legacy IBV_MTU_1024 default). +TEST_F(RdmaTest, v3_server_reply_has_no_mtu_without_client_mtu) { + StartServer(); + + sockaddr_in addr; + bzero((char*)&addr, sizeof(addr)); + addr.sin_family = AF_INET; + addr.sin_port = htons(PORT); + butil::fd_guard sockfd(socket(AF_INET, SOCK_STREAM, 0)); + ASSERT_TRUE(sockfd >= 0); + ASSERT_EQ(0, connect(sockfd, (sockaddr*)&addr, sizeof(sockaddr))); + usleep(100000); + Socket* s = GetSocketFromServer(0); + ASSERT_TRUE(s != NULL); + + // Client hello without MTU field. + rdma::RdmaHello msg = MakeValidV3Hello(); + std::string packet = MakeV3Packet(msg); + ASSERT_EQ((ssize_t)packet.size(), + write(sockfd, packet.data(), packet.size())); + usleep(100000); + + rdma::RdmaHello reply; + ReadServerV3Reply(sockfd, &reply); + + // Server must not advertise MTU when client didn't. + EXPECT_FALSE(reply.has_mtu()); + + uint32_t flags = butil::HostToNet32(0); + ASSERT_EQ((ssize_t)sizeof(flags), write(sockfd, &flags, sizeof(flags))); + usleep(100000); + + sockfd.reset(-1); + usleep(100000); + StopServer(); +} + +// A client hello with an out-of-range MTU (e.g., 0 or > IBV_MTU_4096) must be +// rejected by ValidRdmaHello. Per the handshake-fallback design, the server +// does NOT close the connection; instead it degrades to TCP fallback: the +// endpoint enters S_ACK_WAIT with RDMA_OFF, the reply carries no MTU, and a +// subsequent ACK-without-RDMA finalizes FALLBACK_TCP. +TEST_F(RdmaTest, v3_server_rejects_invalid_mtu) { + StartServer(); + + sockaddr_in addr; + bzero((char*)&addr, sizeof(addr)); + addr.sin_family = AF_INET; + addr.sin_port = htons(PORT); + + // Test with MTU = 0 (below IBV_MTU_256). + butil::fd_guard sockfd1(socket(AF_INET, SOCK_STREAM, 0)); + ASSERT_TRUE(sockfd1 >= 0); + ASSERT_EQ(0, connect(sockfd1, (sockaddr*)&addr, sizeof(sockaddr))); + usleep(100000); + Socket* s = GetSocketFromServer(0); + ASSERT_TRUE(s != NULL); + + rdma::RdmaHello msg = MakeValidV3HelloWithMtu(0); + std::string packet = MakeV3Packet(msg); + ASSERT_EQ((ssize_t)packet.size(), + write(sockfd1, packet.data(), packet.size())); + usleep(100000); + + // The server rejects the hello but falls back to TCP instead of failing. + s = GetSocketFromServer(0); + ASSERT_TRUE(s != NULL); + ASSERT_EQ(rdma::RdmaEndpoint::S_ACK_WAIT, + static_cast(s->_transport.get())->_rdma_ep->_state); + ASSERT_EQ(RdmaTransport::RDMA_OFF, + static_cast(s->_transport.get())->_rdma_state); + ASSERT_FALSE(s->Failed()); + + // The fallback reply must not carry an MTU. + rdma::RdmaHello reply; + ReadServerV3Reply(sockfd1, &reply); + EXPECT_FALSE(reply.has_mtu()); + + // Ack without RDMA so the server finalizes FALLBACK_TCP. + uint32_t flags = butil::HostToNet32(0); + ASSERT_EQ(sizeof(flags), write(sockfd1, &flags, sizeof(flags))); + usleep(100000); + ASSERT_EQ(rdma::RdmaEndpoint::FALLBACK_TCP, + static_cast(s->_transport.get())->_rdma_ep->_state); + ASSERT_FALSE(s->Failed()); + + sockfd1.reset(-1); + usleep(100000); + + // Test with MTU = 6 (above IBV_MTU_4096). + butil::fd_guard sockfd2(socket(AF_INET, SOCK_STREAM, 0)); + ASSERT_TRUE(sockfd2 >= 0); + ASSERT_EQ(0, connect(sockfd2, (sockaddr*)&addr, sizeof(sockaddr))); + usleep(100000); + s = GetSocketFromServer(0); + ASSERT_TRUE(s != NULL); + + msg = MakeValidV3HelloWithMtu(6); + packet = MakeV3Packet(msg); + ASSERT_EQ((ssize_t)packet.size(), + write(sockfd2, packet.data(), packet.size())); + usleep(100000); + + s = GetSocketFromServer(0); + ASSERT_TRUE(s != NULL); + ASSERT_EQ(rdma::RdmaEndpoint::S_ACK_WAIT, + static_cast(s->_transport.get())->_rdma_ep->_state); + ASSERT_EQ(RdmaTransport::RDMA_OFF, + static_cast(s->_transport.get())->_rdma_state); + ASSERT_FALSE(s->Failed()); + + ReadServerV3Reply(sockfd2, &reply); + EXPECT_FALSE(reply.has_mtu()); + + flags = butil::HostToNet32(0); + ASSERT_EQ(sizeof(flags), write(sockfd2, &flags, sizeof(flags))); + usleep(100000); + ASSERT_EQ(rdma::RdmaEndpoint::FALLBACK_TCP, + static_cast(s->_transport.get())->_rdma_ep->_state); + ASSERT_FALSE(s->Failed()); + + sockfd2.reset(-1); + usleep(100000); + + StopServer(); +} + +// Verify the client includes its local MTU in the v3 hello. +TEST_F(RdmaTest, v3_client_hello_includes_mtu) { + HandshakeVersionFlag _hsv(3); + + butil::fd_guard sockfd(butil::tcp_listen(g_ep)); + EXPECT_TRUE(sockfd >= 0); + + Channel channel; + ChannelOptions chan_options; + chan_options.socket_mode = SOCKET_MODE_RDMA; + chan_options.connect_timeout_ms = 500; + chan_options.timeout_ms = 500; + chan_options.max_retry = 0; + ASSERT_EQ(0, channel.Init(g_ep, &chan_options)); + + Controller cntl; + test::EchoRequest req; + test::EchoResponse res; + req.set_message(__FUNCTION__); + google::protobuf::Closure* done = DoNothing(); + ::test::EchoService::Stub(&channel).Echo(&cntl, &req, &res, done); + + butil::fd_guard acc_fd(accept(sockfd, NULL, NULL)); + ASSERT_TRUE(acc_fd >= 0); + + // Read 4B magic + 4B pb_size + body from the client hello. + uint8_t hdr[8]; + ASSERT_EQ(8, read(acc_fd, hdr, 8)); + ASSERT_EQ(0, memcmp(hdr, "RDM3", 4)); + uint32_t pb_size = butil::NetToHost32(*reinterpret_cast(hdr + 4)); + ASSERT_GT(pb_size, 0u); + ASSERT_LE(pb_size, 4096u); + std::string body(pb_size, '\0'); + ASSERT_EQ((ssize_t)pb_size, read(acc_fd, &body[0], pb_size)); + + rdma::RdmaHello hello; + ASSERT_TRUE(hello.ParseFromString(body)); + + // In UT mode (g_skip_rdma_init=true), OpenDevice is never called so + // g_active_mtu stays at the default IBV_MTU_1024. SendLocalHello calls + // GetRdmaActiveMtu() unconditionally, so the client hello includes mtu=1024. + EXPECT_TRUE(hello.has_mtu()); + EXPECT_EQ(IBV_MTU_1024, hello.mtu()); + + bthread_id_join(cntl.call_id()); +} + TEST_F(RdmaTest, try_global_disable_rdma) { StartServer(); rdma::g_rdma_available.store(false, butil::memory_order_relaxed);