VertxConnection message handling fast path: avoid creating a pending queue when handling a message in VertxConnection when it is not necessary.

Author: vietj

vertx-core/src/main/java/io/vertx/core/net/impl/VertxConnection.java

@@ -61,7 +61,8 @@ public class VertxConnection extends ConnectionBase {
   private Handler<Void> shutdownHandler;
 
   // State accessed exclusively from the event loop thread
-  private final Deque<Object> pending;
+  private Deque<Object> pending;
+  private boolean reentrant;
   private boolean read;
   private boolean needsFlush;
   private boolean draining;
@@ -75,7 +76,6 @@ public VertxConnection(ContextInternal context, ChannelHandlerContext chctx) {
     this.channelWritable = chctx.channel().isWritable();
     this.messageQueue = new InternalMessageChannel(chctx.channel().eventLoop());
     this.voidPromise = new VoidChannelPromise(chctx.channel(), false);
-    this.pending = new ArrayDeque<>();
     this.autoRead = true;
   }
 
@@ -218,26 +218,59 @@ void channelWritabilityChanged() {
    * This method is exclusively called by {@code VertxHandler} to read a message on the event-loop thread.
    */
   final void read(Object msg) {
-    read = true;
     if (METRICS_ENABLED) {
       reportBytesRead(msg);
     }
+    read = true;
+    if (!reentrant && !paused && pending == null) {
+      // Fast path
+      reentrant = true;
+      try {
+        handleMessage(msg);
+      } finally {
+        reentrant = false;
+      }
+    } else {
+      addPending(msg);
+    }
+  }
+
+  private void addPending(Object msg) {
+    if (pending == null) {
+      pending = new ArrayDeque<>();
+    }
     pending.add(msg);
-    checkPendingMessages();
+    if (!reentrant) {
+      checkPendingMessages();
+    }
   }
 
   /**
    * This method is exclusively called by {@code VertxHandler} to signal read completion on the event-loop thread.
    */
   final void readComplete() {
     if (read) {
-      checkPendingMessages();
+      if (pending != null) {
+        checkPendingMessages();
+      }
       read = false;
       checkFlush();
       checkAutoRead();
     }
   }
 
+  private void checkPendingMessages() {
+    Object msg;
+    reentrant = true;
+    try {
+      while (!paused && (msg = pending.poll()) != null) {
+        handleMessage(msg);
+      }
+    } finally {
+      reentrant = false;
+    }
+  }
+
   public final void doPause() {
     assert chctx.executor().inEventLoop();
     paused = true;
@@ -272,25 +305,18 @@ private void checkFlush() {
 
   private void checkAutoRead() {
     if (autoRead) {
-      if (pending.size() >= 8) {
+      if (pending != null && pending.size() >= 8) {
         autoRead = false;
         chctx.channel().config().setAutoRead(false);
       }
     } else {
-      if (pending.isEmpty()) {
+      if (pending == null || pending.isEmpty()) {
         autoRead = true;
         chctx.channel().config().setAutoRead(true);
       }
     }
   }
 
-  private void checkPendingMessages() {
-    Object msg;
-    while (!paused && (msg = pending.poll()) != null) {
-      handleMessage(msg);
-    }
-  }
-
   /**
    * Like {@link #write(Object, boolean, ChannelPromise)}.
    */

vertx-core/src/test/java/io/vertx/tests/net/ConnectionBaseTest.java

@@ -420,6 +420,37 @@ public void flush(ChannelHandlerContext ctx) {
     await();
   }
 
+  @Test
+  public void testReentrantRead() throws Exception {
+    connectHandler = conn -> {
+      VertxConnection vertxConn = (VertxConnection) conn;
+      ChannelHandlerContext ctx = conn.channelHandlerContext();
+      ChannelPipeline pipeline = ctx.pipeline();
+      AtomicInteger reentrant = new AtomicInteger();
+      conn.messageHandler(msg -> {
+        assertEquals(0, reentrant.getAndIncrement());
+        switch (((ByteBuf)msg).toString(StandardCharsets.UTF_8)) {
+          case "inbound-1":
+            pipeline.fireChannelRead(Unpooled.copiedBuffer("inbound-2", StandardCharsets.UTF_8));
+            break;
+          case "inbound-2":
+            conn.end(Buffer.buffer("outbound-1"));
+            break;
+        }
+        reentrant.decrementAndGet();
+      });
+    };
+    NetSocket so = awaitFuture(client.connect(1234, "localhost"));
+    Buffer received = Buffer.buffer();
+    so.handler(received::appendBuffer);
+    so.closeHandler(v -> {
+      assertEquals("outbound-1", received.toString());
+      testComplete();
+    });
+    so.write("inbound-1").await();
+    await();
+  }
+
   @Test
   public void testResumeWhenRead() throws Exception {
     connectHandler = conn -> {
@@ -434,13 +465,16 @@ public void testResumeWhenRead() throws Exception {
             vertxConn.doResume();
             break;
           case "inbound-2":
-            conn.close();
+            conn.end(Buffer.buffer("outbound-1"));
             break;
         }
       });
     };
     NetSocket so = awaitFuture(client.connect(1234, "localhost"));
+    Buffer received = Buffer.buffer();
+    so.handler(received::appendBuffer);
     so.closeHandler(v -> {
+      assertEquals("outbound-1", received.toString());
       testComplete();
     });
     so.write("inbound-1").await();