Improve CKMSQuantiles and address memory leak

CKMSQuantiles is copied from an implementation of 2012, where it states that a ‘HACK’ was done, admitting a space leak.
This leak has been noticed several times (#422, #550, #654).
By correctly applying the algorithm from the paper we fix the leak.

I have added unit-tests to show that the behaviour is correct.
I have also added a Benchmark in the benchmark module showing the difference with the old and current
implementation.

According to my benchmarks, is in the new implementation a `get` of a quantile that has ‘seen’ 1 million elements 440 times faster.
Inserting 1 million elements is 3.5 times faster.

While going through the CKMS paper and the Java implementation I have added remarks and snippets
from the paper, to clarify why certain choices are made.

Signed-off-by: Jens <[email protected]>

Fix assertion in HistogramTest

Median of 1..11 = 6

Signed-off-by: Jens <[email protected]>

Address PR remarks

Signed-off-by: Jens <[email protected]>

benchmarks/pom.xml

@@ -27,12 +27,12 @@
         <dependency>
             <groupId>org.openjdk.jmh</groupId>
             <artifactId>jmh-core</artifactId>
-            <version>1.3.2</version>
+            <version>1.34</version>
         </dependency>
         <dependency>
             <groupId>org.openjdk.jmh</groupId>
             <artifactId>jmh-generator-annprocess</artifactId>
-            <version>1.3.2</version>
+            <version>1.34</version>
         </dependency>
         <dependency>
             <groupId>javax.annotation</groupId>

benchmarks/src/main/java/io/prometheus/client/CKMSQuantileBenchmark.java

@@ -0,0 +1,108 @@
+package io.prometheus.client;
+
+import io.prometheus.client.CKMSQuantiles.Quantile;
+import org.openjdk.jmh.annotations.*;
+import org.openjdk.jmh.infra.Blackhole;
+import org.openjdk.jmh.runner.Runner;
+import org.openjdk.jmh.runner.RunnerException;
+import org.openjdk.jmh.runner.options.Options;
+import org.openjdk.jmh.runner.options.OptionsBuilder;
+
+import java.util.*;
+import java.util.concurrent.TimeUnit;
+
+public class CKMSQuantileBenchmark {
+
+    @State(Scope.Benchmark)
+    public static class EmptyBenchmarkState {
+        @Param({"10000", "100000", "1000000"})
+        public int value;
+
+        CKMSQuantiles ckmsQuantiles;
+
+        List<Quantile> quantiles;
+        Random rand = new Random(0);
+
+        long[] shuffle;
+
+        @Setup(Level.Trial)
+        public void setup() {
+            quantiles = new ArrayList<Quantile>();
+            quantiles.add(new Quantile(0.50, 0.050));
+            quantiles.add(new Quantile(0.90, 0.010));
+            quantiles.add(new Quantile(0.95, 0.005));
+            quantiles.add(new Quantile(0.99, 0.001));
+
+
+            shuffle = new long[value];
+            for (int i = 0; i < shuffle.length; i++) {
+                shuffle[i] = i;
+            }
+            Collections.shuffle(Arrays.asList(shuffle), rand);
+
+        }
+    }
+
+    @Benchmark
+    @BenchmarkMode({Mode.AverageTime})
+    @OutputTimeUnit(TimeUnit.MILLISECONDS)
+    public void ckmsQuantileInsertBenchmark(Blackhole blackhole, EmptyBenchmarkState state) {
+        CKMSQuantiles q = new CKMSQuantiles(state.quantiles.toArray(new Quantile[]{}));
+        for (long l : state.shuffle) {
+            q.insert(l);
+        }
+    }
+
+    @State(Scope.Benchmark)
+    public static class PrefilledBenchmarkState {
+        public int value = 1000000;
+
+        CKMSQuantiles ckmsQuantiles;
+
+        List<Quantile> quantiles;
+        Random rand = new Random(0);
+
+        long[] shuffle;
+
+        @Setup(Level.Trial)
+        public void setup() {
+            quantiles = new ArrayList<Quantile>();
+            quantiles.add(new Quantile(0.50, 0.050));
+            quantiles.add(new Quantile(0.90, 0.010));
+            quantiles.add(new Quantile(0.95, 0.005));
+            quantiles.add(new Quantile(0.99, 0.001));
+
+
+            shuffle = new long[value];
+            for (int i = 0; i < shuffle.length; i++) {
+                shuffle[i] = i;
+            }
+            Collections.shuffle(Arrays.asList(shuffle), rand);
+            ckmsQuantiles = new CKMSQuantiles(quantiles.toArray(new Quantile[]{}));
+            for (long l : shuffle) {
+                ckmsQuantiles.insert(l);
+            }
+
+        }
+    }
+
+    @Benchmark
+    @BenchmarkMode({Mode.AverageTime})
+    @OutputTimeUnit(TimeUnit.NANOSECONDS)
+    public void ckmsQuantileGetBenchmark(Blackhole blackhole, PrefilledBenchmarkState state) {
+        blackhole.consume(state.ckmsQuantiles.get(0.95));
+    }
+
+    public static void main(String[] args) throws RunnerException {
+
+        Options opt = new OptionsBuilder()
+                .include(CKMSQuantileBenchmark.class.getSimpleName())
+                .warmupIterations(5)
+                .measurementIterations(4)
+                .threads(1)
+                .forks(1)
+                .build();
+
+        new Runner(opt).run();
+    }
+}

simpleclient/pom.xml

@@ -54,5 +54,11 @@
       <version>4.13.2</version>
       <scope>test</scope>
     </dependency>
+      <dependency>
+          <groupId>org.apache.commons</groupId>
+          <artifactId>commons-math3</artifactId>
+          <version>3.2</version>
+          <scope>test</scope>
+      </dependency>
   </dependencies>
 </project>