mirror of https://github.com/grpc/grpc-java.git
core: use List instead of Set for drainedSubstreams
This improves latency performance (retry enabled) for NETTY transport by about 2us. For INPROCESS, NETTY_LOCAL, OKHTTP transports, the improvement seems much less than 2us. ``` before Benchmark (direct) (transport) Mode Cnt Score Error Units TransportBenchmark.unaryCall1024 true NETTY sample 260801 76566.536 ± 189.439 ns/op TransportBenchmark.unaryCall1024:unaryCall1024·p0.00 true NETTY sample 60480.000 ns/op TransportBenchmark.unaryCall1024:unaryCall1024·p0.50 true NETTY sample 75264.000 ns/op TransportBenchmark.unaryCall1024:unaryCall1024·p0.90 true NETTY sample 85504.000 ns/op TransportBenchmark.unaryCall1024:unaryCall1024·p0.95 true NETTY sample 87424.000 ns/op TransportBenchmark.unaryCall1024:unaryCall1024·p0.99 true NETTY sample 100864.000 ns/op TransportBenchmark.unaryCall1024:unaryCall1024·p0.999 true NETTY sample 140800.000 ns/op TransportBenchmark.unaryCall1024:unaryCall1024·p0.9999 true NETTY sample 205547.469 ns/op TransportBenchmark.unaryCall1024:unaryCall1024·p1.00 true NETTY sample 3915776.000 ns/op TransportBenchmark.unaryCall1024 false NETTY sample 208352 95865.619 ± 113.142 ns/op TransportBenchmark.unaryCall1024:unaryCall1024·p0.00 false NETTY sample 72576.000 ns/op TransportBenchmark.unaryCall1024:unaryCall1024·p0.50 false NETTY sample 93568.000 ns/op TransportBenchmark.unaryCall1024:unaryCall1024·p0.90 false NETTY sample 105728.000 ns/op TransportBenchmark.unaryCall1024:unaryCall1024·p0.95 false NETTY sample 109568.000 ns/op TransportBenchmark.unaryCall1024:unaryCall1024·p0.99 false NETTY sample 124544.000 ns/op TransportBenchmark.unaryCall1024:unaryCall1024·p0.999 false NETTY sample 161792.000 ns/op TransportBenchmark.unaryCall1024:unaryCall1024·p0.9999 false NETTY sample 230520.448 ns/op TransportBenchmark.unaryCall1024:unaryCall1024·p1.00 false NETTY sample 3997696.000 ns/op ``` ``` after Benchmark (direct) (transport) Mode Cnt Score Error Units TransportBenchmark.unaryCall1024 true NETTY sample 269471 74104.666 ± 182.514 ns/op TransportBenchmark.unaryCall1024:unaryCall1024·p0.00 true NETTY sample 60992.000 ns/op TransportBenchmark.unaryCall1024:unaryCall1024·p0.50 true NETTY sample 70912.000 ns/op TransportBenchmark.unaryCall1024:unaryCall1024·p0.90 true NETTY sample 83584.000 ns/op TransportBenchmark.unaryCall1024:unaryCall1024·p0.95 true NETTY sample 85888.000 ns/op TransportBenchmark.unaryCall1024:unaryCall1024·p0.99 true NETTY sample 100224.000 ns/op TransportBenchmark.unaryCall1024:unaryCall1024·p0.999 true NETTY sample 142848.000 ns/op TransportBenchmark.unaryCall1024:unaryCall1024·p0.9999 true NETTY sample 489527.706 ns/op TransportBenchmark.unaryCall1024:unaryCall1024·p1.00 true NETTY sample 4300800.000 ns/op TransportBenchmark.unaryCall1024 false NETTY sample 216000 92468.337 ± 90.229 ns/op TransportBenchmark.unaryCall1024:unaryCall1024·p0.00 false NETTY sample 68480.000 ns/op TransportBenchmark.unaryCall1024:unaryCall1024·p0.50 false NETTY sample 89472.000 ns/op TransportBenchmark.unaryCall1024:unaryCall1024·p0.90 false NETTY sample 103680.000 ns/op TransportBenchmark.unaryCall1024:unaryCall1024·p0.95 false NETTY sample 107008.000 ns/op TransportBenchmark.unaryCall1024:unaryCall1024·p0.99 false NETTY sample 120960.000 ns/op TransportBenchmark.unaryCall1024:unaryCall1024·p0.999 false NETTY sample 159232.000 ns/op TransportBenchmark.unaryCall1024:unaryCall1024·p0.9999 false NETTY sample 272076.493 ns/op TransportBenchmark.unaryCall1024:unaryCall1024·p1.00 false NETTY sample 2662400.000 ns/op ```
This commit is contained in:
ZHANG Dapeng
GitHub
parent
c8aa9f70ca
commit
74532acce0
|
|
@ -33,10 +33,8 @@ import java.io.InputStream;
|
||||||
import java.util.ArrayList;
|
import java.util.ArrayList;
|
||||||
import java.util.Collection;
|
import java.util.Collection;
|
||||||
import java.util.Collections;
|
import java.util.Collections;
|
||||||
import java.util.HashSet;
|
|
||||||
import java.util.List;
|
import java.util.List;
|
||||||
import java.util.Random;
|
import java.util.Random;
|
||||||
import java.util.Set;
|
|
||||||
import java.util.concurrent.Executor;
|
import java.util.concurrent.Executor;
|
||||||
import java.util.concurrent.Future;
|
import java.util.concurrent.Future;
|
||||||
import java.util.concurrent.ScheduledExecutorService;
|
import java.util.concurrent.ScheduledExecutorService;
|
||||||
|
|
@ -78,7 +76,7 @@ abstract class RetriableStream<ReqT> implements ClientStream {
|
||||||
private final Throttle throttle;
|
private final Throttle throttle;
|
||||||
|
|
||||||
private volatile State state = new State(
|
private volatile State state = new State(
|
||||||
new ArrayList<BufferEntry>(), Collections.<Substream>emptySet(), null, false, false);
|
new ArrayList<BufferEntry>(8), Collections.<Substream>emptyList(), null, false, false);
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Either transparent retry happened or reached server's application logic.
|
* Either transparent retry happened or reached server's application logic.
|
||||||
|
|
@ -218,10 +216,11 @@ abstract class RetriableStream<ReqT> implements ClientStream {
|
||||||
|
|
||||||
int stop = Math.min(index + chunk, savedState.buffer.size());
|
int stop = Math.min(index + chunk, savedState.buffer.size());
|
||||||
if (list == null) {
|
if (list == null) {
|
||||||
list = new ArrayList<BufferEntry>(stop - index);
|
list = new ArrayList<BufferEntry>(savedState.buffer.subList(index, stop));
|
||||||
|
} else {
|
||||||
|
list.clear();
|
||||||
|
list.addAll(savedState.buffer.subList(index, stop));
|
||||||
}
|
}
|
||||||
list.clear();
|
|
||||||
list.addAll(savedState.buffer.subList(index, stop));
|
|
||||||
index = stop;
|
index = stop;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -709,7 +708,7 @@ abstract class RetriableStream<ReqT> implements ClientStream {
|
||||||
boolean passThrough) {
|
boolean passThrough) {
|
||||||
this.buffer = buffer;
|
this.buffer = buffer;
|
||||||
this.drainedSubstreams =
|
this.drainedSubstreams =
|
||||||
Collections.unmodifiableCollection(checkNotNull(drainedSubstreams, "drainedSubstreams"));
|
checkNotNull(drainedSubstreams, "drainedSubstreams");
|
||||||
this.winningSubstream = winningSubstream;
|
this.winningSubstream = winningSubstream;
|
||||||
this.cancelled = cancelled;
|
this.cancelled = cancelled;
|
||||||
this.passThrough = passThrough;
|
this.passThrough = passThrough;
|
||||||
|
|
@ -738,10 +737,17 @@ abstract class RetriableStream<ReqT> implements ClientStream {
|
||||||
State substreamDrained(Substream substream) {
|
State substreamDrained(Substream substream) {
|
||||||
checkState(!passThrough, "Already passThrough");
|
checkState(!passThrough, "Already passThrough");
|
||||||
|
|
||||||
Set<Substream> drainedSubstreams = new HashSet<Substream>(this.drainedSubstreams);
|
Collection<Substream> drainedSubstreams;
|
||||||
|
|
||||||
if (!substream.closed) {
|
if (substream.closed) {
|
||||||
|
drainedSubstreams = this.drainedSubstreams;
|
||||||
|
} else if (this.drainedSubstreams.isEmpty()) {
|
||||||
|
// optimize for 0-retry, which is most of the cases.
|
||||||
|
drainedSubstreams = Collections.singletonList(substream);
|
||||||
|
} else {
|
||||||
|
drainedSubstreams = new ArrayList<Substream>(this.drainedSubstreams);
|
||||||
drainedSubstreams.add(substream);
|
drainedSubstreams.add(substream);
|
||||||
|
drainedSubstreams = Collections.unmodifiableCollection(drainedSubstreams);
|
||||||
}
|
}
|
||||||
|
|
||||||
boolean passThrough = winningSubstream != null;
|
boolean passThrough = winningSubstream != null;
|
||||||
|
|
@ -762,8 +768,9 @@ abstract class RetriableStream<ReqT> implements ClientStream {
|
||||||
State substreamClosed(Substream substream) {
|
State substreamClosed(Substream substream) {
|
||||||
substream.closed = true;
|
substream.closed = true;
|
||||||
if (this.drainedSubstreams.contains(substream)) {
|
if (this.drainedSubstreams.contains(substream)) {
|
||||||
Set<Substream> drainedSubstreams = new HashSet<Substream>(this.drainedSubstreams);
|
Collection<Substream> drainedSubstreams = new ArrayList<Substream>(this.drainedSubstreams);
|
||||||
drainedSubstreams.remove(substream);
|
drainedSubstreams.remove(substream);
|
||||||
|
drainedSubstreams = Collections.unmodifiableCollection(drainedSubstreams);
|
||||||
return new State(buffer, drainedSubstreams, winningSubstream, cancelled, passThrough);
|
return new State(buffer, drainedSubstreams, winningSubstream, cancelled, passThrough);
|
||||||
} else {
|
} else {
|
||||||
return this;
|
return this;
|
||||||
|
|
@ -777,12 +784,14 @@ abstract class RetriableStream<ReqT> implements ClientStream {
|
||||||
|
|
||||||
boolean passThrough = false;
|
boolean passThrough = false;
|
||||||
List<BufferEntry> buffer = this.buffer;
|
List<BufferEntry> buffer = this.buffer;
|
||||||
Collection<Substream> drainedSubstreams = Collections.emptySet();
|
Collection<Substream> drainedSubstreams;
|
||||||
|
|
||||||
if (this.drainedSubstreams.contains(winningSubstream)) {
|
if (this.drainedSubstreams.contains(winningSubstream)) {
|
||||||
passThrough = true;
|
passThrough = true;
|
||||||
buffer = null;
|
buffer = null;
|
||||||
drainedSubstreams = Collections.singleton(winningSubstream);
|
drainedSubstreams = Collections.singleton(winningSubstream);
|
||||||
|
} else {
|
||||||
|
drainedSubstreams = Collections.emptyList();
|
||||||
}
|
}
|
||||||
|
|
||||||
return new State(buffer, drainedSubstreams, winningSubstream, cancelled, passThrough);
|
return new State(buffer, drainedSubstreams, winningSubstream, cancelled, passThrough);
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue