archive/jdk
1
0
Fork 0
mirror of https://github.com/openjdk/jdk.git synced 2025-08-27 06:45:07 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions

8223353: (ch) Change channel close implementation to not wait for I/O threads

Browse source 
Reviewed-by: dfuchs, chegar
This commit is contained in:
Alan Bateman 2019-05-08 08:15:04 +01:00
parent 260ae30b14
commit 94d1d0d3d4
6 changed files with 478 additions and 669 deletions
Download patch file Download diff file Expand all files Collapse all files

13
src/java.base/share/classes/java/nio/channels/spi/AbstractInterruptibleChannel.java
View file

@ -30,7 +30,6 @@ import java.nio.channels.AsynchronousCloseException;
import java.nio.channels.Channel; import java.nio.channels.Channel;
import java.nio.channels.ClosedByInterruptException; import java.nio.channels.ClosedByInterruptException;
import java.nio.channels.InterruptibleChannel; import java.nio.channels.InterruptibleChannel;
import java.util.concurrent.locks.ReentrantLock;
import jdk.internal.access.SharedSecrets; import jdk.internal.access.SharedSecrets;
import sun.nio.ch.Interruptible; import sun.nio.ch.Interruptible;
@ -86,7 +85,7 @@ import sun.nio.ch.Interruptible;
public abstract class AbstractInterruptibleChannel public abstract class AbstractInterruptibleChannel
implements Channel, InterruptibleChannel implements Channel, InterruptibleChannel
{ {
private final ReentrantLock closeLock = new ReentrantLock(); private final Object closeLock = new Object();
private volatile boolean closed; private volatile boolean closed;
/** /**
@ -106,14 +105,11 @@ public abstract class AbstractInterruptibleChannel
* If an I/O error occurs * If an I/O error occurs
*/ */
public final void close() throws IOException { public final void close() throws IOException {
closeLock.lock(); synchronized (closeLock) {
try {
if (closed) if (closed)
return; return;
closed = true; closed = true;
implCloseChannel(); implCloseChannel();
} finally {
closeLock.unlock();
} }
} }
@ -157,8 +153,7 @@ public abstract class AbstractInterruptibleChannel
if (interruptor == null) { if (interruptor == null) {
interruptor = new Interruptible() { interruptor = new Interruptible() {
public void interrupt(Thread target) { public void interrupt(Thread target) {
closeLock.lock(); synchronized (closeLock) {
try {
if (closed) if (closed)
return; return;
closed = true; closed = true;
@ -166,8 +161,6 @@ public abstract class AbstractInterruptibleChannel
try { try {
AbstractInterruptibleChannel.this.implCloseChannel(); AbstractInterruptibleChannel.this.implCloseChannel();
} catch (IOException x) { } } catch (IOException x) { }
} finally {
closeLock.unlock();
} }
}}; }};
} }

280
src/java.base/share/classes/sun/nio/ch/DatagramChannelImpl.java
View file

@ -53,7 +53,6 @@ import java.util.Collections;
import java.util.HashSet; import java.util.HashSet;
import java.util.Objects; import java.util.Objects;
import java.util.Set; import java.util.Set;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock; import java.util.concurrent.locks.ReentrantLock;
import sun.net.ResourceManager; import sun.net.ResourceManager;
@ -90,8 +89,7 @@ class DatagramChannelImpl
// Lock held by any thread that modifies the state fields declared below // Lock held by any thread that modifies the state fields declared below
// DO NOT invoke a blocking I/O operation while holding this lock! // DO NOT invoke a blocking I/O operation while holding this lock!
private final ReentrantLock stateLock = new ReentrantLock(); private final Object stateLock = new Object();
private final Condition stateCondition = stateLock.newCondition();
// -- The following fields are protected by stateLock // -- The following fields are protected by stateLock
@ -99,8 +97,7 @@ class DatagramChannelImpl
private static final int ST_UNCONNECTED = 0; private static final int ST_UNCONNECTED = 0;
private static final int ST_CONNECTED = 1; private static final int ST_CONNECTED = 1;
private static final int ST_CLOSING = 2; private static final int ST_CLOSING = 2;
private static final int ST_KILLPENDING = 3; private static final int ST_CLOSED = 3;
private static final int ST_KILLED = 4;
private int state; private int state;
// IDs of native threads doing reads and writes, for signalling // IDs of native threads doing reads and writes, for signalling
@ -181,11 +178,8 @@ class DatagramChannelImpl
: StandardProtocolFamily.INET; : StandardProtocolFamily.INET;
this.fd = fd; this.fd = fd;
this.fdVal = IOUtil.fdVal(fd); this.fdVal = IOUtil.fdVal(fd);
stateLock.lock(); synchronized (stateLock) {
try {
this.localAddress = Net.localAddress(fd); this.localAddress = Net.localAddress(fd);
} finally {
stateLock.unlock();
} }
} }
@ -197,36 +191,27 @@ class DatagramChannelImpl
@Override @Override
public DatagramSocket socket() { public DatagramSocket socket() {
stateLock.lock(); synchronized (stateLock) {
try {
if (socket == null) if (socket == null)
socket = DatagramSocketAdaptor.create(this); socket = DatagramSocketAdaptor.create(this);
return socket; return socket;
} finally {
stateLock.unlock();
} }
} }
@Override @Override
public SocketAddress getLocalAddress() throws IOException { public SocketAddress getLocalAddress() throws IOException {
stateLock.lock(); synchronized (stateLock) {
try {
ensureOpen(); ensureOpen();
// Perform security check before returning address // Perform security check before returning address
return Net.getRevealedLocalAddress(localAddress); return Net.getRevealedLocalAddress(localAddress);
} finally {
stateLock.unlock();
} }
} }
@Override @Override
public SocketAddress getRemoteAddress() throws IOException { public SocketAddress getRemoteAddress() throws IOException {
stateLock.lock(); synchronized (stateLock) {
try {
ensureOpen(); ensureOpen();
return remoteAddress; return remoteAddress;
} finally {
stateLock.unlock();
} }
} }
@ -238,8 +223,7 @@ class DatagramChannelImpl
if (!supportedOptions().contains(name)) if (!supportedOptions().contains(name))
throw new UnsupportedOperationException("'" + name + "' not supported"); throw new UnsupportedOperationException("'" + name + "' not supported");
stateLock.lock(); synchronized (stateLock) {
try {
ensureOpen(); ensureOpen();
if (name == StandardSocketOptions.IP_TOS || if (name == StandardSocketOptions.IP_TOS ||
@ -279,8 +263,6 @@ class DatagramChannelImpl
// remaining options don't need any special handling // remaining options don't need any special handling
Net.setSocketOption(fd, Net.UNSPEC, name, value); Net.setSocketOption(fd, Net.UNSPEC, name, value);
return this; return this;
} finally {
stateLock.unlock();
} }
} }
@ -293,8 +275,7 @@ class DatagramChannelImpl
if (!supportedOptions().contains(name)) if (!supportedOptions().contains(name))
throw new UnsupportedOperationException("'" + name + "' not supported"); throw new UnsupportedOperationException("'" + name + "' not supported");
stateLock.lock(); synchronized (stateLock) {
try {
ensureOpen(); ensureOpen();
if (name == StandardSocketOptions.IP_TOS || if (name == StandardSocketOptions.IP_TOS ||
@ -333,8 +314,6 @@ class DatagramChannelImpl
// no special handling // no special handling
return (T) Net.getSocketOption(fd, Net.UNSPEC, name); return (T) Net.getSocketOption(fd, Net.UNSPEC, name);
} finally {
stateLock.unlock();
} }
} }
@ -382,8 +361,7 @@ class DatagramChannelImpl
begin(); begin();
} }
SocketAddress remote; SocketAddress remote;
stateLock.lock(); synchronized (stateLock) {
try {
ensureOpen(); ensureOpen();
remote = remoteAddress; remote = remoteAddress;
if ((remote == null) && mustBeConnected) if ((remote == null) && mustBeConnected)
@ -392,8 +370,6 @@ class DatagramChannelImpl
bindInternal(null); bindInternal(null);
if (blocking) if (blocking)
readerThread = NativeThread.current(); readerThread = NativeThread.current();
} finally {
stateLock.unlock();
} }
return remote; return remote;
} }
@ -407,15 +383,11 @@ class DatagramChannelImpl
throws AsynchronousCloseException throws AsynchronousCloseException
{ {
if (blocking) { if (blocking) {
stateLock.lock(); synchronized (stateLock) {
try {
readerThread = 0; readerThread = 0;
// notify any thread waiting in implCloseSelectableChannel
if (state == ST_CLOSING) { if (state == ST_CLOSING) {
stateCondition.signalAll(); tryFinishClose();
} }
} finally {
stateLock.unlock();
} }
// remove hook for Thread.interrupt // remove hook for Thread.interrupt
end(completed); end(completed);
@ -708,8 +680,7 @@ class DatagramChannelImpl
begin(); begin();
} }
SocketAddress remote; SocketAddress remote;
stateLock.lock(); synchronized (stateLock) {
try {
ensureOpen(); ensureOpen();
remote = remoteAddress; remote = remoteAddress;
if ((remote == null) && mustBeConnected) if ((remote == null) && mustBeConnected)
@ -718,8 +689,6 @@ class DatagramChannelImpl
bindInternal(null); bindInternal(null);
if (blocking) if (blocking)
writerThread = NativeThread.current(); writerThread = NativeThread.current();
} finally {
stateLock.unlock();
} }
return remote; return remote;
} }
@ -733,15 +702,11 @@ class DatagramChannelImpl
throws AsynchronousCloseException throws AsynchronousCloseException
{ {
if (blocking) { if (blocking) {
stateLock.lock(); synchronized (stateLock) {
try {
writerThread = 0; writerThread = 0;
// notify any thread waiting in implCloseSelectableChannel
if (state == ST_CLOSING) { if (state == ST_CLOSING) {
stateCondition.signalAll(); tryFinishClose();
} }
} finally {
stateLock.unlock();
} }
// remove hook for Thread.interrupt // remove hook for Thread.interrupt
end(completed); end(completed);
@ -810,12 +775,9 @@ class DatagramChannelImpl
try { try {
writeLock.lock(); writeLock.lock();
try { try {
stateLock.lock(); synchronized (stateLock) {
try {
ensureOpen(); ensureOpen();
IOUtil.configureBlocking(fd, block); IOUtil.configureBlocking(fd, block);
} finally {
stateLock.unlock();
} }
} finally { } finally {
writeLock.unlock(); writeLock.unlock();
@ -826,20 +788,14 @@ class DatagramChannelImpl
} }
InetSocketAddress localAddress() { InetSocketAddress localAddress() {
stateLock.lock(); synchronized (stateLock) {
try {
return localAddress; return localAddress;
} finally {
stateLock.unlock();
} }
} }
InetSocketAddress remoteAddress() { InetSocketAddress remoteAddress() {
stateLock.lock(); synchronized (stateLock) {
try {
return remoteAddress; return remoteAddress;
} finally {
stateLock.unlock();
} }
} }
@ -849,14 +805,11 @@ class DatagramChannelImpl
try { try {
writeLock.lock(); writeLock.lock();
try { try {
stateLock.lock(); synchronized (stateLock) {
try {
ensureOpen(); ensureOpen();
if (localAddress != null) if (localAddress != null)
throw new AlreadyBoundException(); throw new AlreadyBoundException();
bindInternal(local); bindInternal(local);
} finally {
stateLock.unlock();
} }
} finally { } finally {
writeLock.unlock(); writeLock.unlock();
@ -868,7 +821,7 @@ class DatagramChannelImpl
} }
private void bindInternal(SocketAddress local) throws IOException { private void bindInternal(SocketAddress local) throws IOException {
assert stateLock.isHeldByCurrentThread() && (localAddress == null); assert Thread.holdsLock(stateLock )&& (localAddress == null);
InetSocketAddress isa; InetSocketAddress isa;
if (local == null) { if (local == null) {
@ -891,11 +844,8 @@ class DatagramChannelImpl
@Override @Override
public boolean isConnected() { public boolean isConnected() {
stateLock.lock(); synchronized (stateLock) {
try {
return (state == ST_CONNECTED); return (state == ST_CONNECTED);
} finally {
stateLock.unlock();
} }
} }
@ -917,8 +867,7 @@ class DatagramChannelImpl
try { try {
writeLock.lock(); writeLock.lock();
try { try {
stateLock.lock(); synchronized (stateLock) {
try {
ensureOpen(); ensureOpen();
if (state == ST_CONNECTED) if (state == ST_CONNECTED)
throw new AlreadyConnectedException(); throw new AlreadyConnectedException();
@ -952,9 +901,6 @@ class DatagramChannelImpl
IOUtil.configureBlocking(fd, true); IOUtil.configureBlocking(fd, true);
} }
} }
} finally {
stateLock.unlock();
} }
} finally { } finally {
writeLock.unlock(); writeLock.unlock();
@ -971,8 +917,7 @@ class DatagramChannelImpl
try { try {
writeLock.lock(); writeLock.lock();
try { try {
stateLock.lock(); synchronized (stateLock) {
try {
if (!isOpen() || (state != ST_CONNECTED)) if (!isOpen() || (state != ST_CONNECTED))
return this; return this;
@ -986,8 +931,6 @@ class DatagramChannelImpl
// refresh local address // refresh local address
localAddress = Net.localAddress(fd); localAddress = Net.localAddress(fd);
} finally {
stateLock.unlock();
} }
} finally { } finally {
writeLock.unlock(); writeLock.unlock();
@ -1035,8 +978,7 @@ class DatagramChannelImpl
if (sm != null) if (sm != null)
sm.checkMulticast(group); sm.checkMulticast(group);
stateLock.lock(); synchronized (stateLock) {
try {
ensureOpen(); ensureOpen();
// check the registry to see if we are already a member of the group // check the registry to see if we are already a member of the group
@ -1091,8 +1033,6 @@ class DatagramChannelImpl
registry.add(key); registry.add(key);
return key; return key;
} finally {
stateLock.unlock();
} }
} }
@ -1118,8 +1058,7 @@ class DatagramChannelImpl
void drop(MembershipKeyImpl key) { void drop(MembershipKeyImpl key) {
assert key.channel() == this; assert key.channel() == this;
stateLock.lock(); synchronized (stateLock) {
try {
if (!key.isValid()) if (!key.isValid())
return; return;
@ -1140,8 +1079,6 @@ class DatagramChannelImpl
key.invalidate(); key.invalidate();
registry.remove(key); registry.remove(key);
} finally {
stateLock.unlock();
} }
} }
@ -1155,8 +1092,7 @@ class DatagramChannelImpl
assert key.channel() == this; assert key.channel() == this;
assert key.sourceAddress() == null; assert key.sourceAddress() == null;
stateLock.lock(); synchronized (stateLock) {
try {
if (!key.isValid()) if (!key.isValid())
throw new IllegalStateException("key is no longer valid"); throw new IllegalStateException("key is no longer valid");
if (source.isAnyLocalAddress()) if (source.isAnyLocalAddress())
@ -1182,8 +1118,6 @@ class DatagramChannelImpl
// ancient kernel // ancient kernel
throw new UnsupportedOperationException(); throw new UnsupportedOperationException();
} }
} finally {
stateLock.unlock();
} }
} }
@ -1194,8 +1128,7 @@ class DatagramChannelImpl
assert key.channel() == this; assert key.channel() == this;
assert key.sourceAddress() == null; assert key.sourceAddress() == null;
stateLock.lock(); synchronized (stateLock) {
try {
if (!key.isValid()) if (!key.isValid())
throw new IllegalStateException("key is no longer valid"); throw new IllegalStateException("key is no longer valid");
@ -1215,116 +1148,117 @@ class DatagramChannelImpl
// should not happen // should not happen
throw new AssertionError(ioe); throw new AssertionError(ioe);
} }
} finally {
stateLock.unlock();
} }
} }
/** /**
* Invoked by implCloseChannel to close the channel. * Closes the socket if there are no I/O operations in progress and the
* * channel is not registered with a Selector.
* 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 private boolean tryClose() throws IOException {
protected void implCloseSelectableChannel() throws IOException { assert Thread.holdsLock(stateLock) && state == ST_CLOSING;
assert !isOpen(); if ((readerThread == 0) && (writerThread == 0) && !isRegistered()) {
state = ST_CLOSED;
try {
nd.close(fd);
} finally {
// notify resource manager
ResourceManager.afterUdpClose();
}
return true;
} else {
return false;
}
}
boolean blocking; /**
boolean interrupted = false; * Invokes tryClose to attempt to close the socket.
*
// set state to ST_CLOSING and invalid membership keys * This method is used for deferred closing by I/O and Selector operations.
stateLock.lock(); */
private void tryFinishClose() {
try { try {
tryClose();
} catch (IOException ignore) { }
}
/**
* Closes this channel when configured in blocking mode.
*
* If there is an I/O operation in progress then the socket is pre-closed
* and the I/O threads signalled, in which case the final close is deferred
* until all I/O operations complete.
*/
private void implCloseBlockingMode() throws IOException {
synchronized (stateLock) {
assert state < ST_CLOSING; assert state < ST_CLOSING;
blocking = isBlocking();
state = ST_CLOSING; state = ST_CLOSING;
// if member of any multicast groups then invalidate the keys // if member of any multicast groups then invalidate the keys
if (registry != null) if (registry != null)
registry.invalidateAll(); registry.invalidateAll();
} finally {
stateLock.unlock();
}
// wait for any outstanding I/O operations to complete if (!tryClose()) {
if (blocking) {
stateLock.lock();
try {
assert state == ST_CLOSING;
long reader = readerThread; long reader = readerThread;
long writer = writerThread; long writer = writerThread;
if (reader != 0 || writer != 0) { if (reader != 0 || writer != 0) {
nd.preClose(fd); nd.preClose(fd);
if (reader != 0) if (reader != 0)
NativeThread.signal(reader); NativeThread.signal(reader);
if (writer != 0) if (writer != 0)
NativeThread.signal(writer); NativeThread.signal(writer);
// wait for blocking I/O operations to end
while (readerThread != 0 || writerThread != 0) {
try {
stateCondition.await();
} catch (InterruptedException e) {
interrupted = true;
}
}
} }
} finally {
stateLock.unlock();
} }
}
}
/**
* Closes this channel when configured in non-blocking mode.
*
* If the channel is registered with a Selector then the close is deferred
* until the channel is flushed from all Selectors.
*/
private void implCloseNonBlockingMode() throws IOException {
synchronized (stateLock) {
assert state < ST_CLOSING;
state = ST_CLOSING;
// if member of any multicast groups then invalidate the keys
if (registry != null)
registry.invalidateAll();
}
// wait for any read/write operations to complete before trying to close
readLock.lock();
readLock.unlock();
writeLock.lock();
writeLock.unlock();
synchronized (stateLock) {
if (state == ST_CLOSING) {
tryClose();
}
}
}
/**
* Invoked by implCloseChannel to close the channel.
*/
@Override
protected void implCloseSelectableChannel() throws IOException {
assert !isOpen();
if (isBlocking()) {
implCloseBlockingMode();
} else { } else {
// non-blocking mode: wait for read/write to complete implCloseNonBlockingMode();
readLock.lock();
try {
writeLock.lock();
writeLock.unlock();
} finally {
readLock.unlock();
}
} }
// set state to ST_KILLPENDING
stateLock.lock();
try {
assert state == ST_CLOSING;
state = ST_KILLPENDING;
} finally {
stateLock.unlock();
}
// close socket if not registered with Selector
if (!isRegistered())
kill();
// restore interrupt status
if (interrupted)
Thread.currentThread().interrupt();
} }
@Override @Override
public void kill() throws IOException { public void kill() {
stateLock.lock(); synchronized (stateLock) {
try { if (state == ST_CLOSING) {
if (state == ST_KILLPENDING) { tryFinishClose();
state = ST_KILLED;
try {
nd.close(fd);
} finally {
// notify resource manager
ResourceManager.afterUdpClose();
}
} }
} finally {
stateLock.unlock();
} }
} }

219
src/java.base/share/classes/sun/nio/ch/ServerSocketChannelImpl.java
View file

@ -46,7 +46,6 @@ import java.util.Collections;
import java.util.HashSet; import java.util.HashSet;
import java.util.Objects; import java.util.Objects;
import java.util.Set; import java.util.Set;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock; import java.util.concurrent.locks.ReentrantLock;
import sun.net.NetHooks; import sun.net.NetHooks;
@ -72,16 +71,14 @@ class ServerSocketChannelImpl
// Lock held by any thread that modifies the state fields declared below // Lock held by any thread that modifies the state fields declared below
// DO NOT invoke a blocking I/O operation while holding this lock! // DO NOT invoke a blocking I/O operation while holding this lock!
private final ReentrantLock stateLock = new ReentrantLock(); private final Object stateLock = new Object();
private final Condition stateCondition = stateLock.newCondition();
// -- The following fields are protected by stateLock // -- The following fields are protected by stateLock
// Channel state, increases monotonically // Channel state, increases monotonically
private static final int ST_INUSE = 0; private static final int ST_INUSE = 0;
private static final int ST_CLOSING = 1; private static final int ST_CLOSING = 1;
private static final int ST_KILLPENDING = 2; private static final int ST_CLOSED = 2;
private static final int ST_KILLED = 3;
private int state; private int state;
// ID of native thread currently blocked in this channel, for signalling // ID of native thread currently blocked in this channel, for signalling
@ -112,11 +109,8 @@ class ServerSocketChannelImpl
this.fd = fd; this.fd = fd;
this.fdVal = IOUtil.fdVal(fd); this.fdVal = IOUtil.fdVal(fd);
if (bound) { if (bound) {
stateLock.lock(); synchronized (stateLock) {
try {
localAddress = Net.localAddress(fd); localAddress = Net.localAddress(fd);
} finally {
stateLock.unlock();
} }
} }
} }
@ -129,26 +123,20 @@ class ServerSocketChannelImpl
@Override @Override
public ServerSocket socket() { public ServerSocket socket() {
stateLock.lock(); synchronized (stateLock) {
try {
if (socket == null) if (socket == null)
socket = ServerSocketAdaptor.create(this); socket = ServerSocketAdaptor.create(this);
return socket; return socket;
} finally {
stateLock.unlock();
} }
} }
@Override @Override
public SocketAddress getLocalAddress() throws IOException { public SocketAddress getLocalAddress() throws IOException {
stateLock.lock(); synchronized (stateLock) {
try {
ensureOpen(); ensureOpen();
return (localAddress == null) return (localAddress == null)
? null ? null
: Net.getRevealedLocalAddress(localAddress); : Net.getRevealedLocalAddress(localAddress);
} finally {
stateLock.unlock();
} }
} }
@ -159,8 +147,7 @@ class ServerSocketChannelImpl
Objects.requireNonNull(name); Objects.requireNonNull(name);
if (!supportedOptions().contains(name)) if (!supportedOptions().contains(name))
throw new UnsupportedOperationException("'" + name + "' not supported"); throw new UnsupportedOperationException("'" + name + "' not supported");
stateLock.lock(); synchronized (stateLock) {
try {
ensureOpen(); ensureOpen();
if (name == StandardSocketOptions.SO_REUSEADDR && Net.useExclusiveBind()) { if (name == StandardSocketOptions.SO_REUSEADDR && Net.useExclusiveBind()) {
@ -171,8 +158,6 @@ class ServerSocketChannelImpl
Net.setSocketOption(fd, Net.UNSPEC, name, value); Net.setSocketOption(fd, Net.UNSPEC, name, value);
} }
return this; return this;
} finally {
stateLock.unlock();
} }
} }
@ -185,8 +170,7 @@ class ServerSocketChannelImpl
if (!supportedOptions().contains(name)) if (!supportedOptions().contains(name))
throw new UnsupportedOperationException("'" + name + "' not supported"); throw new UnsupportedOperationException("'" + name + "' not supported");
stateLock.lock(); synchronized (stateLock) {
try {
ensureOpen(); ensureOpen();
if (name == StandardSocketOptions.SO_REUSEADDR && Net.useExclusiveBind()) { if (name == StandardSocketOptions.SO_REUSEADDR && Net.useExclusiveBind()) {
// SO_REUSEADDR emulated when using exclusive bind // SO_REUSEADDR emulated when using exclusive bind
@ -194,8 +178,6 @@ class ServerSocketChannelImpl
} }
// no options that require special handling // no options that require special handling
return (T) Net.getSocketOption(fd, Net.UNSPEC, name); return (T) Net.getSocketOption(fd, Net.UNSPEC, name);
} finally {
stateLock.unlock();
} }
} }
@ -221,8 +203,7 @@ class ServerSocketChannelImpl
@Override @Override
public ServerSocketChannel bind(SocketAddress local, int backlog) throws IOException { public ServerSocketChannel bind(SocketAddress local, int backlog) throws IOException {
stateLock.lock(); synchronized (stateLock) {
try {
ensureOpen(); ensureOpen();
if (localAddress != null) if (localAddress != null)
throw new AlreadyBoundException(); throw new AlreadyBoundException();
@ -236,8 +217,6 @@ class ServerSocketChannelImpl
Net.bind(fd, isa.getAddress(), isa.getPort()); Net.bind(fd, isa.getAddress(), isa.getPort());
Net.listen(fd, backlog < 1 ? 50 : backlog); Net.listen(fd, backlog < 1 ? 50 : backlog);
localAddress = Net.localAddress(fd); localAddress = Net.localAddress(fd);
} finally {
stateLock.unlock();
} }
return this; return this;
} }
@ -251,15 +230,12 @@ class ServerSocketChannelImpl
private void begin(boolean blocking) throws ClosedChannelException { private void begin(boolean blocking) throws ClosedChannelException {
if (blocking) if (blocking)
begin(); // set blocker to close channel if interrupted begin(); // set blocker to close channel if interrupted
stateLock.lock(); synchronized (stateLock) {
try {
ensureOpen(); ensureOpen();
if (localAddress == null) if (localAddress == null)
throw new NotYetBoundException(); throw new NotYetBoundException();
if (blocking) if (blocking)
thread = NativeThread.current(); thread = NativeThread.current();
} finally {
stateLock.unlock();
} }
} }
@ -273,15 +249,11 @@ class ServerSocketChannelImpl
throws AsynchronousCloseException throws AsynchronousCloseException
{ {
if (blocking) { if (blocking) {
stateLock.lock(); synchronized (stateLock) {
try {
thread = 0; thread = 0;
// notify any thread waiting in implCloseSelectableChannel
if (state == ST_CLOSING) { if (state == ST_CLOSING) {
stateCondition.signalAll(); tryFinishClose();
} }
} finally {
stateLock.unlock();
} }
end(completed); end(completed);
} }
@ -405,101 +377,99 @@ class ServerSocketChannelImpl
*/ */
private void lockedConfigureBlocking(boolean block) throws IOException { private void lockedConfigureBlocking(boolean block) throws IOException {
assert acceptLock.isHeldByCurrentThread(); assert acceptLock.isHeldByCurrentThread();
stateLock.lock(); synchronized (stateLock) {
try {
ensureOpen(); ensureOpen();
IOUtil.configureBlocking(fd, block); IOUtil.configureBlocking(fd, block);
} finally { }
stateLock.unlock(); }
/**
* Closes the socket if there are no accept in progress and the channel is
* not registered with a Selector.
*/
private boolean tryClose() throws IOException {
assert Thread.holdsLock(stateLock) && state == ST_CLOSING;
if ((thread == 0) && !isRegistered()) {
state = ST_CLOSED;
nd.close(fd);
return true;
} else {
return false;
}
}
/**
* Invokes tryClose to attempt to close the socket.
*
* This method is used for deferred closing by I/O and Selector operations.
*/
private void tryFinishClose() {
try {
tryClose();
} catch (IOException ignore) { }
}
/**
* Closes this channel when configured in blocking mode.
*
* If there is an accept in progress then the socket is pre-closed and the
* accept thread is signalled, in which case the final close is deferred
* until the accept aborts.
*/
private void implCloseBlockingMode() throws IOException {
synchronized (stateLock) {
assert state < ST_CLOSING;
state = ST_CLOSING;
if (!tryClose()) {
long th = thread;
if (th != 0) {
nd.preClose(fd);
NativeThread.signal(th);
}
}
}
}
/**
* Closes this channel when configured in non-blocking mode.
*
* If the channel is registered with a Selector then the close is deferred
* until the channel is flushed from all Selectors.
*/
private void implCloseNonBlockingMode() throws IOException {
synchronized (stateLock) {
assert state < ST_CLOSING;
state = ST_CLOSING;
}
// wait for any accept to complete before trying to close
acceptLock.lock();
acceptLock.unlock();
synchronized (stateLock) {
if (state == ST_CLOSING) {
tryClose();
}
} }
} }
/** /**
* Invoked by implCloseChannel to close the channel. * 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 @Override
protected void implCloseSelectableChannel() throws IOException { protected void implCloseSelectableChannel() throws IOException {
assert !isOpen(); assert !isOpen();
if (isBlocking()) {
boolean interrupted = false; implCloseBlockingMode();
boolean blocking;
// set state to ST_CLOSING
stateLock.lock();
try {
assert state < ST_CLOSING;
state = ST_CLOSING;
blocking = isBlocking();
} finally {
stateLock.unlock();
}
// wait for any outstanding accept to complete
if (blocking) {
stateLock.lock();
try {
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 {
stateCondition.await();
} catch (InterruptedException e) {
interrupted = true;
}
}
}
} finally {
stateLock.unlock();
}
} else { } else {
// non-blocking mode: wait for accept to complete implCloseNonBlockingMode();
acceptLock.lock();
acceptLock.unlock();
} }
// set state to ST_KILLPENDING
stateLock.lock();
try {
assert state == ST_CLOSING;
state = ST_KILLPENDING;
} finally {
stateLock.unlock();
}
// close socket if not registered with Selector
if (!isRegistered())
kill();
// restore interrupt status
if (interrupted)
Thread.currentThread().interrupt();
} }
@Override @Override
public void kill() throws IOException { public void kill() {
stateLock.lock(); synchronized (stateLock) {
try { if (state == ST_CLOSING) {
if (state == ST_KILLPENDING) { tryFinishClose();
state = ST_KILLED;
nd.close(fd);
} }
} finally {
stateLock.unlock();
} }
} }
@ -507,11 +477,8 @@ class ServerSocketChannelImpl
* Returns true if channel's socket is bound * Returns true if channel's socket is bound
*/ */
boolean isBound() { boolean isBound() {
stateLock.lock(); synchronized (stateLock) {
try {
return localAddress != null; return localAddress != null;
} finally {
stateLock.unlock();
} }
} }
@ -519,11 +486,8 @@ class ServerSocketChannelImpl
* Returns the local address, or null if not bound * Returns the local address, or null if not bound
*/ */
InetSocketAddress localAddress() { InetSocketAddress localAddress() {
stateLock.lock(); synchronized (stateLock) {
try {
return localAddress; return localAddress;
} finally {
stateLock.unlock();
} }
} }
@ -589,16 +553,13 @@ class ServerSocketChannelImpl
if (!isOpen()) { if (!isOpen()) {
sb.append("closed"); sb.append("closed");
} else { } else {
stateLock.lock(); synchronized (stateLock) {
try {
InetSocketAddress addr = localAddress; InetSocketAddress addr = localAddress;
if (addr == null) { if (addr == null) {
sb.append("unbound"); sb.append("unbound");
} else { } else {
sb.append(Net.getRevealedLocalAddressAsString(addr)); sb.append(Net.getRevealedLocalAddressAsString(addr));
} }
} finally {
stateLock.unlock();
} }
} }
sb.append(']'); sb.append(']');

299
src/java.base/share/classes/sun/nio/ch/SocketChannelImpl.java
View file

@ -53,7 +53,6 @@ import java.util.Collections;
import java.util.HashSet; import java.util.HashSet;
import java.util.Objects; import java.util.Objects;
import java.util.Set; import java.util.Set;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock; import java.util.concurrent.locks.ReentrantLock;
import sun.net.ConnectionResetException; import sun.net.ConnectionResetException;
@ -84,8 +83,7 @@ class SocketChannelImpl
// Lock held by any thread that modifies the state fields declared below // Lock held by any thread that modifies the state fields declared below
// DO NOT invoke a blocking I/O operation while holding this lock! // DO NOT invoke a blocking I/O operation while holding this lock!
private final ReentrantLock stateLock = new ReentrantLock(); private final Object stateLock = new Object();
private final Condition stateCondition = stateLock.newCondition();
// Input/Output closed // Input/Output closed
private volatile boolean isInputClosed; private volatile boolean isInputClosed;
@ -104,8 +102,7 @@ class SocketChannelImpl
private static final int ST_CONNECTIONPENDING = 1; private static final int ST_CONNECTIONPENDING = 1;
private static final int ST_CONNECTED = 2; private static final int ST_CONNECTED = 2;
private static final int ST_CLOSING = 3; private static final int ST_CLOSING = 3;
private static final int ST_KILLPENDING = 4; private static final int ST_CLOSED = 4;
private static final int ST_KILLED = 5;
private volatile int state; // need stateLock to change private volatile int state; // need stateLock to change
// IDs of native threads doing reads and writes, for signalling // IDs of native threads doing reads and writes, for signalling
@ -137,11 +134,8 @@ class SocketChannelImpl
this.fd = fd; this.fd = fd;
this.fdVal = IOUtil.fdVal(fd); this.fdVal = IOUtil.fdVal(fd);
if (bound) { if (bound) {
stateLock.lock(); synchronized (stateLock) {
try {
this.localAddress = Net.localAddress(fd); this.localAddress = Net.localAddress(fd);
} finally {
stateLock.unlock();
} }
} }
} }
@ -154,13 +148,10 @@ class SocketChannelImpl
super(sp); super(sp);
this.fd = fd; this.fd = fd;
this.fdVal = IOUtil.fdVal(fd); this.fdVal = IOUtil.fdVal(fd);
stateLock.lock(); synchronized (stateLock) {
try {
this.localAddress = Net.localAddress(fd); this.localAddress = Net.localAddress(fd);
this.remoteAddress = isa; this.remoteAddress = isa;
this.state = ST_CONNECTED; this.state = ST_CONNECTED;
} finally {
stateLock.unlock();
} }
} }
@ -197,35 +188,26 @@ class SocketChannelImpl
@Override @Override
public Socket socket() { public Socket socket() {
stateLock.lock(); synchronized (stateLock) {
try {
if (socket == null) if (socket == null)
socket = SocketAdaptor.create(this); socket = SocketAdaptor.create(this);
return socket; return socket;
} finally {
stateLock.unlock();
} }
} }
@Override @Override
public SocketAddress getLocalAddress() throws IOException { public SocketAddress getLocalAddress() throws IOException {
stateLock.lock(); synchronized (stateLock) {
try {
ensureOpen(); ensureOpen();
return Net.getRevealedLocalAddress(localAddress); return Net.getRevealedLocalAddress(localAddress);
} finally {
stateLock.unlock();
} }
} }
@Override @Override
public SocketAddress getRemoteAddress() throws IOException { public SocketAddress getRemoteAddress() throws IOException {
stateLock.lock(); synchronized (stateLock) {
try {
ensureOpen(); ensureOpen();
return remoteAddress; return remoteAddress;
} finally {
stateLock.unlock();
} }
} }
@ -237,8 +219,7 @@ class SocketChannelImpl
if (!supportedOptions().contains(name)) if (!supportedOptions().contains(name))
throw new UnsupportedOperationException("'" + name + "' not supported"); throw new UnsupportedOperationException("'" + name + "' not supported");
stateLock.lock(); synchronized (stateLock) {
try {
ensureOpen(); ensureOpen();
if (name == StandardSocketOptions.IP_TOS) { if (name == StandardSocketOptions.IP_TOS) {
@ -257,8 +238,6 @@ class SocketChannelImpl
// no options that require special handling // no options that require special handling
Net.setSocketOption(fd, name, value); Net.setSocketOption(fd, name, value);
return this; return this;
} finally {
stateLock.unlock();
} }
} }
@ -271,8 +250,7 @@ class SocketChannelImpl
if (!supportedOptions().contains(name)) if (!supportedOptions().contains(name))
throw new UnsupportedOperationException("'" + name + "' not supported"); throw new UnsupportedOperationException("'" + name + "' not supported");
stateLock.lock(); synchronized (stateLock) {
try {
ensureOpen(); ensureOpen();
if (name == StandardSocketOptions.SO_REUSEADDR && Net.useExclusiveBind()) { if (name == StandardSocketOptions.SO_REUSEADDR && Net.useExclusiveBind()) {
@ -289,8 +267,6 @@ class SocketChannelImpl
// no options that require special handling // no options that require special handling
return (T) Net.getSocketOption(fd, name); return (T) Net.getSocketOption(fd, name);
} finally {
stateLock.unlock();
} }
} }
@ -332,13 +308,10 @@ class SocketChannelImpl
// set hook for Thread.interrupt // set hook for Thread.interrupt
begin(); begin();
stateLock.lock(); synchronized (stateLock) {
try {
ensureOpenAndConnected(); ensureOpenAndConnected();
// record thread so it can be signalled if needed // record thread so it can be signalled if needed
readerThread = NativeThread.current(); readerThread = NativeThread.current();
} finally {
stateLock.unlock();
} }
} else { } else {
ensureOpenAndConnected(); ensureOpenAndConnected();
@ -355,15 +328,11 @@ class SocketChannelImpl
throws AsynchronousCloseException throws AsynchronousCloseException
{ {
if (blocking) { if (blocking) {
stateLock.lock(); synchronized (stateLock) {
try {
readerThread = 0; readerThread = 0;
// notify any thread waiting in implCloseSelectableChannel
if (state == ST_CLOSING) { if (state == ST_CLOSING) {
stateCondition.signalAll(); tryFinishClose();
} }
} finally {
stateLock.unlock();
} }
// remove hook for Thread.interrupt // remove hook for Thread.interrupt
end(completed); end(completed);
@ -467,15 +436,12 @@ class SocketChannelImpl
// set hook for Thread.interrupt // set hook for Thread.interrupt
begin(); begin();
stateLock.lock(); synchronized (stateLock) {
try {
ensureOpenAndConnected(); ensureOpenAndConnected();
if (isOutputClosed) if (isOutputClosed)
throw new ClosedChannelException(); throw new ClosedChannelException();
// record thread so it can be signalled if needed // record thread so it can be signalled if needed
writerThread = NativeThread.current(); writerThread = NativeThread.current();
} finally {
stateLock.unlock();
} }
} else { } else {
ensureOpenAndConnected(); ensureOpenAndConnected();
@ -492,15 +458,11 @@ class SocketChannelImpl
throws AsynchronousCloseException throws AsynchronousCloseException
{ {
if (blocking) { if (blocking) {
stateLock.lock(); synchronized (stateLock) {
try {
writerThread = 0; writerThread = 0;
// notify any thread waiting in implCloseSelectableChannel
if (state == ST_CLOSING) { if (state == ST_CLOSING) {
stateCondition.signalAll(); tryFinishClose();
} }
} finally {
stateLock.unlock();
} }
// remove hook for Thread.interrupt // remove hook for Thread.interrupt
end(completed); end(completed);
@ -613,12 +575,9 @@ class SocketChannelImpl
*/ */
private void lockedConfigureBlocking(boolean block) throws IOException { private void lockedConfigureBlocking(boolean block) throws IOException {
assert readLock.isHeldByCurrentThread() || writeLock.isHeldByCurrentThread(); assert readLock.isHeldByCurrentThread() || writeLock.isHeldByCurrentThread();
stateLock.lock(); synchronized (stateLock) {
try {
ensureOpen(); ensureOpen();
IOUtil.configureBlocking(fd, block); IOUtil.configureBlocking(fd, block);
} finally {
stateLock.unlock();
} }
} }
@ -626,11 +585,8 @@ class SocketChannelImpl
* Returns the local address, or null if not bound * Returns the local address, or null if not bound
*/ */
InetSocketAddress localAddress() { InetSocketAddress localAddress() {
stateLock.lock(); synchronized (stateLock) {
try {
return localAddress; return localAddress;
} finally {
stateLock.unlock();
} }
} }
@ -638,11 +594,8 @@ class SocketChannelImpl
* Returns the remote address, or null if not connected * Returns the remote address, or null if not connected
*/ */
InetSocketAddress remoteAddress() { InetSocketAddress remoteAddress() {
stateLock.lock(); synchronized (stateLock) {
try {
return remoteAddress; return remoteAddress;
} finally {
stateLock.unlock();
} }
} }
@ -652,8 +605,7 @@ class SocketChannelImpl
try { try {
writeLock.lock(); writeLock.lock();
try { try {
stateLock.lock(); synchronized (stateLock) {
try {
ensureOpen(); ensureOpen();
if (state == ST_CONNECTIONPENDING) if (state == ST_CONNECTIONPENDING)
throw new ConnectionPendingException(); throw new ConnectionPendingException();
@ -668,8 +620,6 @@ class SocketChannelImpl
NetHooks.beforeTcpBind(fd, isa.getAddress(), isa.getPort()); NetHooks.beforeTcpBind(fd, isa.getAddress(), isa.getPort());
Net.bind(fd, isa.getAddress(), isa.getPort()); Net.bind(fd, isa.getAddress(), isa.getPort());
localAddress = Net.localAddress(fd); localAddress = Net.localAddress(fd);
} finally {
stateLock.unlock();
} }
} finally { } finally {
writeLock.unlock(); writeLock.unlock();
@ -706,8 +656,7 @@ class SocketChannelImpl
// set hook for Thread.interrupt // set hook for Thread.interrupt
begin(); begin();
} }
stateLock.lock(); synchronized (stateLock) {
try {
ensureOpen(); ensureOpen();
int state = this.state; int state = this.state;
if (state == ST_CONNECTED) if (state == ST_CONNECTED)
@ -725,8 +674,6 @@ class SocketChannelImpl
// record thread so it can be signalled if needed // record thread so it can be signalled if needed
readerThread = NativeThread.current(); readerThread = NativeThread.current();
} }
} finally {
stateLock.unlock();
} }
} }
@ -743,14 +690,11 @@ class SocketChannelImpl
endRead(blocking, completed); endRead(blocking, completed);
if (completed) { if (completed) {
stateLock.lock(); synchronized (stateLock) {
try {
if (state == ST_CONNECTIONPENDING) { if (state == ST_CONNECTIONPENDING) {
localAddress = Net.localAddress(fd); localAddress = Net.localAddress(fd);
state = ST_CONNECTED; state = ST_CONNECTED;
} }
} finally {
stateLock.unlock();
} }
} }
} }
@ -823,8 +767,7 @@ class SocketChannelImpl
// set hook for Thread.interrupt // set hook for Thread.interrupt
begin(); begin();
} }
stateLock.lock(); synchronized (stateLock) {
try {
ensureOpen(); ensureOpen();
if (state != ST_CONNECTIONPENDING) if (state != ST_CONNECTIONPENDING)
throw new NoConnectionPendingException(); throw new NoConnectionPendingException();
@ -832,8 +775,6 @@ class SocketChannelImpl
// record thread so it can be signalled if needed // record thread so it can be signalled if needed
readerThread = NativeThread.current(); readerThread = NativeThread.current();
} }
} finally {
stateLock.unlock();
} }
} }
@ -850,14 +791,11 @@ class SocketChannelImpl
endRead(blocking, completed); endRead(blocking, completed);
if (completed) { if (completed) {
stateLock.lock(); synchronized (stateLock) {
try {
if (state == ST_CONNECTIONPENDING) { if (state == ST_CONNECTIONPENDING) {
localAddress = Net.localAddress(fd); localAddress = Net.localAddress(fd);
state = ST_CONNECTED; state = ST_CONNECTED;
} }
} finally {
stateLock.unlock();
} }
} }
} }
@ -904,90 +842,89 @@ class SocketChannelImpl
} }
/** /**
* Invoked by implCloseChannel to close the channel. * Closes the socket if there are no I/O operations in progress and the
* * channel is not registered with a Selector.
* 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 private boolean tryClose() throws IOException {
protected void implCloseSelectableChannel() throws IOException { assert Thread.holdsLock(stateLock) && state == ST_CLOSING;
assert !isOpen(); if ((readerThread == 0) && (writerThread == 0) && !isRegistered()) {
state = ST_CLOSED;
boolean blocking; nd.close(fd);
boolean connected; return true;
boolean interrupted = false; } else {
return false;
// set state to ST_CLOSING
stateLock.lock();
try {
assert state < ST_CLOSING;
blocking = isBlocking();
connected = (state == ST_CONNECTED);
state = ST_CLOSING;
} finally {
stateLock.unlock();
} }
}
// wait for any outstanding I/O operations to complete /**
if (blocking) { * Invokes tryClose to attempt to close the socket.
stateLock.lock(); *
try { * This method is used for deferred closing by I/O and Selector operations.
assert state == ST_CLOSING; */
private void tryFinishClose() {
try {
tryClose();
} catch (IOException ignore) { }
}
/**
* Closes this channel when configured in blocking mode.
*
* If there is an I/O operation in progress then the socket is pre-closed
* and the I/O threads signalled, in which case the final close is deferred
* until all I/O operations complete.
*
* 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.
*/
private void implCloseBlockingMode() throws IOException {
synchronized (stateLock) {
assert state < ST_CLOSING;
state = ST_CLOSING;
if (!tryClose()) {
long reader = readerThread; long reader = readerThread;
long writer = writerThread; long writer = writerThread;
if (reader != 0 || writer != 0) { if (reader != 0 || writer != 0) {
nd.preClose(fd); nd.preClose(fd);
connected = false; // fd is no longer connected socket
if (reader != 0) if (reader != 0)
NativeThread.signal(reader); NativeThread.signal(reader);
if (writer != 0) if (writer != 0)
NativeThread.signal(writer); NativeThread.signal(writer);
// wait for blocking I/O operations to end
while (readerThread != 0 || writerThread != 0) {
try {
stateCondition.await();
} catch (InterruptedException e) {
interrupted = true;
}
}
} }
} finally {
stateLock.unlock();
}
} 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 /**
stateLock.lock(); * Closes this channel when configured in non-blocking mode.
try { *
assert state == ST_CLOSING; * If the channel is registered with a Selector then the close is deferred
// if connected and the channel is registered with a Selector then * until the channel is flushed from all Selectors.
// shutdown the output if possible so that the peer reads EOF. If *
// SO_LINGER is enabled and set to a non-zero value then it needs to * If the socket is connected and the channel is registered with a Selector
// be disabled so that the Selector does not wait when it closes * then the socket is shutdown for writing so that the peer reads EOF. In
// the socket. * addition, if SO_LINGER is set to a non-zero value then it is disabled so
if (connected && isRegistered()) { * that the deferred close does not wait.
*/
private void implCloseNonBlockingMode() throws IOException {
boolean connected;
synchronized (stateLock) {
assert state < ST_CLOSING;
connected = (state == ST_CONNECTED);
state = ST_CLOSING;
}
// wait for any read/write operations to complete
readLock.lock();
readLock.unlock();
writeLock.lock();
writeLock.unlock();
// if the socket cannot be closed because it's registered with a Selector
// then shutdown the socket for writing.
synchronized (stateLock) {
if (state == ST_CLOSING && !tryClose() && connected && isRegistered()) {
try { try {
SocketOption<Integer> opt = StandardSocketOptions.SO_LINGER; SocketOption<Integer> opt = StandardSocketOptions.SO_LINGER;
int interval = (int) Net.getSocketOption(fd, Net.UNSPEC, opt); int interval = (int) Net.getSocketOption(fd, Net.UNSPEC, opt);
@ -1000,37 +937,34 @@ class SocketChannelImpl
} }
} catch (IOException ignore) { } } catch (IOException ignore) { }
} }
state = ST_KILLPENDING;
} finally {
stateLock.unlock();
} }
}
// close socket if not registered with Selector /**
if (!isRegistered()) * Invoked by implCloseChannel to close the channel.
kill(); */
@Override
// restore interrupt status protected void implCloseSelectableChannel() throws IOException {
if (interrupted) assert !isOpen();
Thread.currentThread().interrupt(); if (isBlocking()) {
implCloseBlockingMode();
} else {
implCloseNonBlockingMode();
}
} }
@Override @Override
public void kill() throws IOException { public void kill() {
stateLock.lock(); synchronized (stateLock) {
try { if (state == ST_CLOSING) {
if (state == ST_KILLPENDING) { tryFinishClose();
state = ST_KILLED;
nd.close(fd);
} }
} finally {
stateLock.unlock();
} }
} }
@Override @Override
public SocketChannel shutdownInput() throws IOException { public SocketChannel shutdownInput() throws IOException {
stateLock.lock(); synchronized (stateLock) {
try {
ensureOpen(); ensureOpen();
if (!isConnected()) if (!isConnected())
throw new NotYetConnectedException(); throw new NotYetConnectedException();
@ -1042,15 +976,12 @@ class SocketChannelImpl
isInputClosed = true; isInputClosed = true;
} }
return this; return this;
} finally {
stateLock.unlock();
} }
} }
@Override @Override
public SocketChannel shutdownOutput() throws IOException { public SocketChannel shutdownOutput() throws IOException {
stateLock.lock(); synchronized (stateLock) {
try {
ensureOpen(); ensureOpen();
if (!isConnected()) if (!isConnected())
throw new NotYetConnectedException(); throw new NotYetConnectedException();
@ -1062,8 +993,6 @@ class SocketChannelImpl
isOutputClosed = true; isOutputClosed = true;
} }
return this; return this;
} finally {
stateLock.unlock();
} }
} }
@ -1300,16 +1229,13 @@ class SocketChannelImpl
* Return the number of bytes in the socket input buffer. * Return the number of bytes in the socket input buffer.
*/ */
int available() throws IOException { int available() throws IOException {
stateLock.lock(); synchronized (stateLock) {
try {
ensureOpenAndConnected(); ensureOpenAndConnected();
if (isInputClosed) { if (isInputClosed) {
return 0; return 0;
} else { } else {
return Net.available(fd); return Net.available(fd);
} }
} finally {
stateLock.unlock();
} }
} }
@ -1389,8 +1315,7 @@ class SocketChannelImpl
if (!isOpen()) if (!isOpen())
sb.append("closed"); sb.append("closed");
else { else {
stateLock.lock(); synchronized (stateLock) {
try {
switch (state) { switch (state) {
case ST_UNCONNECTED: case ST_UNCONNECTED:
sb.append("unconnected"); sb.append("unconnected");
@ -1415,8 +1340,6 @@ class SocketChannelImpl
sb.append(" remote="); sb.append(" remote=");
sb.append(remoteAddress().toString()); sb.append(remoteAddress().toString());
} }
} finally {
stateLock.unlock();
} }
} }
sb.append(']'); sb.append(']');

