/*
 * Copyright (c) 2001, 2020, Oracle and/or its affiliates. All rights reserved.
 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
 *
 * This code is free software; you can redistribute it and/or modify it
 * under the terms of the GNU General Public License version 2 only, as
 * published by the Free Software Foundation.  Oracle designates this
 * particular file as subject to the "Classpath" exception as provided
 * by Oracle in the LICENSE file that accompanied this code.
 *
 * This code is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
 * version 2 for more details (a copy is included in the LICENSE file that
 * accompanied this code).
 *
 * You should have received a copy of the GNU General Public License version
 * 2 along with this work; if not, write to the Free Software Foundation,
 * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
 *
 * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
 * or visit www.oracle.com if you need additional information or have any
 * questions.
 */

package sun.nio.ch;

import java.io.FileDescriptor;
import java.io.IOException;
import java.io.UncheckedIOException;
import java.lang.invoke.MethodHandles;
import java.lang.invoke.VarHandle;
import java.lang.ref.Cleaner.Cleanable;
import java.lang.reflect.Method;
import java.net.DatagramSocket;
import java.net.Inet4Address;
import java.net.Inet6Address;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.NetworkInterface;
import java.net.PortUnreachableException;
import java.net.ProtocolFamily;
import java.net.SocketAddress;
import java.net.SocketException;
import java.net.SocketOption;
import java.net.SocketTimeoutException;
import java.net.StandardProtocolFamily;
import java.net.StandardSocketOptions;
import java.nio.ByteBuffer;
import java.nio.channels.AlreadyBoundException;
import java.nio.channels.AlreadyConnectedException;
import java.nio.channels.AsynchronousCloseException;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.DatagramChannel;
import java.nio.channels.IllegalBlockingModeException;
import java.nio.channels.MembershipKey;
import java.nio.channels.NotYetConnectedException;
import java.nio.channels.SelectionKey;
import java.nio.channels.spi.AbstractSelectableChannel;
import java.nio.channels.spi.SelectorProvider;
import java.security.AccessController;
import java.security.PrivilegedExceptionAction;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Consumer;

import jdk.internal.ref.CleanerFactory;
import sun.net.ResourceManager;
import sun.net.ext.ExtendedSocketOptions;
import sun.net.util.IPAddressUtil;

An implementation of DatagramChannels.
/**
 * An implementation of DatagramChannels.
 */

