Recognize InterruptedException, and ensure Ctags does block indefinitely

Author: idodeclare

src/org/opensolaris/opengrok/analysis/AnalyzerGuru.java

@@ -383,10 +383,11 @@ public static FileAnalyzer getAnalyzer(InputStream in, String file) throws IOExc
      * @param fa The analyzer to use on the file
      * @param xrefOut Where to write the xref (possibly {@code null})
      * @throws IOException If an exception occurs while collecting the data
+     * @throws InterruptedException if a timeout occurs
      */
     public void populateDocument(Document doc, File file, String path,
-            FileAnalyzer fa, Writer xrefOut)
-            throws IOException {
+        FileAnalyzer fa, Writer xrefOut) throws IOException,
+            InterruptedException {
 
         String date = DateTools.timeToString(file.lastModified(),
                 DateTools.Resolution.MILLISECOND);

src/org/opensolaris/opengrok/analysis/Ctags.java

@@ -32,6 +32,9 @@
 import java.io.StringReader;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 import org.opensolaris.opengrok.configuration.RuntimeEnvironment;
@@ -48,6 +51,16 @@ public class Ctags {
     private static final Logger LOGGER = LoggerFactory.getLogger(Ctags.class);
     private static final long SIGNALWAIT_MS = 5000;
 
+    /**
+     * Wait up to 60 seconds for ctags to run, which is a very long time for a
+     * single run but is really intended for detecting blocked ctags processes
+     * as happened with @tuxillo's testing of
+     * <a href="https://github.com/oracle/opengrok/pull/1936">
+     * Feature/deferred_ops</a>.
+     */
+    private static final long CTAGSWAIT_S = 60;
+
+    private final ScheduledThreadPoolExecutor schedExecutor;
     private final CtagsBuffer buffer = new CtagsBuffer();
     private final Object syncRoot = new Object();
     private volatile boolean closing;
@@ -65,6 +78,19 @@ public class Ctags {
 
     private boolean junit_testing = false;
 
+    /**
+     * Initializes an instance using the specified, required executor which
+     * will be used to track ctags timeouts to avoid blockages.
+     * <p>(The {@code schedExecutor} is not owned by {@link Ctags}.)
+     * @param schedExecutor required, defined instance
+     */
+    public Ctags(ScheduledThreadPoolExecutor schedExecutor) {
+        if (schedExecutor == null) {
+            throw new IllegalArgumentException("`schedExecutor' is null");
+        }
+        this.schedExecutor = schedExecutor;
+    }
+
     /**
      * Gets a value indicating if a subprocess of ctags was started and either:
      * 1) it is not alive; or 2) any necessary supporting thread is not alive.
@@ -311,56 +337,33 @@ private void initialize() throws IOException {
 
             @Override
             public void run() {
-                StringBuilder sb = new StringBuilder();
                 try (BufferedReader error = new BufferedReader(
                         new InputStreamReader(ctags.getErrorStream()))) {
                     String s;
                     while ((s = error.readLine()) != null) {
-                        sb.append(s);
-                        sb.append('\n');
+                        if (s.length() > 0) {
+                            LOGGER.log(Level.WARNING, "Error from ctags: {0}",
+                                s);
+                        }
                         if (closing) break;
                     }
                 } catch (IOException exp) {
                     LOGGER.log(Level.WARNING, "Got an exception reading ctags error stream: ", exp);
                 }
-                if (sb.length() > 0) {
-                    LOGGER.log(Level.WARNING, "Error from ctags: {0}", sb.toString());
-                }
             }
         });
         errThread.setDaemon(true);
         errThread.start();
 
-        outThread = new Thread(() -> {
-            while (!closing) {
-                synchronized(syncRoot) {
-                    while (!signalled && !closing) {
-                        try {
-                            syncRoot.wait(SIGNALWAIT_MS);
-                        } catch (InterruptedException e) {
-                            LOGGER.log(Level.WARNING,
-                                "Ctags direct client unexpectedly interrupted");
-                        }
-                    }
-                    signalled = false;
-                }
-                if (closing) return;
-
-                CtagsReader rdr = new CtagsReader();
-                readTags(rdr);
-                Definitions defs = rdr.getDefinitions();
-                try {
-                    buffer.put(defs);
-                } catch (InterruptedException ex) {
-                    // ignore
-                }
-            }
-        });
-        outThread.setDaemon(true);
-        outThread.start();
+        if (outThread == null || !outThread.isAlive()) {
+            outThread = new Thread(() -> outThreadStart());
+            outThread.setDaemon(true);
+            outThread.start();
+        }
     }
 
-    public Definitions doCtags(String file) throws IOException {
+    public Definitions doCtags(String file) throws IOException,
+            InterruptedException {
         boolean ctagsRunning = false;
         if (ctags != null) {
             try {
@@ -380,23 +383,36 @@ public Definitions doCtags(String file) throws IOException {
             initialize();
         }
 
-        Definitions ret = null;
-        if (file.length() > 0 && !"\n".equals(file)) {
-            synchronized(syncRoot) {
-                signalled = true;
-                syncRoot.notify();
-            }
+        if (file.length() < 1 || "\n".equals(file)) return null;
+
+        synchronized(syncRoot) {
+            signalled = true;
+            syncRoot.notify();
+        }
 
+        Thread clientThread = Thread.currentThread();
+        ScheduledFuture futureTimeout = schedExecutor.schedule(() -> {
+            clientThread.interrupt();
+            outThread.interrupt();
+            LOGGER.log(Level.WARNING, "Ctags futureTimeout executed.");
+        }, CTAGSWAIT_S, TimeUnit.SECONDS);
+
+        Definitions ret;
+        try {
             ctagsIn.write(file);
+            if (Thread.interrupted()) throw new InterruptedException("write()");
             ctagsIn.flush();
-            try {
-                ret = buffer.take();
-            } catch (InterruptedException e) {
-                LOGGER.log(Level.WARNING,
-                    "Ctags indirect client unexpectedly interrupted");
-            }
+            if (Thread.interrupted()) throw new InterruptedException("flush()");
+            ret = buffer.take();
+        } finally {
+            futureTimeout.cancel(false);
         }
 
+        /*
+         * If the timeout could not be canceled in time, then act as if no
+         * results were obtained.
+         */
+        if (Thread.interrupted()) throw new InterruptedException("late");
         return ret;
     }
 
@@ -442,15 +458,23 @@ public void destroy() {
         };
 
         CtagsReader rdr = new CtagsReader();
-        readTags(rdr);
+        try {
+            readTags(rdr);
+        } catch (InterruptedException ex) {
+            LOGGER.log(Level.SEVERE, "readTags() test", ex);
+        }
         Definitions ret = rdr.getDefinitions();
         return ret;
     }
 
-    private void readTags(CtagsReader reader) {
+    private void readTags(CtagsReader reader) throws InterruptedException {
         try {
             do {
                 String tagLine = ctagsOut.readLine();
+                if (Thread.interrupted()) {
+                    throw new InterruptedException("readLine()");
+                }
+
                 //log.fine("Tagline:-->" + tagLine+"<----ONELINE");
                 if (tagLine == null) {
                     if (!junit_testing) {
@@ -480,9 +504,52 @@ private void readTags(CtagsReader reader) {
 
                 reader.readLine(tagLine);
             } while (true);
+        } catch (InterruptedException e) {
+            throw e;
         } catch (Exception e) {
             LOGGER.log(Level.WARNING, "CTags parsing problem: ", e);
         }
         LOGGER.severe("CTag reader cycle was interrupted!");
     }
+
+    private void outThreadStart() {
+        while (!closing) {
+            synchronized(syncRoot) {
+                while (!signalled && !closing) {
+                    try {
+                        syncRoot.wait(SIGNALWAIT_MS);
+                    } catch (InterruptedException e) {
+                        LOGGER.log(Level.WARNING,
+                            "Ctags outThread unexpectedly interrupted--{0}",
+                            e.getMessage());
+                        /*
+                        * Terminating this thread will cause this Ctags
+                        * instance, which appears to be blocked, to be
+                        * reported as closed by isClosed().
+                        */
+                        return;
+                    }
+                }
+                signalled = false;
+            }
+            if (closing) return;
+            
+            CtagsReader rdr = new CtagsReader();
+            try {
+                readTags(rdr);
+                Definitions defs = rdr.getDefinitions();
+                buffer.put(defs);
+            } catch (InterruptedException ex) {
+                LOGGER.log(Level.WARNING,
+                    "Ctags outThread unexpectedly interrupted--{0}",
+                    ex.getMessage());
+                /*
+                * Terminating this thread will cause this Ctags instance,
+                * which appears to be blocked, to be reported as closed by
+                * isClosed().
+                */
+                return;
+            }
+        }
+    }
 }

src/org/opensolaris/opengrok/analysis/FileAnalyzer.java

@@ -179,8 +179,10 @@ public String getFileTypeName() {
      * @param src the input data source
      * @param xrefOut where to write the xref (may be {@code null})
      * @throws IOException if any I/O error
+     * @throws InterruptedException if a timeout occurs
      */
-    public void analyze(Document doc, StreamSource src, Writer xrefOut) throws IOException {
+    public void analyze(Document doc, StreamSource src, Writer xrefOut)
+            throws IOException, InterruptedException {
         // not used
     }
 

src/org/opensolaris/opengrok/analysis/archive/BZip2Analyzer.java

@@ -19,6 +19,7 @@
 
 /*
  * Copyright (c) 2005, 2017, Oracle and/or its affiliates. All rights reserved.
+ * Portions Copyright (c) 2017, Chris Fraire <[email protected]>.
  */
 package org.opensolaris.opengrok.analysis.archive;
 
@@ -57,7 +58,8 @@ protected BZip2Analyzer(FileAnalyzerFactory factory) {
     private FileAnalyzer fa;
 
     @Override
-    public void analyze(Document doc, StreamSource src, Writer xrefOut) throws IOException {
+    public void analyze(Document doc, StreamSource src, Writer xrefOut)
+            throws IOException, InterruptedException {
         StreamSource bzSrc = wrap(src);
         String path = doc.get("path");
         if (path != null

src/org/opensolaris/opengrok/analysis/archive/GZIPAnalyzer.java

@@ -19,6 +19,7 @@
 
 /*
  * Copyright (c) 2005, 2017, Oracle and/or its affiliates. All rights reserved.
+ * Portions Copyright (c) 2017, Chris Fraire <[email protected]>.
  */
 package org.opensolaris.opengrok.analysis.archive;
 
@@ -63,7 +64,8 @@ protected GZIPAnalyzer(FileAnalyzerFactory factory) {
     private FileAnalyzer fa;
 
     @Override
-    public void analyze(Document doc, StreamSource src, Writer xrefOut) throws IOException {
+    public void analyze(Document doc, StreamSource src, Writer xrefOut)
+            throws IOException, InterruptedException {
         StreamSource gzSrc = wrap(src);
         String path = doc.get("path");
         if (path != null

src/org/opensolaris/opengrok/analysis/plain/AbstractSourceCodeAnalyzer.java

@@ -19,6 +19,7 @@
 
 /*
  * Copyright (c) 2012, 2017 Oracle and/or its affiliates. All rights reserved.
+ * Portions Copyright (c) 2017, Chris Fraire <[email protected]>.
  */
 package org.opensolaris.opengrok.analysis.plain;
 
@@ -66,9 +67,4 @@ protected AbstractSourceCodeAnalyzer(FileAnalyzerFactory factory) {
      */
     @Override
     protected abstract JFlexXref newXref(Reader reader);
-
-    @Override
-    public void analyze(Document doc, StreamSource src, Writer xrefOut) throws IOException {
-        super.analyze(doc, src, xrefOut);
-    }        
 }

src/org/opensolaris/opengrok/analysis/plain/PlainAnalyzer.java

@@ -73,7 +73,8 @@ protected Reader getReader(InputStream stream) throws IOException {
     }
 
     @Override
-    public void analyze(Document doc, StreamSource src, Writer xrefOut) throws IOException {
+    public void analyze(Document doc, StreamSource src, Writer xrefOut)
+            throws IOException, InterruptedException {
         Definitions defs = null;
 
         doc.add(new TextField(QueryBuilder.FULL, getReader(src.getStream())));