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

core: fix bug RetriableStream cancel() racing with start() #8386

Merged
merged 2 commits into from Aug 6, 2021
Merged

core: fix bug RetriableStream cancel() racing with start() #8386

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
58a964e
core: fix bug RetriableStream cancel() racing with start()
dapengzhang0 Aug 5, 2021
8782060
move cancellationStatus assignment to a more meaningful place
dapengzhang0 Aug 6, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
43 changes: 29 additions & 14 deletions core/src/main/java/io/grpc/internal/RetriableStream.java
View file
Open in desktop
Expand Up @@ -104,6 +104,7 @@ abstract class RetriableStream<ReqT> implements ClientStream {
@GuardedBy("lock")
private FutureCanceller scheduledHedging;
private long nextBackoffIntervalNanos;
private Status cancellationStatus;

RetriableStream(
MethodDescriptor<ReqT, ?> method, Metadata headers,
Expand Down Expand Up @@ -244,14 +245,16 @@ private void drain(Substream substream) {
int index = 0;
int chunk = 0x80;
List<BufferEntry> list = null;
boolean streamStarted = false;

while (true) {
State savedState;

synchronized (lock) {
savedState = state;
if (savedState.winningSubstream != null && savedState.winningSubstream != substream) {
// committed but not me
if (savedState.winningSubstream != null && savedState.winningSubstream != substream
&& streamStarted) {
Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If streamStarted is false here, we can also short-circuit by returning directly.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Not quite. At the moment, every created stream has to be used. Things like DelayedClientTransport.newStream() add the stream to their tracking list, so you can't just drop the stream on the floor.

// committed but not me, to be cancelled
break;
}
if (index == savedState.buffer.size()) { // I'm drained
Expand All @@ -275,17 +278,22 @@ private void drain(Substream substream) {

for (BufferEntry bufferEntry : list) {
savedState = state;
if (savedState.winningSubstream != null && savedState.winningSubstream != substream) {
// committed but not me
if (savedState.winningSubstream != null && savedState.winningSubstream != substream
&& streamStarted) {
// committed but not me, to be cancelled
break;
}
if (savedState.cancelled) {
if (savedState.cancelled && streamStarted) {
checkState(
savedState.winningSubstream == substream,
"substream should be CANCELLED_BECAUSE_COMMITTED already");
substream.stream.cancel(cancellationStatus);
return;
}
bufferEntry.runWith(substream);
if (bufferEntry instanceof RetriableStream.StartEntry) {
streamStarted = true;
}
}
}

Expand All @@ -299,6 +307,13 @@ private void drain(Substream substream) {
@Nullable
abstract Status prestart();

class StartEntry implements BufferEntry {
@Override
public void runWith(Substream substream) {
substream.stream.start(new Sublistener(substream));
}
}

/** Starts the first PRC attempt. */
@Override
public final void start(ClientStreamListener listener) {
Expand All @@ -311,13 +326,6 @@ public final void start(ClientStreamListener listener) {
return;
}

class StartEntry implements BufferEntry {
@Override
public void runWith(Substream substream) {
substream.stream.start(new Sublistener(substream));
}
}

synchronized (lock) {
state.buffer.add(new StartEntry());
}
Expand Down Expand Up @@ -450,11 +458,18 @@ public final void cancel(Status reason) {
return;
}

state.winningSubstream.stream.cancel(reason);
Substream winningSubstreamToCancel = null;
synchronized (lock) {
// This is not required, but causes a short-circuit in the draining process.
if (state.drainedSubstreams.contains(state.winningSubstream)) {
winningSubstreamToCancel = state.winningSubstream;
} else { // the winningSubstream will be cancelled while draining
cancellationStatus = reason;
}
state = state.cancelled();
}
if (winningSubstreamToCancel != null) {
winningSubstreamToCancel.stream.cancel(reason);
}
}

private void delayOrExecute(BufferEntry bufferEntry) {
Expand Down