diff --git a/src/proto/streams/flow_control.rs b/src/proto/streams/flow_control.rs index 3e85b4b..a9b2999 100644 --- a/src/proto/streams/flow_control.rs +++ b/src/proto/streams/flow_control.rs @@ -1,6 +1,27 @@ use frame::Reason; -use frame::Reason::*; -use proto::*; +use proto::{MAX_WINDOW_SIZE, WindowSize}; + +// We don't want to send WINDOW_UPDATE frames for tiny changes, but instead +// aggregate them when the changes are significant. Many implementations do +// this by keeping a "ratio" of the update version the allowed window size. +// +// While some may wish to represent this ratio as percentage, using a f32, +// we skip having to deal with float math and stick to integers. To do so, +// the "ratio" is represented by 2 i32s, split into the numerator and +// denominator. For example, a 50% ratio is simply represented as 1/2. +// +// An example applying this ratio: If a stream has an allowed window size of +// 100 bytes, WINDOW_UPDATE frames are scheduled when the unclaimed change +// becomes greater than 1/2, or 50 bytes. +const UNCLAIMED_NUMERATOR: i32 = 1; +const UNCLAIMED_DENOMINATOR: i32 = 2; + +#[test] +fn sanity_unclaimed_ratio() { + assert!(UNCLAIMED_NUMERATOR < UNCLAIMED_DENOMINATOR); + assert!(UNCLAIMED_NUMERATOR >= 0); + assert!(UNCLAIMED_DENOMINATOR > 0); +} #[derive(Copy, Clone, Debug)] pub struct FlowControl { @@ -51,17 +72,29 @@ impl FlowControl { self.available += capacity; } - /// Returns the number of bytes available but not assigned to the window. + /// If a WINDOW_UPDATE frame should be sent, returns a positive number + /// representing the increment to be used. + /// + /// If there is no available bytes to be reclaimed, or the number of + /// available bytes does not reach the threshold, this returns `None`. /// /// This represents pending outbound WINDOW_UPDATE frames. - pub fn unclaimed_capacity(&self) -> WindowSize { + pub fn unclaimed_capacity(&self) -> Option { let available = self.available as i32; if self.window_size >= available { - return 0; + return None; } - (available - self.window_size) as WindowSize + let unclaimed = available - self.window_size; + let threshold = self.window_size / UNCLAIMED_DENOMINATOR + * UNCLAIMED_NUMERATOR; + + if unclaimed < threshold { + None + } else { + Some(unclaimed as WindowSize) + } } /// Increase the window size. @@ -71,11 +104,11 @@ impl FlowControl { let (val, overflow) = self.window_size.overflowing_add(sz as i32); if overflow { - return Err(FlowControlError); + return Err(Reason::FlowControlError); } if val > MAX_WINDOW_SIZE as i32 { - return Err(FlowControlError); + return Err(Reason::FlowControlError); } trace!("inc_window; sz={}; old={}; new={}", sz, self.window_size, val); diff --git a/src/proto/streams/recv.rs b/src/proto/streams/recv.rs index 40db92d..d0c9772 100644 --- a/src/proto/streams/recv.rs +++ b/src/proto/streams/recv.rs @@ -214,6 +214,7 @@ impl Recv task: &mut Option) -> Result<(), UserError> { + trace!("release_capacity; size={}", capacity); if capacity > stream.in_flight_recv_data { // TODO: Handle error unimplemented!(); @@ -226,11 +227,13 @@ impl Recv self.flow.assign_capacity(capacity); stream.recv_flow.assign_capacity(capacity); - // Queue the stream for sending the WINDOW_UPDATE frame. - self.pending_window_updates.push(stream); + if stream.recv_flow.unclaimed_capacity().is_some() { + // Queue the stream for sending the WINDOW_UPDATE frame. + self.pending_window_updates.push(stream); - if let Some(task) = task.take() { - task.notify(); + if let Some(task) = task.take() { + task.notify(); + } } Ok(()) @@ -493,9 +496,7 @@ impl Recv -> Poll<(), io::Error> where T: AsyncWrite, { - let incr = self.flow.unclaimed_capacity(); - - if incr > 0 { + if let Some(incr) = self.flow.unclaimed_capacity() { let frame = frame::WindowUpdate::new(StreamId::zero(), incr); // Ensure the codec has capacity @@ -536,9 +537,7 @@ impl Recv } // TODO: de-dup - let incr = stream.recv_flow.unclaimed_capacity(); - - if incr > 0 { + if let Some(incr) = stream.recv_flow.unclaimed_capacity() { // Create the WINDOW_UPDATE frame let frame = frame::WindowUpdate::new(stream.id, incr); diff --git a/src/proto/streams/streams.rs b/src/proto/streams/streams.rs index 6480148..3aac220 100644 --- a/src/proto/streams/streams.rs +++ b/src/proto/streams/streams.rs @@ -478,7 +478,7 @@ impl StreamRef me.actions.recv.poll_trailers(&mut stream) } - /// Releases recv capacity back to the peer. This will result in sending + /// Releases recv capacity back to the peer. This may result in sending /// WINDOW_UPDATE frames on both the stream and connection. pub fn release_capacity(&mut self, capacity: WindowSize) -> Result<(), UserError> diff --git a/tests/flow_control.rs b/tests/flow_control.rs index 841579a..9ff672a 100644 --- a/tests/flow_control.rs +++ b/tests/flow_control.rs @@ -49,6 +49,161 @@ fn send_data_without_requesting_capacity() { h2.wait().unwrap(); } +#[test] +fn release_capacity_sends_window_update() { + let _ = ::env_logger::init(); + + let payload = vec![0; 65_535]; + + let mock = mock_io::Builder::new() + .handshake() + .write(&[ + // GET / + 0, 0, 16, 1, 5, 0, 0, 0, 1, 130, 135, 65, 139, 157, 41, + 172, 75, 143, 168, 233, 25, 151, 33, 233, 132, + ]) + .write(frames::SETTINGS_ACK) + // Read response + .read(&[0, 0, 1, 1, 4, 0, 0, 0, 1, 0x88]) + .read(&[ + // DATA + 0, 64, 0, 0, 0, 0, 0, 0, 1, + ]) + .read(&payload[0..16_384]) + .read(&[ + // DATA + 0, 64, 0, 0, 0, 0, 0, 0, 1, + ]) + .read(&payload[16_384..16_384*2]) + .read(&[ + // DATA + 0, 64, 0, 0, 0, 0, 0, 0, 1, + ]) + .read(&payload[16_384*2..16_384*3]) + .write(&[0, 0, 4, 8, 0, 0, 0, 0, 0, 0, 0, 128, 0]) + .write(&[0, 0, 4, 8, 0, 0, 0, 0, 1, 0, 0, 128, 0]) + .read(&[ + // DATA + 0, 63, 255, 0, 1, 0, 0, 0, 1, + ]) + .read(&payload[16_384*3..16_384*4 - 1]) + + // we send a 2nd stream in order to test the window update is + // is actually written to the socket + .write(&[ + // GET / + 0, 0, 4, 1, 5, 0, 0, 0, 3, 130, 135, 190, 132, + ]) + .read(&[0, 0, 1, 1, 5, 0, 0, 0, 3, 0x88]) + .build(); + + let mut h2 = Client::handshake(mock) + .wait().unwrap(); + + let request = Request::builder() + .method(Method::GET) + .uri("https://http2.akamai.com/") + .body(()).unwrap(); + + let mut stream = h2.request(request, true).unwrap(); + + // Get the response + let resp = h2.run(poll_fn(|| stream.poll_response())).unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + + // read some body to use up window size to below half + let mut body = resp.into_parts().1; + let buf = h2.run(poll_fn(|| body.poll())).unwrap().unwrap(); + assert_eq!(buf.len(), 16_384); + let buf = h2.run(poll_fn(|| body.poll())).unwrap().unwrap(); + assert_eq!(buf.len(), 16_384); + let buf = h2.run(poll_fn(|| body.poll())).unwrap().unwrap(); + assert_eq!(buf.len(), 16_384); + + // release some capacity to send a window_update + body.release_capacity(buf.len() * 2).unwrap(); + + let buf = h2.run(poll_fn(|| body.poll())).unwrap().unwrap(); + assert_eq!(buf.len(), 16_383); + + + // send a 2nd stream to force flushing of window updates + let request = Request::builder() + .method(Method::GET) + .uri("https://http2.akamai.com/") + .body(()).unwrap(); + let mut stream = h2.request(request, true).unwrap(); + let resp = h2.run(poll_fn(|| stream.poll_response())).unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + + h2.wait().unwrap(); +} + +#[test] +fn release_capacity_of_small_amount_does_not_send_window_update() { + let _ = ::env_logger::init(); + + let payload = [0; 16]; + + let mock = mock_io::Builder::new() + .handshake() + .write(&[ + // GET / + 0, 0, 16, 1, 5, 0, 0, 0, 1, 130, 135, 65, 139, 157, 41, + 172, 75, 143, 168, 233, 25, 151, 33, 233, 132, + ]) + .write(frames::SETTINGS_ACK) + // Read response + .read(&[0, 0, 1, 1, 4, 0, 0, 0, 1, 0x88]) + .read(&[ + // DATA + 0, 0, 16, 0, 1, 0, 0, 0, 1, + ]) + .read(&payload[..]) + // write() or WINDOW_UPDATE purposefully excluded + + // we send a 2nd stream in order to test the window update is + // is actually written to the socket + .write(&[ + // GET / + 0, 0, 4, 1, 5, 0, 0, 0, 3, 130, 135, 190, 132, + ]) + .read(&[0, 0, 1, 1, 5, 0, 0, 0, 3, 0x88]) + .build(); + + let mut h2 = Client::handshake(mock) + .wait().unwrap(); + + let request = Request::builder() + .method(Method::GET) + .uri("https://http2.akamai.com/") + .body(()).unwrap(); + + let mut stream = h2.request(request, true).unwrap(); + + // Get the response + let resp = h2.run(poll_fn(|| stream.poll_response())).unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + + let mut body = resp.into_parts().1; + let buf = h2.run(poll_fn(|| body.poll())).unwrap().unwrap(); + + // release some capacity to send a window_update + body.release_capacity(buf.len()).unwrap(); + + // send a 2nd stream to force flushing of window updates + let request = Request::builder() + .method(Method::GET) + .uri("https://http2.akamai.com/") + .body(()).unwrap(); + let mut stream = h2.request(request, true).unwrap(); + let resp = h2.run(poll_fn(|| stream.poll_response())).unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + + + h2.wait().unwrap(); +} + #[test] #[ignore] fn expand_window_sends_window_update() {