diff --git a/distribution/server/src/assemble/LICENSE.bin.txt b/distribution/server/src/assemble/LICENSE.bin.txt index 786c19171928d..7c42dc260ea38 100644 --- a/distribution/server/src/assemble/LICENSE.bin.txt +++ b/distribution/server/src/assemble/LICENSE.bin.txt @@ -311,6 +311,9 @@ The Apache Software License, Version 2.0 - io.netty-netty-tcnative-boringssl-static-2.0.52.Final-osx-x86_64.jar - io.netty-netty-tcnative-boringssl-static-2.0.52.Final-windows-x86_64.jar - io.netty-netty-tcnative-classes-2.0.52.Final.jar + - io.netty.incubator-netty-incubator-transport-classes-io_uring-0.0.15.Final.jar + - io.netty.incubator-netty-incubator-transport-native-io_uring-0.0.15.Final-linux-x86_64.jar + - io.netty.incubator-netty-incubator-transport-native-io_uring-0.0.15.Final-linux-aarch_64.jar * Prometheus client - io.prometheus.jmx-collector-0.16.1.jar - io.prometheus-simpleclient-0.16.0.jar diff --git a/pom.xml b/pom.xml index 53b7b2b2ee72b..88b6933cb75d2 100644 --- a/pom.xml +++ b/pom.xml @@ -127,6 +127,7 @@ flexible messaging model and an intuitive client API. 5.1.0 4.1.77.Final 2.0.52.Final + 0.0.15.Final 9.4.48.v20220622 2.5.2 2.34 @@ -642,6 +643,24 @@ flexible messaging model and an intuitive client API. ${netty.version} + + io.netty.incubator + netty-incubator-transport-native-io_uring + ${netty-iouring.version} + + + io.netty.incubator + netty-incubator-transport-native-io_uring + ${netty-iouring.version} + linux-x86_64 + + + io.netty.incubator + netty-incubator-transport-native-io_uring + ${netty-iouring.version} + linux-aarch_64 + + com.beust jcommander diff --git a/pulsar-broker-shaded/pom.xml b/pulsar-broker-shaded/pom.xml index c3dfc658aad88..274afda2f0380 100644 --- a/pulsar-broker-shaded/pom.xml +++ b/pulsar-broker-shaded/pom.xml @@ -119,6 +119,7 @@ com.google.*:* com.fasterxml.jackson.*:* io.netty:* + io.netty.incubator:* org.apache.pulsar:pulsar-common org.apache.bookkeeper:circe-checksum com.yahoo.datasketches:sketches-core diff --git a/pulsar-client-admin-shaded/pom.xml b/pulsar-client-admin-shaded/pom.xml index 2cb7b85bef0c2..b54082c9cf135 100644 --- a/pulsar-client-admin-shaded/pom.xml +++ b/pulsar-client-admin-shaded/pom.xml @@ -125,6 +125,7 @@ com.google.code.gson:gson com.fasterxml.jackson.core io.netty:* + io.netty.incubator:* org.apache.pulsar:pulsar-common org.apache.bookkeeper:* com.yahoo.datasketches:sketches-core diff --git a/pulsar-client-all/pom.xml b/pulsar-client-all/pom.xml index 1adaafbb90dff..11f46b9270431 100644 --- a/pulsar-client-all/pom.xml +++ b/pulsar-client-all/pom.xml @@ -176,6 +176,7 @@ io.perfmark:* com.yahoo.datasketches:* io.netty:* + io.netty.incubator:* com.squareup.*:* com.google.*:* commons-*:* diff --git a/pulsar-client-shaded/pom.xml b/pulsar-client-shaded/pom.xml index 87957c719e4d0..dd1f2e194850b 100644 --- a/pulsar-client-shaded/pom.xml +++ b/pulsar-client-shaded/pom.xml @@ -150,6 +150,7 @@ com.fasterxml.jackson.core:jackson-core com.fasterxml.jackson.dataformat io.netty:* + io.netty.incubator:* io.perfmark:* org.eclipse.jetty:* com.yahoo.datasketches:* diff --git a/pulsar-common/pom.xml b/pulsar-common/pom.xml index 30a250423605d..7e26eebf3fe9a 100644 --- a/pulsar-common/pom.xml +++ b/pulsar-common/pom.xml @@ -131,6 +131,18 @@ netty-tcnative-boringssl-static + + io.netty.incubator + netty-incubator-transport-native-io_uring + linux-x86_64 + + + + io.netty.incubator + netty-incubator-transport-native-io_uring + linux-aarch_64 + + io.netty netty-codec-haproxy diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/netty/EventLoopUtil.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/netty/EventLoopUtil.java index a2a9269092d43..3ed88947ccc26 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/util/netty/EventLoopUtil.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/netty/EventLoopUtil.java @@ -35,43 +35,61 @@ import io.netty.channel.socket.nio.NioDatagramChannel; import io.netty.channel.socket.nio.NioServerSocketChannel; import io.netty.channel.socket.nio.NioSocketChannel; +import io.netty.incubator.channel.uring.IOUring; +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 java.util.concurrent.CompletableFuture; import java.util.concurrent.ThreadFactory; import lombok.extern.slf4j.Slf4j; import org.apache.bookkeeper.common.util.affinity.CpuAffinity; +import org.apache.commons.lang3.StringUtils; @SuppressWarnings("checkstyle:JavadocType") @Slf4j public class EventLoopUtil { + private static final String ENABLE_IO_URING = "enable.io_uring"; + /** * @return an EventLoopGroup suitable for the current platform */ public static EventLoopGroup newEventLoopGroup(int nThreads, boolean enableBusyWait, ThreadFactory threadFactory) { if (Epoll.isAvailable()) { - if (!enableBusyWait) { - // Regular Epoll based event loop - return new EpollEventLoopGroup(nThreads, threadFactory); - } + String enableIoUring = System.getProperty(ENABLE_IO_URING); - // With low latency setting, put the Netty event loop on busy-wait loop to reduce cost of - // context switches - EpollEventLoopGroup eventLoopGroup = new EpollEventLoopGroup(nThreads, threadFactory, - () -> (selectSupplier, hasTasks) -> SelectStrategy.BUSY_WAIT); + // By default, io_uring will not be enabled, even if available. The environment variable will be used: + // enable.io_uring=1 + if (StringUtils.equalsAnyIgnoreCase(enableIoUring, "1", "true")) { + // Throw exception if IOUring cannot be used + IOUring.ensureAvailability(); + return new IOUringEventLoopGroup(nThreads, threadFactory); + } else { + if (!enableBusyWait) { + // Regular Epoll based event loop + return new EpollEventLoopGroup(nThreads, threadFactory); + } - // Enable CPU affinity on IO threads - for (int i = 0; i < nThreads; i++) { - eventLoopGroup.next().submit(() -> { - try { - CpuAffinity.acquireCore(); - } catch (Throwable t) { - log.warn("Failed to acquire CPU core for thread {} {}", Thread.currentThread().getName(), - t.getMessage(), t); - } - }); - } + // With low latency setting, put the Netty event loop on busy-wait loop to reduce cost of + // context switches + EpollEventLoopGroup eventLoopGroup = new EpollEventLoopGroup(nThreads, threadFactory, + () -> (selectSupplier, hasTasks) -> SelectStrategy.BUSY_WAIT); - return eventLoopGroup; + // Enable CPU affinity on IO threads + for (int i = 0; i < nThreads; i++) { + eventLoopGroup.next().submit(() -> { + try { + CpuAffinity.acquireCore(); + } catch (Throwable t) { + log.warn("Failed to acquire CPU core for thread {} {}", Thread.currentThread().getName(), + t.getMessage(), t); + } + }); + } + + return eventLoopGroup; + } } else { // Fallback to NIO return new NioEventLoopGroup(nThreads, threadFactory); @@ -85,7 +103,9 @@ public static EventLoopGroup newEventLoopGroup(int nThreads, boolean enableBusyW * @return */ public static Class getClientSocketChannelClass(EventLoopGroup eventLoopGroup) { - if (eventLoopGroup instanceof EpollEventLoopGroup) { + if (eventLoopGroup instanceof IOUringEventLoopGroup) { + return IOUringSocketChannel.class; + } else if (eventLoopGroup instanceof EpollEventLoopGroup) { return EpollSocketChannel.class; } else { return NioSocketChannel.class; @@ -93,7 +113,9 @@ public static Class getClientSocketChannelClass(EventLo } public static Class getServerSocketChannelClass(EventLoopGroup eventLoopGroup) { - if (eventLoopGroup instanceof EpollEventLoopGroup) { + if (eventLoopGroup instanceof IOUringEventLoopGroup) { + return IOUringServerSocketChannel.class; + } else if (eventLoopGroup instanceof EpollEventLoopGroup) { return EpollServerSocketChannel.class; } else { return NioServerSocketChannel.class; @@ -101,7 +123,9 @@ public static Class getServerSocketChannelClass(E } public static Class getDatagramChannelClass(EventLoopGroup eventLoopGroup) { - if (eventLoopGroup instanceof EpollEventLoopGroup) { + if (eventLoopGroup instanceof IOUringEventLoopGroup) { + return IOUringDatagramChannel.class; + } else if (eventLoopGroup instanceof EpollEventLoopGroup) { return EpollDatagramChannel.class; } else { return NioDatagramChannel.class; diff --git a/pulsar-sql/presto-distribution/LICENSE b/pulsar-sql/presto-distribution/LICENSE index ea7a1c02deede..fa3b1ba7de6bb 100644 --- a/pulsar-sql/presto-distribution/LICENSE +++ b/pulsar-sql/presto-distribution/LICENSE @@ -256,6 +256,9 @@ The Apache Software License, Version 2.0 - netty-transport-native-unix-common-4.1.77.Final.jar - netty-transport-native-unix-common-4.1.77.Final-linux-x86_64.jar - netty-codec-http2-4.1.77.Final.jar + - netty-incubator-transport-classes-io_uring-0.0.15.Final.jar + - netty-incubator-transport-native-io_uring-0.0.15.Final-linux-x86_64.jar + - netty-incubator-transport-native-io_uring-0.0.15.Final-linux-aarch_64.jar * GRPC - grpc-api-1.45.1.jar - grpc-context-1.45.1.jar