From 29ef00f24ec4efed977890243ff504f026a38104 Mon Sep 17 00:00:00 2001
From: Andreas Mager <andreas@mager.eu>
Date: Thu, 18 May 2017 15:38:39 +0200
Subject: [PATCH] broker/NettyAcceptor: epoll added

---
 ChangeLog.txt                                 |  1 +
 broker/pom.xml                                | 10 +++++++++
 .../java/io/moquette/BrokerConstants.java     |  1 +
 .../moquette/server/netty/NettyAcceptor.java  | 22 ++++++++++++++++---
 distribution/src/main/resources/moquette.conf | 12 +++++++++-
 5 files changed, 42 insertions(+), 4 deletions(-)

diff --git a/ChangeLog.txt b/ChangeLog.txt
index 3b9acaa0..a4b5f9b5 100644
--- a/ChangeLog.txt
+++ b/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)
     
diff --git a/broker/pom.xml b/broker/pom.xml
index a0a256a6..3a67af93 100644
--- a/broker/pom.xml
+++ b/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>
diff --git a/broker/src/main/java/io/moquette/BrokerConstants.java b/broker/src/main/java/io/moquette/BrokerConstants.java
index 47edb890..220dd4a8 100644
--- a/broker/src/main/java/io/moquette/BrokerConstants.java
+++ b/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() {
diff --git a/broker/src/main/java/io/moquette/server/netty/NettyAcceptor.java b/broker/src/main/java/io/moquette/server/netty/NettyAcceptor.java
index c438c1b7..71bf48dc 100644
--- a/broker/src/main/java/io/moquette/server/netty/NettyAcceptor.java
+++ b/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
diff --git a/distribution/src/main/resources/moquette.conf b/distribution/src/main/resources/moquette.conf
index b1eaea90..c3fd7b9c 100644
--- a/distribution/src/main/resources/moquette.conf
+++ b/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
\ No newline at end of file
+# 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