Add '/profiler monitoring' command to monitor tick times, and '--only-ticks-over' argument to filter profiling output to ticks lasting over a certain duration

8 files changed, 497 insertions, 78 deletions

diff --git a/common/src/main/java/me/lucko/spark/common/CommandHandler.java b/common/src/main/java/me/lucko/spark/common/CommandHandler.java
index 6664c18..a9bbe61 100644
--- a/common/src/main/java/me/lucko/spark/common/CommandHandler.java
+++ b/common/src/main/java/me/lucko/spark/common/CommandHandler.java
@@ -10,6 +10,7 @@ import me.lucko.spark.profiler.Sampler;
 import me.lucko.spark.profiler.SamplerBuilder;
 import me.lucko.spark.profiler.ThreadDumper;
 import me.lucko.spark.profiler.ThreadGrouper;
+import me.lucko.spark.profiler.TickCounter;
 
 import java.io.IOException;
 import java.util.ArrayList;
@@ -33,6 +34,8 @@ public abstract class CommandHandler<T> {
 
     /** The URL of the viewer frontend */
     private static final String VIEWER_URL = "https://sparkprofiler.github.io/?";
+    /** The prefix used in all messages */
+    private static final String PREFIX = "&8[&fspark&8] &7";
 
     /**
      * The {@link Timer} being used by the {@link #activeSampler}.
@@ -43,17 +46,25 @@ public abstract class CommandHandler<T> {
     private final Object[] activeSamplerMutex = new Object[0];
     /** The WarmRoast instance currently running, if any */
     private Sampler activeSampler = null;
+    /** The tick monitor instance currently running, if any */
+    private ReportingTickMonitor activeTickMonitor = null;
 
 
     // abstract methods implemented by each platform
 
     protected abstract void sendMessage(T sender, String message);
-    protected abstract void sendLink(T sender, String url);
+    protected abstract void sendMessage(String message);
+    protected abstract void sendLink(String url);
     protected abstract void runAsync(Runnable r);
     protected abstract ThreadDumper getDefaultThreadDumper();
+    protected abstract TickCounter newTickCounter();
 
     private void sendPrefixedMessage(T sender, String message) {
-        sendMessage(sender, "&8[&fspark&8] &7" + message);
+        sendMessage(sender, PREFIX + message);
+    }
+
+    private void sendPrefixedMessage(String message) {
+        sendMessage(PREFIX + message);
     }
 
     public void handleCommand(T sender, String[] args) {
@@ -79,6 +90,9 @@ public abstract class CommandHandler<T> {
                 case "paste":
                     handleStop(sender);
                     break;
+                case "monitoring":
+                    handleMonitoring(sender, arguments);
+                    break;
                 default:
                     sendInfo(sender);
                     break;
@@ -95,9 +109,11 @@ public abstract class CommandHandler<T> {
         sendMessage(sender, "       &8[&7--thread&8 <thread name>]");
         sendMessage(sender, "       &8[&7--not-combined]");
         sendMessage(sender, "       &8[&7--interval&8 <interval millis>]");
+        sendMessage(sender, "       &8[&7--only-ticks-over&8 <tick length millis>]");
         sendMessage(sender, "&b&l> &7/profiler info");
         sendMessage(sender, "&b&l> &7/profiler stop");
         sendMessage(sender, "&b&l> &7/profiler cancel");
+        sendMessage(sender, "&b&l> &7/profiler monitoring");
     }
 
     private void handleStart(T sender, List<String> args) {
@@ -136,6 +152,17 @@ public abstract class CommandHandler<T> {
             threadGrouper = ThreadGrouper.BY_POOL;
         }
 
+        int ticksOver = parseInt(arguments, "only-ticks-over", "o");
+        TickCounter tickCounter = null;
+        if (ticksOver != -1) {
+            try {
+                tickCounter = newTickCounter();
+            } catch (UnsupportedOperationException e) {
+                sendPrefixedMessage(sender, "&cTick counting is not supported on BungeeCord!");
+                return;
+            }
+        }
+
         Sampler sampler;
         synchronized (this.activeSamplerMutex) {
             if (this.activeSampler != null) {
@@ -143,7 +170,7 @@ public abstract class CommandHandler<T> {
                 return;
             }
 
-            sendPrefixedMessage(sender, "&7Initializing a new profiler, please wait...");
+            sendPrefixedMessage("&7Initializing a new profiler, please wait...");
 
             SamplerBuilder builder = new SamplerBuilder();
             builder.threadDumper(threadDumper);
@@ -152,13 +179,16 @@ public abstract class CommandHandler<T> {
                 builder.completeAfter(timeoutSeconds, TimeUnit.SECONDS);
             }
             builder.samplingInterval(intervalMillis);
+            if (ticksOver != -1) {
+                builder.ticksOver(ticksOver, tickCounter);
+            }
             sampler = this.activeSampler = builder.start(this.samplingThread);
 
-            sendPrefixedMessage(sender, "&bProfiler now active!");
+            sendPrefixedMessage("&bProfiler now active!");
             if (timeoutSeconds == -1) {
-                sendPrefixedMessage(sender, "&7Use '/profiler stop' to stop profiling and upload the results.");
+                sendPrefixedMessage("&7Use '/profiler stop' to stop profiling and upload the results.");
             } else {
-                sendPrefixedMessage(sender, "&7The results will be automatically returned after the profiler has been running for " + timeoutSeconds + " seconds.");
+                sendPrefixedMessage("&7The results will be automatically returned after the profiler has been running for " + timeoutSeconds + " seconds.");
             }
         }
 
@@ -167,7 +197,7 @@ public abstract class CommandHandler<T> {
         // send message if profiling fails
         future.whenCompleteAsync((s, throwable) -> {
             if (throwable != null) {
-                sendPrefixedMessage(sender, "&cSampling operation failed unexpectedly. Error: " + throwable.toString());
+                sendPrefixedMessage("&cSampling operation failed unexpectedly. Error: " + throwable.toString());
                 throwable.printStackTrace();
             }
         });
@@ -184,8 +214,8 @@ public abstract class CommandHandler<T> {
         // await the result
         if (timeoutSeconds != -1) {
             future.thenAcceptAsync(s -> {
-                sendPrefixedMessage(sender, "&7The active sampling operation has completed! Uploading results...");
-                handleUpload(sender, s);
+                sendPrefixedMessage("&7The active sampling operation has completed! Uploading results...");
+                handleUpload(s);
             });
         }
     }
@@ -215,8 +245,8 @@ public abstract class CommandHandler<T> {
                 sendPrefixedMessage(sender, "&7There isn't an active sampling task running.");
             } else {
                 this.activeSampler.cancel();
-                sendPrefixedMessage(sender, "&7The active sampling operation has been stopped! Uploading results...");
-                handleUpload(sender, this.activeSampler);
+                sendPrefixedMessage("&7The active sampling operation has been stopped! Uploading results...");
+                handleUpload(this.activeSampler);
                 this.activeSampler = null;
             }
         }
@@ -229,25 +259,59 @@ public abstract class CommandHandler<T> {
             } else {
                 this.activeSampler.cancel();
                 this.activeSampler = null;
-                sendPrefixedMessage(sender, "&bThe active sampling task has been cancelled.");
+                sendPrefixedMessage("&bThe active sampling task has been cancelled.");
             }
         }
     }
 
-    private void handleUpload(T sender, Sampler sampler) {
+    private void handleUpload(Sampler sampler) {
         runAsync(() -> {
             JsonObject output = sampler.formOutput();
             try {
                 String pasteId = Bytebin.postContent(output);
-                sendPrefixedMessage(sender, "&bSampling results:");
-                sendLink(sender, VIEWER_URL + pasteId);
+                sendPrefixedMessage("&bSampling results:");
+                sendLink(VIEWER_URL + pasteId);
             } catch (IOException e) {
-                sendPrefixedMessage(sender, "&cAn error occurred whilst uploading the results.");
+                sendPrefixedMessage("&cAn error occurred whilst uploading the results.");
                 e.printStackTrace();
             }
         });
     }
 
+    private void handleMonitoring(T sender, List<String> args) {
+        SetMultimap<String, String> arguments = parseArguments(args);
+
+        if (this.activeTickMonitor == null) {
+
+            int threshold = parseInt(arguments, "threshold", "t");
+            if (threshold == -1) {
+                threshold = 100;
+            }
+
+            try {
+                TickCounter tickCounter = newTickCounter();
+                this.activeTickMonitor = new ReportingTickMonitor(tickCounter, threshold);
+            } catch (UnsupportedOperationException e) {
+                sendPrefixedMessage(sender, "&cNot supported on BungeeCord!");
+            }
+        } else {
+            this.activeTickMonitor.close();
+            this.activeTickMonitor = null;
+            sendPrefixedMessage("&7Tick monitor disabled.");
+        }
+    }
+
+    private class ReportingTickMonitor extends TickMonitor {
+        public ReportingTickMonitor(TickCounter tickCounter, int percentageChangeThreshold) {
+            super(tickCounter, percentageChangeThreshold);
+        }
+
+        @Override
+        protected void sendMessage(String message) {
+            sendPrefixedMessage(message);
+        }
+    }
+
     private int parseInt(SetMultimap<String, String> arguments, String longArg, String shortArg) {
         Iterator<String> it = Sets.union(arguments.get(longArg), arguments.get(shortArg)).iterator();
         if (it.hasNext()) {
diff --git a/common/src/main/java/me/lucko/spark/common/TickMonitor.java b/common/src/main/java/me/lucko/spark/common/TickMonitor.java
new file mode 100644
index 0000000..e6342c4
--- /dev/null
+++ b/common/src/main/java/me/lucko/spark/common/TickMonitor.java
@@ -0,0 +1,87 @@
+package me.lucko.spark.common;
+
+import me.lucko.spark.profiler.TickCounter;
+
+import java.text.DecimalFormat;
+import java.util.DoubleSummaryStatistics;
+
+public abstract class TickMonitor implements Runnable {
+    private static final DecimalFormat df = new DecimalFormat("#.##");
+
+    private final TickCounter tickCounter;
+    private final int percentageChangeThreshold;
+
+    // data
+    private double lastTickTime = 0;
+    private State state = null;
+    private DoubleSummaryStatistics averageTickTime = new DoubleSummaryStatistics();
+
+    public TickMonitor(TickCounter tickCounter, int percentageChangeThreshold) {
+        this.tickCounter = tickCounter;
+        this.percentageChangeThreshold = percentageChangeThreshold;
+
+        this.tickCounter.start();
+        this.tickCounter.addTickTask(this);
+    }
+
+    protected abstract void sendMessage(String message);
+
+    public void close() {
+        this.tickCounter.close();
+    }
+
+    @Override
+    public void run() {
+        double now = ((double) System.nanoTime()) / 1000000d;
+
+        // init
+        if (this.state == null) {
+            this.state = State.SETUP;
+            this.lastTickTime = now;
+            sendMessage("Tick monitor started. Before the monitor becomes fully active, the server's " +
+                    "average tick rate will be calculated over a period of 100 ticks (approx 5 seconds).");
+            return;
+        }
+
+        // find the diff
+        double diff = now - this.lastTickTime;
+        this.lastTickTime = now;
+
+        // form averages
+        if (this.state == State.SETUP) {
+            this.averageTickTime.accept(diff);
+
+            // move onto the next state
+            if (this.averageTickTime.getCount() >= 100) {
+
+                sendMessage("&bAnalysis is now complete.");
+                sendMessage("&f> &7Max: " + df.format(this.averageTickTime.getMax()) + "ms");
+                sendMessage("&f> &7Min: " + df.format(this.averageTickTime.getMin()) + "ms");
+                sendMessage("&f> &7Avg: " + df.format(this.averageTickTime.getAverage()) + "ms");
+                sendMessage("Starting now, any ticks with >" + this.percentageChangeThreshold + "% increase in " +
+                        "duration compared to the average will be reported.");
+                this.state = State.MONITORING;
+            }
+        }
+
+        if (this.state == State.MONITORING) {
+            double avg = this.averageTickTime.getAverage();
+
+            double increase = diff - avg;
+            if (increase <= 0) {
+                return;
+            }
+
+            double percentageChange = (increase * 100d) / avg;
+            if (percentageChange > this.percentageChangeThreshold) {
+                sendMessage("&7Tick &8#" + this.tickCounter.getCurrentTick() + " &7lasted &b" + df.format(diff) + "&7 milliseconds. " +
+                        "&7(a &b" + df.format(percentageChange) + "% &7increase from average)");
+            }
+        }
+    }
+
+    private enum State {
+        SETUP,
+        MONITORING
+    }
+}
diff --git a/common/src/main/java/me/lucko/spark/profiler/AsyncDataAggregator.java b/common/src/main/java/me/lucko/spark/profiler/AsyncDataAggregator.java
new file mode 100644
index 0000000..9a4090e
--- /dev/null
+++ b/common/src/main/java/me/lucko/spark/profiler/AsyncDataAggregator.java
@@ -0,0 +1,77 @@
+package me.lucko.spark.profiler;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Implementation of {@link DataAggregator} that makes use of a "worker" thread pool for inserting
+ * data.
+ */
+public class AsyncDataAggregator implements DataAggregator {
+
+    /** A map of root stack nodes for each thread with sampling data */
+    private final Map<String, StackNode> threadData = new ConcurrentHashMap<>();
+
+    /** The worker pool for inserting stack nodes */
+    private final ExecutorService workerPool;
+
+    /** The instance used to group threads together */
+    private final ThreadGrouper threadGrouper;
+
+    /** The interval to wait between sampling, in milliseconds */
+    private final int interval;
+
+    public AsyncDataAggregator(ExecutorService workerPool, ThreadGrouper threadGrouper, int interval) {
+        this.workerPool = workerPool;
+        this.threadGrouper = threadGrouper;
+        this.interval = interval;
+    }
+
+    @Override
+    public void insertData(String threadName, StackTraceElement[] stack) {
+        // form the queued data
+        QueuedThreadInfo queuedData = new QueuedThreadInfo(threadName, stack);
+        // schedule insertion of the data
+        this.workerPool.execute(queuedData);
+    }
+
+    @Override
+    public Map<String, StackNode> getData() {
+        // wait for all pending data to be inserted
+        this.workerPool.shutdown();
+        try {
+            this.workerPool.awaitTermination(15, TimeUnit.SECONDS);
+        } catch (InterruptedException e) {
+            e.printStackTrace();
+        }
+
+        return this.threadData;
+    }
+
+    void insertData(QueuedThreadInfo data) {
+        try {
+            String group = this.threadGrouper.getGroup(data.threadName);
+            StackNode node = this.threadData.computeIfAbsent(group, StackNode::new);
+            node.log(data.stack, this.interval);
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+    }
+
+    private final class QueuedThreadInfo implements Runnable {
+        private final String threadName;
+        private final StackTraceElement[] stack;
+
+        QueuedThreadInfo(String threadName, StackTraceElement[] stack) {
+            this.threadName = threadName;
+            this.stack = stack;
+        }
+
+        @Override
+        public void run() {
+            insertData(this);
+        }
+    }
+}
diff --git a/common/src/main/java/me/lucko/spark/profiler/DataAggregator.java b/common/src/main/java/me/lucko/spark/profiler/DataAggregator.java
new file mode 100644
index 0000000..1afa52c
--- /dev/null
+++ b/common/src/main/java/me/lucko/spark/profiler/DataAggregator.java
@@ -0,0 +1,32 @@
+package me.lucko.spark.profiler;
+
+import java.util.Map;
+
+/**
+ * Aggregates sampling data.
+ */
+public interface DataAggregator {
+
+    /**
+     * Called before the sampler begins to insert data
+     */
+    default void start() {
+
+    }
+
+    /**
+     * Forms the output data
+     *
+     * @return the output data
+     */
+    Map<String, StackNode> getData();
+
+    /**
+     * Inserts sampling data into this aggregator
+     *
+     * @param threadName the name of the thread
+     * @param stack the call stack
+     */
+    void insertData(String threadName, StackTraceElement[] stack);
+
+}
diff --git a/common/src/main/java/me/lucko/spark/profiler/Sampler.java b/common/src/main/java/me/lucko/spark/profiler/Sampler.java
index d03b4b6..44cf445 100644
--- a/common/src/main/java/me/lucko/spark/profiler/Sampler.java
+++ b/common/src/main/java/me/lucko/spark/profiler/Sampler.java
@@ -31,10 +31,8 @@ import java.util.Map;
 import java.util.Timer;
 import java.util.TimerTask;
 import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
-import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 
 /**
@@ -43,35 +41,39 @@ import java.util.concurrent.atomic.AtomicInteger;
 public class Sampler extends TimerTask {
     private static final AtomicInteger THREAD_ID = new AtomicInteger(0);
 
-    /** The thread management interface for the current JVM */
-    private final ThreadMXBean threadBean = ManagementFactory.getThreadMXBean();
-
-    /** A map of root stack nodes for each thread with sampling data */
-    private final Map<String, StackNode> threadData = new ConcurrentHashMap<>();
-
     /** The worker pool for inserting stack nodes */
     private final ExecutorService workerPool = Executors.newFixedThreadPool(
             6, new ThreadFactoryBuilder().setNameFormat("spark-worker-" + THREAD_ID.getAndIncrement()).build()
     );
 
+    /** The thread management interface for the current JVM */
+    private final ThreadMXBean threadBean = ManagementFactory.getThreadMXBean();
+    /** The instance used to generate thread information for use in sampling */
+    private final ThreadDumper threadDumper;
+    /** Responsible for aggregating and then outputting collected sampling data */
+    private final DataAggregator dataAggregator;
+
     /** A future to encapsulation the completion of this sampler instance */
     private final CompletableFuture<Sampler> future = new CompletableFuture<>();
 
     /** The interval to wait between sampling, in milliseconds */
     private final int interval;
-    /** The instance used to generate thread information for use in sampling */
-    private final ThreadDumper threadDumper;
-    /** The instance used to group threads together */
-    private final ThreadGrouper threadGrouper;
     /** The time when sampling first began */
     private long startTime = -1;
     /** The unix timestamp (in millis) when this sampler should automatically complete.*/
     private final long endTime; // -1 for nothing
     
     public Sampler(int interval, ThreadDumper threadDumper, ThreadGrouper threadGrouper, long endTime) {
+        this.threadDumper = threadDumper;
+        this.dataAggregator = new AsyncDataAggregator(this.workerPool, threadGrouper, interval);
         this.interval = interval;
+        this.endTime = endTime;
+    }
+
+    public Sampler(int interval, ThreadDumper threadDumper, ThreadGrouper threadGrouper, long endTime, TickCounter tickCounter, int tickLengthThreshold) {
         this.threadDumper = threadDumper;
-        this.threadGrouper = threadGrouper;
+        this.dataAggregator = new TickedDataAggregator(this.workerPool, tickCounter, threadGrouper, interval, tickLengthThreshold);
+        this.interval = interval;
         this.endTime = endTime;
     }
 
@@ -81,35 +83,9 @@ public class Sampler extends TimerTask {
      * @param samplingThread the timer to schedule the sampling on
      */
     public void start(Timer samplingThread) {
-        samplingThread.scheduleAtFixedRate(this, 0, this.interval);
         this.startTime = System.currentTimeMillis();
-    }
-
-    private void insertData(QueuedThreadInfo data) {
-        try {
-            String group = this.threadGrouper.getGroup(data.threadName);
-            StackNode node = this.threadData.computeIfAbsent(group, StackNode::new);
-            node.log(data.stack, Sampler.this.interval);
-        } catch (Exception e) {
-            e.printStackTrace();
-        }
-    }
-
-    /**
-     * Gets the sampling data recorded by this instance.
-     *
-     * @return the data
-     */
-    public Map<String, StackNode> getData() {
-        // wait for all pending data to be inserted
-        this.workerPool.shutdown();
-        try {
-            this.workerPool.awaitTermination(15, TimeUnit.SECONDS);
-        } catch (InterruptedException e) {
-            e.printStackTrace();
-        }
-
-        return this.threadData;
+        this.dataAggregator.start();
+        samplingThread.scheduleAtFixedRate(this, 0, this.interval);
     }
 
     public long getStartTime() {
@@ -145,10 +121,7 @@ public class Sampler extends TimerTask {
                     continue;
                 }
 
-                // form the queued data
-                QueuedThreadInfo queuedData = new QueuedThreadInfo(threadName, stack);
-                // schedule insertion of the data
-                this.workerPool.execute(queuedData);
+                this.dataAggregator.insertData(threadName, stack);
             }
         } catch (Throwable t) {
             this.future.completeExceptionally(t);
@@ -161,7 +134,7 @@ public class Sampler extends TimerTask {
 
         JsonArray threads = new JsonArray();
 
-        List<Map.Entry<String, StackNode>> data = new ArrayList<>(getData().entrySet());
+        List<Map.Entry<String, StackNode>> data = new ArrayList<>(this.dataAggregator.getData().entrySet());
         data.sort(Map.Entry.comparingByKey());
 
         for (Map.Entry<String, StackNode> entry : data) {
@@ -177,19 +150,4 @@ public class Sampler extends TimerTask {
         return out;
     }
 
-    private final class QueuedThreadInfo implements Runnable {
-        private final String threadName;
-        private final StackTraceElement[] stack;
-
-        private QueuedThreadInfo(String threadName, StackTraceElement[] stack) {
-            this.threadName = threadName;
-            this.stack = stack;
-        }
-
-        @Override
-        public void run() {
-            insertData(this);
-        }
-    }
-
 }
diff --git a/common/src/main/java/me/lucko/spark/profiler/SamplerBuilder.java b/common/src/main/java/me/lucko/spark/profiler/SamplerBuilder.java
index fa2898b..4c16d50 100644
--- a/common/src/main/java/me/lucko/spark/profiler/SamplerBuilder.java
+++ b/common/src/main/java/me/lucko/spark/profiler/SamplerBuilder.java
@@ -15,6 +15,9 @@ public class SamplerBuilder {
     private ThreadDumper threadDumper = ThreadDumper.ALL;
     private ThreadGrouper threadGrouper = ThreadGrouper.BY_NAME;
 
+    private int ticksOver = -1;
+    private TickCounter tickCounter = null;
+
     public SamplerBuilder() {
     }
 
@@ -39,8 +42,20 @@ public class SamplerBuilder {
         return this;
     }
 
+    public SamplerBuilder ticksOver(int ticksOver, TickCounter tickCounter) {
+        this.ticksOver = ticksOver;
+        this.tickCounter = tickCounter;
+        return this;
+    }
+
     public Sampler start(Timer samplingThread) {
-        Sampler sampler = new Sampler(this.samplingInterval, this.threadDumper, this.threadGrouper, this.timeout);
+        Sampler sampler;
+        if (this.ticksOver != -1 && this.tickCounter != null) {
+            sampler = new Sampler(this.samplingInterval, this.threadDumper, this.threadGrouper, this.timeout, this.tickCounter, this.ticksOver);
+        } else {
+            sampler = new Sampler(this.samplingInterval, this.threadDumper, this.threadGrouper, this.timeout);
+        }
+
         sampler.start(samplingThread);
         return sampler;
     }
diff --git a/common/src/main/java/me/lucko/spark/profiler/TickCounter.java b/common/src/main/java/me/lucko/spark/profiler/TickCounter.java
new file mode 100644
index 0000000..53a9c27
--- /dev/null
+++ b/common/src/main/java/me/lucko/spark/profiler/TickCounter.java
@@ -0,0 +1,39 @@
+package me.lucko.spark.profiler;
+
+/**
+ * A hook with the game's "tick loop".
+ */
+public interface TickCounter {
+
+    /**
+     * Starts the counter
+     */
+    void start();
+
+    /**
+     * Stops the counter
+     */
+    void close();
+
+    /**
+     * Gets the current tick number
+     *
+     * @return the current tick
+     */
+    long getCurrentTick();
+
+    /**
+     * Adds a task to be called each time the tick increments
+     *
+     * @param runnable the task
+     */
+    void addTickTask(Runnable runnable);
+
+    /**
+     * Removes a tick task
+     *
+     * @param runnable the task
+     */
+    void removeTickTask(Runnable runnable);
+
+}
diff --git a/common/src/main/java/me/lucko/spark/profiler/TickedDataAggregator.java b/common/src/main/java/me/lucko/spark/profiler/TickedDataAggregator.java
new file mode 100644
index 0000000..abca4b3
--- /dev/null
+++ b/common/src/main/java/me/lucko/spark/profiler/TickedDataAggregator.java
@@ -0,0 +1,147 @@
+package me.lucko.spark.profiler;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Implementation of {@link DataAggregator} which supports only including sampling data from "ticks"
+ * which exceed a certain threshold in duration.
+ */
+public class TickedDataAggregator implements DataAggregator {
+
+    /** A map of root stack nodes for each thread with sampling data */
+    private final Map<String, StackNode> threadData = new ConcurrentHashMap<>();
+
+    /** The worker pool for inserting stack nodes */
+    private final ExecutorService workerPool;
+
+    /** Used to monitor the current "tick" of the server */
+    private final TickCounter tickCounter;
+
+    /** The instance used to group threads together */
+    private final ThreadGrouper threadGrouper;
+
+    /** The interval to wait between sampling, in milliseconds */
+    private final int interval;
+
+    /** Tick durations under this threshold will not be inserted */
+    private final int tickLengthThreshold;
+
+    /** The expected number of samples in each tick */
+    private final int expectedSize;
+
+    // state
+    private long currentTick = -1;
+    private TickList currentData = new TickList(0);
+
+    public TickedDataAggregator(ExecutorService workerPool, TickCounter tickCounter, ThreadGrouper threadGrouper, int interval, int tickLengthThreshold) {
+        this.workerPool = workerPool;
+        this.tickCounter = tickCounter;
+        this.threadGrouper = threadGrouper;
+        this.interval = interval;
+        this.tickLengthThreshold = tickLengthThreshold;
+        // 50 millis in a tick, plus 10 so we have a bit of room to go over
+        this.expectedSize = (50 / interval) + 10;
+    }
+
+    // this is effectively synchronized by the Timer instance in Sampler
+    @Override
+    public void insertData(String threadName, StackTraceElement[] stack) {
+        long tick = this.tickCounter.getCurrentTick();
+        if (this.currentTick != tick) {
+            pushCurrentTick();
+            this.currentTick = tick;
+            this.currentData = new TickList(this.expectedSize);
+        }
+
+        // form the queued data
+        QueuedThreadInfo queuedData = new QueuedThreadInfo(threadName, stack);
+        // insert it
+        this.currentData.addData(queuedData);
+    }
+
+    private void pushCurrentTick() {
+        TickList currentData = this.currentData;
+
+        // approximate how long the tick lasted
+        int tickLengthMillis = currentData.getList().size() * this.interval;
+
+        // don't push data below the threshold
+        if (tickLengthMillis < this.tickLengthThreshold) {
+            return;
+        }
+
+        this.workerPool.submit(currentData);
+    }
+
+    @Override
+    public void start() {
+        this.tickCounter.start();
+    }
+
+    @Override
+    public Map<String, StackNode> getData() {
+        // push the current tick
+        pushCurrentTick();
+
+        // close the tick counter
+        this.tickCounter.close();
+
+        // wait for all pending data to be inserted
+        this.workerPool.shutdown();
+        try {
+            this.workerPool.awaitTermination(15, TimeUnit.SECONDS);
+        } catch (InterruptedException e) {
+            e.printStackTrace();
+        }
+
+        return this.threadData;
+    }
+
+    void insertData(List<QueuedThreadInfo> dataList) {
+        for (QueuedThreadInfo data : dataList) {
+            try {
+                String group = this.threadGrouper.getGroup(data.threadName);
+                StackNode node = this.threadData.computeIfAbsent(group, StackNode::new);
+                node.log(data.stack, this.interval);
+            } catch (Exception e) {
+                e.printStackTrace();
+            }
+        }
+    }
+
+    private final class TickList implements Runnable {
+        private final List<QueuedThreadInfo> list;
+
+        TickList(int expectedSize) {
+            this.list = new ArrayList<>(expectedSize);
+        }
+
+        @Override
+        public void run() {
+            insertData(this.list);
+        }
+
+        public List<QueuedThreadInfo> getList() {
+            return this.list;
+        }
+
+        public void addData(QueuedThreadInfo data) {
+            this.list.add(data);
+        }
+    }
+
+    private static final class QueuedThreadInfo {
+        private final String threadName;
+        private final StackTraceElement[] stack;
+
+        QueuedThreadInfo(String threadName, StackTraceElement[] stack) {
+            this.threadName = threadName;
+            this.stack = stack;
+        }
+    }
+}