Tonic: Server side streaming buffering clarification

[stefandanaita]

Hello guys! I've implemented a gRPC server that's doing server side streaming and I have a question around the channel's buffer size.

Setting the buffer size to 4, like it currently is in the documentation (let (mut tx, rx) = mpsc::channel(req.buffer_size as usize);) will push the streams to the client in bursts of 5.

However, what I'm currently trying to achieve is "real-time" between when the stream is sent to the sink and when it actually gets pushed to the client, i.e no buffering. And so I tried setting the buffer to 0, hoping that the stream will get pushed as soon as the sink is flushed (which is done automatically by calling tx.send(...).await;). No luck though, by doing it this way, a stream only gets pushed to the client as soon as the next stream is sent to the sink.

I would expect that by setting the buffer to 0, the streams don't get buffered at all. Am I doing something wrong? How can I achieve what I am looking for with tonic?

To replicate this, you can checkout this mock repository - mock-grpc and send the following request:

{
  "number_of_streams": 5,
  "buffer_size": 0,
  "ttl": 10
}

In the ideal scenario, you should see all the 5 streams coming instantly and the connection to be dropped by the server after 10 seconds. However, you will only get 4 streams instantly and the 5th stream only when the connection is dropped.

This means that by setting the buffer size to 0, every stream only gets pushed out when another one is processed by the sink, and the last one is pushed out when the server ends the request.

Thanks a lot for your help!

[stefandanaita]

Lucio suggested on Discord to use unbounded channels: unbounded channels branch. However, with this approach it's even worse than bounded channels with buffer = 0.

In the bounded channels case, the number of streams buffered are equal to whatever you set as the channel buffer + 1. For unbounded channels, weirdly enough I only get the streams (all of them at once) on the client as soon as the server closes the connection.

[LucioFranco]

@seanmonstar is hyper doing any extra buffering here?

This might also have to do with h2 window size. You may want to consider playing around with that.

[seanmonstar]

I'm sorry I don't fully understand the context. Hyper does have a small buffer to handle the HTTP/2 framing, and if a stream runs of available flow control, it will try to stop pulling from the user's body stream, to propagate back pressure, until the peer gives some more window space.

[LucioFranco]

From tonics side we just basically map prost::Message -> impl Buf and pass to hyper.

[stefandanaita]

I'm sorry I don't fully understand the context. Hyper does have a small buffer to handle the HTTP/2 framing, and if a stream runs of available flow control, it will try to stop pulling from the user's body stream, to propagate back pressure, until the peer gives some more window space.

It's unlikely that the stream runs out of available flow control. At least not for my mock example repository above where I'm basically streaming 5 tiny streams. It's weird that the number of streams buffered when using mpsc::channel(<VALUE>) is equal to VALUE + 1.

So for mpsc::channel(5), we'll have 6 streams buffered and the 7th stream will trigger the first 6 to be sent
..for mpsc::channel(1), we'll have 2 streams buffered, 3rd triggering the first 2..
and for mpsc::channel(0), there's always 1 stream buffered 😢

And when using unbounded channels, which to my understanding should stream right away, nothing is streamed until the connection is ended by the server.

[LucioFranco]

@stefandanaita do you have a demo I could try with a client that shows the behavior? Another thing that would be helpful is to turn on logs and get the RUST_LOG=h2=debug logs.

[fluxxu]

@LucioFranco
I can reproduce the issue with payload

{
  "number_of_streams": 5,
  "buffer_size": 0,
  "ttl": 10
}

here is the log: (RUST_LOG=h2=debug)

fluxxu@Fluxs-MacBook-Pro mock-grpc % cargo run
    Finished dev [unoptimized + debuginfo] target(s) in 0.30s
     Running `target/debug/mock-grpc`
