Copilot commented on code in PR #3422:
URL: https://github.com/apache/brpc/pull/3422#discussion_r3696129413
##########
src/brpc/stream.cpp:
##########
@@ -72,285 +68,239 @@ Stream::~Stream() {
CHECK(_host_socket == NULL);
bthread_mutex_destroy(&_connect_mutex);
bthread_mutex_destroy(&_congestion_control_mutex);
- bthread_id_list_destroy(&_writable_wait_list);
}
int Stream::Create(const StreamOptions &options,
- const StreamSettings *remote_settings,
+ const StreamSettings* remote_settings,
StreamId *id, bool parse_rpc_response) {
- Stream* s = new Stream();
- s->_host_socket = NULL;
- s->_fake_socket_weak_ref = NULL;
- s->_connected = false;
- s->_options = options;
- s->_closed = false;
- s->_error_code = 0;
- s->_cur_buf_size = options.max_buf_size > 0 ? options.max_buf_size : 0;
+ return VersionedRefWithId<Stream>::Create(
+ id, options, remote_settings, parse_rpc_response);
+}
+
+int Stream::OnCreated(const StreamOptions& options,
+ const StreamSettings* remote_settings,
+ bool parse_rpc_response) {
+ _host_socket = NULL;
+ _connected.store(false, butil::memory_order_relaxed);
+ _options = options;
+ _error_code = 0;
+ _error_text.clear();
+ _pending_writes.clear();
+ _produced = 0;
+ _remote_consumed = 0;
+ _local_consumed = 0;
+ _atomic_local_consumed.store(0, butil::memory_order_relaxed);
+ _parse_rpc_response = parse_rpc_response;
+ _pending_buf = NULL;
+ _start_idle_timer_us = 0;
+ _idle_timer = 0;
+ _remote_settings.Clear();
+
+ _cur_buf_size = options.max_buf_size > 0 ? options.max_buf_size : 0;
if (options.max_buf_size > 0 && options.min_buf_size >
options.max_buf_size) {
// set 0 if min_buf_size is invalid.
- s->_options.min_buf_size = 0;
+ _options.min_buf_size = 0;
LOG(WARNING) << "options.min_buf_size is larger than
options.max_buf_size, it will be set to 0.";
}
- if (FLAGS_socket_max_streams_unconsumed_bytes > 0 &&
s->_options.min_buf_size > 0) {
- s->_cur_buf_size = s->_options.min_buf_size;
+ if (FLAGS_socket_max_streams_unconsumed_bytes > 0 && _options.min_buf_size
> 0) {
+ _cur_buf_size = _options.min_buf_size;
}
if (remote_settings != NULL) {
- s->_remote_settings.MergeFrom(*remote_settings);
- }
- s->_parse_rpc_response = parse_rpc_response;
- if (bthread_id_list_init(&s->_writable_wait_list, 8, 8/*FIXME*/)) {
- delete s;
- return -1;
+ _remote_settings.MergeFrom(*remote_settings);
}
+
+ CHECK_EQ(0, bthread_id_list_init(&_writable_wait_list, 8, 8/*FIXME*/));
+
bthread::ExecutionQueueOptions q_opt;
q_opt.bthread_attr
= FLAGS_usercode_in_pthread ? BTHREAD_ATTR_PTHREAD :
BTHREAD_ATTR_NORMAL;
- if (bthread::execution_queue_start(&s->_consumer_queue, &q_opt, Consume,
s) != 0) {
+ if (bthread::execution_queue_start(&_consumer_queue, &q_opt, Consume,
this) != 0) {
LOG(FATAL) << "Fail to create ExecutionQueue";
- delete s;
- return -1;
- }
- SocketOptions sock_opt;
- sock_opt.conn = s;
- SocketId fake_sock_id;
- if (Socket::Create(sock_opt, &fake_sock_id) != 0) {
- s->BeforeRecycle(NULL);
return -1;
}
- SocketUniquePtr ptr;
- CHECK_EQ(0, Socket::Address(fake_sock_id, &ptr));
- s->_fake_socket_weak_ref = ptr.get();
- s->_id = fake_sock_id;
- *id = s->id();
+
+ // The consumer queue holds one reference to this Stream.
+ AddReference();
return 0;
}
-void Stream::BeforeRecycle(Socket *) {
- // No one holds reference now, so we don't need lock here
- bthread_id_list_reset(&_writable_wait_list, ECONNRESET);
- if (_connected) {
- // Send CLOSE frame
- RPC_VLOG << "Send close frame";
- CHECK(_host_socket != NULL);
- policy::SendStreamClose(_host_socket,
- _remote_settings.stream_id(), id());
+void Stream::OnFailed(int error_code, const std::string& error_text) {
+ bool connected = false;
+ {
+ // Record the error for on_failed callback fired in Consume(), and
discard
+ // any writes buffered before connecting.
+ BAIDU_SCOPED_LOCK(_connect_mutex);
+ _error_code = error_code;
+ _error_text = error_text;
+ connected = _connected.load(butil::memory_order_relaxed);
+ _pending_writes.clear();
}
- if (_host_socket) {
- _host_socket->RemoveStream(id());
+ // Wake up all threads blocked on writable.
+ bthread_id_list_reset(&_writable_wait_list, ECONNRESET);
+
+ // Serialize the host Socket membership removal with SetHostSocket().
+ // SetFailed() marks this Stream failed before entering OnFailed(), so a
+ // later SetHostSocket() observes Failed() and cannot add it back.
+ {
+ BAIDU_SCOPED_LOCK(_connect_mutex);
+ if (connected) {
+ RPC_VLOG << "Send close frame";
+ CHECK(_host_socket != NULL);
+ policy::SendStreamClose(
+ _host_socket, _remote_settings.stream_id(), id());
+ }
+ if (_host_socket != NULL) {
+ _host_socket->RemoveStream(id());
+ }
}
Review Comment:
When a stream is SetFailed/closed while it still has outstanding unconsumed
bytes, `_host_socket->_total_streams_unconsumed_size` is never decremented for
this stream. Because later FEEDBACK frames won’t be processed (Stream::Address
fails once the stream is failed), the socket-level total can remain
artificially high and skew `FLAGS_socket_max_streams_unconsumed_bytes`
congestion behavior for other streams on the same socket.
##########
test/brpc_streaming_rpc_unittest.cpp:
##########
@@ -866,11 +850,13 @@ TEST_F(StreamingRpcTest,
segment_stream_data_automatically) {
brpc::SocketUniquePtr host_socket_ptr;
{
- brpc::SocketUniquePtr ptr;
- ASSERT_EQ(0, brpc::Socket::Address(request_stream, &ptr));
- brpc::Stream *s = (brpc::Stream *)ptr->conn();
- ASSERT_TRUE(s->_host_socket != NULL);
- s->_host_socket->ReAddress(&host_socket_ptr);
+ brpc::StreamUniquePtr ptr;
+ ASSERT_EQ(0, brpc::Stream::Address(request_stream, &ptr));
+ brpc::Stream *s = ptr.get();
+ brpc::Socket* host_socket =
+ s->_host_socket;
+ ASSERT_TRUE(host_socket != NULL);
Review Comment:
This test reads the non-atomic `s->_host_socket` without first performing
the acquire load on `_connected` that the other updated tests rely on to safely
publish `_host_socket`. This can become a data race if connection establishment
is still in flight when this block runs.
##########
src/brpc/stream.cpp:
##########
@@ -42,23 +42,19 @@ BRPC_VALIDATE_GFLAG(stream_write_max_segment_size,
PositiveInteger);
const static butil::IOBuf *TIMEOUT_TASK = (butil::IOBuf*)-1L;
-Stream::Stream()
- : _host_socket(NULL)
- , _fake_socket_weak_ref(NULL)
+Stream::Stream(Forbidden f)
+ : VersionedRefWithId<Stream>(f)
+ , _host_socket(NULL)
, _connected(false)
- , _closed(false)
, _error_code(0)
Review Comment:
`_pending_buf` is used in the destructor and in `BeforeRecycled()`, but it
is no longer initialized in the constructor initializer list. Today
`OnCreated()` sets it to NULL before any failure point, but leaving it
uninitialized makes the class fragile (e.g., if future changes add an
early-return before `_pending_buf` is set, recycling could delete an
indeterminate pointer).
--
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]
