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

Do not reuse gRPC connections that fail with native Netty errors #23150

Closed
wants to merge 1 commit into from
Closed

Do not reuse gRPC connections that fail with native Netty errors #23150

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
1 change: 1 addition & 0 deletions src/main/java/com/google/devtools/build/lib/remote/grpc/BUILD
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,5 +20,6 @@ java_library(
"//third_party:jsr305",
"//third_party:rxjava3",
"//third_party/grpc-java:grpc-jar",
"@maven//:io_netty_netty_transport_native_unix_common",
],
)
40 changes: 37 additions & 3 deletions src/main/java/com/google/devtools/build/lib/remote/grpc/SharedConnectionFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,12 @@
import com.google.devtools.build.lib.concurrent.ThreadSafety.ThreadSafe;
import io.grpc.CallOptions;
import io.grpc.ClientCall;
import io.grpc.ForwardingClientCall.SimpleForwardingClientCall;
import io.grpc.ForwardingClientCallListener.SimpleForwardingClientCallListener;
import io.grpc.Metadata;
import io.grpc.MethodDescriptor;
import io.grpc.Status;
import io.netty.channel.unix.Errors;
import io.reactivex.rxjava3.core.Single;
import io.reactivex.rxjava3.disposables.Disposable;
import io.reactivex.rxjava3.functions.Action;
Expand Down Expand Up @@ -118,7 +123,13 @@ public Single<SharedConnection> create() {
.map(
conn ->
new SharedConnection(
conn, /* onClose= */ () -> tokenBucket.addToken(token))));
conn,
/* onClose= */ () -> tokenBucket.addToken(token),
/* onFatalError= */ () -> {
synchronized (this) {
connectionAsyncSubject = null;
}
})));
}

/** Returns current number of available connections. */
Expand All @@ -130,16 +141,33 @@ public int numAvailableConnections() {
public static class SharedConnection implements Connection {
private final Connection connection;
private final Action onClose;
private final Runnable onFatalError;

public SharedConnection(Connection connection, Action onClose) {
public SharedConnection(Connection connection, Action onClose, Runnable onFatalError) {
this.connection = connection;
this.onClose = onClose;
this.onFatalError = onFatalError;
}

@Override
public <ReqT, RespT> ClientCall<ReqT, RespT> call(
MethodDescriptor<ReqT, RespT> method, CallOptions options) {
return connection.call(method, options);
return new SimpleForwardingClientCall<>(connection.call(method, options)) {
@Override
public void start(Listener<RespT> responseListener, Metadata headers) {
super.start(
new SimpleForwardingClientCallListener<>(responseListener) {
@Override
public void onClose(Status status, Metadata trailers) {
if (isFatalError(status.getCause())) {
onFatalError.run();
}
super.onClose(status, trailers);
}
},
headers);
}
};
}

@Override
Expand All @@ -155,5 +183,11 @@ public void close() throws IOException {
public Connection getUnderlyingConnection() {
return connection;
}

private static boolean isFatalError(@Nullable Throwable t) {
// A low-level netty error indicates that the connection is fundamentally broken
// and should not be reused for retries.
return t instanceof Errors.NativeIoException;
Copy link
Collaborator Author

Choose a reason for hiding this comment

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

We could extend this to cover certain status codes, but I wanted to start with something limited enough to not introduce performance regressions.

}
}
}
2 changes: 2 additions & 0 deletions src/test/java/com/google/devtools/build/lib/remote/grpc/BUILD
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,5 +29,7 @@ java_test(
"//third_party:mockito",
"//third_party:rxjava3",
"//third_party:truth",
"@maven//:io_grpc_grpc_api",
"@maven//:io_netty_netty_transport_native_unix_common",
],
)
82 changes: 82 additions & 0 deletions src/test/java/com/google/devtools/build/lib/remote/grpc/SharedConnectionFactoryTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,16 +14,28 @@
package com.google.devtools.build.lib.remote.grpc;

import static com.google.common.truth.Truth.assertThat;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;

import com.google.devtools.build.lib.remote.grpc.SharedConnectionFactory.SharedConnection;
import com.google.devtools.build.lib.remote.util.RxNoGlobalErrorsRule;
import io.grpc.CallOptions;
import io.grpc.ClientCall;
import io.grpc.Metadata;
import io.grpc.MethodDescriptor;
import io.grpc.Status;
import io.netty.channel.unix.Errors;
import io.reactivex.rxjava3.core.Single;
import io.reactivex.rxjava3.observers.TestObserver;
import java.io.IOException;
import java.io.InputStream;
import java.util.ArrayDeque;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.Semaphore;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
Expand Down Expand Up @@ -145,6 +157,76 @@ public void create_concurrentCreate_shareConnections() throws InterruptedExcepti
verify(connectionFactory, times(1)).create();
}

@Test
public void create_belowMaxConcurrency_fatalErrorPreventsReuse() throws IOException {
Connection brokenConnection =
new Connection() {
@Override
public <ReqT, RespT> ClientCall<ReqT, RespT> call(
MethodDescriptor<ReqT, RespT> method, CallOptions options) {
var call = mock(ClientCall.class);
doAnswer(
invocationOnMock -> {
((ClientCall.Listener) invocationOnMock.getArgument(0))
Copy link
Collaborator Author

Choose a reason for hiding this comment

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

This is more of a whitebox test than I would have liked, but I didn't find another way to simulate a connection failure. Open to suggestions here to make this more realistic.

.onClose(
Status.fromThrowable(mock(Errors.NativeIoException.class)),
new Metadata());
return null;
})
.when(call)
.start(any(), any());
return call;
}

@Override
public void close() {}
};
Connection newConnection = mock(Connection.class);
Queue<Connection> connectionsToCreate =
new ArrayDeque<>(List.of(brokenConnection, newConnection));
when(connectionFactory.create())
.thenAnswer(invocation -> Single.just(connectionsToCreate.remove()));

SharedConnectionFactory factory = new SharedConnectionFactory(connectionFactory, 2);

TestObserver<SharedConnection> observer1 = factory.create().test();
assertThat(factory.numAvailableConnections()).isEqualTo(1);
observer1
.assertValue(conn -> conn.getUnderlyingConnection() == brokenConnection)
.assertComplete();

// Submit a call on the first connection and have it fail.
MethodDescriptor.Marshaller<byte[]> nullMarshaller =
new MethodDescriptor.Marshaller<>() {
@Override
public InputStream stream(byte[] bytes) {
return null;
}

@Override
public byte[] parse(InputStream inputStream) {
return null;
}
};
try (Connection firstConnection = observer1.values().getFirst()) {
var call =
firstConnection.call(
MethodDescriptor.newBuilder(nullMarshaller, nullMarshaller)
.setType(MethodDescriptor.MethodType.CLIENT_STREAMING)
.setFullMethodName("testMethod")
.build(),
CallOptions.DEFAULT);
ClientCall.Listener<byte[]> listener = new ClientCall.Listener<>() {};
call.start(listener, new io.grpc.Metadata());
listener.onClose(Status.fromThrowable(mock(Errors.NativeIoException.class)), new Metadata());
}

// Validate that the connection is not reused.
TestObserver<SharedConnection> observer2 = factory.create().test();
observer2.assertValue(conn -> conn.getUnderlyingConnection() == newConnection).assertComplete();
assertThat(factory.numAvailableConnections()).isEqualTo(1);
}

@Test
public void create_afterLastFailed_success() {
AtomicInteger times = new AtomicInteger(0);
Expand Down
Loading