Back to index...

	/*
	* 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.io.UncheckedIOException;
	import java.lang.ref.Cleaner.Cleanable;
	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 Cleanable cleaner;

	// set to true when the socket is in non-blocking mode
	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 to blocking mode. This method is a no-op if the
	* socket is already in blocking mode.
	* @throws IOException if closed or there is an I/O error changing the mode
	*/
	private void configureBlocking(FileDescriptor fd) throws IOException {
	assert readLock.isHeldByCurrentThread();
	if (nonBlocking) {
	synchronized (stateLock) {
	ensureOpen();
	IOUtil.configureBlocking(fd, true);
	nonBlocking = false;
	}
	}
	}

	/**
	* Configures the socket to non-blocking mode. This method is a no-op if the
	* socket is already in non-blocking mode.
	* @throws IOException if closed or there is an I/O error changing the mode
	*/
	private void configureNonBlocking(FileDescriptor fd) throws IOException {
	assert readLock.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)
	tryFinishClose();
	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;
	} finally {
	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
	configureNonBlocking(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)
	tryFinishClose();
	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 a 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 a 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;
	}
	Runnable closer = closerFor(fd, stream);
	this.fd = fd;
	this.stream = stream;
	this.cleaner = CleanerFactory.cleaner().register(this, closer);
	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);
	}

	// save the remote address/port
	this.address = address;
	this.port = port;

	readerThread = NativeThread.current();
	return fd;
	}
	}

	/**
	* Marks the end of a connect operation that may have blocked.
	* @throws SocketException is the socket is closed
	*/
	private void endConnect(FileDescriptor fd, boolean completed) throws IOException {
	synchronized (stateLock) {
	readerThread = 0;
	int state = this.state;
	if (state == ST_CLOSING)
	tryFinishClose();
	if (completed && state == ST_CONNECTING) {
	this.state = ST_CONNECTED;
	localport = Net.localAddress(fd).getPort();
	} else if (!completed && state >= ST_CLOSING) {
	throw new SocketException("Socket closed");
	}
	}
	}

	/**
	* Waits for a connection attempt to finish with a timeout
	* @throws SocketTimeoutException if the connect timeout elapses
	*/
	private boolean timedFinishConnect(FileDescriptor fd, long nanos) throws IOException {
	long startNanos = System.nanoTime();
	boolean polled = Net.pollConnectNow(fd);
	while (!polled && isOpen()) {
	long remainingNanos = nanos - (System.nanoTime() - startNanos);
	if (remainingNanos <= 0) {
	throw new SocketTimeoutException("Connect timed out");
	}
	park(fd, Net.POLLOUT, remainingNanos);
	polled = Net.pollConnectNow(fd);
	}
	return polled && isOpen();
	}

	/**
	* Attempts to establish a connection to the given socket address with a
	* timeout. Closes the socket if connection cannot be established.
	* @throws IOException if the address is not a resolved InetSocketAddress or
	* the connection cannot be established
	*/
	@Override
	protected void connect(SocketAddress remote, int millis) throws IOException {
	// SocketImpl connect only specifies IOException
	if (!(remote instanceof InetSocketAddress))
	throw new IOException("Unsupported address type");
	InetSocketAddress isa = (InetSocketAddress) remote;
	if (isa.isUnresolved()) {
	throw new UnknownHostException(isa.getHostName());
	}

	InetAddress address = isa.getAddress();
	if (address.isAnyLocalAddress())
	address = InetAddress.getLocalHost();
	int port = isa.getPort();

	ReentrantLock connectLock = readLock;
	try {
	connectLock.lock();
	try {
	boolean connected = false;
	FileDescriptor fd = beginConnect(address, port);
	try {

	// configure socket to non-blocking mode when there is a timeout
	if (millis > 0) {
	configureNonBlocking(fd);
	}

	int n = Net.connect(fd, address, port);
	if (n > 0) {
	// connection established
	connected = true;
	} else {
	assert IOStatus.okayToRetry(n);
	if (millis > 0) {
	// finish connect with timeout
	long nanos = MILLISECONDS.toNanos(millis);
	connected = timedFinishConnect(fd, nanos);
	} else {
	// finish connect, no timeout
	boolean polled = false;
	while (!polled && isOpen()) {
	park(fd, Net.POLLOUT);
	polled = Net.pollConnectNow(fd);
	}
	connected = polled && isOpen();
	}
	}

	// restore socket to blocking mode
	if (connected && millis > 0) {
	configureBlocking(fd);
	}

	} finally {
	endConnect(fd, connected);
	}
	} finally {
	connectLock.unlock();
	}
	} catch (IOException ioe) {
	close();
	throw SocketExceptions.of(ioe, isa);
	}
	}

	@Override
	protected void connect(String host, int port) throws IOException {
	connect(new InetSocketAddress(host, port), 0);
	}

	@Override
	protected void connect(InetAddress address, int port) throws IOException {
	connect(new InetSocketAddress(address, port), 0);
	}

	@Override
	protected void bind(InetAddress host, int port) throws IOException {
	synchronized (stateLock) {
	ensureOpen();
	if (localport != 0)
	throw new SocketException("Already bound");
	NetHooks.beforeTcpBind(fd, host, port);
	Net.bind(fd, host, port);
	// set the address field to the given host address to keep
	// compatibility with PlainSocketImpl. When binding to 0.0.0.0
	// then the actual local address will be ::0 when IPv6 is enabled.
	address = host;
	localport = Net.localAddress(fd).getPort();
	}
	}

	@Override
	protected void listen(int backlog) throws IOException {
	synchronized (stateLock) {
	ensureOpen();
	if (localport == 0)
	throw new SocketException("Not bound");
	Net.listen(fd, backlog < 1 ? 50 : backlog);
	}
	}

	/**
	* Marks the beginning of an accept operation that might block.
	* @throws SocketException if the socket is closed
	*/
	private FileDescriptor beginAccept() throws SocketException {
	synchronized (stateLock) {
	ensureOpen();
	if (!stream)
	throw new SocketException("Not a stream socket");
	if (localport == 0)
	throw new SocketException("Not bound");
	readerThread = NativeThread.current();
	return fd;
	}
	}

	/**
	* Marks the end of an accept operation that may have blocked.
	* @throws SocketException is the socket is closed
	*/
	private void endAccept(boolean completed) throws SocketException {
	synchronized (stateLock) {
	int state = this.state;
	readerThread = 0;
	if (state == ST_CLOSING)
	tryFinishClose();
	if (!completed && state >= ST_CLOSING)
	throw new SocketException("Socket closed");
	}
	}

	/**
	* Accepts a new connection with a timeout.
	* @throws SocketTimeoutException if the accept timeout elapses
	*/
	private int timedAccept(FileDescriptor fd,
	FileDescriptor newfd,
	InetSocketAddress[] isaa,
	long nanos)
	throws IOException
	{
	long startNanos = System.nanoTime();
	int n = Net.accept(fd, newfd, isaa);
	while (n == IOStatus.UNAVAILABLE && isOpen()) {
	long remainingNanos = nanos - (System.nanoTime() - startNanos);
	if (remainingNanos <= 0) {
	throw new SocketTimeoutException("Accept timed out");
	}
	park(fd, Net.POLLIN, remainingNanos);
	n = Net.accept(fd, newfd, isaa);
	}
	return n;
	}

	/**
	* Accepts a new connection so that the given SocketImpl is connected to
	* the peer. The SocketImpl must be a newly created NioSocketImpl.
	*/
	@Override
	protected void accept(SocketImpl si) throws IOException {
	NioSocketImpl nsi = (NioSocketImpl) si;
	if (nsi.state != ST_NEW)
	throw new SocketException("Not a newly created SocketImpl");

	FileDescriptor newfd = new FileDescriptor();
	InetSocketAddress[] isaa = new InetSocketAddress[1];

	// acquire the lock, adjusting the timeout for cases where several
	// threads are accepting connections and there is a timeout set
	ReentrantLock acceptLock = readLock;
	int timeout = this.timeout;
	long remainingNanos = 0;
	if (timeout > 0) {
	remainingNanos = tryLock(acceptLock, timeout, MILLISECONDS);
	if (remainingNanos <= 0) {
	assert !acceptLock.isHeldByCurrentThread();
	throw new SocketTimeoutException("Accept timed out");
	}
	} else {
	acceptLock.lock();
	}

	// accept a connection
	try {
	int n = 0;
	FileDescriptor fd = beginAccept();
	try {
	if (remainingNanos > 0) {
	// accept with timeout
	configureNonBlocking(fd);
	n = timedAccept(fd, newfd, isaa, remainingNanos);
	} else {
	// accept, no timeout
	n = Net.accept(fd, newfd, isaa);
	while (IOStatus.okayToRetry(n) && isOpen()) {
	park(fd, Net.POLLIN);
	n = Net.accept(fd, newfd, isaa);
	}
	}
	} finally {
	endAccept(n > 0);
	assert IOStatus.check(n);
	}
	} finally {
	acceptLock.unlock();
	}

	// get local address and configure accepted socket to blocking mode
	InetSocketAddress localAddress;
	try {
	localAddress = Net.localAddress(newfd);
	IOUtil.configureBlocking(newfd, true);
	} catch (IOException ioe) {
	nd.close(newfd);
	throw ioe;
	}

	// set the fields
	Runnable closer = closerFor(newfd, true);
	synchronized (nsi.stateLock) {
	nsi.fd = newfd;
	nsi.stream = true;
	nsi.cleaner = CleanerFactory.cleaner().register(nsi, closer);
	nsi.localport = localAddress.getPort();
	nsi.address = isaa[0].getAddress();
	nsi.port = isaa[0].getPort();
	nsi.state = ST_CONNECTED;
	}
	}

	@Override
	protected InputStream getInputStream() {
	return new InputStream() {
	@Override
	public int read() throws IOException {
	byte[] a = new byte[1];
	int n = read(a, 0, 1);
	return (n > 0) ? (a[0] & 0xff) : -1;
	}
	@Override
	public int read(byte[] b, int off, int len) throws IOException {
	return NioSocketImpl.this.read(b, off, len);
	}
	@Override
	public int available() throws IOException {
	return NioSocketImpl.this.available();
	}
	@Override
	public void close() throws IOException {
	NioSocketImpl.this.close();
	}
	};
	}

	@Override
	protected OutputStream getOutputStream() {
	return new OutputStream() {
	@Override
	public void write(int b) throws IOException {
	byte[] a = new byte[]{(byte) b};
	write(a, 0, 1);
	}
	@Override
	public void write(byte[] b, int off, int len) throws IOException {
	NioSocketImpl.this.write(b, off, len);
	}
	@Override
	public void close() throws IOException {
	NioSocketImpl.this.close();
	}
	};
	}

	@Override
	protected int available() throws IOException {
	synchronized (stateLock) {
	ensureOpenAndConnected();
	if (isInputClosed) {
	return 0;
	} else {
	return Net.available(fd);
	}
	}
	}

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

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

	/**
	* Closes the socket. If there are I/O operations in progress then the
	* socket is pre-closed and the threads are signalled. The socket will be
	* closed when the last I/O operation aborts.
	*/
	@Override
	protected void close() throws IOException {
	synchronized (stateLock) {
	int state = this.state;
	if (state >= ST_CLOSING)
	return;
	if (state == ST_NEW) {
	// stillborn
	this.state = ST_CLOSED;
	return;
	}
	this.state = ST_CLOSING;

	// shutdown output when linger interval not set to 0
	try {
	var SO_LINGER = StandardSocketOptions.SO_LINGER;
	if ((int) Net.getSocketOption(fd, SO_LINGER) != 0) {
	Net.shutdown(fd, Net.SHUT_WR);
	}
	} catch (IOException ignore) { }

	// attempt to close the socket. If there are I/O operations in progress
	// then the socket is pre-closed and the thread(s) signalled. The
	// last thread will close the file descriptor.
	if (!tryClose()) {
	nd.preClose(fd);
	long reader = readerThread;
	if (reader != 0)
	NativeThread.signal(reader);
	long writer = writerThread;
	if (writer != 0)
	NativeThread.signal(writer);
	}
	}
	}

	// the socket options supported by client and server sockets
	private static volatile Set<SocketOption<?>> clientSocketOptions;
	private static volatile Set<SocketOption<?>> serverSocketOptions;

	@Override
	protected Set<SocketOption<?>> supportedOptions() {
	Set<SocketOption<?>> options = (server) ? serverSocketOptions : clientSocketOptions;
	if (options == null) {
	options = new HashSet<>();
	options.add(StandardSocketOptions.SO_RCVBUF);
	options.add(StandardSocketOptions.SO_REUSEADDR);
	if (server) {
	// IP_TOS added for server socket to maintain compatibility
	options.add(StandardSocketOptions.IP_TOS);
	options.addAll(ExtendedSocketOptions.serverSocketOptions());
	} else {
	options.add(StandardSocketOptions.IP_TOS);
	options.add(StandardSocketOptions.SO_KEEPALIVE);
	options.add(StandardSocketOptions.SO_SNDBUF);
	options.add(StandardSocketOptions.SO_LINGER);
	options.add(StandardSocketOptions.TCP_NODELAY);
	options.addAll(ExtendedSocketOptions.clientSocketOptions());
	}
	if (Net.isReusePortAvailable())
	options.add(StandardSocketOptions.SO_REUSEPORT);
	options = Collections.unmodifiableSet(options);
	if (server) {
	serverSocketOptions = options;
	} else {
	clientSocketOptions = options;
	}
	}
	return options;
	}

	@Override
	protected <T> void setOption(SocketOption<T> opt, T value) throws IOException {
	if (!supportedOptions().contains(opt))
	throw new UnsupportedOperationException("'" + opt + "' not supported");
	if (!opt.type().isInstance(value))
	throw new IllegalArgumentException("Invalid value '" + value + "'");
	synchronized (stateLock) {
	ensureOpen();
	if (opt == StandardSocketOptions.IP_TOS) {
	// maps to IP_TOS or IPV6_TCLASS
	Net.setSocketOption(fd, family(), opt, value);
	} else if (opt == StandardSocketOptions.SO_REUSEADDR) {
	boolean b = (boolean) value;
	if (Net.useExclusiveBind()) {
	isReuseAddress = b;
	} else {
	Net.setSocketOption(fd, opt, b);
	}
	} else {
	// option does not need special handling
	Net.setSocketOption(fd, opt, value);
	}
	}
	}

	@SuppressWarnings("unchecked")
	protected <T> T getOption(SocketOption<T> opt) throws IOException {
	if (!supportedOptions().contains(opt))
	throw new UnsupportedOperationException("'" + opt + "' not supported");
	synchronized (stateLock) {
	ensureOpen();
	if (opt == StandardSocketOptions.IP_TOS) {
	return (T) Net.getSocketOption(fd, family(), opt);
	} else if (opt == StandardSocketOptions.SO_REUSEADDR) {
	if (Net.useExclusiveBind()) {
	return (T) Boolean.valueOf(isReuseAddress);
	} else {
	return (T) Net.getSocketOption(fd, opt);
	}
	} else {
	// option does not need special handling
	return (T) Net.getSocketOption(fd, opt);
	}
	}
	}

	private boolean booleanValue(Object value, String desc) throws SocketException {
	if (!(value instanceof Boolean))
	throw new SocketException("Bad value for " + desc);
	return (boolean) value;
	}

	private int intValue(Object value, String desc) throws SocketException {
	if (!(value instanceof Integer))
	throw new SocketException("Bad value for " + desc);
	return (int) value;
	}

	@Override
	public void setOption(int opt, Object value) throws SocketException {
	synchronized (stateLock) {
	ensureOpen();
	try {
	switch (opt) {
	case SO_LINGER: {
	// the value is "false" to disable, or linger interval to enable
	int i;
	if (value instanceof Boolean && ((boolean) value) == false) {
	i = -1;
	} else {
	i = intValue(value, "SO_LINGER");
	}
	Net.setSocketOption(fd, StandardSocketOptions.SO_LINGER, i);
	break;
	}
	case SO_TIMEOUT: {
	int i = intValue(value, "SO_TIMEOUT");
	if (i < 0)
	throw new IllegalArgumentException("timeout < 0");
	timeout = i;
	break;
	}
	case IP_TOS: {
	int i = intValue(value, "IP_TOS");
	Net.setSocketOption(fd, family(), StandardSocketOptions.IP_TOS, i);
	break;
	}
	case TCP_NODELAY: {
	boolean b = booleanValue(value, "TCP_NODELAY");
	Net.setSocketOption(fd, StandardSocketOptions.TCP_NODELAY, b);
	break;
	}
	case SO_SNDBUF: {
	int i = intValue(value, "SO_SNDBUF");
	if (i <= 0)
	throw new SocketException("SO_SNDBUF <= 0");
	Net.setSocketOption(fd, StandardSocketOptions.SO_SNDBUF, i);
	break;
	}
	case SO_RCVBUF: {
	int i = intValue(value, "SO_RCVBUF");
	if (i <= 0)
	throw new SocketException("SO_RCVBUF <= 0");
	Net.setSocketOption(fd, StandardSocketOptions.SO_RCVBUF, i);
	break;
	}
	case SO_KEEPALIVE: {
	boolean b = booleanValue(value, "SO_KEEPALIVE");
	Net.setSocketOption(fd, StandardSocketOptions.SO_KEEPALIVE, b);
	break;
	}
	case SO_OOBINLINE: {
	boolean b = booleanValue(value, "SO_OOBINLINE");
	Net.setSocketOption(fd, ExtendedSocketOption.SO_OOBINLINE, b);
	break;
	}
	case SO_REUSEADDR: {
	boolean b = booleanValue(value, "SO_REUSEADDR");
	if (Net.useExclusiveBind()) {
	isReuseAddress = b;
	} else {
	Net.setSocketOption(fd, StandardSocketOptions.SO_REUSEADDR, b);
	}
	break;
	}
	case SO_REUSEPORT: {
	if (!Net.isReusePortAvailable())
	throw new SocketException("SO_REUSEPORT not supported");
	boolean b = booleanValue(value, "SO_REUSEPORT");
	Net.setSocketOption(fd, StandardSocketOptions.SO_REUSEPORT, b);
	break;
	}
	default:
	throw new SocketException("Unknown option " + opt);
	}
	} catch (SocketException e) {
	throw e;
	} catch (IllegalArgumentException \| IOException e) {
	throw new SocketException(e.getMessage());
	}
	}
	}

	@Override
	public Object getOption(int opt) throws SocketException {
	synchronized (stateLock) {
	ensureOpen();
	try {
	switch (opt) {
	case SO_TIMEOUT:
	return timeout;
	case TCP_NODELAY:
	return Net.getSocketOption(fd, StandardSocketOptions.TCP_NODELAY);
	case SO_OOBINLINE:
	return Net.getSocketOption(fd, ExtendedSocketOption.SO_OOBINLINE);
	case SO_LINGER: {
	// return "false" when disabled, linger interval when enabled
	int i = (int) Net.getSocketOption(fd, StandardSocketOptions.SO_LINGER);
	if (i == -1) {
	return Boolean.FALSE;
	} else {
	return i;
	}
	}
	case SO_REUSEADDR:
	if (Net.useExclusiveBind()) {
	return isReuseAddress;
	} else {
	return Net.getSocketOption(fd, StandardSocketOptions.SO_REUSEADDR);
	}
	case SO_BINDADDR:
	return Net.localAddress(fd).getAddress();
	case SO_SNDBUF:
	return Net.getSocketOption(fd, StandardSocketOptions.SO_SNDBUF);
	case SO_RCVBUF:
	return Net.getSocketOption(fd, StandardSocketOptions.SO_RCVBUF);
	case IP_TOS:
	return Net.getSocketOption(fd, family(), StandardSocketOptions.IP_TOS);
	case SO_KEEPALIVE:
	return Net.getSocketOption(fd, StandardSocketOptions.SO_KEEPALIVE);
	case SO_REUSEPORT:
	if (!Net.isReusePortAvailable())
	throw new SocketException("SO_REUSEPORT not supported");
	return Net.getSocketOption(fd, StandardSocketOptions.SO_REUSEPORT);
	default:
	throw new SocketException("Unknown option " + opt);
	}
	} catch (SocketException e) {
	throw e;
	} catch (IllegalArgumentException \| IOException e) {
	throw new SocketException(e.getMessage());
	}
	}
	}

	@Override
	protected void shutdownInput() throws IOException {
	synchronized (stateLock) {
	ensureOpenAndConnected();
	if (!isInputClosed) {
	Net.shutdown(fd, Net.SHUT_RD);
	isInputClosed = true;
	}
	}
	}

	@Override
	protected void shutdownOutput() throws IOException {
	synchronized (stateLock) {
	ensureOpenAndConnected();
	if (!isOutputClosed) {
	Net.shutdown(fd, Net.SHUT_WR);
	isOutputClosed = true;
	}
	}
	}

	@Override
	protected boolean supportsUrgentData() {
	return true;
	}

	@Override
	protected void sendUrgentData(int data) throws IOException {
	writeLock.lock();
	try {
	int n = 0;
	FileDescriptor fd = beginWrite();
	try {
	do {
	n = Net.sendOOB(fd, (byte) data);
	} while (n == IOStatus.INTERRUPTED && isOpen());
	if (n == IOStatus.UNAVAILABLE) {
	throw new SocketException("No buffer space available");
	}
	} finally {
	endWrite(n > 0);
	}
	} finally {
	writeLock.unlock();
	}
	}

	/**
	* Returns an action to close the given file descriptor.
	*/
	private static Runnable closerFor(FileDescriptor fd, boolean stream) {
	if (stream) {
	return () -> {
	try {
	nd.close(fd);
	} catch (IOException ioe) {
	throw new UncheckedIOException(ioe);
	}
	};
	} else {
	return () -> {
	try {
	nd.close(fd);
	} catch (IOException ioe) {
	throw new UncheckedIOException(ioe);
	} finally {
	// decrement
	ResourceManager.afterUdpClose();
	}
	};
	}
	}

	/**
	* Attempts to acquire the given lock within the given waiting time.
	* @return the remaining time in nanoseconds when the lock is acquired, zero
	* or less if the lock was not acquired before the timeout expired
	*/
	private static long tryLock(ReentrantLock lock, long timeout, TimeUnit unit) {
	assert timeout > 0;
	boolean interrupted = false;
	long nanos = NANOSECONDS.convert(timeout, unit);
	long remainingNanos = nanos;
	long startNanos = System.nanoTime();
	boolean acquired = false;
	while (!acquired && (remainingNanos > 0)) {
	try {
	acquired = lock.tryLock(remainingNanos, NANOSECONDS);
	} catch (InterruptedException e) {
	interrupted = true;
	}
	remainingNanos = nanos - (System.nanoTime() - startNanos);
	}
	if (acquired && remainingNanos <= 0L)
	lock.unlock(); // release lock if timeout has expired
	if (interrupted)
	Thread.currentThread().interrupt();
	return remainingNanos;
	}

	/**
	* Returns the socket protocol family.
	*/
	private static ProtocolFamily family() {
	if (Net.isIPv6Available()) {
	return StandardProtocolFamily.INET6;
	} else {
	return StandardProtocolFamily.INET;
	}
	}
	}

Back to index...