diff --git a/artemis-cli/pom.xml b/artemis-cli/pom.xml index be5d702d3d4..997830c49a7 100644 --- a/artemis-cli/pom.xml +++ b/artemis-cli/pom.xml @@ -239,6 +239,13 @@ + + org.apache.maven.plugins + maven-surefire-plugin + + ${activemq-surefire-argline} + + diff --git a/artemis-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/bin/artemis b/artemis-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/bin/artemis index d50d56e4836..90bc8634c43 100755 --- a/artemis-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/bin/artemis +++ b/artemis-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/bin/artemis @@ -119,6 +119,9 @@ if [ -f "$ARTEMIS_OOME_DUMP" ] ; then mv $ARTEMIS_OOME_DUMP $ARTEMIS_OOME_DUMP.bkp fi +# Netty needs access to unsafe, but this is turned off in Java 25 by default +$JAVACMD --sun-misc-unsafe-memory-access=allow --version > /dev/null 2>&1 && ALLOW_UNSAFE="--sun-misc-unsafe-memory-access=allow" + exec "$JAVACMD" \ $LOGGING_ARGS \ $JAVA_ARGS \ @@ -132,6 +135,7 @@ exec "$JAVACMD" \ -Djava.io.tmpdir="$ARTEMIS_INSTANCE/tmp" \ -Ddata.dir="$ARTEMIS_DATA_DIR" \ -Dartemis.instance.etc="$ARTEMIS_INSTANCE_ETC" \ + $ALLOW_UNSAFE \ $DEBUG_ARGS \ $JAVA_ARGS_APPEND \ org.apache.activemq.artemis.boot.Artemis "$@" diff --git a/artemis-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/etc/artemis-utility.profile b/artemis-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/etc/artemis-utility.profile index a724cd6c496..15d2043b588 100644 --- a/artemis-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/etc/artemis-utility.profile +++ b/artemis-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/etc/artemis-utility.profile @@ -24,7 +24,7 @@ if [ -z "$LOGGING_ARGS" ]; then fi if [ -z "$JAVA_ARGS" ]; then - JAVA_ARGS="-Dlog4j2.disableJmx=true --add-opens java.base/jdk.internal.misc=ALL-UNNAMED ${java-utility-opts}" + JAVA_ARGS="-Dlog4j2.disableJmx=true --add-opens java.base/jdk.internal.misc=ALL-UNNAMED --enable-native-access=ALL-UNNAMED ${java-utility-opts}" fi # Uncomment to enable remote debugging diff --git a/artemis-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/etc/artemis.profile b/artemis-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/etc/artemis.profile index 6da714fe09a..59ba1e3337f 100644 --- a/artemis-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/etc/artemis.profile +++ b/artemis-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/etc/artemis.profile @@ -30,7 +30,7 @@ HAWTIO_ROLES='${role}' # Java Opts if [ -z "$JAVA_ARGS" ]; then - JAVA_ARGS="-XX:AutoBoxCacheMax=20000 -XX:+PrintClassHistogram -XX:+UseG1GC -XX:+UseStringDeduplication -Xms512M -Xmx${java-memory} -Dhawtio.disableProxy=true -Dhawtio.realm=activemq -Dhawtio.offline=true -Dhawtio.rolePrincipalClasses=org.apache.activemq.artemis.spi.core.security.jaas.RolePrincipal -Dhawtio.http.strictTransportSecurity=max-age=31536000;includeSubDomains;preload -Djolokia.policyLocation=classpath:jolokia-access.xml -Dlog4j2.disableJmx=true --add-opens java.base/jdk.internal.misc=ALL-UNNAMED ${java-opts}" + JAVA_ARGS="-XX:AutoBoxCacheMax=20000 -XX:+PrintClassHistogram -XX:+UseG1GC -XX:+UseStringDeduplication -Xms512M -Xmx${java-memory} -Dhawtio.disableProxy=true -Dhawtio.realm=activemq -Dhawtio.offline=true -Dhawtio.rolePrincipalClasses=org.apache.activemq.artemis.spi.core.security.jaas.RolePrincipal -Dhawtio.http.strictTransportSecurity=max-age=31536000;includeSubDomains;preload -Djolokia.policyLocation=classpath:jolokia-access.xml -Dlog4j2.disableJmx=true --add-opens java.base/jdk.internal.misc=ALL-UNNAMED --enable-native-access=ALL-UNNAMED ${java-opts}" fi # Uncomment to enable logging for Safepoint JVM pauses diff --git a/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/SpawnedVMSupport.java b/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/SpawnedVMSupport.java index c0781c24c55..76a8188fff6 100644 --- a/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/SpawnedVMSupport.java +++ b/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/SpawnedVMSupport.java @@ -204,6 +204,12 @@ public static Process spawnVM(String classPath, commandList.add(jacocoAgent); } + String javaVersion = System.getProperty("java.version"); + if (javaVersion.startsWith("24") || javaVersion.startsWith("25")) { + commandList.add("--enable-native-access=ALL-UNNAMED"); + commandList.add("--sun-misc-unsafe-memory-access=allow"); + } + commandList.add(className); for (String arg : args) { commandList.add(arg); diff --git a/artemis-commons/src/test/java/org/apache/activemq/artemis/utils/ByteUtilTest.java b/artemis-commons/src/test/java/org/apache/activemq/artemis/utils/ByteUtilTest.java index c9cf99d3940..7ea90a58c86 100644 --- a/artemis-commons/src/test/java/org/apache/activemq/artemis/utils/ByteUtilTest.java +++ b/artemis-commons/src/test/java/org/apache/activemq/artemis/utils/ByteUtilTest.java @@ -223,6 +223,7 @@ public void shouldZeroesDirectByteBuffer() { @Test public void shouldZeroesLimitedDirectByteBuffer() { + assumeTrue(PlatformDependent.hasUnsafe()); final byte one = (byte) 1; final int capacity = 64; final int bytes = 32; diff --git a/artemis-core-client-osgi/pom.xml b/artemis-core-client-osgi/pom.xml index 5f629447965..cd74d3020ff 100644 --- a/artemis-core-client-osgi/pom.xml +++ b/artemis-core-client-osgi/pom.xml @@ -70,7 +70,7 @@ org.glassfish.json*;resolution:=optional, de.dentrassi.crypto.pem;resolution:=optional, - io.netty.buffer;io.netty.*;version="[4.1,5)", + io.netty.*;version="[4.2,5)", * <_exportcontents>org.apache.activemq.artemis.*;-noimport:=true diff --git a/artemis-core-client/pom.xml b/artemis-core-client/pom.xml index 4bd18e7f6b2..49bd10a54a0 100644 --- a/artemis-core-client/pom.xml +++ b/artemis-core-client/pom.xml @@ -89,6 +89,15 @@ io.netty netty-transport-classes-kqueue + + io.netty + netty-transport-native-io_uring + ${netty-transport-native-io_uring-classifier} + + + io.netty + netty-transport-classes-io_uring + io.netty netty-codec-http @@ -109,10 +118,6 @@ io.netty netty-handler-proxy - - io.netty - netty-codec - io.netty netty-codec-socks diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/ActiveMQClientLogger.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/ActiveMQClientLogger.java index e7c1c7108df..a47f408ec3d 100644 --- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/ActiveMQClientLogger.java +++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/ActiveMQClientLogger.java @@ -349,4 +349,10 @@ public interface ActiveMQClientLogger { @LogMessage(id = 214036, value = "Connection closure to {} has been detected: {} [code={}]", level = LogMessage.Level.INFO) void connectionClosureDetected(String remoteAddress, String message, ActiveMQExceptionType type); + + @LogMessage(id = 214037, value = "Unable to check IoUring availability ", level = LogMessage.Level.WARN) + void unableToCheckIoUringAvailability(Throwable e); + + @LogMessage(id = 214038, value = "IoUring is not available, please add to the classpath or configure useIoUring=false to remove this warning", level = LogMessage.Level.WARN) + void unableToCheckIoUringAvailabilitynoClass(); } diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/CheckDependencies.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/CheckDependencies.java index 4a90401dcca..e56c2ab764d 100644 --- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/CheckDependencies.java +++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/CheckDependencies.java @@ -19,6 +19,7 @@ import io.netty.channel.epoll.Epoll; import io.netty.channel.kqueue.KQueue; +import io.netty.channel.uring.IoUring; import org.apache.activemq.artemis.core.client.ActiveMQClientLogger; import org.apache.activemq.artemis.utils.Env; @@ -51,4 +52,17 @@ public static final boolean isKQueueAvailable() { return false; } } + + public static final boolean isIoUringAvailable() { + try { + return Env.isLinuxOs() && IoUring.isAvailable(); + } catch (NoClassDefFoundError noClassDefFoundError) { + ActiveMQClientLogger.LOGGER.unableToCheckIoUringAvailabilitynoClass(); + return false; + } catch (Throwable e) { + ActiveMQClientLogger.LOGGER.unableToCheckIoUringAvailability(e); + return false; + } + } + } diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/NettyConnector.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/NettyConnector.java index 898dbb9ae71..3bafd8b1e4b 100644 --- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/NettyConnector.java +++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/NettyConnector.java @@ -16,13 +16,12 @@ */ package org.apache.activemq.artemis.core.remoting.impl.netty; -import static org.apache.activemq.artemis.core.remoting.impl.netty.TransportConstants.NETTY_HTTP_HEADER_PREFIX; - import javax.net.ssl.SNIHostName; import javax.net.ssl.SSLContext; import javax.net.ssl.SSLEngine; import javax.net.ssl.SSLParameters; import java.io.IOException; +import java.lang.invoke.MethodHandles; import java.net.ConnectException; import java.net.InetAddress; import java.net.InetSocketAddress; @@ -63,16 +62,19 @@ import io.netty.channel.ChannelPipeline; import io.netty.channel.ChannelPromise; import io.netty.channel.EventLoopGroup; +import io.netty.channel.MultiThreadIoEventLoopGroup; import io.netty.channel.SimpleChannelInboundHandler; import io.netty.channel.WriteBufferWaterMark; -import io.netty.channel.epoll.EpollEventLoopGroup; +import io.netty.channel.epoll.EpollIoHandler; import io.netty.channel.epoll.EpollSocketChannel; import io.netty.channel.group.ChannelGroup; import io.netty.channel.group.DefaultChannelGroup; -import io.netty.channel.kqueue.KQueueEventLoopGroup; +import io.netty.channel.kqueue.KQueueIoHandler; import io.netty.channel.kqueue.KQueueSocketChannel; -import io.netty.channel.nio.NioEventLoopGroup; +import io.netty.channel.nio.NioIoHandler; import io.netty.channel.socket.nio.NioSocketChannel; +import io.netty.channel.uring.IoUringIoHandler; +import io.netty.channel.uring.IoUringSocketChannel; import io.netty.handler.codec.base64.Base64; import io.netty.handler.codec.http.DefaultFullHttpRequest; import io.netty.handler.codec.http.DefaultHttpRequest; @@ -92,11 +94,11 @@ import io.netty.handler.codec.http.LastHttpContent; import io.netty.handler.codec.http.cookie.ClientCookieDecoder; import io.netty.handler.codec.http.cookie.Cookie; -import io.netty.handler.ssl.SslContext; import io.netty.handler.codec.socksx.SocksVersion; import io.netty.handler.proxy.ProxyHandler; import io.netty.handler.proxy.Socks4ProxyHandler; import io.netty.handler.proxy.Socks5ProxyHandler; +import io.netty.handler.ssl.SslContext; import io.netty.handler.ssl.SslHandler; import io.netty.resolver.NoopAddressResolverGroup; import io.netty.util.AttributeKey; @@ -128,8 +130,8 @@ import org.apache.activemq.artemis.utils.PasswordMaskingUtil; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.lang.invoke.MethodHandles; +import static org.apache.activemq.artemis.core.remoting.impl.netty.TransportConstants.NETTY_HTTP_HEADER_PREFIX; import static org.apache.activemq.artemis.utils.Base64.encodeBytes; public class NettyConnector extends AbstractConnector { @@ -137,6 +139,7 @@ public class NettyConnector extends AbstractConnector { public static String NIO_CONNECTOR_TYPE = "NIO"; public static String EPOLL_CONNECTOR_TYPE = "EPOLL"; public static String KQUEUE_CONNECTOR_TYPE = "KQUEUE"; + public static String IOURING_CONNECTOR_TYPE = "IO_URING"; private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass()); @@ -295,6 +298,8 @@ public class NettyConnector extends AbstractConnector { private boolean useKQueue; + private boolean useIoUring; + private int remotingThreads; private boolean useGlobalWorkerPool; @@ -404,6 +409,7 @@ public NettyConnector(final Map configuration, useEpoll = ConfigurationHelper.getBooleanProperty(TransportConstants.USE_EPOLL_PROP_NAME, TransportConstants.DEFAULT_USE_EPOLL, configuration); useKQueue = ConfigurationHelper.getBooleanProperty(TransportConstants.USE_KQUEUE_PROP_NAME, TransportConstants.DEFAULT_USE_KQUEUE, configuration); + useIoUring = ConfigurationHelper.getBooleanProperty(TransportConstants.USE_IOURING_PROP_NAME, TransportConstants.DEFAULT_USE_IOURING, configuration); useServlet = ConfigurationHelper.getBooleanProperty(TransportConstants.USE_SERVLET_PROP_NAME, TransportConstants.DEFAULT_USE_SERVLET, configuration); host = ConfigurationHelper.getStringProperty(TransportConstants.HOST_PROP_NAME, TransportConstants.DEFAULT_HOST, configuration); @@ -528,27 +534,43 @@ public synchronized void start() { return; } - if (remotingThreads == -1) { + boolean defaultRemotingThreads = remotingThreads == -1; + + if (defaultRemotingThreads) { // Default to number of cores * 3 remotingThreads = Runtime.getRuntime().availableProcessors() * 3; } String connectorType; - if (useEpoll && CheckDependencies.isEpollAvailable()) { + if (useIoUring && CheckDependencies.isIoUringAvailable()) { + //IO_URING should default to 1 remotingThread unless specified in config + remotingThreads = defaultRemotingThreads ? 1 : remotingThreads; + + if (useGlobalWorkerPool) { + group = SharedEventLoopGroup.getInstance((threadFactory -> new MultiThreadIoEventLoopGroup(remotingThreads, threadFactory, IoUringIoHandler.newFactory()))); + } else { + group = new MultiThreadIoEventLoopGroup(remotingThreads, IoUringIoHandler.newFactory()); + } + + connectorType = IOURING_CONNECTOR_TYPE; + channelClazz = IoUringSocketChannel.class; + + logger.debug("Connector {} using native io_uring", this); + } else if (useEpoll && CheckDependencies.isEpollAvailable()) { if (useGlobalWorkerPool) { - group = SharedEventLoopGroup.getInstance((threadFactory -> new EpollEventLoopGroup(remotingThreads, threadFactory))); + group = SharedEventLoopGroup.getInstance((threadFactory -> new MultiThreadIoEventLoopGroup(remotingThreads, threadFactory, EpollIoHandler.newFactory()))); } else { - group = new EpollEventLoopGroup(remotingThreads); + group = new MultiThreadIoEventLoopGroup(remotingThreads, EpollIoHandler.newFactory()); } connectorType = EPOLL_CONNECTOR_TYPE; channelClazz = EpollSocketChannel.class; logger.debug("Connector {} using native epoll", this); } else if (useKQueue && CheckDependencies.isKQueueAvailable()) { if (useGlobalWorkerPool) { - group = SharedEventLoopGroup.getInstance((threadFactory -> new KQueueEventLoopGroup(remotingThreads, threadFactory))); + group = SharedEventLoopGroup.getInstance((threadFactory -> new MultiThreadIoEventLoopGroup(remotingThreads, threadFactory, KQueueIoHandler.newFactory()))); } else { - group = new KQueueEventLoopGroup(remotingThreads); + group = new MultiThreadIoEventLoopGroup(remotingThreads, KQueueIoHandler.newFactory()); } connectorType = KQUEUE_CONNECTOR_TYPE; channelClazz = KQueueSocketChannel.class; @@ -556,10 +578,10 @@ public synchronized void start() { } else { if (useGlobalWorkerPool) { channelClazz = NioSocketChannel.class; - group = SharedEventLoopGroup.getInstance((threadFactory -> new NioEventLoopGroup(remotingThreads, threadFactory))); + group = SharedEventLoopGroup.getInstance((threadFactory -> new MultiThreadIoEventLoopGroup(remotingThreads, threadFactory, NioIoHandler.newFactory()))); } else { channelClazz = NioSocketChannel.class; - group = new NioEventLoopGroup(remotingThreads); + group = new MultiThreadIoEventLoopGroup(remotingThreads, NioIoHandler.newFactory()); } connectorType = NIO_CONNECTOR_TYPE; channelClazz = NioSocketChannel.class; diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/TransportConstants.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/TransportConstants.java index 641d1448e91..3dbc349b7ba 100644 --- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/TransportConstants.java +++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/TransportConstants.java @@ -70,6 +70,8 @@ public class TransportConstants { public static final String USE_KQUEUE_PROP_NAME = "useKQueue"; + public static final String USE_IOURING_PROP_NAME = "useIoUring"; + /** * @deprecated Use USE_GLOBAL_WORKER_POOL_PROP_NAME */ @@ -218,6 +220,8 @@ public class TransportConstants { public static final boolean DEFAULT_USE_KQUEUE = true; + public static final boolean DEFAULT_USE_IOURING = false; + public static final boolean DEFAULT_USE_INVM = false; public static final boolean DEFAULT_USE_SERVLET = false; @@ -426,6 +430,7 @@ private static int parseDefaultVariable(String variableName, int defaultValue) { allowableAcceptorKeys.add(TransportConstants.USE_NIO_PROP_NAME); allowableAcceptorKeys.add(TransportConstants.USE_EPOLL_PROP_NAME); allowableAcceptorKeys.add(TransportConstants.USE_KQUEUE_PROP_NAME); + allowableAcceptorKeys.add(TransportConstants.USE_IOURING_PROP_NAME); allowableAcceptorKeys.add(TransportConstants.USE_INVM_PROP_NAME); //noinspection deprecation allowableAcceptorKeys.add(TransportConstants.PROTOCOL_PROP_NAME); @@ -502,6 +507,7 @@ private static int parseDefaultVariable(String variableName, int defaultValue) { allowableConnectorKeys.add(TransportConstants.USE_NIO_GLOBAL_WORKER_POOL_PROP_NAME); allowableConnectorKeys.add(TransportConstants.USE_EPOLL_PROP_NAME); allowableConnectorKeys.add(TransportConstants.USE_KQUEUE_PROP_NAME); + allowableConnectorKeys.add(TransportConstants.USE_IOURING_PROP_NAME); allowableConnectorKeys.add(TransportConstants.USE_GLOBAL_WORKER_POOL_PROP_NAME); allowableConnectorKeys.add(TransportConstants.HOST_PROP_NAME); allowableConnectorKeys.add(TransportConstants.PORT_PROP_NAME); diff --git a/artemis-distribution/src/main/resources/bin/artemis b/artemis-distribution/src/main/resources/bin/artemis index 29dce4abc7b..f0685f48df6 100755 --- a/artemis-distribution/src/main/resources/bin/artemis +++ b/artemis-distribution/src/main/resources/bin/artemis @@ -90,10 +90,15 @@ if $cygwin ; then CLASSPATH=`cygpath --windows "$CLASSPATH"` fi +# Netty needs access to unsafe, but this is turned off in Java 25 by default +$JAVACMD --sun-misc-unsafe-memory-access=allow --version > /dev/null 2>&1 && ALLOW_UNSAFE="--sun-misc-unsafe-memory-access=allow" + exec "$JAVACMD" $JAVA_ARGS $ARTEMIS_CLUSTER_PROPS \ -classpath "$CLASSPATH" \ -Dartemis.home="$ARTEMIS_HOME" \ -Djava.library.path="$ARTEMIS_HOME/bin/lib/linux-$(uname -m)" \ + --enable-native-access=ALL-UNNAMED \ + $ALLOW_UNSAFE \ $DEBUG_ARGS \ $JAVA_ARGS_APPEND \ org.apache.activemq.artemis.boot.Artemis "$@" diff --git a/artemis-features/src/main/resources/features.xml b/artemis-features/src/main/resources/features.xml index d4756422d90..534c10f54b4 100644 --- a/artemis-features/src/main/resources/features.xml +++ b/artemis-features/src/main/resources/features.xml @@ -33,7 +33,8 @@ mvn:io.netty/netty-resolver/${netty.version} mvn:io.netty/netty-transport/${netty.version} mvn:io.netty/netty-buffer/${netty.version} - mvn:io.netty/netty-codec/${netty.version} + mvn:io.netty/netty-codec-base/${netty.version} + mvn:io.netty/netty-codec-compression/${netty.version} mvn:io.netty/netty-codec-socks/${netty.version} mvn:io.netty/netty-codec-haproxy/${netty.version} mvn:io.netty/netty-codec-http/${netty.version} @@ -44,6 +45,8 @@ mvn:io.netty/netty-transport-native-epoll/${netty.version} mvn:io.netty/netty-transport-classes-kqueue/${netty.version} mvn:io.netty/netty-transport-native-kqueue/${netty.version} + mvn:io.netty/netty-transport-classes-io_uring/${netty.version} + mvn:io.netty/netty-transport-native-io_uring/${netty.version} mvn:io.netty/netty-transport-native-unix-common/${netty.version} diff --git a/artemis-jms-client-osgi/pom.xml b/artemis-jms-client-osgi/pom.xml index d1e37929984..92c22ccaba0 100644 --- a/artemis-jms-client-osgi/pom.xml +++ b/artemis-jms-client-osgi/pom.xml @@ -78,7 +78,7 @@ org.glassfish.json*;resolution:=optional, de.dentrassi.crypto.pem;resolution:=optional, - io.netty.buffer;io.netty.*;version="[4.1,5)", + io.netty.*;version="[4.2,5)", * <_exportcontents>org.apache.activemq.artemis.*;-noimport:=true diff --git a/artemis-pom/pom.xml b/artemis-pom/pom.xml index 94a373723a3..e098502c823 100644 --- a/artemis-pom/pom.xml +++ b/artemis-pom/pom.xml @@ -412,12 +412,6 @@ ${netty.version} - - io.netty - netty-codec - ${netty.version} - - io.netty netty-codec-http @@ -486,6 +480,19 @@ ${netty-transport-native-kqueue-classifier} + + io.netty + netty-transport-classes-io_uring + ${netty.version} + + + + io.netty + netty-transport-native-io_uring + ${netty.version} + ${netty-transport-native-io_uring-classifier} + + io.netty netty-tcnative-boringssl-static diff --git a/artemis-protocols/artemis-jakarta-openwire-protocol/pom.xml b/artemis-protocols/artemis-jakarta-openwire-protocol/pom.xml index 4d3c1229bd7..839b81746cb 100644 --- a/artemis-protocols/artemis-jakarta-openwire-protocol/pom.xml +++ b/artemis-protocols/artemis-jakarta-openwire-protocol/pom.xml @@ -90,10 +90,6 @@ io.netty netty-transport - - io.netty - netty-codec - org.osgi osgi.cmpn diff --git a/artemis-protocols/artemis-mqtt-protocol/pom.xml b/artemis-protocols/artemis-mqtt-protocol/pom.xml index 6d398b70f5d..f19bccf2558 100644 --- a/artemis-protocols/artemis-mqtt-protocol/pom.xml +++ b/artemis-protocols/artemis-mqtt-protocol/pom.xml @@ -69,10 +69,6 @@ io.netty netty-transport - - io.netty - netty-codec - io.netty netty-common diff --git a/artemis-protocols/artemis-openwire-protocol/pom.xml b/artemis-protocols/artemis-openwire-protocol/pom.xml index 9248f1d05e5..3a1cd42062c 100644 --- a/artemis-protocols/artemis-openwire-protocol/pom.xml +++ b/artemis-protocols/artemis-openwire-protocol/pom.xml @@ -96,10 +96,6 @@ io.netty netty-transport - - io.netty - netty-codec - org.osgi osgi.cmpn diff --git a/artemis-server-osgi/pom.xml b/artemis-server-osgi/pom.xml index 2470b3d3a9e..085ec59dc08 100644 --- a/artemis-server-osgi/pom.xml +++ b/artemis-server-osgi/pom.xml @@ -128,7 +128,7 @@ org.glassfish.json*;resolution:=optional, org.postgresql*;resolution:=optional, de.dentrassi.crypto.pem;resolution:=optional, - io.netty.buffer;io.netty.*;version="[4.1,5)", + io.netty.*;version="[4.2,5)", java.net.http*;resolution:=optional, com.sun.net.httpserver*;resolution:=optional, * diff --git a/artemis-server/pom.xml b/artemis-server/pom.xml index 7e3edd71389..ad508512f9c 100644 --- a/artemis-server/pom.xml +++ b/artemis-server/pom.xml @@ -136,10 +136,6 @@ io.netty netty-transport-classes-kqueue - - io.netty - netty-codec - commons-beanutils commons-beanutils diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/NettyAcceptor.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/NettyAcceptor.java index c1e209d1d7f..9639d420141 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/NettyAcceptor.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/NettyAcceptor.java @@ -49,19 +49,22 @@ import io.netty.channel.ChannelPipeline; import io.netty.channel.DefaultEventLoopGroup; import io.netty.channel.EventLoopGroup; +import io.netty.channel.MultiThreadIoEventLoopGroup; import io.netty.channel.ServerChannel; import io.netty.channel.WriteBufferWaterMark; -import io.netty.channel.epoll.EpollEventLoopGroup; +import io.netty.channel.epoll.EpollIoHandler; import io.netty.channel.epoll.EpollServerSocketChannel; import io.netty.channel.group.ChannelGroup; import io.netty.channel.group.ChannelGroupFuture; import io.netty.channel.group.DefaultChannelGroup; -import io.netty.channel.kqueue.KQueueEventLoopGroup; +import io.netty.channel.kqueue.KQueueIoHandler; import io.netty.channel.kqueue.KQueueServerSocketChannel; import io.netty.channel.local.LocalAddress; import io.netty.channel.local.LocalServerChannel; -import io.netty.channel.nio.NioEventLoopGroup; +import io.netty.channel.nio.NioIoHandler; import io.netty.channel.socket.nio.NioServerSocketChannel; +import io.netty.channel.uring.IoUringIoHandler; +import io.netty.channel.uring.IoUringServerSocketChannel; import io.netty.handler.codec.haproxy.HAProxyMessageDecoder; import io.netty.handler.ssl.SslContext; import io.netty.handler.ssl.SslHandler; @@ -117,6 +120,7 @@ public class NettyAcceptor extends AbstractAcceptor { public static final String NIO_ACCEPTOR_TYPE = "NIO"; public static final String EPOLL_ACCEPTOR_TYPE = "EPOLL"; public static final String KQUEUE_ACCEPTOR_TYPE = "KQUEUE"; + public static final String IOURING_ACCEPTOR_TYPE = "IO_URING"; static { // Disable default Netty leak detection if the Netty leak detection level system properties are not in use @@ -155,6 +159,8 @@ public class NettyAcceptor extends AbstractAcceptor { private final boolean useKQueue; + private final boolean useIoUring; + private final ProtocolHandler protocolHandler; private final String host; @@ -299,6 +305,7 @@ public NettyAcceptor(final String name, useEpoll = ConfigurationHelper.getBooleanProperty(TransportConstants.USE_EPOLL_PROP_NAME, TransportConstants.DEFAULT_USE_EPOLL, configuration); useKQueue = ConfigurationHelper.getBooleanProperty(TransportConstants.USE_KQUEUE_PROP_NAME, TransportConstants.DEFAULT_USE_KQUEUE, configuration); + useIoUring = ConfigurationHelper.getBooleanProperty(TransportConstants.USE_IOURING_PROP_NAME, TransportConstants.DEFAULT_USE_IOURING, configuration); backlog = ConfigurationHelper.getIntProperty(TransportConstants.BACKLOG_PROP_NAME, -1, configuration); useInvm = ConfigurationHelper.getBooleanProperty(TransportConstants.USE_INVM_PROP_NAME, TransportConstants.DEFAULT_USE_INVM, configuration); @@ -448,19 +455,35 @@ public synchronized void start() throws Exception { eventLoopGroup = new DefaultEventLoopGroup(); } else { ThreadFactory threadFactory = SecurityManagerShim.doPrivileged((PrivilegedAction) () -> new ActiveMQThreadFactory(threadFactoryGroupName, true, ClientSessionFactoryImpl.class.getClassLoader())); - if (useEpoll && CheckDependencies.isEpollAvailable()) { + + boolean defaultRemotingThreads = remotingThreads == -1; + + if (defaultRemotingThreads) { + // Default to number of cores * 3 + remotingThreads = Runtime.getRuntime().availableProcessors() * 3; + } + + if (useIoUring && CheckDependencies.isIoUringAvailable()) { + //IO_URING should default to 1 remotingThread unless specified in config + remotingThreads = defaultRemotingThreads ? 1 : remotingThreads; + + channelClazz = IoUringServerSocketChannel.class; + eventLoopGroup = new MultiThreadIoEventLoopGroup(remotingThreads, threadFactory, IoUringIoHandler.newFactory()); + acceptorType = IOURING_ACCEPTOR_TYPE; + logger.debug("Acceptor using native io_uring"); + } else if (useEpoll && CheckDependencies.isEpollAvailable()) { channelClazz = EpollServerSocketChannel.class; - eventLoopGroup = new EpollEventLoopGroup(remotingThreads, threadFactory); + eventLoopGroup = new MultiThreadIoEventLoopGroup(remotingThreads, threadFactory, EpollIoHandler.newFactory()); acceptorType = EPOLL_ACCEPTOR_TYPE; logger.debug("Acceptor {} using native epoll", name); } else if (useKQueue && CheckDependencies.isKQueueAvailable()) { channelClazz = KQueueServerSocketChannel.class; - eventLoopGroup = new KQueueEventLoopGroup(remotingThreads, threadFactory); + eventLoopGroup = new MultiThreadIoEventLoopGroup(remotingThreads, threadFactory, KQueueIoHandler.newFactory()); acceptorType = KQUEUE_ACCEPTOR_TYPE; logger.debug("Acceptor {} using native kqueue", name); } else { channelClazz = NioServerSocketChannel.class; - eventLoopGroup = new NioEventLoopGroup(remotingThreads, threadFactory); + eventLoopGroup = new MultiThreadIoEventLoopGroup(remotingThreads, threadFactory, NioIoHandler.newFactory()); acceptorType = NIO_ACCEPTOR_TYPE; logger.debug("Acceptor {} using nio", name); } diff --git a/artemis-server/src/test/java/org/apache/activemq/artemis/core/persistence/impl/journal/JournalStorageManagerTest.java b/artemis-server/src/test/java/org/apache/activemq/artemis/core/persistence/impl/journal/JournalStorageManagerTest.java index 8be81812e03..17a999b79d0 100644 --- a/artemis-server/src/test/java/org/apache/activemq/artemis/core/persistence/impl/journal/JournalStorageManagerTest.java +++ b/artemis-server/src/test/java/org/apache/activemq/artemis/core/persistence/impl/journal/JournalStorageManagerTest.java @@ -32,6 +32,7 @@ import java.util.concurrent.Executors; import java.util.stream.Stream; +import io.netty.util.internal.PlatformDependent; import org.apache.activemq.artemis.api.core.ActiveMQBuffer; import org.apache.activemq.artemis.api.core.ActiveMQBuffers; import org.apache.activemq.artemis.core.config.Configuration; @@ -102,6 +103,7 @@ public void testFixJournalFileSize() throws Exception { @TestTemplate public void testAddBytesToLargeMessageNotLeakingByteBuffer() throws Exception { + assumeTrue(PlatformDependent.hasUnsafe()); if (journalType == JournalType.ASYNCIO) { assumeTrue(AIOSequentialFileFactory.isSupported(), "AIO is not supported on this platform"); } diff --git a/artemis-web/src/test/java/org/apache/activemq/cli/test/WebServerComponentTest.java b/artemis-web/src/test/java/org/apache/activemq/cli/test/WebServerComponentTest.java index d2b512f6e45..c7afd266da2 100644 --- a/artemis-web/src/test/java/org/apache/activemq/cli/test/WebServerComponentTest.java +++ b/artemis-web/src/test/java/org/apache/activemq/cli/test/WebServerComponentTest.java @@ -58,8 +58,9 @@ import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelInitializer; import io.netty.channel.EventLoopGroup; +import io.netty.channel.MultiThreadIoEventLoopGroup; import io.netty.channel.SimpleChannelInboundHandler; -import io.netty.channel.nio.NioEventLoopGroup; +import io.netty.channel.nio.NioIoHandler; import io.netty.channel.socket.nio.NioSocketChannel; import io.netty.handler.codec.http.DefaultFullHttpRequest; import io.netty.handler.codec.http.HttpClientCodec; @@ -1083,7 +1084,7 @@ private void createRandomJettyFiles(File dir, int num, List collector) thr } private Channel getChannel(int port, ClientHandler clientHandler) throws InterruptedException { - EventLoopGroup group = new NioEventLoopGroup(); + EventLoopGroup group = new MultiThreadIoEventLoopGroup(NioIoHandler.newFactory()); Bootstrap bootstrap = new Bootstrap(); bootstrap.group(group).channel(NioSocketChannel.class).handler(new ChannelInitializer() { @Override @@ -1096,7 +1097,7 @@ protected void initChannel(Channel ch) throws Exception { } private Channel getSslChannel(int port, SslHandler sslHandler, ClientHandler clientHandler) throws InterruptedException { - EventLoopGroup group = new NioEventLoopGroup(); + EventLoopGroup group = new MultiThreadIoEventLoopGroup(NioIoHandler.newFactory()); Bootstrap bootstrap = new Bootstrap(); bootstrap.group(group).channel(NioSocketChannel.class).handler(new ChannelInitializer() { @Override diff --git a/docs/user-manual/configuring-transports.adoc b/docs/user-manual/configuring-transports.adoc index 056eed7a46c..98131471c8c 100644 --- a/docs/user-manual/configuring-transports.adoc +++ b/docs/user-manual/configuring-transports.adoc @@ -244,7 +244,7 @@ These Native transports add features specific to a particular platform, generate Both Clients and Server can benefit from this. -Current Supported Platforms. +Currently supported platforms: * Linux running 64bit JVM * MacOS running 64bit JVM @@ -255,7 +255,7 @@ If running on an unsupported platform or if there are any issues loading native ==== Linux Native Transport -On supported Linux platforms Epoll is used, @see https://en.wikipedia.org/wiki/Epoll. +On supported Linux platforms Epoll can be used, @see https://en.wikipedia.org/wiki/Epoll. The following properties are specific to this native transport: @@ -264,6 +264,24 @@ enables the use of epoll if a supported linux platform is running a 64bit JVM is Setting this to `false` will force the use of Java NIO instead of epoll. Default is `true` +Additionally, IO_URING can be used, @see https://en.wikipedia.org/wiki/Io_uring. + +The following properties are specific to this native transport: + +useIoUring:: +enables the use of IO_URING if a supported linux platform running a 64bit JVM is detected. +Setting this to `false` will attempt the use of `epoll`, then finally falling back to using Java NIO. +Default is `false` + +[WARNING] +==== +[#io_uring-warning] +IO_URING support is a recent addition to the broker and should be considered `experimental` at this stage. +Using it _could_ introduce unwanted side effects. As such, thorough testing and verification are advised before use in any production or otherwise critical environment. + +Netty has provided a https://github.com/netty/netty/tree/4.2/transport-native-io_uring#faq[FAQ] that may be helpful. +==== + ==== MacOS Native Transport On supported MacOS platforms KQueue is used, @see https://en.wikipedia.org/wiki/Kqueue. diff --git a/pom.xml b/pom.xml index 34306414805..67454363277 100644 --- a/pom.xml +++ b/pom.xml @@ -119,7 +119,7 @@ 12.1.1 5.20.0 4.0.5 - 4.1.128.Final + 4.2.7.Final 2.2.2 5.9.0 3.9.4 @@ -255,13 +255,19 @@ ${activemq.basedir}/artemis-distribution/target/apache-artemis-${project.version}-bin/apache-artemis-${project.version} - -Dbrokerconfig.maxDiskUsage=100 -Dorg.apache.activemq.artemis.core.remoting.impl.netty.TransportConstants.DEFAULT_QUIET_PERIOD=0 -Dorg.apache.activemq.artemis.core.remoting.impl.netty.TransportConstants.DEFAULT_SHUTDOWN_TIMEOUT=0 - -Djava.library.path="${activemq.basedir}/target/bin/lib/linux-x86_64:${activemq.basedir}/target/bin/lib/linux-i686" -Djgroups.bind_addr=localhost - -Djava.net.preferIPv4Stack=true -Dbasedir=${basedir} + + -Dbrokerconfig.maxDiskUsage=100 + -Dorg.apache.activemq.artemis.core.remoting.impl.netty.TransportConstants.DEFAULT_QUIET_PERIOD=0 + -Dorg.apache.activemq.artemis.core.remoting.impl.netty.TransportConstants.DEFAULT_SHUTDOWN_TIMEOUT=0 + -Djava.library.path="${activemq.basedir}/target/bin/lib/linux-x86_64:${activemq.basedir}/target/bin/lib/linux-i686" + -Djgroups.bind_addr=localhost + -Djava.net.preferIPv4Stack=true + -Dbasedir=${basedir} -Djdk.attach.allowAttachSelf=true -Dartemis.distribution.output="${artemis.distribution.output}" -Dlog4j2.configurationFile="file:${activemq.basedir}/tests/config/${logging.config}" - + + ${project.basedir} false @@ -271,6 +277,7 @@ linux-x86_64 osx-x86_64 + linux-x86_64 false @@ -737,6 +744,30 @@ + + jdk-17-23 + + [17, 23] + + + + ${initial-activemq-surefire-argline} + + + + + jdk-25 + + [24, 25] + + + + ${initial-activemq-surefire-argline} + --enable-native-access=ALL-UNNAMED + --sun-misc-unsafe-memory-access=allow + + + diff --git a/tests/artemis-test-support/src/main/java/org/apache/activemq/artemis/tests/util/TcpProxy.java b/tests/artemis-test-support/src/main/java/org/apache/activemq/artemis/tests/util/TcpProxy.java index 65c7b9ae2d2..002506c4adb 100644 --- a/tests/artemis-test-support/src/main/java/org/apache/activemq/artemis/tests/util/TcpProxy.java +++ b/tests/artemis-test-support/src/main/java/org/apache/activemq/artemis/tests/util/TcpProxy.java @@ -36,7 +36,8 @@ import io.netty.channel.ChannelOption; import io.netty.channel.ChannelPipeline; import io.netty.channel.EventLoopGroup; -import io.netty.channel.nio.NioEventLoopGroup; +import io.netty.channel.MultiThreadIoEventLoopGroup; +import io.netty.channel.nio.NioIoHandler; import io.netty.channel.socket.SocketChannel; import io.netty.channel.socket.nio.NioServerSocketChannel; import io.netty.handler.logging.LogLevel; @@ -144,8 +145,8 @@ public void run() { logger.info("Proxying {} to {}", localPort, remotePort); // Configure the bootstrap. - EventLoopGroup bossGroup = new NioEventLoopGroup(1); - EventLoopGroup workerGroup = new NioEventLoopGroup(); + EventLoopGroup bossGroup = new MultiThreadIoEventLoopGroup(1, NioIoHandler.newFactory()); + EventLoopGroup workerGroup = new MultiThreadIoEventLoopGroup(NioIoHandler.newFactory()); try { ServerBootstrap b = new ServerBootstrap(); b.group(bossGroup, workerGroup) diff --git a/tests/artemis-test-support/src/main/java/org/apache/activemq/transport/netty/NettyTcpTransport.java b/tests/artemis-test-support/src/main/java/org/apache/activemq/transport/netty/NettyTcpTransport.java index 234a46fa76d..9280b8b8434 100644 --- a/tests/artemis-test-support/src/main/java/org/apache/activemq/transport/netty/NettyTcpTransport.java +++ b/tests/artemis-test-support/src/main/java/org/apache/activemq/transport/netty/NettyTcpTransport.java @@ -17,20 +17,13 @@ package org.apache.activemq.transport.netty; import java.io.IOException; +import java.lang.invoke.MethodHandles; import java.net.URI; import java.security.Principal; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.function.Function; -import static java.util.function.Function.identity; - -import io.netty.channel.ChannelPromise; -import io.netty.util.ReferenceCounted; -import org.apache.activemq.transport.amqp.client.util.IOExceptionSupport; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; -import java.lang.invoke.MethodHandles; import io.netty.bootstrap.Bootstrap; import io.netty.buffer.ByteBuf; @@ -42,15 +35,23 @@ import io.netty.channel.ChannelInitializer; import io.netty.channel.ChannelOption; import io.netty.channel.ChannelPipeline; +import io.netty.channel.ChannelPromise; import io.netty.channel.EventLoopGroup; import io.netty.channel.FixedRecvByteBufAllocator; +import io.netty.channel.MultiThreadIoEventLoopGroup; import io.netty.channel.SimpleChannelInboundHandler; -import io.netty.channel.nio.NioEventLoopGroup; +import io.netty.channel.nio.NioIoHandler; import io.netty.channel.socket.nio.NioSocketChannel; import io.netty.handler.logging.LoggingHandler; import io.netty.handler.ssl.SslHandler; +import io.netty.util.ReferenceCounted; import io.netty.util.concurrent.Future; import io.netty.util.concurrent.GenericFutureListener; +import org.apache.activemq.transport.amqp.client.util.IOExceptionSupport; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import static java.util.function.Function.identity; /** * TCP based transport that uses Netty as the underlying IO layer. @@ -125,7 +126,7 @@ public void connect() throws IOException { sslHandler = null; } - group = new NioEventLoopGroup(1); + group = new MultiThreadIoEventLoopGroup(1, NioIoHandler.newFactory()); bootstrap = new Bootstrap(); bootstrap.group(group); diff --git a/tests/integration-tests-isolated/src/test/java/org/apache/activemq/artemis/tests/integration/isolated/web/WebServerComponentTest.java b/tests/integration-tests-isolated/src/test/java/org/apache/activemq/artemis/tests/integration/isolated/web/WebServerComponentTest.java index 9b470c8ae6d..444c6fb0d78 100644 --- a/tests/integration-tests-isolated/src/test/java/org/apache/activemq/artemis/tests/integration/isolated/web/WebServerComponentTest.java +++ b/tests/integration-tests-isolated/src/test/java/org/apache/activemq/artemis/tests/integration/isolated/web/WebServerComponentTest.java @@ -16,11 +16,6 @@ */ package org.apache.activemq.artemis.tests.integration.isolated.web; -import static org.junit.jupiter.api.Assertions.assertEquals; -import static org.junit.jupiter.api.Assertions.assertFalse; -import static org.junit.jupiter.api.Assertions.assertNull; -import static org.junit.jupiter.api.Assertions.assertTrue; - import java.io.BufferedReader; import java.io.File; import java.io.FileReader; @@ -37,8 +32,9 @@ import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelInitializer; import io.netty.channel.EventLoopGroup; +import io.netty.channel.MultiThreadIoEventLoopGroup; import io.netty.channel.SimpleChannelInboundHandler; -import io.netty.channel.nio.NioEventLoopGroup; +import io.netty.channel.nio.NioIoHandler; import io.netty.channel.socket.nio.NioSocketChannel; import io.netty.handler.codec.http.DefaultFullHttpRequest; import io.netty.handler.codec.http.HttpClientCodec; @@ -62,6 +58,11 @@ import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + /** * This test leaks a thread named org.eclipse.jetty.util.RolloverFileOutputStream which is why it is isolated now. In * the future Jetty might fix this. @@ -211,7 +212,7 @@ public void testLargeResponseHeaderConfiguration() throws Exception { } private Channel getChannel(int port, ClientHandler clientHandler) throws InterruptedException { - EventLoopGroup group = new NioEventLoopGroup(); + EventLoopGroup group = new MultiThreadIoEventLoopGroup(NioIoHandler.newFactory()); Bootstrap bootstrap = new Bootstrap(); bootstrap.group(group).channel(NioSocketChannel.class).handler(new ChannelInitializer() { @Override diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/http/HttpAuthorityTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/http/HttpAuthorityTest.java index 925ede3a77c..73832632e5d 100644 --- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/http/HttpAuthorityTest.java +++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/http/HttpAuthorityTest.java @@ -26,8 +26,9 @@ import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelInitializer; import io.netty.channel.EventLoopGroup; +import io.netty.channel.MultiThreadIoEventLoopGroup; import io.netty.channel.SimpleChannelInboundHandler; -import io.netty.channel.nio.NioEventLoopGroup; +import io.netty.channel.nio.NioIoHandler; import io.netty.channel.socket.SocketChannel; import io.netty.channel.socket.nio.NioServerSocketChannel; import io.netty.handler.codec.http.DefaultFullHttpResponse; @@ -51,8 +52,8 @@ public void testHttpAuthority() throws Exception { int port = 61616; CountDownLatch requestTested = new CountDownLatch(1); AtomicBoolean failed = new AtomicBoolean(false); - EventLoopGroup bossGroup = new NioEventLoopGroup(); - EventLoopGroup workerGroup = new NioEventLoopGroup(); + EventLoopGroup bossGroup = new MultiThreadIoEventLoopGroup(NioIoHandler.newFactory()); + EventLoopGroup workerGroup = new MultiThreadIoEventLoopGroup(NioIoHandler.newFactory()); try { ServerBootstrap bootstrap = new ServerBootstrap(); bootstrap.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class).childHandler(new ChannelInitializer() { @@ -88,6 +89,7 @@ protected void channelRead0(ChannelHandlerContext ctx, FullHttpRequest req) { } assertTrue(requestTested.await(500, TimeUnit.MILLISECONDS)); assertFalse(failed.get()); + future.channel().close(); } finally { workerGroup.shutdownGracefully(); bossGroup.shutdownGracefully(); diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/transports/netty/NettyConnectorWithHTTPUpgradeTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/transports/netty/NettyConnectorWithHTTPUpgradeTest.java index ba17a06c8b6..1b414b023d1 100644 --- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/transports/netty/NettyConnectorWithHTTPUpgradeTest.java +++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/transports/netty/NettyConnectorWithHTTPUpgradeTest.java @@ -24,11 +24,14 @@ import java.util.Map; import io.netty.bootstrap.ServerBootstrap; +import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelInitializer; import io.netty.channel.ChannelPipeline; +import io.netty.channel.EventLoopGroup; +import io.netty.channel.MultiThreadIoEventLoopGroup; import io.netty.channel.SimpleChannelInboundHandler; -import io.netty.channel.nio.NioEventLoopGroup; +import io.netty.channel.nio.NioIoHandler; import io.netty.channel.socket.SocketChannel; import io.netty.channel.socket.nio.NioServerSocketChannel; import io.netty.handler.codec.http.DefaultFullHttpResponse; @@ -107,8 +110,9 @@ public NettyConnectorWithHTTPUpgradeTest(Boolean useSSL) { private ServerLocator locator; private String acceptorName; - private NioEventLoopGroup bossGroup; - private NioEventLoopGroup workerGroup; + private ChannelFuture channelFuture; + private EventLoopGroup bossGroup; + private EventLoopGroup workerGroup; private String SERVER_SIDE_KEYSTORE = "server-keystore.jks"; private String CLIENT_SIDE_TRUSTSTORE = "server-ca-truststore.jks"; @@ -214,8 +218,8 @@ public void HTTPUpgradeConnectorUsingNormalAcceptor() throws Exception { } private void startWebServer(int port) throws Exception { - bossGroup = new NioEventLoopGroup(); - workerGroup = new NioEventLoopGroup(); + bossGroup = new MultiThreadIoEventLoopGroup(NioIoHandler.newFactory()); + workerGroup = new MultiThreadIoEventLoopGroup(NioIoHandler.newFactory()); ServerBootstrap b = new ServerBootstrap(); final SSLContext context; if (useSSL) { @@ -276,11 +280,12 @@ public void channelReadComplete(ChannelHandlerContext ctx) throws Exception { ctx.flush(); } }); - b.bind(port).sync(); + channelFuture = b.bind(port).sync(); } private void stopWebServer() { - bossGroup.shutdownGracefully(); + channelFuture.channel().close(); workerGroup.shutdownGracefully(); + bossGroup.shutdownGracefully(); } } diff --git a/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/remoting/impl/netty/SocksProxyTest.java b/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/remoting/impl/netty/SocksProxyTest.java index e84f92d47e5..1997e417f9f 100644 --- a/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/remoting/impl/netty/SocksProxyTest.java +++ b/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/remoting/impl/netty/SocksProxyTest.java @@ -16,13 +16,6 @@ */ package org.apache.activemq.artemis.tests.unit.core.remoting.impl.netty; -import static org.junit.jupiter.api.Assertions.assertFalse; -import static org.junit.jupiter.api.Assertions.assertNotNull; -import static org.junit.jupiter.api.Assertions.assertNull; -import static org.junit.jupiter.api.Assertions.assertSame; -import static org.junit.jupiter.api.Assertions.assertTrue; -import static org.junit.jupiter.api.Assumptions.assumeTrue; - import java.net.InetAddress; import java.net.InetSocketAddress; import java.net.NetworkInterface; @@ -38,7 +31,9 @@ import io.netty.bootstrap.ServerBootstrap; import io.netty.channel.ChannelInitializer; import io.netty.channel.ChannelPipeline; -import io.netty.channel.nio.NioEventLoopGroup; +import io.netty.channel.EventLoopGroup; +import io.netty.channel.MultiThreadIoEventLoopGroup; +import io.netty.channel.nio.NioIoHandler; import io.netty.channel.socket.SocketChannel; import io.netty.channel.socket.nio.NioServerSocketChannel; import io.netty.handler.proxy.Socks5ProxyHandler; @@ -58,6 +53,13 @@ import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.junit.jupiter.api.Assumptions.assumeTrue; + public class SocksProxyTest extends ActiveMQTestBase { private static final int SOCKS_PORT = 1080; @@ -66,8 +68,8 @@ public class SocksProxyTest extends ActiveMQTestBase { private ExecutorService threadPool; private ScheduledExecutorService scheduledThreadPool; - private NioEventLoopGroup bossGroup; - private NioEventLoopGroup workerGroup; + private EventLoopGroup bossGroup; + private EventLoopGroup workerGroup; @Override @BeforeEach @@ -264,8 +266,8 @@ public void connectionReadyForWrites(Object connectionID, boolean ready) { } private void startSocksProxy() throws Exception { - bossGroup = new NioEventLoopGroup(); - workerGroup = new NioEventLoopGroup(); + bossGroup = new MultiThreadIoEventLoopGroup(NioIoHandler.newFactory()); + workerGroup = new MultiThreadIoEventLoopGroup(NioIoHandler.newFactory()); ServerBootstrap b = new ServerBootstrap(); b.group(bossGroup, workerGroup);