Introduce an Extrema aggregator.

RELNOTES: None
PiperOrigin-RevId: 187370833

2 files changed, 21 insertions, 34 deletions

diff --git a/src/main/java/com/google/devtools/build/lib/profiler/BUILD b/src/main/java/com/google/devtools/build/lib/profiler/BUILD
index bb1231e064..e0a8dfb7b6 100644
--- a/src/main/java/com/google/devtools/build/lib/profiler/BUILD
+++ b/src/main/java/com/google/devtools/build/lib/profiler/BUILD
@@ -16,6 +16,7 @@ java_library(
         "//src/main/java/com/google/devtools/build/lib:base-util",
         "//src/main/java/com/google/devtools/build/lib:os_util",
         "//src/main/java/com/google/devtools/build/lib/clock",
+        "//src/main/java/com/google/devtools/build/lib/collect",
         "//src/main/java/com/google/devtools/build/lib/concurrent",
         "//src/main/java/com/google/devtools/build/lib/shell",
         "//src/main/java/com/google/devtools/common/options",
diff --git a/src/main/java/com/google/devtools/build/lib/profiler/Profiler.java b/src/main/java/com/google/devtools/build/lib/profiler/Profiler.java
index d453967041..b58c21d9c2 100644
--- a/src/main/java/com/google/devtools/build/lib/profiler/Profiler.java
+++ b/src/main/java/com/google/devtools/build/lib/profiler/Profiler.java
@@ -20,6 +20,7 @@ import com.google.common.base.Predicate;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Iterables;
 import com.google.devtools.build.lib.clock.Clock;
+import com.google.devtools.build.lib.collect.Extrema;
 import com.google.devtools.build.lib.concurrent.ThreadSafety.ThreadCompatible;
 import com.google.devtools.build.lib.concurrent.ThreadSafety.ThreadSafe;
 import com.google.devtools.build.lib.profiler.PredicateBasedStatRecorder.RecorderAndPredicate;
@@ -36,7 +37,6 @@ import java.util.IdentityHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
-import java.util.PriorityQueue;
 import java.util.Queue;
 import java.util.Timer;
 import java.util.TimerTask;
@@ -332,51 +332,40 @@ public final class Profiler {
   /**
    * Aggregator class that keeps track of the slowest tasks of the specified type.
    *
-   * <p><code>priorityQueues</p> is sharded so that all threads need not compete for the same
-   * lock if they do the same operation at the same time. Access to the individual queues is
-   * synchronized on the queue objects themselves.
+   * <p><code>extremaAggregators</p> is sharded so that all threads need not compete for the same
+   * lock if they do the same operation at the same time. Access to an individual {@link Extrema}
+   * is synchronized on the {@link Extrema} instance itself.
    */
   private static final class SlowestTaskAggregator {
     private static final int SHARDS = 16;
     private final int size;
 
     @SuppressWarnings({"unchecked", "rawtypes"})
-    private final PriorityQueue<SlowTask>[] priorityQueues = new PriorityQueue[SHARDS];
+    private final Extrema<SlowTask>[] extremaAggregators = new Extrema[SHARDS];
 
     SlowestTaskAggregator(int size) {
       this.size = size;
 
       for (int i = 0; i < SHARDS; i++) {
-        priorityQueues[i] = new PriorityQueue<>(size + 1);
+        extremaAggregators[i] = Extrema.max(size);
       }
     }
 
     // @ThreadSafe
     void add(TaskData taskData) {
-      PriorityQueue<SlowTask> queue =
-          priorityQueues[(int) (Thread.currentThread().getId() % SHARDS)];
-      synchronized (queue) {
-        if (queue.size() == size) {
-          // Optimization: check if we are faster than the fastest element. If we are, we would
-          // be the ones to fall off the end of the queue, therefore, we can safely return early.
-          if (queue.peek().getDurationNanos() > taskData.duration) {
-            return;
-          }
-
-          queue.add(new SlowTask(taskData));
-          queue.remove();
-        } else {
-          queue.add(new SlowTask(taskData));
-        }
+      Extrema<SlowTask> extrema =
+          extremaAggregators[(int) (Thread.currentThread().getId() % SHARDS)];
+      synchronized (extrema) {
+        extrema.aggregate(new SlowTask(taskData));
       }
     }
 
     // @ThreadSafe
     void clear() {
       for (int i = 0; i < SHARDS; i++) {
-        PriorityQueue<SlowTask> queue = priorityQueues[i];
-        synchronized (queue) {
-          queue.clear();
+        Extrema<SlowTask> extrema = extremaAggregators[i];
+        synchronized (extrema) {
+          extrema.clear();
         }
       }
     }
@@ -384,19 +373,16 @@ public final class Profiler {
     // @ThreadSafe
     Iterable<SlowTask> getSlowestTasks() {
       // This is slow, but since it only happens during the end of the invocation, it's OK
-      PriorityQueue<SlowTask> merged = new PriorityQueue<>(size * SHARDS);
+      Extrema mergedExtrema = Extrema.max(size * SHARDS);
       for (int i = 0; i < SHARDS; i++) {
-        PriorityQueue<SlowTask> queue = priorityQueues[i];
-        synchronized (queue) {
-          merged.addAll(queue);
+        Extrema<SlowTask> extrema = extremaAggregators[i];
+        synchronized (extrema) {
+          for (SlowTask task : extrema.getExtremeElements()) {
+            mergedExtrema.aggregate(task);
+          }
         }
       }
-
-      while (merged.size() > size) {
-        merged.remove();
-      }
-
-      return merged;
+      return mergedExtrema.getExtremeElements();
     }
   }