Delete downloadFile in favor of prefetchFiles in AbstractActionInputP…

…refetcher. This simplifies the interface and makes it less likely that we get the subtle synchronization logic wrong in the future. Due to these changes, AbstractActionInputPrefetcher is now too low-level to handle the presentation of a BulkTransferException to the user, so the logic to convert it into an EnvironmentalExecException is lifted into AbstractSpawnStrategy. PiperOrigin-RevId: 516816172 Change-Id: I6e4b57fb90d9c84821dee36ae29c92cee02a906f
fweikert · May 25, 2023 · 1783aca · 1783aca
1 parent df558be
commit 1783aca
Show file tree

Hide file tree

Showing 11 changed files with 125 additions and 199 deletions.
diff --git a/src/main/java/com/google/devtools/build/lib/exec/AbstractSpawnStrategy.java b/src/main/java/com/google/devtools/build/lib/exec/AbstractSpawnStrategy.java
@@ -15,10 +15,12 @@
 package com.google.devtools.build.lib.exec;
 
 import static com.google.common.util.concurrent.Futures.immediateVoidFuture;
+import static com.google.common.util.concurrent.MoreExecutors.directExecutor;
 
 import com.google.common.base.Preconditions;
 import com.google.common.base.Strings;
 import com.google.common.collect.ImmutableList;
+import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
 import com.google.devtools.build.lib.actions.ActionContext;
 import com.google.devtools.build.lib.actions.ActionExecutionContext;
@@ -48,6 +50,7 @@
 import com.google.devtools.build.lib.exec.SpawnRunner.SpawnExecutionContext;
 import com.google.devtools.build.lib.profiler.Profiler;
 import com.google.devtools.build.lib.profiler.SilentCloseable;
+import com.google.devtools.build.lib.remote.common.BulkTransferException;
 import com.google.devtools.build.lib.server.FailureDetails;
 import com.google.devtools.build.lib.server.FailureDetails.FailureDetail;
 import com.google.devtools.build.lib.server.FailureDetails.Spawn.Code;
