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

equal deleted inserted replaced

-:b43cc3b9ef40
+:13b67c1420b8
 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.Condition;
 import java.util.concurrent.locks.ReentrantLock;
 import sun.net.ResourceManager;
 import sun.net.ext.ExtendedSocketOptions;
 // Lock held by current writing or connecting thread
 private final ReentrantLock writeLock = new ReentrantLock();
 // Lock held by any thread that modifies the state fields declared below
 // DO NOT invoke a blocking I/O operation while holding this lock!
-private final Object stateLock = new Object();
+private final ReentrantLock stateLock = new ReentrantLock();
+private final Condition stateCondition = stateLock.newCondition();
 // -- The following fields are protected by stateLock
 // State (does not necessarily increase monotonically)
 private static final int ST_UNCONNECTED = 0;
 this.family = Net.isIPv6Available()
 ? StandardProtocolFamily.INET6
 : StandardProtocolFamily.INET;
 this.fd = fd;
 this.fdVal = IOUtil.fdVal(fd);
-synchronized (stateLock) {
+stateLock.lock();
+try {
 this.localAddress = Net.localAddress(fd);
+} finally {
+stateLock.unlock();
 }
 }
 // @throws ClosedChannelException if channel is closed
 private void ensureOpen() throws ClosedChannelException {
 throw new ClosedChannelException();
 }
 @Override
 public DatagramSocket socket() {
-synchronized (stateLock) {
+stateLock.lock();
+try {
 if (socket == null)
 socket = DatagramSocketAdaptor.create(this);
 return socket;
+} finally {
+stateLock.unlock();
 }
 }
 @Override
 public SocketAddress getLocalAddress() throws IOException {
-synchronized (stateLock) {
+stateLock.lock();
+try {
 ensureOpen();
 // Perform security check before returning address
 return Net.getRevealedLocalAddress(localAddress);
+} finally {
+stateLock.unlock();
 }
 }
 @Override
 public SocketAddress getRemoteAddress() throws IOException {
-synchronized (stateLock) {
+stateLock.lock();
+try {
 ensureOpen();
 return remoteAddress;
+} finally {
+stateLock.unlock();
 }
 }
 @Override
 public <T> DatagramChannel setOption(SocketOption<T> name, T value)
 {
 Objects.requireNonNull(name);
 if (!supportedOptions().contains(name))
 throw new UnsupportedOperationException("'" + name + "' not supported");
-synchronized (stateLock) {
+stateLock.lock();
+try {
 ensureOpen();
 if (name == StandardSocketOptions.IP_TOS ||
 name == StandardSocketOptions.IP_MULTICAST_TTL ||
 name == StandardSocketOptions.IP_MULTICAST_LOOP)
 }
 // remaining options don't need any special handling
 Net.setSocketOption(fd, Net.UNSPEC, name, value);
 return this;
+} finally {
+stateLock.unlock();
 }
 }
 @Override
 @SuppressWarnings("unchecked")
 {
 Objects.requireNonNull(name);
 if (!supportedOptions().contains(name))
 throw new UnsupportedOperationException("'" + name + "' not supported");
-synchronized (stateLock) {
+stateLock.lock();
+try {
 ensureOpen();
 if (name == StandardSocketOptions.IP_TOS ||
 name == StandardSocketOptions.IP_MULTICAST_TTL ||
 name == StandardSocketOptions.IP_MULTICAST_LOOP)
 return (T)Boolean.valueOf(isReuseAddress);
 }
 // no special handling
 return (T) Net.getSocketOption(fd, Net.UNSPEC, name);
+} finally {
+stateLock.unlock();
 }
 }
 private static class DefaultOptionsHolder {
 static final Set<SocketOption<?>> defaultOptions = defaultOptions();
 if (blocking) {
 // set hook for Thread.interrupt
 begin();
 }
 SocketAddress remote;
-synchronized (stateLock) {
+stateLock.lock();
+try {
 ensureOpen();
 remote = remoteAddress;
 if ((remote == null) && mustBeConnected)
 throw new NotYetConnectedException();
 if (localAddress == null)
 bindInternal(null);
 if (blocking)
 readerThread = NativeThread.current();
+} finally {
+stateLock.unlock();
 }
 return remote;
 }
 /**
 */
 private void endRead(boolean blocking, boolean completed)
 throws AsynchronousCloseException
 {
 if (blocking) {
-synchronized (stateLock) {
+stateLock.lock();
+try {
 readerThread = 0;
 // notify any thread waiting in implCloseSelectableChannel
 if (state == ST_CLOSING) {
-stateLock.notifyAll();
+stateCondition.signalAll();
 }
+} finally {
+stateLock.unlock();
 }
 // remove hook for Thread.interrupt
 end(completed);
 }
 }
 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, connected);
-n = receive(fd, dst, connected);
+if (blocking) {
-} while ((n == IOStatus.INTERRUPTED) && isOpen());
+while (IOStatus.okayToRetry(n) && isOpen()) {
-if (n == IOStatus.UNAVAILABLE)
+park(Net.POLLIN);
+n = receive(fd, dst, connected);
+}
+} else if (n == IOStatus.UNAVAILABLE) {
 return null;
+}
 } else {
 // Cannot receive into user's buffer when running with a
 // security manager and not connected
 bb = Util.getTemporaryDirectBuffer(dst.remaining());
 for (;;) {
-do {
+n = receive(fd, bb, connected);
-n = receive(fd, bb, connected);
+if (blocking) {
-} while ((n == IOStatus.INTERRUPTED) && isOpen());
+while (IOStatus.okayToRetry(n) && isOpen()) {
-if (n == IOStatus.UNAVAILABLE)
+park(Net.POLLIN);
+n = receive(fd, bb, connected);
+}
+} else if (n == IOStatus.UNAVAILABLE) {
 return null;
+}
 InetSocketAddress isa = (InetSocketAddress)sender;
 try {
 sm.checkAccept(isa.getAddress().getHostAddress(),
 isa.getPort());
 } catch (SecurityException se) {
 if (n > 0)
 bb.position(pos + n);
 return n;
 }
+@Override
 public int send(ByteBuffer src, SocketAddress target)
 throws IOException
 {
 Objects.requireNonNull(src);
 InetSocketAddress isa = Net.checkAddress(target, family);
 if (remote != null) {
 // connected
 if (!target.equals(remote)) {
 throw new AlreadyConnectedException();
 }
-do {
+n = IOUtil.write(fd, src, -1, nd);
-n = IOUtil.write(fd, src, -1, nd);
+if (blocking) {
-} while ((n == IOStatus.INTERRUPTED) && isOpen());
+while (IOStatus.okayToRetry(n) && isOpen()) {
+park(Net.POLLOUT);
+n = IOUtil.write(fd, src, -1, nd);
+}
+}
 } else {
 // not connected
 SecurityManager sm = System.getSecurityManager();
 if (sm != null) {
 InetAddress ia = isa.getAddress();
 sm.checkMulticast(ia);
 } else {
 sm.checkConnect(ia.getHostAddress(), isa.getPort());
 }
 }
-do {
+n = send(fd, src, isa);
-n = send(fd, src, isa);
+if (blocking) {
-} while ((n == IOStatus.INTERRUPTED) && isOpen());
+while (IOStatus.okayToRetry(n) && isOpen()) {
+park(Net.POLLOUT);
+n = send(fd, src, isa);
+}
+}
 }
 } finally {
 endWrite(blocking, n > 0);
 assert IOStatus.check(n);
 }
 try {
 boolean blocking = isBlocking();
 int n = 0;
 try {
 beginRead(blocking, true);
-do {
+n = IOUtil.read(fd, buf, -1, nd);
-n = IOUtil.read(fd, buf, -1, nd);
+if (blocking) {
-} while ((n == IOStatus.INTERRUPTED) && isOpen());
+while (IOStatus.okayToRetry(n) && isOpen()) {
+park(Net.POLLIN);
+n = IOUtil.read(fd, buf, -1, nd);
+}
+}
 } finally {
 endRead(blocking, n > 0);
 assert IOStatus.check(n);
 }
 return IOStatus.normalize(n);
 try {
 boolean blocking = isBlocking();
 long n = 0;
 try {
 beginRead(blocking, true);
-do {
+n = IOUtil.read(fd, dsts, offset, length, nd);
-n = IOUtil.read(fd, dsts, offset, length, nd);
+if (blocking) {
-} while ((n == IOStatus.INTERRUPTED) && isOpen());
+while (IOStatus.okayToRetry(n) && isOpen()) {
+park(Net.POLLIN);
+n = IOUtil.read(fd, dsts, offset, length, nd);
+}
+}
 } finally {
 endRead(blocking, n > 0);
 assert IOStatus.check(n);
 }
 return IOStatus.normalize(n);
 if (blocking) {
 // set hook for Thread.interrupt
 begin();
 }
 SocketAddress remote;
-synchronized (stateLock) {
+stateLock.lock();
+try {
 ensureOpen();
 remote = remoteAddress;
 if ((remote == null) && mustBeConnected)
 throw new NotYetConnectedException();
 if (localAddress == null)
 bindInternal(null);
 if (blocking)
 writerThread = NativeThread.current();
+} finally {
+stateLock.unlock();
 }
 return remote;
 }
 /**
 */
 private void endWrite(boolean blocking, boolean completed)
 throws AsynchronousCloseException
 {
 if (blocking) {
-synchronized (stateLock) {
+stateLock.lock();
+try {
 writerThread = 0;
 // notify any thread waiting in implCloseSelectableChannel
 if (state == ST_CLOSING) {
-stateLock.notifyAll();
+stateCondition.signalAll();
 }
+} finally {
+stateLock.unlock();
 }
 // remove hook for Thread.interrupt
 end(completed);
 }
 }
 try {
 boolean blocking = isBlocking();
 int n = 0;
 try {
 beginWrite(blocking, true);
-do {
+n = IOUtil.write(fd, buf, -1, nd);
-n = IOUtil.write(fd, buf, -1, nd);
+if (blocking) {
-} while ((n == IOStatus.INTERRUPTED) && isOpen());
+while (IOStatus.okayToRetry(n) && isOpen()) {
+park(Net.POLLOUT);
+n = IOUtil.write(fd, buf, -1, nd);
+}
+}
 } finally {
 endWrite(blocking, n > 0);
 assert IOStatus.check(n);
 }
 return IOStatus.normalize(n);
 try {
 boolean blocking = isBlocking();
 long n = 0;
 try {
 beginWrite(blocking, true);
-do {
+n = IOUtil.write(fd, srcs, offset, length, nd);
-n = IOUtil.write(fd, srcs, offset, length, nd);
+if (blocking) {
-} while ((n == IOStatus.INTERRUPTED) && isOpen());
+while (IOStatus.okayToRetry(n) && isOpen()) {
+park(Net.POLLOUT);
+n = IOUtil.write(fd, srcs, offset, length, nd);
+}
+}
 } finally {
 endWrite(blocking, n > 0);
 assert IOStatus.check(n);
 }
 return IOStatus.normalize(n);
 protected void implConfigureBlocking(boolean block) throws IOException {
 readLock.lock();
 try {
 writeLock.lock();
 try {
-synchronized (stateLock) {
+stateLock.lock();
+try {
 ensureOpen();
 IOUtil.configureBlocking(fd, block);
+} finally {
+stateLock.unlock();
 }
 } finally {
 writeLock.unlock();
 }
 } finally {
 readLock.unlock();
 }
 }
 InetSocketAddress localAddress() {
-synchronized (stateLock) {
+stateLock.lock();
+try {
 return localAddress;
+} finally {
+stateLock.unlock();
 }
 }
 InetSocketAddress remoteAddress() {
-synchronized (stateLock) {
+stateLock.lock();
+try {
 return remoteAddress;
+} finally {
+stateLock.unlock();
 }
 }
 @Override
 public DatagramChannel bind(SocketAddress local) throws IOException {
 readLock.lock();
 try {
 writeLock.lock();
 try {
-synchronized (stateLock) {
+stateLock.lock();
+try {
 ensureOpen();
 if (localAddress != null)
 throw new AlreadyBoundException();
 bindInternal(local);
+} finally {
+stateLock.unlock();
 }
 } finally {
 writeLock.unlock();
 }
 } finally {
 }
 return this;
 }
 private void bindInternal(SocketAddress local) throws IOException {
-assert Thread.holdsLock(stateLock) && (localAddress == null);
+assert stateLock.isHeldByCurrentThread() && (localAddress == null);
 InetSocketAddress isa;
 if (local == null) {
 // only Inet4Address allowed with IPv4 socket
 if (family == StandardProtocolFamily.INET) {
 localAddress = Net.localAddress(fd);
 }
 @Override
 public boolean isConnected() {
-synchronized (stateLock) {
+stateLock.lock();
+try {
 return (state == ST_CONNECTED);
+} finally {
+stateLock.unlock();
 }
 }
 @Override
 public DatagramChannel connect(SocketAddress sa) throws IOException {
 readLock.lock();
 try {
 writeLock.lock();
 try {
-synchronized (stateLock) {
+stateLock.lock();
+try {
 ensureOpen();
 if (state == ST_CONNECTED)
 throw new AlreadyConnectedException();
 int n = Net.connect(family,
 if (blocking) {
 IOUtil.configureBlocking(fd, false);
 }
 try {
 ByteBuffer buf = ByteBuffer.allocate(100);
-while (receive(buf) != null) {
+while (receive(fd, buf, false) > 0) {
 buf.clear();
 }
 } finally {
 if (blocking) {
 IOUtil.configureBlocking(fd, true);
 }
 }
+} finally {
+stateLock.unlock();
 }
 } finally {
 writeLock.unlock();
 }
 } finally {
 public DatagramChannel disconnect() throws IOException {
 readLock.lock();
 try {
 writeLock.lock();
 try {
-synchronized (stateLock) {
+stateLock.lock();
+try {
 if (!isOpen() || (state != ST_CONNECTED))
 return this;
 // disconnect socket
 boolean isIPv6 = (family == StandardProtocolFamily.INET6);
 remoteAddress = null;
 state = ST_UNCONNECTED;
 // refresh local address
 localAddress = Net.localAddress(fd);
+} finally {
+stateLock.unlock();
 }
 } finally {
 writeLock.unlock();
 }
 } finally {
 SecurityManager sm = System.getSecurityManager();
 if (sm != null)
 sm.checkMulticast(group);
-synchronized (stateLock) {
+stateLock.lock();
+try {
 ensureOpen();
 // check the registry to see if we are already a member of the group
 if (registry == null) {
 registry = new MembershipRegistry();
 groupAddress, targetAddress, sourceAddress);
 }
 registry.add(key);
 return key;
+} finally {
+stateLock.unlock();
 }
 }
 @Override
 public MembershipKey join(InetAddress group,
 // package-private
 void drop(MembershipKeyImpl key) {
 assert key.channel() == this;
-synchronized (stateLock) {
+stateLock.lock();
+try {
 if (!key.isValid())
 return;
 try {
 if (key instanceof MembershipKeyImpl.Type6) {
 throw new AssertionError(ioe);
 }
 key.invalidate();
 registry.remove(key);
+} finally {
+stateLock.unlock();
 }
 }
 /**
 * Block datagrams from given source if a memory to receive all
 throws IOException
 {
 assert key.channel() == this;
 assert key.sourceAddress() == null;
-synchronized (stateLock) {
+stateLock.lock();
+try {
 if (!key.isValid())
 throw new IllegalStateException("key is no longer valid");
 if (source.isAnyLocalAddress())
 throw new IllegalArgumentException("Source address is a wildcard address");
 if (source.isMulticastAddress())
 }
 if (n == IOStatus.UNAVAILABLE) {
 // ancient kernel
 throw new UnsupportedOperationException();
 }
+} finally {
+stateLock.unlock();
 }
 }
 /**
 * Unblock given source.
 */
 void unblock(MembershipKeyImpl key, InetAddress source) {
 assert key.channel() == this;
 assert key.sourceAddress() == null;
-synchronized (stateLock) {
+stateLock.lock();
+try {
 if (!key.isValid())
 throw new IllegalStateException("key is no longer valid");
 try {
 if (key instanceof MembershipKeyImpl.Type6) {
 }
 } catch (IOException ioe) {
 // should not happen
 throw new AssertionError(ioe);
 }
+} finally {
+stateLock.unlock();
 }
 }
 /**
 * Invoked by implCloseChannel to close the channel.
 boolean blocking;
 boolean interrupted = false;
 // set state to ST_CLOSING and invalid membership keys
-synchronized (stateLock) {
+stateLock.lock();
+try {
 assert state < ST_CLOSING;
 blocking = isBlocking();
 state = ST_CLOSING;
 // if member of any multicast groups then invalidate the keys
 if (registry != null)
 registry.invalidateAll();
+} finally {
+stateLock.unlock();
 }
 // wait for any outstanding I/O operations to complete
 if (blocking) {
-synchronized (stateLock) {
+stateLock.lock();
+try {
 assert state == ST_CLOSING;
 long reader = readerThread;
 long writer = writerThread;
 if (reader != 0 || writer != 0) {
 nd.preClose(fd);
 NativeThread.signal(writer);
 // wait for blocking I/O operations to end
 while (readerThread != 0 || writerThread != 0) {
 try {
-stateLock.wait();
+stateCondition.await();
 } catch (InterruptedException e) {
 interrupted = true;
 }
 }
 }
+} finally {
+stateLock.unlock();
 }
 } else {
 // non-blocking mode: wait for read/write to complete
 readLock.lock();
 try {
 readLock.unlock();
 }
 }
 // set state to ST_KILLPENDING
-synchronized (stateLock) {
+stateLock.lock();
+try {
 assert state == ST_CLOSING;
 state = ST_KILLPENDING;
+} finally {
+stateLock.unlock();
 }
 // close socket if not registered with Selector
 if (!isRegistered())
 kill();
 Thread.currentThread().interrupt();
 }
 @Override
 public void kill() throws IOException {
-synchronized (stateLock) {
+stateLock.lock();
+try {
 if (state == ST_KILLPENDING) {
 state = ST_KILLED;
 try {
 nd.close(fd);
 } finally {
 // notify resource manager
 ResourceManager.afterUdpClose();
 }
 }
+} finally {
+stateLock.unlock();
 }
 }
 @SuppressWarnings("deprecation")
 protected void finalize() throws IOException {

changeset 54620	13b67c1420b8
parent 53419	eac105e3ec13
child 54754	193a8f1a4f3b