Decouple RemoteRewoundActionSynchronizer from RemoteActionInputFetcher by depending on AbstractActionInputPrefetcher. PiperOrigin-RevId: 970627798 Change-Id: I3a2015b89dbbfd9701d7adbba63c65335ae279ae
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 index 2c83d1c..75ba0a3 100644 --- a/src/main/java/com/google/devtools/build/lib/remote/AbstractActionInputPrefetcher.java +++ b/src/main/java/com/google/devtools/build/lib/remote/AbstractActionInputPrefetcher.java
@@ -66,6 +66,7 @@ import java.io.IOException; import java.util.ArrayList; import java.util.Arrays; +import java.util.Collection; import java.util.List; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.atomic.AtomicBoolean; @@ -86,6 +87,7 @@ private final AsyncTaskCache.NoResult<Path> downloadCache = AsyncTaskCache.NoResult.create(); private final TempPathGenerator tempPathGenerator; private final OutputPermissions outputPermissions; + private final ConcurrentArtifactPathTrie rewoundActionOutputs = new ConcurrentArtifactPathTrie(); protected final Path execRoot; protected final RemoteOutputChecker remoteOutputChecker; @@ -286,7 +288,16 @@ * If true, then all previously acquired knowledge of the file system state of this path (e.g. the * existence of tree artifact directories or previously downloaded files) must be discarded. */ - protected abstract boolean forceRefetch(Path path); + protected boolean forceRefetch(Path path) { + // Caches for download operations and output directory creation need to be disregarded for the + // outputs of rewound actions as they may have been deleted after they were first created. + // Compare as fragments since execRoot may be located on a file system overlaying the host file + // system where downloads are written to. + PathFragment execRootFragment = execRoot.asFragment(); + PathFragment pathFragment = path.asFragment(); + return pathFragment.startsWith(execRootFragment) + && rewoundActionOutputs.contains(pathFragment.relativeTo(execRootFragment)); + } /** * Downloads file to the given path via its metadata. @@ -895,4 +906,25 @@ public RemoteOutputChecker getRemoteOutputChecker() { return remoteOutputChecker; } + + public void handleRewoundActionOutputs(Collection<Artifact> outputs) { + // SkyframeActionExecutor#prepareForRewinding does *not* invalidate outputDirectoryHelper + // because action file systems correspond to an ActionFileSystemType with + // inMemoryFileSystem() == true. While it is true that resetting outputDirectoryHelper isn't + // necessary to undo the caching of output directory creation during action preparation, we + // still need to reset here since outputDirectoryHelper is also used by + // AbstractActionInputPrefetcher. + if (outputDirectoryHelper != null) { + outputDirectoryHelper.invalidateTreeArtifactDirectoryCreation(outputs); + } + for (Artifact output : outputs) { + // Action templates have TreeFileArtifacts as outputs, which isn't supported by the trie. We + // only need to track the tree artifacts themselves. + if (output instanceof TreeFileArtifact) { + rewoundActionOutputs.add(output.getParent()); + } else { + rewoundActionOutputs.add(output); + } + } + } }
diff --git a/src/main/java/com/google/devtools/build/lib/remote/BUILD b/src/main/java/com/google/devtools/build/lib/remote/BUILD index bd86616..ffa5bcc 100644 --- a/src/main/java/com/google/devtools/build/lib/remote/BUILD +++ b/src/main/java/com/google/devtools/build/lib/remote/BUILD
@@ -304,12 +304,10 @@ deps = [ ":abstract_action_input_prefetcher", ":combined_cache", - ":concurrent_artifact_path_trie", ":remote_output_checker", "//src/main/java/com/google/devtools/build/lib/actions:action_input", "//src/main/java/com/google/devtools/build/lib/actions:action_input_prefetcher", "//src/main/java/com/google/devtools/build/lib/actions:action_output_directory_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/actions:virtual_action_input", "//src/main/java/com/google/devtools/build/lib/events", @@ -319,7 +317,6 @@ "//src/main/java/com/google/devtools/build/lib/remote/util:tracing_metadata_utils", "//src/main/java/com/google/devtools/build/lib/util:temp_path_generator", "//src/main/java/com/google/devtools/build/lib/vfs", - "//src/main/java/com/google/devtools/build/lib/vfs:pathfragment", "//third_party/java/guava:base", "//third_party/java/guava:concurrent", "//third_party/java/jsr305_annotations", @@ -956,6 +953,7 @@ name = "abstract_action_input_prefetcher", srcs = ["AbstractActionInputPrefetcher.java"], deps = [ + ":concurrent_artifact_path_trie", ":remote_output_checker", ":subtree_materializer", "//src/main/java/com/google/devtools/build/lib/actions", @@ -1035,7 +1033,7 @@ name = "remote_rewound_action_synchronizer", srcs = ["RemoteRewoundActionSynchronizer.java"], deps = [ - ":remote_action_input_fetcher", + ":abstract_action_input_prefetcher", "//src/main/java/com/google/devtools/build/lib/actions", "//src/main/java/com/google/devtools/build/lib/actions:action_lookup_data", "//src/main/java/com/google/devtools/build/lib/actions:artifacts",
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 index cafab2d..6b84a1b 100644 --- a/src/main/java/com/google/devtools/build/lib/remote/RemoteActionInputFetcher.java +++ b/src/main/java/com/google/devtools/build/lib/remote/RemoteActionInputFetcher.java
@@ -24,7 +24,6 @@ import com.google.devtools.build.lib.actions.ActionExecutionMetadata; import com.google.devtools.build.lib.actions.ActionInput; import com.google.devtools.build.lib.actions.ActionOutputDirectoryHelper; -import com.google.devtools.build.lib.actions.Artifact; import com.google.devtools.build.lib.actions.FileArtifactValue; import com.google.devtools.build.lib.actions.FileStateType; import com.google.devtools.build.lib.actions.VirtualActionInput; @@ -36,10 +35,8 @@ import com.google.devtools.build.lib.util.TempPathGenerator; 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 com.google.devtools.build.lib.vfs.Symlinks; import java.io.IOException; -import java.util.Collection; import javax.annotation.Nullable; /** @@ -54,7 +51,6 @@ private final String buildRequestId; private final String commandId; private final CombinedCache combinedCache; - private final ConcurrentArtifactPathTrie rewoundActionOutputs = new ConcurrentArtifactPathTrie(); RemoteActionInputFetcher( Reporter reporter, @@ -98,18 +94,6 @@ } @Override - protected boolean forceRefetch(Path path) { - // Caches for download operations and output directory creation need to be disregarded for the - // outputs of rewound actions as they may have been deleted after they were first created. - // Compare as fragments since execRoot may be located on a file system overlaying the host file - // system where downloads are written to. - PathFragment execRootFragment = execRoot.asFragment(); - PathFragment pathFragment = path.asFragment(); - return pathFragment.startsWith(execRootFragment) - && rewoundActionOutputs.contains(pathFragment.relativeTo(execRootFragment)); - } - - @Override protected ListenableFuture<Void> doDownloadFile( @Nullable ActionExecutionMetadata action, Reporter reporter, @@ -171,22 +155,4 @@ }), directExecutor()); } - - public void handleRewoundActionOutputs(Collection<Artifact> outputs) { - // SkyframeActionExecutor#prepareForRewinding does *not* call this method because the - // RemoteActionFileSystem corresponds to an ActionFileSystemType with inMemoryFileSystem() == - // true. While it is true that resetting outputDirectoryHelper isn't necessary to undo the - // caching of output directory creation during action preparation, we still need to reset here - // since outputDirectoryHelper is also used by AbstractActionInputPrefetcher. - outputDirectoryHelper.invalidateTreeArtifactDirectoryCreation(outputs); - for (Artifact output : outputs) { - // Action templates have TreeFileArtifacts as outputs, which isn't supported by the trie. We - // only need to track the tree artifacts themselves. - if (output instanceof Artifact.TreeFileArtifact) { - rewoundActionOutputs.add(output.getParent()); - } else { - rewoundActionOutputs.add(output); - } - } - } }
diff --git a/src/main/java/com/google/devtools/build/lib/remote/RemoteRewoundActionSynchronizer.java b/src/main/java/com/google/devtools/build/lib/remote/RemoteRewoundActionSynchronizer.java index b8789d6..5141e2f 100644 --- a/src/main/java/com/google/devtools/build/lib/remote/RemoteRewoundActionSynchronizer.java +++ b/src/main/java/com/google/devtools/build/lib/remote/RemoteRewoundActionSynchronizer.java
@@ -14,7 +14,6 @@ package com.google.devtools.build.lib.remote; - import com.github.benmanes.caffeine.cache.Caffeine; import com.github.benmanes.caffeine.cache.LoadingCache; import com.google.common.collect.ImmutableList; @@ -41,13 +40,13 @@ * by actual files on disk and requires synchronization to ensure that action outputs aren't deleted * while they are being read. */ -final class RemoteRewoundActionSynchronizer implements RewoundActionSynchronizer { +public final class RemoteRewoundActionSynchronizer implements RewoundActionSynchronizer { /** A task with a cancellation callback. */ public interface Cancellable { void cancel() throws InterruptedException; } - private final RemoteActionInputFetcher actionInputFetcher; + private final AbstractActionInputPrefetcher actionInputFetcher; private final ConcurrentHashMap<ActionLookupData, Cancellable> outputUploadTasks = new ConcurrentHashMap<>(); @@ -75,7 +74,7 @@ // are no longer needed. @Nullable private volatile LoadingCache<ActionLookupData, ReadWriteLock> fineLocks; - public RemoteRewoundActionSynchronizer(RemoteActionInputFetcher actionInputFetcher) { + public RemoteRewoundActionSynchronizer(AbstractActionInputPrefetcher actionInputFetcher) { this.actionInputFetcher = actionInputFetcher; }
diff --git a/src/test/java/com/google/devtools/build/lib/remote/BUILD b/src/test/java/com/google/devtools/build/lib/remote/BUILD index b61243e..4322f43 100644 --- a/src/test/java/com/google/devtools/build/lib/remote/BUILD +++ b/src/test/java/com/google/devtools/build/lib/remote/BUILD
@@ -322,6 +322,7 @@ "//src/main/java/com/google/devtools/build/lib/analysis:analysis_cluster", "//src/main/java/com/google/devtools/build/lib/skyframe:action_execution_value", "//src/main/java/com/google/devtools/build/lib/skyframe:tree_artifact_value", + "//src/main/java/com/google/devtools/build/lib/skyframe/rewinding:action_rewound_event", "//src/main/java/com/google/devtools/build/lib/util:command", "//src/main/java/com/google/devtools/build/lib/util:os", "//src/main/java/com/google/devtools/build/lib/util/io:out-err",
diff --git a/src/test/java/com/google/devtools/build/lib/remote/BuildWithoutTheBytesIntegrationTest.java b/src/test/java/com/google/devtools/build/lib/remote/BuildWithoutTheBytesIntegrationTest.java index ef444e4..6fab21b 100644 --- a/src/test/java/com/google/devtools/build/lib/remote/BuildWithoutTheBytesIntegrationTest.java +++ b/src/test/java/com/google/devtools/build/lib/remote/BuildWithoutTheBytesIntegrationTest.java
@@ -55,6 +55,7 @@ @ClassRule @Rule public static final WorkerInstance worker = IntegrationTestUtils.createWorker(); @TestParameter public boolean useDiskCache; + private Path diskCacheDir; @Override protected ImmutableList<Class<? extends OptionsBase>> getStartupOptionClasses() { @@ -89,7 +90,8 @@ } if (useDiskCache) { - addOptions("--disk_cache=" + UUID.randomUUID()); + diskCacheDir = getWorkspace().getRelative(UUID.randomUUID().toString()); + addOptions("--disk_cache=" + diskCacheDir.getPathString()); } } @@ -142,8 +144,15 @@ @Override protected void evictAllBlobs() throws Exception { worker.reset(); - if (useDiskCache) { - addOptions("--disk_cache=" + UUID.randomUUID()); + if (useDiskCache && diskCacheDir != null) { + Path casDir = diskCacheDir.getRelative("cas"); + if (casDir.exists()) { + casDir.deleteTreesBelow(); + } + Path acDir = diskCacheDir.getRelative("ac"); + if (acDir.exists()) { + acDir.deleteTreesBelow(); + } } } @@ -425,7 +434,6 @@ assertOutputsDoNotExist("//a:hello"); } - @Test public void leaseExtension() throws Exception { // The lease service is only used when action rewinding is disabled. @@ -699,7 +707,6 @@ buildTarget("//:gen"); } - @Test public void remoteFilesExpiredBetweenBuilds(@TestParameter boolean actionRewinding) throws Exception {
diff --git a/src/test/java/com/google/devtools/build/lib/remote/BuildWithoutTheBytesIntegrationTestBase.java b/src/test/java/com/google/devtools/build/lib/remote/BuildWithoutTheBytesIntegrationTestBase.java index d7afa23..174d114 100644 --- a/src/test/java/com/google/devtools/build/lib/remote/BuildWithoutTheBytesIntegrationTestBase.java +++ b/src/test/java/com/google/devtools/build/lib/remote/BuildWithoutTheBytesIntegrationTestBase.java
@@ -24,8 +24,10 @@ import com.google.common.collect.ImmutableList; import com.google.common.collect.ImmutableMap; +import com.google.common.eventbus.AllowConcurrentEvents; import com.google.common.eventbus.Subscribe; import com.google.devtools.build.lib.actions.ActionExecutedEvent; +import com.google.devtools.build.lib.actions.ActionStartedEvent; import com.google.devtools.build.lib.actions.Artifact; import com.google.devtools.build.lib.actions.BuildFailedException; import com.google.devtools.build.lib.actions.CachedActionEvent; @@ -34,6 +36,7 @@ import com.google.devtools.build.lib.buildtool.util.BuildIntegrationTestCase; import com.google.devtools.build.lib.skyframe.ActionExecutionValue; import com.google.devtools.build.lib.skyframe.TreeArtifactValue; +import com.google.devtools.build.lib.skyframe.rewinding.ActionRewoundEvent; import com.google.devtools.build.lib.testutil.TestUtils; import com.google.devtools.build.lib.util.CommandBuilder; import com.google.devtools.build.lib.util.OS; @@ -46,6 +49,7 @@ import java.io.IOException; import java.util.ArrayList; import java.util.List; +import java.util.concurrent.CountDownLatch; import org.junit.Test; /** Base class for integration tests for BwoB. */ @@ -2240,6 +2244,7 @@ disableActionRewinding(); addOptions("--strategy_regexp=.*bar=local"); var error = assertThrows(BuildFailedException.class, () -> buildTarget("//a:bar")); + assertThat(error).hasMessageThat().contains("Lost inputs no longer available remotely"); assertThat(error).hasMessageThat().contains(String.format("%s/%s", hashCode, bytes.length)); assertThat(error.getDetailedExitCode().getExitCode().getNumericExitCode()).isEqualTo(39); } @@ -2519,6 +2524,132 @@ assertValidOutputFile("a/foo.out/file-inside", "hello world"); } + @Test + public void actionRewinding_concurrentConsumersOfRewoundAction() throws Exception { + enableActionRewinding(); + + // Arrange: Prepare workspace where action 'foo' generates two outputs: + // 'foo1.out' (consumed by bar1) and 'foo2.out' (consumed concurrently by bar2). + write( + "a/BUILD", + """ + genrule( + name = "foo", + outs = [ + "foo1.out", + "foo2.out", + ], + cmd = "seq 1 500 > $(location foo1.out); seq 1 500 > $(location foo2.out)", + ) + + genrule( + name = "bar1", + srcs = ["foo1.out"], + outs = ["bar1.out"], + cmd = "cat $(location foo1.out) > $@", + ) + + genrule( + name = "bar2", + srcs = [ + "foo2.out", + "bar2.in", + ], + outs = ["bar2.out"], + cmd = "while [ ! -f a/bar2.marker ]; do cat $(location foo2.out) > /dev/null || exit 1; sleep 0.02; done; cat $(location foo2.out) $(location bar2.in) > $@", + ) + """); + write("a/bar2.in", "bar2"); + + // Clean build: build foo remotely so intermediate outputs foo1.out and foo2.out are in CAS + buildTarget("//a:foo"); + assertOutputDoesNotExist("a/foo1.out"); + assertOutputDoesNotExist("a/foo2.out"); + + // Act: Run bar1 and bar2 concurrently. + // When bar2 starts executing (and prefetches foo2.out), evict blobs from CAS. + // bar1 starts after eviction, sees foo1.out missing from CAS, and triggers action rewinding + // for foo. + // When foo is rewound, foo's preparation deletes foo2.out from disk while bar2 is reading it. + // Without synchronization, bar2 fails on missing foo2.out. + addOptions("--strategy_regexp=.*bar=local", "--jobs=4"); + + CountDownLatch bar2Started = new CountDownLatch(1); + CountDownLatch evictionFinished = new CountDownLatch(1); + CountDownLatch bar1Rewound = new CountDownLatch(1); + Path markerPath = getWorkspace().getRelative("a/bar2.marker"); + + runtimeWrapper.registerSubscriber( + new Object() { + @Subscribe + @AllowConcurrentEvents + public void actionStarted(ActionStartedEvent event) { + String label = event.getAction().getOwner().getLabel().toString(); + if (label.equals("//a:bar2")) { + bar2Started.countDown(); + } else if (label.equals("//a:bar1")) { + try { + // Ensure bar2 has started and eviction has completed before bar1 attempts + // prefetching + bar2Started.await(); + evictionFinished.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new RuntimeException(e); + } + } + } + + @Subscribe + @AllowConcurrentEvents + public void actionRewound(ActionRewoundEvent event) { + if (event + .getFailedRewoundAction() + .getOwner() + .getLabel() + .toString() + .equals("//a:bar1")) { + bar1Rewound.countDown(); + try { + FileSystemUtils.createEmptyFile(markerPath); + } catch (IOException e) { + throw new RuntimeException(e); + } + } + } + }); + + new Thread( + () -> { + try { + bar2Started.await(); + evictAllBlobs(); + evictionFinished.countDown(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new RuntimeException(e); + } catch (Exception e) { + throw new RuntimeException(e); + } + }) + .start(); + + try { + buildTarget("//a:bar1", "//a:bar2"); + } finally { + // Assert test preconditions: ensure eviction completed, bar2 was actively running, + // and action rewinding was legitimately triggered for bar1. + assertThat(evictionFinished.getCount()).isEqualTo(0); + assertThat(bar2Started.getCount()).isEqualTo(0); + assertThat(bar1Rewound.getCount()).isEqualTo(0); + } + waitDownloads(); + + // Assert: Both targets succeed + assertThat(getOutputPath("a/bar1.out").exists()).isTrue(); + assertThat(getOutputPath("a/bar2.out").exists()).isTrue(); + } + protected void restartServer() throws Exception { // Simulates a server restart createRuntimeWrapper();