Merge pull request #88 from elandau/bugfix/gradient2_drift

Fix gradient2 drift
Netflix · Oct 4, 2018 · 23299c8 · 23299c8
2 parents 6c95e64 + cbed3ee
commit 23299c8
Show file tree

Hide file tree

Showing 5 changed files with 44 additions and 13 deletions.
diff --git a/...rrency-limits-core/src/main/java/com/netflix/concurrency/limits/limit/Gradient2Limit.java b/...rrency-limits-core/src/main/java/com/netflix/concurrency/limits/limit/Gradient2Limit.java
@@ -21,6 +21,7 @@
 import com.netflix.concurrency.limits.internal.EmptyMetricRegistry;
 import com.netflix.concurrency.limits.limit.measurement.ExpAvgMeasurement;
 import com.netflix.concurrency.limits.limit.measurement.Measurement;
+import com.netflix.concurrency.limits.limit.measurement.SingleMeasurement;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -205,7 +206,7 @@ private Gradient2Limit(Builder builder) {
         this.minLimit = builder.minLimit;
         this.queueSize = builder.queueSize;
         this.smoothing = builder.smoothing;
-        this.shortRtt = new ExpAvgMeasurement(builder.shortWindow,10);
+        this.shortRtt = new SingleMeasurement(); // new ExpAvgMeasurement(builder.shortWindow,10);
         this.longRtt = new ExpAvgMeasurement(builder.longWindow,10);
         this.maxDriftIntervals = builder.shortWindow * builder.driftMultiplier;
 
@@ -237,6 +238,7 @@ public int _update(final long startTime, final long rtt, final int inflight, fin
             }
         } else {
             intervalsAbove = 0;
+            this.longRtt.update(ignore -> (longRtt + shortRtt) / 2);
         }
 
         shortRttSampleListener.addSample(shortRtt);

diff --git a/...ore/src/main/java/com/netflix/concurrency/limits/limit/measurement/ExpAvgMeasurement.java b/...ore/src/main/java/com/netflix/concurrency/limits/limit/measurement/ExpAvgMeasurement.java
@@ -15,29 +15,30 @@
  */
 package com.netflix.concurrency.limits.limit.measurement;
 
+import java.util.function.BiFunction;
 import java.util.function.Function;
 
 public class ExpAvgMeasurement implements Measurement {
     private Double value = 0.0;
-    private Double sum = 0.0;
     private final int window;
     private final int warmupWindow;
     private int count = 0;
+    private BiFunction<Double, Double, Double> warmupFunc;
 
     public ExpAvgMeasurement(int window, int warmupWindow) {
         this.window = window;
         this.warmupWindow = warmupWindow;
+        this.warmupFunc = Double::min;
     }
 
     @Override
     public Number add(Number sample) {
         if (count == 0) {
             count++;
-            sum = value = sample.doubleValue();
+            value = sample.doubleValue();
         } else if (count < warmupWindow) {
             count++;
-            sum += sample.doubleValue();
-            value = sum.doubleValue() / count;
+            value = warmupFunc.apply(value, sample.doubleValue());
         } else {
             double factor = factor(window);
             value = value * (1-factor) + sample.doubleValue() * factor;
@@ -58,7 +59,6 @@ public Number get() {
     public void reset() {
         value = 0.0;
         count = 0;
-        sum = 0.0;
     }
 
     @Override

diff --git a/...ore/src/main/java/com/netflix/concurrency/limits/limit/measurement/SingleMeasurement.java b/...ore/src/main/java/com/netflix/concurrency/limits/limit/measurement/SingleMeasurement.java
@@ -0,0 +1,27 @@
+package com.netflix.concurrency.limits.limit.measurement;
+
+import java.util.function.Function;
+
+public class SingleMeasurement implements Measurement {
+    private Number value = null;
+
+    @Override
+    public Number add(Number sample) {
+        return value = sample;
+    }
+
+    @Override
+    public Number get() {
+        return value;
+    }
+
+    @Override
+    public void reset() {
+        value = null;
+    }
+
+    @Override
+    public void update(Function<Number, Number> operation) {
+        value = operation.apply(value);
+    }
+}
diff --git a/...limits-core/src/test/java/com/netflix/concurrency/limits/limit/ExpAvgMeasurementTest.java b/...limits-core/src/test/java/com/netflix/concurrency/limits/limit/ExpAvgMeasurementTest.java
@@ -12,13 +12,13 @@ public class ExpAvgMeasurementTest {
     public void testWarmup() {
         ExpAvgMeasurement avg = new ExpAvgMeasurement(100, 10);
 
-        double expected[] = new double[]{10.0, 10.5, 11, 11.5, 12, 12.5, 13, 13.5, 14, 14.5};
+        double expected[] = new double[]{10.0, 10, 10, 10, 10, 10, 10, 10, 10, 10};
         for (int i = 0; i < 10; i++) {
             double value = avg.add(i + 10).doubleValue();
             Assert.assertEquals(expected[i], avg.get().doubleValue(), 0.01);
         }
 
         avg.add(100);
-        Assert.assertEquals(16.19, avg.get().doubleValue(), 0.01);
+        Assert.assertEquals(11.7, avg.get().doubleValue(), 0.1);
     }
 }
diff --git a/...limits-grpc/src/test/java/com/netflix/concurrency/limits/grpc/server/example/Example.java b/...limits-grpc/src/test/java/com/netflix/concurrency/limits/grpc/server/example/Example.java
@@ -15,7 +15,9 @@
 
 public class Example {
     public static void main(String[] args) throws IOException {
-        final Gradient2Limit limit = Gradient2Limit.newBuilder().build();
+        final Gradient2Limit limit = Gradient2Limit.newBuilder()
+                .longWindow(100)
+                .build();
 
         // Create a server
         final TestServer server = TestServer.newBuilder()
@@ -34,10 +36,10 @@ public static void main(String[] args) throws IOException {
         final LatencyCollector latency = new LatencyCollector();
 
         final Driver driver = Driver.newBuilder()
-            .exponentialRps(75,  200, TimeUnit.SECONDS)
-            .exponentialRps(100, 200, TimeUnit.SECONDS)
-            .exponentialRps(200, 200, TimeUnit.SECONDS)
-            .exponentialRps(100, 200, TimeUnit.SECONDS)
+            .exponentialRps(50,  400, TimeUnit.SECONDS)
+            .exponentialRps(100, 400, TimeUnit.SECONDS)
+            .exponentialRps(200, 400, TimeUnit.SECONDS)
+            .exponentialRps(100, 400, TimeUnit.SECONDS)
             .latencyAccumulator(latency)
             .runtime(1, TimeUnit.HOURS)
             .port(server.getPort())