changeset 51236:ed8de3d0cd28

8199433: (se) select(Consumer<SelectionKey> action) as alternative to selected-key set Reviewed-by: bpb
author alanb
date Sat, 16 Jun 2018 10:10:54 +0100
parents 3fbae7b9ddb5
children 95c0644a1c47
files src/java.base/linux/classes/sun/nio/ch/EPollSelectorImpl.java src/java.base/macosx/classes/sun/nio/ch/KQueueSelectorImpl.java src/java.base/share/classes/java/nio/channels/SelectionKey.java src/java.base/share/classes/java/nio/channels/Selector.java src/java.base/share/classes/sun/nio/ch/SelectorImpl.java src/java.base/solaris/classes/sun/nio/ch/DevPollSelectorImpl.java src/java.base/solaris/classes/sun/nio/ch/EventPortSelectorImpl.java src/java.base/unix/classes/sun/nio/ch/PollSelectorImpl.java src/java.base/windows/classes/sun/nio/ch/WindowsSelectorImpl.java test/jdk/java/nio/channels/Selector/SelectWithConsumer.java
diffstat 10 files changed, 1188 insertions(+), 157 deletions(-) [+]
line wrap: on
line diff
--- a/src/java.base/linux/classes/sun/nio/ch/EPollSelectorImpl.java	Fri Jun 15 12:25:53 2018 +0200
+++ b/src/java.base/linux/classes/sun/nio/ch/EPollSelectorImpl.java	Sat Jun 16 10:10:54 2018 +0100
@@ -27,6 +27,7 @@
 
 import java.io.IOException;
 import java.nio.channels.ClosedSelectorException;
+import java.nio.channels.SelectionKey;
 import java.nio.channels.Selector;
 import java.nio.channels.spi.SelectorProvider;
 import java.util.ArrayDeque;
@@ -34,6 +35,7 @@
 import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
+import java.util.function.Consumer;
 
 import static sun.nio.ch.EPoll.EPOLLIN;
 import static sun.nio.ch.EPoll.EPOLL_CTL_ADD;
@@ -97,7 +99,9 @@
     }
 
     @Override
-    protected int doSelect(long timeout) throws IOException {
+    protected int doSelect(Consumer<SelectionKey> action, long timeout)
+        throws IOException
+    {
         assert Thread.holdsLock(this);
 
         // epoll_wait timeout is int
@@ -130,7 +134,7 @@
             end(blocking);
         }
         processDeregisterQueue();
-        return updateSelectedKeys(numEntries);
+        return processEvents(numEntries, action);
     }
 
     /**
@@ -171,13 +175,13 @@
     }
 
     /**
-     * Update the keys of file descriptors that were polled and add them to
-     * the selected-key set.
+     * Process the polled events.
      * If the interrupt fd has been selected, drain it and clear the interrupt.
      */
-    private int updateSelectedKeys(int numEntries) throws IOException {
+    private int processEvents(int numEntries, Consumer<SelectionKey> action)
+        throws IOException
+    {
         assert Thread.holdsLock(this);
-        assert Thread.holdsLock(nioSelectedKeys());
 
         boolean interrupted = false;
         int numKeysUpdated = 0;
@@ -190,17 +194,7 @@
                 SelectionKeyImpl ski = fdToKey.get(fd);
                 if (ski != null) {
                     int rOps = EPoll.getEvents(event);
-                    if (selectedKeys.contains(ski)) {
-                        if (ski.translateAndUpdateReadyOps(rOps)) {
-                            numKeysUpdated++;
-                        }
-                    } else {
-                        ski.translateAndSetReadyOps(rOps);
-                        if ((ski.nioReadyOps() & ski.nioInterestOps()) != 0) {
-                            selectedKeys.add(ski);
-                            numKeysUpdated++;
-                        }
-                    }
+                    numKeysUpdated += processReadyEvents(rOps, ski, action);
                 }
             }
         }
--- a/src/java.base/macosx/classes/sun/nio/ch/KQueueSelectorImpl.java	Fri Jun 15 12:25:53 2018 +0200
+++ b/src/java.base/macosx/classes/sun/nio/ch/KQueueSelectorImpl.java	Sat Jun 16 10:10:54 2018 +0100
@@ -27,6 +27,7 @@
 
 import java.io.IOException;
 import java.nio.channels.ClosedSelectorException;
+import java.nio.channels.SelectionKey;
 import java.nio.channels.Selector;
 import java.nio.channels.spi.SelectorProvider;
 import java.util.ArrayDeque;
@@ -34,6 +35,7 @@
 import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
+import java.util.function.Consumer;
 
 import static sun.nio.ch.KQueue.EVFILT_READ;
 import static sun.nio.ch.KQueue.EVFILT_WRITE;
@@ -100,7 +102,9 @@
     }
 
     @Override
-    protected int doSelect(long timeout) throws IOException {
+    protected int doSelect(Consumer<SelectionKey> action, long timeout)
+        throws IOException
+    {
         assert Thread.holdsLock(this);
 
         long to = Math.min(timeout, Integer.MAX_VALUE);  // max kqueue timeout
@@ -132,7 +136,7 @@
             end(blocking);
         }
         processDeregisterQueue();
-        return updateSelectedKeys(numEntries);
+        return processEvents(numEntries, action);
     }
 
     /**
@@ -180,13 +184,13 @@
     }
 
     /**
-     * Update the keys of file descriptors that were polled and add them to
-     * the selected-key set.
+     * Process the polled events.
      * If the interrupt fd has been selected, drain it and clear the interrupt.
      */
-    private int updateSelectedKeys(int numEntries) throws IOException {
+    private int processEvents(int numEntries, Consumer<SelectionKey> action)
+        throws IOException
+    {
         assert Thread.holdsLock(this);
-        assert Thread.holdsLock(nioSelectedKeys());
 
         int numKeysUpdated = 0;
         boolean interrupted = false;
@@ -214,22 +218,10 @@
                     } else if (filter == EVFILT_WRITE) {
                         rOps |= Net.POLLOUT;
                     }
-
-                    if (selectedKeys.contains(ski)) {
-                        if (ski.translateAndUpdateReadyOps(rOps)) {
-                            // file descriptor may be polled more than once per poll
-                            if (ski.lastPolled != pollCount) {
-                                numKeysUpdated++;
-                                ski.lastPolled = pollCount;
-                            }
-                        }
-                    } else {
-                        ski.translateAndSetReadyOps(rOps);
-                        if ((ski.nioReadyOps() & ski.nioInterestOps()) != 0) {
-                            selectedKeys.add(ski);
-                            numKeysUpdated++;
-                            ski.lastPolled = pollCount;
-                        }
+                    int updated = processReadyEvents(rOps, ski, action);
+                    if (updated > 0 && ski.lastPolled != pollCount) {
+                        numKeysUpdated++;
+                        ski.lastPolled = pollCount;
                     }
                 }
             }
--- a/src/java.base/share/classes/java/nio/channels/SelectionKey.java	Fri Jun 15 12:25:53 2018 +0200
+++ b/src/java.base/share/classes/java/nio/channels/SelectionKey.java	Sat Jun 16 10:10:54 2018 +0100
@@ -291,7 +291,7 @@
      * detects that the corresponding channel is ready for reading, has reached
      * end-of-stream, has been remotely shut down for further reading, or has
      * an error pending, then it will add {@code OP_READ} to the key's
-     * ready-operation set and add the key to its selected-key&nbsp;set.  </p>
+     * ready-operation set.  </p>
      */
     public static final int OP_READ = 1 << 0;
 
@@ -303,8 +303,7 @@
      * href="Selector.html#selop">selection operation</a>.  If the selector
      * detects that the corresponding channel is ready for writing, has been
      * remotely shut down for further writing, or has an error pending, then it
-     * will add {@code OP_WRITE} to the key's ready set and add the key to its
-     * selected-key&nbsp;set.  </p>
+     * will add {@code OP_WRITE} to the key's ready set.  </p>
      */
     public static final int OP_WRITE = 1 << 2;
 
@@ -316,8 +315,7 @@
      * href="Selector.html#selop">selection operation</a>.  If the selector
      * detects that the corresponding socket channel is ready to complete its
      * connection sequence, or has an error pending, then it will add
