Copilot commented on code in PR #3480:
URL: https://github.com/apache/brpc/pull/3480#discussion_r3849013688
##########
src/brpc/input_messenger.cpp:
##########
@@ -369,7 +369,16 @@ void InputMessenger::OnNewMessages(Socket* m) {
if (messenger->ProcessNewMessage(m, nr, read_eof, received_us,
base_realtime, last_msg) < 0) {
return;
- }
+ }
+ // If the transport switched its edge trigger during parsing (e.g.,
+ // RDMA handshake completed and edge trigger changed to
+ // OnNewDataFromTcp), stop reading to avoid racing with the new
+ // edge trigger handler on _read_buf. Drain _nevent so future
+ // epoll events can schedule the new edge trigger handler.
+ if (m->_transport->ShouldStopReading()) {
+ while (m->MoreReadEvents(&progress)) {}
+ return;
+ }
Review Comment:
`OnNewMessages` can return early when `ShouldStopReading()` becomes true,
but if this iteration already detected EOF (`read_eof == true`), the early
return skips the `m->SetEOF()` at the end of the function. That can leave the
socket unmarked EOF even though the peer has closed the TCP fd.
##########
src/brpc/rdma/rdma_endpoint.cpp:
##########
@@ -633,16 +629,28 @@ ParseResult
RdmaEndpoint::ExecuteServerHandshake(butil::IOBuf* source, Socket* s
<< s->description();
ep->_state.store(FAILED, butil::memory_order_relaxed);
s->reset_parsing_context(nullptr);
+ rdma_transport->_on_edge_trigger =
rdma::RdmaEndpoint::OnNewDataFromTcp;
return MakeParseError(PARSE_ERROR_ABSOLUTELY_WRONG);
}
LOG_IF(INFO, FLAGS_rdma_trace_verbose)
<< "Server handshake ends (use rdma v" << ep->_handshake_version
<< ") on " << s->description();
rdma_transport->_rdma_state = RdmaTransport::RDMA_ON;
- ep->_state.store(ESTABLISHED, butil::memory_order_relaxed);
+ // Clear any residual TCP data so it cannot pollute the RDMA recv
+ // stream. HandleCompletion appends (not overwrites) to _read_buf,
+ // so leftover bytes would become a prefix to RDMA data and break
+ // parsing. This clear is safe because HandleCompletion only writes
+ // _read_buf after seeing ESTABLISHED (acquire), which is stored
+ // below (release) — strictly after this clear.
+ source->clear();
+ ep->_state.store(ESTABLISHED, butil::memory_order_release);
s->reset_parsing_context(nullptr);
- return MakeParseError(PARSE_ERROR_TRY_OTHERS);
+ rdma_transport->_on_edge_trigger = rdma::RdmaEndpoint::OnNewDataFromTcp;
+ // Return NOT_ENOUGH_DATA (not TRY_OTHERS) so that OnNewMessages stops
+ // processing _read_buf immediately, before PollCq starts writing RDMA
+ // data into _read_buf.
+ return MakeParseError(PARSE_ERROR_NOT_ENOUGH_DATA);
Review Comment:
After a successful RDMA handshake, this parser returns
`PARSE_ERROR_NOT_ENOUGH_DATA`, which causes `CutInputMessage` to keep the
RDMA-handshake handler as the socket's preferred parser. That can delay parsing
of the first real post-handshake message (especially if the first RDMA payload
is < `HELLO_MAGIC_LEN`) because the handshake parser will keep returning
NOT_ENOUGH_DATA until enough bytes accumulate to reject the magic. Clear the
preferred index when transitioning to ESTABLISHED so protocol detection starts
fresh for the actual payload protocol.
--
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]
