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

equal deleted inserted replaced

-:b43cc3b9ef40
+:13b67c1420b8
 import java.net.ProtocolFamily;
 import java.net.Socket;
 import java.net.SocketAddress;
 import java.net.SocketException;
 import java.net.SocketOption;
+import java.net.SocketTimeoutException;
 import java.net.StandardProtocolFamily;
 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.ConnectionPendingException;
+import java.nio.channels.IllegalBlockingModeException;
 import java.nio.channels.NoConnectionPendingException;
 import java.nio.channels.NotYetConnectedException;
 import java.nio.channels.SelectionKey;
 import java.nio.channels.SocketChannel;
 import java.nio.channels.spi.SelectorProvider;
 import java.util.Collections;
 import java.util.HashSet;
 import java.util.Objects;
 import java.util.Set;
+import java.util.concurrent.locks.Condition;
 import java.util.concurrent.locks.ReentrantLock;
 import sun.net.ConnectionResetException;
 import sun.net.NetHooks;
 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();
 // Input/Output closed
 private volatile boolean isInputClosed;
 private volatile boolean isOutputClosed;
 {
 super(sp);
 this.fd = fd;
 this.fdVal = IOUtil.fdVal(fd);
 if (bound) {
-synchronized (stateLock) {
+stateLock.lock();
+try {
 this.localAddress = Net.localAddress(fd);
+} finally {
+stateLock.unlock();
 }
 }
 }
 // Constructor for sockets obtained from server sockets
 throws IOException
 {
 super(sp);
 this.fd = fd;
 this.fdVal = IOUtil.fdVal(fd);
-synchronized (stateLock) {
+stateLock.lock();
+try {
 this.localAddress = Net.localAddress(fd);
 this.remoteAddress = isa;
 this.state = ST_CONNECTED;
+} finally {
+stateLock.unlock();
 }
 }
 /**
 * Checks that the channel is open.
 }
 }
 @Override
 public Socket socket() {
-synchronized (stateLock) {
+stateLock.lock();
+try {
 if (socket == null)
 socket = SocketAdaptor.create(this);
 return socket;
+} finally {
+stateLock.unlock();
 }
 }
 @Override
 public SocketAddress getLocalAddress() throws IOException {
-synchronized (stateLock) {
+stateLock.lock();
+try {
 ensureOpen();
 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> SocketChannel 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) {
 ProtocolFamily family = Net.isIPv6Available() ?
 StandardProtocolFamily.INET6 : StandardProtocolFamily.INET;
 }
 // no options that require special handling
 Net.setSocketOption(fd, 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.SO_REUSEADDR && Net.useExclusiveBind()) {
 // SO_REUSEADDR emulated when using exclusive bind
 return (T)Boolean.valueOf(isReuseAddress);
 return (T) Net.getSocketOption(fd, family, name);
 }
 // no options that require special handling
 return (T) Net.getSocketOption(fd, name);
+} finally {
+stateLock.unlock();
 }
 }
 private static class DefaultOptionsHolder {
 static final Set<SocketOption<?>> defaultOptions = defaultOptions();
 private void beginRead(boolean blocking) throws ClosedChannelException {
 if (blocking) {
 // set hook for Thread.interrupt
 begin();
-synchronized (stateLock) {
+stateLock.lock();
+try {
 ensureOpenAndConnected();
 // record thread so it can be signalled if needed
 readerThread = NativeThread.current();
+} finally {
+stateLock.unlock();
 }
 } else {
 ensureOpenAndConnected();
 }
 }
 */
 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);
 }
 }
 // check if input is shutdown
 if (isInputClosed)
 return IOStatus.EOF;
