aggregate WINDOW_UPDATE frames until change is over 50% available

src/proto/streams/flow_control.rs

@@ -1,6 +1,27 @@
 use frame::Reason;
-use frame::Reason::*;
+use proto::{MAX_WINDOW_SIZE, WindowSize};
-use proto::*;
+
+// 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<WindowSize> {
         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);

src/proto/streams/recv.rs

@@ -214,6 +214,7 @@ impl<B, P> Recv<B, P>
                             task: &mut Option<Task>)
         -> Result<(), UserError>
     {
+        trace!("release_capacity; size={}", capacity);
         if capacity > stream.in_flight_recv_data {
             // TODO: Handle error
             unimplemented!();
@@ -226,11 +227,13 @@ impl<B, P> Recv<B, P>
         self.flow.assign_capacity(capacity);
         stream.recv_flow.assign_capacity(capacity);
 
-        // Queue the stream for sending the WINDOW_UPDATE frame.
+        if stream.recv_flow.unclaimed_capacity().is_some() {
-        self.pending_window_updates.push(stream);
+            // Queue the stream for sending the WINDOW_UPDATE frame.
+            self.pending_window_updates.push(stream);
 
-        if let Some(task) = task.take() {
+            if let Some(task) = task.take() {
-            task.notify();
+                task.notify();
+            }
         }
 
         Ok(())
@@ -493,9 +496,7 @@ impl<B, P> Recv<B, P>
         -> Poll<(), io::Error>
         where T: AsyncWrite,
     {
-        let incr = self.flow.unclaimed_capacity();
+        if let Some(incr) = self.flow.unclaimed_capacity() {
-
-        if incr > 0 {
             let frame = frame::WindowUpdate::new(StreamId::zero(), incr);
 
             // Ensure the codec has capacity
@@ -536,9 +537,7 @@ impl<B, P> Recv<B, P>
             }
 
             // TODO: de-dup
-            let incr = stream.recv_flow.unclaimed_capacity();
+            if let Some(incr) = stream.recv_flow.unclaimed_capacity() {
-
-            if incr > 0 {
                 // Create the WINDOW_UPDATE frame
                 let frame = frame::WindowUpdate::new(stream.id, incr);
 

src/proto/streams/streams.rs

@@ -478,7 +478,7 @@ impl<B, P> StreamRef<B, P>
         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>

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() {