jdk-sandbox: comparison src/java.base/share/classes/sun/nio/ch/SocketChannelImpl.java

equal deleted inserted replaced

-:a406a9c451a0
+:ce06058197a4
 import java.nio.channels.SelectionKey;
 import java.nio.channels.SocketChannel;
 import java.nio.channels.spi.SelectorProvider;
 import java.util.Collections;
 import java.util.HashSet;
+import java.util.Objects;
 import java.util.Set;
 import java.util.concurrent.locks.ReentrantLock;
 import sun.net.NetHooks;
 import sun.net.ext.ExtendedSocketOptions;
 class SocketChannelImpl
 extends SocketChannel
 implements SelChImpl
 {
 // Used to make native read and write calls
 private static NativeDispatcher nd;
 // Our file descriptor object
 private final FileDescriptor fd;
 private final int fdVal;
-// IDs of native threads doing reads and writes, for signalling
-private volatile long readerThread;
-private volatile long writerThread;
 // 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();
+// Input/Output closed
+private volatile boolean isInputClosed;
+private volatile boolean isOutputClosed;
 // -- The following fields are protected by stateLock
 // set true when exclusive binding is on and SO_REUSEADDR is emulated
 private boolean isReuseAddress;
 // State, increases monotonically
-private static final int ST_UNINITIALIZED = -1;
 private static final int ST_UNCONNECTED = 0;
-private static final int ST_PENDING = 1;
+private static final int ST_CONNECTIONPENDING = 1;
 private static final int ST_CONNECTED = 2;
-private static final int ST_KILLPENDING = 3;
+private static final int ST_CLOSING = 3;
-private static final int ST_KILLED = 4;
+private static final int ST_KILLPENDING = 4;
-private int state = ST_UNINITIALIZED;
+private static final int ST_KILLED = 5;
+private int state;
+// IDs of native threads doing reads and writes, for signalling
+private long readerThread;
+private long writerThread;
 // Binding
 private InetSocketAddress localAddress;
 private InetSocketAddress remoteAddress;
-// Input/Output open
-private boolean isInputOpen = true;
-private boolean isOutputOpen = true;
 // Socket adaptor, created on demand
 private Socket socket;
 // -- End of fields protected by stateLock
 //
 SocketChannelImpl(SelectorProvider sp) throws IOException {
 super(sp);
 this.fd = Net.socket(true);
 this.fdVal = IOUtil.fdVal(fd);
-this.state = ST_UNCONNECTED;
+}
-}
+SocketChannelImpl(SelectorProvider sp, FileDescriptor fd, boolean bound)
-SocketChannelImpl(SelectorProvider sp,
-FileDescriptor fd,
-boolean bound)
 throws IOException
 {
 super(sp);
 this.fd = fd;
 this.fdVal = IOUtil.fdVal(fd);
-this.state = ST_UNCONNECTED;
+if (bound) {
-if (bound)
+synchronized (stateLock) {
 this.localAddress = Net.localAddress(fd);
+}
+}
 }
 // Constructor for sockets obtained from server sockets
 //
