Allow AbstractQueueVisitor implementations to introspect on active wo…

…rker count and possibly use that value to make dynamic decisions around scheduling.

--
MOS_MIGRATED_REVID=116351222

src/main/java/com/google/devtools/build/lib/concurrent/AbstractQueueVisitor.java

@@ -409,7 +409,7 @@ public final void awaitQuiescence(boolean interruptWorkers) throws InterruptedEx
   /** Schedules a call. Called in a worker thread if concurrent. */
   @Override
   public final void execute(Runnable runnable) {
-    if (concurrent) {
+    if (runConcurrently()) {
       WrappedRunnable wrappedRunnable = new WrappedRunnable(runnable);
       try {
         // It's impossible for this increment to result in remainingTasks.get <= 0 because
@@ -434,6 +434,22 @@ public final void execute(Runnable runnable) {
     }
   }
 
+  /**
+   * Subclasses may override this to make dynamic decisiouns about whether to run tasks
+   * asynchronously versus in-thread.
+   */
+  protected boolean runConcurrently() {
+    return concurrent;
+  }
+
+  /**
+   * Returns an approximate count of how many threads in the queue visitor's thread pool are
+   * occupied with tasks.
+   */
+  protected final int activeParallelTasks() {
+    return jobs.asMap().size();
+  }
+
   protected void executeRunnable(Runnable runnable) {
     executorService.execute(runnable);
   }

src/test/java/com/google/devtools/build/lib/concurrent/AbstractQueueVisitorTest.java

@@ -55,6 +55,8 @@ public void simpleCounter() throws Exception {
     counter.enqueue();
     counter.awaitQuiescence(/*interruptWorkers=*/ false);
     assertSame(10, counter.getCount());
+    assertSame(0, counter.activeParallelTasks());
+    assertSame(1, counter.getMaxRunningConcurrently());
   }
 
   @Test
@@ -483,6 +485,7 @@ private static class CountingQueueVisitor extends AbstractQueueVisitor {
     private final static String THREAD_NAME = "BlazeTest CountingQueueVisitor";
 
     private int theInt = 0;
+    private int maxRunningConcurrently = 0;
     private final Object lock = new Object();
 
     public CountingQueueVisitor() {
@@ -503,6 +506,10 @@ public void run() {
                   theInt++;
                   enqueue();
                 }
+                int concurrentTasks = activeParallelTasks();
+                if (concurrentTasks > maxRunningConcurrently) {
+                  maxRunningConcurrently = concurrentTasks;
+                }
               }
             }
           });
@@ -511,6 +518,10 @@ public void run() {
     public int getCount() {
       return theInt;
     }
+
+    public int getMaxRunningConcurrently() {
+      return maxRunningConcurrently;
+    }
   }
 
   private static class ConcreteQueueVisitor extends AbstractQueueVisitor {