Make BuildEventStreamer flushable before first event

Make the BuildEventStreamer capable of handling the flush() method, even
before the initial build event is seen. This happens, if a lot (more than
10240 bytes) of output is produced before the BuildStartingEvent is generated-
This is not unlikely, as the whole option processing happens before the build
starts.

As the promise of flush() is that the OutErrProvider is cleared, and that all
output between two flush() get into separate events, numbered in order, we have
to temporarily store a list of stdout/stderr String pairs.

RELNOTES: None.
PiperOrigin-RevId: 160137478

2 files changed, 130 insertions, 7 deletions

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 66aaa7fe36..e10444b325 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
@@ -56,6 +56,7 @@ import com.google.devtools.build.lib.events.Event;
 import com.google.devtools.build.lib.events.EventHandler;
 import com.google.devtools.build.lib.events.Reporter;
 import com.google.devtools.build.lib.rules.extra.ExtraAction;
+import com.google.devtools.build.lib.util.Pair;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.HashMap;
@@ -86,6 +87,7 @@ public class BuildEventStreamer implements EventHandler {
   private Set<BuildEventId> announcedEvents;
   private final Set<BuildEventId> postedEvents = new HashSet<>();
   private final Set<BuildEventId> configurationsPosted = new HashSet<>();
+  private List<Pair<String, String>> bufferedStdoutStderrPairs = new ArrayList<>();
   private final Multimap<BuildEventId, BuildEvent> pendingEvents = HashMultimap.create();
   private int progressCount;
   private final CountingArtifactGroupNamer artifactGroupNamer = new CountingArtifactGroupNamer();
@@ -168,6 +170,7 @@ public class BuildEventStreamer implements EventHandler {
   private void post(BuildEvent event) {
     BuildEvent linkEvent = null;
     BuildEventId id = event.getEventId();
+    List<BuildEvent> flushEvents = null;
 
     synchronized (this) {
       if (announcedEvents == null) {
@@ -182,6 +185,14 @@ public class BuildEventStreamer implements EventHandler {
         if (reporter != null) {
           reporter.post(new AnnounceBuildEventTransportsEvent(transports));
         }
+
+        if (!bufferedStdoutStderrPairs.isEmpty()) {
+          flushEvents = new ArrayList<>(bufferedStdoutStderrPairs.size());
+          for (Pair<String, String> outErrPair : bufferedStdoutStderrPairs) {
+            flushEvents.add(flushStdoutStderrEvent(outErrPair.getFirst(), outErrPair.getSecond()));
+          }
+        }
+        bufferedStdoutStderrPairs = null;
       } else {
         if (!announcedEvents.contains(id)) {
           String out = null;
@@ -215,6 +226,14 @@ public class BuildEventStreamer implements EventHandler {
       }
       transport.sendBuildEvent(event, artifactGroupNamer);
     }
+
+    if (flushEvents != null) {
+      for (BuildEvent flushEvent : flushEvents) {
+        for (BuildEventTransport transport : transports) {
+          transport.sendBuildEvent(flushEvent, artifactGroupNamer);
+        }
+      }
+    }
   }
 
   /** Clear pending events by generating aborted events for all their requisits. */
@@ -383,8 +402,16 @@ public class BuildEventStreamer implements EventHandler {
     }
   }
 
+  private synchronized BuildEvent flushStdoutStderrEvent(String out, String err) {
+    BuildEvent updateEvent = ProgressEvent.progressUpdate(progressCount, out, err);
+    progressCount++;
+    announcedEvents.addAll(updateEvent.getChildrenEvents());
+    postedEvents.add(updateEvent.getEventId());
+    return updateEvent;
+  }
+
   void flush() {
-    BuildEvent updateEvent;
+    BuildEvent updateEvent = null;
     synchronized (this) {
       String out = null;
       String err = null;
@@ -392,13 +419,16 @@ public class BuildEventStreamer implements EventHandler {
         out = outErrProvider.getOut();
         err = outErrProvider.getErr();
       }
-      updateEvent = ProgressEvent.progressUpdate(progressCount, out, err);
-      progressCount++;
-      announcedEvents.addAll(updateEvent.getChildrenEvents());
-      postedEvents.add(updateEvent.getEventId());
+      if (announcedEvents != null) {
+        updateEvent = flushStdoutStderrEvent(out, err);
+      } else {
+        bufferedStdoutStderrPairs.add(Pair.of(out, err));
+      }
     }
-    for (BuildEventTransport transport : transports) {
-      transport.sendBuildEvent(updateEvent, artifactGroupNamer);
+    if (updateEvent != null) {
+      for (BuildEventTransport transport : transports) {
+        transport.sendBuildEvent(updateEvent, artifactGroupNamer);
+      }
     }
   }
 
diff --git a/src/test/java/com/google/devtools/build/lib/runtime/BuildEventStreamerTest.java b/src/test/java/com/google/devtools/build/lib/runtime/BuildEventStreamerTest.java
index 76272294fa..9fae49273a 100644
--- a/src/test/java/com/google/devtools/build/lib/runtime/BuildEventStreamerTest.java
+++ b/src/test/java/com/google/devtools/build/lib/runtime/BuildEventStreamerTest.java
@@ -628,4 +628,97 @@ public class BuildEventStreamerTest extends FoundationTestCase {
     assertThat(allEventsSeen.get(5).getEventId()).isEqualTo(BuildEventId.progressId(2));
     assertThat(allEventsSeen.get(6)).isEqualTo(secondWithConfiguration);
   }
+
+  @Test
+  public void testEarlyFlush() throws Exception {
+    // Verify that the streamer can handle early calls to flush() and still correctly
+    // reports stdout and stderr in the build-event stream.
+    RecordingBuildEventTransport transport = new RecordingBuildEventTransport();
+    BuildEventStreamer streamer =
+        new BuildEventStreamer(ImmutableSet.<BuildEventTransport>of(transport), reporter);
+    BuildEventStreamer.OutErrProvider outErr =
+        Mockito.mock(BuildEventStreamer.OutErrProvider.class);
+    String firstStdoutMsg = "Some text that was written to stdout.";
+    String firstStderrMsg = "The UI text that bazel wrote to stderr.";
+    String secondStdoutMsg = "More text that was written to stdout, still before the start event.";
+    String secondStderrMsg = "More text written to stderr, still before the start event.";
+    when(outErr.getOut()).thenReturn(firstStdoutMsg).thenReturn(secondStdoutMsg);
+    when(outErr.getErr()).thenReturn(firstStderrMsg).thenReturn(secondStderrMsg);
+    BuildEvent startEvent =
+        new GenericBuildEvent(
+            testId("Initial"),
+            ImmutableSet.<BuildEventId>of(ProgressEvent.INITIAL_PROGRESS_UPDATE));
+
+    streamer.registerOutErrProvider(outErr);
+    streamer.flush();
+    streamer.flush();
+    streamer.buildEvent(startEvent);
+
+    List<BuildEvent> eventsSeen = transport.getEvents();
+    assertThat(eventsSeen).hasSize(3);
+    assertThat(eventsSeen.get(0).getEventId()).isEqualTo(startEvent.getEventId());
+    BuildEvent progressEvent = eventsSeen.get(1);
+    assertThat(progressEvent.getEventId()).isEqualTo(ProgressEvent.INITIAL_PROGRESS_UPDATE);
+    BuildEventStreamProtos.BuildEvent progressEventProto = transport.getEventProtos().get(1);
+    assertThat(progressEventProto.getProgress().getStdout()).isEqualTo(firstStdoutMsg);
+    assertThat(progressEventProto.getProgress().getStderr()).isEqualTo(firstStderrMsg);
+    BuildEventStreamProtos.BuildEvent secondProgressEventProto = transport.getEventProtos().get(2);
+    assertThat(secondProgressEventProto.getProgress().getStdout()).isEqualTo(secondStdoutMsg);
+    assertThat(secondProgressEventProto.getProgress().getStderr()).isEqualTo(secondStderrMsg);
+
+    // As there is only one progress event, the OutErrProvider should be queried
+    // only once per flush() for stdout and stderr.
+    verify(outErr, times(2)).getOut();
+    verify(outErr, times(2)).getErr();
+  }
+
+  @Test
+  public void testEarlyFlushBadInitialEvent() throws Exception {
+    // Verify that an early flush works correctly with an unusual start event.
+    // In this case, we expect 3 events in the stream, in that order:
+    // - an artifical progress event as initial event, to properly link in
+    //   all events
+    // - the unusal first event we have seen, and
+    // - a progress event reporting the flushed messages.
+    RecordingBuildEventTransport transport = new RecordingBuildEventTransport();
+    BuildEventStreamer streamer =
+        new BuildEventStreamer(ImmutableSet.<BuildEventTransport>of(transport), reporter);
+    BuildEventStreamer.OutErrProvider outErr =
+        Mockito.mock(BuildEventStreamer.OutErrProvider.class);
+    String stdoutMsg = "Some text that was written to stdout.";
+    String stderrMsg = "The UI text that bazel wrote to stderr.";
+    when(outErr.getOut()).thenReturn(stdoutMsg);
+    when(outErr.getErr()).thenReturn(stderrMsg);
+
+    BuildEvent unexpectedStartEvent =
+        new GenericBuildEvent(testId("unexpected start"), ImmutableSet.<BuildEventId>of());
+
+    streamer.registerOutErrProvider(outErr);
+    streamer.flush();
+    streamer.buildEvent(unexpectedStartEvent);
+
+    List<BuildEvent> eventsSeen = transport.getEvents();
+    assertThat(eventsSeen).hasSize(3);
+
+    BuildEvent initial = eventsSeen.get(0);
+    assertThat(initial.getEventId()).isEqualTo(ProgressEvent.INITIAL_PROGRESS_UPDATE);
+    BuildEventStreamProtos.BuildEvent initialProto = transport.getEventProtos().get(0);
+    assertThat(initialProto.getProgress().getStdout()).isEmpty();
+    assertThat(initialProto.getProgress().getStderr()).isEmpty();
+
+    assertThat(eventsSeen.get(1).getEventId()).isEqualTo(unexpectedStartEvent.getEventId());
+    assertWithMessage("Unexpected event should be linked")
+        .that(initial.getChildrenEvents().contains(unexpectedStartEvent.getEventId()))
+        .isTrue();
+
+    BuildEventStreamProtos.BuildEvent progressProto = transport.getEventProtos().get(2);
+    assertThat(progressProto.getProgress().getStdout()).isEqualTo(stdoutMsg);
+    assertThat(progressProto.getProgress().getStderr()).isEqualTo(stderrMsg);
+    assertWithMessage("flushed progress should be linked")
+        .that(initial.getChildrenEvents().contains(eventsSeen.get(2).getEventId()))
+        .isTrue();
+
+    verify(outErr, times(1)).getOut();
+    verify(outErr, times(1)).getErr();
+  }
 }