class DatagramChannelImpl
    extends DatagramChannel
    implements SelChImpl
{
    // Used to make native read and write calls
    private static final NativeDispatcher nd = new DatagramDispatcher();

    // true if interruptible (can be false to emulate legacy DatagramSocket)
    private final boolean interruptible;

    // The protocol family of the socket
    private final ProtocolFamily family;

    // Our file descriptor
    private final FileDescriptor fd;
    private final int fdVal;

    // Native sockaddrs and cached InetSocketAddress for receive, protected by readLock
    private NativeSocketAddress sourceSockAddr;
    private NativeSocketAddress cachedSockAddr;
    private InetSocketAddress cachedInetSocketAddress;

    // Native sockaddr and cached objects for send, protected by writeLock
    private final NativeSocketAddress targetSockAddr;
    private InetSocketAddress previousTarget;
    private int previousSockAddrLength;

    // Cleaner to close file descriptor and free native socket address
    private final Cleanable cleaner;

    // Lock held by current reading or connecting thread
    private final ReentrantLock readLock = new ReentrantLock();

    // Lock held by current writing or connecting thread
    private final ReentrantLock writeLock = new ReentrantLock();

    // Lock held by any thread that modifies the state fields declared below
    // DO NOT invoke a blocking I/O operation while holding this lock!
    private final Object stateLock = new Object();

    // -- The following fields are protected by stateLock

    // State (does not necessarily increase monotonically)
    private static final int ST_UNCONNECTED = 0;
    private static final int ST_CONNECTED = 1;
    private static final int ST_CLOSING = 2;
    private static final int ST_CLOSED = 3;
    private int state;

    // IDs of native threads doing reads and writes, for signalling
    private long readerThread;
    private long writerThread;

    // Local and remote (connected) address
    private InetSocketAddress localAddress;
    private InetSocketAddress remoteAddress;

    // Local address prior to connecting
    private InetSocketAddress initialLocalAddress;

    // Socket adaptor, created lazily
    private static final VarHandle SOCKET;
    static {
        try {
            MethodHandles.Lookup l = MethodHandles.lookup();
            SOCKET = l.findVarHandle(DatagramChannelImpl.class, "socket", DatagramSocket.class);
        } catch (Exception e) {
            throw new InternalError(e);
        }
    }
    private volatile DatagramSocket socket;

    // Multicast support
    private MembershipRegistry registry;

    // set true when socket is bound and SO_REUSEADDRESS is emulated
    private boolean reuseAddressEmulated;

    // set true/false when socket is already bound and SO_REUSEADDR is emulated
    private boolean isReuseAddress;

    // -- End of fields protected by stateLock


    DatagramChannelImpl(SelectorProvider sp, boolean interruptible) throws IOException {
        this(sp, (Net.isIPv6Available()
                ? StandardProtocolFamily.INET6
                : StandardProtocolFamily.INET),
                interruptible);
    }

    DatagramChannelImpl(SelectorProvider sp, ProtocolFamily family, boolean interruptible)
        throws IOException
    {
        super(sp);

        Objects.requireNonNull(family, "'family' is null");
        if ((family != StandardProtocolFamily.INET) &&
                (family != StandardProtocolFamily.INET6)) {
            throw new UnsupportedOperationException("Protocol family not supported");
        }
        if (family == StandardProtocolFamily.INET6 && !Net.isIPv6Available()) {
            throw new UnsupportedOperationException("IPv6 not available");
        }

        FileDescriptor fd = null;
        NativeSocketAddress[] sockAddrs = null;

        ResourceManager.beforeUdpCreate();
        boolean initialized = false;
        try {
            this.interruptible = interruptible;
            this.family = family;
            this.fd = fd = Net.socket(family, false);
            this.fdVal = IOUtil.fdVal(fd);

            sockAddrs = NativeSocketAddress.allocate(3);
            readLock.lock();
            try {
                this.sourceSockAddr = sockAddrs[0];
                this.cachedSockAddr = sockAddrs[1];
            } finally {
                readLock.unlock();
            }
            this.targetSockAddr = sockAddrs[2];

            initialized = true;
        } finally {
            if (!initialized) {
                if (sockAddrs != null) NativeSocketAddress.freeAll(sockAddrs);
                if (fd != null) nd.close(fd);
                ResourceManager.afterUdpClose();
            }
        }

        Runnable releaser = releaserFor(fd, sockAddrs);
        this.cleaner = CleanerFactory.cleaner().register(this, releaser);
    }

    DatagramChannelImpl(SelectorProvider sp, FileDescriptor fd)
        throws IOException
    {
        super(sp);

        NativeSocketAddress[] sockAddrs = null;

        ResourceManager.beforeUdpCreate();
        boolean initialized = false;
        try {
            this.interruptible = true;
            this.family = Net.isIPv6Available()
                    ? StandardProtocolFamily.INET6
                    : StandardProtocolFamily.INET;
            this.fd = fd;
            this.fdVal = IOUtil.fdVal(fd);

            sockAddrs = NativeSocketAddress.allocate(3);
            readLock.lock();
            try {
                this.sourceSockAddr = sockAddrs[0];
                this.cachedSockAddr = sockAddrs[1];
            } finally {
                readLock.unlock();
            }
            this.targetSockAddr = sockAddrs[2];

            initialized = true;
        } finally {
            if (!initialized) {
                if (sockAddrs != null) NativeSocketAddress.freeAll(sockAddrs);
                nd.close(fd);
                ResourceManager.afterUdpClose();
            }
        }

        Runnable releaser = releaserFor(fd, sockAddrs);
        this.cleaner = CleanerFactory.cleaner().register(this, releaser);

        synchronized (stateLock) {
            this.localAddress = Net.localAddress(fd);
        }
    }

    // @throws ClosedChannelException if channel is closed
    private void ensureOpen() throws ClosedChannelException {
        if (!isOpen())
            throw new ClosedChannelException();
    }

    @Override
    public DatagramSocket socket() {
        DatagramSocket socket = this.socket;
        if (socket == null) {
            socket = DatagramSocketAdaptor.create(this);
            if (!SOCKET.compareAndSet(this, null, socket)) {
                socket = this.socket;
            }
        }
        return socket;
    }

    @Override
    public SocketAddress getLocalAddress() throws IOException {
        synchronized (stateLock) {
            ensureOpen();
            // Perform security check before returning address
            return Net.getRevealedLocalAddress(localAddress);
        }
    }

    @Override
    public SocketAddress getRemoteAddress() throws IOException {
        synchronized (stateLock) {
            ensureOpen();
            return remoteAddress;
        }
    }

    
Returns the protocol family to specify to set/getSocketOption for the
given socket option.
/**
     * Returns the protocol family to specify to set/getSocketOption for the
     * given socket option.
     */
    private ProtocolFamily familyFor(SocketOption<?> name) {
        assert Thread.holdsLock(stateLock);

        // unspecified (most options)
        if (SocketOptionRegistry.findOption(name, Net.UNSPEC) != null)
            return Net.UNSPEC;

        // IPv4 socket
        if (family == StandardProtocolFamily.INET)
            return StandardProtocolFamily.INET;

        // IPv6 socket that is unbound
        if (localAddress == null)
            return StandardProtocolFamily.INET6;

        // IPv6 socket bound to wildcard or IPv6 address
        InetAddress address = localAddress.getAddress();
        if (address.isAnyLocalAddress() || (address instanceof Inet6Address))
            return StandardProtocolFamily.INET6;

        // IPv6 socket bound to IPv4 address
        if (Net.canUseIPv6OptionsWithIPv4LocalAddress()) {
            // IPV6_XXX options can be used
            return StandardProtocolFamily.INET6;
        } else {
            // IPV6_XXX options cannot be used
            return StandardProtocolFamily.INET;
        }
    }

    @Override
    public <T> DatagramChannel setOption(SocketOption<T> name, T value)
        throws IOException
    {
        Objects.requireNonNull(name);
        if (!supportedOptions().contains(name))
            throw new UnsupportedOperationException("'" + name + "' not supported");
        if (!name.type().isInstance(value))
            throw new IllegalArgumentException("Invalid value '" + value + "'");

        synchronized (stateLock) {
            ensureOpen();

            ProtocolFamily family = familyFor(name);

            // Some platforms require both IPV6_XXX and IP_XXX socket options to
            // be set when the channel's socket is IPv6 and it is used to send
            // IPv4 multicast datagrams. The IP_XXX socket options are set on a
            // best effort basis.
            boolean needToSetIPv4Option = (family != Net.UNSPEC)
                    && (this.family == StandardProtocolFamily.INET6)
                    && Net.shouldSetBothIPv4AndIPv6Options();

            // outgoing multicast interface
            if (name == StandardSocketOptions.IP_MULTICAST_IF) {
                assert family != Net.UNSPEC;
                NetworkInterface interf = (NetworkInterface) value;
                if (family == StandardProtocolFamily.INET6) {
                    int index = interf.getIndex();
                    if (index == -1)
                        throw new IOException("Network interface cannot be identified");
                    Net.setInterface6(fd, index);
                }
                if (family == StandardProtocolFamily.INET || needToSetIPv4Option) {
                    // need IPv4 address to identify interface
                    Inet4Address target = Net.anyInet4Address(interf);
                    if (target != null) {
                        try {
                            Net.setInterface4(fd, Net.inet4AsInt(target));
                        } catch (IOException ioe) {
                            if (family == StandardProtocolFamily.INET) throw ioe;
                        }
                    } else if (family == StandardProtocolFamily.INET) {
                        throw new IOException("Network interface not configured for IPv4");
                    }
                }
                return this;
            }

            // SO_REUSEADDR needs special handling as it may be emulated
            if (name == StandardSocketOptions.SO_REUSEADDR
                && Net.useExclusiveBind() && localAddress != null) {
                reuseAddressEmulated = true;
                this.isReuseAddress = (Boolean)value;
            }

            // remaining options don't need any special handling
            Net.setSocketOption(fd, family, name, value);
            if (needToSetIPv4Option && family != StandardProtocolFamily.INET) {
                try {
                    Net.setSocketOption(fd, StandardProtocolFamily.INET, name, value);
                } catch (IOException ignore) { }
            }

            return this;
        }
    }

    @Override
    @SuppressWarnings("unchecked")
    public <T> T getOption(SocketOption<T> name)
        throws IOException
    {
        Objects.requireNonNull(name);
        if (!supportedOptions().contains(name))
            throw new UnsupportedOperationException("'" + name + "' not supported");

        synchronized (stateLock) {
            ensureOpen();

            ProtocolFamily family = familyFor(name);

            if (name == StandardSocketOptions.IP_MULTICAST_IF) {
                if (family == StandardProtocolFamily.INET) {
                    int address = Net.getInterface4(fd);
                    if (address == 0)
                        return null;    // default interface

                    InetAddress ia = Net.inet4FromInt(address);
                    NetworkInterface ni = NetworkInterface.getByInetAddress(ia);
                    if (ni == null)
                        throw new IOException("Unable to map address to interface");
                    return (T) ni;
                } else {
                    int index = Net.getInterface6(fd);
                    if (index == 0)
                        return null;    // default interface

                    NetworkInterface ni = NetworkInterface.getByIndex(index);
                    if (ni == null)
                        throw new IOException("Unable to map index to interface");
                    return (T) ni;
                }
            }

            if (name == StandardSocketOptions.SO_REUSEADDR && reuseAddressEmulated) {
                return (T) Boolean.valueOf(isReuseAddress);
            }

            // no special handling
            return (T) Net.getSocketOption(fd, family, name);
        }
    }

    private static class DefaultOptionsHolder {
        static final Set<SocketOption<?>> defaultOptions = defaultOptions();

        private static Set<SocketOption<?>> defaultOptions() {
            HashSet<SocketOption<?>> set = new HashSet<>();
            set.add(StandardSocketOptions.SO_SNDBUF);
            set.add(StandardSocketOptions.SO_RCVBUF);
            set.add(StandardSocketOptions.SO_REUSEADDR);
            if (Net.isReusePortAvailable()) {
                set.add(StandardSocketOptions.SO_REUSEPORT);
            }
            set.add(StandardSocketOptions.SO_BROADCAST);
            set.add(StandardSocketOptions.IP_TOS);
            set.add(StandardSocketOptions.IP_MULTICAST_IF);
            set.add(StandardSocketOptions.IP_MULTICAST_TTL);
            set.add(StandardSocketOptions.IP_MULTICAST_LOOP);
            set.addAll(ExtendedSocketOptions.datagramSocketOptions());
            return Collections.unmodifiableSet(set);
        }
    }

    @Override
    public final Set<SocketOption<?>> supportedOptions() {
        return DefaultOptionsHolder.defaultOptions;
    }

    
Marks the beginning of a read operation that might block.
Params:
blocking – true if configured blocking
mustBeConnected – true if the socket must be connected
Throws:
ClosedChannelException – if the channel is closed
NotYetConnectedException – if mustBeConnected and not connected
IOException – if socket not bound and cannot be bound
Returns:
remote address if connected/**
     * Marks the beginning of a read operation that might block.
     *
     * @param blocking true if configured blocking
     * @param mustBeConnected true if the socket must be connected
     * @return remote address if connected
     * @throws ClosedChannelException if the channel is closed
     * @throws NotYetConnectedException if mustBeConnected and not connected
     * @throws IOException if socket not bound and cannot be bound
     */
    private SocketAddress beginRead(boolean blocking, boolean mustBeConnected)
        throws IOException
    {
        if (blocking && interruptible) {
            // set hook for Thread.interrupt
            begin();
        }
        SocketAddress remote;
        synchronized (stateLock) {
            ensureOpen();
            remote = remoteAddress;
            if ((remote == null) && mustBeConnected)
                throw new NotYetConnectedException();
            if (localAddress == null)
                bindInternal(null);
            if (blocking)
                readerThread = NativeThread.current();
        }
        return remote;
    }

    
Marks the end of a read operation that may have blocked.
Throws:
AsynchronousCloseException – if the channel was closed asynchronously/**
     * Marks the end of a read operation that may have blocked.
     *
     * @throws AsynchronousCloseException if the channel was closed asynchronously
     */
    private void endRead(boolean blocking, boolean completed)
        throws AsynchronousCloseException
    {
        if (blocking) {
            synchronized (stateLock) {
                readerThread = 0;
                if (state == ST_CLOSING) {
                    tryFinishClose();
                }
            }
            if (interruptible) {
                // remove hook for Thread.interrupt (may throw AsynchronousCloseException)
                end(completed);
            } else if (!completed && !isOpen()) {
                throw new AsynchronousCloseException();
            }
        }
    }

    @Override
    public SocketAddress receive(ByteBuffer dst) throws IOException {
        if (dst.isReadOnly())
            throw new IllegalArgumentException("Read-only buffer");
        readLock.lock();
        try {
            boolean blocking = isBlocking();
            SocketAddress sender = null;
            try {
                SocketAddress remote = beginRead(blocking, false);
                boolean connected = (remote != null);
                SecurityManager sm = System.getSecurityManager();
                if (connected || (sm == null)) {
                    // connected or no security manager
                    int n = receive(dst, connected);
                    if (blocking) {
                        while (IOStatus.okayToRetry(n) && isOpen()) {
                            park(Net.POLLIN);
                            n = receive(dst, connected);
                        }
                    }
                    if (n >= 0) {
                        // sender address is in socket address buffer
                        sender = sourceSocketAddress();
                    }
                } else {
                    // security manager and unconnected
                    sender = untrustedReceive(dst);
                }
                return sender;
            } finally {
                endRead(blocking, (sender != null));
            }
        } finally {
            readLock.unlock();
        }
    }

    
Receives a datagram into an untrusted buffer. When there is a security
manager set, and the socket is not connected, datagrams have to be received
into a buffer that is not accessible to the user. The datagram is copied
into the user's buffer when the sender address is accepted by the security
manager.
/**
     * Receives a datagram into an untrusted buffer. When there is a security
     * manager set, and the socket is not connected, datagrams have to be received
     * into a buffer that is not accessible to the user. The datagram is copied
     * into the user's buffer when the sender address is accepted by the security
     * manager.
     */
    private SocketAddress untrustedReceive(ByteBuffer dst) throws IOException {
        SecurityManager sm = System.getSecurityManager();
        assert readLock.isHeldByCurrentThread()
                && sm != null && remoteAddress == null;

        ByteBuffer bb = Util.getTemporaryDirectBuffer(dst.remaining());
        try {
            boolean blocking = isBlocking();
            for (;;) {
                int n = receive(bb, false);
                if (blocking) {
                    while (IOStatus.okayToRetry(n) && isOpen()) {
                        park(Net.POLLIN);
                        n = receive(bb, false);
                    }
                }
                if (n >= 0) {
                    // sender address is in socket address buffer
                    InetSocketAddress isa = sourceSocketAddress();
                    try {
                        sm.checkAccept(isa.getAddress().getHostAddress(), isa.getPort());
                        bb.flip();
                        dst.put(bb);
                        return isa;
                    } catch (SecurityException se) {
                        // ignore datagram
                        bb.clear();
                    }
                } else {
                    return null;
                }
            }
        } finally {
            Util.releaseTemporaryDirectBuffer(bb);
        }
    }

    
Receives a datagram into the given buffer.
Throws:
IllegalBlockingModeException – if the channel is non-blocking
SocketTimeoutException – if the timeout elapses
API Note:
This method is for use by the socket adaptor. The buffer is
assumed to be trusted, meaning it is not accessible to user code./**
     * Receives a datagram into the given buffer.
     *
     * @apiNote This method is for use by the socket adaptor. The buffer is
     * assumed to be trusted, meaning it is not accessible to user code.
     *
     * @throws IllegalBlockingModeException if the channel is non-blocking
     * @throws SocketTimeoutException if the timeout elapses
     */
    SocketAddress blockingReceive(ByteBuffer dst, long nanos) throws IOException {
        readLock.lock();
        try {
            ensureOpen();
            if (!isBlocking())
                throw new IllegalBlockingModeException();
            SecurityManager sm = System.getSecurityManager();
            boolean connected = isConnected();
            SocketAddress sender;
            do {
                if (nanos > 0) {
                    sender = trustedBlockingReceive(dst, nanos);
                } else {
                    sender = trustedBlockingReceive(dst);
                }
                // check sender when security manager set and not connected
                if (sm != null && !connected) {
                    InetSocketAddress isa = (InetSocketAddress) sender;
                    try {
                        sm.checkAccept(isa.getAddress().getHostAddress(), isa.getPort());
                    } catch (SecurityException e) {
                        sender = null;
                    }
                }
            } while (sender == null);
            return sender;
        } finally {
            readLock.unlock();
        }
    }

    
Receives a datagram into given buffer. This method is used to support
the socket adaptor. The buffer is assumed to be trusted.
Throws:
SocketTimeoutException – if the timeout elapses/**
     * Receives a datagram into given buffer. This method is used to support
     * the socket adaptor. The buffer is assumed to be trusted.
     * @throws SocketTimeoutException if the timeout elapses
     */
    private SocketAddress trustedBlockingReceive(ByteBuffer dst)
        throws IOException
    {
        assert readLock.isHeldByCurrentThread() && isBlocking();
        SocketAddress sender = null;
        try {
            SocketAddress remote = beginRead(true, false);
            boolean connected = (remote != null);
            int n = receive(dst, connected);
            while (IOStatus.okayToRetry(n) && isOpen()) {
                park(Net.POLLIN);
                n = receive(dst, connected);
            }
            if (n >= 0) {
                // sender address is in socket address buffer
                sender = sourceSocketAddress();
            }
            return sender;
        } finally {
            endRead(true, (sender != null));
        }
    }

    
Receives a datagram into given buffer with a timeout. This method is
used to support the socket adaptor. The buffer is assumed to be trusted.
Throws:
SocketTimeoutException – if the timeout elapses/**
     * Receives a datagram into given buffer with a timeout. This method is
     * used to support the socket adaptor. The buffer is assumed to be trusted.
     * @throws SocketTimeoutException if the timeout elapses
     */
    private SocketAddress trustedBlockingReceive(ByteBuffer dst, long nanos)
        throws IOException
    {
        assert readLock.isHeldByCurrentThread() && isBlocking();
        SocketAddress sender = null;
        try {
            SocketAddress remote = beginRead(true, false);
            boolean connected = (remote != null);

            // change socket to non-blocking
            lockedConfigureBlocking(false);
            try {
                long startNanos = System.nanoTime();
                int n = receive(dst, connected);
                while (n == IOStatus.UNAVAILABLE && isOpen()) {
                    long remainingNanos = nanos - (System.nanoTime() - startNanos);
                    if (remainingNanos <= 0) {
                        throw new SocketTimeoutException("Receive timed out");
                    }
                    park(Net.POLLIN, remainingNanos);
                    n = receive(dst, connected);
                }
                if (n >= 0) {
                    // sender address is in socket address buffer
                    sender = sourceSocketAddress();
                }
                return sender;
            } finally {
                // restore socket to blocking mode (if channel is open)
                tryLockedConfigureBlocking(true);
            }
        } finally {
            endRead(true, (sender != null));
        }
    }

    private int receive(ByteBuffer dst, boolean connected) throws IOException {
        int pos = dst.position();
        int lim = dst.limit();
        assert (pos <= lim);
        int rem = (pos <= lim ? lim - pos : 0);
        if (dst instanceof DirectBuffer && rem > 0)
            return receiveIntoNativeBuffer(dst, rem, pos, connected);

        // Substitute a native buffer. If the supplied buffer is empty
        // we must instead use a nonempty buffer, otherwise the call
        // will not block waiting for a datagram on some platforms.
        int newSize = Math.max(rem, 1);
        ByteBuffer bb = Util.getTemporaryDirectBuffer(newSize);
        try {
            int n = receiveIntoNativeBuffer(bb, newSize, 0, connected);
            bb.flip();
            if (n > 0 && rem > 0)
                dst.put(bb);
            return n;
        } finally {
            Util.releaseTemporaryDirectBuffer(bb);
        }
    }

    private int receiveIntoNativeBuffer(ByteBuffer bb, int rem, int pos,
                                        boolean connected)
        throws IOException
    {
        int n = receive0(fd,
                         ((DirectBuffer)bb).address() + pos, rem,
                         sourceSockAddr.address(),
                         connected);
        if (n > 0)
            bb.position(pos + n);
        return n;
    }

    
Return an InetSocketAddress to represent the source/sender socket address
in sourceSockAddr. Returns the cached InetSocketAddress if the source
address is the same as the cached address.
/**
     * Return an InetSocketAddress to represent the source/sender socket address
     * in sourceSockAddr. Returns the cached InetSocketAddress if the source
     * address is the same as the cached address.
     */
    private InetSocketAddress sourceSocketAddress() throws IOException {
        assert readLock.isHeldByCurrentThread();
        if (cachedInetSocketAddress != null && sourceSockAddr.equals(cachedSockAddr)) {
            return cachedInetSocketAddress;
        }
        InetSocketAddress isa = sourceSockAddr.decode();
        // swap sourceSockAddr and cachedSockAddr
        NativeSocketAddress tmp = cachedSockAddr;
        cachedSockAddr = sourceSockAddr;
        sourceSockAddr = tmp;
        cachedInetSocketAddress = isa;
        return isa;
    }

    @Override
    public int send(ByteBuffer src, SocketAddress target)
        throws IOException
    {
        Objects.requireNonNull(src);
        InetSocketAddress isa = Net.checkAddress(target, family);

        writeLock.lock();
        try {
            boolean blocking = isBlocking();
            int n;
            boolean completed = false;
            try {
                SocketAddress remote = beginWrite(blocking, false);
                if (remote != null) {
                    // connected
                    if (!target.equals(remote)) {
                        throw new AlreadyConnectedException();
                    }
                    n = IOUtil.write(fd, src, -1, nd);
                    if (blocking) {
                        while (IOStatus.okayToRetry(n) && isOpen()) {
                            park(Net.POLLOUT);
                            n = IOUtil.write(fd, src, -1, nd);
                        }
                    }
                    completed = (n > 0);
                } else {
                    // not connected
                    SecurityManager sm = System.getSecurityManager();
                    InetAddress ia = isa.getAddress();
                    if (sm != null) {
                        if (ia.isMulticastAddress()) {
                            sm.checkMulticast(ia);
                        } else {
                            sm.checkConnect(ia.getHostAddress(), isa.getPort());
                        }
                    }
                    if (ia.isLinkLocalAddress())
                        isa = IPAddressUtil.toScopedAddress(isa);
                    if (isa.getPort() == 0)
                        throw new SocketException("Can't send to port 0");
                    n = send(fd, src, isa);
                    if (blocking) {
                        while (IOStatus.okayToRetry(n) && isOpen()) {
                            park(Net.POLLOUT);
                            n = send(fd, src, isa);
                        }
                    }
                    completed = (n >= 0);
                }
            } finally {
                endWrite(blocking, completed);
            }
            assert n >= 0 || n == IOStatus.UNAVAILABLE;
            return IOStatus.normalize(n);
        } finally {
            writeLock.unlock();
        }
    }

    
Sends a datagram from the bytes in given buffer.
Throws:
IllegalBlockingModeException – if the channel is non-blocking
API Note:
This method is for use by the socket adaptor./**
     * Sends a datagram from the bytes in given buffer.
     *
     * @apiNote This method is for use by the socket adaptor.
     *
     * @throws IllegalBlockingModeException if the channel is non-blocking
     */
    void blockingSend(ByteBuffer src, SocketAddress target) throws IOException {
        writeLock.lock();
        try {
            ensureOpen();
            if (!isBlocking())
                throw new IllegalBlockingModeException();
            send(src, target);
        } finally {
            writeLock.unlock();
        }
    }

    private int send(FileDescriptor fd, ByteBuffer src, InetSocketAddress target)
        throws IOException
    {
        if (src instanceof DirectBuffer)
            return sendFromNativeBuffer(fd, src, target);

        // Substitute a native buffer
        int pos = src.position();
        int lim = src.limit();
        assert (pos <= lim);
        int rem = (pos <= lim ? lim - pos : 0);

        ByteBuffer bb = Util.getTemporaryDirectBuffer(rem);
        try {
            bb.put(src);
            bb.flip();
            // Do not update src until we see how many bytes were written
            src.position(pos);

            int n = sendFromNativeBuffer(fd, bb, target);
            if (n > 0) {
                // now update src
                src.position(pos + n);
            }
            return n;
        } finally {
            Util.releaseTemporaryDirectBuffer(bb);
        }
    }

    private int sendFromNativeBuffer(FileDescriptor fd, ByteBuffer bb,
                                     InetSocketAddress target)
        throws IOException
    {
        int pos = bb.position();
        int lim = bb.limit();
        assert (pos <= lim);
        int rem = (pos <= lim ? lim - pos : 0);

        int written;
        try {
            int addressLen = targetSocketAddress(target);
            written = send0(fd, ((DirectBuffer)bb).address() + pos, rem,
                            targetSockAddr.address(), addressLen);
        } catch (PortUnreachableException pue) {
            if (isConnected())
                throw pue;
            written = rem;
        }
        if (written > 0)
            bb.position(pos + written);
        return written;
    }

    
Encodes the given InetSocketAddress into targetSockAddr, returning the
length of the sockaddr structure (sizeof struct sockaddr or sockaddr6).
/**
     * Encodes the given InetSocketAddress into targetSockAddr, returning the
     * length of the sockaddr structure (sizeof struct sockaddr or sockaddr6).
     */
    private int targetSocketAddress(InetSocketAddress isa) {
        assert writeLock.isHeldByCurrentThread();
        // Nothing to do if target address is already in the buffer. Use
        // identity rather than equals as Inet6Address.equals ignores scope_id.
        if (isa == previousTarget)
            return previousSockAddrLength;
        previousTarget = null;
        int len = targetSockAddr.encode(family, isa);
        previousTarget = isa;
        previousSockAddrLength = len;
        return len;
    }

    @Override
    public int read(ByteBuffer buf) throws IOException {
        Objects.requireNonNull(buf);

        readLock.lock();
        try {
            boolean blocking = isBlocking();
            int n = 0;
            try {
                beginRead(blocking, true);
                n = IOUtil.read(fd, buf, -1, nd);
                if (blocking) {
                    while (IOStatus.okayToRetry(n) && isOpen()) {
                        park(Net.POLLIN);
                        n = IOUtil.read(fd, buf, -1, nd);
                    }
                }
            } finally {
                endRead(blocking, n > 0);
                assert IOStatus.check(n);
            }
            return IOStatus.normalize(n);
        } finally {
            readLock.unlock();
        }
    }

    @Override
    public long read(ByteBuffer[] dsts, int offset, int length)
        throws IOException
    {
        Objects.checkFromIndexSize(offset, length, dsts.length);

        readLock.lock();
        try {
            boolean blocking = isBlocking();
            long n = 0;
            try {
                beginRead(blocking, true);
                n = IOUtil.read(fd, dsts, offset, length, nd);
                if (blocking) {
                    while (IOStatus.okayToRetry(n)  && isOpen()) {
                        park(Net.POLLIN);
                        n = IOUtil.read(fd, dsts, offset, length, nd);
                    }
                }
            } finally {
                endRead(blocking, n > 0);
                assert IOStatus.check(n);
            }
            return IOStatus.normalize(n);
        } finally {
            readLock.unlock();
        }
    }

    
Marks the beginning of a write operation that might block.
Params:
blocking – true if configured blocking
mustBeConnected – true if the socket must be connected
Throws:
ClosedChannelException – if the channel is closed
NotYetConnectedException – if mustBeConnected and not connected
IOException – if socket not bound and cannot be bound
Returns:
remote address if connected/**
     * Marks the beginning of a write operation that might block.
     * @param blocking true if configured blocking
     * @param mustBeConnected true if the socket must be connected
     * @return remote address if connected
     * @throws ClosedChannelException if the channel is closed
     * @throws NotYetConnectedException if mustBeConnected and not connected
     * @throws IOException if socket not bound and cannot be bound
     */
    private SocketAddress beginWrite(boolean blocking, boolean mustBeConnected)
        throws IOException
    {
        if (blocking && interruptible) {
            // set hook for Thread.interrupt
            begin();
        }
        SocketAddress remote;
        synchronized (stateLock) {
            ensureOpen();
            remote = remoteAddress;
            if ((remote == null) && mustBeConnected)
                throw new NotYetConnectedException();
            if (localAddress == null)
                bindInternal(null);
            if (blocking)
                writerThread = NativeThread.current();
        }
        return remote;
    }

    
Marks the end of a write operation that may have blocked.
Throws:
AsynchronousCloseException – if the channel was closed asynchronously/**
     * Marks the end of a write operation that may have blocked.
     *
     * @throws AsynchronousCloseException if the channel was closed asynchronously
     */
    private void endWrite(boolean blocking, boolean completed)
        throws AsynchronousCloseException
    {
        if (blocking) {
            synchronized (stateLock) {
                writerThread = 0;
                if (state == ST_CLOSING) {
                    tryFinishClose();
                }
            }

            if (interruptible) {
                // remove hook for Thread.interrupt (may throw AsynchronousCloseException)
                end(completed);
            } else if (!completed && !isOpen()) {
                throw new AsynchronousCloseException();
            }
        }
    }

    @Override
    public int write(ByteBuffer buf) throws IOException {
        Objects.requireNonNull(buf);

        writeLock.lock();
        try {
            boolean blocking = isBlocking();
            int n = 0;
            try {
                beginWrite(blocking, true);
                n = IOUtil.write(fd, buf, -1, nd);
                if (blocking) {
                    while (IOStatus.okayToRetry(n) && isOpen()) {
                        park(Net.POLLOUT);
                        n = IOUtil.write(fd, buf, -1, nd);
                    }
                }
            } finally {
                endWrite(blocking, n > 0);
                assert IOStatus.check(n);
            }
            return IOStatus.normalize(n);
        } finally {
            writeLock.unlock();
        }
    }

    @Override
    public long write(ByteBuffer[] srcs, int offset, int length)
        throws IOException
    {
        Objects.checkFromIndexSize(offset, length, srcs.length);

        writeLock.lock();
        try {
            boolean blocking = isBlocking();
            long n = 0;
            try {
                beginWrite(blocking, true);
                n = IOUtil.write(fd, srcs, offset, length, nd);
                if (blocking) {
                    while (IOStatus.okayToRetry(n) && isOpen()) {
                        park(Net.POLLOUT);
                        n = IOUtil.write(fd, srcs, offset, length, nd);
                    }
                }
            } finally {
                endWrite(blocking, n > 0);
                assert IOStatus.check(n);
            }
            return IOStatus.normalize(n);
        } finally {
            writeLock.unlock();
        }
    }

    @Override
    protected void implConfigureBlocking(boolean block) throws IOException {
        readLock.lock();
        try {
            writeLock.lock();
            try {
                lockedConfigureBlocking(block);
            } finally {
                writeLock.unlock();
            }
        } finally {
            readLock.unlock();
        }
    }

    
Adjusts the blocking mode. readLock or writeLock must already be held.
/**
     * Adjusts the blocking mode. readLock or writeLock must already be held.
     */
    private void lockedConfigureBlocking(boolean block) throws IOException {
        assert readLock.isHeldByCurrentThread() || writeLock.isHeldByCurrentThread();
        synchronized (stateLock) {
            ensureOpen();
            IOUtil.configureBlocking(fd, block);
        }
    }

    
Adjusts the blocking mode if the channel is open. readLock or writeLock
must already be held.
Returns:
true if the blocking mode was adjusted, false if the blocking mode was not adjusted because the channel is closed/**
     * Adjusts the blocking mode if the channel is open. readLock or writeLock
     * must already be held.
     *
     * @return {@code true} if the blocking mode was adjusted, {@code false} if
     *         the blocking mode was not adjusted because the channel is closed
     */
    private boolean tryLockedConfigureBlocking(boolean block) throws IOException {
        assert readLock.isHeldByCurrentThread() || writeLock.isHeldByCurrentThread();
        synchronized (stateLock) {
            if (isOpen()) {
                IOUtil.configureBlocking(fd, block);
                return true;
            } else {
                return false;
            }
        }
    }

    InetSocketAddress localAddress() {
        synchronized (stateLock) {
            return localAddress;
        }
    }

    InetSocketAddress remoteAddress() {
        synchronized (stateLock) {
            return remoteAddress;
        }
    }

    @Override
    public DatagramChannel bind(SocketAddress local) throws IOException {
        readLock.lock();
        try {
            writeLock.lock();
            try {
                synchronized (stateLock) {
                    ensureOpen();
                    if (localAddress != null)
                        throw new AlreadyBoundException();
                    bindInternal(local);
                }
            } finally {
                writeLock.unlock();
            }
        } finally {
            readLock.unlock();
        }
        return this;
    }

    private void bindInternal(SocketAddress local) throws IOException {
        assert Thread.holdsLock(stateLock )&& (localAddress == null);

        InetSocketAddress isa;
        if (local == null) {
            // only Inet4Address allowed with IPv4 socket
            if (family == StandardProtocolFamily.INET) {
                isa = new InetSocketAddress(InetAddress.getByName("0.0.0.0"), 0);
            } else {
                isa = new InetSocketAddress(0);
            }
        } else {
            isa = Net.checkAddress(local, family);
        }
        SecurityManager sm = System.getSecurityManager();
        if (sm != null)
            sm.checkListen(isa.getPort());

        Net.bind(family, fd, isa.getAddress(), isa.getPort());
        localAddress = Net.localAddress(fd);
    }

    @Override
    public boolean isConnected() {
        synchronized (stateLock) {
            return (state == ST_CONNECTED);
        }
    }

    @Override
    public DatagramChannel connect(SocketAddress sa) throws IOException {
        return connect(sa, true);
    }

    
Connects the channel's socket.
Params:
sa – the remote address to which this channel is to be connected
check – true to check if the channel is already connected./**
     * Connects the channel's socket.
     *
     * @param sa the remote address to which this channel is to be connected
     * @param check true to check if the channel is already connected.
     */
    DatagramChannel connect(SocketAddress sa, boolean check) throws IOException {
        InetSocketAddress isa = Net.checkAddress(sa, family);
        SecurityManager sm = System.getSecurityManager();
        if (sm != null) {
            InetAddress ia = isa.getAddress();
            if (ia.isMulticastAddress()) {
                sm.checkMulticast(ia);
            } else {
                sm.checkConnect(ia.getHostAddress(), isa.getPort());
                sm.checkAccept(ia.getHostAddress(), isa.getPort());
            }
        }

        readLock.lock();
        try {
            writeLock.lock();
            try {
                synchronized (stateLock) {
                    ensureOpen();
                    if (check && state == ST_CONNECTED)
                        throw new AlreadyConnectedException();
                    if (isa.getPort() == 0)
                        throw new SocketException("Can't connect to port 0");

                    // ensure that the socket is bound
                    if (localAddress == null) {
                        bindInternal(null);
                    }

                    // capture local address before connect
                    initialLocalAddress = localAddress;

                    int n = Net.connect(family,
                                        fd,
                                        isa.getAddress(),
                                        isa.getPort());
                    if (n <= 0)
                        throw new Error();      // Can't happen

                    // connected
                    remoteAddress = isa;
                    state = ST_CONNECTED;

                    // refresh local address
                    localAddress = Net.localAddress(fd);

                    // flush any packets already received.
                    boolean blocking = isBlocking();
                    if (blocking) {
                        IOUtil.configureBlocking(fd, false);
                    }
                    try {
                        ByteBuffer buf = ByteBuffer.allocate(100);
                        while (receive(buf, false) >= 0) {
                            buf.clear();
                        }
                    } finally {
                        if (blocking) {
                            IOUtil.configureBlocking(fd, true);
                        }
                    }
                }
            } finally {
                writeLock.unlock();
            }
        } finally {
            readLock.unlock();
        }
        return this;
    }

    @Override
    public DatagramChannel disconnect() throws IOException {
        readLock.lock();
        try {
            writeLock.lock();
            try {
                synchronized (stateLock) {
                    if (!isOpen() || (state != ST_CONNECTED))
                        return this;

                    // disconnect socket
                    boolean isIPv6 = (family == StandardProtocolFamily.INET6);
                    disconnect0(fd, isIPv6);

                    // no longer connected
                    remoteAddress = null;
                    state = ST_UNCONNECTED;

                    // refresh localAddress, should be same as it was prior to connect
                    localAddress = Net.localAddress(fd);
                    try {
                        if (!localAddress.equals(initialLocalAddress)) {
                            // Workaround connect(2) issues on Linux and macOS
                            repairSocket(initialLocalAddress);
                            assert (localAddress != null)
                                    && localAddress.equals(Net.localAddress(fd))
                                    && localAddress.equals(initialLocalAddress);
                        }
                    } finally {
                        initialLocalAddress = null;
                    }
                }
            } finally {
                writeLock.unlock();
            }
        } finally {
            readLock.unlock();
        }
        return this;
    }

    
"Repair" the channel's socket after a disconnect that didn't restore the
local address.
On Linux, connect(2) dissolves the association but changes the local port
to 0 when it was initially bound to an ephemeral port. The workaround here
is to rebind to the original port.
On macOS, connect(2) dissolves the association but rebinds the socket to
the wildcard address when it was initially bound to a specific address.
The workaround here is to re-create the socket.
/**
     * "Repair" the channel's socket after a disconnect that didn't restore the
     * local address.
     *
     * On Linux, connect(2) dissolves the association but changes the local port
     * to 0 when it was initially bound to an ephemeral port. The workaround here
     * is to rebind to the original port.
     *
     * On macOS, connect(2) dissolves the association but rebinds the socket to
     * the wildcard address when it was initially bound to a specific address.
     * The workaround here is to re-create the socket.
     */
    private void repairSocket(InetSocketAddress target)
        throws IOException
    {
        assert Thread.holdsLock(stateLock);

        // Linux: try to bind the socket to the original address/port
        if (localAddress.getPort() == 0) {
            assert localAddress.getAddress().equals(target.getAddress());
            Net.bind(family, fd, target.getAddress(), target.getPort());
            localAddress = Net.localAddress(fd);
            return;
        }

        // capture the value of all existing socket options
        Map<SocketOption<?>, Object> map = new HashMap<>();
        for (SocketOption<?> option : supportedOptions()) {
            Object value = getOption(option);
            if (value != null) {
                map.put(option, value);
            }
        }

        // macOS: re-create the socket.
        FileDescriptor newfd = Net.socket(family, false);
        try {
            // copy the socket options that are protocol family agnostic
            for (Map.Entry<SocketOption<?>, Object> e : map.entrySet()) {
                SocketOption<?> option = e.getKey();
                if (SocketOptionRegistry.findOption(option, Net.UNSPEC) != null) {
                    Object value = e.getValue();
                    try {
                        Net.setSocketOption(newfd, Net.UNSPEC, option, value);
                    } catch (IOException ignore) { }
                }
            }

            // copy the blocking mode
            if (!isBlocking()) {
                IOUtil.configureBlocking(newfd, false);
            }

            // dup this channel's socket to the new socket. If this succeeds then
            // fd will reference the new socket. If it fails then it will still
            // reference the old socket.
            nd.dup(newfd, fd);
        } finally {
            // release the file descriptor
            nd.close(newfd);
        }

        // bind to the original local address
        try {
            Net.bind(family, fd, target.getAddress(), target.getPort());
        } catch (IOException ioe) {
            // bind failed, socket is left unbound
            localAddress = null;
            throw ioe;
        }

        // restore local address
        localAddress = Net.localAddress(fd);

        // restore all socket options (including those set in first pass)
        for (Map.Entry<SocketOption<?>, Object> e : map.entrySet()) {
            @SuppressWarnings("unchecked")
            SocketOption<Object> option = (SocketOption<Object>) e.getKey();
            Object value = e.getValue();
            try {
                setOption(option, value);
            } catch (IOException ignore) { }
        }

        // restore multicast group membership
        MembershipRegistry registry = this.registry;
        if (registry != null) {
            registry.forEach(k -> {
                if (k instanceof MembershipKeyImpl.Type6) {
                    MembershipKeyImpl.Type6 key6 = (MembershipKeyImpl.Type6) k;
                    Net.join6(fd, key6.groupAddress(), key6.index(), key6.source());
                } else {
                    MembershipKeyImpl.Type4 key4 = (MembershipKeyImpl.Type4) k;
                    Net.join4(fd, key4.groupAddress(), key4.interfaceAddress(), key4.source());
                }
            });
        }

        // reset registration in all Selectors that this channel is registered with
        AbstractSelectableChannels.forEach(this, SelectionKeyImpl::reset);
    }

    
Defines static methods to access AbstractSelectableChannel non-public members.
/**
     * Defines static methods to access AbstractSelectableChannel non-public members.
     */
    private static class AbstractSelectableChannels {
        private static final Method FOREACH;
        static {
            try {
                PrivilegedExceptionAction<Method> pae = () -> {
                    Method m = AbstractSelectableChannel.class.getDeclaredMethod("forEach", Consumer.class);
                    m.setAccessible(true);
                    return m;
                };
                FOREACH = AccessController.doPrivileged(pae);
            } catch (Exception e) {
                throw new InternalError(e);
            }
        }
        static void forEach(AbstractSelectableChannel ch, Consumer<SelectionKeyImpl> action) {
            try {
                FOREACH.invoke(ch, action);
            } catch (Exception e) {
                throw new InternalError(e);
            }
        }
    }

    
Joins channel's socket to the given group/interface and
optional source address.
/**
     * Joins channel's socket to the given group/interface and
     * optional source address.
     */
    private MembershipKey innerJoin(InetAddress group,
                                    NetworkInterface interf,
                                    InetAddress source)
        throws IOException
    {
        if (!group.isMulticastAddress())
            throw new IllegalArgumentException("Group not a multicast address");

        // check multicast address is compatible with this socket
        if (group instanceof Inet4Address) {
            if (family == StandardProtocolFamily.INET6 && !Net.canIPv6SocketJoinIPv4Group())
                throw new IllegalArgumentException("IPv6 socket cannot join IPv4 multicast group");
        } else if (group instanceof Inet6Address) {
            if (family != StandardProtocolFamily.INET6)
                throw new IllegalArgumentException("Only IPv6 sockets can join IPv6 multicast group");
        } else {
            throw new IllegalArgumentException("Address type not supported");
        }

        // check source address
        if (source != null) {
            if (source.isAnyLocalAddress())
                throw new IllegalArgumentException("Source address is a wildcard address");
            if (source.isMulticastAddress())
                throw new IllegalArgumentException("Source address is multicast address");
            if (source.getClass() != group.getClass())
                throw new IllegalArgumentException("Source address is different type to group");
        }

        SecurityManager sm = System.getSecurityManager();
        if (sm != null)
            sm.checkMulticast(group);

        synchronized (stateLock) {
            ensureOpen();

            // check the registry to see if we are already a member of the group
            if (registry == null) {
                registry = new MembershipRegistry();
            } else {
                // return existing membership key
                MembershipKey key = registry.checkMembership(group, interf, source);
                if (key != null)
                    return key;
            }

            MembershipKeyImpl key;
            if ((family == StandardProtocolFamily.INET6) &&
                ((group instanceof Inet6Address) || Net.canJoin6WithIPv4Group()))
            {
                int index = interf.getIndex();
                if (index == -1)
                    throw new IOException("Network interface cannot be identified");

                // need multicast and source address as byte arrays
                byte[] groupAddress = Net.inet6AsByteArray(group);
                byte[] sourceAddress = (source == null) ? null :
                    Net.inet6AsByteArray(source);

                // join the group
                int n = Net.join6(fd, groupAddress, index, sourceAddress);
                if (n == IOStatus.UNAVAILABLE)
                    throw new UnsupportedOperationException();

                key = new MembershipKeyImpl.Type6(this, group, interf, source,
                                                  groupAddress, index, sourceAddress);

            } else {
                // need IPv4 address to identify interface
                Inet4Address target = Net.anyInet4Address(interf);
                if (target == null)
                    throw new IOException("Network interface not configured for IPv4");

                int groupAddress = Net.inet4AsInt(group);
                int targetAddress = Net.inet4AsInt(target);
                int sourceAddress = (source == null) ? 0 : Net.inet4AsInt(source);

                // join the group
                int n = Net.join4(fd, groupAddress, targetAddress, sourceAddress);
                if (n == IOStatus.UNAVAILABLE)
                    throw new UnsupportedOperationException();

                key = new MembershipKeyImpl.Type4(this, group, interf, source,
                                                  groupAddress, targetAddress, sourceAddress);
            }

            registry.add(key);
            return key;
        }
    }

    @Override
    public MembershipKey join(InetAddress group,
                              NetworkInterface interf)
        throws IOException
    {
        return innerJoin(group, interf, null);
    }

    @Override
    public MembershipKey join(InetAddress group,
                              NetworkInterface interf,
                              InetAddress source)
        throws IOException
    {
        Objects.requireNonNull(source);
        return innerJoin(group, interf, source);
    }

    // package-private
    void drop(MembershipKeyImpl key) {
        assert key.channel() == this;

        synchronized (stateLock) {
            if (!key.isValid())
                return;

            try {
                if (key instanceof MembershipKeyImpl.Type6) {
                    MembershipKeyImpl.Type6 key6 =
                        (MembershipKeyImpl.Type6)key;
                    Net.drop6(fd, key6.groupAddress(), key6.index(), key6.source());
                } else {
                    MembershipKeyImpl.Type4 key4 = (MembershipKeyImpl.Type4)key;
                    Net.drop4(fd, key4.groupAddress(), key4.interfaceAddress(),
                        key4.source());
                }
            } catch (IOException ioe) {
                // should not happen
                throw new AssertionError(ioe);
            }

            key.invalidate();
            registry.remove(key);
        }
    }

    
Finds an existing membership of a multicast group. Returns null if this
channel's socket is not a member of the group.
API Note:
This method is for use by the socket adaptor/**
     * Finds an existing membership of a multicast group. Returns null if this
     * channel's socket is not a member of the group.
     *
     * @apiNote This method is for use by the socket adaptor
     */
    MembershipKey findMembership(InetAddress group, NetworkInterface interf) {
        synchronized (stateLock) {
            if (registry != null) {
                return registry.checkMembership(group, interf, null);
            } else {
                return null;
            }
        }
    }

    
Block datagrams from the given source.
/**
     * Block datagrams from the given source.
     */
    void block(MembershipKeyImpl key, InetAddress source)
        throws IOException
    {
        assert key.channel() == this;
        assert key.sourceAddress() == null;

        synchronized (stateLock) {
            if (!key.isValid())
                throw new IllegalStateException("key is no longer valid");
            if (source.isAnyLocalAddress())
                throw new IllegalArgumentException("Source address is a wildcard address");
            if (source.isMulticastAddress())
                throw new IllegalArgumentException("Source address is multicast address");
            if (source.getClass() != key.group().getClass())
                throw new IllegalArgumentException("Source address is different type to group");

            int n;
            if (key instanceof MembershipKeyImpl.Type6) {
                 MembershipKeyImpl.Type6 key6 =
                    (MembershipKeyImpl.Type6)key;
                n = Net.block6(fd, key6.groupAddress(), key6.index(),
                               Net.inet6AsByteArray(source));
            } else {
                MembershipKeyImpl.Type4 key4 =
                    (MembershipKeyImpl.Type4)key;
                n = Net.block4(fd, key4.groupAddress(), key4.interfaceAddress(),
                               Net.inet4AsInt(source));
            }
            if (n == IOStatus.UNAVAILABLE) {
                // ancient kernel
                throw new UnsupportedOperationException();
            }
        }
    }

    
Unblock the given source.
/**
     * Unblock the given source.
     */
    void unblock(MembershipKeyImpl key, InetAddress source) {
        assert key.channel() == this;
        assert key.sourceAddress() == null;

        synchronized (stateLock) {
            if (!key.isValid())
                throw new IllegalStateException("key is no longer valid");

            try {
                if (key instanceof MembershipKeyImpl.Type6) {
                    MembershipKeyImpl.Type6 key6 =
                        (MembershipKeyImpl.Type6)key;
                    Net.unblock6(fd, key6.groupAddress(), key6.index(),
                                 Net.inet6AsByteArray(source));
                } else {
                    MembershipKeyImpl.Type4 key4 =
                        (MembershipKeyImpl.Type4)key;
                    Net.unblock4(fd, key4.groupAddress(), key4.interfaceAddress(),
                                 Net.inet4AsInt(source));
                }
            } catch (IOException ioe) {
                // should not happen
                throw new AssertionError(ioe);
            }
        }
    }

    
Closes the socket if there are no I/O operations in progress and the
channel is not registered with a Selector.
/**
     * Closes the socket if there are no I/O operations in progress and the
     * channel is not registered with a Selector.
     */
    private boolean tryClose() throws IOException {
        assert Thread.holdsLock(stateLock) && state == ST_CLOSING;
        if ((readerThread == 0) && (writerThread == 0) && !isRegistered()) {
            state = ST_CLOSED;
            try {
                // close socket
                cleaner.clean();
            } catch (UncheckedIOException ioe) {
                throw ioe.getCause();
            }
            return true;
        } else {
            return false;
        }
    }

    
Invokes tryClose to attempt to close the socket.
This method is used for deferred closing by I/O and Selector operations.
/**
     * Invokes tryClose to attempt to close the socket.
     *
     * This method is used for deferred closing by I/O and Selector operations.
     */
    private void tryFinishClose() {
        try {
            tryClose();
        } catch (IOException ignore) { }
    }

    
Closes this channel when configured in blocking mode.
If there is an I/O operation in progress then the socket is pre-closed
and the I/O threads signalled, in which case the final close is deferred
until all I/O operations complete.
/**
     * Closes this channel when configured in blocking mode.
     *
     * If there is an I/O operation in progress then the socket is pre-closed
     * and the I/O threads signalled, in which case the final close is deferred
     * until all I/O operations complete.
     */
    private void implCloseBlockingMode() throws IOException {
        synchronized (stateLock) {
            assert state < ST_CLOSING;
            state = ST_CLOSING;

            // if member of any multicast groups then invalidate the keys
            if (registry != null)
                registry.invalidateAll();

            if (!tryClose()) {
                long reader = readerThread;
                long writer = writerThread;
                if (reader != 0 || writer != 0) {
                    nd.preClose(fd);
                    if (reader != 0)
                        NativeThread.signal(reader);
                    if (writer != 0)
                        NativeThread.signal(writer);
                }
            }
        }
    }

    
Closes this channel when configured in non-blocking mode.
If the channel is registered with a Selector then the close is deferred
until the channel is flushed from all Selectors.
/**
     * Closes this channel when configured in non-blocking mode.
     *
     * If the channel is registered with a Selector then the close is deferred
     * until the channel is flushed from all Selectors.
     */
    private void implCloseNonBlockingMode() throws IOException {
        synchronized (stateLock) {
            assert state < ST_CLOSING;
            state = ST_CLOSING;

            // if member of any multicast groups then invalidate the keys
            if (registry != null)
                registry.invalidateAll();
        }

        // wait for any read/write operations to complete before trying to close
        readLock.lock();
        readLock.unlock();
        writeLock.lock();
        writeLock.unlock();
        synchronized (stateLock) {
            if (state == ST_CLOSING) {
                tryClose();
            }
        }
    }

    
Invoked by implCloseChannel to close the channel.
/**
     * Invoked by implCloseChannel to close the channel.
     */
    @Override
    protected void implCloseSelectableChannel() throws IOException {
        assert !isOpen();
        if (isBlocking()) {
            implCloseBlockingMode();
        } else {
            implCloseNonBlockingMode();
        }
    }

    @Override
    public void kill() {
        synchronized (stateLock) {
            if (state == ST_CLOSING) {
                tryFinishClose();
            }
        }
    }

    
Translates native poll revent set into a ready operation set
/**
     * Translates native poll revent set into a ready operation set
     */
    public boolean translateReadyOps(int ops, int initialOps, SelectionKeyImpl ski) {
        int intOps = ski.nioInterestOps();
        int oldOps = ski.nioReadyOps();
        int newOps = initialOps;

        if ((ops & Net.POLLNVAL) != 0) {
            // This should only happen if this channel is pre-closed while a
            // selection operation is in progress
            // ## Throw an error if this channel has not been pre-closed
            return false;
        }

        if ((ops & (Net.POLLERR | Net.POLLHUP)) != 0) {
            newOps = intOps;
            ski.nioReadyOps(newOps);
            return (newOps & ~oldOps) != 0;
        }

        if (((ops & Net.POLLIN) != 0) &&
            ((intOps & SelectionKey.OP_READ) != 0))
            newOps |= SelectionKey.OP_READ;

        if (((ops & Net.POLLOUT) != 0) &&
            ((intOps & SelectionKey.OP_WRITE) != 0))
            newOps |= SelectionKey.OP_WRITE;

        ski.nioReadyOps(newOps);
        return (newOps & ~oldOps) != 0;
    }

    public boolean translateAndUpdateReadyOps(int ops, SelectionKeyImpl ski) {
        return translateReadyOps(ops, ski.nioReadyOps(), ski);
    }

    public boolean translateAndSetReadyOps(int ops, SelectionKeyImpl ski) {
        return translateReadyOps(ops, 0, ski);
    }

    
Translates an interest operation set into a native poll event set
/**
     * Translates an interest operation set into a native poll event set
     */
    public int translateInterestOps(int ops) {
        int newOps = 0;
        if ((ops & SelectionKey.OP_READ) != 0)
            newOps |= Net.POLLIN;
        if ((ops & SelectionKey.OP_WRITE) != 0)
            newOps |= Net.POLLOUT;
        if ((ops & SelectionKey.OP_CONNECT) != 0)
            newOps |= Net.POLLIN;
        return newOps;
    }

    public FileDescriptor getFD() {
        return fd;
    }

    public int getFDVal() {
        return fdVal;
    }

    
Returns an action to release the given file descriptor and socket addresses.
/**
     * Returns an action to release the given file descriptor and socket addresses.
     */
    private static Runnable releaserFor(FileDescriptor fd, NativeSocketAddress... sockAddrs) {
        return () -> {
            try {
                nd.close(fd);
            } catch (IOException ioe) {
                throw new UncheckedIOException(ioe);
            } finally {
                // decrement socket count and release memory
                ResourceManager.afterUdpClose();
                NativeSocketAddress.freeAll(sockAddrs);
            }
        };
    }

    // -- Native methods --

    private static native void disconnect0(FileDescriptor fd, boolean isIPv6)
        throws IOException;

    private static native int receive0(FileDescriptor fd, long address, int len,
                                       long senderAddress, boolean connected)
        throws IOException;

    private static native int send0(FileDescriptor fd, long address, int len,
                                    long targetAddress, int targetAddressLen)
        throws IOException;

    static {
        IOUtil.load();
    }
}