Optimize SlidingTimeWindowMovingAverages sumBuckets (#4936)

`SlidingTimeWindowMovingAverages` `sumBuckets()` method could be optimized to perform indexed list access and remove allocations as it currently allocates a `LongAdder` as well as one or two streams. If one is using the 1, 5, or 15 minute rates on hot paths, these can be unnecessary expensive allocations as well as less optimized computations.

We can avoid the allocations completely and accumulate the sum directly to a `long` via optimized direct indexed list access.


[MovingAverageBenchmarks](https://github.com/palantir/tritium/blob/davids/OptimizedSlidingTimeWindowMovingAverages/tritium-jmh/src/jmh/java/com/palantir/tritium/microbenchmarks/MovingAverageBenchmarks.java) demonstrates the difference in implementations in both execution time (22x faster) and allocations:


```
Benchmark                                             (recordings)                                    (type)  Mode  Cnt     Score     Error   Units
MovingAverageBenchmarks.getM1Rate                               10           SlidingTimeWindowMovingAverages  avgt   20  1364.969 ±  12.040   ns/op
MovingAverageBenchmarks.getM1Rate:gc.alloc.rate.norm            10           SlidingTimeWindowMovingAverages  avgt   20   688.037 ±   0.001    B/op
MovingAverageBenchmarks.getM1Rate                               10  OptimizedSlidingTimeWindowMovingAverages  avgt   20    59.182 ±   0.343   ns/op
MovingAverageBenchmarks.getM1Rate:gc.alloc.rate.norm            10  OptimizedSlidingTimeWindowMovingAverages  avgt   20     0.002 ±   0.001    B/op
MovingAverageBenchmarks.getM1Rate                             1000           SlidingTimeWindowMovingAverages  avgt   20  1401.864 ± 107.134   ns/op
MovingAverageBenchmarks.getM1Rate:gc.alloc.rate.norm          1000           SlidingTimeWindowMovingAverages  avgt   20   688.038 ±   0.003    B/op
MovingAverageBenchmarks.getM1Rate                             1000  OptimizedSlidingTimeWindowMovingAverages  avgt   20    61.157 ±   1.393   ns/op
MovingAverageBenchmarks.getM1Rate:gc.alloc.rate.norm          1000  OptimizedSlidingTimeWindowMovingAverages  avgt   20     0.002 ±   0.001    B/op
```

Author: schlosna

metrics-core/src/main/java/com/codahale/metrics/SlidingTimeWindowMovingAverages.java

@@ -2,7 +2,6 @@
 
 import java.time.Duration;
 import java.time.Instant;
-import java.util.ArrayList;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.LongAdder;
@@ -39,7 +38,7 @@ public class SlidingTimeWindowMovingAverages implements MovingAverages {
      * One counter per time bucket/slot (i.e. per second, see TICK_INTERVAL) for the entire
      * time window (i.e. 15 minutes, see TIME_WINDOW_DURATION_MINUTES)
      */
-    private ArrayList<LongAdder> buckets;
+    private final LongAdder[] buckets;
 
     /**
      * Index into buckets, pointing at the bucket containing the oldest counts
@@ -79,9 +78,9 @@ public SlidingTimeWindowMovingAverages(Clock clock) {
         final long startTime = clock.getTick();
         lastTick = new AtomicLong(startTime);
 
-        buckets = new ArrayList<>(NUMBER_OF_BUCKETS);
+        buckets = new LongAdder[NUMBER_OF_BUCKETS];
         for (int i = 0; i < NUMBER_OF_BUCKETS; i++) {
-            buckets.add(new LongAdder());
+            buckets[i] = new LongAdder();
         }
         bucketBaseTime = Instant.ofEpochSecond(0L, startTime);
         oldestBucketTime = bucketBaseTime;
@@ -91,7 +90,7 @@ public SlidingTimeWindowMovingAverages(Clock clock) {
 
     @Override
     public void update(long n) {
-        buckets.get(currentBucketIndex).add(n);
+        buckets[currentBucketIndex].add(n);
     }
 
     @Override
@@ -162,14 +161,14 @@ private void cleanOldBuckets(Instant currentTick) {
     private void cleanBucketRange(int fromIndex, int toIndex) {
         if (fromIndex < toIndex) {
             for (int i = fromIndex; i < toIndex; i++) {
-                buckets.get(i).reset();
+                buckets[i].reset();
             }
         } else {
             for (int i = fromIndex; i < NUMBER_OF_BUCKETS; i++) {
-                buckets.get(i).reset();
+                buckets[i].reset();
             }
             for (int i = 0; i < toIndex; i++) {
-                buckets.get(i).reset();
+                buckets[i].reset();
             }
         }
     }
@@ -179,19 +178,20 @@ private long sumBuckets(Instant toTime, int numberOfBuckets) {
         // increment toIndex to include the current bucket into the sum
         int toIndex = normalizeIndex(calculateIndexOfTick(toTime) + 1);
         int fromIndex = normalizeIndex(toIndex - numberOfBuckets);
-        LongAdder adder = new LongAdder();
+        long sum = 0;
 
         if (fromIndex < toIndex) {
-            buckets.stream()
-                    .skip(fromIndex)
-                    .limit(toIndex - fromIndex)
-                    .mapToLong(LongAdder::longValue)
-                    .forEach(adder::add);
+            for (int i = fromIndex; i < toIndex; i++) {
+                sum += buckets[i].longValue();
+            }
         } else {
-            buckets.stream().limit(toIndex).mapToLong(LongAdder::longValue).forEach(adder::add);
-            buckets.stream().skip(fromIndex).mapToLong(LongAdder::longValue).forEach(adder::add);
+            for (int i = fromIndex; i < NUMBER_OF_BUCKETS; i++) {
+                sum += buckets[i].longValue();
+            }
+            for (int i = 0; i < toIndex; i++) {
+                sum += buckets[i].longValue();
+            }
         }
-        long retval = adder.longValue();
-        return retval;
+        return sum;
     }
 }