@@ -246,13 +249,35 @@ public int getId() {
     public ListenableFuture<Void> prefetchInputs()
         throws IOException, ForbiddenActionInputException {
       if (Spawns.shouldPrefetchInputsForLocalExecution(spawn)) {
-        return actionExecutionContext
-            .getActionInputPrefetcher()
-            .prefetchFiles(
-                getInputMapping(PathFragment.EMPTY_FRAGMENT, /* willAccessRepeatedly= */ true)
-                    .values(),
-                getMetadataProvider(),
-                Priority.MEDIUM);
+        return Futures.catchingAsync(
+            actionExecutionContext
+                .getActionInputPrefetcher()
+                .prefetchFiles(
+                    getInputMapping(PathFragment.EMPTY_FRAGMENT, /* willAccessRepeatedly= */ true)
+                        .values(),
+                    getMetadataProvider(),
+                    Priority.MEDIUM),
+            BulkTransferException.class,
+            (BulkTransferException e) -> {
+              if (BulkTransferException.allCausedByCacheNotFoundException(e)) {
+                var code =
+                    executionOptions.useNewExitCodeForLostInputs
+                        ? Code.REMOTE_CACHE_EVICTED
+                        : Code.REMOTE_CACHE_FAILED;
+                throw new EnvironmentalExecException(
+                    e,
+                    FailureDetail.newBuilder()
+                        .setMessage(
+                            "Failed to fetch blobs because they do not exist remotely."
+                                + " Build without the Bytes does not work if your remote"
+                                + " cache evicts blobs during builds.")
+                        .setSpawn(FailureDetails.Spawn.newBuilder().setCode(code))
+                        .build());
+              } else {
+                throw e;
+              }
+            },
+            directExecutor());
       }
 
       return immediateVoidFuture();

diff --git a/src/main/java/com/google/devtools/build/lib/exec/BUILD b/src/main/java/com/google/devtools/build/lib/exec/BUILD
@@ -27,6 +27,7 @@ java_library(
         "//src/main/java/com/google/devtools/build/lib/actions:artifacts",
         "//src/main/java/com/google/devtools/build/lib/events",
         "//src/main/java/com/google/devtools/build/lib/profiler",
+        "//src/main/java/com/google/devtools/build/lib/remote/common:bulk_transfer_exception",
         "//src/main/java/com/google/devtools/build/lib/util:command",
         "//src/main/java/com/google/devtools/build/lib/util/io",
         "//src/main/java/com/google/devtools/build/lib/vfs",

diff --git a/src/main/java/com/google/devtools/build/lib/remote/AbstractActionInputPrefetcher.java b/src/main/java/com/google/devtools/build/lib/remote/AbstractActionInputPrefetcher.java
@@ -22,7 +22,6 @@
 import static com.google.devtools.build.lib.remote.util.RxFutures.toListenableFuture;
 import static com.google.devtools.build.lib.remote.util.RxUtils.mergeBulkTransfer;
 import static com.google.devtools.build.lib.remote.util.RxUtils.toTransferResult;
-import static com.google.devtools.build.lib.remote.util.Utils.getFromFuture;
 
 import com.google.auto.value.AutoValue;
 import com.google.common.annotations.VisibleForTesting;
@@ -277,11 +276,6 @@ protected abstract ListenableFuture<Void> doDownloadFile(
 
   protected void prefetchVirtualActionInput(VirtualActionInput input) throws IOException {}
 
-  /** Transforms the error encountered during the prefetch . */
-  protected Completable onErrorResumeNext(Throwable error) {
-    return Completable.error(error);
-  }
-
   /**
    * Fetches remotely stored action outputs, that are inputs to this spawn, and stores them under
    * their path in the output base.
@@ -323,8 +317,7 @@ public ListenableFuture<Void> prefetchFiles(
 
     Completable prefetch =
         Completable.using(
-                () -> dirCtx, ctx -> mergeBulkTransfer(transfers), DirectoryContext::close)
-            .onErrorResumeNext(this::onErrorResumeNext);
+            () -> dirCtx, ctx -> mergeBulkTransfer(transfers), DirectoryContext::close);
 
     return toListenableFuture(prefetch);
   }
@@ -427,30 +420,11 @@ private Symlink maybeGetSymlink(
     return null;
   }
 
-  /**
-   * Downloads file into the {@code path} with its metadata.
-   *
-   * <p>The file will be written into a temporary file and moved to the final destination after the
-   * download finished.
-   */
-  private Completable downloadFileRx(
-      DirectoryContext dirCtx,
-      Path path,
-      @Nullable Path treeRoot,
-      @Nullable ActionInput actionInput,
-      FileArtifactValue metadata,
-      Priority priority) {
-    if (!canDownloadFile(path, metadata)) {
-      return Completable.complete();
-    }
-    return downloadFileNoCheckRx(dirCtx, path, treeRoot, actionInput, metadata, priority);
-  }
-
   private Completable downloadFileNoCheckRx(
       DirectoryContext dirCtx,
       Path path,
       @Nullable Path treeRoot,
-      @Nullable ActionInput actionInput,
+      ActionInput actionInput,
       FileArtifactValue metadata,
       Priority priority) {
     if (path.isSymbolicLink()) {
@@ -492,7 +466,7 @@ private Completable downloadFileNoCheckRx(
                 /* eager= */ false)
             .doOnError(
                 error -> {
-                  if (error instanceof CacheNotFoundException && actionInput != null) {
+                  if (error instanceof CacheNotFoundException) {
                     missingActionInputs.add(actionInput);
                   }
                 });
@@ -508,37 +482,6 @@ private Completable downloadFileNoCheckRx(
             }));
   }
 
-  /**
-   * Download file to the {@code path} with given metadata. Blocking await for the download to
-   * complete.
-   *
-   * <p>The file will be written into a temporary file and moved to the final destination after the
-   * download finished.
-   */
-  public void downloadFile(Path path, @Nullable ActionInput actionInput, FileArtifactValue metadata)
-      throws IOException, InterruptedException {
-    getFromFuture(downloadFileAsync(path.asFragment(), actionInput, metadata, Priority.CRITICAL));
-  }
-
-  protected ListenableFuture<Void> downloadFileAsync(
-      PathFragment path,
-      @Nullable ActionInput actionInput,
-      FileArtifactValue metadata,
-      Priority priority) {
-    return toListenableFuture(
-        Completable.using(
-            DirectoryContext::new,
-            dirCtx ->
-                downloadFileRx(
-                    dirCtx,
-                    execRoot.getFileSystem().getPath(path),
-                    /* treeRoot= */ null,
-                    actionInput,
-                    metadata,
-                    priority),
-            DirectoryContext::close));
-  }
-
   private void finalizeDownload(
       DirectoryContext dirCtx, @Nullable Path treeRoot, Path tmpPath, Path finalPath)
       throws IOException {

diff --git a/src/main/java/com/google/devtools/build/lib/remote/BUILD b/src/main/java/com/google/devtools/build/lib/remote/BUILD
@@ -207,6 +207,7 @@ java_library(
     deps = [
         ":abstract_action_input_prefetcher",
         "//src/main/java/com/google/devtools/build/lib/actions",
+        "//src/main/java/com/google/devtools/build/lib/actions:action_input_helper",
         "//src/main/java/com/google/devtools/build/lib/actions:artifacts",
         "//src/main/java/com/google/devtools/build/lib/actions:file_metadata",
         "//src/main/java/com/google/devtools/build/lib/analysis:analysis_cluster",
@@ -220,6 +221,7 @@ java_library(
         "//src/main/java/com/google/devtools/build/lib/skyframe:tree_artifact_value",
         "//src/main/java/com/google/devtools/build/lib/util",
         "//src/main/java/com/google/devtools/build/lib/vfs",
+        "//src/main/java/com/google/devtools/build/lib/vfs:pathfragment",
         "//src/main/java/com/google/devtools/build/skyframe",
         "//src/main/java/com/google/devtools/build/skyframe:skyframe-objects",
         "//third_party:flogger",

diff --git a/src/main/java/com/google/devtools/build/lib/remote/RemoteActionFileSystem.java b/src/main/java/com/google/devtools/build/lib/remote/RemoteActionFileSystem.java
@@ -20,12 +20,15 @@
 import static com.google.common.base.Preconditions.checkState;
 import static com.google.common.collect.ImmutableMap.toImmutableMap;
 import static com.google.common.collect.Streams.stream;
+import static com.google.devtools.build.lib.remote.util.Utils.getFromFuture;
 
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import com.google.devtools.build.lib.actions.ActionInput;
+import com.google.devtools.build.lib.actions.ActionInputHelper;
 import com.google.devtools.build.lib.actions.ActionInputMap;
+import com.google.devtools.build.lib.actions.ActionInputPrefetcher.Priority;
 import com.google.devtools.build.lib.actions.Artifact;
 import com.google.devtools.build.lib.actions.Artifact.SpecialArtifact;
 import com.google.devtools.build.lib.actions.Artifact.TreeFileArtifact;
@@ -566,10 +569,6 @@ public ActionInput getInput(String execPath) {
     return null;
   }
 
-  ActionInput getActionInput(PathFragment path) {
-    return getInput(path.relativeTo(execRoot).getPathString());
-  }
-
   @Nullable
   @Override
   public FileArtifactValue getMetadata(ActionInput input) throws IOException {
@@ -605,7 +604,7 @@ private FileArtifactValue getMetadataByExecPath(PathFragment execPath) {
   }
 
   @Nullable
-  RemoteFileArtifactValue getRemoteMetadata(PathFragment path) {
+  private RemoteFileArtifactValue getRemoteMetadata(PathFragment path) {
     if (!isOutput(path)) {
       return null;
     }
@@ -631,15 +630,21 @@ private TreeArtifactValue getRemoteTreeMetadata(PathFragment path) {
   }
 
   private void downloadFileIfRemote(PathFragment path) throws IOException {
-    FileArtifactValue m = getRemoteMetadata(path);
-    if (m != null) {
-      try {
-        inputFetcher.downloadFile(delegateFs.getPath(path), getActionInput(path), m);
-      } catch (InterruptedException e) {
-        Thread.currentThread().interrupt();
-        throw new IOException(
-            String.format("Received interrupt while fetching file '%s'", path), e);
+    if (!isRemote(path)) {
+      return;
+    }
+    PathFragment execPath = path.relativeTo(execRoot);
+    try {
+      ActionInput input = getInput(execPath.getPathString());
+      if (input == null) {
+        // For undeclared outputs, getInput returns null as there's no artifact associated with the
+        // path. Therefore, we synthesize one here just so we're able to call prefetchFiles.
+        input = ActionInputHelper.fromPath(execPath);
       }
+      getFromFuture(inputFetcher.prefetchFiles(ImmutableList.of(input), this, Priority.CRITICAL));
+    } catch (InterruptedException e) {
+      Thread.currentThread().interrupt();
+      throw new IOException(String.format("Received interrupt while fetching file '%s'", path), e);
     }
   }
 

diff --git a/src/main/java/com/google/devtools/build/lib/remote/RemoteActionInputFetcher.java b/src/main/java/com/google/devtools/build/lib/remote/RemoteActionInputFetcher.java
@@ -21,25 +21,19 @@
 import com.google.common.base.Preconditions;
 import com.google.common.collect.ImmutableList;
 import com.google.common.util.concurrent.ListenableFuture;
-import com.google.devtools.build.lib.actions.EnvironmentalExecException;
 import com.google.devtools.build.lib.actions.FileArtifactValue;
 import com.google.devtools.build.lib.actions.cache.VirtualActionInput;
 import com.google.devtools.build.lib.events.ExtendedEventHandler.FetchProgress;
 import com.google.devtools.build.lib.events.Reporter;
 import com.google.devtools.build.lib.exec.SpawnProgressEvent;
 import com.google.devtools.build.lib.remote.RemoteCache.DownloadProgressReporter;
-import com.google.devtools.build.lib.remote.common.BulkTransferException;
 import com.google.devtools.build.lib.remote.common.RemoteActionExecutionContext;
 import com.google.devtools.build.lib.remote.util.DigestUtil;
 import com.google.devtools.build.lib.remote.util.TempPathGenerator;
 import com.google.devtools.build.lib.remote.util.TracingMetadataUtils;
-import com.google.devtools.build.lib.server.FailureDetails;
-import com.google.devtools.build.lib.server.FailureDetails.FailureDetail;
-import com.google.devtools.build.lib.server.FailureDetails.Spawn.Code;
 import com.google.devtools.build.lib.vfs.OutputPermissions;
 import com.google.devtools.build.lib.vfs.Path;
 import com.google.devtools.build.lib.vfs.PathFragment;
-import io.reactivex.rxjava3.core.Completable;
 import java.io.IOException;
 import java.util.regex.Pattern;
 
@@ -54,7 +48,6 @@ class RemoteActionInputFetcher extends AbstractActionInputPrefetcher {
   private final String buildRequestId;
   private final String commandId;
   private final RemoteCache remoteCache;
-  private final boolean useNewExitCodeForLostInputs;
 
   RemoteActionInputFetcher(
       Reporter reporter,
@@ -64,13 +57,11 @@ class RemoteActionInputFetcher extends AbstractActionInputPrefetcher {
       Path execRoot,
       TempPathGenerator tempPathGenerator,
       ImmutableList<Pattern> patternsToDownload,
-      OutputPermissions outputPermissions,
-      boolean useNewExitCodeForLostInputs) {
+      OutputPermissions outputPermissions) {
     super(reporter, execRoot, tempPathGenerator, patternsToDownload, outputPermissions);
     this.buildRequestId = Preconditions.checkNotNull(buildRequestId);
     this.commandId = Preconditions.checkNotNull(commandId);
     this.remoteCache = Preconditions.checkNotNull(remoteCache);
-    this.useNewExitCodeForLostInputs = useNewExitCodeForLostInputs;
   }
 
   @Override
@@ -141,25 +132,4 @@ public boolean isFinished() {
       return progress.finished();
     }
   }
-
-  @Override
-  protected Completable onErrorResumeNext(Throwable error) {
-    if (error instanceof BulkTransferException) {
-      if (((BulkTransferException) error).allCausedByCacheNotFoundException()) {
-        var code =
-            useNewExitCodeForLostInputs ? Code.REMOTE_CACHE_EVICTED : Code.REMOTE_CACHE_FAILED;
-        error =
-            new EnvironmentalExecException(
-                (BulkTransferException) error,
-                FailureDetail.newBuilder()
-                    .setMessage(
-                        "Failed to fetch blobs because they do not exist remotely."
-                            + " Build without the Bytes does not work if your remote"
-                            + " cache evicts blobs during builds")
-                    .setSpawn(FailureDetails.Spawn.newBuilder().setCode(code))
-                    .build());
-      }
-    }
-    return Completable.error(error);
-  }
 }