[7.2.0] Support profiling in an virtual thread. (#22375) Cherry-pick commits: - 10777215a95224f8dfd163572d98a301862a277e - 10e02a15a6f5b915b7a8d25ad7a4be3481210037
diff --git a/src/main/java/com/google/devtools/build/lib/profiler/JsonTraceFileWriter.java b/src/main/java/com/google/devtools/build/lib/profiler/JsonTraceFileWriter.java index 4206cbe..d9c15f8 100644 --- a/src/main/java/com/google/devtools/build/lib/profiler/JsonTraceFileWriter.java +++ b/src/main/java/com/google/devtools/build/lib/profiler/JsonTraceFileWriter.java
@@ -85,7 +85,9 @@ } public void enqueue(TraceData data) { - if (!metadataPosted.get()) { + // We assign a virtual lane for virtual thread and the metadata for the virtual lane is posted + // at creation time. + if (!Thread.currentThread().isVirtual() && !metadataPosted.get()) { metadataPosted.set(Boolean.TRUE); queue.add(new ThreadMetadata()); }
diff --git a/src/main/java/com/google/devtools/build/lib/profiler/Profiler.java b/src/main/java/com/google/devtools/build/lib/profiler/Profiler.java index a0ad097..24bf3b6 100644 --- a/src/main/java/com/google/devtools/build/lib/profiler/Profiler.java +++ b/src/main/java/com/google/devtools/build/lib/profiler/Profiler.java
@@ -22,8 +22,6 @@ import com.google.common.collect.ImmutableSet; import com.google.common.collect.Iterables; import com.google.common.collect.Maps; -import com.google.common.util.concurrent.ListenableFuture; -import com.google.common.util.concurrent.MoreExecutors; import com.google.devtools.build.lib.actions.ResourceEstimator; import com.google.devtools.build.lib.bugreport.BugReporter; import com.google.devtools.build.lib.clock.Clock; @@ -45,11 +43,12 @@ import java.util.LinkedHashMap; import java.util.List; import java.util.Map; -import java.util.Objects; -import java.util.PriorityQueue; +import java.util.Queue; import java.util.Set; import java.util.UUID; +import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.AtomicReference; import java.util.function.Supplier; @@ -276,8 +275,7 @@ // @ThreadSafe void add(TaskData taskData) { - Extrema<SlowTask> extrema = - extremaAggregators[(int) (Thread.currentThread().getId() % SHARDS)]; + Extrema<SlowTask> extrema = extremaAggregators[(int) (taskData.threadId % SHARDS)]; synchronized (extrema) { extrema.aggregate(new SlowTask(taskData)); } @@ -583,6 +581,8 @@ aggregator.clear(); } } + + multiLaneGenerator.reset(); } /** Returns true iff profiling is currently enabled. */ @@ -624,37 +624,41 @@ * @param type task type * @param description task description. May be stored until end of build. */ - private void logTask( - long threadId, long startTimeNanos, long duration, ProfilerTask type, String description) { - checkNotNull(description); - checkState(!description.isEmpty(), "No description -> not helpful"); - if (duration < 0) { - // See note in Clock#nanoTime, which is used by Profiler#nanoTimeMaybe. - duration = 0; - } + private void logTask(long startTimeNanos, long duration, ProfilerTask type, String description) { + var threadId = borrowLaneAndGetLaneId(); + try { + checkNotNull(description); + checkState(!description.isEmpty(), "No description -> not helpful"); + if (duration < 0) { + // See note in Clock#nanoTime, which is used by Profiler#nanoTimeMaybe. + duration = 0; + } - StatRecorder statRecorder = tasksHistograms[type.ordinal()]; - if (collectTaskHistograms && statRecorder != null) { - statRecorder.addStat((int) Duration.ofNanos(duration).toMillis(), description); - } + StatRecorder statRecorder = tasksHistograms[type.ordinal()]; + if (collectTaskHistograms && statRecorder != null) { + statRecorder.addStat((int) Duration.ofNanos(duration).toMillis(), description); + } - if (isActive() && startTimeNanos >= 0 && isProfiling(type)) { - // Store instance fields as local variables so they are not nulled out from under us by - // #clear. - JsonTraceFileWriter currentWriter = writerRef.get(); - if (wasTaskSlowEnoughToRecord(type, duration)) { - TaskData data = new TaskData(threadId, startTimeNanos, type, description); - data.durationNanos = duration; - if (currentWriter != null) { - currentWriter.enqueue(data); - } + if (isActive() && startTimeNanos >= 0 && isProfiling(type)) { + // Store instance fields as local variables so they are not nulled out from under us by + // #clear. + JsonTraceFileWriter currentWriter = writerRef.get(); + if (wasTaskSlowEnoughToRecord(type, duration)) { + TaskData data = new TaskData(threadId, startTimeNanos, type, description); + data.durationNanos = duration; + if (currentWriter != null) { + currentWriter.enqueue(data); + } - SlowestTaskAggregator aggregator = slowestTasks[type.ordinal()]; + SlowestTaskAggregator aggregator = slowestTasks[type.ordinal()]; - if (aggregator != null) { - aggregator.add(data); + if (aggregator != null) { + aggregator.add(data); + } } } + } finally { + releaseLane(); } } @@ -669,12 +673,7 @@ */ public void logSimpleTask(long startTimeNanos, ProfilerTask type, String description) { if (clock != null) { - logTask( - Thread.currentThread().getId(), - startTimeNanos, - clock.nanoTime() - startTimeNanos, - type, - description); + logTask(startTimeNanos, clock.nanoTime() - startTimeNanos, type, description); } } @@ -692,12 +691,7 @@ */ public void logSimpleTask( long startTimeNanos, long stopTimeNanos, ProfilerTask type, String description) { - logTask( - Thread.currentThread().getId(), - startTimeNanos, - stopTimeNanos - startTimeNanos, - type, - description); + logTask(startTimeNanos, stopTimeNanos - startTimeNanos, type, description); } /** @@ -712,12 +706,12 @@ */ public void logSimpleTaskDuration( long startTimeNanos, Duration duration, ProfilerTask type, String description) { - logTask(Thread.currentThread().getId(), startTimeNanos, duration.toNanos(), type, description); + logTask(startTimeNanos, duration.toNanos(), type, description); } /** Used to log "events" happening at a specific time - tasks with zero duration. */ public void logEventAtTime(long atTimeNanos, ProfilerTask type, String description) { - logTask(Thread.currentThread().getId(), atTimeNanos, 0, type, description); + logTask(atTimeNanos, 0, type, description); } /** Used to log "events" - tasks with zero duration. */ @@ -726,9 +720,16 @@ logEventAtTime(clock.nanoTime(), type, description); } - private SilentCloseable reallyProfile(long laneId, ProfilerTask type, String description) { + private SilentCloseable reallyProfile(ProfilerTask type, String description) { final long startTimeNanos = clock.nanoTime(); - return () -> completeTask(laneId, startTimeNanos, type, description); + long laneId = borrowLaneAndGetLaneId(); + return () -> { + try { + completeTask(laneId, startTimeNanos, type, description); + } finally { + releaseLane(); + } + }; } /** @@ -751,11 +752,7 @@ * @param description task description. May be stored until the end of the build. */ public SilentCloseable profile(ProfilerTask type, String description) { - return profile(Thread.currentThread().getId(), type, description); - } - - private SilentCloseable profile(long laneId, ProfilerTask type, String description) { - return (isActive() && isProfiling(type)) ? reallyProfile(laneId, type, description) : NOP; + return (isActive() && isProfiling(type)) ? reallyProfile(type, description) : NOP; } /** @@ -763,11 +760,7 @@ * profiling. */ public SilentCloseable profile(ProfilerTask type, Supplier<String> description) { - return profile(Thread.currentThread().getId(), type, description); - } - - private SilentCloseable profile(long laneId, ProfilerTask type, Supplier<String> description) { - return (isActive() && isProfiling(type)) ? reallyProfile(laneId, type, description.get()) : NOP; + return (isActive() && isProfiling(type)) ? reallyProfile(type, description.get()) : NOP; } /** @@ -807,15 +800,21 @@ checkNotNull(description); if (isActive() && isProfiling(type)) { final long startTimeNanos = clock.nanoTime(); - return () -> + var laneId = borrowLaneAndGetLaneId(); + return () -> { + try { completeAction( - Thread.currentThread().getId(), + laneId, startTimeNanos, type, description, mnemonic, includePrimaryOutput ? primaryOutput : null, includeTargetLabel ? targetLabel : null); + } finally { + releaseLane(); + } + }; } else { return NOP; } @@ -833,6 +832,15 @@ || (type == ProfilerTask.INFO && "discoverInputs".equals(taskData.description)); } + public void completeTask(long startTimeNanos, ProfilerTask type, String description) { + var laneId = borrowLaneAndGetLaneId(); + try { + completeTask(laneId, startTimeNanos, type, description); + } finally { + releaseLane(); + } + } + /** Records the end of the task. */ private void completeTask( long laneId, long startTimeNanos, ProfilerTask type, String description) { @@ -905,136 +913,131 @@ } } - static class ProfilerTaskType { - private final String format; + private final AtomicLong nextLaneId = new AtomicLong(1_000_000); + private final MultiLaneGenerator multiLaneGenerator = new MultiLaneGenerator(); - public ProfilerTaskType(String format) { - this.format = format; + private class MultiLaneGenerator { + private final Map<String, LaneGenerator> laneGenerators = Maps.newConcurrentMap(); + + private Lane acquire(String prefix) { + checkState(isActive()); + var laneGenerator = + laneGenerators.computeIfAbsent(prefix, unused -> new LaneGenerator(prefix)); + return laneGenerator.acquire(); } - public String getName(long index) { - return String.format(format, index); + private void release(String prefix, Lane lane) { + checkState(isActive()); + var laneGenerator = checkNotNull(laneGenerators.get(prefix)); + laneGenerator.release(lane); + } + + private void reset() { + multiLaneGenerator.laneGenerators.clear(); + } + } + + private static class Lane implements Comparable<Lane> { + private final long id; + private int refCount; + + private Lane(long id) { + this.id = id; } @Override - public boolean equals(Object obj) { - if (!(obj instanceof ProfilerTaskType)) { - return false; + public int compareTo(Lane o) { + return Long.compare(id, o.id); + } + } + + private class LaneGenerator { + private final String prefix; + + private final Queue<Lane> availableLanes = new ConcurrentLinkedQueue<>(); + + private final AtomicInteger count = new AtomicInteger(0); + + private LaneGenerator(String prefix) { + this.prefix = prefix; + } + + public Lane acquire() { + var lane = availableLanes.poll(); + // It might create more virtual lanes, but it's fine for our purpose. + if (lane == null) { + long newLaneId = nextLaneId.getAndIncrement(); + int newLaneIndex = count.getAndIncrement(); + String newLaneName = prefix + newLaneIndex + " (Virtual)"; + var threadMetadata = new ThreadMetadata(newLaneName, newLaneId); + var writer = Profiler.this.writerRef.get(); + if (writer != null) { + writer.enqueue(threadMetadata); + } + lane = new Lane(newLaneId); } + return lane; + } - if (this == obj) { - return true; + public void release(Lane lane) { + availableLanes.offer(lane); + } + } + + private final ThreadLocal<String> virtualThreadPrefix = + ThreadLocal.withInitial(this::guessThreadPrefix); + private final ThreadLocal<Lane> borrowedLane = + ThreadLocal.withInitial( + () -> { + var prefix = virtualThreadPrefix.get(); + var lane = multiLaneGenerator.acquire(prefix); + checkState(lane.refCount == 0); + return lane; + }); + + private long borrowLaneAndGetLaneId() { + var currentThread = Thread.currentThread(); + var threadId = currentThread.threadId(); + if (!currentThread.isVirtual() || !isActive()) { + return threadId; + } + + var lane = borrowedLane.get(); + lane.refCount += 1; + return lane.id; + } + + private void releaseLane() { + var currentThread = Thread.currentThread(); + if (!currentThread.isVirtual() || !isActive()) { + return; + } + + var lane = borrowedLane.get(); + lane.refCount -= 1; + checkState(lane.refCount >= 0); + if (lane.refCount == 0) { + borrowedLane.remove(); + var prefix = virtualThreadPrefix.get(); + multiLaneGenerator.release(prefix, lane); + } + } + + private String guessThreadPrefix() { + var currentThread = Thread.currentThread(); + checkState(currentThread.isVirtual()); + var threadName = currentThread.getName(); + + // Assume the thread name has format "prefix%d" + for (int i = threadName.length() - 1; i > 0; i--) { + var ch = threadName.charAt(i); + if (ch < '0' || ch > '9') { + if (i < threadName.length() - 1) { + return threadName.substring(0, i + 1); + } } - - var that = (ProfilerTaskType) obj; - return Objects.equals(this.format, that.format); } - @Override - public int hashCode() { - return Objects.hashCode(format); - } - } - - public ProfilerTaskType createTaskType(String format) { - return new ProfilerTaskType(format); - } - - /** - * A profiler has a specific lane id. All trace events emitted by it are associated with its lane - * id instead of Thread.currentThread().getId(). - */ - public static class ScopedProfiler { - private final boolean active; - private final long laneId; - - public ScopedProfiler(boolean active, long laneId) { - this.active = active; - this.laneId = laneId; - } - - public SilentCloseable profile(String description) { - if (!active) { - return () -> {}; - } - return Profiler.instance().profile(laneId, ProfilerTask.INFO, description); - } - } - - /** - * An interface used to supply a {@link ListenableFuture} with the given {@link ScopedProfiler}. - */ - public interface FutureSupplier<T> { - ListenableFuture<T> get(ScopedProfiler profiler); - } - - public <T> ListenableFuture<T> profileAsync( - ProfilerTaskType type, String description, FutureSupplier<T> futureSupplier) { - if (!(isActive() && isProfiling(ProfilerTask.INFO))) { - return futureSupplier.get(new ScopedProfiler(/* active= */ false, 0)); - } - - long laneId = laneIdGenerator.acquire(type); - final long startTimeNanos = clock.nanoTime(); - var scopedProfiler = new ScopedProfiler(/* active= */ true, laneId); - var future = futureSupplier.get(scopedProfiler); - future.addListener( - () -> { - long endTimeNanos = clock.nanoTime(); - long duration = endTimeNanos - startTimeNanos; - recordTask( - new TaskData(laneId, startTimeNanos, duration, ProfilerTask.INFO, description)); - laneIdGenerator.release(type, laneId); - }, - MoreExecutors.directExecutor()); - return future; - } - - private static final long LANE_ID_BASE = 1_000_000; - private final AtomicLong nextLaneId = new AtomicLong(LANE_ID_BASE); - private final TaskTypeLaneIdGenerator laneIdGenerator = new TaskTypeLaneIdGenerator(); - - private class TaskTypeLaneIdGenerator { - private final Map<ProfilerTaskType, LaneIdGenerator> typeToLaneIdGenerator = - Maps.newConcurrentMap(); - - public long acquire(ProfilerTaskType type) { - var laneIdGenerator = - typeToLaneIdGenerator.computeIfAbsent(type, unused -> new LaneIdGenerator(type)); - return laneIdGenerator.acquire(); - } - - public void release(ProfilerTaskType type, long laneId) { - var laneIdGenerator = checkNotNull(typeToLaneIdGenerator.get(type)); - laneIdGenerator.release(laneId); - } - } - - private class LaneIdGenerator { - private final ProfilerTaskType type; - private final PriorityQueue<Long> availableLaneIds = new PriorityQueue<>(); - - private int count = 0; - - private LaneIdGenerator(ProfilerTaskType type) { - this.type = type; - } - - public synchronized long acquire() { - if (!availableLaneIds.isEmpty()) { - return availableLaneIds.poll(); - } - var newLaneId = Profiler.this.nextLaneId.getAndIncrement(); - var threadMetadata = new ThreadMetadata(type.getName(count++), newLaneId, LANE_ID_BASE); - var writer = Profiler.this.writerRef.get(); - if (writer != null) { - writer.enqueue(threadMetadata); - } - return newLaneId; - } - - public synchronized void release(long laneId) { - availableLaneIds.add(laneId); - } + return "Other"; } }
diff --git a/src/main/java/com/google/devtools/build/lib/profiler/ThreadMetadata.java b/src/main/java/com/google/devtools/build/lib/profiler/ThreadMetadata.java index 99b1a39..54ca49e 100644 --- a/src/main/java/com/google/devtools/build/lib/profiler/ThreadMetadata.java +++ b/src/main/java/com/google/devtools/build/lib/profiler/ThreadMetadata.java
@@ -24,6 +24,12 @@ private final long threadId; private final long sortIndex; + public ThreadMetadata(String readableName, long threadId) { + this.readableName = readableName; + this.threadId = threadId; + this.sortIndex = getSortIndex(readableName); + } + public ThreadMetadata(String readableName, long threadId, long sortIndex) { this.readableName = readableName; this.threadId = threadId;