Replace the HTTP2 client stream outbound backpressure workaround with…

… the use of the outbound write queue
eclipse-vertx · Jan 12, 2024 · 80dde74 · 80dde74
1 parent cf50ac4
commit 80dde74
Showing 1 changed file with 7 additions and 28 deletions.
diff --git a/src/main/java/io/vertx/core/http/impl/Http2ClientConnection.java b/src/main/java/io/vertx/core/http/impl/Http2ClientConnection.java
@@ -242,14 +242,11 @@ static abstract class Stream extends VertxHttp2Stream<Http2ClientConnection> {
     protected Handler<Throwable> exceptionHandler;
     protected Handler<HttpClientPush> pushHandler;
     protected Handler<Void> closeHandler;
-    protected long writeWindow;
-    protected final long windowSize;
 
     Stream(Http2ClientConnection conn, ContextInternal context, boolean push) {
       super(conn, context);
 
       this.push = push;
-      this.windowSize = conn.getWindowSize();
     }
 
     void onContinue() {
@@ -450,8 +447,8 @@ public boolean writeQueueFull() {
     }
 
     @Override
-    public synchronized boolean isNotWritable() {
-      return writeWindow > windowSize;
+    public boolean isNotWritable() {
+      return !messageQueue.isWritable();
     }
 
     @Override
@@ -510,6 +507,10 @@ void handleReset(long errorCode) {
 
     @Override
     void handleWriteQueueDrained() {
+      Handler<Void> handler = drainHandler;
+      if (handler != null) {
+        context.dispatch(null, handler);
+      }
     }
 
     @Override
@@ -634,29 +635,7 @@ private void createStream(HttpRequestHead head, Http2Headers headers) throws Htt
     public Future<Void> writeBuffer(ByteBuf buf, boolean end) {
       Promise<Void> promise = context.promise();
       writeData(buf, end, promise);
-      Future<Void> fut = promise.future();
-      if (buf != null) {
-        int size = buf.readableBytes();
-        synchronized (this) {
-          writeWindow += size;
-        }
-        fut = fut.andThen(ar -> {
-          Handler<Void> drainHandler;
-          synchronized (this) {
-            boolean full = writeWindow > windowSize;
-            writeWindow -= size;
-            if (full && writeWindow <= windowSize) {
-              drainHandler = this.drainHandler;
-            } else {
-              drainHandler = null;
-            }
-          }
-          if (drainHandler != null) {
-            drainHandler.handle(null);
-          }
-        });
-      }
-      return fut;
+      return promise.future();
     }
 
     @Override