Copilot commented on code in PR #3505:
URL: https://github.com/apache/brpc/pull/3505#discussion_r3894468524
##########
test/brpc_rdma_unittest.cpp:
##########
@@ -719,6 +730,234 @@ TEST_F(RdmaTest, client_send_data_on_tcp_after_ack_send) {
StopServer();
}
+// Build a well-formed v2 client hello: "RDMA" followed by the 36B body.
+static void MakeV2ClientHello(uint8_t (&data)[rdma::HELLO_V2_MSG_LEN_MIN]) {
+ rdma::v2_wire::HelloMessage msg{};
+ msg.msg_len = rdma::HELLO_V2_MSG_LEN_MIN;
+ msg.hello_ver = rdma::HELLO_V2_VERSION;
+ msg.impl_ver = rdma::IMPL_V2_VERSION;
+ msg.sq_size = 16;
+ msg.rq_size = 16;
+ msg.block_size = 8192;
+ msg.qp_num = 0;
+ msg.gid = rdma::GetRdmaGid();
+ memcpy(data, "RDMA", 4);
+ msg.Serialize(data + 4);
+}
+
+// Connect, push a well-formed v2 hello and read back the server's reply, which
+// leaves the server in S_ACK_WAIT waiting for the 4B ACK.
+static void HandshakeUntilAckWait(butil::fd_guard* sockfd) {
+ sockaddr_in addr;
+ bzero((char*)&addr, sizeof(addr));
+ addr.sin_family = AF_INET;
+ addr.sin_port = htons(PORT);
+ sockfd->reset(socket(AF_INET, SOCK_STREAM, 0));
+ ASSERT_TRUE(*sockfd >= 0);
+ ASSERT_EQ(0, connect(*sockfd, (sockaddr*)&addr, sizeof(sockaddr)));
Review Comment:
`HandshakeUntilAckWait` never sets `addr.sin_addr`, so `connect()` targets
0.0.0.0 and is expected to fail on many systems. Also, `read()`/`write()` are
not guaranteed to transfer the full buffer in one call; asserting an exact byte
count will make the test flaky. Set `addr.sin_addr` to the loopback (or use
`g_ep`), and replace single `read`/`write` calls with a small loop that
reads/writes exactly `HELLO_V2_MSG_LEN_MIN` bytes (or use a helper that does
so).
##########
test/brpc_rdma_unittest.cpp:
##########
@@ -719,6 +730,234 @@ TEST_F(RdmaTest, client_send_data_on_tcp_after_ack_send) {
StopServer();
}
+// Build a well-formed v2 client hello: "RDMA" followed by the 36B body.
+static void MakeV2ClientHello(uint8_t (&data)[rdma::HELLO_V2_MSG_LEN_MIN]) {
+ rdma::v2_wire::HelloMessage msg{};
+ msg.msg_len = rdma::HELLO_V2_MSG_LEN_MIN;
+ msg.hello_ver = rdma::HELLO_V2_VERSION;
+ msg.impl_ver = rdma::IMPL_V2_VERSION;
+ msg.sq_size = 16;
+ msg.rq_size = 16;
+ msg.block_size = 8192;
+ msg.qp_num = 0;
+ msg.gid = rdma::GetRdmaGid();
+ memcpy(data, "RDMA", 4);
+ msg.Serialize(data + 4);
+}
+
+// Connect, push a well-formed v2 hello and read back the server's reply, which
+// leaves the server in S_ACK_WAIT waiting for the 4B ACK.
+static void HandshakeUntilAckWait(butil::fd_guard* sockfd) {
+ sockaddr_in addr;
+ bzero((char*)&addr, sizeof(addr));
+ addr.sin_family = AF_INET;
+ addr.sin_port = htons(PORT);
+ sockfd->reset(socket(AF_INET, SOCK_STREAM, 0));
+ ASSERT_TRUE(*sockfd >= 0);
+ ASSERT_EQ(0, connect(*sockfd, (sockaddr*)&addr, sizeof(sockaddr)));
+
+ uint8_t hello[rdma::HELLO_V2_MSG_LEN_MIN];
+ MakeV2ClientHello(hello);
+ ASSERT_EQ((ssize_t)sizeof(hello), write(*sockfd, hello, sizeof(hello)));
+ usleep(100000); // wait for server to handle the msg
+ uint8_t reply[rdma::HELLO_V2_MSG_LEN_MIN];
+ ASSERT_EQ((ssize_t)sizeof(reply), read(*sockfd, reply, sizeof(reply)));
Review Comment:
`HandshakeUntilAckWait` never sets `addr.sin_addr`, so `connect()` targets
0.0.0.0 and is expected to fail on many systems. Also, `read()`/`write()` are
not guaranteed to transfer the full buffer in one call; asserting an exact byte
count will make the test flaky. Set `addr.sin_addr` to the loopback (or use
`g_ep`), and replace single `read`/`write` calls with a small loop that
reads/writes exactly `HELLO_V2_MSG_LEN_MIN` bytes (or use a helper that does
so).
##########
test/brpc_rdma_unittest.cpp:
##########
@@ -719,6 +730,234 @@ TEST_F(RdmaTest, client_send_data_on_tcp_after_ack_send) {
StopServer();
}
+// Build a well-formed v2 client hello: "RDMA" followed by the 36B body.
+static void MakeV2ClientHello(uint8_t (&data)[rdma::HELLO_V2_MSG_LEN_MIN]) {
+ rdma::v2_wire::HelloMessage msg{};
+ msg.msg_len = rdma::HELLO_V2_MSG_LEN_MIN;
+ msg.hello_ver = rdma::HELLO_V2_VERSION;
+ msg.impl_ver = rdma::IMPL_V2_VERSION;
+ msg.sq_size = 16;
+ msg.rq_size = 16;
+ msg.block_size = 8192;
+ msg.qp_num = 0;
+ msg.gid = rdma::GetRdmaGid();
+ memcpy(data, "RDMA", 4);
+ msg.Serialize(data + 4);
+}
+
+// Connect, push a well-formed v2 hello and read back the server's reply, which
+// leaves the server in S_ACK_WAIT waiting for the 4B ACK.
+static void HandshakeUntilAckWait(butil::fd_guard* sockfd) {
+ sockaddr_in addr;
+ bzero((char*)&addr, sizeof(addr));
+ addr.sin_family = AF_INET;
+ addr.sin_port = htons(PORT);
+ sockfd->reset(socket(AF_INET, SOCK_STREAM, 0));
+ ASSERT_TRUE(*sockfd >= 0);
+ ASSERT_EQ(0, connect(*sockfd, (sockaddr*)&addr, sizeof(sockaddr)));
+
+ uint8_t hello[rdma::HELLO_V2_MSG_LEN_MIN];
+ MakeV2ClientHello(hello);
+ ASSERT_EQ((ssize_t)sizeof(hello), write(*sockfd, hello, sizeof(hello)));
+ usleep(100000); // wait for server to handle the msg
Review Comment:
These fixed `usleep(100000)` waits make the unit tests timing-dependent and
prone to flakes under load/slow CI. Prefer polling with a bounded timeout for
the expected state transition (e.g., wait until `transport->_rdma_ep->_state`
reaches the target) or use an explicit synchronization signal from the server
side when possible.
##########
test/brpc_rdma_unittest.cpp:
##########
@@ -719,6 +730,234 @@ TEST_F(RdmaTest, client_send_data_on_tcp_after_ack_send) {
StopServer();
}
+// Build a well-formed v2 client hello: "RDMA" followed by the 36B body.
+static void MakeV2ClientHello(uint8_t (&data)[rdma::HELLO_V2_MSG_LEN_MIN]) {
+ rdma::v2_wire::HelloMessage msg{};
+ msg.msg_len = rdma::HELLO_V2_MSG_LEN_MIN;
+ msg.hello_ver = rdma::HELLO_V2_VERSION;
+ msg.impl_ver = rdma::IMPL_V2_VERSION;
+ msg.sq_size = 16;
+ msg.rq_size = 16;
+ msg.block_size = 8192;
+ msg.qp_num = 0;
+ msg.gid = rdma::GetRdmaGid();
+ memcpy(data, "RDMA", 4);
+ msg.Serialize(data + 4);
+}
+
+// Connect, push a well-formed v2 hello and read back the server's reply, which
+// leaves the server in S_ACK_WAIT waiting for the 4B ACK.
+static void HandshakeUntilAckWait(butil::fd_guard* sockfd) {
+ sockaddr_in addr;
+ bzero((char*)&addr, sizeof(addr));
+ addr.sin_family = AF_INET;
+ addr.sin_port = htons(PORT);
+ sockfd->reset(socket(AF_INET, SOCK_STREAM, 0));
+ ASSERT_TRUE(*sockfd >= 0);
+ ASSERT_EQ(0, connect(*sockfd, (sockaddr*)&addr, sizeof(sockaddr)));
+
+ uint8_t hello[rdma::HELLO_V2_MSG_LEN_MIN];
+ MakeV2ClientHello(hello);
+ ASSERT_EQ((ssize_t)sizeof(hello), write(*sockfd, hello, sizeof(hello)));
+ usleep(100000); // wait for server to handle the msg
+ uint8_t reply[rdma::HELLO_V2_MSG_LEN_MIN];
+ ASSERT_EQ((ssize_t)sizeof(reply), read(*sockfd, reply, sizeof(reply)));
+}
+
+// A client is free to pipeline its first request right behind the handshake
+// ACK. Only the 4B ACK belongs to the handshake. Whatever follows it must be
+// handed over to the real protocol instead of dropping the connection.
+TEST_F(RdmaTest, server_accepts_data_pipelined_behind_fallback_ack) {
+ StartServer();
+
+ butil::fd_guard sockfd;
+ ASSERT_NO_FATAL_FAILURE(HandshakeUntilAckWait(&sockfd));
+ Socket* s = GetSocketFromServer(0);
+ ASSERT_TRUE(s != nullptr);
+ auto* transport = static_cast<RdmaTransport*>(s->_transport.get());
+ ASSERT_EQ(rdma::RdmaEndpoint::S_ACK_WAIT, transport->_rdma_ep->_state);
+
+ // An ACK asking for TCP, plus the first 4 bytes of a baidu_std request.
One
+ // write, so that both end up in the same read on the server.
+ uint8_t ack_and_data[rdma::HELLO_ACK_LEN + 4];
+ const uint32_t flags = butil::HostToNet32(0);
+ memcpy(ack_and_data, &flags, rdma::HELLO_ACK_LEN);
+ memcpy(ack_and_data + rdma::HELLO_ACK_LEN, "PRPC", 4);
+ ASSERT_EQ((ssize_t)sizeof(ack_and_data),
+ write(sockfd, ack_and_data, sizeof(ack_and_data)));
+ usleep(100000); // wait for server to handle the msg
Review Comment:
These fixed `usleep(100000)` waits make the unit tests timing-dependent and
prone to flakes under load/slow CI. Prefer polling with a bounded timeout for
the expected state transition (e.g., wait until `transport->_rdma_ep->_state`
reaches the target) or use an explicit synchronization signal from the server
side when possible.
##########
src/brpc/rdma/rdma_endpoint.cpp:
##########
@@ -1164,6 +1265,43 @@ int RdmaEndpoint::DoAllocateResources() {
return 0;
}
+int RdmaEndpoint::StartCqEvents() {
+ CHECK_EQ(InputMessengerProcessor::STREAM_NONE,
_socket->parsing_stream_type())
+ << "StartCqEvents() called while " << *_socket << " is parsing";
Review Comment:
Using `CHECK` here will abort the entire process if this invariant is
violated in production (even if the connection could otherwise be failed
gracefully). Consider downgrading to `DCHECK` plus returning `-1`/failing the
socket, or explicitly handling this as a recoverable error path to avoid
turning a single-connection race into a process-wide crash.
##########
test/brpc_rdma_unittest.cpp:
##########
@@ -719,6 +730,234 @@ TEST_F(RdmaTest, client_send_data_on_tcp_after_ack_send) {
StopServer();
}
+// Build a well-formed v2 client hello: "RDMA" followed by the 36B body.
+static void MakeV2ClientHello(uint8_t (&data)[rdma::HELLO_V2_MSG_LEN_MIN]) {
+ rdma::v2_wire::HelloMessage msg{};
+ msg.msg_len = rdma::HELLO_V2_MSG_LEN_MIN;
+ msg.hello_ver = rdma::HELLO_V2_VERSION;
+ msg.impl_ver = rdma::IMPL_V2_VERSION;
+ msg.sq_size = 16;
+ msg.rq_size = 16;
+ msg.block_size = 8192;
+ msg.qp_num = 0;
+ msg.gid = rdma::GetRdmaGid();
+ memcpy(data, "RDMA", 4);
+ msg.Serialize(data + 4);
+}
+
+// Connect, push a well-formed v2 hello and read back the server's reply, which
+// leaves the server in S_ACK_WAIT waiting for the 4B ACK.
+static void HandshakeUntilAckWait(butil::fd_guard* sockfd) {
+ sockaddr_in addr;
+ bzero((char*)&addr, sizeof(addr));
+ addr.sin_family = AF_INET;
+ addr.sin_port = htons(PORT);
+ sockfd->reset(socket(AF_INET, SOCK_STREAM, 0));
+ ASSERT_TRUE(*sockfd >= 0);
+ ASSERT_EQ(0, connect(*sockfd, (sockaddr*)&addr, sizeof(sockaddr)));
+
+ uint8_t hello[rdma::HELLO_V2_MSG_LEN_MIN];
+ MakeV2ClientHello(hello);
+ ASSERT_EQ((ssize_t)sizeof(hello), write(*sockfd, hello, sizeof(hello)));
+ usleep(100000); // wait for server to handle the msg
+ uint8_t reply[rdma::HELLO_V2_MSG_LEN_MIN];
+ ASSERT_EQ((ssize_t)sizeof(reply), read(*sockfd, reply, sizeof(reply)));
+}
+
+// A client is free to pipeline its first request right behind the handshake
+// ACK. Only the 4B ACK belongs to the handshake. Whatever follows it must be
+// handed over to the real protocol instead of dropping the connection.
+TEST_F(RdmaTest, server_accepts_data_pipelined_behind_fallback_ack) {
+ StartServer();
+
+ butil::fd_guard sockfd;
+ ASSERT_NO_FATAL_FAILURE(HandshakeUntilAckWait(&sockfd));
+ Socket* s = GetSocketFromServer(0);
+ ASSERT_TRUE(s != nullptr);
+ auto* transport = static_cast<RdmaTransport*>(s->_transport.get());
+ ASSERT_EQ(rdma::RdmaEndpoint::S_ACK_WAIT, transport->_rdma_ep->_state);
+
+ // An ACK asking for TCP, plus the first 4 bytes of a baidu_std request.
One
+ // write, so that both end up in the same read on the server.
+ uint8_t ack_and_data[rdma::HELLO_ACK_LEN + 4];
+ const uint32_t flags = butil::HostToNet32(0);
+ memcpy(ack_and_data, &flags, rdma::HELLO_ACK_LEN);
+ memcpy(ack_and_data + rdma::HELLO_ACK_LEN, "PRPC", 4);
+ ASSERT_EQ((ssize_t)sizeof(ack_and_data),
+ write(sockfd, ack_and_data, sizeof(ack_and_data)));
+ usleep(100000); // wait for server to handle the msg
+
+ // The handshake took the ACK only and left "PRPC" to baidu_std, which is
+ // now waiting for the rest of its 12B header. So the connection lives on
+ // with those 4 bytes still buffered.
+ ASSERT_EQ(rdma::RdmaEndpoint::FALLBACK_TCP, transport->_rdma_ep->_state);
+ ASSERT_EQ(RdmaTransport::RDMA_OFF, transport->_rdma_state);
+ ASSERT_TRUE(GetSocketFromServer(0) != nullptr);
+ ASSERT_EQ(4u, s->fd_input_processor().read_buf().size());
+
+ sockfd.reset(-1);
+ usleep(100000); // wait for server to handle the msg
+ ASSERT_EQ(nullptr, GetSocketFromServer(0));
+
+ StopServer();
+}
+
+// Once RDMA is on, the TCP fd is no longer an RPC channel, so bytes trailing
+// the ACK can only be a protocol error.
+TEST_F(RdmaTest, server_rejects_data_pipelined_behind_rdma_ack) {
+ StartServer();
+
+ butil::fd_guard sockfd;
+ ASSERT_NO_FATAL_FAILURE(HandshakeUntilAckWait(&sockfd));
+ Socket* s = GetSocketFromServer(0);
+ ASSERT_TRUE(s != nullptr);
+ auto* transport = static_cast<RdmaTransport*>(s->_transport.get());
+ ASSERT_EQ(rdma::RdmaEndpoint::S_ACK_WAIT, transport->_rdma_ep->_state);
+
+ uint8_t ack_and_data[rdma::HELLO_ACK_LEN + 4];
+ const uint32_t flags = butil::HostToNet32(rdma::HELLO_ACK_RDMA_OK);
+ memcpy(ack_and_data, &flags, rdma::HELLO_ACK_LEN);
+ memcpy(ack_and_data + rdma::HELLO_ACK_LEN, "PRPC", 4);
+ ASSERT_EQ((ssize_t)sizeof(ack_and_data),
+ write(sockfd, ack_and_data, sizeof(ack_and_data)));
+ usleep(100000); // wait for server to handle the msg
+
+ // Note that `transport->_rdma_ep` is gone by now: dropping the connection
+ // recycles the Socket, and RdmaTransport::Release() deletes the endpoint.
+ ASSERT_EQ(nullptr, GetSocketFromServer(0));
+
+ StopServer();
+}
+
+// Once RDMA is on, the server must stop parsing its TCP fd altogether.
+TEST_F(RdmaTest, server_stops_parsing_tcp_fd_once_rdma_is_on) {
+ StartServer();
+
+ butil::fd_guard sockfd;
+ ASSERT_NO_FATAL_FAILURE(HandshakeUntilAckWait(&sockfd));
+ Socket* s = GetSocketFromServer(0);
+ ASSERT_TRUE(s != nullptr);
+ auto* transport = static_cast<RdmaTransport*>(s->_transport.get());
+ ASSERT_EQ(rdma::RdmaEndpoint::S_ACK_WAIT, transport->_rdma_ep->_state);
+
+ // A bare ACK asking for RDMA. Nothing trails it, so the handshake ends in
+ // ESTABLISHED instead of being rejected (see the test above).
+ const uint32_t flags = butil::HostToNet32(rdma::HELLO_ACK_RDMA_OK);
+ ASSERT_EQ((ssize_t)sizeof(flags), write(sockfd, &flags, sizeof(flags)));
+ usleep(100000); // wait for server to handle the msg
+ ASSERT_EQ(rdma::RdmaEndpoint::ESTABLISHED, transport->_rdma_ep->_state);
+ ASSERT_EQ(RdmaTransport::RDMA_ON, transport->_rdma_state);
+ ASSERT_TRUE(GetSocketFromServer(0) != nullptr);
+
+ ASSERT_EQ(4, write(sockfd, "PRPC", 4));
+ usleep(100000); // wait for server to handle the msg
+ ASSERT_EQ(nullptr, GetSocketFromServer(0));
+
+ StopServer();
+}
+
+// The same bytes on the stream carried by the QP are a real RPC, and the
handler
+// must decline so that CutInputMessage() moves on to the protocol handlers.
+TEST_F(RdmaTest, server_parses_qp_stream_after_rdma_is_on) {
+ StartServer();
+
+ butil::fd_guard sockfd;
+ ASSERT_NO_FATAL_FAILURE(HandshakeUntilAckWait(&sockfd));
+ Socket* s = GetSocketFromServer(0);
+ ASSERT_TRUE(s != nullptr);
+ auto* transport = static_cast<RdmaTransport*>(s->_transport.get());
+ ASSERT_EQ(rdma::RdmaEndpoint::S_ACK_WAIT, transport->_rdma_ep->_state);
+
+ const uint32_t flags = butil::HostToNet32(rdma::HELLO_ACK_RDMA_OK);
+ ASSERT_EQ((ssize_t)sizeof(flags), write(sockfd, &flags, sizeof(flags)));
+ usleep(100000); // wait for server to handle the msg
+ ASSERT_EQ(rdma::RdmaEndpoint::ESTABLISHED, transport->_rdma_ep->_state);
+
+ InputMessengerProcessor& qp_stream = transport->_rdma_ep->_input_processor;
+ ASSERT_TRUE(qp_stream.read_buf().empty());
+ qp_stream.read_buf().append("PRPC");
+ InputMessageClosure last_msg;
+ ASSERT_EQ(0, qp_stream.ProcessNewMessage(4, false,
butil::gettimeofday_us(), 0, last_msg));
Review Comment:
`ProcessNewMessage()` expects `received_us` in the same time domain used
elsewhere (`cpuwide_time_us()`), and `base_realtime` is normally derived from
`gettimeofday_us() - received_us`. Passing `gettimeofday_us()` as `received_us`
and `0` as `base_realtime` can corrupt `_last_readtime_us` (used for idle
timeout and debug deltas) and introduce hard-to-diagnose time-related behavior.
Use `butil::cpuwide_time_us()` for `received_us` and compute `base_realtime`
consistently.
##########
src/brpc/rdma/rdma_endpoint.cpp:
##########
@@ -237,36 +247,88 @@ void RdmaEndpoint::OnNewDataFromTcp(Socket* m) {
ep->_read_butex->fetch_add(1, butil::memory_order_release);
bthread::butex_wake(ep->_read_butex);
} else if (state == FALLBACK_TCP){ // handshake finishes
- InputMessenger::OnNewMessages(m);
+ InputMessenger::OnNewMessages(s);
return;
} else if (state == ESTABLISHED) {
- uint8_t tmp;
- ssize_t nr = read(ep->_socket->fd(), &tmp, 1);
- if (nr == 0) {
- ep->_socket->SetEOF();
- return;
- }
- if (nr > 0) {
- LOG(WARNING) << "Read unexpected data from " << ep->_socket;
- ep->_socket->SetFailed(EPROTO, "Read unexpected data from %s",
- ep->_socket->description().c_str());
+ if (!ep->HandleTcpEventAfterEstablished()) {
return;
}
+ }
+ if (!s->MoreReadEvents(&progress)) {
+ break;
+ }
+ }
+}
- if (errno != EAGAIN) {
+void RdmaEndpoint::OnNewDataFromTcpAtServer(Socket* _socket) {
+ auto* rdma_transport =
static_cast<RdmaTransport*>(_socket->_transport.get());
+ RdmaEndpoint* ep = rdma_transport->GetRdmaEp();
Review Comment:
The parameter name `_socket` shadows/visually conflicts with the member
`RdmaEndpoint::_socket`, and later code uses both `_socket` and `ep->_socket`,
which is easy to misread and increases the chance of mistakes. Rename the
parameter to `s`/`sock`/`tcp_socket` to clearly distinguish it from the
endpoint member.
##########
src/brpc/input_messenger_processor.cpp:
##########
@@ -0,0 +1,291 @@
+// 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 "butil/logging.h"
+#include "butil/binary_printer.h"
+#include "bthread/unstable.h"
+#include "brpc/options.pb.h"
+#include "brpc/transport.h"
+#include "brpc/input_messenger_processor.h"
+#include "brpc/input_messenger.h"
+
+namespace brpc {
+
+DECLARE_uint64(max_body_size);
+
+const size_t MSG_SIZE_WINDOW = 10; // Take last so many message into stat.
+const size_t MIN_ONCE_READ = 4096;
+const size_t MAX_ONCE_READ = 524288;
+
+static const char* StreamTypeName(InputMessengerProcessor::StreamType type) {
+ switch (type) {
+ case InputMessengerProcessor::STREAM_NONE: return "none";
+ case InputMessengerProcessor::STREAM_TCP_FD: return "tcp_fd";
+ case InputMessengerProcessor::STREAM_RDMA_QP: return "rdma_qp";
+ }
+ return "unknown";
+}
+
+InputMessengerProcessor::ParsingStreamGuard::ParsingStreamGuard(Socket*
socket, StreamType type)
+ : _socket(socket) {
+ CHECK_NE(STREAM_NONE, type)
+ << "Parsing through a processor that was never Init()ed, " << *socket;
+ CHECK_EQ(STREAM_NONE, socket->parsing_stream_type())
+ << "Two input streams of " << *socket << " are parsing at the same
time: "
+ << StreamTypeName(socket->parsing_stream_type()) << " and "
+ << StreamTypeName(type);
Review Comment:
If `ProcessNewMessage()` is ever called before `Init()`, `socket` will be
null and the first failing `CHECK` streams `*socket`, which will dereference
null and crash before printing the diagnostic. To make the guard robust, add an
explicit `CHECK(socket != nullptr)` (with a message that does not dereference
it) before using `*socket`, and ideally also validate the processor was
initialized (e.g., `_stream_type != STREAM_NONE`).
##########
test/brpc_rdma_unittest.cpp:
##########
@@ -719,6 +730,234 @@ TEST_F(RdmaTest, client_send_data_on_tcp_after_ack_send) {
StopServer();
}
+// Build a well-formed v2 client hello: "RDMA" followed by the 36B body.
+static void MakeV2ClientHello(uint8_t (&data)[rdma::HELLO_V2_MSG_LEN_MIN]) {
+ rdma::v2_wire::HelloMessage msg{};
+ msg.msg_len = rdma::HELLO_V2_MSG_LEN_MIN;
+ msg.hello_ver = rdma::HELLO_V2_VERSION;
+ msg.impl_ver = rdma::IMPL_V2_VERSION;
+ msg.sq_size = 16;
+ msg.rq_size = 16;
+ msg.block_size = 8192;
+ msg.qp_num = 0;
+ msg.gid = rdma::GetRdmaGid();
+ memcpy(data, "RDMA", 4);
+ msg.Serialize(data + 4);
+}
+
+// Connect, push a well-formed v2 hello and read back the server's reply, which
+// leaves the server in S_ACK_WAIT waiting for the 4B ACK.
+static void HandshakeUntilAckWait(butil::fd_guard* sockfd) {
+ sockaddr_in addr;
+ bzero((char*)&addr, sizeof(addr));
+ addr.sin_family = AF_INET;
+ addr.sin_port = htons(PORT);
+ sockfd->reset(socket(AF_INET, SOCK_STREAM, 0));
+ ASSERT_TRUE(*sockfd >= 0);
+ ASSERT_EQ(0, connect(*sockfd, (sockaddr*)&addr, sizeof(sockaddr)));
+
+ uint8_t hello[rdma::HELLO_V2_MSG_LEN_MIN];
+ MakeV2ClientHello(hello);
+ ASSERT_EQ((ssize_t)sizeof(hello), write(*sockfd, hello, sizeof(hello)));
+ usleep(100000); // wait for server to handle the msg
+ uint8_t reply[rdma::HELLO_V2_MSG_LEN_MIN];
+ ASSERT_EQ((ssize_t)sizeof(reply), read(*sockfd, reply, sizeof(reply)));
+}
+
+// A client is free to pipeline its first request right behind the handshake
+// ACK. Only the 4B ACK belongs to the handshake. Whatever follows it must be
+// handed over to the real protocol instead of dropping the connection.
+TEST_F(RdmaTest, server_accepts_data_pipelined_behind_fallback_ack) {
+ StartServer();
+
+ butil::fd_guard sockfd;
+ ASSERT_NO_FATAL_FAILURE(HandshakeUntilAckWait(&sockfd));
+ Socket* s = GetSocketFromServer(0);
+ ASSERT_TRUE(s != nullptr);
+ auto* transport = static_cast<RdmaTransport*>(s->_transport.get());
+ ASSERT_EQ(rdma::RdmaEndpoint::S_ACK_WAIT, transport->_rdma_ep->_state);
+
+ // An ACK asking for TCP, plus the first 4 bytes of a baidu_std request.
One
+ // write, so that both end up in the same read on the server.
+ uint8_t ack_and_data[rdma::HELLO_ACK_LEN + 4];
+ const uint32_t flags = butil::HostToNet32(0);
+ memcpy(ack_and_data, &flags, rdma::HELLO_ACK_LEN);
+ memcpy(ack_and_data + rdma::HELLO_ACK_LEN, "PRPC", 4);
+ ASSERT_EQ((ssize_t)sizeof(ack_and_data),
+ write(sockfd, ack_and_data, sizeof(ack_and_data)));
+ usleep(100000); // wait for server to handle the msg
+
+ // The handshake took the ACK only and left "PRPC" to baidu_std, which is
+ // now waiting for the rest of its 12B header. So the connection lives on
+ // with those 4 bytes still buffered.
+ ASSERT_EQ(rdma::RdmaEndpoint::FALLBACK_TCP, transport->_rdma_ep->_state);
+ ASSERT_EQ(RdmaTransport::RDMA_OFF, transport->_rdma_state);
+ ASSERT_TRUE(GetSocketFromServer(0) != nullptr);
+ ASSERT_EQ(4u, s->fd_input_processor().read_buf().size());
+
+ sockfd.reset(-1);
+ usleep(100000); // wait for server to handle the msg
+ ASSERT_EQ(nullptr, GetSocketFromServer(0));
+
+ StopServer();
+}
+
+// Once RDMA is on, the TCP fd is no longer an RPC channel, so bytes trailing
+// the ACK can only be a protocol error.
+TEST_F(RdmaTest, server_rejects_data_pipelined_behind_rdma_ack) {
+ StartServer();
+
+ butil::fd_guard sockfd;
+ ASSERT_NO_FATAL_FAILURE(HandshakeUntilAckWait(&sockfd));
+ Socket* s = GetSocketFromServer(0);
+ ASSERT_TRUE(s != nullptr);
+ auto* transport = static_cast<RdmaTransport*>(s->_transport.get());
+ ASSERT_EQ(rdma::RdmaEndpoint::S_ACK_WAIT, transport->_rdma_ep->_state);
+
+ uint8_t ack_and_data[rdma::HELLO_ACK_LEN + 4];
+ const uint32_t flags = butil::HostToNet32(rdma::HELLO_ACK_RDMA_OK);
+ memcpy(ack_and_data, &flags, rdma::HELLO_ACK_LEN);
+ memcpy(ack_and_data + rdma::HELLO_ACK_LEN, "PRPC", 4);
+ ASSERT_EQ((ssize_t)sizeof(ack_and_data),
+ write(sockfd, ack_and_data, sizeof(ack_and_data)));
+ usleep(100000); // wait for server to handle the msg
Review Comment:
These fixed `usleep(100000)` waits make the unit tests timing-dependent and
prone to flakes under load/slow CI. Prefer polling with a bounded timeout for
the expected state transition (e.g., wait until `transport->_rdma_ep->_state`
reaches the target) or use an explicit synchronization signal from the server
side when possible.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
