Copilot commented on code in PR #3480:
URL: https://github.com/apache/brpc/pull/3480#discussion_r3842288894
##########
src/brpc/transport.h:
##########
@@ -51,6 +51,11 @@ class Transport {
virtual void QueueMessage(InputMessageClosure& input_msg, int*
num_bthread_created, bool last_msg) = 0;
virtual void Debug(std::ostream &os) = 0;
+ // Returns true if OnNewMessages should stop its read loop immediately
+ // (e.g., RDMA transport after handshake completes and edge trigger
+ // is switched to OnNewDataFromTcp). Default: never stop.
+ virtual bool ShouldStopReading() const { return false; }
Review Comment:
Adding a new virtual method to a base class changes the vtable and can break
ABI for any out-of-tree `Transport` implementations or users linking against a
prebuilt brpc binary. If ABI stability is a goal here, consider an alternative
that avoids vtable changes (e.g., a non-virtual capability queried via existing
hooks or a callback registered in `Socket`/`InputMessenger`), or explicitly
document/bump the ABI/soname for this release.
##########
src/brpc/rdma/rdma_endpoint.cpp:
##########
@@ -948,7 +970,7 @@ ssize_t RdmaEndpoint::HandleCompletion(ibv_wc& wc) {
if (wc.byte_len > 0) {
SendAck(1);
}
- return wc.byte_len;
+ return bytes_written;
Review Comment:
When a receive completion arrives before `ESTABLISHED`, the code drops the
payload but still acknowledges (`SendAck(1)`). This creates silent stream data
loss: the peer will consider data delivered/consumed, but the server discards
it, which can corrupt the application protocol after the handshake completes. A
safer behavior would be to treat this as a protocol violation and fail/close
the connection (or avoid acknowledging dropped bytes), or buffer the data
outside `_read_buf` until `ESTABLISHED` and only then deliver it.
##########
src/brpc/rdma_transport.cpp:
##########
@@ -70,10 +70,19 @@ int RdmaTransport::Reset(int32_t expected_nref) {
if (_rdma_ep) {
_rdma_ep->Reset();
_rdma_state = RDMA_UNKNOWN;
+ if (_socket->CreatedByConnect()) {
+ _on_edge_trigger = rdma::RdmaEndpoint::OnNewDataFromTcp;
+ } else {
+ _on_edge_trigger = InputMessenger::OnNewMessages;
+ }
}
return 0;
}
+bool RdmaTransport::ShouldStopReading() const {
+ return _rdma_state == RDMA_ON;
+}
Review Comment:
`ShouldStopReading()` is called from `InputMessenger::OnNewMessages()` and
reads `_rdma_state`. In this PR, `_rdma_state` is also written from
handshake/completion paths, so if `_rdma_state` is not atomic (or otherwise
synchronized), this introduces a data race. To make the stop condition
thread-safe and consistent with the rest of the memory-ordering work in this
PR, `_rdma_state` should be stored/loaded with proper synchronization (e.g.,
make it `std::atomic` and use at least `acquire` on reads / `release` on
transitions), or derive the stop condition from an already-atomic state
(`ep->_state`).
##########
src/brpc/rdma/rdma_endpoint.cpp:
##########
@@ -916,16 +924,30 @@ 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(WARNING) << "RDMA recv completion in non-ESTABLISHED state
"
+ << GetStateStr() << ", drop "
+ << wc.byte_len << " bytes from "
+ << _socket->description();
} else {
- // Copy data when the receive data is really small
- _socket->_read_buf.append(_rbuf_data[_rq_received],
wc.byte_len);
+ if (zerocopy) {
+ _rbuf[_rq_received].cutn(&_socket->_read_buf, wc.byte_len);
+ } else {
+ // Copy data when the receive data is really small
+ _socket->_read_buf.append(_rbuf_data[_rq_received],
wc.byte_len);
+ }
+ bytes_written = wc.byte_len;
}
Review Comment:
When a receive completion arrives before `ESTABLISHED`, the code drops the
payload but still acknowledges (`SendAck(1)`). This creates silent stream data
loss: the peer will consider data delivered/consumed, but the server discards
it, which can corrupt the application protocol after the handshake completes. A
safer behavior would be to treat this as a protocol violation and fail/close
the connection (or avoid acknowledging dropped bytes), or buffer the data
outside `_read_buf` until `ESTABLISHED` and only then deliver it.
##########
src/brpc/rdma/rdma_endpoint.cpp:
##########
@@ -916,16 +924,30 @@ 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(WARNING) << "RDMA recv completion in non-ESTABLISHED state
"
+ << GetStateStr() << ", drop "
+ << wc.byte_len << " bytes from "
+ << _socket->description();
Review Comment:
This `LOG(WARNING)` can become extremely noisy if a peer misbehaves or if
there’s a transient state window under load (it may log per completion).
Consider rate-limiting (e.g., `LOG_EVERY_N`, `LOG_FIRST_N`, or a throttled
logger) and/or gating via an existing verbosity flag to avoid flooding logs in
production.
##########
src/brpc/rdma/rdma_endpoint.cpp:
##########
@@ -916,16 +924,30 @@ 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(WARNING) << "RDMA recv completion in non-ESTABLISHED state
"
+ << GetStateStr() << ", drop "
+ << wc.byte_len << " bytes from "
+ << _socket->description();
} else {
- // Copy data when the receive data is really small
- _socket->_read_buf.append(_rbuf_data[_rq_received],
wc.byte_len);
+ if (zerocopy) {
+ _rbuf[_rq_received].cutn(&_socket->_read_buf, wc.byte_len);
+ } else {
+ // Copy data when the receive data is really small
+ _socket->_read_buf.append(_rbuf_data[_rq_received],
wc.byte_len);
+ }
+ bytes_written = wc.byte_len;
}
Review Comment:
Previously, the receive path always 'drained' the completion bytes from the
per-RQ buffer (`cutn` in the zero-copy case) or copied them out. With the new
drop path, the code no longer advances/drains any per-slot bookkeeping in the
zero-copy case when not `ESTABLISHED`. If `_rbuf[_rq_received]` relies on
`cutn()` (or an equivalent reset) to keep its internal offsets/state consistent
across reposts, skipping it may cause stale data to be re-exposed or internal
state to grow unexpectedly. Consider explicitly consuming/resetting the per-RQ
buffer state even when dropping (without touching `_socket->_read_buf`).
--
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]
