Use IntUnaryOperator / DoubleUnaryOperator in VegasLimit

Switch to IntUnaryOperator and DoubleUnaryOperator in VegasLimit to avoid unnecessary boxing.
Additionally, reduced repeated reads of volatile fields.

Author: kilink

concurrency-limits-core/src/main/java/com/netflix/concurrency/limits/limit/VegasLimit.java

@@ -19,14 +19,15 @@
 import com.netflix.concurrency.limits.MetricRegistry;
 import com.netflix.concurrency.limits.MetricRegistry.SampleListener;
 import com.netflix.concurrency.limits.internal.EmptyMetricRegistry;
-import com.netflix.concurrency.limits.internal.Preconditions;
 import com.netflix.concurrency.limits.limit.functions.Log10RootFunction;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.util.concurrent.ThreadLocalRandom;
 import java.util.concurrent.TimeUnit;
+import java.util.function.DoubleUnaryOperator;
 import java.util.function.Function;
+import java.util.function.IntUnaryOperator;
 
 /**
  * Limiter based on TCP Vegas where the limit increases by alpha if the queue_use is small ({@literal <} alpha)
@@ -41,19 +42,19 @@
 public class VegasLimit extends AbstractLimit {
     private static final Logger LOG = LoggerFactory.getLogger(VegasLimit.class);
 
-    private static final Function<Integer, Integer> LOG10 = Log10RootFunction.create(0);
+    private static final IntUnaryOperator LOG10 = Log10RootFunction.create(0);
 
     public static class Builder {
         private int initialLimit = 20;
         private int maxConcurrency = 1000;
         private MetricRegistry registry = EmptyMetricRegistry.INSTANCE;
         private double smoothing = 1.0;
 
-        private Function<Integer, Integer> alphaFunc = (limit) -> 3 * LOG10.apply(limit.intValue());
-        private Function<Integer, Integer> betaFunc = (limit) -> 6 * LOG10.apply(limit.intValue());
-        private Function<Integer, Integer> thresholdFunc = (limit) -> LOG10.apply(limit.intValue());
-        private Function<Double, Double> increaseFunc = (limit) -> limit + LOG10.apply(limit.intValue());
-        private Function<Double, Double> decreaseFunc = (limit) -> limit - LOG10.apply(limit.intValue());
+        private IntUnaryOperator alphaFunc = (limit) -> 3 * LOG10.applyAsInt(limit);
+        private IntUnaryOperator betaFunc = (limit) -> 6 * LOG10.applyAsInt(limit);
+        private IntUnaryOperator thresholdFunc = LOG10;
+        private DoubleUnaryOperator increaseFunc = (limit) -> limit + LOG10.applyAsInt((int) limit);
+        private DoubleUnaryOperator decreaseFunc = (limit) -> limit - LOG10.applyAsInt((int) limit);
         private int probeMultiplier = 30;
 
         private Builder() {
@@ -74,13 +75,31 @@ public Builder alpha(int alpha) {
             this.alphaFunc = (ignore) -> alpha;
             return this;
         }
-        
+
+        /**
+         * @deprecated use {@link #thresholdFunction(IntUnaryOperator)}
+         */
+        @Deprecated
         public Builder threshold(Function<Integer, Integer> threshold) {
+            this.thresholdFunc = threshold::apply;
+            return this;
+        }
+
+        public Builder thresholdFunction(IntUnaryOperator threshold) {
             this.thresholdFunc = threshold;
             return this;
         }
-        
+
+        /**
+         * @deprecated use {@link #alphaFunction(IntUnaryOperator)}
+         */
+        @Deprecated
         public Builder alpha(Function<Integer, Integer> alpha) {
+            this.alphaFunc = alpha::apply;
+            return this;
+        }
+
+        public Builder alphaFunction(IntUnaryOperator alpha) {
             this.alphaFunc = alpha;
             return this;
         }
@@ -89,18 +108,45 @@ public Builder beta(int beta) {
             this.betaFunc = (ignore) -> beta;
             return this;
         }
-        
+
+        /**
+         * @deprecated use {@link #betaFunction(IntUnaryOperator)}
+         */
+        @Deprecated
         public Builder beta(Function<Integer, Integer> beta) {
+            this.betaFunc = beta::apply;
+            return this;
+        }
+
+        public Builder betaFunction(IntUnaryOperator beta) {
             this.betaFunc = beta;
             return this;
         }
-        
+
+        /**
+         * @deprecated use {@link #increaseFunction(DoubleUnaryOperator)}
+         */
+        @Deprecated
         public Builder increase(Function<Double, Double> increase) {
+            this.increaseFunc = increase::apply;
+            return this;
+        }
+
+        public Builder increaseFunction(DoubleUnaryOperator increase) {
             this.increaseFunc = increase;
             return this;
         }
