Merge pull request #24 from Netflix/feature/min_threshold

Eliminated unusually low values that skew the limit estimation

Author: elandau

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

@@ -9,6 +9,7 @@
 import com.netflix.concurrency.limits.Limit;
 import com.netflix.concurrency.limits.MetricIds;
 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.SquareRootFunction;
@@ -26,35 +27,84 @@ public final class GradientLimit implements Limit {
     public static class Builder {
         private int initialLimit = 20;
         private int maxConcurrency = 1000;
+        private long minRttThreshold = TimeUnit.MILLISECONDS.toNanos(1);
+        
         private double smoothing = 0.2;
         private Function<Integer, Integer> queueSize = SquareRootFunction.create(4);
         private MetricRegistry registry = EmptyMetricRegistry.INSTANCE;
 
+        /**
+         * Minimum threshold for accepting a new rtt sample.  Any RTT lower than this threshold
+         * will be discarded.
+         *  
+         * @param minRttTreshold
+         * @param units
+         * @return Chainable builder
+         */
+        public Builder minRttThreshold(long minRttTreshold, TimeUnit units) {
+            this.minRttThreshold = units.toMillis(minRttTreshold);
+            return this;
+        }
+        
+        /**
+         * Initial limit used by the limiter
+         * @param initialLimit
+         * @return Chainable builder
+         */
         public Builder initialLimit(int initialLimit) {
             this.initialLimit = initialLimit;
             return this;
         }
 
+        /**
+         * Maximum allowable concurrency.  Any estimated concurrency will be capped
+         * at this value
+         * @param maxConcurrency
+         * @return Chainable builder
+         */
         public Builder maxConcurrency(int maxConcurrency) {
             this.maxConcurrency = maxConcurrency;
             return this;
         }
 
+        /**
+         * Fixed amount the estimated limit can grow while latencies remain low
+         * @param queueSize
+         * @return Chainable builder
+         */
         public Builder queueSize(int queueSize) {
             this.queueSize = (ignore) -> queueSize;
             return this;
         }
-        
+
+        /**
+         * Function to dynamically determine the amount the estimated limit can grow while
+         * latencies remain low as a function of the current limit.
+         * @param queueSize
+         * @return Chainable builder
+         */
         public Builder queueSize(Function<Integer, Integer> queueSize) {
             this.queueSize = queueSize;
             return this;
         }
 
+        /**
+         * Smoothing factor to limit how aggressively the estimated limit can shrink
+         * when queuing has been detected.
+         * @param smoothing Value of 0.0 to 1.0 where 1.0 means the limit is completely
+         *  replicated by the new estimate.
+         * @return Chainable builder
+         */
         public Builder smoothing(double smoothing) {
             this.smoothing = smoothing;
             return this;
         }
 
+        /**
+         * Registry for reporting metrics about the limiter's internal state.
+         * @param registry
+         * @return Chainable builder
+         */
         public Builder metricRegistry(MetricRegistry registry) {
             this.registry = registry;
             return this;
@@ -93,22 +143,35 @@ public static GradientLimit newDefault() {
 
     private final double smoothing;
 
+    private final SampleListener sampleRttMetric;
+
+    private final long minRttThreshold;
+
     private GradientLimit(Builder builder) {
         this.estimatedLimit = builder.initialLimit;
         this.maxLimit = builder.maxConcurrency;
         this.queueSize = builder.queueSize;
         this.smoothing = builder.smoothing;
+        this.minRttThreshold = builder.minRttThreshold;
+        
+        this.sampleRttMetric = builder.registry.registerDistribution("sample_rtt");
     }
 
     @Override
     public synchronized void update(long rtt, int maxInFlight) {
         Preconditions.checkArgument(rtt > 0, "rtt must be >0 but got " + rtt);
 
+        if (rtt < minRttThreshold) {
+            return;
+        }
+        
         if (rtt_noload == 0 || rtt < rtt_noload) {
             LOG.debug("New MinRTT {}", rtt);
             rtt_noload = rtt;
         }
 
+        sampleRttMetric.addSample(rtt);
+        
         final double queueSize = this.queueSize.apply((int)this.estimatedLimit);
         final double gradient = (double)rtt_noload / rtt;
         double newLimit;