1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
|
// Copyright Yahoo. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
package ai.vespa.feed.client.impl;
import ai.vespa.feed.client.HttpResponse;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicLong;
import static java.lang.Math.log;
import static java.lang.Math.max;
import static java.lang.Math.min;
import static java.lang.Math.pow;
import static java.lang.Math.random;
/**
* Samples latency as a function of inflight requests, and regularly adjusts to the optimal value.
*
* @author jonmv
*/
public class DynamicThrottler extends StaticThrottler {
private final AtomicLong ok = new AtomicLong(0);
private final AtomicLong targetInflight;
private final double weight = 0.7;
private final double[] throughputs = new double[128];
private long startNanos = System.nanoTime();
private long sent = 0;
public DynamicThrottler(FeedClientBuilderImpl builder) {
super(builder);
targetInflight = new AtomicLong(8 * minInflight);
}
@Override
public void sent(long __, CompletableFuture<HttpResponse> ___) {
double currentInflight = targetInflight.get();
if (++sent * sent * sent < 1e2 * currentInflight * currentInflight)
return;
sent = 0;
double elapsedNanos = -startNanos + (startNanos = System.nanoTime());
double currentThroughput = ok.getAndSet(0) / elapsedNanos;
// Use buckets for throughput over inflight, along the log-scale, in [minInflight, maxInflight).
int index = (int) (throughputs.length * log(max(1, min(255, currentInflight / minInflight)))
/ log(256)); // 4096 (server max streams per connection) / 16 (our min per connection)
throughputs[index] = currentThroughput;
// Loop over throughput measurements and pick the one which optimises throughput and latency.
double choice = currentInflight;
double max = -1;
for (int i = throughputs.length; i-- > 0; ) {
if (throughputs[i] == 0) continue; // Skip unknown values.
double inflight = minInflight * pow(256, (i + 0.5) / throughputs.length);
double objective = throughputs[i] * pow(inflight, (weight - 1)); // Optimise throughput (weight), but also latency (1 - weight).
if (objective > max) {
max = objective;
choice = inflight;
}
}
long target = (long) ((random() * 0.20 + 0.92) * choice); // Random walk, skewed towards increase.
targetInflight.set(max(minInflight, min(maxInflight, target)));
}
@Override
public void success() {
super.success();
ok.incrementAndGet();
}
@Override
public void throttled(long inflight) {
super.throttled(inflight);
}
@Override
public long targetInflight() {
return min(super.targetInflight(), targetInflight.get());
}
}
|
