Skip to content

Commit

Permalink
Retry on HTTP remote cache fetch failure
Browse files Browse the repository at this point in the history 
Bazel's previous behavior was to rebuild an artifact locally if fetching
it from an HTTP remote cache failed. This behavior is different from
GRPC remote cache case where Bazel will retry the fetch.

The lack of retry is an issue for multiple reasons: On one hand
rebuilding locally can be slower than fetching from the remote cache, on
the other hand if a build action is not bit reproducible, as is the case
with some compilers, then the local rebuild will trigger cache misses on
further build actions that depend on the current artifact.

This change aims to avoid theses issues by retrying the fetch in the
HTTP cache case similarly to how the GRPC cache client does it.

Some care needs to be taken due to the design of Bazel's internal remote
cache client API. For a fetch the client is given an `OutputStream`
object that it is expected to write the fetched data to. This may be a
temporary file on disk that will be moved to the final location after
the fetch completed. On retry, we need to be careful to not duplicate
previously written data when writing into this `OutputStream`. Due to
the generality of the `OutputStream` interface we cannot reset the file
handle or write pointer to start fresh. Instead, this change follows the
same pattern used in the GRPC cache client. Namely, keep track of the
data previously written and continue from that offset on retry.

With this change the HTTP cache client will attempt to fetch the data
from the remote cache via an HTTP range request. So that the server only
needs to send the data that is still missing. If the server replies with
a 206 Partial Content response, then we write the received data directly
into the output stream, if the server does not support range requests
and instead replies with the full data, then we drop the duplicate
prefix and only write into the output stream from the required offset.
  • Loading branch information
