Skip to content

Commit

Permalink
Merge branch 'release-6.4.0' into 19312-incompatible_merge_fixed_and_…
Browse files Browse the repository at this point in the history 
…default_shell_env
  • Loading branch information
@meteorcloudy
meteorcloudy authored Aug 29, 2023
2 parents a788f72 + f5b0fd5 commit 463f83a
Show file tree
Hide file tree
Showing 11 changed files with 451 additions and 45 deletions.
23 changes: 17 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 @@ -59,15 +59,22 @@ 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 @@ -90,7 +97,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 @@ -109,6 +117,7 @@ private static RemoteCacheClient createHttp(
options.remoteVerifyDownloads,
ImmutableList.copyOf(options.remoteHeaders),
digestUtil,
retrier,
creds,
authAndTlsOptions);
} else {
Expand All @@ -122,6 +131,7 @@ private static RemoteCacheClient createHttp(
options.remoteVerifyDownloads,
ImmutableList.copyOf(options.remoteHeaders),
digestUtil,
retrier,
creds,
authAndTlsOptions);
}
Expand All @@ -148,9 +158,10 @@ private static RemoteCacheClient createDiskAndHttpCache(
RemoteOptions options,
Credentials cred,
AuthAndTLSOptions authAndTlsOptions,
DigestUtil digestUtil)
DigestUtil digestUtil,
RemoteRetrier retrier)
throws IOException {
RemoteCacheClient httpCache = createHttp(options, cred, authAndTlsOptions, digestUtil);
RemoteCacheClient httpCache = createHttp(options, cred, authAndTlsOptions, digestUtil, retrier);
return createDiskAndRemoteClient(
workingDirectory,
diskCachePath,
Expand Down
39 changes: 38 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 @@ -66,6 +66,7 @@
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.RemoteBuildEventUploadMode;
import com.google.devtools.build.lib.remote.options.RemoteOptions;
Expand Down Expand Up @@ -105,17 +106,21 @@
import io.grpc.Channel;
import io.grpc.ClientInterceptor;
import io.grpc.ManagedChannel;
import io.netty.handler.codec.DecoderException;
import io.netty.handler.codec.http.HttpResponseStatus;
import io.reactivex.rxjava3.plugins.RxJavaPlugins;
import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.nio.channels.ClosedChannelException;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.function.Predicate;
import java.util.regex.Pattern;
import javax.annotation.Nullable;

Expand Down Expand Up @@ -221,6 +226,36 @@ private static ServerCapabilities getAndVerifyServerCapabilities(
return capabilities;
}

public static final Predicate<? super Exception> RETRIABLE_HTTP_ERRORS =
e -> {
boolean retry = false;
if (e instanceof ClosedChannelException) {
retry = true;
} else if (e instanceof HttpException) {
int status = ((HttpException) e).response().status().code();
retry =
status == HttpResponseStatus.INTERNAL_SERVER_ERROR.code()
|| status == HttpResponseStatus.BAD_GATEWAY.code()
|| status == HttpResponseStatus.SERVICE_UNAVAILABLE.code()
|| status == HttpResponseStatus.GATEWAY_TIMEOUT.code();
} else if (e instanceof IOException) {
String msg = e.getMessage().toLowerCase();
if (msg.contains("connection reset by peer")) {
retry = true;
} else if (msg.contains("operation timed out")) {
retry = true;
}
} else {
// Workaround for a netty bug: https://github.com/netty/netty/issues/11815. Remove this
// once it is fixed in the upstream.
if (e instanceof DecoderException
&& e.getMessage().endsWith("functions:OPENSSL_internal:BAD_DECRYPT")) {
retry = true;
}
}
return retry;
};

private void initHttpAndDiskCache(
CommandEnvironment env,
Credentials credentials,
Expand All @@ -235,7 +270,9 @@ private void initHttpAndDiskCache(
credentials,
authAndTlsOptions,
Preconditions.checkNotNull(env.getWorkingDirectory(), "workingDirectory"),
digestUtil);
digestUtil,
new RemoteRetrier(
remoteOptions, RETRIABLE_HTTP_ERRORS, 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 @@ -20,6 +20,7 @@ java_library(
deps = [
"//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:Retrier",
"//src/main/java/com/google/devtools/build/lib/remote/common",
"//src/main/java/com/google/devtools/build/lib/remote/common:cache_not_found_exception",
"//src/main/java/com/google/devtools/build/lib/remote/util",
Expand Down
12 changes: 11 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,8 @@ public Digest digest() {
public OutputStream out() {
return out;
}

public long offset() {
return offset;
}
}
65 changes: 51 additions & 14 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 @@ -81,8 +82,10 @@
import java.util.List;
import java.util.Map.Entry;
import java.util.NoSuchElementException;
import java.util.Optional;
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 +132,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 +154,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 +167,7 @@ public static HttpCacheClient create(
verifyDownloads,
extraHttpHeaders,
digestUtil,
retrier,
creds,
authAndTlsOptions,
null);
Expand All @@ -175,6 +181,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 +196,7 @@ public static HttpCacheClient create(
verifyDownloads,
extraHttpHeaders,
digestUtil,
retrier,
creds,
authAndTlsOptions,
domainSocketAddress);
Expand All @@ -202,6 +210,7 @@ public static HttpCacheClient create(
verifyDownloads,
extraHttpHeaders,
digestUtil,
retrier,
creds,
authAndTlsOptions,
domainSocketAddress);
Expand All @@ -219,6 +228,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 +294,7 @@ public void channelCreated(Channel ch) {
this.extraHttpHeaders = extraHttpHeaders;
this.verifyDownloads = verifyDownloads;
this.digestUtil = digestUtil;
this.retrier = retrier;
}

@SuppressWarnings("FutureReturnValueIgnored")
Expand Down Expand Up @@ -441,8 +452,11 @@ public ListenableFuture<Void> downloadBlob(
RemoteActionExecutionContext context, Digest digest, OutputStream out) {
final DigestOutputStream digestOut =
verifyDownloads ? digestUtil.newDigestOutputStream(out) : null;
final AtomicLong casBytesDownloaded = new AtomicLong();
return Futures.transformAsync(
get(digest, digestOut != null ? digestOut : out, /* casDownload= */ true),
retrier.executeAsync(
() ->
get(digest, digestOut != null ? digestOut : out, Optional.of(casBytesDownloaded))),
(v) -> {
try {
if (digestOut != null) {
Expand All @@ -458,7 +472,8 @@ public ListenableFuture<Void> downloadBlob(
}

@SuppressWarnings("FutureReturnValueIgnored")
private ListenableFuture<Void> get(Digest digest, final OutputStream out, boolean casDownload) {
private ListenableFuture<Void> get(
Digest digest, final OutputStream out, Optional<AtomicLong> casBytesDownloaded) {
final AtomicBoolean dataWritten = new AtomicBoolean();
OutputStream wrappedOut =
new OutputStream() {
Expand All @@ -469,12 +484,18 @@ 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);
if (casBytesDownloaded.isPresent()) {
casBytesDownloaded.get().addAndGet(length);
}
out.write(b, offset, length);
}

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

Expand All @@ -483,7 +504,12 @@ public void flush() throws IOException {
out.flush();
}
};
DownloadCommand downloadCmd = new DownloadCommand(uri, casDownload, digest, wrappedOut);
long offset = 0;
if (casBytesDownloaded.isPresent()) {
offset = casBytesDownloaded.get().get();
}
DownloadCommand downloadCmd =
new DownloadCommand(uri, casBytesDownloaded.isPresent(), digest, wrappedOut, offset);
SettableFuture<Void> outerF = SettableFuture.create();
acquireDownloadChannel()
.addListener(
Expand Down Expand Up @@ -575,8 +601,11 @@ private void getAfterCredentialRefresh(DownloadCommand cmd, SettableFuture<Void>
public ListenableFuture<CachedActionResult> downloadActionResult(
RemoteActionExecutionContext context, ActionKey actionKey, boolean inlineOutErr) {
return Futures.transform(
Utils.downloadAsActionResult(
actionKey, (digest, out) -> get(digest, out, /* casDownload= */ false)),
retrier.executeAsync(
() ->
Utils.downloadAsActionResult(
actionKey,
(digest, out) -> get(digest, out, /* casBytesDownloaded= */ Optional.empty()))),
CachedActionResult::remote,
MoreExecutors.directExecutor());
}
Expand Down Expand Up @@ -670,20 +699,28 @@ 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 463f83a

Please sign in to comment.