Read ctags output from a separate thread

tuxillo's testing showed that some Ctags
instances seemed to hang upon writing to their
stdin.

java.lang.Process documentation advises that
"failure to promptly ... read the output stream
of the subprocess may cause [it] ... to block, or
even deadlock." Using another thread can help to
mitigate the possible deadlock.

Author: idodeclare

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

@@ -46,8 +46,15 @@
 public class Ctags {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(Ctags.class);
+    private static final long SIGNALWAIT_MS = 5000;
 
+    private final CtagsBuffer buffer = new CtagsBuffer();
+    private final Object syncRoot = new Object();
+    private volatile boolean closing;
+    private volatile boolean signalled;
     private Process ctags;
+    Thread errThread;
+    Thread outThread;
     private OutputStreamWriter ctagsIn;
     private BufferedReader ctagsOut;
     private static final String CTAGS_FILTER_TERMINATOR = "__ctags_done_with_file__";
@@ -77,6 +84,7 @@ public void setCTagsExtraOptionsFile(String CTagsExtraOptionsFile) {
     public void close() throws IOException {
         IOUtils.close(ctagsIn);
         if (ctags != null) {
+            closing = true;
             LOGGER.log(Level.FINE, "Destroying ctags command");
             ctags.destroy();
         }
@@ -292,7 +300,7 @@ private void initialize() throws IOException {
         ctagsIn = new OutputStreamWriter(ctags.getOutputStream());
         ctagsOut = new BufferedReader(new InputStreamReader(ctags.getInputStream()));
 
-        Thread errThread = new Thread(new Runnable() {
+        errThread = new Thread(new Runnable() {
 
             @Override
             public void run() {
@@ -303,6 +311,7 @@ public void run() {
                     while ((s = error.readLine()) != null) {
                         sb.append(s);
                         sb.append('\n');
+                        if (closing) break;
                     }
                 } catch (IOException exp) {
                     LOGGER.log(Level.WARNING, "Got an exception reading ctags error stream: ", exp);
@@ -314,6 +323,34 @@ public void run() {
         });
         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();
     }
 
     public Definitions doCtags(String file) throws IOException {
@@ -338,12 +375,19 @@ public Definitions doCtags(String file) throws IOException {
 
         Definitions ret = null;
         if (file.length() > 0 && !"\n".equals(file)) {
-            //log.fine("doing >" + file + "<");
+            synchronized(syncRoot) {
+                signalled = true;
+                syncRoot.notify();
+            }
+
             ctagsIn.write(file);
             ctagsIn.flush();
-            CtagsReader rdr = new CtagsReader();
-            readTags(rdr);
-            ret = rdr.getDefinitions();
+            try {
+                ret = buffer.take();
+            } catch (InterruptedException e) {
+                LOGGER.log(Level.WARNING,
+                    "Ctags indirect client unexpectedly interrupted");
+            }
         }
 
         return ret;

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

@@ -0,0 +1,70 @@
+/*
+ * CDDL HEADER START
+ *
+ * The contents of this file are subject to the terms of the
+ * Common Development and Distribution License (the "License").
+ * You may not use this file except in compliance with the License.
+ *
+ * See LICENSE.txt included in this distribution for the specific
+ * language governing permissions and limitations under the License.
+ *
+ * When distributing Covered Code, include this CDDL HEADER in each
+ * file and include the License file at LICENSE.txt.
+ * If applicable, add the following below this CDDL HEADER, with the
+ * fields enclosed by brackets "[]" replaced with your own identifying
+ * information: Portions Copyright [yyyy] [name of copyright owner]
+ *
+ * CDDL HEADER END
+ */
+
+/*
+ * Copyright (c) 2017, Oracle and/or its affiliates. All rights reserved.
+ * Portions Copyright (c) 2017, Chris Fraire <[email protected]>.
+ */
+
+package org.opensolaris.opengrok.analysis;
+
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+
+/**
+ * Represents a blocking buffer to allow threads marshaling to and from a
+ * {@link Ctags} process to hand off results in a synchronized manner.
+ */
+public class CtagsBuffer {
+    private final Lock lock = new ReentrantLock();
+    private final Condition notFull  = lock.newCondition();
+    private final Condition notEmpty = lock.newCondition();
+
+    // A buffer of length=1 is all that's needed.
+    private final Definitions[] items = new Definitions[1];
+    private volatile int putptr, takeptr, count;
+
+    public void put(Definitions defs) throws InterruptedException {
+        lock.lock();
+        try {
+            while (count == items.length) notFull.await();
+            items[putptr] = defs;
+            if (++putptr == items.length) putptr = 0;
+            ++count;
+            notEmpty.signal();
+        } finally {
+           lock.unlock();
+        }
+    }
+
+    public Definitions take() throws InterruptedException {
+        lock.lock();
+        try {
+            while (count == 0) notEmpty.await();
+            Definitions defs = items[takeptr];
+            if (++takeptr == items.length) takeptr = 0;
+            --count;
+            notFull.signal();
+            return defs;
+        } finally {
+            lock.unlock();
+        }
+    }
+}