Pull upload(ActionResult) into super class.
Remove the custom upload(ActionResult) implementations from SimpleBlobStoreActionCache and GrpcRemoteCache.
This is a big step towards merging SimpleBlobStoreActionCache and GrpcRemoteCache.
Closes #9167.
PiperOrigin-RevId: 264563384
diff --git a/src/main/java/com/google/devtools/build/lib/remote/AbstractRemoteActionCache.java b/src/main/java/com/google/devtools/build/lib/remote/AbstractRemoteActionCache.java
index d53bb72..b5dbc4c 100644
--- a/src/main/java/com/google/devtools/build/lib/remote/AbstractRemoteActionCache.java
+++ b/src/main/java/com/google/devtools/build/lib/remote/AbstractRemoteActionCache.java
@@ -29,6 +29,7 @@
import build.bazel.remote.execution.v2.Tree;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
+import com.google.common.base.Throwables;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
@@ -79,6 +80,7 @@
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
+import java.util.concurrent.ExecutionException;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import javax.annotation.Nullable;
@@ -120,20 +122,8 @@
abstract ActionResult getCachedActionResult(ActionKey actionKey)
throws IOException, InterruptedException;
- /**
- * Upload the result of a locally executed action to the remote cache.
- *
- * @throws IOException if there was an error uploading to the remote cache
- * @throws ExecException if uploading any of the action outputs is not supported
- */
- abstract void upload(
- SimpleBlobStore.ActionKey actionKey,
- Action action,
- Command command,
- Path execRoot,
- Collection<Path> files,
- FileOutErr outErr)
- throws ExecException, IOException, InterruptedException;
+ protected abstract void setCachedActionResult(ActionKey actionKey, ActionResult action)
+ throws IOException, InterruptedException;
/**
* Uploads a file
@@ -157,6 +147,116 @@
*/
protected abstract ListenableFuture<Void> uploadBlob(Digest digest, ByteString data);
+ protected abstract ImmutableSet<Digest> getMissingDigests(Iterable<Digest> digests)
+ throws IOException, InterruptedException;
+
+ /**
+ * Upload the result of a locally executed action to the remote cache.
+ *
+ * @throws IOException if there was an error uploading to the remote cache
+ * @throws ExecException if uploading any of the action outputs is not supported
+ */
+ public ActionResult upload(
+ ActionKey actionKey,
+ Action action,
+ Command command,
+ Path execRoot,
+ Collection<Path> outputs,
+ FileOutErr outErr,
+ int exitCode)
+ throws ExecException, IOException, InterruptedException {
+ ActionResult.Builder resultBuilder = ActionResult.newBuilder();
+ uploadOutputs(execRoot, actionKey, action, command, outputs, outErr, resultBuilder);
+ resultBuilder.setExitCode(exitCode);
+ ActionResult result = resultBuilder.build();
+ if (exitCode == 0 && !action.getDoNotCache()) {
+ setCachedActionResult(actionKey, result);
+ }
+ return result;
+ }
+
+ public ActionResult upload(
+ ActionKey actionKey,
+ Action action,
+ Command command,
+ Path execRoot,
+ Collection<Path> outputs,
+ FileOutErr outErr)
+ throws ExecException, IOException, InterruptedException {
+ return upload(actionKey, action, command, execRoot, outputs, outErr, /* exitCode= */ 0);
+ }
+
+ private void uploadOutputs(
+ Path execRoot,
+ ActionKey actionKey,
+ Action action,
+ Command command,
+ Collection<Path> files,
+ FileOutErr outErr,
+ ActionResult.Builder result)
+ throws ExecException, IOException, InterruptedException {
+ UploadManifest manifest =
+ new UploadManifest(
+ digestUtil,
+ result,
+ execRoot,
+ options.incompatibleRemoteSymlinks,
+ options.allowSymlinkUpload);
+ manifest.addFiles(files);
+ manifest.setStdoutStderr(outErr);
+ manifest.addAction(actionKey, action, command);
+
+ Map<Digest, Path> digestToFile = manifest.getDigestToFile();
+ Map<Digest, ByteString> digestToBlobs = manifest.getDigestToBlobs();
+ Collection<Digest> digests = new ArrayList<>();
+ digests.addAll(digestToFile.keySet());
+ digests.addAll(digestToBlobs.keySet());
+
+ ImmutableSet<Digest> digestsToUpload = getMissingDigests(digests);
+ ImmutableList.Builder<ListenableFuture<Void>> uploads = ImmutableList.builder();
+ for (Digest digest : digestsToUpload) {
+ Path file = digestToFile.get(digest);
+ if (file != null) {
+ uploads.add(uploadFile(digest, file));
+ } else {
+ ByteString blob = digestToBlobs.get(digest);
+ if (blob == null) {
+ String message = "FindMissingBlobs call returned an unknown digest: " + digest;
+ throw new IOException(message);
+ }
+ uploads.add(uploadBlob(digest, blob));
+ }
+ }
+
+ waitForUploads(uploads.build());
+
+ if (manifest.getStderrDigest() != null) {
+ result.setStderrDigest(manifest.getStderrDigest());
+ }
+ if (manifest.getStdoutDigest() != null) {
+ result.setStdoutDigest(manifest.getStdoutDigest());
+ }
+ }
+
+ private static void waitForUploads(List<ListenableFuture<Void>> uploads)
+ throws IOException, InterruptedException {
+ try {
+ for (ListenableFuture<Void> upload : uploads) {
+ upload.get();
+ }
+ } catch (ExecutionException e) {
+ // TODO(buchgr): Add support for cancellation and factor this method out to be shared
+ // between ByteStreamUploader as well.
+ Throwable cause = e.getCause();
+ Throwables.throwIfInstanceOf(cause, IOException.class);
+ Throwables.throwIfInstanceOf(cause, InterruptedException.class);
+ if (cause != null) {
+ throw new IOException(cause);
+ }
+ throw new IOException(e);
+ }
+ }
+
/**
* Downloads a blob with a content hash {@code digest} to {@code out}.
*
diff --git a/src/main/java/com/google/devtools/build/lib/remote/GrpcRemoteCache.java b/src/main/java/com/google/devtools/build/lib/remote/GrpcRemoteCache.java
index e6c5c63..e9fdb16 100644
--- a/src/main/java/com/google/devtools/build/lib/remote/GrpcRemoteCache.java
+++ b/src/main/java/com/google/devtools/build/lib/remote/GrpcRemoteCache.java
@@ -17,11 +17,9 @@
import static com.google.common.base.Strings.isNullOrEmpty;
import static java.lang.String.format;
-import build.bazel.remote.execution.v2.Action;
import build.bazel.remote.execution.v2.ActionCacheGrpc;
import build.bazel.remote.execution.v2.ActionCacheGrpc.ActionCacheBlockingStub;
import build.bazel.remote.execution.v2.ActionResult;
-import build.bazel.remote.execution.v2.Command;
import build.bazel.remote.execution.v2.ContentAddressableStorageGrpc;
import build.bazel.remote.execution.v2.ContentAddressableStorageGrpc.ContentAddressableStorageFutureStub;
import build.bazel.remote.execution.v2.Digest;
@@ -38,7 +36,6 @@
import com.google.common.base.Ascii;
import com.google.common.base.Preconditions;
import com.google.common.base.Throwables;
-import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Iterables;
import com.google.common.collect.Maps;
@@ -50,7 +47,6 @@
import com.google.common.util.concurrent.MoreExecutors;
import com.google.common.util.concurrent.SettableFuture;
import com.google.devtools.build.lib.actions.ActionInput;
-import com.google.devtools.build.lib.actions.ExecException;
import com.google.devtools.build.lib.concurrent.ThreadSafety.ThreadSafe;
import com.google.devtools.build.lib.remote.RemoteRetrier.ProgressiveBackoff;
import com.google.devtools.build.lib.remote.common.SimpleBlobStore.ActionKey;
@@ -58,7 +54,6 @@
import com.google.devtools.build.lib.remote.options.RemoteOptions;
import com.google.devtools.build.lib.remote.util.DigestUtil;
import com.google.devtools.build.lib.remote.util.TracingMetadataUtils;
-import com.google.devtools.build.lib.util.io.FileOutErr;
import com.google.devtools.build.lib.vfs.Path;
import com.google.protobuf.ByteString;
import com.google.protobuf.Message;
@@ -70,7 +65,6 @@
import java.io.IOException;
import java.io.OutputStream;
import java.util.ArrayList;
-import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutionException;
@@ -166,17 +160,8 @@
|| Ascii.toLowerCase(options.remoteCache).startsWith("https://"));
}
- private ListenableFuture<FindMissingBlobsResponse> getMissingDigests(
- FindMissingBlobsRequest request) throws IOException, InterruptedException {
- Context ctx = Context.current();
- try {
- return retrier.executeAsync(() -> ctx.call(() -> casFutureStub().findMissingBlobs(request)));
- } catch (StatusRuntimeException e) {
- throw new IOException(e);
- }
- }
-
- private ImmutableSet<Digest> getMissingDigests(Iterable<Digest> digests)
+ @Override
+ protected ImmutableSet<Digest> getMissingDigests(Iterable<Digest> digests)
throws IOException, InterruptedException {
if (Iterables.isEmpty(digests)) {
return ImmutableSet.of();
@@ -208,6 +193,16 @@
return result.build();
}
+ private ListenableFuture<FindMissingBlobsResponse> getMissingDigests(
+ FindMissingBlobsRequest request) throws IOException, InterruptedException {
+ Context ctx = Context.current();
+ try {
+ return retrier.executeAsync(() -> ctx.call(() -> casFutureStub().findMissingBlobs(request)));
+ } catch (StatusRuntimeException e) {
+ throw new IOException(e);
+ }
+ }
+
/**
* Ensures that the tree structure of the inputs, the input files themselves, and the command are
* available in the remote cache, such that the tree can be reassembled and executed on another
@@ -377,32 +372,6 @@
}
@Override
- public void upload(
- ActionKey actionKey,
- Action action,
- Command command,
- Path execRoot,
- Collection<Path> files,
- FileOutErr outErr)
- throws ExecException, IOException, InterruptedException {
- ActionResult.Builder result = ActionResult.newBuilder();
- upload(execRoot, actionKey, action, command, files, outErr, result);
- try {
- retrier.execute(
- () ->
- acBlockingStub()
- .updateActionResult(
- UpdateActionResultRequest.newBuilder()
- .setInstanceName(options.remoteInstanceName)
- .setActionDigest(actionKey.getDigest())
- .setActionResult(result)
- .build()));
- } catch (StatusRuntimeException e) {
- throw new IOException(e);
- }
- }
-
- @Override
protected ListenableFuture<Void> uploadFile(Digest digest, Path path) {
return uploader.uploadBlobAsync(
HashCode.fromString(digest.getHash()),
@@ -418,77 +387,6 @@
/* forceUpload= */ true);
}
- void upload(
- Path execRoot,
- ActionKey actionKey,
- Action action,
- Command command,
- Collection<Path> files,
- FileOutErr outErr,
- ActionResult.Builder result)
- throws ExecException, IOException, InterruptedException {
- UploadManifest manifest =
- new UploadManifest(
- digestUtil,
- result,
- execRoot,
- options.incompatibleRemoteSymlinks,
- options.allowSymlinkUpload);
- manifest.addFiles(files);
- manifest.setStdoutStderr(outErr);
- manifest.addAction(actionKey, action, command);
-
- Map<Digest, Path> digestToFile = manifest.getDigestToFile();
- Map<Digest, ByteString> digestToBlobs = manifest.getDigestToBlobs();
- Collection<Digest> digests = new ArrayList<>();
- digests.addAll(digestToFile.keySet());
- digests.addAll(digestToBlobs.keySet());
-
- ImmutableSet<Digest> digestsToUpload = getMissingDigests(digests);
- ImmutableList.Builder<ListenableFuture<Void>> uploads = ImmutableList.builder();
- for (Digest digest : digestsToUpload) {
- Path file = digestToFile.get(digest);
- if (file != null) {
- uploads.add(uploadFile(digest, file));
- } else {
- ByteString blob = digestToBlobs.get(digest);
- if (blob == null) {
- String message = "FindMissingBlobs call returned an unknown digest: " + digest;
- throw new IOException(message);
- }
- uploads.add(uploadBlob(digest, blob));
- }
- }
-
- waitForUploads(uploads.build());
-
- if (manifest.getStderrDigest() != null) {
- result.setStderrDigest(manifest.getStderrDigest());
- }
- if (manifest.getStdoutDigest() != null) {
- result.setStdoutDigest(manifest.getStdoutDigest());
- }
- }
-
- private static void waitForUploads(List<ListenableFuture<Void>> uploads)
- throws IOException, InterruptedException {
- try {
- for (ListenableFuture<Void> upload : uploads) {
- upload.get();
- }
- } catch (ExecutionException e) {
- // TODO(buchgr): Add support for cancellation and factor this method out to be shared
- // between ByteStreamUploader as well.
- Throwable cause = e.getCause();
- Throwables.throwIfInstanceOf(cause, IOException.class);
- Throwables.throwIfInstanceOf(cause, InterruptedException.class);
- if (cause != null) {
- throw new IOException(cause);
- }
- throw new IOException(e);
- }
- }
-
// Execution Cache API
@Override
@@ -511,4 +409,22 @@
throw new IOException(e);
}
}
+
+ @Override
+ protected void setCachedActionResult(ActionKey actionKey, ActionResult result)
+ throws IOException, InterruptedException {
+ try {
+ retrier.execute(
+ () ->
+ acBlockingStub()
+ .updateActionResult(
+ UpdateActionResultRequest.newBuilder()
+ .setInstanceName(options.remoteInstanceName)
+ .setActionDigest(actionKey.getDigest())
+ .setActionResult(result)
+ .build()));
+ } catch (StatusRuntimeException e) {
+ throw new IOException(e);
+ }
+ }
}
diff --git a/src/main/java/com/google/devtools/build/lib/remote/SimpleBlobStoreActionCache.java b/src/main/java/com/google/devtools/build/lib/remote/SimpleBlobStoreActionCache.java
index fba3ba0..8822c0d 100644
--- a/src/main/java/com/google/devtools/build/lib/remote/SimpleBlobStoreActionCache.java
+++ b/src/main/java/com/google/devtools/build/lib/remote/SimpleBlobStoreActionCache.java
@@ -14,35 +14,30 @@
package com.google.devtools.build.lib.remote;
-import build.bazel.remote.execution.v2.Action;
import build.bazel.remote.execution.v2.ActionResult;
-import build.bazel.remote.execution.v2.Command;
import build.bazel.remote.execution.v2.Digest;
import build.bazel.remote.execution.v2.Directory;
import build.bazel.remote.execution.v2.DirectoryNode;
import build.bazel.remote.execution.v2.FileNode;
+import com.google.common.collect.ImmutableSet;
import com.google.common.hash.HashingOutputStream;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.MoreExecutors;
import com.google.common.util.concurrent.SettableFuture;
-import com.google.devtools.build.lib.actions.ExecException;
import com.google.devtools.build.lib.concurrent.ThreadSafety.ThreadSafe;
import com.google.devtools.build.lib.remote.common.SimpleBlobStore;
import com.google.devtools.build.lib.remote.common.SimpleBlobStore.ActionKey;
import com.google.devtools.build.lib.remote.options.RemoteOptions;
import com.google.devtools.build.lib.remote.util.DigestUtil;
import com.google.devtools.build.lib.remote.util.Utils;
-import com.google.devtools.build.lib.util.io.FileOutErr;
import com.google.devtools.build.lib.vfs.Path;
import com.google.protobuf.ByteString;
import com.google.protobuf.InvalidProtocolBufferException;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.OutputStream;
-import java.util.Collection;
-import java.util.Map;
import javax.annotation.Nullable;
/**
@@ -77,60 +72,6 @@
}
@Override
- public void upload(
- SimpleBlobStore.ActionKey actionKey,
- Action action,
- Command command,
- Path execRoot,
- Collection<Path> files,
- FileOutErr outErr)
- throws ExecException, IOException, InterruptedException {
- ActionResult.Builder result = ActionResult.newBuilder();
- upload(result, actionKey, action, command, execRoot, files, /* uploadAction= */ true);
- if (outErr.getErrorPath().exists()) {
- Digest stdErrDigest = digestUtil.compute(outErr.getErrorPath());
- getFromFuture(uploadFile(stdErrDigest, outErr.getErrorPath()));
- result.setStderrDigest(stdErrDigest);
- }
- if (outErr.getOutputPath().exists()) {
- Digest stdoutDigest = digestUtil.compute(outErr.getOutputPath());
- getFromFuture(uploadFile(stdoutDigest, outErr.getOutputPath()));
- result.setStdoutDigest(stdoutDigest);
- }
- blobStore.putActionResult(actionKey, result.build());
- }
-
- public void upload(
- ActionResult.Builder result,
- SimpleBlobStore.ActionKey actionKey,
- Action action,
- Command command,
- Path execRoot,
- Collection<Path> files,
- boolean uploadAction)
- throws ExecException, IOException, InterruptedException {
- UploadManifest manifest =
- new UploadManifest(
- digestUtil,
- result,
- execRoot,
- options.incompatibleRemoteSymlinks,
- options.allowSymlinkUpload);
- manifest.addFiles(files);
- if (uploadAction) {
- manifest.addAction(actionKey, action, command);
- }
-
- for (Map.Entry<Digest, Path> entry : manifest.getDigestToFile().entrySet()) {
- getFromFuture(uploadFile(entry.getKey(), entry.getValue()));
- }
-
- for (Map.Entry<Digest, ByteString> entry : manifest.getDigestToBlobs().entrySet()) {
- getFromFuture(uploadBlob(entry.getKey(), entry.getValue()));
- }
- }
-
- @Override
public ListenableFuture<Void> uploadFile(Digest digest, Path file) {
return blobStore.uploadFile(digest, file);
}
@@ -140,6 +81,11 @@
return blobStore.uploadBlob(digest, data);
}
+ @Override
+ protected ImmutableSet<Digest> getMissingDigests(Iterable<Digest> digests) {
+ return ImmutableSet.copyOf(digests);
+ }
+
public boolean containsKey(Digest digest) throws IOException, InterruptedException {
return blobStore.contains(digest.getHash());
}
