netty: Add Executor When Creating SslContext

This PR is to add one more Executor parameter when creating the SslContext.
In Netty, we already have this implementation for passing Executor when creating SslContext: netty/netty#8847. This extra Executor is used to take some time-consuming tasks when doing SSL handshake. However, in current gRPC implementation, we are not using this API.

In this PR, the relevant changes are:

1. get the executorPool from ChannelBuilder or ServerBuilder

2. pass the executorPool all the way down to ClientTlsHandler

3. fill executorPool in when creating SslHandler

Author: ZhenLian

core/src/main/java/io/grpc/internal/AbstractManagedChannelImplBuilder.java

@@ -657,4 +657,11 @@ private T thisT() {
     T thisT = (T) this;
     return thisT;
   }
+
+  /**
+   * Returns the internal offload executor pool for offloading tasks.
+   */
+  protected ObjectPool<? extends Executor> getOffloadExecutorPool() {
+    return this.offloadExecutorPool;
+  }
 }

core/src/main/java/io/grpc/internal/AbstractServerImplBuilder.java

@@ -297,4 +297,11 @@ public List<ServerServiceDefinition> getServices() {
       return null;
     }
   }
+
+  /**
+   * Returns the internal ExecutorPool for offloading tasks.
+   */
+  protected ObjectPool<? extends Executor> getExecutorPool() {
+    return this.executorPool;
+  }
 }

netty/src/main/java/io/grpc/netty/InternalProtocolNegotiators.java

@@ -159,7 +159,7 @@ public static ChannelHandler grpcNegotiationHandler(GrpcHttp2ConnectionHandler n
 
   public static ChannelHandler clientTlsHandler(
       ChannelHandler next, SslContext sslContext, String authority) {
-    return new ClientTlsHandler(next, sslContext, authority);
+    return new ClientTlsHandler(next, sslContext, authority, null);
   }
 
   public static class ProtocolNegotiationHandler

netty/src/main/java/io/grpc/netty/NettyChannelBuilder.java

@@ -51,6 +51,7 @@
 import java.net.SocketAddress;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.concurrent.Executor;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import javax.annotation.CheckReturnValue;
@@ -398,7 +399,8 @@ protected ClientTransportFactory buildTransportFactory() {
           throw new RuntimeException(ex);
         }
       }
