broker/NettyAcceptor: epoll added

ChangeLog.txt

@@ -1,4 +1,5 @@
 Version 0.10:
+    Netty epoll system added
     Separated in memory and MapDB storage implementations
     Cleanly separated inflight part (inbound and outbound)
     

broker/pom.xml

@@ -73,6 +73,16 @@
             <version>${netty.version}</version>
         </dependency>
 
+        <dependency>
+            <groupId>io.netty</groupId>
+            <artifactId>netty-transport-native-epoll</artifactId>
+            <version>${netty.version}</version>
+            <classifier>linux-x86_64</classifier>
+            <!-- 
+            <classifier>${os.detected.name}-${os.detected.arch}</classifier>
+             -->
+        </dependency>
+
         <dependency>
             <groupId>commons-codec</groupId>
             <artifactId>commons-codec</artifactId>

broker/src/main/java/io/moquette/BrokerConstants.java

@@ -56,6 +56,7 @@ public final class BrokerConstants {
     public static final String NETTY_TCP_NODELAY_PROPERTY_NAME = "netty.tcp_nodelay";
     public static final String NETTY_SO_KEEPALIVE_PROPERTY_NAME = "netty.so_keepalive";
     public static final String NETTY_CHANNEL_TIMEOUT_SECONDS_PROPERTY_NAME = "netty.channel_timeout.seconds";
+    public static final String NETTY_EPOLL_PROPERTY_NAME = "netty.epoll";
     public static final String STORAGE_CLASS_NAME = "storage_class";
 
     private BrokerConstants() {

broker/src/main/java/io/moquette/server/netty/NettyAcceptor.java

@@ -25,7 +25,10 @@ import io.moquette.spi.security.ISslContextCreator;
 import io.netty.bootstrap.ServerBootstrap;
 import io.netty.buffer.ByteBuf;
 import io.netty.channel.*;
+import io.netty.channel.epoll.EpollEventLoopGroup;
+import io.netty.channel.epoll.EpollServerSocketChannel;
 import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.ServerSocketChannel;
 import io.netty.channel.socket.SocketChannel;
 import io.netty.channel.socket.nio.NioServerSocketChannel;
 import io.netty.handler.codec.MessageToMessageDecoder;
@@ -99,6 +102,8 @@ public class NettyAcceptor implements ServerAcceptor {
     private boolean nettySoKeepalive;
     private int nettyChannelTimeoutSeconds;
 
+    private Class<? extends ServerSocketChannel> channelClass;
+
     @Override
     public void initialize(ProtocolProcessor processor, IConfig props, ISslContextCreator sslCtxCreator)
             throws IOException {
@@ -114,8 +119,19 @@ public class NettyAcceptor implements ServerAcceptor {
         nettyChannelTimeoutSeconds = Integer
                 .parseInt(props.getProperty(BrokerConstants.NETTY_CHANNEL_TIMEOUT_SECONDS_PROPERTY_NAME, "10"));
 
-        m_bossGroup = new NioEventLoopGroup();
-        m_workerGroup = new NioEventLoopGroup();
+        boolean epoll = Boolean.parseBoolean(props.getProperty(BrokerConstants.NETTY_EPOLL_PROPERTY_NAME, "false"));
+        if (epoll) {
+            LOG.info("Netty is using epoll.");
+            m_bossGroup = new EpollEventLoopGroup();
+            m_workerGroup = new EpollEventLoopGroup();
+            channelClass = EpollServerSocketChannel.class;
+        } else {
+            LOG.info("Netty is using nio.");
+            m_bossGroup = new NioEventLoopGroup();
+            m_workerGroup = new NioEventLoopGroup();
+            channelClass = NioServerSocketChannel.class;
+        }
+
         final NettyMQTTHandler handler = new NettyMQTTHandler(processor);
 
         initializePlainTCPTransport(handler, props);
@@ -136,7 +152,7 @@ public class NettyAcceptor implements ServerAcceptor {
     private void initFactory(String host, int port, String protocol, final PipelineInitializer pipeliner) {
         LOG.info("Initializing server. Protocol={}", protocol);
         ServerBootstrap b = new ServerBootstrap();
-        b.group(m_bossGroup, m_workerGroup).channel(NioServerSocketChannel.class)
+        b.group(m_bossGroup, m_workerGroup).channel(channelClass)
                 .childHandler(new ChannelInitializer<SocketChannel>() {
 
                     @Override

distribution/src/main/resources/moquette.conf

@@ -99,4 +99,14 @@ password_file config/password_file.conf
 #       interval between flushes of MapDB storage to disk. It's in
 #       seconds, if not specified defaults is 30 s.
 #*********************************************************************
-# autosave_interval 120
+# autosave_interval 120
+
+#*********************************************************************
+# Netty Configuration
+#*********************************************************************
+#
+# Linux systems can use epoll instead of nio. To get a performance 
+# gain and reduced GC.
+# http://netty.io/wiki/native-transports.html for more information
+#
+# netty.epoll=true