Create a "transport" Executor for grpc-binder Channels

Author: jdcormie

binder/src/main/java/io/grpc/binder/BinderChannelBuilder.java

@@ -210,6 +210,25 @@ public BinderChannelBuilder scheduledExecutorService(
     return this;
   }
 
+  /**
+   * Provides a custom Executor for potentially expensive, but not blocking, transport work.
+   *
+   * <p>This is an optional parameter. If the user has not provided a scheduled executor service
+   * when the channel is built, the builder will use a static cached thread pool.
+   *
+   * <p>The channel does not take ownership of 'transportExecutor'. Callers must shut it down at
+   * some point *after* every Channel built with in has terminated.
+   *
+   * <p>'transportExecutor' must not be directExecutor() or similar.
+   *
+   * @return this
+   */
+  public BinderChannelBuilder transportExecutor(Executor transportExecutor) {
+    transportFactoryBuilder.setTransportExecutorPool(
+        new FixedObjectPool<>(checkNotNull(transportExecutor, "transportExecutor")));
+    return this;
+  }
+
   /**
    * Provides a custom {@link Executor} for accessing this application's main thread.
    *

binder/src/main/java/io/grpc/binder/internal/BinderClientTransportFactory.java

@@ -48,6 +48,7 @@ public final class BinderClientTransportFactory implements ClientTransportFactor
   final BinderChannelCredentials channelCredentials;
   final Executor mainThreadExecutor;
   final ObjectPool<ScheduledExecutorService> scheduledExecutorPool;
+  final ObjectPool<? extends Executor> transportExecutorPool;
   final ObjectPool<? extends Executor> offloadExecutorPool;
   final SecurityPolicy securityPolicy;
   @Nullable final UserHandle defaultTargetUserHandle;
@@ -58,6 +59,7 @@ public final class BinderClientTransportFactory implements ClientTransportFactor
 
   ScheduledExecutorService executorService;
   Executor offloadExecutor;
+  Executor transportExecutor;
   private boolean closed;
 
   private BinderClientTransportFactory(Builder builder) {
@@ -69,6 +71,8 @@ private BinderClientTransportFactory(Builder builder) {
             : ContextCompat.getMainExecutor(sourceContext);
     scheduledExecutorPool = checkNotNull(builder.scheduledExecutorPool);
     offloadExecutorPool = checkNotNull(builder.offloadExecutorPool);
+    transportExecutorPool = builder.transportExecutorPool != null ?
+        builder.transportExecutorPool : builder.offloadExecutorPool;
     securityPolicy = checkNotNull(builder.securityPolicy);
     defaultTargetUserHandle = builder.defaultTargetUserHandle;
     bindServiceFlags = checkNotNull(builder.bindServiceFlags);
@@ -78,6 +82,7 @@ private BinderClientTransportFactory(Builder builder) {
 
     executorService = scheduledExecutorPool.getObject();
     offloadExecutor = offloadExecutorPool.getObject();
+    transportExecutor = transportExecutorPool.getObject();
   }
 
   @Override
@@ -103,6 +108,7 @@ public SwapChannelCredentialsResult swapChannelCredentials(ChannelCredentials ch
   public void close() {
     closed = true;
     executorService = scheduledExecutorPool.returnObject(executorService);
+    transportExecutor = transportExecutorPool.returnObject(transportExecutor);
     offloadExecutor = offloadExecutorPool.returnObject(offloadExecutor);
   }
 
@@ -120,6 +126,7 @@ public static final class Builder implements ClientTransportFactoryBuilder {
     // Optional.
     BinderChannelCredentials channelCredentials = BinderChannelCredentials.forDefault();
     Executor mainThreadExecutor; // Default filled-in at build time once sourceContext is decided.
+    ObjectPool<? extends Executor> transportExecutorPool; // Default filled-in at build time.
     ObjectPool<ScheduledExecutorService> scheduledExecutorPool =
         SharedResourcePool.forResource(GrpcUtil.TIMER_SERVICE);
     SecurityPolicy securityPolicy = SecurityPolicies.internalOnly();
@@ -139,6 +146,11 @@ public Builder setSourceContext(Context sourceContext) {
       return this;
     }
 
+    public Builder setTransportExecutorPool(ObjectPool<? extends Executor> transportExecutorPool) {
+      this.transportExecutorPool = checkNotNull(transportExecutorPool, "transportExecutorPool");
+      return this;
+    }
+
     public Builder setOffloadExecutorPool(ObjectPool<? extends Executor> offloadExecutorPool) {
       this.offloadExecutorPool = checkNotNull(offloadExecutorPool, "offloadExecutorPool");
       return this;

binder/src/main/java/io/grpc/binder/internal/BinderTransport.java

@@ -559,8 +559,10 @@ final void handleAcknowledgedBytes(long numBytes) {
   public static final class BinderClientTransport extends BinderTransport
       implements ConnectionClientTransport, Bindable.Observer {
 
+    private final ObjectPool<? extends Executor> transportExecutorPool;
     private final ObjectPool<? extends Executor> offloadExecutorPool;
     private final Executor offloadExecutor;
+    private final Executor transportExecutor;
     private final SecurityPolicy securityPolicy;
     private final Bindable serviceBinding;
 
@@ -598,8 +600,10 @@ public BinderClientTransport(
               factory.inboundParcelablePolicy),
           factory.binderDecorator,
           buildLogId(factory.sourceContext, targetAddress));
+      this.transportExecutorPool = factory.transportExecutorPool;
       this.offloadExecutorPool = factory.offloadExecutorPool;
       this.securityPolicy = factory.securityPolicy;
+      this.transportExecutor = transportExecutorPool.getObject();
       this.offloadExecutor = offloadExecutorPool.getObject();
       this.readyTimeoutMillis = factory.readyTimeoutMillis;
       numInUseStreams = new AtomicInteger();
@@ -621,13 +625,14 @@ public BinderClientTransport(
     @Override
     void releaseExecutors() {
       super.releaseExecutors();
+      transportExecutorPool.returnObject(transportExecutor);
       offloadExecutorPool.returnObject(offloadExecutor);
     }
 
     @Override
     public synchronized void onBound(IBinder binder) {
       sendSetupTransaction(
-          binderDecorator.decorate(OneWayBinderProxy.wrap(binder, offloadExecutor)));
+          binderDecorator.decorate(OneWayBinderProxy.wrap(binder, transportExecutor)));
     }
 
     @Override
@@ -788,7 +793,7 @@ public void onFailure(Throwable t) {
                   handleAuthResult(t);
                 }
               },
-              offloadExecutor);
+              transportExecutor);
         }
       }
     }
@@ -797,7 +802,7 @@ private synchronized void handleAuthResult(IBinder binder, Status authorization)
       if (inState(TransportState.SETUP)) {
         if (!authorization.isOk()) {
           shutdownInternal(authorization, true);
-        } else if (!setOutgoingBinder(OneWayBinderProxy.wrap(binder, offloadExecutor))) {
+        } else if (!setOutgoingBinder(OneWayBinderProxy.wrap(binder, transportExecutor))) {
           shutdownInternal(
               Status.UNAVAILABLE.withDescription("Failed to observe outgoing binder"), true);
         } else {