Add --netty-epoll option

Author: acogoluegnes

pom.xml

@@ -62,6 +62,7 @@
     <micrometer.version>1.15.2</micrometer.version>
     <jgroups.version>5.4.8.Final</jgroups.version>
     <jgroups-kubernetes.version>2.0.2.Final</jgroups-kubernetes.version>
+    <netty.version>4.2.3.Final</netty.version>
     <gson.version>2.13.1</gson.version>
     <resilience4j.version>2.1.0</resilience4j.version>
     <logback.version>1.3.15</logback.version>
@@ -169,6 +170,12 @@
       <artifactId>jgroups-kubernetes</artifactId>
       <version>${jgroups-kubernetes.version}</version>
     </dependency>
+    <dependency>
+      <groupId>io.netty</groupId>
+      <artifactId>netty-transport-native-epoll</artifactId>
+      <version>${netty.version}</version>
+      <classifier>linux-x86_64</classifier>
+    </dependency>
 
     <dependency>
       <groupId>org.junit.jupiter</groupId>

src/main/java/com/rabbitmq/perf/PerfTest.java

@@ -42,7 +42,9 @@
 import com.rabbitmq.perf.metrics.MetricsFormatterFactory.Context;
 import io.micrometer.core.instrument.composite.CompositeMeterRegistry;
 import io.netty.channel.EventLoopGroup;
+import io.netty.channel.IoHandlerFactory;
 import io.netty.channel.MultiThreadIoEventLoopGroup;
+import io.netty.channel.epoll.EpollIoHandler;
 import io.netty.channel.nio.NioIoHandler;
 import java.io.*;
 import java.math.BigDecimal;
@@ -855,10 +857,13 @@ private static ConnectionFactory configureNettyIfRequested(
       CommandLineProxy cmd, ConnectionFactory factory, ShutdownService shutdownService) {
     if (netty(cmd)) {
       int nbThreads = Utils.intArg(cmd, "ntyt", -1);
+      boolean epoll = hasOption(cmd, "ntyep");
+      IoHandlerFactory ioHandlerFactory =
+          epoll ? EpollIoHandler.newFactory() : NioIoHandler.newFactory();
       EventLoopGroup eventLoopGroup =
           nbThreads > 0
-              ? new MultiThreadIoEventLoopGroup(nbThreads, NioIoHandler.newFactory())
-              : new MultiThreadIoEventLoopGroup(NioIoHandler.newFactory());
+              ? new MultiThreadIoEventLoopGroup(nbThreads, ioHandlerFactory)
+              : new MultiThreadIoEventLoopGroup(ioHandlerFactory);
       shutdownService.wrap(() -> eventLoopGroup.shutdownGracefully(0, 0, TimeUnit.SECONDS));
       factory.netty().eventLoopGroup(eventLoopGroup);
     }
@@ -868,7 +873,8 @@ private static ConnectionFactory configureNettyIfRequested(
   private static boolean netty(CommandLineProxy cmd) {
     boolean netty = hasOption(cmd, "nty");
     int nbThreads = Utils.intArg(cmd, "ntyt", -1);
-    return netty || nbThreads > 0;
+    boolean epoll = hasOption(cmd, "ntyep");
+    return netty || nbThreads > 0 || epoll;
   }
 
   static MulticastSet.CompletionHandler getCompletionHandler(
@@ -1445,6 +1451,12 @@ static Options getOptions() {
             "netty-threads",
             true,
             "number of Netty threads to use (default is Netty's default)"));
+    options.addOption(
+        new Option(
+            "ntyep",
+            "netty-epoll",
+            false,
+            "use Netty's native epoll transport (Linux x86-64 only)"));
 
     return options;
   }

src/test/java/com/rabbitmq/perf/it/PerfTestIT.java

@@ -33,6 +33,9 @@
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.EnabledIfSystemProperty;
+import org.junit.jupiter.api.condition.EnabledOnOs;
+import org.junit.jupiter.api.condition.OS;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.EnumSource;
 
@@ -105,6 +108,16 @@ void tlsWithDefaultSslContextShouldFail(IoLayer ioLayer) throws Exception {
     assertThat(consoleOutput()).contains("test stopped");
   }
 
+  @Test
+  @EnabledOnOs(OS.LINUX)
+  @EnabledIfSystemProperty(named = "os.arch", matches = "amd64")
+  void nativeEpollWorksOnLinux() throws Exception {
+    int messageCount = 1000;
+    run(builder().pmessages(messageCount).cmessages(messageCount).nettyThreads(1).nettyEpoll());
+    waitRunEnds();
+    assertThat(consoleOutput()).contains("starting", "stopped", "avg");
+  }
+
   private static void waitOneSecond() throws InterruptedException {
     wait(Duration.ofSeconds(1));
   }
@@ -154,24 +167,30 @@ ArgumentsBuilder nettyThreads(int threads) {
     }
 
     ArgumentsBuilder uri(String uri) {
-      this.arguments.put("uri", uri);
-      return this;
+      return this.argument("uri", uri);
     }
 
     ArgumentsBuilder sni(String sni) {
-      this.arguments.put("server-name-indication", sni);
-      return this;
+      return argument("server-name-indication", sni);
     }
 
     ArgumentsBuilder useDefaultSslContext() {
-      this.arguments.put("use-default-ssl-context", "true");
-      return this;
+      return argument("use-default-ssl-context", "true");
+    }
+
+    public ArgumentsBuilder nettyEpoll() {
+      return argument("netty-epoll", "true");
     }
 
     ArgumentsBuilder rate(int rate) {
       return intArgument("rate", rate);
     }
 
+    private ArgumentsBuilder argument(String key, String value) {
+      this.arguments.put(key, value);
+      return this;
+    }
+
     private ArgumentsBuilder intArgument(String key, int value) {
       this.arguments.put(key, String.valueOf(value));
       return this;