/*

 * Copyright (c) 2019, 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.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.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.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 the connect method is used to establish a connection with a

 * timeout then the socket is configured non-blocking for the connect attempt,

 * and then restored to blocking mode when the connection is established.

 * If the accept or read methods are used with a timeout then the socket is

 * configured non-blocking and is never restored. When in non-blocking mode,

 * operations that don't complete immediately will poll the socket and preserve

 * the semantics of blocking operations.

 */

public final class NioSocketImpl extends SocketImpl implements PlatformSocketImpl {

    private static final NativeDispatcher nd = new SocketDispatcher();

    // 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;

    // set by configureNonBlockingForever when the socket changed to 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, protected by stateLock

    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;

    // used by read to emulate legacy behavior, protected by readLock

    private boolean readEOF;

    private boolean connectionReset;

    /**

     * Creates an 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 {

        int state = this.state;

        if (state == ST_NEW)

            throw new SocketException("Socket not created");

        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 = NANOSECONDS.toMillis(nanos);

        }

        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);

    }

    /**

     * Configures the socket's blocking mode except when socket has been

     * configured non-blocking by {@code configureNonBlockingForever}.

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

     */

    private void configureBlocking(FileDescriptor fd, boolean block) throws IOException {

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

        if (!nonBlocking) {

            synchronized (stateLock) {

                if (!nonBlocking) {

                    ensureOpen();

                    IOUtil.configureBlocking(fd, block);

                }

            }

        }

    }

    /**

     * Configures the socket to be non-blocking. Once configured to non-blocking

     * by this method then the blocking mode cannot be changed back to blocking.

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

     */

    private void configureNonBlockingForever(FileDescriptor fd) throws IOException {

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

        if (!nonBlocking) {

            synchronized (stateLock) {

                ensureOpen();

                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)

                tryClose();

            if (!completed && state >= ST_CLOSING)

                throw new SocketException("Socket closed");

        }

    }

    /**

     * Attempts 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;

            Util.offerFirstTemporaryDirectBuffer(dst);

        }

    }

    /**

     * Reads bytes from the socket into the given byte array with a timeout.

     * @throws SocketTimeoutException if the read timeout elapses

     */

    private int timedRead(FileDescriptor fd, byte[] b, int off, int len, long nanos)

        throws IOException

    {

        long startNanos = System.nanoTime();

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

        while (n == IOStatus.UNAVAILABLE && isOpen()) {

            long remainingNanos = nanos - (System.nanoTime() - startNanos);

            if (remainingNanos <= 0) {

                throw new SocketTimeoutException("Read timed out");

            }

            park(fd, Net.POLLIN, remainingNanos);

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

        }

        return n;

    }

    /**

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

     * @return the number of bytes read or -1 at EOF

     * @throws SocketException if the socket is closed or a socket I/O error occurs

     * @throws SocketTimeoutException if the read timeout elapses

     */

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

        int n = 0;

        FileDescriptor fd = beginRead();

        try {

            if (connectionReset)

                throw new SocketException("Connection reset");

            if (isInputClosed)

                return -1;

            int timeout = this.timeout;

            if (timeout > 0) {

                // read with timeout

                configureNonBlockingForever(fd);

                n = timedRead(fd, b, off, len, MILLISECONDS.toNanos(timeout));

            } else {

                // read, no timeout

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

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

                    park(fd, Net.POLLIN);

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

                }

            }

            return n;

        } catch (SocketTimeoutException e) {

            throw e;

        } catch (ConnectionResetException e) {

            connectionReset = true;

            throw new SocketException("Connection reset");

        } catch (IOException ioe) {

            throw new SocketException(ioe.getMessage());

        } finally {

            endRead(n > 0);

        }

    }

    /**

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

     * @return the number of bytes read or -1 at EOF

     * @throws IndexOutOfBoundsException if the bound checks fail

     * @throws SocketException if the socket is closed or a socket I/O error occurs

     * @throws SocketTimeoutException if the read timeout elapses

     */

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

        Objects.checkFromIndexSize(off, len, b.length);

        if (len == 0) {

            return 0;

        } else {

            readLock.lock();

            try {

                // emulate legacy behavior to return -1, even if socket is closed

                if (readEOF)

                    return -1;

                // read up to MAX_BUFFER_SIZE bytes

                int size = Math.min(len, MAX_BUFFER_SIZE);

                int n = implRead(b, off, size);

                if (n == -1)

                    readEOF = true;

                return n;

            } 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)

                tryClose();

            if (!completed && state >= ST_CLOSING)

                throw new SocketException("Socket closed");

        }

    }

    /**

     * Attempts to write a sequence of bytes to the 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 the socket from the given byte array.

     * @return the number of bytes written

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

     */

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

        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;

        } catch (IOException ioe) {

            throw new SocketException(ioe.getMessage());

        } finally {

            endWrite(n > 0);

        }

    }

    /**

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

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

     */

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

        Objects.checkFromIndexSize(off, len, b.length);

        if (len > 0) {

            writeLock.lock();

            try {

                int pos = off;

                int end = off + len;

                while (pos < end) {

                    // write up to MAX_BUFFER_SIZE bytes

                    int size = Math.min((end - pos), MAX_BUFFER_SIZE);

                    int n = implWrite(b, pos, size);

                    pos += n;

                }

            } finally {

                writeLock.unlock();

            }

        }

    }

    /**

     * Creates the socket.

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

     */

    @Override

    protected void create(boolean stream) throws IOException {

        synchronized (stateLock) {

            if (state != ST_NEW)

                throw new IOException("Already created");

            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;

        }

    }

    /**

     * 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_NEW)

                    throw new SocketException("Not created");

                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);