Refactor profiler

- move the save method to an inner class
- don't use a timer, use a blocking queue instead
- add a format enum (in anticipation of adding a json output format)
- update the test to use an in memory buffer, and avoid FoundationTestCase

Compared to the original https://github.com/bazelbuild/bazel/commit/15b8c259db111012b4642287172cb4d1d82151f3, it contains these changes:
- Make it so we don't create a queue if we are not going to write any
  data! The queue is now owned by the writer, and if there is no writer, there
  is no queue.

  This was causing a memory regression because slowest task profiling is
  enabled by default, in which case the profiler is started with no output
  file. In that case, there's no thread that is emptying the queue, but the
  queue was still created by default.

- add additional tests for slowest task and histogram handling; these also
  provide coverage for the case where the profiler is started without an output
  stream
- move all the writer thread handling into the inner class
- make writer access thread-safe
- add a bunch of documentation

PiperOrigin-RevId: 200212978

1 files changed, 161 insertions, 75 deletions

diff --git a/src/test/java/com/google/devtools/build/lib/profiler/ProfilerTest.java b/src/test/java/com/google/devtools/build/lib/profiler/ProfilerTest.java
index ae1d572596..0fee59186d 100644
--- a/src/test/java/com/google/devtools/build/lib/profiler/ProfilerTest.java
+++ b/src/test/java/com/google/devtools/build/lib/profiler/ProfilerTest.java
@@ -14,7 +14,9 @@
 package com.google.devtools.build.lib.profiler;
 
 import static com.google.common.truth.Truth.assertThat;
+import static com.google.devtools.build.lib.profiler.Profiler.Format.BINARY_BAZEL_FORMAT;
 import static java.nio.charset.StandardCharsets.ISO_8859_1;
+import static org.junit.Assert.fail;
 
 import com.google.common.collect.ImmutableList;
 import com.google.common.io.ByteStreams;
@@ -23,15 +25,15 @@ import com.google.devtools.build.lib.clock.Clock;
 import com.google.devtools.build.lib.profiler.Profiler.ProfiledTaskKinds;
 import com.google.devtools.build.lib.profiler.Profiler.SlowTask;
 import com.google.devtools.build.lib.profiler.analysis.ProfileInfo;
-import com.google.devtools.build.lib.testutil.FoundationTestCase;
 import com.google.devtools.build.lib.testutil.ManualClock;
 import com.google.devtools.build.lib.testutil.Suite;
 import com.google.devtools.build.lib.testutil.TestSpec;
 import com.google.devtools.build.lib.testutil.TestUtils;
-import com.google.devtools.build.lib.vfs.FileSystemUtils;
-import com.google.devtools.build.lib.vfs.Path;
-import java.io.InputStream;
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
 import java.io.OutputStream;
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
@@ -40,6 +42,7 @@ import java.util.zip.Deflater;
 import java.util.zip.DeflaterOutputStream;
 import java.util.zip.Inflater;
 import java.util.zip.InflaterInputStream;
+import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -50,29 +53,43 @@ import org.junit.runners.JUnit4;
  */
 @TestSpec(size = Suite.MEDIUM_TESTS) // testConcurrentProfiling takes ~700ms, testProfiler 100ms.
 @RunWith(JUnit4.class)
