Ensure discarded HTTP/2 data frames free flow-control space. (#4033)

#3915

okhttp-tests/src/test/java/okhttp3/internal/http2/Http2ConnectionTest.java

@@ -192,6 +192,39 @@ public final class Http2ConnectionTest {
  assertTrue(Arrays.equals("fghi".getBytes("UTF-8"), data2.data));
  }
 
+ /**
+ * Confirm that we account for discarded data frames. It's possible that data frames are in-flight
+ * just prior to us canceling a stream.
+ */
+ @Test public void discardedDataFramesAreCounted() throws Exception {
+ // Write the mocking script.
+ peer.sendFrame().settings(new Settings());
+ peer.acceptFrame(); // ACK
+ peer.acceptFrame(); // SYN_STREAM 3
+ peer.sendFrame().headers(3, headerEntries("a", "apple"));
+ peer.sendFrame().data(false, 3, data(1024), 1024);
+ peer.acceptFrame(); // RST_STREAM
+ peer.sendFrame().data(true, 3, data(1024), 1024);
+ peer.acceptFrame(); // RST_STREAM
+ peer.play();
+
+ Http2Connection connection = connect(peer);
+ Http2Stream stream1 = connection.newStream(headerEntries("b", "bark"), false);
+ Source source = stream1.getSource();
+ Buffer buffer = new Buffer();
+ while (buffer.size() != 1024) source.read(buffer, 1024);
+ stream1.close(ErrorCode.CANCEL);
+
+ InFrame frame1 = peer.takeFrame();
+ assertEquals(Http2.TYPE_HEADERS, frame1.type);
+ InFrame frame2 = peer.takeFrame();
+ assertEquals(Http2.TYPE_RST_STREAM, frame2.type);
+ InFrame frame3 = peer.takeFrame();
+ assertEquals(Http2.TYPE_RST_STREAM, frame3.type);
+
+ assertEquals(2048, connection.unacknowledgedBytesRead);
+ }
+
  @Test public void receiveGoAwayHttp2() throws Exception {
  // write the mocking script
  peer.sendFrame().settings(new Settings());

okhttp-tests/src/test/java/okhttp3/internal/http2/HttpOverHttp2Test.java

@@ -334,6 +334,39 @@ private static OkHttpClient buildHttp2Client() {
  assertEquals("abc", response2.body().string());
  }
 
+ @Test public void connectionWindowUpdateOnClose() throws Exception {
+ server.enqueue(new MockResponse()
+ .setBody(new Buffer().write(new byte[Http2Connection.OKHTTP_CLIENT_WINDOW_SIZE + 1])));
+ server.enqueue(new MockResponse()
+ .setBody("abc"));
+
+ Call call1 = client.newCall(new Request.Builder()
+ .url(server.url("/"))
+ .build());
+ Response response1 = call1.execute();
+
+ // Wait until the server has completely filled the stream and connection flow-control windows.
+ int expectedFrameCount = Http2Connection.OKHTTP_CLIENT_WINDOW_SIZE / 16384;
+ int dataFrameCount = 0;
+ while (dataFrameCount < expectedFrameCount) {
+ String log = http2Handler.take();
+ if (log.equals("FINE: << 0x00000003 16384 DATA ")) {
+ dataFrameCount++;
+ }
+ }
+
+ // Cancel the call and close the response body. This should discard the buffered data and update
+ // the connnection flow-control window.
+ call1.cancel();
+ response1.close();
+
+ Call call2 = client.newCall(new Request.Builder()
+ .url(server.url("/"))
+ .build());
+ Response response2 = call2.execute();
+ assertEquals("abc", response2.body().string());
+ }
+
  /** https:/square/okhttp/issues/373 */
  @Test @Ignore public void synchronousRequest() throws Exception {
  server.enqueue(new MockResponse().setBody("A"));

okhttp-urlconnection/src/test/java/okhttp3/UrlConnectionCacheTest.java

@@ -1079,7 +1079,7 @@ private RecordedRequest assertClientSuppliedCondition(MockResponse seed, String
  Date lastModifiedDate = new Date(System.currentTimeMillis() + TimeUnit.HOURS.toMillis(-1));
  Date servedDate = new Date(System.currentTimeMillis() + TimeUnit.HOURS.toMillis(-2));
  DateFormat dateFormat = new SimpleDateFormat("EEE dd-MMM-yyyy HH:mm:ss z", Locale.US);
- dateFormat.setTimeZone(TimeZone.getTimeZone("EDT"));
+ dateFormat.setTimeZone(TimeZone.getTimeZone("America/New_York"));
  String lastModifiedString = dateFormat.format(lastModifiedDate);
  String servedString = dateFormat.format(servedDate);
 

okhttp/src/main/java/okhttp3/internal/http2/Http2Connection.java

@@ -200,6 +200,14 @@ public synchronized int maxConcurrentStreams() {
  return peerSettings.getMaxConcurrentStreams(Integer.MAX_VALUE);
  }
 
+ synchronized void updateConnectionFlowControl(long read) {
+ unacknowledgedBytesRead += read;
+ if (unacknowledgedBytesRead >= okHttpSettings.getInitialWindowSize() / 2) {
+ writeWindowUpdateLater(0, unacknowledgedBytesRead);
+ unacknowledgedBytesRead = 0;
+ }
+ }
+
  /**
  * Returns a new server-initiated stream.
  *
@@ -622,6 +630,7 @@ class ReaderRunnable extends NamedRunnable implements Http2Reader.Handler {
  Http2Stream dataStream = getStream(streamId);
  if (dataStream == null) {
  writeSynResetLater(streamId, ErrorCode.PROTOCOL_ERROR);
+ updateConnectionFlowControl(length);
  source.skip(length);
  return;
  }

okhttp/src/main/java/okhttp3/internal/http2/Http2Stream.java

@@ -358,15 +358,7 @@ private final class FramingSource implements Source {
 
  if (read != -1) {
  // Update connection.unacknowledgedBytesRead outside the stream lock.
- synchronized (connection) { // Multiple application threads may hit this section.
- connection.unacknowledgedBytesRead += read;
- if (connection.unacknowledgedBytesRead
- >= connection.okHttpSettings.getInitialWindowSize() / 2) {
- connection.writeWindowUpdateLater(0, connection.unacknowledgedBytesRead);
- connection.unacknowledgedBytesRead = 0;
- }
- }
-
+ updateConnectionFlowControl(read);
  return read;
  }
 
@@ -381,6 +373,11 @@ private final class FramingSource implements Source {
  return -1; // This source is exhausted.
  }
 
+ private void updateConnectionFlowControl(long read) {
+ assert (!Thread.holdsLock(Http2Stream.this));
+ connection.updateConnectionFlowControl(read);
+ }
+
  /** Returns once the source is either readable or finished. */
  private void waitUntilReadable() throws IOException {
  readTimeout.enter();
@@ -438,11 +435,16 @@ void receive(BufferedSource in, long byteCount) throws IOException {
  }
 
  @Override public void close() throws IOException {
+ long bytesDiscarded;
  synchronized (Http2Stream.this) {
  closed = true;
+ bytesDiscarded = readBuffer.size();
  readBuffer.clear();
  Http2Stream.this.notifyAll();
  }
+ if (bytesDiscarded > 0) {
+ updateConnectionFlowControl(bytesDiscarded);
+ }
  cancelStreamIfNecessary();
  }
  }