protocols/streaming-response #1947

wngr · 2021-02-05T07:54:49Z

this adds a streaming-response crate, which provides a NetworkBehaviour similar
to the request-response protocol to have a generic messaging mechanism
between two peers. The most notable difference to request-response and
the main motivation for its implementation is to provide an open-ended
streaming mechanism for the responses: a consumer's request can result
in any number of individual response frames, until said stream is
finalized. Conceptually, it is somewhat similar to grpc streaming.

Posting this here as a draft PR for discussion, whether you would be willing
to accept such a dedicated crate, or whether it would make sense to fold this into
request-response.

Related discussions #1727 #1942

mxinden

Thank you for the pull request @wngr.

I am in favor of adding a stream-oriented protocol abstraction to rust-libp2p.

whether you would be willing to accept such a dedicated crate, or whether it would make sense to fold this into request-response.

Opposite to @romanb gut feeling, with the simple API interface (request(), respond(), finish_response()) you are proposing in this pull request in mind, I would expect adding stream-oriented responses to libp2p-request-response directly not to be too intrusive, put simply, replacing the oneshot between behaviour and handler with an mpsc channel. That said, my gut feeling also tells me that Roamn's gut feeling is usually right.

I will need to try adding stream-oriented responses to libp2p-request-response before giving a better review here. I might get around doing so in the next couple of weeks. In case someone else has time before that, I would appreciate a pull request with a proof-of-concept.

wngr · 2021-03-26T10:14:41Z

IMHO the biggest design mismatch is the double use of the SubstreamProtocol's timeout for both upgrading to the proper protocol and waiting for a response. For streaming based protocols, a response might take an arbitrary amount of time to be generated. So my first instinct would be to decouple those two. Once that is done, bolting the API I proposed in d716a43 should be straight-forward.
To be backwards compatible, another layer of timeout handling would need to be added though..

Thinking about this, I tend to come back to the same point: The inherent abstraction of the current request-response protocol is an RPC call. The caller expects a timely answer from a server (.. eerm peer). Whereas the streaming-response could be more considered to be a typed subscription to some service, with no protocol-level expectation about timely responses etc (apart from establishing the connection channel). So it's more directed than pubsub, but less rigid than an RPC call.

Not sure to which conclusion that leads, just leaving my thoughts here. Maybe it's better to have some duplication than try to come up with a silver bullet..

mxinden · 2021-04-21T12:27:04Z

Sorry for the slow response time here.

the biggest design mismatch is the double use of the SubstreamProtocol's timeout for both upgrading to the proper protocol and waiting for a response.

Maybe it's better to have some duplication than try to come up with a silver bullet..

I tried patching libp2p-request-response to allow streamed responses to a single request which turned out harder than expected, thus I agree with the above. Sorry for leading the discussion astray.

