Connection driven rpc backpressure - #15
Conversation
kvc0
left a comment
kvc0
left a comment
There was a problem hiding this comment.
Thanks for the improvement! I have a few notes to consider
|
I ended up adding Then downstream that resulted in moving Codec from SocketService down to ConnectionService to make the generics less messy - again, I think ConnectionService was already de-facto associated with a codec and I'm just making that relationship explicit. That adds a little bit to the surface of the breaking API changes here, but I think it's manageable based on the example crates and in service of what I'm trying to do here. |
kvc0
left a comment
kvc0
left a comment
There was a problem hiding this comment.
This looks great to me. I have a request, since it's a breaking change in the same neighborhood.
|
|
||
| /// Create a new message codec for a connection. | ||
| fn codec(&self) -> Self::Codec; | ||
| fn codec(&self) -> <Self::ConnectionService as ConnectionService>::Codec; |
There was a problem hiding this comment.
since we're doing a breaking change, let's give implementors more control: Now is a good time to make codec() and new_stream_service() take &mut self. &mut is available at the framework consumer sites, and giving &mut to users allows them more synchronization-free options.
| produced += 1; | ||
| } | ||
| // Then poll ready RPCs until we either run out of room to send responses or run out of rpcs. | ||
| while let Some(permit) = outbound.reserve() { | ||
| match pin!(&mut me.rpcs).poll_next(context) { |
2 participants
BREAKING: notable semantic/API change to protosocket and protosocket-rpc API traits.
The main motivation of this PR is to reintroduce backpressure between a server's connection outbound write buffer and RPC processing, which was lost in the v1 refactor.
Essentially, this moves the "connection service polls a FuturesUnordered via
poll" implementation that all server implementers I'm aware of were doing into the connection service itself. This is important, because it allows the connection service to conditionally advance the request futures only when there's somewhere for the responses to actually go: the serialization queue must have room, which is itself backpressured by the connection's write buffer.RPC Server implementers now just return a future which will yield a response for each unary/streaming request, rather than being responsible for bookkeeping the production and return of responses themselves.
As before, they can always opt to decouple request processing from the connection via spawning.
Overall, I believe this should lighten the work for implementers and internalize concerns like message-id bookkeeping, as bonuses.
The underlying machinery is also available to any raw protosocket implementers who want to engage with backpressure.