Support --only-ticks-over with async-profiler

spark-common/src/main/java/me/lucko/spark/common/sampler/SamplerBuilder.java

@@ -112,8 +112,8 @@ public Sampler start(SparkPlatform platform) throws UnsupportedOperationExceptio
             throw new IllegalArgumentException("samplingInterval = " + this.samplingInterval);
         }
 
-        boolean canUseAsyncProfiler = AsyncProfilerAccess.getInstance(platform).checkSupported(platform);
         boolean onlyTicksOverMode = this.ticksOver != -1 && this.tickHook != null;
+        boolean canUseAsyncProfiler = AsyncProfilerAccess.getInstance(platform).checkSupported(platform) && (!onlyTicksOverMode || platform.getTickReporter() != null);
 
         if (this.mode == SamplerMode.ALLOCATION) {
             if (!canUseAsyncProfiler || !AsyncProfilerAccess.getInstance(platform).checkAllocationProfilingSupported(platform)) {
@@ -122,12 +122,9 @@ public Sampler start(SparkPlatform platform) throws UnsupportedOperationExceptio
             if (this.ignoreSleeping) {
                 platform.getPlugin().log(Level.WARNING, "Ignoring sleeping threads is not supported in allocation profiling mode. Sleeping threads will be included in the results.");
             }
-            if (onlyTicksOverMode) {
-                platform.getPlugin().log(Level.WARNING, "'Only-ticks-over' is not supported in allocation profiling mode.");
-            }
         }
 
-        if (onlyTicksOverMode || this.forceJavaSampler) {
+        if (this.forceJavaSampler) {
             canUseAsyncProfiler = false;
         }
 
@@ -139,14 +136,24 @@ public Sampler start(SparkPlatform platform) throws UnsupportedOperationExceptio
         SamplerSettings settings = new SamplerSettings(interval, this.threadDumper, this.threadGrouper.get(), this.autoEndTime, this.background, this.ignoreSleeping);
 
         Sampler sampler;
-        if (this.mode == SamplerMode.ALLOCATION) {
-            sampler = new AsyncSampler(platform, settings, new SampleCollector.Allocation(interval, this.allocLiveOnly));
-        } else if (canUseAsyncProfiler) {
-            sampler = new AsyncSampler(platform, settings, new SampleCollector.Execution(interval));
-        } else if (onlyTicksOverMode) {
-            sampler = new JavaSampler(platform, settings, this.tickHook, this.ticksOver);
+        if (canUseAsyncProfiler) {
+            SampleCollector<?> collector;
+            if (this.mode == SamplerMode.ALLOCATION) {
+                collector = new SampleCollector.Allocation(interval, this.allocLiveOnly);
+            } else {
+                collector = new SampleCollector.Execution(interval);
+            }
+            if (onlyTicksOverMode) {
+                sampler = new AsyncSampler(platform, settings, collector, this.ticksOver);
+            } else {
+                sampler = new AsyncSampler(platform, settings, collector);
+            }
         } else {
-            sampler = new JavaSampler(platform, settings);
+            if (onlyTicksOverMode) {
+                sampler = new JavaSampler(platform, settings, this.tickHook, this.ticksOver);
+            } else {
+                sampler = new JavaSampler(platform, settings);
+            }
         }
 
         sampler.start();

spark-common/src/main/java/me/lucko/spark/common/sampler/async/AsyncDataAggregator.java

@@ -29,7 +29,7 @@
 /**
  * Data aggregator for {@link AsyncSampler}.
  */
-public class AsyncDataAggregator extends AbstractDataAggregator {
+public class AsyncDataAggregator extends AbstractDataAggregator implements AutoCloseable {
 
     /** A describer for async-profiler stack trace elements. */
     private static final StackTraceNode.Describer<AsyncStackTraceElement> STACK_TRACE_DESCRIBER = (element, parent) ->
@@ -79,4 +79,8 @@ private static boolean isSleeping(ProfileSegment element) {
         return false;
     }
 
+    @Override
+    public void close() {
+
+    }
 }

spark-common/src/main/java/me/lucko/spark/common/sampler/async/AsyncProfilerJob.java

@@ -84,6 +84,8 @@ static AsyncProfilerJob createNew(AsyncProfilerAccess access, AsyncProfiler prof
     private int window;
     /** If the profiler should run in quiet mode */
     private boolean quiet;
+    /** If the profiler needs to use the same clock as {@link System#nanoTime()} */
+    private boolean forceNanoTime;
 
     /** The file used by async-profiler to output data */
     private Path outputFile;
@@ -117,12 +119,13 @@ private void checkActive() {
     }
 
     // Initialise the job
-    public void init(SparkPlatform platform, SampleCollector<?> collector, ThreadDumper threadDumper, int window, boolean quiet) {
+    public void init(SparkPlatform platform, SampleCollector<?> collector, ThreadDumper threadDumper, int window, boolean quiet, boolean forceNanoTime) {
         this.platform = platform;
         this.sampleCollector = collector;
         this.threadDumper = threadDumper;
         this.window = window;
         this.quiet = quiet;
+        this.forceNanoTime = forceNanoTime;
     }
 
     /**
@@ -151,6 +154,9 @@ public void start() {
             if (this.threadDumper instanceof ThreadDumper.Specific) {
                 command.add("filter");
             }
+            if (this.forceNanoTime) {
+                command.add("clock=monotonic");
+            }
 
             // start the profiler
             String resp = execute(command.build()).trim();

spark-common/src/main/java/me/lucko/spark/common/sampler/async/AsyncSampler.java

@@ -27,6 +27,7 @@
 import me.lucko.spark.common.sampler.SamplerSettings;
 import me.lucko.spark.common.sampler.SamplerType;
 import me.lucko.spark.common.sampler.window.ProfilingWindowUtils;
+import me.lucko.spark.common.sampler.window.WindowStatisticsCollector;
 import me.lucko.spark.common.tick.TickHook;
 import me.lucko.spark.common.util.SparkThreadFactory;
 import me.lucko.spark.common.ws.ViewerSocket;
@@ -66,10 +67,18 @@ public class AsyncSampler extends AbstractSampler {
     private ScheduledFuture<?> socketStatisticsTask;
 
     public AsyncSampler(SparkPlatform platform, SamplerSettings settings, SampleCollector<?> collector) {
+        this(platform, settings, collector, new AsyncDataAggregator(settings.threadGrouper(), settings.ignoreSleeping()));
+    }
+
+    public AsyncSampler(SparkPlatform platform, SamplerSettings settings, SampleCollector<?> collector, int tickLengthThreshold) {
+        this(platform, settings, collector, new TickedAsyncDataAggregator(settings.threadGrouper(), settings.ignoreSleeping(), platform.getTickReporter(), tickLengthThreshold));
+    }
+
+    private AsyncSampler(SparkPlatform platform, SamplerSettings settings, SampleCollector<?> collector, AsyncDataAggregator dataAggregator) {
         super(platform, settings);
         this.sampleCollector = collector;
+        this.dataAggregator = dataAggregator;
         this.profilerAccess = AsyncProfilerAccess.getInstance(platform);
-        this.dataAggregator = new AsyncDataAggregator(settings.threadGrouper(), settings.ignoreSleeping());
         this.scheduler = Executors.newSingleThreadScheduledExecutor(
                 new ThreadFactoryBuilder()
                         .setNameFormat("spark-async-sampler-worker-thread")
@@ -87,13 +96,18 @@ public void start() {
 
         TickHook tickHook = this.platform.getTickHook();
         if (tickHook != null) {
-            this.windowStatisticsCollector.startCountingTicks(tickHook);
+            if (this.dataAggregator instanceof TickedAsyncDataAggregator) {
+                WindowStatisticsCollector.ExplicitTickCounter counter = this.windowStatisticsCollector.startCountingTicksExplicit(tickHook);
+                ((TickedAsyncDataAggregator) this.dataAggregator).setTickCounter(counter);
+            } else {
+                this.windowStatisticsCollector.startCountingTicks(tickHook);
+            };
         }
 
         int window = ProfilingWindowUtils.windowNow();
 
         AsyncProfilerJob job = this.profilerAccess.startNewProfilerJob();
-        job.init(this.platform, this.sampleCollector, this.threadDumper, window, this.background);
+        job.init(this.platform, this.sampleCollector, this.threadDumper, window, this.background, this.dataAggregator instanceof TickedAsyncDataAggregator);
         job.start();
         this.windowStatisticsCollector.recordWindowStartTime(window);
         this.currentJob = job;
@@ -131,7 +145,7 @@ private void rotateProfilerJob() {
                 // start a new job
                 int window = previousJob.getWindow() + 1;
                 AsyncProfilerJob newJob = this.profilerAccess.startNewProfilerJob();
-                newJob.init(this.platform, this.sampleCollector, this.threadDumper, window, this.background);
+                newJob.init(this.platform, this.sampleCollector, this.threadDumper, window, this.background, this.dataAggregator instanceof TickedAsyncDataAggregator);
                 newJob.start();
                 this.windowStatisticsCollector.recordWindowStartTime(window);
                 this.currentJob = newJob;

spark-common/src/main/java/me/lucko/spark/common/sampler/async/ProfileSegment.java

@@ -43,13 +43,16 @@ public class ProfileSegment {
     private final long value;
     /** The state of the thread. {@value #UNKNOWN_THREAD_STATE} if state is unknown */
     private final String threadState;
+    /** The time at which this segment was recorded, as if it was produced by {@link System#nanoTime()} */
+    private final long time;
 
-    private ProfileSegment(int nativeThreadId, String threadName, AsyncStackTraceElement[] stackTrace, long value, String threadState) {
+    private ProfileSegment(int nativeThreadId, String threadName, AsyncStackTraceElement[] stackTrace, long value, String threadState, long time) {
         this.nativeThreadId = nativeThreadId;
         this.threadName = threadName;
         this.stackTrace = stackTrace;
         this.value = value;
         this.threadState = threadState;
+        this.time = time;
     }
 
     public int getNativeThreadId() {
@@ -72,6 +75,10 @@ public String getThreadState() {
         return this.threadState;
     }
 
+    public long getTime() {
+        return this.time;
+    }
+
     public static ProfileSegment parseSegment(JfrReader reader, JfrReader.Event sample, String threadName, long value) {
         JfrReader.StackTrace stackTrace = reader.stackTraces.get(sample.stackTraceId);
         int len = stackTrace != null ? stackTrace.methods.length : 0;
@@ -86,7 +93,7 @@ public static ProfileSegment parseSegment(JfrReader reader, JfrReader.Event samp
             threadState = reader.threadStates.get(executionSample.threadState);
         }
 
-        return new ProfileSegment(sample.tid, threadName, stack, value, threadState);
+        return new ProfileSegment(sample.tid, threadName, stack, value, threadState, sample.time);
     }
 
     private static AsyncStackTraceElement parseStackFrame(JfrReader reader, long methodId) {

spark-common/src/main/java/me/lucko/spark/common/sampler/async/TickedAsyncDataAggregator.java

@@ -0,0 +1,115 @@
+/*
+ * This file is part of spark.
+ *
+ *  Copyright (c) lucko (Luck) <[email protected]>
+ *  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 <http://www.gnu.org/licenses/>.
+ */
+
+package me.lucko.spark.common.sampler.async;
+
+import me.lucko.spark.common.sampler.ThreadGrouper;
+import me.lucko.spark.common.sampler.window.WindowStatisticsCollector;
+import me.lucko.spark.common.tick.TickReporter;
+import me.lucko.spark.proto.SparkSamplerProtos;
+
+import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.TimeUnit;
+
+public class TickedAsyncDataAggregator extends AsyncDataAggregator {
+
+    /** The ticks that exceeded the threshold, cleared one-by-one when inserting data */
+    private final Queue<ExceededTick> ticksOver = new ConcurrentLinkedQueue<>();
+
+    /** The callback called when this aggregator is closed, to clean up resources */
+    private final Runnable closeCallback;
+
+    /** Tick durations under this threshold will not be inserted, measured in microseconds */
+    private final long tickLengthThreshold;
+
+    /** Counts the number of ticks aggregated */
+    private WindowStatisticsCollector.ExplicitTickCounter tickCounter;
+
+    protected TickedAsyncDataAggregator(ThreadGrouper threadGrouper, boolean ignoreSleeping, TickReporter tickReporter, int tickLengthThreshold) {
+        super(threadGrouper, ignoreSleeping);
+        this.tickLengthThreshold = TimeUnit.MILLISECONDS.toMicros(tickLengthThreshold);
+        TickReporter.Callback tickReporterCallback = duration -> {
+            if (duration > tickLengthThreshold) {
+                long end = System.nanoTime();
+                long start = (long) (end - (duration * 1_000_000)); // ms to ns
+                this.ticksOver.add(new ExceededTick(start, end));
+                this.tickCounter.increment();
+            }
+        };
+        tickReporter.addCallback(tickReporterCallback);
+        this.closeCallback = () -> tickReporter.removeCallback(tickReporterCallback);
+    }
+
+    @Override
+    public void insertData(ProfileSegment element, int window) {
+        // with async-profiler clock=monotonic, the event time uses the same clock
+        // as System.nanoTime(), so we can compare it directly
+        long time = element.getTime();
+        while (true) {
+            ExceededTick earliestExceeding = ticksOver.peek();
+            if (earliestExceeding == null) {
+                // no tick over threshold anymore
+                return;
+            } else if (time - earliestExceeding.start < 0) {
+                // segment happened before current exceeding
+                return;
+            } else if (earliestExceeding.end - time < 0) {
+                // segment happened after current exceeding,
+                // but it might fall into the next one
+                ticksOver.remove();
+            } else {
+                // segment falls exactly into exceeding, record it
+                break;
+            }
+        }
+        super.insertData(element, window);
+    }
+
+    @Override
+    public SparkSamplerProtos.SamplerMetadata.DataAggregator getMetadata() {
+        return SparkSamplerProtos.SamplerMetadata.DataAggregator.newBuilder()
+                .setType(SparkSamplerProtos.SamplerMetadata.DataAggregator.Type.TICKED)
+                .setThreadGrouper(this.threadGrouper.asProto())
+                .setTickLengthThreshold(this.tickLengthThreshold)
+                .setNumberOfIncludedTicks(this.tickCounter.getTotalCountedTicks())
+                .build();
+
+    }
+
+    @Override
+    public void close() {
+        this.closeCallback.run();
+    }
+
+    public void setTickCounter(WindowStatisticsCollector.ExplicitTickCounter counter) {
+        this.tickCounter = counter;
+    }
+
+    private static final class ExceededTick {
+        private final long start;
+        private final long end;
+
+        ExceededTick(long start, long end) {
+            this.start = start;
+            this.end = end;
+        }
+    }
+}