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

equal deleted inserted replaced

-:f14852315495
+:1d3139252c1c
 *
 * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
 * or visit www.oracle.com if you need additional information or have any
 * questions.
 */
 package sun.nio.ch;
 import java.io.IOException;
-import java.nio.channels.*;
+import java.nio.channels.ClosedSelectorException;
-import java.nio.channels.spi.*;
+import java.nio.channels.SelectableChannel;
-import java.util.*;
+import java.nio.channels.SelectionKey;
+import java.nio.channels.Selector;
+import java.nio.channels.spi.SelectorProvider;
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Deque;
+import java.util.Iterator;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import jdk.internal.misc.Unsafe;
 /**
-* An implementation of Selector for Solaris.
+* Selector implementation based on poll
 */
-class PollSelectorImpl
+class PollSelectorImpl extends SelectorImpl {
-extends AbstractPollSelectorImpl
-{
+// initial capacity of poll array
+private static final int INITIAL_CAPACITY = 16;
-// File descriptors used for interrupt
-private int fd0;
+// poll array, grows as needed
-private int fd1;
+private int pollArrayCapacity = INITIAL_CAPACITY;
+private int pollArraySize;
-// Lock for interrupt triggering and clearing
+private AllocatedNativeObject pollArray;
-private Object interruptLock = new Object();
-private boolean interruptTriggered = false;
+// file descriptors used for interrupt
+private final int fd0;
-/**
+private final int fd1;
-* Package private constructor called by factory method in
-* the abstract superclass Selector.
+// keys for file descriptors in poll array, synchronize on selector
-*/
+private final List<SelectionKeyImpl> pollKeys = new ArrayList<>();
+// pending updates, queued by putEventOps
+private final Object updateLock = new Object();
+private final Deque<SelectionKeyImpl> updateKeys = new ArrayDeque<>();
+private final Deque<Integer> updateOps = new ArrayDeque<>();
+// interrupt triggering and clearing
+private final Object interruptLock = new Object();
+private boolean interruptTriggered;
 PollSelectorImpl(SelectorProvider sp) throws IOException {
-super(sp, 1, 1);
+super(sp);
-long pipeFds = IOUtil.makePipe(false);
-fd0 = (int) (pipeFds >>> 32);
+int size = pollArrayCapacity * SIZE_POLLFD;
-fd1 = (int) pipeFds;
+this.pollArray = new AllocatedNativeObject(size, false);
 try {
-pollWrapper = new PollArrayWrapper(INIT_CAP);
+long fds = IOUtil.makePipe(false);
-pollWrapper.initInterrupt(fd0, fd1);
+this.fd0 = (int) (fds >>> 32);
-channelArray = new SelectionKeyImpl[INIT_CAP];
+this.fd1 = (int) fds;
-} catch (Throwable t) {
+} catch (IOException ioe) {
-try {
+pollArray.free();
-FileDispatcherImpl.closeIntFD(fd0);
+throw ioe;
-} catch (IOException ioe0) {
+}
-t.addSuppressed(ioe0);
-}
+// wakeup support
-try {
+synchronized (this) {
-FileDispatcherImpl.closeIntFD(fd1);
+setFirst(fd0, Net.POLLIN);
-} catch (IOException ioe1) {
+}
-t.addSuppressed(ioe1);
+}
-}
-throw t;
+private void ensureOpen() {
-}
+if (!isOpen())
-}
-protected int doSelect(long timeout)
-throws IOException
-{
-if (channelArray == null)
 throw new ClosedSelectorException();
+}
+@Override
+protected int doSelect(long timeout) throws IOException {
+assert Thread.holdsLock(this);
+processUpdateQueue();
 processDeregisterQueue();
 try {
 begin();
-pollWrapper.poll(totalChannels, 0, timeout);
+int to = (int) Math.min(timeout, Integer.MAX_VALUE); // max poll timeout
+boolean timedPoll = (to > 0);
+int numPolled;
+do {
+long startTime = timedPoll ? System.nanoTime() : 0;
+numPolled = poll(pollArray.address(), pollArraySize, to);
+if (numPolled == IOStatus.INTERRUPTED && timedPoll) {
+// timed poll interrupted so need to adjust timeout
+long adjust = System.nanoTime() - startTime;
+to -= TimeUnit.MILLISECONDS.convert(adjust, TimeUnit.NANOSECONDS);
+if (to <= 0) {
+// timeout expired so no retry
+numPolled = 0;
+}
+}
+} while (numPolled == IOStatus.INTERRUPTED);
+assert numPolled <= pollArraySize;
 } finally {
 end();
 }
 processDeregisterQueue();
-int numKeysUpdated = updateSelectedKeys();
+return updateSelectedKeys();
-if (pollWrapper.getReventOps(0) != 0) {
+}
-// Clear the wakeup pipe
-pollWrapper.putReventOps(0, 0);
+/**
-synchronized (interruptLock) {
+* Process changes to the interest ops.
-IOUtil.drain(fd0);
+*/
-interruptTriggered = false;
+private void processUpdateQueue() {
+assert Thread.holdsLock(this);
+synchronized (updateLock) {
+assert updateKeys.size() == updateOps.size();
+SelectionKeyImpl ski;
+while ((ski = updateKeys.pollFirst()) != null) {
+int ops = updateOps.pollFirst();
+if (ski.isValid()) {
+int index = ski.getIndex();
+assert index >= 0 && index < pollArraySize;
+if (index > 0) {
+assert pollKeys.get(index) == ski;
+if (ops == 0) {
+remove(ski);
+} else {
+update(ski, ops);
+}
+} else if (ops != 0) {
+add(ski, ops);
+}
+}
 }
 }
+}
+/**
+* Update the keys whose fd's have been selected by kqueue.
+* Add the ready keys to the selected key set.
+* If the interrupt fd has been selected, drain it and clear the interrupt.
+*/
+private int updateSelectedKeys() throws IOException {
+assert Thread.holdsLock(this);
+assert Thread.holdsLock(nioSelectedKeys());
+assert pollArraySize > 0 && pollArraySize == pollKeys.size();
+int numKeysUpdated = 0;
+for (int i = 1; i < pollArraySize; i++) {
+int rOps = getReventOps(i);
+if (rOps != 0) {
+SelectionKeyImpl ski = pollKeys.get(i);
+assert ski.channel.getFDVal() == getDescriptor(i);
+if (ski.isValid()) {
+if (selectedKeys.contains(ski)) {
+if (ski.channel.translateAndSetReadyOps(rOps, ski)) {
+numKeysUpdated++;
+}
+} else {
+ski.channel.translateAndSetReadyOps(rOps, ski);
+if ((ski.nioReadyOps() & ski.nioInterestOps()) != 0) {
+selectedKeys.add(ski);
+numKeysUpdated++;
+}
+}
+}
+}
+}
+// check for interrupt
+if (getReventOps(0) != 0) {
+assert getDescriptor(0) == fd0;
+clearInterrupt();
+}
 return numKeysUpdated;
 }
-protected void implCloseInterrupt() throws IOException {
+@Override
+protected void implClose() throws IOException {
+assert !isOpen();
+assert Thread.holdsLock(this);
+assert Thread.holdsLock(nioKeys());
 // prevent further wakeup
 synchronized (interruptLock) {
 interruptTriggered = true;
 }
+pollArray.free();
 FileDispatcherImpl.closeIntFD(fd0);
 FileDispatcherImpl.closeIntFD(fd1);
-fd0 = -1;
-fd1 = -1;
+// Deregister channels
-pollWrapper.release(0);
+Iterator<SelectionKey> i = keys.iterator();
-}
+while (i.hasNext()) {
+SelectionKeyImpl ski = (SelectionKeyImpl)i.next();
+ski.setIndex(-1);
+deregister(ski);
+SelectableChannel selch = ski.channel();
+if (!selch.isOpen() && !selch.isRegistered())
+((SelChImpl)selch).kill();
+i.remove();
+}
+}
+@Override
+protected void implRegister(SelectionKeyImpl ski) {
+assert ski.getIndex() == 0;
+assert Thread.holdsLock(nioKeys());
+ensureOpen();
+keys.add(ski);
+}
+@Override
+protected void implDereg(SelectionKeyImpl ski) throws IOException {
+assert !ski.isValid();
+assert Thread.holdsLock(this);
+assert Thread.holdsLock(nioKeys());
+assert Thread.holdsLock(nioSelectedKeys());
+// remove from poll array
+int index = ski.getIndex();
+if (index > 0) {
+remove(ski);
+}
+// remove from selected-key and key set
+selectedKeys.remove(ski);
+keys.remove(ski);
+// remove from channel's key set
+deregister(ski);
+SelectableChannel selch = ski.channel();
+if (!selch.isOpen() && !selch.isRegistered())
+((SelChImpl) selch).kill();
+}
+@Override
+public void putEventOps(SelectionKeyImpl ski, int ops) {
+ensureOpen();
+synchronized (updateLock) {
+updateOps.addLast(ops);   // ops first in case adding the key fails
+updateKeys.addLast(ski);
+}
+}
+@Override
 public Selector wakeup() {
 synchronized (interruptLock) {
 if (!interruptTriggered) {
-pollWrapper.interrupt();
+try {
+IOUtil.write1(fd1, (byte)0);
+} catch (IOException ioe) {
+throw new InternalError(ioe);
+}
 interruptTriggered = true;
 }
 }
 return this;
 }
+private void clearInterrupt() throws IOException {
+synchronized (interruptLock) {
+IOUtil.drain(fd0);
+interruptTriggered = false;
+}
+}
+/**
+* Sets the first pollfd enty in the poll array to the given fd
+*/
+private void setFirst(int fd, int ops) {
+assert pollArraySize == 0;
+assert pollKeys.isEmpty();
+putDescriptor(0, fd);
+putEventOps(0, ops);
+pollArraySize = 1;
+pollKeys.add(null);  // dummy element
+}
+/**
+* Adds a pollfd entry to the poll array, expanding the poll array if needed.
+*/
+private void add(SelectionKeyImpl ski, int ops) {
+expandIfNeeded();
+int index = pollArraySize;
+assert index > 0;
+putDescriptor(index, ski.channel.getFDVal());
+putEventOps(index, ops);
+putReventOps(index, 0);
+ski.setIndex(index);
+pollArraySize++;
+pollKeys.add(ski);
+assert pollKeys.size() == pollArraySize;
+}
+/**
+* Update the events of pollfd entry.
+*/
+private void update(SelectionKeyImpl ski, int ops) {
+int index = ski.getIndex();
+assert index > 0 && index < pollArraySize;
+assert getDescriptor(index) == ski.channel.getFDVal();
+putEventOps(index, ops);
+}
+/**
+* Removes a pollfd entry from the poll array
+*/
+private void remove(SelectionKeyImpl ski) {
+int index = ski.getIndex();
+assert index > 0 && index < pollArraySize;
+assert getDescriptor(index) == ski.channel.getFDVal();
+// replace pollfd at index with the last pollfd in array
+int lastIndex = pollArraySize - 1;
+if (lastIndex != index) {
+SelectionKeyImpl lastKey = pollKeys.get(lastIndex);
+assert lastKey.getIndex() == lastIndex;
+int lastFd = getDescriptor(lastIndex);
+int lastOps = getEventOps(lastIndex);
+int lastRevents = getReventOps(lastIndex);
+assert lastKey.channel.getFDVal() == lastFd;
+putDescriptor(index, lastFd);
+putEventOps(index, lastOps);
+putReventOps(index, lastRevents);
+pollKeys.set(index, lastKey);
+lastKey.setIndex(index);
+}
+pollKeys.remove(lastIndex);
+pollArraySize--;
+assert pollKeys.size() == pollArraySize;
+ski.setIndex(0);
+}
+/**
+* Expand poll array if at capacity
+*/
+private void expandIfNeeded() {
+if (pollArraySize == pollArrayCapacity) {
+int oldSize = pollArrayCapacity * SIZE_POLLFD;
+int newCapacity = pollArrayCapacity + INITIAL_CAPACITY;
+int newSize = newCapacity * SIZE_POLLFD;
+AllocatedNativeObject newPollArray = new AllocatedNativeObject(newSize, false);
+Unsafe.getUnsafe().copyMemory(pollArray.address(), newPollArray.address(), oldSize);
+pollArray.free();
+pollArray = newPollArray;
+pollArrayCapacity = newCapacity;
+}
+}
+private static final short SIZE_POLLFD   = 8;
+private static final short FD_OFFSET     = 0;
+private static final short EVENT_OFFSET  = 4;
+private static final short REVENT_OFFSET = 6;
+private void putDescriptor(int i, int fd) {
+int offset = SIZE_POLLFD * i + FD_OFFSET;
+pollArray.putInt(offset, fd);
+}
+private int getDescriptor(int i) {
+int offset = SIZE_POLLFD * i + FD_OFFSET;
+return pollArray.getInt(offset);
+}
+private void putEventOps(int i, int event) {
+int offset = SIZE_POLLFD * i + EVENT_OFFSET;
+pollArray.putShort(offset, (short)event);
+}
+private int getEventOps(int i) {
+int offset = SIZE_POLLFD * i + EVENT_OFFSET;
+return pollArray.getShort(offset);
+}
+private void putReventOps(int i, int revent) {
+int offset = SIZE_POLLFD * i + REVENT_OFFSET;
+pollArray.putShort(offset, (short)revent);
+}
+private int getReventOps(int i) {
+int offset = SIZE_POLLFD * i + REVENT_OFFSET;
+return pollArray.getShort(offset);
+}
+private static native int poll(long pollAddress, int numfds, int timeout);
+static {
+IOUtil.load();
+}
 }

changeset 49417	1d3139252c1c
parent 49290	07779973cbe2
child 49493	814bd31f8da0