ARTEMIS-994 Support Netty Native Epoll on Linux

artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/DelegatingEventLoopGroup.java

@@ -0,0 +1,198 @@
+/*
+ * 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.remoting.impl.netty;
+
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Spliterator;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.function.Consumer;
+
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelPromise;
+import io.netty.channel.EventLoop;
+import io.netty.channel.EventLoopGroup;
+import io.netty.util.concurrent.EventExecutor;
+import io.netty.util.concurrent.Future;
+
+public class DelegatingEventLoopGroup implements EventLoopGroup {
+
+   private final EventLoopGroup delegate;
+
+   public DelegatingEventLoopGroup(EventLoopGroup eventLoopGroup) {
+      this.delegate = eventLoopGroup;
+   }
+
+   @Override
+   public EventLoop next() {
+      return delegate.next();
+   }
+
+   @Override
+   public ChannelFuture register(Channel channel) {
+      return delegate.register(channel);
+   }
+
+   @Override
+   public ChannelFuture register(ChannelPromise channelPromise) {
+      return delegate.register(channelPromise);
+   }
+
+   @Override
+   @Deprecated
+   public ChannelFuture register(Channel channel, ChannelPromise channelPromise) {
+      return delegate.register(channel, channelPromise);
+   }
+
+   @Override
+   public boolean isShuttingDown() {
+      return delegate.isShuttingDown();
+   }
+
+   @Override
+   public Future<?> shutdownGracefully() {
+      return delegate.shutdownGracefully();
+   }
+
+   @Override
+   public Future<?> shutdownGracefully(long l, long l1, TimeUnit timeUnit) {
+      return delegate.shutdownGracefully(l, l1, timeUnit);
+   }
+
+   @Override
+   public Future<?> terminationFuture() {
+      return delegate.terminationFuture();
+   }
+
+   @Override
+   @Deprecated
+   public void shutdown() {
+      delegate.shutdown();
+   }
+
+   @Override
+   @Deprecated
+   public List<Runnable> shutdownNow() {
+      return delegate.shutdownNow();
+   }
+
+   @Override
+   public Iterator<EventExecutor> iterator() {
+      return delegate.iterator();
+   }
+
+   @Override
+   public Future<?> submit(Runnable runnable) {
+      return delegate.submit(runnable);
+   }
+
+   @Override
+   public <T> Future<T> submit(Runnable runnable, T t) {
+      return delegate.submit(runnable, t);
+   }
+
+   @Override
+   public <T> Future<T> submit(Callable<T> callable) {
+      return delegate.submit(callable);
+   }
+
+   @Override
+   public io.netty.util.concurrent.ScheduledFuture<?> schedule(Runnable runnable, long l, TimeUnit timeUnit) {
+      return delegate.schedule(runnable, l, timeUnit);
+   }
+
+   @Override
+   public <V> io.netty.util.concurrent.ScheduledFuture<V> schedule(Callable<V> callable, long l, TimeUnit timeUnit) {
+      return delegate.schedule(callable, l, timeUnit);
+   }
+
+   @Override
+   public io.netty.util.concurrent.ScheduledFuture<?> scheduleAtFixedRate(Runnable runnable,
+                                                                          long l,
+                                                                          long l1,
+                                                                          TimeUnit timeUnit) {
+      return delegate.scheduleAtFixedRate(runnable, l, l1, timeUnit);
+   }
+
+   @Override
+   public io.netty.util.concurrent.ScheduledFuture<?> scheduleWithFixedDelay(Runnable runnable,
+                                                                             long l,
+                                                                             long l1,
+                                                                             TimeUnit timeUnit) {
+      return delegate.scheduleWithFixedDelay(runnable, l, l1, timeUnit);
+   }
+
+   @Override
+   public boolean isShutdown() {
+      return delegate.isShutdown();
+   }
+
+   @Override
+   public boolean isTerminated() {
+      return delegate.isTerminated();
+   }
+
+   @Override
+   public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
+      return delegate.awaitTermination(timeout, unit);
+   }
+
+   @Override
+   public <T> List<java.util.concurrent.Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) throws InterruptedException {
+      return delegate.invokeAll(tasks);
+   }
+
+   @Override
+   public <T> List<java.util.concurrent.Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
+                                                             long timeout,
+                                                             TimeUnit unit) throws InterruptedException {
+      return delegate.invokeAll(tasks, timeout, unit);
+   }
+
+   @Override
+   public <T> T invokeAny(Collection<? extends Callable<T>> tasks) throws InterruptedException, ExecutionException {
+      return delegate.invokeAny(tasks);
+   }
+
+   @Override
+   public <T> T invokeAny(Collection<? extends Callable<T>> tasks,
+                          long timeout,
+                          TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
+      return delegate.invokeAny(tasks, timeout, unit);
+   }
+
+   @Override
+   public void execute(Runnable command) {
+      delegate.execute(command);
+   }
+
+   @Override
+   public void forEach(Consumer<? super EventExecutor> action) {
+      delegate.forEach(action);
+   }
+
+   @Override
+   public Spliterator<EventExecutor> spliterator() {
+      return delegate.spliterator();
+   }
+
+}

artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/NettyConnection.java

@@ -215,7 +215,7 @@ public ActiveMQBuffer createTransportBuffer(final int size) {
 
    @Override
    public ActiveMQBuffer createTransportBuffer(final int size, boolean pooled) {
-      return new ChannelBufferWrapper(PartialPooledByteBufAllocator.INSTANCE.directBuffer(size), true);
+      return new ChannelBufferWrapper(channel.alloc().directBuffer(size), true);
    }
 
    @Override

artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/NettyConnector.java

@@ -58,6 +58,9 @@
 import io.netty.channel.ChannelPromise;
 import io.netty.channel.EventLoopGroup;
 import io.netty.channel.SimpleChannelInboundHandler;
+import io.netty.channel.epoll.Epoll;
+import io.netty.channel.epoll.EpollEventLoopGroup;
+import io.netty.channel.epoll.EpollSocketChannel;
 import io.netty.channel.group.ChannelGroup;
 import io.netty.channel.group.DefaultChannelGroup;
 import io.netty.channel.nio.NioEventLoopGroup;
@@ -218,6 +221,12 @@ public class NettyConnector extends AbstractConnector {
 
    private boolean useNioGlobalWorkerPool;
 
+   private boolean useEpoll;
+
+   private int epollRemotingThreads;
+
+   private boolean useEpollGlobalWorkerPool;
+
    private ScheduledExecutorService scheduledThreadPool;
 
    private Executor closeExecutor;
@@ -288,6 +297,13 @@ public NettyConnector(final Map<String, Object> configuration,
 
       useNioGlobalWorkerPool = ConfigurationHelper.getBooleanProperty(TransportConstants.USE_NIO_GLOBAL_WORKER_POOL_PROP_NAME, TransportConstants.DEFAULT_USE_NIO_GLOBAL_WORKER_POOL, configuration);
 
+      useEpoll = ConfigurationHelper.getBooleanProperty(TransportConstants.USE_EPOLL_PROP_NAME, TransportConstants.DEFAULT_USE_EPOLL, configuration);
+
+      epollRemotingThreads = ConfigurationHelper.getIntProperty(TransportConstants.EPOLL_REMOTING_THREADS_PROPNAME, -1, configuration);
+
+      useEpollGlobalWorkerPool = ConfigurationHelper.getBooleanProperty(TransportConstants.USE_EPOLL_GLOBAL_WORKER_POOL_PROP_NAME, TransportConstants.DEFAULT_USE_EPOLL_GLOBAL_WORKER_POOL, 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);
       port = ConfigurationHelper.getIntProperty(TransportConstants.PORT_PROP_NAME, TransportConstants.DEFAULT_PORT, configuration);
@@ -371,22 +387,46 @@ public synchronized void start() {
          return;
       }
 
-      int threadsToUse;
+      // Default to number of cores * 3
+      int defaultThreadsToUse = Runtime.getRuntime().availableProcessors() * 3;
 
-      if (nioRemotingThreads == -1) {
-         // Default to number of cores * 3
+      if (useEpoll) {
+         if (Epoll.isAvailable()) {
+            int epollThreadsToUse;
+            if (epollRemotingThreads == -1) {
+               epollThreadsToUse = defaultThreadsToUse;
+            } else {
+               epollThreadsToUse = this.epollRemotingThreads;
+            }
+            if (useEpollGlobalWorkerPool) {
+               channelClazz = EpollSocketChannel.class;
+               group = SharedEventLoopGroup.getInstance((threadFactory -> new EpollEventLoopGroup(epollThreadsToUse, threadFactory)));
+            } else {
+               channelClazz = EpollSocketChannel.class;
+               group = new EpollEventLoopGroup(epollThreadsToUse);
+            }
+            logger.info("Connector using native epoll");
 
-         threadsToUse = Runtime.getRuntime().availableProcessors() * 3;
-      } else {
-         threadsToUse = this.nioRemotingThreads;
+         } else {
+            logger.warn("Connector unable to load native epoll, will continue and load nio");
+         }
       }
 
-      if (useNioGlobalWorkerPool) {
-         channelClazz = NioSocketChannel.class;
-         group = SharedNioEventLoopGroup.getInstance(threadsToUse);
-      } else {
-         channelClazz = NioSocketChannel.class;
-         group = new NioEventLoopGroup(threadsToUse);
+      if (channelClazz == null || group == null) {
+         int nioThreadsToUse;
+         if (nioRemotingThreads == -1) {
+            nioThreadsToUse = defaultThreadsToUse;
+         } else {
+            nioThreadsToUse = this.nioRemotingThreads;
+         }
+         if (useNioGlobalWorkerPool) {
+            channelClazz = NioSocketChannel.class;
+            group = SharedEventLoopGroup.getInstance((threadFactory -> new NioEventLoopGroup(nioThreadsToUse, threadFactory)));
+         } else {
+            channelClazz = NioSocketChannel.class;
+            group = new NioEventLoopGroup(nioThreadsToUse);
+         }
+         logger.info("Connector using nio");
       }
       // if we are a servlet wrap the socketChannelFactory
 
@@ -407,7 +447,6 @@ public synchronized void start() {
       }
       bootstrap.option(ChannelOption.SO_KEEPALIVE, true);
       bootstrap.option(ChannelOption.SO_REUSEADDR, true);
-      bootstrap.option(ChannelOption.ALLOCATOR, PartialPooledByteBufAllocator.INSTANCE);
       channelGroup = new DefaultChannelGroup("activemq-connector", GlobalEventExecutor.INSTANCE);
 
       final SSLContext context;
@@ -1054,7 +1093,7 @@ public Bootstrap getBootStrap() {
    }
 
    public static void clearThreadPools() {
-      SharedNioEventLoopGroup.forceShutdown();
+      SharedEventLoopGroup.forceShutdown();
    }
 
    private static ClassLoader getThisClassLoader() {