mxinden
Prototype message patterns: `RecvSend`, `SendRecv` and `Send`
Whilst thinking more about https://github.com/libp2p/rust-libp2p/issues/3130 and the PoC I made in https://github.com/thomaseizinger/rust-async-response-stream, I realized that this could actually be part of asynchronous-codec.
It could be implemented on top but it is also fairly intertwined with the idea of Framed. A good data point here is that we don't need any additional dependencies. All we are doing is defining a Stream utility on top of Framed that makes it easier to implement a request-response message pattern.
The design has slightly evolved from https://github.com/thomaseizinger/rust-async-response-stream. Instead of providing two Futures, we now just implement Stream. For the easy usecase where we close the stream after sending the response, we only ever output one Item: The incoming request and the placeholder for the response. Because it is a Stream, a SelectAll will continue to poll it until it returns None. This allows us to completely hide the message exchange and makes using this really easy:
- Create an instance and push it into a
SelectAll. - Process messages coming in from polling
SelectAll.
I also removed the timeout, I think this should be implemented on top.
Sorry for the lack of documentation!
The asynchronuous-codec crate is useful to define encoding and decoding modules that compose well. Framed can then be used to turn an Async{Read,Write} into a Stream and Sink of messages. We use it extensively inside rust-libp2p to decode network messages from a byte stream.
What this PR adds are "message patterns" on top of a framed stream. So far only one is added but it is easy to envision more. I've started with RecvSend because it is the most interesting one.
Your pseudo-code is on point. My vision is that by implementing Stream, we can drive a RecvSend through a SelectAll as part of ConnectionHandler::poll.
The item returned from RecvSend are the incoming requests plus a "channel" to send the response back. This allows for the response to be provided out-of-bound from the stream. There are various patterns employed in rust-libp2p as to how the response makes it to the stream. I think a channel-design lives at the intersection of convenience (response can be sent from anywhere), minimal boilerplate (no message passing down the layers) and (type)safety (response can only be sent once).
The design goals for this were:
- Minimal boilerplate upon usage. It is intended to be repeated in every
ConnectionHandlerto replace the currentInboundUpgradeandOutboundUpgradetraits. - Good composition. For most use cases, we will want to add a timeout to the entire recv and send operation. By implementing
Stream, we should be able to compose it with a generic wrapper that times out if the inner stream doesn't finish in time. - Flexible for different messages. Because we depend only on
EncoderandDecoder, this can be used with probably any protocol.
Okay, that is really helpful, I think I have a much better understanding of it now, thanks!
Sent some tweaks for this branch in https://github.com/thomaseizinger/asynchronous-codec/pull/1
@mxinden I keep coming back to wanting this almost every time I touch any protocol code. Are you in favor of this?
@mxinden I keep coming back to wanting this almost every time I touch any protocol code. Are you in favor of this?
We would use it in combination with a TimedStreams abstraction that wraps a stream::SelectAll but races each stream against a timeout. Using it together with the RecvSend in here (and other "protocols" like SendRecv or just Recv or Send) gives us the same functionality as InboundUpgrade and OutboundUpgrade just without having to implement any traits but composing existing code together.
I am hesitant: I don't think the usage of Stream is intuitive here. I consider a Stream a consumer from a source, not both a consumer and a source sending data to a destination.
All that said, the abstraction might be worth the confusion that comes with it. It seems to me that you feel strongly about this new abstraction. Which of our rust-libp2p protocols would benefit the most of this pattern? Do you want to integrate this into one of them as a proof-of-concept?
I am hesitant: I don't think the usage of
Streamis intuitive here. I consider aStreama consumer from a source, not both a consumer and a source sending data to a destination.
I think of it as a source of events of a protocol.
All that said, the abstraction might be worth the confusion that comes with it. It seems to me that you feel strongly about this new abstraction. Which of our rust-libp2p protocols would benefit the most of this pattern? Do you want to integrate this into one of them as a proof-of-concept?
Every one that sends messages. The idea has the potential to replace every state machine for sending messages in rust-libp2p with a use of the appropriate "message pattern". Currently we only have RecvSend implemented here but it is easy enough to envision additional ones, SendRecv for example.
Do you want to integrate this into one of them as a proof-of-concept?
Here you go: https://github.com/libp2p/rust-libp2p/pull/3610
Code duplication in here is currently horrendous but that can be fixed later.
I am hesitant: I don't think the usage of
Streamis intuitive here. I consider aStreama consumer from a source, not both a consumer and a source sending data to a destination.All that said, the abstraction might be worth the confusion that comes with it. It seems to me that you feel strongly about this new abstraction. Which of our rust-libp2p protocols would benefit the most of this pattern? Do you want to integrate this into one of them as a proof-of-concept?
I've written about this - to some extent - here: https://github.com/libp2p/rust-libp2p/discussions/3603
The usage of Stream is somewhat accidental actually. It happens to be the closest thing to a generator interface that we have on stable and has widespread use.
If we had a generator interface, I'd use that!
I might have a go at re-writing these using async/await and see whether they turn out to be cleaner. Most likely they will at the cost of an allocation but I'd be surprised if we can even measure that in the context of network IO.
I might have a go at re-writing these using
async/awaitand see whether they turn out to be cleaner. Most likely they will at the cost of an allocation but I'd be surprised if we can even measure that in the context of network IO.
It doesn't really work very well because Rust doesn't have native generators yet.