/*

 * Copyright (c) 2018, 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.InputStream;

import java.io.OutputStream;

import java.lang.invoke.MethodHandles;

import java.lang.invoke.VarHandle;

import java.lang.reflect.Field;

import java.net.InetAddress;

import java.net.InetSocketAddress;

import java.net.ProtocolFamily;

import java.net.SocketAddress;

import java.net.SocketException;

import java.net.SocketImpl;

import java.net.SocketOption;

import java.net.SocketTimeoutException;

import java.net.StandardProtocolFamily;

import java.net.StandardSocketOptions;

import java.net.UnknownHostException;

import java.nio.ByteBuffer;

import java.security.AccessController;

import java.security.PrivilegedActionException;

import java.security.PrivilegedExceptionAction;

import java.util.Collections;

import java.util.HashSet;

import java.util.Objects;

import java.util.Set;

import java.util.concurrent.TimeUnit;

import java.util.concurrent.locks.ReentrantLock;

import jdk.internal.access.SharedSecrets;

import jdk.internal.ref.CleanerFactory;

import sun.net.ConnectionResetException;

import sun.net.NetHooks;

import sun.net.PlatformSocketImpl;

import sun.net.ResourceManager;

import sun.net.ext.ExtendedSocketOptions;

import sun.net.util.SocketExceptions;

import static java.util.concurrent.TimeUnit.MILLISECONDS;

import static java.util.concurrent.TimeUnit.NANOSECONDS;

/**

 * NIO based SocketImpl.

 *

 * This implementation attempts to be compatible with legacy PlainSocketImpl,

 * including behavior and exceptions that are not specified by SocketImpl.

 *

 * The underlying socket used by this SocketImpl is initially configured

 * blocking. If a connect, accept or read is attempted with a timeout then the

 * socket is changed to non-blocking mode. When in non-blocking mode, operations

 * that don't complete immediately will poll the socket.

 */

public final class NioSocketImpl extends SocketImpl implements PlatformSocketImpl {

    private static final NativeDispatcher nd = new SocketDispatcher(true);

    // The maximum number of bytes to read/write per syscall to avoid needing

    // a huge buffer from the temporary buffer cache

    private static final int MAX_BUFFER_SIZE = 128 * 1024;

    // true if this is a SocketImpl for a ServerSocket

    private final boolean server;

    // Lock held when reading (also used when accepting or connecting)

    private final ReentrantLock readLock = new ReentrantLock();

    // Lock held when writing

    private final ReentrantLock writeLock = new ReentrantLock();

    // The stateLock for read/changing state

    private final Object stateLock = new Object();

    private static final int ST_NEW = 0;

    private static final int ST_UNCONNECTED = 1;

    private static final int ST_CONNECTING = 2;

    private static final int ST_CONNECTED = 3;

    private static final int ST_CLOSING = 4;

    private static final int ST_CLOSED = 5;

    private volatile int state;  // need stateLock to change

    // set by SocketImpl.create, protected by stateLock

    private boolean stream;

    private FileDescriptorCloser closer;

    // lazily set to true when the socket is configured non-blocking

    private volatile boolean nonBlocking;

    // used by connect/read/write/accept, protected by stateLock

    private long readerThread;

    private long writerThread;

    // used when SO_REUSEADDR is emulated

    private boolean isReuseAddress;

    // read or accept timeout in millis

    private volatile int timeout;

    // flags to indicate if the connection is shutdown for input and output

    private volatile boolean isInputClosed;

    private volatile boolean isOutputClosed;

    // socket input/output streams

    private volatile InputStream in;

    private volatile OutputStream out;

    private static final VarHandle IN, OUT;

    static {

        try {

            MethodHandles.Lookup l = MethodHandles.lookup();

            IN = l.findVarHandle(NioSocketImpl.class, "in", InputStream.class);

            OUT = l.findVarHandle(NioSocketImpl.class, "out", OutputStream.class);

        } catch (Exception e) {

            throw new InternalError(e);

        }

    }

    /**

     * Creates a instance of this SocketImpl.

     * @param server true if this is a SocketImpl for a ServerSocket

     */

    public NioSocketImpl(boolean server) {

        this.server = server;

    }

    /**

     * Returns true if the socket is open.

     */

    private boolean isOpen() {

        return state < ST_CLOSING;

    }

    /**

     * Throws SocketException if the socket is not open.

     */

    private void ensureOpen() throws SocketException {

        if (state >= ST_CLOSING)

            throw new SocketException("Socket closed");

    }

    /**

     * Throws SocketException if the socket is not open and connected.

     */

    private void ensureOpenAndConnected() throws SocketException {

        int state = this.state;

        if (state < ST_CONNECTED)

            throw new SocketException("Not connected");

        if (state > ST_CONNECTED)

            throw new SocketException("Socket closed");

    }

    /**

     * Disables the current thread for scheduling purposes until the socket is

     * ready for I/O, or is asynchronously closed, for up to the specified

     * waiting time.

     * @throws IOException if an I/O error occurs

     */

