Add --native-epoll option to performance tool

Fixes #270

Author: acogoluegnes

pom.xml

@@ -639,6 +639,13 @@
           <version>${amqp-client.version}</version>
         </dependency>
 
+        <dependency>
+          <groupId>io.netty</groupId>
+          <artifactId>netty-transport-native-epoll</artifactId>
+          <version>${netty.version}</version>
+          <classifier>linux-x86_64</classifier>
+        </dependency>
+
       </dependencies>
       <build>
         <finalName>${finalName}</finalName>

src/main/java/com/rabbitmq/stream/perf/StreamPerfTest.java

@@ -54,10 +54,13 @@
 import io.micrometer.core.instrument.Tag;
 import io.micrometer.core.instrument.Tags;
 import io.micrometer.core.instrument.composite.CompositeMeterRegistry;
+import io.netty.bootstrap.Bootstrap;
 import io.netty.buffer.ByteBufAllocator;
 import io.netty.buffer.ByteBufAllocatorMetric;
 import io.netty.buffer.ByteBufAllocatorMetricProvider;
 import io.netty.channel.EventLoopGroup;
+import io.netty.channel.epoll.EpollEventLoopGroup;
+import io.netty.channel.epoll.EpollSocketChannel;
 import io.netty.channel.nio.NioEventLoopGroup;
 import io.netty.handler.ssl.SslContextBuilder;
 import io.netty.handler.ssl.SslHandler;
@@ -441,6 +444,12 @@ public void setMaxSegmentSize(ByteCapacity in) {
       converter = Utils.ByteCapacityTypeConverter.class)
   private ByteCapacity requestedMaxFrameSize;
 
+  @CommandLine.Option(
+      names = {"--native-epoll", "-ne"},
+      description = "use Netty's native epoll transport (Linux x86-64 only)",
+      defaultValue = "false")
+  private boolean nativeEpoll;
+
   private MetricsCollector metricsCollector;
   private PerformanceMetrics performanceMetrics;
   private List<Monitoring> monitorings;
@@ -664,7 +673,14 @@ public Integer call() throws Exception {
         }
       }
 
-      this.eventLoopGroup = new NioEventLoopGroup();
+      java.util.function.Consumer<Bootstrap> bootstrapCustomizer;
+      if (this.nativeEpoll) {
+        this.eventLoopGroup = new EpollEventLoopGroup();
+        bootstrapCustomizer = b -> b.channel(EpollSocketChannel.class);
+      } else {
+        this.eventLoopGroup = new NioEventLoopGroup();
+        bootstrapCustomizer = b -> {};
+      }
 
       EnvironmentBuilder environmentBuilder =
           Environment.builder()
@@ -676,6 +692,7 @@ public Integer call() throws Exception {
               .netty()
               .byteBufAllocator(byteBufAllocator)
               .eventLoopGroup(eventLoopGroup)
+              .bootstrapCustomizer(bootstrapCustomizer)
               .environmentBuilder()
               .codec(codec)
               .maxProducersByConnection(this.producersByConnection)

src/test/java/com/rabbitmq/stream/perf/StreamPerfTestTest.java

@@ -60,6 +60,9 @@
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.TestInfo;
+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.api.extension.ExtendWith;
 
 @ExtendWith(TestUtils.StreamTestInfrastructureExtension.class)
@@ -496,6 +499,18 @@ void singleActiveConsumersOnSuperStream() throws Exception {
                     .isEqualTo(Constants.RESPONSE_CODE_STREAM_DOES_NOT_EXIST));
   }
 
+  @Test
+  @EnabledOnOs(OS.LINUX)
+  @EnabledIfSystemProperty(named = "os.arch", matches = "amd64")
+  void nativeEpollWorksOnLinux() throws Exception {
+    Future<?> run = run(builder().nativeEpoll());
+    waitUntilStreamExists(s);
+    waitOneSecond();
+    run.cancel(true);
+    waitRunEnds();
+    assertThat(streamExists(s)).isTrue();
+  }
+
   private static <T> Consumer<T> wrap(CallableConsumer<T> action) {
     return t -> {
       try {
@@ -708,6 +723,11 @@ ArgumentsBuilder time(int time) {
       return this;
     }
 
+    ArgumentsBuilder nativeEpoll() {
+      arguments.put("native-epoll", "");
+      return this;
+    }
+
     String build() {
       return this.arguments.entrySet().stream()
           .map(e -> "--" + e.getKey() + (e.getValue().isEmpty() ? "" : (" " + e.getValue())))