-SocketChannelImpl(SelectorProvider sp,
+SocketChannelImpl(SelectorProvider sp, FileDescriptor fd, InetSocketAddress isa)
-FileDescriptor fd, InetSocketAddress remote)
 throws IOException
 {
 super(sp);
 this.fd = fd;
 this.fdVal = IOUtil.fdVal(fd);
-this.state = ST_CONNECTED;
+synchronized (stateLock) {
 this.localAddress = Net.localAddress(fd);
-this.remoteAddress = remote;
+this.remoteAddress = isa;
-}
+this.state = ST_CONNECTED;
+}
+}
+// @throws ClosedChannelException if channel is closed
+private void ensureOpen() throws ClosedChannelException {
+if (!isOpen())
+throw new ClosedChannelException();
+}
+@Override
 public Socket socket() {
 synchronized (stateLock) {
 if (socket == null)
 socket = SocketAdaptor.create(this);
 return socket;
 }
 @Override
 public SocketAddress getLocalAddress() throws IOException {
 synchronized (stateLock) {
-if (!isOpen())
+ensureOpen();
-throw new ClosedChannelException();
+return Net.getRevealedLocalAddress(localAddress);
-return  Net.getRevealedLocalAddress(localAddress);
 }
 }
 @Override
 public SocketAddress getRemoteAddress() throws IOException {
 synchronized (stateLock) {
-if (!isOpen())
+ensureOpen();
-throw new ClosedChannelException();
 return remoteAddress;
 }
 }
 @Override
 public <T> SocketChannel setOption(SocketOption<T> name, T value)
 throws IOException
 {
-if (name == null)
+Objects.requireNonNull(name);
-throw new NullPointerException();
 if (!supportedOptions().contains(name))
 throw new UnsupportedOperationException("'" + name + "' not supported");
 synchronized (stateLock) {
-if (!isOpen())
+ensureOpen();
-throw new ClosedChannelException();
 if (name == StandardSocketOptions.IP_TOS) {
 ProtocolFamily family = Net.isIPv6Available() ?
 StandardProtocolFamily.INET6 : StandardProtocolFamily.INET;
 Net.setSocketOption(fd, family, name, value);
 @Override
 @SuppressWarnings("unchecked")
 public <T> T getOption(SocketOption<T> name)
 throws IOException
 {
-if (name == null)
+Objects.requireNonNull(name);
-throw new NullPointerException();
 if (!supportedOptions().contains(name))
 throw new UnsupportedOperationException("'" + name + "' not supported");
 synchronized (stateLock) {
-if (!isOpen())
+ensureOpen();
-throw new ClosedChannelException();
+if (name == StandardSocketOptions.SO_REUSEADDR && Net.useExclusiveBind()) {
-if (name == StandardSocketOptions.SO_REUSEADDR &&
-Net.useExclusiveBind())
-{
 // SO_REUSEADDR emulated when using exclusive bind
 return (T)Boolean.valueOf(isReuseAddress);
 }
 // special handling for IP_TOS: always return 0 when IPv6
 private static class DefaultOptionsHolder {
 static final Set<SocketOption<?>> defaultOptions = defaultOptions();
 private static Set<SocketOption<?>> defaultOptions() {
-HashSet<SocketOption<?>> set = new HashSet<>(8);
+HashSet<SocketOption<?>> set = new HashSet<>();
 set.add(StandardSocketOptions.SO_SNDBUF);
 set.add(StandardSocketOptions.SO_RCVBUF);
 set.add(StandardSocketOptions.SO_KEEPALIVE);
 set.add(StandardSocketOptions.SO_REUSEADDR);
 if (Net.isReusePortAvailable()) {
 set.add(StandardSocketOptions.SO_LINGER);
 set.add(StandardSocketOptions.TCP_NODELAY);
 // additional options required by socket adaptor
 set.add(StandardSocketOptions.IP_TOS);
 set.add(ExtendedSocketOption.SO_OOBINLINE);
-ExtendedSocketOptions extendedOptions =
+set.addAll(ExtendedSocketOptions.getInstance().options());
-ExtendedSocketOptions.getInstance();
-set.addAll(extendedOptions.options());
 return Collections.unmodifiableSet(set);
 }
 }
 @Override
 public final Set<SocketOption<?>> supportedOptions() {
 return DefaultOptionsHolder.defaultOptions;
 }
-private boolean ensureReadOpen() throws ClosedChannelException {
+/**
-synchronized (stateLock) {
+* Marks the beginning of a read operation that might block.
-if (!isOpen())
+*
-throw new ClosedChannelException();
+* @throws ClosedChannelException if the channel is closed
-if (!isConnected())
+* @throws NotYetConnectedException if the channel is not yet connected
+*/
+private void beginRead(boolean blocking) throws ClosedChannelException {
+if (blocking) {
+// set hook for Thread.interrupt
+begin();
+}
+synchronized (stateLock) {
+ensureOpen();
+if (state != ST_CONNECTED)
 throw new NotYetConnectedException();
-if (!isInputOpen)
+if (blocking)
-return false;
+readerThread = NativeThread.current();
-else
+}
-return true;
+}
-}
-}
+/**
+* Marks the end of a read operation that may have blocked.
-private void ensureWriteOpen() throws ClosedChannelException {
+*
-synchronized (stateLock) {
+* @throws AsynchronousCloseException if the channel was closed due to this
-if (!isOpen())
+* thread being interrupted on a blocking read operation.
-throw new ClosedChannelException();
+*/
-if (!isOutputOpen)
+private void endRead(boolean blocking, boolean completed)
-throw new ClosedChannelException();
+throws AsynchronousCloseException
-if (!isConnected())
+{
-throw new NotYetConnectedException();
+if (blocking) {
-}
+synchronized (stateLock) {
-}
+readerThread = 0;
+// notify any thread waiting in implCloseSelectableChannel
-private void readerCleanup() throws IOException {
+if (state == ST_CLOSING) {
-synchronized (stateLock) {
+stateLock.notifyAll();
-readerThread = 0;
+}
-if (state == ST_KILLPENDING)
+}
-kill();
+// remove hook for Thread.interrupt
-}
+end(completed);
 }
+}
-private void writerCleanup() throws IOException {
-synchronized (stateLock) {
+@Override
-writerThread = 0;
-if (state == ST_KILLPENDING)
-kill();
-}
-}
 public int read(ByteBuffer buf) throws IOException {
+Objects.requireNonNull(buf);
-if (buf == null)
-throw new NullPointerException();
 readLock.lock();
 try {
-if (!ensureReadOpen())
+boolean blocking = isBlocking();
-return -1;
 int n = 0;
 try {
+beginRead(blocking);
-// Set up the interruption machinery; see
-// AbstractInterruptibleChannel for details
+// check if input is shutdown
-//
+if (isInputClosed)
-begin();
+return IOStatus.EOF;
-synchronized (stateLock) {
+if (blocking) {
-if (!isOpen()) {
+do {
-// Either the current thread is already interrupted, so
+n = IOUtil.read(fd, buf, -1, nd);
-// begin() closed the channel, or another thread closed the
+} while (n == IOStatus.INTERRUPTED && isOpen());
-// channel since we checked it a few bytecodes ago.  In
+} else {
-// either case the value returned here is irrelevant since
-// the invocation of end() in the finally block will throw
-// an appropriate exception.
-//
-return 0;
-}
-// Save this thread so that it can be signalled on those
-// platforms that require it
-//
-readerThread = NativeThread.current();
-}
-// Between the previous test of isOpen() and the return of the
-// IOUtil.read invocation below, this channel might be closed
-// or this thread might be interrupted.  We rely upon the
-// implicit synchronization point in the kernel read() call to
-// make sure that the right thing happens.  In either case the
-// implCloseSelectableChannel method is ultimately invoked in
-// some other thread, so there are three possibilities:
-//
-//   - implCloseSelectableChannel() invokes nd.preClose()
-//     before this thread invokes read(), in which case the
-//     read returns immediately with either EOF or an error,
-//     the latter of which will cause an IOException to be
-//     thrown.
-//
-//   - implCloseSelectableChannel() invokes nd.preClose() after
-//     this thread is blocked in read().  On some operating
-//     systems (e.g., Solaris and Windows) this causes the read
-//     to return immediately with either EOF or an error
-//     indication.
-//
-//   - implCloseSelectableChannel() invokes nd.preClose() after
-//     this thread is blocked in read() but the operating
-//     system (e.g., Linux) doesn't support preemptive close,
-//     so implCloseSelectableChannel() proceeds to signal this
-//     thread, thereby causing the read to return immediately
-//     with IOStatus.INTERRUPTED.
-//
-// In all three cases the invocation of end() in the finally
-// clause will notice that the channel has been closed and
-// throw an appropriate exception (AsynchronousCloseException
-// or ClosedByInterruptException) if necessary.
-//
-// *There is A fourth possibility. implCloseSelectableChannel()
-// invokes nd.preClose(), signals reader/writer thred and quickly
-// moves on to nd.close() in kill(), which does a real close.
-// Then a third thread accepts a new connection, opens file or
-// whatever that causes the released "fd" to be recycled. All
-// above happens just between our last isOpen() check and the
-// next kernel read reached, with the recycled "fd". The solution
-// is to postpone the real kill() if there is a reader or/and
-// writer thread(s) over there "waiting", leave the cleanup/kill
-// to the reader or writer thread. (the preClose() still happens
-// so the connection gets cut off as usual).
-//
-// For socket channels there is the additional wrinkle that
-// asynchronous shutdown works much like asynchronous close,
-// except that the channel is shutdown rather than completely
-// closed.  This is analogous to the first two cases above,
-// except that the shutdown operation plays the role of
-// nd.preClose().
-for (;;) {
 n = IOUtil.read(fd, buf, -1, nd);
-if ((n == IOStatus.INTERRUPTED) && isOpen()) {
+}
-// The system call was interrupted but the channel
-// is still open, so retry
-continue;
-}
-return IOStatus.normalize(n);
-}
 } finally {
-readerCleanup();        // Clear reader thread
+endRead(blocking, n > 0);
-// The end method, which is defined in our superclass
+if (n <= 0 && isInputClosed)
-// AbstractInterruptibleChannel, resets the interruption
+return IOStatus.EOF;
-// machinery.  If its argument is true then it returns
+}
-// normally; otherwise it checks the interrupt and open state
+return IOStatus.normalize(n);
-// of this channel and throws an appropriate exception if
-// necessary.
-//
-// So, if we actually managed to do any I/O in the above try
-// block then we pass true to the end method.  We also pass
-// true if the channel was in non-blocking mode when the I/O
-// operation was initiated but no data could be transferred;
-// this prevents spurious exceptions from being thrown in the
-// rare event that a channel is closed or a thread is
-// interrupted at the exact moment that a non-blocking I/O
-// request is made.
-//
-end(n > 0 || (n == IOStatus.UNAVAILABLE));
-// Extra case for socket channels: Asynchronous shutdown
-//
-synchronized (stateLock) {
-if ((n <= 0) && (!isInputOpen))
-return IOStatus.EOF;
-}
-assert IOStatus.check(n);
-}
 } finally {
 readLock.unlock();
 }
 }
+@Override
 public long read(ByteBuffer[] dsts, int offset, int length)
 throws IOException
 {
-if ((offset < 0) || (length < 0) || (offset > dsts.length - length))
+Objects.checkFromIndexSize(offset, length, dsts.length);
-throw new IndexOutOfBoundsException();
 readLock.lock();
 try {
-if (!ensureReadOpen())
+boolean blocking = isBlocking();
-return -1;
 long n = 0;
 try {
-begin();
+beginRead(blocking);
-synchronized (stateLock) {
-if (!isOpen())
+// check if input is shutdown
-return 0;
+if (isInputClosed)
-readerThread = NativeThread.current();
+return IOStatus.EOF;
-}
+if (blocking) {
-for (;;) {
+do {
+n = IOUtil.read(fd, dsts, offset, length, nd);
+} while (n == IOStatus.INTERRUPTED && isOpen());
+} else {
 n = IOUtil.read(fd, dsts, offset, length, nd);
-if ((n == IOStatus.INTERRUPTED) && isOpen())
-continue;
-return IOStatus.normalize(n);
 }
 } finally {
-readerCleanup();
+endRead(blocking, n > 0);
-end(n > 0 || (n == IOStatus.UNAVAILABLE));
+if (n <= 0 && isInputClosed)
-synchronized (stateLock) {
+return IOStatus.EOF;
-if ((n <= 0) && (!isInputOpen))
+}
-return IOStatus.EOF;
+return IOStatus.normalize(n);
-}
-assert IOStatus.check(n);
-}
 } finally {
 readLock.unlock();
 }
 }
+/**
+* Marks the beginning of a write operation that might block.
+*
+* @throws ClosedChannelException if the channel is closed or output shutdown
+* @throws NotYetConnectedException if the channel is not yet connected
+*/
+private void beginWrite(boolean blocking) throws ClosedChannelException {
+if (blocking) {
+// set hook for Thread.interrupt
+begin();
+}
+synchronized (stateLock) {
+ensureOpen();
+if (isOutputClosed)
+throw new ClosedChannelException();
+if (state != ST_CONNECTED)
+throw new NotYetConnectedException();
+if (blocking)
+writerThread = NativeThread.current();
+}
+}
+/**
+* Marks the end of a write operation that may have blocked.
+*
+* @throws AsynchronousCloseException if the channel was closed due to this
+* thread being interrupted on a blocking write operation.
+*/
+private void endWrite(boolean blocking, boolean completed)
+throws AsynchronousCloseException
+{
+if (blocking) {
+synchronized (stateLock) {
+writerThread = 0;
+// notify any thread waiting in implCloseSelectableChannel
+if (state == ST_CLOSING) {
+stateLock.notifyAll();
+}
+}
+// remove hook for Thread.interrupt
+end(completed);
+}
+}
+@Override
 public int write(ByteBuffer buf) throws IOException {
-if (buf == null)
+Objects.requireNonNull(buf);
-throw new NullPointerException();
 writeLock.lock();
 try {
-ensureWriteOpen();
+boolean blocking = isBlocking();
 int n = 0;
 try {
-begin();
+beginWrite(blocking);
-synchronized (stateLock) {
+if (blocking) {
-if (!isOpen())
+do {
-return 0;
+n = IOUtil.write(fd, buf, -1, nd);
-writerThread = NativeThread.current();
+} while (n == IOStatus.INTERRUPTED && isOpen());
-}
+} else {
-for (;;) {
 n = IOUtil.write(fd, buf, -1, nd);
-if ((n == IOStatus.INTERRUPTED) && isOpen())
-continue;
-return IOStatus.normalize(n);
 }
 } finally {
-writerCleanup();
+endWrite(blocking, n > 0);
-end(n > 0 || (n == IOStatus.UNAVAILABLE));
+if (n <= 0 && isOutputClosed)
-synchronized (stateLock) {
+throw new AsynchronousCloseException();
-if ((n <= 0) && (!isOutputOpen))
+}
-throw new AsynchronousCloseException();
+return IOStatus.normalize(n);
-}
-assert IOStatus.check(n);
-}
 } finally {
 writeLock.unlock();
 }
 }
+@Override
 public long write(ByteBuffer[] srcs, int offset, int length)
 throws IOException
 {
-if ((offset < 0) || (length < 0) || (offset > srcs.length - length))
+Objects.checkFromIndexSize(offset, length, srcs.length);
-throw new IndexOutOfBoundsException();
 writeLock.lock();
 try {
-ensureWriteOpen();
+boolean blocking = isBlocking();
 long n = 0;
 try {
-begin();
+beginWrite(blocking);
-synchronized (stateLock) {
+if (blocking) {
-if (!isOpen())
+do {
-return 0;
+n = IOUtil.write(fd, srcs, offset, length, nd);
-writerThread = NativeThread.current();
+} while (n == IOStatus.INTERRUPTED && isOpen());
-}
+} else {
-for (;;) {
 n = IOUtil.write(fd, srcs, offset, length, nd);
-if ((n == IOStatus.INTERRUPTED) && isOpen())
-continue;
-return IOStatus.normalize(n);
 }
 } finally {
-writerCleanup();
+endWrite(blocking, n > 0);
-end((n > 0) || (n == IOStatus.UNAVAILABLE));
+if (n <= 0 && isOutputClosed)
-synchronized (stateLock) {
+throw new AsynchronousCloseException();
-if ((n <= 0) && (!isOutputOpen))
+}
-throw new AsynchronousCloseException();
+return IOStatus.normalize(n);
-}
-assert IOStatus.check(n);
-}
 } finally {
 writeLock.unlock();
 }
 }
-// package-private
+/**
+* Writes a byte of out of band data.
+*/
 int sendOutOfBandData(byte b) throws IOException {
 writeLock.lock();
 try {
-ensureWriteOpen();
+boolean blocking = isBlocking();
 int n = 0;
 try {
-begin();
+beginWrite(blocking);
-synchronized (stateLock) {
+if (blocking) {
-if (!isOpen())
+do {
-return 0;
+n = sendOutOfBandData(fd, b);
-writerThread = NativeThread.current();
+} while (n == IOStatus.INTERRUPTED && isOpen());
-}
+} else {
-for (;;) {
 n = sendOutOfBandData(fd, b);
-if ((n == IOStatus.INTERRUPTED) && isOpen())
-continue;
-return IOStatus.normalize(n);
 }
 } finally {
-writerCleanup();
+endWrite(blocking, n > 0);
-end((n > 0) || (n == IOStatus.UNAVAILABLE));
+if (n <= 0 && isOutputClosed)
-synchronized (stateLock) {
+throw new AsynchronousCloseException();
-if ((n <= 0) && (!isOutputOpen))
+}
-throw new AsynchronousCloseException();
+return IOStatus.normalize(n);
-}
-assert IOStatus.check(n);
-}
 } finally {
 writeLock.unlock();
 }
 }
+@Override
 protected void implConfigureBlocking(boolean block) throws IOException {
-IOUtil.configureBlocking(fd, block);
+readLock.lock();
-}
+try {
+writeLock.lock();
-public InetSocketAddress localAddress() {
+try {
+synchronized (stateLock) {
+ensureOpen();
+IOUtil.configureBlocking(fd, block);
+}
+} finally {
+writeLock.unlock();
+}
+} finally {
+readLock.unlock();
+}
+}
+/**
+* Returns the local address, or null if not bound
+*/
+InetSocketAddress localAddress() {
 synchronized (stateLock) {
 return localAddress;
 }
 }
-public SocketAddress remoteAddress() {
+/**
+* Returns the remote address, or null if not connected
+*/
+InetSocketAddress remoteAddress() {
 synchronized (stateLock) {
 return remoteAddress;
 }
 }
 readLock.lock();
 try {
 writeLock.lock();
 try {
 synchronized (stateLock) {
-if (!isOpen())
+ensureOpen();
-throw new ClosedChannelException();
+if (state == ST_CONNECTIONPENDING)
-if (state == ST_PENDING)
 throw new ConnectionPendingException();
 if (localAddress != null)
 throw new AlreadyBoundException();
 InetSocketAddress isa = (local == null) ?
 new InetSocketAddress(0) : Net.checkAddress(local);
 readLock.unlock();
 }
 return this;
 }
+@Override
 public boolean isConnected() {
 synchronized (stateLock) {
 return (state == ST_CONNECTED);
 }
 }
+@Override
 public boolean isConnectionPending() {
 synchronized (stateLock) {
-return (state == ST_PENDING);
+return (state == ST_CONNECTIONPENDING);
 }
 }
-void ensureOpenAndUnconnected() throws IOException { // package-private
+/**
-synchronized (stateLock) {
+* Marks the beginning of a connect operation that might block.
-if (!isOpen())
+*
-throw new ClosedChannelException();
+* @throws ClosedChannelException if the channel is closed
+* @throws AlreadyConnectedException if already connected
+* @throws ConnectionPendingException is a connection is pending
+*/
+private void beginConnect(boolean blocking) throws ClosedChannelException {
+if (blocking) {
+// set hook for Thread.interrupt
+begin();
+}
+synchronized (stateLock) {
+ensureOpen();
 if (state == ST_CONNECTED)
 throw new AlreadyConnectedException();
-if (state == ST_PENDING)
+if (state == ST_CONNECTIONPENDING)
 throw new ConnectionPendingException();
-}
+if (blocking)
-}
+readerThread = NativeThread.current();
+}
+}
+/**
+* Marks the end of a connect operation that may have blocked.
+*
+* @throws AsynchronousCloseException if the channel was closed due to this
+* thread being interrupted on a blocking connect operation.
+*/
+private void endConnect(boolean blocking, boolean completed)
+throws AsynchronousCloseException
+{
+endRead(blocking, completed);
+}
+@Override
 public boolean connect(SocketAddress sa) throws IOException {
+InetSocketAddress isa = Net.checkAddress(sa);
+SecurityManager sm = System.getSecurityManager();
+if (sm != null)
+sm.checkConnect(isa.getAddress().getHostAddress(), isa.getPort());
 readLock.lock();
 try {
 writeLock.lock();
 try {
-ensureOpenAndUnconnected();
+// notify before-connect hook
-InetSocketAddress isa = Net.checkAddress(sa);
+synchronized (stateLock) {
-SecurityManager sm = System.getSecurityManager();
+if (state == ST_UNCONNECTED && localAddress == null) {
-if (sm != null)
+NetHooks.beforeTcpConnect(fd, isa.getAddress(), isa.getPort());
-sm.checkConnect(isa.getAddress().getHostAddress(),
+}
-isa.getPort());
+}
-synchronized (blockingLock()) {
-int n = 0;
+InetAddress ia = isa.getAddress();
+if (ia.isAnyLocalAddress())
+ia = InetAddress.getLocalHost();
+int n = 0;
+boolean blocking = isBlocking();
+try {
 try {
-try {
+beginConnect(blocking);
-begin();
+if (blocking) {
-synchronized (stateLock) {
+do {
-if (!isOpen()) {
+n = Net.connect(fd, ia, isa.getPort());
-return false;
+} while (n == IOStatus.INTERRUPTED && isOpen());
-}
+} else {
-// notify hook only if unbound
+n = Net.connect(fd, ia, isa.getPort());
-if (localAddress == null) {
-NetHooks.beforeTcpConnect(fd,
-isa.getAddress(),
-isa.getPort());
-}
-readerThread = NativeThread.current();
-}
-for (;;) {
-InetAddress ia = isa.getAddress();
-if (ia.isAnyLocalAddress())
-ia = InetAddress.getLocalHost();
-n = Net.connect(fd,
-ia,
-isa.getPort());
-if ((n == IOStatus.INTERRUPTED) && isOpen())
-continue;
-break;
-}
-} finally {
-readerCleanup();
-end((n > 0) || (n == IOStatus.UNAVAILABLE));
-assert IOStatus.check(n);
 }
-} catch (IOException x) {
+} finally {
-// If an exception was thrown, close the channel after
+endConnect(blocking, n > 0);
-// invoking end() so as to avoid bogus
-// AsynchronousCloseExceptions
-close();
-throw x;
 }
-synchronized (stateLock) {
+} catch (IOException x) {
-remoteAddress = isa;
+// connect failed, close socket
-if (n > 0) {
+close();
+throw x;
-// Connection succeeded; disallow further
+}
-// invocation
-state = ST_CONNECTED;
+// connection may be established
-if (isOpen())
+synchronized (stateLock) {
-localAddress = Net.localAddress(fd);
+if (!isOpen())
-return true;
+throw new AsynchronousCloseException();
-}
+remoteAddress = isa;
-// If nonblocking and no exception then connection
+if (n > 0) {
-// pending; disallow another invocation
+// connected established
-if (!isBlocking())
+localAddress = Net.localAddress(fd);
-state = ST_PENDING;
+state = ST_CONNECTED;
-else
+return true;
-assert false;
+} else {
+// connection pending
+assert !blocking;
+state = ST_CONNECTIONPENDING;
+return false;
 }
 }
-return false;
 } finally {
 writeLock.unlock();
 }
 } finally {
 readLock.unlock();
 }
 }
+/**
+* Marks the beginning of a finishConnect operation that might block.
+*
+* @throws ClosedChannelException if the channel is closed
+* @throws NoConnectionPendingException if no connection is pending
+*/
+private void beginFinishConnect(boolean blocking) throws ClosedChannelException {
+if (blocking) {
+// set hook for Thread.interrupt
+begin();
+}
+synchronized (stateLock) {
+ensureOpen();
+if (state != ST_CONNECTIONPENDING)
+throw new NoConnectionPendingException();
+if (blocking)
+readerThread = NativeThread.current();
+}
+}
+/**
+* Marks the end of a finishConnect operation that may have blocked.
+*
+* @throws AsynchronousCloseException if the channel was closed due to this
+* thread being interrupted on a blocking connect operation.
+*/
+private void endFinishConnect(boolean blocking, boolean completed)
+throws AsynchronousCloseException
+{
+endRead(blocking, completed);
+}
+@Override
 public boolean finishConnect() throws IOException {
 readLock.lock();
 try {
 writeLock.lock();
 try {
+// already connected?
+synchronized (stateLock) {
+if (state == ST_CONNECTED)
+return true;
+}
+int n = 0;
+boolean blocking = isBlocking();
+try {
+try {
+beginFinishConnect(blocking);
+if (blocking) {
+do {
+n = checkConnect(fd, true);
+} while (n == 0 || (n == IOStatus.INTERRUPTED) && isOpen());
+} else {
+n = checkConnect(fd, false);
+}
+} finally {
+endFinishConnect(blocking, n > 0);
+}
+} catch (IOException x) {
+close();
+throw x;
+}
+// post finishConnect, connection may be established
 synchronized (stateLock) {
 if (!isOpen())
-throw new ClosedChannelException();
+throw new AsynchronousCloseException();
-if (state == ST_CONNECTED)
+if (n > 0) {
+// connection established
+localAddress = Net.localAddress(fd);
+state = ST_CONNECTED;
 return true;
-if (state != ST_PENDING)
+} else {
-throw new NoConnectionPendingException();
+// connection still pending
-}
+assert !blocking;
-int n = 0;
+return false;
-try {
-try {
-begin();
-synchronized (blockingLock()) {
-synchronized (stateLock) {
-if (!isOpen()) {
-return false;
-}
-readerThread = NativeThread.current();
-}
-if (!isBlocking()) {
-for (;;) {
-n = checkConnect(fd, false);
-if ((n == IOStatus.INTERRUPTED) && isOpen())
-continue;
-break;
-}
-} else {
-for (;;) {
-n = checkConnect(fd, true);
-if (n == 0) {
-// Loop in case of
-// spurious notifications
-continue;
-}
-if ((n == IOStatus.INTERRUPTED) && isOpen())
-continue;
-break;
-}
-}
-}
-} finally {
-synchronized (stateLock) {
-readerThread = 0;
-if (state == ST_KILLPENDING) {
-kill();
-// poll()/getsockopt() does not report
-// error (throws exception, with n = 0)
-// on Linux platform after dup2 and
-// signal-wakeup. Force n to 0 so the
-// end() can throw appropriate exception
-n = 0;
-}
-}
-end((n > 0) || (n == IOStatus.UNAVAILABLE));
-assert IOStatus.check(n);
 }
-} catch (IOException x) {
+}
-// If an exception was thrown, close the channel after
-// invoking end() so as to avoid bogus
-// AsynchronousCloseExceptions
-close();
-throw x;
-}
-if (n > 0) {
-synchronized (stateLock) {
-state = ST_CONNECTED;
-if (isOpen())
-localAddress = Net.localAddress(fd);
-}
-return true;
-}
-return false;
 } finally {
 writeLock.unlock();
 }
 } finally {
 readLock.unlock();
 }
 }
+/**
+* Invoked by implCloseChannel to close the channel.
+*
+* This method waits for outstanding I/O operations to complete. When in
+* blocking mode, the socket is pre-closed and the threads in blocking I/O
+* operations are signalled to ensure that the outstanding I/O operations
+* complete quickly.
+*
+* If the socket is connected then it is shutdown by this method. The
+* shutdown ensures that the peer reads EOF for the case that the socket is
+* not pre-closed or closed by this method.
+*
+* The socket is closed by this method when it is not registered with a
+* Selector. Note that a channel configured blocking may be registered with
+* a Selector. This arises when a key is canceled and the channel configured
+* to blocking mode before the key is flushed from the Selector.
+*/
+@Override
+protected void implCloseSelectableChannel() throws IOException {
+assert !isOpen();
+boolean blocking;
+boolean connected;
+boolean interrupted = false;
+// set state to ST_CLOSING
+synchronized (stateLock) {
+assert state < ST_CLOSING;
+blocking = isBlocking();
+connected = (state == ST_CONNECTED);
+state = ST_CLOSING;
+}
+// wait for any outstanding I/O operations to complete
+if (blocking) {
+synchronized (stateLock) {
+assert state == ST_CLOSING;
+long reader = readerThread;
+long writer = writerThread;
+if (reader != 0 || writer != 0) {
+nd.preClose(fd);
+connected = false; // fd is no longer connected socket
+if (reader != 0)
+NativeThread.signal(reader);
+if (writer != 0)
+NativeThread.signal(writer);
+// wait for blocking I/O operations to end
+while (readerThread != 0 || writerThread != 0) {
+try {
+stateLock.wait();
+} catch (InterruptedException e) {
+interrupted = true;
+}
+}
+}
+}
+} else {
+// non-blocking mode: wait for read/write to complete
+readLock.lock();
+try {
+writeLock.lock();
+writeLock.unlock();
+} finally {
+readLock.unlock();
+}
+}
+// set state to ST_KILLPENDING
+synchronized (stateLock) {
+assert state == ST_CLOSING;
+// if connected, and the channel is registered with a Selector, we
+// shutdown the output so that the peer reads EOF
+if (connected && isRegistered()) {
+try {
+Net.shutdown(fd, Net.SHUT_WR);
+} catch (IOException ignore) { }
+}
+state = ST_KILLPENDING;
+}
+// close socket if not registered with Selector
+if (!isRegistered())
+kill();
+// restore interrupt status
+if (interrupted)
+Thread.currentThread().interrupt();
+}
+@Override
+public void kill() throws IOException {
+synchronized (stateLock) {
+if (state == ST_KILLPENDING) {
+state = ST_KILLED;
+nd.close(fd);
+}
+}
+}
 @Override
 public SocketChannel shutdownInput() throws IOException {
 synchronized (stateLock) {
-if (!isOpen())
+ensureOpen();
-throw new ClosedChannelException();
 if (!isConnected())
 throw new NotYetConnectedException();
-if (isInputOpen) {
+if (!isInputClosed) {
 Net.shutdown(fd, Net.SHUT_RD);
-if (readerThread != 0)
+long thread = readerThread;
-NativeThread.signal(readerThread);
+if (thread != 0)
-isInputOpen = false;
+NativeThread.signal(thread);
+isInputClosed = true;
 }
 return this;
 }
 }
 @Override
 public SocketChannel shutdownOutput() throws IOException {
 synchronized (stateLock) {
-if (!isOpen())
+ensureOpen();
-throw new ClosedChannelException();
 if (!isConnected())
 throw new NotYetConnectedException();
-if (isOutputOpen) {
+if (!isOutputClosed) {
 Net.shutdown(fd, Net.SHUT_WR);
-if (writerThread != 0)
+long thread = writerThread;
-NativeThread.signal(writerThread);
+if (thread != 0)
-isOutputOpen = false;
+NativeThread.signal(thread);
+isOutputClosed = true;
 }
 return this;
 }
 }
-public boolean isInputOpen() {
+boolean isInputOpen() {
-synchronized (stateLock) {
+return !isInputClosed;
-return isInputOpen;
+}
-}
-}
+boolean isOutputOpen() {
+return !isOutputClosed;
-public boolean isOutputOpen() {
+}
-synchronized (stateLock) {
-return isOutputOpen;
+/**
-}
+* Poll this channel's socket for reading up to the given timeout.
-}
+* @return {@code true} if the socket is polled
+*/
-// AbstractInterruptibleChannel synchronizes invocations of this method
+boolean pollRead(long timeout) throws IOException {
-// using AbstractInterruptibleChannel.closeLock, and also ensures that this
+boolean blocking = isBlocking();
-// method is only ever invoked once.  Before we get to this method, isOpen
+assert Thread.holdsLock(blockingLock()) && blocking;
-// (which is volatile) will have been set to false.
-//
+readLock.lock();
-protected void implCloseSelectableChannel() throws IOException {
+try {
-synchronized (stateLock) {
+boolean polled = false;
-isInputOpen = false;
+try {
-isOutputOpen = false;
+beginRead(blocking);
+int n = Net.poll(fd, Net.POLLIN, timeout);
-// Close the underlying file descriptor and dup it to a known fd
+polled = (n > 0);
-// that's already closed.  This prevents other operations on this
+} finally {
-// channel from using the old fd, which might be recycled in the
+endRead(blocking, polled);
-// meantime and allocated to an entirely different channel.
+}
-//
+return polled;
-if (state != ST_KILLED)
+} finally {
-nd.preClose(fd);
+readLock.unlock();
+}
-// Signal native threads, if needed.  If a target thread is not
+}
-// currently blocked in an I/O operation then no harm is done since
-// the signal handler doesn't actually do anything.
+/**
-//
+* Poll this channel's socket for a connection, up to the given timeout.
-if (readerThread != 0)
+* @return {@code true} if the socket is polled
-NativeThread.signal(readerThread);
+*/
+boolean pollConnected(long timeout) throws IOException {
-if (writerThread != 0)
+boolean blocking = isBlocking();
-NativeThread.signal(writerThread);
+assert Thread.holdsLock(blockingLock()) && blocking;
-// If this channel is not registered then it's safe to close the fd
+readLock.lock();
-// immediately since we know at this point that no thread is
+try {
-// blocked in an I/O operation upon the channel and, since the
+writeLock.lock();
-// channel is marked closed, no thread will start another such
+try {
-// operation.  If this channel is registered then we don't close
+boolean polled = false;
-// the fd since it might be in use by a selector.  In that case
+try {
-// closing this channel caused its keys to be cancelled, so the
+beginFinishConnect(blocking);
-// last selector to deregister a key for this channel will invoke
+int n = Net.poll(fd, Net.POLLCONN, timeout);
-// kill() to close the fd.
+polled = (n > 0);
-//
+} finally {
-if (!isRegistered())
+endFinishConnect(blocking, polled);
-kill();
+}
-}
+return polled;
-}
+} finally {
+writeLock.unlock();
-public void kill() throws IOException {
+}
-synchronized (stateLock) {
+} finally {
-if (state == ST_KILLED)
+readLock.unlock();
-return;
-if (state == ST_UNINITIALIZED) {
-state = ST_KILLED;
-return;
-}
-assert !isOpen() && !isRegistered();
-// Postpone the kill if there is a waiting reader
-// or writer thread. See the comments in read() for
-// more detailed explanation.
-if (readerThread == 0 && writerThread == 0) {
-nd.close(fd);
-state = ST_KILLED;
-} else {
-state = ST_KILLPENDING;
-}
 }
 }
 /**
 * Translates native poll revent ops into a ready operation ops
 (state == ST_CONNECTED))
 newOps |= SelectionKey.OP_READ;
 if (((ops & Net.POLLCONN) != 0) &&
 ((intOps & SelectionKey.OP_CONNECT) != 0) &&
-((state == ST_UNCONNECTED) || (state == ST_PENDING))) {
+((state == ST_UNCONNECTED) || (state == ST_CONNECTIONPENDING))) {
 newOps |= SelectionKey.OP_CONNECT;
 }
 if (((ops & Net.POLLOUT) != 0) &&
 ((intOps & SelectionKey.OP_WRITE) != 0) &&
 return translateReadyOps(ops, sk.nioReadyOps(), sk);
 }
 public boolean translateAndSetReadyOps(int ops, SelectionKeyImpl sk) {
 return translateReadyOps(ops, 0, sk);
-}
-// package-private
-int poll(int events, long timeout) throws IOException {
-assert Thread.holdsLock(blockingLock()) && !isBlocking();
-readLock.lock();
-try {
-int n = 0;
-try {
-begin();
-synchronized (stateLock) {
-if (!isOpen())
-return 0;
-readerThread = NativeThread.current();
-}
-n = Net.poll(fd, events, timeout);
-} finally {
-readerCleanup();
-end(n > 0);
-}
-return n;
-} finally {
-readLock.unlock();
-}
 }
 /**
 * Translates an interest operation set into a native poll event set
 */
 synchronized (stateLock) {
 switch (state) {
 case ST_UNCONNECTED:
 sb.append("unconnected");
 break;
-case ST_PENDING:
+case ST_CONNECTIONPENDING:
 sb.append("connection-pending");
 break;
 case ST_CONNECTED:
 sb.append("connected");
-if (!isInputOpen)
+if (isInputClosed)
 sb.append(" ishut");
-if (!isOutputOpen)
+if (isOutputClosed)
 sb.append(" oshut");
 break;
 }
 InetSocketAddress addr = localAddress();
 if (addr != null) {

changeset 49001	ce06058197a4
parent 48761	74c1fa26435a
child 49141	ac95c7a76132