diff options
Diffstat (limited to 'src')
12 files changed, 1000 insertions, 268 deletions
diff --git a/src/main/java/com/google/devtools/build/lib/remote/GrpcActionCache.java b/src/main/java/com/google/devtools/build/lib/remote/GrpcActionCache.java index 7f0a04812f..23899b5549 100644 --- a/src/main/java/com/google/devtools/build/lib/remote/GrpcActionCache.java +++ b/src/main/java/com/google/devtools/build/lib/remote/GrpcActionCache.java @@ -21,6 +21,7 @@ import com.google.bytestream.ByteStreamProto.ReadRequest; import com.google.bytestream.ByteStreamProto.ReadResponse; import com.google.bytestream.ByteStreamProto.WriteRequest; import com.google.bytestream.ByteStreamProto.WriteResponse; +import com.google.common.annotations.VisibleForTesting; import com.google.common.base.Supplier; import com.google.common.base.Suppliers; import com.google.common.collect.ImmutableList; @@ -53,6 +54,7 @@ import com.google.devtools.remoteexecution.v1test.UpdateActionResultRequest; import com.google.protobuf.ByteString; import io.grpc.Channel; import io.grpc.Status; +import io.grpc.StatusException; import io.grpc.StatusRuntimeException; import io.grpc.protobuf.StatusProto; import io.grpc.stub.StreamObserver; @@ -60,7 +62,11 @@ import java.io.IOException; import java.io.OutputStream; import java.util.ArrayList; import java.util.Collection; +import java.util.Collections; +import java.util.HashSet; import java.util.Iterator; +import java.util.List; +import java.util.Set; import java.util.UUID; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; @@ -72,66 +78,54 @@ public class GrpcActionCache implements RemoteActionCache { private final RemoteOptions options; private final ChannelOptions channelOptions; private final Channel channel; + private final Retrier retrier; + // All gRPC stubs are reused. + private final Supplier<ContentAddressableStorageBlockingStub> casBlockingStub; + private final Supplier<ByteStreamBlockingStub> bsBlockingStub; + private final Supplier<ByteStreamStub> bsStub; + private final Supplier<ActionCacheBlockingStub> acBlockingStub; + @VisibleForTesting public GrpcActionCache(Channel channel, ChannelOptions channelOptions, RemoteOptions options) { this.options = options; this.channelOptions = channelOptions; this.channel = channel; + this.retrier = new Retrier(options); + casBlockingStub = + Suppliers.memoize( + () -> + ContentAddressableStorageGrpc.newBlockingStub(channel) + .withCallCredentials(channelOptions.getCallCredentials()) + .withDeadlineAfter(options.remoteTimeout, TimeUnit.SECONDS)); + bsBlockingStub = + Suppliers.memoize( + () -> + ByteStreamGrpc.newBlockingStub(channel) + .withCallCredentials(channelOptions.getCallCredentials()) + .withDeadlineAfter(options.remoteTimeout, TimeUnit.SECONDS)); + bsStub = + Suppliers.memoize( + () -> + ByteStreamGrpc.newStub(channel) + .withCallCredentials(channelOptions.getCallCredentials()) + .withDeadlineAfter(options.remoteTimeout, TimeUnit.SECONDS)); + acBlockingStub = + Suppliers.memoize( + () -> + ActionCacheGrpc.newBlockingStub(channel) + .withCallCredentials(channelOptions.getCallCredentials()) + .withDeadlineAfter(options.remoteTimeout, TimeUnit.SECONDS)); } @Override public void close() {} - // All gRPC stubs are reused. - private final Supplier<ContentAddressableStorageBlockingStub> casBlockingStub = - Suppliers.memoize( - new Supplier<ContentAddressableStorageBlockingStub>() { - @Override - public ContentAddressableStorageBlockingStub get() { - return ContentAddressableStorageGrpc.newBlockingStub(channel) - .withCallCredentials(channelOptions.getCallCredentials()) - .withDeadlineAfter(options.remoteTimeout, TimeUnit.SECONDS); - } - }); - - private final Supplier<ByteStreamBlockingStub> bsBlockingStub = - Suppliers.memoize( - new Supplier<ByteStreamBlockingStub>() { - @Override - public ByteStreamBlockingStub get() { - return ByteStreamGrpc.newBlockingStub(channel) - .withCallCredentials(channelOptions.getCallCredentials()) - .withDeadlineAfter(options.remoteTimeout, TimeUnit.SECONDS); - } - }); - - private final Supplier<ByteStreamStub> bsStub = - Suppliers.memoize( - new Supplier<ByteStreamStub>() { - @Override - public ByteStreamStub get() { - return ByteStreamGrpc.newStub(channel) - .withCallCredentials(channelOptions.getCallCredentials()) - .withDeadlineAfter(options.remoteTimeout, TimeUnit.SECONDS); - } - }); - - private final Supplier<ActionCacheBlockingStub> acBlockingStub = - Suppliers.memoize( - new Supplier<ActionCacheBlockingStub>() { - @Override - public ActionCacheBlockingStub get() { - return ActionCacheGrpc.newBlockingStub(channel) - .withCallCredentials(channelOptions.getCallCredentials()) - .withDeadlineAfter(options.remoteTimeout, TimeUnit.SECONDS); - } - }); - public static boolean isRemoteCacheOptions(RemoteOptions options) { return options.remoteCache != null; } - private ImmutableSet<Digest> getMissingDigests(Iterable<Digest> digests) { + private ImmutableSet<Digest> getMissingDigests(Iterable<Digest> digests) + throws IOException, InterruptedException { FindMissingBlobsRequest.Builder request = FindMissingBlobsRequest.newBuilder() .setInstanceName(options.remoteInstanceName) @@ -139,7 +133,8 @@ public class GrpcActionCache implements RemoteActionCache { if (request.getBlobDigestsCount() == 0) { return ImmutableSet.of(); } - FindMissingBlobsResponse response = casBlockingStub.get().findMissingBlobs(request.build()); + FindMissingBlobsResponse response = + retrier.execute(() -> casBlockingStub.get().findMissingBlobs(request.build())); return ImmutableSet.copyOf(response.getMissingBlobDigestsList()); } @@ -150,7 +145,7 @@ public class GrpcActionCache implements RemoteActionCache { @Override public void ensureInputsPresent( TreeNodeRepository repository, Path execRoot, TreeNode root, Command command) - throws IOException, InterruptedException { + throws IOException, InterruptedException { repository.computeMerkleDigests(root); // TODO(olaola): avoid querying all the digests, only ask for novel subtrees. ImmutableSet<Digest> missingDigests = getMissingDigests(repository.getAllDigests(root)); @@ -165,20 +160,23 @@ public class GrpcActionCache implements RemoteActionCache { BatchUpdateBlobsRequest.Builder treeBlobRequest = BatchUpdateBlobsRequest.newBuilder().setInstanceName(options.remoteInstanceName); for (Directory d : treeNodes) { - final byte[] data = d.toByteArray(); + byte[] data = d.toByteArray(); treeBlobRequest .addRequestsBuilder() .setContentDigest(Digests.computeDigest(data)) .setData(ByteString.copyFrom(data)); } - BatchUpdateBlobsResponse response = - casBlockingStub.get().batchUpdateBlobs(treeBlobRequest.build()); - // TODO(olaola): handle retries on transient errors. - for (BatchUpdateBlobsResponse.Response r : response.getResponsesList()) { - if (!Status.fromCodeValue(r.getStatus().getCode()).isOk()) { - throw StatusProto.toStatusRuntimeException(r.getStatus()); - } - } + retrier.execute( + () -> { + BatchUpdateBlobsResponse response = + casBlockingStub.get().batchUpdateBlobs(treeBlobRequest.build()); + for (BatchUpdateBlobsResponse.Response r : response.getResponsesList()) { + if (!Status.fromCodeValue(r.getStatus().getCode()).isOk()) { + throw StatusProto.toStatusRuntimeException(r.getStatus()); + } + } + return null; + }); } uploadBlob(command.toByteArray()); if (!actionInputs.isEmpty()) { @@ -204,7 +202,7 @@ public class GrpcActionCache implements RemoteActionCache { */ @Override public void download(ActionResult result, Path execRoot, FileOutErr outErr) - throws IOException, CacheNotFoundException { + throws IOException, InterruptedException, CacheNotFoundException { for (OutputFile file : result.getOutputFilesList()) { Path path = execRoot.getRelative(file.getPath()); FileSystemUtils.createDirectoryAndParents(path.getParentDirectory()); @@ -213,15 +211,21 @@ public class GrpcActionCache implements RemoteActionCache { // Handle empty file locally. FileSystemUtils.writeContent(path, new byte[0]); } else { - try (OutputStream stream = path.getOutputStream()) { - if (!file.getContent().isEmpty()) { + if (!file.getContent().isEmpty()) { + try (OutputStream stream = path.getOutputStream()) { file.getContent().writeTo(stream); - } else { - Iterator<ReadResponse> replies = readBlob(digest); - while (replies.hasNext()) { - replies.next().getData().writeTo(stream); - } } + } else { + retrier.execute( + () -> { + try (OutputStream stream = path.getOutputStream()) { + Iterator<ReadResponse> replies = readBlob(digest); + while (replies.hasNext()) { + replies.next().getData().writeTo(stream); + } + return null; + } + }); } } path.setExecutable(file.getIsExecutable()); @@ -234,7 +238,7 @@ public class GrpcActionCache implements RemoteActionCache { } private void downloadOutErr(ActionResult result, FileOutErr outErr) - throws IOException, CacheNotFoundException { + throws IOException, InterruptedException, CacheNotFoundException { if (!result.getStdoutRaw().isEmpty()) { result.getStdoutRaw().writeTo(outErr.getOutputStream()); outErr.getOutputStream().flush(); @@ -272,20 +276,21 @@ public class GrpcActionCache implements RemoteActionCache { } @Override - public void upload( - ActionKey actionKey, Path execRoot, Collection<Path> files, FileOutErr outErr) - throws IOException, InterruptedException { + public void upload(ActionKey actionKey, Path execRoot, Collection<Path> files, FileOutErr outErr) + throws IOException, InterruptedException { ActionResult.Builder result = ActionResult.newBuilder(); upload(execRoot, files, outErr, result); try { - acBlockingStub - .get() - .updateActionResult( - UpdateActionResultRequest.newBuilder() - .setInstanceName(options.remoteInstanceName) - .setActionDigest(actionKey.getDigest()) - .setActionResult(result) - .build()); + retrier.execute( + () -> + acBlockingStub + .get() + .updateActionResult( + UpdateActionResultRequest.newBuilder() + .setInstanceName(options.remoteInstanceName) + .setActionDigest(actionKey.getDigest()) + .setActionResult(result) + .build())); } catch (StatusRuntimeException e) { if (e.getStatus().getCode() != Status.Code.UNIMPLEMENTED) { throw e; @@ -294,7 +299,7 @@ public class GrpcActionCache implements RemoteActionCache { } void upload(Path execRoot, Collection<Path> files, FileOutErr outErr, ActionResult.Builder result) - throws IOException, InterruptedException { + throws IOException, InterruptedException { ArrayList<Digest> digests = new ArrayList<>(); Chunker.Builder b = new Chunker.Builder(); for (Path file : files) { @@ -324,7 +329,7 @@ public class GrpcActionCache implements RemoteActionCache { .setDigest(digests.get(index++)) .setIsExecutable(file.isExecutable()); } - // TODO(ulfjack): Use the Chunker also for stdout / stderr. + // TODO(olaola): inline small stdout/stderr here. if (outErr.getErrorPath().exists()) { Digest stderr = uploadFileContents(outErr.getErrorPath()); result.setStderrDigest(stderr); @@ -356,8 +361,7 @@ public class GrpcActionCache implements RemoteActionCache { * * @return The key for fetching the file contents blob from cache. */ - Digest uploadFileContents( - ActionInput input, Path execRoot, ActionInputFileCache inputCache) + Digest uploadFileContents(ActionInput input, Path execRoot, ActionInputFileCache inputCache) throws IOException, InterruptedException { Digest digest = Digests.getDigestFromInputCache(input, inputCache); ImmutableSet<Digest> missing = getMissingDigests(ImmutableList.of(digest)); @@ -369,80 +373,98 @@ public class GrpcActionCache implements RemoteActionCache { private void uploadChunks(int numItems, Chunker.Builder chunkerBuilder) throws InterruptedException, IOException { - final CountDownLatch finishLatch = new CountDownLatch(numItems); - final AtomicReference<RuntimeException> exception = new AtomicReference<>(null); - StreamObserver<WriteRequest> requestObserver = null; String resourceName = ""; if (!options.remoteInstanceName.isEmpty()) { resourceName += options.remoteInstanceName + "/"; } + Retrier.Backoff backoff = retrier.newBackoff(); Chunker chunker = chunkerBuilder.build(); - while (chunker.hasNext()) { - Chunker.Chunk chunk = chunker.next(); - final Digest digest = chunk.getDigest(); - long offset = chunk.getOffset(); - WriteRequest.Builder request = WriteRequest.newBuilder(); - if (offset == 0) { // Beginning of new upload. - numItems--; - request.setResourceName( - resourceName - + "uploads/" - + UUID.randomUUID() - + "/blobs/" - + digest.getHash() - + "/" - + digest.getSizeBytes()); - // The batches execute simultaneously. - requestObserver = - bsStub - .get() - .write( - new StreamObserver<WriteResponse>() { - private long bytesLeft = digest.getSizeBytes(); + while (true) { // Retry until either uploaded everything or raised an exception. + CountDownLatch finishLatch = new CountDownLatch(numItems); + AtomicReference<IOException> crashException = new AtomicReference<>(null); + List<Status> errors = Collections.synchronizedList(new ArrayList<Status>()); + Set<Digest> failedDigests = Collections.synchronizedSet(new HashSet<Digest>()); + StreamObserver<WriteRequest> requestObserver = null; + while (chunker.hasNext()) { + Chunker.Chunk chunk = chunker.next(); + Digest digest = chunk.getDigest(); + long offset = chunk.getOffset(); + WriteRequest.Builder request = WriteRequest.newBuilder(); + if (offset == 0) { // Beginning of new upload. + numItems--; + request.setResourceName( + String.format( + "%s/uploads/%s/blobs/%s/%d", + resourceName, UUID.randomUUID(), digest.getHash(), digest.getSizeBytes())); + // The batches execute simultaneously. + requestObserver = + bsStub + .get() + .write( + new StreamObserver<WriteResponse>() { + private long bytesLeft = digest.getSizeBytes(); + + @Override + public void onNext(WriteResponse reply) { + bytesLeft -= reply.getCommittedSize(); + } - @Override - public void onNext(WriteResponse reply) { - bytesLeft -= reply.getCommittedSize(); - } + @Override + public void onError(Throwable t) { + // In theory, this can be any error, even though it's supposed to usually + // be only StatusException or StatusRuntimeException. We have to check + // for other errors, in order to not accidentally retry them! + if (!(t instanceof StatusRuntimeException + || t instanceof StatusException)) { + crashException.compareAndSet(null, new IOException(t)); + } - @Override - public void onError(Throwable t) { - exception.compareAndSet( - null, new StatusRuntimeException(Status.fromThrowable(t))); - finishLatch.countDown(); - } + failedDigests.add(digest); + errors.add(Status.fromThrowable(t)); + finishLatch.countDown(); + } - @Override - public void onCompleted() { - if (bytesLeft != 0) { - exception.compareAndSet( - null, new RuntimeException("Server did not commit all data.")); + @Override + public void onCompleted() { + // This can actually happen even if we did not send all the bytes, + // if the server has and is able to reuse parts of the uploaded blob. + finishLatch.countDown(); } - finishLatch.countDown(); - } - }); + }); + } + byte[] data = chunk.getData(); + boolean finishWrite = offset + data.length == digest.getSizeBytes(); + request + .setData(ByteString.copyFrom(data)) + .setWriteOffset(offset) + .setFinishWrite(finishWrite); + requestObserver.onNext(request.build()); + if (finishWrite) { + requestObserver.onCompleted(); + } + if (finishLatch.getCount() <= numItems) { + // Current RPC errored before we finished sending. + if (!finishWrite) { + chunker.advanceInput(); + } + } } - byte[] data = chunk.getData(); - boolean finishWrite = offset + data.length == digest.getSizeBytes(); - request.setData(ByteString.copyFrom(data)).setWriteOffset(offset).setFinishWrite(finishWrite); - requestObserver.onNext(request.build()); - if (finishWrite) { - requestObserver.onCompleted(); + finishLatch.await(options.remoteTimeout, TimeUnit.SECONDS); + if (crashException.get() != null) { + throw crashException.get(); // Re-throw the exception that is supposed to never happen. } - if (finishLatch.getCount() <= numItems) { - // Current RPC errored before we finished sending. - if (!finishWrite) { - chunker.advanceInput(); - } + if (failedDigests.isEmpty()) { + return; // Successfully sent everything. } - } - finishLatch.await(options.remoteTimeout, TimeUnit.SECONDS); - if (exception.get() != null) { - throw exception.get(); // Re-throw the first encountered exception. + retrier.onFailures(backoff, errors); // This will throw when out of retries. + // We don't have to synchronize on failedDigests now, because after finishLatch.await we're + // back to single threaded execution. + chunker = chunkerBuilder.onlyUseDigests(failedDigests).build(); + numItems = failedDigests.size(); } } - Digest uploadBlob(byte[] blob) throws InterruptedException { + Digest uploadBlob(byte[] blob) throws IOException, InterruptedException { Digest digest = Digests.computeDigest(blob); ImmutableSet<Digest> missing = getMissingDigests(ImmutableList.of(digest)); try { @@ -456,19 +478,24 @@ public class GrpcActionCache implements RemoteActionCache { } } - byte[] downloadBlob(Digest digest) throws CacheNotFoundException { + byte[] downloadBlob(Digest digest) + throws IOException, InterruptedException, CacheNotFoundException { if (digest.getSizeBytes() == 0) { return new byte[0]; } - Iterator<ReadResponse> replies = readBlob(digest); byte[] result = new byte[(int) digest.getSizeBytes()]; - int offset = 0; - while (replies.hasNext()) { - ByteString data = replies.next().getData(); - data.copyTo(result, offset); - offset += data.size(); - } - Preconditions.checkState(digest.getSizeBytes() == offset); + retrier.execute( + () -> { + Iterator<ReadResponse> replies = readBlob(digest); + int offset = 0; + while (replies.hasNext()) { + ByteString data = replies.next().getData(); + data.copyTo(result, offset); + offset += data.size(); + } + Preconditions.checkState(digest.getSizeBytes() == offset); + return null; + }); return result; } @@ -476,17 +503,20 @@ public class GrpcActionCache implements RemoteActionCache { /** Returns a cached result for a given Action digest, or null if not found in cache. */ @Override - public ActionResult getCachedActionResult(ActionKey actionKey) { + public ActionResult getCachedActionResult(ActionKey actionKey) + throws IOException, InterruptedException { try { - return acBlockingStub - .get() - .getActionResult( - GetActionResultRequest.newBuilder() - .setInstanceName(options.remoteInstanceName) - .setActionDigest(actionKey.getDigest()) - .build()); - } catch (StatusRuntimeException e) { - if (e.getStatus().getCode() == Status.Code.NOT_FOUND) { + return retrier.execute( + () -> + acBlockingStub + .get() + .getActionResult( + GetActionResultRequest.newBuilder() + .setInstanceName(options.remoteInstanceName) + .setActionDigest(actionKey.getDigest()) + .build())); + } catch (RetryException e) { + if (e.causedByStatusCode(Status.Code.NOT_FOUND)) { return null; } throw e; diff --git a/src/main/java/com/google/devtools/build/lib/remote/GrpcRemoteExecutor.java b/src/main/java/com/google/devtools/build/lib/remote/GrpcRemoteExecutor.java index 815b9237d0..51bc0b6fa7 100644 --- a/src/main/java/com/google/devtools/build/lib/remote/GrpcRemoteExecutor.java +++ b/src/main/java/com/google/devtools/build/lib/remote/GrpcRemoteExecutor.java @@ -16,6 +16,7 @@ package com.google.devtools.build.lib.remote; import com.google.common.base.Supplier; import com.google.common.base.Suppliers; +import com.google.devtools.build.lib.actions.UserExecException; import com.google.devtools.build.lib.concurrent.ThreadSafety.ThreadSafe; import com.google.devtools.build.lib.util.Preconditions; import com.google.devtools.remoteexecution.v1test.ExecuteRequest; @@ -24,7 +25,6 @@ import com.google.devtools.remoteexecution.v1test.ExecutionGrpc; import com.google.devtools.remoteexecution.v1test.ExecutionGrpc.ExecutionBlockingStub; import com.google.longrunning.Operation; import com.google.protobuf.InvalidProtocolBufferException; -import com.google.protobuf.util.Durations; import com.google.rpc.Status; import com.google.watcher.v1.Change; import com.google.watcher.v1.ChangeBatch; @@ -32,7 +32,10 @@ import com.google.watcher.v1.Request; import com.google.watcher.v1.WatcherGrpc; import com.google.watcher.v1.WatcherGrpc.WatcherBlockingStub; import io.grpc.Channel; +import io.grpc.Status.Code; +import io.grpc.StatusRuntimeException; import io.grpc.protobuf.StatusProto; +import java.io.IOException; import java.util.Iterator; import java.util.concurrent.TimeUnit; import javax.annotation.Nullable; @@ -43,18 +46,11 @@ public class GrpcRemoteExecutor { private final RemoteOptions options; private final ChannelOptions channelOptions; private final Channel channel; + private final Retrier retrier; - // Reuse the gRPC stub. - private final Supplier<ExecutionBlockingStub> execBlockingStub = - Suppliers.memoize( - new Supplier<ExecutionBlockingStub>() { - @Override - public ExecutionBlockingStub get() { - return ExecutionGrpc.newBlockingStub(channel) - .withCallCredentials(channelOptions.getCallCredentials()) - .withDeadlineAfter(options.remoteTimeout, TimeUnit.SECONDS); - } - }); + // Reuse the gRPC stubs. + private final Supplier<ExecutionBlockingStub> execBlockingStub; + private final Supplier<WatcherBlockingStub> watcherBlockingStub; public static boolean isRemoteExecutionOptions(RemoteOptions options) { return options.remoteExecutor != null; @@ -64,68 +60,94 @@ public class GrpcRemoteExecutor { this.options = options; this.channelOptions = channelOptions; this.channel = channel; + this.retrier = new Retrier(options); + execBlockingStub = + Suppliers.memoize( + () -> + ExecutionGrpc.newBlockingStub(channel) + .withCallCredentials(channelOptions.getCallCredentials()) + .withDeadlineAfter(options.remoteTimeout, TimeUnit.SECONDS)); + // Do not set a deadline on this call, because it is hard to estimate in + // advance how much time we should give the server to execute an action + // remotely (scheduling, queing, optional retries, etc.) + // It is the server's responsibility to respect the Action timeout field. + watcherBlockingStub = + Suppliers.memoize( + () -> + WatcherGrpc.newBlockingStub(channel) + .withCallCredentials(channelOptions.getCallCredentials())); } - private @Nullable ExecuteResponse getOperationResponse(Operation op) { + private @Nullable ExecuteResponse getOperationResponse(Operation op) + throws IOException, UserExecException { if (op.getResultCase() == Operation.ResultCase.ERROR) { - throw StatusProto.toStatusRuntimeException(op.getError()); + StatusRuntimeException e = StatusProto.toStatusRuntimeException(op.getError()); + if (e.getStatus().getCode() == Code.DEADLINE_EXCEEDED) { + // This was caused by the command itself exceeding the timeout, + // therefore it is not retriable. + // TODO(olaola): this should propagate a timeout SpawnResult instead of raising. + throw new UserExecException("Remote execution time out", true); + } + throw e; } if (op.getDone()) { Preconditions.checkState(op.getResultCase() != Operation.ResultCase.RESULT_NOT_SET); try { return op.getResponse().unpack(ExecuteResponse.class); } catch (InvalidProtocolBufferException e) { - throw new RuntimeException(e); + throw new IOException(e); } } return null; } - public ExecuteResponse executeRemotely(ExecuteRequest request) { - Operation op = execBlockingStub.get().execute(request); + public ExecuteResponse executeRemotely(ExecuteRequest request) + throws InterruptedException, IOException, UserExecException { + Operation op = retrier.execute(() -> execBlockingStub.get().execute(request)); ExecuteResponse resp = getOperationResponse(op); if (resp != null) { return resp; } - int actionSeconds = (int) Durations.toSeconds(request.getAction().getTimeout()); - WatcherBlockingStub stub = - WatcherGrpc.newBlockingStub(channel) - .withCallCredentials(channelOptions.getCallCredentials()) - .withDeadlineAfter(options.remoteTimeout + actionSeconds, TimeUnit.SECONDS); Request wr = Request.newBuilder().setTarget(op.getName()).build(); - Iterator<ChangeBatch> replies = stub.watch(wr); - while (replies.hasNext()) { - ChangeBatch cb = replies.next(); - for (Change ch : cb.getChangesList()) { - switch (ch.getState()) { - case INITIAL_STATE_SKIPPED: - continue; - case ERROR: - try { - throw StatusProto.toStatusRuntimeException(ch.getData().unpack(Status.class)); - } catch (InvalidProtocolBufferException e) { - throw new RuntimeException(e); - } - case DOES_NOT_EXIST: - throw new RuntimeException( - String.format("Operation %s lost on the remote server.", op.getName())); - case EXISTS: - try { - op = ch.getData().unpack(Operation.class); - } catch (InvalidProtocolBufferException e) { - throw new RuntimeException(e); - } - resp = getOperationResponse(op); - if (resp != null) { - return resp; + return retrier.execute( + () -> { + Iterator<ChangeBatch> replies = watcherBlockingStub.get().watch(wr); + while (replies.hasNext()) { + ChangeBatch cb = replies.next(); + for (Change ch : cb.getChangesList()) { + switch (ch.getState()) { + case INITIAL_STATE_SKIPPED: + continue; + case ERROR: + try { + throw StatusProto.toStatusRuntimeException(ch.getData().unpack(Status.class)); + } catch (InvalidProtocolBufferException e) { + throw new RuntimeException(e); + } + case DOES_NOT_EXIST: + // TODO(olaola): either make this retriable, or use a different exception. + throw new IOException( + String.format("Operation %s lost on the remote server.", op.getName())); + case EXISTS: + Operation o; + try { + o = ch.getData().unpack(Operation.class); + } catch (InvalidProtocolBufferException e) { + throw new RuntimeException(e); + } + ExecuteResponse r = getOperationResponse(o); + if (r != null) { + return r; + } + continue; + default: + // This can only happen if the enum gets unexpectedly extended. + throw new IOException(String.format("Illegal change state: %s", ch.getState())); + } } - continue; - default: - throw new RuntimeException(String.format("Illegal change state: %s", ch.getState())); - } - } - } - throw new RuntimeException( - String.format("Watch request for %s terminated with no result.", op.getName())); + } + throw new IOException( + String.format("Watch request for %s terminated with no result.", op.getName())); + }); } } diff --git a/src/main/java/com/google/devtools/build/lib/remote/RemoteActionCache.java b/src/main/java/com/google/devtools/build/lib/remote/RemoteActionCache.java index 994c23e1e6..f8c53268c5 100644 --- a/src/main/java/com/google/devtools/build/lib/remote/RemoteActionCache.java +++ b/src/main/java/com/google/devtools/build/lib/remote/RemoteActionCache.java @@ -58,15 +58,14 @@ interface RemoteActionCache { */ // TODO(olaola): will need to amend to include the TreeNodeRepository for updating. void download(ActionResult result, Path execRoot, FileOutErr outErr) - throws IOException, CacheNotFoundException; - + throws IOException, InterruptedException, CacheNotFoundException; /** * Attempts to look up the given action in the remote cache and return its result, if present. * Returns {@code null} if there is no such entry. Note that a successful result from this method * does not guarantee the availability of the corresponding output files in the remote cache. */ @Nullable - ActionResult getCachedActionResult(ActionKey actionKey); + ActionResult getCachedActionResult(ActionKey actionKey) throws IOException, InterruptedException; /** * Upload the result of a locally executed action to the cache by uploading any necessary files, diff --git a/src/main/java/com/google/devtools/build/lib/remote/RemoteOptions.java b/src/main/java/com/google/devtools/build/lib/remote/RemoteOptions.java index 07aaff11f7..d472adae09 100644 --- a/src/main/java/com/google/devtools/build/lib/remote/RemoteOptions.java +++ b/src/main/java/com/google/devtools/build/lib/remote/RemoteOptions.java @@ -154,4 +154,52 @@ public final class RemoteOptions extends OptionsBase { help = "Value to pass as instance_name in the remote execution API." ) public String remoteInstanceName; + + @Option( + name = "experimental_remote_retry", + defaultValue = "true", + category = "remote", + help = "Whether to retry transient remote execution/cache errors." + ) + public boolean experimentalRemoteRetry; + + @Option( + name = "experimental_remote_retry_start_delay_millis", + defaultValue = "100", + category = "remote", + help = "The initial delay before retrying a transient error." + ) + public long experimentalRemoteRetryStartDelayMillis; + + @Option( + name = "experimental_remote_retry_max_delay_millis", + defaultValue = "5000", + category = "remote", + help = "The maximum delay before retrying a transient error." + ) + public long experimentalRemoteRetryMaxDelayMillis; + + @Option( + name = "experimental_remote_retry_max_attempts", + defaultValue = "5", + category = "remote", + help = "The maximum number of attempts to retry a transient error." + ) + public int experimentalRemoteRetryMaxAttempts; + + @Option( + name = "experimental_remote_retry_multiplier", + defaultValue = "2", + category = "remote", + help = "The multiplier by which to increase the retry delay on transient errors." + ) + public double experimentalRemoteRetryMultiplier; + + @Option( + name = "experimental_remote_retry_jitter", + defaultValue = "0.1", + category = "remote", + help = "The random factor to apply to retry delays on transient errors." + ) + public double experimentalRemoteRetryJitter; } diff --git a/src/main/java/com/google/devtools/build/lib/remote/RemoteSpawnStrategy.java b/src/main/java/com/google/devtools/build/lib/remote/RemoteSpawnStrategy.java index c0952c0282..8426ae3864 100644 --- a/src/main/java/com/google/devtools/build/lib/remote/RemoteSpawnStrategy.java +++ b/src/main/java/com/google/devtools/build/lib/remote/RemoteSpawnStrategy.java @@ -297,13 +297,7 @@ final class RemoteSpawnStrategy implements SpawnActionContext { verboseFailures, spawn.getArguments(), spawn.getEnvironment(), cwd); throw new UserExecException(message + ": Exit " + result.getExitCode()); } - } catch (IOException e) { - throw new UserExecException("Unexpected IO error.", e); - } catch (InterruptedException e) { - eventHandler.handle(Event.warn(mnemonic + " remote work interrupted (" + e + ")")); - Thread.currentThread().interrupt(); - throw e; - } catch (StatusRuntimeException e) { + } catch (RetryException e) { String stackTrace = ""; if (verboseFailures) { stackTrace = "\n" + Throwables.getStackTraceAsString(e); @@ -312,7 +306,7 @@ final class RemoteSpawnStrategy implements SpawnActionContext { if (remoteOptions.remoteLocalFallback) { execLocally(spawn, actionExecutionContext, remoteCache, actionKey); } else { - throw new UserExecException(e); + throw new UserExecException(e.getCause()); } } catch (CacheNotFoundException e) { // TODO(olaola): handle this exception by reuploading / reexecuting the action remotely. @@ -322,6 +316,8 @@ final class RemoteSpawnStrategy implements SpawnActionContext { } else { throw new UserExecException(e); } + } catch (IOException e) { + throw new UserExecException("Unexpected IO error.", e); } catch (UnsupportedOperationException e) { eventHandler.handle( Event.warn(mnemonic + " unsupported operation for action cache (" + e + ")")); diff --git a/src/main/java/com/google/devtools/build/lib/remote/Retrier.java b/src/main/java/com/google/devtools/build/lib/remote/Retrier.java new file mode 100644 index 0000000000..981af46199 --- /dev/null +++ b/src/main/java/com/google/devtools/build/lib/remote/Retrier.java @@ -0,0 +1,338 @@ +// 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.remote; + +import com.google.common.annotations.VisibleForTesting; +import com.google.common.base.Predicate; +import com.google.common.base.Predicates; +import com.google.common.base.Supplier; +import com.google.common.base.Throwables; +import com.google.common.collect.ImmutableList; +import com.google.devtools.build.lib.util.Preconditions; +import io.grpc.Status; +import io.grpc.StatusException; +import io.grpc.StatusRuntimeException; +import java.time.Duration; +import java.util.List; +import java.util.concurrent.Callable; +import java.util.concurrent.ThreadLocalRandom; +import java.util.concurrent.TimeUnit; + +/** + * Supports execution with retries on particular gRPC Statuses. The retrier is ThreadSafe. + * + * <p>Example usage: The simple use-case is to call retrier.execute, e.g: + * + * <pre> + * foo = retrier.execute( + * new Callable<Foo>() { + * @Override + * public Foo call() { + * return grpcStub.getFoo(fooRequest); + * } + * }); + * </pre> + * + * <p>When you need to retry multiple asynchronous calls, you can do: + * + * <pre> + * Retrier.Backoff backoff = retrier.newBackoff(); + * List<Status> errors = Collections.synchronizedList(new ArrayList<Status>()); + * while (true) { + * CountDownLatch finishLatch = new CountDownLatch(items.size()); + * for (Item item : items) { + * requestObserver = myStub.asyncCall( + * request, + * new StreamObserver<Response>() { + * ... + * + * @Override + * public void onError(Throwable t) { + * // Need to handle non Status errors here! + * errors.add(Status.fromThrowable(t)); + * finishLatch.countDown(); + * } + * @Override + * public void onCompleted() { + * finishLatch.countDown(); + * } + * }); + * requestObserver.onNext(i1); + * requestObserver.onNext(i2); + * ... + * requestObserver.onCompleted(); + * } + * finishLatch.await(someTime, TimeUnit.SECONDS); + * if (errors.isEmpty()) { + * return; + * } + * retrier.onFailures(backoff, errors); // Sleep once for the whole batch of failures. + * items = failingItems; // this needs to be collected from the observers as well. + * } + * </pre> + * + * <p>This retries the multiple calls in bulk. Another way to do it is retry each call separately as + * it occurs: + * + * <pre> + * class RetryingObserver extends StreamObserver<Response> { + * private final CountDownLatch finishLatch; + * private final Backoff backoff; + * private final AtomicReference<RuntimeException> exception; + * + * RetryingObserver( + * CountDownLatch finishLatch, Backoff backoff, AtomicReference<RuntimeException> exception) { + * this.finishLatch = finishLatch; + * this.backoff = backoff; + * this.exception = exception; + * } + * + * @Override + * public void onError(Throwable t) { + * // Need to handle non Status errors here first! + * try { + * retrier.onFailure(backoff, Status.fromThrowable(t)); + * + * // This assumes you passed through the relevant info to recreate the original request: + * requestObserver = myStub.asyncCall( + * request, + * new RetryingObserver(finishLatch, backoff)); // Recursion! + * requestObserver.onNext(i1); + * requestObserver.onNext(i2); + * ... + * requestObserver.onCompleted(); + * + * } catch (RetryException e) { + * exception.compareAndSet(null, e); + * finishLatch.countDown(); + * } + * } + * @Override + * public void onCompleted() { + * finishLatch.countDown(); + * } + * } + * + * Retrier.Backoff backoff = retrier.newBackoff(); + * List<Status> errors = Collections.synchronizedList(new ArrayList<Status>()); + * while (true) { + * CountDownLatch finishLatch = new CountDownLatch(items.size()); + * for (Item item : items) { + * requestObserver = myStub.asyncCall( + * request, + * new RetryingObserver(finishLatch, backoff)); + * requestObserver.onNext(i1); + * requestObserver.onNext(i2); + * ... + * requestObserver.onCompleted(); + * } + * finishLatch.await(someTime, TimeUnit.SECONDS); + * if (exception.get() != null) { + * throw exception.get(); // Re-throw the first encountered exception. + * } + * } + * </pre> + * + * In both cases you need to instantiate and keep a Backoff object, and use onFailure(s) to retry. + */ +public class Retrier { + /** + * Backoff is a stateful object providing a sequence of durations that are used to time delays + * between retries. It is not ThreadSafe. The reason that Backoff needs to be stateful, rather + * than a static map of attempt number to delay, is to enable using the retrier via the manual + * onFailure(backoff, e) method (see multiple async gRPC calls example above). + */ + public interface Backoff { + + /** Indicates that no more retries should be made for use in {@link #nextDelayMillis()}. */ + static final long STOP = -1L; + + /** Returns the next delay in milliseconds, or < 0 if we should not continue retrying. */ + long nextDelayMillis(); + + /** + * Returns the number of calls to {@link #nextDelayMillis()} thus far, not counting any calls + * that returned STOP. + */ + int getRetryAttempts(); + + /** + * Creates a Backoff supplier for a Backoff which does not support any retries. Both the + * Supplier and the Backoff are stateless and thread-safe. + */ + static final Supplier<Backoff> NO_RETRIES = + () -> + new Backoff() { + @Override + public long nextDelayMillis() { + return STOP; + } + + @Override + public int getRetryAttempts() { + return 0; + } + }; + + /** + * Creates a Backoff supplier for an optionally jittered exponential backoff. The supplier is + * ThreadSafe (non-synchronized calls to get() are fine), but the returned Backoff is not. + * + * @param initial The initial backoff duration. + * @param max The maximum backoff duration. + * @param multiplier The amount the backoff should increase in each iteration. Must be >1. + * @param jitter The amount the backoff should be randomly varied (0-1), with 0 providing no + * jitter, and 1 providing a duration that is 0-200% of the non-jittered duration. + * @param maxAttempts Maximal times to attempt a retry 0 means no retries. + */ + static Supplier<Backoff> exponential( + Duration initial, Duration max, double multiplier, double jitter, int maxAttempts) { + Preconditions.checkArgument(multiplier > 1, "multipler must be > 1"); + Preconditions.checkArgument(jitter >= 0 && jitter <= 1, "jitter must be in the range (0, 1)"); + Preconditions.checkArgument(maxAttempts >= 0, "maxAttempts must be >= 0"); + return () -> + new Backoff() { + private final long maxMillis = max.toMillis(); + private long nextDelayMillis = initial.toMillis(); + private int attempts = 0; + + @Override + public long nextDelayMillis() { + if (attempts == maxAttempts) { + return STOP; + } + attempts++; + double jitterRatio = jitter * (ThreadLocalRandom.current().nextDouble(2.0) - 1); + long result = (long) (nextDelayMillis * (1 + jitterRatio)); + // Advance current by the non-jittered result. + nextDelayMillis = (long) (nextDelayMillis * multiplier); + if (nextDelayMillis > maxMillis) { + nextDelayMillis = maxMillis; + } + return result; + } + + @Override + public int getRetryAttempts() { + return attempts; + } + }; + } + } + + public static final Predicate<Status> DEFAULT_IS_RETRIABLE = + st -> { + switch (st.getCode()) { + case CANCELLED: + return !Thread.currentThread().isInterrupted(); + case UNKNOWN: + case DEADLINE_EXCEEDED: + case ABORTED: + case INTERNAL: + case UNAVAILABLE: + case UNAUTHENTICATED: + return true; + default: + return false; + } + }; + + public static final Predicate<Status> RETRY_ALL = Predicates.alwaysTrue(); + public static final Predicate<Status> RETRY_NONE = Predicates.alwaysFalse(); + public static final Retrier NO_RETRIES = new Retrier(Backoff.NO_RETRIES, RETRY_NONE); + + private final Supplier<Backoff> backoffSupplier; + private final Predicate<Status> isRetriable; + + @VisibleForTesting + Retrier(Supplier<Backoff> backoffSupplier, Predicate<Status> isRetriable) { + this.backoffSupplier = backoffSupplier; + this.isRetriable = isRetriable; + } + + public Retrier(RemoteOptions options) { + this( + options.experimentalRemoteRetry + ? Backoff.exponential( + Duration.ofMillis(options.experimentalRemoteRetryStartDelayMillis), + Duration.ofMillis(options.experimentalRemoteRetryMaxDelayMillis), + options.experimentalRemoteRetryMultiplier, + options.experimentalRemoteRetryJitter, + options.experimentalRemoteRetryMaxAttempts) + : Backoff.NO_RETRIES, + DEFAULT_IS_RETRIABLE); + } + + /** + * Executes the given callable in a loop, retrying on retryable errors, as defined by the current + * backoff/retry policy. Will raise the last encountered retriable error, or the first + * non-retriable error. + * + * @param c The callable to execute. + */ + public <T> T execute(Callable<T> c) throws InterruptedException, RetryException { + Backoff backoff = backoffSupplier.get(); + while (true) { + try { + return c.call(); + } catch (StatusException | StatusRuntimeException e) { + onFailure(backoff, Status.fromThrowable(e)); + } catch (Exception e) { + // Generic catch because Callable is declared to throw Exception. + Throwables.throwIfUnchecked(e); + throw new RetryException(e, backoff.getRetryAttempts()); + } + } + } + + @VisibleForTesting + void sleep(long timeMillis) throws InterruptedException { + Preconditions.checkArgument( + timeMillis >= 0L, "timeMillis must not be negative: %s", timeMillis); + TimeUnit.MILLISECONDS.sleep(timeMillis); + } + + public Backoff newBackoff() { + return backoffSupplier.get(); + } + + public void onFailure(Backoff backoff, Status s) throws RetryException, InterruptedException { + onFailures(backoff, ImmutableList.of(s)); + } + + /** + * Handles failures according to the current backoff/retry policy. If any of the errors are not + * retriable, the first such error is thrown. Otherwise, if backoff still allows, this sleeps for + * the specified duration. Otherwise, the first error is thrown. + * + * @param backoff The backoff object to get delays from. + * @param errors The errors that occurred (must be non-empty). + */ + public void onFailures(Backoff backoff, List<Status> errors) + throws InterruptedException, RetryException { + Preconditions.checkArgument(!errors.isEmpty(), "errors must be non-empty"); + long delay = backoff.nextDelayMillis(); + for (Status st : errors) { + if (st.getCode() == Status.Code.CANCELLED && Thread.currentThread().isInterrupted()) { + Thread.currentThread().interrupt(); + throw new InterruptedException(); + } + if (delay < 0 || !isRetriable.apply(st)) { + throw new RetryException(st.asRuntimeException(), backoff.getRetryAttempts()); + } + } + sleep(delay); + } +} diff --git a/src/main/java/com/google/devtools/build/lib/remote/RetryException.java b/src/main/java/com/google/devtools/build/lib/remote/RetryException.java new file mode 100644 index 0000000000..6e9890afd6 --- /dev/null +++ b/src/main/java/com/google/devtools/build/lib/remote/RetryException.java @@ -0,0 +1,43 @@ +// 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.remote; + +import io.grpc.Status.Code; +import io.grpc.StatusRuntimeException; +import java.io.IOException; + +/** An exception to indicate failed retry attempts. */ +public final class RetryException extends IOException { + private final int attempts; + + RetryException(Throwable cause, int retryAttempts) { + super(cause); + this.attempts = retryAttempts + 1; + } + + public int getAttempts() { + return attempts; + } + + public boolean causedByStatusCode(Code code) { + return getCause() instanceof StatusRuntimeException + && ((StatusRuntimeException) getCause()).getStatus().getCode() == code; + } + + @Override + public String toString() { + return String.format("after %d attempts: %s", attempts, getCause()); + } +} diff --git a/src/test/java/com/google/devtools/build/lib/remote/FakeImmutableCacheByteStreamImpl.java b/src/test/java/com/google/devtools/build/lib/remote/FakeImmutableCacheByteStreamImpl.java index 4de1498207..e418a29af6 100644 --- a/src/test/java/com/google/devtools/build/lib/remote/FakeImmutableCacheByteStreamImpl.java +++ b/src/test/java/com/google/devtools/build/lib/remote/FakeImmutableCacheByteStreamImpl.java @@ -21,11 +21,16 @@ import com.google.bytestream.ByteStreamProto.ReadResponse; import com.google.common.collect.ImmutableMap; import com.google.devtools.remoteexecution.v1test.Digest; import com.google.protobuf.ByteString; +import io.grpc.Status; import io.grpc.stub.StreamObserver; +import java.util.HashMap; import java.util.Map; class FakeImmutableCacheByteStreamImpl extends ByteStreamImplBase { private final Map<ReadRequest, ReadResponse> cannedReplies; + private final Map<ReadRequest, Integer> numErrors; + // Start returning the correct response after this number of errors is reached. + private static final int MAX_ERRORS = 3; public FakeImmutableCacheByteStreamImpl(Map<Digest, String> contents) { ImmutableMap.Builder<ReadRequest, ReadResponse> b = ImmutableMap.builder(); @@ -37,6 +42,7 @@ class FakeImmutableCacheByteStreamImpl extends ByteStreamImplBase { ReadResponse.newBuilder().setData(ByteString.copyFromUtf8(e.getValue())).build()); } cannedReplies = b.build(); + numErrors = new HashMap<>(); } public FakeImmutableCacheByteStreamImpl(Digest digest, String contents) { @@ -50,7 +56,13 @@ class FakeImmutableCacheByteStreamImpl extends ByteStreamImplBase { @Override public void read(ReadRequest request, StreamObserver<ReadResponse> responseObserver) { assertThat(cannedReplies.containsKey(request)).isTrue(); - responseObserver.onNext(cannedReplies.get(request)); - responseObserver.onCompleted(); + int errCount = numErrors.getOrDefault(request, 0); + if (errCount < MAX_ERRORS) { + numErrors.put(request, errCount + 1); + responseObserver.onError(Status.UNAVAILABLE.asRuntimeException()); // Retriable error. + } else { + responseObserver.onNext(cannedReplies.get(request)); + responseObserver.onCompleted(); + } } } diff --git a/src/test/java/com/google/devtools/build/lib/remote/GrpcActionCacheTest.java b/src/test/java/com/google/devtools/build/lib/remote/GrpcActionCacheTest.java index 52752f478d..cbfecca1d2 100644 --- a/src/test/java/com/google/devtools/build/lib/remote/GrpcActionCacheTest.java +++ b/src/test/java/com/google/devtools/build/lib/remote/GrpcActionCacheTest.java @@ -28,6 +28,7 @@ import com.google.bytestream.ByteStreamProto.WriteResponse; import com.google.common.collect.ImmutableList; import com.google.devtools.build.lib.actions.ActionInputHelper; import com.google.devtools.build.lib.authandtls.AuthAndTLSOptions; +import com.google.devtools.build.lib.remote.Digests.ActionKey; import com.google.devtools.build.lib.testutil.Scratch; import com.google.devtools.build.lib.util.io.FileOutErr; import com.google.devtools.build.lib.vfs.FileSystem; @@ -36,11 +37,14 @@ import com.google.devtools.build.lib.vfs.FileSystemUtils; import com.google.devtools.build.lib.vfs.Path; import com.google.devtools.build.lib.vfs.inmemoryfs.InMemoryFileSystem; import com.google.devtools.common.options.Options; +import com.google.devtools.remoteexecution.v1test.ActionCacheGrpc.ActionCacheImplBase; import com.google.devtools.remoteexecution.v1test.ActionResult; import com.google.devtools.remoteexecution.v1test.ContentAddressableStorageGrpc.ContentAddressableStorageImplBase; import com.google.devtools.remoteexecution.v1test.Digest; import com.google.devtools.remoteexecution.v1test.FindMissingBlobsRequest; import com.google.devtools.remoteexecution.v1test.FindMissingBlobsResponse; +import com.google.devtools.remoteexecution.v1test.GetActionResultRequest; +import com.google.devtools.remoteexecution.v1test.UpdateActionResultRequest; import com.google.protobuf.ByteString; import io.grpc.CallOptions; import io.grpc.Channel; @@ -49,6 +53,7 @@ import io.grpc.ClientInterceptor; import io.grpc.ClientInterceptors; import io.grpc.MethodDescriptor; import io.grpc.Server; +import io.grpc.Status; import io.grpc.inprocess.InProcessChannelBuilder; import io.grpc.inprocess.InProcessServerBuilder; import io.grpc.stub.StreamObserver; @@ -210,17 +215,23 @@ public class GrpcActionCacheTest { } @Test - public void testUploadBlobCacheHit() throws Exception { + public void testUploadBlobCacheHitWithRetries() throws Exception { final GrpcActionCache client = newClient(); final Digest digest = Digests.computeDigestUtf8("abcdefg"); serviceRegistry.addService( new ContentAddressableStorageImplBase() { + private int numErrors = 4; + @Override public void findMissingBlobs( FindMissingBlobsRequest request, StreamObserver<FindMissingBlobsResponse> responseObserver) { - responseObserver.onNext(FindMissingBlobsResponse.getDefaultInstance()); - responseObserver.onCompleted(); + if (numErrors-- <= 0) { + responseObserver.onNext(FindMissingBlobsResponse.getDefaultInstance()); + responseObserver.onCompleted(); + } else { + responseObserver.onError(Status.UNAVAILABLE.asRuntimeException()); + } } }); assertThat(client.uploadBlob("abcdefg".getBytes(UTF_8))).isEqualTo(digest); @@ -337,6 +348,30 @@ public class GrpcActionCacheTest { }; } + private Answer<StreamObserver<WriteRequest>> blobChunkedWriteAnswerError() { + return new Answer<StreamObserver<WriteRequest>>() { + @Override + @SuppressWarnings("unchecked") + public StreamObserver<WriteRequest> answer(final InvocationOnMock invocation) { + return new StreamObserver<WriteRequest>() { + @Override + public void onNext(WriteRequest request) { + ((StreamObserver<WriteResponse>) invocation.getArguments()[0]) + .onError(Status.UNAVAILABLE.asRuntimeException()); + } + + @Override + public void onCompleted() {} + + @Override + public void onError(Throwable t) { + fail("An unexpected client-side error occurred: " + t); + } + }; + } + }; + } + @Test public void testUploadBlobMultipleChunks() throws Exception { final Digest digest = Digests.computeDigestUtf8("abcdef"); @@ -364,7 +399,7 @@ public class GrpcActionCacheTest { } @Test - public void testUploadAllResultsCacheHits() throws Exception { + public void testUploadCacheHits() throws Exception { final GrpcActionCache client = newClient(); final Digest fooDigest = fakeFileCache.createScratchInput(ActionInputHelper.fromPath("a/foo"), "xyz"); @@ -404,51 +439,89 @@ public class GrpcActionCacheTest { } @Test - public void testUploadAllResultsCacheMisses() throws Exception { + public void testUploadCacheMissesWithRetries() throws Exception { final GrpcActionCache client = newClient(); final Digest fooDigest = fakeFileCache.createScratchInput(ActionInputHelper.fromPath("a/foo"), "xyz"); final Digest barDigest = fakeFileCache.createScratchInput(ActionInputHelper.fromPath("bar"), "x"); + final Digest bazDigest = + fakeFileCache.createScratchInput(ActionInputHelper.fromPath("baz"), "z"); final Path fooFile = execRoot.getRelative("a/foo"); final Path barFile = execRoot.getRelative("bar"); + final Path bazFile = execRoot.getRelative("baz"); + ActionKey actionKey = Digests.unsafeActionKeyFromDigest(fooDigest); // Could be any key. barFile.setExecutable(true); serviceRegistry.addService( new ContentAddressableStorageImplBase() { + private int numErrors = 4; + @Override public void findMissingBlobs( FindMissingBlobsRequest request, StreamObserver<FindMissingBlobsResponse> responseObserver) { + if (numErrors-- <= 0) { + responseObserver.onNext(FindMissingBlobsResponse.getDefaultInstance()); + responseObserver.onCompleted(); + } else { + responseObserver.onError(Status.UNAVAILABLE.asRuntimeException()); + } + } + }); + ActionResult.Builder rb = ActionResult.newBuilder(); + rb.addOutputFilesBuilder().setPath("a/foo").setDigest(fooDigest); + rb.addOutputFilesBuilder().setPath("bar").setDigest(barDigest).setIsExecutable(true); + rb.addOutputFilesBuilder().setPath("baz").setDigest(bazDigest); + ActionResult result = rb.build(); + serviceRegistry.addService( + new ActionCacheImplBase() { + private int numErrors = 4; + + @Override + public void updateActionResult( + UpdateActionResultRequest request, StreamObserver<ActionResult> responseObserver) { assertThat(request) .isEqualTo( - FindMissingBlobsRequest.newBuilder() - .addBlobDigests(fooDigest) - .addBlobDigests(barDigest) + UpdateActionResultRequest.newBuilder() + .setActionDigest(fooDigest) + .setActionResult(result) .build()); - // Both are missing. - responseObserver.onNext( - FindMissingBlobsResponse.newBuilder() - .addMissingBlobDigests(fooDigest) - .addMissingBlobDigests(barDigest) - .build()); - responseObserver.onCompleted(); + if (numErrors-- <= 0) { + responseObserver.onNext(result); + responseObserver.onCompleted(); + } else { + responseObserver.onError(Status.UNAVAILABLE.asRuntimeException()); + } } }); ByteStreamImplBase mockByteStreamImpl = Mockito.mock(ByteStreamImplBase.class); serviceRegistry.addService(mockByteStreamImpl); when(mockByteStreamImpl.write(Mockito.<StreamObserver<WriteResponse>>anyObject())) - .thenAnswer(blobChunkedWriteAnswer("xyz", 3)) - .thenAnswer(blobChunkedWriteAnswer("x", 1)); + .thenAnswer(blobChunkedWriteAnswerError()) // Error out for foo. + .thenAnswer(blobChunkedWriteAnswer("x", 1)) // Upload bar successfully. + .thenAnswer(blobChunkedWriteAnswerError()) // Error out for baz. + .thenAnswer(blobChunkedWriteAnswer("xyz", 3)) // Retry foo successfully. + .thenAnswer(blobChunkedWriteAnswerError()) // Error out for baz again. + .thenAnswer(blobChunkedWriteAnswer("z", 1)); // Retry baz successfully. + + client.upload(actionKey, execRoot, ImmutableList.<Path>of(fooFile, barFile, bazFile), outErr); + } - ActionResult.Builder result = ActionResult.newBuilder(); - client.upload(execRoot, ImmutableList.<Path>of(fooFile, barFile), outErr, result); - ActionResult.Builder expectedResult = ActionResult.newBuilder(); - expectedResult.addOutputFilesBuilder().setPath("a/foo").setDigest(fooDigest); - expectedResult - .addOutputFilesBuilder() - .setPath("bar") - .setDigest(barDigest) - .setIsExecutable(true); - assertThat(result.build()).isEqualTo(expectedResult.build()); + @Test + public void testGetCachedActionResultWithRetries() throws Exception { + final GrpcActionCache client = newClient(); + ActionKey actionKey = Digests.unsafeActionKeyFromDigest(Digests.computeDigestUtf8("key")); + serviceRegistry.addService( + new ActionCacheImplBase() { + private int numErrors = 4; + + @Override + public void getActionResult( + GetActionResultRequest request, StreamObserver<ActionResult> responseObserver) { + responseObserver.onError( + (numErrors-- <= 0 ? Status.NOT_FOUND : Status.UNAVAILABLE).asRuntimeException()); + } + }); + assertThat(client.getCachedActionResult(actionKey)).isNull(); } } diff --git a/src/test/java/com/google/devtools/build/lib/remote/GrpcRemoteExecutionClientTest.java b/src/test/java/com/google/devtools/build/lib/remote/GrpcRemoteExecutionClientTest.java index 1183a03488..1a34a663d1 100644 --- a/src/test/java/com/google/devtools/build/lib/remote/GrpcRemoteExecutionClientTest.java +++ b/src/test/java/com/google/devtools/build/lib/remote/GrpcRemoteExecutionClientTest.java @@ -58,17 +58,15 @@ import com.google.devtools.remoteexecution.v1test.GetActionResultRequest; import com.google.longrunning.Operation; import com.google.protobuf.Any; import com.google.protobuf.ByteString; -import com.google.rpc.Code; -import com.google.rpc.Status; import com.google.watcher.v1.Change; import com.google.watcher.v1.ChangeBatch; import com.google.watcher.v1.Request; import com.google.watcher.v1.WatcherGrpc.WatcherImplBase; import io.grpc.Channel; import io.grpc.Server; +import io.grpc.Status; import io.grpc.inprocess.InProcessChannelBuilder; import io.grpc.inprocess.InProcessServerBuilder; -import io.grpc.protobuf.StatusProto; import io.grpc.stub.StreamObserver; import io.grpc.util.MutableHandlerRegistry; import java.io.IOException; @@ -285,6 +283,30 @@ public class GrpcRemoteExecutionClientTest { }; } + private Answer<StreamObserver<WriteRequest>> blobWriteAnswerError() { + return new Answer<StreamObserver<WriteRequest>>() { + @Override + @SuppressWarnings("unchecked") + public StreamObserver<WriteRequest> answer(final InvocationOnMock invocation) { + return new StreamObserver<WriteRequest>() { + @Override + public void onNext(WriteRequest request) { + ((StreamObserver<WriteResponse>) invocation.getArguments()[0]) + .onError(Status.UNAVAILABLE.asRuntimeException()); + } + + @Override + public void onCompleted() {} + + @Override + public void onError(Throwable t) { + fail("An unexpected client-side error occurred: " + t); + } + }; + } + }; + } + @Test public void remotelyExecute() throws Exception { serviceRegistry.addService( @@ -292,9 +314,7 @@ public class GrpcRemoteExecutionClientTest { @Override public void getActionResult( GetActionResultRequest request, StreamObserver<ActionResult> responseObserver) { - responseObserver.onError( - StatusProto.toStatusRuntimeException( - Status.newBuilder().setCode(Code.NOT_FOUND.getNumber()).build())); + responseObserver.onError(Status.NOT_FOUND.asRuntimeException()); } }); final ActionResult actionResult = @@ -359,15 +379,16 @@ public class GrpcRemoteExecutionClientTest { } @Test - public void remotelyExecuteWithWatch() throws Exception { + public void remotelyExecuteWithWatchAndRetries() throws Exception { serviceRegistry.addService( new ActionCacheImplBase() { + private int numErrors = 4; + @Override public void getActionResult( GetActionResultRequest request, StreamObserver<ActionResult> responseObserver) { responseObserver.onError( - StatusProto.toStatusRuntimeException( - Status.newBuilder().setCode(Code.NOT_FOUND.getNumber()).build())); + (numErrors-- <= 0 ? Status.NOT_FOUND : Status.UNAVAILABLE).asRuntimeException()); } }); final ActionResult actionResult = @@ -378,17 +399,29 @@ public class GrpcRemoteExecutionClientTest { final String opName = "operations/xyz"; serviceRegistry.addService( new ExecutionImplBase() { + private int numErrors = 4; + @Override public void execute(ExecuteRequest request, StreamObserver<Operation> responseObserver) { - responseObserver.onNext(Operation.newBuilder().setName(opName).build()); - responseObserver.onCompleted(); + if (numErrors-- <= 0) { + responseObserver.onNext(Operation.newBuilder().setName(opName).build()); + responseObserver.onCompleted(); + } else { + responseObserver.onError(Status.UNAVAILABLE.asRuntimeException()); + } } }); serviceRegistry.addService( new WatcherImplBase() { + private int numErrors = 4; + @Override public void watch(Request request, StreamObserver<ChangeBatch> responseObserver) { assertThat(request.getTarget()).isEqualTo(opName); + if (numErrors-- > 0) { + responseObserver.onError(Status.UNAVAILABLE.asRuntimeException()); + return; + } // Some optional initial state. responseObserver.onNext( ChangeBatch.newBuilder() @@ -443,10 +476,17 @@ public class GrpcRemoteExecutionClientTest { final Digest cmdDigest = Digests.computeDigest(command); serviceRegistry.addService( new ContentAddressableStorageImplBase() { + private int numErrors = 4; + @Override public void findMissingBlobs( FindMissingBlobsRequest request, StreamObserver<FindMissingBlobsResponse> responseObserver) { + if (numErrors-- > 0) { + responseObserver.onError(Status.UNAVAILABLE.asRuntimeException()); + return; + } + FindMissingBlobsResponse.Builder b = FindMissingBlobsResponse.newBuilder(); final Set<Digest> requested = ImmutableSet.copyOf(request.getBlobDigestsList()); if (requested.contains(cmdDigest)) { @@ -463,8 +503,11 @@ public class GrpcRemoteExecutionClientTest { ByteStreamImplBase mockByteStreamImpl = Mockito.mock(ByteStreamImplBase.class); when(mockByteStreamImpl.write(Mockito.<StreamObserver<WriteResponse>>anyObject())) - .thenAnswer(blobWriteAnswer(command.toByteArray())) - .thenAnswer(blobWriteAnswer("xyz".getBytes(UTF_8))); + .thenAnswer(blobWriteAnswerError()) // Error on command upload. + .thenAnswer(blobWriteAnswer(command.toByteArray())) // Upload command successfully. + .thenAnswer(blobWriteAnswerError()) // Error on the input file. + .thenAnswer(blobWriteAnswerError()) // Error on the input file again. + .thenAnswer(blobWriteAnswer("xyz".getBytes(UTF_8))); // Upload input file successfully. serviceRegistry.addService(mockByteStreamImpl); SpawnResult result = client.exec(simpleSpawn, simplePolicy); diff --git a/src/test/java/com/google/devtools/build/lib/remote/RetrierTest.java b/src/test/java/com/google/devtools/build/lib/remote/RetrierTest.java new file mode 100644 index 0000000000..ac54c26da7 --- /dev/null +++ b/src/test/java/com/google/devtools/build/lib/remote/RetrierTest.java @@ -0,0 +1,112 @@ +// Copyright 2015 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.remote; + +import static com.google.common.truth.Truth.assertThat; +import static org.junit.Assert.fail; +import static org.mockito.Mockito.when; + +import com.google.common.collect.Range; +import io.grpc.Status; +import java.time.Duration; +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; +import org.mockito.Mockito; + +/** Tests for {@link Retrier}. */ +@RunWith(JUnit4.class) +public class RetrierTest { + + interface Foo { + public String foo(); + } + + private Foo fooMock; + + @Before + public void setUp() { + fooMock = Mockito.mock(Foo.class); + } + + @Test + public void testExponentialBackoff() throws Exception { + Retrier.Backoff backoff = + Retrier.Backoff.exponential( + Duration.ofSeconds(1), Duration.ofSeconds(10), 2, 0, 6) + .get(); + assertThat(backoff.nextDelayMillis()).isEqualTo(1000); + assertThat(backoff.nextDelayMillis()).isEqualTo(2000); + assertThat(backoff.nextDelayMillis()).isEqualTo(4000); + assertThat(backoff.nextDelayMillis()).isEqualTo(8000); + assertThat(backoff.nextDelayMillis()).isEqualTo(10000); + assertThat(backoff.nextDelayMillis()).isEqualTo(10000); + assertThat(backoff.nextDelayMillis()).isEqualTo(Retrier.Backoff.STOP); + } + + @Test + public void testExponentialBackoffJittered() throws Exception { + Retrier.Backoff backoff = + Retrier.Backoff.exponential( + Duration.ofSeconds(1), Duration.ofSeconds(10), 2, 0.1, 6) + .get(); + assertThat(backoff.nextDelayMillis()).isIn(Range.closedOpen(900L, 1100L)); + assertThat(backoff.nextDelayMillis()).isIn(Range.closedOpen(1800L, 2200L)); + assertThat(backoff.nextDelayMillis()).isIn(Range.closedOpen(3600L, 4400L)); + assertThat(backoff.nextDelayMillis()).isIn(Range.closedOpen(7200L, 8800L)); + assertThat(backoff.nextDelayMillis()).isIn(Range.closedOpen(9000L, 11000L)); + assertThat(backoff.nextDelayMillis()).isIn(Range.closedOpen(9000L, 11000L)); + assertThat(backoff.nextDelayMillis()).isEqualTo(Retrier.Backoff.STOP); + } + + void assertThrows(Retrier retrier, int attempts) throws InterruptedException { + try { + retrier.execute(() -> fooMock.foo()); + fail(); + } catch (RetryException e) { + assertThat(e.getAttempts()).isEqualTo(attempts); + } + } + + @Test + public void testNoRetries() throws Exception { + Retrier retrier = Mockito.spy(Retrier.NO_RETRIES); + Mockito.doNothing().when(retrier).sleep(Mockito.anyLong()); + when(fooMock.foo()) + .thenReturn("bla") + .thenThrow(Status.Code.UNKNOWN.toStatus().asRuntimeException()); + assertThat(retrier.execute(() -> fooMock.foo())).isEqualTo("bla"); + assertThrows(retrier, 1); + Mockito.verify(fooMock, Mockito.times(2)).foo(); + } + + @Test + public void testRepeatedRetriesReset() throws Exception { + Retrier retrier = + Mockito.spy( + new Retrier( + Retrier.Backoff.exponential( + Duration.ofSeconds(1), Duration.ofSeconds(10), 2, 0, 2), + Retrier.RETRY_ALL)); + Mockito.doNothing().when(retrier).sleep(Mockito.anyLong()); + when(fooMock.foo()).thenThrow(Status.Code.UNKNOWN.toStatus().asRuntimeException()); + assertThrows(retrier, 3); + assertThrows(retrier, 3); + Mockito.verify(retrier, Mockito.times(2)).sleep(1000); + Mockito.verify(retrier, Mockito.times(2)).sleep(2000); + Mockito.verify(fooMock, Mockito.times(6)).foo(); + } +} + diff --git a/src/tools/remote_worker/src/main/java/com/google/devtools/build/remote/WatcherServer.java b/src/tools/remote_worker/src/main/java/com/google/devtools/build/remote/WatcherServer.java index 301e6a45ed..3449e6e417 100644 --- a/src/tools/remote_worker/src/main/java/com/google/devtools/build/remote/WatcherServer.java +++ b/src/tools/remote_worker/src/main/java/com/google/devtools/build/remote/WatcherServer.java @@ -294,6 +294,7 @@ final class WatcherServer extends WatcherImplBase { return timeoutSeconds > 0 && wallTimeMillis / 1000.0 > timeoutSeconds; } + @Override public void watch(Request wr, StreamObserver<ChangeBatch> responseObserver) { final String opName = wr.getTarget(); if (!operationsCache.containsKey(opName)) { @@ -303,6 +304,7 @@ final class WatcherServer extends WatcherImplBase { .setCode(Code.NOT_FOUND.getNumber()) .setMessage("Operation not found: " + opName) .build())); + return; } ExecuteRequest request = operationsCache.get(opName); Path tempRoot = workPath.getRelative("build-" + opName); @@ -336,7 +338,21 @@ final class WatcherServer extends WatcherImplBase { logger.log(WARNING, "Cache miss on {0}.", e.getMissingDigest()); responseObserver.onError(StatusUtils.notFoundError(e.getMissingDigest())); } catch (StatusRuntimeException e) { - responseObserver.onError(e); + // In particular, command DEADLINE_EXCEEDED errors should go in the Operation.error field to + // distinguish them from gRPC request DEADLINE_EXCEEDED. + responseObserver.onNext( + ChangeBatch.newBuilder() + .addChanges( + Change.newBuilder() + .setState(Change.State.EXISTS) + .setData( + Any.pack( + Operation.newBuilder() + .setName(opName) + .setError(StatusProto.fromThrowable(e)) + .build())) + .build()) + .build()); } catch (IllegalArgumentException e) { responseObserver.onError( StatusProto.toStatusRuntimeException( |