http2: send client conn flow control bytes back immediately

HTTP/2 server must return client connection flow control bytes
back immediately on receiving data frame. Connection flow control
updates must not depend on whether data has been read by application
handler or not. This prevents fast streams from starving due to
slow streams.

Author: jronak

http2/server.go

@@ -1772,10 +1772,18 @@ func (sc *serverConn) processData(f *DataFrame) error {
 			st.bodyBytes += int64(len(data))
 		}
 
+		// Connection level flow control window update must be
+		// sent immediately after receiving data frame.
+		// This separates connection's flow control from body reads
+		// as connection's flow control must not depend on whether
+		// body has been read by application handler or not.
+		// This prevents fast streams from starving due to other
+		// slow streams.
+		sc.sendWindowUpdate32(nil, int32(f.Length))
+
 		// Return any padded flow control now, since we won't
 		// refund it later on body reads.
 		if pad := int32(f.Length) - int32(len(data)); pad > 0 {
-			sc.sendWindowUpdate32(nil, pad)
 			sc.sendWindowUpdate32(st, pad)
 		}
 	}
@@ -2317,7 +2325,6 @@ func (sc *serverConn) noteBodyReadFromHandler(st *stream, n int, err error) {
 
 func (sc *serverConn) noteBodyRead(st *stream, n int) {
 	sc.serveG.check()
-	sc.sendWindowUpdate(nil, n) // conn-level
 	if st.state != stateHalfClosedRemote && st.state != stateClosed {
 		// Don't send this WINDOW_UPDATE if the stream is closed
 		// remotely.

http2/server_test.go

@@ -1251,19 +1251,21 @@ func TestServer_Handler_Sends_WindowUpdate(t *testing.T) {
 		EndHeaders:    true,
 	})
 	st.writeData(1, false, []byte("abcdef"))
+	// Expect to immediately get connection level 6 bytes back.
+	st.wantWindowUpdate(0, 6)
+
 	puppet.do(readBodyHandler(t, "abc"))
-	st.wantWindowUpdate(0, 3)
 	st.wantWindowUpdate(1, 3)
 
 	puppet.do(readBodyHandler(t, "def"))
-	st.wantWindowUpdate(0, 3)
 	st.wantWindowUpdate(1, 3)
 
 	st.writeData(1, true, []byte("ghijkl")) // END_STREAM here
+	st.wantWindowUpdate(0, 6)
+
+	// no more stream-level window updates, since END_STREAM
 	puppet.do(readBodyHandler(t, "ghi"))
 	puppet.do(readBodyHandler(t, "jkl"))
-	st.wantWindowUpdate(0, 3)
-	st.wantWindowUpdate(0, 3) // no more stream-level, since END_STREAM
 }
 
 // the version of the TestServer_Handler_Sends_WindowUpdate with padding.
@@ -1286,17 +1288,17 @@ func TestServer_Handler_Sends_WindowUpdate_Padding(t *testing.T) {
 	})
 	st.writeDataPadded(1, false, []byte("abcdef"), []byte{0, 0, 0, 0})
 
-	// Expect to immediately get our 5 bytes of padding back for
-	// both the connection and stream (4 bytes of padding + 1 byte of length)
-	st.wantWindowUpdate(0, 5)
+	// Expect to immediately get our 6 bytes of data + 5 bytes of padding
+	//  back at connection level.
+	// (4 bytes of padding + 1 byte of padding length)
+	st.wantWindowUpdate(0, 11)
+	// Expect to immediately get our 5 bytes of padding back at stream level.
 	st.wantWindowUpdate(1, 5)
 
 	puppet.do(readBodyHandler(t, "abc"))
-	st.wantWindowUpdate(0, 3)
 	st.wantWindowUpdate(1, 3)
 
 	puppet.do(readBodyHandler(t, "def"))
-	st.wantWindowUpdate(0, 3)
 	st.wantWindowUpdate(1, 3)
 }