Add method getRdepsUnboundedInUniverseParallel to StremableQueryEnvironment

RELNOTES: None
PiperOrigin-RevId: 167335614

1 files changed, 14 insertions, 12 deletions

diff --git a/src/main/java/com/google/devtools/build/lib/query2/ParallelVisitor.java b/src/main/java/com/google/devtools/build/lib/query2/ParallelVisitor.java
index 0bb9f1b8ab..e81aadd162 100644
--- a/src/main/java/com/google/devtools/build/lib/query2/ParallelVisitor.java
+++ b/src/main/java/com/google/devtools/build/lib/query2/ParallelVisitor.java
@@ -40,11 +40,14 @@ import java.util.concurrent.TimeUnit;
  *
  * <p>The visitor uses an AbstractQueueVisitor backed by a ThreadPoolExecutor with a thread pool NOT
  * part of the global query evaluation pool to avoid starvation.
+ *
+ * @param <T> the type of objects to visit
+ * @param <V> the type of visitation results to process
  */
 @ThreadSafe
-public abstract class ParallelVisitor<T> {
+public abstract class ParallelVisitor<T, V> {
   private final Uniquifier<T> uniquifier;
-  private final Callback<Target> callback;
+  private final Callback<V> callback;
   private final int visitBatchSize;
 
   private final VisitingTaskExecutor executor;
@@ -80,7 +83,7 @@ public abstract class ParallelVisitor<T> {
    *
    * <p>TODO(shazh): Revisit the choice of task target based on real-prod performance.
    */
-  private static final long MIN_PENDING_TASKS = 3 * SkyQueryEnvironment.DEFAULT_THREAD_COUNT;
+  private static final long MIN_PENDING_TASKS = 3L * SkyQueryEnvironment.DEFAULT_THREAD_COUNT;
 
   /**
    * Fail fast on RuntimeExceptions, including {@code RuntimeInterruptedException} and {@code
@@ -114,8 +117,7 @@ public abstract class ParallelVisitor<T> {
           /*workQueue=*/ new BlockingStack<Runnable>(),
           new ThreadFactoryBuilder().setNameFormat("parallel-visitor %d").build());
 
-  protected ParallelVisitor(
-      Uniquifier<T> uniquifier, Callback<Target> callback, int visitBatchSize) {
+  protected ParallelVisitor(Uniquifier<T> uniquifier, Callback<V> callback, int visitBatchSize) {
     this.uniquifier = uniquifier;
     this.callback = callback;
     this.visitBatchSize = visitBatchSize;
@@ -125,7 +127,7 @@ public abstract class ParallelVisitor<T> {
 
   /** Factory for {@link ParallelVisitor} instances. */
   public interface Factory {
-    ParallelVisitor<?> create();
+    ParallelVisitor<?, ?> create();
   }
 
   /**
@@ -154,11 +156,11 @@ public abstract class ParallelVisitor<T> {
   }
 
   /**
-   * Forwards the given {@code keysToUseForResult}'s contribution to the set of {@link Target}s in
-   * the full visitation to the given {@link Callback}.
+   * Forwards the given {@code keysToUseForResult}'s contribution to the set of results in the full
+   * visitation to the given {@link Callback}.
    */
-  protected abstract void processResultantTargets(
-      Iterable<SkyKey> keysToUseForResult, Callback<Target> callback)
+  protected abstract void processPartialResults(
+      Iterable<SkyKey> keysToUseForResult, Callback<V> callback)
       throws QueryException, InterruptedException;
 
   /** Gets the {@link Visit} representing the local visitation of the given {@code values}. */
@@ -227,7 +229,7 @@ public abstract class ParallelVisitor<T> {
 
     @Override
     protected void process() throws QueryException, InterruptedException {
-      processResultantTargets(keysToUseForResult, callback);
+      processPartialResults(keysToUseForResult, callback);
     }
   }
 
@@ -312,7 +314,7 @@ public abstract class ParallelVisitor<T> {
     @Override
     public void process(Iterable<Target> partialResult)
         throws QueryException, InterruptedException {
-      ParallelVisitor<?> visitor = visitorFactory.create();
+      ParallelVisitor<?, ?> visitor = visitorFactory.create();
       // TODO(nharmata): It's not ideal to have an operation like this in #process that blocks on
       // another, potentially expensive computation. Refactor to something like "processAsync".
       visitor.visitAndWaitForCompletion(