-     * {@code OP_CONNECT} to the key's ready set and add the key to its
-     * selected-key&nbsp;set.  </p>
+     * {@code OP_CONNECT} to the key's ready set.  </p>
      */
     public static final int OP_CONNECT = 1 << 3;
 
@@ -329,8 +327,7 @@
      * href="Selector.html#selop">selection operation</a>.  If the selector
      * detects that the corresponding server-socket channel is ready to accept
      * another connection, or has an error pending, then it will add
-     * {@code OP_ACCEPT} to the key's ready set and add the key to its
-     * selected-key&nbsp;set.  </p>
+     * {@code OP_ACCEPT} to the key's ready set.  </p>
      */
     public static final int OP_ACCEPT = 1 << 4;
 
--- a/src/java.base/share/classes/java/nio/channels/Selector.java	Fri Jun 15 12:25:53 2018 +0200
+++ b/src/java.base/share/classes/java/nio/channels/Selector.java	Sat Jun 16 10:10:54 2018 +0100
@@ -28,7 +28,9 @@
 import java.io.Closeable;
 import java.io.IOException;
 import java.nio.channels.spi.SelectorProvider;
+import java.util.Objects;
 import java.util.Set;
+import java.util.function.Consumer;
 
 
 /**
@@ -56,7 +58,8 @@
  *
  *   <li><p> The <i>selected-key set</i> is the set of keys such that each
  *   key's channel was detected to be ready for at least one of the operations
- *   identified in the key's interest set during a prior selection operation.
+ *   identified in the key's interest set during a prior selection operation
+ *   that adds keys or updates keys in the set.
  *   This set is returned by the {@link #selectedKeys() selectedKeys} method.
  *   The selected-key set is always a subset of the key set. </p></li>
  *
@@ -92,6 +95,27 @@
  * <a id="selop"></a>
  * <h2>Selection</h2>
  *
+ * <p> A selection operation queries the underlying operating system for an
+ * update as to the readiness of each registered channel to perform any of the
+ * operations identified by its key's interest set.  There are two forms of
+ * selection operation:
+ *
+ * <ol>
+ *
+ *   <li><p> The {@link #select()}, {@link #select(long)}, and {@link #selectNow()}
+ *   methods add the keys of channels ready to perform an operation to the
+ *   selected-key set, or update the ready-operation set of keys already in the
+ *   selected-key set. </p></li>
+ *
+ *   <li><p> The {@link #select(Consumer)}, {@link #select(Consumer, long)}, and
+ *   {@link #selectNow(Consumer)} methods perform an <i>action</i> on the key
+ *   of each channel that is ready to perform an operation.  These methods do
+ *   not add to the selected-key set. </p></li>
+ *
+ * </ol>
+ *
+ * <h3>Selection operations that add to the selected-key set</h3>
+ *
  * <p> During each selection operation, keys may be added to and removed from a
  * selector's selected-key set and may be removed from its key and
  * cancelled-key sets.  Selection is performed by the {@link #select()}, {@link
@@ -141,6 +165,45 @@
  * difference between the three selection methods. </p>
  *
  *
+ * <h3>Selection operations that perform an action on selected keys</h3>
+ *
+ * <p> During each selection operation, keys may be removed from the selector's
+ * key, selected-key, and cancelled-key sets.  Selection is performed by the
+ * {@link #select(Consumer)}, {@link #select(Consumer,long)}, and {@link
+ * #selectNow(Consumer)} methods, and involves three steps:  </p>
+ *
+ * <ol>
+ *
+ *   <li><p> Each key in the cancelled-key set is removed from each key set of
+ *   which it is a member, and its channel is deregistered.  This step leaves
+ *   the cancelled-key set empty. </p></li>
+ *
+ *   <li><p> The underlying operating system is queried for an update as to the
+ *   readiness of each remaining channel to perform any of the operations
+ *   identified by its key's interest set as of the moment that the selection
+ *   operation began.
+ *
+ *   <p> For a channel that is ready for at least one such operation, the
+ *   ready-operation set of the channel's key is set to identify exactly the
+ *   operations for which the channel is ready and the <i>action</i> specified
+ *   to the {@code select} method is invoked to consume the channel's key.  Any
+ *   readiness information previously recorded in the ready set is discarded
+ *   prior to invoking the <i>action</i>.
+ *
+ *   <p> Alternatively, where a channel is ready for more than one operation,
+ *   the <i>action</i> may be invoked more than once with the channel's key and
+ *   ready-operation set modified to a subset of the operations for which the
+ *   channel is ready.  Where the <i>action</i> is invoked more than once for
+ *   the same key then its ready-operation set never contains operation bits
+ *   that were contained in the set at previous calls to the <i>action</i>
+ *   in the same selection operation.  </p></li>
+ *
+ *   <li><p> If any keys were added to the cancelled-key set while step (2) was
+ *   in progress then they are processed as in step (1). </p></li>
+ *
+ * </ol>
+ *
+ *
  * <h2>Concurrency</h2>
  *
  * <p> A Selector and its key set are safe for use by multiple concurrent
@@ -156,13 +219,12 @@
  *
  * <p> Keys may be cancelled and channels may be closed at any time.  Hence the
  * presence of a key in one or more of a selector's key sets does not imply
- * that the key is valid or that its channel is open.  Application code should
+ * that the key is valid or that its channel is open. Application code should
  * be careful to synchronize and check these conditions as necessary if there
  * is any possibility that another thread will cancel a key or close a channel.
  *
- * <p> A thread blocked in one of the {@link #select()} or {@link
- * #select(long)} methods may be interrupted by some other thread in one of
- * three ways:
+ * <p> A thread blocked in a selection operation may be interrupted by some
+ * other thread in one of three ways:
  *
  * <ul>
  *
@@ -356,18 +418,188 @@
     public abstract int select() throws IOException;
 
     /**
+     * Selects and performs an action on the keys whose corresponding channels
+     * are ready for I/O operations.
+     *
+     * <p> This method performs a blocking <a href="#selop">selection
+     * operation</a>.  It wakes up from querying the operating system only when
+     * at least one channel is selected, this selector's {@link #wakeup wakeup}
+     * method is invoked, the current thread is interrupted, or the given
+     * timeout period expires, whichever comes first.
+     *
+     * <p> The specified <i>action</i>'s {@link Consumer#accept(Object) accept}
+     * method is invoked with the key for each channel that is ready to perform
+     * an operation identified by its key's interest set.  The {@code accept}
+     * method may be invoked more than once for the same key but with the
+     * ready-operation set containing a subset of the operations for which the
+     * channel is ready (as described above).  The {@code accept} method is
+     * invoked while synchronized on the selector and its selected-key set.
+     * Great care must be taken to avoid deadlocking with other threads that
+     * also synchronize on these objects.  Selection operations are not reentrant
+     * in general and consequently the <i>action</i> should take great care not
+     * to attempt a selection operation on the same selector.  The behavior when
+     * attempting a reentrant selection operation is implementation specific and
+     * therefore not specified.  If the <i>action</i> closes the selector then
+     * {@code ClosedSelectorException} is thrown when the action completes.
+     * The <i>action</i> is not prohibited from closing channels registered with
+     * the selector, nor prohibited from cancelling keys or changing a key's
+     * interest set.  If a channel is selected but its key is cancelled or its
+     * interest set changed before the <i>action</i> is performed on the key
+     * then it is implementation specific as to whether the <i>action</i> is
+     * invoked (it may be invoked with an {@link SelectionKey#isValid() invalid}
+     * key).  Exceptions thrown by the action are relayed to the caller.
+     *
+     * <p> This method does not offer real-time guarantees: It schedules the
+     * timeout as if by invoking the {@link Object#wait(long)} method.
+     *
+     * @implSpec The default implementation removes all keys from the
+     * selected-key set, invokes {@link #select(long) select(long)} with the
+     * given timeout and then performs the action for each key added to the
+     * selected-key set.  The default implementation does not detect the action
+     * performing a reentrant selection operation.  The selected-key set may
+     * or may not be empty on completion of the default implementation.
+     *
+     * @param  action   The action to perform
+     *
+     * @param  timeout  If positive, block for up to {@code timeout}
+     *                  milliseconds, more or less, while waiting for a
+     *                  channel to become ready; if zero, block indefinitely;
+     *                  must not be negative
+     *
+     * @return  The number of unique keys consumed, possibly zero
+     *
+     * @throws  IOException
+     *          If an I/O error occurs
+     *
+     * @throws  ClosedSelectorException
+     *          If this selector is closed or is closed by the action
+     *
+     * @throws  IllegalArgumentException
+     *          If the value of the timeout argument is negative
+     *
+     * @since 11
+     */
+    public int select(Consumer<SelectionKey> action, long timeout)
+        throws IOException
+    {
+        if (timeout < 0)
+            throw new IllegalArgumentException("Negative timeout");
+        return doSelect(Objects.requireNonNull(action), timeout);
+    }
+
+    /**
+     * Selects and performs an action on the keys whose corresponding channels
+     * are ready for I/O operations.
+     *
+     * <p> This method performs a blocking <a href="#selop">selection
+     * operation</a>.  It wakes up from querying the operating system only when
+     * at least one channel is selected, this selector's {@link #wakeup wakeup}
+     * method is invoked, or the current thread is interrupted, whichever comes
+     * first.
+     *
+     * <p> This method is equivalent to invoking the 2-arg
+     * {@link #select(Consumer, long) select} method with a timeout of {@code 0}
+     * to block indefinitely.  </p>
+     *
+     * @implSpec The default implementation invokes the 2-arg {@code select}
+     * method with a timeout of {@code 0}.
+     *
+     * @param  action   The action to perform
+     *
+     * @return  The number of unique keys consumed, possibly zero
+     *
+     * @throws  IOException
+     *          If an I/O error occurs
+     *
+     * @throws  ClosedSelectorException
+     *          If this selector is closed or is closed by the action
+     *
+     * @since 11
+     */
+    public int select(Consumer<SelectionKey> action) throws IOException {
+        return select(action, 0);
+    }
+
+    /**
+     * Selects and performs an action on the keys whose corresponding channels
+     * are ready for I/O operations.
+     *
+     * <p> This method performs a non-blocking <a href="#selop">selection
+     * operation</a>.
+     *
+     * <p> Invoking this method clears the effect of any previous invocations
+     * of the {@link #wakeup wakeup} method.  </p>
+     *
+     * @implSpec The default implementation removes all keys from the
+     * selected-key set, invokes {@link #selectNow() selectNow()} and then
+     * performs the action for each key added to the selected-key set.  The
+     * default implementation does not detect the action performing a reentrant
+     * selection operation.  The selected-key set may or may not be empty on
+     * completion of the default implementation.
+     *
+     * @param  action   The action to perform
+     *
+     * @return  The number of unique keys consumed, possibly zero
+     *
+     * @throws  IOException
+     *          If an I/O error occurs
+     *
+     * @throws  ClosedSelectorException
+     *          If this selector is closed or is closed by the action
+     *
+     * @since 11
+     */
+    public int selectNow(Consumer<SelectionKey> action) throws IOException {
+        return doSelect(Objects.requireNonNull(action), -1);
+    }
+
+    /**
+     * Default implementation of select(Consumer) and selectNow(Consumer).
+     */
+    private int doSelect(Consumer<SelectionKey> action, long timeout)
+        throws IOException
+    {
+        synchronized (this) {
+            Set<SelectionKey> selectedKeys = selectedKeys();
+            synchronized (selectedKeys) {
+                selectedKeys.clear();
+                int numKeySelected;
+                if (timeout < 0) {
+                    numKeySelected = selectNow();
+                } else {
+                    numKeySelected = select(timeout);
+                }
+
+                // copy selected-key set as action may remove keys
+                Set<SelectionKey> keysToConsume = Set.copyOf(selectedKeys);
+                assert keysToConsume.size() == numKeySelected;
+                selectedKeys.clear();
+
+                // invoke action for each selected key
+                keysToConsume.forEach(k -> {
+                    action.accept(k);
+                    if (!isOpen())
+                        throw new ClosedSelectorException();
+                });
+
+                return numKeySelected;
+            }
+        }
+    }
+
+
+    /**
      * Causes the first selection operation that has not yet returned to return
      * immediately.
      *
-     * <p> If another thread is currently blocked in an invocation of the
-     * {@link #select()} or {@link #select(long)} methods then that invocation
-     * will return immediately.  If no selection operation is currently in
-     * progress then the next invocation of one of these methods will return
-     * immediately unless the {@link #selectNow()} method is invoked in the
-     * meantime.  In any case the value returned by that invocation may be
-     * non-zero.  Subsequent invocations of the {@link #select()} or {@link
-     * #select(long)} methods will block as usual unless this method is invoked
-     * again in the meantime.
+     * <p> If another thread is currently blocked in a selection operation then
+     * that invocation will return immediately.  If no selection operation is
+     * currently in progress then the next invocation of a selection operation
+     * will return immediately unless {@link #selectNow()} or {@link
+     * #selectNow(Consumer)} is invoked in the meantime.  In any case the value
+     * returned by that invocation may be non-zero.  Subsequent selection
+     * operations will block as usual unless this method is invoked again in the
+     * meantime.
      *
      * <p> Invoking this method more than once between two successive selection
      * operations has the same effect as invoking it just once.  </p>
@@ -398,5 +630,4 @@
      *          If an I/O error occurs
      */
     public abstract void close() throws IOException;
