Copilot commented on code in PR #3626:
URL: https://github.com/apache/brpc/pull/3626#discussion_r4355504203
##########
test/brpc_urma_unittest.cpp:
##########
@@ -41,6 +43,92 @@ 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();
Review Comment:
The repeated test setup for sizing `_sbuf` and assigning
`_local_window_capacity` is duplicated across multiple tests. To reduce
repetition and make future adjustments less error-prone, consider factoring
this into a small helper (e.g., a fixture method) that initializes the
endpoint’s SQ buffer/window fields consistently.
##########
test/brpc_urma_unittest.cpp:
##########
@@ -41,6 +43,92 @@ 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();
Review Comment:
The repeated test setup for sizing `_sbuf` and assigning
`_local_window_capacity` is duplicated across multiple tests. To reduce
repetition and make future adjustments less error-prone, consider factoring
this into a small helper (e.g., a fixture method) that initializes the
endpoint’s SQ buffer/window fields consistently.
##########
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:
The capacity check is no longer atomic with the credit increment: `old` is
loaded once and checked, but `_sq_window_size.fetch_add()` can still push the
value beyond `_local_window_capacity` if multiple completion handlers run
concurrently (each can observe `old < capacity` and then increment). The
previous CAS loop prevented this by re-checking on contention. Consider
restoring a CAS loop, or using `fetch_add`’s returned prior value to validate
bounds and undo/report if it would exceed capacity.
##########
test/brpc_urma_unittest.cpp:
##########
@@ -41,6 +43,92 @@ 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();
Review Comment:
The repeated test setup for sizing `_sbuf` and assigning
`_local_window_capacity` is duplicated across multiple tests. To reduce
repetition and make future adjustments less error-prone, consider factoring
this into a small helper (e.g., a fixture method) that initializes the
endpoint’s SQ buffer/window fields consistently.
--
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]
