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 693b17569722..bbec6d082051 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 @@ -124,6 +124,11 @@ if [ -n "$ARTEMIS_SHELL_HISTORY" ]; then SHELL_HISTORY_ARG="-Dartemis.shell.history=${ARTEMIS_SHELL_HISTORY}" fi +# Netty 4.2 only uses sun.misc.Unsafe on JDK 24+ when it's explicitly permitted; the NIO/MAPPED journals +# rely on it, so enable it there to avoid a broker startup failure. The option is unrecognized before JDK 23, +# so it's only added when recognized. +$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 \ @@ -138,6 +143,7 @@ exec "$JAVACMD" \ -Ddata.dir="$ARTEMIS_DATA_DIR" \ -Dartemis.instance.etc="$ARTEMIS_INSTANCE_ETC" \ $SHELL_HISTORY_ARG \ + $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/bin/artemis.cmd b/artemis-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/bin/artemis.cmd index b64ad80d4374..375b189ff822 100644 --- a/artemis-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/bin/artemis.cmd +++ b/artemis-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/bin/artemis.cmd @@ -74,6 +74,11 @@ move /Y %ARTEMIS_OOME_DUMP% %ARTEMIS_OOME_DUMP%.bkp rem "Create full JVM Args" set JVM_ARGS=%LOGGING_ARGS% set JVM_ARGS=%JVM_ARGS% %JAVA_ARGS% + +rem "Netty 4.2 only uses sun.misc.Unsafe on JDK 24+ when it's explicitly permitted; the NIO/MAPPED journals" +rem "rely on it, so enable it there to avoid a broker startup failure. The option is unrecognized before JDK 23," +rem "so it's only added when recognized." +"%_JAVACMD%" --sun-misc-unsafe-memory-access=allow --version >nul 2>&1 && set JVM_ARGS=%JVM_ARGS% --sun-misc-unsafe-memory-access=allow if not "%ARTEMIS_CLUSTER_PROPS%"=="" set JVM_ARGS=%JVM_ARGS% %ARTEMIS_CLUSTER_PROPS% set JVM_ARGS=%JVM_ARGS% -classpath %ARTEMIS_HOME%\lib\artemis-boot.jar set JVM_ARGS=%JVM_ARGS% -Dartemis.home=%ARTEMIS_HOME% 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 e23f776e7d85..44155962340f 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 @@ -28,7 +28,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-utility.profile.cmd b/artemis-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/etc/artemis-utility.profile.cmd index f2ca8ba63667..26e4190c2fcf 100644 --- a/artemis-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/etc/artemis-utility.profile.cmd +++ b/artemis-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/etc/artemis-utility.profile.cmd @@ -22,7 +22,7 @@ set ARTEMIS_DATA_DIR="${artemis.instance.data}" IF "%LOGGING_ARGS%"=="" (set LOGGING_ARGS=-Dlog4j2.configurationFile=log4j2-utility.properties) -IF "%JAVA_ARGS%"=="" (set JAVA_ARGS=-Dlog4j2.disableJmx=true --add-opens java.base/jdk.internal.misc=ALL-UNNAMED ${java-utility-opts}) +IF "%JAVA_ARGS%"=="" (set JAVA_ARGS=-Dlog4j2.disableJmx=true --add-opens java.base/jdk.internal.misc=ALL-UNNAMED --enable-native-access=ALL-UNNAMED ${java-utility-opts}) rem Uncomment to enable remote debugging rem set DEBUG_ARGS=-agentlib:jdwp=transport=dt_socket,server=y,suspend=y,address=5005 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 687cdd5acf1d..23d4a133fe29 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-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/etc/artemis.profile.cmd b/artemis-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/etc/artemis.profile.cmd index c5d4b0ec3e64..40b191cb027e 100644 --- a/artemis-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/etc/artemis.profile.cmd +++ b/artemis-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/etc/artemis.profile.cmd @@ -25,7 +25,7 @@ rem Cluster Properties: Used to pass arguments to Apache Artemis which can be re rem set ARTEMIS_CLUSTER_PROPS=-Dactivemq.remoting.default.port=61617 -Dactivemq.remoting.amqp.port=5673 -Dactivemq.remoting.stomp.port=61614 -Dactivemq.remoting.hornetq.port=5446 rem Java Opts -IF "%JAVA_ARGS%"=="" (set 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 -Dhawtio.roles=${role} ${java-opts} ) +IF "%JAVA_ARGS%"=="" (set 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 -Dhawtio.roles=${role} ${java-opts} ) rem Logs Safepoints JVM pauses: Uncomment to enable them rem In addition to the traditional GC logs you could enable some JVM flags to know any meaningful and "hidden" pause that could diff --git a/artemis-commons/pom.xml b/artemis-commons/pom.xml index a3d948048e24..0e8a7b50d1c1 100644 --- a/artemis-commons/pom.xml +++ b/artemis-commons/pom.xml @@ -83,6 +83,25 @@ io.netty netty-transport + + io.netty + netty-transport-classes-epoll + + + io.netty + netty-transport-classes-kqueue + + + io.netty + netty-transport-classes-io_uring + true + + + io.netty + netty-transport-native-io_uring + ${netty-transport-native-io_uring-classifier} + test + commons-beanutils commons-beanutils diff --git a/artemis-commons/src/main/java/org/apache/activemq/artemis/logs/ActiveMQUtilLogger.java b/artemis-commons/src/main/java/org/apache/activemq/artemis/logs/ActiveMQUtilLogger.java index 540b3a5d39ed..3b0dbfa9d380 100644 --- a/artemis-commons/src/main/java/org/apache/activemq/artemis/logs/ActiveMQUtilLogger.java +++ b/artemis-commons/src/main/java/org/apache/activemq/artemis/logs/ActiveMQUtilLogger.java @@ -77,4 +77,22 @@ public interface ActiveMQUtilLogger { @LogMessage(id = 202017, value = "Algorithm two-way is deprecated and will be removed from the default codec in a future version. Use a custom codec instead. Consult the manual for details.", level = LogMessage.Level.WARN) void deprecatedDefaultCodecTwoWayAlgorithm(); + + @LogMessage(id = 202018, value = "Unable to check KQueue availability ", level = LogMessage.Level.WARN) + void unableToCheckKQueueAvailability(Throwable e); + + @LogMessage(id = 202019, value = "KQueue is not available, please add to the classpath or configure useKQueue=false to remove this warning", level = LogMessage.Level.WARN) + void unableToCheckKQueueAvailabilityNoClass(); + + @LogMessage(id = 202020, value = "Unable to check Epoll availability ", level = LogMessage.Level.WARN) + void unableToCheckEpollAvailability(Throwable e); + + @LogMessage(id = 202021, value = "Epoll is not available, please add to the classpath or configure useEpoll=false to remove this warning", level = LogMessage.Level.WARN) + void unableToCheckEpollAvailabilityNoClass(); + + @LogMessage(id = 202022, value = "Unable to check IoUring availability ", level = LogMessage.Level.WARN) + void unableToCheckIoUringAvailability(Throwable e); + + @LogMessage(id = 202023, 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-commons/src/main/java/org/apache/activemq/artemis/utils/ByteUtil.java b/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/ByteUtil.java index 64e09abd8978..0859f6091380 100644 --- a/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/ByteUtil.java +++ b/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/ByteUtil.java @@ -520,8 +520,11 @@ private static void uncheckedZeros(final ByteBuffer buffer, int offset, int byte Arrays.fill(buffer.array(), start, start + bytes, zero); } else { //slow path + // absolute put() checks index against limit, but we zero up to capacity + final ByteBuffer dup = buffer.duplicate(); + dup.clear(); for (int i = 0; i < bytes; i++) { - buffer.put(i + offset, zero); + dup.put(i + offset, zero); } } } diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/CheckDependencies.java b/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/CheckDependencies.java similarity index 61% rename from artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/CheckDependencies.java rename to artemis-commons/src/main/java/org/apache/activemq/artemis/utils/CheckDependencies.java index f31922eecac5..ae57f5e1afca 100644 --- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/CheckDependencies.java +++ b/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/CheckDependencies.java @@ -14,17 +14,14 @@ * See the License for the specific language governing permissions and * limitations under the License. */ - -package org.apache.activemq.artemis.core.remoting.impl.netty; +package org.apache.activemq.artemis.utils; import io.netty.channel.epoll.Epoll; import io.netty.channel.kqueue.KQueue; -import org.apache.activemq.artemis.core.client.ActiveMQClientLogger; -import org.apache.activemq.artemis.utils.Env; +import org.apache.activemq.artemis.logs.ActiveMQUtilLogger; /** - * This class will check for Epoll or KQueue is available, and return false in case of NoClassDefFoundError it could be - * improved to check for other cases eventually. + * This class will check if certain dependencies are available, and return false in case of NoClassDefFoundError */ public class CheckDependencies { @@ -32,10 +29,10 @@ public static final boolean isEpollAvailable() { try { return Env.isLinuxOs() && Epoll.isAvailable(); } catch (NoClassDefFoundError noClassDefFoundError) { - ActiveMQClientLogger.LOGGER.unableToCheckEpollAvailabilitynoClass(); + ActiveMQUtilLogger.LOGGER.unableToCheckEpollAvailabilityNoClass(); return false; } catch (Throwable e) { - ActiveMQClientLogger.LOGGER.unableToCheckEpollAvailability(e); + ActiveMQUtilLogger.LOGGER.unableToCheckEpollAvailability(e); return false; } } @@ -44,10 +41,22 @@ public static final boolean isKQueueAvailable() { try { return Env.isMacOs() && KQueue.isAvailable(); } catch (NoClassDefFoundError noClassDefFoundError) { - ActiveMQClientLogger.LOGGER.unableToCheckKQueueAvailabilityNoClass(); + ActiveMQUtilLogger.LOGGER.unableToCheckKQueueAvailabilityNoClass(); + return false; + } catch (Throwable e) { + ActiveMQUtilLogger.LOGGER.unableToCheckKQueueAvailability(e); + return false; + } + } + + public static final boolean isIoUringAvailable() { + try { + return Env.isLinuxOs() && NettyIoUringSupport.isAvailable(); + } catch (NoClassDefFoundError noClassDefFoundError) { + ActiveMQUtilLogger.LOGGER.unableToCheckIoUringAvailabilityNoClass(); return false; } catch (Throwable e) { - ActiveMQClientLogger.LOGGER.unableToCheckKQueueAvailability(e); + ActiveMQUtilLogger.LOGGER.unableToCheckIoUringAvailability(e); return false; } } diff --git a/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/NettyIoUringSupport.java b/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/NettyIoUringSupport.java new file mode 100644 index 000000000000..f9e0a76b8129 --- /dev/null +++ b/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/NettyIoUringSupport.java @@ -0,0 +1,44 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.activemq.artemis.utils; + +import io.netty.channel.Channel; +import io.netty.channel.IoHandlerFactory; +import io.netty.channel.ServerChannel; +import io.netty.channel.uring.IoUring; +import io.netty.channel.uring.IoUringIoHandler; +import io.netty.channel.uring.IoUringServerSocketChannel; +import io.netty.channel.uring.IoUringSocketChannel; + +public final class NettyIoUringSupport { + + public static boolean isAvailable() { + return IoUring.isAvailable(); + } + + public static IoHandlerFactory newHandlerFactory() { + return IoUringIoHandler.newFactory(); + } + + public static Class socketChannelClass() { + return IoUringSocketChannel.class; + } + + public static Class serverSocketChannelClass() { + return IoUringServerSocketChannel.class; + } +} 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 c0781c24c552..a3f9a318bd38 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); } + if (Runtime.version().feature() >= 24) { + commandList.add("--enable-native-access=ALL-UNNAMED"); + // the NIO/MAPPED journals rely on unsafe + 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/NettyIoUringSupportTest.java b/artemis-commons/src/test/java/org/apache/activemq/artemis/utils/NettyIoUringSupportTest.java new file mode 100644 index 000000000000..b9ac2822b8d6 --- /dev/null +++ b/artemis-commons/src/test/java/org/apache/activemq/artemis/utils/NettyIoUringSupportTest.java @@ -0,0 +1,59 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.activemq.artemis.utils; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assumptions.assumeTrue; + +import org.junit.jupiter.api.Test; + +/** + * Sanity checks for {@link NettyIoUringSupport}, the single class that references the Netty io_uring transport types. + * The io_uring classes are on this module's (optional) test classpath, so the class-level assertions below always run; + * the native-availability assertion is guarded because it needs a supported Linux platform with the native lib loaded. + */ +public class NettyIoUringSupportTest { + + @Test + public void testChannelClassesAreIoUringTypes() { + assertEquals("io.netty.channel.uring.IoUringSocketChannel", NettyIoUringSupport.socketChannelClass().getName()); + assertEquals("io.netty.channel.uring.IoUringServerSocketChannel", NettyIoUringSupport.serverSocketChannelClass().getName()); + } + + @Test + public void testAvailabilityCheckDoesNotThrow() { + // this must always return cleanly (true or false); it must never propagate an exception, since callers rely on it + // as the gate before touching any io_uring type + boolean available = CheckDependencies.isIoUringAvailable(); + + // when Env reports a non-Linux platform io_uring can never be considered available + if (!Env.isLinuxOs()) { + assertEquals(false, available); + } + } + + @Test + public void testHandlerFactoryWhenAvailable() { + assumeTrue(CheckDependencies.isIoUringAvailable()); + + // on a platform where io_uring is available the isolated factory/channel accessors must produce usable objects + assertNotNull(NettyIoUringSupport.newHandlerFactory()); + assertNotNull(NettyIoUringSupport.socketChannelClass()); + assertNotNull(NettyIoUringSupport.serverSocketChannelClass()); + } +} diff --git a/artemis-core-client-osgi/pom.xml b/artemis-core-client-osgi/pom.xml index eac31bb82693..81eeba835b1e 100644 --- a/artemis-core-client-osgi/pom.xml +++ b/artemis-core-client-osgi/pom.xml @@ -71,7 +71,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 55e4c5af364f..7bef87d5bcda 100644 --- a/artemis-core-client/pom.xml +++ b/artemis-core-client/pom.xml @@ -89,6 +89,9 @@ io.netty netty-transport-classes-kqueue + io.netty netty-codec-http @@ -109,10 +112,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 ce5007d0a08d..45cfe82caefc 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 @@ -28,7 +28,7 @@ /** * Logger Codes 210000 - 218999 */ -@LogBundle(projectCode = "AMQ", regexID = "21[0-8][0-9]{3}", retiredIDs = {211001, 211002, 211003, 212000, 212006, 212029, 212074, 212078, 214012, 214023, 214024, 214026, 214027, 214028, 214029}) +@LogBundle(projectCode = "AMQ", regexID = "21[0-8][0-9]{3}", retiredIDs = {211001, 211002, 211003, 212000, 212006, 212029, 212071, 212073, 212074, 212075, 212076, 212078, 214012, 214023, 214024, 214026, 214027, 214028, 214029}) public interface ActiveMQClientLogger { ActiveMQClientLogger LOGGER = BundleFactory.newBundle(ActiveMQClientLogger.class, ActiveMQClientLogger.class.getPackage().getName()); @@ -236,21 +236,9 @@ public interface ActiveMQClientLogger { @LogMessage(id = 212070, value = "Unable to initialize VersionLoader ", level = LogMessage.Level.WARN) void unableToInitVersionLoader(Throwable e); - @LogMessage(id = 212071, value = "Unable to check Epoll availability ", level = LogMessage.Level.WARN) - void unableToCheckEpollAvailability(Throwable e); - @LogMessage(id = 212072, value = "Failed to change channel state to ReadyForWriting ", level = LogMessage.Level.WARN) void failedToSetChannelReadyForWriting(Throwable e); - @LogMessage(id = 212073, value = "Unable to check KQueue availability ", level = LogMessage.Level.WARN) - void unableToCheckKQueueAvailability(Throwable e); - - @LogMessage(id = 212075, value = "KQueue is not available, please add to the classpath or configure useKQueue=false to remove this warning", level = LogMessage.Level.WARN) - void unableToCheckKQueueAvailabilityNoClass(); - - @LogMessage(id = 212076, value = "Epoll is not available, please add to the classpath or configure useEpoll=false to remove this warning", level = LogMessage.Level.WARN) - void unableToCheckEpollAvailabilitynoClass(); - @LogMessage(id = 212077, value = "Timed out waiting to receive initial broadcast from cluster. Retry {} of {}", level = LogMessage.Level.WARN) void broadcastTimeout(int retry, int maxretry); 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 ff0525163957..f74bab3d33bd 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,15 +62,16 @@ 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.handler.codec.base64.Base64; import io.netty.handler.codec.http.DefaultFullHttpRequest; @@ -92,11 +92,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; @@ -122,14 +122,16 @@ import org.apache.activemq.artemis.spi.core.remoting.ssl.OpenSSLContextFactoryProvider; import org.apache.activemq.artemis.spi.core.remoting.ssl.SSLContextConfig; import org.apache.activemq.artemis.spi.core.remoting.ssl.SSLContextFactoryProvider; +import org.apache.activemq.artemis.utils.CheckDependencies; +import org.apache.activemq.artemis.utils.NettyIoUringSupport; import org.apache.activemq.artemis.utils.ConfigurationHelper; import org.apache.activemq.artemis.utils.FutureLatch; import org.apache.activemq.artemis.utils.IPV6Util; 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()); @@ -301,6 +304,8 @@ public class NettyConnector extends AbstractConnector { private boolean useKQueue; + private boolean useIoUring; + private int remotingThreads; private boolean useGlobalWorkerPool; @@ -410,6 +415,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); @@ -551,27 +557,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, NettyIoUringSupport.newHandlerFactory()))); + } else { + group = new MultiThreadIoEventLoopGroup(remotingThreads, NettyIoUringSupport.newHandlerFactory()); + } + + connectorType = IOURING_CONNECTOR_TYPE; + channelClazz = NettyIoUringSupport.socketChannelClass(); + + 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; @@ -579,10 +601,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; @@ -756,14 +778,14 @@ public void initChannel(Channel channel) throws Exception { engine.setEnabledProtocols(originalProtocols); } - if (verifyHost) { - SSLParameters sslParameters = engine.getSSLParameters(); - sslParameters.setEndpointIdentificationAlgorithm("HTTPS"); - engine.setSSLParameters(sslParameters); - } + // Set the endpoint identification algorithm explicitly (rather than only when enabling host + // verification) so the behavior doesn't depend on the SSL provider's default. + SSLParameters sslParameters = engine.getSSLParameters(); + sslParameters.setEndpointIdentificationAlgorithm(verifyHost ? "HTTPS" : null); + engine.setSSLParameters(sslParameters); if (sniHost != null) { - SSLParameters sslParameters = engine.getSSLParameters(); + sslParameters = engine.getSSLParameters(); sslParameters.setServerNames(Arrays.asList(new SNIHostName(sniHost))); engine.setSSLParameters(sslParameters); } 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 1876b723a4c7..7f9d804a94d2 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 @@ -74,6 +74,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 */ @@ -228,6 +230,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; @@ -443,6 +447,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); @@ -523,6 +528,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 8f44f9589313..9381da38613b 100755 --- a/artemis-distribution/src/main/resources/bin/artemis +++ b/artemis-distribution/src/main/resources/bin/artemis @@ -94,6 +94,7 @@ 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 \ $DEBUG_ARGS \ $JAVA_ARGS_APPEND \ org.apache.activemq.artemis.boot.Artemis "$@" diff --git a/artemis-distribution/src/main/resources/bin/artemis.cmd b/artemis-distribution/src/main/resources/bin/artemis.cmd index 273bf851e290..47a2097937d7 100755 --- a/artemis-distribution/src/main/resources/bin/artemis.cmd +++ b/artemis-distribution/src/main/resources/bin/artemis.cmd @@ -52,6 +52,7 @@ set JVM_ARGS=%JAVA_ARGS% if not "%ARTEMIS_CLUSTER_PROPS%"=="" set JVM_ARGS=%JVM_ARGS% %ARTEMIS_CLUSTER_PROPS% set JVM_ARGS=%JVM_ARGS% -classpath %ARTEMIS_HOME%\lib\artemis-boot.jar set JVM_ARGS=%JVM_ARGS% -Dartemis.home=%ARTEMIS_HOME% +set JVM_ARGS=%JVM_ARGS% --enable-native-access=ALL-UNNAMED if not "%DEBUG_ARGS%"=="" set JVM_ARGS=%JVM_ARGS% %DEBUG_ARGS% if not "%JAVA_ARGS_APPEND%"=="" set JVM_ARGS=%JVM_ARGS% %JAVA_ARGS_APPEND% diff --git a/artemis-features/src/main/resources/features.xml b/artemis-features/src/main/resources/features.xml index c00193f8e640..6a9cf4dbbd67 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 130dbe62a8f6..cee6ee6e4432 100644 --- a/artemis-jms-client-osgi/pom.xml +++ b/artemis-jms-client-osgi/pom.xml @@ -79,7 +79,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-journal/src/main/java/org/apache/activemq/artemis/core/io/mapped/MappedFile.java b/artemis-journal/src/main/java/org/apache/activemq/artemis/core/io/mapped/MappedFile.java index f8159d1ee790..942527037053 100644 --- a/artemis-journal/src/main/java/org/apache/activemq/artemis/core/io/mapped/MappedFile.java +++ b/artemis-journal/src/main/java/org/apache/activemq/artemis/core/io/mapped/MappedFile.java @@ -27,6 +27,7 @@ import io.netty.buffer.Unpooled; import io.netty.util.internal.PlatformDependent; import org.apache.activemq.artemis.core.buffers.impl.ChannelBufferWrapper; +import org.apache.activemq.artemis.core.io.util.DirectByteBufferReleaser; import org.apache.activemq.artemis.core.journal.EncodingSupport; import org.apache.activemq.artemis.utils.PowerOf2Util; import org.apache.activemq.artemis.utils.Env; @@ -241,8 +242,8 @@ public void close() { } catch (IOException e) { throw new IllegalStateException(e); } finally { - //unmap in a deterministic way: do not rely on GC to do it - PlatformDependent.freeDirectBuffer(this.buffer); + //unmap in a deterministic way when possible: falls back to GC-triggered cleanup if native freeing is unavailable + DirectByteBufferReleaser.freeDirectBuffer(this.buffer); } } } diff --git a/artemis-journal/src/main/java/org/apache/activemq/artemis/core/io/mapped/MappedSequentialFileFactory.java b/artemis-journal/src/main/java/org/apache/activemq/artemis/core/io/mapped/MappedSequentialFileFactory.java index 8f144598c042..91cc1e1b17dc 100644 --- a/artemis-journal/src/main/java/org/apache/activemq/artemis/core/io/mapped/MappedSequentialFileFactory.java +++ b/artemis-journal/src/main/java/org/apache/activemq/artemis/core/io/mapped/MappedSequentialFileFactory.java @@ -19,11 +19,11 @@ import java.io.File; import java.nio.ByteBuffer; -import io.netty.util.internal.PlatformDependent; import org.apache.activemq.artemis.core.io.AbstractSequentialFileFactory; import org.apache.activemq.artemis.core.io.IOCriticalErrorListener; import org.apache.activemq.artemis.core.io.SequentialFile; import org.apache.activemq.artemis.core.io.util.ByteBufferPool; +import org.apache.activemq.artemis.core.io.util.DirectByteBufferReleaser; import org.apache.activemq.artemis.utils.PowerOf2Util; import org.apache.activemq.artemis.utils.ByteUtil; import org.apache.activemq.artemis.utils.Env; @@ -90,7 +90,7 @@ public ByteBuffer allocateDirectBuffer(final int size) { @Override public void releaseDirectBuffer(ByteBuffer buffer) { - PlatformDependent.freeDirectBuffer(buffer); + DirectByteBufferReleaser.freeDirectBuffer(buffer); } public MappedSequentialFileFactory enableBufferReuse() { diff --git a/artemis-journal/src/main/java/org/apache/activemq/artemis/core/io/nio/NIOSequentialFileFactory.java b/artemis-journal/src/main/java/org/apache/activemq/artemis/core/io/nio/NIOSequentialFileFactory.java index ead344b8f037..3c53a5993e0a 100644 --- a/artemis-journal/src/main/java/org/apache/activemq/artemis/core/io/nio/NIOSequentialFileFactory.java +++ b/artemis-journal/src/main/java/org/apache/activemq/artemis/core/io/nio/NIOSequentialFileFactory.java @@ -27,6 +27,7 @@ import org.apache.activemq.artemis.core.io.IOCriticalErrorListener; import org.apache.activemq.artemis.core.io.SequentialFile; import org.apache.activemq.artemis.core.io.util.ByteBufferPool; +import org.apache.activemq.artemis.core.io.util.DirectByteBufferReleaser; import org.apache.activemq.artemis.utils.PowerOf2Util; import org.apache.activemq.artemis.utils.Env; import org.apache.activemq.artemis.utils.critical.CriticalAnalyzer; @@ -134,9 +135,7 @@ public ByteBuffer allocateDirectBuffer(final int size) { @Override public void releaseDirectBuffer(ByteBuffer buffer) { - if (buffer.isDirect()) { - PlatformDependent.freeDirectBuffer(buffer); - } + DirectByteBufferReleaser.freeDirectBuffer(buffer); } @Override @@ -165,7 +164,17 @@ public void releaseBuffer(ByteBuffer buffer) { @Override public void clearBuffer(final ByteBuffer buffer) { if (buffer.isDirect()) { - PlatformDependent.setMemory(PlatformDependent.directBufferAddress(buffer), buffer.limit(), (byte) 0); + if (PlatformDependent.hasUnsafe()) { + PlatformDependent.setMemory(PlatformDependent.directBufferAddress(buffer), buffer.limit(), (byte) 0); + } else { + final int position = buffer.position(); + final byte[] zeros = new byte[Math.min(buffer.limit(), 8192)]; + buffer.position(0); + while (buffer.hasRemaining()) { + buffer.put(zeros, 0, Math.min(zeros.length, buffer.remaining())); + } + buffer.position(position); + } } else { Arrays.fill(buffer.array(), buffer.arrayOffset(), buffer.limit(), (byte) 0); } diff --git a/artemis-journal/src/main/java/org/apache/activemq/artemis/core/io/util/DirectByteBufferReleaser.java b/artemis-journal/src/main/java/org/apache/activemq/artemis/core/io/util/DirectByteBufferReleaser.java new file mode 100644 index 000000000000..a639281c274a --- /dev/null +++ b/artemis-journal/src/main/java/org/apache/activemq/artemis/core/io/util/DirectByteBufferReleaser.java @@ -0,0 +1,82 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.activemq.artemis.core.io.util; + +import java.lang.invoke.MethodHandles; +import java.nio.ByteBuffer; + +import io.netty.util.internal.PlatformDependent; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Best-effort helper to eagerly free direct {@link ByteBuffer}s that were allocated through the JDK, e.g. via + * {@link ByteBuffer#allocateDirect(int)} or {@link java.nio.channels.FileChannel#map}. + *

