Work around buggy MyersDiff by killing threadsv2.1.6-rc1

The JGIt MyersDiff class contains a bug that triggers an infinite loop
on only certain input files.  Most source code is able to be processed
in a reasonable time bound, but some just steal a thread and never
return to the caller.

Implement a custom thread pool that is used to invoke MyersDiff for
the intraline difference data.  If the worker thread doesn't end
within the configured time bound (default of 5 seconds), Gerrit
removes the worker from the pool and tries to kill the worker with the
unsafe Thread.stop() method.

A custom thread pool is used to try and make Thread.stop() safe by
having the amount of data accessed by each worker thread be limited to
only the "constant" inputs supplied by the cache lookup request, and
the result that the thread would return.

If any locks are released early as a result of ThreadDeath going up
the worker thread stack at worst only the incoming or outgoing queues
that are private to that worker will be corrupted.  Since these queues
are private to the worker, and to the thread that is currently
borrowing this worker from the pool (and who is also now killing it),
we can safely ensure that the queues won't be touched after the
Thread.stop() request is made.  This wouldn't be true if we reused any
of the java.util.concurrent thread pool utilities.

This change doesn't actually fix the MyersDiff bug, so we're leaving
issue 487 open.  It does however reduce the impact by trying to abort
the runaway thread, and still show the file with intraline difference
support disabled on just that one file.

Bug: issue 487
Change-Id: I6cbfdd0acc6f7e612a29ed789efe9da591a45273
Signed-off-by: Shawn O. Pearce <sop@google.com>

1 files changed, 176 insertions, 0 deletions

diff --git a/gerrit-server/src/main/java/com/google/gerrit/server/patch/IntraLineLoader.java b/gerrit-server/src/main/java/com/google/gerrit/server/patch/IntraLineLoader.java
index 0d3afdeed4..0ac1af2a18 100644
--- a/gerrit-server/src/main/java/com/google/gerrit/server/patch/IntraLineLoader.java
+++ b/gerrit-server/src/main/java/com/google/gerrit/server/patch/IntraLineLoader.java
@@ -16,24 +16,200 @@
 package com.google.gerrit.server.patch;
 
 import com.google.gerrit.server.cache.EntryCreator;
+import com.google.gerrit.server.config.ConfigUtil;
+import com.google.gerrit.server.config.GerritServerConfig;
+import com.google.inject.Inject;
 
 import org.eclipse.jgit.diff.Edit;
 import org.eclipse.jgit.diff.MyersDiff;
 import org.eclipse.jgit.diff.ReplaceEdit;
