flow control per stream

Author: YifeiZhuang

okhttp/src/main/java/io/grpc/okhttp/OkHttpClientStream.java

@@ -281,11 +281,10 @@ public void deframeFailed(Throwable cause) {
     @Override
     @GuardedBy("lock")
     public void bytesRead(int processedBytes) {
-      processedWindow -= processedBytes;
       if (processedWindow <= initialWindowSize * Utils.DEFAULT_WINDOW_UPDATE_RATIO) {
         int delta = initialWindowSize - processedWindow;
-        window += delta;
         processedWindow += delta;
+        window = initialWindowSize;
         frameWriter.windowUpdate(id(), delta);
       }
     }
@@ -321,11 +320,12 @@ public void transportHeadersReceived(List<Header> headers, boolean endOfStream)
      * Must be called with holding the transport lock.
      */
     @GuardedBy("lock")
-    public void transportDataReceived(okio.Buffer frame, boolean endOfStream) {
+    public void transportDataReceived(okio.Buffer frame, boolean endOfStream, int paddingLen) {
       // We only support 16 KiB frames, and the max permitted in HTTP/2 is 16 MiB. This is verified
       // in OkHttp's Http2 deframer. In addition, this code is after the data has been read.
       int length = (int) frame.size();
       window -= length;
+      processedWindow -= (length + paddingLen);
       if (window < 0) {
         frameWriter.rstStream(id(), ErrorCode.FLOW_CONTROL_ERROR);
         transport.finishStream(

okhttp/src/main/java/io/grpc/okhttp/OkHttpClientTransport.java

@@ -1167,7 +1167,7 @@ public void data(boolean inFinished, int streamId, BufferedSource in, int length
         synchronized (lock) {
           // TODO(b/145386688): This access should be guarded by 'stream.transportState().lock';
           // instead found: 'OkHttpClientTransport.this.lock'
-          stream.transportState().transportDataReceived(buf, inFinished);
+          stream.transportState().transportDataReceived(buf, inFinished, paddedLength - length);
         }
       }
 

okhttp/src/test/java/io/grpc/okhttp/OkHttpClientTransportTest.java

@@ -792,14 +792,14 @@ public void windowUpdate() throws Exception {
     frameHandler().data(false, 3, buffer, messageFrameLength, messageFrameLength);
 
     verify(frameWriter, timeout(TIME_OUT_MS))
-        .windowUpdate(eq(3), eq((long) 2 * messageFrameLength));
+        .windowUpdate(eq(3), eq((long) 2 * messageFrameLength + paddingLength));
 
     // Stream 2 receives another message
     buffer = createMessageFrame(fakeMessage);
     frameHandler().data(false, 5, buffer, messageFrameLength, messageFrameLength);
 
     verify(frameWriter, timeout(TIME_OUT_MS))
-        .windowUpdate(eq(5), eq((long) 2 * messageFrameLength));
+        .windowUpdate(eq(5), eq((long) 2 * messageFrameLength + paddingLength));
     verify(frameWriter, timeout(TIME_OUT_MS))
         .windowUpdate(eq(0), eq((long) 2 * messageFrameLength));