Reduce CPU usage of gradle run (#49055) (#49102)

The RunTask is responsible for logging output from nodes to the console
and also stays active since we want the cluster to keep running.
However, the implementation of the logging and waiting resulted in a
spin loop that continually polls for data to have been written to one
of the nodes' output files. On my laptop, this causes an idle
invocation of `gradle run` to consume an entire core.

The JDK provides a method to be notified of changes to files through
the use of a WatchService. While a WatchService based implementation
for logging and waiting works, a delay of up to ten seconds is
encountered when running on macOS. This is due to the lack of a native
WatchService implementation that uses kqueue or FSEvents; the current
WatchService implementation in the JDK uses polling with a default
interval of ten seconds. While the interval can be changed
programmatically it is not an acceptable solution due to the need to
access the com.sun.nio.file.SensitivityWatchEventModifier enum, which
is in an internal package.

The change in this commit instead introduces a check to see if any data
was available to read and log. If no data is available in any of the
node output files, the thread sleeps for 100ms. This is enough time to
prevent consuming large amounts of cpu while still providing output to
the console in a timely fashion.

buildSrc/src/main/java/org/elasticsearch/gradle/testclusters/RunTask.java

@@ -7,11 +7,12 @@ import org.gradle.api.tasks.TaskAction;
 import org.gradle.api.tasks.options.Option;
 
 import java.io.BufferedReader;
+import java.io.Closeable;
 import java.io.IOException;
 import java.nio.file.Files;
-import java.util.HashSet;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Map;
-import java.util.Set;
 import java.util.stream.Collectors;
 
 public class RunTask extends DefaultTestClustersTask {
@@ -66,18 +67,55 @@ public class RunTask extends DefaultTestClustersTask {
 
     @TaskAction
     public void runAndWait() throws IOException {
-        Set<BufferedReader> toRead = new HashSet<>();
-        for (ElasticsearchCluster cluster : getClusters()) {
-            for (ElasticsearchNode node : cluster.getNodes()) {
-                toRead.add(Files.newBufferedReader(node.getEsStdoutFile()));
-            }
-        }
-        while (Thread.currentThread().isInterrupted() == false) {
-            for (BufferedReader bufferedReader : toRead) {
-                if (bufferedReader.ready()) {
-                    logger.lifecycle(bufferedReader.readLine());
+        List<BufferedReader> toRead = new ArrayList<>();
+        try {
+            for (ElasticsearchCluster cluster : getClusters()) {
+                for (ElasticsearchNode node : cluster.getNodes()) {
+                    BufferedReader reader = Files.newBufferedReader(node.getEsStdoutFile());
+                    toRead.add(reader);
                 }
             }
+
+            while (Thread.currentThread().isInterrupted() == false) {
+                boolean readData = false;
+                for (BufferedReader bufferedReader : toRead) {
+                    if (bufferedReader.ready()) {
+                        readData = true;
+                        logger.lifecycle(bufferedReader.readLine());
+                    }
+                }
+
+                if (readData == false) {
+                    // no data was ready to be consumed and rather than continuously spinning, pause
+                    // for some time to avoid excessive CPU usage. Ideally we would use the JDK
+                    // WatchService to receive change notifications but the WatchService does not have
+                    // a native MacOS implementation and instead relies upon polling with possible
+                    // delays up to 10s before a notification is received. See JDK-7133447.
+                    try {
+                        Thread.sleep(100L);
+                    } catch (InterruptedException e) {
+                        Thread.currentThread().interrupt();
+                        return;
+                    }
+                }
+            }
+        } finally {
+            Exception thrown = null;
+            for (Closeable closeable : toRead) {
+                try {
+                    closeable.close();
+                } catch (Exception e) {
+                    if (thrown == null) {
+                        thrown = e;
+                    } else {
+                        thrown.addSuppressed(e);
+                    }
+                }
+            }
+
+            if (thrown != null) {
+                logger.debug("exception occurred during close of stdout file readers", thrown);
+            }
         }
     }
 }