Refactor streaming rpc

[chenBright]

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:

• 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.

[Copilot]

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.

[Copilot]

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;