Skip to content

Commit

Permalink
Remote: Display download progress when actions are downloading output…
Browse files Browse the repository at this point in the history
…s from remote cache.

Normally, when executing action with remote cache/execution, the UI only display the "remote"/"remote-cache" strategy:
```
[500 / 1000] 500 actions, 3 running
    [Sched] Executing genrule //:test-1;
    Executing genrule //:test-2; 2s remote
    Executing genrule //:test-3; 3s remote ...
```

However, it doesn't tell users what is happening under the hood. #13555 fixed the confusion which the UI display the action is scheduling while it is actually downloading the outputs.

With this change, Bazel will display the downloads if action is downloading outputs. e.g.
```
[500 / 1000] 500 actions, 3 running
    [Sched] Executing genrule //:test-1; 1s remote
    Executing genrule //:test-2; Downloading 2.out, 20.1 KiB / 100 KiB; 2s remote
    Executing genrule //:test-3; 3s remote ...
```

Add a generic `ActionProgressEvent` which can be reported within action execution to display detailed execution progress for that action.

Closes #13557.

PiperOrigin-RevId: 383224334
  • Loading branch information
coeuvre authored and Copybara-Service committed Jul 6, 2021
1 parent ec3d4bd commit 0f812eb
Show file tree
Hide file tree
Showing 16 changed files with 619 additions and 36 deletions.
@@ -0,0 +1,39 @@
// Copyright 2021 The Bazel Authors. All rights reserved.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package com.google.devtools.build.lib.actions;

import com.google.auto.value.AutoValue;
import com.google.devtools.build.lib.events.ExtendedEventHandler.ProgressLike;

/** Notifications for the progress of an in-flight action. */
@AutoValue
public abstract class ActionProgressEvent implements ProgressLike {

public static ActionProgressEvent create(
ActionExecutionMetadata action, String progressId, String progress, boolean finished) {
return new AutoValue_ActionProgressEvent(action, progressId, progress, finished);
}

/** Gets the metadata associated with the action being scheduled. */
public abstract ActionExecutionMetadata action();

/** The id that uniquely determines the progress among all progress events within an action. */
public abstract String progressId();

/** Human readable description of the progress. */
public abstract String progress();

/** Whether the download progress reported about is finished already. */
public abstract boolean finished();
}
Expand Up @@ -328,7 +328,6 @@ public void report(ProgressStatus progress) {
return;
}