-        
+
+        /**
+         * @deprecated use {@link #decreaseFunction(DoubleUnaryOperator)}
+         */
+        @Deprecated
         public Builder decrease(Function<Double, Double> decrease) {
+            this.decreaseFunc = decrease::apply;
+            return this;
+        }
+
+        public Builder decreaseFunction(DoubleUnaryOperator decrease) {
             this.decreaseFunc = decrease;
             return this;
         }
@@ -164,11 +210,11 @@ public static VegasLimit newDefault() {
     private final int maxLimit; 
 
     private final double smoothing;
-    private final Function<Integer, Integer> alphaFunc;
-    private final Function<Integer, Integer> betaFunc;
-    private final Function<Integer, Integer> thresholdFunc;
-    private final Function<Double, Double> increaseFunc;
-    private final Function<Double, Double> decreaseFunc;
+    private final IntUnaryOperator alphaFunc;
+    private final IntUnaryOperator betaFunc;
+    private final IntUnaryOperator thresholdFunc;
+    private final DoubleUnaryOperator increaseFunc;
+    private final DoubleUnaryOperator decreaseFunc;
     private final SampleListener rttSampleListener;
     private final int probeMultiplier;
     private int probeCount = 0;
@@ -201,69 +247,77 @@ private boolean shouldProbe() {
 
     @Override
     protected int _update(long startTime, long rtt, int inflight, boolean didDrop) {
-        Preconditions.checkArgument(rtt > 0, "rtt must be >0 but got " + rtt);
+        if (rtt <= 0) {
+            throw new IllegalArgumentException("rtt must be >0 but got " + rtt);
+        }
 
         probeCount++;
         if (shouldProbe()) {
-            LOG.debug("Probe MinRTT {}", TimeUnit.NANOSECONDS.toMicros(rtt) / 1000.0);
+            if (LOG.isDebugEnabled()) {
+                LOG.debug("Probe MinRTT {}", TimeUnit.NANOSECONDS.toMicros(rtt) / 1000.0);
+            }
             resetProbeJitter();
             probeCount = 0;
             rtt_noload = rtt;
-            return (int)estimatedLimit;
+            return (int) estimatedLimit;
         }
-        
+
+        long rtt_noload = this.rtt_noload;
         if (rtt_noload == 0 || rtt < rtt_noload) {
-            LOG.debug("New MinRTT {}", TimeUnit.NANOSECONDS.toMicros(rtt) / 1000.0);
-            rtt_noload = rtt;
-            return (int)estimatedLimit;
+            if (LOG.isDebugEnabled()) {
+                LOG.debug("New MinRTT {}", TimeUnit.NANOSECONDS.toMicros(rtt) / 1000.0);
+            }
+            this.rtt_noload = rtt;
+            return (int) estimatedLimit;
         }
-        
+
         rttSampleListener.addLongSample(rtt_noload);
 
-        return updateEstimatedLimit(rtt, inflight, didDrop);
+        return updateEstimatedLimit(rtt, rtt_noload, inflight, didDrop);
     }
 
-    private int updateEstimatedLimit(long rtt, int inflight, boolean didDrop) {
-        final int queueSize = (int) Math.ceil(estimatedLimit * (1 - (double)rtt_noload / rtt));
+    private int updateEstimatedLimit(long rtt, long rtt_noload, int inflight, boolean didDrop) {
+        double estimatedLimit = this.estimatedLimit;
+        final int queueSize = (int) Math.ceil(estimatedLimit * (1 - (double) rtt_noload / rtt));
 
         double newLimit;
         // Treat any drop (i.e timeout) as needing to reduce the limit
         if (didDrop) {
-            newLimit = decreaseFunc.apply(estimatedLimit);
+            newLimit = decreaseFunc.applyAsDouble(estimatedLimit);
         // Prevent upward drift if not close to the limit
         } else if (inflight * 2 < estimatedLimit) {
-            return (int)estimatedLimit;
+            return (int) estimatedLimit;
         } else {
-            int alpha = alphaFunc.apply((int)estimatedLimit);
-            int beta = betaFunc.apply((int)estimatedLimit);
-            int threshold = this.thresholdFunc.apply((int)estimatedLimit);
+            int alpha = alphaFunc.applyAsInt((int) estimatedLimit);
+            int beta = betaFunc.applyAsInt((int) estimatedLimit);
+            int threshold = thresholdFunc.applyAsInt((int) estimatedLimit);
 
             // Aggressive increase when no queuing
             if (queueSize <= threshold) {
                 newLimit = estimatedLimit + beta;
             // Increase the limit if queue is still manageable
             } else if (queueSize < alpha) {
-                newLimit = increaseFunc.apply(estimatedLimit);
+                newLimit = increaseFunc.applyAsDouble(estimatedLimit);
             // Detecting latency so decrease
             } else if (queueSize > beta) {
-                newLimit = decreaseFunc.apply(estimatedLimit);
+                newLimit = decreaseFunc.applyAsDouble(estimatedLimit);
             // We're within he sweet spot so nothing to do
             } else {
-                return (int)estimatedLimit;
+                return (int) estimatedLimit;
             }
         }
 
         newLimit = Math.max(1, Math.min(maxLimit, newLimit));
         newLimit = (1 - smoothing) * estimatedLimit + smoothing * newLimit;
-        if ((int)newLimit != (int)estimatedLimit && LOG.isDebugEnabled()) {
+        if ((int) newLimit != (int) estimatedLimit && LOG.isDebugEnabled()) {
             LOG.debug("New limit={} minRtt={} ms winRtt={} ms queueSize={}",
-                    (int)newLimit,
+                    (int) newLimit,
                     TimeUnit.NANOSECONDS.toMicros(rtt_noload) / 1000.0,
                     TimeUnit.NANOSECONDS.toMicros(rtt) / 1000.0,
                     queueSize);
         }
-        estimatedLimit = newLimit;
-        return (int)estimatedLimit;
+        this.estimatedLimit = newLimit;
+        return (int) newLimit;
     }
 
     @Override

concurrency-limits-core/src/main/java/com/netflix/concurrency/limits/limit/functions/Log10RootFunction.java

@@ -15,18 +15,19 @@
  */
 package com.netflix.concurrency.limits.limit.functions;
 
-import java.util.function.Function;
-import java.util.stream.IntStream;
+import java.util.function.IntUnaryOperator;
 
 /**
  * Function used by limiters to calculate thredsholds using log10 of the current limit.  
  * Here we pre-compute the log10 of numbers up to 1000 as an optimization.
  */
-public final class Log10RootFunction implements Function<Integer, Integer> {
+public final class Log10RootFunction implements IntUnaryOperator {
     static final int[] lookup = new int[1000];
 
     static {
-        IntStream.range(0, 1000).forEach(i -> lookup[i] = Math.max(1, (int)Math.log10(i)));
+        for (int i = 0; i < lookup.length; i++) {
+            lookup[i] = Math.max(1, (int) Math.log10(i));
+        }
     }
 
     private static final Log10RootFunction INSTANCE = new Log10RootFunction();
@@ -37,12 +38,12 @@ public final class Log10RootFunction implements Function<Integer, Integer> {
      * @param baseline
      * @return
      */
-    public static Function<Integer, Integer> create(int baseline) {
-        return INSTANCE.andThen(t -> t + baseline);
+    public static IntUnaryOperator create(int baseline) {
+        return baseline == 0 ? INSTANCE : INSTANCE.andThen(t -> t + baseline);
     }
-    
+
     @Override
-    public Integer apply(Integer t) {
-        return t < 1000 ? lookup[t] : (int)Math.log10(t);
+    public int applyAsInt(int t) {
+        return t < 1000 ? lookup[t] : (int) Math.log10(t);
     }
 }

concurrency-limits-core/src/test/java/com/netflix/concurrency/limits/limit/VegasLimitTest.java

@@ -58,7 +58,7 @@ public void noChangeIfWithinThresholds() {
     @Test
     public void decreaseSmoothing() {
         VegasLimit limit = VegasLimit.newBuilder()
-            .decrease(current -> current / 2)
+            .decreaseFunction(current -> current / 2)
             .smoothing(0.5)
             .initialLimit(100)
             .maxConcurrency(200)
@@ -80,7 +80,7 @@ public void decreaseSmoothing() {
     @Test
     public void decreaseWithoutSmoothing() {
         VegasLimit limit = VegasLimit.newBuilder()
-            .decrease(current -> current / 2)
+            .decreaseFunction(current -> current / 2)
             .initialLimit(100)
             .maxConcurrency(200)
             .build();

concurrency-limits-core/src/test/java/com/netflix/concurrency/limits/limit/functions/Log10RootFunctionTest.java

@@ -3,24 +3,24 @@
 import org.junit.Assert;
 import org.junit.Test;
 
-import java.util.function.Function;
+import java.util.function.IntUnaryOperator;
 
 public class Log10RootFunctionTest {
     @Test
     public void test0Index() {
-        Function<Integer, Integer> func = Log10RootFunction.create(0);
-        Assert.assertEquals(1, func.apply(0).intValue());
+        IntUnaryOperator func = Log10RootFunction.create(0);
+        Assert.assertEquals(1, func.applyAsInt(0));
     }
 
     @Test
     public void testInRange() {
-        Function<Integer, Integer> func = Log10RootFunction.create(0);
-        Assert.assertEquals(2, func.apply(100).intValue());
+        IntUnaryOperator func = Log10RootFunction.create(0);
+        Assert.assertEquals(2, func.applyAsInt(100));
     }
 
     @Test
     public void testOutofLookupRange() {
-        Function<Integer, Integer> func = Log10RootFunction.create(0);
-        Assert.assertEquals(4, func.apply(10000).intValue());
+        IntUnaryOperator func = Log10RootFunction.create(0);
+        Assert.assertEquals(4, func.applyAsInt(10000));
     }
 }