+ * Freeing such a buffer eagerly is only an optimization to release native memory promptly rather than waiting for the + * buffer's {@link java.lang.ref.Cleaner} to run during GC. On JDK 24+ running without {@code sun.misc.Unsafe}, Netty is + * unable to free "arbitrary" (JDK-allocated) direct buffers and {@link PlatformDependent#freeDirectBuffer(ByteBuffer)} + * throws {@link UnsupportedOperationException}. In that case the eager free is skipped and the GC-triggered + * {@link java.lang.ref.Cleaner} associated with the buffer reclaims the native memory instead. + */ +public final class DirectByteBufferReleaser { + + private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass()); + + private static final boolean CAN_FREE_DIRECT_BUFFER = probeCanFreeDirectBuffer(); + + private DirectByteBufferReleaser() { + } + + /** + * Eagerly frees the given direct {@code buffer} if the platform supports it. Does nothing for {@code null}, + * non-direct buffers, or when the platform is unable to free JDK-allocated direct buffers (in which case the memory + * is reclaimed by the GC-triggered {@link java.lang.ref.Cleaner} associated with the buffer). + *

+ * {@link PlatformDependent#freeDirectBuffer} is deprecated in favor of + * {@link io.netty.util.internal.CleanableDirectBuffer#clean()}, but that replacement can only free buffers Netty + * itself allocated via {@link PlatformDependent#allocateDirect(int)}. Artemis passes buffers it allocated through + * the JDK ({@link ByteBuffer#allocateDirect} / {@link java.nio.channels.FileChannel#map}), so freeDirectBuffer + * remains the only Netty API able to free them; hence the suppression. + */ + @SuppressWarnings("deprecation") + public static void freeDirectBuffer(ByteBuffer buffer) { + if (CAN_FREE_DIRECT_BUFFER && buffer != null && buffer.isDirect()) { + PlatformDependent.freeDirectBuffer(buffer); + } + } + + /** + * @return whether native (eager) freeing of direct buffers is available on the current platform + */ + public static boolean canFreeDirectBuffer() { + return CAN_FREE_DIRECT_BUFFER; + } + + @SuppressWarnings("deprecation") + private static boolean probeCanFreeDirectBuffer() { + // Probe with a JDK-allocated direct buffer, matching how Artemis allocates the buffers passed to this class. + final ByteBuffer probe = ByteBuffer.allocateDirect(1); + try { + PlatformDependent.freeDirectBuffer(probe); + return true; + } catch (Throwable t) { + logger.debug("Unable to eagerly free direct ByteBuffers; native memory will be reclaimed by the GC-triggered Cleaner instead. On JDK 24+ enabling sun.misc.Unsafe (e.g. --sun-misc-unsafe-memory-access=allow) restores eager freeing.", t); + return false; + } + } +} diff --git a/artemis-journal/src/main/java/org/apache/activemq/artemis/core/io/util/ThreadLocalByteBufferPool.java b/artemis-journal/src/main/java/org/apache/activemq/artemis/core/io/util/ThreadLocalByteBufferPool.java index ff3b1011b0c8..23bd32aec3bf 100644 --- a/artemis-journal/src/main/java/org/apache/activemq/artemis/core/io/util/ThreadLocalByteBufferPool.java +++ b/artemis-journal/src/main/java/org/apache/activemq/artemis/core/io/util/ThreadLocalByteBufferPool.java @@ -19,7 +19,6 @@ import java.nio.ByteBuffer; import java.util.Objects; -import io.netty.util.internal.PlatformDependent; import org.apache.activemq.artemis.utils.PowerOf2Util; import org.apache.activemq.artemis.utils.ByteUtil; import org.apache.activemq.artemis.utils.Env; @@ -56,23 +55,18 @@ public ByteBuffer borrow(final int size, boolean zeroed) { @Override public void release(ByteBuffer buffer) { Objects.requireNonNull(buffer); - boolean directBuffer = buffer.isDirect(); - if (directBuffer == direct && !buffer.isReadOnly()) { + if (buffer.isDirect() == direct && !buffer.isReadOnly()) { final ByteBuffer byteBuffer = bytesPool.get(); if (byteBuffer != buffer) { //replace with the current pooled only if greater or null if (byteBuffer == null || buffer.capacity() > byteBuffer.capacity()) { if (byteBuffer != null) { //free the smaller one - if (directBuffer) { - PlatformDependent.freeDirectBuffer(byteBuffer); - } + DirectByteBufferReleaser.freeDirectBuffer(byteBuffer); } bytesPool.set(buffer); } else { - if (directBuffer) { - PlatformDependent.freeDirectBuffer(buffer); - } + DirectByteBufferReleaser.freeDirectBuffer(buffer); } } } diff --git a/artemis-journal/src/test/java/org/apache/activemq/artemis/core/io/util/DirectByteBufferReleaserTest.java b/artemis-journal/src/test/java/org/apache/activemq/artemis/core/io/util/DirectByteBufferReleaserTest.java new file mode 100644 index 000000000000..96d56cddf960 --- /dev/null +++ b/artemis-journal/src/test/java/org/apache/activemq/artemis/core/io/util/DirectByteBufferReleaserTest.java @@ -0,0 +1,86 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.activemq.artemis.core.io.util; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +import java.io.File; +import java.nio.file.Files; +import java.util.ArrayList; +import java.util.List; + +import org.apache.activemq.artemis.core.io.SequentialFile; +import org.apache.activemq.artemis.core.io.nio.NIOSequentialFileFactory; +import org.apache.activemq.artemis.utils.SpawnedVMSupport; +import org.junit.jupiter.api.Test; + +/** + * Regression test for the NIO journal direct-buffer release path when {@code sun.misc.Unsafe} is unavailable. In that + * case Netty cannot free JDK-allocated ("arbitrary") direct buffers, so the eager free must be skipped (letting the + * GC-triggered {@link java.lang.ref.Cleaner} reclaim the memory) rather than throwing + * {@link UnsupportedOperationException}. + *

+ * On JDK 24+ at runtime by default Netty 4.2 avoids {@code sun.misc.Unsafe} (so it runs with {@code hasUnsafe=false}) + * unless the JVM is started with {@code --sun-misc-unsafe-memory-access=allow}. The same no-Unsafe path is also reached + * when Unsafe is explicitly disabled ({@code -Dio.netty.noUnsafe=true}) or on a future JDK where + * {@code sun.misc.Unsafe} is gone entirely. + */ +public class DirectByteBufferReleaserTest { + + // runs in the spawned child JVM (started with -Dio.netty.noUnsafe=true to force Netty's no-Unsafe path) + public static void main(String[] arg) { + try { + // exercise the exact path that failed during Create auto-tune: SyncCalculation -> NIOSequentialFile.fill + File dir = Files.createTempDirectory("DirectByteBufferReleaserTest").toFile(); + dir.deleteOnExit(); + NIOSequentialFileFactory factory = new NIOSequentialFileFactory(dir, 1); + factory.start(); + SequentialFile file = factory.createSequentialFile("release.dat"); + file.open(); + // allocates a direct ByteBuffer and then releases it; the release must not throw without Unsafe + file.fill(1024 * 1024); + file.close(); + // force a real release of a direct buffer through the factory (bypassing the pool) + factory.releaseDirectBuffer(factory.allocateDirectBuffer(1024)); + factory.stop(); + System.exit(0); + } catch (Throwable e) { + e.printStackTrace(); + System.exit(100); + } + } + + @Test + public void releaseDirectBufferWithoutUnsafe() throws Exception { + final String javaPath = new File(new File(System.getProperty("java.home"), "bin"), "java").getAbsolutePath(); + final List command = new ArrayList<>(); + command.add(javaPath); + command.add("-cp"); + command.add(SpawnedVMSupport.getClassPath()); + // force Netty's no-Unsafe path (hasUnsafe=false); in that configuration Netty cannot free the journal's + // JDK-allocated ("arbitrary") direct buffers and PlatformDependent.freeDirectBuffer throws, which is exactly + // the case DirectByteBufferReleaser must catch and skip + command.add("-Dio.netty.noUnsafe=true"); + command.add("-Djava.io.tmpdir=" + System.getProperty("java.io.tmpdir", "./tmp")); + command.add(DirectByteBufferReleaserTest.class.getName()); + + final ProcessBuilder builder = new ProcessBuilder(command); + builder.inheritIO(); + final Process process = builder.start(); + assertEquals(0, process.waitFor(), "releasing a direct buffer without Unsafe/native-access must not throw"); + } +} diff --git a/artemis-pom/pom.xml b/artemis-pom/pom.xml index cf5183713ad3..ac8e66d90161 100644 --- a/artemis-pom/pom.xml +++ b/artemis-pom/pom.xml @@ -458,6 +458,12 @@ ${netty.version} + + io.netty + netty-codec-base + ${netty.version} + + io.netty netty-codec-http @@ -526,6 +532,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 365f7c299115..ba68ae5e1c20 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 9355de690ded..7a01af67f6c3 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 96161074c2e4..3fcd992688c2 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 5241c9b89b7f..f157169c8302 100644 --- a/artemis-server-osgi/pom.xml +++ b/artemis-server-osgi/pom.xml @@ -129,7 +129,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, com.nimbusds.jose*;resolution:=optional, diff --git a/artemis-server/pom.xml b/artemis-server/pom.xml index a756ec82d1db..785ee78f4a81 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 1bd19470e876..5ed541338d16 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,18 +49,19 @@ 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.handler.codec.haproxy.HAProxyMessageDecoder; import io.netty.handler.ssl.SslContext; @@ -98,6 +99,8 @@ import org.apache.activemq.artemis.spi.core.remoting.ssl.SSLContextConfig; import org.apache.activemq.artemis.spi.core.remoting.ssl.SSLContextFactoryProvider; import org.apache.activemq.artemis.utils.ActiveMQThreadFactory; +import org.apache.activemq.artemis.utils.CheckDependencies; +import org.apache.activemq.artemis.utils.NettyIoUringSupport; import org.apache.activemq.artemis.utils.ConfigurationHelper; import org.apache.activemq.artemis.utils.ProxyProtocolUtil; import org.apache.activemq.artemis.utils.collections.TypedProperties; @@ -118,6 +121,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 @@ -160,6 +164,8 @@ public class NettyAcceptor extends AbstractAcceptor { private final boolean useKQueue; + private final boolean useIoUring; + private final ProtocolHandler protocolHandler; private final String host; @@ -312,6 +318,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); @@ -501,19 +508,35 @@ protected void internalStart() 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 = NettyIoUringSupport.serverSocketChannelClass(); + eventLoopGroup = new MultiThreadIoEventLoopGroup(remotingThreads, threadFactory, NettyIoUringSupport.newHandlerFactory()); + 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); } @@ -730,14 +753,14 @@ public SslHandler getSslHandler(ByteBufAllocator alloc, String peerHost, int pee engine.setEnabledProtocols(set.toArray(new String[set.size()])); - if (verifyHost) { - SSLParameters sslParameters = engine.getSSLParameters(); - sslParameters.setEndpointIdentificationAlgorithm("HTTPS"); - engine.setSSLParameters(sslParameters); - } + // Set the endpoint identification algorithm explicitly (rather than only when enabling host verification) so + // the behavior doesn't depend on the SSL provider's default. + SSLParameters sslParameters = engine.getSSLParameters(); + sslParameters.setEndpointIdentificationAlgorithm(verifyHost ? "HTTPS" : null); + engine.setSSLParameters(sslParameters); if (sniHost != null) { - SSLParameters sslParameters = engine.getSSLParameters(); + sslParameters = engine.getSSLParameters(); sslParameters.setSNIMatchers(Arrays.asList(SNIHostName.createSNIMatcher(sniHost))); engine.setSSLParameters(sslParameters); } 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 8be81812e037..17a999b79d04 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 5cf42b5d8670..79f509220a17 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 @@ -59,8 +59,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; @@ -1148,7 +1149,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 @@ -1161,7 +1162,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/artemis-website/pom.xml b/artemis-website/pom.xml index e14178b1d552..4c00766c1608 100644 --- a/artemis-website/pom.xml +++ b/artemis-website/pom.xml @@ -76,6 +76,14 @@ ${project.version} provided + + + io.netty + netty-transport-classes-io_uring + provided + diff --git a/docs/user-manual/configuring-transports.adoc b/docs/user-manual/configuring-transports.adoc index e2aeac6005d6..7836fa030ca3 100644 --- a/docs/user-manual/configuring-transports.adoc +++ b/docs/user-manual/configuring-transports.adoc @@ -250,7 +250,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 @@ -261,7 +261,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: @@ -270,6 +270,47 @@ 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. +==== + +Because IO_URING is disabled by default and experimental its Netty dependencies are *not* bundled with the broker or the clients, so they must be added explicitly before setting `useIoUring` to `true`. +Everything needed lives in two jars, both matching the Netty version the broker/client ships (see the `netty-*` jars under the distribution's `lib` directory): + +* https://central.sonatype.com/artifact/io.netty/netty-transport-native-io_uring[`io.netty:netty-transport-native-io_uring`] with the `linux-x86_64` classifier +* https://central.sonatype.com/artifact/io.netty/netty-transport-classes-io_uring[`io.netty:netty-transport-classes-io_uring`] + +To enable it on a *client* using Maven add just the native dependency alongside your existing xref:client-classpath.adoc[client dependency]: + +[,xml,subs="normal"] +---- + + io.netty + netty-transport-native-io_uring + _netty-version_ + linux-x86_64 + +---- + +To enable it on the *broker* (or a client not using Maven) add both aforementioned jars to the classpath. +See xref:using-server.adoc#adding-runtime-dependencies[here] for details on adding the jars to the broker's classpath. + +If these jars are not present the broker/client silently falls back to `epoll` and then Java NIO, exactly as if `useIoUring` were `false`. + ==== MacOS Native Transport On supported MacOS platforms KQueue is used, @see https://en.wikipedia.org/wiki/Kqueue. diff --git a/docs/user-manual/versions.adoc b/docs/user-manual/versions.adoc index 518c121fbbb9..ef765cad78bb 100644 --- a/docs/user-manual/versions.adoc +++ b/docs/user-manual/versions.adoc @@ -15,6 +15,35 @@ NOTE: Follow the general upgrade procedure outlined in the xref:upgrading.adoc#u :sectnums!: +== Version 2.57.0 + +https://issues.apache.org/jira/secure/ReleaseNote.jspa?projectId=12315920&version=12357464[Full release notes] + +=== Highlights + +* Netty was upgraded to 4.2. +The most notable change is a new default "adaptive" buffer allocator (which was experimental in 4.1). +It automatically tunes itself to the observed workload. +If you happen to observe a performance regression for your particular workload you can restore the previous allocator by passing `-Dio.netty.allocator.type=pooled` to the JVM on startup. +* Experimental support for the native https://en.wikipedia.org/wiki/Io_uring[io_uring] transport on Linux was added. +See the xref:configuring-transports.adoc#linux-native-transport[Linux native transport] documentation for details. + +=== Upgrading from 2.56.0 + +* Due to https://issues.apache.org/jira/browse/ARTEMIS-5734[ARTEMIS-5734] Artemis ships Netty 4.2 which no longer uses `sun.misc.Unsafe` by default on JDK 24+. +Previous versions of the broker shipped Netty 4.1, which used `sun.misc.Unsafe` unconditionally and therefore triggered the one-time "terminally deprecated method in sun.misc.Unsafe has been called" warning printed by JDK 24+. +Netty 4.2 only uses `sun.misc.Unsafe` when the JVM is started with `--sun-misc-unsafe-memory-access=allow`. +See Netty's https://netty.io/wiki/java-24-and-sun.misc.unsafe.html[Java 24 and sun.misc.Unsafe] documentation for further details. ++ +That said, the broker currently relies on direct-memory operations from Netty which in turn rely on `sun.misc.Unsafe`. +Freshly created brokers and brokers upgraded via xref:upgrading.adoc#upgrading-tool[`artemis upgrade`] get this enabled automatically via the `--sun-misc-unsafe-memory-access=allow` setting on JDK 24+. +If you instead upgrade by swapping the broker home directory in place then, when running on JDK 24+, you must add `--sun-misc-unsafe-memory-access=allow` to `JAVA_ARGS` in `etc/artemis.profile` (or `etc/artemis.profile.cmd` on Windows) yourself. +This is a stopgap: `sun.misc.Unsafe` (and the `--sun-misc-unsafe-memory-access` flag) is being removed from future JDKs, and a subsequent Artemis release will remove the dependency. ++ +Optimal, safe performance for Netty 4.2 on JDK 24+ relies on the JVM's native access being enabled. +Freshly created brokers and brokers updated via xref:upgrading.adoc#upgrading-tool[`artemis upgrade`] get this enabled automatically via the `--enable-native-access=ALL-UNNAMED` setting. +If you instead upgrade by swapping the broker home directory in place then, when running on JDK 24+, you must add it to `JAVA_ARGS` in `etc/artemis.profile` (or `etc/artemis.profile.cmd` on Windows) yourself. + == Version 2.56.0 https://issues.apache.org/jira/secure/ReleaseNote.jspa?projectId=12315920&version=12357208[Full release notes] diff --git a/pom.xml b/pom.xml index 282458e75d25..ad27df70f0aa 100644 --- a/pom.xml +++ b/pom.xml @@ -122,7 +122,7 @@ 12.3.1 5.23.0 4.0.7 - 4.1.137.Final + 4.2.17.Final 2.2.2 5.9.0 3.9.5 @@ -240,6 +240,7 @@ false true + + + io.netty + netty-transport-native-io_uring + ${netty-transport-native-io_uring-classifier} + test + org.fusesource.hawtbuf hawtbuf 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 925ede3a77cf..73832632e5d1 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/remoting/IoUringTransportTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/remoting/IoUringTransportTest.java new file mode 100644 index 000000000000..b01646bf2303 --- /dev/null +++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/remoting/IoUringTransportTest.java @@ -0,0 +1,104 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.activemq.artemis.tests.integration.remoting; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assumptions.assumeTrue; + +import java.lang.reflect.Field; +import java.util.HashMap; +import java.util.Map; + +import org.apache.activemq.artemis.api.core.QueueConfiguration; +import org.apache.activemq.artemis.api.core.RoutingType; +import org.apache.activemq.artemis.api.core.SimpleString; +import org.apache.activemq.artemis.api.core.TransportConfiguration; +import org.apache.activemq.artemis.api.core.client.ActiveMQClient; +import org.apache.activemq.artemis.api.core.client.ClientConsumer; +import org.apache.activemq.artemis.api.core.client.ClientMessage; +import org.apache.activemq.artemis.api.core.client.ClientProducer; +import org.apache.activemq.artemis.api.core.client.ClientSession; +import org.apache.activemq.artemis.api.core.client.ClientSessionFactory; +import org.apache.activemq.artemis.api.core.client.ServerLocator; +import org.apache.activemq.artemis.core.config.Configuration; +import org.apache.activemq.artemis.core.remoting.impl.netty.TransportConstants; +import org.apache.activemq.artemis.core.server.ActiveMQServer; +import org.apache.activemq.artemis.tests.util.ActiveMQTestBase; +import org.apache.activemq.artemis.utils.CheckDependencies; +import org.apache.activemq.artemis.utils.RandomUtil; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +public class IoUringTransportTest extends ActiveMQTestBase { + + private static final SimpleString QUEUE = SimpleString.of("IoUringTransportTestQueue"); + + private Map ioUringParams; + + @Override + @BeforeEach + public void setUp() throws Exception { + // skip before we build anything if the platform can't do io_uring + assumeTrue(CheckDependencies.isIoUringAvailable(), "io_uring is not available on this platform"); + super.setUp(); + + // force io_uring as the only native option so a working transfer unambiguously proves io_uring was used + ioUringParams = new HashMap<>(); + ioUringParams.put(TransportConstants.USE_IOURING_PROP_NAME, true); + ioUringParams.put(TransportConstants.USE_EPOLL_PROP_NAME, false); + ioUringParams.put(TransportConstants.USE_KQUEUE_PROP_NAME, false); + } + + @Test + public void testSendReceiveOverIoUring() throws Exception { + Configuration config = createBasicConfig().addAcceptorConfiguration(new TransportConfiguration(NETTY_ACCEPTOR_FACTORY, ioUringParams, "netty")); + ActiveMQServer server = addServer(createServer(false, config)); + server.start(); + + // the acceptor must have genuinely selected io_uring, not silently fallen back to NIO + assertUsingIoUring(server); + + TransportConfiguration connectorConfig = new TransportConfiguration(NETTY_CONNECTOR_FACTORY, ioUringParams); + ServerLocator locator = addServerLocator(ActiveMQClient.createServerLocatorWithoutHA(connectorConfig)); + ClientSessionFactory sf = createSessionFactory(locator); + ClientSession session = addClientSession(sf.createSession(false, true, true)); + + session.createQueue(QueueConfiguration.of(QUEUE).setRoutingType(RoutingType.ANYCAST).setDurable(false)); + + String text = RandomUtil.randomUUIDString(); + ClientProducer producer = session.createProducer(QUEUE); + ClientMessage message = session.createMessage(false); + message.getBodyBuffer().writeString(text); + producer.send(message); + + ClientConsumer consumer = session.createConsumer(QUEUE); + session.start(); + ClientMessage received = consumer.receive(5000); + assertNotNull(received, "expected a message over the io_uring transport"); + assertEquals(text, received.getBodyBuffer().readString()); + received.acknowledge(); + } + + private void assertUsingIoUring(ActiveMQServer server) throws Exception { + Object acceptor = server.getRemotingService().getAcceptor("netty"); + Field channelClazzField = acceptor.getClass().getDeclaredField("channelClazz"); + channelClazzField.setAccessible(true); + Class channelClazz = (Class) channelClazzField.get(acceptor); + assertEquals("io.netty.channel.uring.IoUringServerSocketChannel", channelClazz.getName(), "acceptor did not select the io_uring transport"); + } +} diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/ssl/CoreClientOverOneWaySSLTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/ssl/CoreClientOverOneWaySSLTest.java index dd382ccef965..2b3d26651bd8 100644 --- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/ssl/CoreClientOverOneWaySSLTest.java +++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/ssl/CoreClientOverOneWaySSLTest.java @@ -21,9 +21,12 @@ import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.junit.jupiter.api.Assertions.fail; +import static org.junit.jupiter.api.Assumptions.assumeTrue; import javax.net.ssl.SSLContext; import javax.net.ssl.SSLEngine; + +import io.netty.handler.ssl.OpenSsl; import java.util.Arrays; import java.util.Collection; import java.util.HashMap; @@ -500,6 +503,44 @@ public void testOneWaySSLVerifyHostNegative() throws Exception { } } + @TestTemplate + public void testOneWaySSLVerifyHostDisabledOpenSSL() throws Exception { + // Netty 4.2 enables endpoint identification (host verification) by default for client SslContexts, including the + // OpenSSL provider. This verifies that verifyHost=false actually disables it: connecting to a server whose + // certificate CN does not match the host must succeed. + assumeTrue(OpenSsl.isAvailable()); + // the "unknown" server keystore stores its key under the "unknown-server" alias, not "server", so the + // useKeystoreAlias variant can't set up a working server here (unrelated to host verification) + assumeTrue(!useKeystoreAlias); + + createCustomSslServer(true); + String text = RandomUtil.randomUUIDString(); + + tc.getParams().put(TransportConstants.SSL_ENABLED_PROP_NAME, true); + tc.getParams().put(TransportConstants.SSL_PROVIDER, TransportConstants.OPENSSL_PROVIDER); + tc.getParams().put(TransportConstants.TRUSTSTORE_PROVIDER_PROP_NAME, storeProvider); + tc.getParams().put(TransportConstants.TRUSTSTORE_TYPE_PROP_NAME, storeType); + tc.getParams().put(TransportConstants.TRUSTSTORE_PATH_PROP_NAME, CLIENT_SIDE_TRUSTSTORE); + tc.getParams().put(TransportConstants.TRUSTSTORE_PASSWORD_PROP_NAME, PASSWORD); + tc.getParams().put(TransportConstants.VERIFY_HOST_PROP_NAME, false); + + ServerLocator locator = addServerLocator(ActiveMQClient.createServerLocatorWithoutHA(tc)); + ClientSessionFactory sf = addSessionFactory(createSessionFactory(locator)); + ClientSession session = addClientSession(sf.createSession(false, true, true)); + session.createQueue(QueueConfiguration.of(CoreClientOverOneWaySSLTest.QUEUE).setDurable(false)); + ClientProducer producer = addClientProducer(session.createProducer(CoreClientOverOneWaySSLTest.QUEUE)); + + ClientMessage message = createTextMessage(session, text); + producer.send(message); + + ClientConsumer consumer = addClientConsumer(session.createConsumer(CoreClientOverOneWaySSLTest.QUEUE)); + session.start(); + + Message m = consumer.receive(1000); + assertNotNull(m); + assertEquals(text, m.getBodyBuffer().readString()); + } + @TestTemplate public void testOneWaySSLReloaded() throws Exception { createCustomSslServer(); 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 e8df611f7cc2..4e92d2908c12 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 2fff314794db..42d6e13e1677 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);