Copilot commented on code in PR #3626:
URL: https://github.com/apache/brpc/pull/3626#discussion_r4335139402
##########
src/brpc/urma/urma_endpoint.cpp:
##########
@@ -952,28 +952,24 @@ ssize_t UrmaEndpoint::HandleCompletion(const urma_cr_t&
cr) {
uint16_t wnd = 1; // We signal every WR (complete_enable=1).
uint16_t old =
_sq_window_size.load(butil::memory_order_relaxed);
- while (true) {
- if (old >= _local_window_capacity) {
- LOG(WARNING)
- << "URMA send completion exceeds SQ window: old=" << old
- << " increment=" << wnd
- << " capacity=" << _local_window_capacity
- << " user_ctx=" << cr.user_ctx
- << " on " << _socket->description();
- errno = EPROTO;
- return -1;
- }
- if (_sq_window_size.compare_exchange_weak(
- old, static_cast<uint16_t>(old + wnd),
- butil::memory_order_relaxed)) {
- break;
- }
+ if (old >= _local_window_capacity) {
Review Comment:
The capacity guard compares `old` against capacity, but the log message
includes an `increment` and the code is structured as if `wnd` could vary. If
`wnd` ever becomes > 1, `old < capacity` is not sufficient (it should reject
when `old + wnd > capacity`, taking care to avoid uint16 overflow during the
addition). Even if `wnd` is currently always 1, tightening the check makes the
code robust to future changes and aligns the validation with the logged fields.
##########
test/brpc_urma_unittest.cpp:
##########
@@ -41,6 +43,91 @@ extern butil::atomic<bool> g_urma_available;
} // namespace urma
} // namespace brpc
+class UrmaSendCompletionTest : public ::testing::Test {
+protected:
+ void SetUp() override {
+ SocketOptions options;
+ SocketId id;
+ ASSERT_EQ(0, Socket::Create(options, &id));
+ ASSERT_EQ(0, Socket::Address(id, &_socket));
+ }
+
+ void TearDown() override {
+ if (_socket) {
+ _socket->SetFailed();
+ }
+ }
+
+ static urma_cr_t SendCompletion() {
+ urma_cr_t cr{};
+ cr.status = URMA_CR_SUCCESS;
+ cr.user_ctx = 1;
+ return cr;
+ }
+
+ SocketUniquePtr _socket;
+};
+
+TEST_F(UrmaSendCompletionTest, reclaim_buffer_before_publishing_credit) {
+ char payload = 'x';
+ bool released = false;
+ urma::UrmaEndpoint endpoint(_socket.get());
+ endpoint._sbuf.resize(endpoint._sq_size -
+ urma::UrmaEndpoint::RESERVED_WR_NUM);
+ endpoint._local_window_capacity = endpoint._sbuf.size();
+ endpoint._sq_window_size.store(0, butil::memory_order_relaxed);
Review Comment:
These tests directly read/write `UrmaEndpoint` internals (e.g., `_sbuf`,
`_sq_size`, `_local_window_capacity`, `_sq_window_size`, `_sq_sent`). If this
required making previously-private members accessible, it weakens encapsulation
and makes it harder to change `UrmaEndpoint` safely. Prefer adding a small,
test-only hook (e.g., a `FRIEND_TEST` and narrowly scoped accessors/mutators)
or a dedicated test fixture/helper inside the URMA module that can drive the
state without exposing these fields broadly.
##########
test/brpc_urma_unittest.cpp:
##########
@@ -41,6 +43,91 @@ extern butil::atomic<bool> g_urma_available;
} // namespace urma
} // namespace brpc
+class UrmaSendCompletionTest : public ::testing::Test {
+protected:
+ void SetUp() override {
+ SocketOptions options;
+ SocketId id;
+ ASSERT_EQ(0, Socket::Create(options, &id));
+ ASSERT_EQ(0, Socket::Address(id, &_socket));
+ }
+
+ void TearDown() override {
+ if (_socket) {
+ _socket->SetFailed();
+ }
+ }
+
+ static urma_cr_t SendCompletion() {
+ urma_cr_t cr{};
+ cr.status = URMA_CR_SUCCESS;
+ cr.user_ctx = 1;
+ return cr;
+ }
+
+ SocketUniquePtr _socket;
+};
+
+TEST_F(UrmaSendCompletionTest, reclaim_buffer_before_publishing_credit) {
+ char payload = 'x';
+ bool released = false;
+ urma::UrmaEndpoint endpoint(_socket.get());
+ endpoint._sbuf.resize(endpoint._sq_size -
+ urma::UrmaEndpoint::RESERVED_WR_NUM);
+ endpoint._local_window_capacity = endpoint._sbuf.size();
+ endpoint._sq_window_size.store(0, butil::memory_order_relaxed);
+ ASSERT_EQ(0, endpoint._sbuf[0].append_user_data(
+ &payload, sizeof(payload), [&](void*) {
+ released = true;
+ // Observe the credit at the exact point the old buffer is freed.
+ // Publishing first lets the writer reuse a slot still being
cleared.
+ EXPECT_EQ(0, endpoint._sq_window_size.load(
+ butil::memory_order_relaxed));
+ }));
+
+ ASSERT_EQ(0, endpoint.HandleCompletion(SendCompletion()));
+ EXPECT_TRUE(released);
+ EXPECT_TRUE(endpoint._sbuf[0].empty());
+ EXPECT_EQ(1, endpoint._sq_sent);
+ EXPECT_EQ(1, endpoint._sq_window_size.load(butil::memory_order_relaxed));
+}
+
+TEST_F(UrmaSendCompletionTest, reclaim_last_slot_and_wrap_cursor) {
+ urma::UrmaEndpoint endpoint(_socket.get());
+ endpoint._sbuf.resize(endpoint._sq_size -
+ urma::UrmaEndpoint::RESERVED_WR_NUM);
+ endpoint._local_window_capacity = endpoint._sbuf.size();
+ endpoint._sq_sent = endpoint._sbuf.size() - 1;
+ const uint16_t old_credit = endpoint._local_window_capacity - 1;
+ endpoint._sq_window_size.store(old_credit, butil::memory_order_relaxed);
+ ASSERT_EQ(0, endpoint._sbuf[endpoint._sq_sent].append("last"));
+ ASSERT_EQ(0, endpoint._sbuf[0].append("next"));
+
+ ASSERT_EQ(0, endpoint.HandleCompletion(SendCompletion()));
+ EXPECT_EQ(0, endpoint._sq_sent);
+ EXPECT_TRUE(endpoint._sbuf.back().empty());
+ EXPECT_EQ("next", endpoint._sbuf[0].to_string());
+ EXPECT_EQ(endpoint._local_window_capacity,
+ endpoint._sq_window_size.load(butil::memory_order_relaxed));
+}
+
+TEST_F(UrmaSendCompletionTest, reject_full_window_before_reclaiming_buffer) {
+ urma::UrmaEndpoint endpoint(_socket.get());
+ endpoint._sbuf.resize(endpoint._sq_size -
+ urma::UrmaEndpoint::RESERVED_WR_NUM);
+ endpoint._local_window_capacity = endpoint._sbuf.size();
+ endpoint._sq_window_size.store(endpoint._local_window_capacity,
+ butil::memory_order_relaxed);
+ ASSERT_EQ(0, endpoint._sbuf[0].append("keep"));
+
+ EXPECT_EQ(-1, endpoint.HandleCompletion(SendCompletion()));
+ EXPECT_EQ(EPROTO, errno);
Review Comment:
To make the errno assertion more robust, explicitly set `errno = 0`
immediately before calling `HandleCompletion()`. This avoids a false pass/fail
if future refactors return `-1` without setting `errno` on some paths, or if
errno is changed by intervening code.
##########
src/brpc/urma/urma_endpoint.cpp:
##########
@@ -952,28 +952,24 @@ ssize_t UrmaEndpoint::HandleCompletion(const urma_cr_t&
cr) {
uint16_t wnd = 1; // We signal every WR (complete_enable=1).
uint16_t old =
_sq_window_size.load(butil::memory_order_relaxed);
- while (true) {
- if (old >= _local_window_capacity) {
- LOG(WARNING)
- << "URMA send completion exceeds SQ window: old=" << old
- << " increment=" << wnd
- << " capacity=" << _local_window_capacity
- << " user_ctx=" << cr.user_ctx
- << " on " << _socket->description();
- errno = EPROTO;
- return -1;
- }
- if (_sq_window_size.compare_exchange_weak(
- old, static_cast<uint16_t>(old + wnd),
- butil::memory_order_relaxed)) {
- break;
- }
+ if (old >= _local_window_capacity) {
+ LOG(WARNING)
+ << "URMA send completion exceeds SQ window: old=" << old
+ << " increment=" << wnd
+ << " capacity=" << _local_window_capacity
+ << " user_ctx=" << cr.user_ctx
+ << " on " << _socket->description();
+ errno = EPROTO;
+ return -1;
}
for (uint16_t i = 0; i < wnd; ++i) {
_sbuf[_sq_sent].clear();
_sq_sent = (_sq_sent + 1) % (_sq_size - RESERVED_WR_NUM);
}
butil::subtle::MemoryBarrier();
+ // Reclaim the buffer before publishing the SQ credit. Otherwise the
+ // writer may reuse this slot while the completion thread clears it.
+ _sq_window_size.fetch_add(wnd, butil::memory_order_relaxed);
Review Comment:
Changing the SQ credit update from a CAS loop to `load` + `fetch_add`
removes the atomic capacity check. If `HandleCompletion()` can be invoked
concurrently, multiple threads may all observe `old < _local_window_capacity`
and then each `fetch_add`, allowing `_sq_window_size` to exceed capacity
without triggering the EPROTO path. To preserve the capacity invariant under
concurrency, use an atomic update that checks-and-updates as one operation
(e.g., restore a compare-exchange loop that validates `old + wnd <= capacity`),
or otherwise validate based on the returned value from the atomic op in a way
that remains correct under concurrent updates.
--
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]
