DefaultHttp2RemoteFlowController not allocating all available bytes

Motivation:
DefaultHttp2RemoteFlowController's allocation algorithm may not allocate all bytes that are available in the connection window. If the 'fair share' based upon weight is not fully used by sibling nodes it was not correctly re-distributed to other sibilings which may be able to utilize part / all of that share.

Modifications:
- Add a unit test which demonstrates the issue.
- Modify the allocation algorithm to ensure all available bytes are allocated.

Result:
Fixes netty#4266

codec-http2/src/main/java/io/netty/handler/codec/http2/DefaultHttp2RemoteFlowController.java

@@ -288,7 +288,12 @@ private int maxUsableChannelBytes() {
         return min(connectionState().windowSize(), useableBytes);
     }
 
-    private int writableBytes(int requestedBytes) {
+    /**
+     * Package private for testing purposes only!
+     * @param requestedBytes The desired amount of bytes.
+     * @return The amount of bytes that can be supported by underlying {@link Channel} without queuing "too-much".
+     */
+    final int writableBytes(int requestedBytes) {
         return Math.min(requestedBytes, maxUsableChannelBytes());
     }
 
@@ -386,15 +391,6 @@ public boolean visit(Http2Stream child) throws Http2Exception {
                 bytesAllocated += bytesForChild;
                 nextConnectionWindow -= bytesForChild;
                 bytesForTree -= bytesForChild;
-
-                // If this subtree still wants to send then re-insert into children list and re-consider for next
-                // iteration. This is needed because we don't yet know if all the peers will be able to use
-                // all of their "fair share" of the connection window, and if they don't use it then we should
-                // divide their unused shared up for the peers who still want to send.
-                if (nextConnectionWindow > 0 && state.streamableBytesForTree() > 0) {
-                    stillHungry(child);
-                    nextTotalWeight += child.weight();
-                }
             }
 
             // Allocate any remaining bytes to the children of this stream.
@@ -404,7 +400,18 @@ public boolean visit(Http2Stream child) throws Http2Exception {
                 nextConnectionWindow -= childBytesAllocated;
             }
 
-            return nextConnectionWindow > 0;
+            if (nextConnectionWindow > 0) {
+                // If this subtree still wants to send then it should be re-considered to take bytes that are unused by
+                // sibling nodes. This is needed because we don't yet know if all the peers will be able to use all of
+                // their "fair share" of the connection window, and if they don't use it then we should divide their
+                // unused shared up for the peers who still want to send.
+                if (state.streamableBytesForTree() > 0) {
+                    stillHungry(child);
+                }
+                return true;
+            }
+
+            return false;
         }
 
         void feedHungryChildren() throws Http2Exception {
@@ -438,15 +445,16 @@ void feedHungryChildren() throws Http2Exception {
          * Indicates that the given child is still hungry (i.e. still has streamable bytes that can
          * fit within the current connection window).
          */
-        void stillHungry(Http2Stream child) {
+        private void stillHungry(Http2Stream child) {
             ensureSpaceIsAllocated(nextTail);
             stillHungry[nextTail++] = child;
+            nextTotalWeight += child.weight();
         }
 
         /**
          * Ensures that the {@link #stillHungry} array is properly sized to hold the given index.
          */
-        void ensureSpaceIsAllocated(int index) {
+        private void ensureSpaceIsAllocated(int index) {
             if (stillHungry == null) {
                 // Initial size is 1/4 the number of children. Clipping the minimum at 2, which will over allocate if
                 // maxSize == 1 but if this was true we shouldn't need to re-allocate because the 1 child should get

codec-http2/src/test/java/io/netty/handler/codec/http2/DefaultHttp2RemoteFlowControllerTest.java

@@ -755,6 +755,56 @@ public void reprioritizeShouldAdjustOutboundFlow() throws Http2Exception {
         verify(listener, times(1)).streamWritten(stream(STREAM_D), 5);
     }
 
+    /**
+     * Test that the maximum allowed amount the flow controller allows to be sent is always fully allocated if
+     * the streams have at least this much data to send. See https://github.com/netty/netty/issues/4266.
+     * <pre>
+     *            0
+     *          / | \
+     *        /   |   \
+     *      A(0) B(0) C(0)
+     *     /
+     *    D(> allowed to send in 1 allocation attempt)
+     * </pre>
+     */
+    @Test
+    public void unstreamableParentsShouldFeedHungryChildren() throws Http2Exception {
+        // Max all connection windows. We don't want this being a limiting factor in the test.
+        maxStreamWindow(CONNECTION_STREAM_ID);
+        maxStreamWindow(STREAM_A);
+        maxStreamWindow(STREAM_B);
+        maxStreamWindow(STREAM_C);
+        maxStreamWindow(STREAM_D);
+
+        // Setup the priority tree.
+        setPriority(STREAM_A, 0, (short) 32, false);
+        setPriority(STREAM_B, 0, (short) 16, false);
+        setPriority(STREAM_C, 0, (short) 16, false);
+        setPriority(STREAM_D, STREAM_A, (short) 16, false);
+
+        // The bytesBeforeUnwritable defaults to Long.MAX_VALUE, we need to leave room to send enough data to exceed
+        // the writableBytes, and so we must reduce this value to something no-zero.
+        when(channel.bytesBeforeUnwritable()).thenReturn(1L);
+
+        // Calculate the max amount of data the flow controller will allow to be sent now.
+        final int writableBytes = controller.writableBytes(window(CONNECTION_STREAM_ID));
+
+        // This is insider knowledge into how writePendingBytes works. Because the algorithm will keep looping while
+        // the channel is writable, we simulate that the channel will become unwritable after the first write.
+        when(channel.isWritable()).thenReturn(false);
+
+        // Send enough so it can not be completely written out
+        final int expectedUnsentAmount = 1;
+        // Make sure we don't overflow
+        assertTrue(Integer.MAX_VALUE - expectedUnsentAmount > writableBytes);
+        FakeFlowControlled dataD = new FakeFlowControlled(writableBytes + expectedUnsentAmount);
+        sendData(STREAM_D, dataD);
+        controller.writePendingBytes();
+
+        dataD.assertPartiallyWritten(writableBytes);
+        verify(listener, times(1)).streamWritten(eq(stream(STREAM_D)), eq(writableBytes));
+    }
+
     /**
      * In this test, we root all streams at the connection, and then verify that data is split appropriately based on
      * weight (all available data is the same).
@@ -1431,6 +1481,10 @@ private void exhaustStreamWindow(int streamId) throws Http2Exception {
         incrementWindowSize(streamId, -window(streamId));
     }
 
+    private void maxStreamWindow(int streamId) throws Http2Exception {
+        incrementWindowSize(streamId, Http2CodecUtil.MAX_INITIAL_WINDOW_SIZE - window(streamId));
+    }
+
     private int window(int streamId) throws Http2Exception {
         return controller.windowSize(stream(streamId));
     }