You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@brpc.apache.org by "chenBright (via GitHub)" <gi...@apache.org> on 2026/08/01 16:54:53 UTC
[PR] Refactor streaming rpc (brpc)
chenBright opened a new pull request, #3422:
URL: https://github.com/apache/brpc/pull/3422
### What problem does this PR solve?
Issue Number: resolve N/A
Problem Summary:
The previous Stream implementation created a fake `Socket` for every Stream to reuse `SocketId`, reference counting, object lookup, and the Socket write queue.
This introduced unnecessary coupling and an inefficient write path:
```text
StreamWrite -> fake Socket queue -> Stream frame packing -> host Socket queue -> network
```
As a result, each Stream message passed through two Socket queues and could require two bthread scheduling operations before being sent. It also made Stream lifecycle management depend on `SocketConnection` and fake-Socket-specific behavior, increasing the complexity of failure handling, reference management, and object recycling.
Since `VersionedRefWithId` already provides versioned IDs, O(1) lookup, reference counting, and deferred recycling, creating a fake Socket solely for these capabilities is unnecessary.
### What is changed and the side effects?
Changed:
- Refactored `Stream` to inherit directly from `VersionedRefWithId<Stream>`.
- Removed the fake Socket and the `SocketConnection` dependency from Stream.
- Simplified the write path to pack a complete Stream message and enqueue it directly to the real host Socket:
```text
StreamWrite -> Stream frame packing -> host Socket queue -> network
```
Side effects:
- Performance effects:
- Removes one queue traversal from every Stream write.
- Avoids the potential extra bthread scheduling operation introduced by the fake Socket.
- Reduces the Stream lookup path from `StreamId -> fake Socket -> Stream` to `StreamId -> Stream`.
- Breaking backward compatibility:
---
### Check List:
- Please make sure your changes are compilable.
- When providing us with a new feature, it is best to add related tests.
- Please follow [Contributor Covenant Code of Conduct](https://github.com/apache/brpc/blob/master/CODE_OF_CONDUCT.md).
--
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: dev-unsubscribe@brpc.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@brpc.apache.org
For additional commands, e-mail: dev-help@brpc.apache.org
Re: [PR] Refactor streaming rpc (brpc)
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
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: dev-unsubscribe@brpc.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@brpc.apache.org
For additional commands, e-mail: dev-help@brpc.apache.org
Re: [PR] Refactor streaming rpc (brpc)
Posted by "wwbmmm (via GitHub)" <gi...@apache.org>.
wwbmmm merged PR #3422:
URL: https://github.com/apache/brpc/pull/3422
--
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: dev-unsubscribe@brpc.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@brpc.apache.org
For additional commands, e-mail: dev-help@brpc.apache.org