Support netty io_uring

bom/pom.xml

@@ -62,6 +62,8 @@
         <module.deploy.skip>true</module.deploy.skip>
         <!-- Fabric8 for Kubernetes -->
         <fabric8_kubernetes_version>6.9.2</fabric8_kubernetes_version>
+        <!-- io_uring -->
+        <netty-iouring.version>0.0.21.Final</netty-iouring.version>
     </properties>
 
     <dependencyManagement>
@@ -613,6 +615,25 @@
                 <version>0.16.0</version>
                 <scope>test</scope>
             </dependency>
+
+            <!-- io_uring -->
+            <dependency>
+                <groupId>io.netty.incubator</groupId>
+                <artifactId>netty-incubator-transport-classes-io_uring</artifactId>
+                <version>${netty-iouring.version}</version>
+            </dependency>
+            <dependency>
+                <groupId>io.netty.incubator</groupId>
+                <artifactId>netty-incubator-transport-native-io_uring</artifactId>
+                <version>${netty-iouring.version}</version>
+                <classifier>linux-x86_64</classifier>
+            </dependency>
+            <dependency>
+                <groupId>io.netty.incubator</groupId>
+                <artifactId>netty-incubator-transport-native-io_uring</artifactId>
+                <version>${netty-iouring.version}</version>
+                <classifier>linux-aarch_64</classifier>
+            </dependency>
         </dependencies>
     </dependencyManagement>
 

core/api/src/main/java/com/alipay/sofa/rpc/common/RpcOptions.java

