Implement --bes_upload_mode=fully_async.
Unlike NOWAIT_FOR_UPLOAD_COMPLETE, this mode does not block the subsequent build on server-side acknowledgement.
All file uploads should be complete by the time we unblock, so there should not be a risk of a racy read.
In case of server shutdown, we attempt to block on acknowledgement just as in NOWAIT_FOR_UPLOAD_COMPLETE.
RELNOTES: None
PiperOrigin-RevId: 244867998
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 da511ff..2d41ab4 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
@@ -28,6 +28,7 @@
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.google.common.util.concurrent.Uninterruptibles;
import com.google.devtools.build.lib.authandtls.AuthAndTLSOptions;
+import com.google.devtools.build.lib.buildeventservice.BuildEventServiceOptions.BesUploadMode;
import com.google.devtools.build.lib.buildeventservice.client.BuildEventServiceClient;
import com.google.devtools.build.lib.buildeventstream.AnnounceBuildEventTransportsEvent;
import com.google.devtools.build.lib.buildeventstream.BuildEventArtifactUploader;
@@ -96,6 +97,15 @@
private ImmutableMap<BuildEventTransport, ListenableFuture<Void>> closeFuturesMap =
ImmutableMap.of();
+ /**
+ * Holds the half-close futures for the upload of each transport. The completion of the half-close
+ * indicates that the client has sent all of the data to the server and is just waiting for
+ * acknowledgement. The client must still keep the data buffered locally in case acknowledgement
+ * fails.
+ */
+ private ImmutableMap<BuildEventTransport, ListenableFuture<Void>> halfCloseFuturesMap =
+ ImmutableMap.of();
+
// TODO(lpino): Use Optional instead of @Nullable for the members below.
@Nullable private OutErr outErr;
@Nullable private ImmutableSet<BuildEventTransport> bepTransports;
@@ -147,6 +157,7 @@
.values()
.forEach(closeFuture -> closeFuture.cancel(/* mayInterruptIfRunning= */ true));
closeFuturesMap = ImmutableMap.of();
+ halfCloseFuturesMap = ImmutableMap.of();
}
private void waitForPreviousInvocation() {
@@ -157,8 +168,12 @@
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.
+ ImmutableMap<BuildEventTransport, ListenableFuture<Void>> futureMap =
+ besOptions.besUploadMode == BesUploadMode.FULLY_ASYNC
+ ? halfCloseFuturesMap
+ : closeFuturesMap;
Uninterruptibles.getUninterruptibly(
- Futures.allAsList(closeFuturesMap.values()),
+ Futures.allAsList(futureMap.values()),
getMaxWaitForPreviousInvocation().getSeconds(),
TimeUnit.SECONDS);
} catch (TimeoutException exception) {
@@ -278,6 +293,7 @@
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.
+ googleLogger.atInfo().log("Closing pending build event transports");
Uninterruptibles.getUninterruptibly(
Futures.allAsList(streamer.getCloseFuturesMap().values()));
} catch (ExecutionException e) {
@@ -380,12 +396,14 @@
private void closeBepTransports() throws AbruptExitException {
closeFuturesMap = streamer.getCloseFuturesMap();
+ halfCloseFuturesMap = streamer.getHalfClosedMap();
switch (besOptions.besUploadMode) {
case WAIT_FOR_UPLOAD_COMPLETE:
waitForBuildEventTransportsToClose();
return;
case NOWAIT_FOR_UPLOAD_COMPLETE:
+ case FULLY_ASYNC:
// When running asynchronously notify the UI immediately since we won't wait for the
// uploads to close.
for (BuildEventTransport bepTransport : bepTransports) {
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 ed22d50..7bf8231 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
@@ -139,8 +139,15 @@
/** Determines the mode that will be used to upload data to the Build Event Service. */
public enum BesUploadMode {
+ /** Block at the end of the build waiting for the upload to complete */
WAIT_FOR_UPLOAD_COMPLETE,
- NOWAIT_FOR_UPLOAD_COMPLETE
+ /** Block at the beginning of the next build waiting for upload completion */
+ NOWAIT_FOR_UPLOAD_COMPLETE,
+ /**
+ * Block at the beginning of the next build waiting for the client to finish uploading the data,
+ * but possibly not blocking on the server acknowledgement.
+ */
+ FULLY_ASYNC,
}
/** Converter for {@link BesUploadMode} */
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 30f4374..ca3b4e6 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
@@ -131,6 +131,8 @@
@GuardedBy("lock")
private SettableFuture<Void> closeFuture;
+ private final SettableFuture<Void> halfCloseFuture = SettableFuture.create();
+
/**
* The thread that calls the lifecycle RPCs and does the build event upload. It's started lazily
* on the first call to {@link #enqueueEvent(BuildEvent)} or {@link #close()} (which ever comes
@@ -459,6 +461,7 @@
lastEvent.getSequenceNumber(), lastEvent.getCreationTime());
streamContext.sendOverStream(request);
streamContext.halfCloseStream();
+ halfCloseFuture.set(null);
}
break;
diff --git a/src/main/java/com/google/devtools/build/lib/buildeventstream/BuildEventTransport.java b/src/main/java/com/google/devtools/build/lib/buildeventstream/BuildEventTransport.java
index 782d36a..7d15fab 100644
--- a/src/main/java/com/google/devtools/build/lib/buildeventstream/BuildEventTransport.java
+++ b/src/main/java/com/google/devtools/build/lib/buildeventstream/BuildEventTransport.java
@@ -46,15 +46,32 @@
/**
* Initiates a close. Callers may listen to the returned future to be notified when the close is
- * complete i.e. wait for all build events to be sent. The future may contain any information
- * about possible transport errors.
+ * complete i.e. wait for all build events to be sent and acknowledged. The future may contain any
+ * information about possible transport errors.
*
* <p>This method might be called multiple times without any effect after the first call.
*
- * <p>This method should not throw any exceptions.
+ * <p>This method should not throw any exceptions, but the returned Future might.
*/
ListenableFuture<Void> close();
+ /**
+ * Returns the status of half-close. Callers may listen to the return future to be notified when
+ * the half-close is complete
+ *
+ * <p>Half-close indicates that all client-side data is transmitted but still waiting on
+ * server-side acknowledgement. The client must buffer the information in case the server fails to
+ * acknowledge.
+ *
+ * <p>Implementations may choose to return the full close Future via {@link #close()} if there is
+ * no sensible half-close state.
+ *
+ * <p>This should be only called after {@link #close()}.
+ */
+ default ListenableFuture<Void> getHalfCloseFuture() {
+ return close();
+ }
+
@VisibleForTesting
@Nullable
BuildEventArtifactUploader getUploader();
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 560dd18..93d7da1 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
@@ -43,6 +43,7 @@
import java.util.concurrent.ExecutionException;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.logging.Level;
import java.util.logging.Logger;
import javax.annotation.concurrent.ThreadSafe;
@@ -85,6 +86,7 @@
private final Function<BuildEventStreamProtos.BuildEvent, byte[]> serializeFunc;
private final BuildEventArtifactUploader uploader;
+ private final AtomicBoolean isClosed = new AtomicBoolean();
@VisibleForTesting
final BlockingQueue<ListenableFuture<BuildEventStreamProtos.BuildEvent>> pendingWrites =
@@ -170,7 +172,9 @@
}
public ListenableFuture<Void> close() {
- if (closeFuture.isDone()) {
+ if (isClosed.getAndSet(true)) {
+ return closeFuture;
+ } else if (closeFuture.isDone()) {
return closeFuture;
}
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 ef4bdb9..742aa2b 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
@@ -120,6 +120,15 @@
ImmutableMap.of();
/**
+ * Holds the half-close futures for the upload of each transport. The completion of the half-close
+ * indicates that the client has sent all of the data to the server and is just waiting for
+ * acknowledgement. The client must still keep the data buffered locally in case acknowledgement
+ * fails.
+ */
+ private ImmutableMap<BuildEventTransport, ListenableFuture<Void>> halfCloseFuturesMap =
+ ImmutableMap.of();
+
+ /**
* Provider for stdout and stderr output.
*/
public interface OutErrProvider {
@@ -341,10 +350,16 @@
ImmutableMap.Builder<BuildEventTransport, ListenableFuture<Void>> closeFuturesMapBuilder =
ImmutableMap.builder();
for (final BuildEventTransport transport : transports) {
- ListenableFuture<Void> closeFuture = transport.close();
- closeFuturesMapBuilder.put(transport, closeFuture);
+ closeFuturesMapBuilder.put(transport, transport.close());
}
closeFuturesMap = closeFuturesMapBuilder.build();
+
+ ImmutableMap.Builder<BuildEventTransport, ListenableFuture<Void>> halfCloseFuturesMapBuilder =
+ ImmutableMap.builder();
+ for (final BuildEventTransport transport : transports) {
+ halfCloseFuturesMapBuilder.put(transport, transport.getHalfCloseFuture());
+ }
+ halfCloseFuturesMap = halfCloseFuturesMapBuilder.build();
}
private void maybeReportArtifactSet(
@@ -685,6 +700,19 @@
return closeFuturesMap;
}
+ /**
+ * Returns the map from BEP transports to their corresponding half-close futures.
+ *
+ * <p>Half-close indicates that all client-side data is transmitted but still waiting on
+ * server-side acknowledgement. The client must buffer the information in case the server fails to
+ * acknowledge.
+ *
+ * <p>If this method is called before calling {@link #close()} then it will return an empty map.
+ */
+ public synchronized ImmutableMap<BuildEventTransport, ListenableFuture<Void>> getHalfClosedMap() {
+ return halfCloseFuturesMap;
+ }
+
/** A builder for {@link BuildEventStreamer}. */
public static class Builder {
private Set<BuildEventTransport> buildEventTransports;
