Copilot commented on code in PR #3480:
URL: https://github.com/apache/brpc/pull/3480#discussion_r3840618109
##########
src/brpc/rdma/rdma_endpoint.cpp:
##########
@@ -920,7 +928,19 @@ ssize_t RdmaEndpoint::HandleCompletion(ibv_wc& wc) {
if (wc.byte_len < (uint32_t)FLAGS_rdma_zerocopy_min_size) {
zerocopy = false;
}
- CHECK_NE(_state.load(butil::memory_order_relaxed), FALLBACK_TCP);
+ // Don't write to _read_buf until the handshake is fully done
+ // (ESTABLISHED). During the handshake (S_ACK_WAIT etc.), the
+ // main socket's OnNewMessages is driving the handshake via
+ // _read_buf; PollCq writing to _read_buf concurrently corrupts
+ // the IOBuf (non-thread-safe).
+ if (_state.load(butil::memory_order_acquire) != ESTABLISHED) {
+ LOG(WARNING) << "RDMA recv completion in non-ESTABLISHED state
"
+ << GetStateStr() << ", drop "
+ << wc.byte_len << " bytes from "
+ << _socket->description();
+ PostRecv(1, zerocopy);
+ return 0;
+ }
Review Comment:
In the non-ESTABLISHED recv-completion path, PostRecv() failures are
ignored, and the early return skips SendAck(1). If the peer sends RDMA data
early (the scenario this change is explicitly handling), not acking the
reposted recv WR can stall the peer’s flow-control window; ignoring PostRecv
errors can also mask an unrecoverable transport failure.
##########
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 reaching ESTABLISHED, returning PARSE_ERROR_NOT_ENOUGH_DATA does not
stop InputMessenger::OnNewMessages from continuing its read loop (it will call
Socket::DoRead again, which mutates Socket::_read_buf). Since HandleCompletion
may start appending to _read_buf immediately after the ESTABLISHED store, there
is still a possible data race between PollCq and OnNewMessages during this
transition. Consider adding a mechanism to make OnNewMessages exit immediately
once the edge-trigger is switched (e.g., a check in OnNewMessages to break when
transport->GetOnEdgeTrigger() is no longer OnNewMessages, or another explicit
stop-reading signal) before publishing ESTABLISHED to PollCq.
--
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]