@aherrmann
aherrmann committed Feb 9, 2022
1 parent 6244713 commit aa96766
Show file tree
Hide file tree
Showing 7 changed files with 181 additions and 82 deletions.
17 changes: 11 additions & 6 deletions src/main/java/com/google/devtools/build/lib/remote/RemoteCacheClientFactory.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -58,15 +58,16 @@ public static RemoteCacheClient create(
@Nullable Credentials creds,
AuthAndTLSOptions authAndTlsOptions,
Path workingDirectory,
DigestUtil digestUtil)
DigestUtil digestUtil,
RemoteRetrier retrier)
throws IOException {
Preconditions.checkNotNull(workingDirectory, "workingDirectory");
if (isHttpCache(options) && isDiskCache(options)) {
return createDiskAndHttpCache(
workingDirectory, options.diskCache, options, creds, authAndTlsOptions, digestUtil);
workingDirectory, options.diskCache, options, creds, authAndTlsOptions, digestUtil, retrier);
}
if (isHttpCache(options)) {
return createHttp(options, creds, authAndTlsOptions, digestUtil);
return createHttp(options, creds, authAndTlsOptions, digestUtil, retrier);
}
if (isDiskCache(options)) {
return createDiskCache(
Expand All @@ -85,7 +86,8 @@ private static RemoteCacheClient createHttp(
RemoteOptions options,
Credentials creds,
AuthAndTLSOptions authAndTlsOptions,
DigestUtil digestUtil) {
DigestUtil digestUtil,
RemoteRetrier retrier) {
Preconditions.checkNotNull(options.remoteCache, "remoteCache");

try {
Expand All @@ -104,6 +106,7 @@ private static RemoteCacheClient createHttp(
options.remoteVerifyDownloads,
ImmutableList.copyOf(options.remoteHeaders),
digestUtil,
retrier,
creds,
authAndTlsOptions);
} else {
Expand All @@ -117,6 +120,7 @@ private static RemoteCacheClient createHttp(
options.remoteVerifyDownloads,
ImmutableList.copyOf(options.remoteHeaders),
digestUtil,
retrier,
creds,
authAndTlsOptions);
}
Expand Down Expand Up @@ -145,15 +149,16 @@ private static RemoteCacheClient createDiskAndHttpCache(
RemoteOptions options,
Credentials cred,
AuthAndTLSOptions authAndTlsOptions,
DigestUtil digestUtil)
DigestUtil digestUtil,
RemoteRetrier retrier)
throws IOException {
Path cacheDir =
workingDirectory.getRelative(Preconditions.checkNotNull(diskCachePath, "diskCachePath"));
if (!cacheDir.exists()) {
cacheDir.createDirectoryAndParents();
}

RemoteCacheClient httpCache = createHttp(options, cred, authAndTlsOptions, digestUtil);
RemoteCacheClient httpCache = createHttp(options, cred, authAndTlsOptions, digestUtil, retrier);
return createDiskAndRemoteClient(
workingDirectory,
diskCachePath,
Expand Down
24 changes: 23 additions & 1 deletion src/main/java/com/google/devtools/build/lib/remote/RemoteModule.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -64,9 +64,11 @@
import com.google.devtools.build.lib.exec.SpawnStrategyRegistry;
import com.google.devtools.build.lib.packages.TargetUtils;
import com.google.devtools.build.lib.remote.RemoteServerCapabilities.ServerCapabilitiesRequirement;
import com.google.devtools.build.lib.remote.common.CacheNotFoundException;
import com.google.devtools.build.lib.remote.common.RemoteCacheClient;
import com.google.devtools.build.lib.remote.common.RemoteExecutionClient;
import com.google.devtools.build.lib.remote.downloader.GrpcRemoteDownloader;
import com.google.devtools.build.lib.remote.http.HttpException;
import com.google.devtools.build.lib.remote.logging.LoggingInterceptor;
import com.google.devtools.build.lib.remote.options.RemoteOptions;
import com.google.devtools.build.lib.remote.options.RemoteOutputsMode;
Expand Down Expand Up @@ -103,6 +105,7 @@
import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.nio.channels.ClosedChannelException;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
Expand Down Expand Up @@ -233,7 +236,26 @@ private void initHttpAndDiskCache(
creds,
authAndTlsOptions,
Preconditions.checkNotNull(env.getWorkingDirectory(), "workingDirectory"),
digestUtil);
digestUtil,
new RemoteRetrier(
remoteOptions,
(e) -> {
boolean retry = false;
if (e instanceof ClosedChannelException) {
retry = true;
} else if (e instanceof HttpException) {
retry = true;
} else if (e instanceof IOException) {
String msg = e.getMessage().toLowerCase();
if (msg.contains("connection reset by peer")) {
retry = true;
}
}
return retry;
},
retryScheduler,
Retrier.ALLOW_ALL_CALLS)
);
} catch (IOException e) {
handleInitFailure(env, e, Code.CACHE_INIT_FAILURE);
return;
Expand Down
1 change: 1 addition & 0 deletions src/main/java/com/google/devtools/build/lib/remote/http/BUILD
View file
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@ java_library(
"//src/main/java/com/google/devtools/build/lib/analysis:blaze_version_info",
"//src/main/java/com/google/devtools/build/lib/authandtls",
"//src/main/java/com/google/devtools/build/lib/remote/common",
"//src/main/java/com/google/devtools/build/lib/remote:Retrier",
"//src/main/java/com/google/devtools/build/lib/remote/util",
"//src/main/java/com/google/devtools/build/lib/vfs",
"//third_party:auth",
Expand Down
10 changes: 9 additions & 1 deletion src/main/java/com/google/devtools/build/lib/remote/http/DownloadCommand.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -25,12 +25,18 @@ final class DownloadCommand {
private final boolean casDownload;
private final Digest digest;
private final OutputStream out;
private final long offset;

DownloadCommand(URI uri, boolean casDownload, Digest digest, OutputStream out) {
DownloadCommand(URI uri, boolean casDownload, Digest digest, OutputStream out, long offset) {
this.uri = Preconditions.checkNotNull(uri);
this.casDownload = casDownload;
this.digest = Preconditions.checkNotNull(digest);
this.out = Preconditions.checkNotNull(out);
this.offset = offset;
}

DownloadCommand(URI uri, boolean casDownload, Digest digest, OutputStream out) {
this(uri, casDownload, digest, out, 0);
}

public URI uri() {
Expand All @@ -48,4 +54,6 @@ public Digest digest() {
public OutputStream out() {
return out;
}

public long offset() { return offset; }
}
132 changes: 74 additions & 58 deletions src/main/java/com/google/devtools/build/lib/remote/http/HttpCacheClient.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@
import com.google.common.util.concurrent.MoreExecutors;
import com.google.common.util.concurrent.SettableFuture;
import com.google.devtools.build.lib.authandtls.AuthAndTLSOptions;
import com.google.devtools.build.lib.remote.RemoteRetrier;
import com.google.devtools.build.lib.remote.common.CacheNotFoundException;
import com.google.devtools.build.lib.remote.common.RemoteActionExecutionContext;
import com.google.devtools.build.lib.remote.common.RemoteCacheClient;
Expand Down Expand Up @@ -83,6 +84,7 @@
import java.util.NoSuchElementException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Function;
import java.util.regex.Pattern;
import javax.annotation.Nullable;
Expand Down Expand Up @@ -129,6 +131,7 @@ public final class HttpCacheClient implements RemoteCacheClient {
private final boolean useTls;
private final boolean verifyDownloads;
private final DigestUtil digestUtil;
private final RemoteRetrier retrier;

private final Object closeLock = new Object();

Expand All @@ -150,6 +153,7 @@ public static HttpCacheClient create(
boolean verifyDownloads,
ImmutableList<Entry<String, String>> extraHttpHeaders,
DigestUtil digestUtil,
RemoteRetrier retrier,
@Nullable final Credentials creds,
AuthAndTLSOptions authAndTlsOptions)
throws Exception {
Expand All @@ -162,6 +166,7 @@ public static HttpCacheClient create(
verifyDownloads,
extraHttpHeaders,
digestUtil,
retrier,
creds,
authAndTlsOptions,
null);
Expand All @@ -175,6 +180,7 @@ public static HttpCacheClient create(
boolean verifyDownloads,
ImmutableList<Entry<String, String>> extraHttpHeaders,
DigestUtil digestUtil,
RemoteRetrier retrier,
@Nullable final Credentials creds,
AuthAndTLSOptions authAndTlsOptions)
throws Exception {
Expand All @@ -189,6 +195,7 @@ public static HttpCacheClient create(
verifyDownloads,
extraHttpHeaders,
digestUtil,
retrier,
creds,
authAndTlsOptions,
domainSocketAddress);
Expand All @@ -202,6 +209,7 @@ public static HttpCacheClient create(
verifyDownloads,
extraHttpHeaders,
digestUtil,
retrier,
creds,
authAndTlsOptions,
domainSocketAddress);
Expand All @@ -219,6 +227,7 @@ private HttpCacheClient(
boolean verifyDownloads,
ImmutableList<Entry<String, String>> extraHttpHeaders,
DigestUtil digestUtil,
RemoteRetrier retrier,
@Nullable final Credentials creds,
AuthAndTLSOptions authAndTlsOptions,
@Nullable SocketAddress socketAddress)
Expand Down Expand Up @@ -284,6 +293,7 @@ public void channelCreated(Channel ch) {
this.extraHttpHeaders = extraHttpHeaders;
this.verifyDownloads = verifyDownloads;
this.digestUtil = digestUtil;
this.retrier = retrier;
}

@SuppressWarnings("FutureReturnValueIgnored")
Expand Down Expand Up @@ -460,6 +470,7 @@ public ListenableFuture<Void> downloadBlob(
@SuppressWarnings("FutureReturnValueIgnored")
private ListenableFuture<Void> get(Digest digest, final OutputStream out, boolean casDownload) {
final AtomicBoolean dataWritten = new AtomicBoolean();
AtomicLong bytesDownloaded = new AtomicLong();
OutputStream wrappedOut =
new OutputStream() {
// OutputStream.close() does nothing, which is what we want to ensure that the
Expand All @@ -469,12 +480,14 @@ private ListenableFuture<Void> get(Digest digest, final OutputStream out, boolea
@Override
public void write(byte[] b, int offset, int length) throws IOException {
dataWritten.set(true);
bytesDownloaded.addAndGet(length);
out.write(b, offset, length);
}

@Override
public void write(int b) throws IOException {
dataWritten.set(true);
bytesDownloaded.incrementAndGet();
out.write(b);
}

Expand All @@ -483,57 +496,59 @@ public void flush() throws IOException {
out.flush();
}
};
DownloadCommand downloadCmd = new DownloadCommand(uri, casDownload, digest, wrappedOut);
SettableFuture<Void> outerF = SettableFuture.create();
acquireDownloadChannel()
.addListener(
(Future<Channel> channelPromise) -> {
if (!channelPromise.isSuccess()) {
outerF.setException(channelPromise.cause());
return;
}

Channel ch = channelPromise.getNow();
ch.writeAndFlush(downloadCmd)
.addListener(
(f) -> {
try {
if (f.isSuccess()) {
outerF.set(null);
} else {
Throwable cause = f.cause();
// cause can be of type HttpException, because Netty uses
// Unsafe.throwException to
// re-throw a checked exception that hasn't been declared in the method
// signature.
if (cause instanceof HttpException) {
HttpResponse response = ((HttpException) cause).response();
if (!dataWritten.get() && authTokenExpired(response)) {
// The error is due to an auth token having expired. Let's try
// again.
try {
refreshCredentials();
getAfterCredentialRefresh(downloadCmd, outerF);
return;
} catch (IOException e) {
cause.addSuppressed(e);
} catch (RuntimeException e) {
logger.atWarning().withCause(e).log("Unexpected exception");
cause.addSuppressed(e);
}
} else if (cacheMiss(response.status())) {
outerF.setException(new CacheNotFoundException(digest));
return;
}
}
outerF.setException(cause);
}
} finally {
releaseDownloadChannel(ch);
return retrier.executeAsync(() -> {
DownloadCommand downloadCmd = new DownloadCommand(uri, casDownload, digest, wrappedOut, bytesDownloaded.get());
SettableFuture<Void> outerF = SettableFuture.create();
acquireDownloadChannel()
.addListener(
(Future<Channel> channelPromise) -> {
if (!channelPromise.isSuccess()) {
outerF.setException(channelPromise.cause());
return;
}

Channel ch = channelPromise.getNow();
ch.writeAndFlush(downloadCmd)
.addListener(
(f) -> {
try {
if (f.isSuccess()) {
outerF.set(null);
} else {
Throwable cause = f.cause();
// cause can be of type HttpException, because Netty uses
// Unsafe.throwException to
// re-throw a checked exception that hasn't been declared in the method
// signature.
if (cause instanceof HttpException) {
HttpResponse response = ((HttpException) cause).response();
if (!dataWritten.get() && authTokenExpired(response)) {
// The error is due to an auth token having expired. Let's try
// again.
try {
refreshCredentials();
getAfterCredentialRefresh(downloadCmd, outerF);
return;
} catch (IOException e) {
cause.addSuppressed(e);
} catch (RuntimeException e) {
logger.atWarning().withCause(e).log("Unexpected exception");
cause.addSuppressed(e);
}
} else if (cacheMiss(response.status())) {
outerF.setException(new CacheNotFoundException(digest));
return;
}
}
outerF.setException(cause);
}
} finally {
releaseDownloadChannel(ch);
}
});
});
});
return outerF;
return outerF;
});
}

@SuppressWarnings("FutureReturnValueIgnored")
Expand Down Expand Up @@ -670,20 +685,21 @@ private void uploadAfterCredentialRefresh(UploadCommand upload, SettableFuture<V
@Override
public ListenableFuture<Void> uploadFile(
RemoteActionExecutionContext context, Digest digest, Path file) {
try {
return uploadAsync(
digest.getHash(), digest.getSizeBytes(), file.getInputStream(), /* casUpload= */ true);
} catch (IOException e) {
// Can be thrown from file.getInputStream.
return Futures.immediateFailedFuture(e);
}
return retrier.executeAsync(() -> {
try {
return uploadAsync(digest.getHash(), digest.getSizeBytes(), file.getInputStream(), /* casUpload= */ true);
} catch (IOException e) {
// Can be thrown from file.getInputStream.
return Futures.immediateFailedFuture(e);
}
});
}

@Override
public ListenableFuture<Void> uploadBlob(
RemoteActionExecutionContext context, Digest digest, ByteString data) {
return uploadAsync(
digest.getHash(), digest.getSizeBytes(), data.newInput(), /* casUpload= */ true);
return retrier.executeAsync(() -> uploadAsync(
digest.getHash(), digest.getSizeBytes(), data.newInput(), /* casUpload= */ true));
}

@Override
Expand Down
Loading

0 comments on commit aa96766

Please sign in to comment.