@@ -396,6 +396,12 @@ public class RpcOptions {
      * 默认开启epoll？
      */
     public static final String TRANSPORT_USE_EPOLL                      = "transport.use.epoll";
+
+    /**
+     * 是否开始io_uring
+     */
+    public static final String TRANSPORT_USE_IO_URING                   = "transport.use.ioUring";
+
     /**
      * 默认服务端 数据包限制
      */

remoting/remoting-http/pom.xml

@@ -46,6 +46,22 @@
           <artifactId>netty-tcnative-boringssl-static</artifactId>
           <classifier>${os.detected.classifier}</classifier>
         </dependency>
+
+        <!-- io_uring -->
+        <dependency>
+            <groupId>io.netty.incubator</groupId>
+            <artifactId>netty-incubator-transport-classes-io_uring</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.netty.incubator</groupId>
+            <artifactId>netty-incubator-transport-native-io_uring</artifactId>
+            <classifier>linux-x86_64</classifier>
+        </dependency>
+        <dependency>
+            <groupId>io.netty.incubator</groupId>
+            <artifactId>netty-incubator-transport-native-io_uring</artifactId>
+            <classifier>linux-aarch_64</classifier>
+        </dependency>
     </dependencies>
 
     <build>

remoting/remoting-http/src/main/java/com/alipay/sofa/rpc/transport/http/AbstractHttp2ClientTransport.java

@@ -51,8 +51,6 @@
 import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelOption;
 import io.netty.channel.EventLoopGroup;
-import io.netty.channel.epoll.EpollSocketChannel;
-import io.netty.channel.socket.nio.NioSocketChannel;
 import io.netty.handler.codec.http.DefaultFullHttpRequest;
 import io.netty.handler.codec.http.FullHttpRequest;
 import io.netty.handler.codec.http.HttpHeaderNames;
@@ -145,7 +143,7 @@ public void connect() {
             int port = providerInfo.getPort();
             Bootstrap b = new Bootstrap();
             b.group(workerGroup);
-            b.channel(transportConfig.isUseEpoll() ? EpollSocketChannel.class : NioSocketChannel.class);
+            b.channel(NettyHelper.socketChannel());
             b.option(ChannelOption.SO_KEEPALIVE, true);
             b.remoteAddress(host, port);
             b.handler(initializer);

remoting/remoting-http/src/main/java/com/alipay/sofa/rpc/transport/netty/NettyHelper.java

@@ -29,7 +29,13 @@
 import io.netty.channel.ChannelHandlerContext;
 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.channel.socket.SocketChannel;
+import io.netty.channel.socket.nio.NioSocketChannel;
+import io.netty.incubator.channel.uring.IOUring;
+import io.netty.incubator.channel.uring.IOUringEventLoopGroup;
+import io.netty.incubator.channel.uring.IOUringSocketChannel;
 import io.netty.util.Attribute;
 import io.netty.util.AttributeKey;
 
@@ -42,6 +48,7 @@
 import static com.alipay.sofa.rpc.common.RpcConfigs.getIntValue;
 import static com.alipay.sofa.rpc.common.RpcOptions.TRANSPORT_CLIENT_IO_THREADS;
 import static com.alipay.sofa.rpc.common.RpcOptions.TRANSPORT_USE_EPOLL;
+import static com.alipay.sofa.rpc.common.RpcOptions.TRANSPORT_USE_IO_URING;
 
 /**
  * @author <a href="mailto:[email protected]">GengZhang</a>
@@ -188,16 +195,38 @@ public synchronized static EventLoopGroup getClientIOEventLoopGroup() {
                 clientIoThreads : // 用户配置
                 Math.max(4, SystemInfo.getCpuCores() + 1); // 默认cpu+1,至少4个
             NamedThreadFactory threadName = new NamedThreadFactory("CLI-IO", true);
-            boolean useEpoll = getBooleanValue(TRANSPORT_USE_EPOLL);
-            clientIOEventLoopGroup = useEpoll ? new EpollEventLoopGroup(threads, threadName)
-                : new NioEventLoopGroup(threads, threadName);
+            clientIOEventLoopGroup = eventLoopGroup(threads, threadName);
             refCounter.putIfAbsent(clientIOEventLoopGroup, new AtomicInteger(0));
             // SelectStrategyFactory 未设置
         }
         refCounter.get(clientIOEventLoopGroup).incrementAndGet();
         return clientIOEventLoopGroup;
     }
 
+    public synchronized static EventLoopGroup eventLoopGroup(int threads, NamedThreadFactory threadName) {
+        boolean useEpoll = getBooleanValue(TRANSPORT_USE_EPOLL);
+        boolean useIoUring = getBooleanValue(TRANSPORT_USE_IO_URING);
+        if (useEpoll) {
+            return new EpollEventLoopGroup(threads, threadName);
+        } else if (useIoUring && SystemInfo.isLinux() && IOUring.isAvailable()) {
+            return new IOUringEventLoopGroup(threads, threadName);
+        } else {
+            return new NioEventLoopGroup(threads, threadName);
+        }
+    }
+
+    public synchronized static Class<? extends SocketChannel> socketChannel() {
+        boolean useEpoll = getBooleanValue(TRANSPORT_USE_EPOLL);
+        boolean useIoUring = getBooleanValue(TRANSPORT_USE_IO_URING);
+        if (useEpoll) {
+            return EpollSocketChannel.class;
+        } else if (useIoUring && SystemInfo.isLinux() && IOUring.isAvailable()) {
+            return IOUringSocketChannel.class;
+        } else {
+            return NioSocketChannel.class;
+        }
+    }
+
     /**
      * 关闭客户端IO线程池
      */

remoting/remoting-http/src/test/java/com/alipay/sofa/rpc/transport/netty/NettyHelperTest.java

@@ -0,0 +1,53 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.alipay.sofa.rpc.transport.netty;
+
+import com.alipay.sofa.rpc.common.struct.NamedThreadFactory;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.channel.socket.nio.NioSocketChannel;
+import io.netty.incubator.channel.uring.IOUring;
+import io.netty.incubator.channel.uring.IOUringEventLoopGroup;
+import io.netty.incubator.channel.uring.IOUringSocketChannel;
+import org.junit.Assert;
+import org.junit.Test;
+
+import static com.alipay.sofa.rpc.common.RpcOptions.TRANSPORT_USE_IO_URING;
+
+/**
+ * @author chengming
+ * @version NettyHelperTest.java, v 0.1 2024年03月18日 2:35 PM chengming
+ */
+public class NettyHelperTest {
+
+    @Test
+    public void testEventLoopGroup() {
+        System.setProperty("os.name", "linux111");
+        System.setProperty(TRANSPORT_USE_IO_URING, "true");
+
+        EventLoopGroup eventLoopGroup = NettyHelper.eventLoopGroup(1, new NamedThreadFactory("test", true));
+        Class<? extends SocketChannel> socketChannel = NettyHelper.socketChannel();
+        if (IOUring.isAvailable()) {
+            Assert.assertTrue(eventLoopGroup instanceof IOUringEventLoopGroup);
+            Assert.assertEquals(IOUringSocketChannel.class, socketChannel);
+        } else {
+            Assert.assertTrue(eventLoopGroup instanceof NioEventLoopGroup);
+            Assert.assertEquals(NioSocketChannel.class, socketChannel);
+        }
+    }
+}