// TODO(ulfjack): We should report more details to the UI.
ExtendedEventHandler eventHandler = actionExecutionContext.getEventHandler();
progress.postTo(eventHandler, action);
}
Expand Down
1 change: 1 addition & 0 deletions src/main/java/com/google/devtools/build/lib/exec/BUILD
Expand Up @@ -269,6 +269,7 @@ java_library(
srcs = [
"SpawnCheckingCacheEvent.java",
"SpawnExecutingEvent.java",
"SpawnProgressEvent.java",
"SpawnRunner.java",
"SpawnSchedulingEvent.java",
],
Expand Down
@@ -0,0 +1,43 @@
// Copyright 2021 The Bazel Authors. All rights reserved.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package com.google.devtools.build.lib.exec;

import com.google.auto.value.AutoValue;
import com.google.devtools.build.lib.actions.ActionExecutionMetadata;
import com.google.devtools.build.lib.actions.ActionProgressEvent;
import com.google.devtools.build.lib.events.ExtendedEventHandler;
import com.google.devtools.build.lib.exec.SpawnRunner.ProgressStatus;

/** The {@link SpawnRunner} is making some progress. */
@AutoValue
public abstract class SpawnProgressEvent implements ProgressStatus {

public static SpawnProgressEvent create(String resourceId, String progress, boolean finished) {
return new AutoValue_SpawnProgressEvent(resourceId, progress, finished);
}

/** The id that uniquely determines the progress among all progress events for this spawn. */
abstract String progressId();

/** Human readable description of the progress. */
abstract String progress();

/** Whether the progress reported about is finished already. */
abstract boolean finished();

@Override
public void postTo(ExtendedEventHandler eventHandler, ActionExecutionMetadata action) {
eventHandler.post(ActionProgressEvent.create(action, progressId(), progress(), finished()));
}
}
127 changes: 122 additions & 5 deletions src/main/java/com/google/devtools/build/lib/remote/RemoteCache.java
Expand Up @@ -15,6 +15,8 @@

import static com.google.common.util.concurrent.Futures.immediateFuture;
import static com.google.common.util.concurrent.MoreExecutors.directExecutor;
import static com.google.devtools.build.lib.remote.common.ProgressStatusListener.NO_ACTION;
import static com.google.devtools.build.lib.remote.util.Utils.bytesCountToDisplayString;
import static com.google.devtools.build.lib.remote.util.Utils.getFromFuture;

import build.bazel.remote.execution.v2.Action;
Expand Down Expand Up @@ -50,6 +52,7 @@
import com.google.devtools.build.lib.actions.UserExecException;
import com.google.devtools.build.lib.actions.cache.MetadataInjector;
import com.google.devtools.build.lib.concurrent.ThreadSafety;
import com.google.devtools.build.lib.exec.SpawnProgressEvent;
import com.google.devtools.build.lib.exec.SpawnRunner.SpawnExecutionContext;
import com.google.devtools.build.lib.profiler.Profiler;
import com.google.devtools.build.lib.profiler.SilentCloseable;
Expand All @@ -58,6 +61,7 @@
import com.google.devtools.build.lib.remote.RemoteCache.ActionResultMetadata.SymlinkMetadata;
import com.google.devtools.build.lib.remote.common.LazyFileOutputStream;
import com.google.devtools.build.lib.remote.common.OutputDigestMismatchException;
import com.google.devtools.build.lib.remote.common.ProgressStatusListener;
import com.google.devtools.build.lib.remote.common.RemoteActionExecutionContext;
import com.google.devtools.build.lib.remote.common.RemoteActionFileArtifactValue;
import com.google.devtools.build.lib.remote.common.RemoteCacheClient;
Expand Down Expand Up @@ -91,6 +95,9 @@
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.atomic.AtomicLong;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import javax.annotation.Nullable;
Expand Down Expand Up @@ -314,6 +321,50 @@ private static Path toTmpDownloadPath(Path actualPath) {
return actualPath.getParentDirectory().getRelative(actualPath.getBaseName() + ".tmp");
}

static class DownloadProgressReporter {
private static final Pattern PATTERN = Pattern.compile("^bazel-out/[^/]+/[^/]+/");
private final ProgressStatusListener listener;
private final String id;
private final String file;
private final String totalSize;
private final AtomicLong downloadedBytes = new AtomicLong(0);

DownloadProgressReporter(ProgressStatusListener listener, String file, long totalSize) {
this.listener = listener;
this.id = file;
this.totalSize = bytesCountToDisplayString(totalSize);

Matcher matcher = PATTERN.matcher(file);
this.file = matcher.replaceFirst("");
}

void started() {
reportProgress(false, false);
}

void downloadedBytes(int count) {
downloadedBytes.addAndGet(count);
reportProgress(true, false);
}

void finished() {
reportProgress(true, true);
}

private void reportProgress(boolean includeBytes, boolean finished) {
String progress;
if (includeBytes) {
progress =
String.format(
"Downloading %s, %s / %s",
file, bytesCountToDisplayString(downloadedBytes.get()), totalSize);
} else {
progress = String.format("Downloading %s", file);
}
listener.onProgressStatus(SpawnProgressEvent.create(id, progress, finished));
}
}

/**
* Download the output files and directory trees of a remotely executed action to the local
* machine, as well stdin / stdout to the given files.
Expand All @@ -330,7 +381,8 @@ public void download(
RemotePathResolver remotePathResolver,
ActionResult result,
FileOutErr origOutErr,
OutputFilesLocker outputFilesLocker)
OutputFilesLocker outputFilesLocker,
ProgressStatusListener progressStatusListener)
throws ExecException, IOException, InterruptedException {
ActionResultMetadata metadata = parseActionResultMetadata(context, remotePathResolver, result);

Expand All @@ -347,7 +399,11 @@ public void download(
context,
remotePathResolver.localPathToOutputPath(file.path()),
toTmpDownloadPath(file.path()),
file.digest());
file.digest(),
new DownloadProgressReporter(
progressStatusListener,
remotePathResolver.localPathToOutputPath(file.path()),
file.digest().getSizeBytes()));
return Futures.transform(download, (d) -> file, directExecutor());
} catch (IOException e) {
return Futures.<FileMetadata>immediateFailedFuture(e);
Expand Down Expand Up @@ -499,10 +555,14 @@ private void createSymlinks(Iterable<SymlinkMetadata> symlinks) throws IOExcepti
}

public ListenableFuture<Void> downloadFile(
RemoteActionExecutionContext context, String outputPath, Path localPath, Digest digest)
RemoteActionExecutionContext context,
String outputPath,
Path localPath,
Digest digest,
DownloadProgressReporter reporter)
throws IOException {
SettableFuture<Void> outerF = SettableFuture.create();
ListenableFuture<Void> f = downloadFile(context, localPath, digest);
ListenableFuture<Void> f = downloadFile(context, localPath, digest, reporter);
Futures.addCallback(
f,
new FutureCallback<Void>() {
Expand All @@ -529,6 +589,16 @@ public void onFailure(Throwable throwable) {
/** Downloads a file (that is not a directory). The content is fetched from the digest. */
public ListenableFuture<Void> downloadFile(
RemoteActionExecutionContext context, Path path, Digest digest) throws IOException {
return downloadFile(context, path, digest, new DownloadProgressReporter(NO_ACTION, "", 0));
}

/** Downloads a file (that is not a directory). The content is fetched from the digest. */
public ListenableFuture<Void> downloadFile(
RemoteActionExecutionContext context,
Path path,
Digest digest,
DownloadProgressReporter reporter)
throws IOException {
Preconditions.checkNotNull(path.getParentDirectory()).createDirectoryAndParents();
if (digest.getSizeBytes() == 0) {
// Handle empty file locally.
Expand All @@ -549,7 +619,9 @@ public ListenableFuture<Void> downloadFile(
return COMPLETED_SUCCESS;
}

OutputStream out = new LazyFileOutputStream(path);
reporter.started();
OutputStream out = new ReportingOutputStream(new LazyFileOutputStream(path), reporter);

SettableFuture<Void> outerF = SettableFuture.create();
ListenableFuture<Void> f = cacheProtocol.downloadBlob(context, digest, out);
Futures.addCallback(
Expand All @@ -560,6 +632,7 @@ public void onSuccess(Void result) {
try {
out.close();
outerF.set(null);
reporter.finished();
} catch (IOException e) {
outerF.setException(e);
} catch (RuntimeException e) {
Expand All @@ -572,6 +645,7 @@ public void onSuccess(Void result) {
public void onFailure(Throwable t) {
try {
out.close();
reporter.finished();
} catch (IOException e) {
if (t != e) {
t.addSuppressed(e);
Expand Down Expand Up @@ -1100,6 +1174,49 @@ private static FailureDetail createFailureDetail(String message, Code detailedCo
.build();
}

/**
* An {@link OutputStream} that reports all the write operations with {@link
* DownloadProgressReporter}.
*/
private static class ReportingOutputStream extends OutputStream {

private final OutputStream out;
private final DownloadProgressReporter reporter;

ReportingOutputStream(OutputStream out, DownloadProgressReporter reporter) {
this.out = out;
this.reporter = reporter;
}

@Override
public void write(byte[] b) throws IOException {
out.write(b);
reporter.downloadedBytes(b.length);
}

@Override
public void write(byte[] b, int off, int len) throws IOException {
out.write(b, off, len);
reporter.downloadedBytes(len);
}

@Override
public void write(int b) throws IOException {
out.write(b);
reporter.downloadedBytes(1);
}

@Override
public void flush() throws IOException {
out.flush();
}

@Override
public void close() throws IOException {
out.close();
}
}

/** In-memory representation of action result metadata. */
static class ActionResultMetadata {

Expand Down
Expand Up @@ -386,7 +386,8 @@ public InMemoryOutput downloadOutputs(RemoteAction action, RemoteActionResult re
remotePathResolver,
result.actionResult,
action.spawnExecutionContext.getFileOutErr(),
action.spawnExecutionContext::lockOutputFiles);
action.spawnExecutionContext::lockOutputFiles,
action.spawnExecutionContext::report);
} else {
PathFragment inMemoryOutputPath = getInMemoryOutputPath(action.spawn);
inMemoryOutput =
Expand Down
@@ -0,0 +1,29 @@
// Copyright 2021 The Bazel Authors. All rights reserved.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package com.google.devtools.build.lib.remote.common;

import com.google.devtools.build.lib.exec.SpawnRunner.ProgressStatus;

/** An interface that is used to receive {@link ProgressStatus} updates during spawn execution. */
@FunctionalInterface
public interface ProgressStatusListener {

void onProgressStatus(ProgressStatus progress);

/** A {@link ProgressStatusListener} that does nothing. */
ProgressStatusListener NO_ACTION =
progress -> {
// Intentionally left empty
};
}

0 comments on commit 0f812eb

Please sign in to comment.