Skip to content

Commit a7c65af

Browse files
iancha1992coeuvre
iancha1992
and
coeuvre
authored
[7.1.2] Don't upload remote input to remote cache (#21941)
and when it's missing, treat it as remote cache eviction. Also revert the workaround for #19513. Fixes #21777. Potential fix for #21626 and #21778. Closes #21825. PiperOrigin-RevId: 619877088 Change-Id: Ib1204de8440b780e5a6ee6a563a87da08f196ca5 Commit eda0fe4 Co-authored-by: Chi Wang <[email protected]>
1 parent 33a2ce6 commit a7c65af

19 files changed

+240
-61
lines changed

‎src/main/java/com/google/devtools/build/lib/remote/AbstractActionInputPrefetcher.java

Lines changed: 2 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -46,6 +46,7 @@
4646
import com.google.devtools.build.lib.profiler.Profiler;
4747
import com.google.devtools.build.lib.profiler.ProfilerTask;
4848
import com.google.devtools.build.lib.remote.common.CacheNotFoundException;
49+
import com.google.devtools.build.lib.remote.common.LostInputsEvent;
4950
import com.google.devtools.build.lib.remote.util.AsyncTaskCache;
5051
import com.google.devtools.build.lib.remote.util.TempPathGenerator;
5152
import com.google.devtools.build.lib.vfs.FileSymlinkLoopException;
@@ -79,8 +80,6 @@ public abstract class AbstractActionInputPrefetcher implements ActionInputPrefet
7980
protected final Path execRoot;
8081
protected final RemoteOutputChecker remoteOutputChecker;
8182

82-
private final Set<ActionInput> missingActionInputs = Sets.newConcurrentHashSet();
83-
8483
private final ActionOutputDirectoryHelper outputDirectoryHelper;
8584

8685
/** The state of a directory tracked by {@link DirectoryTracker}, as explained below. */
@@ -538,7 +537,7 @@ private Completable downloadFileNoCheckRx(
538537
.doOnError(
539538
error -> {
540539
if (error instanceof CacheNotFoundException) {
541-
missingActionInputs.add(actionInput);
540+
reporter.post(new LostInputsEvent());
542541
}
543542
}));
544543

@@ -700,10 +699,6 @@ public void flushOutputTree() throws InterruptedException {
700699
downloadCache.awaitInProgressTasks();
701700
}
702701

703-
public ImmutableSet<ActionInput> getMissingActionInputs() {
704-
return ImmutableSet.copyOf(missingActionInputs);
705-
}
706-
707702
public RemoteOutputChecker getRemoteOutputChecker() {
708703
return remoteOutputChecker;
709704
}

