Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -87,6 +87,7 @@ public void registerCommands(Consumer<Command> 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)
Expand All @@ -101,7 +102,7 @@ public void registerCommands(Consumer<Command> 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
}
Expand Down Expand Up @@ -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");
Expand Down Expand Up @@ -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()
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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)
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -58,7 +67,7 @@ public LongToDoubleFunction valueTransformer() {
return this.valueTransformer;
}

public int defaultInterval() {
public double defaultInterval() {
return this.defaultInterval;
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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 {
Expand All @@ -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;
Expand All @@ -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;
}

Expand All @@ -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) {
Expand Down Expand Up @@ -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();
}
Expand Down Expand Up @@ -216,7 +237,8 @@ private static Set<ProfilingEvent> getSupportedEvents(AsyncProfiler profiler) {

public enum ProfilingEvent {
WALL(Events.WALL),
ALLOC(Events.ALLOC);
ALLOC(Events.ALLOC),
LOCK(Events.LOCK);

private final String id;

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -250,7 +250,7 @@ private <E extends JfrReader.Event> 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);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand Down Expand Up @@ -55,10 +57,11 @@ public interface SampleCollector<E extends Event> {
/**
* 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.
Expand Down Expand Up @@ -94,7 +97,7 @@ public Class<ExecutionSample> eventClass() {
}

@Override
public long measure(ExecutionSample event) {
public long measure(JfrReader reader, ExecutionSample event) {
return event.value() * this.intervalMicroseconds;
}

Expand Down Expand Up @@ -140,7 +143,7 @@ public Class<AllocationSample> eventClass() {
}

@Override
public long measure(AllocationSample event) {
public long measure(JfrReader reader, AllocationSample event) {
return event.value();
}

Expand All @@ -150,4 +153,53 @@ public SamplerMode getMode() {
}
}

/**
* Sample collector for lock (contention) profiles.
*
* <p>Measures the time threads spend waiting to enter contended monitors
* (synchronized blocks/methods) or parked waiting for
* {@code java.util.concurrent} locks.</p>
*/
final class Lock implements SampleCollector<ContendedLock> {
private final int thresholdMicroseconds; // time in microseconds

public Lock(int thresholdMicroseconds) {
this.thresholdMicroseconds = thresholdMicroseconds;
}

@Override
public Collection<String> 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<ContendedLock> 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;
}
}

}
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand Down
1 change: 1 addition & 0 deletions spark-common/src/main/proto/spark/spark_sampler.proto
Original file line number Diff line number Diff line change
Expand Up @@ -71,6 +71,7 @@ message SamplerMetadata {
enum SamplerMode {
EXECUTION = 0;
ALLOCATION = 1;
LOCK = 2;
}

enum SamplerEngine {
Expand Down
Loading