Fix RDMA server core dump on concurrent access to Socket::_read_buf

[chenBright]

What problem does this PR solve?

Issue Number: resolve #3479

Problem Summary:

What is changed and the side effects?

Changed:

1. One input stream, one buffer. New InputMessengerProcessor holds the
   state of a single input stream: its butil::IOPortal and the message-size statistics
   that size the next read. Socket::_read_buf, _last_msg_size and _avg_msg_size
   move into it, and InputMessenger::CutInputMessage() / ProcessNewMessage()
   become its methods. A Socket owns one for its fd, a RdmaEndpoint owns one for
   its QP. Socket::DoRead() now takes the destination IOPortal* instead of always
   filling _read_buf.

2. The server stops parsing the TCP fd once RDMA is on (scenario 1).
   RdmaTransport::Init() installs RdmaEndpoint::OnNewDataFromTcp for both sides
   rather than only for the client; it dispatches on Socket::CreatedByConnect().
   In ESTABLISHED the fd is only probed for EOF, never parsed. This is needed on
   top of the buffer split because preferred_index and parsing_context are
   still per-Socket, so two streams must not parse at once.

3. CQ events start after the handshake, not during it (scenarios 2).
   RdmaEndpoint::StartCqEvents() is split out of DoAllocateResources(). The
   server calls it from OnNewDataFromTcpAtServer() once OnNewMessages() has
   returned, the client from ProcessHandshakeAtClient(). No CQE is lost by
   deferring: BringUpQp() fills the RQ and DoAllocateResources() arms both CQs
   before the QP reaches RTS, and adding an already readable fd to an edge-triggered
   epoll reports it immediately.

4. The handshake ACK no longer swallows what follows it (scenario 3). Phase
   2 consumes exactly HELLO_ACK_LEN bytes and leaves the rest to the real
   protocol 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 the
   handshake.

Side effects:

• Performance effects:

• Breaking backward compatibility:

---

Check List:

• Please make sure your changes are compilable.
• When providing us with a new feature, it is best to add related tests.
• Please follow Contributor Covenant Code of Conduct.

[bzs1118]

Thx for your solution. I tested it in our environment and it works.

[Copilot]

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 InputMessengerProcessor to own per-input-stream buffering and message-size stats; Socket and RdmaEndpoint each 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.