Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

ISSUE-159 VegasLimit for batch processing #161

Open
wants to merge 1 commit into
base: main
Choose a base branch
from
Open

ISSUE-159 VegasLimit for batch processing #161

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
27 changes: 23 additions & 4 deletions concurrency-limits-core/src/main/java/com/netflix/concurrency/limits/limit/VegasLimit.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -48,7 +48,7 @@ public static class Builder {
private int maxConcurrency = 1000;
private MetricRegistry registry = EmptyMetricRegistry.INSTANCE;
private double smoothing = 1.0;

private double bufferFactor = 0.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());
Expand Down Expand Up @@ -104,7 +104,13 @@ public Builder decrease(Function<Double, Double> decrease) {
this.decreaseFunc = decrease;
return this;
}


public Builder bufferFactor(double bufferFactor) {
Preconditions.checkArgument(bufferFactor >= 0.0, "buffer factor must >= 0.0");
this.bufferFactor = bufferFactor;
return this;
}

public Builder smoothing(double smoothing) {
this.smoothing = smoothing;
return this;
Expand Down Expand Up @@ -154,7 +160,8 @@ public static VegasLimit newDefault() {
private volatile double estimatedLimit;

private volatile long rtt_noload = 0;


private volatile long pauseUpdateUntil = 0;
/**
* Maximum allowed limit providing an upper bound failsafe
*/
Expand All @@ -166,20 +173,24 @@ public static VegasLimit newDefault() {
private final Function<Integer, Integer> thresholdFunc;
private final Function<Double, Double> increaseFunc;
private final Function<Double, Double> decreaseFunc;
private final double bufferFactor;
private final int initialLimit;
private final SampleListener rttSampleListener;
private final int probeMultiplier;
private int probeCount = 0;
private double probeJitter;

private VegasLimit(Builder builder) {
super(builder.initialLimit);
this.initialLimit = builder.initialLimit;
this.estimatedLimit = builder.initialLimit;
this.maxLimit = builder.maxConcurrency;
this.alphaFunc = builder.alphaFunc;
this.betaFunc = builder.betaFunc;
this.increaseFunc = builder.increaseFunc;
this.decreaseFunc = builder.decreaseFunc;
this.thresholdFunc = builder.thresholdFunc;
this.bufferFactor = builder.bufferFactor;
this.smoothing = builder.smoothing;
this.probeMultiplier = builder.probeMultiplier;

Expand All @@ -206,6 +217,10 @@ protected int _update(long startTime, long rtt, int inflight, boolean didDrop) {
resetProbeJitter();
probeCount = 0;
rtt_noload = rtt;
if (bufferFactor > 0.0) {
pauseUpdateUntil = startTime + rtt + (long)(rtt * (bufferFactor / (bufferFactor + 1)));
estimatedLimit = Math.max(initialLimit, Math.ceil(estimatedLimit / (bufferFactor + 1)));
}
return (int)estimatedLimit;
}

Expand All @@ -217,11 +232,15 @@ protected int _update(long startTime, long rtt, int inflight, boolean didDrop) {

rttSampleListener.addSample(rtt_noload);

if (pauseUpdateUntil != 0 && pauseUpdateUntil > startTime) {
return (int)estimatedLimit;
}

return updateEstimatedLimit(rtt, inflight, didDrop);
}

private int updateEstimatedLimit(long rtt, int inflight, boolean didDrop) {
final int queueSize = (int) Math.ceil(estimatedLimit * (1 - (double)rtt_noload / rtt));
final int queueSize = (int) Math.ceil(estimatedLimit * (1 - (double)rtt_noload * (bufferFactor + 1) / rtt));

double newLimit;
// Treat any drop (i.e timeout) as needing to reduce the limit
Expand Down
45 changes: 44 additions & 1 deletion ...rrency-limits-core/src/test/java/com/netflix/concurrency/limits/limit/VegasLimitTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,9 @@
import junit.framework.Assert;
import org.junit.Test;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.TimeUnit;

public class VegasLimitTest {
Expand All @@ -15,6 +18,17 @@ public static VegasLimit create() {
.maxConcurrency(20)
.build();
}

public static VegasLimit create(double bufferFactor) {
return VegasLimit.newBuilder()
.alpha(3)
.beta(6)
.smoothing(1.0)
.initialLimit(10)
.bufferFactor(bufferFactor)
.maxConcurrency(20)
.build();
}

@Test
public void largeLimitIncrease() {
Expand Down Expand Up @@ -45,6 +59,15 @@ public void decreaseLimit() {
limit.onSample(0, TimeUnit.MILLISECONDS.toNanos(50), 11, false);
Assert.assertEquals(9, limit.getLimit());
}

@Test
public void decreaseLimitWithBufferFactor() {
VegasLimit limit = create(1.0);
limit.onSample(0, TimeUnit.MILLISECONDS.toNanos(10), 10, false);
Assert.assertEquals(10, limit.getLimit());
limit.onSample(0, TimeUnit.MILLISECONDS.toNanos(50), 11, false);
Assert.assertEquals(10, limit.getLimit());
}

@Test
public void noChangeIfWithinThresholds() {
Expand All @@ -54,7 +77,16 @@ public void noChangeIfWithinThresholds() {
limit.onSample(0, TimeUnit.MILLISECONDS.toNanos(14), 14, false);
Assert.assertEquals(10, limit.getLimit());
}


@Test
public void noChangeIfWithinThresholdsWithBuffer() {
VegasLimit limit = create(1.0);
limit.onSample(0, TimeUnit.MILLISECONDS.toNanos(10), 10, false);
Assert.assertEquals(10, limit.getLimit());
limit.onSample(0, TimeUnit.MILLISECONDS.toNanos(5), 14, false);
Assert.assertEquals(10, limit.getLimit());
}

@Test
public void decreaseSmoothing() {
VegasLimit limit = VegasLimit.newBuilder()
Expand Down Expand Up @@ -97,4 +129,15 @@ public void decreaseWithoutSmoothing() {
limit.onSample(0, TimeUnit.MILLISECONDS.toNanos(20), 100, false);
Assert.assertEquals(25, limit.getLimit());
}

@Test
public void pauseUpdateWhenProbe() {
VegasLimit limit = create(1.0);
List<Integer> limits = new ArrayList<>();
limit.notifyOnChange(v -> limits.add(v));
for (int i = 0; i < 600; ++i) {
limit.onSample(0, TimeUnit.MILLISECONDS.toNanos(10), 100, false);
}
Assert.assertEquals(Arrays.asList(16,20,10), limits);
}
}