| // Copyright 2016 The Bazel Authors. All rights reserved. |
| // |
| // Licensed under the Apache License, Version 2.0 (the "License"); |
| // you may not use this file except in compliance with the License. |
| // You may obtain a copy of the License at |
| // |
| // http://www.apache.org/licenses/LICENSE-2.0 |
| // |
| // Unless required by applicable law or agreed to in writing, software |
| // distributed under the License is distributed on an "AS IS" BASIS, |
| // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| // See the License for the specific language governing permissions and |
| // limitations under the License. |
| |
| package com.google.devtools.build.lib.server; |
| |
| import com.google.common.annotations.VisibleForTesting; |
| import com.google.common.base.Preconditions; |
| import com.google.common.collect.ImmutableList; |
| import com.google.common.net.InetAddresses; |
| import com.google.common.util.concurrent.ThreadFactoryBuilder; |
| import com.google.common.util.concurrent.Uninterruptibles; |
| import com.google.devtools.build.lib.clock.Clock; |
| import com.google.devtools.build.lib.concurrent.ThreadSafety.Immutable; |
| import com.google.devtools.build.lib.runtime.BlazeCommandDispatcher; |
| import com.google.devtools.build.lib.runtime.BlazeCommandDispatcher.LockingMode; |
| import com.google.devtools.build.lib.runtime.BlazeCommandResult; |
| import com.google.devtools.build.lib.runtime.BlazeRuntime; |
| import com.google.devtools.build.lib.runtime.proto.InvocationPolicyOuterClass.InvocationPolicy; |
| import com.google.devtools.build.lib.server.CommandManager.RunningCommand; |
| import com.google.devtools.build.lib.server.CommandProtos.CancelRequest; |
| import com.google.devtools.build.lib.server.CommandProtos.CancelResponse; |
| import com.google.devtools.build.lib.server.CommandProtos.PingRequest; |
| import com.google.devtools.build.lib.server.CommandProtos.PingResponse; |
| import com.google.devtools.build.lib.server.CommandProtos.RunRequest; |
| import com.google.devtools.build.lib.server.CommandProtos.RunResponse; |
| import com.google.devtools.build.lib.server.CommandProtos.StartupOption; |
| import com.google.devtools.build.lib.util.ExitCode; |
| import com.google.devtools.build.lib.util.Pair; |
| import com.google.devtools.build.lib.util.io.OutErr; |
| import com.google.devtools.build.lib.vfs.FileSystemUtils; |
| 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.Server; |
| import io.grpc.StatusRuntimeException; |
| import io.grpc.netty.NettyServerBuilder; |
| import io.grpc.stub.ServerCallStreamObserver; |
| import io.grpc.stub.StreamObserver; |
| import java.io.IOException; |
| import java.io.OutputStream; |
| import java.io.PrintWriter; |
| import java.io.StringWriter; |
| import java.net.InetSocketAddress; |
| import java.nio.charset.StandardCharsets; |
| import java.security.SecureRandom; |
| import java.util.ArrayList; |
| import java.util.List; |
| import java.util.Optional; |
| import java.util.concurrent.Exchanger; |
| import java.util.concurrent.ExecutionException; |
| import java.util.concurrent.ExecutorService; |
| import java.util.concurrent.Executors; |
| import java.util.concurrent.Future; |
| import java.util.concurrent.LinkedBlockingQueue; |
| import java.util.concurrent.TimeUnit; |
| import java.util.concurrent.atomic.AtomicBoolean; |
| import java.util.concurrent.atomic.AtomicLong; |
| import java.util.concurrent.atomic.AtomicReference; |
| import java.util.logging.Logger; |
| |
| /** |
| * gRPC server class. |
| * |
| * <p>Only this class should depend on gRPC so that we only need to exclude this during |
| * bootstrapping. |
| * |
| * <p>This class is a little complicated and rich in multithreading, so an explanation of its |
| * innards follows. |
| * |
| * <p>We use the direct executor for gRPC so that it calls our methods directly on its event handler |
| * threads (which it creates itself). This is acceptable for {@code ping()} and {@code cancel()} |
| * because they run very quickly. For {@code run()}, we transfer the call to our own threads in |
| * {@code commandExecutorPool}. We do this instead of setting an executor on the server object |
| * because gRPC insists on serializing calls within a single RPC call, which means that the Runnable |
| * passed to {@code setOnReadyHandler} doesn't get called while the main RPC method is running, |
| * which means we can't use flow control, which we need so that gRPC doesn't buffer an unbounded |
| * amount of outgoing data. |
| * |
| * <p>Two threads are spawned for each command: one that handles the command in {@code |
| * commandExecutorPool} and one that streams the result back to the client in {@code |
| * streamExecutorPool}. |
| * |
| * <p>In addition to these threads, we maintain one extra thread for handling the server timeout and |
| * an interrupt watcher thread is started for each interrupt request that logs if it takes too long |
| * to take effect. |
| * |
| * <p>Each running RPC has a UUID associated with it that is used to identify it when a client wants |
| * to cancel it. Cancellation is done by the client sending the server a {@code cancel()} RPC call |
| * which results in the main thread of the command being interrupted. |
| */ |
| public class GrpcServerImpl implements RPCServer { |
| private static final Logger logger = Logger.getLogger(GrpcServerImpl.class.getName()); |
| private final boolean shutdownOnLowSysMem; |
| |
| /** |
| * Factory class. Instantiated by reflection. |
| * |
| * <p>Used so that method calls using reflection are as simple as possible. |
| */ |
| public static class Factory implements RPCServer.Factory { |
| @Override |
| public RPCServer create( |
| BlazeCommandDispatcher dispatcher, |
| Clock clock, |
| int port, |
| Path serverDirectory, |
| int maxIdleSeconds, |
| boolean shutdownOnLowSysMem, |
| boolean idleServerTasks) |
| throws IOException { |
| return new GrpcServerImpl( |
| dispatcher, |
| clock, |
| port, |
| serverDirectory, |
| maxIdleSeconds, |
| shutdownOnLowSysMem, |
| idleServerTasks); |
| } |
| } |
| |
| @VisibleForTesting |
| enum StreamType { |
| STDOUT, |
| STDERR, |
| } |
| |
| /** Actions {@link GrpcSink} can do. */ |
| private enum SinkThreadAction { |
| DISCONNECT, |
| FINISH, |
| READY, |
| SEND, |
| } |
| |
| /** |
| * Sent back and forth between threads wanting to write to the client stream and the stream |
| * handler thread. |
| */ |
| @Immutable |
| private static final class SinkThreadItem { |
| private final boolean success; |
| private final RunResponse message; |
| |
| private SinkThreadItem(boolean success, RunResponse message) { |
| this.success = success; |
| this.message = message; |
| } |
| } |
| |
| /** |
| * A class that handles communicating through a gRPC interface for a streaming rpc call. |
| * |
| * <p>It can do four things: |
| * <li>Send a response message over the wire. If the channel is ready, it's sent immediately, if |
| * it's not, blocks until it is. Note that there can always be only one thread in {@link |
| * #offer(RunResponse)} because it's synchronized. This results in the associated streams |
| * blocking if gRPC is not ready, which is how we implement pushback. |
| * <li>Be notified that gRPC is ready. If there is a pending message, it is then sent. |
| * <li>Be notified that the client disconnected. In this case, an {@link IOException} is reported |
| * and the thread from which the stream was written to is interrupted so that the server |
| * becomes free as soon as possible. |
| * <li>Processing can be terminated. It is reported whether the client disconnected before. |
| */ |
| @VisibleForTesting |
| static class GrpcSink { |
| private final LinkedBlockingQueue<SinkThreadAction> actionQueue; |
| private final Exchanger<SinkThreadItem> exchanger; |
| private final ServerCallStreamObserver<RunResponse> observer; |
| private final Future<?> future; |
| private final AtomicReference<Thread> commandThread = new AtomicReference<>(); |
| private final AtomicBoolean disconnected = new AtomicBoolean(false); |
| private final AtomicLong receivedEventCount = new AtomicLong(0); |
| |
| @VisibleForTesting |
| GrpcSink( |
| final String rpcCommandName, |
| ServerCallStreamObserver<RunResponse> observer, |
| ExecutorService executor) { |
| // This queue is intentionally unbounded: we always act on it fairly quickly so filling up |
| // RAM is not a concern but we don't want to block in the gRPC cancel/onready handlers. |
| this.actionQueue = new LinkedBlockingQueue<>(); |
| this.exchanger = new Exchanger<>(); |
| this.observer = observer; |
| this.observer.setOnCancelHandler( |
| () -> { |
| Thread commandThread = GrpcSink.this.commandThread.get(); |
| if (commandThread != null) { |
| logger.info( |
| String.format( |
| "Interrupting thread %s due to the streaming %s call being cancelled " |
| + "(likely client hang up or explicit gRPC-level cancellation)", |
| commandThread.getName(), rpcCommandName)); |
| commandThread.interrupt(); |
| } |
| |
| actionQueue.offer(SinkThreadAction.DISCONNECT); |
| }); |
| this.observer.setOnReadyHandler(() -> actionQueue.offer(SinkThreadAction.READY)); |
| this.future = executor.submit(GrpcSink.this::call); |
| } |
| |
| @VisibleForTesting |
| long getReceivedEventCount() { |
| return receivedEventCount.get(); |
| } |
| |
| @VisibleForTesting |
| void setCommandThread(Thread thread) { |
| Thread old = commandThread.getAndSet(thread); |
| if (old != null) { |
| throw new IllegalStateException(String.format("Command state set twice (thread %s ->%s)", |
| old.getName(), Thread.currentThread().getName())); |
| } |
| } |
| |
| /** |
| * Sends an item to the client. |
| * |
| * @return true if the item was sent successfully, false if the connection to the client was |
| * lost |
| */ |
| @VisibleForTesting |
| synchronized boolean offer(RunResponse item) { |
| SinkThreadItem queueItem = new SinkThreadItem(false, item); |
| actionQueue.offer(SinkThreadAction.SEND); |
| return exchange(queueItem, false).success; |
| } |
| |
| private boolean disconnected() { |
| return disconnected.get(); |
| } |
| |
| @VisibleForTesting |
| boolean finish() { |
| actionQueue.offer(SinkThreadAction.FINISH); |
| try { |
| Uninterruptibles.getUninterruptibly(future); |
| } catch (ExecutionException e) { |
| throw new IllegalStateException(e); |
| } |
| |
| // Reset the interrupted bit so that it doesn't stay set for the next command that is handled |
| // by this thread |
| Thread.interrupted(); |
| return disconnected(); |
| } |
| |
| private SinkThreadItem exchange(SinkThreadItem item, boolean swallowInterrupts) { |
| boolean interrupted = false; |
| SinkThreadItem result; |
| while (true) { |
| try { |
| result = exchanger.exchange(item); |
| break; |
| } catch (InterruptedException e) { |
| interrupted = true; |
| } |
| } |
| |
| if (interrupted && !swallowInterrupts) { |
| Thread.currentThread().interrupt(); |
| } |
| |
| return result; |
| } |
| |
| private void sendPendingItem() { |
| SinkThreadItem item = exchange(new SinkThreadItem(true, null), true); |
| try { |
| observer.onNext(item.message); |
| } catch (StatusRuntimeException e) { |
| // The RPC was cancelled e.g. by the client terminating unexpectedly. We'll eventually get |
| // notified about this and interrupt the command thread, but in the meantime, we can just |
| // ignore the error; the client is dead, so there isn't anyone to talk to so swallowing the |
| // output is fine. |
| logger.info( |
| String.format( |
| "Client cancelled command for streamer thread %s", |
| Thread.currentThread().getName())); |
| } |
| } |
| |
| /** Main function of the streamer thread. */ |
| private void call() { |
| boolean itemPending = false; |
| |
| while (true) { |
| SinkThreadAction action; |
| action = Uninterruptibles.takeUninterruptibly(actionQueue); |
| receivedEventCount.incrementAndGet(); |
| switch (action) { |
| case FINISH: |
| if (itemPending) { |
| exchange(new SinkThreadItem(false, null), true); |
| itemPending = false; |
| } |
| |
| // Reset the interrupted bit so that it doesn't stay set for the next command that is |
| // handled by this thread |
| Thread.interrupted(); |
| return; |
| |
| case READY: |
| if (itemPending) { |
| sendPendingItem(); |
| itemPending = false; |
| } |
| break; |
| |
| case DISCONNECT: |
| logger.info( |
| "Client disconnected for stream thread " + Thread.currentThread().getName()); |
| disconnected.set(true); |
| if (itemPending) { |
| exchange(new SinkThreadItem(false, null), true); |
| itemPending = false; |
| } |
| break; |
| |
| case SEND: |
| if (disconnected()) { |
| exchange(new SinkThreadItem(false, null), true); |
| } else if (observer.isReady()) { |
| sendPendingItem(); |
| } else { |
| itemPending = true; |
| } |
| } |
| } |
| } |
| } |
| |
| /** |
| * An output stream that forwards the data written to it over the gRPC command stream. |
| * |
| * <p>Note that wraping this class with a {@code Channel} can cause a deadlock if there is an |
| * {@link OutputStream} in between that synchronizes both on {@code #close()} and {@code #write()} |
| * because then if an interrupt happens in {@link GrpcSink#exchange(SinkThreadItem, boolean)}, |
| * the thread on which {@code interrupt()} was called will wait until the {@code Channel} closes |
| * itself while holding a lock for interrupting the thread on which {@code #exchange()} is |
| * being executed and that thread will hold a lock that is needed for the {@code Channel} to be |
| * closed and call {@code interrupt()} in {@code #exchange()}, which will in turn try to acquire |
| * the interrupt lock. |
| */ |
| @VisibleForTesting |
| static class RpcOutputStream extends OutputStream { |
| private static final int CHUNK_SIZE = 8192; |
| |
| // Store commandId and responseCookie as ByteStrings to avoid String -> UTF8 bytes conversion |
| // for each serialized chunk of output. |
| private final ByteString commandIdBytes; |
| private final ByteString responseCookieBytes; |
| |
| private final StreamType type; |
| private final GrpcSink sink; |
| |
| RpcOutputStream(String commandId, String responseCookie, StreamType type, GrpcSink sink) { |
| this.commandIdBytes = ByteString.copyFromUtf8(commandId); |
| this.responseCookieBytes = ByteString.copyFromUtf8(responseCookie); |
| this.type = type; |
| this.sink = sink; |
| } |
| |
| @Override |
| public synchronized void write(byte[] b, int off, int inlen) throws IOException { |
| for (int i = 0; i < inlen; i += CHUNK_SIZE) { |
| ByteString input = ByteString.copyFrom(b, off + i, Math.min(CHUNK_SIZE, inlen - i)); |
| RunResponse.Builder response = RunResponse |
| .newBuilder() |
| .setCookieBytes(responseCookieBytes) |
| .setCommandIdBytes(commandIdBytes); |
| |
| switch (type) { |
| case STDOUT: response.setStandardOutput(input); break; |
| case STDERR: response.setStandardError(input); break; |
| default: throw new IllegalStateException(); |
| } |
| |
| // Send the chunk to the streamer thread. May block. |
| if (!sink.offer(response.build())) { |
| // Client disconnected. Terminate the current command as soon as possible. Note that |
| // throwing IOException is not enough because we are in the habit of swallowing it. Note |
| // that when gRPC notifies us about the disconnection (see the call to setOnCancelHandler) |
| // we interrupt the command thread, which should be enough to make the server come around |
| // as soon as possible. |
| logger.info( |
| String.format( |
| "Client disconnected received for command %s on thread %s", |
| commandIdBytes.toStringUtf8(), Thread.currentThread().getName())); |
| throw new IOException("Client disconnected"); |
| } |
| } |
| } |
| |
| @Override |
| public void write(int byteAsInt) throws IOException { |
| byte b = (byte) byteAsInt; // make sure we work with bytes in comparisons |
| write(new byte[] {b}, 0, 1); |
| } |
| } |
| |
| /** |
| * A thread that watches if the PID file changes and shuts down the server immediately if so. |
| */ |
| private class PidFileWatcherThread extends Thread { |
| private boolean shuttingDown = false; |
| |
| private PidFileWatcherThread() { |
| super("pid-file-watcher"); |
| setDaemon(true); |
| } |
| |
| // The synchronized block is here so that if the "PID file deleted" timer kicks in during a |
| // regular shutdown, they don't race. |
| private synchronized void signalShutdown() { |
| shuttingDown = true; |
| } |
| |
| @Override |
| public void run() { |
| while (true) { |
| Uninterruptibles.sleepUninterruptibly(5, TimeUnit.SECONDS); |
| boolean ok = false; |
| try { |
| String pidFileContents = new String(FileSystemUtils.readContentAsLatin1(pidFile)); |
| ok = pidFileContents.equals(pidInFile); |
| } catch (IOException e) { |
| logger.info("Cannot read PID file: " + e.getMessage()); |
| // Handled by virtue of ok not being set to true |
| } |
| |
| if (!ok) { |
| synchronized (PidFileWatcherThread.this) { |
| if (shuttingDown) { |
| logger.warning("PID file deleted or overwritten but shutdown is already in progress"); |
| break; |
| } |
| |
| shuttingDown = true; |
| // Someone overwrote the PID file. Maybe it's another server, so shut down as quickly |
| // as possible without even running the shutdown hooks (that would delete it) |
| logger.severe("PID file deleted or overwritten, exiting as quickly as possible"); |
| Runtime.getRuntime().halt(ExitCode.BLAZE_INTERNAL_ERROR.getNumericExitCode()); |
| } |
| } |
| } |
| } |
| } |
| |
| // These paths are all relative to the server directory |
| private static final String PORT_FILE = "command_port"; |
| private static final String REQUEST_COOKIE_FILE = "request_cookie"; |
| private static final String RESPONSE_COOKIE_FILE = "response_cookie"; |
| |
| private static final AtomicBoolean runShutdownHooks = new AtomicBoolean(true); |
| |
| private final CommandManager commandManager; |
| private final BlazeCommandDispatcher dispatcher; |
| private final ExecutorService streamExecutorPool; |
| private final ExecutorService commandExecutorPool; |
| private final Clock clock; |
| private final Path serverDirectory; |
| private final String requestCookie; |
| private final String responseCookie; |
| private final int maxIdleSeconds; |
| private final PidFileWatcherThread pidFileWatcherThread; |
| private final Path pidFile; |
| private final String pidInFile; |
| private final List<Path> filesToDeleteAtExit = new ArrayList<>(); |
| private final int port; |
| |
| private Server server; |
| private boolean serving; |
| |
| public GrpcServerImpl( |
| BlazeCommandDispatcher dispatcher, |
| Clock clock, |
| int port, |
| Path serverDirectory, |
| int maxIdleSeconds, |
| boolean shutdownOnLowSysMem, |
| boolean doIdleServerTasks) |
| throws IOException { |
| Runtime.getRuntime().addShutdownHook(new Thread() { |
| @Override |
| public void run() { |
| shutdownHook(); |
| } |
| }); |
| |
| // server.pid was written in the C++ launcher after fork() but before exec() . |
| // The client only accesses the pid file after connecting to the socket |
| // which ensures that it gets the correct pid value. |
| pidFile = serverDirectory.getRelative("server.pid.txt"); |
| pidInFile = new String(FileSystemUtils.readContentAsLatin1(pidFile)); |
| deleteAtExit(pidFile); |
| |
| this.dispatcher = dispatcher; |
| this.clock = clock; |
| this.serverDirectory = serverDirectory; |
| this.port = port; |
| this.maxIdleSeconds = maxIdleSeconds; |
| this.shutdownOnLowSysMem = shutdownOnLowSysMem; |
| this.serving = false; |
| |
| this.streamExecutorPool = |
| Executors.newCachedThreadPool( |
| new ThreadFactoryBuilder().setNameFormat("grpc-stream-%d").setDaemon(true).build()); |
| |
| this.commandExecutorPool = |
| Executors.newCachedThreadPool( |
| new ThreadFactoryBuilder().setNameFormat("grpc-command-%d").setDaemon(true).build()); |
| |
| SecureRandom random = new SecureRandom(); |
| requestCookie = generateCookie(random, 16); |
| responseCookie = generateCookie(random, 16); |
| |
| pidFileWatcherThread = new PidFileWatcherThread(); |
| pidFileWatcherThread.start(); |
| commandManager = new CommandManager(doIdleServerTasks); |
| } |
| |
| private static String generateCookie(SecureRandom random, int byteCount) { |
| byte[] bytes = new byte[byteCount]; |
| random.nextBytes(bytes); |
| StringBuilder result = new StringBuilder(); |
| for (byte b : bytes) { |
| result.append(Integer.toHexString(b + 128)); |
| } |
| |
| return result.toString(); |
| } |
| |
| /** |
| * This is called when the server is shut down as a result of a "clean --expunge". |
| * |
| * <p>In this case, no files should be deleted on shutdown hooks, since clean also deletes the |
| * lock file, and there is a small possibility of the following sequence of events: |
| * |
| * <ol> |
| * <li>Client 1 runs "blaze clean --expunge" |
| * <li>Client 2 runs a command and waits for client 1 to finish |
| * <li>The clean command deletes everything including the lock file |
| * <li>Client 2 starts running and since the output base is empty, starts up a new server, which |
| * creates its own socket and PID files |
| * <li>The server used by client runs its shutdown hooks, deleting the PID files created by the |
| * new server |
| * </ol> |
| * |
| * It also disables the "die when the PID file changes" handler so that it doesn't kill the server |
| * while the "clean --expunge" command is running. |
| */ |
| @Override |
| public void prepareForAbruptShutdown() { |
| disableShutdownHooks(); |
| pidFileWatcherThread.signalShutdown(); |
| } |
| |
| @Override |
| public void interrupt() { |
| commandManager.interruptInflightCommands(); |
| } |
| |
| @Override |
| public void serve() throws IOException { |
| Preconditions.checkState(!serving); |
| |
| // For reasons only Apple knows, you cannot bind to IPv4-localhost when you run in a sandbox |
| // that only allows loopback traffic, but binding to IPv6-localhost works fine. This would |
| // however break on systems that don't support IPv6. So what we'll do is to try to bind to IPv6 |
| // and if that fails, try again with IPv4. |
| InetSocketAddress address = new InetSocketAddress("[::1]", port); |
| try { |
| server = |
| NettyServerBuilder.forAddress(address) |
| .addService(commandServer) |
| .directExecutor() |
| .build() |
| .start(); |
| } catch (IOException e) { |
| address = new InetSocketAddress("127.0.0.1", port); |
| server = |
| NettyServerBuilder.forAddress(address) |
| .addService(commandServer) |
| .directExecutor() |
| .build() |
| .start(); |
| } |
| |
| if (maxIdleSeconds > 0) { |
| Thread timeoutAndMemoryCheckingThread = |
| new Thread( |
| new ServerWatcherRunnable( |
| server, maxIdleSeconds, shutdownOnLowSysMem, commandManager)); |
| timeoutAndMemoryCheckingThread.setName("grpc-timeout-and-memory"); |
| timeoutAndMemoryCheckingThread.setDaemon(true); |
| timeoutAndMemoryCheckingThread.start(); |
| } |
| serving = true; |
| |
| writeServerFile( |
| PORT_FILE, InetAddresses.toUriString(address.getAddress()) + ":" + server.getPort()); |
| writeServerFile(REQUEST_COOKIE_FILE, requestCookie); |
| writeServerFile(RESPONSE_COOKIE_FILE, responseCookie); |
| |
| try { |
| server.awaitTermination(); |
| } catch (InterruptedException e) { |
| // TODO(lberki): Handle SIGINT in a reasonable way |
| throw new IllegalStateException(e); |
| } |
| } |
| |
| private void writeServerFile(String name, String contents) throws IOException { |
| Path file = serverDirectory.getChild(name); |
| FileSystemUtils.writeContentAsLatin1(file, contents); |
| deleteAtExit(file); |
| } |
| |
| protected void disableShutdownHooks() { |
| runShutdownHooks.set(false); |
| } |
| |
| private void shutdownHook() { |
| if (!runShutdownHooks.get()) { |
| return; |
| } |
| |
| List<Path> files; |
| synchronized (filesToDeleteAtExit) { |
| files = new ArrayList<>(filesToDeleteAtExit); |
| } |
| for (Path path : files) { |
| try { |
| path.delete(); |
| } catch (IOException e) { |
| printStack(e); |
| } |
| } |
| } |
| |
| /** |
| * Schedule the specified file for (attempted) deletion at JVM exit. |
| */ |
| protected void deleteAtExit(final Path path) { |
| synchronized (filesToDeleteAtExit) { |
| filesToDeleteAtExit.add(path); |
| } |
| } |
| |
| static void printStack(IOException e) { |
| /* |
| * Hopefully this never happens. It's not very nice to just write this |
| * to the user's console, but I'm not sure what better choice we have. |
| */ |
| StringWriter err = new StringWriter(); |
| PrintWriter printErr = new PrintWriter(err); |
| printErr.println("=======[BAZEL SERVER: ENCOUNTERED IO EXCEPTION]======="); |
| e.printStackTrace(printErr); |
| printErr.println("====================================================="); |
| logger.severe(err.toString()); |
| } |
| |
| private void executeCommand( |
| RunRequest request, StreamObserver<RunResponse> observer, GrpcSink sink) { |
| sink.setCommandThread(Thread.currentThread()); |
| |
| if (!request.getCookie().equals(requestCookie) || request.getClientDescription().isEmpty()) { |
| try { |
| observer.onNext( |
| RunResponse.newBuilder() |
| .setExitCode(ExitCode.LOCAL_ENVIRONMENTAL_ERROR.getNumericExitCode()) |
| .build()); |
| observer.onCompleted(); |
| } catch (StatusRuntimeException e) { |
| logger.info("Client cancelled command while rejecting it: " + e.getMessage()); |
| } |
| return; |
| } |
| |
| // There is a small period of time between calling setOnCancelHandler() and setCommandThread() |
| // during which the command thread is not interrupted when a cancel is signaled. Cover that |
| // case by explicitly checking for disconnection here. |
| if (sink.disconnected()) { |
| return; |
| } |
| |
| String commandId; |
| BlazeCommandResult result; |
| |
| // TODO(b/63925394): This information needs to be passed to the GotOptionsEvent, which does not |
| // currently have the explicit startup options. See Improved Command Line Reporting design doc |
| // for details. |
| // Convert the startup options record to Java strings, source first. |
| ImmutableList.Builder<Pair<String, String>> startupOptions = ImmutableList.builder(); |
| for (StartupOption option : request.getStartupOptionsList()) { |
| // UTF-8 won't do because we want to be able to pass arbitrary binary strings. |
| // Not that the internals of Bazel handle that correctly, but why not make at least this |
| // little part correct? |
| startupOptions.add(new Pair<>( |
| option.getSource().toString(StandardCharsets.ISO_8859_1), |
| option.getOption().toString(StandardCharsets.ISO_8859_1))); |
| } |
| |
| try (RunningCommand command = commandManager.create()) { |
| commandId = command.getId(); |
| |
| try { |
| // Send the client the command id as soon as we know it. |
| observer.onNext( |
| RunResponse.newBuilder() |
| .setCookie(responseCookie) |
| .setCommandId(commandId) |
| .build()); |
| } catch (StatusRuntimeException e) { |
| logger.info( |
| "The client cancelled the command before receiving the command id: " + e.getMessage()); |
| } |
| |
| OutErr rpcOutErr = |
| OutErr.create( |
| new RpcOutputStream(command.getId(), responseCookie, StreamType.STDOUT, sink), |
| new RpcOutputStream(command.getId(), responseCookie, StreamType.STDERR, sink)); |
| |
| try { |
| // UTF-8 won't do because we want to be able to pass arbitrary binary strings. |
| // Not that the internals of Bazel handle that correctly, but why not make at least this |
| // little part correct? |
| ImmutableList<String> args = request.getArgList().stream() |
| .map(arg -> arg.toString(StandardCharsets.ISO_8859_1)) |
| .collect(ImmutableList.toImmutableList()); |
| |
| InvocationPolicy policy = InvocationPolicyParser.parsePolicy(request.getInvocationPolicy()); |
| logger.info(BlazeRuntime.getRequestLogString(args)); |
| result = |
| dispatcher.exec( |
| policy, |
| args, |
| rpcOutErr, |
| request.getBlockForLock() ? LockingMode.WAIT : LockingMode.ERROR_OUT, |
| request.getClientDescription(), |
| clock.currentTimeMillis(), |
| Optional.of(startupOptions.build())); |
| } catch (OptionsParsingException e) { |
| rpcOutErr.printErrLn(e.getMessage()); |
| result = BlazeCommandResult.exitCode(ExitCode.COMMAND_LINE_ERROR); |
| } |
| } catch (InterruptedException e) { |
| result = BlazeCommandResult.exitCode(ExitCode.INTERRUPTED); |
| commandId = ""; // The default value, the client will ignore it |
| } |
| |
| if (sink.finish()) { |
| // Client disconnected. Then we are not allowed to call any methods on the observer. |
| logger.info( |
| String.format( |
| "Client disconnected before we could send exit code for command %s", commandId)); |
| return; |
| } |
| |
| // There is a chance that an Uninterruptibles#getUninterruptibly() leaves us with the |
| // interrupt bit set. So we just reset the interruption state here to make these cancel |
| // requests not have any effect outside of command execution (after the try block above, |
| // the cancel request won't find the thread to interrupt) |
| Thread.interrupted(); |
| |
| if (result.shutdown()) { |
| server.shutdown(); |
| } |
| |
| RunResponse.Builder response = RunResponse.newBuilder() |
| .setCookie(responseCookie) |
| .setCommandId(commandId) |
| .setFinished(true) |
| .setTerminationExpected(result.shutdown()); |
| |
| if (result.getExecRequest() != null) { |
| response.setExitCode(0); |
| response.setExecRequest(result.getExecRequest()); |
| } else { |
| response.setExitCode(result.getExitCode().getNumericExitCode()); |
| } |
| |
| try { |
| observer.onNext(response.build()); |
| observer.onCompleted(); |
| } catch (StatusRuntimeException e) { |
| // The client cancelled the call. Log an error and go on. |
| logger.info( |
| String.format( |
| "Client cancelled command %s just right before its end: %s", |
| commandId, e.getMessage())); |
| } |
| } |
| |
| private final CommandServerGrpc.CommandServerImplBase commandServer = |
| new CommandServerGrpc.CommandServerImplBase() { |
| @Override |
| public void run(final RunRequest request, final StreamObserver<RunResponse> observer) { |
| final GrpcSink sink = |
| new GrpcSink( |
| "Run", (ServerCallStreamObserver<RunResponse>) observer, streamExecutorPool); |
| // Switch to our own threads so that onReadyStateHandler can be called (see class-level |
| // comment) |
| commandExecutorPool.execute(() -> executeCommand(request, observer, sink)); |
| } |
| |
| @Override |
| public void ping(PingRequest pingRequest, StreamObserver<PingResponse> streamObserver) { |
| Preconditions.checkState(serving); |
| |
| try (RunningCommand command = commandManager.create()) { |
| PingResponse.Builder response = PingResponse.newBuilder(); |
| if (pingRequest.getCookie().equals(requestCookie)) { |
| response.setCookie(responseCookie); |
| } |
| |
| streamObserver.onNext(response.build()); |
| streamObserver.onCompleted(); |
| } |
| } |
| |
| @Override |
| public void cancel( |
| final CancelRequest request, final StreamObserver<CancelResponse> streamObserver) { |
| logger.info(String.format("Got CancelRequest for command id %s", request.getCommandId())); |
| if (!request.getCookie().equals(requestCookie)) { |
| streamObserver.onCompleted(); |
| return; |
| } |
| |
| // Actually performing the cancellation can result in some blocking which we don't want |
| // to do on the dispatcher thread, instead offload to command pool. |
| commandExecutorPool.execute(() -> doCancel(request, streamObserver)); |
| } |
| |
| private void doCancel( |
| CancelRequest request, StreamObserver<CancelResponse> streamObserver) { |
| commandManager.doCancel(request); |
| try { |
| streamObserver.onNext(CancelResponse.newBuilder().setCookie(responseCookie).build()); |
| streamObserver.onCompleted(); |
| } catch (StatusRuntimeException e) { |
| // There is no one to report the failure to |
| logger.info( |
| "Client cancelled RPC of cancellation request for " + request.getCommandId()); |
| } |
| } |
| }; |
| } |
