Fix RDMA server core dump on concurrent access to Socket::_read_buf - #3505
Fix RDMA server core dump on concurrent access to Socket::_read_buf#3505chenBright wants to merge 1 commit into
Conversation
5478b96 to
c4c0afb
Compare
|
Thx for your solution. I tested it in our environment and it works. |
There was a problem hiding this comment.
Pull request overview
Warning
Copilot couldn't run its full agentic review because it didn't start before the timeout. Make sure your repository has a runner available, or add a copilot-code-review.yml file specifying one with the runs-on attribute. See the docs for more details.
This PR addresses RDMA server crashes caused by concurrent parsing/access to Socket::_read_buf by splitting per-stream input buffering/parsing state and tightening handshake/CQ event ordering so only one stream parses through a given Socket context at a time.
Changes:
- Introduces
InputMessengerProcessorto own per-input-stream buffering and message-size stats;SocketandRdmaEndpointeach own a processor for their respective streams. - Updates RDMA handshake/CQ event startup so the server stops parsing the TCP fd once RDMA is established, and CQ polling begins only after the handshake completes.
- Adds/expands RDMA handshake and churn tests to cover pipelining behind ACK, per-stream separation, and repeated connection teardown scenarios.
Reviewed changes
Copilot reviewed 13 out of 13 changed files in this pull request and generated 9 comments.
Show a summary per file
| File | Description |
|---|---|
| test/brpc_rdma_unittest.cpp | Adds handshake pipeline tests, stream-separation checks, and repeated churn/E2E tests for RDMA/TCP mixes. |
| src/brpc/ubshm/ub_endpoint.cpp | Switches UBShm input buffering/parsing to Socket::fd_input_processor(). |
| src/brpc/socket.h | Adds InputMessengerProcessor ownership for fd stream and introduces parsing_stream_type tracking API. |
| src/brpc/socket.cpp | Initializes/uses fd input processor, updates DoRead to accept a destination buffer, and adapts debug output. |
| src/brpc/rdma_transport.cpp | Adjusts edge-trigger callback selection to route TCP events through RDMA endpoint when present. |
| src/brpc/rdma/rdma_handshake_server.cpp | Clarifies fallback-handshake behavior/comments around phase 2 consumption and context reset. |
| src/brpc/rdma/rdma_endpoint.h | Adds client/server TCP event split and a dedicated input processor for the QP stream; documents CQ event start constraints. |
| src/brpc/rdma/rdma_endpoint.cpp | Implements per-side TCP event handling, defers CQ event start, rejects unexpected bytes on TCP fd post-RDMA, and parses QP stream via endpoint processor. |
| src/brpc/input_messenger_processor.h | New: declares per-stream buffering/parsing/state holder used by Socket and RdmaEndpoint. |
| src/brpc/input_messenger_processor.cpp | New: moves CutInputMessage/ProcessNewMessage logic from InputMessenger into the processor and adds parsing-stream guard. |
| src/brpc/input_messenger.h | Makes InputMessengerProcessor a friend and removes now-moved private parsing methods. |
| src/brpc/input_messenger.cpp | Uses Socket::fd_input_processor() for reads/parsing and removes duplicated parsing logic. |
| src/brpc/input_message_base.h | Grants InputMessengerProcessor friend access for message dispatch fields. |
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
| 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 reply[rdma::HELLO_V2_MSG_LEN_MIN]; | ||
| ASSERT_EQ((ssize_t)sizeof(reply), read(*sockfd, reply, sizeof(reply))); |
| 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 |
| 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 |
| 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 |
| InputMessageClosure last_msg; | ||
| ASSERT_EQ(0, qp_stream.ProcessNewMessage(4, false, butil::gettimeofday_us(), 0, last_msg)); |
| CHECK_EQ(InputMessengerProcessor::STREAM_NONE, _socket->parsing_stream_type()) | ||
| << "StartCqEvents() called while " << *_socket << " is parsing"; |
| 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); |
| void RdmaEndpoint::OnNewDataFromTcpAtServer(Socket* _socket) { | ||
| auto* rdma_transport = static_cast<RdmaTransport*>(_socket->_transport.get()); | ||
| RdmaEndpoint* ep = rdma_transport->GetRdmaEp(); |
What problem does this PR solve?
Issue Number: resolve #3479
Problem Summary:
What is changed and the side effects?
Changed:
One input stream, one buffer. New
InputMessengerProcessorholds thestate of a single input stream: its
butil::IOPortaland the message-size statisticsthat size the next read.
Socket::_read_buf,_last_msg_sizeand_avg_msg_sizemove into it, and
InputMessenger::CutInputMessage()/ProcessNewMessage()become its methods. A
Socketowns one for its fd, aRdmaEndpointowns one forits QP.
Socket::DoRead()now takes the destinationIOPortal*instead of alwaysfilling
_read_buf.The server stops parsing the TCP fd once RDMA is on (scenario 1).
RdmaTransport::Init()installsRdmaEndpoint::OnNewDataFromTcpfor both sidesrather than only for the client; it dispatches on
Socket::CreatedByConnect().In
ESTABLISHEDthe fd is only probed for EOF, never parsed. This is needed ontop of the buffer split because
preferred_indexandparsing_contextarestill per-Socket, so two streams must not parse at once.
CQ events start after the handshake, not during it (scenarios 2).
RdmaEndpoint::StartCqEvents()is split out ofDoAllocateResources(). Theserver calls it from
OnNewDataFromTcpAtServer()onceOnNewMessages()hasreturned, the client from
ProcessHandshakeAtClient(). No CQE is lost bydeferring:
BringUpQp()fills the RQ andDoAllocateResources()arms both CQsbefore the QP reaches RTS, and adding an already readable fd to an edge-triggered
epoll reports it immediately.
The handshake ACK no longer swallows what follows it (scenario 3). Phase
2 consumes exactly
HELLO_ACK_LENbytes and leaves the rest to the realprotocol when the connection falls back to TCP. When RDMA is on the fd is not an
RPC channel any more, so bytes on it stay a protocol error, and a guard at the
top of
ExecuteServerHandshake()turns them away once the endpoint has left thehandshake.
Side effects:
Performance effects:
Breaking backward compatibility:
Check List: