Fix a few bugs related to non event-loop thread writes.

- Incorrect HTTP server metrics report for non event-loop thread writes #5222 - HTTP/2 push is not flushed when written from a non event-loop thread #5223
eclipse-vertx · Jun 4, 2024 · aec1041 · aec1041
1 parent 2c594e5
commit aec1041
Show file tree

Hide file tree

Showing 7 changed files with 98 additions and 38 deletions.
diff --git a/src/main/java/io/vertx/core/http/impl/Http2ServerStream.java b/src/main/java/io/vertx/core/http/impl/Http2ServerStream.java
@@ -10,6 +10,7 @@
  */
 package io.vertx.core.http.impl;
 
+import io.netty.channel.EventLoop;
 import io.netty.handler.codec.http.HttpHeaderNames;
 import io.netty.handler.codec.http.HttpHeaderValues;
 import io.netty.handler.codec.http2.Http2Headers;
@@ -19,12 +20,10 @@
 import io.vertx.core.buffer.Buffer;
 import io.vertx.core.http.HttpFrame;
 import io.vertx.core.http.HttpMethod;
-import io.vertx.core.http.HttpServerRequest;
 import io.vertx.core.http.StreamPriority;
 import io.vertx.core.http.impl.headers.Http2HeadersAdaptor;
 import io.vertx.core.impl.ContextInternal;
 import io.vertx.core.net.HostAndPort;
-import io.vertx.core.net.impl.HostAndPortImpl;
 import io.vertx.core.spi.metrics.HttpServerMetrics;
 import io.vertx.core.spi.metrics.Metrics;
 import io.vertx.core.spi.observability.HttpRequest;
@@ -242,13 +241,23 @@ public Object metric() {
     return metric;
   }
 
-  public HttpServerRequest routed(String route) {
+  public void routed(String route) {
     if (METRICS_ENABLED) {
-      HttpServerMetrics metrics = conn.metrics();
-      if (metrics != null && !responseEnded) {
-        metrics.requestRouted(metric, route);
+      EventLoop eventLoop = vertx.getOrCreateContext().nettyEventLoop();
+      synchronized (this) {
+        if (shouldQueue(eventLoop)) {
+          queueForWrite(eventLoop, () -> routedInternal(route));
+          return;
+        }
       }
+      routedInternal(route);
+    }
+  }
+
+  private void routedInternal(String route) {
+    HttpServerMetrics metrics = conn.metrics();
+    if (metrics != null && !responseEnded) {
+      metrics.requestRouted(metric, route);
     }
-    return null;
   }
 }
diff --git a/src/main/java/io/vertx/core/http/impl/VertxHttp2ConnectionHandler.java b/src/main/java/io/vertx/core/http/impl/VertxHttp2ConnectionHandler.java
@@ -335,14 +335,8 @@ io.netty.util.concurrent.Future<Integer> writePushPromise(int streamId, Http2Hea
         future.setFailure(fut.cause());
       }
     });
-    EventExecutor executor = chctx.executor();
-    if (executor.inEventLoop()) {
-      _writePushPromise(streamId, promisedStreamId, headers, promise);
-    } else {
-      executor.execute(() -> {
-        _writePushPromise(streamId, promisedStreamId, headers, promise);
-      });
-    }
+    _writePushPromise(streamId, promisedStreamId, headers, promise);
+    checkFlush();
     return future;
   }
 

diff --git a/src/main/java/io/vertx/core/http/impl/VertxHttp2Stream.java b/src/main/java/io/vertx/core/http/impl/VertxHttp2Stream.java
@@ -71,7 +71,6 @@ abstract class VertxHttp2Stream<C extends Http2ConnectionBase> {
             conn.consumeCredits(this.stream, len);
           }
         });
-        bytesRead += data.length();
         handleData(data);
       }
     });
@@ -117,6 +116,7 @@ void onHeaders(Http2Headers headers, StreamPriority streamPriority) {
   }
 
   void onData(Buffer data) {
+    bytesRead += data.length();
     conn.reportBytesRead(data.length());
     context.execute(data, pending::write);
   }
@@ -191,10 +191,10 @@ final void writeHeaders(Http2Headers headers, boolean end, boolean checkFlush, H
 
   void doWriteHeaders(Http2Headers headers, boolean end, boolean checkFlush, Handler<AsyncResult<Void>> handler) {
     FutureListener<Void> promise = handler == null ? null : context.promise(handler);
-    conn.handler.writeHeaders(stream, headers, end, priority.getDependency(), priority.getWeight(), priority.isExclusive(), checkFlush, promise);
     if (end) {
       endWritten();
     }
+    conn.handler.writeHeaders(stream, headers, end, priority.getDependency(), priority.getWeight(), priority.isExclusive(), checkFlush, promise);
   }
 
   protected void endWritten() {
@@ -241,10 +241,10 @@ void doWriteData(ByteBuf buf, boolean end, Handler<AsyncResult<Void>> handler) {
     bytesWritten += numOfBytes;
     conn.reportBytesWritten(numOfBytes);
     FutureListener<Void> promise = handler == null ? null : context.promise(handler);
-    conn.handler.writeData(stream, chunk, end, promise);
     if (end) {
       endWritten();
     }
+    conn.handler.writeData(stream, chunk, end, promise);
   }
 
   final void writeReset(long code) {

diff --git a/src/test/java/io/vertx/core/http/Http1xMetricsTest.java b/src/test/java/io/vertx/core/http/Http1xMetricsTest.java
@@ -10,14 +10,19 @@
  */
 package io.vertx.core.http;
 
+import io.vertx.core.ThreadingModel;
 import org.junit.Test;
 
 import java.util.concurrent.CountDownLatch;
 
 public class Http1xMetricsTest extends HttpMetricsTestBase {
 
   public Http1xMetricsTest() {
-    super(HttpVersion.HTTP_1_1);
+    this(ThreadingModel.EVENT_LOOP);
+  }
+
+  protected Http1xMetricsTest(ThreadingModel threadingModel) {
+    super(HttpVersion.HTTP_1_1, threadingModel);
   }
 
   @Test

diff --git a/src/test/java/io/vertx/core/http/Http1xWorkerMetricsTest.java b/src/test/java/io/vertx/core/http/Http1xWorkerMetricsTest.java
@@ -0,0 +1,23 @@
+/*
+ * Copyright (c) 2011-2019 Contributors to the Eclipse Foundation
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License 2.0 which is available at
+ * http://www.eclipse.org/legal/epl-2.0, or the Apache License, Version 2.0
+ * which is available at https://www.apache.org/licenses/LICENSE-2.0.
+ *
+ * SPDX-License-Identifier: EPL-2.0 OR Apache-2.0
+ */
+package io.vertx.core.http;
+
+import io.vertx.core.ThreadingModel;
+import org.junit.Test;
+
+import java.util.concurrent.CountDownLatch;
+
+public class Http1xWorkerMetricsTest extends Http1xMetricsTest {
+
+  public Http1xWorkerMetricsTest() {
+    super(ThreadingModel.EVENT_LOOP);
+  }
+}
diff --git a/src/test/java/io/vertx/core/http/Http2MetricsTest.java b/src/test/java/io/vertx/core/http/Http2MetricsTest.java
@@ -10,8 +10,10 @@
  */
 package io.vertx.core.http;
 
+import io.vertx.core.ThreadingModel;
 import io.vertx.test.core.TestUtils;
 import io.vertx.test.fakemetrics.*;
+import org.junit.Ignore;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.junit.runners.Parameterized;
@@ -27,19 +29,21 @@ public class Http2MetricsTest extends HttpMetricsTestBase {
   public static Collection<Object[]> params() {
     ArrayList<Object[]> params = new ArrayList<>();
     // h2
-    params.add(new Object[] { Http2TestBase.createHttp2ClientOptions(), Http2TestBase.createHttp2ServerOptions(HttpTestBase.DEFAULT_HTTP_PORT, HttpTestBase.DEFAULT_HTTP_HOST) });
+    params.add(new Object[] { Http2TestBase.createHttp2ClientOptions(), Http2TestBase.createHttp2ServerOptions(HttpTestBase.DEFAULT_HTTP_PORT, HttpTestBase.DEFAULT_HTTP_HOST), ThreadingModel.EVENT_LOOP });
+    // h2 + worker
+    params.add(new Object[] { Http2TestBase.createHttp2ClientOptions(), Http2TestBase.createHttp2ServerOptions(HttpTestBase.DEFAULT_HTTP_PORT, HttpTestBase.DEFAULT_HTTP_HOST), ThreadingModel.WORKER });
     // h2c with upgrade
-    params.add(new Object[] { new HttpClientOptions().setProtocolVersion(HttpVersion.HTTP_2).setHttp2ClearTextUpgrade(true), new HttpServerOptions().setPort(HttpTestBase.DEFAULT_HTTP_PORT).setHost(HttpTestBase.DEFAULT_HTTP_HOST) });
-    // h2c directq
-    params.add(new Object[] { new HttpClientOptions().setProtocolVersion(HttpVersion.HTTP_2).setHttp2ClearTextUpgrade(false), new HttpServerOptions().setPort(HttpTestBase.DEFAULT_HTTP_PORT).setHost(HttpTestBase.DEFAULT_HTTP_HOST) });
+    params.add(new Object[] { new HttpClientOptions().setProtocolVersion(HttpVersion.HTTP_2).setHttp2ClearTextUpgrade(true), new HttpServerOptions().setPort(HttpTestBase.DEFAULT_HTTP_PORT).setHost(HttpTestBase.DEFAULT_HTTP_HOST), ThreadingModel.EVENT_LOOP  });
+    // h2c direct
+    params.add(new Object[] { new HttpClientOptions().setProtocolVersion(HttpVersion.HTTP_2).setHttp2ClearTextUpgrade(false), new HttpServerOptions().setPort(HttpTestBase.DEFAULT_HTTP_PORT).setHost(HttpTestBase.DEFAULT_HTTP_HOST), ThreadingModel.EVENT_LOOP  });
     return params;
   }
 
   private HttpClientOptions clientOptions;
   private HttpServerOptions serverOptions;
 
-  public Http2MetricsTest(HttpClientOptions clientOptions, HttpServerOptions serverOptions) {
-    super(HttpVersion.HTTP_2);
+  public Http2MetricsTest(HttpClientOptions clientOptions, HttpServerOptions serverOptions, ThreadingModel threadingModel) {
+    super(HttpVersion.HTTP_2, threadingModel);
 
     this.clientOptions = clientOptions;
     this.serverOptions = serverOptions.setHandle100ContinueAutomatically(true);
@@ -70,10 +74,11 @@ public void testPushPromise() throws Exception {
         AtomicInteger numBuffer = new AtomicInteger(numBuffers);
         vertx.setPeriodic(1, timerID -> {
           if (numBuffer.getAndDecrement() == 0) {
-            pushedResp.end();
-            assertNull(serverMetrics.getResponseMetric("/wibble"));
+            pushedResp.end().onComplete(onSuccess(v -> {
+              assertNull(serverMetrics.getResponseMetric("/wibble"));
+              complete();
+            }));
             vertx.cancelTimer(timerID);
-            complete();
           } else {
             pushedResp.write(TestUtils.randomBuffer(1000));
           }

diff --git a/src/test/java/io/vertx/core/http/HttpMetricsTestBase.java b/src/test/java/io/vertx/core/http/HttpMetricsTestBase.java
@@ -13,10 +13,12 @@
 
 import io.vertx.core.Context;
 import io.vertx.core.Future;
+import io.vertx.core.ThreadingModel;
 import io.vertx.core.VertxOptions;
 import io.vertx.core.buffer.Buffer;
 import io.vertx.core.http.impl.HttpClientImpl;
 import io.vertx.core.http.impl.HttpServerRequestInternal;
+import io.vertx.core.impl.VertxInternal;
 import io.vertx.core.metrics.MetricsOptions;
 import io.vertx.core.net.NetClient;
 import io.vertx.core.net.SocketAddress;
@@ -46,9 +48,19 @@
 public abstract class HttpMetricsTestBase extends HttpTestBase {
 
   private final HttpVersion protocol;
+  private final ThreadingModel threadingModel;
 
-  public HttpMetricsTestBase(HttpVersion protocol) {
+  public HttpMetricsTestBase(HttpVersion protocol, ThreadingModel threadingModel) {
     this.protocol = protocol;
+    this.threadingModel = threadingModel;
+  }
+
+  @Override
+  protected void startServer(SocketAddress bindAddress, Context context, HttpServer server) throws Exception {
+    if (threadingModel == ThreadingModel.WORKER) {
+      context = ((VertxInternal) vertx).createWorkerContext();
+    }
+    super.startServer(bindAddress, context, server);
   }
 
   @Override
@@ -83,6 +95,8 @@ public void testHttpMetricsLifecycle() throws Exception {
       assertTrue(serverMetric.get().socket.connected.get());
       assertNull(serverMetric.get().route.get());
       req.routed("/route/:param");
+      // Worker can wait
+      assertWaitUntil(() -> serverMetric.get().route.get() != null);
       assertEquals("/route/:param", serverMetric.get().route.get());
       req.bodyHandler(buff -> {
         assertEquals(contentLength, buff.length());
@@ -93,14 +107,21 @@ public void testHttpMetricsLifecycle() throws Exception {
         vertx.setPeriodic(1, timerID -> {
           Buffer chunk = TestUtils.randomBuffer(chunkSize);
           if (numBuffer.decrementAndGet() == 0) {
-            resp.end(chunk);
-            assertTrue(serverMetric.get().responseEnded.get());
-            assertEquals(contentLength, serverMetric.get().bytesWritten.get());
-            assertNull(serverMetrics.getRequestMetric(req));
+            resp
+              .end(chunk)
+              .onComplete(onSuccess(v -> {
+                assertTrue(serverMetric.get().responseEnded.get());
+                assertFalse(serverMetric.get().failed.get());
+                assertEquals(contentLength, serverMetric.get().bytesWritten.get());
+                assertNull(serverMetrics.getRequestMetric(req));
+              }));
             vertx.cancelTimer(timerID);
           } else {
-            resp.write(chunk);
-            assertSame(serverMetric.get().response.get(), resp);
+            resp
+              .write(chunk)
+              .onComplete(onSuccess(v -> {
+                assertSame(serverMetric.get().response.get(), resp);
+              }));
           }
         });
       });
@@ -205,9 +226,7 @@ public void testHttpClientLifecycle() throws Exception {
         });
       });
     });
-    CountDownLatch listenLatch = new CountDownLatch(1);
-    server.listen(HttpTestBase.DEFAULT_HTTP_PORT, "localhost", onSuccess(s -> { listenLatch.countDown(); }));
-    awaitLatch(listenLatch);
+    startServer(testAddress);
     FakeHttpClientMetrics clientMetrics = FakeMetricsBase.getMetrics(client);
     CountDownLatch responseBeginLatch = new CountDownLatch(1);
     CountDownLatch responseEndLatch = new CountDownLatch(1);
@@ -301,8 +320,13 @@ public void testRouteMetrics() throws Exception {
       HttpServerMetric metric = metrics.getRequestMetric(req);
       assertNull(metric.route.get());
       req.routed("MyRoute");
+      // Worker can wait
+      assertWaitUntil(() -> metric.route.get() != null);
       assertEquals("MyRoute", metric.route.get());
+      metric.route.set(null);
       req.routed("MyRoute - rerouted");
+      // Worker can wait
+      assertWaitUntil(() -> metric.route.get() != null);
       assertEquals("MyRoute - rerouted", metric.route.get());
       req.response().end();
       testComplete();