8198562: (ch) Separate blocking and non-blocking code paths (part 1)
8198754: (ch) Separate blocking and non-blocking code paths (part 2)
Reviewed-by: bpb
--- a/src/java.base/share/classes/java/lang/System.java Tue Feb 27 23:11:26 2018 -0800
+++ b/src/java.base/share/classes/java/lang/System.java Wed Feb 28 09:54:38 2018 +0000
@@ -2080,8 +2080,8 @@
E[] getEnumConstantsShared(Class<E> klass) {
return klass.getEnumConstantsShared();
}
- public void blockedOn(Thread t, Interruptible b) {
- t.blockedOn(b);
+ public void blockedOn(Interruptible b) {
+ Thread.blockedOn(b);
}
public void registerShutdownHook(int slot, boolean registerShutdownInProgress, Runnable hook) {
Shutdown.add(slot, registerShutdownInProgress, hook);
--- a/src/java.base/share/classes/java/lang/Thread.java Tue Feb 27 23:11:26 2018 -0800
+++ b/src/java.base/share/classes/java/lang/Thread.java Wed Feb 28 09:54:38 2018 +0000
@@ -231,9 +231,10 @@
/* Set the blocker field; invoked via jdk.internal.misc.SharedSecrets
* from java.nio code
*/
- void blockedOn(Interruptible b) {
- synchronized (blockerLock) {
- blocker = b;
+ static void blockedOn(Interruptible b) {
+ Thread me = Thread.currentThread();
+ synchronized (me.blockerLock) {
+ me.blocker = b;
}
}
@@ -1006,18 +1007,22 @@
* @spec JSR-51
*/
public void interrupt() {
- if (this != Thread.currentThread())
+ Thread me = Thread.currentThread();
+ if (this != me)
checkAccess();
- synchronized (blockerLock) {
- Interruptible b = blocker;
- if (b != null) {
- interrupt0(); // Just to set the interrupt flag
- b.interrupt(this);
- return;
+ // set interrupt status
+ interrupt0();
+
+ // thread may be blocked in an I/O operation
+ if (this != me && blocker != null) {
+ synchronized (blockerLock) {
+ Interruptible b = blocker;
+ if (b != null) {
+ b.interrupt(this);
+ }
}
}
- interrupt0();
}
/**
--- a/src/java.base/share/classes/java/nio/channels/spi/AbstractInterruptibleChannel.java Tue Feb 27 23:11:26 2018 -0800
+++ b/src/java.base/share/classes/java/nio/channels/spi/AbstractInterruptibleChannel.java Wed Feb 28 09:54:38 2018 +0000
@@ -1,5 +1,5 @@
/*
- * Copyright (c) 2000, 2017, Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2000, 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
@@ -205,6 +205,6 @@
// -- jdk.internal.misc.SharedSecrets --
static void blockedOn(Interruptible intr) { // package-private
- SharedSecrets.getJavaLangAccess().blockedOn(Thread.currentThread(), intr);
+ SharedSecrets.getJavaLangAccess().blockedOn(intr);
}
}
--- a/src/java.base/share/classes/jdk/internal/misc/JavaLangAccess.java Tue Feb 27 23:11:26 2018 -0800
+++ b/src/java.base/share/classes/jdk/internal/misc/JavaLangAccess.java Wed Feb 28 09:54:38 2018 +0000
@@ -1,5 +1,5 @@
/*
- * Copyright (c) 2003, 2017, Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2003, 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
@@ -100,9 +100,9 @@
<E extends Enum<E>> E[] getEnumConstantsShared(Class<E> klass);
/**
- * Set thread's blocker field.
+ * Set current thread's blocker field.
*/
- void blockedOn(Thread t, Interruptible b);
+ void blockedOn(Interruptible b);
/**
* Registers a shutdown hook.
--- a/src/java.base/share/classes/sun/nio/ch/DatagramChannelImpl.java Tue Feb 27 23:11:26 2018 -0800
+++ b/src/java.base/share/classes/sun/nio/ch/DatagramChannelImpl.java Wed Feb 28 09:54:38 2018 +0000
@@ -41,15 +41,17 @@
import java.net.StandardSocketOptions;
import java.nio.ByteBuffer;
import java.nio.channels.AlreadyBoundException;
+import java.nio.channels.AlreadyConnectedException;
+import java.nio.channels.AsynchronousCloseException;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.DatagramChannel;
import java.nio.channels.MembershipKey;
import java.nio.channels.NotYetConnectedException;
import java.nio.channels.SelectionKey;
-import java.nio.channels.UnsupportedAddressTypeException;
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;
@@ -64,21 +66,16 @@
extends DatagramChannel
implements SelChImpl
{
-
// Used to make native read and write calls
private static NativeDispatcher nd = new DatagramDispatcher();
+ // The protocol family of the socket
+ private final ProtocolFamily family;
+
// Our file descriptor
private final FileDescriptor fd;
private final int fdVal;
- // The protocol family of the socket
- private final ProtocolFamily family;
-
- // IDs of native threads doing reads and writes, for signalling
- private volatile long readerThread;
- private volatile long writerThread;
-
// Cached InetAddress and port for unconnected DatagramChannels
// used by receive0
private InetAddress cachedSenderInetAddress;
@@ -97,13 +94,18 @@
// -- The following fields are protected by stateLock
// State (does not necessarily increase monotonically)
- private static final int ST_UNINITIALIZED = -1;
private static final int ST_UNCONNECTED = 0;
private static final int ST_CONNECTED = 1;
- private static final int ST_KILLED = 2;
- private int state = ST_UNINITIALIZED;
+ private static final int ST_CLOSING = 2;
+ private static final int ST_KILLPENDING = 3;
+ private static final int ST_KILLED = 4;
+ private int state;
- // Binding
+ // IDs of native threads doing reads and writes, for signalling
+ private long readerThread;
+ private long writerThread;
+
+ // Binding and remote address (when connected)
private InetSocketAddress localAddress;
private InetSocketAddress remoteAddress;
@@ -127,11 +129,11 @@
super(sp);
ResourceManager.beforeUdpCreate();
try {
- this.family = Net.isIPv6Available() ?
- StandardProtocolFamily.INET6 : StandardProtocolFamily.INET;
+ this.family = Net.isIPv6Available()
+ ? StandardProtocolFamily.INET6
+ : StandardProtocolFamily.INET;
this.fd = Net.socket(family, false);
this.fdVal = IOUtil.fdVal(fd);
- this.state = ST_UNCONNECTED;
} catch (IOException ioe) {
ResourceManager.afterUdpClose();
throw ioe;
@@ -142,13 +144,10 @@
throws IOException
{
super(sp);
+ Objects.requireNonNull(family, "'family' is null");
if ((family != StandardProtocolFamily.INET) &&
- (family != StandardProtocolFamily.INET6))
- {
- if (family == null)
- throw new NullPointerException("'family' is null");
- else
- throw new UnsupportedOperationException("Protocol family not supported");
+ (family != StandardProtocolFamily.INET6)) {
+ throw new UnsupportedOperationException("Protocol family not supported");
}
if (family == StandardProtocolFamily.INET6) {
if (!Net.isIPv6Available()) {
@@ -161,7 +160,6 @@
this.family = family;
this.fd = Net.socket(family, false);
this.fdVal = IOUtil.fdVal(fd);
- this.state = ST_UNCONNECTED;
} catch (IOException ioe) {
ResourceManager.afterUdpClose();
throw ioe;
@@ -176,14 +174,23 @@
// increment UDP count to match decrement when closing
ResourceManager.beforeUdpCreate();
- this.family = Net.isIPv6Available() ?
- StandardProtocolFamily.INET6 : StandardProtocolFamily.INET;
+ this.family = Net.isIPv6Available()
+ ? StandardProtocolFamily.INET6
+ : StandardProtocolFamily.INET;
this.fd = fd;
this.fdVal = IOUtil.fdVal(fd);
- this.state = ST_UNCONNECTED;
- this.localAddress = Net.localAddress(fd);
+ synchronized (stateLock) {
+ this.localAddress = Net.localAddress(fd);
+ }
}
+ // @throws ClosedChannelException if channel is closed
+ private void ensureOpen() throws ClosedChannelException {
+ if (!isOpen())
+ throw new ClosedChannelException();
+ }
+
+ @Override
public DatagramSocket socket() {
synchronized (stateLock) {
if (socket == null)
@@ -195,8 +202,7 @@
@Override
public SocketAddress getLocalAddress() throws IOException {
synchronized (stateLock) {
- if (!isOpen())
- throw new ClosedChannelException();
+ ensureOpen();
// Perform security check before returning address
return Net.getRevealedLocalAddress(localAddress);
}
@@ -205,8 +211,7 @@
@Override
public SocketAddress getRemoteAddress() throws IOException {
synchronized (stateLock) {
- if (!isOpen())
- throw new ClosedChannelException();
+ ensureOpen();
return remoteAddress;
}
}
@@ -215,8 +220,7 @@
public <T> DatagramChannel setOption(SocketOption<T> name, T value)
throws IOException
{
- if (name == null)
- throw new NullPointerException();
+ Objects.requireNonNull(name);
if (!supportedOptions().contains(name))
throw new UnsupportedOperationException("'" + name + "' not supported");
@@ -251,9 +255,8 @@
}
return this;
}
- if (name == StandardSocketOptions.SO_REUSEADDR &&
- Net.useExclusiveBind() && localAddress != null)
- {
+ if (name == StandardSocketOptions.SO_REUSEADDR
+ && Net.useExclusiveBind() && localAddress != null) {
reuseAddressEmulated = true;
this.isReuseAddress = (Boolean)value;
}
@@ -269,8 +272,7 @@
public <T> T getOption(SocketOption<T> name)
throws IOException
{
- if (name == null)
- throw new NullPointerException();
+ Objects.requireNonNull(name);
if (!supportedOptions().contains(name))
throw new UnsupportedOperationException("'" + name + "' not supported");
@@ -307,9 +309,7 @@
}
}
- if (name == StandardSocketOptions.SO_REUSEADDR &&
- reuseAddressEmulated)
- {
+ if (name == StandardSocketOptions.SO_REUSEADDR && reuseAddressEmulated) {
return (T)Boolean.valueOf(isReuseAddress);
}
@@ -322,7 +322,7 @@
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_REUSEADDR);
@@ -334,9 +334,7 @@
set.add(StandardSocketOptions.IP_MULTICAST_IF);
set.add(StandardSocketOptions.IP_MULTICAST_TTL);
set.add(StandardSocketOptions.IP_MULTICAST_LOOP);
- ExtendedSocketOptions extendedOptions =
- ExtendedSocketOptions.getInstance();
- set.addAll(extendedOptions.options());
+ set.addAll(ExtendedSocketOptions.getInstance().options());
return Collections.unmodifiableSet(set);
}
}
@@ -346,33 +344,78 @@
return DefaultOptionsHolder.defaultOptions;
}
- private void ensureOpen() throws ClosedChannelException {
- if (!isOpen())
- throw new ClosedChannelException();
+ /**
+ * Marks the beginning of a read operation that might block.
+ *
+ * @param blocking true if configured blocking
+ * @param mustBeConnected true if the socket must be connected
+ * @return remote address if connected
+ * @throws ClosedChannelException if the channel is closed
+ * @throws NotYetConnectedException if mustBeConnected and not connected
+ * @throws IOException if socket not bound and cannot be bound
+ */
+ private SocketAddress beginRead(boolean blocking, boolean mustBeConnected)
+ throws IOException
+ {
+ if (blocking) {
+ // set hook for Thread.interrupt
+ begin();
+ }
+ SocketAddress remote;
+ synchronized (stateLock) {
+ ensureOpen();
+ remote = remoteAddress;
+ if ((remote == null) && mustBeConnected)
+ throw new NotYetConnectedException();
+ if (localAddress == null)
+ bindInternal(null);
+ if (blocking)
+ readerThread = NativeThread.current();
+ }
+ return remote;
+ }
+
+ /**
+ * Marks the end of a read operation that may have blocked.
+ *
+ * @throws AsynchronousCloseException if the channel was closed asynchronously
+ */
+ private void endRead(boolean blocking, boolean completed)
+ throws AsynchronousCloseException
+ {
+ if (blocking) {
+ synchronized (stateLock) {
+ readerThread = 0;
+ // notify any thread waiting in implCloseSelectableChannel
+ if (state == ST_CLOSING) {
+ stateLock.notifyAll();
+ }
+ }
+ // remove hook for Thread.interrupt
+ end(completed);
+ }
}
private SocketAddress sender; // Set by receive0 (## ugh)
+ @Override
public SocketAddress receive(ByteBuffer dst) throws IOException {
if (dst.isReadOnly())
throw new IllegalArgumentException("Read-only buffer");
+
readLock.lock();
try {
- ensureOpen();
- // Socket was not bound before attempting receive
- if (localAddress() == null)
- bind(null);
+ boolean blocking = isBlocking();
int n = 0;
ByteBuffer bb = null;
try {
- begin();
- if (!isOpen())
- return null;
- SecurityManager security = System.getSecurityManager();
- readerThread = NativeThread.current();
- if (isConnected() || (security == null)) {
+ SocketAddress remote = beginRead(blocking, false);
+ boolean connected = (remote != null);
+ SecurityManager sm = System.getSecurityManager();
+ if (connected || (sm == null)) {
+ // connected or no security manager
do {
- n = receive(fd, dst);
+ n = receive(fd, dst, connected);
} while ((n == IOStatus.INTERRUPTED) && isOpen());
if (n == IOStatus.UNAVAILABLE)
return null;
@@ -382,15 +425,14 @@
bb = Util.getTemporaryDirectBuffer(dst.remaining());
for (;;) {
do {
- n = receive(fd, bb);
+ n = receive(fd, bb, connected);
} while ((n == IOStatus.INTERRUPTED) && isOpen());
if (n == IOStatus.UNAVAILABLE)
return null;
InetSocketAddress isa = (InetSocketAddress)sender;
try {
- security.checkAccept(
- isa.getAddress().getHostAddress(),
- isa.getPort());
+ sm.checkAccept(isa.getAddress().getHostAddress(),
+ isa.getPort());
} catch (SecurityException se) {
// Ignore packet
bb.clear();
@@ -402,12 +444,12 @@
break;
}
}
+ assert sender != null;
return sender;
} finally {
if (bb != null)
Util.releaseTemporaryDirectBuffer(bb);
- readerThread = 0;
- end((n > 0) || (n == IOStatus.UNAVAILABLE));
+ endRead(blocking, n > 0);
assert IOStatus.check(n);
}
} finally {
@@ -415,7 +457,7 @@
}
}
- private int receive(FileDescriptor fd, ByteBuffer dst)
+ private int receive(FileDescriptor fd, ByteBuffer dst, boolean connected)
throws IOException
{
int pos = dst.position();
@@ -423,7 +465,7 @@
assert (pos <= lim);
int rem = (pos <= lim ? lim - pos : 0);
if (dst instanceof DirectBuffer && rem > 0)
- return receiveIntoNativeBuffer(fd, dst, rem, pos);
+ return receiveIntoNativeBuffer(fd, dst, rem, pos, connected);
// Substitute a native buffer. If the supplied buffer is empty
// we must instead use a nonempty buffer, otherwise the call
@@ -431,7 +473,7 @@
int newSize = Math.max(rem, 1);
ByteBuffer bb = Util.getTemporaryDirectBuffer(newSize);
try {
- int n = receiveIntoNativeBuffer(fd, bb, newSize, 0);
+ int n = receiveIntoNativeBuffer(fd, bb, newSize, 0, connected);
bb.flip();
if (n > 0 && rem > 0)
dst.put(bb);
@@ -442,11 +484,10 @@
}
private int receiveIntoNativeBuffer(FileDescriptor fd, ByteBuffer bb,
- int rem, int pos)
+ int rem, int pos, boolean connected)
throws IOException
{
- int n = receive0(fd, ((DirectBuffer)bb).address() + pos, rem,
- isConnected());
+ int n = receive0(fd, ((DirectBuffer)bb).address() + pos, rem, connected);
if (n > 0)
bb.position(pos + n);
return n;
@@ -455,59 +496,44 @@
public int send(ByteBuffer src, SocketAddress target)
throws IOException
{
- if (src == null)
- throw new NullPointerException();
+ Objects.requireNonNull(src);
+ InetSocketAddress isa = Net.checkAddress(target, family);
writeLock.lock();
try {
- ensureOpen();
- InetSocketAddress isa = Net.checkAddress(target);
- InetAddress ia = isa.getAddress();
- if (ia == null)
- throw new IOException("Target address not resolved");
- synchronized (stateLock) {
- if (!isConnected()) {
- if (target == null)
- throw new NullPointerException();
+ boolean blocking = isBlocking();
+ int n = 0;
+ try {
+ SocketAddress remote = beginWrite(blocking, false);
+ if (remote != null) {
+ // connected
+ if (!target.equals(remote)) {
+ throw new IllegalArgumentException(
+ "Connected address not equal to target address");
+ }
+ do {
+ n = IOUtil.write(fd, src, -1, nd);
+ } while ((n == IOStatus.INTERRUPTED) && isOpen());
+ } else {
+ // not connected
SecurityManager sm = System.getSecurityManager();
if (sm != null) {
+ InetAddress ia = isa.getAddress();
if (ia.isMulticastAddress()) {
sm.checkMulticast(ia);
} else {
- sm.checkConnect(ia.getHostAddress(),
- isa.getPort());
+ sm.checkConnect(ia.getHostAddress(), isa.getPort());
}
}
- } else { // Connected case; Check address then write
- if (!target.equals(remoteAddress)) {
- throw new IllegalArgumentException(
- "Connected address not equal to target address");
- }
- return write(src);
+ do {
+ n = send(fd, src, isa);
+ } while ((n == IOStatus.INTERRUPTED) && isOpen());
}
- }
-
- int n = 0;
- try {
- begin();
- if (!isOpen())
- return 0;
- writerThread = NativeThread.current();
- do {
- n = send(fd, src, isa);
- } while ((n == IOStatus.INTERRUPTED) && isOpen());
-
- synchronized (stateLock) {
- if (isOpen() && (localAddress == null)) {
- localAddress = Net.localAddress(fd);
- }
- }
- return IOStatus.normalize(n);
} finally {
- writerThread = 0;
- end((n > 0) || (n == IOStatus.UNAVAILABLE));
+ endWrite(blocking, n > 0);
assert IOStatus.check(n);
}
+ return IOStatus.normalize(n);
} finally {
writeLock.unlock();
}
@@ -567,141 +593,180 @@
return written;
}
+ @Override
public int read(ByteBuffer buf) throws IOException {
- if (buf == null)
- throw new NullPointerException();
+ Objects.requireNonNull(buf);
+
readLock.lock();
try {
- synchronized (stateLock) {
- ensureOpen();
- if (!isConnected())
- throw new NotYetConnectedException();
- }
+ boolean blocking = isBlocking();
int n = 0;
try {
- begin();
- if (!isOpen())
- return 0;
- readerThread = NativeThread.current();
+ beginRead(blocking, true);
do {
n = IOUtil.read(fd, buf, -1, nd);
} while ((n == IOStatus.INTERRUPTED) && isOpen());
- return IOStatus.normalize(n);
+
} finally {
- readerThread = 0;
- end((n > 0) || (n == IOStatus.UNAVAILABLE));
+ endRead(blocking, n > 0);
assert IOStatus.check(n);
}
+ return IOStatus.normalize(n);
+ } finally {
+ readLock.unlock();
+ }
+ }
+
+ @Override
+ public long read(ByteBuffer[] dsts, int offset, int length)
+ throws IOException
+ {
+ Objects.checkFromIndexSize(offset, length, dsts.length);
+
+ readLock.lock();
+ try {
+ boolean blocking = isBlocking();
+ long n = 0;
+ try {
+ beginRead(blocking, true);
+ do {
+ n = IOUtil.read(fd, dsts, offset, length, nd);
+ } while ((n == IOStatus.INTERRUPTED) && isOpen());
+
+ } finally {
+ endRead(blocking, n > 0);
+ assert IOStatus.check(n);
+ }
+ return IOStatus.normalize(n);
} finally {
readLock.unlock();
}
}
- public long read(ByteBuffer[] dsts, int offset, int length)
+ /**
+ * Marks the beginning of a write operation that might block.
+ * @param blocking true if configured blocking
+ * @param mustBeConnected true if the socket must be connected
+ * @return remote address if connected
+ * @throws ClosedChannelException if the channel is closed
+ * @throws NotYetConnectedException if mustBeConnected and not connected
+ * @throws IOException if socket not bound and cannot be bound
+ */
+ private SocketAddress beginWrite(boolean blocking, boolean mustBeConnected)
throws IOException
{
- if ((offset < 0) || (length < 0) || (offset > dsts.length - length))
- throw new IndexOutOfBoundsException();
+ if (blocking) {
+ // set hook for Thread.interrupt
+ begin();
+ }
+ SocketAddress remote;
+ synchronized (stateLock) {
+ ensureOpen();
+ remote = remoteAddress;
+ if ((remote == null) && mustBeConnected)
+ throw new NotYetConnectedException();
+ if (localAddress == null)
+ bindInternal(null);
+ if (blocking)
+ writerThread = NativeThread.current();
+ }
+ return remote;
+ }
+
+ /**
+ * Marks the end of a write operation that may have blocked.
+ *
+ * @throws AsynchronousCloseException if the channel was closed asynchronously
+ */
+ 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 {
+ Objects.requireNonNull(buf);
+
+ writeLock.lock();
+ try {
+ boolean blocking = isBlocking();
+ int n = 0;
+ try {
+ beginWrite(blocking, true);
+ do {
+ n = IOUtil.write(fd, buf, -1, nd);
+ } while ((n == IOStatus.INTERRUPTED) && isOpen());
+ } finally {
+ endWrite(blocking, n > 0);
+ assert IOStatus.check(n);
+ }
+ return IOStatus.normalize(n);
+ } finally {
+ writeLock.unlock();
+ }
+ }
+
+ @Override
+ public long write(ByteBuffer[] srcs, int offset, int length)
+ throws IOException
+ {
+ Objects.checkFromIndexSize(offset, length, srcs.length);
+
+ writeLock.lock();
+ try {
+ boolean blocking = isBlocking();
+ long n = 0;
+ try {
+ beginWrite(blocking, true);
+ do {
+ n = IOUtil.write(fd, srcs, offset, length, nd);
+ } while ((n == IOStatus.INTERRUPTED) && isOpen());
+ } finally {
+ endWrite(blocking, n > 0);
+ assert IOStatus.check(n);
+ }
+ return IOStatus.normalize(n);
+ } finally {
+ writeLock.unlock();
+ }
+ }
+
+ @Override
+ protected void implConfigureBlocking(boolean block) throws IOException {
readLock.lock();
try {
- synchronized (stateLock) {
- ensureOpen();
- if (!isConnected())
- throw new NotYetConnectedException();
- }
- long n = 0;
+ writeLock.lock();
try {
- begin();
- if (!isOpen())
- return 0;
- readerThread = NativeThread.current();
- do {
- n = IOUtil.read(fd, dsts, offset, length, nd);
- } while ((n == IOStatus.INTERRUPTED) && isOpen());
- return IOStatus.normalize(n);
+ synchronized (stateLock) {
+ ensureOpen();
+ IOUtil.configureBlocking(fd, block);
+ }
} finally {
- readerThread = 0;
- end((n > 0) || (n == IOStatus.UNAVAILABLE));
- assert IOStatus.check(n);
+ writeLock.unlock();
}
} finally {
readLock.unlock();
}
}
- public int write(ByteBuffer buf) throws IOException {
- if (buf == null)
- throw new NullPointerException();
- writeLock.lock();
- try {
- synchronized (stateLock) {
- ensureOpen();
- if (!isConnected())
- throw new NotYetConnectedException();
- }
- int n = 0;
- try {
- begin();
- if (!isOpen())
- return 0;
- writerThread = NativeThread.current();
- do {
- n = IOUtil.write(fd, buf, -1, nd);
- } while ((n == IOStatus.INTERRUPTED) && isOpen());
- return IOStatus.normalize(n);
- } finally {
- writerThread = 0;
- end((n > 0) || (n == IOStatus.UNAVAILABLE));
- assert IOStatus.check(n);
- }
- } finally {
- writeLock.unlock();
- }
- }
-
- public long write(ByteBuffer[] srcs, int offset, int length)
- throws IOException
- {
- if ((offset < 0) || (length < 0) || (offset > srcs.length - length))
- throw new IndexOutOfBoundsException();
- writeLock.lock();
- try {
- synchronized (stateLock) {
- ensureOpen();
- if (!isConnected())
- throw new NotYetConnectedException();
- }
- long n = 0;
- try {
- begin();
- if (!isOpen())
- return 0;
- writerThread = NativeThread.current();
- do {
- n = IOUtil.write(fd, srcs, offset, length, nd);
- } while ((n == IOStatus.INTERRUPTED) && isOpen());
- return IOStatus.normalize(n);
- } finally {
- writerThread = 0;
- end((n > 0) || (n == IOStatus.UNAVAILABLE));
- assert IOStatus.check(n);
- }
- } finally {
- writeLock.unlock();
- }
- }
-
- protected void implConfigureBlocking(boolean block) throws IOException {
- IOUtil.configureBlocking(fd, block);
- }
-
- public SocketAddress localAddress() {
+ InetSocketAddress localAddress() {
synchronized (stateLock) {
return localAddress;
}
}
- public SocketAddress remoteAddress() {
+ InetSocketAddress remoteAddress() {
synchronized (stateLock) {
return remoteAddress;
}
@@ -717,30 +782,7 @@
ensureOpen();
if (localAddress != null)
throw new AlreadyBoundException();
- InetSocketAddress isa;
- if (local == null) {
- // only Inet4Address allowed with IPv4 socket
- if (family == StandardProtocolFamily.INET) {
- isa = new InetSocketAddress(InetAddress.getByName("0.0.0.0"), 0);
- } else {
- isa = new InetSocketAddress(0);
- }
- } else {
- isa = Net.checkAddress(local);
-
- // only Inet4Address allowed with IPv4 socket
- if (family == StandardProtocolFamily.INET) {
- InetAddress addr = isa.getAddress();
- if (!(addr instanceof Inet4Address))
- throw new UnsupportedAddressTypeException();
- }
- }
- SecurityManager sm = System.getSecurityManager();
- if (sm != null) {
- sm.checkListen(isa.getPort());
- }
- Net.bind(family, fd, isa.getAddress(), isa.getPort());
- localAddress = Net.localAddress(fd);
+ bindInternal(local);
}
} finally {
writeLock.unlock();
@@ -751,34 +793,58 @@
return this;
}
+ private void bindInternal(SocketAddress local) throws IOException {
+ assert Thread.holdsLock(stateLock) && (localAddress == null);
+
+ InetSocketAddress isa;
+ if (local == null) {
+ // only Inet4Address allowed with IPv4 socket
+ if (family == StandardProtocolFamily.INET) {
+ isa = new InetSocketAddress(InetAddress.getByName("0.0.0.0"), 0);
+ } else {
+ isa = new InetSocketAddress(0);
+ }
+ } else {
+ isa = Net.checkAddress(local, family);
+ }
+ SecurityManager sm = System.getSecurityManager();
+ if (sm != null)
+ sm.checkListen(isa.getPort());
+
+ Net.bind(family, fd, isa.getAddress(), isa.getPort());
+ localAddress = Net.localAddress(fd);
+ }
+
+ @Override
public boolean isConnected() {
synchronized (stateLock) {
return (state == ST_CONNECTED);
}
}
- void ensureOpenAndUnconnected() throws IOException { // package-private
- synchronized (stateLock) {
- if (!isOpen())
- throw new ClosedChannelException();
- if (state != ST_UNCONNECTED)
- throw new IllegalStateException("Connect already invoked");
- }
- }
-
@Override
public DatagramChannel connect(SocketAddress sa) throws IOException {
+ InetSocketAddress isa = Net.checkAddress(sa, family);
+ SecurityManager sm = System.getSecurityManager();
+ if (sm != null) {
+ InetAddress ia = isa.getAddress();
+ if (ia.isMulticastAddress()) {
+ sm.checkMulticast(ia);
+ } else {
+ sm.checkConnect(ia.getHostAddress(), isa.getPort());
+ sm.checkAccept(ia.getHostAddress(), isa.getPort());
+ }
+ }
+
readLock.lock();
try {
writeLock.lock();
try {
synchronized (stateLock) {
- ensureOpenAndUnconnected();
- InetSocketAddress isa = Net.checkAddress(sa);
- SecurityManager sm = System.getSecurityManager();
- if (sm != null)
- sm.checkConnect(isa.getAddress().getHostAddress(),
- isa.getPort());
+ ensureOpen();
+ if (state == ST_CONNECTED)
+ throw new AlreadyConnectedException();
+
int n = Net.connect(family,
fd,
isa.getAddress(),
@@ -786,31 +852,26 @@
if (n <= 0)
throw new Error(); // Can't happen
- // Connection succeeded; disallow further invocation
- state = ST_CONNECTED;
+ // connected
remoteAddress = isa;
- sender = isa;
- cachedSenderInetAddress = isa.getAddress();
- cachedSenderPort = isa.getPort();
+ state = ST_CONNECTED;
- // set or refresh local address
+ // refresh local address
localAddress = Net.localAddress(fd);
// flush any packets already received.
- synchronized (blockingLock()) {
- boolean blocking = isBlocking();
- try {
- ByteBuffer tmpBuf = ByteBuffer.allocate(100);
- if (blocking) {
- configureBlocking(false);
- }
- do {
- tmpBuf.clear();
- } while (receive(tmpBuf) != null);
- } finally {
- if (blocking) {
- configureBlocking(true);
- }
+ boolean blocking = isBlocking();
+ if (blocking) {
+ IOUtil.configureBlocking(fd, false);
+ }
+ try {
+ ByteBuffer buf = ByteBuffer.allocate(100);
+ while (receive(buf) != null) {
+ buf.clear();
+ }
+ } finally {
+ if (blocking) {
+ IOUtil.configureBlocking(fd, true);
}
}
}
@@ -823,21 +884,21 @@
return this;
}
+ @Override
public DatagramChannel disconnect() throws IOException {
readLock.lock();
try {
writeLock.lock();
try {
synchronized (stateLock) {
- if (!isConnected() || !isOpen())
+ if (!isOpen() || (state != ST_CONNECTED))
return this;
- InetSocketAddress isa = remoteAddress;
- SecurityManager sm = System.getSecurityManager();
- if (sm != null)
- sm.checkConnect(isa.getAddress().getHostAddress(),
- isa.getPort());
+
+ // disconnect socket
boolean isIPv6 = (family == StandardProtocolFamily.INET6);
disconnect0(fd, isIPv6);
+
+ // no longer connected
remoteAddress = null;
state = ST_UNCONNECTED;
@@ -891,8 +952,7 @@
sm.checkMulticast(group);
synchronized (stateLock) {
- if (!isOpen())
- throw new ClosedChannelException();
+ ensureOpen();
// check the registry to see if we are already a member of the group
if (registry == null) {
@@ -963,8 +1023,7 @@
InetAddress source)
throws IOException
{
- if (source == null)
- throw new NullPointerException("source address is null");
+ Objects.requireNonNull(source);
return innerJoin(group, interf, source);
}
@@ -1065,37 +1124,99 @@
}
}
+ /**
+ * 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.
+ *
+ * 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 interrupted = false;
+
+ // set state to ST_CLOSING and invalid membership keys
synchronized (stateLock) {
- if (state != ST_KILLED)
- nd.preClose(fd);
- ResourceManager.afterUdpClose();
+ assert state < ST_CLOSING;
+ blocking = isBlocking();
+ state = ST_CLOSING;
- // if member of mulitcast group then invalidate all keys
+ // if member of any multicast groups then invalidate the keys
if (registry != null)
registry.invalidateAll();
+ }
- long th;
- if ((th = readerThread) != 0)
- NativeThread.signal(th);
- if ((th = writerThread) != 0)
- NativeThread.signal(th);
- if (!isRegistered())
- kill();
+ // 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);
+
+ 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;
+ 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_KILLED)
- return;
- if (state == ST_UNINITIALIZED) {
+ if (state == ST_KILLPENDING) {
state = ST_KILLED;
- return;
+ try {
+ nd.close(fd);
+ } finally {
+ // notify resource manager
+ ResourceManager.afterUdpClose();
+ }
}
- assert !isOpen() && !isRegistered();
- nd.close(fd);
- state = ST_KILLED;
}
}
@@ -1148,26 +1269,25 @@
return translateReadyOps(ops, 0, sk);
}
- // package-private
- int poll(int events, long timeout) throws IOException {
- assert Thread.holdsLock(blockingLock()) && !isBlocking();
+ /**
+ * Poll this channel's socket for reading up to the given timeout.
+ * @return {@code true} if the socket is polled
+ */
+ boolean pollRead(long timeout) throws IOException {
+ boolean blocking = isBlocking();
+ assert Thread.holdsLock(blockingLock()) && blocking;
readLock.lock();
try {
- int n = 0;
+ boolean polled = false;
try {
- begin();
- synchronized (stateLock) {
- if (!isOpen())
- return 0;
- readerThread = NativeThread.current();
- }
- n = Net.poll(fd, events, timeout);
+ beginRead(blocking, false);
+ int n = Net.poll(fd, Net.POLLIN, timeout);
+ polled = (n > 0);
} finally {
- readerThread = 0;
- end(n > 0);
+ endRead(blocking, polled);
}
- return n;
+ return polled;
} finally {
readLock.unlock();
}
@@ -1216,5 +1336,4 @@
IOUtil.load();
initIDs();
}
-
}
--- a/src/java.base/share/classes/sun/nio/ch/DatagramSocketAdaptor.java Tue Feb 27 23:11:26 2018 -0800
+++ b/src/java.base/share/classes/sun/nio/ch/DatagramSocketAdaptor.java Wed Feb 28 09:54:38 2018 +0000
@@ -41,6 +41,7 @@
import java.nio.channels.ClosedChannelException;
import java.nio.channels.DatagramChannel;
import java.nio.channels.IllegalBlockingModeException;
+import java.util.Objects;
// Make a datagram-socket channel look like a datagram socket.
@@ -53,7 +54,6 @@
public class DatagramSocketAdaptor
extends DatagramSocket
{
-
// The channel being adapted
private final DatagramChannelImpl dc;
@@ -63,7 +63,7 @@
// ## super will create a useless impl
private DatagramSocketAdaptor(DatagramChannelImpl dc) throws IOException {
// Invoke the DatagramSocketAdaptor(SocketAddress) constructor,
- // passing a dummy DatagramSocketImpl object to aovid any native
+ // passing a dummy DatagramSocketImpl object to avoid any native
// resource allocation in super class and invoking our bind method
// before the dc field is initialized.
super(dummyDatagramSocket);
@@ -87,10 +87,10 @@
throw new IllegalArgumentException("connect: " + port);
if (remote == null)
throw new IllegalArgumentException("connect: null address");
- if (isClosed())
- return;
try {
dc.connect(remote);
+ } catch (ClosedChannelException e) {
+ // ignore
} catch (Exception x) {
Net.translateToSocketException(x);
}
@@ -115,8 +115,7 @@
}
public void connect(SocketAddress remote) throws SocketException {
- if (remote == null)
- throw new IllegalArgumentException("Address can't be null");
+ Objects.requireNonNull(remote, "Address can't be null");
connectInternal(remote);
}
@@ -137,15 +136,13 @@
}
public InetAddress getInetAddress() {
- return (isConnected()
- ? Net.asInetSocketAddress(dc.remoteAddress()).getAddress()
- : null);
+ InetSocketAddress remote = dc.remoteAddress();
+ return (remote != null) ? remote.getAddress() : null;
}
public int getPort() {
- return (isConnected()
- ? Net.asInetSocketAddress(dc.remoteAddress()).getPort()
- : -1);
+ InetSocketAddress remote = dc.remoteAddress();
+ return (remote != null) ? remote.getPort() : -1;
}
public void send(DatagramPacket p) throws IOException {
@@ -161,8 +158,7 @@
if (p.getAddress() == null) {
// Legacy DatagramSocket will send in this case
// and set address and port of the packet
- InetSocketAddress isa = (InetSocketAddress)
- dc.remoteAddress();
+ InetSocketAddress isa = dc.remoteAddress();
p.setPort(isa.getPort());
p.setAddress(isa.getAddress());
dc.write(bb);
@@ -181,36 +177,24 @@
}
}
- // Must hold dc.blockingLock()
- //
private SocketAddress receive(ByteBuffer bb) throws IOException {
- if (timeout == 0) {
- return dc.receive(bb);
- }
+ assert Thread.holdsLock(dc.blockingLock()) && dc.isBlocking();
- dc.configureBlocking(false);
- try {
- SocketAddress sender;
- if ((sender = dc.receive(bb)) != null)
- return sender;
- long to = timeout;
+ long to = this.timeout;
+ if (to == 0) {
+ return dc.receive(bb);
+ } else {
for (;;) {
if (!dc.isOpen())
- throw new ClosedChannelException();
+ throw new ClosedChannelException();
long st = System.currentTimeMillis();
- int result = dc.poll(Net.POLLIN, to);
- if (result > 0 && ((result & Net.POLLIN) != 0)) {
- if ((sender = dc.receive(bb)) != null)
- return sender;
+ if (dc.pollRead(to)) {
+ return dc.receive(bb);
}
to -= System.currentTimeMillis() - st;
if (to <= 0)
throw new SocketTimeoutException();
}
- } finally {
- try {
- dc.configureBlocking(true);
- } catch (ClosedChannelException e) { }
}
}
@@ -236,10 +220,10 @@
public InetAddress getLocalAddress() {
if (isClosed())
return null;
- SocketAddress local = dc.localAddress();
+ InetSocketAddress local = dc.localAddress();
if (local == null)
local = new InetSocketAddress(0);
- InetAddress result = ((InetSocketAddress)local).getAddress();
+ InetAddress result = local.getAddress();
SecurityManager sm = System.getSecurityManager();
if (sm != null) {
try {
@@ -255,9 +239,9 @@
if (isClosed())
return -1;
try {
- SocketAddress local = dc.getLocalAddress();
+ InetSocketAddress local = dc.localAddress();
if (local != null) {
- return ((InetSocketAddress)local).getPort();
+ return local.getPort();
}
} catch (Exception x) {
}
--- a/src/java.base/share/classes/sun/nio/ch/IOUtil.java Tue Feb 27 23:11:26 2018 -0800
+++ b/src/java.base/share/classes/sun/nio/ch/IOUtil.java Wed Feb 28 09:54:38 2018 +0000
@@ -55,8 +55,7 @@
throws IOException
{
if (src instanceof DirectBuffer) {
- return writeFromNativeBuffer(fd, src, position,
- directIO, alignment, nd);
+ return writeFromNativeBuffer(fd, src, position, directIO, alignment, nd);
}
// Substitute a native buffer
@@ -77,8 +76,7 @@
// Do not update src until we see how many bytes were written
src.position(pos);
- int n = writeFromNativeBuffer(fd, bb, position,
- directIO, alignment, nd);
+ int n = writeFromNativeBuffer(fd, bb, position, directIO, alignment, nd);
if (n > 0) {
// now update src
src.position(pos + n);
@@ -232,8 +230,7 @@
if (dst.isReadOnly())
throw new IllegalArgumentException("Read-only buffer");
if (dst instanceof DirectBuffer)
- return readIntoNativeBuffer(fd, dst, position,
- directIO, alignment, nd);
+ return readIntoNativeBuffer(fd, dst, position, directIO, alignment, nd);
// Substitute a native buffer
ByteBuffer bb;
@@ -245,8 +242,7 @@
bb = Util.getTemporaryDirectBuffer(rem);
}
try {
- int n = readIntoNativeBuffer(fd, bb, position,
- directIO, alignment,nd);
+ int n = readIntoNativeBuffer(fd, bb, position, directIO, alignment,nd);
bb.flip();
if (n > 0)
dst.put(bb);
--- a/src/java.base/share/classes/sun/nio/ch/MembershipKeyImpl.java Tue Feb 27 23:11:26 2018 -0800
+++ b/src/java.base/share/classes/sun/nio/ch/MembershipKeyImpl.java Wed Feb 28 09:54:38 2018 +0000
@@ -1,5 +1,5 @@
/*
- * Copyright (c) 2008, 2009, Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2008, 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
@@ -25,10 +25,11 @@
package sun.nio.ch;
-import java.nio.channels.*;
import java.net.InetAddress;
import java.net.NetworkInterface;
import java.io.IOException;
+import java.nio.channels.MembershipKey;
+import java.nio.channels.MulticastChannel;
import java.util.HashSet;
/**
@@ -46,7 +47,7 @@
private volatile boolean invalid;
// lock used when creating or accessing blockedSet
- private Object stateLock = new Object();
+ private final Object stateLock = new Object();
// set of source addresses that are blocked
private HashSet<InetAddress> blockedSet;
--- a/src/java.base/share/classes/sun/nio/ch/Net.java Tue Feb 27 23:11:26 2018 -0800
+++ b/src/java.base/share/classes/sun/nio/ch/Net.java Wed Feb 28 09:54:38 2018 +0000
@@ -1,5 +1,5 @@
/*
- * Copyright (c) 2000, 2016, Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2000, 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
@@ -25,12 +25,30 @@
package sun.nio.ch;
-import java.io.*;
-import java.net.*;
-import java.nio.channels.*;
-import java.util.*;
+import java.io.FileDescriptor;
+import java.io.IOException;
+import java.net.Inet4Address;
+import java.net.Inet6Address;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.NetworkInterface;
+import java.net.ProtocolFamily;
+import java.net.SocketAddress;
+import java.net.SocketException;
+import java.net.SocketOption;
+import java.net.StandardProtocolFamily;
+import java.net.StandardSocketOptions;
+import java.net.UnknownHostException;
+import java.nio.channels.AlreadyBoundException;
+import java.nio.channels.ClosedChannelException;
+import java.nio.channels.NotYetBoundException;
+import java.nio.channels.NotYetConnectedException;
+import java.nio.channels.UnresolvedAddressException;
+import java.nio.channels.UnsupportedAddressTypeException;
import java.security.AccessController;
import java.security.PrivilegedAction;
+import java.util.Enumeration;
+
import sun.net.ext.ExtendedSocketOptions;
import sun.security.action.GetPropertyAction;
@@ -116,6 +134,16 @@
return isa;
}
+ static InetSocketAddress checkAddress(SocketAddress sa, ProtocolFamily family) {
+ InetSocketAddress isa = checkAddress(sa);
+ if (family == StandardProtocolFamily.INET) {
+ InetAddress addr = isa.getAddress();
+ if (!(addr instanceof Inet4Address))
+ throw new UnsupportedAddressTypeException();
+ }
+ return isa;
+ }
+
static InetSocketAddress asInetSocketAddress(SocketAddress sa) {
if (!(sa instanceof InetSocketAddress))
throw new UnsupportedAddressTypeException();
--- a/src/java.base/share/classes/sun/nio/ch/SelChImpl.java Tue Feb 27 23:11:26 2018 -0800
+++ b/src/java.base/share/classes/sun/nio/ch/SelChImpl.java Wed Feb 28 09:54:38 2018 +0000
@@ -1,5 +1,5 @@
/*
- * Copyright (c) 2000, 2012, Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2000, 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
@@ -29,7 +29,6 @@
import java.io.FileDescriptor;
import java.io.IOException;
-
/**
* An interface that allows translation (and more!).
*
@@ -50,7 +49,7 @@
* contains at least one bit that the previous value did not
* contain
*/
- public boolean translateAndUpdateReadyOps(int ops, SelectionKeyImpl sk);
+ boolean translateAndUpdateReadyOps(int ops, SelectionKeyImpl sk);
/**
* Sets the specified ops if present in interestOps. The specified
@@ -60,7 +59,7 @@
* contains at least one bit that the previous value did not
* contain
*/
- public boolean translateAndSetReadyOps(int ops, SelectionKeyImpl sk);
+ boolean translateAndSetReadyOps(int ops, SelectionKeyImpl sk);
void translateAndSetInterestOps(int ops, SelectionKeyImpl sk);
--- a/src/java.base/share/classes/sun/nio/ch/ServerSocketAdaptor.java Tue Feb 27 23:11:26 2018 -0800
+++ b/src/java.base/share/classes/sun/nio/ch/ServerSocketAdaptor.java Wed Feb 28 09:54:38 2018 +0000
@@ -34,7 +34,6 @@
import java.net.SocketException;
import java.net.SocketTimeoutException;
import java.net.StandardSocketOptions;
-import java.nio.channels.ClosedChannelException;
import java.nio.channels.IllegalBlockingModeException;
import java.nio.channels.NotYetBoundException;
import java.nio.channels.ServerSocketChannel;
@@ -51,7 +50,6 @@
class ServerSocketAdaptor // package-private
extends ServerSocket
{
-
// The channel being adapted
private final ServerSocketChannelImpl ssc;
@@ -67,13 +65,10 @@
}
// ## super will create a useless impl
- private ServerSocketAdaptor(ServerSocketChannelImpl ssc)
- throws IOException
- {
+ private ServerSocketAdaptor(ServerSocketChannelImpl ssc) throws IOException {
this.ssc = ssc;
}
-
public void bind(SocketAddress local) throws IOException {
bind(local, 50);
}
@@ -89,26 +84,31 @@
}
public InetAddress getInetAddress() {
- if (!ssc.isBound())
+ InetSocketAddress local = ssc.localAddress();
+ if (local == null) {
return null;
- return Net.getRevealedLocalAddress(ssc.localAddress()).getAddress();
-
+ } else {
+ return Net.getRevealedLocalAddress(local).getAddress();
+ }
}
public int getLocalPort() {
- if (!ssc.isBound())
+ InetSocketAddress local = ssc.localAddress();
+ if (local == null) {
return -1;
- return Net.asInetSocketAddress(ssc.localAddress()).getPort();
+ } else {
+ return local.getPort();
+ }
}
-
public Socket accept() throws IOException {
synchronized (ssc.blockingLock()) {
try {
if (!ssc.isBound())
throw new NotYetBoundException();
- if (timeout == 0) {
+ long to = this.timeout;
+ if (to == 0) {
// for compatibility reasons: accept connection if available
// when configured non-blocking
SocketChannel sc = ssc.accept();
@@ -119,28 +119,15 @@
if (!ssc.isBlocking())
throw new IllegalBlockingModeException();
- ssc.configureBlocking(false);
- try {
- SocketChannel sc;
- if ((sc = ssc.accept()) != null)
- return sc.socket();
- long to = timeout;
- for (;;) {
- if (!ssc.isOpen())
- throw new ClosedChannelException();
- long st = System.currentTimeMillis();
- int result = ssc.poll(Net.POLLIN, to);
- if (result > 0 && ((sc = ssc.accept()) != null))
- return sc.socket();
- to -= System.currentTimeMillis() - st;
- if (to <= 0)
- throw new SocketTimeoutException();
- }
- } finally {
- try {
- ssc.configureBlocking(true);
- } catch (ClosedChannelException e) { }
+ for (;;) {
+ long st = System.currentTimeMillis();
+ if (ssc.pollAccept(to))
+ return ssc.accept().socket();
+ to -= System.currentTimeMillis() - st;
+ if (to <= 0)
+ throw new SocketTimeoutException();
}
+
} catch (Exception x) {
Net.translateException(x);
assert false;
@@ -216,5 +203,4 @@
return -1; // Never happens
}
}
-
}
--- a/src/java.base/share/classes/sun/nio/ch/ServerSocketChannelImpl.java Tue Feb 27 23:11:26 2018 -0800
+++ b/src/java.base/share/classes/sun/nio/ch/ServerSocketChannelImpl.java Wed Feb 28 09:54:38 2018 +0000
@@ -35,6 +35,7 @@
import java.net.StandardProtocolFamily;
import java.net.StandardSocketOptions;
import java.nio.channels.AlreadyBoundException;
+import java.nio.channels.AsynchronousCloseException;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.NotYetBoundException;
import java.nio.channels.SelectionKey;
@@ -43,6 +44,7 @@
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;
@@ -56,7 +58,6 @@
extends ServerSocketChannel
implements SelChImpl
{
-
// Used to make native close and configure calls
private static NativeDispatcher nd;
@@ -64,10 +65,7 @@
private final FileDescriptor fd;
private final int fdVal;
- // ID of native thread currently blocked in this channel, for signalling
- private volatile long thread;
-
- // Lock held by thread currently blocked in this channel
+ // Lock held by thread currently blocked on this channel
private final ReentrantLock acceptLock = new ReentrantLock();
// Lock held by any thread that modifies the state fields declared below
@@ -77,10 +75,14 @@
// -- The following fields are protected by stateLock
// Channel state, increases monotonically
- private static final int ST_UNINITIALIZED = -1;
private static final int ST_INUSE = 0;
- private static final int ST_KILLED = 1;
- private int state = ST_UNINITIALIZED;
+ private static final int ST_CLOSING = 1;
+ private static final int ST_KILLPENDING = 2;
+ private static final int ST_KILLED = 3;
+ private int state;
+
+ // ID of native thread currently blocked in this channel, for signalling
+ private long thread;
// Binding
private InetSocketAddress localAddress; // null => unbound
@@ -98,22 +100,28 @@
super(sp);
this.fd = Net.serverSocket(true);
this.fdVal = IOUtil.fdVal(fd);
- this.state = ST_INUSE;
}
- ServerSocketChannelImpl(SelectorProvider sp,
- FileDescriptor fd,
- boolean bound)
+ ServerSocketChannelImpl(SelectorProvider sp, FileDescriptor fd, boolean bound)
throws IOException
{
super(sp);
this.fd = fd;
this.fdVal = IOUtil.fdVal(fd);
- this.state = ST_INUSE;
- if (bound)
- localAddress = Net.localAddress(fd);
+ if (bound) {
+ synchronized (stateLock) {
+ localAddress = Net.localAddress(fd);
+ }
+ }
}
+ // @throws ClosedChannelException if channel is closed
+ private void ensureOpen() throws ClosedChannelException {
+ if (!isOpen())
+ throw new ClosedChannelException();
+ }
+
+ @Override
public ServerSocket socket() {
synchronized (stateLock) {
if (socket == null)
@@ -125,11 +133,10 @@
@Override
public SocketAddress getLocalAddress() throws IOException {
synchronized (stateLock) {
- if (!isOpen())
- throw new ClosedChannelException();
- return localAddress == null ? localAddress
- : Net.getRevealedLocalAddress(
- Net.asInetSocketAddress(localAddress));
+ ensureOpen();
+ return (localAddress == null)
+ ? null
+ : Net.getRevealedLocalAddress(localAddress);
}
}
@@ -137,13 +144,11 @@
public <T> ServerSocketChannel setOption(SocketOption<T> name, T value)
throws IOException
{
- if (name == null)
- throw new NullPointerException();
+ Objects.requireNonNull(name);
if (!supportedOptions().contains(name))
throw new UnsupportedOperationException("'" + name + "' not supported");
synchronized (stateLock) {
- if (!isOpen())
- throw new ClosedChannelException();
+ ensureOpen();
if (name == StandardSocketOptions.IP_TOS) {
ProtocolFamily family = Net.isIPv6Available() ?
@@ -152,9 +157,7 @@
return this;
}
- if (name == StandardSocketOptions.SO_REUSEADDR &&
- Net.useExclusiveBind())
- {
+ if (name == StandardSocketOptions.SO_REUSEADDR && Net.useExclusiveBind()) {
// SO_REUSEADDR emulated when using exclusive bind
isReuseAddress = (Boolean)value;
} else {
@@ -170,17 +173,13 @@
public <T> T getOption(SocketOption<T> name)
throws IOException
{
- if (name == null)
- throw new NullPointerException();
+ Objects.requireNonNull(name);
if (!supportedOptions().contains(name))
throw new UnsupportedOperationException("'" + name + "' not supported");
synchronized (stateLock) {
- if (!isOpen())
- throw new ClosedChannelException();
- if (name == StandardSocketOptions.SO_REUSEADDR &&
- Net.useExclusiveBind())
- {
+ ensureOpen();
+ if (name == StandardSocketOptions.SO_REUSEADDR && Net.useExclusiveBind()) {
// SO_REUSEADDR emulated when using exclusive bind
return (T)Boolean.valueOf(isReuseAddress);
}
@@ -193,7 +192,7 @@
static final Set<SocketOption<?>> defaultOptions = defaultOptions();
private static Set<SocketOption<?>> defaultOptions() {
- HashSet<SocketOption<?>> set = new HashSet<>(2);
+ HashSet<SocketOption<?>> set = new HashSet<>();
set.add(StandardSocketOptions.SO_RCVBUF);
set.add(StandardSocketOptions.SO_REUSEADDR);
if (Net.isReusePortAvailable()) {
@@ -209,35 +208,23 @@
return DefaultOptionsHolder.defaultOptions;
}
- public boolean isBound() {
- synchronized (stateLock) {
- return localAddress != null;
- }
- }
-
- public InetSocketAddress localAddress() {
- synchronized (stateLock) {
- return localAddress;
- }
- }
-
@Override
public ServerSocketChannel bind(SocketAddress local, int backlog) throws IOException {
acceptLock.lock();
try {
- if (!isOpen())
- throw new ClosedChannelException();
- if (isBound())
- throw new AlreadyBoundException();
- InetSocketAddress isa = (local == null) ? new InetSocketAddress(0) :
- Net.checkAddress(local);
- SecurityManager sm = System.getSecurityManager();
- if (sm != null)
- sm.checkListen(isa.getPort());
- NetHooks.beforeTcpBind(fd, isa.getAddress(), isa.getPort());
- Net.bind(fd, isa.getAddress(), isa.getPort());
- Net.listen(fd, backlog < 1 ? 50 : backlog);
synchronized (stateLock) {
+ ensureOpen();
+ if (localAddress != null)
+ throw new AlreadyBoundException();
+ InetSocketAddress isa = (local == null)
+ ? new InetSocketAddress(0)
+ : Net.checkAddress(local);
+ SecurityManager sm = System.getSecurityManager();
+ if (sm != null)
+ sm.checkListen(isa.getPort());
+ NetHooks.beforeTcpBind(fd, isa.getAddress(), isa.getPort());
+ Net.bind(fd, isa.getAddress(), isa.getPort());
+ Net.listen(fd, backlog < 1 ? 50 : backlog);
localAddress = Net.localAddress(fd);
}
} finally {
@@ -246,47 +233,78 @@
return this;
}
+ /**
+ * Marks the beginning of an I/O operation that might block.
+ *
+ * @throws ClosedChannelException if the channel is closed
+ * @throws NotYetBoundException if the channel's socket has not been bound yet
+ */
+ private void begin(boolean blocking) throws ClosedChannelException {
+ if (blocking)
+ begin(); // set blocker to close channel if interrupted
+ synchronized (stateLock) {
+ ensureOpen();
+ if (localAddress == null)
+ throw new NotYetBoundException();
+ if (blocking)
+ thread = NativeThread.current();
+ }
+ }
+
+ /**
+ * Marks the end of an I/O operation that may have blocked.
+ *
+ * @throws AsynchronousCloseException if the channel was closed due to this
+ * thread being interrupted on a blocking I/O operation.
+ */
+ private void end(boolean blocking, boolean completed)
+ throws AsynchronousCloseException
+ {
+ if (blocking) {
+ synchronized (stateLock) {
+ thread = 0;
+ // notify any thread waiting in implCloseSelectableChannel
+ if (state == ST_CLOSING) {
+ stateLock.notifyAll();
+ }
+ }
+ end(completed);
+ }
+ }
+
+ @Override
public SocketChannel accept() throws IOException {
acceptLock.lock();
try {
- if (!isOpen())
- throw new ClosedChannelException();
- if (!isBound())
- throw new NotYetBoundException();
- SocketChannel sc = null;
-
int n = 0;
FileDescriptor newfd = new FileDescriptor();
InetSocketAddress[] isaa = new InetSocketAddress[1];
+ boolean blocking = isBlocking();
try {
- begin();
- if (!isOpen())
- return null;
- thread = NativeThread.current();
- for (;;) {
+ begin(blocking);
+ do {
n = accept(this.fd, newfd, isaa);
- if ((n == IOStatus.INTERRUPTED) && isOpen())
- continue;
- break;
- }
+ } while (n == IOStatus.INTERRUPTED && isOpen());
} finally {
- thread = 0;
- end(n > 0);
+ end(blocking, n > 0);
assert IOStatus.check(n);
}
if (n < 1)
return null;
+ // newly accepted socket is initially in blocking mode
IOUtil.configureBlocking(newfd, true);
+
InetSocketAddress isa = isaa[0];
- sc = new SocketChannelImpl(provider(), newfd, isa);
+ SocketChannel sc = new SocketChannelImpl(provider(), newfd, isa);
+
+ // check permitted to accept connections from the remote address
SecurityManager sm = System.getSecurityManager();
if (sm != null) {
try {
- sm.checkAccept(isa.getAddress().getHostAddress(),
- isa.getPort());
+ sm.checkAccept(isa.getAddress().getHostAddress(), isa.getPort());
} catch (SecurityException x) {
sc.close();
throw x;
@@ -299,33 +317,133 @@
}
}
+ @Override
protected void implConfigureBlocking(boolean block) throws IOException {
- IOUtil.configureBlocking(fd, block);
- }
-
- protected void implCloseSelectableChannel() throws IOException {
- synchronized (stateLock) {
- if (state != ST_KILLED)
- nd.preClose(fd);
- long th = thread;
- if (th != 0)
- NativeThread.signal(th);
- if (!isRegistered())
- kill();
+ acceptLock.lock();
+ try {
+ synchronized (stateLock) {
+ ensureOpen();
+ IOUtil.configureBlocking(fd, block);
+ }
+ } finally {
+ acceptLock.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.
+ *
+ * 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 interrupted = false;
+ boolean blocking;
+
+ // set state to ST_CLOSING
+ synchronized (stateLock) {
+ assert state < ST_CLOSING;
+ state = ST_CLOSING;
+ blocking = isBlocking();
+ }
+
+ // wait for any outstanding accept to complete
+ if (blocking) {
+ synchronized (stateLock) {
+ assert state == ST_CLOSING;
+ long th = thread;
+ if (th != 0) {
+ nd.preClose(fd);
+ NativeThread.signal(th);
+
+ // wait for accept operation to end
+ while (thread != 0) {
+ try {
+ stateLock.wait();
+ } catch (InterruptedException e) {
+ interrupted = true;
+ }
+ }
+ }
+ }
+ } else {
+ // non-blocking mode: wait for accept to complete
+ acceptLock.lock();
+ acceptLock.unlock();
+ }
+
+ // set state to ST_KILLPENDING
+ synchronized (stateLock) {
+ assert state == ST_CLOSING;
+ 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_KILLED)
- return;
- if (state == ST_UNINITIALIZED) {
+ if (state == ST_KILLPENDING) {
state = ST_KILLED;
- return;
+ nd.close(fd);
}
- assert !isOpen() && !isRegistered();
- nd.close(fd);
- state = ST_KILLED;
+ }
+ }
+
+ /**
+ * Returns true if channel's socket is bound
+ */
+ boolean isBound() {
+ synchronized (stateLock) {
+ return localAddress != null;
+ }
+ }
+
+ /**
+ * Returns the local address, or null if not bound
+ */
+ InetSocketAddress localAddress() {
+ synchronized (stateLock) {
+ return localAddress;
+ }
+ }
+
+ /**
+ * Poll this channel's socket for a new connection up to the given timeout.
+ * @return {@code true} if there is a connection to accept
+ */
+ boolean pollAccept(long timeout) throws IOException {
+ assert Thread.holdsLock(blockingLock()) && isBlocking();
+ acceptLock.lock();
+ try {
+ boolean polled = false;
+ try {
+ begin(true);
+ int n = Net.poll(fd, Net.POLLIN, timeout);
+ polled = (n > 0);
+ } finally {
+ end(true, polled);
+ }
+ return polled;
+ } finally {
+ acceptLock.unlock();
}
}
@@ -367,31 +485,6 @@
return translateReadyOps(ops, 0, sk);
}
- // package-private
- int poll(int events, long timeout) throws IOException {
- assert Thread.holdsLock(blockingLock()) && !isBlocking();
-
- acceptLock.lock();
- try {
- int n = 0;
- try {
- begin();
- synchronized (stateLock) {
- if (!isOpen())
- return 0;
- thread = NativeThread.current();
- }
- n = Net.poll(fd, events, timeout);
- } finally {
- thread = 0;
- end(n > 0);
- }
- return n;
- } finally {
- acceptLock.unlock();
- }
- }
-
/**
* Translates an interest operation set into a native poll event set
*/
@@ -421,7 +514,7 @@
sb.append("closed");
} else {
synchronized (stateLock) {
- InetSocketAddress addr = localAddress();
+ InetSocketAddress addr = localAddress;
if (addr == null) {
sb.append("unbound");
} else {
@@ -438,7 +531,8 @@
*
* @implNote Wrap native call to allow instrumentation.
*/
- private int accept(FileDescriptor ssfd, FileDescriptor newfd,
+ private int accept(FileDescriptor ssfd,
+ FileDescriptor newfd,
InetSocketAddress[] isaa)
throws IOException
{
@@ -452,7 +546,8 @@
// Returns 1 on success, or IOStatus.UNAVAILABLE (if non-blocking and no
// connections are pending) or IOStatus.INTERRUPTED.
//
- private native int accept0(FileDescriptor ssfd, FileDescriptor newfd,
+ private native int accept0(FileDescriptor ssfd,
+ FileDescriptor newfd,
InetSocketAddress[] isaa)
throws IOException;
--- a/src/java.base/share/classes/sun/nio/ch/SocketAdaptor.java Tue Feb 27 23:11:26 2018 -0800
+++ b/src/java.base/share/classes/sun/nio/ch/SocketAdaptor.java Wed Feb 28 09:54:38 2018 +0000
@@ -44,16 +44,10 @@
import java.nio.channels.SocketChannel;
import java.security.AccessController;
import java.security.PrivilegedExceptionAction;
-import java.util.concurrent.TimeUnit;
+import static java.util.concurrent.TimeUnit.*;
// Make a socket channel look like a socket.
//
-// The only aspects of java.net.Socket-hood that we don't attempt to emulate
-// here are the interrupted-I/O exceptions (which our Solaris implementations
-// attempt to support) and the sending of urgent data. Otherwise an adapted
-// socket should look enough like a real java.net.Socket to fool most of the
-// developers most of the time, right down to the exception message strings.
-//
// The methods in this class are defined in exactly the same order as in
// java.net.Socket so as to simplify tracking future changes to that class.
//
@@ -61,7 +55,6 @@
class SocketAdaptor
extends Socket
{
-
// The channel being adapted
private final SocketChannelImpl sc;
@@ -102,40 +95,42 @@
throw new IllegalBlockingModeException();
try {
+ // no timeout
if (timeout == 0) {
sc.connect(remote);
return;
}
+ // timed connect
sc.configureBlocking(false);
try {
if (sc.connect(remote))
return;
- long timeoutNanos =
- TimeUnit.NANOSECONDS.convert(timeout,
- TimeUnit.MILLISECONDS);
- for (;;) {
- if (!sc.isOpen())
- throw new ClosedChannelException();
- long startTime = System.nanoTime();
-
- int result = sc.poll(Net.POLLCONN, timeout);
- if (result > 0 && sc.finishConnect())
- break;
- timeoutNanos -= System.nanoTime() - startTime;
- if (timeoutNanos <= 0) {
- try {
- sc.close();
- } catch (IOException x) { }
- throw new SocketTimeoutException();
- }
- }
} finally {
try {
sc.configureBlocking(true);
} catch (ClosedChannelException e) { }
}
+ long timeoutNanos = NANOSECONDS.convert(timeout, MILLISECONDS);
+ long to = timeout;
+ for (;;) {
+ long startTime = System.nanoTime();
+ if (sc.pollConnected(to)) {
+ boolean connected = sc.finishConnect();
+ assert connected;
+ break;
+ }
+ timeoutNanos -= System.nanoTime() - startTime;
+ if (timeoutNanos <= 0) {
+ try {
+ sc.close();
+ } catch (IOException x) { }
+ throw new SocketTimeoutException();
+ }
+ to = MILLISECONDS.convert(timeoutNanos, NANOSECONDS);
+ }
+
} catch (Exception x) {
Net.translateException(x, true);
}
@@ -152,11 +147,11 @@
}
public InetAddress getInetAddress() {
- SocketAddress remote = sc.remoteAddress();
+ InetSocketAddress remote = sc.remoteAddress();
if (remote == null) {
return null;
} else {
- return ((InetSocketAddress)remote).getAddress();
+ return remote.getAddress();
}
}
@@ -171,20 +166,20 @@
}
public int getPort() {
- SocketAddress remote = sc.remoteAddress();
+ InetSocketAddress remote = sc.remoteAddress();
if (remote == null) {
return 0;
} else {
- return ((InetSocketAddress)remote).getPort();
+ return remote.getPort();
}
}
public int getLocalPort() {
- SocketAddress local = sc.localAddress();
+ InetSocketAddress local = sc.localAddress();
if (local == null) {
return -1;
} else {
- return ((InetSocketAddress)local).getPort();
+ return local.getPort();
}
}
@@ -202,34 +197,22 @@
if (!sc.isBlocking())
throw new IllegalBlockingModeException();
- if (timeout == 0)
+ // no timeout
+ long to = SocketAdaptor.this.timeout;
+ if (to == 0)
return sc.read(bb);
- sc.configureBlocking(false);
- try {
- int n;
- if ((n = sc.read(bb)) != 0)
- return n;
- long timeoutNanos =
- TimeUnit.NANOSECONDS.convert(timeout,
- TimeUnit.MILLISECONDS);
- for (;;) {
- if (!sc.isOpen())
- throw new ClosedChannelException();
- long startTime = System.nanoTime();
- int result = sc.poll(Net.POLLIN, timeout);
- if (result > 0) {
- if ((n = sc.read(bb)) != 0)
- return n;
- }
- timeoutNanos -= System.nanoTime() - startTime;
- if (timeoutNanos <= 0)
- throw new SocketTimeoutException();
+ // timed read
+ long timeoutNanos = NANOSECONDS.convert(to, MILLISECONDS);
+ for (;;) {
+ long startTime = System.nanoTime();
+ if (sc.pollRead(to)) {
+ return sc.read(bb);
}
- } finally {
- try {
- sc.configureBlocking(true);
- } catch (ClosedChannelException e) { }
+ timeoutNanos -= System.nanoTime() - startTime;
+ if (timeoutNanos <= 0)
+ throw new SocketTimeoutException();
+ to = MILLISECONDS.convert(timeoutNanos, NANOSECONDS);
}
}
}
@@ -453,5 +436,4 @@
public boolean isOutputShutdown() {
return !sc.isOutputOpen();
}
-
}
--- a/src/java.base/share/classes/sun/nio/ch/SocketChannelImpl.java Tue Feb 27 23:11:26 2018 -0800
+++ b/src/java.base/share/classes/sun/nio/ch/SocketChannelImpl.java Wed Feb 28 09:54:38 2018 +0000
@@ -48,6 +48,7 @@
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;
@@ -62,7 +63,6 @@
extends SocketChannel
implements SelChImpl
{
-
// Used to make native read and write calls
private static NativeDispatcher nd;
@@ -70,10 +70,6 @@
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();
@@ -84,28 +80,32 @@
// 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_KILLED = 4;
- private int state = ST_UNINITIALIZED;
+ private static final int ST_CLOSING = 3;
+ private static final int ST_KILLPENDING = 4;
+ 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;
@@ -118,36 +118,43 @@
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)
- this.localAddress = Net.localAddress(fd);
+ if (bound) {
+ synchronized (stateLock) {
+ this.localAddress = Net.localAddress(fd);
+ }
+ }
}
// Constructor for sockets obtained from server sockets
//
- SocketChannelImpl(SelectorProvider sp,
- FileDescriptor fd, InetSocketAddress remote)
+ SocketChannelImpl(SelectorProvider sp, FileDescriptor fd, InetSocketAddress isa)
throws IOException
{
super(sp);
this.fd = fd;
this.fdVal = IOUtil.fdVal(fd);
- this.state = ST_CONNECTED;
- this.localAddress = Net.localAddress(fd);
- this.remoteAddress = remote;
+ synchronized (stateLock) {
+ this.localAddress = Net.localAddress(fd);
+ 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)
@@ -159,17 +166,15 @@
@Override
public SocketAddress getLocalAddress() throws IOException {
synchronized (stateLock) {
- if (!isOpen())
- throw new ClosedChannelException();
- return Net.getRevealedLocalAddress(localAddress);
+ ensureOpen();
+ return Net.getRevealedLocalAddress(localAddress);
}
}
@Override
public SocketAddress getRemoteAddress() throws IOException {
synchronized (stateLock) {
- if (!isOpen())
- throw new ClosedChannelException();
+ ensureOpen();
return remoteAddress;
}
}
@@ -178,14 +183,12 @@
public <T> SocketChannel setOption(SocketOption<T> name, T value)
throws IOException
{
- if (name == null)
- throw new NullPointerException();
+ Objects.requireNonNull(name);
if (!supportedOptions().contains(name))
throw new UnsupportedOperationException("'" + name + "' not supported");
synchronized (stateLock) {
- if (!isOpen())
- throw new ClosedChannelException();
+ ensureOpen();
if (name == StandardSocketOptions.IP_TOS) {
ProtocolFamily family = Net.isIPv6Available() ?
@@ -211,18 +214,14 @@
public <T> T getOption(SocketOption<T> name)
throws IOException
{
- if (name == null)
- throw new NullPointerException();
+ Objects.requireNonNull(name);
if (!supportedOptions().contains(name))
throw new UnsupportedOperationException("'" + name + "' not supported");
synchronized (stateLock) {
- if (!isOpen())
- throw new ClosedChannelException();
+ ensureOpen();
- 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);
}
@@ -243,7 +242,7 @@
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);
@@ -256,9 +255,7 @@
// additional options required by socket adaptor
set.add(StandardSocketOptions.IP_TOS);
set.add(ExtendedSocketOption.SO_OOBINLINE);
- ExtendedSocketOptions extendedOptions =
- ExtendedSocketOptions.getInstance();
- set.addAll(extendedOptions.options());
+ set.addAll(ExtendedSocketOptions.getInstance().options());
return Collections.unmodifiableSet(set);
}
}
@@ -268,329 +265,277 @@
return DefaultOptionsHolder.defaultOptions;
}
- private boolean ensureReadOpen() throws ClosedChannelException {
+ /**
+ * Marks the beginning of a read operation that might block.
+ *
+ * @throws ClosedChannelException if the channel is closed
+ * @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) {
- if (!isOpen())
- throw new ClosedChannelException();
- if (!isConnected())
+ ensureOpen();
+ if (state != ST_CONNECTED)
throw new NotYetConnectedException();
- if (!isInputOpen)
- return false;
- else
- return true;
+ if (blocking)
+ readerThread = NativeThread.current();
}
}
- private void ensureWriteOpen() throws ClosedChannelException {
- synchronized (stateLock) {
- if (!isOpen())
- throw new ClosedChannelException();
- if (!isOutputOpen)
- throw new ClosedChannelException();
- if (!isConnected())
- throw new NotYetConnectedException();
+ /**
+ * Marks the end of a read operation that may have blocked.
+ *
+ * @throws AsynchronousCloseException if the channel was closed due to this
+ * thread being interrupted on a blocking read operation.
+ */
+ private void endRead(boolean blocking, boolean completed)
+ throws AsynchronousCloseException
+ {
+ if (blocking) {
+ synchronized (stateLock) {
+ readerThread = 0;
+ // notify any thread waiting in implCloseSelectableChannel
+ if (state == ST_CLOSING) {
+ stateLock.notifyAll();
+ }
+ }
+ // remove hook for Thread.interrupt
+ end(completed);
}
}
- private void readerCleanup() throws IOException {
- synchronized (stateLock) {
- readerThread = 0;
- if (state == ST_KILLPENDING)
- kill();
+ @Override
+ public int read(ByteBuffer buf) throws IOException {
+ Objects.requireNonNull(buf);
+
+ readLock.lock();
+ try {
+ boolean blocking = isBlocking();
+ int n = 0;
+ try {
+ beginRead(blocking);
+
+ // check if input is shutdown
+ if (isInputClosed)
+ return IOStatus.EOF;
+
+ if (blocking) {
+ do {
+ n = IOUtil.read(fd, buf, -1, nd);
+ } while (n == IOStatus.INTERRUPTED && isOpen());
+ } else {
+ n = IOUtil.read(fd, buf, -1, nd);
+ }
+ } finally {
+ endRead(blocking, n > 0);
+ if (n <= 0 && isInputClosed)
+ return IOStatus.EOF;
+ }
+ return IOStatus.normalize(n);
+ } finally {
+ readLock.unlock();
}
}
- private void writerCleanup() throws IOException {
- synchronized (stateLock) {
- writerThread = 0;
- if (state == ST_KILLPENDING)
- kill();
- }
- }
-
- public int read(ByteBuffer buf) throws IOException {
-
- if (buf == null)
- throw new NullPointerException();
+ @Override
+ public long read(ByteBuffer[] dsts, int offset, int length)
+ throws IOException
+ {
+ Objects.checkFromIndexSize(offset, length, dsts.length);
readLock.lock();
try {
- if (!ensureReadOpen())
- return -1;
+ boolean blocking = isBlocking();
+ long n = 0;
+ try {
+ beginRead(blocking);
+
+ // check if input is shutdown
+ if (isInputClosed)
+ return IOStatus.EOF;
+
+ if (blocking) {
+ do {
+ n = IOUtil.read(fd, dsts, offset, length, nd);
+ } while (n == IOStatus.INTERRUPTED && isOpen());
+ } else {
+ n = IOUtil.read(fd, dsts, offset, length, nd);
+ }
+ } finally {
+ endRead(blocking, n > 0);
+ if (n <= 0 && isInputClosed)
+ return IOStatus.EOF;
+ }
+ return IOStatus.normalize(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 {
+ Objects.requireNonNull(buf);
+
+ writeLock.lock();
+ try {
+ boolean blocking = isBlocking();
int n = 0;
try {
-
- // Set up the interruption machinery; see
- // AbstractInterruptibleChannel for details
- //
- begin();
+ beginWrite(blocking);
+ if (blocking) {
+ do {
+ n = IOUtil.write(fd, buf, -1, nd);
+ } while (n == IOStatus.INTERRUPTED && isOpen());
+ } else {
+ n = IOUtil.write(fd, buf, -1, nd);
+ }
+ } finally {
+ endWrite(blocking, n > 0);
+ if (n <= 0 && isOutputClosed)
+ throw new AsynchronousCloseException();
+ }
+ return IOStatus.normalize(n);
+ } finally {
+ writeLock.unlock();
+ }
+ }
- synchronized (stateLock) {
- if (!isOpen()) {
- // Either the current thread is already interrupted, so
- // begin() closed the channel, or another thread closed the
- // channel since we checked it a few bytecodes ago. In
- // 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();
- }
+ @Override
+ public long write(ByteBuffer[] srcs, int offset, int length)
+ throws IOException
+ {
+ Objects.checkFromIndexSize(offset, length, srcs.length);
- // 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);
+ writeLock.lock();
+ try {
+ boolean blocking = isBlocking();
+ long n = 0;
+ try {
+ beginWrite(blocking);
+ if (blocking) {
+ do {
+ n = IOUtil.write(fd, srcs, offset, length, nd);
+ } while (n == IOStatus.INTERRUPTED && isOpen());
+ } else {
+ n = IOUtil.write(fd, srcs, offset, length, nd);
}
-
} finally {
- readerCleanup(); // Clear reader thread
- // The end method, which is defined in our superclass
- // AbstractInterruptibleChannel, resets the interruption
- // machinery. If its argument is true then it returns
- // normally; otherwise it checks the interrupt and open state
- // 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));
+ endWrite(blocking, n > 0);
+ if (n <= 0 && isOutputClosed)
+ throw new AsynchronousCloseException();
+ }
+ return IOStatus.normalize(n);
+ } finally {
+ writeLock.unlock();
+ }
+ }
- // Extra case for socket channels: Asynchronous shutdown
- //
+ /**
+ * Writes a byte of out of band data.
+ */
+ int sendOutOfBandData(byte b) throws IOException {
+ writeLock.lock();
+ try {
+ boolean blocking = isBlocking();
+ int n = 0;
+ try {
+ beginWrite(blocking);
+ if (blocking) {
+ do {
+ n = sendOutOfBandData(fd, b);
+ } while (n == IOStatus.INTERRUPTED && isOpen());
+ } else {
+ n = sendOutOfBandData(fd, b);
+ }
+ } finally {
+ endWrite(blocking, n > 0);
+ if (n <= 0 && isOutputClosed)
+ throw new AsynchronousCloseException();
+ }
+ return IOStatus.normalize(n);
+ } finally {
+ writeLock.unlock();
+ }
+ }
+
+ @Override
+ protected void implConfigureBlocking(boolean block) throws IOException {
+ readLock.lock();
+ try {
+ writeLock.lock();
+ try {
synchronized (stateLock) {
- if ((n <= 0) && (!isInputOpen))
- return IOStatus.EOF;
+ ensureOpen();
+ IOUtil.configureBlocking(fd, block);
}
-
- assert IOStatus.check(n);
-
+ } finally {
+ writeLock.unlock();
}
} finally {
readLock.unlock();
}
}
- public long read(ByteBuffer[] dsts, int offset, int length)
- throws IOException
- {
- if ((offset < 0) || (length < 0) || (offset > dsts.length - length))
- throw new IndexOutOfBoundsException();
- readLock.lock();
- try {
- if (!ensureReadOpen())
- return -1;
- long n = 0;
- try {
- begin();
- synchronized (stateLock) {
- if (!isOpen())
- return 0;
- readerThread = NativeThread.current();
- }
-
- for (;;) {
- n = IOUtil.read(fd, dsts, offset, length, nd);
- if ((n == IOStatus.INTERRUPTED) && isOpen())
- continue;
- return IOStatus.normalize(n);
- }
- } finally {
- readerCleanup();
- end(n > 0 || (n == IOStatus.UNAVAILABLE));
- synchronized (stateLock) {
- if ((n <= 0) && (!isInputOpen))
- return IOStatus.EOF;
- }
- assert IOStatus.check(n);
- }
- } finally {
- readLock.unlock();
- }
- }
-
- public int write(ByteBuffer buf) throws IOException {
- if (buf == null)
- throw new NullPointerException();
- writeLock.lock();
- try {
- ensureWriteOpen();
- int n = 0;
- try {
- begin();
- synchronized (stateLock) {
- if (!isOpen())
- return 0;
- writerThread = NativeThread.current();
- }
- for (;;) {
- n = IOUtil.write(fd, buf, -1, nd);
- if ((n == IOStatus.INTERRUPTED) && isOpen())
- continue;
- return IOStatus.normalize(n);
- }
- } finally {
- writerCleanup();
- end(n > 0 || (n == IOStatus.UNAVAILABLE));
- synchronized (stateLock) {
- if ((n <= 0) && (!isOutputOpen))
- throw new AsynchronousCloseException();
- }
- assert IOStatus.check(n);
- }
- } finally {
- writeLock.unlock();
- }
- }
-
- public long write(ByteBuffer[] srcs, int offset, int length)
- throws IOException
- {
- if ((offset < 0) || (length < 0) || (offset > srcs.length - length))
- throw new IndexOutOfBoundsException();
- writeLock.lock();
- try {
- ensureWriteOpen();
- long n = 0;
- try {
- begin();
- synchronized (stateLock) {
- if (!isOpen())
- return 0;
- writerThread = NativeThread.current();
- }
- for (;;) {
- n = IOUtil.write(fd, srcs, offset, length, nd);
- if ((n == IOStatus.INTERRUPTED) && isOpen())
- continue;
- return IOStatus.normalize(n);
- }
- } finally {
- writerCleanup();
- end((n > 0) || (n == IOStatus.UNAVAILABLE));
- synchronized (stateLock) {
- if ((n <= 0) && (!isOutputOpen))
- throw new AsynchronousCloseException();
- }
- assert IOStatus.check(n);
- }
- } finally {
- writeLock.unlock();
- }
- }
-
- // package-private
- int sendOutOfBandData(byte b) throws IOException {
- writeLock.lock();
- try {
- ensureWriteOpen();
- int n = 0;
- try {
- begin();
- synchronized (stateLock) {
- if (!isOpen())
- return 0;
- writerThread = NativeThread.current();
- }
- for (;;) {
- n = sendOutOfBandData(fd, b);
- if ((n == IOStatus.INTERRUPTED) && isOpen())
- continue;
- return IOStatus.normalize(n);
- }
- } finally {
- writerCleanup();
- end((n > 0) || (n == IOStatus.UNAVAILABLE));
- synchronized (stateLock) {
- if ((n <= 0) && (!isOutputOpen))
- throw new AsynchronousCloseException();
- }
- assert IOStatus.check(n);
- }
- } finally {
- writeLock.unlock();
- }
- }
-
- protected void implConfigureBlocking(boolean block) throws IOException {
- IOUtil.configureBlocking(fd, block);
- }
-
- public InetSocketAddress localAddress() {
+ /**
+ * 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;
}
@@ -603,9 +548,8 @@
writeLock.lock();
try {
synchronized (stateLock) {
- if (!isOpen())
- throw new ClosedChannelException();
- if (state == ST_PENDING)
+ ensureOpen();
+ if (state == ST_CONNECTIONPENDING)
throw new ConnectionPendingException();
if (localAddress != null)
throw new AlreadyBoundException();
@@ -628,101 +572,115 @@
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
+ /**
+ * Marks the beginning of a connect operation that might block.
+ *
+ * @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) {
- if (!isOpen())
- throw new ClosedChannelException();
+ 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();
- InetSocketAddress isa = Net.checkAddress(sa);
- SecurityManager sm = System.getSecurityManager();
- if (sm != null)
- sm.checkConnect(isa.getAddress().getHostAddress(),
- isa.getPort());
- synchronized (blockingLock()) {
- int n = 0;
- try {
- try {
- begin();
- synchronized (stateLock) {
- if (!isOpen()) {
- return false;
- }
- // notify hook only if unbound
- 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) {
- // If an exception was thrown, close the channel after
- // invoking end() so as to avoid bogus
- // AsynchronousCloseExceptions
- close();
- throw x;
- }
- synchronized (stateLock) {
- remoteAddress = isa;
- if (n > 0) {
-
- // Connection succeeded; disallow further
- // invocation
- state = ST_CONNECTED;
- if (isOpen())
- localAddress = Net.localAddress(fd);
- return true;
- }
- // If nonblocking and no exception then connection
- // pending; disallow another invocation
- if (!isBlocking())
- state = ST_PENDING;
- else
- assert false;
+ // notify before-connect hook
+ synchronized (stateLock) {
+ if (state == ST_UNCONNECTED && localAddress == null) {
+ NetHooks.beforeTcpConnect(fd, isa.getAddress(), isa.getPort());
}
}
- return false;
+
+ InetAddress ia = isa.getAddress();
+ if (ia.isAnyLocalAddress())
+ ia = InetAddress.getLocalHost();
+
+ int n = 0;
+ boolean blocking = isBlocking();
+ try {
+ try {
+ beginConnect(blocking);
+ if (blocking) {
+ do {
+ n = Net.connect(fd, ia, isa.getPort());
+ } while (n == IOStatus.INTERRUPTED && isOpen());
+ } else {
+ n = Net.connect(fd, ia, isa.getPort());
+ }
+ } finally {
+ endConnect(blocking, n > 0);
+ }
+ } catch (IOException x) {
+ // connect failed, close socket
+ close();
+ throw x;
+ }
+
+ // connection may be established
+ synchronized (stateLock) {
+ if (!isOpen())
+ throw new AsynchronousCloseException();
+ remoteAddress = isa;
+ if (n > 0) {
+ // connected established
+ localAddress = Net.localAddress(fd);
+ state = ST_CONNECTED;
+ return true;
+ } else {
+ // connection pending
+ assert !blocking;
+ state = ST_CONNECTIONPENDING;
+ return false;
+ }
+ }
} finally {
writeLock.unlock();
}
@@ -731,83 +689,85 @@
}
}
+ /**
+ * 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 (!isOpen())
- throw new ClosedChannelException();
if (state == ST_CONNECTED)
return true;
- if (state != ST_PENDING)
- throw new NoConnectionPendingException();
}
+
int n = 0;
+ boolean blocking = isBlocking();
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;
- }
- }
+ beginFinishConnect(blocking);
+ if (blocking) {
+ do {
+ n = checkConnect(fd, true);
+ } while (n == 0 || (n == IOStatus.INTERRUPTED) && isOpen());
+ } else {
+ n = checkConnect(fd, false);
}
} 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);
+ endFinishConnect(blocking, n > 0);
}
} 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) {
+
+ // post finishConnect, connection may be established
+ synchronized (stateLock) {
+ if (!isOpen())
+ throw new AsynchronousCloseException();
+ if (n > 0) {
+ // connection established
+ localAddress = Net.localAddress(fd);
state = ST_CONNECTED;
- if (isOpen())
- localAddress = Net.localAddress(fd);
+ return true;
+ } else {
+ // connection still pending
+ assert !blocking;
+ return false;
}
- return true;
}
- return false;
} finally {
writeLock.unlock();
}
@@ -816,18 +776,119 @@
}
}
+ /**
+ * 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())
- throw new ClosedChannelException();
+ ensureOpen();
if (!isConnected())
throw new NotYetConnectedException();
- if (isInputOpen) {
+ if (!isInputClosed) {
Net.shutdown(fd, Net.SHUT_RD);
- if (readerThread != 0)
- NativeThread.signal(readerThread);
- isInputOpen = false;
+ long thread = readerThread;
+ if (thread != 0)
+ NativeThread.signal(thread);
+ isInputClosed = true;
}
return this;
}
@@ -836,94 +897,78 @@
@Override
public SocketChannel shutdownOutput() throws IOException {
synchronized (stateLock) {
- if (!isOpen())
- throw new ClosedChannelException();
+ ensureOpen();
if (!isConnected())
throw new NotYetConnectedException();
- if (isOutputOpen) {
+ if (!isOutputClosed) {
Net.shutdown(fd, Net.SHUT_WR);
- if (writerThread != 0)
- NativeThread.signal(writerThread);
- isOutputOpen = false;
+ long thread = writerThread;
+ if (thread != 0)
+ NativeThread.signal(thread);
+ isOutputClosed = true;
}
return this;
}
}
- public boolean isInputOpen() {
- synchronized (stateLock) {
- return isInputOpen;
- }
+ boolean isInputOpen() {
+ return !isInputClosed;
+ }
+
+ 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
+ */
+ boolean pollRead(long timeout) throws IOException {
+ boolean blocking = isBlocking();
+ assert Thread.holdsLock(blockingLock()) && blocking;
+
+ readLock.lock();
+ try {
+ boolean polled = false;
+ try {
+ beginRead(blocking);
+ int n = Net.poll(fd, Net.POLLIN, timeout);
+ polled = (n > 0);
+ } finally {
+ endRead(blocking, polled);
+ }
+ return polled;
+ } finally {
+ readLock.unlock();
}
}
- // AbstractInterruptibleChannel synchronizes invocations of this method
- // using AbstractInterruptibleChannel.closeLock, and also ensures that this
- // method is only ever invoked once. Before we get to this method, isOpen
- // (which is volatile) will have been set to false.
- //
- protected void implCloseSelectableChannel() throws IOException {
- synchronized (stateLock) {
- isInputOpen = false;
- isOutputOpen = false;
-
- // Close the underlying file descriptor and dup it to a known fd
- // that's already closed. This prevents other operations on this
- // channel from using the old fd, which might be recycled in the
- // meantime and allocated to an entirely different channel.
- //
- if (state != ST_KILLED)
- nd.preClose(fd);
-
- // 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.
- //
- if (readerThread != 0)
- NativeThread.signal(readerThread);
-
- if (writerThread != 0)
- NativeThread.signal(writerThread);
+ /**
+ * Poll this channel's socket for a connection, up to the given timeout.
+ * @return {@code true} if the socket is polled
+ */
+ boolean pollConnected(long timeout) throws IOException {
+ boolean blocking = isBlocking();
+ assert Thread.holdsLock(blockingLock()) && blocking;
- // If this channel is not registered then it's safe to close the fd
- // immediately since we know at this point that no thread is
- // blocked in an I/O operation upon the channel and, since the
- // channel is marked closed, no thread will start another such
- // operation. If this channel is registered then we don't close
- // the fd since it might be in use by a selector. In that case
- // closing this channel caused its keys to be cancelled, so the
- // last selector to deregister a key for this channel will invoke
- // kill() to close the fd.
- //
- if (!isRegistered())
- kill();
- }
- }
-
- public void kill() throws IOException {
- synchronized (stateLock) {
- if (state == ST_KILLED)
- return;
- if (state == ST_UNINITIALIZED) {
- state = ST_KILLED;
- return;
+ readLock.lock();
+ try {
+ writeLock.lock();
+ try {
+ boolean polled = false;
+ try {
+ beginFinishConnect(blocking);
+ int n = Net.poll(fd, Net.POLLCONN, timeout);
+ polled = (n > 0);
+ } finally {
+ endFinishConnect(blocking, polled);
+ }
+ return polled;
+ } finally {
+ writeLock.unlock();
}
- 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;
- }
+ } finally {
+ readLock.unlock();
}
}
@@ -956,7 +1001,7 @@
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;
}
@@ -977,31 +1022,6 @@
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
*/
@@ -1037,14 +1057,14 @@
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;
}
--- a/src/java.base/unix/classes/sun/nio/ch/SinkChannelImpl.java Tue Feb 27 23:11:26 2018 -0800
+++ b/src/java.base/unix/classes/sun/nio/ch/SinkChannelImpl.java Wed Feb 28 09:54:38 2018 +0000
@@ -28,31 +28,26 @@
import java.io.FileDescriptor;
import java.io.IOException;
import java.nio.ByteBuffer;
+import java.nio.channels.AsynchronousCloseException;
import java.nio.channels.ClosedChannelException;
+import java.nio.channels.NotYetConnectedException;
import java.nio.channels.Pipe;
import java.nio.channels.SelectionKey;
import java.nio.channels.spi.SelectorProvider;
+import java.util.Objects;
import java.util.concurrent.locks.ReentrantLock;
-
class SinkChannelImpl
extends Pipe.SinkChannel
implements SelChImpl
{
-
// Used to make native read and write calls
private static final NativeDispatcher nd = new FileDispatcherImpl();
// The file descriptor associated with this channel
private final FileDescriptor fd;
-
- // fd value needed for dev/poll. This value will remain valid
- // even after the value in the file descriptor object has been set to -1
private final int fdVal;
- // ID of native thread doing write, for signalling
- private volatile long thread;
-
// Lock held by current writing thread
private final ReentrantLock writeLock = new ReentrantLock();
@@ -63,10 +58,14 @@
// -- The following fields are protected by stateLock
// Channel state
- private static final int ST_UNINITIALIZED = -1;
private static final int ST_INUSE = 0;
- private static final int ST_KILLED = 1;
- private volatile int state = ST_UNINITIALIZED;
+ private static final int ST_CLOSING = 1;
+ private static final int ST_KILLPENDING = 2;
+ private static final int ST_KILLED = 3;
+ private int state;
+
+ // ID of native thread doing write, for signalling
+ private long thread;
// -- End of fields protected by stateLock
@@ -83,39 +82,88 @@
super(sp);
this.fd = fd;
this.fdVal = IOUtil.fdVal(fd);
- this.state = ST_INUSE;
}
+ /**
+ * Invoked by implCloseChannel to close the channel.
+ */
+ @Override
protected void implCloseSelectableChannel() throws IOException {
+ assert !isOpen();
+
+ boolean interrupted = false;
+ boolean blocking;
+
+ // set state to ST_CLOSING
synchronized (stateLock) {
- if (state != ST_KILLED)
- nd.preClose(fd);
- long th = thread;
- if (th != 0)
- NativeThread.signal(th);
- if (!isRegistered())
- kill();
+ assert state < ST_CLOSING;
+ state = ST_CLOSING;
+ blocking = isBlocking();
+ }
+
+ // wait for any outstanding write to complete
+ if (blocking) {
+ synchronized (stateLock) {
+ assert state == ST_CLOSING;
+ long th = thread;
+ if (th != 0) {
+ nd.preClose(fd);
+ NativeThread.signal(th);
+
+ // wait for write operation to end
+ while (thread != 0) {
+ try {
+ stateLock.wait();
+ } catch (InterruptedException e) {
+ interrupted = true;
+ }
+ }
+ }
+ }
+ } else {
+ // non-blocking mode: wait for write to complete
+ writeLock.lock();
+ writeLock.unlock();
+ }
+
+ // set state to ST_KILLPENDING
+ synchronized (stateLock) {
+ assert state == ST_CLOSING;
+ 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) {
+ assert thread == 0;
+ if (state == ST_KILLPENDING) {
+ state = ST_KILLED;
+ nd.close(fd);
+ }
}
}
- public void kill() throws IOException {
- synchronized (stateLock) {
- if (state == ST_KILLED)
- return;
- if (state == ST_UNINITIALIZED) {
- state = ST_KILLED;
- return;
+ @Override
+ protected void implConfigureBlocking(boolean block) throws IOException {
+ writeLock.lock();
+ try {
+ synchronized (stateLock) {
+ IOUtil.configureBlocking(fd, block);
}
- assert !isOpen() && !isRegistered();
- nd.close(fd);
- state = ST_KILLED;
+ } finally {
+ writeLock.unlock();
}
}
- protected void implConfigureBlocking(boolean block) throws IOException {
- IOUtil.configureBlocking(fd, block);
- }
-
public boolean translateReadyOps(int ops, int initialOps,
SelectionKeyImpl sk) {
int intOps = sk.nioInterestOps();// Do this just once, it synchronizes
@@ -153,67 +201,95 @@
sk.selector.putEventOps(sk, ops);
}
- private void ensureOpen() throws IOException {
- if (!isOpen())
- throw new ClosedChannelException();
+ /**
+ * Marks the beginning of a write operation that might block.
+ *
+ * @throws ClosedChannelException if the channel is closed
+ * @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) {
+ if (!isOpen())
+ throw new ClosedChannelException();
+ if (blocking)
+ thread = 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) {
+ thread = 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 src) throws IOException {
+ Objects.requireNonNull(src);
+
writeLock.lock();
try {
- ensureOpen();
+ boolean blocking = isBlocking();
int n = 0;
try {
- begin();
- if (!isOpen())
- return 0;
- thread = NativeThread.current();
+ beginWrite(blocking);
do {
n = IOUtil.write(fd, src, -1, nd);
} while ((n == IOStatus.INTERRUPTED) && isOpen());
- return IOStatus.normalize(n);
} finally {
- thread = 0;
- end((n > 0) || (n == IOStatus.UNAVAILABLE));
+ endWrite(blocking, n > 0);
assert IOStatus.check(n);
}
+ return IOStatus.normalize(n);
} finally {
writeLock.unlock();
}
}
- public long write(ByteBuffer[] srcs) throws IOException {
- if (srcs == null)
- throw new NullPointerException();
+ @Override
+ public long write(ByteBuffer[] srcs, int offset, int length) throws IOException {
+ Objects.checkFromIndexSize(offset, length, srcs.length);
writeLock.lock();
try {
- ensureOpen();
+ boolean blocking = isBlocking();
long n = 0;
try {
- begin();
- if (!isOpen())
- return 0;
- thread = NativeThread.current();
+ beginWrite(blocking);
do {
- n = IOUtil.write(fd, srcs, nd);
+ n = IOUtil.write(fd, srcs, offset, length, nd);
} while ((n == IOStatus.INTERRUPTED) && isOpen());
- return IOStatus.normalize(n);
} finally {
- thread = 0;
- end((n > 0) || (n == IOStatus.UNAVAILABLE));
+ endWrite(blocking, n > 0);
assert IOStatus.check(n);
}
+ return IOStatus.normalize(n);
} finally {
writeLock.unlock();
}
}
- public long write(ByteBuffer[] srcs, int offset, int length)
- throws IOException
- {
- if ((offset < 0) || (length < 0) || (offset > srcs.length - length))
- throw new IndexOutOfBoundsException();
- return write(Util.subsequence(srcs, offset, length));
+ @Override
+ public long write(ByteBuffer[] srcs) throws IOException {
+ return write(srcs, 0, srcs.length);
}
}
--- a/src/java.base/unix/classes/sun/nio/ch/SourceChannelImpl.java Tue Feb 27 23:11:26 2018 -0800
+++ b/src/java.base/unix/classes/sun/nio/ch/SourceChannelImpl.java Wed Feb 28 09:54:38 2018 +0000
@@ -28,31 +28,26 @@
import java.io.FileDescriptor;
import java.io.IOException;
import java.nio.ByteBuffer;
+import java.nio.channels.AsynchronousCloseException;
import java.nio.channels.ClosedChannelException;
+import java.nio.channels.NotYetConnectedException;
import java.nio.channels.Pipe;
import java.nio.channels.SelectionKey;
import java.nio.channels.spi.SelectorProvider;
+import java.util.Objects;
import java.util.concurrent.locks.ReentrantLock;
-
class SourceChannelImpl
extends Pipe.SourceChannel
implements SelChImpl
{
-
// Used to make native read and write calls
private static final NativeDispatcher nd = new FileDispatcherImpl();
// The file descriptor associated with this channel
private final FileDescriptor fd;
-
- // fd value needed for dev/poll. This value will remain valid
- // even after the value in the file descriptor object has been set to -1
private final int fdVal;
- // ID of native thread doing read, for signalling
- private volatile long thread;
-
// Lock held by current reading thread
private final ReentrantLock readLock = new ReentrantLock();
@@ -63,10 +58,14 @@
// -- The following fields are protected by stateLock
// Channel state
- private static final int ST_UNINITIALIZED = -1;
private static final int ST_INUSE = 0;
- private static final int ST_KILLED = 1;
- private volatile int state = ST_UNINITIALIZED;
+ private static final int ST_CLOSING = 1;
+ private static final int ST_KILLPENDING = 2;
+ private static final int ST_KILLED = 3;
+ private int state;
+
+ // ID of native thread doing read, for signalling
+ private long thread;
// -- End of fields protected by stateLock
@@ -83,39 +82,88 @@
super(sp);
this.fd = fd;
this.fdVal = IOUtil.fdVal(fd);
- this.state = ST_INUSE;
}
+ /**
+ * Invoked by implCloseChannel to close the channel.
+ */
+ @Override
protected void implCloseSelectableChannel() throws IOException {
+ assert !isOpen();
+
+ boolean interrupted = false;
+ boolean blocking;
+
+ // set state to ST_CLOSING
synchronized (stateLock) {
- if (state != ST_KILLED)
- nd.preClose(fd);
- long th = thread;
- if (th != 0)
- NativeThread.signal(th);
- if (!isRegistered())
- kill();
+ assert state < ST_CLOSING;
+ state = ST_CLOSING;
+ blocking = isBlocking();
+ }
+
+ // wait for any outstanding read to complete
+ if (blocking) {
+ synchronized (stateLock) {
+ assert state == ST_CLOSING;
+ long th = thread;
+ if (th != 0) {
+ nd.preClose(fd);
+ NativeThread.signal(th);
+
+ // wait for read operation to end
+ while (thread != 0) {
+ try {
+ stateLock.wait();
+ } catch (InterruptedException e) {
+ interrupted = true;
+ }
+ }
+ }
+ }
+ } else {
+ // non-blocking mode: wait for read to complete
+ readLock.lock();
+ readLock.unlock();
+ }
+
+ // set state to ST_KILLPENDING
+ synchronized (stateLock) {
+ assert state == ST_CLOSING;
+ 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) {
+ assert thread == 0;
+ if (state == ST_KILLPENDING) {
+ state = ST_KILLED;
+ nd.close(fd);
+ }
}
}
- public void kill() throws IOException {
- synchronized (stateLock) {
- if (state == ST_KILLED)
- return;
- if (state == ST_UNINITIALIZED) {
- state = ST_KILLED;
- return;
+ @Override
+ protected void implConfigureBlocking(boolean block) throws IOException {
+ readLock.lock();
+ try {
+ synchronized (stateLock) {
+ IOUtil.configureBlocking(fd, block);
}
- assert !isOpen() && !isRegistered();
- nd.close(fd);
- state = ST_KILLED;
+ } finally {
+ readLock.unlock();
}
}
- protected void implConfigureBlocking(boolean block) throws IOException {
- IOUtil.configureBlocking(fd, block);
- }
-
public boolean translateReadyOps(int ops, int initialOps,
SelectionKeyImpl sk) {
int intOps = sk.nioInterestOps(); // Do this just once, it synchronizes
@@ -153,68 +201,95 @@
sk.selector.putEventOps(sk, ops);
}
- private void ensureOpen() throws IOException {
- if (!isOpen())
- throw new ClosedChannelException();
+ /**
+ * Marks the beginning of a read operation that might block.
+ *
+ * @throws ClosedChannelException if the channel is closed
+ * @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) {
+ if (!isOpen())
+ throw new ClosedChannelException();
+ if (blocking)
+ thread = NativeThread.current();
+ }
}
+ /**
+ * Marks the end of a read operation that may have blocked.
+ *
+ * @throws AsynchronousCloseException if the channel was closed due to this
+ * thread being interrupted on a blocking read operation.
+ */
+ private void endRead(boolean blocking, boolean completed)
+ throws AsynchronousCloseException
+ {
+ if (blocking) {
+ synchronized (stateLock) {
+ thread = 0;
+ // notify any thread waiting in implCloseSelectableChannel
+ if (state == ST_CLOSING) {
+ stateLock.notifyAll();
+ }
+ }
+ // remove hook for Thread.interrupt
+ end(completed);
+ }
+ }
+
+ @Override
public int read(ByteBuffer dst) throws IOException {
+ Objects.requireNonNull(dst);
readLock.lock();
try {
- ensureOpen();
+ boolean blocking = isBlocking();
int n = 0;
try {
- begin();
- if (!isOpen())
- return 0;
- thread = NativeThread.current();
+ beginRead(blocking);
do {
n = IOUtil.read(fd, dst, -1, nd);
} while ((n == IOStatus.INTERRUPTED) && isOpen());
- return IOStatus.normalize(n);
} finally {
- thread = 0;
- end((n > 0) || (n == IOStatus.UNAVAILABLE));
+ endRead(blocking, n > 0);
assert IOStatus.check(n);
}
+ return IOStatus.normalize(n);
} finally {
readLock.unlock();
}
}
- public long read(ByteBuffer[] dsts, int offset, int length)
- throws IOException
- {
- if ((offset < 0) || (length < 0) || (offset > dsts.length - length))
- throw new IndexOutOfBoundsException();
- return read(Util.subsequence(dsts, offset, length));
- }
-
- public long read(ByteBuffer[] dsts) throws IOException {
- if (dsts == null)
- throw new NullPointerException();
+ @Override
+ public long read(ByteBuffer[] dsts, int offset, int length) throws IOException {
+ Objects.checkFromIndexSize(offset, length, dsts.length);
readLock.lock();
try {
- ensureOpen();
+ boolean blocking = isBlocking();
long n = 0;
try {
- begin();
- if (!isOpen())
- return 0;
- thread = NativeThread.current();
+ beginRead(blocking);
do {
- n = IOUtil.read(fd, dsts, nd);
+ n = IOUtil.read(fd, dsts, offset, length, nd);
} while ((n == IOStatus.INTERRUPTED) && isOpen());
- return IOStatus.normalize(n);
} finally {
- thread = 0;
- end((n > 0) || (n == IOStatus.UNAVAILABLE));
+ endRead(blocking, n > 0);
assert IOStatus.check(n);
}
+ return IOStatus.normalize(n);
} finally {
readLock.unlock();
}
}
+
+ @Override
+ public long read(ByteBuffer[] dsts) throws IOException {
+ return read(dsts, 0, dsts.length);
+ }
}