From bf37b2fd1b8f3518273d0533d18313eae0bcb305 Mon Sep 17 00:00:00 2001 From: Fabian Meumertzheim Date: Mon, 10 Aug 2026 13:17:32 +0200 Subject: [PATCH 1/3] Unlock rewinding locks on exceptions and add profiler spans This guards against interrupts and Errors causing locks to be held indefinitely. Along the way add more profiler spans. --- .../RemoteRewoundActionSynchronizer.java | 23 +++++++++++++++---- 1 file changed, 19 insertions(+), 4 deletions(-) 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 b14263d6ddb3c7..960680d762d69a 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 @@ -149,7 +149,10 @@ private SilentCloseable enterActionPreparationForRewinding(Action action) if (localCoarseLock != null) { // This is the first time a rewound action has attempted to prepare for its execution. // Switch to using the fine locks under the protection of the coarse write lock. - localCoarseLock.writeLock().lockInterruptibly(); + try (SilentCloseable c = + Profiler.instance().profile(ProfilerTask.ACTION_LOCK, "action.prepareFirstRewinding")) { + localCoarseLock.writeLock().lockInterruptibly(); + } try { // Check again under the lock to avoid a race between multiple rewound actions attempting // to prepare for execution at the same time. @@ -166,6 +169,8 @@ private SilentCloseable enterActionPreparationForRewinding(Action action) // (https://github.com/openjdk/jdk/blob/b349f661ea5f14b258191134714a7e712c90ef3e/src/java.base/share/classes/java/util/concurrent/locks/StampedLock.java#L1039), // TODO: Investigate the effect of fair locks on build wall time. .build((ActionLookupData unused) -> new StampedLock().asReadWriteLock()); + // Must be assigned after fineLocks as lockArtifactsForConsumption relies on a null + // coarseLock implying a non-null fineLocks. coarseLock = null; } } finally { @@ -174,8 +179,18 @@ private SilentCloseable enterActionPreparationForRewinding(Action action) } var writeLock = fineLocks.get(outputKeyFor(action)).writeLock(); - writeLock.lockInterruptibly(); - prepareOutputsForRewinding(action); + try (SilentCloseable c = + Profiler.instance() + .profile(ProfilerTask.ACTION_LOCK, "action.awaitRewoundActionConsumers")) { + writeLock.lockInterruptibly(); + } + try (SilentCloseable c = + Profiler.instance().profile(ProfilerTask.INFO, "action.prepareOutputsForRewinding")) { + prepareOutputsForRewinding(action); + } catch (Throwable t) { + writeLock.unlock(); + throw t; + } return writeLock::unlock; } @@ -265,7 +280,7 @@ private SilentCloseable lockArtifactsForConsumption( readLock.lockInterruptibly(); locksToUnlockBuilder.add(readLock); } - } catch (InterruptedException e) { + } catch (Throwable e) { for (var readLock : locksToUnlockBuilder.build()) { readLock.unlock(); } From a532aa9092b0e3584841e6e4f43a5d544783e4c0 Mon Sep 17 00:00:00 2001 From: Fabian Meumertzheim Date: Mon, 10 Aug 2026 13:21:06 +0200 Subject: [PATCH 2/3] Guard ActionTemplate outputs by the key of the template Actions generated by an `ActionTemplate` have `TreeFileArtifact`s as outputs, whose generating action key is that of the expansion action. Consumers, however, depend on the parent tree artifact and thus lock the key of the *template*, so a rewound expansion action acquired a write lock that no consumer ever contended on and could prepare for its re-execution while its outputs were still being read. Both the outputs of a rewound action and the inputs of any action are now mapped to the generating action key of the outermost tree artifact containing them, which is the template for the outputs of an expansion. This key is shared by all output trees of a template and is thus still a single lock per action - the proof of deadlock freedom relies on this. Since the actions of a single expansion may form a chain, an action can have an input that is guarded by the very key it holds the write lock of. As the locks aren't reentrant, a rewound action skips the read lock of that key. Also keys `outputUploadTasks` by `ActionLookupData` instead of by the `Action` instance, which isn't stable across rewinding for actions generated by an `ActionTemplate`. --- .../RemoteRewoundActionSynchronizer.java | 127 +++++--- .../lib/skyframe/SkyframeActionExecutor.java | 5 +- .../devtools/build/lib/vfs/OutputService.java | 5 +- .../build/lib/skyframe/rewinding/BUILD | 1 + .../lib/skyframe/rewinding/RewindingTest.java | 5 + .../rewinding/RewindingTestsHelper.java | 273 +++++++++++++++++- 6 files changed, 374 insertions(+), 42 deletions(-) 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 960680d762d69a..b8789d63950e37 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,6 +14,7 @@ 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; @@ -47,7 +48,7 @@ public interface Cancellable { } private final RemoteActionInputFetcher actionInputFetcher; - private final ConcurrentHashMap outputUploadTasks = + private final ConcurrentHashMap outputUploadTasks = new ConcurrentHashMap<>(); // A single coarse lock is used to synchronize rewound actions (writers) and both rewound and @@ -92,33 +93,40 @@ on any (rewound or non-rewound) action executions while it holds read locks and * Nodes are given by the currently active Skyframe action execution threads, each of which is identified with the action it is (or will be) executing. Actions are in one-to-one correspondence with the ActionLookupData that is used as the key in the fine locks map. - * For each pair of actions A_1 and A_2, there is an edge from A_1 to A_2 labeled with XY(A_3) - if A_1 is waiting for the X lock of A_3 and A_2 currently holds the Y lock of A_3, where X and - Y are either R (for read) or W (for write). The resulting graph may have parallel edges with - distinct labels. + * For each pair of actions A_1 and A_2, there is an edge from A_1 to A_2 labeled with XY(K) + if A_1 is waiting for the X lock of the key K and A_2 currently holds the Y lock of K, where X + and Y are either R (for read) or W (for write). The resulting graph may have parallel edges + with distinct labels. + + Say that an action A "covers" a key K if A is the action identified by K, or if K identifies an + ActionTemplate and A is one of its expanded actions. By construction of outputKeyFor, the write + lock of K is only ever acquired by actions covering K, and every action covers exactly one key. - Let C be any directed cycle in the graph representing a deadlock, let A_1 -[XY(A_3)]-> A_2 be an + Let C be any directed cycle in the graph representing a deadlock, let A_1 -[XY(K)]-> A_2 be an edge in C and consider the following cases for the pair XY: * RR: Since a read-write lock whose read lock is held by at least one thread doesn't block any other thread from acquiring its read lock, this case doesn't occur. - * WW: The write lock of A_3 is only ever (attempted to be) acquired by A_3 itself when it is - rewound, which means that the edge would necessarily be of the shape A_3 -[WW(A_3)]-> A_3. - But this isn't possible since the write lock for an action is only acquired in one place ( - enterActionPreparationForRewinding) and not recursively. - * WR: In this case, A_1 attempts to acquire a write lock, which only happens when A_1 is a - rewound action about to prepare for its (re-)execution. This means that the edge is - necessarily of the shape A_1 -[WR(A_1)]-> A_2. While a rewound action is waiting for its - own write lock in enterActionPreparation, it doesn't hold any locks since - enterActionExecution hasn't been called yet in SkyframeActionExecutor and all past - executions of the action have released all their locks due to use of try-with-resources. - This means that A_1 can't have any incoming edges in the wait-for graph, which is a - contradiction to the assumption that it is contained in the directed cycle C. - - We conclude that XY = RW. Since the write lock of A_3 is only ever acquired by A_3 itself, all - edges in C are of the form A_1 -[RW(A_2)]-> A_2. But by construction of inputKeysFor, the - action A_1 is attempting to acquire the read locks of all its inputs' generating actions, and - thus the action A_1 depends on one of the outputs of A_2 (*). + * WW and WR: In both cases, A_1 attempts to acquire a write lock, which only happens when A_1 is + a rewound action about to prepare for its (re-)execution. While a rewound action is waiting + for a write lock in enterActionPreparation, it doesn't hold any locks: enterActionExecution + hasn't been called yet in SkyframeActionExecutor, it only ever acquires the single write + lock it is waiting for, and all past executions of the action have released all their locks + due to use of try-with-resources. This means that A_1 can't have any incoming edges in the + wait-for graph, which is a contradiction to the assumption that it is contained in the + directed cycle C. + + We conclude that XY = RW, so all edges in C are of the form A_1 -[RW(K)]-> A_2 with A_2 covering + K. Since every node of C also has an incoming edge, every node of C holds a write lock and thus + covers the key of that lock. + + By construction of inputKeysFor, A_1 is waiting for R(K) because it has an input guarded by K, + which is either an output of the action identified by K, or a file in a tree artifact declared + by the ActionTemplate identified by K. In the latter case, if the input is an individual file + rather than the tree artifact itself, then A_1 is an expanded action of that template and thus + covers K - but A_1 covers exactly one key, namely the one of the write lock it holds, which A_2 + holds instead. So A_1 depends on the tree artifact in its entirety and thus on all actions + covering K, in particular on A_2 (*). Applied to all edges of C, we conclude that there is a corresponding directed cycle in the action graph, which is a contradiction since Bazel disallows dependency cycles. @@ -127,6 +135,12 @@ thus the action A_1 depends on one of the outputs of A_2 (*). * The proof would not go through at (*) if fineLocks were replaced by a Striped lock structure with a fixed number of locks. In fact, this gives rise to a deadlock if the number of stripes is at least 2, but low enough that distinct generating actions hash to the same stripe. + * It is crucial that an action only ever acquires a single write lock: a rewound action holding + one write lock while waiting for another could deadlock with a reader acquiring the same two + locks in the opposite order, and readers acquire their locks in an arbitrary order. + * A rewound action must skip the read lock of the key guarding its own outputs, which it already + holds the write lock of: the locks aren't reentrant, so an expanded action consuming the + outputs of another action from the same expansion would otherwise deadlock with itself. */ @Override @@ -199,7 +213,7 @@ private SilentCloseable enterActionPreparationForRewinding(Action action) * their prefetching state. */ private void prepareOutputsForRewinding(Action action) throws InterruptedException { - Cancellable task = outputUploadTasks.remove(action); + Cancellable task = outputUploadTasks.remove(actionKeyFor(action)); if (task != null) { task.cancel(); } @@ -207,12 +221,18 @@ private void prepareOutputsForRewinding(Action action) throws InterruptedExcepti } @Override - public SilentCloseable enterActionExecution(Action action, InputMetadataProvider metadataProvider) + public SilentCloseable enterActionExecution( + Action action, boolean wasRewound, InputMetadataProvider metadataProvider) throws InterruptedException { try (SilentCloseable c = Profiler.instance().profile(ProfilerTask.ACTION_LOCK, "action.enterActionExecution")) { return lockArtifactsForConsumption( - () -> action.getInputs().toList().iterator(), metadataProvider); + action.getInputs().toList(), + metadataProvider, + // A rewound action already holds the write lock on the key guarding its outputs and the + // locks aren't reentrant. Actions generated by an ActionTemplate can consume the outputs + // of other actions from the same expansion, which are guarded by the same key. + wasRewound ? outputKeyFor(action) : null); } } @@ -226,7 +246,8 @@ public SilentCloseable enterProcessOutputsAndGetLostArtifacts( try (SilentCloseable c = Profiler.instance() .profile(ProfilerTask.ACTION_LOCK, "action.enterProcessOutputsAndGetLostArtifacts")) { - return lockArtifactsForConsumption(importantOutputs, fullMetadataProvider); + return lockArtifactsForConsumption( + importantOutputs, fullMetadataProvider, /* writeLockedKey= */ null); } } @@ -238,7 +259,7 @@ public void registerOutputUploadTask(ActionExecutionMetadata action, Cancellable // We don't expect to have multiple output upload tasks for the same action registered at the // same time. outputUploadTasks.merge( - action, + actionKeyFor(action), task, (oldTask, newTask) -> { throw new IllegalStateException( @@ -248,7 +269,9 @@ public void registerOutputUploadTask(ActionExecutionMetadata action, Cancellable } private SilentCloseable lockArtifactsForConsumption( - Iterable artifacts, InputMetadataProvider metadataProvider) + Iterable artifacts, + InputMetadataProvider metadataProvider, + @Nullable ActionLookupData writeLockedKey) throws InterruptedException { var localCoarseLock = coarseLock; if (localCoarseLock != null) { @@ -271,7 +294,7 @@ private SilentCloseable lockArtifactsForConsumption( localCoarseLock.readLock().unlock(); } var allReadWriteLocks = - localFineLocks.getAll(inputKeysFor(artifacts, metadataProvider)).values(); + localFineLocks.getAll(inputKeysFor(artifacts, metadataProvider, writeLockedKey)).values(); var locksToUnlockBuilder = ImmutableList.builderWithExpectedSize(allReadWriteLocks.size()); try { @@ -291,7 +314,9 @@ private SilentCloseable lockArtifactsForConsumption( } private static Iterable inputKeysFor( - Iterable artifacts, InputMetadataProvider metadataProvider) { + Iterable artifacts, + InputMetadataProvider metadataProvider, + @Nullable ActionLookupData writeLockedKey) { var allArtifacts = Iterables.concat( artifacts, @@ -299,12 +324,44 @@ private static Iterable inputKeysFor( Iterables.transform( metadataProvider.getRunfilesTrees(), runfilesTree -> runfilesTree.getArtifacts().toList()))); - return Iterables.transform( - Iterables.filter(allArtifacts, artifact -> artifact instanceof DerivedArtifact), - artifact -> ((DerivedArtifact) artifact).getGeneratingActionKey()); + var result = + Iterables.transform( + Iterables.filter(allArtifacts, artifact -> artifact instanceof DerivedArtifact), + artifact -> lockKeyFor((DerivedArtifact) artifact)); + if (writeLockedKey == null) { + return result; + } + return Iterables.filter(result, key -> !key.equals(writeLockedKey)); } - private static ActionLookupData outputKeyFor(Action action) { + /** Returns the key that uniquely identifies the given action. */ + private static ActionLookupData actionKeyFor(ActionExecutionMetadata action) { return ((DerivedArtifact) action.getPrimaryOutput()).getGeneratingActionKey(); } + + /** + * Returns the key of the lock that guards the given artifact, which is the generating action key + * of the outermost tree artifact containing it, or its own if it isn't contained in one. + * + *

This is the artifact's own generating action key except for the outputs of an {@link + * com.google.devtools.build.lib.actions.ActionTemplate} expansion, which are guarded by the key + * of the template: they are only ever consumed as part of a tree artifact the template declares, + * either by actions outside the expansion, which depend on that tree artifact, or by other + * actions of the same expansion, which depend on individual files in it. + */ + private static ActionLookupData lockKeyFor(DerivedArtifact artifact) { + var outermost = artifact; + for (var parent = artifact.getParent(); parent != null; parent = parent.getParent()) { + outermost = parent; + } + return outermost.getGeneratingActionKey(); + } + + /** + * Returns the key of the lock that guards the outputs of the given action, which is the key its + * consumers acquire the read lock of. + */ + private static ActionLookupData outputKeyFor(Action action) { + return lockKeyFor((DerivedArtifact) action.getPrimaryOutput()); + } } diff --git a/src/main/java/com/google/devtools/build/lib/skyframe/SkyframeActionExecutor.java b/src/main/java/com/google/devtools/build/lib/skyframe/SkyframeActionExecutor.java index 956f51e6b9c3cd..ddc2b97d3b42ac 100644 --- a/src/main/java/com/google/devtools/build/lib/skyframe/SkyframeActionExecutor.java +++ b/src/main/java/com/google/devtools/build/lib/skyframe/SkyframeActionExecutor.java @@ -1141,8 +1141,9 @@ public ActionStepOrResult run(Environment env) } env.getListener().post(event); var rewoundActionSynchronizer = outputService.getRewoundActionSynchronizer(); + boolean wasRewound = wasRewound(action); try (SilentCloseable outerLock = - rewoundActionSynchronizer.enterActionPreparation(action, wasRewound(action))) { + rewoundActionSynchronizer.enterActionPreparation(action, wasRewound)) { if (actionFileSystemType().shouldDoEagerActionPrep()) { try (SilentCloseable d = Profiler.instance().profile(ProfilerTask.INFO, "action.prepare")) { @@ -1177,7 +1178,7 @@ public ActionStepOrResult run(Environment env) try (SilentCloseable innerLock = rewoundActionSynchronizer.enterActionExecution( - action, actionExecutionContext.getInputMetadataProvider())) { + action, wasRewound, actionExecutionContext.getInputMetadataProvider())) { return executeAction(env.getListener(), action); } } diff --git a/src/main/java/com/google/devtools/build/lib/vfs/OutputService.java b/src/main/java/com/google/devtools/build/lib/vfs/OutputService.java index 6b034956d90076..4b93b730e75e09 100644 --- a/src/main/java/com/google/devtools/build/lib/vfs/OutputService.java +++ b/src/main/java/com/google/devtools/build/lib/vfs/OutputService.java @@ -291,7 +291,8 @@ SilentCloseable enterActionPreparation(Action action, boolean wasRewound) throws InterruptedException; /** Guards an action from the beginning to the end of its {@link Action#execute execution}. */ - SilentCloseable enterActionExecution(Action action, InputMetadataProvider metadataProvider) + SilentCloseable enterActionExecution( + Action action, boolean wasRewound, InputMetadataProvider metadataProvider) throws InterruptedException; /** @@ -307,7 +308,7 @@ public SilentCloseable enterActionPreparation(Action action, boolean wasRewound) @Override public SilentCloseable enterActionExecution( - Action action, InputMetadataProvider metadataProvider) { + Action action, boolean wasRewound, InputMetadataProvider metadataProvider) { return () -> {}; } }; diff --git a/src/test/java/com/google/devtools/build/lib/skyframe/rewinding/BUILD b/src/test/java/com/google/devtools/build/lib/skyframe/rewinding/BUILD index a96f75249d4487..7296b03461aa2f 100644 --- a/src/test/java/com/google/devtools/build/lib/skyframe/rewinding/BUILD +++ b/src/test/java/com/google/devtools/build/lib/skyframe/rewinding/BUILD @@ -38,6 +38,7 @@ java_library( "//src/main/java/com/google/devtools/build/lib/skyframe:aspect_key_creator", "//src/main/java/com/google/devtools/build/lib/skyframe/rewinding", "//src/main/java/com/google/devtools/build/lib/skyframe/rewinding:lost_important_output_handler_module", + "//src/main/java/com/google/devtools/build/lib/util:os", "//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", diff --git a/src/test/java/com/google/devtools/build/lib/skyframe/rewinding/RewindingTest.java b/src/test/java/com/google/devtools/build/lib/skyframe/rewinding/RewindingTest.java index dc12784dafd741..7467ca06ce1218 100644 --- a/src/test/java/com/google/devtools/build/lib/skyframe/rewinding/RewindingTest.java +++ b/src/test/java/com/google/devtools/build/lib/skyframe/rewinding/RewindingTest.java @@ -266,6 +266,11 @@ public void treeArtifactRewound_oneFileLost() throws Exception { helper.runTreeArtifactRewound_oneFileLost_spawnFailed(); } + @Test + public void actionTemplateExpansionRewound_notConcurrentWithTreeConsumers() throws Exception { + helper.runActionTemplateExpansionRewound_notConcurrentWithTreeConsumers(); + } + @Test public void generatedRunfilesRewound_allFilesLost() throws Exception { helper.runGeneratedRunfilesRewound_allFilesLost_spawnFailed(); diff --git a/src/test/java/com/google/devtools/build/lib/skyframe/rewinding/RewindingTestsHelper.java b/src/test/java/com/google/devtools/build/lib/skyframe/rewinding/RewindingTestsHelper.java index 142d1bcfd17834..fb7be69562a599 100644 --- a/src/test/java/com/google/devtools/build/lib/skyframe/rewinding/RewindingTestsHelper.java +++ b/src/test/java/com/google/devtools/build/lib/skyframe/rewinding/RewindingTestsHelper.java @@ -23,6 +23,7 @@ import static com.google.devtools.build.lib.vfs.FileSystemUtils.readContentAsLatin1; import static com.google.devtools.build.lib.vfs.FileSystemUtils.writeContent; import static java.util.Arrays.stream; +import static java.util.stream.Collectors.joining; import static org.junit.Assert.assertThrows; import static org.junit.Assert.fail; import static org.mockito.ArgumentMatchers.any; @@ -33,6 +34,7 @@ import com.google.common.collect.ImmutableMultiset; import com.google.common.collect.ImmutableSetMultimap; import com.google.common.collect.Iterables; +import com.google.common.eventbus.AllowConcurrentEvents; import com.google.common.eventbus.Subscribe; import com.google.common.flogger.GoogleLogger; import com.google.common.util.concurrent.Uninterruptibles; @@ -49,6 +51,7 @@ import com.google.devtools.build.lib.actions.FilesetOutputSymlink; import com.google.devtools.build.lib.actions.LostInputsExecException; import com.google.devtools.build.lib.actions.Spawn; +import com.google.devtools.build.lib.actions.SpawnExecutedEvent; import com.google.devtools.build.lib.actions.SpawnResult; import com.google.devtools.build.lib.analysis.AspectCompleteEvent; import com.google.devtools.build.lib.analysis.TargetCompleteEvent; @@ -79,6 +82,7 @@ import com.google.devtools.build.lib.testutil.SpawnController.SpawnShim; import com.google.devtools.build.lib.testutil.SpawnInputUtils; import com.google.devtools.build.lib.testutil.TestConstants; +import com.google.devtools.build.lib.util.OS; import com.google.devtools.build.lib.vfs.Path; import com.google.devtools.build.lib.vfs.PathFragment; import com.google.devtools.build.skyframe.NodeEntry.DirtyType; @@ -1664,6 +1668,259 @@ final void runTreeArtifactRewoundWhenTreeFilesLost( assertArtifactKey(rewoundKeys.get(2), "tree/_pic_objs/consumes_tree/make_cc_dir"); } + /** + * The number of actions that consume the tree artifact populated by an action template expansion + * concurrently in {@link #runActionTemplateExpansionRewound_notConcurrentWithTreeConsumers}. + * + *

The more of them are reading it when the expansion is rewound, the more likely it is that a + * missing mutual exclusion is detected. + */ + private static final int TREE_CONSUMER_COUNT = 8; + + /** + * A tool that copies its second argument to its first. + * + *

On Windows, {@code cmd.exe} treats forward slashes as option prefixes, so the paths have to + * be translated before they are passed to {@code copy}. + */ + private static final String COPY_TOOL_SCRIPT = + OS.getCurrent() == OS.WINDOWS + ? """ + @echo off + set "OUT=%~1" + set "IN=%~2" + copy /Y "%IN:/=\\%" "%OUT:/=\\%" >NUL + """ + : """ + #!/bin/bash + cp "$2" "$1" + """; + + /** + * Verifies that an action generated by an {@link + * com.google.devtools.build.lib.actions.ActionTemplate} does not prepare for its re-execution + * while a consumer of the tree artifact it populates is running. + * + *

The outputs of such an action are {@link + * com.google.devtools.build.lib.actions.Artifact.TreeFileArtifact}s owned by the expansion, + * whereas its consumers depend on the parent tree artifact, which is owned by the template. A + * rewound expansion action must therefore synchronize on the template rather than on itself. + * + *

The rewound action also consumes the output of another action of the same expansion, which + * is a special case that only happens with expansions and must not result in deadlocks. + */ + public final void runActionTemplateExpansionRewound_notConcurrentWithTreeConsumers() throws Exception { + // All consumers and the action that reports the lost input have to run concurrently for the + // expansion to be rewound while the tree artifact is being read. + ensureMinimumJobs(TREE_CONSUMER_COUNT + 1); + testCase.addOptions("--experimental_allow_map_directory"); + testCase.write( + "foo/defs.bzl", + """ + def _copy_tool_impl(ctx): + tool = ctx.actions.declare_file(ctx.attr.name + ".bat") + ctx.actions.write(tool, r\"\"\"COPY_TOOL_SCRIPT\"\"\", is_executable = True) + return DefaultInfo(files = depset([tool]), executable = tool) + + copy_tool = rule(implementation = _copy_tool_impl, executable = True) + + def _map_impl(template_ctx, input_directories, output_directories, tools, **kwargs): + for child in input_directories["seed"].children: + # The two actions form a chain within the expansion, so the second one consumes a + # file of the very tree artifact it populates. + mid = template_ctx.declare_file( + child.basename + ".mid", + directory = output_directories["mapped"], + ) + args = template_ctx.args() + args.add_all([mid, child]) + template_ctx.run( + inputs = [child], + outputs = [mid], + executable = tools["copy_tool"], + arguments = [args], + progress_message = "Mapping foo/mapped_dir (1)", + ) + out = template_ctx.declare_file( + child.basename + ".out", + directory = output_directories["mapped"], + ) + args = template_ctx.args() + args.add_all([out, mid]) + template_ctx.run( + inputs = [mid], + outputs = [out], + executable = tools["copy_tool"], + arguments = [args], + progress_message = "Mapping foo/mapped_dir (2)", + ) + + def _mapped_tree_impl(ctx): + seed = ctx.actions.declare_directory("seed_dir") + ctx.actions.run_shell( + outputs = [seed], + command = "echo seed > $1/f1", + arguments = [seed.path], + progress_message = "Seeding foo/seed_dir", + ) + mapped = ctx.actions.declare_directory("mapped_dir") + ctx.actions.map_directory( + implementation = _map_impl, + input_directories = {"seed": seed}, + output_directories = {"mapped": mapped}, + tools = {"copy_tool": ctx.attr._copy_tool.files_to_run}, + # Ensure that the rewound expansion action re-executes its spawn instead of + # picking up the result of its first execution from the cache. + execution_requirements = {"no-cache": "1"}, + ) + return DefaultInfo(files = depset([mapped])) + + mapped_tree = rule( + implementation = _mapped_tree_impl, + attrs = { + "_copy_tool": attr.label( + default = ":copy_tool", + executable = True, + cfg = "exec", + ), + }, + ) + + def _consumer_impl(ctx): + out = ctx.actions.declare_file(ctx.attr.name + ".out") + ctx.actions.run_shell( + inputs = ctx.files.srcs, + outputs = [out], + command = "echo consumed > $1", + arguments = [out.path], + progress_message = "Consuming //foo:" + ctx.attr.name, + ) + return DefaultInfo(files = depset([out])) + + consumer = rule( + implementation = _consumer_impl, + attrs = {"srcs": attr.label_list(allow_files = True)}, + ) + """ + .replace("COPY_TOOL_SCRIPT", COPY_TOOL_SCRIPT)); + testCase.write( + "foo/BUILD", + """ + load(":defs.bzl", "consumer", "copy_tool", "mapped_tree") + + copy_tool(name = "copy_tool") + + mapped_tree(name = "mapped_tree") + + genrule( + name = "warmup_gen", + outs = ["warmup.out"], + cmd = "echo warmup > $@", + tags = ["no-cache"], + ) + + genrule( + name = "warmup_consumer", + srcs = ["warmup.out"], + outs = ["warmup_consumed.out"], + cmd = "cp $< $@", + ) + + consumer( + name = "losing_consumer", + srcs = [ + "warmup_consumed.out", + ":mapped_tree", + ], + ) + """ + + IntStream.range(0, TREE_CONSUMER_COUNT) + .mapToObj( + i -> + """ + consumer( + name = "consumer_%d", + srcs = [ + "warmup_consumed.out", + ":mapped_tree", + ], + ) + """ + .formatted(i)) + .collect(joining("\n"))); + + // The first rewound action of a build waits for all in-flight actions to finish before it + // prepares for its re-execution, which would mask the behavior under test. Rewind an unrelated + // action first; all consumers depend on its output and thus can't start any earlier. + addSpawnShim( + "Executing genrule //foo:warmup_consumer", + (spawn, context) -> createLostInputsExecException(spawn, context, "warmup.out")); + + // A consumer holds the read lock on the tree artifact from before its spawn shim runs until + // after its spawn has been executed, so the count is a lower bound on the number of consumers + // that are reading the tree artifact. + AtomicInteger consumersReadingTree = new AtomicInteger(); + AtomicInteger maxConsumersReadingDuringExpansion = new AtomicInteger(); + testCase.getRuntimeWrapper() + .registerSubscriber( + new Object() { + @Subscribe + @AllowConcurrentEvents + @SuppressWarnings("unused") + public void accept(SpawnExecutedEvent event) { + if (event.getActionMetadata().describe().startsWith("Consuming //foo:consumer_")) { + consumersReadingTree.decrementAndGet(); + } + } + }); + for (int i = 0; i < TREE_CONSUMER_COUNT; i++) { + addSpawnShim( + "Consuming //foo:consumer_" + i, + (spawn, context) -> { + consumersReadingTree.incrementAndGet(); + return ExecResult.delegate(); + }); + } + // The expansion actions are the only writers, so they must never observe a reader. Both of them + // run twice: once initially and once after the expansion has been rewound. + for (String expansionAction : + ImmutableList.of("Mapping foo/mapped_dir (1)", "Mapping foo/mapped_dir (2)")) { + for (int run = 0; run < 2; run++) { + addSpawnShim( + expansionAction, + (spawn, context) -> { + maxConsumersReadingDuringExpansion.accumulateAndGet( + consumersReadingTree.get(), Math::max); + return ExecResult.delegate(); + }); + } + } + // Fails without executing a spawn, so that the expansion is rewound while the other consumers + // are as likely as possible to still be reading the tree artifact. The consumers deliberately + // don't wait for each other: making them do so would deadlock whenever Skyframe can't run all + // of them concurrently. + addSpawnShim( + "Consuming //foo:losing_consumer", + (spawn, context) -> { + SpecialArtifact mappedTree = SpawnInputUtils.getTreeArtifactWithName(spawn, "mapped_dir"); + return createLostInputsExecException( + context, SpawnInputUtils.getExpandedToArtifact("f1.out", mappedTree, spawn, context)); + }); + + testCase.buildTarget("//foo:all"); + + verifyAllSpawnShimsConsumed(); + assertWithMessage( + "the rewound action template expansion prepared for its re-execution while consumers" + + " were still reading the tree artifact it populates") + .that(maxConsumersReadingDuringExpansion.get()) + .isEqualTo(0); + // Rewinding an expanded action re-expands the template, so both actions of the chain re-run. + var executedSpawns = ImmutableMultiset.copyOf(getExecutedSpawnDescriptions()); + assertThat(executedSpawns).hasCount("Mapping foo/mapped_dir (1)", 2); + assertThat(executedSpawns).hasCount("Mapping foo/mapped_dir (2)", 2); + } + public final void runGeneratedRunfilesRewound_allFilesLost_spawnFailed() throws Exception { // This test demonstrates that rewinding works when an action fails due to lost inputs which are // generated files in the action's runfiles. Rewinding must propagate across the runfiles tree @@ -3316,10 +3573,20 @@ static boolean isActionExecutionKey(Object key, Label label) { * CPU. */ private void ensureMultipleJobs() throws Exception { + ensureMinimumJobs(2); + } + + /** + * Ensures that the value of the {@code --jobs} flag is at least {@code minJobs}. + * + *

Note that the default value for {@code --jobs} is automatically calculated based on host + * CPU. + */ + private void ensureMinimumJobs(int minJobs) throws Exception { int autoJobs = new JobsConverter().convert("auto"); - if (autoJobs == 1) { - logger.atInfo().log("Setting --jobs=2 (was 1)"); - testCase.addOptions("--jobs=2"); + if (autoJobs < minJobs) { + logger.atInfo().log("Setting --jobs=%s (was %s)", minJobs, autoJobs); + testCase.addOptions("--jobs=" + minJobs); } else { logger.atInfo().log("Keeping default value of --jobs=%s", autoJobs); } From 5344e4b0d3ff31bcf50177539e43cd03e199d0dd Mon Sep 17 00:00:00 2001 From: Fabian Meumertzheim Date: Mon, 10 Aug 2026 13:21:46 +0200 Subject: [PATCH 3/3] Fix correctness issues in RemoteRewoundActionSynchronizer Fixes two issues in the synchronization of rewound actions with the consumers of their outputs: * Output upload tasks were only ever removed from `outputUploadTasks` when the corresponding action was rewound, so they accumulated (and retained the action as well as its spawn result) for the duration of the build. Since `registerOutputUploadTask` threw on a second registration for the same action, an action uploading its outputs twice would also have failed the build. Tasks are now unregistered on completion and multiple concurrent tasks per action are permitted. * Cancelling an output upload before the background task started running blocked forever: the body of a task submitted to an `ExecutorService` is skipped entirely if its future is cancelled before it starts, so neither the latch nor the completion callback (which closes the in-flight execution in `RemoteSpawnCache`) ever ran. The upload is now started via `execute` and cancelled by interrupting its thread, which always runs the body and thus the completion callback. --- .../lib/remote/RemoteExecutionService.java | 127 ++++++++++++++---- .../RemoteRewoundActionSynchronizer.java | 55 ++++++-- 2 files changed, 144 insertions(+), 38 deletions(-) diff --git a/src/main/java/com/google/devtools/build/lib/remote/RemoteExecutionService.java b/src/main/java/com/google/devtools/build/lib/remote/RemoteExecutionService.java index 1588e8dfafa97d..ac5711f7e72319 100644 --- a/src/main/java/com/google/devtools/build/lib/remote/RemoteExecutionService.java +++ b/src/main/java/com/google/devtools/build/lib/remote/RemoteExecutionService.java @@ -153,8 +153,10 @@ import java.util.concurrent.CancellationException; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; +import java.util.concurrent.Executor; import java.util.concurrent.Executors; import java.util.concurrent.Phaser; +import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.Semaphore; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; @@ -1809,36 +1811,111 @@ public void uploadOutputs( if (remoteOptions.getRemoteCacheAsync() && !action.getSpawn().getResourceOwner().mayModifySpawnOutputsAfterExecution()) { - var uploadDone = new CountDownLatch(1); - var future = - backgroundTaskExecutor.submit( - () -> { - try { - doUploadOutputs(action, spawnResult, onUploadComplete); - } catch (ExecException e) { - reportUploadError(e); - } catch (InterruptedException ignored) { - // ThreadPerTaskExecutor does not care about interrupt status. - } finally { - uploadDone.countDown(); - } - }); - - if (outputService instanceof RemoteOutputService remoteOutputService - && remoteOutputService.getRewoundActionSynchronizer() - instanceof RemoteRewoundActionSynchronizer remoteRewoundActionSynchronizer) { - remoteRewoundActionSynchronizer.registerOutputUploadTask( - action.getRemoteActionExecutionContext().getSpawnOwner(), - () -> { - future.cancel(true); - uploadDone.await(); - }); - } + new OutputUploadTask(action, spawnResult, onUploadComplete).start(backgroundTaskExecutor); } else { doUploadOutputs(action, spawnResult, onUploadComplete); } } + @Nullable + private RemoteRewoundActionSynchronizer getRewoundActionSynchronizer() { + if (outputService instanceof RemoteOutputService remoteOutputService + && remoteOutputService.getRewoundActionSynchronizer() + instanceof RemoteRewoundActionSynchronizer rewoundActionSynchronizer) { + return rewoundActionSynchronizer; + } + return null; + } + + /** + * A cancellable background upload of an action's outputs. + * + *

Registers itself with the {@link RemoteRewoundActionSynchronizer}, if there is one, before + * the upload starts and unregisters itself when it is done, so that a rewinding of the action + * waits for uploads that are still in flight. + * + *

Ensures that the completion callback runs exactly once and that {@link #cancel} only returns + * once the upload no longer accesses the action's outputs. + */ + private final class OutputUploadTask implements RemoteRewoundActionSynchronizer.Cancellable { + private final RemoteAction action; + private final SpawnResult spawnResult; + private final Runnable onUploadComplete; + private final ActionExecutionMetadata spawnOwner; + @Nullable private final RemoteRewoundActionSynchronizer rewoundActionSynchronizer; + private final CountDownLatch done = new CountDownLatch(1); + + // The thread running the upload, published by the upload itself before it reads cancelled. + @Nullable private volatile Thread thread; + // Set by cancel before it reads thread. + private volatile boolean cancelled; + + OutputUploadTask(RemoteAction action, SpawnResult spawnResult, Runnable onUploadComplete) { + this.action = action; + this.spawnResult = spawnResult; + this.onUploadComplete = onUploadComplete; + this.spawnOwner = action.getRemoteActionExecutionContext().getSpawnOwner(); + this.rewoundActionSynchronizer = getRewoundActionSynchronizer(); + } + + /** Registers the task for cancellation and starts the upload on the given executor. */ + void start(Executor executor) { + // Register before starting the upload so that it can't unregister itself before it has been + // registered. + if (rewoundActionSynchronizer != null) { + rewoundActionSynchronizer.registerOutputUploadTask(spawnOwner, this); + } + try { + // Runs the upload rather than submitting it: the body of a task submitted to an + // ExecutorService is skipped entirely if its future is cancelled before it starts, which + // would leave the completion callback unrun and cancel waiting forever. + executor.execute(this::upload); + } catch (RejectedExecutionException e) { + // The upload will never run, so complete the task on its behalf. + finish(); + throw e; + } + } + + private void upload() { + // Publish the thread before reading cancelled, which cancel writes before it reads the + // thread. At least one of the two thus observes the other and the upload is either skipped + // or interrupted. + thread = Thread.currentThread(); + try { + if (cancelled) { + onUploadComplete.run(); + } else { + doUploadOutputs(action, spawnResult, onUploadComplete); + } + } catch (ExecException e) { + reportUploadError(e); + } catch (InterruptedException ignored) { + // ThreadPerTaskExecutor does not care about interrupt status. + } finally { + finish(); + } + } + + @Override + public void cancel() throws InterruptedException { + cancelled = true; + var localThread = thread; + if (localThread != null) { + localThread.interrupt(); + } + done.await(); + } + + /** Signals that the upload no longer accesses the action's outputs. */ + private void finish() { + done.countDown(); + if (rewoundActionSynchronizer != null) { + rewoundActionSynchronizer.unregisterOutputUploadTask(spawnOwner, this); + } + } + } + private void doUploadOutputs( RemoteAction action, SpawnResult spawnResult, Runnable onUploadComplete) throws ExecException, InterruptedException { 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 b8789d63950e37..c59a4c3a8e046c 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 @@ -42,13 +42,22 @@ * while they are being read. */ final class RemoteRewoundActionSynchronizer implements RewoundActionSynchronizer { - /** A task with a cancellation callback. */ + /** A cancellable task that operates on the outputs of an action. */ public interface Cancellable { + /** + * Cancels the task and returns only once it no longer accesses the outputs of the action it has + * been registered for, which are about to be regenerated by a rewound execution of that action. + * + *

May be called after the task has completed, in which case it returns immediately. + */ void cancel() throws InterruptedException; } private final RemoteActionInputFetcher actionInputFetcher; - private final ConcurrentHashMap outputUploadTasks = + + // An action generally has at most one such task in flight, but nothing prevents an action from + // executing multiple spawns whose outputs are uploaded concurrently. + private final ConcurrentHashMap> outputUploadTasks = new ConcurrentHashMap<>(); // A single coarse lock is used to synchronize rewound actions (writers) and both rewound and @@ -213,9 +222,11 @@ private SilentCloseable enterActionPreparationForRewinding(Action action) * their prefetching state. */ private void prepareOutputsForRewinding(Action action) throws InterruptedException { - Cancellable task = outputUploadTasks.remove(actionKeyFor(action)); - if (task != null) { - task.cancel(); + ImmutableList tasks = outputUploadTasks.remove(actionKeyFor(action)); + if (tasks != null) { + for (Cancellable task : tasks) { + task.cancel(); + } } actionInputFetcher.handleRewoundActionOutputs(action.getOutputs()); } @@ -254,17 +265,35 @@ public SilentCloseable enterProcessOutputsAndGetLostArtifacts( /** * Registers a cancellation callback for an upload of action outputs that may still be running * after the action has completed. + * + *

Must be paired with a call to {@link #unregisterOutputUploadTask} once the upload has + * completed so that the task doesn't remain registered (and thus retained) for the rest of the + * build. */ public void registerOutputUploadTask(ActionExecutionMetadata action, Cancellable task) { - // We don't expect to have multiple output upload tasks for the same action registered at the - // same time. - outputUploadTasks.merge( + // The task is added to the list inside the mapping function so that registration is atomic + // with respect to the removal of the entry in prepareOutputsForRewinding. + outputUploadTasks.compute( + actionKeyFor(action), + (unusedKey, tasks) -> + tasks == null + ? ImmutableList.of(task) + : ImmutableList.builderWithExpectedSize(tasks.size() + 1) + .addAll(tasks) + .add(task) + .build()); + } + + /** Unregisters a task previously registered via {@link #registerOutputUploadTask}. */ + public void unregisterOutputUploadTask(ActionExecutionMetadata action, Cancellable task) { + outputUploadTasks.computeIfPresent( actionKeyFor(action), - task, - (oldTask, newTask) -> { - throw new IllegalStateException( - "Attempted to register multiple output upload tasks for %s: %s and %s" - .formatted(action, oldTask, newTask)); + (unusedKey, tasks) -> { + // Identity comparison: a task is only ever registered once, and a task registered by a + // re-execution of the action must not be unregistered by its predecessor. + var remainingTasks = + tasks.stream().filter(t -> t != task).collect(ImmutableList.toImmutableList()); + return remainingTasks.isEmpty() ? null : remainingTasks; }); }