Pass credentials/fds via UNIX sockets. SOCK_SEQPACKET support.

src/one/nio/net/JavaDatagramSocket.java

@@ -143,6 +143,16 @@ public final int write(ByteBuffer src) throws IOException {
         return ch.write(src);
     }
 
+    @Override
+    public int sendMsg(Msg msg, int flags) throws IOException {
+        throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public int recvMsg(Msg msg, int flags) throws IOException {
+        throw new UnsupportedOperationException();
+    }
+
     @Override
     public final void setBlocking(boolean blocking) {
         try {

src/one/nio/net/JavaServerSocket.java

@@ -122,6 +122,16 @@ public final int write(ByteBuffer src) throws IOException {
         throw new UnsupportedOperationException();
     }
 
+    @Override
+    public int sendMsg(Msg msg, int flags) throws IOException {
+        throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public int recvMsg(Msg msg, int flags) throws IOException {
+        throw new UnsupportedOperationException();
+    }
+
     @Override
     public final void setBlocking(boolean blocking) {
         try {

src/one/nio/net/JavaSocket.java

@@ -152,6 +152,16 @@ public final int write(ByteBuffer src) throws IOException {
         return ch.write(src);
     }
 
+    @Override
+    public int sendMsg(Msg msg, int flags) throws IOException {
+        throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public int recvMsg(Msg msg, int flags) throws IOException {
+        throw new UnsupportedOperationException();
+    }
+
     @Override
     public final void setBlocking(boolean blocking) {
         try {

src/one/nio/net/Msg.java

@@ -0,0 +1,79 @@
+/*
+ * Copyright 2021 Odnoklassniki Ltd, Mail.Ru Group
+ *
+ * Licensed 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 one.nio.net;
+
+import java.util.Arrays;
+
+public final class Msg {
+    public static final int SCM_RIGHTS = 1;
+    public static final int SCM_CREDENTIALS = 2;
+
+    private byte[] data;
+    private int cmsgType;
+    private int[] cmsgData;
+
+    public Msg(int capacity) {
+        this.data = new byte[capacity];
+    }
+
+    public Msg(byte[] data) {
+        this.data = data;
+    }
+
+    public byte[] data() {
+        return data;
+    }
+
+    public int cmsgType() {
+        return cmsgType;
+    }
+
+    public int[] cmsgData() {
+        return cmsgData;
+    }
+
+    public Msg withCmsg(int type, int... data) {
+        if (cmsgType != 0) {
+            throw new IllegalStateException("cmsg already set");
+        }
+        cmsgType = type;
+        cmsgData = data;
+        return this;
+    }
+
+    public Msg withFd(int fd) {
+        return withCmsg(SCM_RIGHTS, fd);
+    }
+
+    public Msg withFd(int... fds) {
+        return withCmsg(SCM_RIGHTS, fds);
+    }
+
+    public Msg withCred(int pid, int uid, int gid) {
+        return withCmsg(SCM_CREDENTIALS, pid, uid, gid);
+    }
+
+    @Override
+    public String toString() {
+        StringBuilder sb = new StringBuilder("Msg(").append(data == null ? 0 : data.length);
+        if (cmsgType != 0) {
+            sb.append(cmsgType == SCM_RIGHTS ? " + SCM_RIGHTS" : " + SCM_CREDENTIALS");
+            sb.append(Arrays.toString(cmsgData));
+        }
+        return sb.append(')').toString();
+    }
+}

src/one/nio/net/NativeSocket.java

@@ -22,7 +22,6 @@
 
 import java.io.IOException;
 import java.io.RandomAccessFile;
-import java.lang.reflect.Field;
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
 import java.net.UnknownHostException;
@@ -31,22 +30,13 @@
 class NativeSocket extends Socket {
     private static final int INET_FAMILY = initNatives(Boolean.getBoolean("java.net.preferIPv4Stack"));
 
-    private static final Field ARRAY_FIELD;
-    private static final Field OFFSET_FIELD;
-
-    static {
-        ARRAY_FIELD = JavaInternals.getField(ByteBuffer.class, "hb");
-        OFFSET_FIELD = JavaInternals.getField(ByteBuffer.class, "offset");
-    }
+    private static final long ARRAY_FIELD = JavaInternals.fieldOffset(ByteBuffer.class, "hb");
+    private static final long OFFSET_FIELD = JavaInternals.fieldOffset(ByteBuffer.class, "offset");
 
     int fd;
 
-    NativeSocket(int domain, boolean datagram) throws IOException {
-        this.fd = socket0(domain, datagram);
-    }
-
-    NativeSocket(boolean datagram) throws IOException {
-        this.fd = socket0(INET_FAMILY, datagram);
+    NativeSocket(int domain, int type) throws IOException {
+        this.fd = socket0(domain != 0 ? domain : INET_FAMILY, type);
     }
 
     NativeSocket(int fd) {
@@ -157,11 +147,11 @@ private int sendTo(ByteBuffer src, int flags, Object address, int port) throws I
     }
 
     private byte[] getArray(ByteBuffer src) throws IllegalAccessException {
-        return (byte[]) ARRAY_FIELD.get(src);
+        return (byte[]) JavaInternals.unsafe.getObject(src, ARRAY_FIELD);
     }
 
     private int getOffset(ByteBuffer src) throws IllegalAccessException {
-        return OFFSET_FIELD.getInt(src);
+        return JavaInternals.unsafe.getInt(src, OFFSET_FIELD);
     }
 
     @Override
@@ -235,6 +225,16 @@ public int write(ByteBuffer src) throws IOException {
         return bytes;
     }
 
+    @Override
+    public int sendMsg(Msg msg, int flags) throws IOException {
+        return sendMsg0(msg.data(), msg.cmsgType(), msg.cmsgData(), flags);
+    }
+
+    @Override
+    public int recvMsg(Msg msg, int flags) throws IOException {
+        return recvMsg0(msg.data(), msg, flags);
+    }
+
     @Override
     public final native void setBlocking(boolean blocking);
 
@@ -310,7 +310,7 @@ static Object toNativeAddr(String host, int port) throws UnknownHostException {
 
     private static native int initNatives(boolean preferIPv4);
 
-    private static native int socket0(int domain, boolean datagram) throws IOException;
+    private static native int socket0(int domain, int type) throws IOException;
 
     final native void connect0(Object address, int port) throws IOException;
     final native void bind0(Object address, int port) throws IOException;
@@ -320,4 +320,6 @@ static Object toNativeAddr(String host, int port) throws UnknownHostException {
     final native int sendTo1(long buf, int size, int flags, Object address, int port) throws IOException;
     final native int recvFrom0(byte[] data, int offset, int maxSize, int flags, AddressHolder holder) throws IOException;
     final native int recvFrom1(long buf, int maxSize, int flags, AddressHolder holder) throws IOException;
+    final native int sendMsg0(byte[] data, int cmsgType, int[] cmsgData, int flags) throws IOException;
+    final native int recvMsg0(byte[] data, Msg msg, int flags) throws IOException;
 }

src/one/nio/net/Socket.java

@@ -32,6 +32,13 @@ public abstract class Socket implements ByteChannel {
     public static final int AF_INET  = 2;
     public static final int AF_INET6 = 10;
 
+    // Socket types
+    public static final int SOCK_STREAM    = 1;
+    public static final int SOCK_DGRAM     = 2;
+    public static final int SOCK_RAW       = 3;
+    public static final int SOCK_RDM       = 4;
+    public static final int SOCK_SEQPACKET = 5;
+
     // Use when the address has no port (i.e. for AF_UNIX address)
     public static final int NO_PORT = -1;
 
@@ -57,6 +64,29 @@ public abstract class Socket implements ByteChannel {
     public static final int IPTOS_THROUGHPUT  = 0x08;
     public static final int IPTOS_LOWDELAY    = 0x10;
 
+    // Socket options
+    public static final int SO_DEBUG     = 1;
+    public static final int SO_REUSEADDR = 2;
+    public static final int SO_TYPE      = 3;
+    public static final int SO_ERROR     = 4;
+    public static final int SO_DONTROUTE = 5;
+    public static final int SO_BROADCAST = 6;
+    public static final int SO_SNDBUF    = 7;
+    public static final int SO_RCVBUF    = 8;
+    public static final int SO_KEEPALIVE = 9;
+    public static final int SO_OOBINLINE = 10;
+    public static final int SO_NO_CHECK  = 11;
+    public static final int SO_PRIORITY  = 12;
+    public static final int SO_LINGER    = 13;
+    public static final int SO_BSDCOMPAT = 14;
+    public static final int SO_REUSEPORT = 15;
+    public static final int SO_PASSCRED  = 16;
+    public static final int SO_PEERCRED  = 17;
+    public static final int SO_RCVLOWAT  = 18;
+    public static final int SO_SNDLOWAT  = 19;
+    public static final int SO_RCVTIMEO  = 20;
+    public static final int SO_SNDTIMEO  = 21;
+
     // TCP socket options
     public static final int TCP_NODELAY      = 1;
     public static final int TCP_MAXSEG       = 2;
@@ -88,6 +118,8 @@ public abstract class Socket implements ByteChannel {
     public abstract void readFully(byte[] data, int offset, int count) throws IOException;
     public abstract InetSocketAddress recv(ByteBuffer dst, int flags) throws IOException;
     public abstract long sendFile(RandomAccessFile file, long offset, long count) throws IOException;
+    public abstract int sendMsg(Msg msg, int flags) throws IOException;
+    public abstract int recvMsg(Msg msg, int flags) throws IOException;
     public abstract void setBlocking(boolean blocking);
     public abstract boolean isBlocking();
     public abstract void setTimeout(int timeout);
@@ -149,23 +181,23 @@ public int read(byte[] data, int offset, int count) throws IOException {
     }
 
     public static Socket create() throws IOException {
-        return NativeLibrary.IS_SUPPORTED ? new NativeSocket(false) : new JavaSocket();
+        return NativeLibrary.IS_SUPPORTED ? new NativeSocket(0, SOCK_STREAM) : new JavaSocket();
     }
 
     public static Socket createServerSocket() throws IOException {
-        return NativeLibrary.IS_SUPPORTED ? new NativeSocket(false) : new JavaServerSocket();
+        return NativeLibrary.IS_SUPPORTED ? new NativeSocket(0, SOCK_STREAM) : new JavaServerSocket();
     }
 
     public static Socket createDatagramSocket() throws IOException {
-        return NativeLibrary.IS_SUPPORTED ? new NativeSocket(true) : new JavaDatagramSocket();
+        return NativeLibrary.IS_SUPPORTED ? new NativeSocket(0, SOCK_DGRAM) : new JavaDatagramSocket();
     }
 
-    public static Socket createUnixSocket() throws IOException {
+    public static Socket createUnixSocket(int type) throws IOException {
         if (!NativeLibrary.IS_SUPPORTED) {
             throw new IOException("Unix sockets are supported in native mode only");
         }
 
-        return new NativeSocket(AF_UNIX, true);
+        return new NativeSocket(AF_UNIX, type);
     }
 
     public static Socket connectInet(InetAddress address, int port) throws IOException {
@@ -186,7 +218,7 @@ public static Socket connectUnix(File unixPath) throws IOException {
             throw new IOException("Unix sockets are supported in native mode only");
         }
 
-        NativeSocket sock = new NativeSocket(AF_UNIX, false);
+        NativeSocket sock = new NativeSocket(AF_UNIX, SOCK_STREAM);
         sock.connect(unixPath.getAbsolutePath(), NO_PORT);
         return sock;
     }
@@ -196,7 +228,7 @@ public static Socket bindUnix(File unixPath, int backlog) throws IOException {
             throw new IOException("Unix sockets are supported in native mode only");
         }
 
-        NativeSocket sock = new NativeSocket(AF_UNIX, false);
+        NativeSocket sock = new NativeSocket(AF_UNIX, SOCK_STREAM);
         sock.bind(unixPath.getAbsolutePath(), NO_PORT, backlog);
         sock.listen(backlog);
         return sock;

src/one/nio/net/native/jni_util.h

@@ -14,6 +14,8 @@
  * limitations under the License.
  */
 
+#pragma once
+
 #define MAX_STACK_BUF 65536
 #define SIG_WAKEUP (__SIGRTMAX - 2)