Propagate context in the gRPC server
Using an executorPool to run the thread so we only need to handle context propagation in one place.
@@ -44,6 +44,7 @@ import com.google.devtools.build.lib.vfs.Path;
import com.google.devtools.common.options.InvocationPolicyParser;
import com.google.devtools.common.options.OptionsParsingException;
import com.google.protobuf.ByteString;
+import io.grpc.Context;
import io.grpc.Server;
import io.grpc.StatusRuntimeException;
import io.grpc.netty.NettyServerBuilder;
@@ -59,7 +60,7 @@ import java.security.SecureRandom;
import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
-import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -321,7 +322,7 @@ public class GrpcServerImpl extends CommandServerGrpc.CommandServerImplBase impl
private final CommandManager commandManager;
private final CommandDispatcher dispatcher;
- private final ExecutorService commandExecutorPool;
+ private final Executor commandExecutorPool;
private final Clock clock;
private final Path serverDirectory;
private final String requestCookie;
@@ -366,8 +367,12 @@ public class GrpcServerImpl extends CommandServerGrpc.CommandServerImplBase impl
this.serving = false;
this.commandExecutorPool =
+ Context.currentContextExecutor(
Executors.newCachedThreadPool(
- new ThreadFactoryBuilder().setNameFormat("grpc-command-%d").setDaemon(true).build());
+ new ThreadFactoryBuilder()
+ .setNameFormat("grpc-command-%d")
+ .setDaemon(true)
+ .build()));
this.requestCookie = requestCookie;
this.responseCookie = responseCookie;
@@ -644,7 +649,7 @@ public class GrpcServerImpl extends CommandServerGrpc.CommandServerImplBase impl
((ServerCallStreamObserver<RunResponse>) observer);
BlockingStreamObserver<RunResponse> blockingStreamObserver =
new BlockingStreamObserver<>(serverCallStreamObserver);
- new Thread(() -> executeCommand(request, blockingStreamObserver), "command-thread").start();
+ commandExecutorPool.execute(() -> executeCommand(request, blockingStreamObserver));
}
@Override
