chenBright commented on code in PR #3480:
URL: https://github.com/apache/brpc/pull/3480#discussion_r3860301246
##########
src/brpc/input_messenger.cpp:
##########
@@ -369,7 +376,19 @@ 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)) {}
+ if (read_eof) {
+ m->SetEOF();
+ }
+ return;
+ }
Review Comment:
- `ShouldStopReading()` is more like a patch than an abstraction, which
makes this hard to maintain going forward.
- "stop the read loop immediately" leaves undefined whether the caller
must drain _nevent, call `SetEOF()`, flush `last_msg`, or skip the
`_read_buf.length()` read.
- Introducing any new state later (a reconnect/migration state, or an
intermediate one between the ACK and ESTABLISHED) means remembering to update
the gate in HandleCompletion, the condition here, and every terminal path that
assigns _on_edge_trigger. The five scattered _on_edge_trigger assignments are
already easy to forget: omitting one on a newly added terminal path silently
reintroduces the race.
- Draining `_nevent` discards events under EPOLLET, the server never learns
the peer is gone. .
##########
src/brpc/rdma/rdma_endpoint.cpp:
##########
@@ -916,16 +929,31 @@ ssize_t RdmaEndpoint::HandleCompletion(ibv_wc& wc) {
}
case IBV_WC_RECV: { // recv completion
// Please note that only the first wc.byte_len bytes is valid
+ ssize_t bytes_written = 0;
if (wc.byte_len > 0) {
if (wc.byte_len < (uint32_t)FLAGS_rdma_zerocopy_min_size) {
zerocopy = false;
}
- CHECK_NE(_state.load(butil::memory_order_relaxed), FALLBACK_TCP);
- if (zerocopy) {
- _rbuf[_rq_received].cutn(&_socket->_read_buf, wc.byte_len);
+ // 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). Fall through to handle imm
+ // data, re-post recv WR, and send ack normally.
+ if (_state.load(butil::memory_order_acquire) != ESTABLISHED) {
+ LOG_EVERY_N(WARNING, 100)
+ << "RDMA recv completion in non-ESTABLISHED state "
+ << GetStateStr() << ", drop "
+ << wc.byte_len << " bytes from "
+ << _socket->description();
Review Comment:
Dropping RDMA payload is not acceptable.
--
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]
