Repository: eclipse-vertx/vert.x

Problem statement:
Support IO_URING transport
I'm implementing support for https://github.com/netty/netty-incubator-transport-io_uring on Vert-x `Transport`, see https://github.com/franz1981/vert.x/tree/4.2.1_iouring for some initial experiment.


Hints:

Patch:
diff --git a/src/main/java/io/vertx/core/Vertx.java b/src/main/java/io/vertx/core/Vertx.java
--- a/src/main/java/io/vertx/core/Vertx.java
+++ b/src/main/java/io/vertx/core/Vertx.java
@@ -660,6 +660,12 @@ default <T> Future<T> executeBlocking(Handler<Promise<T>> blockingCodeHandler) {
   @CacheReturn
   boolean isNativeTransportEnabled();
 
+  /**
+   * @return the error (if any) that cause the unavailability of native transport when {@link #isNativeTransportEnabled()}  returns {@code false}.
+   */
+  @CacheReturn
+  Throwable unavailableNativeTransportCause();
+
   /**
    * Set a default exception handler for {@link Context}, set on {@link Context#exceptionHandler(Handler)} at creation.
    *
diff --git a/src/main/java/io/vertx/core/http/impl/Http1xUpgradeToH2CHandler.java b/src/main/java/io/vertx/core/http/impl/Http1xUpgradeToH2CHandler.java
--- a/src/main/java/io/vertx/core/http/impl/Http1xUpgradeToH2CHandler.java
+++ b/src/main/java/io/vertx/core/http/impl/Http1xUpgradeToH2CHandler.java
@@ -78,14 +78,16 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception
             if (settings != null) {
               if (initializer.context.isEventLoopContext()) {
                 ChannelPipeline pipeline = ctx.pipeline();
+                if (pipeline.get("chunkedWriter") != null) {
+                  pipeline.remove("chunkedWriter");
+                }
                 DefaultFullHttpResponse res = new DefaultFullHttpResponse(HTTP_1_1, SWITCHING_PROTOCOLS, Unpooled.EMPTY_BUFFER, false);
                 res.headers().add(HttpHeaderNames.CONNECTION, HttpHeaderValues.UPGRADE);
                 res.headers().add(HttpHeaderNames.UPGRADE, Http2CodecUtil.HTTP_UPGRADE_PROTOCOL_NAME);
-                ctx.writeAndFlush(res);
+                ctx.write(res);
                 pipeline.remove("httpEncoder");
                 if (isCompressionSupported) {
                   pipeline.remove("deflater");
-                  pipeline.remove("chunkedWriter");
                 }
                 if (isDecompressionSupported) {
                   pipeline.remove("inflater");
diff --git a/src/main/java/io/vertx/core/http/impl/HttpServerWorker.java b/src/main/java/io/vertx/core/http/impl/HttpServerWorker.java
--- a/src/main/java/io/vertx/core/http/impl/HttpServerWorker.java
+++ b/src/main/java/io/vertx/core/http/impl/HttpServerWorker.java
@@ -270,7 +270,7 @@ private void configureHttp1OrH2C(ChannelPipeline pipeline, SslChannelProvider ss
     if (options.isCompressionSupported()) {
       pipeline.addLast("deflater", new HttpChunkContentCompressor(compressionOptions));
     }
-    if (options.isSsl() || options.isCompressionSupported()) {
+    if (options.isSsl() || options.isCompressionSupported() || !vertx.transport().supportFileRegion()) {
       // only add ChunkedWriteHandler when SSL is enabled otherwise it is not needed as FileRegion is used.
       pipeline.addLast("chunkedWriter", new ChunkedWriteHandler());       // For large file / sendfile support
     }
diff --git a/src/main/java/io/vertx/core/impl/VertxImpl.java b/src/main/java/io/vertx/core/impl/VertxImpl.java
--- a/src/main/java/io/vertx/core/impl/VertxImpl.java
+++ b/src/main/java/io/vertx/core/impl/VertxImpl.java
@@ -326,6 +326,14 @@ public boolean isNativeTransportEnabled() {
     return transport != Transport.JDK;
   }
 
+  @Override
+  public Throwable unavailableNativeTransportCause() {
+    if (isNativeTransportEnabled()) {
+      return null;
+    }
+    return transport.unavailabilityCause();
+  }
+
   public FileSystem fileSystem() {
     return fileSystem;
   }
diff --git a/src/main/java/io/vertx/core/net/impl/ConnectionBase.java b/src/main/java/io/vertx/core/net/impl/ConnectionBase.java
--- a/src/main/java/io/vertx/core/net/impl/ConnectionBase.java
+++ b/src/main/java/io/vertx/core/net/impl/ConnectionBase.java
@@ -14,7 +14,7 @@
 import io.netty.buffer.Unpooled;
 import io.netty.channel.*;
 import io.netty.handler.ssl.SslHandler;
-import io.netty.handler.stream.ChunkedFile;
+import io.netty.handler.stream.ChunkedNioFile;
 import io.netty.handler.timeout.IdleStateEvent;
 import io.netty.util.AttributeKey;
 import io.netty.util.ReferenceCountUtil;
@@ -409,7 +409,7 @@ protected void handleIdle(IdleStateEvent event) {
   protected abstract void handleInterestedOpsChanged();
 
   protected boolean supportsFileRegion() {
-    return !isSsl();
+    return vertx.transport().supportFileRegion() && !isSsl();
   }
 
   protected void reportBytesRead(Object msg) {
@@ -505,7 +505,7 @@ public final ChannelFuture sendFile(RandomAccessFile raf, long offset, long leng
     ChannelPromise writeFuture = chctx.newPromise();
     if (!supportsFileRegion()) {
       // Cannot use zero-copy
-      writeToChannel(new ChunkedFile(raf, offset, length, 8192), writeFuture);
+      writeToChannel(new ChunkedNioFile(raf.getChannel(), offset, length, 8192), writeFuture);
     } else {
       // No encryption - use zero-copy.
       sendFileRegion(raf, offset, length, writeFuture);
diff --git a/src/main/java/io/vertx/core/net/impl/NetClientImpl.java b/src/main/java/io/vertx/core/net/impl/NetClientImpl.java
--- a/src/main/java/io/vertx/core/net/impl/NetClientImpl.java
+++ b/src/main/java/io/vertx/core/net/impl/NetClientImpl.java
@@ -95,7 +95,7 @@ protected void initChannel(ChannelPipeline pipeline) {
     if (logEnabled) {
       pipeline.addLast("logging", new LoggingHandler(options.getActivityLogDataFormat()));
     }
-    if (options.isSsl()) {
+    if (options.isSsl() || !vertx.transport().supportFileRegion()) {
       // only add ChunkedWriteHandler when SSL is enabled otherwise it is not needed as FileRegion is used.
       pipeline.addLast("chunkedWriter", new ChunkedWriteHandler());       // For large file / sendfile support
     }
diff --git a/src/main/java/io/vertx/core/net/impl/NetServerImpl.java b/src/main/java/io/vertx/core/net/impl/NetServerImpl.java
--- a/src/main/java/io/vertx/core/net/impl/NetServerImpl.java
+++ b/src/main/java/io/vertx/core/net/impl/NetServerImpl.java
@@ -255,8 +255,8 @@ protected void initChannel(ChannelPipeline pipeline, boolean ssl) {
     if (options.getLogActivity()) {
       pipeline.addLast("logging", new LoggingHandler(options.getActivityLogDataFormat()));
     }
-    if (ssl) {
-      // only add ChunkedWriteHandler when SSL is enabled otherwise it is not needed as FileRegion is used.
+    if (ssl || !vertx.transport().supportFileRegion()) {
+      // only add ChunkedWriteHandler when SSL is enabled or FileRegion isn't supported
       pipeline.addLast("chunkedWriter", new ChunkedWriteHandler());       // For large file / sendfile support
     }
     int idleTimeout = options.getIdleTimeout();
diff --git a/src/main/java/io/vertx/core/net/impl/transport/EpollTransport.java b/src/main/java/io/vertx/core/net/impl/transport/EpollTransport.java
--- a/src/main/java/io/vertx/core/net/impl/transport/EpollTransport.java
+++ b/src/main/java/io/vertx/core/net/impl/transport/EpollTransport.java
@@ -62,6 +62,7 @@ public static void setPendingFastOpenRequestsThreshold(int value) {
   }
 
   EpollTransport() {
+    super(true);
   }
 
   @Override
diff --git a/src/main/java/io/vertx/core/net/impl/transport/IOUringTransport.java b/src/main/java/io/vertx/core/net/impl/transport/IOUringTransport.java
new file mode 100644
--- /dev/null
+++ b/src/main/java/io/vertx/core/net/impl/transport/IOUringTransport.java
@@ -0,0 +1,160 @@
+/*
+ * 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.net.impl.transport;
+
+import java.net.SocketAddress;
+import java.util.concurrent.ThreadFactory;
+
+import io.netty.bootstrap.Bootstrap;
+import io.netty.bootstrap.ServerBootstrap;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFactory;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.ServerChannel;
+import io.netty.channel.socket.DatagramChannel;
+import io.netty.channel.socket.InternetProtocolFamily;
+import io.netty.channel.unix.DomainSocketAddress;
+import io.netty.incubator.channel.uring.IOUring;
+import io.netty.incubator.channel.uring.IOUringChannelOption;
+import io.netty.incubator.channel.uring.IOUringDatagramChannel;
+import io.netty.incubator.channel.uring.IOUringEventLoopGroup;
+import io.netty.incubator.channel.uring.IOUringServerSocketChannel;
+import io.netty.incubator.channel.uring.IOUringSocketChannel;
+import io.vertx.core.datagram.DatagramSocketOptions;
+import io.vertx.core.net.ClientOptionsBase;
+import io.vertx.core.net.NetServerOptions;
+import io.vertx.core.net.impl.SocketAddressImpl;
+
+public class IOUringTransport extends Transport {
+
+  private static volatile int pendingFastOpenRequestsThreshold = 256;
+
+  /**
+   * Return the number of of pending TFO connections in SYN-RCVD state for TCP_FASTOPEN.
+   * <p>
+   * {@see #setPendingFastOpenRequestsThreshold}
+   */
+  public static int getPendingFastOpenRequestsThreshold() {
+    return pendingFastOpenRequestsThreshold;
+  }
+
+  /**
+   * Set the number of of pending TFO connections in SYN-RCVD state for TCP_FASTOPEN
+   * <p/>
+   * If this value goes over a certain limit the server disables all TFO connections.
+   */
+  public static void setPendingFastOpenRequestsThreshold(int value) {
+    if (value < 0) {
+      throw new IllegalArgumentException("Invalid " + value);
+    }
+    pendingFastOpenRequestsThreshold = value;
+  }
+
+  IOUringTransport() {
+    super(false);
+  }
+
+  @Override
+  public boolean supportFileRegion() {
+    return false;
+  }
+
+  @Override
+  public SocketAddress convert(io.vertx.core.net.SocketAddress address) {
+    if (address.isDomainSocket()) {
+      throw new IllegalArgumentException("Domain socket not supported by IOUring transport");
+    }
+    return super.convert(address);
+  }
+
+  @Override
+  public io.vertx.core.net.SocketAddress convert(SocketAddress address) {
+    if (address instanceof DomainSocketAddress) {
+      return new SocketAddressImpl(((DomainSocketAddress) address).path());
+    }
+    return super.convert(address);
+  }
+
+  @Override
+  public boolean isAvailable() {
+    return IOUring.isAvailable();
+  }
+
+  @Override
+  public Throwable unavailabilityCause() {
+    return IOUring.unavailabilityCause();
+  }
+
+  @Override
+  public EventLoopGroup eventLoopGroup(int type, int nThreads, ThreadFactory threadFactory, int ignoredIoRatio) {
+    IOUringEventLoopGroup eventLoopGroup = new IOUringEventLoopGroup(nThreads, threadFactory);
+    return eventLoopGroup;
+  }
+
+  @Override
+  public DatagramChannel datagramChannel() {
+    return new IOUringDatagramChannel();
+  }
+
+  @Override
+  public DatagramChannel datagramChannel(InternetProtocolFamily family) {
+    return new IOUringDatagramChannel();
+  }
+
+  @Override
+  public ChannelFactory<? extends Channel> channelFactory(boolean domainSocket) {
+    if (domainSocket) {
+      throw new IllegalArgumentException();
+    }
+    return IOUringSocketChannel::new;
+  }
+
+  @Override
+  public ChannelFactory<? extends ServerChannel> serverChannelFactory(boolean domainSocket) {
+    if (domainSocket) {
+      throw new IllegalArgumentException();
+    }
+    return IOUringServerSocketChannel::new;
+  }
+
+  @Override
+  public void configure(DatagramChannel channel, DatagramSocketOptions options) {
+    channel.config().setOption(IOUringChannelOption.SO_REUSEPORT, options.isReusePort());
+    super.configure(channel, options);
+  }
+
+  @Override
+  public void configure(NetServerOptions options, boolean domainSocket, ServerBootstrap bootstrap) {
+    if (domainSocket) {
+      throw new IllegalArgumentException();
+    }
+    bootstrap.option(IOUringChannelOption.SO_REUSEPORT, options.isReusePort());
+    if (options.isTcpFastOpen()) {
+      bootstrap.option(IOUringChannelOption.TCP_FASTOPEN, options.isTcpFastOpen() ? pendingFastOpenRequestsThreshold : 0);
+    }
+    bootstrap.childOption(IOUringChannelOption.TCP_QUICKACK, options.isTcpQuickAck());
+    bootstrap.childOption(IOUringChannelOption.TCP_CORK, options.isTcpCork());
+    super.configure(options, false, bootstrap);
+  }
+
+  @Override
+  public void configure(ClientOptionsBase options, boolean domainSocket, Bootstrap bootstrap) {
+    if (domainSocket) {
+      throw new IllegalArgumentException();
+    }
+    if (options.isTcpFastOpen()) {
+      bootstrap.option(IOUringChannelOption.TCP_FASTOPEN_CONNECT, options.isTcpFastOpen());
+    }
+    bootstrap.option(IOUringChannelOption.TCP_QUICKACK, options.isTcpQuickAck());
+    bootstrap.option(IOUringChannelOption.TCP_CORK, options.isTcpCork());
+    super.configure(options, false, bootstrap);
+  }
+}
diff --git a/src/main/java/io/vertx/core/net/impl/transport/KQueueTransport.java b/src/main/java/io/vertx/core/net/impl/transport/KQueueTransport.java
--- a/src/main/java/io/vertx/core/net/impl/transport/KQueueTransport.java
+++ b/src/main/java/io/vertx/core/net/impl/transport/KQueueTransport.java
@@ -34,6 +34,7 @@
 class KQueueTransport extends Transport {
 
   KQueueTransport() {
+    super(true);
   }
 
   @Override
diff --git a/src/main/java/io/vertx/core/net/impl/transport/Transport.java b/src/main/java/io/vertx/core/net/impl/transport/Transport.java
--- a/src/main/java/io/vertx/core/net/impl/transport/Transport.java
+++ b/src/main/java/io/vertx/core/net/impl/transport/Transport.java
@@ -47,7 +47,11 @@ public class Transport {
   /**
    * The JDK transport, always there.
    */
-  public static Transport JDK = new Transport();
+  public static Transport JDK = new Transport(false);
+
+  public boolean supportFileRegion() {
+    return true;
+  }
 
   /**
    * The native transport, it may be {@code null} or failed.
@@ -64,6 +68,16 @@ public static Transport nativeTransport() {
     } catch (Throwable ignore) {
       // Jar not here
     }
+    try {
+      Transport ioUring = new IOUringTransport();
+      if (ioUring.isAvailable()) {
+        return ioUring;
+      } else {
+        transport = ioUring;
+      }
+    } catch (Throwable ignore) {
+      // Jar not here
+    }
     try {
       Transport kqueue = new KQueueTransport();
       if (kqueue.isAvailable()) {
@@ -90,7 +104,18 @@ public static Transport transport(boolean preferNative) {
     }
   }
 
+  private final boolean supportsDomainSockets;
+
   protected Transport() {
+    this(false);
+  }
+
+  protected Transport(boolean supportsDomainSockets) {
+    this.supportsDomainSockets = supportsDomainSockets;
+  }
+
+  public boolean supportsDomainSockets() {
+    return supportsDomainSockets;
   }
 
   /**


Test patch:
diff --git a/src/test/java/io/vertx/core/datagram/DatagramTest.java b/src/test/java/io/vertx/core/datagram/DatagramTest.java
--- a/src/test/java/io/vertx/core/datagram/DatagramTest.java
+++ b/src/test/java/io/vertx/core/datagram/DatagramTest.java
@@ -27,6 +27,7 @@
 import io.vertx.test.core.TestUtils;
 import io.vertx.test.core.VertxTestBase;
 import io.vertx.test.netty.TestLoggerFactory;
+import org.junit.Assert;
 import org.junit.Assume;
 import org.junit.Test;
 
@@ -177,33 +178,42 @@ public void testPauseResume() {
     peer2 = vertx.createDatagramSocket(new DatagramSocketOptions());
     peer2.exceptionHandler(t -> fail(t.getMessage()));
     peer2.listen(1234, "127.0.0.1", ar -> {
-      Buffer buffer = TestUtils.randomBuffer(128);
-      AtomicBoolean received = new AtomicBoolean();
-      peer2.handler(packet -> received.set(true));
+      final AtomicBoolean suspendedReceive = new AtomicBoolean();
+      peer2.handler(packet -> suspendedReceive.set(true));
       peer2.pause();
+      Buffer buffer = TestUtils.randomBuffer(128);
       peer1.send(buffer, 1234, "127.0.0.1", ar2 -> {
         assertTrue(ar2.succeeded());
       });
-      vertx.setTimer(1000, l -> {
-        AtomicInteger count = new AtomicInteger();
+      final int MAX_FAILED_ATTEMPTS = 10;
+      vertx.setTimer(1000, ignore -> {
+        Assert.assertFalse(suspendedReceive.get());
+        AtomicBoolean resumedReceive = new AtomicBoolean();
         peer2.handler(packet -> {
-          switch (count.getAndIncrement()) {
-            case 0:
-              assertEquals(buffer, packet.data());
-              peer1.send(buffer, 1234, "127.0.0.1", ar2 -> {
-                assertTrue(ar2.succeeded());
-              });
-              break;
-            case 1:
-              assertFalse(received.get());
-              assertEquals(buffer, packet.data());
-              testComplete();
-              break;
-            default:
-              fail();
+          Assert.assertEquals(buffer, packet.data());
+          if (resumedReceive.compareAndSet(false, true)) {
+            testComplete();
           }
         });
         peer2.resume();
+        peer1.send(buffer, 1234, "127.0.0.1", ar2 -> {
+          assertTrue(ar2.succeeded());
+        });
+        AtomicInteger failedAttempts = new AtomicInteger();
+        vertx.setPeriodic(1000, l -> {
+          if (resumedReceive.get()) {
+            vertx.cancelTimer(l.longValue());
+            return;
+          }
+          if (failedAttempts.incrementAndGet() == MAX_FAILED_ATTEMPTS) {
+            vertx.cancelTimer(l.longValue());
+            fail("failed to receive any packet while resumed: retried " + MAX_FAILED_ATTEMPTS + " times");
+            return;
+          }
+          peer1.send(buffer, 1234, "127.0.0.1", ar2 -> {
+            assertTrue(ar2.succeeded());
+          });
+        });
       });
     });
     await();
diff --git a/src/test/java/io/vertx/core/http/HttpTest.java b/src/test/java/io/vertx/core/http/HttpTest.java
--- a/src/test/java/io/vertx/core/http/HttpTest.java
+++ b/src/test/java/io/vertx/core/http/HttpTest.java
@@ -39,6 +39,7 @@
 import io.vertx.core.http.impl.ServerCookie;
 import io.vertx.core.http.impl.headers.HeadersMultiMap;
 import io.vertx.core.impl.Utils;
+import io.vertx.core.impl.VertxInternal;
 import io.vertx.core.net.NetClient;
 import io.vertx.core.net.NetClientOptions;
 import io.vertx.core.net.NetServerOptions;
@@ -176,6 +177,7 @@ public void testListenSocketAddress() {
   public void testListenDomainSocketAddress() throws Exception {
     Vertx vx = Vertx.vertx(new VertxOptions().setPreferNativeTransport(true));
     Assume.assumeTrue("Native transport must be enabled", vx.isNativeTransportEnabled());
+    Assume.assumeTrue("Transport must support domain sockets", ((VertxInternal) vx).transport().supportsDomainSockets());
     int len = 3;
     waitFor(len * len);
     List<SocketAddress> addresses = new ArrayList<>();
diff --git a/src/test/java/io/vertx/core/net/NetTest.java b/src/test/java/io/vertx/core/net/NetTest.java
--- a/src/test/java/io/vertx/core/net/NetTest.java
+++ b/src/test/java/io/vertx/core/net/NetTest.java
@@ -1945,8 +1945,8 @@ void run(boolean shouldPass) {
 
   @Test
   public void testListenDomainSocketAddress() throws Exception {
-    Vertx vx = Vertx.vertx(new VertxOptions().setPreferNativeTransport(true));
-    Assume.assumeTrue("Native transport must be enabled", vx.isNativeTransportEnabled());
+    VertxInternal vx = (VertxInternal) Vertx.vertx(new VertxOptions().setPreferNativeTransport(true));
+    Assume.assumeTrue("Transport must support domain sockets", vx.transport().supportsDomainSockets());
     int len = 3;
     waitFor(len * len);
     List<SocketAddress> addresses = new ArrayList<>();
diff --git a/src/test/java/io/vertx/it/TransportTest.java b/src/test/java/io/vertx/it/TransportTest.java
--- a/src/test/java/io/vertx/it/TransportTest.java
+++ b/src/test/java/io/vertx/it/TransportTest.java
@@ -50,6 +50,12 @@ public void testNoNative() {
     } catch (ClassNotFoundException ignore) {
       // Expected
     }
+    try {
+      classLoader.loadClass("io.netty.incubator.channel.uring.IOUring");
+      fail("Was not expected to load IOUring class");
+    } catch (ClassNotFoundException ignore) {
+      // Expected
+    }
     testNetServer(new VertxOptions());
     assertFalse(vertx.isNativeTransportEnabled());
   }