167
src/java.base/unix/classes/sun/nio/ch/SinkChannelImpl.java
View file

@ -35,7 +35,6 @@ import java.nio.channels.Pipe;
import java.nio.channels.SelectionKey; import java.nio.channels.SelectionKey;
import java.nio.channels.spi.SelectorProvider; import java.nio.channels.spi.SelectorProvider;
import java.util.Objects; import java.util.Objects;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock; import java.util.concurrent.locks.ReentrantLock;
class SinkChannelImpl class SinkChannelImpl
@ -54,16 +53,14 @@ class SinkChannelImpl
// Lock held by any thread that modifies the state fields declared below // Lock held by any thread that modifies the state fields declared below
// DO NOT invoke a blocking I/O operation while holding this lock! // DO NOT invoke a blocking I/O operation while holding this lock!
private final ReentrantLock stateLock = new ReentrantLock(); private final Object stateLock = new Object();
private final Condition stateCondition = stateLock.newCondition();
// -- The following fields are protected by stateLock // -- The following fields are protected by stateLock
// Channel state // Channel state
private static final int ST_INUSE = 0; private static final int ST_INUSE = 0;
private static final int ST_CLOSING = 1; private static final int ST_CLOSING = 1;
private static final int ST_KILLPENDING = 2; private static final int ST_CLOSED = 2;
private static final int ST_KILLED = 3;
private int state; private int state;
// ID of native thread doing write, for signalling // ID of native thread doing write, for signalling
@ -86,83 +83,93 @@ class SinkChannelImpl
this.fdVal = IOUtil.fdVal(fd); this.fdVal = IOUtil.fdVal(fd);
} }
/**
* Closes the write end of the pipe if there are no write operation in
* progress and the channel is not registered with a Selector.
*/
private boolean tryClose() throws IOException {
assert Thread.holdsLock(stateLock) && state == ST_CLOSING;
if (thread == 0 && !isRegistered()) {
state = ST_CLOSED;
nd.close(fd);
return true;
} else {
return false;
}
}
/**
* Invokes tryClose to attempt to close the write end of the pipe.
*
* This method is used for deferred closing by I/O and Selector operations.
*/
private void tryFinishClose() {
try {
tryClose();
} catch (IOException ignore) { }
}
/**
* Closes this channel when configured in blocking mode.
*
* If there is a write operation in progress then the write-end of the pipe
* is pre-closed and the writer is signalled, in which case the final close
* is deferred until the writer aborts.
*/
private void implCloseBlockingMode() throws IOException {
synchronized (stateLock) {
assert state < ST_CLOSING;
state = ST_CLOSING;
if (!tryClose()) {
long th = thread;
if (th != 0) {
nd.preClose(fd);
NativeThread.signal(th);
}
}
}
}
/**
* Closes this channel when configured in non-blocking mode.
*
* If the channel is registered with a Selector then the close is deferred
* until the channel is flushed from all Selectors.
*/
private void implCloseNonBlockingMode() throws IOException {
synchronized (stateLock) {
assert state < ST_CLOSING;
state = ST_CLOSING;
}
// wait for any write operation to complete before trying to close
writeLock.lock();
writeLock.unlock();
synchronized (stateLock) {
if (state == ST_CLOSING) {
tryClose();
}
}
}
/** /**
* Invoked by implCloseChannel to close the channel. * Invoked by implCloseChannel to close the channel.
*/ */
@Override @Override
protected void implCloseSelectableChannel() throws IOException { protected void implCloseSelectableChannel() throws IOException {
assert !isOpen(); assert !isOpen();
if (isBlocking()) {
boolean interrupted = false; implCloseBlockingMode();
boolean blocking;
// set state to ST_CLOSING
stateLock.lock();
try {
assert state < ST_CLOSING;
state = ST_CLOSING;
blocking = isBlocking();
} finally {
stateLock.unlock();
}
// wait for any outstanding write to complete
if (blocking) {
stateLock.lock();
try {
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 {
stateCondition.await();
} catch (InterruptedException e) {
interrupted = true;
}
}
}
} finally {
stateLock.unlock();
}
} else { } else {
// non-blocking mode: wait for write to complete implCloseNonBlockingMode();
writeLock.lock();
writeLock.unlock();
} }
// set state to ST_KILLPENDING
stateLock.lock();
try {
assert state == ST_CLOSING;
state = ST_KILLPENDING;
} finally {
stateLock.unlock();
}
// close socket if not registered with Selector
if (!isRegistered())
kill();
// restore interrupt status
if (interrupted)
Thread.currentThread().interrupt();
} }
@Override @Override
public void kill() throws IOException { public void kill() {
stateLock.lock(); synchronized (stateLock) {
try { if (state == ST_CLOSING) {
assert thread == 0; tryFinishClose();
if (state == ST_KILLPENDING) {
state = ST_KILLED;
nd.close(fd);
} }
} finally {
stateLock.unlock();
} }
} }
@ -170,11 +177,10 @@ class SinkChannelImpl
protected void implConfigureBlocking(boolean block) throws IOException { protected void implConfigureBlocking(boolean block) throws IOException {
writeLock.lock(); writeLock.lock();
try { try {
stateLock.lock(); synchronized (stateLock) {
try { if (!isOpen())
throw new ClosedChannelException();
IOUtil.configureBlocking(fd, block); IOUtil.configureBlocking(fd, block);
} finally {
stateLock.unlock();
} }
} finally { } finally {
writeLock.unlock(); writeLock.unlock();
@ -229,14 +235,11 @@ class SinkChannelImpl
// set hook for Thread.interrupt // set hook for Thread.interrupt
begin(); begin();
} }
stateLock.lock(); synchronized (stateLock) {
try {
if (!isOpen()) if (!isOpen())
throw new ClosedChannelException(); throw new ClosedChannelException();
if (blocking) if (blocking)
thread = NativeThread.current(); thread = NativeThread.current();
} finally {
stateLock.unlock();
} }
} }
@ -250,15 +253,11 @@ class SinkChannelImpl
throws AsynchronousCloseException throws AsynchronousCloseException
{ {
if (blocking) { if (blocking) {
stateLock.lock(); synchronized (stateLock) {
try {
thread = 0; thread = 0;
// notify any thread waiting in implCloseSelectableChannel
if (state == ST_CLOSING) { if (state == ST_CLOSING) {
stateCondition.signalAll(); tryFinishClose();
} }
} finally {
stateLock.unlock();
} }
// remove hook for Thread.interrupt // remove hook for Thread.interrupt
end(completed); end(completed);

169
src/java.base/unix/classes/sun/nio/ch/SourceChannelImpl.java
View file

@ -35,7 +35,6 @@ import java.nio.channels.Pipe;
import java.nio.channels.SelectionKey; import java.nio.channels.SelectionKey;
import java.nio.channels.spi.SelectorProvider; import java.nio.channels.spi.SelectorProvider;
import java.util.Objects; import java.util.Objects;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock; import java.util.concurrent.locks.ReentrantLock;
class SourceChannelImpl class SourceChannelImpl
@ -54,16 +53,14 @@ class SourceChannelImpl
// Lock held by any thread that modifies the state fields declared below // Lock held by any thread that modifies the state fields declared below
// DO NOT invoke a blocking I/O operation while holding this lock! // DO NOT invoke a blocking I/O operation while holding this lock!
private final ReentrantLock stateLock = new ReentrantLock(); private final Object stateLock = new Object();
private final Condition stateCondition = stateLock.newCondition();
// -- The following fields are protected by stateLock // -- The following fields are protected by stateLock
// Channel state // Channel state
private static final int ST_INUSE = 0; private static final int ST_INUSE = 0;
private static final int ST_CLOSING = 1; private static final int ST_CLOSING = 1;
private static final int ST_KILLPENDING = 2; private static final int ST_CLOSED = 2;
private static final int ST_KILLED = 3;
private int state; private int state;
// ID of native thread doing read, for signalling // ID of native thread doing read, for signalling
@ -86,83 +83,93 @@ class SourceChannelImpl
this.fdVal = IOUtil.fdVal(fd); this.fdVal = IOUtil.fdVal(fd);
} }
/**
* Closes the read end of the pipe if there are no read operation in
* progress and the channel is not registered with a Selector.
*/
private boolean tryClose() throws IOException {
assert Thread.holdsLock(stateLock) && state == ST_CLOSING;
if (thread == 0 && !isRegistered()) {
state = ST_CLOSED;
nd.close(fd);
return true;
} else {
return false;
}
}
/**
* Invokes tryClose to attempt to close the read end of the pipe.
*
* This method is used for deferred closing by I/O and Selector operations.
*/
private void tryFinishClose() {
try {
tryClose();
} catch (IOException ignore) { }
}
/**
* Closes this channel when configured in blocking mode.
*
* If there is a read operation in progress then the read-end of the pipe
* is pre-closed and the reader is signalled, in which case the final close
* is deferred until the reader aborts.
*/
private void implCloseBlockingMode() throws IOException {
synchronized (stateLock) {
assert state < ST_CLOSING;
state = ST_CLOSING;
if (!tryClose()) {
long th = thread;
if (th != 0) {
nd.preClose(fd);
NativeThread.signal(th);
}
}
}
}
/**
* Closes this channel when configured in non-blocking mode.
*
* If the channel is registered with a Selector then the close is deferred
* until the channel is flushed from all Selectors.
*/
private void implCloseNonBlockingMode() throws IOException {
synchronized (stateLock) {
assert state < ST_CLOSING;
state = ST_CLOSING;
}
// wait for any read operation to complete before trying to close
readLock.lock();
readLock.unlock();
synchronized (stateLock) {
if (state == ST_CLOSING) {
tryClose();
}
}
}
/** /**
* Invoked by implCloseChannel to close the channel. * Invoked by implCloseChannel to close the channel.
*/ */
@Override @Override
protected void implCloseSelectableChannel() throws IOException { protected void implCloseSelectableChannel() throws IOException {
assert !isOpen(); assert !isOpen();
if (isBlocking()) {
boolean interrupted = false; implCloseBlockingMode();
boolean blocking;
// set state to ST_CLOSING
stateLock.lock();
try {
assert state < ST_CLOSING;
state = ST_CLOSING;
blocking = isBlocking();
} finally {
stateLock.unlock();
}
// wait for any outstanding read to complete
if (blocking) {
stateLock.lock();
try {
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 {
stateCondition.await();
} catch (InterruptedException e) {
interrupted = true;
}
}
}
} finally {
stateLock.unlock();
}
} else { } else {
// non-blocking mode: wait for read to complete implCloseNonBlockingMode();
readLock.lock();
readLock.unlock();
} }
// set state to ST_KILLPENDING
stateLock.lock();
try {
assert state == ST_CLOSING;
state = ST_KILLPENDING;
} finally {
stateLock.unlock();
}
// close socket if not registered with Selector
if (!isRegistered())
kill();
// restore interrupt status
if (interrupted)
Thread.currentThread().interrupt();
} }
@Override @Override
public void kill() throws IOException { public void kill() {
stateLock.lock(); synchronized (stateLock) {
try { assert !isOpen();
assert thread == 0; if (state == ST_CLOSING) {
if (state == ST_KILLPENDING) { tryFinishClose();
state = ST_KILLED;
nd.close(fd);
} }
} finally {
stateLock.unlock();
} }
} }
@ -170,11 +177,10 @@ class SourceChannelImpl
protected void implConfigureBlocking(boolean block) throws IOException { protected void implConfigureBlocking(boolean block) throws IOException {
readLock.lock(); readLock.lock();
try { try {
stateLock.lock(); synchronized (stateLock) {
try { if (!isOpen())
throw new ClosedChannelException();
IOUtil.configureBlocking(fd, block); IOUtil.configureBlocking(fd, block);
} finally {
stateLock.unlock();
} }
} finally { } finally {
readLock.unlock(); readLock.unlock();
@ -229,14 +235,11 @@ class SourceChannelImpl
// set hook for Thread.interrupt // set hook for Thread.interrupt
begin(); begin();
} }
stateLock.lock(); synchronized (stateLock) {
try {
if (!isOpen()) if (!isOpen())
throw new ClosedChannelException(); throw new ClosedChannelException();
if (blocking) if (blocking)
thread = NativeThread.current(); thread = NativeThread.current();
} finally {
stateLock.unlock();
} }
} }
@ -250,15 +253,11 @@ class SourceChannelImpl
throws AsynchronousCloseException throws AsynchronousCloseException
{ {
if (blocking) { if (blocking) {
stateLock.lock(); synchronized (stateLock) {
try {
thread = 0; thread = 0;
// notify any thread waiting in implCloseSelectableChannel
if (state == ST_CLOSING) { if (state == ST_CLOSING) {
stateCondition.signalAll(); tryFinishClose();
} }
} finally {
stateLock.unlock();
} }
// remove hook for Thread.interrupt // remove hook for Thread.interrupt
end(completed); end(completed);