+n = IOUtil.read(fd, buf, -1, nd);
 if (blocking) {
-do {
+while (IOStatus.okayToRetry(n) && isOpen()) {
+park(Net.POLLIN);
 n = IOUtil.read(fd, buf, -1, nd);
-} while (n == IOStatus.INTERRUPTED && isOpen());
+}
-} else {
-n = IOUtil.read(fd, buf, -1, nd);
 }
 } catch (ConnectionResetException e) {
 connectionReset = true;
 throwConnectionReset();
 } finally {
 // check if input is shutdown
 if (isInputClosed)
 return IOStatus.EOF;
+n = IOUtil.read(fd, dsts, offset, length, nd);
 if (blocking) {
-do {
+while (IOStatus.okayToRetry(n) && isOpen()) {
+park(Net.POLLIN);
 n = IOUtil.read(fd, dsts, offset, length, nd);
-} while (n == IOStatus.INTERRUPTED && isOpen());
+}
-} else {
-n = IOUtil.read(fd, dsts, offset, length, nd);
 }
 } catch (ConnectionResetException e) {
 connectionReset = true;
 throwConnectionReset();
 } finally {
 private void beginWrite(boolean blocking) throws ClosedChannelException {
 if (blocking) {
 // set hook for Thread.interrupt
 begin();
-synchronized (stateLock) {
+stateLock.lock();
+try {
 ensureOpenAndConnected();
 if (isOutputClosed)
 throw new ClosedChannelException();
 // record thread so it can be signalled if needed
 writerThread = NativeThread.current();
+} finally {
+stateLock.unlock();
 }
 } else {
 ensureOpenAndConnected();
 }
 }
 */
 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);
+n = IOUtil.write(fd, buf, -1, nd);
 if (blocking) {
-do {
+while (IOStatus.okayToRetry(n) && isOpen()) {
+park(Net.POLLOUT);
 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();
 try {
 boolean blocking = isBlocking();
 long n = 0;
 try {
 beginWrite(blocking);
+n = IOUtil.write(fd, srcs, offset, length, nd);
 if (blocking) {
-do {
+while (IOStatus.okayToRetry(n) && isOpen()) {
+park(Net.POLLOUT);
 n = IOUtil.write(fd, srcs, offset, length, nd);
-} while (n == IOStatus.INTERRUPTED && isOpen());
+}
-} else {
-n = IOUtil.write(fd, srcs, offset, length, nd);
 }
 } finally {
 endWrite(blocking, n > 0);
 if (n <= 0 && isOutputClosed)
 throw new AsynchronousCloseException();
 protected void implConfigureBlocking(boolean block) throws IOException {
 readLock.lock();
 try {
 writeLock.lock();
 try {
-synchronized (stateLock) {
+lockedConfigureBlocking(block);
-ensureOpen();
-IOUtil.configureBlocking(fd, block);
-}
 } finally {
 writeLock.unlock();
 }
 } finally {
 readLock.unlock();
 }
 }
 /**
+* Adjust the blocking mode while holding the readLock or writeLock.
+*/
+private void lockedConfigureBlocking(boolean block) throws IOException {
+assert readLock.isHeldByCurrentThread() || writeLock.isHeldByCurrentThread();
+stateLock.lock();
+try {
+ensureOpen();
+IOUtil.configureBlocking(fd, block);
+} finally {
+stateLock.unlock();
+}
+}
+/**
 * Returns the local address, or null if not bound
 */
 InetSocketAddress localAddress() {
-synchronized (stateLock) {
+stateLock.lock();
+try {
 return localAddress;
+} finally {
+stateLock.unlock();
 }
 }
 /**
 * Returns the remote address, or null if not connected
 */
 InetSocketAddress remoteAddress() {
-synchronized (stateLock) {
+stateLock.lock();
+try {
 return remoteAddress;
+} finally {
+stateLock.unlock();
 }
 }
 @Override
 public SocketChannel bind(SocketAddress local) throws IOException {
 readLock.lock();
 try {
 writeLock.lock();
 try {
-synchronized (stateLock) {
+stateLock.lock();
+try {
 ensureOpen();
 if (state == ST_CONNECTIONPENDING)
 throw new ConnectionPendingException();
 if (localAddress != null)
 throw new AlreadyBoundException();
 sm.checkListen(isa.getPort());
 }
 NetHooks.beforeTcpBind(fd, isa.getAddress(), isa.getPort());
 Net.bind(fd, isa.getAddress(), isa.getPort());
 localAddress = Net.localAddress(fd);
+} finally {
+stateLock.unlock();
 }
 } finally {
 writeLock.unlock();
 }
 } finally {
 {
 if (blocking) {
 // set hook for Thread.interrupt
 begin();
 }
-synchronized (stateLock) {
+stateLock.lock();
+try {
 ensureOpen();
 int state = this.state;
 if (state == ST_CONNECTED)
 throw new AlreadyConnectedException();
 if (state == ST_CONNECTIONPENDING)
 if (blocking) {
 // record thread so it can be signalled if needed
 readerThread = NativeThread.current();
 }
+} finally {
+stateLock.unlock();
 }
 }
 /**
 * Marks the end of a connect operation that may have blocked.
 throws IOException
 {
 endRead(blocking, completed);
 if (completed) {
-synchronized (stateLock) {
+stateLock.lock();
+try {
 if (state == ST_CONNECTIONPENDING) {
 localAddress = Net.localAddress(fd);
 state = ST_CONNECTED;
 }
-}
+} finally {
-}
+stateLock.unlock();
 }
+}
-@Override
+}
-public boolean connect(SocketAddress sa) throws IOException {
+/**
+* Checks the remote address to which this channel is to be connected.
+*/
+private InetSocketAddress checkRemote(SocketAddress sa) throws IOException {
 InetSocketAddress isa = Net.checkAddress(sa);
 SecurityManager sm = System.getSecurityManager();
-if (sm != null)
+if (sm != null) {
 sm.checkConnect(isa.getAddress().getHostAddress(), isa.getPort());
+}
-InetAddress ia = isa.getAddress();
+if (isa.getAddress().isAnyLocalAddress()) {
-if (ia.isAnyLocalAddress())
+return new InetSocketAddress(InetAddress.getLocalHost(), isa.getPort());
-ia = InetAddress.getLocalHost();
+} else {
+return isa;
+}
+}
+@Override
+public boolean connect(SocketAddress remote) throws IOException {
+InetSocketAddress isa = checkRemote(remote);
 try {
 readLock.lock();
 try {
 writeLock.lock();
 try {
-int n = 0;
 boolean blocking = isBlocking();
+boolean connected = false;
 try {
 beginConnect(blocking, isa);
-do {
+int n = Net.connect(fd, isa.getAddress(), isa.getPort());
-n = Net.connect(fd, ia, isa.getPort());
+if (n > 0) {
-} while (n == IOStatus.INTERRUPTED && isOpen());
+connected = true;
+} else if (blocking) {
+assert IOStatus.okayToRetry(n);
+boolean polled = false;
+while (!polled && isOpen()) {
+park(Net.POLLOUT);
+polled = Net.pollConnectNow(fd);
+}
+connected = polled && isOpen();
+}
 } finally {
-endConnect(blocking, (n > 0));
+endConnect(blocking, connected);
 }
-assert IOStatus.check(n);
+return connected;
-return n > 0;
 } finally {
 writeLock.unlock();
 }
 } finally {
 readLock.unlock();
 private void beginFinishConnect(boolean blocking) throws ClosedChannelException {
 if (blocking) {
 // set hook for Thread.interrupt
 begin();
 }
-synchronized (stateLock) {
+stateLock.lock();
+try {
 ensureOpen();
 if (state != ST_CONNECTIONPENDING)
 throw new NoConnectionPendingException();
 if (blocking) {
 // record thread so it can be signalled if needed
 readerThread = NativeThread.current();
 }
+} finally {
+stateLock.unlock();
 }
 }
 /**
 * Marks the end of a finishConnect operation that may have blocked.
 throws IOException
 {
 endRead(blocking, completed);
 if (completed) {
-synchronized (stateLock) {
+stateLock.lock();
+try {
 if (state == ST_CONNECTIONPENDING) {
 localAddress = Net.localAddress(fd);
 state = ST_CONNECTED;
 }
+} finally {
+stateLock.unlock();
 }
 }
 }
 @Override
 boolean blocking = isBlocking();
 boolean connected = false;
 try {
 beginFinishConnect(blocking);
+boolean polled = Net.pollConnectNow(fd);
 if (blocking) {
-do {
+while (!polled && isOpen()) {
-connected = Net.pollConnect(fd, -1);
+park(Net.POLLOUT);
-} while (!connected && isOpen());
+polled = Net.pollConnectNow(fd);
-} else {
+}
-connected = Net.pollConnect(fd, 0);
 }
+connected = polled && isOpen();
 } finally {
 endFinishConnect(blocking, connected);
 }
 assert (blocking && connected) ^ !blocking;
 return connected;
 boolean blocking;
 boolean connected;
 boolean interrupted = false;
 // set state to ST_CLOSING
-synchronized (stateLock) {
+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) {
-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;
 // if connected and the channel is registered with a Selector then
 // 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
 // be disabled so that the Selector does not wait when it closes
 Net.shutdown(fd, Net.SHUT_WR);
 }
 } catch (IOException ignore) { }
 }
 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;
 nd.close(fd);
 }
+} finally {
+stateLock.unlock();
 }
 }
 @Override
 public SocketChannel shutdownInput() throws IOException {
-synchronized (stateLock) {
+stateLock.lock();
+try {
 ensureOpen();
 if (!isConnected())
 throw new NotYetConnectedException();
 if (!isInputClosed) {
 Net.shutdown(fd, Net.SHUT_RD);
 if (thread != 0)
 NativeThread.signal(thread);
 isInputClosed = true;
 }
 return this;
+} finally {
+stateLock.unlock();
 }
 }
 @Override
 public SocketChannel shutdownOutput() throws IOException {
-synchronized (stateLock) {
+stateLock.lock();
+try {
 ensureOpen();
 if (!isConnected())
 throw new NotYetConnectedException();
 if (!isOutputClosed) {
 Net.shutdown(fd, Net.SHUT_WR);
 if (thread != 0)
 NativeThread.signal(thread);
 isOutputClosed = true;
 }
 return this;
+} finally {
+stateLock.unlock();
 }
 }
 boolean isInputOpen() {
 return !isInputClosed;
 boolean isOutputOpen() {
 return !isOutputClosed;
 }
 /**
-* Poll this channel's socket for reading up to the given timeout.
+* Waits for a connection attempt to finish with a timeout
-* @return {@code true} if the socket is polled
+* @throws SocketTimeoutException if the connect timeout elapses
 */
-boolean pollRead(long timeout) throws IOException {
+private boolean finishTimedConnect(long nanos) throws IOException {
-boolean blocking = isBlocking();
+long startNanos = System.nanoTime();
-assert Thread.holdsLock(blockingLock()) && blocking;
+boolean polled = Net.pollConnectNow(fd);
+while (!polled && isOpen()) {
+long remainingNanos = nanos - (System.nanoTime() - startNanos);
+if (remainingNanos <= 0) {
+throw new SocketTimeoutException("Connect timed out");
+}
+park(Net.POLLOUT, remainingNanos);
+polled = Net.pollConnectNow(fd);
+}
+return polled && isOpen();
+}
+/**
+* Attempts to establish a connection to the given socket address with a
+* timeout. Closes the socket if connection cannot be established.
+*
+* @apiNote This method is for use by the socket adaptor.
+*
+* @throws IllegalBlockingModeException if the channel is non-blocking
+* @throws SocketTimeoutException if the read timeout elapses
+*/
+void blockingConnect(SocketAddress remote, long nanos) throws IOException {
+InetSocketAddress isa = checkRemote(remote);
+try {
+readLock.lock();
+try {
+writeLock.lock();
+try {
+if (!isBlocking())
+throw new IllegalBlockingModeException();
+boolean connected = false;
+try {
+beginConnect(true, isa);
+// change socket to non-blocking
+lockedConfigureBlocking(false);
+try {
+int n = Net.connect(fd, isa.getAddress(), isa.getPort());
+connected = (n > 0) ? true : finishTimedConnect(nanos);
+} finally {
+// restore socket to blocking mode
+lockedConfigureBlocking(true);
+}
+} finally {
+endConnect(true, connected);
+}
+} finally {
+writeLock.unlock();
+}
+} finally {
+readLock.unlock();
+}
+} catch (IOException ioe) {
+// connect failed, close the channel
+close();
+throw SocketExceptions.of(ioe, isa);
+}
+}
+/**
+* Attempts to read bytes from the socket into the given byte array.
+*/
+private int tryRead(byte[] b, int off, int len) throws IOException {
+ByteBuffer dst = Util.getTemporaryDirectBuffer(len);
+assert dst.position() == 0;
+try {
+int n = nd.read(fd, ((DirectBuffer)dst).address(), len);
+if (n > 0) {
+dst.get(b, off, n);
+}
+return n;
+} finally{
+Util.offerFirstTemporaryDirectBuffer(dst);
+}
+}
+/**
+* Reads bytes from the socket into the given byte array with a timeout.
+* @throws SocketTimeoutException if the read timeout elapses
+*/
+private int timedRead(byte[] b, int off, int len, long nanos) throws IOException {
+long startNanos = System.nanoTime();
+int n = tryRead(b, off, len);
+while (n == IOStatus.UNAVAILABLE && isOpen()) {
+long remainingNanos = nanos - (System.nanoTime() - startNanos);
+if (remainingNanos <= 0) {
+throw new SocketTimeoutException("Read timed out");
+}
+park(Net.POLLIN, remainingNanos);
+n = tryRead(b, off, len);
+}
+return n;
+}
+/**
+* Reads bytes from the socket into the given byte array.
+*
+* @apiNote This method is for use by the socket adaptor.
+*
+* @throws IllegalBlockingModeException if the channel is non-blocking
+* @throws SocketTimeoutException if the read timeout elapses
+*/
+int blockingRead(byte[] b, int off, int len, long nanos) throws IOException {
+Objects.checkFromIndexSize(off, len, b.length);
+if (len == 0) {
+// nothing to do
+return 0;
+}
 readLock.lock();
 try {
-boolean polled = false;
+// check that channel is configured blocking
-try {
+if (!isBlocking())
-beginRead(blocking);
+throw new IllegalBlockingModeException();
-int events = Net.poll(fd, Net.POLLIN, timeout);
-polled = (events != 0);
+int n = 0;
-} finally {
+try {
-endRead(blocking, polled);
+beginRead(true);
-}
-return polled;
+// check if connection has been reset
+if (connectionReset)
+throwConnectionReset();
+// check if input is shutdown
+if (isInputClosed)
+return IOStatus.EOF;
+if (nanos > 0) {
+// change socket to non-blocking
+lockedConfigureBlocking(false);
+try {
+n = timedRead(b, off, len, nanos);
+} finally {
+// restore socket to blocking mode
+lockedConfigureBlocking(true);
+}
+} else {
+// read, no timeout
+n = tryRead(b, off, len);
+while (IOStatus.okayToRetry(n) && isOpen()) {
+park(Net.POLLIN);
+n = tryRead(b, off, len);
+}
+}
+} catch (ConnectionResetException e) {
+connectionReset = true;
+throwConnectionReset();
+} finally {
+endRead(true, n > 0);
+if (n <= 0 && isInputClosed)
+return IOStatus.EOF;
+}
+assert n > 0 || n == -1;
+return n;
 } finally {
 readLock.unlock();
 }
 }
 /**
-* Poll this channel's socket for a connection, up to the given timeout.
+* Attempts to write a sequence of bytes to the socket from the given
-* @return {@code true} if the socket is polled
+* byte array.
 */
-boolean pollConnected(long timeout) throws IOException {
+private int tryWrite(byte[] b, int off, int len) throws IOException {
-boolean blocking = isBlocking();
+ByteBuffer src = Util.getTemporaryDirectBuffer(len);
-assert Thread.holdsLock(blockingLock()) && blocking;
+assert src.position() == 0;
+try {
-readLock.lock();
+src.put(b, off, len);
-try {
+return nd.write(fd, ((DirectBuffer)src).address(), len);
-writeLock.lock();
+} finally {
-try {
+Util.offerFirstTemporaryDirectBuffer(src);
-boolean polled = false;
+}
-try {
+}
-beginFinishConnect(blocking);
-int events = Net.poll(fd, Net.POLLCONN, timeout);
+/**
-polled = (events != 0);
+* Writes a sequence of bytes to the socket from the given byte array.
-} finally {
+*
-// invoke endFinishConnect with completed = false so that
+* @apiNote This method is for use by the socket adaptor.
-// the state is not changed to ST_CONNECTED. The socket
+*/
-// adaptor will use finishConnect to finish.
+void blockingWriteFully(byte[] b, int off, int len) throws IOException {
-endFinishConnect(blocking, /*completed*/false);
+Objects.checkFromIndexSize(off, len, b.length);
-}
+if (len == 0) {
-return polled;
+// nothing to do
-} finally {
+return;
-writeLock.unlock();
+}
-}
-} finally {
+writeLock.lock();
-readLock.unlock();
+try {
+// check that channel is configured blocking
+if (!isBlocking())
+throw new IllegalBlockingModeException();
+// loop until all bytes have been written
+int pos = off;
+int end = off + len;
+beginWrite(true);
+try {
+while (pos < end && isOpen()) {
+int size = end - pos;
+int n = tryWrite(b, pos, size);
+while (IOStatus.okayToRetry(n) && isOpen()) {
+park(Net.POLLOUT);
+n = tryWrite(b, pos, size);
+}
+if (n > 0) {
+pos += n;
+}
+}
+} finally {
+endWrite(true, pos >= end);
+}
+} finally {
+writeLock.unlock();
 }
 }
 /**
 * Return the number of bytes in the socket input buffer.
 */
 int available() throws IOException {
-synchronized (stateLock) {
+stateLock.lock();
+try {
 ensureOpenAndConnected();
 if (isInputClosed) {
 return 0;
 } else {
 return Net.available(fd);
 }
+} finally {
+stateLock.unlock();
 }
 }
 /**
 * Translates native poll revent ops into a ready operation ops
 sb.append(this.getClass().getSuperclass().getName());
 sb.append('[');
 if (!isOpen())
 sb.append("closed");
 else {
-synchronized (stateLock) {
+stateLock.lock();
+try {
 switch (state) {
 case ST_UNCONNECTED:
 sb.append("unconnected");
 break;
 case ST_CONNECTIONPENDING:
 }
 if (remoteAddress() != null) {
 sb.append(" remote=");
 sb.append(remoteAddress().toString());
 }
+} finally {
+stateLock.unlock();
 }
 }
 sb.append(']');
 return sb.toString();
 }

changeset 54620	13b67c1420b8
parent 54246	f04e3492fd88
child 54754	193a8f1a4f3b
child 57339	40fdbdd92617