Add --netty-kqueue option

Author: acogoluegnes

pom.xml

@@ -176,6 +176,12 @@
       <version>${netty.version}</version>
       <classifier>linux-x86_64</classifier>
     </dependency>
+    <dependency>
+      <groupId>io.netty</groupId>
+      <artifactId>netty-transport-native-kqueue</artifactId>
+      <version>${netty.version}</version>
+      <classifier>osx-aarch_64</classifier>
+    </dependency>
 
     <dependency>
       <groupId>org.junit.jupiter</groupId>

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

@@ -17,6 +17,7 @@
 
 import static com.rabbitmq.client.ConnectionFactory.computeDefaultTlsProtocol;
 import static com.rabbitmq.perf.OptionsUtils.forEach;
+import static com.rabbitmq.perf.Tuples.pair;
 import static com.rabbitmq.perf.Utils.strArg;
 import static java.lang.String.format;
 import static java.util.Arrays.asList;
@@ -33,6 +34,7 @@
 import com.rabbitmq.client.impl.DefaultCredentialsRefreshService.DefaultCredentialsRefreshServiceBuilder;
 import com.rabbitmq.client.impl.DefaultExceptionHandler;
 import com.rabbitmq.perf.Metrics.ConfigurationContext;
+import com.rabbitmq.perf.Tuples.Pair;
 import com.rabbitmq.perf.Utils.GsonOAuth2ClientCredentialsGrantCredentialsProvider;
 import com.rabbitmq.perf.metrics.CompositeMetricsFormatter;
 import com.rabbitmq.perf.metrics.CsvMetricsFormatter;
@@ -42,12 +44,18 @@
 import com.rabbitmq.perf.metrics.MetricsFormatterFactory.Context;
 import io.micrometer.core.instrument.composite.CompositeMeterRegistry;
 import io.netty.bootstrap.Bootstrap;
+import io.netty.channel.Channel;
 import io.netty.channel.EventLoopGroup;
 import io.netty.channel.IoHandlerFactory;
 import io.netty.channel.MultiThreadIoEventLoopGroup;
+import io.netty.channel.epoll.Epoll;
 import io.netty.channel.epoll.EpollIoHandler;
 import io.netty.channel.epoll.EpollSocketChannel;
+import io.netty.channel.kqueue.KQueue;
+import io.netty.channel.kqueue.KQueueIoHandler;
+import io.netty.channel.kqueue.KQueueSocketChannel;
 import io.netty.channel.nio.NioIoHandler;
+import io.netty.channel.socket.nio.NioSocketChannel;
 import java.io.*;
 import java.math.BigDecimal;
 import java.nio.charset.Charset;
@@ -63,6 +71,7 @@
 import java.util.function.BooleanSupplier;
 import java.util.function.Consumer;
 import java.util.function.Function;
+import java.util.function.Supplier;
 import java.util.stream.Collectors;
 import javax.net.ssl.SSLContext;
 import javax.net.ssl.TrustManager;
