Introduce QuiescingExecutor

This interface (mostly) encapsulates what the ValueVisitor expects
from the AbstractQueueVisitor class it currently inherits from. This
makes it easier for a future CL to change ValueVisitor's strategy of
code reuse from inheritance to composition.

RELNOTES:

--
MOS_MIGRATED_REVID=106728863

2 files changed, 63 insertions, 37 deletions

diff --git a/src/main/java/com/google/devtools/build/lib/concurrent/AbstractQueueVisitor.java b/src/main/java/com/google/devtools/build/lib/concurrent/AbstractQueueVisitor.java
index 51f054746c..9f49798bde 100644
--- a/src/main/java/com/google/devtools/build/lib/concurrent/AbstractQueueVisitor.java
+++ b/src/main/java/com/google/devtools/build/lib/concurrent/AbstractQueueVisitor.java
@@ -32,27 +32,10 @@ import java.util.logging.Level;
 import java.util.logging.Logger;
 
 /**
- * AbstractQueueVisitor is a wrapper around {@link ExecutorService} which delays service shutdown
- * until entire visitation is complete. This is useful for cases in which worker tasks may submit
- * additional tasks.
- *
- * <p>Consider the following example:
- * <pre>
- *   ThreadPoolExecutor executor = <...>
- *   executor.submit(myRunnableTask);
- *   executor.shutdown();
- *   executor.awaitTermination();
- * </pre>
- *
- * <p>This won't work properly if {@code myRunnableTask} submits additional
- * tasks to the executor, because it may already have shut down
- * by that point.
- *
- * <p>AbstractQueueVisitor supports interruption. If the main thread is
- * interrupted, tasks will no longer be added to the queue, and the
- * {@link #work(boolean)} method will throw {@link InterruptedException}.
+ * AbstractQueueVisitor is a {@link QuiescingExecutor} implementation that wraps an {@link
+ * ExecutorService}.
  */
-public class AbstractQueueVisitor {
+public class AbstractQueueVisitor implements QuiescingExecutor {
 
   /**
    * Default factory function for constructing {@link ThreadPoolExecutor}s. The {@link
@@ -114,7 +97,7 @@ public class AbstractQueueVisitor {
 
   /**
    * If {@link #concurrent} is {@code true}, then this is a counter of the number of {@link
-   * Runnable}s {@link #enqueue}-d that have not finished evaluation.
+   * Runnable}s {@link #execute}-d that have not finished evaluation.
    */
   private final AtomicLong remainingTasks = new AtomicLong(0);
 
@@ -124,13 +107,13 @@ public class AbstractQueueVisitor {
 
   /**
    * The {@link ExecutorService}. If !{@code concurrent}, always {@code null}. Created lazily on
-   * first call to {@link #enqueue(Runnable)}, and removed after call to {@link #work(boolean)}.
+   * first call to {@link #execute(Runnable)}, and removed after call to {@link #awaitQuiescence}.
    */
   private final ExecutorService pool;
 
   /**
    * Flag used to record when the main thread (the thread which called
-   * {@link #work(boolean)}) is interrupted.
+   * {@link #awaitQuiescence(boolean)}) is interrupted.
    *
    * When this is true, adding tasks to the thread pool will
    * fail quietly as a part of the process of shutting down the
@@ -341,19 +324,9 @@ public class AbstractQueueVisitor {
     this(true, parallelism, keepAlive, units, false, true, poolName, EXECUTOR_FACTORY);
   }
 
-  /**
-   * Executes all tasks on the queue, and optionally shuts the pool down and deletes it.
-   *
-   * <p>Throws (the same) unchecked exception if any worker thread failed unexpectedly. If the pool
-   * is interrupted and a worker also throws an unchecked exception, the unchecked exception is
-   * rethrown, since it may indicate a programming bug. If callers handle the unchecked exception,
-   * they may check the interrupted bit to see if the pool was interrupted.
-   *
-   * @param interruptWorkers if true, interrupt worker threads if main thread gets an interrupt or
-   *        if a worker throws a critical error (see {@link #classifyError(Throwable)}. If false,
-   *        just wait for them to terminate normally.
-   */
-  protected final void work(boolean interruptWorkers) throws InterruptedException {
+
+  @Override
+  public final void awaitQuiescence(boolean interruptWorkers) throws InterruptedException {
     if (concurrent) {
       awaitTermination(interruptWorkers);
     } else {
@@ -367,7 +340,8 @@ public class AbstractQueueVisitor {
    * Schedules a call.
    * Called in a worker thread if concurrent.
    */
-  protected final void enqueue(Runnable runnable) {
+  @Override
+  public final void execute(Runnable runnable) {
     if (concurrent) {
       AtomicBoolean ranTask = new AtomicBoolean(false);
       try {
diff --git a/src/main/java/com/google/devtools/build/lib/concurrent/QuiescingExecutor.java b/src/main/java/com/google/devtools/build/lib/concurrent/QuiescingExecutor.java
new file mode 100644
index 0000000000..721c47a714
--- /dev/null
+++ b/src/main/java/com/google/devtools/build/lib/concurrent/QuiescingExecutor.java
@@ -0,0 +1,52 @@
+// Copyright 2015 The Bazel Authors. All rights reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//    http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+package com.google.devtools.build.lib.concurrent;
+
+import java.util.concurrent.Executor;
+
+/**
+ * QuiescingExecutor is an {@link Executor} which supports waiting until all submitted tasks are
+ * complete. This is useful when tasks may submit additional tasks.
+ *
+ * <p>Consider the following example:
+ * <pre>
+ *   ThreadPoolExecutor executor = <...>
+ *   executor.submit(myRunnableTask);
+ *   executor.shutdown();
+ *   executor.awaitTermination();
+ * </pre>
+ *
+ * <p>This won't work properly if {@code myRunnableTask} submits additional tasks to the
+ * executor, because it may already have shut down by that point.
+ *
+ * <p>QuiescingExecutor supports interruption. If the main thread is interrupted, tasks will no
+ * longer be started, and the {@link #awaitQuiescence} method will throw {@link
+ * InterruptedException}.
+ */
+public interface QuiescingExecutor extends Executor {
+
+  /**
+   * Waits for all tasks to complete. If the {@link QuiescingExecutor} owns its own {@link
+   * java.util.concurrent.ExecutorService}, the service will also be shutdown.
+   *
+   * <p>Throws (the same) unchecked exception if any worker thread failed unexpectedly. If the main
+   * thread is interrupted and a worker also throws an unchecked exception, the unchecked exception
+   * is rethrown, since it may indicate a programming bug. If callers handle the unchecked
+   * exception, they may check the interrupted bit to see if the pool was interrupted.
+   *
+   * @param interruptWorkers if true, interrupt worker threads if main thread gets an interrupt.
+   *                         If false, just wait for them to terminate normally.
+   */
+  void awaitQuiescence(boolean interruptWorkers) throws InterruptedException;
+}