    private void park(FileDescriptor fd, int event, long nanos) throws IOException {

        long millis;

        if (nanos == 0) {

            millis = -1;

        } else {

            millis = MILLISECONDS.convert(nanos, NANOSECONDS);

        }

        Net.poll(fd, event, millis);

    }

    /**

     * Disables the current thread for scheduling purposes until the socket is

     * ready for I/O or is asynchronously closed.

     * @throws IOException if an I/O error occurs

     */

    private void park(FileDescriptor fd, int event) throws IOException {

        park(fd, event, 0);

    }

    /**

     * Ensures that the socket is configured non-blocking when a timeout is specified.

     * @throws IOException if there is an I/O error changing the blocking mode

     */

    private void configureNonBlockingIfNeeded(FileDescriptor fd, int timeout)

        throws IOException

    {

        if (timeout > 0 && !nonBlocking) {

            assert readLock.isHeldByCurrentThread() || writeLock.isHeldByCurrentThread();

            IOUtil.configureBlocking(fd, false);

            nonBlocking = true;

        }

    }

    /**

     * Marks the beginning of a read operation that might block.

     * @throws SocketException if the socket is closed or not connected

     */

    private FileDescriptor beginRead() throws SocketException {

        synchronized (stateLock) {

            ensureOpenAndConnected();

            readerThread = NativeThread.current();

            return fd;

        }

    }

    /**

     * Marks the end of a read operation that may have blocked.

     * @throws SocketException is the socket is closed

     */

    private void endRead(boolean completed) throws SocketException {

        synchronized (stateLock) {

            readerThread = 0;

            int state = this.state;

            if (state == ST_CLOSING)

                stateLock.notifyAll();

            if (!completed && state >= ST_CLOSING)

                throw new SocketException("Socket closed");

        }

    }

    /**

     * Try to read bytes from the socket into the given byte array.

     */

    private int tryRead(FileDescriptor fd, byte[] b, int off, int len)

        throws IOException

    {

        ByteBuffer dst = Util.getTemporaryDirectBuffer(len);

        assert dst.position() == 0;

        try {

            int n = nd.read(fd, ((DirectBuffer)dst).address(), len);

            if (n > 0) {

                dst.get(b, off, n);

            }

            return n;

        } finally{

            Util.offerFirstTemporaryDirectBuffer(dst);

        }

    }

    /**

     * Reads bytes from the socket into the given byte array.

     * @return the number of bytes read

     * @throws IOException if the socket is closed or an I/O occurs

     * @throws SocketTimeoutException if the read timeout elapses

     */

    private int read(byte[] b, int off, int len) throws IOException {

        readLock.lock();

        try {

            int timeout = this.timeout;

            int n = 0;

            FileDescriptor fd = beginRead();

            try {

                if (isInputClosed)

                    return IOStatus.EOF;

                configureNonBlockingIfNeeded(fd, timeout);

                n = tryRead(fd, b, off, len);

                if (IOStatus.okayToRetry(n) && isOpen()) {

                    if (timeout > 0) {

                        // read with timeout

                        long nanos = NANOSECONDS.convert(timeout, TimeUnit.MILLISECONDS);

                        do {

                            long startTime = System.nanoTime();

                            park(fd, Net.POLLIN, nanos);

                            n = tryRead(fd, b, off, len);

                            if (n == IOStatus.UNAVAILABLE) {

                                nanos -= System.nanoTime() - startTime;

                                if (nanos <= 0)

                                    throw new SocketTimeoutException("Read timed out");

                            }

                        } while (n == IOStatus.UNAVAILABLE && isOpen());

                    } else {

                        // read, no timeout

                        do {

                            park(fd, Net.POLLIN);

                            n = tryRead(fd, b, off, len);

                        } while (IOStatus.okayToRetry(n) && isOpen());

                    }

                }

                return n;

            } finally {

                endRead(n > 0);

            }

        } finally {

            readLock.unlock();

        }

    }

    /**

     * Marks the beginning of a write operation that might block.

     * @throws SocketException if the socket is closed or not connected

     */

    private FileDescriptor beginWrite() throws SocketException {

        synchronized (stateLock) {

            ensureOpenAndConnected();

            writerThread = NativeThread.current();

            return fd;

        }

    }

    /**

     * Marks the end of a write operation that may have blocked.

     * @throws SocketException is the socket is closed

     */

    private void endWrite(boolean completed) throws SocketException {

        synchronized (stateLock) {

            writerThread = 0;

            int state = this.state;

            if (state == ST_CLOSING)

                stateLock.notifyAll();

            if (!completed && state >= ST_CLOSING)

                throw new SocketException("Socket closed");

        }

    }

    /**

     * Try to write a sequence of bytes to this socket from the given byte array

     */

    private int tryWrite(FileDescriptor fd, byte[] b, int off, int len)

        throws IOException

    {

        ByteBuffer src = Util.getTemporaryDirectBuffer(len);

        assert src.position() == 0;

        try {

            src.put(b, off, len);

            return nd.write(fd, ((DirectBuffer)src).address(), len);

        } finally {

            Util.offerFirstTemporaryDirectBuffer(src);

        }

    }