@@ -860,10 +869,10 @@ private static void configureNettyIfRequested(
     if (netty(cmd)) {
       int nbThreads = Utils.intArg(cmd, "ntyt", -1);
       boolean epoll = hasOption(cmd, "ntyep");
-      IoHandlerFactory ioHandlerFactory =
-          epoll ? EpollIoHandler.newFactory() : NioIoHandler.newFactory();
-      Consumer<Bootstrap> bootstrapCustomizer =
-          epoll ? b -> b.channel(EpollSocketChannel.class) : b -> {};
+      boolean kqueue = hasOption(cmd, "ntykq");
+      Pair<IoHandlerFactory, Class<? extends Channel>> io = nettyIo(epoll, kqueue);
+      IoHandlerFactory ioHandlerFactory = io.v1();
+      Consumer<Bootstrap> bootstrapCustomizer = b -> b.channel(io.v2());
       EventLoopGroup eventLoopGroup =
           nbThreads > 0
               ? new MultiThreadIoEventLoopGroup(nbThreads, ioHandlerFactory)
@@ -873,11 +882,32 @@ private static void configureNettyIfRequested(
     }
   }
 
+  private static Pair<IoHandlerFactory, Class<? extends Channel>> nettyIo(
+      boolean epoll, boolean kqueue) {
+    Supplier<Pair<IoHandlerFactory, Class<? extends Channel>>> result =
+        () -> pair(NioIoHandler.newFactory(), NioSocketChannel.class);
+    if (epoll) {
+      if (Epoll.isAvailable()) {
+        result = () -> pair(EpollIoHandler.newFactory(), EpollSocketChannel.class);
+      } else {
+        LOGGER.warn("epoll not available, using Java NIO instead");
+      }
+    } else if (kqueue) {
+      if (KQueue.isAvailable()) {
+        result = () -> pair(KQueueIoHandler.newFactory(), KQueueSocketChannel.class);
+      } else {
+        LOGGER.warn("kqueue not available, using Java NIO instead");
+      }
+    }
+    return result.get();
+  }
+
   private static boolean netty(CommandLineProxy cmd) {
     boolean netty = hasOption(cmd, "nty");
     int nbThreads = Utils.intArg(cmd, "ntyt", -1);
     boolean epoll = hasOption(cmd, "ntyep");
-    return netty || nbThreads > 0 || epoll;
+    boolean kqueue = hasOption(cmd, "ntykq");
+    return netty || nbThreads > 0 || epoll || kqueue;
   }
 
   static MulticastSet.CompletionHandler getCompletionHandler(
@@ -1460,7 +1490,12 @@ static Options getOptions() {
             "netty-epoll",
             false,
             "use Netty's native epoll transport (Linux x86-64 only)"));
-
+    options.addOption(
+        new Option(
+            "ntykq",
+            "netty-kqueue",
+            false,
+            "use Netty's native kqueue transport (macOS aarch64 only)"));
     return options;
   }
 

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

@@ -0,0 +1,44 @@
+// Copyright (c) 2024 Broadcom. All Rights Reserved.
+// The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries.
+//
+// This software, the RabbitMQ Java client library, is triple-licensed under the
+// Mozilla Public License 2.0 ("MPL"), the GNU General Public License version 2
+// ("GPL") and the Apache License version 2 ("ASL"). For the MPL, please see
+// LICENSE-MPL-RabbitMQ. For the GPL, please see LICENSE-GPL2.  For the ASL,
+// please see LICENSE-APACHE2.
+//
+// This software is distributed on an "AS IS" basis, WITHOUT WARRANTY OF ANY KIND,
+// either express or implied. See the LICENSE file for specific language governing
+// rights and limitations of this software.
+//
+// If you have any questions regarding licensing, please contact us at
+// [email protected].
+package com.rabbitmq.perf;
+
+final class Tuples {
+
+  private Tuples() {}
+
+  public static <A, B> Pair<A, B> pair(A v1, B v2) {
+    return new Pair<>(v1, v2);
+  }
+
+  public static class Pair<A, B> {
+
+    private final A v1;
+    private final B v2;
+
+    private Pair(A v1, B v2) {
+      this.v1 = v1;
+      this.v2 = v2;
+    }
+
+    public A v1() {
+      return this.v1;
+    }
+
+    public B v2() {
+      return this.v2;
+    }
+  }
+}

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

@@ -118,6 +118,16 @@ void nativeEpollWorksOnLinux() throws Exception {
     assertThat(consoleOutput()).contains("starting", "stopped", "avg");
   }
 
+  @Test
+  @EnabledOnOs(OS.MAC)
+  @EnabledIfSystemProperty(named = "os.arch", matches = "aarch64")
+  void nativeKqueueWorksOnMacOs() throws Exception {
+    int messageCount = 1000;
+    run(builder().pmessages(messageCount).cmessages(messageCount).nettyThreads(1).nettyKqueue());
+    waitRunEnds();
+    assertThat(consoleOutput()).contains("starting", "stopped", "avg");
+  }
+
   private static void waitOneSecond() throws InterruptedException {
     wait(Duration.ofSeconds(1));
   }
@@ -178,10 +188,14 @@ ArgumentsBuilder useDefaultSslContext() {
       return argument("use-default-ssl-context", "true");
     }
 
-    public ArgumentsBuilder nettyEpoll() {
+    ArgumentsBuilder nettyEpoll() {
       return argument("netty-epoll", "true");
     }
 
+    ArgumentsBuilder nettyKqueue() {
+      return argument("netty-kqueue", "true");
+    }
+
     ArgumentsBuilder rate(int rate) {
       return intArgument("rate", rate);
     }