From 6949b2706f2fbde9b39705608afa8695bfa1fefb Mon Sep 17 00:00:00 2001 From: Beanes Date: Sat, 19 Sep 2026 17:54:40 +0200 Subject: [PATCH] Add lock profiler mode (--lock) Adds a new sampler mode backed by async-profiler's "lock" event, which measures the time threads spend waiting on contended monitors (synchronized) and parked on java.util.concurrent locks. - new SamplerMode.LOCK / proto SamplerMode.LOCK = 2 - SampleCollector.Lock reads jdk.JavaMonitorEnter + jdk.ThreadPark events and converts their duration from ticks to nanoseconds - `/spark profiler start --lock`, with --interval used as the contention threshold in milliseconds (default 10us, same as async-profiler) - --ignore-sleeping is disabled in lock mode, as parked threads are exactly what is being measured --- .../common/command/modules/SamplerModule.java | 14 +- .../spark/common/sampler/SamplerBuilder.java | 41 ++++- .../spark/common/sampler/SamplerMode.java | 15 +- .../sampler/async/AsyncProfilerAccess.java | 24 ++- .../sampler/async/AsyncProfilerJob.java | 2 +- .../common/sampler/async/SampleCollector.java | 58 ++++++- .../common/sampler/async/jfr/JfrReader.java | 2 +- .../src/main/proto/spark/spark_sampler.proto | 1 + .../spark/common/sampler/LockSamplerTest.java | 152 ++++++++++++++++++ 9 files changed, 289 insertions(+), 20 deletions(-) create mode 100644 spark-common/src/test/java/me/lucko/spark/common/sampler/LockSamplerTest.java diff --git a/spark-common/src/main/java/me/lucko/spark/common/command/modules/SamplerModule.java b/spark-common/src/main/java/me/lucko/spark/common/command/modules/SamplerModule.java index 72b3100c744..cd8ee23a4ce 100644 --- a/spark-common/src/main/java/me/lucko/spark/common/command/modules/SamplerModule.java +++ b/spark-common/src/main/java/me/lucko/spark/common/command/modules/SamplerModule.java @@ -87,6 +87,7 @@ public void registerCommands(Consumer consumer) { .argumentUsage("start", "only-ticks-over", "tick length millis") .argumentUsage("start", "interval", "interval millis") .argumentUsage("start", "alloc", null) + .argumentUsage("start", "lock", null) .argumentUsage("stop", "", null) .argumentUsage("cancel", "", null) .executor(this::profiler) @@ -101,7 +102,7 @@ public void registerCommands(Consumer consumer) { } if (subCommand.equals("start")) { opts = new ArrayList<>(Arrays.asList("--timeout", "--regex", "--combine-all", - "--not-combined", "--interval", "--only-ticks-over", "--force-java-sampler", "--alloc", "--alloc-live-only")); + "--not-combined", "--interval", "--only-ticks-over", "--force-java-sampler", "--alloc", "--alloc-live-only", "--lock")); opts.removeAll(arguments); opts.add("--thread"); // allowed multiple times } @@ -183,7 +184,14 @@ private void profilerStart(SparkPlatform platform, CommandSender sender, Command "Consider setting a timeout value over 30 seconds.")); } - SamplerMode mode = arguments.boolFlag("alloc") ? SamplerMode.ALLOCATION : SamplerMode.EXECUTION; + boolean alloc = arguments.boolFlag("alloc"); + boolean lock = arguments.boolFlag("lock"); + if (alloc && lock) { + resp.replyPrefixed(text("The --alloc and --lock flags cannot be used together. Please choose one profiling mode.", RED)); + return; + } + + SamplerMode mode = alloc ? SamplerMode.ALLOCATION : lock ? SamplerMode.LOCK : SamplerMode.EXECUTION; boolean allocLiveOnly = arguments.boolFlag("alloc-live-only"); double interval = arguments.doubleFlag("interval"); @@ -257,7 +265,7 @@ private void profilerStart(SparkPlatform platform, CommandSender sender, Command platform.getSamplerContainer().setActiveSampler(sampler); resp.broadcastPrefixed(text() - .append(text((mode == SamplerMode.ALLOCATION ? "Allocation Profiler" : "Profiler") + " is now running!", GOLD)) + .append(text((mode == SamplerMode.ALLOCATION ? "Allocation Profiler" : mode == SamplerMode.LOCK ? "Lock Profiler" : "Profiler") + " is now running!", GOLD)) .append(space()) .append(text("(" + (sampler instanceof AsyncSampler ? "async" : "built-in java") + ")", DARK_GRAY)) .build() diff --git a/spark-common/src/main/java/me/lucko/spark/common/sampler/SamplerBuilder.java b/spark-common/src/main/java/me/lucko/spark/common/sampler/SamplerBuilder.java index 2023063830e..6f1f422167f 100644 --- a/spark-common/src/main/java/me/lucko/spark/common/sampler/SamplerBuilder.java +++ b/spark-common/src/main/java/me/lucko/spark/common/sampler/SamplerBuilder.java @@ -39,7 +39,7 @@ public class SamplerBuilder { private SamplerMode mode = SamplerMode.EXECUTION; - private double samplingInterval = -1; // milliseconds or bytes depending on the mode + private double samplingInterval = -1; // milliseconds (execution, lock) or bytes (allocation) depending on the mode private boolean ignoreSleeping = false; private boolean forceJavaSampler = false; private boolean allocLiveOnly = false; @@ -127,23 +127,48 @@ public Sampler start(SparkPlatform platform) throws UnsupportedOperationExceptio } } + boolean ignoreSleeping = this.ignoreSleeping; + + if (this.mode == SamplerMode.LOCK) { + if (this.forceJavaSampler) { + throw new UnsupportedOperationException("Lock profiling is not supported by the built-in Java sampler."); + } + if (!canUseAsyncProfiler || !asyncProfiler.checkLockProfilingSupported(platform)) { + throw new UnsupportedOperationException("Lock profiling is not supported on your system. Check the console for more info."); + } + if (ignoreSleeping) { + // threads waiting for a lock are often parked, which is exactly what we want to measure! + platform.getPlugin().log(Level.WARNING, "Ignoring sleeping threads is not supported in lock profiling mode. Sleeping threads will be included in the results."); + ignoreSleeping = false; + } + } + if (this.forceJavaSampler) { canUseAsyncProfiler = false; } // microseconds or bytes depending on the mode - int interval = (int) (this.mode == SamplerMode.EXECUTION ? - this.samplingInterval * 1000d : // convert to microseconds - this.samplingInterval + int interval = (int) (this.mode == SamplerMode.ALLOCATION ? + this.samplingInterval : + this.samplingInterval * 1000d // convert to microseconds ); - SamplerSettings settings = new SamplerSettings(interval, this.threadDumper, this.threadGrouper.get(), this.autoEndTime, this.background, this.ignoreSleeping); + SamplerSettings settings = new SamplerSettings(interval, this.threadDumper, this.threadGrouper.get(), this.autoEndTime, this.background, ignoreSleeping); Sampler sampler; if (canUseAsyncProfiler) { - SampleCollector collector = this.mode == SamplerMode.ALLOCATION - ? new SampleCollector.Allocation(interval, this.allocLiveOnly) - : new SampleCollector.Execution(interval); + SampleCollector collector; + switch (this.mode) { + case ALLOCATION: + collector = new SampleCollector.Allocation(interval, this.allocLiveOnly); + break; + case LOCK: + collector = new SampleCollector.Lock(interval); + break; + default: + collector = new SampleCollector.Execution(interval); + break; + } sampler = onlyTicksOverMode ? new AsyncSampler(platform, settings, collector, this.ticksOver) diff --git a/spark-common/src/main/java/me/lucko/spark/common/sampler/SamplerMode.java b/spark-common/src/main/java/me/lucko/spark/common/sampler/SamplerMode.java index f9a6e4167a2..0deedc4ead1 100644 --- a/spark-common/src/main/java/me/lucko/spark/common/sampler/SamplerMode.java +++ b/spark-common/src/main/java/me/lucko/spark/common/sampler/SamplerMode.java @@ -42,13 +42,22 @@ public enum SamplerMode { }, 524287, // 512 KiB SamplerMetadata.SamplerMode.ALLOCATION + ), + + LOCK( + value -> { + // convert the duration from nanoseconds -> milliseconds + return value / 1000000d; + }, + 0.01, // ms (10 microseconds) + SamplerMetadata.SamplerMode.LOCK ); private final LongToDoubleFunction valueTransformer; - private final int defaultInterval; + private final double defaultInterval; private final SamplerMetadata.SamplerMode proto; - SamplerMode(LongToDoubleFunction valueTransformer, int defaultInterval, SamplerMetadata.SamplerMode proto) { + SamplerMode(LongToDoubleFunction valueTransformer, double defaultInterval, SamplerMetadata.SamplerMode proto) { this.valueTransformer = valueTransformer; this.defaultInterval = defaultInterval; this.proto = proto; @@ -58,7 +67,7 @@ public LongToDoubleFunction valueTransformer() { return this.valueTransformer; } - public int defaultInterval() { + public double defaultInterval() { return this.defaultInterval; } diff --git a/spark-common/src/main/java/me/lucko/spark/common/sampler/async/AsyncProfilerAccess.java b/spark-common/src/main/java/me/lucko/spark/common/sampler/async/AsyncProfilerAccess.java index 93cba859fe4..9e7bbd901e1 100644 --- a/spark-common/src/main/java/me/lucko/spark/common/sampler/async/AsyncProfilerAccess.java +++ b/spark-common/src/main/java/me/lucko/spark/common/sampler/async/AsyncProfilerAccess.java @@ -62,6 +62,8 @@ public static synchronized AsyncProfilerAccess getInstance(SparkPlatform platfor private final ProfilingEvent profilingEvent; /** The event to use for allocation profiling */ private final ProfilingEvent allocationProfilingEvent; + /** The event to use for lock profiling */ + private final ProfilingEvent lockProfilingEvent; /** If profiler is null, contains the reason why setup failed */ private final Exception setupException; @@ -70,6 +72,7 @@ public static synchronized AsyncProfilerAccess getInstance(SparkPlatform platfor AsyncProfiler profiler; ProfilingEvent profilingEvent = null; ProfilingEvent allocationProfilingEvent = null; + ProfilingEvent lockProfilingEvent = null; Exception setupException = null; try { @@ -81,6 +84,11 @@ public static synchronized AsyncProfilerAccess getInstance(SparkPlatform platfor allocationProfilingEvent = ProfilingEvent.ALLOC; } + // lock profiler + if (supportedEvents.contains(ProfilingEvent.LOCK)) { + lockProfilingEvent = ProfilingEvent.LOCK; + } + // normal profiler if (supportedEvents.contains(ProfilingEvent.WALL)) { profilingEvent = ProfilingEvent.WALL; @@ -96,6 +104,7 @@ public static synchronized AsyncProfilerAccess getInstance(SparkPlatform platfor this.profiler = profiler; this.profilingEvent = profilingEvent; this.allocationProfilingEvent = allocationProfilingEvent; + this.lockProfilingEvent = lockProfilingEvent; this.setupException = setupException; } @@ -114,6 +123,10 @@ public ProfilingEvent getAllocationProfilingEvent() { return this.allocationProfilingEvent; } + public ProfilingEvent getLockProfilingEvent() { + return this.lockProfilingEvent; + } + public boolean checkSupported(SparkPlatform platform) { if (this.setupException != null) { if (this.setupException instanceof UnsupportedSystemException) { @@ -147,6 +160,14 @@ public boolean checkAllocationProfilingSupported(SparkPlatform platform) { return supported; } + public boolean checkLockProfilingSupported(SparkPlatform platform) { + boolean supported = this.lockProfilingEvent != null; + if (!supported && this.profiler != null) { + platform.getPlugin().log(Level.WARNING, "The lock profiling mode is not supported on your system."); + } + return supported; + } + public String getVersion() { return this.profiler.getVersion(); } @@ -216,7 +237,8 @@ private static Set getSupportedEvents(AsyncProfiler profiler) { public enum ProfilingEvent { WALL(Events.WALL), - ALLOC(Events.ALLOC); + ALLOC(Events.ALLOC), + LOCK(Events.LOCK); private final String id; diff --git a/spark-common/src/main/java/me/lucko/spark/common/sampler/async/AsyncProfilerJob.java b/spark-common/src/main/java/me/lucko/spark/common/sampler/async/AsyncProfilerJob.java index 4339d06b858..4d8155fc6c6 100644 --- a/spark-common/src/main/java/me/lucko/spark/common/sampler/async/AsyncProfilerJob.java +++ b/spark-common/src/main/java/me/lucko/spark/common/sampler/async/AsyncProfilerJob.java @@ -250,7 +250,7 @@ private void readSegments(JfrReader reader, SampleCo continue; } - long value = collector.measure(sample); + long value = collector.measure(reader, sample); // parse the segment and give it to the data aggregator ProfileSegment segment = ProfileSegment.parseSegment(reader, sample, threadName, value); diff --git a/spark-common/src/main/java/me/lucko/spark/common/sampler/async/SampleCollector.java b/spark-common/src/main/java/me/lucko/spark/common/sampler/async/SampleCollector.java index 3c3cddc9ad7..0028a70589d 100644 --- a/spark-common/src/main/java/me/lucko/spark/common/sampler/async/SampleCollector.java +++ b/spark-common/src/main/java/me/lucko/spark/common/sampler/async/SampleCollector.java @@ -23,7 +23,9 @@ import com.google.common.collect.ImmutableList; import me.lucko.spark.common.sampler.SamplerMode; import me.lucko.spark.common.sampler.async.AsyncProfilerAccess.ProfilingEvent; +import me.lucko.spark.common.sampler.async.jfr.JfrReader; import me.lucko.spark.common.sampler.async.jfr.JfrReader.AllocationSample; +import me.lucko.spark.common.sampler.async.jfr.JfrReader.ContendedLock; import me.lucko.spark.common.sampler.async.jfr.JfrReader.Event; import me.lucko.spark.common.sampler.async.jfr.JfrReader.ExecutionSample; @@ -55,10 +57,11 @@ public interface SampleCollector { /** * Gets the measurements for a given event * + * @param reader the reader the event was read from * @param event the event * @return the measurement */ - long measure(E event); + long measure(JfrReader reader, E event); /** * Gets the mode for the collector. @@ -94,7 +97,7 @@ public Class eventClass() { } @Override - public long measure(ExecutionSample event) { + public long measure(JfrReader reader, ExecutionSample event) { return event.value() * this.intervalMicroseconds; } @@ -140,7 +143,7 @@ public Class eventClass() { } @Override - public long measure(AllocationSample event) { + public long measure(JfrReader reader, AllocationSample event) { return event.value(); } @@ -150,4 +153,53 @@ public SamplerMode getMode() { } } + /** + * Sample collector for lock (contention) profiles. + * + *

