Use `refCount` based method for virtual lane.
... so no [de]registration is needed for `Profiler` to work with virtual thread.
`Profiler` now assumes virtual thread has name pattern like `prefix%d` and tries to extract the prefix into thread local when the first time it needs a virtual lane.
Remove the `synchronized` block since it harms the performance.
PiperOrigin-RevId: 627697201
Change-Id: I35b48d1b30b8d95e397b559546c47aa5b60c7c40
diff --git a/src/main/java/com/google/devtools/build/lib/bazel/BazelRepositoryModule.java b/src/main/java/com/google/devtools/build/lib/bazel/BazelRepositoryModule.java
index 4757485..70a1500 100644
--- a/src/main/java/com/google/devtools/build/lib/bazel/BazelRepositoryModule.java
+++ b/src/main/java/com/google/devtools/build/lib/bazel/BazelRepositoryModule.java
@@ -78,7 +78,6 @@
import com.google.devtools.build.lib.cmdline.RepositoryName;
import com.google.devtools.build.lib.events.Event;
import com.google.devtools.build.lib.pkgcache.PackageOptions;
-import com.google.devtools.build.lib.profiler.Profiler;
import com.google.devtools.build.lib.rules.repository.LocalRepositoryFunction;
import com.google.devtools.build.lib.rules.repository.LocalRepositoryRule;
import com.google.devtools.build.lib.rules.repository.NewLocalRepositoryFunction;
@@ -333,9 +332,7 @@
Executors.class
.getDeclaredMethod("newThreadPerTaskExecutor", ThreadFactory.class)
.invoke(
- null,
- Profiler.instance()
- .profileableVirtualThreadFactory("starlark-repository-")));
+ null, Thread.ofVirtual().name("starlark-repository-", 0).factory()));
} catch (ReflectiveOperationException e) {
if (repoOptions.workerForRepoFetching == RepositoryOptions.WorkerForRepoFetching.AUTO) {
starlarkRepositoryFunction.setWorkerExecutorService(null);
diff --git a/src/main/java/com/google/devtools/build/lib/dynamic/DynamicExecutionModule.java b/src/main/java/com/google/devtools/build/lib/dynamic/DynamicExecutionModule.java
index c62644d..63fb097 100644
--- a/src/main/java/com/google/devtools/build/lib/dynamic/DynamicExecutionModule.java
+++ b/src/main/java/com/google/devtools/build/lib/dynamic/DynamicExecutionModule.java
@@ -31,7 +31,6 @@
import com.google.devtools.build.lib.exec.ExecutionPolicy;
import com.google.devtools.build.lib.exec.SpawnStrategyRegistry;
import com.google.devtools.build.lib.exec.local.LocalExecutionOptions;
-import com.google.devtools.build.lib.profiler.Profiler;
import com.google.devtools.build.lib.runtime.BlazeModule;
import com.google.devtools.build.lib.runtime.Command;
import com.google.devtools.build.lib.runtime.CommandEnvironment;
@@ -81,7 +80,7 @@
if (buildRequestOptions != null && buildRequestOptions.useAsyncExecution) {
executorService =
Executors.newThreadPerTaskExecutor(
- Profiler.instance().profileableVirtualThreadFactory("dynamic-execution-thread-"));
+ Thread.ofVirtual().name("dynamic-execution-thread-", 0).factory());
} else {
executorService =
Executors.newCachedThreadPool(
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 d5fde8f..4bd8089 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.clock.Clock;
import com.google.devtools.build.lib.collect.Extrema;
import com.google.devtools.build.lib.concurrent.ThreadSafety.ThreadCompatible;
@@ -42,18 +40,17 @@
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.ThreadFactory;
+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;
import java.util.zip.GZIPOutputStream;
import javax.annotation.Nullable;
-import javax.annotation.concurrent.GuardedBy;
/**
* Blaze internal profiler. Provides facility to report various Blaze tasks and store them
@@ -573,6 +570,8 @@
aggregator.clear();
}
}
+
+ multiLaneGenerator.reset();
}
/** Returns true iff profiling is currently enabled. */
@@ -614,37 +613,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();
}
}
@@ -659,7 +662,7 @@
*/
public void logSimpleTask(long startTimeNanos, ProfilerTask type, String description) {
if (clock != null) {
- logTask(getLaneId(), startTimeNanos, clock.nanoTime() - startTimeNanos, type, description);
+ logTask(startTimeNanos, clock.nanoTime() - startTimeNanos, type, description);
}
}
@@ -677,7 +680,7 @@
*/
public void logSimpleTask(
long startTimeNanos, long stopTimeNanos, ProfilerTask type, String description) {
- logTask(getLaneId(), startTimeNanos, stopTimeNanos - startTimeNanos, type, description);
+ logTask(startTimeNanos, stopTimeNanos - startTimeNanos, type, description);
}
/**
@@ -692,12 +695,12 @@
*/
public void logSimpleTaskDuration(
long startTimeNanos, Duration duration, ProfilerTask type, String description) {
- logTask(getLaneId(), 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(getLaneId(), atTimeNanos, 0, type, description);
+ logTask(atTimeNanos, 0, type, description);
}
/** Used to log "events" - tasks with zero duration. */
@@ -706,9 +709,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();
+ }
+ };
}
/**
@@ -731,11 +741,7 @@
* @param description task description. May be stored until the end of the build.
*/
public SilentCloseable profile(ProfilerTask type, String description) {
- return profile(getLaneId(), 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;
}
/**
@@ -743,11 +749,7 @@
* profiling.
*/
public SilentCloseable profile(ProfilerTask type, Supplier<String> description) {
- return profile(getLaneId(), 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;
}
/**
@@ -787,15 +789,21 @@
checkNotNull(description);
if (isActive() && isProfiling(type)) {
final long startTimeNanos = clock.nanoTime();
- return () ->
+ var laneId = borrowLaneAndGetLaneId();
+ return () -> {
+ try {
completeAction(
- getLaneId(),
+ laneId,
startTimeNanos,
type,
description,
mnemonic,
includePrimaryOutput ? primaryOutput : null,
includeTargetLabel ? targetLabel : null);
+ } finally {
+ releaseLane();
+ }
+ };
} else {
return NOP;
}
@@ -813,7 +821,12 @@
}
public void completeTask(long startTimeNanos, ProfilerTask type, String description) {
- completeTask(getLaneId(), startTimeNanos, type, description);
+ var laneId = borrowLaneAndGetLaneId();
+ try {
+ completeTask(laneId, startTimeNanos, type, description);
+ } finally {
+ releaseLane();
+ }
}
/** Records the end of the task. */
@@ -894,110 +907,6 @@
}
}
- static class ProfilerTaskType {
- private final String format;
-
- public ProfilerTaskType(String format) {
- this.format = format;
- }
-
- public String getName(long index) {
- return String.format(format, index);
- }
-
- @Override
- public boolean equals(Object obj) {
- if (!(obj instanceof ProfilerTaskType that)) {
- return false;
- }
-
- if (this == obj) {
- return true;
- }
-
- 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(
- String prefix, String description, FutureSupplier<T> futureSupplier) {
- if (!(isActive() && isProfiling(ProfilerTask.INFO))) {
- return futureSupplier.get(new ScopedProfiler(/* active= */ false, 0));
- }
-
- var lane = multiLaneGenerator.acquire(prefix);
- final long startTimeNanos = clock.nanoTime();
- var scopedProfiler = new ScopedProfiler(/* active= */ true, lane.id());
- var future = futureSupplier.get(scopedProfiler);
- future.addListener(
- () -> {
- long endTimeNanos = clock.nanoTime();
- long duration = endTimeNanos - startTimeNanos;
- recordTask(
- new TaskData(lane.id(), startTimeNanos, duration, ProfilerTask.INFO, description));
- multiLaneGenerator.release(prefix, lane);
- },
- MoreExecutors.directExecutor());
- return future;
- }
-
- private final ThreadLocal<String> virtualThreadPrefix = ThreadLocal.withInitial(() -> null);
- private final ThreadLocal<Lane> borrowedLane = ThreadLocal.withInitial(() -> null);
-
- private void registerVirtualThread(String prefix) {
- var thread = Thread.currentThread();
- var threadId = thread.threadId();
- virtualThreadPrefix.set(prefix);
- thread.setName(prefix + threadId);
- }
-
- private void deregisterVirtualThread() {
- var prefix = checkNotNull(virtualThreadPrefix.get());
- virtualThreadPrefix.remove();
- var lane = borrowedLane.get();
- if (lane != null) {
- borrowedLane.remove();
- multiLaneGenerator.release(prefix, lane);
- }
- }
-
private final AtomicLong nextLaneId = new AtomicLong(1_000_000);
private final MultiLaneGenerator multiLaneGenerator = new MultiLaneGenerator();
@@ -1016,9 +925,20 @@
var laneGenerator = checkNotNull(laneGenerators.get(prefix));
laneGenerator.release(lane);
}
+
+ private void reset() {
+ multiLaneGenerator.laneGenerators.clear();
+ }
}
- private record Lane(long id) implements Comparable<Lane> {
+ private static class Lane implements Comparable<Lane> {
+ private final long id;
+ private int refCount;
+
+ private Lane(long id) {
+ this.id = id;
+ }
+
@Override
public int compareTo(Lane o) {
return Long.compare(id, o.id);
@@ -1028,74 +948,90 @@
private class LaneGenerator {
private final String prefix;
- @GuardedBy("this")
- private final PriorityQueue<Lane> availableLanes = new PriorityQueue<>();
+ private final Queue<Lane> availableLanes = new ConcurrentLinkedQueue<>();
- @GuardedBy("this")
- private int count = 0;
+ private final AtomicInteger count = new AtomicInteger(0);
private LaneGenerator(String prefix) {
this.prefix = prefix;
}
public Lane acquire() {
- long newLaneId;
- String newLaneName;
- synchronized (this) {
- if (!availableLanes.isEmpty()) {
- return availableLanes.poll();
+ 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);
}
-
- newLaneId = nextLaneId.getAndIncrement();
- int newLaneIndex = count++;
- newLaneName = prefix + newLaneIndex + " (Virtual)";
+ lane = new Lane(newLaneId);
}
-
- var threadMetadata = new ThreadMetadata(newLaneName, newLaneId);
- var writer = Profiler.this.writerRef.get();
- if (writer != null) {
- writer.enqueue(threadMetadata);
- }
- return new Lane(newLaneId);
+ return lane;
}
- public synchronized void release(Lane lane) {
- availableLanes.add(lane);
+ public void release(Lane lane) {
+ availableLanes.offer(lane);
}
}
- private long getLaneId() {
+ 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()) {
+ if (!currentThread.isVirtual() || !isActive()) {
return threadId;
}
var lane = borrowedLane.get();
- if (lane == null) {
- var prefix = virtualThreadPrefix.get();
- checkNotNull(
- prefix,
- "Current virtual thread is not registered. Did you use"
- + " Profiler#profileableVirtualThreadFactor to create a VirtualThread?");
- lane = multiLaneGenerator.acquire(prefix);
- borrowedLane.set(lane);
- }
- return lane.id();
+ lane.refCount += 1;
+ return lane.id;
}
- public ThreadFactory profileableVirtualThreadFactory(String prefix) {
- return r ->
- Thread.ofVirtual()
- .unstarted(
- () -> {
- var profiler = Profiler.instance();
- profiler.registerVirtualThread(prefix);
- try {
- r.run();
- } finally {
- profiler.deregisterVirtualThread();
- }
- });
+ 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);
+ }
+ }
+ }
+
+ return "Other";
}
}
diff --git a/src/main/java/com/google/devtools/build/lib/runtime/QuiescingExecutorsImpl.java b/src/main/java/com/google/devtools/build/lib/runtime/QuiescingExecutorsImpl.java
index 325d1bb..2cafbbb 100644
--- a/src/main/java/com/google/devtools/build/lib/runtime/QuiescingExecutorsImpl.java
+++ b/src/main/java/com/google/devtools/build/lib/runtime/QuiescingExecutorsImpl.java
@@ -25,7 +25,6 @@
import com.google.devtools.build.lib.concurrent.QuiescingExecutor;
import com.google.devtools.build.lib.concurrent.QuiescingExecutors;
import com.google.devtools.build.lib.pkgcache.PackageOptions;
-import com.google.devtools.build.lib.profiler.Profiler;
import com.google.devtools.build.skyframe.ParallelEvaluatorErrorClassifier;
import com.google.devtools.common.options.OptionsProvider;
import java.util.concurrent.Executors;
@@ -146,8 +145,7 @@
/* parallelism= */ cpuHeavySkyKeysThreadPoolSize, SKYFRAME_EVALUATOR_CPU_HEAVY),
useAsyncExecution
? Executors.newThreadPerTaskExecutor(
- Profiler.instance()
- .profileableVirtualThreadFactory(SKYFRAME_EVALUATOR_EXECUTION + "-"))
+ Thread.ofVirtual().name(SKYFRAME_EVALUATOR_EXECUTION + "-", 0).factory())
: AbstractQueueVisitor.createExecutorService(
/* parallelism= */ executionParallelism, SKYFRAME_EVALUATOR_EXECUTION),
ExceptionHandlingMode.FAIL_FAST,
