Add support for Unix domain sockets

pom.xml

@@ -658,6 +658,13 @@
       <artifactId>netty-transport-native-unix-common</artifactId>
       <version>${netty.version}</version>
     </dependency>
+    <dependency>
+      <groupId>io.netty</groupId>
+      <artifactId>netty-transport-native-epoll</artifactId>
+      <version>${netty.version}</version>
+      <scope>test</scope>
+      <classifier>${jni.classifier}</classifier>
+    </dependency>
     <dependency>
       <groupId>io.netty</groupId>
       <artifactId>netty-testsuite</artifactId>

src/main/c/netty_io_uring_native.c

@@ -315,6 +315,10 @@ static jint netty_io_uring_afInet6(JNIEnv* env, jclass clazz) {
     return AF_INET6;
 }
 
+static jint netty_io_uring_afUnix(JNIEnv* env, jclass clazz) {
+    return AF_UNIX;
+}
+
 static jint netty_io_uring_sizeofSockaddrIn(JNIEnv* env, jclass clazz) {
     return sizeof(struct sockaddr_in);
 }
@@ -323,6 +327,10 @@ static jint netty_io_uring_sizeofSockaddrIn6(JNIEnv* env, jclass clazz) {
     return sizeof(struct sockaddr_in6);
 }
 
+static jint netty_io_uring_sizeofSockaddrUn(JNIEnv* env, jclass clazz) {
+    return sizeof(struct sockaddr_un);
+}
+
 static jint netty_io_uring_sockaddrInOffsetofSinFamily(JNIEnv* env, jclass clazz) {
     return offsetof(struct sockaddr_in, sin_family);
 }
@@ -359,6 +367,14 @@ static jint netty_io_uring_sockaddrIn6OffsetofSin6ScopeId(JNIEnv* env, jclass cl
     return offsetof(struct sockaddr_in6, sin6_scope_id);
 }
 
+static jint netty_io_uring_sockaddrUnOffsetofSunFamily(JNIEnv* env, jclass clazz) {
+    return offsetof(struct sockaddr_un, sun_family);
+}
+
+static jint netty_io_uring_sockaddrUnOffsetofSunPath(JNIEnv* env, jclass clazz) {
+    return offsetof(struct sockaddr_un, sun_path);
+}
+
 static jint netty_io_uring_in6AddressOffsetofS6Addr(JNIEnv* env, jclass clazz) {
     return offsetof(struct in6_addr, s6_addr);
 }
@@ -492,8 +508,10 @@ static const JNINativeMethod statically_referenced_fixed_method_table[] = {
   { "sockCloexec", "()I", (void *) netty_io_uring_sockCloexec },
   { "afInet", "()I", (void *) netty_io_uring_afInet },
   { "afInet6", "()I", (void *) netty_io_uring_afInet6 },
+  { "afUnix", "()I", (void *) netty_io_uring_afUnix },
   { "sizeofSockaddrIn", "()I", (void *) netty_io_uring_sizeofSockaddrIn },
   { "sizeofSockaddrIn6", "()I", (void *) netty_io_uring_sizeofSockaddrIn6 },
+  { "sizeofSockaddrUn", "()I", (void *) netty_io_uring_sizeofSockaddrUn },
   { "sockaddrInOffsetofSinFamily", "()I", (void *) netty_io_uring_sockaddrInOffsetofSinFamily },
   { "sockaddrInOffsetofSinPort", "()I", (void *) netty_io_uring_sockaddrInOffsetofSinPort },
   { "sockaddrInOffsetofSinAddr", "()I", (void *) netty_io_uring_sockaddrInOffsetofSinAddr },
@@ -503,6 +521,8 @@ static const JNINativeMethod statically_referenced_fixed_method_table[] = {
   { "sockaddrIn6OffsetofSin6Flowinfo", "()I", (void *) netty_io_uring_sockaddrIn6OffsetofSin6Flowinfo },
   { "sockaddrIn6OffsetofSin6Addr", "()I", (void *) netty_io_uring_sockaddrIn6OffsetofSin6Addr },
   { "sockaddrIn6OffsetofSin6ScopeId", "()I", (void *) netty_io_uring_sockaddrIn6OffsetofSin6ScopeId },
+  { "sockaddrUnOffsetofSunFamily", "()I", (void *) netty_io_uring_sockaddrUnOffsetofSunFamily },
+  { "sockaddrUnOffsetofSunPath", "()I", (void *) netty_io_uring_sockaddrUnOffsetofSunPath },
   { "in6AddressOffsetofS6Addr", "()I", (void *) netty_io_uring_in6AddressOffsetofS6Addr },
   { "sizeofSockaddrStorage", "()I", (void *) netty_io_uring_sizeofSockaddrStorage },
   { "sizeofSizeT", "()I", (void *) netty_io_uring_sizeofSizeT },

src/main/java/io/netty/incubator/channel/uring/AbstractIOUringChannel.java

@@ -672,16 +672,14 @@ public void connect(
                 }
 
                 doConnect(remoteAddress, localAddress);
-                InetSocketAddress inetSocketAddress = (InetSocketAddress) remoteAddress;
 
                 remoteAddressMemory = Buffer.allocateDirectWithNativeOrder(Native.SIZEOF_SOCKADDR_STORAGE);
                 long remoteAddressMemoryAddress = Buffer.memoryAddress(remoteAddressMemory);
 
-                SockaddrIn.write(socket.isIpv6(), remoteAddressMemoryAddress, inetSocketAddress);
+                int addrLen = Sockaddr.write(socket.isIpv6(), remoteAddressMemoryAddress, remoteAddress);
 
                 final IOUringSubmissionQueue ioUringSubmissionQueue = submissionQueue();
-                ioUringSubmissionQueue.addConnect(socket.intValue(), remoteAddressMemoryAddress,
-                        Native.SIZEOF_SOCKADDR_STORAGE, (short) 0);
+                ioUringSubmissionQueue.addConnect(socket.intValue(), remoteAddressMemoryAddress, addrLen, (short) 0);
                 ioState |= CONNECT_SCHEDULED;
             } catch (Throwable t) {
                 closeIfClosed();
@@ -785,7 +783,7 @@ protected SocketAddress remoteAddress0() {
     /**
      * Connect to the remote peer
      */
-    private void doConnect(SocketAddress remoteAddress, SocketAddress localAddress) throws Exception {
+    protected void doConnect(SocketAddress remoteAddress, SocketAddress localAddress) throws Exception {
         if (localAddress instanceof InetSocketAddress) {
             checkResolvable((InetSocketAddress) localAddress);
         }

src/main/java/io/netty/incubator/channel/uring/IOUringDomainSocketChannel.java

@@ -0,0 +1,78 @@
+/*
+ * Copyright 2021 The Netty Project
+ *
+ * The Netty Project 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:
+ *
+ *   https://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 io.netty.incubator.channel.uring;
+
+import io.netty.channel.Channel;
+import io.netty.channel.unix.DomainSocketAddress;
+import io.netty.channel.unix.DomainSocketChannel;
+import io.netty.channel.unix.DomainSocketChannelConfig;
+
+import java.net.SocketAddress;
+
+public class IOUringDomainSocketChannel extends AbstractIOUringStreamChannel implements DomainSocketChannel {
+    private final IOUringDomainSocketChannelConfig config;
+    private volatile DomainSocketAddress local;
+    private volatile DomainSocketAddress remote;
+
+    public IOUringDomainSocketChannel() {
+        super(null, LinuxSocket.newSocketDomain(), false);
+        this.config = new IOUringDomainSocketChannelConfig(this);
+    }
+
+    IOUringDomainSocketChannel(Channel parent, LinuxSocket fd, SocketAddress remote) {
+        super(parent, fd, remote);
+        this.config = new IOUringDomainSocketChannelConfig(this);
+    }
+
+    @Override
+    public DomainSocketChannelConfig config() {
+        return config;
+    }
+
+    @Override
+    protected void doBind(SocketAddress localAddress) throws Exception {
+        socket.bind(localAddress);
+        local = (DomainSocketAddress) localAddress;
+    }
+
+    @Override
+    protected void doConnect(SocketAddress remoteAddress, SocketAddress localAddress) throws Exception {
+        super.doConnect(remoteAddress, localAddress);
+
+        local = (DomainSocketAddress) localAddress;
+        remote = (DomainSocketAddress) remoteAddress;
+    }
+
+    @Override
+    public DomainSocketAddress remoteAddress() {
+        return (DomainSocketAddress) super.remoteAddress();
+    }
+
+    @Override
+    protected SocketAddress remoteAddress0() {
+        return remote;
+    }
+
+    @Override
+    public DomainSocketAddress localAddress() {
+        return (DomainSocketAddress) super.localAddress();
+    }
+
+    @Override
+    protected SocketAddress localAddress0() {
+        return local;
+    }
+}

src/main/java/io/netty/incubator/channel/uring/IOUringDomainSocketChannelConfig.java

@@ -0,0 +1,212 @@
+/*
+ * Copyright 2021 The Netty Project
+ *
+ * The Netty Project 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:
+ *
+ *   https://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 io.netty.incubator.channel.uring;
+
+import io.netty.buffer.ByteBufAllocator;
+import io.netty.channel.ChannelOption;
+import io.netty.channel.DefaultChannelConfig;
+import io.netty.channel.MessageSizeEstimator;
+import io.netty.channel.RecvByteBufAllocator;
+import io.netty.channel.WriteBufferWaterMark;
+import io.netty.channel.socket.DuplexChannelConfig;
+import io.netty.channel.unix.DomainSocketChannelConfig;
+import io.netty.channel.unix.DomainSocketReadMode;
+import io.netty.util.internal.ObjectUtil;
+
+import java.io.IOException;
+import java.util.Map;
+
+import static io.netty.channel.ChannelOption.ALLOW_HALF_CLOSURE;
+import static io.netty.channel.ChannelOption.SO_RCVBUF;
+import static io.netty.channel.ChannelOption.SO_SNDBUF;
+import static io.netty.channel.unix.UnixChannelOption.DOMAIN_SOCKET_READ_MODE;
+
+public final class IOUringDomainSocketChannelConfig extends DefaultChannelConfig
+        implements DomainSocketChannelConfig, DuplexChannelConfig {
+    private volatile DomainSocketReadMode mode = DomainSocketReadMode.BYTES;
+    private volatile boolean allowHalfClosure;
+
+    IOUringDomainSocketChannelConfig(AbstractIOUringChannel channel) {
+        super(channel);
+    }
+
+    @Override
+    public Map<ChannelOption<?>, Object> getOptions() {
+        return getOptions(super.getOptions(), DOMAIN_SOCKET_READ_MODE, ALLOW_HALF_CLOSURE, SO_SNDBUF, SO_RCVBUF);
+    }
+
+    @SuppressWarnings("unchecked")
+    @Override
+    public <T> T getOption(ChannelOption<T> option) {
+        if (option == ALLOW_HALF_CLOSURE) {
+            return (T) Boolean.valueOf(isAllowHalfClosure());
+        }
+        if (option == SO_SNDBUF) {
+            return (T) Integer.valueOf(getSendBufferSize());
+        }
+        if (option == SO_RCVBUF) {
+            return (T) Integer.valueOf(getReceiveBufferSize());
+        }
+        return super.getOption(option);
+    }
+
+    @Override
+    public <T> boolean setOption(ChannelOption<T> option, T value) {
+        validate(option, value);
+
+        if (option == ALLOW_HALF_CLOSURE) {
+            setAllowHalfClosure((Boolean) value);
+        } else if (option == SO_SNDBUF) {
+            setSendBufferSize((Integer) value);
+        } else if (option == SO_RCVBUF) {
+            setReceiveBufferSize((Integer) value);
+        } else {
+            return super.setOption(option, value);
+        }
+
+        return true;
+    }
+
+    @Override
+    @Deprecated
+    public IOUringDomainSocketChannelConfig setMaxMessagesPerRead(int maxMessagesPerRead) {
+        super.setMaxMessagesPerRead(maxMessagesPerRead);
+        return this;
+    }
+
+    @Override
+    public IOUringDomainSocketChannelConfig setConnectTimeoutMillis(int connectTimeoutMillis) {
+        super.setConnectTimeoutMillis(connectTimeoutMillis);
+        return this;
+    }
+
+    @Override
+    public IOUringDomainSocketChannelConfig setWriteSpinCount(int writeSpinCount) {
+        super.setWriteSpinCount(writeSpinCount);
+        return this;
+    }
+
+    @Override
+    public IOUringDomainSocketChannelConfig setRecvByteBufAllocator(RecvByteBufAllocator allocator) {
+        super.setRecvByteBufAllocator(allocator);
+        return this;
+    }
+
+    @Override
+    public IOUringDomainSocketChannelConfig setAllocator(ByteBufAllocator allocator) {
+        super.setAllocator(allocator);
+        return this;
+    }
+
+    @Override
+    public IOUringDomainSocketChannelConfig setAutoClose(boolean autoClose) {
+        super.setAutoClose(autoClose);
+        return this;
+    }
+
+    @Override
+    public IOUringDomainSocketChannelConfig setMessageSizeEstimator(MessageSizeEstimator estimator) {
+        super.setMessageSizeEstimator(estimator);
+        return this;
+    }
+
+    @Override
+    @Deprecated
+    public IOUringDomainSocketChannelConfig setWriteBufferLowWaterMark(int writeBufferLowWaterMark) {
+        super.setWriteBufferLowWaterMark(writeBufferLowWaterMark);
+        return this;
+    }
+
+    @Override
+    @Deprecated
+    public IOUringDomainSocketChannelConfig setWriteBufferHighWaterMark(int writeBufferHighWaterMark) {
+        super.setWriteBufferHighWaterMark(writeBufferHighWaterMark);
+        return this;
+    }
+
+    @Override
+    public IOUringDomainSocketChannelConfig setWriteBufferWaterMark(WriteBufferWaterMark writeBufferWaterMark) {
+        super.setWriteBufferWaterMark(writeBufferWaterMark);
+        return this;
+    }
+
+    @Override
+    public IOUringDomainSocketChannelConfig setAutoRead(boolean autoRead) {
+        super.setAutoRead(autoRead);
+        return this;
+    }
+
+    @Override
+    public IOUringDomainSocketChannelConfig setReadMode(DomainSocketReadMode mode) {
+        // File descriptors cannot be transferred with io_uring yet
+        if (mode == DomainSocketReadMode.FILE_DESCRIPTORS) {
+            throw new IllegalArgumentException("File descriptor transfers aren't supported with io_uring");
+        }
+
+        this.mode = ObjectUtil.checkNotNull(mode, "mode");
+        return this;
+    }
+
+    @Override
+    public DomainSocketReadMode getReadMode() {
+        return mode;
+    }
+
+    @Override
+    public boolean isAllowHalfClosure() {
+        return allowHalfClosure;
+    }
+
+    @Override
+    public IOUringDomainSocketChannelConfig setAllowHalfClosure(boolean allowHalfClosure) {
+        this.allowHalfClosure = allowHalfClosure;
+        return this;
+    }
+
+    public int getSendBufferSize() {
+        try {
+            return ((IOUringDomainSocketChannel) channel).socket.getSendBufferSize();
+        } catch (IOException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    public IOUringDomainSocketChannelConfig setSendBufferSize(int sendBufferSize) {
+        try {
+            ((IOUringDomainSocketChannel) channel).socket.setSendBufferSize(sendBufferSize);
+            return this;
+        } catch (IOException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    public int getReceiveBufferSize() {
+        try {
+            return ((IOUringDomainSocketChannel) channel).socket.getReceiveBufferSize();
+        } catch (IOException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    public IOUringDomainSocketChannelConfig setReceiveBufferSize(int receiveBufferSize) {
+        try {
+            ((IOUringDomainSocketChannel) channel).socket.setReceiveBufferSize(receiveBufferSize);
+            return this;
+        } catch (IOException e) {
+            throw new RuntimeException(e);
+        }
+    }
+}