+import org.eclipse.jgit.lib.Config;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.regex.Pattern;
 
 class IntraLineLoader extends EntryCreator<IntraLineDiffKey, IntraLineDiff> {
+  private static final Logger log = LoggerFactory
+      .getLogger(IntraLineLoader.class);
+
   private static final Pattern BLANK_LINE_RE = Pattern
       .compile("^[ \\t]*(|[{}]|/\\*\\*?|\\*)[ \\t]*$");
 
   private static final Pattern CONTROL_BLOCK_START_RE = Pattern
       .compile("[{:][ \\t]*$");
 
+  private final BlockingQueue<Worker> workerPool;
+  private final long timeoutMillis;
+
+  @Inject
+  IntraLineLoader(final @GerritServerConfig Config cfg) {
+    final int workers =
+        cfg.getInt("cache", PatchListCacheImpl.INTRA_NAME, "maxIdleWorkers",
+            Runtime.getRuntime().availableProcessors() * 3 / 2);
+    workerPool = new ArrayBlockingQueue<Worker>(workers, true /* fair */);
+
+    timeoutMillis =
+        ConfigUtil.getTimeUnit(cfg, "cache", PatchListCacheImpl.INTRA_NAME,
+            "timeout", TimeUnit.MILLISECONDS.convert(5, TimeUnit.SECONDS),
+            TimeUnit.MILLISECONDS);
+  }
+
   @Override
   public IntraLineDiff createEntry(IntraLineDiffKey key) throws Exception {
+    Worker w = workerPool.poll();
+    if (w == null) {
+      w = new Worker();
+    }
+
+    Worker.Result r = w.computeWithTimeout(key, timeoutMillis);
+
+    if (r == Worker.Result.TIMEOUT) {
+      // Don't keep this thread. We have to murder it unsafely, which
+      // means its unable to be reused in the future. Return back a
+      // null result, indicating the cache cannot load this key.
+      //
+      return new IntraLineDiff(IntraLineDiff.Status.TIMEOUT);
+    }
+
+    if (!workerPool.offer(w)) {
+      // If the idle worker pool is full, terminate this thread.
+      //
+      w.end();
+    }
+
+    if (r.error != null) {
+      // If there was an error computing the result, carry it
+      // up to the caller so the cache knows this key is invalid.
+      //
+      throw r.error;
+    }
+
+    return r.diff;
+  }
+
+  private static class Worker {
+    private static final AtomicInteger count = new AtomicInteger(1);
+
+    private final ArrayBlockingQueue<Input> input;
+    private final ArrayBlockingQueue<Result> result;
+    private final Thread thread;
+
+    Worker() {
+      input = new ArrayBlockingQueue<Input>(1);
+      result = new ArrayBlockingQueue<Result>(1);
+
+      thread = new Thread(new Runnable() {
+        public void run() {
+          workerLoop();
+        }
+      });
+      thread.setName("IntraLineDiff-" + count.getAndIncrement());
+      thread.setDaemon(true);
+      thread.start();
+    }
+
+    Result computeWithTimeout(IntraLineDiffKey key, long timeoutMillis)
+        throws Exception {
+      if (!input.offer(new Input(key))) {
+        log.error("Cannot enqueue task to thread " + thread.getName());
+        return null;
+      }
+
+      Result r = result.poll(timeoutMillis, TimeUnit.MILLISECONDS);
+      if (r != null) {
+        return r;
+      } else {
+        log.warn(timeoutMillis + " ms timeout reached for IntraLineDiff"
+            + " in project " + key.getProject().get() //
+            + " on commit " + key.getCommit().name() //
+            + " for path " + key.getPath() //
+            + " comparing " + key.getBlobA().name() //
+            + ".." + key.getBlobB().name() //
+            + ".  Killing " + thread.getName());
+        try {
+          thread.stop();
+        } catch (Throwable error) {
+          // Ignore any reason the thread won't stop.
+          log.error("Cannot stop runaway thread " + thread.getName(), error);
+        }
+        return Result.TIMEOUT;
+      }
+    }
+
+    void end() {
+      if (!input.offer(Input.END_THREAD)) {
+        log.error("Cannot gracefully stop thread " + thread.getName());
+      }
+    }
+
+    private void workerLoop() {
+      try {
+        for (;;) {
+          Input in;
+          try {
+            in = input.take();
+          } catch (InterruptedException e) {
+            log.error("Unexpected interrupt on " + thread.getName());
+            continue;
+          }
+
+          if (in == Input.END_THREAD) {
+            return;
+          }
+
+          Result r;
+          try {
+            r = new Result(IntraLineLoader.compute(in.key));
+          } catch (Exception error) {
+            r = new Result(error);
+          }
+
+          if (!result.offer(r)) {
+            log.error("Cannot return result from " + thread.getName());
+          }
+        }
+      } catch (ThreadDeath iHaveBeenShot) {
+        // Handle thread death by gracefully returning to the caller,
+        // allowing the thread to be destroyed.
+      }
+    }
+
+    private static class Input {
+      static final Input END_THREAD = new Input(null);
+
+      final IntraLineDiffKey key;
+
+      Input(IntraLineDiffKey key) {
+        this.key = key;
+      }
+    }
+
+    static class Result {
+      static final Result TIMEOUT = new Result((IntraLineDiff) null);
+
+      final IntraLineDiff diff;
+      final Exception error;
+
+      Result(IntraLineDiff diff) {
+        this.diff = diff;
+        this.error = null;
+      }
+
+      Result(Exception error) {
+        this.diff = null;
+        this.error = error;
+      }
+    }
+  }
+
+  private static IntraLineDiff compute(IntraLineDiffKey key) throws Exception {
     List<Edit> edits = new ArrayList<Edit>(key.getEdits());
     Text aContent = key.getTextA();
     Text bContent = key.getTextB();