-public class ProfilerTest extends FoundationTestCase {
-
-  private Path cacheDir;
+public class ProfilerTest {
   private Profiler profiler = Profiler.instance();
   private ManualClock clock;
 
   @Before
-  public final void createCacheDirectory() throws Exception {
-    cacheDir = scratch.dir("/tmp");
-  }
-
-  @Before
   public final void setManualClock() {
     clock = new ManualClock();
     BlazeClock.setClock(clock);
   }
 
+  @After
+  public void forceStopToAvoidPoisoningTheProfiler() {
+    // If a test does not stop the profiler, e.g., due to a test failure, all subsequent tests fail
+    // because the profiler is still running, so we force-stop the profiler here.
+    try {
+      profiler.stop();
+    } catch (IOException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  private ByteArrayOutputStream start(ProfiledTaskKinds kinds, Profiler.Format format) {
+    ByteArrayOutputStream buffer = new ByteArrayOutputStream();
+    profiler.start(
+        kinds, buffer, format, "test", false, BlazeClock.instance(), BlazeClock.nanoTime());
+    return buffer;
+  }
+
+  private void startUnbuffered(ProfiledTaskKinds kinds) {
+    profiler.start(
+        kinds, null, null, "test", false, BlazeClock.instance(), BlazeClock.nanoTime());
+  }
+
   @Test
   public void testProfilerActivation() throws Exception {
-    Path cacheFile = cacheDir.getRelative("profile1.dat");
     assertThat(profiler.isActive()).isFalse();
-    profiler.start(ProfiledTaskKinds.ALL, cacheFile.getOutputStream(), "basic test", false,
-        BlazeClock.instance(), BlazeClock.instance().nanoTime());
+    start(ProfiledTaskKinds.ALL, BINARY_BAZEL_FORMAT);
     assertThat(profiler.isActive()).isTrue();
 
     profiler.stop();
@@ -81,14 +98,12 @@ public class ProfilerTest extends FoundationTestCase {
 
   @Test
   public void testTaskDetails() throws Exception {
-    Path cacheFile = cacheDir.getRelative("profile1.dat");
-    profiler.start(ProfiledTaskKinds.ALL, cacheFile.getOutputStream(), "basic test", false,
-        BlazeClock.instance(), BlazeClock.instance().nanoTime());
+    ByteArrayOutputStream buffer = start(ProfiledTaskKinds.ALL, BINARY_BAZEL_FORMAT);
     try (SilentCloseable c = profiler.profile(ProfilerTask.ACTION, "action task")) {
       profiler.logEvent(ProfilerTask.INFO, "event");
     }
     profiler.stop();
-    ProfileInfo info = ProfileInfo.loadProfile(cacheFile);
+    ProfileInfo info = ProfileInfo.loadProfile(new ByteArrayInputStream(buffer.toByteArray()));
     info.calculateStats();
 
     ProfileInfo.Task task = info.allTasksById.get(0);
@@ -104,9 +119,7 @@ public class ProfilerTest extends FoundationTestCase {
 
   @Test
   public void testProfiler() throws Exception {
-    Path cacheFile = cacheDir.getRelative("profile1.dat");
-    profiler.start(ProfiledTaskKinds.ALL, cacheFile.getOutputStream(), "basic test", false,
-        BlazeClock.instance(), BlazeClock.instance().nanoTime());
+    ByteArrayOutputStream buffer = start(ProfiledTaskKinds.ALL, BINARY_BAZEL_FORMAT);
     profiler.logSimpleTask(BlazeClock.instance().nanoTime(),
                            ProfilerTask.PHASE, "profiler start");
     try (SilentCloseable c = profiler.profile(ProfilerTask.ACTION, "complex task")) {
@@ -125,7 +138,7 @@ public class ProfilerTest extends FoundationTestCase {
     // all other calls to profiler should be ignored
     profiler.logEvent(ProfilerTask.PHASE, "should be ignored");
 
-    ProfileInfo info = ProfileInfo.loadProfile(cacheFile);
+    ProfileInfo info = ProfileInfo.loadProfile(new ByteArrayInputStream(buffer.toByteArray()));
     info.calculateStats();
     assertThat(info.allTasksById).hasSize(6); // only 5 tasks + finalization should be recorded
 
@@ -160,16 +173,22 @@ public class ProfilerTest extends FoundationTestCase {
 
   @Test
   public void testProfilerRecordingAllEvents() throws Exception {
-    Path cacheFile = cacheDir.getRelative("profile1.dat");
-    profiler.start(ProfiledTaskKinds.ALL, cacheFile.getOutputStream(), "basic test", true,
-        BlazeClock.instance(), BlazeClock.instance().nanoTime());
+    ByteArrayOutputStream buffer = new ByteArrayOutputStream();
+    profiler.start(
+        ProfiledTaskKinds.ALL,
+        buffer,
+        BINARY_BAZEL_FORMAT,
+        "basic test",
+        true,
+        BlazeClock.instance(),
+        BlazeClock.instance().nanoTime());
     try (SilentCloseable c = profiler.profile(ProfilerTask.ACTION, "action task")) {
       // Next task takes less than 10 ms but should be recorded anyway.
       clock.advanceMillis(1);
       profiler.logSimpleTask(BlazeClock.instance().nanoTime(), ProfilerTask.VFS_STAT, "stat1");
     }
     profiler.stop();
-    ProfileInfo info = ProfileInfo.loadProfile(cacheFile);
+    ProfileInfo info = ProfileInfo.loadProfile(new ByteArrayInputStream(buffer.toByteArray()));
     info.calculateStats();
     assertThat(info.allTasksById).hasSize(3); // 2 tasks + finalization should be recorded
 
@@ -182,10 +201,16 @@ public class ProfilerTest extends FoundationTestCase {
 
   @Test
   public void testProfilerRecordingOnlySlowestEvents() throws Exception {
-    Path profileData = cacheDir.getRelative("foo");
+    ByteArrayOutputStream buffer = new ByteArrayOutputStream();
 
-    profiler.start(ProfiledTaskKinds.SLOWEST, profileData.getOutputStream(), "test", true,
-        BlazeClock.instance(), BlazeClock.instance().nanoTime());
+    profiler.start(
+        ProfiledTaskKinds.SLOWEST,
+        buffer,
+        BINARY_BAZEL_FORMAT,
+        "test",
+        true,
+        BlazeClock.instance(),
+        BlazeClock.instance().nanoTime());
     profiler.logSimpleTask(10000, 20000, ProfilerTask.VFS_STAT, "stat");
     profiler.logSimpleTask(20000, 30000, ProfilerTask.REMOTE_EXECUTION, "remote execution");
 
@@ -194,7 +219,7 @@ public class ProfilerTest extends FoundationTestCase {
 
     profiler.stop();
 
-    ProfileInfo info = ProfileInfo.loadProfile(profileData);
+    ProfileInfo info = ProfileInfo.loadProfile(new ByteArrayInputStream(buffer.toByteArray()));
     info.calculateStats();
     assertThat(info.allTasksById).hasSize(1); // only VFS_STAT task should be recorded
 
@@ -203,14 +228,20 @@ public class ProfilerTest extends FoundationTestCase {
   }
 
   @Test
+  public void testSlowestTasks() throws Exception {
+    startUnbuffered(ProfiledTaskKinds.ALL);
+    profiler.logSimpleTaskDuration(
+        Profiler.nanoTimeMaybe(), Duration.ofSeconds(10), ProfilerTask.LOCAL_PARSE, "foo");
+    Iterable<SlowTask> slowestTasks = profiler.getSlowestTasks();
+    assertThat(slowestTasks).hasSize(1);
+    SlowTask task = slowestTasks.iterator().next();
+    assertThat(task.type).isEqualTo(ProfilerTask.LOCAL_PARSE);
+    profiler.stop();
+  }
+
+  @Test
   public void testGetSlowestTasksCapped() throws Exception {
-    profiler.start(
-        ProfiledTaskKinds.SLOWEST,
-        ByteStreams.nullOutputStream(),
-        "test",
-        /*recordAllDurations=*/ true,
-        BlazeClock.instance(),
-        BlazeClock.instance().nanoTime());
+    startUnbuffered(ProfiledTaskKinds.SLOWEST);
 
     // Add some fast tasks - these shouldn't show up in the slowest.
     for (int i = 0; i < ProfilerTask.VFS_STAT.slowestInstancesCount; i++) {
@@ -279,10 +310,15 @@ public class ProfilerTest extends FoundationTestCase {
 
   @Test
   public void testProfilerRecordsNothing() throws Exception {
-    Path profileData = cacheDir.getRelative("foo");
-
-    profiler.start(ProfiledTaskKinds.NONE, profileData.getOutputStream(), "test", true,
-        BlazeClock.instance(), BlazeClock.instance().nanoTime());
+    ByteArrayOutputStream buffer = new ByteArrayOutputStream();
+    profiler.start(
+        ProfiledTaskKinds.NONE,
+        buffer,
+        BINARY_BAZEL_FORMAT,
+        "test",
+        true,
+        BlazeClock.instance(),
+        BlazeClock.instance().nanoTime());
     profiler.logSimpleTask(10000, 20000, ProfilerTask.VFS_STAT, "stat");
 
     assertThat(ProfilerTask.VFS_STAT.collectsSlowestInstances()).isTrue();
@@ -290,16 +326,14 @@ public class ProfilerTest extends FoundationTestCase {
 
     profiler.stop();
 
-    ProfileInfo info = ProfileInfo.loadProfile(profileData);
+    ProfileInfo info = ProfileInfo.loadProfile(new ByteArrayInputStream(buffer.toByteArray()));
     info.calculateStats();
     assertThat(info.allTasksById).isEmpty();
   }
 
   @Test
   public void testConcurrentProfiling() throws Exception {
-    Path cacheFile = cacheDir.getRelative("profile3.dat");
-    profiler.start(ProfiledTaskKinds.ALL, cacheFile.getOutputStream(), "concurrent test", false,
-        BlazeClock.instance(), BlazeClock.instance().nanoTime());
+    ByteArrayOutputStream buffer = start(ProfiledTaskKinds.ALL, BINARY_BAZEL_FORMAT);
 
     long id = Thread.currentThread().getId();
     Thread thread1 = new Thread() {
@@ -329,7 +363,7 @@ public class ProfilerTest extends FoundationTestCase {
     }
     profiler.stop();
 
-    ProfileInfo info = ProfileInfo.loadProfile(cacheFile);
+    ProfileInfo info = ProfileInfo.loadProfile(new ByteArrayInputStream(buffer.toByteArray()));
     info.calculateStats();
     info.analyzeRelationships();
     assertThat(info.allTasksById).hasSize(4 + 10000 + 10000); // total number of tasks
@@ -353,9 +387,7 @@ public class ProfilerTest extends FoundationTestCase {
 
   @Test
   public void testPhaseTasks() throws Exception {
-    Path cacheFile = cacheDir.getRelative("profile4.dat");
-    profiler.start(ProfiledTaskKinds.ALL, cacheFile.getOutputStream(), "phase test", false,
-        BlazeClock.instance(), BlazeClock.instance().nanoTime());
+    ByteArrayOutputStream buffer = start(ProfiledTaskKinds.ALL, BINARY_BAZEL_FORMAT);
     Thread thread1 = new Thread() {
       @Override public void run() {
         for (int i = 0; i < 100; i++) {
@@ -394,7 +426,7 @@ public class ProfilerTest extends FoundationTestCase {
     clock.advanceMillis(1);
     profiler.stop();
 
-    ProfileInfo info = ProfileInfo.loadProfile(cacheFile);
+    ProfileInfo info = ProfileInfo.loadProfile(new ByteArrayInputStream(buffer.toByteArray()));
     info.calculateStats();
     info.analyzeRelationships();
     // number of tasks: INIT(1) + LOAD(1) + Thread1.TEST(100) + ANALYZE(1)
@@ -417,9 +449,7 @@ public class ProfilerTest extends FoundationTestCase {
 
   @Test
   public void testCorruptedFile() throws Exception {
-    Path cacheFile = cacheDir.getRelative("profile5.dat");
-    profiler.start(ProfiledTaskKinds.ALL, cacheFile.getOutputStream(), "phase test", false,
-        BlazeClock.instance(), BlazeClock.instance().nanoTime());
+    ByteArrayOutputStream buffer = start(ProfiledTaskKinds.ALL, BINARY_BAZEL_FORMAT);
     for (int i = 0; i < 100; i++) {
       try (SilentCloseable c = profiler.profile(ProfilerTask.INFO, "outer task " + i)) {
         clock.advanceMillis(1);
@@ -428,14 +458,12 @@ public class ProfilerTest extends FoundationTestCase {
     }
     profiler.stop();
 
-    ProfileInfo info = ProfileInfo.loadProfile(cacheFile);
+    ProfileInfo info = ProfileInfo.loadProfile(new ByteArrayInputStream(buffer.toByteArray()));
     info.calculateStats();
     assertThat(info.isCorruptedOrIncomplete()).isFalse();
 
-    Path corruptedFile = cacheDir.getRelative("profile5bad.dat");
-    FileSystemUtils.writeContent(
-        corruptedFile, Arrays.copyOf(FileSystemUtils.readContent(cacheFile), 2000));
-    info = ProfileInfo.loadProfile(corruptedFile);
+    info = ProfileInfo.loadProfile(
+        new ByteArrayInputStream(Arrays.copyOf(buffer.toByteArray(), 2000)));
     info.calculateStats();
     assertThat(info.isCorruptedOrIncomplete()).isTrue();
     // Since root tasks will appear after nested tasks in the profile file and
@@ -447,9 +475,7 @@ public class ProfilerTest extends FoundationTestCase {
 
   @Test
   public void testUnsupportedProfilerRecord() throws Exception {
-    Path dataFile = cacheDir.getRelative("profile5.dat");
-    profiler.start(ProfiledTaskKinds.ALL, dataFile.getOutputStream(), "phase test", false,
-        BlazeClock.instance(), BlazeClock.instance().nanoTime());
+    ByteArrayOutputStream buffer = start(ProfiledTaskKinds.ALL, BINARY_BAZEL_FORMAT);
     try (SilentCloseable c = profiler.profile(ProfilerTask.INFO, "outer task")) {
       profiler.logEvent(ProfilerTask.PHASE, "inner task");
     }
@@ -459,29 +485,28 @@ public class ProfilerTest extends FoundationTestCase {
     profiler.stop();
 
     // Validate our test profile.
-    ProfileInfo info = ProfileInfo.loadProfile(dataFile);
+    ProfileInfo info = ProfileInfo.loadProfile(new ByteArrayInputStream(buffer.toByteArray()));
     info.calculateStats();
     assertThat(info.isCorruptedOrIncomplete()).isFalse();
     assertThat(info.getStatsForType(ProfilerTask.INFO, info.rootTasksById).count).isEqualTo(3);
     assertThat(info.getStatsForType(ProfilerTask.UNKNOWN, info.rootTasksById).count).isEqualTo(0);
 
     // Now replace "TEST" type with something unsupported - e.g. "XXXX".
-    InputStream in = new InflaterInputStream(dataFile.getInputStream(), new Inflater(false), 65536);
-    byte[] buffer = new byte[60000];
-    int len = in.read(buffer);
-    in.close();
-    assertThat(len).isLessThan(buffer.length); // Validate that file was completely decoded.
-    String content = new String(buffer, ISO_8859_1);
+    byte[] deflated = ByteStreams.toByteArray(new InflaterInputStream(
+        new ByteArrayInputStream(buffer.toByteArray()), new Inflater(false), 65536));
+    String content = new String(deflated, ISO_8859_1);
     int infoIndex = content.indexOf("INFO");
     assertThat(infoIndex).isGreaterThan(0);
     content = content.substring(0, infoIndex) + "XXXX" + content.substring(infoIndex + 4);
-    OutputStream out = new DeflaterOutputStream(dataFile.getOutputStream(),
-        new Deflater(Deflater.BEST_SPEED, false), 65536);
+
+    buffer = new ByteArrayOutputStream();
+    OutputStream out =
+        new DeflaterOutputStream(buffer, new Deflater(Deflater.BEST_SPEED, false), 65536);
     out.write(content.getBytes(ISO_8859_1));
     out.close();
 
     // Validate that XXXX records were classified as UNKNOWN.
-    info = ProfileInfo.loadProfile(dataFile);
+    info = ProfileInfo.loadProfile(new ByteArrayInputStream(buffer.toByteArray()));
     info.calculateStats();
     assertThat(info.isCorruptedOrIncomplete()).isFalse();
     assertThat(info.getStatsForType(ProfilerTask.INFO, info.rootTasksById).count).isEqualTo(0);
@@ -505,10 +530,71 @@ public class ProfilerTest extends FoundationTestCase {
         return initialNanoTime - numNanoTimeCalls.addAndGet(1);
       }
     };
-    Path cacheFile = cacheDir.getRelative("profile1.dat");
-    profiler.start(ProfiledTaskKinds.ALL, cacheFile.getOutputStream(),
-        "testResilenceToNonDecreasingNanoTimes", false, badClock, initialNanoTime);
+    profiler.start(
+        ProfiledTaskKinds.ALL,
+        new ByteArrayOutputStream(),
+        BINARY_BAZEL_FORMAT,
+        "testResilenceToNonDecreasingNanoTimes",
+        false,
+        badClock,
+        initialNanoTime);
     profiler.logSimpleTask(badClock.nanoTime(), ProfilerTask.INFO, "some task");
     profiler.stop();
   }
+
+  /** Checks that the histograms are cleared in the stop call. */
+  @Test
+  public void testEmptyTaskHistograms() throws Exception {
+    startUnbuffered(ProfiledTaskKinds.ALL);
+    profiler.logSimpleTaskDuration(
+        Profiler.nanoTimeMaybe(), Duration.ofSeconds(10), ProfilerTask.INFO, "foo");
+    profiler.stop();
+    ImmutableList<StatRecorder> histograms = profiler.getTasksHistograms();
+    for (StatRecorder recorder : histograms) {
+      assertThat(recorder.isEmpty()).isTrue();
+    }
+  }
+
+  @Test
+  public void testTaskHistograms() throws Exception {
+    startUnbuffered(ProfiledTaskKinds.ALL);
+    profiler.logSimpleTaskDuration(
+        Profiler.nanoTimeMaybe(), Duration.ofSeconds(10), ProfilerTask.INFO, "foo");
+    ImmutableList<StatRecorder> histograms = profiler.getTasksHistograms();
+    StatRecorder infoStatRecorder = histograms.get(ProfilerTask.INFO.ordinal());
+    assertThat(infoStatRecorder.isEmpty()).isFalse();
+    // This is the only provided API to get the contents of the StatRecorder.
+    assertThat(infoStatRecorder.toString()).contains("'INFO'");
+    assertThat(infoStatRecorder.toString()).contains("Count: 1");
+    assertThat(infoStatRecorder.toString()).contains("[8192..16384 ms]");
+    // The stop() call is here because the histograms are cleared in the stop call. See the
+    // documentation of {@link Profiler#getTasksHistograms}.
+    profiler.stop();
+  }
+
+  @Test
+  public void testIOExceptionInOutputStream() throws Exception {
+    OutputStream failingOutputStream = new OutputStream() {
+      @Override
+      public void write(int b) throws IOException {
+        throw new IOException("Expected failure.");
+      }
+    };
+    profiler.start(
+        ProfiledTaskKinds.ALL,
+        failingOutputStream,
+        BINARY_BAZEL_FORMAT,
+        "basic test",
+        false,
+        BlazeClock.instance(),
+        BlazeClock.instance().nanoTime());
+    profiler.logSimpleTaskDuration(
+        Profiler.nanoTimeMaybe(), Duration.ofSeconds(10), ProfilerTask.INFO, "foo");
+    try {
+      profiler.stop();
+      fail();
+    } catch (IOException expected) {
+      assertThat(expected).hasMessageThat().isEqualTo("Expected failure.");
+    }
+  }
 }