Allowing Netty TLS bootstrap handler to be sharable.

Fixes #504

Author: nmittler

benchmarks/src/main/java/io/grpc/benchmarks/qps/ClientConfiguration.java

@@ -41,12 +41,11 @@
 
 import io.grpc.testing.PayloadType;
 import io.grpc.testing.RpcType;
+import io.grpc.testing.TestUtils;
 import io.grpc.transport.netty.NettyChannelBuilder;
 
-import java.net.InetAddress;
 import java.net.InetSocketAddress;
 import java.net.SocketAddress;
-import java.net.UnknownHostException;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.LinkedHashSet;
@@ -56,7 +55,6 @@
  * Configuration options for benchmark clients.
  */
 class ClientConfiguration implements Configuration {
- private static final String TESTCA_HOST = "foo.test.google.fr";
  private static final ClientConfiguration DEFAULT = new ClientConfiguration();
 
  Transport transport = Transport.NETTY_NIO;
@@ -117,14 +115,9 @@ protected ClientConfiguration build0(ClientConfiguration config) {
 
  if (config.testca && config.address instanceof InetSocketAddress) {
  // Override the socket address with the host from the testca.
- try {
- InetSocketAddress prevAddress = (InetSocketAddress) config.address;
- InetAddress inetAddress = InetAddress.getByName(prevAddress.getHostName());
- inetAddress = InetAddress.getByAddress(TESTCA_HOST, inetAddress.getAddress());
- config.address = new InetSocketAddress(inetAddress, prevAddress.getPort());
- } catch (UnknownHostException e) {
- throw new RuntimeException(e);
- }
+ InetSocketAddress address = (InetSocketAddress) config.address;
+ config.address = TestUtils.testServerAddress(address.getHostName(),
+ address.getPort());
  }
  }
 

build.gradle

@@ -252,6 +252,9 @@ subprojects {
  test {
  testLogging {
  exceptionFormat = 'full'
+ showExceptions true
+ showCauses true
+ showStackTraces true
  }
  }
 }

interop-testing/build.gradle

@@ -26,12 +26,6 @@ dependencies {
 
 test {
  jvmArgs "-Xbootclasspath/p:" + configurations.alpnboot.asPath
- testLogging {
- exceptionFormat "full"
- showExceptions true
- showCauses true
- showStackTraces true
- }
 }
 
 task test_client(type: CreateStartScripts) {

interop-testing/src/test/java/io/grpc/testing/integration/Http2NettyTest.java

@@ -43,8 +43,6 @@
 import org.junit.runners.JUnit4;
 
 import java.io.IOException;
-import java.net.InetAddress;
-import java.net.InetSocketAddress;
 
 /**
  * Integration tests for GRPC over HTTP2 using the Netty framework.
@@ -73,10 +71,8 @@ public static void stopServer() {
  @Override
  protected ChannelImpl createChannel() {
  try {
- InetAddress address
- = InetAddress.getByAddress("foo.test.google.fr", new byte[] {127, 0, 0, 1});
  return NettyChannelBuilder
- .forAddress(new InetSocketAddress(address, serverPort))
+ .forAddress(TestUtils.testServerAddress(serverPort))
  .sslContext(GrpcSslContexts.forClient().trustManager(
  TestUtils.loadCert("ca.pem")).build())
  .build();

interop-testing/src/test/java/io/grpc/testing/integration/Http2OkHttpTest.java

@@ -71,7 +71,7 @@ public static void stopServer() throws Exception {
  @Override
  protected ChannelImpl createChannel() {
  OkHttpChannelBuilder builder = OkHttpChannelBuilder.forAddress("127.0.0.1", serverPort)
- .overrideHostForAuthority("foo.test.google.fr");
+ .overrideHostForAuthority(TestUtils.TEST_SERVER_HOST);
  try {
  builder.sslSocketFactory(TestUtils.getSslSocketFactoryForCertainCert(
  TestUtils.loadCert("ca.pem")));

netty/build.gradle

@@ -5,7 +5,12 @@ dependencies {
  libraries.netty
 
  // Tests depend on base class defined by core module.
- testCompile project(':grpc-core').sourceSets.test.output
+ testCompile project(':grpc-core').sourceSets.test.output,
+ project(':grpc-testing')
+}
+
+test {
+ jvmArgs "-Xbootclasspath/p:" + configurations.alpnboot.asPath
 }
 
 javadoc.options.links 'https://netty.io/4.1/api/'

netty/src/main/java/io/grpc/transport/netty/NettyClientTransport.java

@@ -37,9 +37,11 @@
 
 import io.grpc.Metadata;
 import io.grpc.MethodDescriptor;
+import io.grpc.Status;
 import io.grpc.transport.ClientStream;
 import io.grpc.transport.ClientStreamListener;
 import io.grpc.transport.ClientTransport;
+
 import io.netty.bootstrap.Bootstrap;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelFuture;
@@ -123,16 +125,27 @@ public ClientStream newStream(MethodDescriptor<?, ?> method, Metadata.Headers he
  Preconditions.checkNotNull(listener, "listener");
 
  // Create the stream.
- NettyClientStream stream = new NettyClientStream(listener, channel, handler);
+ final NettyClientStream stream = new NettyClientStream(listener, channel, handler);
 
  // Convert the headers into Netty HTTP/2 headers.
  AsciiString defaultPath = new AsciiString("/" + method.getName());
  Http2Headers http2Headers = Utils.convertClientHeaders(headers, negotiationHandler.scheme(),
  defaultPath, authority);
 
+ ChannelFutureListener failureListener = new ChannelFutureListener() {
+ @Override
+ public void operationComplete(ChannelFuture future) throws Exception {
+ if (!future.isSuccess()) {
+ // Stream creation failed. Close the stream if not already closed.
+ stream.transportReportStatus(Status.fromThrowable(future.cause()), true,
+ new Metadata.Trailers());
+ }
+ }
+ };
+
  // Write the command requesting the creation of the stream.
  handler.getWriteQueue().enqueue(new CreateStreamCommand(http2Headers, stream),
- !method.getType().clientSendsOneMessage());
+  !method.getType().clientSendsOneMessage()).addListener(failureListener);
  return stream;
  }
 
@@ -159,19 +172,7 @@ public void start(Listener transportListener) {
  channel.closeFuture().addListener(new ChannelFutureListener() {
  @Override
  public void operationComplete(ChannelFuture future) throws Exception {
- if (!future.isSuccess()) {
- // The close failed. Just notify that transport shutdown failed.
- notifyTerminated(future.cause());
- return;
- }
-
- if (handler.connectionError() != null) {
- // The handler encountered a connection error.
- notifyTerminated(handler.connectionError());
- } else {
- // Normal termination of the connection.
- notifyTerminated(null);
- }
+ notifyTerminated(handler.connectionError());
  }
  });
  }

netty/src/main/java/io/grpc/transport/netty/NettyServerTransport.java

@@ -90,15 +90,7 @@ public void start(ServerTransportListener listener) {
  channel.closeFuture().addListener(new ChannelFutureListener() {
  @Override
  public void operationComplete(ChannelFuture future) throws Exception {
- if (!future.isSuccess()) {
- notifyTerminated(future.cause());
- } else if (handler.connectionError() != null) {
- // The handler encountered a connection error.
- notifyTerminated(handler.connectionError());
- } else {
- // Normal termination of the connection.
- notifyTerminated(null);
- }
+ notifyTerminated(handler.connectionError());
  }
  });
 

netty/src/main/java/io/grpc/transport/netty/ProtocolNegotiators.java

@@ -34,6 +34,8 @@
 import com.google.common.base.Preconditions;
 import com.google.common.util.concurrent.SettableFuture;
 
+import io.grpc.Status;
+
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelDuplexHandler;
 import io.netty.channel.ChannelHandler;
@@ -116,31 +118,33 @@ public static ProtocolNegotiator tls(final SslContext sslContext,
  Preconditions.checkNotNull(sslContext, "sslContext");
  Preconditions.checkNotNull(inetAddress, "inetAddress");
 
- final ChannelHandler sslBootstrapHandler = new ChannelHandlerAdapter() {
+ return new ProtocolNegotiator() {
  @Override
- public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
- // TODO(nmittler): This method is currently unsupported for OpenSSL. Need to fix in Netty.
- SSLEngine sslEngine = sslContext.newEngine(ctx.alloc(),
- inetAddress.getHostName(), inetAddress.getPort());
- SSLParameters sslParams = new SSLParameters();
- sslParams.setEndpointIdentificationAlgorithm("HTTPS");
- sslEngine.setSSLParameters(sslParams);
-
- final SettableFuture<Void> completeFuture = SettableFuture.create();
- if (isOpenSsl(sslContext.getClass())) {
- completeFuture.set(null);
- } else {
- // Using JDK SSL
- if (!installJettyTlsProtocolSelection(sslEngine, completeFuture, false)) {
- throw new IllegalStateException("NPN/ALPN extensions not installed");
- }
- }
+ public Handler newHandler(Http2ConnectionHandler handler) {
+ ChannelHandler sslBootstrap = new ChannelHandlerAdapter() {
+ @Override
+ public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
+ // TODO(nmittler): Unsupported for OpenSSL in Netty < 4.1.Beta6.
+ SSLEngine sslEngine = sslContext.newEngine(ctx.alloc(),
+ inetAddress.getHostName(), inetAddress.getPort());
+ SSLParameters sslParams = new SSLParameters();
+ sslParams.setEndpointIdentificationAlgorithm("HTTPS");
+ sslEngine.setSSLParameters(sslParams);
+
+ final SettableFuture<Void> completeFuture = SettableFuture.create();
+ if (isOpenSsl(sslContext.getClass())) {
+ completeFuture.set(null);
+ } else {
+ // Using JDK SSL
+ if (!installJettyTlsProtocolSelection(sslEngine, completeFuture, false)) {
+ throw new IllegalStateException("NPN/ALPN extensions not installed");
+ }
+ }
 
- SslHandler sslHandler = new SslHandler(sslEngine, false);
- sslHandler.handshakeFuture().addListener(
- new GenericFutureListener<Future<? super Channel>>() {
+ SslHandler sslHandler = new SslHandler(sslEngine, false);
+ sslHandler.handshakeFuture().addListener(new GenericFutureListener<Future<Channel>>() {
  @Override
- public void operationComplete(Future<? super Channel> future) throws Exception {
+ public void operationComplete(Future<Channel> future) throws Exception {
  // If an error occurred during the handshake, throw it to the pipeline.
  if (future.isSuccess()) {
  completeFuture.get();
@@ -149,13 +153,10 @@ public void operationComplete(Future<? super Channel> future) throws Exception {
  }
  }
  });
- ctx.pipeline().replace(this, "sslHandler", sslHandler);
- }
- };
- return new ProtocolNegotiator() {
- @Override
- public Handler newHandler(Http2ConnectionHandler handler) {
- return new BufferUntilTlsNegotiatedHandler(sslBootstrapHandler, handler);
+ ctx.pipeline().replace(this, null, sslHandler);
+ }
+ };
+ return new BufferUntilTlsNegotiatedHandler(sslBootstrap, handler);
  }
  };
  }
@@ -190,6 +191,10 @@ public Handler newHandler(Http2ConnectionHandler handler) {
  };
  }
 
+ private static RuntimeException unavailableException(String msg) {
+ return Status.UNAVAILABLE.withDescription(msg).asRuntimeException();
+ }
+
  /**
  * Returns {@code true} if the given class is for use with Netty OpenSsl.
  */
@@ -199,7 +204,7 @@ private static boolean isOpenSsl(Class<?> clazz) {
 
  /**
  * Buffers all writes until either {@link #writeBufferedAndRemove(ChannelHandlerContext)} or
- * {@link #failBufferedAndClose(ChannelHandlerContext)} is called. This handler allows us to
+ * {@link #fail(ChannelHandlerContext, Throwable)} is called. This handler allows us to
  * write to a {@link Channel} before we are allowed to write to it officially i.e.
  * before it's active or the TLS Handshake is complete.
  */
@@ -231,9 +236,14 @@ public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
  super.channelRegistered(ctx);
  }
 
+ @Override
+ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
+ fail(ctx, cause);
+ }
+
  @Override
  public void channelInactive(ChannelHandlerContext ctx) throws Exception {
- failBufferedAndClose(ctx);
+ fail(ctx, unavailableException("Connection broken while performing protocol negotiation"));
  super.channelInactive(ctx);
  }
 
@@ -275,26 +285,28 @@ public void flush(ChannelHandlerContext ctx) {
 
  @Override
  public void close(ChannelHandlerContext ctx, ChannelPromise future) throws Exception {
- failBufferedAndClose(ctx);
+ fail(ctx, unavailableException("Channel closed while performing protocol negotiation"));
  }
 
- protected void failBufferedAndClose(ChannelHandlerContext ctx) {
+ protected final void fail(ChannelHandlerContext ctx, Throwable cause) {
  if (bufferedWrites != null) {
- Exception e = new Exception("Buffered write failed.");
  while (!bufferedWrites.isEmpty()) {
  ChannelWrite write = bufferedWrites.poll();
- write.promise.setFailure(e);
+ write.promise.setFailure(cause);
  }
  bufferedWrites = null;
  }
+
+ log.log(Level.SEVERE, "Transport failed during protocol negotiation", cause);
+
  /**
  * In case something goes wrong ensure that the channel gets closed as the
  * NettyClientTransport relies on the channel's close future to get completed.
  */
  ctx.close();
  }
 
- protected void writeBufferedAndRemove(ChannelHandlerContext ctx) {
+ protected final void writeBufferedAndRemove(ChannelHandlerContext ctx) {
  if (!ctx.channel().isActive() || writing) {
  return;
  }
@@ -349,7 +361,7 @@ public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exc
  if (handshakeEvent.isSuccess()) {
  writeBufferedAndRemove(ctx);
  } else {
- failBufferedAndClose(ctx);
+ fail(ctx, handshakeEvent.cause());
  }
  }
  super.userEventTriggered(ctx, evt);
@@ -413,8 +425,7 @@ public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exc
  if (evt == HttpClientUpgradeHandler.UpgradeEvent.UPGRADE_SUCCESSFUL) {
  writeBufferedAndRemove(ctx);
  } else if (evt == HttpClientUpgradeHandler.UpgradeEvent.UPGRADE_REJECTED) {
- failBufferedAndClose(ctx);
- ctx.pipeline().fireExceptionCaught(new Exception("HTTP/2 upgrade rejected"));
+ fail(ctx, unavailableException("HTTP/2 upgrade rejected"));
  }
  super.userEventTriggered(ctx, evt);
  }

netty/src/main/java/io/grpc/transport/netty/WriteQueue.java

@@ -34,6 +34,7 @@
 import com.google.common.base.Preconditions;
 
 import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelPromise;
 
 import java.util.ArrayDeque;
@@ -92,8 +93,8 @@ void scheduleFlush() {
  * @param flush true if a flush of the write should be schedule, false if a later call to
  * enqueue will schedule the flush.
  */
- void enqueue(Object command, boolean flush) {
- enqueue(command, channel.newPromise(), flush);
+ ChannelFuture enqueue(Object command, boolean flush) {
+ return enqueue(command, channel.newPromise(), flush);
  }
 
  /**
@@ -104,11 +105,12 @@ void enqueue(Object command, boolean flush) {
  * @param flush true if a flush of the write should be schedule, false if a later call to
  * enqueue will schedule the flush.
  */
- void enqueue(Object command, ChannelPromise promise, boolean flush) {
+ ChannelFuture enqueue(Object command, ChannelPromise promise, boolean flush) {
  queue.add(new QueuedCommand(command, promise));
  if (flush) {
  scheduleFlush();
  }
+ return promise;
  }
 
  /**