Skip to content

Refactor streaming rpc - #3422

Open
chenBright wants to merge 1 commit into
apache:masterfrom
chenBright:refactor_stream_rpc
Open

Refactor streaming rpc#3422
chenBright wants to merge 1 commit into
apache:masterfrom
chenBright:refactor_stream_rpc

Conversation

@chenBright

Copy link
Copy Markdown
Contributor

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:

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:

    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:

Copilot AI left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Pull request overview

This PR refactors bRPC’s streaming RPC implementation to remove the per-Stream “fake Socket” indirection and instead manage streams directly via VersionedRefWithId<Stream>, simplifying lookup and reducing the write path to a single host-socket queue. This aims to reduce coupling to SocketConnection, lower per-write scheduling/queue overhead, and streamline lifecycle management.

Changes:

  • Refactored Stream to be a VersionedRefWithId<Stream> object (StreamId becomes VRefId) and updated call sites to use Stream::Address.
  • Simplified the write path by packing stream frames into a single IOBuf and enqueueing directly to the host Socket.
  • Updated protocols/controller paths and unit tests to reflect the new Stream addressing and connection publishing rules.

Reviewed changes

Copilot reviewed 10 out of 10 changed files in this pull request and generated 3 comments.

Show a summary per file
File Description
test/brpc_streaming_rpc_unittest.cpp Updates tests to address Stream directly and adjusts expectations for the new single-write packing behavior.
src/bthread/execution_queue_inl.h Formatting/indentation adjustments only.
src/brpc/versioned_ref_with_id.h Adds reusable member detection + optional-caller helpers; removes virtual destructor to avoid vptr in CRTP.
src/brpc/stream.h Switches StreamId to VRefId and updates includes/signatures accordingly.
src/brpc/stream.cpp Core refactor: Stream lifecycle via VersionedRefWithId, pending-write buffering before connect, single-shot host socket write path, updated failure handling.
src/brpc/stream_impl.h Updates Stream class definition to inherit VersionedRefWithId, adds pending-write buffering structures and new lifecycle hooks.
src/brpc/socket.h Aligns Socket destructor with non-virtual VersionedRefWithId base destructor.
src/brpc/policy/streaming_rpc_protocol.cpp Switches frame dispatch lookup from Socket/fake-conn to Stream::Address and direct Stream::OnReceived.
src/brpc/policy/baidu_rpc_protocol.cpp Updates stream addressing/connection setup during RPC response/request packing.
src/brpc/controller.cpp Updates stream connection setup to use StreamUniquePtr and direct stream pointers.

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

Comment thread src/brpc/stream.cpp
Comment thread test/brpc_streaming_rpc_unittest.cpp Outdated
Comment thread src/brpc/stream.cpp

Copilot AI left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Pull request overview

Copilot reviewed 10 out of 10 changed files in this pull request and generated no new comments.

Suppressed comments (2)

src/brpc/versioned_ref_with_id.h:358

  • The comment for AddressImpl describes both cases as failed_as_well=true, but the second case corresponds to failed_as_well=false (used by Address()). This is misleading for readers and makes it harder to reason about return codes.
    // 1. When `failed_as_well=true', returns 0 on success,
    //    1 on failed object, -1 on recycled.
    // 2. When `failed_as_well=true', returns 0 on success,
    //    -1 when the object was SetFailed().

src/brpc/stream.cpp:568

  • SetConnected() publishes _connected=true with release ordering, but then conditionally sets _connected back to false if Failed() becomes true. This makes _connected non-monotonic and can cause waiters that only poll _connected to miss the brief true transition (spurious timeouts) and contradicts the “one-way transition” assumption used elsewhere for safely reading _host_socket. Consider keeping _connected monotonic and relying on Failed() to represent liveness.
    _connected.store(true, butil::memory_order_release);
    if (Failed()) {
        _connected.store(false, butil::memory_order_relaxed);
        bthread_mutex_unlock(&_connect_mutex);
        return;

@chenBright
chenBright requested a review from wwbmmm August 6, 2026 09:33
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants