native socket

Jvm 中的 Socket

server 监听

public class SocketServer {
    private static final int PORT = 80;
    public static void main(String[] args) throws IOException {
        ServerSocket serverSocket = new ServerSocket(PORT);
        Socket server = serverSocket.accept();
        DataInputStream in = new DataInputStream(server.getInputStream());
        System.out.println(in.readUTF());
        DataOutputStream out = new DataOutputStream(server.getOutputStream());
        out.writeUTF("谢谢连接我：" + server.getLocalSocketAddress() + "\nGoodbye!");
        server.close();
    }
}

public class ServerSocket implements java.io.Closeable {
    SocketImpl getImpl() throws SocketException {
        if (!created)
            createImpl();
        return impl;
    }
    void createImpl() throws SocketException {
        impl.create(true);
        created = true;
    }
}

public class ServerSocket implements java.io.Closeable {
    public void bind(SocketAddress endpoint, int backlog) throws IOException {
        getImpl().bind(epoint.getAddress(), epoint.getPort());
        getImpl().listen(backlog); 
    }
}

socket 方法

public final class NioSocketImpl extends SocketImpl implements PlatformSocketImpl {
    protected void create(boolean stream) throws IOException {
        FileDescriptor fd;
        if (server) {
            fd = Net.serverSocket(true);
        }
        this.fd = fd;
    }
}        

public class Net {
    static FileDescriptor serverSocket(boolean stream) {
        return serverSocket(UNSPEC, stream);
    }
    static FileDescriptor serverSocket(ProtocolFamily family, boolean stream) {
        boolean preferIPv6 = isIPv6Available() &&
            (family != StandardProtocolFamily.INET);
        return IOUtil.newFD(socket0(preferIPv6, stream, true, fastLoopback));
    }
    private static native int socket0(boolean preferIPv6, boolean stream, boolean reuse, boolean fastLoopback);
}

JNIEXPORT jint JNICALL
Java_sun_nio_ch_Net_socket0(JNIEnv *env, jclass cl, jboolean preferIPv6,
                            jboolean stream, jboolean reuse, jboolean ignored) {
    int fd;
    int type = (stream ? SOCK_STREAM : SOCK_DGRAM);
    int domain = (ipv6_available() && preferIPv6) ? AF_INET6 : AF_INET;
    // unix 标准的 socket 接口，用于生成 socket 的文件描述符
    fd = socket(domain, type, 0);
    if (fd < 0) {
        return handleSocketError(env, errno);
    }
    if (reuse) {
        int arg = 1;
        // unix 标准的 socket 接口，用于修改 socket 的配置
        if (setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, (char*)&arg, sizeof(arg)) < 0) {
            close(fd);
            return -1;
        }
    }
    return fd;
}

bind 方法

public final class NioSocketImpl extends SocketImpl implements PlatformSocketImpl {
    protected void bind(InetAddress host, int port) throws IOException {
        Net.bind(fd, host, port);
    }
}

public class Net {
    public static void bind(FileDescriptor fd, InetAddress addr, int port) throws IOException {
        bind(UNSPEC, fd, addr, port);
    }
    static void bind(ProtocolFamily family, FileDescriptor fd, InetAddress addr, int port) throws IOException {
        bind0(fd, preferIPv6, exclusiveBind, addr, port);
    }
    private static native void bind0(FileDescriptor fd, boolean preferIPv6, boolean useExclBind, InetAddress addr, int port) throws IOException;
}

JNIEXPORT void JNICALL
Java_sun_nio_ch_Net_bind0(JNIEnv *env, jclass clazz, jobject fdo, jboolean preferIPv6,
                          jboolean useExclBind, jobject iao, int port){
    SOCKETADDRESS sa;
    int sa_len = 0;
    int rv = 0;
    if (NET_InetAddressToSockaddr(env, iao, port, &sa, &sa_len, preferIPv6) != 0) {
        return;
    }
    rv = NET_Bind(fdval(env, fdo), &sa, sa_len);
    if (rv != 0) {
        handleSocketError(env, errno);
    }
}

int
NET_Bind(int fd, SOCKETADDRESS *sa, int len)
{
    int rv;
    int arg, alen;
    // unix 标准的 socket 接口，用于告诉内核程序将指定监听端口与 fd 关联起来
    rv = bind(fd, &sa->sa, len);
    return rv;
}

listen 方法

public final class NioSocketImpl extends SocketImpl implements PlatformSocketImpl {
    protected void listen(int backlog) throws IOException {
            Net.listen(fd, backlog < 1 ? 50 : backlog);
        }
    }
}

public class Net {
    static native void listen(FileDescriptor fd, int backlog) throws IOException;
}

JNIEXPORT void JNICALL
Java_sun_nio_ch_Net_listen(JNIEnv *env, jclass cl, jobject fdo, jint backlog) {
    // unix 标准的 socket 接口，用于启动监听 
    listen(fdval(env, fdo), backlog);
}

accept 函数

public class ServerSocket implements java.io.Closeable {
    public Socket accept() throws IOException {
        Socket s = new Socket((SocketImpl) null);
        implAccept(s);
        return s;
    }
    
    protected final void implAccept(Socket s) throws IOException {
        SocketImpl si = s.impl;
        if (si == null) {
            si = implAccept();
            s.setImpl(si);
            s.postAccept();
            return;
        }
    }
    
