Implement asynchronous Build Event Service (BES) upload.
We introduce a flag to control the flow of invocations w.r.t. BES upload. This new flag is called --bes_upload_mode and it currently has two possible values:
- WAIT_FOR_UPLOAD_COMPLETE (default, present in before this commit): In this mode Bazel will wait for the communication with the BES to finish.
- NOWAIT_FOR_UPLOAD_COMPLETE (new mode): In this mode Bazel *won't* wait for the BES upload to finish, instead it will finish the invocation as soon as it can. The upload continues in the background and if a new invocation starts Bazel will wait five seconds for the previous upload to finish before (abruptly) cancelling it.
PiperOrigin-RevId: 243591454
diff --git a/src/main/java/com/google/devtools/build/lib/buildeventservice/BUILD b/src/main/java/com/google/devtools/build/lib/buildeventservice/BUILD
index 4049fa5..e037768 100644
--- a/src/main/java/com/google/devtools/build/lib/buildeventservice/BUILD
+++ b/src/main/java/com/google/devtools/build/lib/buildeventservice/BUILD
@@ -39,6 +39,7 @@
"//src/main/java/com/google/devtools/build/lib/vfs",
"//src/main/java/com/google/devtools/common/options",
"//third_party:auto_value",
+ "//third_party:flogger",
"//third_party:guava",
"//third_party:jsr305",
"//third_party/grpc:grpc-jar",
diff --git a/src/main/java/com/google/devtools/build/lib/buildeventservice/BuildEventServiceModule.java b/src/main/java/com/google/devtools/build/lib/buildeventservice/BuildEventServiceModule.java
index b0529d2..671f032 100644
--- a/src/main/java/com/google/devtools/build/lib/buildeventservice/BuildEventServiceModule.java
+++ b/src/main/java/com/google/devtools/build/lib/buildeventservice/BuildEventServiceModule.java
@@ -22,6 +22,7 @@
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
+import com.google.common.flogger.GoogleLogger;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
@@ -69,6 +70,7 @@
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicReference;
import java.util.logging.Level;
import java.util.logging.Logger;
@@ -82,6 +84,9 @@
extends BlazeModule {
private static final Logger logger = Logger.getLogger(BuildEventServiceModule.class.getName());
+ private static final GoogleLogger googleLogger = GoogleLogger.forEnclosingClass();
+ // TODO(lpino): Consider making the wait value configurable.
+ private static final Duration MAX_WAIT_FOR_PREVIOUS_INVOCATION = Duration.ofSeconds(5);
private final AtomicReference<AbruptExitException> pendingAbruptExitException =
new AtomicReference<>();
@@ -90,14 +95,17 @@
private AuthAndTLSOptions authTlsOptions;
private BuildEventStreamOptions besStreamOptions;
private boolean useExperimentalUi;
+ /** Holds the close futures for the upload of each transport */
+ private ImmutableMap<BuildEventTransport, ListenableFuture<Void>> closeFuturesMap =
+ ImmutableMap.of();
// TODO(lpino): Use Optional instead of @Nullable for the members below.
- private @Nullable OutErr outErr;
- private @Nullable ImmutableSet<BuildEventTransport> bepTransports;
- private @Nullable String buildRequestId;
- private @Nullable String invocationId;
- private @Nullable Reporter cmdLineReporter;
- private @Nullable BuildEventStreamer streamer;
+ @Nullable private OutErr outErr;
+ @Nullable private ImmutableSet<BuildEventTransport> bepTransports;
+ @Nullable private String buildRequestId;
+ @Nullable private String invocationId;
+ @Nullable private Reporter cmdLineReporter;
+ @Nullable private BuildEventStreamer streamer;
protected BESOptionsT besOptions;
@@ -128,7 +136,7 @@
// Don't hide unchecked exceptions as part of the error reporting.
Throwables.throwIfUnchecked(exception);
- logger.log(Level.SEVERE, msg, exception);
+ googleLogger.atSevere().withCause(exception).log(msg);
AbruptExitException abruptException = new AbruptExitException(msg, exitCode, exception);
reportCommandLineError(commandLineReporter, exception);
moduleEnvironment.exit(abruptException);
@@ -143,15 +151,65 @@
BuildEventProtocolOptions.class);
}
+ private void cancelPendingUploads() {
+ closeFuturesMap
+ .values()
+ .forEach(closeFuture -> closeFuture.cancel(/* mayInterruptIfRunning= */ true));
+ closeFuturesMap = ImmutableMap.of();
+ }
+
+ private void waitForPreviousInvocation() {
+ AbruptExitException pendingException = pendingAbruptExitException.getAndSet(null);
+ if (pendingException != null) {
+ cmdLineReporter.handle(
+ Event.warn(
+ String.format(
+ "Previous invocation failed to finish Build Event Protocol upload with "
+ + "the following exception: '%s'. "
+ + "Ignoring the failure and starting a new invocation...",
+ pendingException.getMessage())));
+ cancelPendingUploads();
+ return;
+ }
+
+ if (closeFuturesMap.isEmpty()) {
+ return;
+ }
+
+ try {
+ // TODO(b/234994611): Find a way to print a meaningful message when waiting. The current
+ // infrastructure doesn't allow printing messages in the terminal in beforeCommand.
+ Uninterruptibles.getUninterruptibly(
+ Futures.allAsList(closeFuturesMap.values()),
+ MAX_WAIT_FOR_PREVIOUS_INVOCATION.getSeconds(),
+ TimeUnit.SECONDS);
+ } catch (TimeoutException exception) {
+ String msg =
+ String.format(
+ "Pending Build Event Protocol upload took more than %ds to finish. "
+ + "Cancelling and starting a new invocation...",
+ MAX_WAIT_FOR_PREVIOUS_INVOCATION.getSeconds());
+ cmdLineReporter.handle(Event.warn(msg));
+ googleLogger.atWarning().withCause(exception).log(msg);
+ } catch (ExecutionException exception) {
+ String msg =
+ String.format(
+ "Previous Build Event Protocol upload failed with "
+ + "the following exception: '%s'. "
+ + "Ignoring the failure and starting a new invocation...",
+ exception.getMessage());
+ cmdLineReporter.handle(Event.warn(msg));
+ googleLogger.atWarning().withCause(exception).log(msg);
+ } finally {
+ cancelPendingUploads();
+ }
+ }
+
@Override
public void beforeCommand(CommandEnvironment cmdEnv) {
this.invocationId = cmdEnv.getCommandId().toString();
this.buildRequestId = cmdEnv.getBuildRequestId();
this.cmdLineReporter = cmdEnv.getReporter();
- // Reset to null in case afterCommand was not called.
- // TODO(lpino): Remove this statement once {@link BlazeModule#afterCommmand()} is guaranteed
- // to be executed for every invocation.
- this.outErr = null;
OptionsParsingResult parsingResult = cmdEnv.getOptions();
this.besOptions = Preconditions.checkNotNull(parsingResult.getOptions(optionsClass()));
@@ -175,6 +233,8 @@
.select(bepOptions.buildEventUploadStrategy)
.create(cmdEnv));
+ waitForPreviousInvocation();
+
if (!whitelistedCommands(besOptions).contains(cmdEnv.getCommandName())) {
// Exit early if the running command isn't supported.
return;
@@ -230,17 +290,52 @@
return outErr;
}
+ private void forceShutdownBuildEventStreamer() {
+ streamer.close(AbortReason.INTERNAL);
+
+ try {
+ // TODO(b/130148250): Uninterruptibles.getUninterruptibly waits forever if no timeout is
+ // passed. We should fix this by waiting at most the value set by bes_timeout.
+ Uninterruptibles.getUninterruptibly(
+ Futures.allAsList(streamer.getCloseFuturesMap().values()));
+ } catch (ExecutionException e) {
+ googleLogger.atSevere().withCause(e).log("Failed to close a build event transport");
+ LoggingUtil.logToRemote(Level.SEVERE, "Failed to close a build event transport", e);
+ } finally {
+ cancelPendingUploads();
+ }
+ }
+
@Override
public void blazeShutdownOnCrash() {
if (streamer != null) {
- logger.warning("Attempting to close BES streamer on crash");
- streamer.close(AbortReason.INTERNAL);
+ googleLogger.atWarning().log("Attempting to close BES streamer on crash");
+ forceShutdownBuildEventStreamer();
+ }
+ }
- try {
- waitForBuildEventTransportsToClose(streamer.getCloseFuturesMap());
- } catch (AbruptExitException e) {
- LoggingUtil.logToRemote(Level.WARNING, "Failure while waiting for BES close", e);
- }
+ @Override
+ public void blazeShutdown() {
+ AbruptExitException pendingException = pendingAbruptExitException.getAndSet(null);
+ if (pendingException != null) {
+ cancelPendingUploads();
+ return;
+ }
+
+ if (closeFuturesMap.isEmpty()) {
+ return;
+ }
+
+ try {
+ Uninterruptibles.getUninterruptibly(
+ Futures.allAsList(closeFuturesMap.values()),
+ MAX_WAIT_FOR_PREVIOUS_INVOCATION.getSeconds(),
+ TimeUnit.SECONDS);
+ } catch (TimeoutException | ExecutionException exception) {
+ googleLogger.atWarning().withCause(exception).log(
+ "Encountered Exception when closing BEP transports in Blaze's shutting down sequence");
+ } finally {
+ cancelPendingUploads();
}
}
@@ -254,9 +349,7 @@
+ "s."));
}
- private void waitForBuildEventTransportsToClose(
- ImmutableMap<BuildEventTransport, ListenableFuture<Void>> closeFuturesMap)
- throws AbruptExitException {
+ private void waitForBuildEventTransportsToClose() {
final ScheduledExecutorService executor =
Executors.newSingleThreadScheduledExecutor(
new ThreadFactoryBuilder().setNameFormat("bes-notify-ui-%d").build());
@@ -283,16 +376,22 @@
TimeUnit.SECONDS);
}
- // Wait synchronously for all the futures to finish.
try (AutoProfiler p = AutoProfiler.logged("waiting for BES close", logger)) {
+ // TODO(b/130148250): Uninterruptibles.getUninterruptibly waits forever if no timeout is
+ // passed. We should fix this by waiting at most the value set by bes_timeout.
Uninterruptibles.getUninterruptibly(Futures.allAsList(closeFuturesMap.values()));
}
- } catch (ExecutionException exception) {
- throw new AbruptExitException(
- "Failed to close a build event transport",
- ExitCode.TRANSIENT_BUILD_EVENT_SERVICE_UPLOAD_ERROR,
- exception);
+ } catch (ExecutionException e) {
+ this.pendingAbruptExitException.compareAndSet(
+ null,
+ new AbruptExitException(
+ String.format(
+ "Failed to close a build event transport with exception '%s'", e.getMessage()),
+ ExitCode.TRANSIENT_BUILD_EVENT_SERVICE_UPLOAD_ERROR,
+ e));
} finally {
+ cancelPendingUploads();
+
if (waitMessageFuture != null) {
waitMessageFuture.cancel(/* mayInterruptIfRunning= */ true);
}
@@ -303,19 +402,41 @@
}
}
+ private void closeBepTransports() throws AbruptExitException {
+ closeFuturesMap = streamer.getCloseFuturesMap();
+ switch (besOptions.besUploadMode) {
+ case WAIT_FOR_UPLOAD_COMPLETE:
+ waitForBuildEventTransportsToClose();
+ AbruptExitException e = pendingAbruptExitException.getAndSet(null);
+ if (e != null) {
+ throw e;
+ }
+ return;
+
+ case NOWAIT_FOR_UPLOAD_COMPLETE:
+ // When running asynchronously notify the UI immediately since we won't wait for the
+ // uploads to close.
+ for (BuildEventTransport bepTransport : bepTransports) {
+ cmdLineReporter.post(new BuildEventTransportClosedEvent(bepTransport));
+ }
+ return;
+ }
+ throw new IllegalStateException("Unknown BesUploadMode found: " + besOptions.besUploadMode);
+ }
+
@Override
public void afterCommand() throws AbruptExitException {
if (streamer != null) {
if (!streamer.isClosed()) {
// This should not occur, but close with an internal error if a {@link BuildEventStreamer}
// bug manifests as an unclosed streamer.
- logger.warning("Attempting to close BES streamer after command");
+ googleLogger.atWarning().log("Attempting to close BES streamer after command");
String msg = "BES was not properly closed";
LoggingUtil.logToRemote(Level.WARNING, msg, new IllegalStateException(msg));
- streamer.close(AbortReason.INTERNAL);
+ forceShutdownBuildEventStreamer();
}
- waitForBuildEventTransportsToClose(streamer.getCloseFuturesMap());
+ closeBepTransports();
if (!Strings.isNullOrEmpty(besOptions.besBackend)) {
constructAndMaybeReportInvocationIdUrl();
@@ -324,11 +445,6 @@
}
}
- AbruptExitException e = pendingAbruptExitException.getAndSet(null);
- if (e != null) {
- throw e;
- }
-
if (!besStreamOptions.keepBackendConnections) {
clearBesClient();
}
diff --git a/src/main/java/com/google/devtools/build/lib/buildeventservice/BuildEventServiceOptions.java b/src/main/java/com/google/devtools/build/lib/buildeventservice/BuildEventServiceOptions.java
index 2c23ff1..eda45ce 100644
--- a/src/main/java/com/google/devtools/build/lib/buildeventservice/BuildEventServiceOptions.java
+++ b/src/main/java/com/google/devtools/build/lib/buildeventservice/BuildEventServiceOptions.java
@@ -15,6 +15,7 @@
package com.google.devtools.build.lib.buildeventservice;
import com.google.devtools.common.options.Converters;
+import com.google.devtools.common.options.EnumConverter;
import com.google.devtools.common.options.Option;
import com.google.devtools.common.options.OptionDocumentationCategory;
import com.google.devtools.common.options.OptionEffectTag;
@@ -124,4 +125,28 @@
+ " backend. Bazel will output the URL appended by the invocation id to the"
+ " terminal.")
public String besResultsUrl;
+
+ @Option(
+ name = "bes_upload_mode",
+ defaultValue = "WAIT_FOR_UPLOAD_COMPLETE",
+ converter = BesUploadModeConverter.class,
+ documentationCategory = OptionDocumentationCategory.UNDOCUMENTED,
+ effectTags = {OptionEffectTag.EAGERNESS_TO_EXIT},
+ help =
+ "Specifies whether the Build Event Service upload should block the build completion "
+ + "or should end the invocation immediately and finish the upload in the background.")
+ public BesUploadMode besUploadMode;
+
+ /** Determines the mode that will be used to upload data to the Build Event Service. */
+ public enum BesUploadMode {
+ WAIT_FOR_UPLOAD_COMPLETE,
+ NOWAIT_FOR_UPLOAD_COMPLETE
+ }
+
+ /** Converter for {@link BesUploadMode} */
+ public static class BesUploadModeConverter extends EnumConverter<BesUploadMode> {
+ public BesUploadModeConverter() {
+ super(BesUploadMode.class, "Mode for uploading to the Build Event Service");
+ }
+ }
}
diff --git a/src/main/java/com/google/devtools/build/lib/buildeventservice/BuildEventServiceTransport.java b/src/main/java/com/google/devtools/build/lib/buildeventservice/BuildEventServiceTransport.java
index da71445..7707872d 100644
--- a/src/main/java/com/google/devtools/build/lib/buildeventservice/BuildEventServiceTransport.java
+++ b/src/main/java/com/google/devtools/build/lib/buildeventservice/BuildEventServiceTransport.java
@@ -72,7 +72,16 @@
// per API contract it is expected to never fail.
SettableFuture<Void> closeFuture = SettableFuture.create();
ListenableFuture<Void> uploaderCloseFuture = besUploader.close();
- uploaderCloseFuture.addListener(() -> closeFuture.set(null), MoreExecutors.directExecutor());
+ uploaderCloseFuture.addListener(
+ () -> {
+ // Make sure to cancel any pending uploads if the closing is cancelled.
+ if (uploaderCloseFuture.isCancelled()) {
+ besUploader.closeOnCancel();
+ }
+ closeFuture.set(null);
+ },
+ MoreExecutors.directExecutor());
+
return closeFuture;
}
diff --git a/src/main/java/com/google/devtools/build/lib/buildeventservice/BuildEventServiceUploader.java b/src/main/java/com/google/devtools/build/lib/buildeventservice/BuildEventServiceUploader.java
index 2535091..c69ff9a 100644
--- a/src/main/java/com/google/devtools/build/lib/buildeventservice/BuildEventServiceUploader.java
+++ b/src/main/java/com/google/devtools/build/lib/buildeventservice/BuildEventServiceUploader.java
@@ -148,6 +148,9 @@
@GuardedBy("lock")
private boolean interruptCausedByTimeout;
+ @GuardedBy("lock")
+ private boolean interruptCausedByCancel;
+
private StreamContext streamContext;
private BuildEventServiceUploader(
@@ -239,22 +242,34 @@
}
}
- /** Stops the upload immediately. Enqueued events that have not been sent yet will be lost. */
private void closeOnTimeout() {
synchronized (lock) {
+ interruptCausedByTimeout = true;
+ closeNow();
+ }
+ }
+
+ void closeOnCancel() {
+ synchronized (lock) {
+ interruptCausedByCancel = true;
+ closeNow();
+ }
+ }
+
+ /** Stops the upload immediately. Enqueued events that have not been sent yet will be lost. */
+ private void closeNow() {
+ synchronized (lock) {
if (uploadThread != null) {
if (uploadThread.isInterrupted()) {
return;
}
-
- interruptCausedByTimeout = true;
uploadThread.interrupt();
}
}
}
private void logAndExitAbruptly(String message, ExitCode exitCode, Throwable cause) {
- checkState(exitCode != ExitCode.SUCCESS);
+ checkState(!exitCode.equals(ExitCode.SUCCESS));
logger.info(message);
abruptExitCallback.accept(new AbruptExitException(message, exitCode, cause));
}
@@ -291,11 +306,14 @@
logger.info("Aborting the BES upload due to having received an interrupt");
synchronized (lock) {
Preconditions.checkState(
- interruptCausedByTimeout, "Unexpected interrupt on BES uploader thread");
- logAndExitAbruptly(
- "The Build Event Protocol upload timed out",
- ExitCode.TRANSIENT_BUILD_EVENT_SERVICE_UPLOAD_ERROR,
- e);
+ interruptCausedByTimeout || interruptCausedByCancel,
+ "Unexpected interrupt on BES uploader thread");
+ if (interruptCausedByTimeout) {
+ logAndExitAbruptly(
+ "The Build Event Protocol upload timed out",
+ ExitCode.TRANSIENT_BUILD_EVENT_SERVICE_UPLOAD_ERROR,
+ e);
+ }
}
} finally {
// TODO(buchgr): Due to b/113035235 exitFunc needs to be called before the close future
diff --git a/src/main/java/com/google/devtools/build/lib/buildeventstream/transports/BuildEventStreamOptions.java b/src/main/java/com/google/devtools/build/lib/buildeventstream/transports/BuildEventStreamOptions.java
index 4da6c78..5ac4448 100644
--- a/src/main/java/com/google/devtools/build/lib/buildeventstream/transports/BuildEventStreamOptions.java
+++ b/src/main/java/com/google/devtools/build/lib/buildeventstream/transports/BuildEventStreamOptions.java
@@ -116,29 +116,4 @@
+ "of the set as well as the file and uri lengths, which may in turn depend on the "
+ "hash function.")
public int maxNamedSetEntries;
-
- // TODO(ruperts): Remove these public getter methods for consistency with other options classes?
- public String getBuildEventTextFile() {
- return buildEventTextFile;
- }
-
- public String getBuildEventBinaryFile() {
- return buildEventBinaryFile;
- }
-
- public String getBuildEventJsonFile() {
- return buildEventJsonFile;
- }
-
- public boolean getBuildEventTextFilePathConversion() {
- return buildEventTextFilePathConversion;
- }
-
- public boolean getBuildEventBinaryFilePathConversion() {
- return buildEventBinaryFilePathConversion;
- }
-
- public boolean getBuildEventJsonFilePathConversion() {
- return buildEventJsonFilePathConversion;
- }
}
diff --git a/src/main/java/com/google/devtools/build/lib/buildeventstream/transports/FileTransport.java b/src/main/java/com/google/devtools/build/lib/buildeventstream/transports/FileTransport.java
index 55444cb..87f837a 100644
--- a/src/main/java/com/google/devtools/build/lib/buildeventstream/transports/FileTransport.java
+++ b/src/main/java/com/google/devtools/build/lib/buildeventstream/transports/FileTransport.java
@@ -187,6 +187,16 @@
if (closeFuture.isDone()) {
return closeFuture;
}
+
+ // Close abruptly if the closing future is cancelled.
+ closeFuture.addListener(
+ () -> {
+ if (closeFuture.isCancelled()) {
+ closeNow();
+ }
+ },
+ MoreExecutors.directExecutor());
+
try {
pendingWrites.put(CLOSE);
} catch (InterruptedException e) {
diff --git a/src/main/java/com/google/devtools/build/lib/runtime/BuildEventStreamer.java b/src/main/java/com/google/devtools/build/lib/runtime/BuildEventStreamer.java
index f592396..6a56407 100644
--- a/src/main/java/com/google/devtools/build/lib/runtime/BuildEventStreamer.java
+++ b/src/main/java/com/google/devtools/build/lib/runtime/BuildEventStreamer.java
@@ -78,7 +78,7 @@
@ThreadSafe
public class BuildEventStreamer {
private final Collection<BuildEventTransport> transports;
- private final BuildEventStreamOptions options;
+ private final BuildEventStreamOptions besOptions;
@GuardedBy("this")
private Set<BuildEventId> announcedEvents;
@@ -144,7 +144,7 @@
BuildEventStreamOptions options,
CountingArtifactGroupNamer artifactGroupNamer) {
this.transports = transports;
- this.options = options;
+ this.besOptions = options;
this.announcedEvents = null;
this.progressCount = 0;
this.artifactGroupNamer = artifactGroupNamer;
@@ -355,11 +355,11 @@
}
// We only split if the max number of entries is at least 2 (it must be at least a binary tree).
// The method throws for smaller values.
- if (options.maxNamedSetEntries >= 2) {
+ if (besOptions.maxNamedSetEntries >= 2) {
// We only split the event after naming it to avoid splitting the same node multiple times.
// Note that the artifactGroupNames keeps references to the individual pieces, so this can
// double the memory consumption of large nested sets.
- view = view.splitIfExceedsMaximumSize(options.maxNamedSetEntries);
+ view = view.splitIfExceedsMaximumSize(besOptions.maxNamedSetEntries);
}
for (NestedSetView<Artifact> transitive : view.transitives()) {
maybeReportArtifactSet(pathResolver, transitive);
@@ -645,7 +645,7 @@
/** Returns whether an {@link ActionExecutedEvent} should be published. */
private boolean shouldPublishActionExecutedEvent(ActionExecutedEvent event) {
- if (options.publishAllActions) {
+ if (besOptions.publishAllActions) {
return true;
}
if (event.getException() != null) {
@@ -713,3 +713,4 @@
}
}
}
+
