This is an automated email from the ASF dual-hosted git repository.
wwbmmm pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/brpc.git
The following commit(s) were added to refs/heads/master by this push:
new 217f85edd Fix socket overcrowding state race (#3621)
217f85edd is described below
commit 217f85edd471b7ceff9fd878d87f1cca8727bc44
Author: nkysg <[email protected]>
AuthorDate: Sun Oct 25 17:50:15 2026 +0800
Fix socket overcrowding state race (#3621)
* Fix socket overcrowding state race
* Clarify socket overcrowding limit behavior
* Fix socket overcrowding flag consistency
---
src/brpc/socket.cpp | 36 ++++++++++++++--------------
src/brpc/socket.h | 5 +---
test/brpc_http_rpc_protocol_unittest.cpp | 40 +++++++++++++++++++++++++++++++-
3 files changed, 57 insertions(+), 24 deletions(-)
diff --git a/src/brpc/socket.cpp b/src/brpc/socket.cpp
index 57cfd2296..7756c9c3b 100644
--- a/src/brpc/socket.cpp
+++ b/src/brpc/socket.cpp
@@ -89,8 +89,8 @@ DEFINE_int32(ssl_handshake_timeout_ms, 5000,
BRPC_VALIDATE_GFLAG(ssl_handshake_timeout_ms, PassValidate);
DEFINE_int64(socket_max_unwritten_bytes, 64 * 1024 * 1024,
- "Max unwritten bytes in each socket, if the limit is reached,"
- " Socket.Write fails with EOVERCROWDED");
+ "Max unwritten bytes in each socket. Socket.Write fails with "
+ "EOVERCROWDED when the limit is reached.");
DEFINE_int64(socket_max_streams_unconsumed_bytes, 0,
"Max stream receivers' unconsumed bytes in one socket,"
@@ -399,11 +399,7 @@ void Socket::WriteRequest::Setup(Socket* s) {
return;
}
}
- const int64_t before_write =
- s->_unwritten_bytes.fetch_add(data.size(),
butil::memory_order_relaxed);
- if (before_write + (int64_t)data.size() >=
FLAGS_socket_max_unwritten_bytes) {
- s->_overcrowded = true;
- }
+ s->_unwritten_bytes.fetch_add(data.size(),
butil::memory_order_relaxed);
}
const uint32_t pc = pipelined_count();
if (pc) {
@@ -480,7 +476,6 @@ Socket::Socket(Forbidden f)
, _socket_mode(SOCKET_MODE_TCP)
, _connection_type_for_progressive_read(CONNECTION_TYPE_UNKNOWN)
, _controller_released_socket(false)
- , _overcrowded(false)
, _fail_me_at_server_stop(false)
, _logoff_flag(false)
, _error_code(0)
@@ -776,7 +771,6 @@ int Socket::OnCreated(const SocketOptions& options) {
_ssl_ctx = options.initial_ssl_ctx;
_connection_type_for_progressive_read = CONNECTION_TYPE_UNKNOWN;
_controller_released_socket.store(false, butil::memory_order_relaxed);
- _overcrowded = false;
// Maybe non-zero for RTMP connections.
_fail_me_at_server_stop = false;
_logoff_flag.store(false, butil::memory_order_relaxed);
@@ -1663,7 +1657,7 @@ int Socket::Write(butil::IOBuf* data, const WriteOptions*
options_in) {
}
}
- if (!opt.ignore_eovercrowded && _overcrowded) {
+ if (!opt.ignore_eovercrowded && is_overcrowded()) {
return SetError(opt.id_wait, EOVERCROWDED);
}
@@ -1701,7 +1695,7 @@ int Socket::Write(SocketMessagePtr<>& msg, const
WriteOptions* options_in) {
}
}
- if (!opt.ignore_eovercrowded && _overcrowded) {
+ if (!opt.ignore_eovercrowded && is_overcrowded()) {
return SetError(opt.id_wait, EOVERCROWDED);
}
@@ -1878,8 +1872,9 @@ void* Socket::KeepWrite(void* void_arg) {
//if (nw <= 0 || req->data.empty()/*note*/) {
if (nw <= 0) {
// NOTE: Waiting epollout within timeout is a must to force
- // KeepWrite to check and setup pending WriteRequests periodically,
- // which may turn on _overcrowded to stop pending requests from
+ // KeepWrite to check and setup pending WriteRequests periodically
+ // so their bytes contribute to _unwritten_bytes. Subsequent
+ // writes then observe the limit and stop the pending queue from
// growing infinitely.
const timespec duetime =
butil::milliseconds_from_now(WAIT_EPOLLOUT_TIMEOUT_MS);
bool pollin = s->_transport->HasOnEdgeTrigger();
@@ -2418,7 +2413,7 @@ void Socket::DebugSocket(std::ostream& os, SocketId id) {
<< "\nread_buf=" << input_processor.read_buf().size()
<< "\nlast_read_to_now=" << cpuwide_now - ptr->_last_readtime_us << "us"
<< "\nlast_write_to_now=" << cpuwide_now - ptr->_last_writetime_us <<
"us"
- << "\novercrowded=" << ptr->_overcrowded;
+ << "\novercrowded=" << ptr->is_overcrowded();
os << "\nid_wait_list={";
for (size_t i = 0; i < nidsize; ++i) {
if (i) {
@@ -3003,11 +2998,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 limit = FLAGS_socket_max_unwritten_bytes;
+ const int64_t unwritten_bytes =
+ _unwritten_bytes.load(butil::memory_order_relaxed);
+ return limit > 0 && unwritten_bytes >= limit;
}
void Socket::AddOutputBytes(size_t bytes) {
GetOrNewSharedPart()->out_size.fetch_add(bytes,
butil::memory_order_relaxed);
diff --git a/src/brpc/socket.h b/src/brpc/socket.h
index 94f85880d..4e651e075 100644
--- a/src/brpc/socket.h
+++ b/src/brpc/socket.h
@@ -712,7 +712,7 @@ public:
}
// Returns true if the remote side is overcrowded.
- bool is_overcrowded() const { return _overcrowded; }
+ bool is_overcrowded() const;
bthread_keytable_pool_t* keytable_pool() const { return _keytable_pool; }
@@ -1012,9 +1012,6 @@ private:
ConnectionType _connection_type_for_progressive_read;
butil::atomic<bool> _controller_released_socket;
- // True if the socket is too full to write.
- volatile bool _overcrowded;
-
bool _fail_me_at_server_stop;
// Set by SetLogOff
diff --git a/test/brpc_http_rpc_protocol_unittest.cpp
b/test/brpc_http_rpc_protocol_unittest.cpp
index fe02ba0a1..880caf06c 100644
--- a/test/brpc_http_rpc_protocol_unittest.cpp
+++ b/test/brpc_http_rpc_protocol_unittest.cpp
@@ -40,6 +40,7 @@
#include "butil/logging.h"
#include "butil/files/scoped_file.h"
#include "butil/fd_guard.h"
+#include "butil/fd_utility.h"
#include "butil/file_util.h"
#include "brpc/socket.h"
#include "brpc/acceptor.h"
@@ -2289,7 +2290,7 @@ TEST_F(HttpTest, h2_ping_ack_respects_socket_write_cap) {
brpc::Socket::WriteOptions wopt;
wopt.ignore_eovercrowded = false;
ASSERT_EQ(0, sock->Write(&big, &wopt));
- ASSERT_TRUE(sock->_overcrowded);
+ ASSERT_TRUE(sock->is_overcrowded());
brpc::policy::H2Context* ctx =
new brpc::policy::H2Context(sock.get(), nullptr);
@@ -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;
+
+ char buf[8192];
+ const int max_wait_iterations = 1000;
+ for (int i = 0;
+ i < max_wait_iterations && sock->is_overcrowded(); ++i) {
+ ssize_t nread;
+ while ((nread = read(read_fd, buf, sizeof(buf))) > 0) {}
+ ASSERT_EQ(-1, nread);
+ ASSERT_TRUE(errno == EAGAIN || errno == EWOULDBLOCK);
+ bthread_usleep(1000);
+ }
+ ASSERT_FALSE(sock->is_overcrowded());
+}
+
TEST_F(HttpTest, http2_goaway_with_debug_data) {
// GOAWAY payload is Last-Stream-ID(4) | Error Code(4) | Debug Data(*).
const uint32_t payload_size = 16;
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
