jdk-sandbox: comparison src/java.base/unix/classes/sun/nio/ch/SourceChannelImpl.java

equal deleted inserted replaced

-:a406a9c451a0
+:ce06058197a4
 package sun.nio.ch;
 import java.io.FileDescriptor;
 import java.io.IOException;
 import java.nio.ByteBuffer;
+import java.nio.channels.AsynchronousCloseException;
 import java.nio.channels.ClosedChannelException;
+import java.nio.channels.NotYetConnectedException;
 import java.nio.channels.Pipe;
 import java.nio.channels.SelectionKey;
 import java.nio.channels.spi.SelectorProvider;
+import java.util.Objects;
 import java.util.concurrent.locks.ReentrantLock;
 class SourceChannelImpl
 extends Pipe.SourceChannel
 implements SelChImpl
 {
 // Used to make native read and write calls
 private static final NativeDispatcher nd = new FileDispatcherImpl();
 // The file descriptor associated with this channel
 private final FileDescriptor fd;
-// fd value needed for dev/poll. This value will remain valid
-// even after the value in the file descriptor object has been set to -1
 private final int fdVal;
-// ID of native thread doing read, for signalling
-private volatile long thread;
 // Lock held by current reading thread
 private final ReentrantLock readLock = new ReentrantLock();
 // Lock held by any thread that modifies the state fields declared below
 private final Object stateLock = new Object();
 // -- The following fields are protected by stateLock
 // Channel state
-private static final int ST_UNINITIALIZED = -1;
 private static final int ST_INUSE = 0;
-private static final int ST_KILLED = 1;
+private static final int ST_CLOSING = 1;
-private volatile int state = ST_UNINITIALIZED;
+private static final int ST_KILLPENDING = 2;
+private static final int ST_KILLED = 3;
+private int state;
+// ID of native thread doing read, for signalling
+private long thread;
 // -- End of fields protected by stateLock
 public FileDescriptor getFD() {
 SourceChannelImpl(SelectorProvider sp, FileDescriptor fd) {
 super(sp);
 this.fd = fd;
 this.fdVal = IOUtil.fdVal(fd);
-this.state = ST_INUSE;
+}
-}
+/**
+* Invoked by implCloseChannel to close the channel.
+*/
+@Override
 protected void implCloseSelectableChannel() throws IOException {
-synchronized (stateLock) {
+assert !isOpen();
-if (state != ST_KILLED)
-nd.preClose(fd);
+boolean interrupted = false;
-long th = thread;
+boolean blocking;
-if (th != 0)
-NativeThread.signal(th);
+// set state to ST_CLOSING
-if (!isRegistered())
+synchronized (stateLock) {
-kill();
+assert state < ST_CLOSING;
-}
+state = ST_CLOSING;
-}
+blocking = isBlocking();
+}
+// wait for any outstanding read to complete
+if (blocking) {
+synchronized (stateLock) {
+assert state == ST_CLOSING;
+long th = thread;
+if (th != 0) {
+nd.preClose(fd);
+NativeThread.signal(th);
+// wait for read operation to end
+while (thread != 0) {
+try {
+stateLock.wait();
+} catch (InterruptedException e) {
+interrupted = true;
+}
+}
+}
+}
+} else {
+// non-blocking mode: wait for read to complete
+readLock.lock();
+readLock.unlock();
+}
+// set state to ST_KILLPENDING
+synchronized (stateLock) {
+assert state == ST_CLOSING;
+state = ST_KILLPENDING;
+}
+// close socket if not registered with Selector
+if (!isRegistered())
+kill();
+// restore interrupt status
+if (interrupted)
+Thread.currentThread().interrupt();
+}
+@Override
 public void kill() throws IOException {
 synchronized (stateLock) {
-if (state == ST_KILLED)
+assert thread == 0;
-return;
+if (state == ST_KILLPENDING) {
-if (state == ST_UNINITIALIZED) {
 state = ST_KILLED;
-return;
+nd.close(fd);
 }
-assert !isOpen() && !isRegistered();
+}
-nd.close(fd);
+}
-state = ST_KILLED;
-}
+@Override
-}
 protected void implConfigureBlocking(boolean block) throws IOException {
-IOUtil.configureBlocking(fd, block);
+readLock.lock();
+try {
+synchronized (stateLock) {
+IOUtil.configureBlocking(fd, block);
+}
+} finally {
+readLock.unlock();
+}
 }
 public boolean translateReadyOps(int ops, int initialOps,
 SelectionKeyImpl sk) {
 int intOps = sk.nioInterestOps(); // Do this just once, it synchronizes
 if (ops == SelectionKey.OP_READ)
 ops = Net.POLLIN;
 sk.selector.putEventOps(sk, ops);
 }