‎src/main/java/com/google/devtools/build/lib/remote/BUILD

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -96,6 +96,7 @@ java_library(
9696
"//src/main/java/com/google/devtools/build/lib/remote/common",
9797
"//src/main/java/com/google/devtools/build/lib/remote/common:bulk_transfer_exception",
9898
"//src/main/java/com/google/devtools/build/lib/remote/common:cache_not_found_exception",
99+
"//src/main/java/com/google/devtools/build/lib/remote/common:lost_inputs_event",
99100
"//src/main/java/com/google/devtools/build/lib/remote/disk",
100101
"//src/main/java/com/google/devtools/build/lib/remote/downloader",
101102
"//src/main/java/com/google/devtools/build/lib/remote/grpc",
@@ -229,6 +230,7 @@ java_library(
229230
"//src/main/java/com/google/devtools/build/lib/events",
230231
"//src/main/java/com/google/devtools/build/lib/profiler",
231232
"//src/main/java/com/google/devtools/build/lib/remote/common:cache_not_found_exception",
233+
"//src/main/java/com/google/devtools/build/lib/remote/common:lost_inputs_event",
232234
"//src/main/java/com/google/devtools/build/lib/remote/util",
233235
"//src/main/java/com/google/devtools/build/lib/vfs",
234236
"//src/main/java/com/google/devtools/build/lib/vfs:pathfragment",
@@ -245,12 +247,13 @@ java_library(
245247
srcs = ["LeaseService.java"],
246248
deps = [
247249
"//src/main/java/com/google/devtools/build/lib/actions",
248-
"//src/main/java/com/google/devtools/build/lib/actions:artifacts",
249250
"//src/main/java/com/google/devtools/build/lib/actions:file_metadata",
251+
"//src/main/java/com/google/devtools/build/lib/remote/common:lost_inputs_event",
250252
"//src/main/java/com/google/devtools/build/lib/skyframe:action_execution_value",
251253
"//src/main/java/com/google/devtools/build/lib/skyframe:sky_functions",
252254
"//src/main/java/com/google/devtools/build/lib/skyframe:tree_artifact_value",
253255
"//src/main/java/com/google/devtools/build/skyframe",
256+
"//third_party:guava",
254257
"//third_party:jsr305",
255258
],
256259
)

‎src/main/java/com/google/devtools/build/lib/remote/LeaseService.java

Lines changed: 12 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -13,14 +13,15 @@
1313
// limitations under the License.
1414
package com.google.devtools.build.lib.remote;
1515

16-
import com.google.devtools.build.lib.actions.ActionInput;
16+
import com.google.common.eventbus.AllowConcurrentEvents;
17+
import com.google.common.eventbus.Subscribe;
1718
import com.google.devtools.build.lib.actions.FileArtifactValue;
1819
import com.google.devtools.build.lib.actions.cache.ActionCache;
20+
import com.google.devtools.build.lib.remote.common.LostInputsEvent;
1921
import com.google.devtools.build.lib.skyframe.ActionExecutionValue;
2022
import com.google.devtools.build.lib.skyframe.SkyFunctions;
2123
import com.google.devtools.build.lib.skyframe.TreeArtifactValue;
2224
import com.google.devtools.build.skyframe.MemoizingEvaluator;
23-
import java.util.Set;
2425
import java.util.concurrent.atomic.AtomicBoolean;
2526
import javax.annotation.Nullable;
2627

@@ -30,6 +31,7 @@ public class LeaseService {
3031
@Nullable private final ActionCache actionCache;
3132
private final AtomicBoolean leaseExtensionStarted = new AtomicBoolean(false);
3233
@Nullable LeaseExtension leaseExtension;
34+
private final AtomicBoolean hasMissingActionInputs = new AtomicBoolean(false);
3335

3436
public LeaseService(
3537
MemoizingEvaluator memoizingEvaluator,
@@ -48,12 +50,18 @@ public void finalizeAction() {
4850
}
4951
}
5052

51-
public void finalizeExecution(Set<ActionInput> missingActionInputs) {
53+
@AllowConcurrentEvents
54+
@Subscribe
55+
public void onLostInputs(LostInputsEvent event) {
56+
hasMissingActionInputs.set(true);
57+
}
58+
59+
public void finalizeExecution() {
5260
if (leaseExtension != null) {
5361
leaseExtension.stop();
5462
}
5563

56-
if (!missingActionInputs.isEmpty()) {
64+
if (hasMissingActionInputs.getAndSet(false)) {
5765
handleMissingInputs();
5866
}
5967
}

‎src/main/java/com/google/devtools/build/lib/remote/RemoteExecutionCache.java

Lines changed: 78 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,7 @@
1313
// limitations under the License.
1414
package com.google.devtools.build.lib.remote;
1515

16+
import static com.google.common.base.Preconditions.checkNotNull;
1617
import static com.google.common.collect.ImmutableList.toImmutableList;
1718
import static com.google.common.util.concurrent.Futures.immediateFailedFuture;
1819
import static com.google.common.util.concurrent.Futures.immediateFuture;
@@ -22,24 +23,29 @@
2223
import static com.google.devtools.build.lib.remote.util.RxUtils.mergeBulkTransfer;
2324
import static com.google.devtools.build.lib.remote.util.RxUtils.toTransferResult;
2425
import static java.lang.String.format;
25-
import static java.util.concurrent.TimeUnit.SECONDS;
2626

2727
import build.bazel.remote.execution.v2.Digest;
2828
import build.bazel.remote.execution.v2.Directory;
29+
import com.google.common.annotations.VisibleForTesting;
2930
import com.google.common.base.Throwables;
3031
import com.google.common.collect.ImmutableList;
3132
import com.google.common.collect.ImmutableSet;
3233
import com.google.common.collect.Iterables;
34+
import com.google.common.flogger.GoogleLogger;
3335
import com.google.common.util.concurrent.ListenableFuture;
36+
import com.google.devtools.build.lib.events.Reporter;
3437
import com.google.devtools.build.lib.profiler.Profiler;
3538
import com.google.devtools.build.lib.profiler.SilentCloseable;
39+
import com.google.devtools.build.lib.remote.common.CacheNotFoundException;
40+
import com.google.devtools.build.lib.remote.common.LostInputsEvent;
3641
import com.google.devtools.build.lib.remote.common.RemoteActionExecutionContext;
3742
import com.google.devtools.build.lib.remote.common.RemoteCacheClient;
3843
import com.google.devtools.build.lib.remote.merkletree.MerkleTree;
3944
import com.google.devtools.build.lib.remote.merkletree.MerkleTree.PathOrBytes;
4045
import com.google.devtools.build.lib.remote.options.RemoteOptions;
4146
import com.google.devtools.build.lib.remote.util.DigestUtil;
4247
import com.google.devtools.build.lib.remote.util.RxUtils.TransferResult;
48+
import com.google.devtools.build.lib.vfs.Path;
4349
import com.google.protobuf.Message;
4450
import io.reactivex.rxjava3.annotations.NonNull;
4551
import io.reactivex.rxjava3.core.Completable;
@@ -59,13 +65,50 @@
5965
/** A {@link RemoteCache} with additional functionality needed for remote execution. */
6066
public class RemoteExecutionCache extends RemoteCache {
6167

68+
private static final GoogleLogger logger = GoogleLogger.forEnclosingClass();
69+
70+
/**
71+
* An interface used to check whether a given {@link Path} is stored in a remote or a disk cache.
72+
*/
73+
public interface RemotePathChecker {
74+
boolean isRemote(RemoteActionExecutionContext context, Path path) throws IOException;
75+
}
76+
77+
private RemotePathChecker remotePathChecker =
78+
new RemotePathChecker() {
79+
@Override
80+
public boolean isRemote(RemoteActionExecutionContext context, Path path)
81+
throws IOException {
82+
var fs = path.getFileSystem();
83+
if (fs instanceof RemoteActionFileSystem) {
84+
var remoteActionFileSystem = (RemoteActionFileSystem) fs;
85+
if (remoteActionFileSystem.isRemote(path)) {
86+
if (context.getReadCachePolicy().allowDiskCache()) {
87+
try (var inputStream = path.getInputStream()) {
88+
// If the file exists in the disk cache, download it and continue the upload.
89+
return false;
90+
} catch (IOException e) {
91+
logger.atWarning().withCause(e).log(
92+
"Failed to get input stream for %s", path.getPathString());
93+
}
94+
}
95+
return true;
96+
}
97+
}
98+
return false;
99+
}
100+
};
101+
62102
public RemoteExecutionCache(
63-
RemoteCacheClient protocolImpl,
64-
RemoteOptions options,
65-
DigestUtil digestUtil) {
103+
RemoteCacheClient protocolImpl, RemoteOptions options, DigestUtil digestUtil) {
66104
super(protocolImpl, options, digestUtil);
67105
}
68106

107+
@VisibleForTesting
108+
void setRemotePathChecker(RemotePathChecker remotePathChecker) {
109+
this.remotePathChecker = remotePathChecker;
110+
}
111+
69112
/**
70113
* Ensures that the tree structure of the inputs, the input files themselves, and the command are
71114
* available in the remote cache, such that the tree can be reassembled and executed on another
@@ -82,7 +125,8 @@ public void ensureInputsPresent(
82125
RemoteActionExecutionContext context,
83126
MerkleTree merkleTree,
84127
Map<Digest, Message> additionalInputs,
85-
boolean force)
128+
boolean force,
129+
Reporter reporter)
86130
throws IOException, InterruptedException {
87131
Iterable<Digest> merkleTreeAllDigests;
88132
try (SilentCloseable s = Profiler.instance().profile("merkleTree.getAllDigests()")) {
@@ -95,7 +139,7 @@ public void ensureInputsPresent(
95139
}
96140

97141
Flowable<TransferResult> uploads =
98-
createUploadTasks(context, merkleTree, additionalInputs, allDigests, force)
142+
createUploadTasks(context, merkleTree, additionalInputs, allDigests, force, reporter)
99143
.flatMapPublisher(
100144
result ->
101145
Flowable.using(
@@ -113,10 +157,7 @@ public void ensureInputsPresent(
113157
}));
114158

115159
try {
116-
// Workaround for https://github.com/bazelbuild/bazel/issues/19513.
117-
if (!mergeBulkTransfer(uploads).blockingAwait(options.remoteTimeout.getSeconds(), SECONDS)) {
118-
throw new IOException("Timed out when waiting for uploads");
119-
}
160+
mergeBulkTransfer(uploads).blockingAwait();
120161
} catch (RuntimeException e) {
121162
Throwable cause = e.getCause();
122163
if (cause != null) {
@@ -131,7 +172,8 @@ private ListenableFuture<Void> uploadBlob(
131172
RemoteActionExecutionContext context,
132173
Digest digest,
133174
MerkleTree merkleTree,
134-
Map<Digest, Message> additionalInputs) {
175+
Map<Digest, Message> additionalInputs,
176+
Reporter reporter) {
135177
Directory node = merkleTree.getDirectoryByDigest(digest);
136178
if (node != null) {
137179
return cacheProtocol.uploadBlob(context, digest, node.toByteString());
@@ -142,7 +184,20 @@ private ListenableFuture<Void> uploadBlob(
142184
if (file.getBytes() != null) {
143185
return cacheProtocol.uploadBlob(context, digest, file.getBytes());
144186
}
145-
return cacheProtocol.uploadFile(context, digest, file.getPath());
187+
188+
var path = checkNotNull(file.getPath());
189+
try {
190+
if (remotePathChecker.isRemote(context, path)) {
191+
// If we get here, the remote input was determined to exist in the remote or disk cache at
192+
// some point before action execution, but reported to be missing when querying the remote
193+
// for missing action inputs; possibly because it was evicted in the interim.
194+
reporter.post(new LostInputsEvent());
195+
throw new CacheNotFoundException(digest, path.getPathString());
196+
}
197+
} catch (IOException e) {
198+
return immediateFailedFuture(e);
199+
}
200+
return cacheProtocol.uploadFile(context, digest, path);
146201
}
147202

148203
Message message = additionalInputs.get(digest);
@@ -169,14 +224,16 @@ private Single<List<UploadTask>> createUploadTasks(
169224
MerkleTree merkleTree,
170225
Map<Digest, Message> additionalInputs,
171226
Iterable<Digest> allDigests,
172-
boolean force) {
227+
boolean force,
228+
Reporter reporter) {
173229
return Single.using(
174230
() -> Profiler.instance().profile("collect digests"),
175231
ignored ->
176232
Flowable.fromIterable(allDigests)
177233
.flatMapMaybe(
178234
digest ->
179-
maybeCreateUploadTask(context, merkleTree, additionalInputs, digest, force))
235+
maybeCreateUploadTask(
236+
context, merkleTree, additionalInputs, digest, force, reporter))
180237
.collect(toImmutableList()),
181238
SilentCloseable::close);
182239
}
@@ -186,7 +243,8 @@ private Maybe<UploadTask> maybeCreateUploadTask(
186243
MerkleTree merkleTree,
187244
Map<Digest, Message> additionalInputs,
188245
Digest digest,
189-
boolean force) {
246+
boolean force,
247+
Reporter reporter) {
190248
return Maybe.create(
191249
emitter -> {
192250
AsyncSubject<Void> completion = AsyncSubject.create();
@@ -211,7 +269,11 @@ private Maybe<UploadTask> maybeCreateUploadTask(
211269
return toCompletable(
212270
() ->
213271
uploadBlob(
214-
context, uploadTask.digest, merkleTree, additionalInputs),
272+
context,
273+
uploadTask.digest,
274+
merkleTree,
275+
additionalInputs,
276+
reporter),
215277
directExecutor());
216278
}),
217279
/* onAlreadyRunning= */ () -> emitter.onSuccess(uploadTask),

‎src/main/java/com/google/devtools/build/lib/remote/RemoteExecutionService.java

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1477,7 +1477,8 @@ public void uploadInputsIfNotPresent(RemoteAction action, boolean force)
14771477
.withWriteCachePolicy(CachePolicy.REMOTE_CACHE_ONLY), // Only upload to remote cache
14781478
merkleTree,
14791479
additionalInputs,
1480-
force);
1480+
force,
1481+
reporter);
14811482
} finally {
14821483
maybeReleaseRemoteActionBuildingSemaphore();
14831484
}

‎src/main/java/com/google/devtools/build/lib/remote/RemoteModule.java

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -579,7 +579,8 @@ public void beforeCommand(CommandEnvironment env) throws AbruptExitException {
579579
buildRequestId,
580580
invocationId,
581581
remoteOptions.remoteInstanceName,
582-
remoteOptions.remoteAcceptCached));
582+
remoteOptions.remoteAcceptCached,
583+
env.getReporter()));
583584
} else {
584585
if (enableDiskCache) {
585586
try {
@@ -990,6 +991,7 @@ public void executorInit(CommandEnvironment env, BuildRequest request, ExecutorB
990991
env.getSkyframeExecutor().getEvaluator(),
991992
env.getBlazeWorkspace().getPersistentActionCache(),
992993
leaseExtension);
994+
env.getEventBus().register(leaseService);
993995

994996
remoteOutputService.setRemoteOutputChecker(remoteOutputChecker);
995997
remoteOutputService.setActionInputFetcher(actionInputFetcher);

‎src/main/java/com/google/devtools/build/lib/remote/RemoteOutputService.java

Lines changed: 1 addition & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -19,11 +19,9 @@
1919
import com.google.common.collect.ImmutableCollection;
2020
import com.google.common.collect.ImmutableList;
2121
import com.google.common.collect.ImmutableMap;
22-
import com.google.common.collect.ImmutableSet;
2322
import com.google.common.eventbus.Subscribe;
2423
import com.google.devtools.build.lib.actions.Action;
2524
import com.google.devtools.build.lib.actions.ActionExecutionMetadata;
26-
import com.google.devtools.build.lib.actions.ActionInput;
2725
import com.google.devtools.build.lib.actions.ActionInputMap;
2826
import com.google.devtools.build.lib.actions.Artifact;
2927
import com.google.devtools.build.lib.actions.ArtifactPathResolver;
@@ -168,11 +166,7 @@ public void finalizeBuild(boolean buildSuccessful) {
168166
@Subscribe
169167
public void onExecutionPhaseCompleteEvent(ExecutionPhaseCompleteEvent event) {
170168
if (leaseService != null) {
171-
var missingActionInputs = ImmutableSet.<ActionInput>of();
172-
if (actionInputFetcher != null) {
173-
missingActionInputs = actionInputFetcher.getMissingActionInputs();
174-
}
175-
leaseService.finalizeExecution(missingActionInputs);
169+
leaseService.finalizeExecution();
176170
}
177171
}
178172

0 commit comments

Comments
 (0)