Skip to content

Don't hold a lock in DelayedStream when calling realStream #1526

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

Merged
ejona86 merged 1 commit into from
Mar 9, 2016
Merged

Don't hold a lock in DelayedStream when calling realStream #1526

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
6 changes: 4 additions & 2 deletions core/src/main/java/io/grpc/internal/ClientCallImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@

package io.grpc.internal;

import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Preconditions.checkNotNull;
import static com.google.common.base.Preconditions.checkState;
import static com.google.common.util.concurrent.MoreExecutors.directExecutor;
Expand Down Expand Up @@ -208,11 +209,11 @@ public void runInContext() {
stream.setAuthority(callOptions.getAuthority());
}
stream.setCompressor(compressor);

stream.start(new ClientStreamListenerImpl(observer));
if (compressor != Codec.Identity.NONE) {
stream.setMessageCompression(true);
}

stream.start(new ClientStreamListenerImpl(observer));
// Delay any sources of cancellation after start(), because most of the transports are broken if
// they receive cancel before start. Issue #1343 has more details

Expand Down Expand Up @@ -269,6 +270,7 @@ private static Long getRemainingTimeoutNanos(@Nullable Long deadlineNanoTime) {
@Override
public void request(int numMessages) {
Preconditions.checkState(stream != null, "Not started");
checkArgument(numMessages >= 0, "Number requested must be non-negative");
stream.request(numMessages);
}

Expand Down
47 changes: 22 additions & 25 deletions core/src/main/java/io/grpc/internal/DelayedClientTransport.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -80,44 +80,41 @@ public void start(Listener listener) {
@Override
public ClientStream newStream(MethodDescriptor<?, ?> method, Metadata headers) {
Supplier<ClientTransport> supplier = transportSupplier;
if (supplier == null) {
synchronized (lock) {
// Check again, since it may have changed while waiting for lock
supplier = transportSupplier;
if (supplier == null && !shutdown) {
PendingStream pendingStream = new PendingStream(method, headers);
pendingStreams.add(pendingStream);
return pendingStream;
}
}
}
if (supplier != null) {
Copy link
Contributor

@carl-mastrangelo carl-mastrangelo Mar 8, 2016

Choose a reason for hiding this comment

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

What happens if supplier != null and shutdown is true here?

Copy link
Member Author

@ejona86 ejona86 Mar 8, 2016

Choose a reason for hiding this comment

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

After suppler != null, we don't care about shutdown. In the docs for shutdown():

Prevents creating any new streams until {@link #setTransport} is called.

After setTransport() is called, DelayedClientTransport does full delegation ignoring shutdown. This can seem a bit strange, but it is a really useful feature for current code as it allows cleaning up old DelayedClientTransports, is easier to implement, and there aren't any usages yet that would benefit from stricter behavior.

We use the dct.setTransport(realTransport); dct.shutdown(); pattern in both ManagedChannelImpl and TransportSet. Because of the race between obtainActiveTransport() returning and the caller creating a new stream, with a strict behavior it would be very hard to cleanup/shutdown a DelayedClientTransport without spurious failures.

return supplier.get().newStream(method, headers);
}
synchronized (lock) {
// Check again, since it may have changed while waiting for lock
supplier = transportSupplier;
if (supplier != null) {
return supplier.get().newStream(method, headers);
}
if (!shutdown) {
PendingStream pendingStream = new PendingStream(method, headers);
pendingStreams.add(pendingStream);
return pendingStream;
}
}
return new FailingClientStream(Status.UNAVAILABLE.withDescription("transport shutdown"));
}

@Override
public void ping(final PingCallback callback, Executor executor) {
Supplier<ClientTransport> supplier = transportSupplier;
if (supplier == null) {
synchronized (lock) {
// Check again, since it may have changed while waiting for lock
supplier = transportSupplier;
if (supplier == null && !shutdown) {
PendingPing pendingPing = new PendingPing(callback, executor);
pendingPings.add(pendingPing);
return;
}
}
}
if (supplier != null) {
supplier.get().ping(callback, executor);
return;
}
synchronized (lock) {
// Check again, since it may have changed while waiting for lock
supplier = transportSupplier;
if (supplier != null) {
supplier.get().ping(callback, executor);
return;
}
if (!shutdown) {
PendingPing pendingPing = new PendingPing(callback, executor);
pendingPings.add(pendingPing);
return;
}
}
executor.execute(new Runnable() {
@Override public void run() {
callback.onFailure(
Expand Down
Loading