diff options
Diffstat (limited to 'common/src/main/java/me/lucko/spark/profiler')
6 files changed, 330 insertions, 62 deletions
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; + } + } +} |