    private SocketImpl implAccept() throws IOException {
        if (impl instanceof PlatformSocketImpl) {
            return platformImplAccept();
        } 
    }
    private SocketImpl platformImplAccept() throws IOException {
        SocketImpl psi = SocketImpl.createPlatformSocketImpl(false);
        implAccept(psi);
        return psi;
    }
    private void implAccept(SocketImpl si) throws IOException {
        impl.accept(si); 
    }
}

public final class NioSocketImpl extends SocketImpl implements PlatformSocketImpl {
    protected void accept(SocketImpl si) throws IOException {
        FileDescriptor newfd = new FileDescriptor();
        int n = 0; 
        n = Net.accept(fd, newfd, isaa);
        InetSocketAddress localAddress;
        localAddress = Net.localAddress(newfd);
        IOUtil.configureBlocking(newfd, true);
    }
}

public class Net {
    public static native int accept(FileDescriptor fd,
                                    FileDescriptor newfd,
                                    InetSocketAddress[] isaa) throws IOException;
}

JNIEXPORT jint JNICALL
Java_sun_nio_ch_Net_accept(JNIEnv *env, jclass clazz, jobject fdo, jobject newfdo, jobjectArray isaa) {
    jint fd = fdval(env, fdo);
    jint newfd;
    SOCKETADDRESS sa;
    socklen_t sa_len = sizeof(SOCKETADDRESS);
    jobject remote_ia;
    jint remote_port = 0;
    jobject isa;
    for (;;) {
        // unix 标准的 socket 接口。用于接收一个新的连接
        newfd = accept(fd, &sa.sa, &sa_len);
        if (newfd >= 0) {
            break;
        }
        if (errno != ECONNABORTED) {
            break;
        }
    }
    setfdval(env, newfdo, newfd);
    remote_ia = NET_SockaddrToInetAddress(env, &sa, (int *)&remote_port);
    isa = (*env)->NewObject(env, isa_class, isa_ctorID, remote_ia, remote_port);
    (*env)->SetObjectArrayElement(env, isaa, 0, isa);
    return 1;
}

IO 多路复用 - poll

修改 pollfd 列表

class PollSelectorImpl extends SelectorImpl {
    private void putDescriptor(int i, int fd) {
        int offset = SIZE_POLLFD * i + FD_OFFSET;
        pollArray.putInt(offset, fd);
    }    
    private void putEventOps(int i, int event) {
        int offset = SIZE_POLLFD * i + EVENT_OFFSET;
        pollArray.putShort(offset, (short)event);
    }
    
    private void add(SelectionKeyImpl ski, int ops) {
        expandIfNeeded();
        int index = pollArraySize;
        putDescriptor(index, ski.getFDVal());
        putEventOps(index, ops);
        putReventOps(index, 0);
        ski.setIndex(index);
        pollArraySize++;
        pollKeys.add(ski);
    }
    
    private void processUpdateQueue() {
        SelectionKeyImpl ski;
        while ((ski = updateKeys.pollFirst()) != null) {
            int newEvents = ski.translateInterestOps();
            if (ski.isValid()) {
                int index = ski.getIndex();
                if (index > 0) {
                    if (newEvents == 0) {
                        remove(ski);
                    } else {
                        update(ski, newEvents);
                    }
                } else if (newEvents != 0) {
                    add(ski, newEvents);
                }
            }
        }
    }
    public void setEventOps(SelectionKeyImpl ski) {
        updateKeys.addLast(ski);
    }
}

public final class SelectionKeyImpl extends AbstractSelectionKey {
    public SelectionKey interestOps(int ops) {
        int oldOps = (int) INTERESTOPS.getAndSet(this, ops);
        if (ops != oldOps) {
            selector.setEventOps(this);
        }
        return this;
    }
}

调用 poll

public abstract class SelectorImpl extends AbstractSelector {
    @Override
    public final int select(long timeout) throws IOException {
        if (timeout < 0)
            throw new IllegalArgumentException("Negative timeout");
        return lockAndDoSelect(null, (timeout == 0) ? -1 : timeout);
    }    
    private int lockAndDoSelect(Consumer<SelectionKey> action, long timeout) throws IOException {
        return doSelect(action, timeout);
    }
    
}

class PollSelectorImpl extends SelectorImpl {
    @Override
    protected int doSelect(Consumer<SelectionKey> action, long timeout) throws IOException {
        int numPolled;
        do {
            long startTime = timedPoll ? System.nanoTime() : 0;
            numPolled = poll(pollArray.address(), pollArraySize, to);
        } while (numPolled == IOStatus.INTERRUPTED);
        return processEvents(action);
    }
    private static native int poll(long pollAddress, int numfds, int timeout);
}

JNIEXPORT jint JNICALL
Java_sun_nio_ch_PollSelectorImpl_poll(JNIEnv *env, jclass clazz,
                                      jlong address, jint numfds,
                                      jint timeout) {
    struct pollfd *a;
    int res;
    a = (struct pollfd *) jlong_to_ptr(address);
    res = poll(a, numfds, timeout);
    return (jint) res;
}

IO 多路复用 - epoll

java 原生不支持 epoll