In case there are multiple projects that could make use of streaming-response, I am happy to include it in rust-libp2p (//CC @thomaseizinger). In case there is only a single project, I would prefer for it to be maintained outside of the rust-libp2p mono-repo.

It might as well be worth considering making streaming-response more generic, in the sense of allowing streaming on both the requesting and the responding side. See @romanb's suggestion in #1942 (comment):

My first impulse would be to have a dedicated crate for generic streaming-oriented protocols, where any number of "chunks" can be sent in either direction on a single substream.

This would allow both the communication pattern suggested in this pull request, as well as various other streaming based patterns.

wngr · 2021-04-26T12:39:18Z

Thanks for your response nonetheless!

You have an idea of an API sketch for a generic duplex streaming protocol? Wouldn't that just be an opinionated behaviour that wraps some book-keeping into a common package:

Connect to Peer, if not connected
Keep alive connection (and provide some API to figure out which connections to keep open)
Send [u8] to Peer with some metadata
Receive [u8] from Peer with some metadata
..
And higher level APIs (like for example the API proposed in this PR) is layered on top?

Let me know if there is more interest in this direction. If not, I will be publishing the streaming-response behaviour as a separate crate.

mxinden · 2021-04-26T13:17:41Z

You have an idea of an API sketch for a generic duplex streaming protocol? Wouldn't that just be an opinionated behaviour that wraps some book-keeping into a common package:

Pretty much, yes. I would deem this useful as (A) a building block to build e.g. streaming-response or more complex streaming based protocols and (B) a way to prototype protocols without having to bother with the verbose NetworkBehaviour trait.

Connect to Peer, if not connected

Keep alive connection (and provide some API to figure out which connections to keep open)

👍

Send [u8] to Peer with some metadata

Receive [u8] from Peer with some metadata

I was thinking of exposing an AsyncRead + AsyncWrite stream directly.

impl StreamingBehaviour {
  fn new_stream(&mut self, peer_id: PeerId, addr: Option<Multiaddr>) -> RequestId {
    // ...
  }
}

impl NetworkBehaviour for StreamingBehaviour {
    type OutEvent = StreamingEvent;
    // ...
}

enum StreamingEvent {
  NewStream {
    id: RequestId,
    stream: // something that implements AsyncRead + AsyncWrite and allows to be tracked to close an idle connection
  }
}

And higher level APIs (like for example the API proposed in this PR) is layered on top?

Off the top of my head yes, though there might be many caveats that I am missing right now.

Take the above with a grain of salt. After all, I am not currently in need of a streaming based protocol abstraction myself, thus I am having difficulties coming up with a useful abstraction. In case a more generic abstraction like the one suggested above would be useful for multiple real-world protocols, I think this is worth pursuing. If not, I think your more specific streaming-response abstraction is the way to go.

What do you think @wngr?

wngr · 2021-04-30T06:59:17Z

Yup, I think I'm gonna give this a try.

mxinden · 2021-05-31T14:42:32Z

@wngr let me know once you would like another review here.

wngr · 2021-05-31T14:46:55Z

Thanks, will do! The force push was accidental, nothing to see here ..

NetworkBehaviour

wngr · 2021-08-10T19:48:46Z

I finally came around spending some time on this. It's pretty barebones, but I think convenient enough to be of broader use. I borrowed some things from the request-response protocol (dialing of peers, some house-keeping).
Handing out a raw handle implementing Async{Read,Write} was a bit too untyped for me, that's why I added the StreamingCodec -- not sure whether it adds much though.
Ping @mxinden -- let me know if this goes into the direction you imagined.

wngr · 2021-08-10T19:49:46Z

protocols/streaming/tests/smoke.rs

+                    stream.write_all(b"Hello").await.unwrap();
+                    stream.flush().await.unwrap();


Smoke test. Direct usage of Async{Read,Write}.

wngr · 2021-08-10T19:50:01Z

protocols/streaming/tests/codec.rs

+
+    let (peer2_id, trans) = mk_transport();
+    let mut swarm2 = Swarm::new(trans, Streaming::<PongCodec>::default(), peer2_id);
+


Example using a typed codec.

wngr · 2021-08-10T19:50:31Z

protocols/streaming/tests/streaming.rs

+            SwarmEvent::ConnectionClosed { peer_id, .. } => {
+                assert_eq!(peer_id, peer1_id);
+                assert_eq!(rx.next().await.unwrap(), 10);
+                break;


Example demonstrating connection keep-alive.

mxinden · 2021-08-14T11:38:42Z

Thanks @wngr! Review is in progress. Hope to finish sometime later today or next week.

mxinden

Thanks @wngr!

Couple of suggestions below. I still need to take a deeper look at the test code.

mxinden · 2021-08-14T11:21:28Z

protocols/streaming/src/handler.rs

@@ -0,0 +1,306 @@
+use futures::future;


For the sake of consistency, would you mind including license headers? Feel free to mention yourself. In case a file is based on libp2p-request-response, please include the reference to Parity Technologies.

mxinden · 2021-08-14T11:23:43Z

protocols/streaming/src/lib.rs

+    /// connection to be established.
+    pending_outbound_requests: HashMap<PeerId, SmallVec<[OutboundStreamId; 10]>>,
+    config: StreamingConfig,
+    _codec: PhantomData<T>,


Suggested change

_codec: PhantomData<T>,

Is this needed?

mxinden · 2021-08-14T11:28:50Z

protocols/streaming/src/lib.rs

+    }
+
+    fn protocol_name() -> Self::Protocol {
+        b"/streaming/bytes/1.0.0"


I would expect users to use custom protocol names, even when using IdentityCodec. What do you think of making this an optional field on IdentityCodec defaulting to /streaming/bytes/1.0.0?

mxinden · 2021-08-14T11:37:43Z

protocols/streaming/src/lib.rs

+    /// by [`NetworkBehaviour::addresses_of_peer`].
+    ///
+    /// Addresses added in this way are only removed by `remove_address`.
+    pub fn add_address(&mut self, peer: PeerId, address: Multiaddr) {


Related to this pull request: I find the user experience of first having to add an address for a peer and only then being able to open a stream and thus a connection to said peer not ideal. This is not specific to this pull request, e.g. libp2p-request-response and libp2p-dcutr (#2076) have the same indirection.

I am not opposed to the NetworkBehaviour::addresses_of_peer, though I would suggest to extend NetworkBehaviourAction::Dial{Peer,Address}, allowing one to provide a set of addresses right away, potentially later on extended by addresses retrieved through NetworkBehaviour::addresses_of_peer from other NetworkBehaviour implementations.

mxinden · 2021-08-16T11:18:31Z

protocols/streaming/src/handler.rs

+    EMPTY_QUEUE_SHRINK_THRESHOLD,
+};
+
+pub(crate) type RefCount = Arc<PhantomData<u8>>;


Do I understand correctly that the task of the handler is not woken up, when a StreamHandle with a RefCount is dropped? If so, a handler, if not polled due to some other event, might stay in KeepAlive::Yes forever, even though all its streams are closed, correct?

In libp2p-relay I have solved this with the help of a oneshot. Would that be an option here as well?

rust-libp2p/protocols/relay/src/handler.rs

Lines 110 to 119 in 98bc5e6

/// Tracks substreams lend out to other [`RelayHandler`]s or as

/// [`Connection`](protocol::Connection) to the

/// [`RelayTransport`](crate::RelayTransport).

///

/// For each substream to the peer of this handler, there is a future in here that resolves once

/// the given substream is dropped.

///

/// Once all substreams are dropped and this handler has no other work, [`KeepAlive::Until`] can

/// be set, allowing the connection to be closed eventually.

alive_lend_out_substreams: FuturesUnordered<oneshot::Receiver<()>>,

mxinden · 2021-08-16T11:24:59Z

protocols/streaming/src/lib.rs

+    }
+
+    fn inject_disconnected(&mut self, peer: &PeerId) {
+        self.connected.remove(peer);


Given that inject_connection_closed above removes a peer from connected if no more connections exist and given that inject_disconnected is only ever called after inject_connection_closed, this line should never be executed, correct? If so, how about at least adding a debug_asserthere to make sureinject_connection_closed` will properly clean up in the future.

rust-libp2p/swarm/src/lib.rs

Lines 579 to 583 in 98bc5e6

this.behaviour

.inject_connection_closed(&peer_id, &id, &endpoint);

if num_established == 0 {

this.behaviour.inject_disconnected(&peer_id);

}

mxinden · 2021-08-16T11:36:47Z

protocols/streaming/src/handler.rs

+    inbound_stream_id: Arc<AtomicU64>,
+    /// A pending fatal error that results in the connection being closed.
+    pending_error: Option<ProtocolsHandlerUpgrErr<std::convert::Infallible>>,
+    _codec: PhantomData<T>,


Suggested change

_codec: PhantomData<T>,

Is this needed?

mxinden · 2021-08-16T11:37:09Z

protocols/streaming/src/handler.rs

+}
+
+#[derive(Debug, Default)]
+pub struct StreamingProtocol<T: StreamingCodec> {


Suggested change

pub struct StreamingProtocol<T: StreamingCodec> {

pub struct StreamingProtocol<T> {

Is this needed?

mxinden · 2021-08-16T11:37:24Z

protocols/streaming/src/handler.rs

+pub struct StreamingProtocol<T: StreamingCodec> {
+    _codec: PhantomData<T>,
+}
+impl<T: StreamingCodec> StreamingProtocol<T> {


Suggested change

impl<T: StreamingCodec> StreamingProtocol<T> {

impl<T> StreamingProtocol<T> {

Same as above.

wngr · 2021-10-16T17:34:34Z

Thanks for your feedback @mxinden, appreciate it.

My focus in the past months has shifted quite a bit, so it's unlikely I will make much progress on this. Might get warmed up in the future, but no gain in keeping the PR open until then.

0x77dev · 2024-01-29T06:55:24Z

Just to update the thread, the #5027 has been merged into the master branch. You can find an easier method for establishing basic connections between peers in the stream protocol example.

mxinden reviewed Feb 9, 2021

View reviewed changes

wngr force-pushed the streaming-response-protocol branch from 55d4d70 to 585ec0c Compare May 28, 2021 13:28

protocols/streaming: Initial implementation of generic streaming

888f153

NetworkBehaviour

wngr force-pushed the streaming-response-protocol branch from 585ec0c to 888f153 Compare August 10, 2021 19:48

wngr commented Aug 10, 2021

View reviewed changes

wngr mentioned this pull request Aug 12, 2021

Make event_process = false the default on derive(NetworkBehaviour) #2190

Closed

mxinden reviewed Aug 16, 2021

View reviewed changes

wngr closed this Oct 16, 2021

mxinden mentioned this pull request Feb 7, 2022

Blocking code is not allowed in sync #[behaviour(event_process = true)] trait implementation #2474

Closed

b5 mentioned this pull request May 17, 2022

protocols/stream: Add a streaming-response protocol to rust-libp2p #2657

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

protocols/streaming-response #1947

protocols/streaming-response #1947

wngr commented Feb 5, 2021

mxinden left a comment

wngr commented Mar 26, 2021

mxinden commented Apr 21, 2021

wngr commented Apr 26, 2021

mxinden commented Apr 26, 2021

wngr commented Apr 30, 2021

mxinden commented May 31, 2021

wngr commented May 31, 2021

wngr commented Aug 10, 2021

wngr Aug 10, 2021

wngr Aug 10, 2021

wngr Aug 10, 2021

mxinden commented Aug 14, 2021

mxinden left a comment

mxinden Aug 14, 2021

mxinden Aug 14, 2021

mxinden Aug 14, 2021

mxinden Aug 14, 2021

mxinden Aug 16, 2021

mxinden Aug 16, 2021

mxinden Aug 16, 2021

mxinden Aug 16, 2021

mxinden Aug 16, 2021

wngr commented Oct 16, 2021

0x77dev commented Jan 29, 2024

		stream.write_all(b"Hello").await.unwrap();
		stream.flush().await.unwrap();


		let (peer2_id, trans) = mk_transport();
		let mut swarm2 = Swarm::new(trans, Streaming::<PongCodec>::default(), peer2_id);

	/// Tracks substreams lend out to other [`RelayHandler`]s or as
	/// [`Connection`](protocol::Connection) to the
	/// [`RelayTransport`](crate::RelayTransport).
	///
	/// For each substream to the peer of this handler, there is a future in here that resolves once
	/// the given substream is dropped.
	///
	/// Once all substreams are dropped and this handler has no other work, [`KeepAlive::Until`] can
	/// be set, allowing the connection to be closed eventually.
	alive_lend_out_substreams: FuturesUnordered<oneshot::Receiver<()>>,

	this.behaviour
	.inject_connection_closed(&peer_id, &id, &endpoint);
	if num_established == 0 {
	this.behaviour.inject_disconnected(&peer_id);
	}

	pub struct StreamingProtocol<T: StreamingCodec> {
	pub struct StreamingProtocol<T> {

	impl<T: StreamingCodec> StreamingProtocol<T> {
	impl<T> StreamingProtocol<T> {

protocols/streaming-response #1947

protocols/streaming-response #1947

Conversation

wngr commented Feb 5, 2021

mxinden left a comment

Choose a reason for hiding this comment

wngr commented Mar 26, 2021

mxinden commented Apr 21, 2021

wngr commented Apr 26, 2021

mxinden commented Apr 26, 2021

wngr commented Apr 30, 2021

mxinden commented May 31, 2021

wngr commented May 31, 2021

wngr commented Aug 10, 2021

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mxinden commented Aug 14, 2021

mxinden left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

wngr commented Oct 16, 2021

0x77dev commented Jan 29, 2024