8200256: java/nio/channels/AsynchronousChannelGroup/Basic.java fails intermittently

Reviewed-by: chegar

make/mapfiles/libnio/mapfile-linux

@@ -77,6 +77,7 @@ SUNWprivate_1.1 {
 		Java_sun_nio_ch_InheritedChannel_soType0;
                 Java_sun_nio_ch_IOUtil_configureBlocking;
                 Java_sun_nio_ch_IOUtil_drain;
+                Java_sun_nio_ch_IOUtil_drain1;
                 Java_sun_nio_ch_IOUtil_fdVal;
 		Java_sun_nio_ch_IOUtil_fdLimit;
                 Java_sun_nio_ch_IOUtil_initIDs;

make/mapfiles/libnio/mapfile-macosx

@@ -70,6 +70,7 @@ SUNWprivate_1.1 {
 		Java_sun_nio_ch_InheritedChannel_soType0;
                 Java_sun_nio_ch_IOUtil_configureBlocking;
                 Java_sun_nio_ch_IOUtil_drain;
+                Java_sun_nio_ch_IOUtil_drain1;
                 Java_sun_nio_ch_IOUtil_fdVal;
                 Java_sun_nio_ch_IOUtil_fdLimit;
                 Java_sun_nio_ch_IOUtil_initIDs;

make/mapfiles/libnio/mapfile-solaris

@@ -75,6 +75,7 @@ SUNWprivate_1.1 {
 		Java_sun_nio_ch_InheritedChannel_soType0;
                 Java_sun_nio_ch_IOUtil_configureBlocking;
                 Java_sun_nio_ch_IOUtil_drain;
+                Java_sun_nio_ch_IOUtil_drain1;
 		Java_sun_nio_ch_IOUtil_fdLimit;
                 Java_sun_nio_ch_IOUtil_fdVal;
                 Java_sun_nio_ch_IOUtil_initIDs;

src/java.base/linux/classes/sun/nio/ch/EPollPort.java

@@ -164,7 +164,7 @@ final class EPollPort
         if (nThreads == 0) {
             implClose();
         } else {
-            // send interrupt to each thread
+            // send wakeup to each thread
             while (nThreads-- > 0) {
                 wakeup();
             }
@@ -182,11 +182,11 @@ final class EPollPort
             throw new AssertionError();     // should not happen
     }
 
-    /*
+    /**
      * Task to process events from epoll and dispatch to the channel's
      * onEvent handler.
      *
-     * Events are retreived from epoll in batch and offered to a BlockingQueue
+     * Events are retrieved from epoll in batch and offered to a BlockingQueue
      * where they are consumed by handler threads. A special "NEED_TO_POLL"
      * event is used to signal one consumer to re-poll when all events have
      * been consumed.
@@ -200,7 +200,7 @@ final class EPollPort
                         n = EPoll.wait(epfd, address, MAX_EPOLL_EVENTS, -1);
                     } while (n == IOStatus.INTERRUPTED);
 
-                    /*
+                    /**
                      * 'n' events have been read. Here we map them to their
                      * corresponding channel in batch and queue n-1 so that
                      * they can be handled by other handler threads. The last
@@ -215,8 +215,13 @@ final class EPollPort
                             // wakeup
                             if (fd == sp[0]) {
                                 if (wakeupCount.decrementAndGet() == 0) {
-                                    // no more wakeups so drain pipe
+                                    // consume one wakeup byte, never more as this
-                                    IOUtil.drain(sp[0]);
+                                    // would interfere with shutdown when there is
+                                    // a wakeup byte queued to wake each thread
+                                    int nread;
+                                    do {
+                                        nread = IOUtil.drain1(sp[0]);
+                                    } while (nread == IOStatus.INTERRUPTED);
                                 }
 
                                 // queue special event if there are more events

src/java.base/macosx/classes/sun/nio/ch/KQueuePort.java

@@ -160,7 +160,7 @@ final class KQueuePort
         if (nThreads == 0) {
             implClose();
         } else {
-            // send interrupt to each thread
+            // send wakeup to each thread
             while (nThreads-- > 0) {
                 wakeup();
             }
@@ -182,11 +182,11 @@ final class KQueuePort
             throw new InternalError("kevent failed: " + err);  // should not happen
     }
 
-    /*
+    /**
      * Task to process events from kqueue and dispatch to the channel's
      * onEvent handler.
      *
-     * Events are retreived from kqueue in batch and offered to a BlockingQueue
+     * Events are retrieved from kqueue in batch and offered to a BlockingQueue
      * where they are consumed by handler threads. A special "NEED_TO_POLL"
      * event is used to signal one consumer to re-poll when all events have
      * been consumed.
@@ -200,7 +200,7 @@ final class KQueuePort
                         n = KQueue.poll(kqfd, address, MAX_KEVENTS_TO_POLL, -1L);
                     } while (n == IOStatus.INTERRUPTED);
 
-                    /*
+                    /**
                      * 'n' events have been read. Here we map them to their
                      * corresponding channel in batch and queue n-1 so that
                      * they can be handled by other handler threads. The last
@@ -215,8 +215,13 @@ final class KQueuePort
                             // wakeup
                             if (fd == sp[0]) {
                                 if (wakeupCount.decrementAndGet() == 0) {
-                                    // no more wakeups so drain pipe
+                                    // consume one wakeup byte, never more as this
-                                    IOUtil.drain(sp[0]);
+                                    // would interfere with shutdown when there is
+                                    // a wakeup byte queued to wake each thread
+                                    int nread;
+                                    do {
+                                        nread = IOUtil.drain1(sp[0]);
+                                    } while (nread == IOStatus.INTERRUPTED);
                                 }
 
                                 // queue special event if there are more events

src/java.base/share/classes/sun/nio/ch/AsynchronousChannelGroupImpl.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2008, 2017, Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2008, 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
@@ -60,8 +60,8 @@ abstract class AsynchronousChannelGroupImpl
     // associated Executor for timeouts
     private ScheduledThreadPoolExecutor timeoutExecutor;
 
-    // task queue for when using a fixed thread pool. In that case, thread
+    // task queue for when using a fixed thread pool. In that case, a thread
-    // waiting on I/O events must be awokon to poll tasks from this queue.
+    // waiting on I/O events must be awoken to poll tasks from this queue.
     private final Queue<Runnable> taskQueue;
 
     // group shutdown

src/java.base/share/classes/sun/nio/ch/IOUtil.java

@@ -401,8 +401,17 @@ public class IOUtil {
 
     static native int write1(int fd, byte b) throws IOException;
 
+    /**
+     * Read and discard all bytes.
+     */
     static native boolean drain(int fd) throws IOException;
 
+    /**
+     * Read and discard at most one byte
+     * @return the number of bytes read or IOS_INTERRUPTED
+     */
+    static native int drain1(int fd) throws IOException;
+
     public static native void configureBlocking(FileDescriptor fd,
                                                 boolean blocking)
         throws IOException;

src/java.base/unix/native/libnio/ch/IOUtil.c

@@ -104,7 +104,6 @@ Java_sun_nio_ch_IOUtil_makePipe(JNIEnv *env, jobject this, jboolean blocking)
     return ((jlong) fd[0] << 32) | (jlong) fd[1];
 }
 
-
 JNIEXPORT jint JNICALL
 Java_sun_nio_ch_IOUtil_write1(JNIEnv *env, jclass cl, jint fd, jbyte b)
 {
@@ -112,7 +111,6 @@ Java_sun_nio_ch_IOUtil_write1(JNIEnv *env, jclass cl, jint fd, jbyte b)
     return convertReturnVal(env, write(fd, &c, 1), JNI_FALSE);
 }
 
-
 JNIEXPORT jboolean JNICALL
 Java_sun_nio_ch_IOUtil_drain(JNIEnv *env, jclass cl, jint fd)
 {
@@ -130,6 +128,26 @@ Java_sun_nio_ch_IOUtil_drain(JNIEnv *env, jclass cl, jint fd)
     }
 }
 
+JNIEXPORT jint JNICALL
+Java_sun_nio_ch_IOUtil_drain1(JNIEnv *env, jclass cl, jint fd)
+{
+    int res;
+    char buf[1];
+
+    res = read(fd, buf, 1);
+    if (res < 0) {
+        if (errno == EAGAIN) {
+            res = 0;
+        } else if (errno == EINTR) {
+            return IOS_INTERRUPTED;
+        } else {
+            JNU_ThrowIOExceptionWithLastError(env, "read");
+            return IOS_THROWN;
+        }
+    }
+    return res;
+}
+
 JNIEXPORT jint JNICALL
 Java_sun_nio_ch_IOUtil_fdLimit(JNIEnv *env, jclass this)
 {