Copilot commented on code in PR #3621:
URL: https://github.com/apache/brpc/pull/3621#discussion_r4348739323
##########
test/brpc_http_rpc_protocol_unittest.cpp:
##########
@@ -2319,6 +2320,43 @@ TEST_F(HttpTest, h2_ping_ack_respects_socket_write_cap) {
ASSERT_EQ(brpc::PARSE_ERROR_ABSOLUTELY_WRONG, pr.error());
}
+TEST_F(HttpTest, socket_overcrowded_state_tracks_pending_writes) {
+ int fds[2];
+ ASSERT_EQ(0, pipe(fds));
+ butil::fd_guard read_fd(fds[0]);
+ ASSERT_EQ(0, butil::make_non_blocking(read_fd));
+ brpc::SocketOptions options;
+ options.fd = fds[1];
+ brpc::SocketId id;
+ ASSERT_EQ(0, brpc::Socket::Create(options, &id));
+ brpc::SocketUniquePtr sock;
+ ASSERT_EQ(0, brpc::Socket::Address(id, &sock));
+
+ GFLAGS_NAMESPACE::FlagSaver flag_saver;
+ brpc::FLAGS_socket_max_unwritten_bytes = 64;
+
+ butil::IOBuf data;
+ data.append(std::string(1024 * 1024, 'x'));
Review Comment:
The test allocates/appends a 1MB `std::string` to trigger overcrowding with
a 64-byte limit. This is more memory/CPU than needed for the boundary behavior
under test and can slow CI. Consider reducing the payload to the smallest size
that reliably drives `_unwritten_bytes` past the threshold (e.g., a few
pipe-bufferfuls / a smaller fixed size).
##########
test/brpc_http_rpc_protocol_unittest.cpp:
##########
@@ -2319,6 +2320,43 @@ TEST_F(HttpTest, h2_ping_ack_respects_socket_write_cap) {
ASSERT_EQ(brpc::PARSE_ERROR_ABSOLUTELY_WRONG, pr.error());
}
+TEST_F(HttpTest, socket_overcrowded_state_tracks_pending_writes) {
+ int fds[2];
+ ASSERT_EQ(0, pipe(fds));
+ butil::fd_guard read_fd(fds[0]);
+ ASSERT_EQ(0, butil::make_non_blocking(read_fd));
+ brpc::SocketOptions options;
+ options.fd = fds[1];
+ brpc::SocketId id;
+ ASSERT_EQ(0, brpc::Socket::Create(options, &id));
Review Comment:
`fds[1]` is not guarded, so if an `ASSERT_*` fails before `Socket::Create`
takes ownership (or if ownership semantics change), the write-end FD can leak.
Wrap `fds[1]` in a `butil::fd_guard` as well, and explicitly `release()` it
only after successful socket creation if the socket assumes ownership.
##########
test/brpc_http_rpc_protocol_unittest.cpp:
##########
@@ -2319,6 +2320,43 @@ TEST_F(HttpTest, h2_ping_ack_respects_socket_write_cap) {
ASSERT_EQ(brpc::PARSE_ERROR_ABSOLUTELY_WRONG, pr.error());
}
+TEST_F(HttpTest, socket_overcrowded_state_tracks_pending_writes) {
+ int fds[2];
+ ASSERT_EQ(0, pipe(fds));
+ butil::fd_guard read_fd(fds[0]);
+ ASSERT_EQ(0, butil::make_non_blocking(read_fd));
+ brpc::SocketOptions options;
+ options.fd = fds[1];
+ brpc::SocketId id;
+ ASSERT_EQ(0, brpc::Socket::Create(options, &id));
+ brpc::SocketUniquePtr sock;
+ ASSERT_EQ(0, brpc::Socket::Address(id, &sock));
+
+ GFLAGS_NAMESPACE::FlagSaver flag_saver;
+ brpc::FLAGS_socket_max_unwritten_bytes = 64;
+
+ butil::IOBuf data;
+ data.append(std::string(1024 * 1024, 'x'));
+ ASSERT_EQ(0, sock->Write(&data));
+ ASSERT_TRUE(sock->is_overcrowded());
+
+ brpc::FLAGS_socket_max_unwritten_bytes = 0;
+ ASSERT_FALSE(sock->is_overcrowded());
+ brpc::FLAGS_socket_max_unwritten_bytes = 64;
Review Comment:
This test mutates a global flag while the socket may have active background
work (e.g., KeepWrite / transport threads). Since the flag variable is not an
atomic, concurrent reads/writes can be a data race and may also cause
cross-test interference in parallel runs. Prefer avoiding mid-test flag flips
on a live socket (e.g., structure the test to set the flag before
creating/using the socket, or use separate socket instances per flag setting;
if runtime toggling must be tested, add synchronization to ensure no concurrent
readers while updating the flag).
##########
src/brpc/socket.cpp:
##########
@@ -3003,11 +2999,14 @@ void Socket::AddInputMessages(size_t count) {
GetOrNewSharedPart()->in_num_messages.fetch_add(count,
butil::memory_order_relaxed);
}
void Socket::CancelUnwrittenBytes(size_t bytes) {
- const int64_t before_minus =
- _unwritten_bytes.fetch_sub(bytes, butil::memory_order_relaxed);
- if (before_minus < (int64_t)bytes + FLAGS_socket_max_unwritten_bytes) {
- _overcrowded = false;
- }
+ _unwritten_bytes.fetch_sub(bytes, butil::memory_order_relaxed);
+}
+
+bool Socket::is_overcrowded() const {
+ const int64_t unwritten_bytes =
+ _unwritten_bytes.load(butil::memory_order_relaxed);
+ return FLAGS_socket_max_unwritten_bytes > 0 &&
+ unwritten_bytes >= FLAGS_socket_max_unwritten_bytes;
Review Comment:
`FLAGS_socket_max_unwritten_bytes` is read twice, which can yield
inconsistent results if the flag is changed at runtime (e.g., first read > 0,
second read becomes 0), potentially making `is_overcrowded()` return an
incorrect value. Read the flag once into a local `const int64_t limit` and use
`limit` for both comparisons.
--
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]
