Reduce the impact of long GC's and huge spikes in latency by always smoothing updates to the limit. Introduce smoothing to the noload RTT to keep that value more stable.

Author: elandau

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

@@ -28,16 +28,16 @@ public final class GradientLimit implements Limit {
     private static final Logger LOG = LoggerFactory.getLogger(GradientLimit.class);
 
     public static class Builder {
-        private int initialLimit = 100;
+        private int initialLimit = 50;
+        private int minLimit = 1;
         private int maxConcurrency = 1000;
         private long minRttThreshold = TimeUnit.MICROSECONDS.toNanos(1);
 
-        private double smoothing = 0.2;
+        private double smoothing = 0.1;
         private Function<Integer, Integer> queueSize = SquareRootFunction.create(4);
         private MetricRegistry registry = EmptyMetricRegistry.INSTANCE;
         private double rttTolerance = 1.0;
-        
-        private int probeMultiplier = 30;
+        private int probeMultiplier = 10;
 
         /**
          * Minimum threshold for accepting a new rtt sample.  Any RTT lower than this threshold
@@ -62,6 +62,18 @@ public Builder initialLimit(int initialLimit) {
             return this;
         }
 
+        /**
+         * Minimum concurrency limit allowed.  The minimum helps prevent the algorithm from adjust the limit
+         * too far down.  Note that this limit is not desirable when use as backpressure for batch apps.
+         * 
+         * @param minLimit
+         * @return Chainable builder
+         */
+        public Builder minLimit(int minLimit) {
+            this.minLimit = minLimit;
+            return this;
+        }
+        
         /**
          * Tolerance for changes in minimum latency.  
          * @param rttTolerance Value {@literal >}= 1.0 indicating how much change in minimum latency is acceptable
@@ -132,13 +144,13 @@ public Builder metricRegistry(MetricRegistry registry) {
          * The limiter will probe for a new noload RTT every probeMultiplier * current limit
          * iterations.  Default value is 30. Set to -1 to disable 
          * @param probeMultiplier 
-         * @return Chinable builder
+         * @return Chainable builder
          */
         public Builder probeMultiplier(int probeMultiplier) {
             this.probeMultiplier = probeMultiplier;
             return this;
         }
-        
+
         public GradientLimit build() {
             return new GradientLimit(this);
         }
@@ -157,13 +169,15 @@ public static GradientLimit newDefault() {
      */
     private volatile double estimatedLimit;
 
-    private Measurement rttNoLoad = new MinimumMeasurement();
+    private final Measurement rttNoLoad;
 
     /**
      * Maximum allowed limit providing an upper bound failsafe
      */
     private final int maxLimit; 
 
+    private final int minLimit;
+    
     private final Function<Integer, Integer> queueSize;
 
     private final double smoothing;
@@ -185,12 +199,14 @@ public static GradientLimit newDefault() {
     private GradientLimit(Builder builder) {
         this.estimatedLimit = builder.initialLimit;
         this.maxLimit = builder.maxConcurrency;
+        this.minLimit = builder.minLimit;
         this.queueSize = builder.queueSize;
         this.smoothing = builder.smoothing;
         this.minRttThreshold = builder.minRttThreshold;
         this.rttTolerance = builder.rttTolerance;
         this.probeMultiplier = builder.probeMultiplier;
         this.resetRttCounter = nextProbeCountdown();
+        this.rttNoLoad = new SmoothingMinimumMeasurement(builder.smoothing);
 
         this.minRttSampleListener = builder.registry.registerDistribution(MetricIds.MIN_RTT_NAME);
         this.minWindowRttSampleListener = builder.registry.registerDistribution(MetricIds.WINDOW_MIN_RTT_NAME);
@@ -207,53 +223,63 @@ private int nextProbeCountdown() {
 
     @Override
     public synchronized void update(SampleWindow sample) {
-        final long rtt = sample.getCandidateRttNanos();
-        minWindowRttSampleListener.addSample(rtt);
-        
-        Preconditions.checkArgument(rtt > 0, "rtt must be >0 but got " + rtt);
+        Preconditions.checkArgument(sample.getCandidateRttNanos() > 0, "rtt must be >0 but got " + sample.getCandidateRttNanos());
 
-        if (rtt < minRttThreshold) {
+        if (sample.getCandidateRttNanos() < minRttThreshold) {
             return;
         }
+        
+        final long rtt = sample.getCandidateRttNanos(); // rttWindowNoLoad.get().longValue();
+        minWindowRttSampleListener.addSample(rtt);
 
         final double queueSize = this.queueSize.apply((int)this.estimatedLimit);
         queueSizeSampleListener.addSample(queueSize);
 
-        // Reset or probe for a new RTT and a new estimatedLimit.  It's necessary to cut the limit
+        // Reset or probe for a new noload RTT and a new estimatedLimit.  It's necessary to cut the limit
         // in half to avoid having the limit drift upwards when the RTT is probed during heavy load.
         // To avoid decreasing the limit too much we don't allow it to go lower than the queueSize.
         if (probeMultiplier != DISABLED && resetRttCounter-- <= 0) {
             resetRttCounter = nextProbeCountdown();
 
-            estimatedLimit = Math.max(estimatedLimit - queueSize, queueSize);
+            estimatedLimit = Math.max(minLimit, Math.max(estimatedLimit - queueSize, queueSize));
+            rttNoLoad.update(current -> rtt);
             LOG.debug("Probe MinRTT limit={}", getLimit());
-            rttNoLoad.reset();
             return;
         } else if (rttNoLoad.add(rtt)) {
             LOG.debug("New MinRTT {} limit={}", TimeUnit.NANOSECONDS.toMicros(rtt)/1000.0, getLimit());
         }
 
         minRttSampleListener.addSample(rttNoLoad.get());
 
-        final double gradient = Math.max(0.5, Math.min(1.0, rttTolerance * rttNoLoad.get() / rtt));
+        final double gradient;
+        // rtt is still higher than rtt_noload because of smoothing rtt noload updates
+        // set to 1.0 to indicate no queueing
+        if (rttNoLoad.get().doubleValue() > rtt) {
+            gradient = 1.0;
+        } else {
+            gradient = Math.max(0.5, rttTolerance * rttNoLoad.get().doubleValue() / rtt);
+        }
+        
         double newLimit;
+        // Reduce the limit aggressively if there was a drop
         if (sample.didDrop()) {
             newLimit = estimatedLimit/2;
+        // Don't grow the limit because we are app limited
         } else if ((estimatedLimit - sample.getMaxInFlight()) > queueSize) {
             return;
+        // Normal update to the limit
         } else {
             newLimit = estimatedLimit * gradient + queueSize;
         }
 
         newLimit = Math.max(queueSize, Math.min(maxLimit, newLimit));
-        if (newLimit < estimatedLimit) {
-            newLimit = estimatedLimit * (1-smoothing) + smoothing*(newLimit);
-        }
+        newLimit = Math.max(minLimit, estimatedLimit * (1-smoothing) + smoothing*(newLimit));
+        
         if ((int)newLimit != (int)estimatedLimit) {
             if (LOG.isDebugEnabled()) {
                 LOG.debug("New limit={} minRtt={} ms winRtt={} ms queueSize={} gradient={} resetCounter={}", 
                         (int)newLimit, 
-                        TimeUnit.NANOSECONDS.toMicros(rttNoLoad.get())/1000.0, 
+                        TimeUnit.NANOSECONDS.toMicros(rttNoLoad.get().longValue())/1000.0, 
                         TimeUnit.NANOSECONDS.toMicros(rtt)/1000.0,
                         queueSize,
                         gradient,
@@ -269,13 +295,13 @@ public int getLimit() {
     }
 
     public long getRttNoLoad() {
-        return rttNoLoad.get();
+        return rttNoLoad.get().longValue();
     }
 
     @Override
     public String toString() {
         return "GradientLimit [limit=" + (int)estimatedLimit + 
-                ", rtt_noload=" + TimeUnit.MICROSECONDS.toMillis(rttNoLoad.get()) / 1000.0+
+                ", rtt_noload=" + TimeUnit.MICROSECONDS.toMillis(rttNoLoad.get().longValue()) / 1000.0+
                 " ms]";
     }
 }

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

@@ -11,14 +11,14 @@ public interface Measurement {
      * @param sample
      * @return True if internal state was updated
      */
-    boolean add(long sample);
+    boolean add(Number sample);
 
-    long update(Function<Long, Long> func);
+    Number update(Function<Number, Number> func);
 
     /**
      * @return Return the current value
      */
-    long get();
+    Number get();
 
     /**
      * Reset the internal state as if no samples were ever added

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

@@ -3,30 +3,30 @@
 import java.util.function.Function;
 
 public class MinimumMeasurement implements Measurement {
-    private long value = 0;
+    private Double value = 0.0;
 
     @Override
-    public boolean add(long sample) {
-        if (value == 0 || sample < value) {
-            value = sample;
+    public boolean add(Number sample) {
+        if (value == 0.0 || sample.doubleValue() < value) {
+            value = sample.doubleValue();
             return true;
         }
         return false;
     }
 
     @Override
-    public long get() {
+    public Number get() {
         return value;
     }
 
     @Override
     public void reset() {
-        value = 0;
+        value = 0.0;
     }
 
     @Override
-    public long update(Function<Long, Long> func) {
-        value = func.apply(value);
+    public Number update(Function<Number, Number> func) {
+        value = func.apply(value).doubleValue();
         return value;
     }
 }

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

@@ -0,0 +1,49 @@
+package com.netflix.concurrency.limits.limit;
+
+import java.util.function.Function;
+
+/**
+ * Measures the minimum value of a sample set but also adds a smoothing factor
+ */
+public class SmoothingMinimumMeasurement implements Measurement {
+    private Double value = 0.0;
+    private final double ratio;
+    
+    /**
+     * @param smoothing Factor applied to the new measurement using the formula
+     *          value = value * (1-smoothing) + newValue * smoothing
+     */
+    public SmoothingMinimumMeasurement(double smoothing) {
+        this.ratio = smoothing;
+    }
+    
+    @Override
+    public boolean add(Number sample) {
+        if (value == 0) {
+            value = sample.doubleValue();
+            return true;
+        } else if (sample.doubleValue() < value.doubleValue()) {
+            value = value * ratio + sample.doubleValue() * (1-ratio);
+            return true;
+        }
+        
+        return false;
+    }
+
+    @Override
+    public Number get() {
+        return value;
+    }
+
+    @Override
+    public void reset() {
+        value = 0.0;
+    }
+
+    @Override
+    public Number update(Function<Number, Number> func) {
+        double newValue = func.apply(value).doubleValue();
+        value = value * (1-ratio) + newValue * ratio;
+        return value;
+    }
+}

concurrency-limits-grpc/src/test/java/com/netflix/concurrency/limits/grpc/server/example/Example.java

@@ -18,7 +18,6 @@
 import com.netflix.concurrency.limits.grpc.server.ConcurrencyLimitServerInterceptor;
 import com.netflix.concurrency.limits.grpc.server.GrpcServerLimiterBuilder;
 import com.netflix.concurrency.limits.limit.GradientLimit;
-import com.netflix.concurrency.limits.limit.VegasLimit;
 
 import io.grpc.CallOptions;
 import io.grpc.Channel;
@@ -46,20 +45,26 @@ public class Example {
 
 
     public static ServerCallHandler<String, String> createServerHandler(int concurrency) {
-        final ExponentialDistribution distribution = new ExponentialDistribution(10.0);
-        final Supplier<Integer> latency = () -> 100 + (int)distribution.sample();
+        final ExponentialDistribution distribution = new ExponentialDistribution(20.0);
+        final Supplier<Integer> latency = () -> 1 + (int)distribution.sample();
 
         List<Semaphore> semaphores = Arrays.asList(
-                new Semaphore(concurrency, true),
-                new Semaphore(concurrency*2, true),
-                new Semaphore(concurrency*4, true),
-                new Semaphore(concurrency*8, true));
+                new Semaphore(       concurrency,      true),
+                new Semaphore(       concurrency,      true),
+                new Semaphore(       concurrency,      true),
+                new Semaphore(       concurrency,      true),
+                new Semaphore(1, true)
+                );
+        
+        AtomicInteger position = new AtomicInteger(0);
+        Executors.newSingleThreadScheduledExecutor().scheduleAtFixedRate(() -> {
+            position.updateAndGet(current -> (current + 1) % semaphores.size());
+        }, 10, 10, TimeUnit.SECONDS);
 
         return ServerCalls.asyncUnaryCall(new UnaryMethod<String, String>() {
-            AtomicInteger counter = new AtomicInteger();
             @Override
             public void invoke(String req, StreamObserver<String> observer) {
-                Semaphore sem = semaphores.get((counter.incrementAndGet() / 2000) % semaphores.size());
+                Semaphore sem = semaphores.get(position.get());
                 try {
                     sem.acquire();
                     Uninterruptibles.sleepUninterruptibly(latency.get(), TimeUnit.MILLISECONDS);
@@ -86,7 +91,7 @@ public static void main(String[] args) throws IOException {
                     .build(), new ConcurrencyLimitServerInterceptor(new GrpcServerLimiterBuilder()
                             .limiter(builder -> builder
                                     .limit(limit)
-                                    .minWindowTime(200, TimeUnit.MILLISECONDS)
+                                    .minWindowTime(1, TimeUnit.SECONDS)
                                     )
                             .build())
                 ))
@@ -96,9 +101,10 @@ public static void main(String[] args) throws IOException {
         // Report progress
         AtomicInteger dropCount = new AtomicInteger(0);
         AtomicInteger successCount = new AtomicInteger(0);
-        
+        AtomicInteger counter = new AtomicInteger(0);
         Executors.newSingleThreadScheduledExecutor().scheduleAtFixedRate(() -> {
-            System.out.println(MessageFormat.format("{0}, {1}, {2}", limit.getLimit(), successCount.getAndSet(0), dropCount.getAndSet(0)));
+            System.out.println(MessageFormat.format("{0,number,#}, {1}, {2}, {3}", counter.incrementAndGet(), 
+                    limit.getLimit(), successCount.getAndSet(0), dropCount.getAndSet(0)));
         }, 1, 1, TimeUnit.SECONDS);
 
         // Create a client
@@ -107,7 +113,8 @@ public static void main(String[] args) throws IOException {
                 .build();
 
         DriverBuilder.newBuilder()
-            .exponential(3, 2, TimeUnit.SECONDS)
+            .exponential(3, 90, TimeUnit.SECONDS)
+            .exponential(1, 5, TimeUnit.SECONDS)
             .run(1, TimeUnit.HOURS, () -> {
                 ClientCalls.asyncUnaryCall(channel.newCall(METHOD_DESCRIPTOR, CallOptions.DEFAULT.withWaitForReady()), "request",
                         new StreamObserver<String>() {