Update histogram implementation to FastUtil and use 64-bit bucket counts (#108)

* Update histogram implementation to FastUtil and use 64-bit bucket counts.

* Fix histogram iteration.

Author: vjkoskela

pom.xml

@@ -109,6 +109,7 @@
     <apache.httpcore.version>4.4.5</apache.httpcore.version>
     <arpnetworking.commons.version>1.13.3</arpnetworking.commons.version>
     <aspectjrt.version>1.9.1</aspectjrt.version>
+    <asynchttpclient.version>2.0.24</asynchttpclient.version>
     <cglib.version>3.2.5</cglib.version>
     <commons.codec.version>1.10</commons.codec.version>
     <ebean.version>6.8.1</ebean.version>
@@ -118,7 +119,7 @@
     <guice.version>4.1.0</guice.version>
     <hamcrest.version>2.0.0.0</hamcrest.version>
     <hikari.version>2.5.1</hikari.version>
-    <asynchttpclient.version>2.0.24</asynchttpclient.version>
+    <fastutil.version>8.3.1</fastutil.version>
     <h2.version>1.4.193</h2.version>
     <jackson.version>2.9.2</jackson.version>
     <javassist.version>3.22.0-GA</javassist.version>
@@ -136,7 +137,7 @@
     <metrics.client.version>0.11.1</metrics.client.version>
     <metrics.jvm.extra.version>0.11.0</metrics.jvm.extra.version>
     <metrics.http.extra.version>0.11.1</metrics.http.extra.version>
-    <metrics.aggregator.protocol.version>1.0.6</metrics.aggregator.protocol.version>
+    <metrics.aggregator.protocol.version>1.0.8</metrics.aggregator.protocol.version>
     <mockito.version>2.12.0</mockito.version>
     <netty.version>3.10.3.Final</netty.version>
     <netty.all.version>4.0.21.Final</netty.all.version>
@@ -733,6 +734,14 @@
       <artifactId>oval</artifactId>
       <version>${oval.version}</version>
     </dependency>
+    <dependency>
+      <groupId>it.unimi.dsi</groupId>
+      <artifactId>fastutil</artifactId>
+      <version>${fastutil.version}</version>
+      <!-- Jackson support not yet available:
+          https://github.com/FasterXML/jackson-datatypes-collections/issues/38
+      -->
+    </dependency>
     <dependency>
       <groupId>org.javassist</groupId>
       <artifactId>javassist</artifactId>

src/main/java/com/arpnetworking/clusteraggregator/models/CombinedMetricData.java

@@ -276,7 +276,7 @@ public static Builder fromStatisticSetRecord(final Messages.StatisticSetRecord r
                     final HistogramStatistic.Histogram histogram = new HistogramStatistic.Histogram();
                     for (final Messages.SparseHistogramEntry entry : supportingData.getEntriesList()) {
                         final double bucket = entry.getBucket();
-                        final int count = entry.getCount();
+                        final long count = entry.getCount();
                         histogram.recordValue(bucket, count);
                     }
 

src/main/java/com/arpnetworking/tsdcore/sinks/KairosDbSink.java

@@ -35,6 +35,7 @@
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Lists;
+import it.unimi.dsi.fastutil.doubles.Double2LongMap;
 import net.sf.oval.constraint.Min;
 import net.sf.oval.constraint.NotNull;
 import org.joda.time.Period;
@@ -323,8 +324,8 @@ public void serializeHistogram(
                 chunkGenerator.writeNumberField("mean", additionalData.getMean());
                 chunkGenerator.writeNumberField("sum", additionalData.getSum());
                 chunkGenerator.writeObjectFieldStart("bins");
-                for (Map.Entry<Double, Integer> bin : bins.getValues()) {
-                    chunkGenerator.writeNumberField(bin.getKey().toString(), bin.getValue());
+                for (Double2LongMap.Entry bin : bins.getValues()) {
+                    chunkGenerator.writeNumberField(String.valueOf(bin.getDoubleKey()), bin.getLongValue());
                 }
 
                 chunkGenerator.writeEndObject();  //close bins

src/main/java/com/arpnetworking/tsdcore/sinks/circonus/CirconusSinkActor.java

@@ -34,6 +34,7 @@
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
 import com.google.common.collect.Sets;
+import it.unimi.dsi.fastutil.doubles.Double2LongMap;
 import org.joda.time.Period;
 import org.joda.time.format.ISOPeriodFormat;
 import play.libs.ws.StandaloneWSResponse;
@@ -287,9 +288,9 @@ private Map<String, Object> serialize(final Collection<AggregatedData> data) {
                 final HistogramStatistic.HistogramSupportingData histogramSupportingData = (HistogramStatistic.HistogramSupportingData)
                         aggregatedData.getSupportingData();
                 final HistogramStatistic.HistogramSnapshot histogram = histogramSupportingData.getHistogramSnapshot();
-                final ArrayList<String> valueList = new ArrayList<>(histogram.getEntriesCount());
+                final List<String> valueList = new ArrayList<>((int) histogram.getEntriesCount());
                 final MathContext context = new MathContext(2, RoundingMode.DOWN);
-                for (final Map.Entry<Double, Integer> entry : histogram.getValues()) {
+                for (final Double2LongMap.Entry entry : histogram.getValues()) {
                     for (int i = 0; i < entry.getValue(); i++) {
                         final BigDecimal decimal = new BigDecimal(entry.getKey(), context);
                         final String bucketString = String.format("H[%s]=%d", decimal.toPlainString(), entry.getValue());

src/main/java/com/arpnetworking/tsdcore/statistics/HistogramStatistic.java

@@ -20,13 +20,15 @@
 import com.arpnetworking.tsdcore.model.CalculatedValue;
 import com.arpnetworking.tsdcore.model.Quantity;
 import com.arpnetworking.tsdcore.model.Unit;
+import it.unimi.dsi.fastutil.doubles.Double2LongAVLTreeMap;
+import it.unimi.dsi.fastutil.doubles.Double2LongMap;
+import it.unimi.dsi.fastutil.doubles.Double2LongSortedMap;
+import it.unimi.dsi.fastutil.objects.ObjectSortedSet;
 import net.sf.oval.constraint.NotNull;
 
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
-import java.util.Set;
-import java.util.TreeMap;
 
 /**
  * Histogram statistic. This is a supporting statistic and does not produce
@@ -59,6 +61,7 @@ public Quantity calculateAggregations(final List<AggregatedData> aggregations) {
 
     private HistogramStatistic() { }
 
+    private static final int DEFAULT_PRECISION = 7;
     private static final long serialVersionUID = 7060886488604176233L;
 
     /**
@@ -183,9 +186,9 @@ public HistogramSnapshot getHistogramSnapshot() {
         public HistogramSupportingData toUnit(final Unit newUnit) {
             if (_unit.isPresent()) {
                 final Histogram newHistogram = new Histogram();
-                for (final Map.Entry<Double, Integer> entry : _histogramSnapshot.getValues()) {
-                    final Double newBucket = newUnit.convert(entry.getKey(), _unit.get());
-                    newHistogram.recordValue(newBucket, entry.getValue());
+                for (final Double2LongMap.Entry entry : _histogramSnapshot.getValues()) {
+                    final double newBucket = newUnit.convert(entry.getDoubleKey(), _unit.get());
+                    newHistogram.recordValue(newBucket, entry.getLongValue());
                 }
                 return new HistogramSupportingData.Builder()
                         .setHistogramSnapshot(newHistogram.getSnapshot())
@@ -249,14 +252,15 @@ public Builder setUnit(final Optional<Unit> value) {
      */
     public static final class Histogram {
 
+
         /**
          * Records a value into the histogram.
          *
          * @param value The value of the entry.
          * @param count The number of entries at this value.
          */
-        public void recordValue(final double value, final int count) {
-            _data.merge(truncate(value), count, (i, j) -> i + j);
+        public void recordValue(final double value, final long count) {
+            _data.merge(truncateToDouble(value), count, Long::sum);
             _entriesCount += count;
         }
 
@@ -269,29 +273,83 @@ public void recordValue(final double value) {
             recordValue(value, 1);
         }
 
+        /**
+         * Records a packed value into the histogram. The packed values must be at the
+         * precision that this {@link Histogram} was declared with!
+         *
+         * @param packed The packed bucket key.
+         * @param count The number of entries at this value.
+         */
+        public void recordPacked(final long packed, final long count) {
+            recordValue(unpack(packed), count);
+        }
+
         /**
          * Adds a histogram snapshot to this one.
          *
          * @param histogramSnapshot The histogram snapshot to add to this one.
          */
         public void add(final HistogramSnapshot histogramSnapshot) {
-            for (final Map.Entry<Double, Integer> entry : histogramSnapshot._data.entrySet()) {
-                _data.merge(entry.getKey(), entry.getValue(), (i, j) -> i + j);
+            for (final Double2LongMap.Entry entry : histogramSnapshot._data.double2LongEntrySet()) {
+                _data.merge(entry.getDoubleKey(), entry.getLongValue(), Long::sum);
             }
             _entriesCount += histogramSnapshot._entriesCount;
         }
 
         public HistogramSnapshot getSnapshot() {
-            return new HistogramSnapshot(_data, _entriesCount);
+            return new HistogramSnapshot(_data, _entriesCount, _precision);
+        }
+
+        long truncateToLong(final double val) {
+            return Double.doubleToRawLongBits(val) & _truncateMask;
+        }
+
+        double truncateToDouble(final double val) {
+            return Double.longBitsToDouble(truncateToLong(val));
+        }
+
+        long pack(final double val) {
+            final long truncated = truncateToLong(val);
+            final long shifted = truncated >> (MANTISSA_BITS - _precision);
+            return shifted & _packMask;
+        }
+
+        double unpack(final long packed) {
+            return Double.longBitsToDouble(packed << (MANTISSA_BITS - _precision));
         }
 
-        private static double truncate(final double val) {
-            final long mask = 0xffffe00000000000L;
-            return Double.longBitsToDouble(Double.doubleToRawLongBits(val) & mask);
+        /**
+         * Public constructor.
+         */
+        public Histogram() {
+            this(DEFAULT_PRECISION);
+        }
+
+        /**
+         * Public constructor.
+         *
+         * @param precision the bits of mantissa precision in the bucket key
+         */
+        public Histogram(final int precision) {
+            // TODO(ville): Support variable precision histograms end-to-end.
+            if (precision != DEFAULT_PRECISION) {
+                throw new IllegalArgumentException("The stack does not fully support variable precision histograms.");
+            }
+
+            _precision = precision;
+            _truncateMask = BASE_MASK >> _precision;
+            _packMask = (1 << (_precision + EXPONENT_BITS + 1)) - 1;
         }
 
         private int _entriesCount = 0;
-        private final TreeMap<Double, Integer> _data = new TreeMap<>();
+        private final Double2LongSortedMap _data = new Double2LongAVLTreeMap();
+        private final int _precision;
+        private final long _truncateMask;
+        private final int _packMask;
+
+        private static final int MANTISSA_BITS = 52;
+        private static final int EXPONENT_BITS = 11;
+        private static final long BASE_MASK = (1L << (MANTISSA_BITS + EXPONENT_BITS)) >> EXPONENT_BITS;
     }
 
     /**
@@ -300,7 +358,8 @@ private static double truncate(final double val) {
      * @author Brandon Arp (brandon dot arp at inscopemetrics dot com)
      */
     public static final class HistogramSnapshot {
-        private HistogramSnapshot(final TreeMap<Double, Integer> data, final int entriesCount) {
+        private HistogramSnapshot(final Double2LongSortedMap data, final long entriesCount, final int precision) {
+            _precision = precision;
             _entriesCount = entriesCount;
             _data.putAll(data);
         }
@@ -316,24 +375,30 @@ public Double getValueAtPercentile(final double percentile) {
             // The Math.min is for the case where the computation may be just
             // slightly larger than the _entriesCount and prevents an index out of range.
             final int target = (int) Math.min(Math.ceil(_entriesCount * percentile / 100.0D), _entriesCount);
-            int accumulated = 0;
-            for (final Map.Entry<Double, Integer> next : _data.entrySet()) {
-                accumulated += next.getValue();
+            long accumulated = 0;
+            for (final Double2LongMap.Entry next : _data.double2LongEntrySet()) {
+                accumulated += next.getLongValue();
                 if (accumulated >= target) {
-                    return next.getKey();
+                    return next.getDoubleKey();
                 }
             }
             return 0D;
         }
 
-        public int getEntriesCount() {
+        public int getPrecision() {
+            return _precision;
+        }
+
+        public long getEntriesCount() {
             return _entriesCount;
         }
 
-        public Set<Map.Entry<Double, Integer>> getValues() {
-            return _data.entrySet();
+        public ObjectSortedSet<Double2LongMap.Entry> getValues() {
+            return _data.double2LongEntrySet();
         }
-        private int _entriesCount = 0;
-        private final TreeMap<Double, Integer> _data = new TreeMap<>();
+
+        private long _entriesCount = 0;
+        private final Double2LongSortedMap _data = new Double2LongAVLTreeMap();
+        private final int _precision;
     }
 }

src/test/java/com/arpnetworking/tsdcore/statistics/HistogramStatisticTest.java

@@ -18,11 +18,11 @@
 import com.arpnetworking.tsdcore.model.CalculatedValue;
 import com.arpnetworking.tsdcore.model.Quantity;
 import com.arpnetworking.tsdcore.model.Unit;
+import it.unimi.dsi.fastutil.doubles.Double2LongMap;
 import org.junit.Assert;
 import org.junit.Test;
 
 import java.util.Collections;
-import java.util.Map;
 
 /**
  * Tests the HistogramStatistic class.
@@ -40,8 +40,8 @@ public void histogramAccumulateQuantities() {
         final CalculatedValue<HistogramStatistic.HistogramSupportingData> value = accumulator.calculate(Collections.emptyMap());
         final HistogramStatistic.HistogramSupportingData supportingData = value.getData();
         final HistogramStatistic.HistogramSnapshot histogram = supportingData.getHistogramSnapshot();
-        for (final Map.Entry<Double, Integer> entry : histogram.getValues()) {
-            Assert.assertEquals(entry.getValue(), (Integer) 1);
+        for (final Double2LongMap.Entry entry : histogram.getValues()) {
+            Assert.assertEquals(entry.getLongValue(), 1L);
         }
     }
 
@@ -59,8 +59,8 @@ public void histogramAccumulateHistogram() {
         final CalculatedValue<HistogramStatistic.HistogramSupportingData> value = merged.calculate(Collections.emptyMap());
         final HistogramStatistic.HistogramSupportingData supportingData = value.getData();
         final HistogramStatistic.HistogramSnapshot histogram = supportingData.getHistogramSnapshot();
-        for (final Map.Entry<Double, Integer> entry : histogram.getValues()) {
-            Assert.assertEquals(entry.getValue(), (Integer) 1);
+        for (final Double2LongMap.Entry entry : histogram.getValues()) {
+            Assert.assertEquals(entry.getLongValue(), 1L);
         }
     }
 
@@ -90,14 +90,14 @@ public void histogramAccumulateMultipleHistogram() {
         final CalculatedValue<HistogramStatistic.HistogramSupportingData> value = merged.calculate(Collections.emptyMap());
         final HistogramStatistic.HistogramSupportingData supportingData = value.getData();
         final HistogramStatistic.HistogramSnapshot histogram = supportingData.getHistogramSnapshot();
-        for (final Map.Entry<Double, Integer> entry : histogram.getValues()) {
-            final int val = entry.getKey().intValue();
+        for (final Double2LongMap.Entry entry : histogram.getValues()) {
+            final int val = (int) entry.getDoubleKey();
             if (val < 50) {
-                Assert.assertEquals("incorrect value for key " + val, (Integer) 1, entry.getValue());
+                Assert.assertEquals("incorrect value for key " + val, 1L, entry.getLongValue());
             } else if (val <= 100) {
-                Assert.assertEquals("incorrect value for key " + val, (Integer) 2, entry.getValue());
+                Assert.assertEquals("incorrect value for key " + val, 2L, entry.getLongValue());
             } else { // val > 100
-                Assert.assertEquals("incorrect value for key " + val, (Integer) 1, entry.getValue());
+                Assert.assertEquals("incorrect value for key " + val, 1L, entry.getLongValue());
             }
         }
 
@@ -150,9 +150,9 @@ public void histogramAccumulateMultipleHistogramConversion() {
         final CalculatedValue<HistogramStatistic.HistogramSupportingData> value = merged.calculate(Collections.emptyMap());
         final HistogramStatistic.HistogramSupportingData supportingData = value.getData();
         final HistogramStatistic.HistogramSnapshot histogram = supportingData.getHistogramSnapshot();
-        for (final Map.Entry<Double, Integer> entry : histogram.getValues()) {
+        for (final Double2LongMap.Entry entry : histogram.getValues()) {
 
-            Assert.assertTrue(entry.getKey() <= 100);
+            Assert.assertTrue(entry.getDoubleKey() <= 100.0);
         }
 
         Assert.assertEquals(200, histogram.getEntriesCount());
@@ -168,17 +168,17 @@ public void histogramUnitConversion() {
         final CalculatedValue<HistogramStatistic.HistogramSupportingData> value = accumulator.calculate(Collections.emptyMap());
         final HistogramStatistic.HistogramSupportingData supportingData = value.getData();
         HistogramStatistic.HistogramSnapshot histogram = supportingData.getHistogramSnapshot();
-        for (final Map.Entry<Double, Integer> entry : histogram.getValues()) {
-            final int val = entry.getKey().intValue();
+        for (final Double2LongMap.Entry entry : histogram.getValues()) {
+            final int val = (int) entry.getDoubleKey();
             if (val < 990) {
                 Assert.fail("shouldn't see a key this small");
             }
         }
 
         final HistogramStatistic.HistogramSupportingData converted = supportingData.toUnit(Unit.SECOND);
         histogram = converted.getHistogramSnapshot();
-        for (final Map.Entry<Double, Integer> entry : histogram.getValues()) {
-            final int val = entry.getKey().intValue();
+        for (final Double2LongMap.Entry entry : histogram.getValues()) {
+            final int val = (int) entry.getDoubleKey();
             if (val > 100) {
                 Assert.fail("shouldn't see a key this large after unit conversion");
             }