-private void ensureOpen() throws IOException {
+/**
-if (!isOpen())
+* Marks the beginning of a read operation that might block.
-throw new ClosedChannelException();
+*
-}
+* @throws ClosedChannelException if the channel is closed
+* @throws NotYetConnectedException if the channel is not yet connected
+*/
+private void beginRead(boolean blocking) throws ClosedChannelException {
+if (blocking) {
+// set hook for Thread.interrupt
+begin();
+}
+synchronized (stateLock) {
+if (!isOpen())
+throw new ClosedChannelException();
+if (blocking)
+thread = NativeThread.current();
+}
+}
+/**
+* Marks the end of a read operation that may have blocked.
+*
+* @throws AsynchronousCloseException if the channel was closed due to this
+* thread being interrupted on a blocking read operation.
+*/
+private void endRead(boolean blocking, boolean completed)
+throws AsynchronousCloseException
+{
+if (blocking) {
+synchronized (stateLock) {
+thread = 0;
+// notify any thread waiting in implCloseSelectableChannel
+if (state == ST_CLOSING) {
+stateLock.notifyAll();
+}
+}
+// remove hook for Thread.interrupt
+end(completed);
+}
+}
+@Override
 public int read(ByteBuffer dst) throws IOException {
+Objects.requireNonNull(dst);
 readLock.lock();
 try {
-ensureOpen();
+boolean blocking = isBlocking();
 int n = 0;
 try {
-begin();
+beginRead(blocking);
-if (!isOpen())
-return 0;
-thread = NativeThread.current();
 do {
 n = IOUtil.read(fd, dst, -1, nd);
 } while ((n == IOStatus.INTERRUPTED) && isOpen());
-return IOStatus.normalize(n);
 } finally {
-thread = 0;
+endRead(blocking, n > 0);
-end((n > 0) || (n == IOStatus.UNAVAILABLE));
 assert IOStatus.check(n);
 }
+return IOStatus.normalize(n);
 } finally {
 readLock.unlock();
 }
 }
-public long read(ByteBuffer[] dsts, int offset, int length)
+@Override
-throws IOException
+public long read(ByteBuffer[] dsts, int offset, int length) throws IOException {
-{
+Objects.checkFromIndexSize(offset, length, dsts.length);
-if ((offset < 0) || (length < 0) || (offset > dsts.length - length))
-throw new IndexOutOfBoundsException();
-return read(Util.subsequence(dsts, offset, length));
-}
-public long read(ByteBuffer[] dsts) throws IOException {
-if (dsts == null)
-throw new NullPointerException();
 readLock.lock();
 try {
-ensureOpen();
+boolean blocking = isBlocking();
 long n = 0;
 try {
-begin();
+beginRead(blocking);
-if (!isOpen())
-return 0;
-thread = NativeThread.current();
 do {
-n = IOUtil.read(fd, dsts, nd);
+n = IOUtil.read(fd, dsts, offset, length, nd);
 } while ((n == IOStatus.INTERRUPTED) && isOpen());
-return IOStatus.normalize(n);
 } finally {
-thread = 0;
+endRead(blocking, n > 0);
-end((n > 0) || (n == IOStatus.UNAVAILABLE));
 assert IOStatus.check(n);
 }
+return IOStatus.normalize(n);
 } finally {
 readLock.unlock();
 }
 }
+@Override
+public long read(ByteBuffer[] dsts) throws IOException {
+return read(dsts, 0, dsts.length);
+}
 }

changeset 49001	ce06058197a4
parent 48750	ffbb784a8873
child 49493	814bd31f8da0