-
 }
--- a/src/java.base/share/classes/sun/nio/ch/SelectorImpl.java	Fri Jun 15 12:25:53 2018 +0200
+++ b/src/java.base/share/classes/sun/nio/ch/SelectorImpl.java	Sat Jun 16 10:10:54 2018 +0100
@@ -36,8 +36,10 @@
 import java.util.Collections;
 import java.util.HashSet;
 import java.util.Iterator;
+import java.util.Objects;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.Consumer;
 
 
 /**
@@ -51,12 +53,15 @@
     private final Set<SelectionKey> keys;
 
     // The set of keys with data ready for an operation
-    protected final Set<SelectionKey> selectedKeys;
+    private final Set<SelectionKey> selectedKeys;
 
     // Public views of the key sets
     private final Set<SelectionKey> publicKeys;             // Immutable
     private final Set<SelectionKey> publicSelectedKeys;     // Removal allowed, but not addition
 
+    // used to check for reentrancy
+    private boolean inSelect;
+
     protected SelectorImpl(SelectorProvider sp) {
         super(sp);
         keys = ConcurrentHashMap.newKeySet();
@@ -83,13 +88,6 @@
     }
 
     /**
-     * Returns the public view of the selected-key set
-     */
-    protected final Set<SelectionKey> nioSelectedKeys() {
-        return publicSelectedKeys;
-    }
-
-    /**
      * Marks the beginning of a select operation that might block
      */
     protected final void begin(boolean blocking) {
@@ -106,16 +104,27 @@
     /**
      * Selects the keys for channels that are ready for I/O operations.
      *
+     * @param action  the action to perform, can be null
      * @param timeout timeout in milliseconds to wait, 0 to not wait, -1 to
      *                wait indefinitely
      */
-    protected abstract int doSelect(long timeout) throws IOException;
+    protected abstract int doSelect(Consumer<SelectionKey> action, long timeout)
+        throws IOException;
 
-    private int lockAndDoSelect(long timeout) throws IOException {
+    private int lockAndDoSelect(Consumer<SelectionKey> action, long timeout)
+        throws IOException
+    {
         synchronized (this) {
             ensureOpen();
-            synchronized (publicSelectedKeys) {
-                return doSelect(timeout);
+            if (inSelect)
+                throw new IllegalStateException("select in progress");
+            inSelect = true;
+            try {
+                synchronized (publicSelectedKeys) {
+                    return doSelect(action, timeout);
+                }
+            } finally {
+                inSelect = false;
             }
         }
     }
@@ -124,17 +133,39 @@
     public final int select(long timeout) throws IOException {
         if (timeout < 0)
             throw new IllegalArgumentException("Negative timeout");
-        return lockAndDoSelect((timeout == 0) ? -1 : timeout);
+        return lockAndDoSelect(null, (timeout == 0) ? -1 : timeout);
     }
 
     @Override
     public final int select() throws IOException {
-        return lockAndDoSelect(-1);
+        return lockAndDoSelect(null, -1);
     }
 
     @Override
     public final int selectNow() throws IOException {
-        return lockAndDoSelect(0);
+        return lockAndDoSelect(null, 0);
+    }
+
+    @Override
+    public final int select(Consumer<SelectionKey> action, long timeout)
+        throws IOException
+    {
+        Objects.requireNonNull(action);
+        if (timeout < 0)
+            throw new IllegalArgumentException("Negative timeout");
+        return lockAndDoSelect(action, (timeout == 0) ? -1 : timeout);
+    }
+
+    @Override
+    public final int select(Consumer<SelectionKey> action) throws IOException {
+        Objects.requireNonNull(action);
+        return lockAndDoSelect(action, -1);
+    }
+
+    @Override
+    public final int selectNow(Consumer<SelectionKey> action) throws IOException {
+        Objects.requireNonNull(action);
+        return lockAndDoSelect(action, 0);
     }
 
     /**
@@ -240,6 +271,39 @@
     }
 
     /**
+     * Invoked by selection operations to handle ready events. If an action
+     * is specified then it is invoked to handle the key, otherwise the key
+     * is added to the selected-key set (or updated when it is already in the
+     * set).
+     */
+    protected final int processReadyEvents(int rOps,
+                                           SelectionKeyImpl ski,
+                                           Consumer<SelectionKey> action) {
+        if (action != null) {
+            ski.translateAndSetReadyOps(rOps);
+            if ((ski.nioReadyOps() & ski.nioInterestOps()) != 0) {
+                action.accept(ski);
+                ensureOpen();
+                return 1;
+            }
+        } else {
+            assert Thread.holdsLock(publicSelectedKeys);
+            if (selectedKeys.contains(ski)) {
+                if (ski.translateAndUpdateReadyOps(rOps)) {
+                    return 1;
+                }
+            } else {
+                ski.translateAndSetReadyOps(rOps);
+                if ((ski.nioReadyOps() & ski.nioInterestOps()) != 0) {
+                    selectedKeys.add(ski);
+                    return 1;
+                }
+            }
+        }
+        return 0;
+    }
+
+    /**
      * Invoked by interestOps to ensure the interest ops are updated at the
      * next selection operation.
      */
--- a/src/java.base/solaris/classes/sun/nio/ch/DevPollSelectorImpl.java	Fri Jun 15 12:25:53 2018 +0200
+++ b/src/java.base/solaris/classes/sun/nio/ch/DevPollSelectorImpl.java	Sat Jun 16 10:10:54 2018 +0100
@@ -27,6 +27,7 @@
 
 import java.io.IOException;
 import java.nio.channels.ClosedSelectorException;
+import java.nio.channels.SelectionKey;
 import java.nio.channels.Selector;
 import java.nio.channels.spi.SelectorProvider;
 import java.util.ArrayDeque;
@@ -34,6 +35,7 @@
 import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
+import java.util.function.Consumer;
 
 import static sun.nio.ch.DevPollArrayWrapper.NUM_POLLFDS;
 import static sun.nio.ch.DevPollArrayWrapper.POLLREMOVE;
@@ -85,7 +87,9 @@
     }
 
     @Override
-    protected int doSelect(long timeout) throws IOException {
+    protected int doSelect(Consumer<SelectionKey> action, long timeout)
+        throws IOException
+    {
         assert Thread.holdsLock(this);
 
         long to = timeout;
@@ -117,7 +121,7 @@
             end(blocking);
         }
         processDeregisterQueue();
-        return updateSelectedKeys(numEntries);
+        return processEvents(numEntries, action);
     }
 
     /**
@@ -165,13 +169,13 @@
     }
 
     /**
-     * Update the keys of file descriptors that were polled and add them to
-     * the selected-key set.
+     * Process the polled events.
      * If the interrupt fd has been selected, drain it and clear the interrupt.
      */
-    private int updateSelectedKeys(int numEntries) throws IOException {
+    private int processEvents(int numEntries, Consumer<SelectionKey> action)
+        throws IOException
+    {
         assert Thread.holdsLock(this);
-        assert Thread.holdsLock(nioSelectedKeys());
 
         boolean interrupted = false;
         int numKeysUpdated = 0;
@@ -183,17 +187,7 @@
                 SelectionKeyImpl ski = fdToKey.get(fd);
                 if (ski != null) {
                     int rOps = pollWrapper.getReventOps(i);
-                    if (selectedKeys.contains(ski)) {
-                        if (ski.translateAndUpdateReadyOps(rOps)) {
-                            numKeysUpdated++;
-                        }
-                    } else {
-                        ski.translateAndSetReadyOps(rOps);
-                        if ((ski.nioReadyOps() & ski.nioInterestOps()) != 0) {
-                            selectedKeys.add(ski);
-                            numKeysUpdated++;
-                        }
-                    }
+                    numKeysUpdated += processReadyEvents(rOps, ski, action);
                 }
             }
         }
--- a/src/java.base/solaris/classes/sun/nio/ch/EventPortSelectorImpl.java	Fri Jun 15 12:25:53 2018 +0200
+++ b/src/java.base/solaris/classes/sun/nio/ch/EventPortSelectorImpl.java	Sat Jun 16 10:10:54 2018 +0100
@@ -35,6 +35,7 @@
 import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
+import java.util.function.Consumer;
 
 import static sun.nio.ch.SolarisEventPort.PORT_SOURCE_FD;
 import static sun.nio.ch.SolarisEventPort.PORT_SOURCE_USER;
@@ -97,7 +98,9 @@
     }
 
     @Override
-    protected int doSelect(long timeout) throws IOException {
+    protected int doSelect(Consumer<SelectionKey> action, long timeout)
+        throws IOException
+    {
         assert Thread.holdsLock(this);
 
         long to = timeout;
@@ -129,7 +132,7 @@
             end(blocking);
         }
         processDeregisterQueue();
-        return processPortEvents(numEvents);
+        return processPortEvents(numEvents, action);
     }
 
     /**
@@ -166,19 +169,21 @@
     }
 
     /**
-     * Process the port events. This method updates the keys of file descriptors
-     * that were polled. It also re-queues the key so that the file descriptor
-     * is re-associated at the next select operation.
-     *
-     * @return the number of selection keys updated.
+     * Process the polled events and re-queue the selected keys so the file
+     * descriptors are re-associated at the next select operation.
      */
-    private int processPortEvents(int numEvents) throws IOException {
+    private int processPortEvents(int numEvents, Consumer<SelectionKey> action)
+        throws IOException
+    {
         assert Thread.holdsLock(this);
-        assert Thread.holdsLock(nioSelectedKeys());
 
         int numKeysUpdated = 0;
         boolean interrupted = false;
 
+        // Process the polled events while holding the update lock. This allows
+        // keys to be queued for ready file descriptors so they can be
+        // re-associated at the next select. The selected-key can be updated
+        // in this pass.
         synchronized (updateLock) {
             for (int i = 0; i < numEvents; i++) {
                 short source = getSource(i);
@@ -186,22 +191,15 @@
                     int fd = getDescriptor(i);
                     SelectionKeyImpl ski = fdToKey.get(fd);
                     if (ski != null) {
-                        int rOps = getEventOps(i);
-                        if (selectedKeys.contains(ski)) {
-                            if (ski.translateAndUpdateReadyOps(rOps)) {
-                                numKeysUpdated++;
-                            }
-                        } else {
-                            ski.translateAndSetReadyOps(rOps);
-                            if ((ski.nioReadyOps() & ski.nioInterestOps()) != 0) {
-                                selectedKeys.add(ski);
-                                numKeysUpdated++;
-                            }
+                        ski.registeredEvents(0);
+                        updateKeys.addLast(ski);
+
+                        // update selected-key set if no action specified
+                        if (action == null) {
+                            int rOps = getEventOps(i);
+                            numKeysUpdated += processReadyEvents(rOps, ski, null);
                         }
 
-                        // re-queue key so it re-associated at next select
-                        ski.registeredEvents(0);
-                        updateKeys.addLast(ski);
                     }
                 } else if (source == PORT_SOURCE_USER) {
                     interrupted = true;
@@ -211,6 +209,22 @@
             }
         }
 
+        // if an action specified then iterate over the polled events again so
+        // that the action is performed without holding the update lock.
+        if (action != null) {
+            for (int i = 0; i < numEvents; i++) {
+                short source = getSource(i);
+                if (source == PORT_SOURCE_FD) {
+                    int fd = getDescriptor(i);
+                    SelectionKeyImpl ski = fdToKey.get(fd);
+                    if (ski != null) {
+                        int rOps = getEventOps(i);
+                        numKeysUpdated += processReadyEvents(rOps, ski, action);
+                    }
+                }
+            }
+        }
+
         if (interrupted) {
             clearInterrupt();
         }
--- a/src/java.base/unix/classes/sun/nio/ch/PollSelectorImpl.java	Fri Jun 15 12:25:53 2018 +0200
+++ b/src/java.base/unix/classes/sun/nio/ch/PollSelectorImpl.java	Sat Jun 16 10:10:54 2018 +0100
@@ -26,6 +26,7 @@
 
 import java.io.IOException;
 import java.nio.channels.ClosedSelectorException;
+import java.nio.channels.SelectionKey;
 import java.nio.channels.Selector;
 import java.nio.channels.spi.SelectorProvider;
 import java.util.ArrayDeque;
@@ -33,6 +34,7 @@
 import java.util.Deque;
 import java.util.List;
 import java.util.concurrent.TimeUnit;
+import java.util.function.Consumer;
 
 import jdk.internal.misc.Unsafe;
 
@@ -92,7 +94,9 @@
     }
 
     @Override
-    protected int doSelect(long timeout) throws IOException {
+    protected int doSelect(Consumer<SelectionKey> action, long timeout)
+        throws IOException
+    {
         assert Thread.holdsLock(this);
 
         int to = (int) Math.min(timeout, Integer.MAX_VALUE); // max poll timeout
@@ -125,7 +129,7 @@
         }
 
         processDeregisterQueue();
-        return updateSelectedKeys();
+        return processEvents(action);
     }
 
     /**
@@ -157,13 +161,13 @@
     }
 
     /**
-     * Update the keys of file descriptors that were polled and add them to
-     * the selected-key set.
+     * Process the polled events.
      * If the interrupt fd has been selected, drain it and clear the interrupt.
      */
-    private int updateSelectedKeys() throws IOException {
+    private int processEvents(Consumer<SelectionKey> action)
+        throws IOException
+    {
         assert Thread.holdsLock(this);
-        assert Thread.holdsLock(nioSelectedKeys());
         assert pollArraySize > 0 && pollArraySize == pollKeys.size();
 
         int numKeysUpdated = 0;
@@ -173,17 +177,7 @@
                 SelectionKeyImpl ski = pollKeys.get(i);
                 assert ski.getFDVal() == getDescriptor(i);
                 if (ski.isValid()) {
-                    if (selectedKeys.contains(ski)) {
-                        if (ski.translateAndUpdateReadyOps(rOps)) {
-                            numKeysUpdated++;
-                        }
-                    } else {
-                        ski.translateAndSetReadyOps(rOps);
-                        if ((ski.nioReadyOps() & ski.nioInterestOps()) != 0) {
-                            selectedKeys.add(ski);
-                            numKeysUpdated++;
-                        }
-                    }
+                    numKeysUpdated += processReadyEvents(rOps, ski, action);
                 }
             }
         }
--- a/src/java.base/windows/classes/sun/nio/ch/WindowsSelectorImpl.java	Fri Jun 15 12:25:53 2018 +0200
+++ b/src/java.base/windows/classes/sun/nio/ch/WindowsSelectorImpl.java	Sat Jun 16 10:10:54 2018 +0100
@@ -28,6 +28,7 @@
 import java.io.IOException;
 import java.nio.channels.ClosedSelectorException;
 import java.nio.channels.Pipe;
+import java.nio.channels.SelectionKey;
 import java.nio.channels.Selector;
 import java.nio.channels.spi.SelectorProvider;
 import java.util.ArrayDeque;
@@ -36,6 +37,7 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.function.Consumer;
 
 /**
  * A multi-threaded implementation of Selector for Windows.
@@ -139,7 +141,9 @@
     }
 
     @Override
-    protected int doSelect(long timeout) throws IOException {
+    protected int doSelect(Consumer<SelectionKey> action, long timeout)
+        throws IOException
+    {
         assert Thread.holdsLock(this);
         this.timeout = timeout; // set selector timeout
         processUpdateQueue();
@@ -173,7 +177,7 @@
         // Done with poll(). Set wakeupSocket to nonsignaled  for the next run.
         finishLock.checkForException();
         processDeregisterQueue();
-        int updated = updateSelectedKeys();
+        int updated = updateSelectedKeys(action);
         // Done with poll(). Set wakeupSocket to nonsignaled  for the next run.
         resetWakeupSocket();
         return updated;
@@ -349,16 +353,16 @@
         private native int poll0(long pollAddress, int numfds,
              int[] readFds, int[] writeFds, int[] exceptFds, long timeout);
 
-        private int processSelectedKeys(long updateCount) {
+        private int processSelectedKeys(long updateCount, Consumer<SelectionKey> action) {
             int numKeysUpdated = 0;
-            numKeysUpdated += processFDSet(updateCount, readFds,
+            numKeysUpdated += processFDSet(updateCount, action, readFds,
                                            Net.POLLIN,
                                            false);
-            numKeysUpdated += processFDSet(updateCount, writeFds,
+            numKeysUpdated += processFDSet(updateCount, action, writeFds,
                                            Net.POLLCONN |
                                            Net.POLLOUT,
                                            false);
-            numKeysUpdated += processFDSet(updateCount, exceptFds,
+            numKeysUpdated += processFDSet(updateCount, action, exceptFds,
                                            Net.POLLIN |
                                            Net.POLLCONN |
                                            Net.POLLOUT,
@@ -372,7 +376,9 @@
          *
          * me.updateCount <= updateCount
          */
-        private int processFDSet(long updateCount, int[] fds, int rOps,
+        private int processFDSet(long updateCount,
+                                 Consumer<SelectionKey> action,
+                                 int[] fds, int rOps,
                                  boolean isExceptFds)
         {
             int numKeysUpdated = 0;
@@ -401,20 +407,10 @@
                     continue;
                 }
 
-                if (selectedKeys.contains(sk)) { // Key in selected set
-                    if (sk.translateAndUpdateReadyOps(rOps)) {
-                        if (me.updateCount != updateCount) {
-                            me.updateCount = updateCount;
-                            numKeysUpdated++;
-                        }
-                    }
-                } else { // Key is not in selected set yet
-                    sk.translateAndSetReadyOps(rOps);
-                    if ((sk.nioReadyOps() & sk.nioInterestOps()) != 0) {
-                        selectedKeys.add(sk);
-                        me.updateCount = updateCount;
-                        numKeysUpdated++;
-                    }
+                int updated = processReadyEvents(rOps, sk, action);
+                if (updated > 0 && me.updateCount != updateCount) {
+                    me.updateCount = updateCount;
+                    numKeysUpdated++;
                 }
             }
             return numKeysUpdated;
@@ -509,12 +505,12 @@
 
     // Update ops of the corresponding Channels. Add the ready keys to the
     // ready queue.
-    private int updateSelectedKeys() {
+    private int updateSelectedKeys(Consumer<SelectionKey> action) {
         updateCount++;
         int numKeysUpdated = 0;
-        numKeysUpdated += subSelector.processSelectedKeys(updateCount);
+        numKeysUpdated += subSelector.processSelectedKeys(updateCount, action);
         for (SelectThread t: threads) {
-            numKeysUpdated += t.subSelector.processSelectedKeys(updateCount);
+            numKeysUpdated += t.subSelector.processSelectedKeys(updateCount, action);
         }
         return numKeysUpdated;
     }
--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/test/jdk/java/nio/channels/Selector/SelectWithConsumer.java	Sat Jun 16 10:10:54 2018 +0100
@@ -0,0 +1,755 @@
+/*
+ * Copyright (c) 2018, Oracle and/or its affiliates. All rights reserved.
+ * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
+ *
+ * This code is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU General Public License version 2 only, as
+ * published by the Free Software Foundation.
+ *
+ * This code is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
+ * version 2 for more details (a copy is included in the LICENSE file that
+ * accompanied this code).
+ *
+ * You should have received a copy of the GNU General Public License version
+ * 2 along with this work; if not, write to the Free Software Foundation,
+ * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
+ *
+ * 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.
+ */
+
+/* @test
+ * @bug 8199433
+ * @run testng SelectWithConsumer
+ * @summary Unit test for Selector select(Consumer), select(Consumer,long) and
+ *          selectNow(Consumer)
+ */
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.nio.ByteBuffer;
+import java.nio.channels.ClosedSelectorException;
+import java.nio.channels.Pipe;
+import java.nio.channels.SelectionKey;
+import java.nio.channels.Selector;
+import java.nio.channels.ServerSocketChannel;
+import java.nio.channels.SocketChannel;
+import java.nio.channels.WritableByteChannel;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import static java.util.concurrent.TimeUnit.*;
+
+import org.testng.annotations.AfterTest;
+import org.testng.annotations.Test;
+import static org.testng.Assert.*;
+
+@Test
+public class SelectWithConsumer {
+
+    /**
+     * Invoke the select methods that take an action and check that the
+     * accumulated ready ops notified to the action matches the expected ops.
+     */
+    void testActionInvoked(SelectionKey key, int expectedOps) throws Exception {
+        var callerThread = Thread.currentThread();
+        var sel = key.selector();
+        var interestOps = key.interestOps();
+        var notifiedOps = new AtomicInteger();
+
+        // select(Consumer)
+        if (expectedOps == 0)
+            sel.wakeup(); // ensure select does not block
+        notifiedOps.set(0);
+        int n = sel.select(k -> {
+            assertTrue(Thread.currentThread() == callerThread);
+            assertTrue(k == key);
+            int readyOps = key.readyOps();
+            assertTrue((readyOps & interestOps) != 0);
+            assertTrue((readyOps & notifiedOps.get()) == 0);
+            notifiedOps.set(notifiedOps.get() | readyOps);
+        });
+        assertTrue((n == 1) ^ (expectedOps == 0));
+        assertTrue(notifiedOps.get() == expectedOps);
+
+        // select(Consumer, timeout)
+        notifiedOps.set(0);
+        n = sel.select(k -> {
+            assertTrue(Thread.currentThread() == callerThread);
+            assertTrue(k == key);
+            int readyOps = key.readyOps();
+            assertTrue((readyOps & interestOps) != 0);
+            assertTrue((readyOps & notifiedOps.get()) == 0);
+            notifiedOps.set(notifiedOps.get() | readyOps);
+        }, 1000);
+        assertTrue((n == 1) ^ (expectedOps == 0));
+        assertTrue(notifiedOps.get() == expectedOps);
+
+        // selectNow(Consumer)
+        notifiedOps.set(0);
+        n = sel.selectNow(k -> {
+            assertTrue(Thread.currentThread() == callerThread);
+            assertTrue(k == key);
+            int readyOps = key.readyOps();
+            assertTrue((readyOps & interestOps) != 0);
+            assertTrue((readyOps & notifiedOps.get()) == 0);
+            notifiedOps.set(notifiedOps.get() | readyOps);
+        });
+        assertTrue((n == 1) ^ (expectedOps == 0));
+        assertTrue(notifiedOps.get() == expectedOps);
+    }
+
+    /**
+     * Test that an action is performed when a channel is ready for reading.
+     */
+    public void testReadable() throws Exception {
+        Pipe p = Pipe.open();
+        try (Selector sel = Selector.open()) {
+            Pipe.SinkChannel sink = p.sink();
+            Pipe.SourceChannel source = p.source();
+            source.configureBlocking(false);
+            SelectionKey key = source.register(sel, SelectionKey.OP_READ);
+
+            // write to sink to ensure source is readable
+            scheduleWrite(sink, messageBuffer(), 100, MILLISECONDS);
+
+            // test that action is invoked
+            testActionInvoked(key, SelectionKey.OP_READ);
+        } finally {
+            closePipe(p);
+        }
+    }
+
+    /**
+     * Test that an action is performed when a channel is ready for writing.
+     */
+    public void testWritable() throws Exception {
+        Pipe p = Pipe.open();
+        try (Selector sel = Selector.open()) {
+            Pipe.SourceChannel source = p.source();
+            Pipe.SinkChannel sink = p.sink();
+            sink.configureBlocking(false);
+            SelectionKey key = sink.register(sel, SelectionKey.OP_WRITE);
+
+            // test that action is invoked
+            testActionInvoked(key, SelectionKey.OP_WRITE);
+        } finally {
+            closePipe(p);
+        }
+    }
+
+    /**
+     * Test that an action is performed when a channel is ready for both
+     * reading and writing.
+     */
+    public void testReadableAndWriteable() throws Exception {
+        ServerSocketChannel ssc = null;
+        SocketChannel sc = null;
+        SocketChannel peer = null;
+        try (Selector sel = Selector.open()) {
+            ssc = ServerSocketChannel.open().bind(new InetSocketAddress(0));
+            sc = SocketChannel.open(ssc.getLocalAddress());
+            sc.configureBlocking(false);
+            SelectionKey key = sc.register(sel, (SelectionKey.OP_READ |
+                                                 SelectionKey.OP_WRITE));
+
+            // accept connection and write data so the source is readable
+            peer = ssc.accept();
+            peer.write(messageBuffer());
+
+            // test that action is invoked
+            testActionInvoked(key, (SelectionKey.OP_READ | SelectionKey.OP_WRITE));
+        } finally {
+            if (ssc != null) ssc.close();
+            if (sc != null) sc.close();
+            if (peer != null) peer.close();
+        }
+    }
+
+    /**
+     * Test that the action is called for two selected channels
+     */
+    public void testTwoChannels() throws Exception {
+        Pipe p = Pipe.open();
+        try (Selector sel = Selector.open()) {
+            Pipe.SourceChannel source = p.source();
+            Pipe.SinkChannel sink = p.sink();
+            source.configureBlocking(false);
+            sink.configureBlocking(false);
+            SelectionKey key1 = source.register(sel, SelectionKey.OP_READ);
+            SelectionKey key2 = sink.register(sel, SelectionKey.OP_WRITE);
+
+            // write to sink to ensure that the source is readable
+            sink.write(messageBuffer());
+
+            var counter = new AtomicInteger();
+
+            // select(Consumer)
+            counter.set(0);
+            int n = sel.select(k -> {
+                counter.incrementAndGet();
+                if (k == key1) {
+                    assertTrue(k.isReadable());
+                } else if (k == key2) {
+                    assertTrue(k.isWritable());
+                } else {
+                    assertTrue(false);
+                }
+            });
+            assertTrue(n == 2);
+            assertTrue(counter.get() == 2);
+
+            // select(Consumer, timeout)
+            counter.set(0);
+            n = sel.select(k -> {
+                counter.incrementAndGet();
+                if (k == key1) {
+                    assertTrue(k.isReadable());
+                } else if (k == key2) {
+                    assertTrue(k.isWritable());
+                } else {
+                    assertTrue(false);
+                }
+            }, 1000);
+            assertTrue(n == 2);
+            assertTrue(counter.get() == 2);
+
+            // selectNow(Consumer)
+            counter.set(0);
+            n = sel.selectNow(k -> {
+                counter.incrementAndGet();
+                if (k == key1) {
+                    assertTrue(k.isReadable());
+                } else if (k == key2) {
+                    assertTrue(k.isWritable());
+                } else {
+                    assertTrue(false);
+                }
+            });
+            assertTrue(n == 2);
+            assertTrue(counter.get() == 2);
+        } finally {
+            closePipe(p);
+        }
+    }
+
+    /**
+     * Test calling select twice, the action should be invoked each time
+     */
+    public void testRepeatedSelect1() throws Exception {
+        Pipe p = Pipe.open();
+        try (Selector sel = Selector.open()) {
+            Pipe.SourceChannel source = p.source();
+            Pipe.SinkChannel sink = p.sink();
+            source.configureBlocking(false);
+            SelectionKey key = source.register(sel, SelectionKey.OP_READ);
+
+            // write to sink to ensure that the source is readable
+            sink.write(messageBuffer());
+
+            // test that action is invoked
+            testActionInvoked(key, SelectionKey.OP_READ);
+            testActionInvoked(key, SelectionKey.OP_READ);
+
+        } finally {
+            closePipe(p);
+        }
+    }
+
+    /**
+     * Test calling select twice. An I/O operation is performed after the
+     * first select so the channel will not be selected by the second select.
+     */
+    public void testRepeatedSelect2() throws Exception {
+        Pipe p = Pipe.open();
+        try (Selector sel = Selector.open()) {
+            Pipe.SourceChannel source = p.source();
+            Pipe.SinkChannel sink = p.sink();
+            source.configureBlocking(false);
+            SelectionKey key = source.register(sel, SelectionKey.OP_READ);
+
+            // write to sink to ensure that the source is readable
+            sink.write(messageBuffer());
+
+            // test that action is invoked
+            testActionInvoked(key, SelectionKey.OP_READ);
+
+            // read all bytes
+            int n;
+            ByteBuffer bb = ByteBuffer.allocate(100);
+            do {
+                n = source.read(bb);
+                bb.clear();
+            } while (n > 0);
+
+            // test that action is not invoked
+            testActionInvoked(key, 0);
+        } finally {
+            closePipe(p);
+        }
+    }
+
+    /**
+     * Test timeout
+     */
+    public void testTimeout() throws Exception {
+        Pipe p = Pipe.open();
+        try (Selector sel = Selector.open()) {
+            Pipe.SourceChannel source = p.source();
+            Pipe.SinkChannel sink = p.sink();
+            source.configureBlocking(false);
+            source.register(sel, SelectionKey.OP_READ);
+            long start = System.currentTimeMillis();
+            int n = sel.select(k -> assertTrue(false), 1000L);
+            long duration = System.currentTimeMillis() - start;
+            assertTrue(n == 0);
+            assertTrue(duration > 500, "select took " + duration + " ms");
+        } finally {
+            closePipe(p);
+        }
+    }
+
+    /**
+     * Test wakeup prior to select
+     */
+    public void testWakeupBeforeSelect() throws Exception {
+        // select(Consumer)
+        try (Selector sel = Selector.open()) {
+            sel.wakeup();
+            int n = sel.select(k -> assertTrue(false));
+            assertTrue(n == 0);
+        }
+
+        // select(Consumer, timeout)
+        try (Selector sel = Selector.open()) {
+            sel.wakeup();
+            long start = System.currentTimeMillis();
+            int n = sel.select(k -> assertTrue(false), 60*1000);
+            long duration = System.currentTimeMillis() - start;
+            assertTrue(n == 0);
+            assertTrue(duration < 5000, "select took " + duration + " ms");
+        }
+    }
+
+    /**
+     * Test wakeup during select
+     */
+    public void testWakeupDuringSelect() throws Exception {
+        // select(Consumer)
+        try (Selector sel = Selector.open()) {
+            scheduleWakeup(sel, 1, SECONDS);
+            int n = sel.select(k -> assertTrue(false));
+            assertTrue(n == 0);
+        }
+
+        // select(Consumer, timeout)
+        try (Selector sel = Selector.open()) {
+            scheduleWakeup(sel, 1, SECONDS);
+            long start = System.currentTimeMillis();
+            int n = sel.select(k -> assertTrue(false), 60*1000);
+            long duration = System.currentTimeMillis() - start;
+            assertTrue(n == 0);
+            assertTrue(duration > 500 && duration < 10*1000,
+                    "select took " + duration + " ms");
+        }
+    }
+
+    /**
+     * Test invoking select with interrupt status set
+     */
+    public void testInterruptBeforeSelect() throws Exception {
+        // select(Consumer)
+        try (Selector sel = Selector.open()) {
+            Thread.currentThread().interrupt();
+            int n = sel.select(k -> assertTrue(false));
+            assertTrue(n == 0);
+            assertTrue(Thread.currentThread().isInterrupted());
+            assertTrue(sel.isOpen());
+        } finally {
+            Thread.currentThread().interrupted();  // clear interrupt status
+        }
+
+        // select(Consumer, timeout)
+        try (Selector sel = Selector.open()) {
+            Thread.currentThread().interrupt();
+            long start = System.currentTimeMillis();
+            int n = sel.select(k -> assertTrue(false), 60*1000);
+            long duration = System.currentTimeMillis() - start;
+            assertTrue(n == 0);
+            assertTrue(duration < 5000, "select took " + duration + " ms");
+            assertTrue(Thread.currentThread().isInterrupted());
+            assertTrue(sel.isOpen());
+        } finally {
+            Thread.currentThread().interrupted();  // clear interrupt status
+        }
+    }
+
+    /**
+     * Test interrupt thread during select
+     */
+    public void testInterruptDuringSelect() throws Exception {
+        // select(Consumer)
+        try (Selector sel = Selector.open()) {
+            scheduleInterrupt(Thread.currentThread(), 1, SECONDS);
+            int n = sel.select(k -> assertTrue(false));
+            assertTrue(n == 0);
+            assertTrue(Thread.currentThread().isInterrupted());
+            assertTrue(sel.isOpen());
+        } finally {
+            Thread.currentThread().interrupted();  // clear interrupt status
+        }
+
+        // select(Consumer, timeout)
+        try (Selector sel = Selector.open()) {
+            scheduleInterrupt(Thread.currentThread(), 1, SECONDS);
+            long start = System.currentTimeMillis();
+            int n = sel.select(k -> assertTrue(false), 60*1000);
+            long duration = System.currentTimeMillis() - start;
+            assertTrue(n == 0);
+            assertTrue(duration > 500 && duration < 5000,
+                    "select took " + duration + " ms");
+            assertTrue(Thread.currentThread().isInterrupted());
+            assertTrue(sel.isOpen());
+        } finally {
+            Thread.currentThread().interrupted();  // clear interrupt status
+        }
+    }
+
+    /**
+     * Test invoking select on a closed selector
+     */
+    @Test(expectedExceptions = ClosedSelectorException.class)
+    public void testClosedSelector1() throws Exception {
+        Selector sel = Selector.open();
+        sel.close();
+        sel.select(k -> assertTrue(false));
+    }
+    @Test(expectedExceptions = ClosedSelectorException.class)
+    public void testClosedSelector2() throws Exception {
+        Selector sel = Selector.open();
+        sel.close();
+        sel.select(k -> assertTrue(false), 1000);
+    }
+    @Test(expectedExceptions = ClosedSelectorException.class)
+    public void testClosedSelector3() throws Exception {
+        Selector sel = Selector.open();
+        sel.close();
+        sel.selectNow(k -> assertTrue(false));
+    }
+
+    /**
+     * Test closing selector while in a selection operation
+     */
+    public void testCloseDuringSelect() throws Exception {
+        // select(Consumer)
+        try (Selector sel = Selector.open()) {
+            scheduleClose(sel, 3, SECONDS);
+            int n = sel.select(k -> assertTrue(false));
+            assertTrue(n == 0);
+            assertFalse(sel.isOpen());
+        }
+
+        // select(Consumer, timeout)
+        try (Selector sel = Selector.open()) {
+            scheduleClose(sel, 3, SECONDS);
+            long start = System.currentTimeMillis();
+            int n = sel.select(k -> assertTrue(false), 60*1000);
+            long duration = System.currentTimeMillis() - start;
+            assertTrue(n == 0);
+            assertTrue(duration > 2000 && duration < 10*1000,
+                    "select took " + duration + " ms");
+            assertFalse(sel.isOpen());
+        }
+    }
+
+    /**
+     * Test action closing selector
+     */
+    @Test(expectedExceptions = ClosedSelectorException.class)
+    public void testActionClosingSelector() throws Exception {
+        Pipe p = Pipe.open();
+        try (Selector sel = Selector.open()) {
+            Pipe.SourceChannel source = p.source();
+            Pipe.SinkChannel sink = p.sink();
+            source.configureBlocking(false);
+            SelectionKey key = source.register(sel, SelectionKey.OP_READ);
+
+            // write to sink to ensure that the source is readable
+            sink.write(messageBuffer());
+
+            // should relay ClosedSelectorException
+            sel.select(k -> {
+                assertTrue(k == key);
+                try {
+                    sel.close();
+                } catch (IOException ioe) { }
+            });
+        } finally {
+            closePipe(p);
+        }
+    }
+
+    /**
+     * Test that the action is invoked while synchronized on the selector and
+     * its selected-key set.
+     */
+    public void testLocks() throws Exception {
+        Pipe p = Pipe.open();
+        try (Selector sel = Selector.open()) {
+            Pipe.SourceChannel source = p.source();
+            Pipe.SinkChannel sink = p.sink();
+            source.configureBlocking(false);
+            SelectionKey key = source.register(sel, SelectionKey.OP_READ);
+
+            // write to sink to ensure that the source is readable
+            sink.write(messageBuffer());
+
+            // select(Consumer)
+            sel.select(k -> {
+                assertTrue(k == key);
+                assertTrue(Thread.holdsLock(sel));
+                assertFalse(Thread.holdsLock(sel.keys()));
+                assertTrue(Thread.holdsLock(sel.selectedKeys()));
+            });
+
+            // select(Consumer, timeout)
+            sel.select(k -> {
+                assertTrue(k == key);
+                assertTrue(Thread.holdsLock(sel));
+                assertFalse(Thread.holdsLock(sel.keys()));
+                assertTrue(Thread.holdsLock(sel.selectedKeys()));
+            }, 1000L);
+
+            // selectNow(Consumer)
+            sel.selectNow(k -> {
+                assertTrue(k == key);
+                assertTrue(Thread.holdsLock(sel));
+                assertFalse(Thread.holdsLock(sel.keys()));
+                assertTrue(Thread.holdsLock(sel.selectedKeys()));
+            });
+        } finally {
+            closePipe(p);
+        }
+    }
+
+    /**
+     * Test that selection operations remove cancelled keys from the selector's
+     * key and selected-key sets.
+     */
+    public void testCancel() throws Exception {
+        Pipe p = Pipe.open();
+        try (Selector sel = Selector.open()) {
+            Pipe.SinkChannel sink = p.sink();
+            Pipe.SourceChannel source = p.source();
+
+            // write to sink to ensure that the source is readable
+            sink.write(messageBuffer());
+
+            sink.configureBlocking(false);
+            source.configureBlocking(false);
+            SelectionKey key1 = sink.register(sel, SelectionKey.OP_WRITE);
+            SelectionKey key2 = source.register(sel, SelectionKey.OP_READ);
+
+            sel.selectNow();
+            assertTrue(sel.keys().contains(key1));
+            assertTrue(sel.keys().contains(key2));
+            assertTrue(sel.selectedKeys().contains(key1));
+            assertTrue(sel.selectedKeys().contains(key2));
+
+            // cancel key1
+            key1.cancel();
+            int n = sel.selectNow(k -> assertTrue(k == key2));
+            assertTrue(n == 1);
+            assertFalse(sel.keys().contains(key1));
+            assertTrue(sel.keys().contains(key2));
+            assertFalse(sel.selectedKeys().contains(key1));
+            assertTrue(sel.selectedKeys().contains(key2));
+
+            // cancel key2
+            key2.cancel();
+            n = sel.selectNow(k -> assertTrue(false));
+            assertTrue(n == 0);
+            assertFalse(sel.keys().contains(key1));
+            assertFalse(sel.keys().contains(key2));
+            assertFalse(sel.selectedKeys().contains(key1));
+            assertFalse(sel.selectedKeys().contains(key2));
+        } finally {
+            closePipe(p);
+        }
+    }
+
+    /**
+     * Test an action invoking select()
+     */
+    public void testReentrantSelect1() throws Exception {
+        Pipe p = Pipe.open();
+        try (Selector sel = Selector.open()) {
+            Pipe.SinkChannel sink = p.sink();
+            Pipe.SourceChannel source = p.source();
+            source.configureBlocking(false);
+            source.register(sel, SelectionKey.OP_READ);
+
+            // write to sink to ensure that the source is readable
+            scheduleWrite(sink, messageBuffer(), 100, MILLISECONDS);
+
+            int n = sel.select(k -> {
+                try {
+                    sel.select();
+                    assertTrue(false);
+                } catch (IOException ioe) {
+                    throw new RuntimeException(ioe);
+                } catch (IllegalStateException expected) {
+                }
+            });
+            assertTrue(n == 1);
+        } finally {
+            closePipe(p);
+        }
+    }
+
+    /**
+     * Test an action invoking selectNow()
+     */
+    public void testReentrantSelect2() throws Exception {
+        Pipe p = Pipe.open();
+        try (Selector sel = Selector.open()) {
+            Pipe.SinkChannel sink = p.sink();
+            Pipe.SourceChannel source = p.source();
+
+            // write to sink to ensure that the source is readable
+            scheduleWrite(sink, messageBuffer(), 100, MILLISECONDS);
+
+            source.configureBlocking(false);
+            source.register(sel, SelectionKey.OP_READ);
+            int n = sel.select(k -> {
+                try {
+                    sel.selectNow();
+                    assertTrue(false);
+                } catch (IOException ioe) {
+                    throw new RuntimeException(ioe);
+                } catch (IllegalStateException expected) {
+                }
+            });
+            assertTrue(n == 1);
+        } finally {
+            closePipe(p);
+        }
+    }
+
+    /**
+     * Test an action invoking select(Consumer)
+     */
+    public void testReentrantSelect3() throws Exception {
+        Pipe p = Pipe.open();
+        try (Selector sel = Selector.open()) {
+            Pipe.SinkChannel sink = p.sink();
+            Pipe.SourceChannel source = p.source();
+
+            // write to sink to ensure that the source is readable
+            scheduleWrite(sink, messageBuffer(), 100, MILLISECONDS);
+
+            source.configureBlocking(false);
+            source.register(sel, SelectionKey.OP_READ);
+            int n = sel.select(k -> {
+                try {
+                    sel.select(x -> assertTrue(false));
+                    assertTrue(false);
+                } catch (IOException ioe) {
+                    throw new RuntimeException(ioe);
+                } catch (IllegalStateException expected) {
+                }
+            });
+            assertTrue(n == 1);
+        } finally {
+            closePipe(p);
+        }
+    }
+
+    /**
+     * Negative timeout
+     */
+    @Test(expectedExceptions = IllegalArgumentException.class)
+    public void testNegativeTimeout() throws Exception {
+        try (Selector sel = Selector.open()) {
+            sel.select(k -> { }, -1L);
+        }
+    }
+
+    /**
+     * Null action
+     */
+    @Test(expectedExceptions = NullPointerException.class)
+    public void testNull1() throws Exception {
+        try (Selector sel = Selector.open()) {
+            sel.select(null);
+        }
+    }
+    @Test(expectedExceptions = NullPointerException.class)
+    public void testNull2() throws Exception {
+        try (Selector sel = Selector.open()) {
+            sel.select(null, 1000);
+        }
+    }
+    @Test(expectedExceptions = NullPointerException.class)
+    public void testNull3() throws Exception {
+        try (Selector sel = Selector.open()) {
+            sel.selectNow(null);
+        }
+    }
+
+
+    // -- support methods ---
+
+    private final ScheduledExecutorService POOL = Executors.newScheduledThreadPool(1);
+
+    @AfterTest
+    void shutdownThreadPool() {
+        POOL.shutdown();
+    }
+
+    void scheduleWakeup(Selector sel, long delay, TimeUnit unit) {
+        POOL.schedule(() -> sel.wakeup(), delay, unit);
+    }
+
+    void scheduleInterrupt(Thread t, long delay, TimeUnit unit) {
+        POOL.schedule(() -> t.interrupt(), delay, unit);
+    }
+
+    void scheduleClose(Closeable c, long delay, TimeUnit unit) {
+        POOL.schedule(() -> {
+            try {
+                c.close();
+            } catch (IOException ioe) {
+                ioe.printStackTrace();
+            }
+        }, delay, unit);
+    }
+
+    void scheduleWrite(WritableByteChannel sink, ByteBuffer buf, long delay, TimeUnit unit) {
+        POOL.schedule(() -> {
+            try {
+                sink.write(buf);
+            } catch (IOException ioe) {
+                ioe.printStackTrace();
+            }
+        }, delay, unit);
+    }
+
+    static void closePipe(Pipe p) {
+        try { p.sink().close(); } catch (IOException ignore) { }
+        try { p.source().close(); } catch (IOException ignore) { }
+    }
+
+    static ByteBuffer messageBuffer() {
+        try {
+            return ByteBuffer.wrap("message".getBytes("UTF-8"));
+        } catch (Exception e) {
+            throw new RuntimeException(e);
+        }
+    }
+}