50052
GO FOR LAUNCH
Jul 13 15:49:27.304 DEBUG h2::codec::framed_write: send; frame=Settings { flags: (0x0), initial_window_size: 1048576, max_frame_size: 16384 }    
Jul 13 15:49:27.315 DEBUG h2::codec::framed_read: received; frame=Settings { flags: (0x0), enable_push: 0, max_concurrent_streams: 0, initial_window_size: 4194304, max_frame_size: 4194304, max_header_list_size: 8192 }    
Jul 13 15:49:27.315 DEBUG h2::codec::framed_write: send; frame=Settings { flags: (0x1: ACK) }    
Jul 13 15:49:27.315 DEBUG h2::codec::framed_read: received; frame=WindowUpdate { stream_id: StreamId(0), size_increment: 4128769 }    
Jul 13 15:49:27.315 DEBUG h2::codec::framed_read: received; frame=Ping { ack: false, payload: [0, 0, 0, 0, 0, 0, 0, 0] }    
Jul 13 15:49:27.315 DEBUG h2::codec::framed_write: send; frame=Ping { ack: true, payload: [0, 0, 0, 0, 0, 0, 0, 0] }    
Jul 13 15:49:27.315 DEBUG h2::codec::framed_read: received; frame=Headers { stream_id: StreamId(1), flags: (0x4: END_HEADERS) }    
Jul 13 15:49:27.316 DEBUG h2::codec::framed_read: received; frame=WindowUpdate { stream_id: StreamId(1), size_increment: 5 }    
Jul 13 15:49:27.316 DEBUG h2::codec::framed_read: received; frame=Data { stream_id: StreamId(1), flags: (0x1: END_STREAM) }    
Jul 13 15:49:27.316 DEBUG h2::codec::framed_read: received; frame=WindowUpdate { stream_id: StreamId(0), size_increment: 5 }    
Jul 13 15:49:27.316 DEBUG h2::codec::framed_write: send; frame=WindowUpdate { stream_id: StreamId(0), size_increment: 983041 }    
Sending stream 1...
Jul 13 15:49:27.318 DEBUG h2::codec::framed_write: send; frame=Headers { stream_id: StreamId(1), flags: (0x4: END_HEADERS) }    
Sent stream 1 in 815000 nanos
Sending stream 2...
Jul 13 15:49:27.318 DEBUG h2::codec::framed_write: send; frame=Data { stream_id: StreamId(1) }    
Sent stream 2 in 612000 nanos
Sending stream 3...
Jul 13 15:49:27.319 DEBUG h2::codec::framed_write: send; frame=Data { stream_id: StreamId(1) }    
Sent stream 3 in 552000 nanos
Sending stream 4...
Jul 13 15:49:27.320 DEBUG h2::codec::framed_write: send; frame=Data { stream_id: StreamId(1) }    
Sent stream 4 in 533000 nanos
Sending stream 5...
Jul 13 15:49:27.320 DEBUG h2::codec::framed_write: send; frame=Data { stream_id: StreamId(1) }    
Sent stream 5 in 533000 nanos
Jul 13 15:49:38.318 DEBUG h2::codec::framed_write: send; frame=Data { stream_id: StreamId(1) }    
Jul 13 15:49:38.318 DEBUG h2::codec::framed_write: send; frame=Headers { stream_id: StreamId(1), flags: (0x5: END_HEADERS | END_STREAM) }  

[LucioFranco]

@fluxxu do you have code available for this?

[fluxxu]

@fluxxu do you have code available for this?

@LucioFranco
Here
https://github.com/fluxxu/mock-grpc.git

Server:

cargo run --bin server

Client:

cargo run --bin client

[stefandanaita]

Thanks for jumping in @fluxxu! @LucioFranco have you had a chance to try Fluxxu's mock code?

[LucioFranco]

Hey @stefandanaita and @fluxxu thanks for the reproduction! So I can reproduce the issue but nothing is poking out as obvious, I am going to have to dig a bit deeper into h2 and see what is going on. From what I can tell the rx end of the stream is not getting polled after notifying that there is a new message.

cc @seanmonstar we should take a look together when you have some time.

[stefandanaita]

Hey @LucioFranco! Have you had any time to look into this? 😢

[LucioFranco]

I have not had a chance yet, I pinged @seanmonstar to take a look, otherwise, I will try and look next week when I get some time. Work has been quite busy for me sorry :(

[stefandanaita]

Hello @LucioFranco! I know you're pretty busy, but if you find some time at some point, this issue is still valid and could do with some love 😄

[LucioFranco]

@stefandanaita sorry, let me add this to the list of priorities, hopefulyl will get there soon. I am trying to focus on hyper as long as this is not blocking you?