-      negotiator = createProtocolNegotiatorByType(negotiationType, localSslContext);
+      negotiator = createProtocolNegotiatorByType(negotiationType, localSslContext,
+          this.getOffloadExecutorPool());
     }
 
     return new NettyTransportFactory(
@@ -441,14 +443,15 @@ void overrideAuthorityChecker(@Nullable OverrideAuthorityChecker authorityChecke
   @CheckReturnValue
   static ProtocolNegotiator createProtocolNegotiatorByType(
       NegotiationType negotiationType,
-      SslContext sslContext) {
+      SslContext sslContext,
+      ObjectPool<? extends Executor> executorPool) {
     switch (negotiationType) {
       case PLAINTEXT:
         return ProtocolNegotiators.plaintext();
       case PLAINTEXT_UPGRADE:
         return ProtocolNegotiators.plaintextUpgrade();
       case TLS:
-        return ProtocolNegotiators.tls(sslContext);
+        return ProtocolNegotiators.tls(sslContext, executorPool);
       default:
         throw new IllegalArgumentException("Unsupported negotiationType: " + negotiationType);
     }

netty/src/main/java/io/grpc/netty/NettyServerBuilder.java

@@ -533,8 +533,9 @@ protected List<NettyServer> buildTransportServers(
 
     ProtocolNegotiator negotiator = protocolNegotiator;
     if (negotiator == null) {
-      negotiator = sslContext != null ? ProtocolNegotiators.serverTls(sslContext) :
-          ProtocolNegotiators.serverPlaintext();
+      negotiator = sslContext != null
+          ? ProtocolNegotiators.serverTls(sslContext, this.getExecutorPool())
+          : ProtocolNegotiators.serverPlaintext();
     }
 
     List<NettyServer> transportServers = new ArrayList<>(listenAddresses.size());

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

@@ -33,6 +33,7 @@
 import io.grpc.Status;
 import io.grpc.internal.GrpcAttributes;
 import io.grpc.internal.GrpcUtil;
+import io.grpc.internal.ObjectPool;
 import io.netty.channel.ChannelDuplexHandler;
 import io.netty.channel.ChannelFutureListener;
 import io.netty.channel.ChannelHandler;
@@ -58,6 +59,7 @@
 import java.net.SocketAddress;
 import java.net.URI;
 import java.util.Arrays;
+import java.util.concurrent.Executor;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 import javax.annotation.Nullable;
@@ -106,20 +108,35 @@ public static ProtocolNegotiator serverPlaintext() {
 
   /**
    * Create a server TLS handler for HTTP/2 capable of using ALPN/NPN.
+   * @param executorPool a dedicated {@link Executor} pool for time-consuming TLS tasks
    */
-  public static ProtocolNegotiator serverTls(final SslContext sslContext) {
+  public static ProtocolNegotiator serverTls(final SslContext sslContext,
+      final ObjectPool<? extends Executor> executorPool) {
     Preconditions.checkNotNull(sslContext, "sslContext");
+    final Executor executor;
+    if (executorPool != null) {
+      // The handlers here can out-live the {@link ProtocolNegotiator}.
+      // To keep their own reference to executor from executorPool, we use an extra (unused)
+      // reference here forces the executor to stay alive, which prevents it from being re-created
+      // for every connection.
+      executor = executorPool.getObject();
+    } else {
+      executor = null;
+    }
     return new ProtocolNegotiator() {
       @Override
       public ChannelHandler newHandler(GrpcHttp2ConnectionHandler handler) {
         ChannelHandler gnh = new GrpcNegotiationHandler(handler);
-        ChannelHandler sth = new ServerTlsHandler(gnh, sslContext);
+        ChannelHandler sth = new ServerTlsHandler(gnh, sslContext, executorPool);
         return new WaitUntilActiveHandler(sth);
       }
 
       @Override
-      public void close() {}
-
+      public void close() {
+        if (executorPool != null && executor != null) {
+          executorPool.returnObject(executor);
+        }
+      }
 
       @Override
       public AsciiString scheme() {
@@ -128,22 +145,37 @@ public AsciiString scheme() {
     };
   }
 
+  /**
+   * Create a server TLS handler for HTTP/2 capable of using ALPN/NPN.
+   */
+  public static ProtocolNegotiator serverTls(final SslContext sslContext) {
+    return serverTls(sslContext, null);
+  }
+
   static final class ServerTlsHandler extends ChannelInboundHandlerAdapter {
+    private Executor executor;
     private final ChannelHandler next;
     private final SslContext sslContext;
 
     private ProtocolNegotiationEvent pne = ProtocolNegotiationEvent.DEFAULT;
 
-    ServerTlsHandler(ChannelHandler next, SslContext sslContext) {
+    ServerTlsHandler(ChannelHandler next,
+        SslContext sslContext,
+        final ObjectPool<? extends Executor> executorPool) {
       this.sslContext = checkNotNull(sslContext, "sslContext");
       this.next = checkNotNull(next, "next");
+      if (executorPool != null) {
+        this.executor = executorPool.getObject();
+      }
     }
 
     @Override
     public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
       super.handlerAdded(ctx);
       SSLEngine sslEngine = sslContext.newEngine(ctx.alloc());
-      ctx.pipeline().addBefore(ctx.name(), null, new SslHandler(sslEngine, false));
+      ctx.pipeline().addBefore(ctx.name(), /* name= */ null, this.executor != null
+          ? new SslHandler(sslEngine, false, this.executor)
+          : new SslHandler(sslEngine, false));
     }
 
     @Override
@@ -259,11 +291,18 @@ protected void userEventTriggered0(ChannelHandlerContext ctx, Object evt) throws
 
   static final class ClientTlsProtocolNegotiator implements ProtocolNegotiator {
 
-    public ClientTlsProtocolNegotiator(SslContext sslContext) {
+    public ClientTlsProtocolNegotiator(SslContext sslContext,
+        ObjectPool<? extends Executor> executorPool) {
       this.sslContext = checkNotNull(sslContext, "sslContext");
+      this.executorPool = executorPool;
+      if (this.executorPool != null) {
+        this.executor = this.executorPool.getObject();
+      }
     }
 
     private final SslContext sslContext;
+    private final ObjectPool<? extends Executor> executorPool;
+    private Executor executor;
 
     @Override
     public AsciiString scheme() {
@@ -273,26 +312,34 @@ public AsciiString scheme() {
     @Override
     public ChannelHandler newHandler(GrpcHttp2ConnectionHandler grpcHandler) {
       ChannelHandler gnh = new GrpcNegotiationHandler(grpcHandler);
-      ChannelHandler cth = new ClientTlsHandler(gnh, sslContext, grpcHandler.getAuthority());
+      ChannelHandler cth = new ClientTlsHandler(gnh, sslContext, grpcHandler.getAuthority(),
+          this.executor);
       return new WaitUntilActiveHandler(cth);
     }
 
     @Override
-    public void close() {}
+    public void close() {
+      if (this.executorPool != null && this.executor != null) {
+        this.executorPool.returnObject(this.executor);
+      }
+    }
   }
 
   static final class ClientTlsHandler extends ProtocolNegotiationHandler {
 
     private final SslContext sslContext;
     private final String host;
     private final int port;
+    private Executor executor;
 
-    ClientTlsHandler(ChannelHandler next, SslContext sslContext, String authority) {
+    ClientTlsHandler(ChannelHandler next, SslContext sslContext, String authority,
+        Executor executor) {
       super(next);
       this.sslContext = checkNotNull(sslContext, "sslContext");
       HostPort hostPort = parseAuthority(authority);
       this.host = hostPort.host;
       this.port = hostPort.port;
+      this.executor = executor;
     }
 
     @Override
@@ -301,7 +348,9 @@ protected void handlerAdded0(ChannelHandlerContext ctx) {
       SSLParameters sslParams = sslEngine.getSSLParameters();
       sslParams.setEndpointIdentificationAlgorithm("HTTPS");
       sslEngine.setSSLParameters(sslParams);
-      ctx.pipeline().addBefore(ctx.name(), /* name= */ null, new SslHandler(sslEngine, false));
+      ctx.pipeline().addBefore(ctx.name(), /* name= */ null, this.executor != null
+          ? new SslHandler(sslEngine, false, this.executor)
+          : new SslHandler(sslEngine, false));
     }
 
     @Override
@@ -363,13 +412,24 @@ static HostPort parseAuthority(String authority) {
     return new HostPort(host, port);
   }
 
+  /**
+   * Returns a {@link ProtocolNegotiator} that ensures the pipeline is set up so that TLS will
+   * be negotiated, the {@code handler} is added and writes to the {@link io.netty.channel.Channel}
+   * may happen immediately, even before the TLS Handshake is complete.
+   * @param executorPool a dedicated {@link Executor} pool for time-consuming TLS tasks
+   */
+  public static ProtocolNegotiator tls(SslContext sslContext,
+      ObjectPool<? extends Executor> executorPool) {
+    return new ClientTlsProtocolNegotiator(sslContext, executorPool);
+  }
+
   /**
    * Returns a {@link ProtocolNegotiator} that ensures the pipeline is set up so that TLS will
    * be negotiated, the {@code handler} is added and writes to the {@link io.netty.channel.Channel}
    * may happen immediately, even before the TLS Handshake is complete.
    */
   public static ProtocolNegotiator tls(SslContext sslContext) {
-    return new ClientTlsProtocolNegotiator(sslContext);
+    return tls(sslContext, null);
   }
 
   /** A tuple of (host, port). */

netty/src/test/java/io/grpc/netty/NettyChannelBuilderTest.java

@@ -17,11 +17,13 @@
 package io.grpc.netty;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertTrue;
 import static org.mockito.Mockito.mock;
 
 import io.grpc.ManagedChannel;
 import io.grpc.netty.InternalNettyChannelBuilder.OverrideAuthorityChecker;
+import io.grpc.netty.NettyTestUtil.TrackingObjectPoolForTest;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelFactory;
 import io.netty.channel.EventLoopGroup;
@@ -143,26 +145,42 @@ public void failIfSslContextIsNotClient() {
   public void createProtocolNegotiatorByType_plaintext() {
     ProtocolNegotiator negotiator = NettyChannelBuilder.createProtocolNegotiatorByType(
         NegotiationType.PLAINTEXT,
-        noSslContext);
+        noSslContext, null);
     // just check that the classes are the same, and that negotiator is not null.
     assertTrue(negotiator instanceof ProtocolNegotiators.PlaintextProtocolNegotiator);
+    negotiator.close();
   }
 
   @Test
   public void createProtocolNegotiatorByType_plaintextUpgrade() {
     ProtocolNegotiator negotiator = NettyChannelBuilder.createProtocolNegotiatorByType(
         NegotiationType.PLAINTEXT_UPGRADE,
-        noSslContext);
+        noSslContext, null);
     // just check that the classes are the same, and that negotiator is not null.
     assertTrue(negotiator instanceof ProtocolNegotiators.PlaintextUpgradeProtocolNegotiator);
+    negotiator.close();
   }
 
   @Test
   public void createProtocolNegotiatorByType_tlsWithNoContext() {
     thrown.expect(NullPointerException.class);
     NettyChannelBuilder.createProtocolNegotiatorByType(
         NegotiationType.TLS,
-        noSslContext);
+        noSslContext, null);
+  }
+
+  @Test
+  public void createProtocolNegotiatorByType_tlsWithExecutor() throws Exception {
+    TrackingObjectPoolForTest executorPool = new TrackingObjectPoolForTest();
+    assertEquals(false, executorPool.isInUse());
+    SslContext localSslContext = GrpcSslContexts.forClient().build();
+    ProtocolNegotiator negotiator = NettyChannelBuilder.createProtocolNegotiatorByType(
+        NegotiationType.TLS,
+        localSslContext, executorPool);
+    assertEquals(true, executorPool.isInUse());
+    assertNotNull(negotiator);
+    negotiator.close();
+    assertEquals(false, executorPool.isInUse());
   }
 
   @Test

netty/src/test/java/io/grpc/netty/NettyClientTransportTest.java

@@ -65,6 +65,7 @@
 import io.grpc.internal.TransportTracer;
 import io.grpc.internal.testing.TestUtils;
 import io.grpc.netty.NettyChannelBuilder.LocalSocketPicker;
+import io.grpc.netty.NettyTestUtil.TrackingObjectPoolForTest;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelConfig;
 import io.netty.channel.ChannelDuplexHandler;
@@ -671,6 +672,51 @@ public void keepAliveDisabled_shouldNotSetTcpUserTimeout() throws Exception {
     }
   }
 
+  /**
+   * Verifies that we can successfully build a server and client negotiator with tls and the
+   * executor passing in, and without resource leak after closing the negotiator.
+   */
+  @Test
+  public void tlsNegotiationServerExecutorShouldSucceed() throws Exception {
+    // initialize the client and server Executor pool
+    TrackingObjectPoolForTest serverExecutorPool = new TrackingObjectPoolForTest();
+    TrackingObjectPoolForTest clientExecutorPool = new TrackingObjectPoolForTest();
+    assertEquals(false, serverExecutorPool.isInUse());
+    assertEquals(false, clientExecutorPool.isInUse());
+
+    File serverCert = TestUtils.loadCert("server1.pem");
+    File serverKey = TestUtils.loadCert("server1.key");
+    SslContext sslContext = GrpcSslContexts.forServer(serverCert, serverKey)
+        .ciphers(TestUtils.preferredTestCiphers(), SupportedCipherSuiteFilter.INSTANCE)
+        .clientAuth(ClientAuth.NONE)
+        .build();
+    negotiator = ProtocolNegotiators.serverTls(sslContext, serverExecutorPool);
+    startServer();
+    // after starting the server, the Executor in the server pool should be used
+    assertEquals(true, serverExecutorPool.isInUse());
+
+    File caCert = TestUtils.loadCert("ca.pem");
+    File clientCert = TestUtils.loadCert("client.pem");
+    File clientKey = TestUtils.loadCert("client.key");
+    SslContext clientContext = GrpcSslContexts.forClient()
+        .trustManager(caCert)
+        .ciphers(TestUtils.preferredTestCiphers(), SupportedCipherSuiteFilter.INSTANCE)
+        .keyManager(clientCert, clientKey)
+        .build();
+    ProtocolNegotiator negotiator = ProtocolNegotiators.tls(clientContext, clientExecutorPool);
+    // after starting the client, the Executor in the client pool should be used
+    assertEquals(true, clientExecutorPool.isInUse());
+    final NettyClientTransport transport = newTransport(negotiator);
+    callMeMaybe(transport.start(clientTransportListener));
+    Rpc rpc = new Rpc(transport).halfClose();
+    rpc.waitForResponse();
+    // closing the negotiators should return the executors back to pool, and release the resource
+    negotiator.close();
+    assertEquals(false, clientExecutorPool.isInUse());
+    this.negotiator.close();
+    assertEquals(false, serverExecutorPool.isInUse());
+  }
+
   private Throwable getRootCause(Throwable t) {
     if (t.getCause() == null) {
       return t;