Measures the time threads spend waiting to enter contended monitors + * (synchronized blocks/methods) or parked waiting for + * {@code java.util.concurrent} locks.

+ */ + final class Lock implements SampleCollector { + private final int thresholdMicroseconds; // time in microseconds + + public Lock(int thresholdMicroseconds) { + this.thresholdMicroseconds = thresholdMicroseconds; + } + + @Override + public Collection initArguments(AsyncProfilerAccess access) { + ProfilingEvent event = access.getLockProfilingEvent(); + Objects.requireNonNull(event, "event"); + + // async-profiler expects the lock threshold in nanoseconds + long thresholdNanos = this.thresholdMicroseconds * 1000L; + + return ImmutableList.of( + "event=" + event, + "lock=" + thresholdNanos + ); + } + + @Override + public Class eventClass() { + return ContendedLock.class; + } + + @Override + public long measure(JfrReader reader, ContendedLock event) { + // the duration is recorded in ticks, convert to nanoseconds + long ticksPerSec = reader.ticksPerSec; + if (ticksPerSec <= 0 || ticksPerSec == 1_000_000_000L) { + return event.value(); + } + return (long) (event.value() * (1_000_000_000d / ticksPerSec)); + } + + @Override + public SamplerMode getMode() { + return SamplerMode.LOCK; + } + } + } diff --git a/spark-common/src/main/java/me/lucko/spark/common/sampler/async/jfr/JfrReader.java b/spark-common/src/main/java/me/lucko/spark/common/sampler/async/jfr/JfrReader.java index 389633aee1b..b249fdac3b2 100644 --- a/spark-common/src/main/java/me/lucko/spark/common/sampler/async/jfr/JfrReader.java +++ b/spark-common/src/main/java/me/lucko/spark/common/sampler/async/jfr/JfrReader.java @@ -879,7 +879,7 @@ public CPULoad(JfrReader jfr) { } } - static class ContendedLock extends Event { + public static class ContendedLock extends Event { // spark - make public public final long duration; public final int classId; diff --git a/spark-common/src/main/proto/spark/spark_sampler.proto b/spark-common/src/main/proto/spark/spark_sampler.proto index 3a003a605a5..5d02790d647 100644 --- a/spark-common/src/main/proto/spark/spark_sampler.proto +++ b/spark-common/src/main/proto/spark/spark_sampler.proto @@ -71,6 +71,7 @@ message SamplerMetadata { enum SamplerMode { EXECUTION = 0; ALLOCATION = 1; + LOCK = 2; } enum SamplerEngine { diff --git a/spark-common/src/test/java/me/lucko/spark/common/sampler/LockSamplerTest.java b/spark-common/src/test/java/me/lucko/spark/common/sampler/LockSamplerTest.java new file mode 100644 index 00000000000..96ff5cbacdd --- /dev/null +++ b/spark-common/src/test/java/me/lucko/spark/common/sampler/LockSamplerTest.java @@ -0,0 +1,152 @@ +/* + * This file is part of spark. + * + * Copyright (c) lucko (Luck) + * Copyright (c) contributors + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see . + */ + +package me.lucko.spark.common.sampler; + +import me.lucko.spark.common.sampler.async.AsyncSampler; +import me.lucko.spark.common.sampler.source.ClassSourceLookup; +import me.lucko.spark.proto.SparkSamplerProtos; +import me.lucko.spark.proto.SparkSamplerProtos.SamplerMetadata; +import me.lucko.spark.test.plugin.TestCommandSender; +import me.lucko.spark.test.plugin.TestSparkPlugin; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.nio.file.Path; +import java.util.Locale; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.locks.ReentrantLock; +import java.util.stream.Collectors; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.junit.jupiter.api.Assumptions.assumeTrue; + +public class LockSamplerTest { + + private static boolean isAsyncSupportedOs() { + String os = System.getProperty("os.name").toLowerCase(Locale.ROOT).replace(" ", ""); + return os.equals("linux") || os.equals("macosx"); + } + + @Test + public void testLockSampler(@TempDir Path directory) throws Exception { + assumeTrue(isAsyncSupportedOs(), "async profiler is only supported on Linux and macOS"); + + AtomicBoolean running = new AtomicBoolean(true); + Object monitor = new Object(); + ReentrantLock lock = new ReentrantLock(); + + Runnable monitorContender = () -> { + while (running.get()) { + synchronized (monitor) { + sleepQuietly(20); + } + sleepQuietly(2); // give the other thread a chance to take the monitor + } + }; + Runnable lockContender = () -> { + while (running.get()) { + lock.lock(); + try { + sleepQuietly(20); + } finally { + lock.unlock(); + } + sleepQuietly(2); // give the other thread a chance to take the lock + } + }; + + Thread[] threads = new Thread[]{ + new Thread(monitorContender, "Monitor Test Thread 1"), + new Thread(monitorContender, "Monitor Test Thread 2"), + new Thread(lockContender, "Lock Test Thread 1"), + new Thread(lockContender, "Lock Test Thread 2") + }; + for (Thread thread : threads) { + thread.setDaemon(true); + thread.start(); + } + + try (TestSparkPlugin plugin = new TestSparkPlugin(directory)) { + Sampler sampler = new SamplerBuilder() + .mode(SamplerMode.LOCK) + .threadDumper(ThreadDumper.ALL) + .threadGrouper(ThreadGrouper.BY_POOL) + .samplingInterval(SamplerMode.LOCK.defaultInterval()) + .ignoreSleeping(true) // should be ignored in lock mode + .completeAfter(5, TimeUnit.SECONDS) + .start(plugin.platform()); + + assertInstanceOf(AsyncSampler.class, sampler); + assertEquals(SamplerMode.LOCK, sampler.getMode()); + + sampler.getFuture().get(30, TimeUnit.SECONDS); + + Sampler.ExportProps exportProps = new Sampler.ExportProps() + .creator(TestCommandSender.INSTANCE.toData()) + .classSourceLookup(() -> ClassSourceLookup.create(plugin.platform())); + + SparkSamplerProtos.SamplerData proto = sampler.toProto(plugin.platform(), exportProps); + assertNotNull(proto); + assertEquals(SamplerMetadata.SamplerMode.LOCK, proto.getMetadata().getSamplerMode()); + + for (String name : new String[]{"Monitor Test Thread", "Lock Test Thread"}) { + SparkSamplerProtos.ThreadNode thread = proto.getThreadsList().stream() + .filter(t -> t.getName().startsWith(name)) // name is suffixed with the pool size + .findFirst() + .orElseThrow(() -> new AssertionError("Missing thread '" + name + "' in " + proto.getThreadsList().stream().map(SparkSamplerProtos.ThreadNode::getName).collect(Collectors.toList()))); + + // two threads fight over one lock for ~5 seconds, so the pool as a whole should + // have spent a good few seconds waiting. (values are in milliseconds) + double totalTime = thread.getTimesList().stream().mapToDouble(Double::doubleValue).sum(); + System.out.println(name + " spent " + totalTime + "ms waiting for locks"); + assertTrue(totalTime > 1000, name + " total = " + totalTime); + assertTrue(totalTime < 15000, name + " total = " + totalTime); + } + } finally { + running.set(false); + } + } + + @Test + public void testLockSamplerNotSupportedByJavaSampler(@TempDir Path directory) throws Exception { + try (TestSparkPlugin plugin = new TestSparkPlugin(directory)) { + assertThrows(UnsupportedOperationException.class, () -> new SamplerBuilder() + .mode(SamplerMode.LOCK) + .samplingInterval(SamplerMode.LOCK.defaultInterval()) + .forceJavaSampler(true) + .start(plugin.platform())); + } + } + + private static void sleepQuietly(long millis) { + try { + Thread.sleep(millis); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + +}