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

1 files changed, 77 insertions, 70 deletions

diff --git a/src/test/java/com/google/devtools/build/lib/concurrent/AbstractQueueVisitorTest.java b/src/test/java/com/google/devtools/build/lib/concurrent/AbstractQueueVisitorTest.java
index e31cd94fa9..2c62bec0b0 100644
--- a/src/test/java/com/google/devtools/build/lib/concurrent/AbstractQueueVisitorTest.java
+++ b/src/test/java/com/google/devtools/build/lib/concurrent/AbstractQueueVisitorTest.java
@@ -52,7 +52,7 @@ public class AbstractQueueVisitorTest {
   public void simpleCounter() throws Exception {
     CountingQueueVisitor counter = new CountingQueueVisitor();
     counter.enqueue();
-    counter.work(false);
+    counter.awaitQuiescence(/*interruptWorkers=*/ false);
     assertSame(10, counter.getCount());
   }
 
@@ -64,7 +64,7 @@ public class AbstractQueueVisitorTest {
 
     CountingQueueVisitor counter = new CountingQueueVisitor(executor);
     counter.enqueue();
-    counter.work(false);
+    counter.awaitQuiescence(/*interruptWorkers=*/ false);
     assertSame(10, counter.getCount());
 
     executor.shutdown();
@@ -76,7 +76,7 @@ public class AbstractQueueVisitorTest {
     CountingQueueVisitor counter = new CountingQueueVisitor();
     counter.enqueue();
     counter.enqueue();
-    counter.work(false);
+    counter.awaitQuiescence(/*interruptWorkers=*/ false);
     assertSame(10, counter.getCount());
   }
 
@@ -84,17 +84,18 @@ public class AbstractQueueVisitorTest {
   public void exceptionFromWorkerThread() {
     final RuntimeException myException = new IllegalStateException();
     ConcreteQueueVisitor visitor = new ConcreteQueueVisitor();
-    visitor.enqueue(new Runnable() {
-      @Override
-      public void run() {
-        throw myException;
-      }
-    });
+    visitor.execute(
+        new Runnable() {
+          @Override
+          public void run() {
+            throw myException;
+          }
+        });
 
     try {
       // The exception from the worker thread should be
       // re-thrown from the main thread.
-      visitor.work(false);
+      visitor.awaitQuiescence(/*interruptWorkers=*/ false);
       fail();
     } catch (Exception e) {
       assertSame(myException, e);
@@ -125,7 +126,7 @@ public class AbstractQueueVisitorTest {
     CountingQueueVisitor counter = new CountingQueueVisitor(executor);
     counter.enqueue();
     try {
-      counter.work(false);
+      counter.awaitQuiescence(/*interruptWorkers=*/ false);
       fail();
     } catch (Error expected) {
       assertThat(expected).hasMessage("Could not create thread (fakeout)");
@@ -143,20 +144,22 @@ public class AbstractQueueVisitorTest {
     final ConcreteQueueVisitor visitor = new ConcreteQueueVisitor();
     // Use a latch to make sure the thread gets a chance to start.
     final CountDownLatch threadStarted = new CountDownLatch(1);
-    visitor.enqueue(new Runnable() {
-      @Override
-      public void run() {
-        threadStarted.countDown();
-        assertTrue(Uninterruptibles.awaitUninterruptibly(
-            visitor.getInterruptionLatchForTestingOnly(), 2, TimeUnit.SECONDS));
-        throw THROWABLE;
-      }
-    });
+    visitor.execute(
+        new Runnable() {
+          @Override
+          public void run() {
+            threadStarted.countDown();
+            assertTrue(
+                Uninterruptibles.awaitUninterruptibly(
+                    visitor.getInterruptionLatchForTestingOnly(), 2, TimeUnit.SECONDS));
+            throw THROWABLE;
+          }
+        });
     assertTrue(threadStarted.await(TestUtils.WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS));
     // Interrupt will not be processed until work starts.
     Thread.currentThread().interrupt();
     try {
-      visitor.work(/*interruptWorkers=*/true);
+      visitor.awaitQuiescence(/*interruptWorkers=*/ true);
       fail();
     } catch (Exception e) {
       assertEquals(THROWABLE, e);
@@ -172,18 +175,19 @@ public class AbstractQueueVisitorTest {
     final boolean[] workerThreadCompleted = { false };
     final ConcreteQueueVisitor visitor = new ConcreteQueueVisitor();
 
-    visitor.enqueue(new Runnable() {
-      @Override
-      public void run() {
-        try {
-          latch1.countDown();
-          latch2.await();
-          workerThreadCompleted[0] = true;
-        } catch (InterruptedException e) {
-          // Do not set workerThreadCompleted to true
-        }
-      }
-    });
+    visitor.execute(
+        new Runnable() {
+          @Override
+          public void run() {
+            try {
+              latch1.countDown();
+              latch2.await();
+              workerThreadCompleted[0] = true;
+            } catch (InterruptedException e) {
+              // Do not set workerThreadCompleted to true
+            }
+          }
+        });
 
     TestThread interrupterThread = new TestThread() {
       @Override
@@ -199,7 +203,7 @@ public class AbstractQueueVisitorTest {
     interrupterThread.start();
 
     try {
-      visitor.work(false);
+      visitor.awaitQuiescence(/*interruptWorkers=*/ false);
       fail();
     } catch (InterruptedException e) {
       // Expected.
@@ -228,23 +232,24 @@ public class AbstractQueueVisitorTest {
         ? new ConcreteQueueVisitor()
         : new ConcreteQueueVisitor(executor, true);
 
-    visitor.enqueue(new Runnable() {
-      @Override
-      public void run() {
-        try {
-          latch1.countDown();
-          latch2.await();
-        } catch (InterruptedException e) {
-          workerThreadInterrupted[0] = true;
-        }
-      }
-    });
+    visitor.execute(
+        new Runnable() {
+          @Override
+          public void run() {
+            try {
+              latch1.countDown();
+              latch2.await();
+            } catch (InterruptedException e) {
+              workerThreadInterrupted[0] = true;
+            }
+          }
+        });
 
     latch1.await();
     Thread.currentThread().interrupt();
 
     try {
-      visitor.work(true);
+      visitor.awaitQuiescence(/*interruptWorkers=*/ true);
       fail();
     } catch (InterruptedException e) {
       // Expected.
@@ -307,14 +312,14 @@ public class AbstractQueueVisitorTest {
     Runnable ra = awaitAddAndEnqueueRunnable(interrupt, visitor, latchA, visitedList, "a", r1);
     Runnable rb = awaitAddAndEnqueueRunnable(interrupt, visitor, latchB, visitedList, "b", r2);
 
-    visitor.enqueue(ra);
-    visitor.enqueue(rb);
+    visitor.execute(ra);
+    visitor.execute(rb);
     latchA.await();
     latchB.await();
-    visitor.enqueue(interrupt ? interruptingRunnable(Thread.currentThread()) : throwingRunnable());
+    visitor.execute(interrupt ? interruptingRunnable(Thread.currentThread()) : throwingRunnable());
 
     try {
-      visitor.work(false);
+      visitor.awaitQuiescence(/*interruptWorkers=*/ false);
       fail();
     } catch (Exception e) {
       if (interrupt) {
@@ -357,12 +362,12 @@ public class AbstractQueueVisitorTest {
       }
     };
 
-    visitor.enqueue(r1);
+    visitor.execute(r1);
     latch1.await();
-    visitor.enqueue(throwingRunnable());
+    visitor.execute(throwingRunnable());
 
     try {
-      visitor.work(true);
+      visitor.awaitQuiescence(/*interruptWorkers=*/ true);
       fail();
     } catch (Exception e) {
       assertSame(THROWABLE, e);
@@ -404,13 +409,13 @@ public class AbstractQueueVisitorTest {
         }
       }
     };
-    visitor.enqueue(errorRunnable);
-    visitor.enqueue(sleepRunnable);
+    visitor.execute(errorRunnable);
+    visitor.execute(sleepRunnable);
     Error thrownError = null;
     // Interrupt workers on a critical error. That way we can test that visitor.work doesn't wait
     // for all workers to finish if one of them already had a critical error.
     try {
-      visitor.work(/*interruptWorkers=*/true);
+      visitor.awaitQuiescence(/*interruptWorkers=*/ true);
     } catch (Error e) {
       thrownError = e;
     }
@@ -451,16 +456,17 @@ public class AbstractQueueVisitorTest {
         }
 
         try {
-          assertTrue(interrupt
-                     ? visitor.awaitInterruptionForTestingOnly(1, TimeUnit.MINUTES)
-                     : visitor.getExceptionLatchForTestingOnly().await(1, TimeUnit.MINUTES));
+          assertTrue(
+              interrupt
+                  ? visitor.awaitInterruptionForTestingOnly(1, TimeUnit.MINUTES)
+                  : visitor.getExceptionLatchForTestingOnly().await(1, TimeUnit.MINUTES));
         } catch (InterruptedException e) {
           // Unexpected.
           throw new RuntimeException(e);
         }
         list.add(toAdd);
         if (toEnqueue != null) {
-          visitor.enqueue(toEnqueue);
+          visitor.execute(toEnqueue);
         }
       }
     };
@@ -482,17 +488,18 @@ public class AbstractQueueVisitorTest {
     }
 
     public void enqueue() {
-      super.enqueue(new Runnable() {
-        @Override
-        public void run() {
-          synchronized (lock) {
-            if (theInt < 10) {
-              theInt++;
-              enqueue();
+      super.execute(
+          new Runnable() {
+            @Override
+            public void run() {
+              synchronized (lock) {
+                if (theInt < 10) {
+                  theInt++;
+                  enqueue();
+                }
+              }
             }
-          }
-        }
-      });
+          });
     }
 
     public int getCount() {