    /**

     * Writes a sequence of bytes to this socket from the given byte array.

     * @return the number of bytes written

     * @throws IOException if the socket is closed or an I/O occurs

     */

    private int write(byte[] b, int off, int len) throws IOException {

        writeLock.lock();

        try {

            int n = 0;

            FileDescriptor fd = beginWrite();

            try {

                n = tryWrite(fd, b, off, len);

                while (IOStatus.okayToRetry(n) && isOpen()) {

                    park(fd, Net.POLLOUT);

                    n = tryWrite(fd, b, off, len);

                }

                return n;

            } finally {

                endWrite(n > 0);

            }

        } finally {

            writeLock.unlock();

        }

    }

    /**

     * Creates the socket.

     * @param stream {@code true} for a streams socket

     */

    @Override

    protected void create(boolean stream) throws IOException {

        synchronized (stateLock) {

            assert state == ST_NEW;

            if (!stream)

                ResourceManager.beforeUdpCreate();

            FileDescriptor fd;

            try {

                if (server) {

                    assert stream;

                    fd = Net.serverSocket(true);

                } else {

                    fd = Net.socket(stream);

                }

            } catch (IOException ioe) {

                if (!stream)

                    ResourceManager.afterUdpClose();

                throw ioe;

            }

            this.fd = fd;

            this.stream = stream;

            this.closer = FileDescriptorCloser.create(this);

            this.state = ST_UNCONNECTED;

        }

    }

    /**

     * For use by ServerSocket to set the state and other fields after a

     * connection is accepted by a ServerSocket using a custom SocketImpl.

     * The protected fields defined by SocketImpl should be set.

     */

    @Override

    public void postCustomAccept() throws IOException {

        synchronized (stateLock) {

            assert state == ST_NEW;

            assert fd.valid() && localport != 0 && address != null && port != 0;

            IOUtil.configureBlocking(fd, true);

            stream = true;

            closer = FileDescriptorCloser.create(this);

            state = ST_CONNECTED;

        }

    }

    /**

     * For use by ServerSocket to copy the state from this connected SocketImpl

     * to a target SocketImpl. If the target SocketImpl is not a newly created

     * SocketImpl then it is first closed to release any resources. The target

     * SocketImpl becomes the owner of the file descriptor, this SocketImpl

     * is marked as closed and should be discarded.

     */

    @Override

    public void copyTo(SocketImpl si) {

        if (si instanceof NioSocketImpl) {

            NioSocketImpl nsi = (NioSocketImpl) si;

            if (nsi.state != ST_NEW) {

                try {

                    nsi.close();

                } catch (IOException ignore) { }

            }

            synchronized (nsi.stateLock) {

                assert nsi.state == ST_NEW || nsi.state == ST_CLOSED;

                synchronized (this.stateLock) {

                    // this SocketImpl should be connected

                    assert state == ST_CONNECTED && fd.valid()

                        && localport != 0 && address != null && port != 0;

                    // copy fields

                    nsi.stream = this.stream;

                    nsi.fd = this.fd;

                    nsi.localport = this.localport;

                    nsi.address = this.address;

                    nsi.port = this.port;

                    // reset fields; do not reset timeout

                    nsi.nonBlocking = false;

                    nsi.isInputClosed = false;

                    nsi.isOutputClosed = false;

                    nsi.isReuseAddress = false;

                    nsi.state = ST_CONNECTED;

                    // GC'ing of this impl should not close the file descriptor

                    this.closer.disable();

                    this.state = ST_CLOSED;

                    // create new closer to execute when nsi is GC'ed

                    nsi.closer = FileDescriptorCloser.create(nsi);

                }

            }

        } else {

            synchronized (this.stateLock) {

                // this SocketImpl should be connected

                assert state == ST_CONNECTED && fd.valid()

                        && localport != 0 && address != null && port != 0;

                // set fields in foreign impl

                setSocketImplFields(si, fd, localport, address, port);

                // disable closer to prevent GC'ing of this impl from

                // closing the file descriptor

                this.closer.disable();

                this.state = ST_CLOSED;

            }

        }

    }

    /**

     * Marks the beginning of a connect operation that might block.

     * @throws SocketException if the socket is closed or already connected

     */

    private FileDescriptor beginConnect(InetAddress address, int port)

        throws IOException

    {

        synchronized (stateLock) {

            int state = this.state;

            if (state != ST_UNCONNECTED) {

                if (state == ST_CONNECTING)

                    throw new SocketException("Connection in progress");

                if (state == ST_CONNECTED)

                    throw new SocketException("Already connected");

                if (state >= ST_CLOSING)

                    throw new SocketException("Socket closed");

                assert false;

            }

            this.state = ST_CONNECTING;

            // invoke beforeTcpConnect hook if not already bound

            if (localport == 0) {

                NetHooks.beforeTcpConnect(fd, address, port);

            }

            // save the remote address/port

            this.address = address;

            this.port = port;