Update ParallelSkyQueryUtils to use QuiescingExecutor instead of ForkJoinPool

for concurrent visitations.

During BFS visitation of rdeps and rbuildfiles, it uses a centralized pool
(backed by a LinkedBlockingQueue) to store all pending visits, and a
periodically running scheduler to schedule tasks for each pending visit.

--
MOS_MIGRATED_REVID=140398162

3 files changed, 10 insertions, 3 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 01cdca1d3e..45d1c56c15 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
@@ -385,6 +385,11 @@ public class AbstractQueueVisitor implements QuiescingExecutor {
     }
   }
 
+  @Override
+  public long getRemainingTasksCount() {
+    return remainingTasks.get();
+  }
+
   /**
    * Subclasses may override this to make dynamic decisions about whether to run tasks
    * asynchronously versus in-thread.
diff --git a/src/main/java/com/google/devtools/build/lib/concurrent/BlockingStack.java b/src/main/java/com/google/devtools/build/lib/concurrent/BlockingStack.java
index f409bc2d73..93fc3c99f0 100644
--- a/src/main/java/com/google/devtools/build/lib/concurrent/BlockingStack.java
+++ b/src/main/java/com/google/devtools/build/lib/concurrent/BlockingStack.java
@@ -22,11 +22,11 @@ import java.util.concurrent.LinkedBlockingDeque;
 import java.util.concurrent.TimeUnit;
 
 /** A {@link BlockingQueue} with LIFO (last-in-first-out) ordering. */
-class BlockingStack<E> extends AbstractQueue<E> implements BlockingQueue<E> {
+public class BlockingStack<E> extends AbstractQueue<E> implements BlockingQueue<E> {
   // We just restrict to only using the *First methods on the deque, turning it into a stack.
   private final BlockingDeque<E> deque;
 
-  BlockingStack() {
+  public BlockingStack() {
     this.deque = new LinkedBlockingDeque<>();
   }
 
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
index 65718c6205..78bc93dc10 100644
--- a/src/main/java/com/google/devtools/build/lib/concurrent/QuiescingExecutor.java
+++ b/src/main/java/com/google/devtools/build/lib/concurrent/QuiescingExecutor.java
@@ -14,7 +14,6 @@
 package com.google.devtools.build.lib.concurrent;
 
 import com.google.common.annotations.VisibleForTesting;
-
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.Executor;
 
@@ -53,6 +52,9 @@ public interface QuiescingExecutor extends Executor {
    */
   void awaitQuiescence(boolean interruptWorkers) throws InterruptedException;
 
+  /** Return the number of tasks which are not completed (running or waiting to be executed). */
+  long getRemainingTasksCount();
+
   /** Get latch that is released if a task throws an exception. Used only in tests. */
   @VisibleForTesting
   CountDownLatch getExceptionLatchForTestingOnly();