diff options
Diffstat (limited to 'spark-common/src/main/java')
28 files changed, 1791 insertions, 255 deletions
diff --git a/spark-common/src/main/java/me/lucko/spark/common/SparkPlatform.java b/spark-common/src/main/java/me/lucko/spark/common/SparkPlatform.java index cf3b0ee..a6bf332 100644 --- a/spark-common/src/main/java/me/lucko/spark/common/SparkPlatform.java +++ b/spark-common/src/main/java/me/lucko/spark/common/SparkPlatform.java @@ -211,7 +211,7 @@ public class SparkPlatform { if (command.aliases().contains(alias)) { try { command.executor().execute(this, sender, resp, new Arguments(rawArgs)); - } catch (IllegalArgumentException e) { + } catch (Arguments.ParseException e) { resp.replyPrefixed(text(e.getMessage(), RED)); } return; diff --git a/spark-common/src/main/java/me/lucko/spark/common/command/Arguments.java b/spark-common/src/main/java/me/lucko/spark/common/command/Arguments.java index 2b202af..3cd0365 100644 --- a/spark-common/src/main/java/me/lucko/spark/common/command/Arguments.java +++ b/spark-common/src/main/java/me/lucko/spark/common/command/Arguments.java @@ -51,7 +51,7 @@ public class Arguments { if (flag == null || matches) { if (!matches) { - throw new IllegalArgumentException("Expected flag at position " + i + " but got '" + arg + "' instead!"); + throw new ParseException("Expected flag at position " + i + " but got '" + arg + "' instead!"); } // store existing value, if present @@ -83,7 +83,7 @@ public class Arguments { try { return Math.abs(Integer.parseInt(it.next())); } catch (NumberFormatException e) { - throw new IllegalArgumentException("Invalid input for '" + key + "' argument. Please specify a number!"); + throw new ParseException("Invalid input for '" + key + "' argument. Please specify a number!"); } } return -1; // undefined @@ -95,7 +95,7 @@ public class Arguments { try { return Math.abs(Double.parseDouble(it.next())); } catch (NumberFormatException e) { - throw new IllegalArgumentException("Invalid input for '" + key + "' argument. Please specify a number!"); + throw new ParseException("Invalid input for '" + key + "' argument. Please specify a number!"); } } return -1; // undefined @@ -108,4 +108,21 @@ public class Arguments { public boolean boolFlag(String key) { return this.parsedArgs.containsKey(key); } + + public static final class ParseException extends IllegalArgumentException { + public ParseException() { + } + + public ParseException(String s) { + super(s); + } + + public ParseException(String message, Throwable cause) { + super(message, cause); + } + + public ParseException(Throwable cause) { + super(cause); + } + } } 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 eb77b24..cce3169 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 @@ -33,6 +33,7 @@ import me.lucko.spark.common.sampler.SamplerBuilder; import me.lucko.spark.common.sampler.ThreadDumper; import me.lucko.spark.common.sampler.ThreadGrouper; import me.lucko.spark.common.sampler.ThreadNodeOrder; +import me.lucko.spark.common.sampler.async.AsyncSampler; import me.lucko.spark.common.sampler.node.MergeMode; import me.lucko.spark.common.sampler.tick.TickHook; import me.lucko.spark.common.util.MethodDisambiguator; @@ -55,13 +56,13 @@ import static net.kyori.adventure.text.format.NamedTextColor.*; public class SamplerModule implements CommandModule { private static final MediaType SPARK_SAMPLER_MEDIA_TYPE = MediaType.parse("application/x-spark-sampler"); - /** The WarmRoast instance currently running, if any */ + /** The sampler instance currently running, if any */ private Sampler activeSampler = null; @Override public void close() { if (this.activeSampler != null) { - this.activeSampler.cancel(); + this.activeSampler.stop(); this.activeSampler = null; } } @@ -83,6 +84,7 @@ public class SamplerModule implements CommandModule { .argumentUsage("only-ticks-over", "tick length millis") .argumentUsage("ignore-sleeping", null) .argumentUsage("ignore-native", null) + .argumentUsage("force-java-sampler", null) .argumentUsage("order-by-time", null) .argumentUsage("separate-parent-calls", null) .executor((platform, sender, resp, arguments) -> { @@ -118,7 +120,7 @@ public class SamplerModule implements CommandModule { if (this.activeSampler == null) { resp.replyPrefixed(text("There isn't an active sampling task running.")); } else { - this.activeSampler.cancel(); + this.activeSampler.stop(); resp.broadcastPrefixed(text("The active sampling operation has been stopped! Uploading results...")); ThreadNodeOrder threadOrder = arguments.boolFlag("order-by-time") ? ThreadNodeOrder.BY_TIME : ThreadNodeOrder.BY_NAME; String comment = Iterables.getFirst(arguments.stringFlag("comment"), null); @@ -149,6 +151,7 @@ public class SamplerModule implements CommandModule { boolean ignoreSleeping = arguments.boolFlag("ignore-sleeping"); boolean ignoreNative = arguments.boolFlag("ignore-native"); + boolean forceJavaSampler = arguments.boolFlag("force-java-sampler"); Set<String> threads = arguments.stringFlag("thread"); ThreadDumper threadDumper; @@ -201,19 +204,25 @@ public class SamplerModule implements CommandModule { builder.samplingInterval(intervalMillis); builder.ignoreSleeping(ignoreSleeping); builder.ignoreNative(ignoreNative); + builder.forceJavaSampler(forceJavaSampler); if (ticksOver != -1) { builder.ticksOver(ticksOver, tickHook); } Sampler sampler = this.activeSampler = builder.start(); - resp.broadcastPrefixed(text("Profiler now active!", GOLD)); + resp.broadcastPrefixed(text() + .append(text("Profiler now active!", GOLD)) + .append(space()) + .append(text("(" + (sampler instanceof AsyncSampler ? "async" : "built-in java") + ")", DARK_GRAY)) + .build() + ); if (timeoutSeconds == -1) { resp.broadcastPrefixed(text("Use '/" + platform.getPlugin().getCommandName() + " profiler --stop' to stop profiling and upload the results.")); } else { resp.broadcastPrefixed(text("The results will be automatically returned after the profiler has been running for " + timeoutSeconds + " seconds.")); } - CompletableFuture<Sampler> future = this.activeSampler.getFuture(); + CompletableFuture<? extends Sampler> future = this.activeSampler.getFuture(); // send message if profiling fails future.whenCompleteAsync((s, throwable) -> { @@ -253,8 +262,8 @@ public class SamplerModule implements CommandModule { List<String> opts = new ArrayList<>(Arrays.asList("--info", "--stop", "--cancel", "--timeout", "--regex", "--combine-all", "--not-combined", "--interval", - "--only-ticks-over", "--ignore-sleeping", "--ignore-native", "--order-by-time", - "--separate-parent-calls", "--comment")); + "--only-ticks-over", "--ignore-sleeping", "--ignore-native", "--force-java-sampler", + "--order-by-time", "--separate-parent-calls", "--comment")); opts.removeAll(arguments); opts.add("--thread"); // allowed multiple times diff --git a/spark-common/src/main/java/me/lucko/spark/common/sampler/Sampler.java b/spark-common/src/main/java/me/lucko/spark/common/sampler/Sampler.java index e772cb3..5088ed7 100644 --- a/spark-common/src/main/java/me/lucko/spark/common/sampler/Sampler.java +++ b/spark-common/src/main/java/me/lucko/spark/common/sampler/Sampler.java @@ -1,7 +1,6 @@ /* * This file is part of spark. * - * Copyright (C) Albert Pham <http://www.sk89q.com> * Copyright (c) lucko (Luck) <luck@lucko.me> * Copyright (c) contributors * @@ -21,174 +20,66 @@ package me.lucko.spark.common.sampler; -import com.google.common.util.concurrent.ThreadFactoryBuilder; import me.lucko.spark.common.command.sender.CommandSender; import me.lucko.spark.common.platform.PlatformInfo; -import me.lucko.spark.common.sampler.aggregator.DataAggregator; -import me.lucko.spark.common.sampler.aggregator.SimpleDataAggregator; -import me.lucko.spark.common.sampler.aggregator.TickedDataAggregator; import me.lucko.spark.common.sampler.node.MergeMode; import me.lucko.spark.common.sampler.node.ThreadNode; -import me.lucko.spark.common.sampler.tick.TickHook; import me.lucko.spark.proto.SparkProtos.SamplerData; -import me.lucko.spark.proto.SparkProtos.SamplerMetadata; import java.io.ByteArrayOutputStream; import java.io.IOException; import java.io.OutputStream; -import java.lang.management.ManagementFactory; -import java.lang.management.ThreadInfo; -import java.lang.management.ThreadMXBean; -import java.util.ArrayList; import java.util.Comparator; -import java.util.List; import java.util.Map; import java.util.concurrent.CompletableFuture; -import java.util.concurrent.Executors; -import java.util.concurrent.ScheduledExecutorService; -import java.util.concurrent.ScheduledFuture; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicInteger; import java.util.zip.GZIPOutputStream; /** - * Main sampler class. + * Abstract superinterface for all sampler implementations. */ -public class Sampler implements Runnable { - private static final AtomicInteger THREAD_ID = new AtomicInteger(0); - - /** The worker pool for inserting stack nodes */ - private final ScheduledExecutorService workerPool = Executors.newScheduledThreadPool( - 6, new ThreadFactoryBuilder().setNameFormat("spark-worker-" + THREAD_ID.getAndIncrement() + "-%d").build() - ); - - /** The main sampling task */ - private ScheduledFuture<?> task; - - /** 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 microseconds */ - private final int interval; - /** 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, boolean ignoreSleeping, boolean ignoreNative) { - this.threadDumper = threadDumper; - this.dataAggregator = new SimpleDataAggregator(this.workerPool, threadGrouper, interval, ignoreSleeping, ignoreNative); - this.interval = interval; - this.endTime = endTime; - } - - public Sampler(int interval, ThreadDumper threadDumper, ThreadGrouper threadGrouper, long endTime, boolean ignoreSleeping, boolean ignoreNative, TickHook tickHook, int tickLengthThreshold) { - this.threadDumper = threadDumper; - this.dataAggregator = new TickedDataAggregator(this.workerPool, threadGrouper, interval, ignoreSleeping, ignoreNative, tickHook, tickLengthThreshold); - this.interval = interval; - this.endTime = endTime; - } +public interface Sampler { /** * Starts the sampler. */ - public void start() { - this.startTime = System.currentTimeMillis(); - this.task = this.workerPool.scheduleAtFixedRate(this, 0, this.interval, TimeUnit.MICROSECONDS); - } - - public long getStartTime() { - if (this.startTime == -1) { - throw new IllegalStateException("Not yet started"); - } - return this.startTime; - } - - public long getEndTime() { - return this.endTime; - } - - public CompletableFuture<Sampler> getFuture() { - return this.future; - } + void start(); - public void cancel() { - this.task.cancel(false); - } - - @Override - public void run() { - // this is effectively synchronized, the worker pool will not allow this task - // to concurrently execute. - try { - if (this.endTime != -1 && this.endTime <= System.currentTimeMillis()) { - this.future.complete(this); - cancel(); - return; - } - - ThreadInfo[] threadDumps = this.threadDumper.dumpThreads(this.threadBean); - this.workerPool.execute(new InsertDataTask(this.dataAggregator, threadDumps)); - } catch (Throwable t) { - this.future.completeExceptionally(t); - cancel(); - } - } - - private static final class InsertDataTask implements Runnable { - private final DataAggregator dataAggregator; - private final ThreadInfo[] threadDumps; - - InsertDataTask(DataAggregator dataAggregator, ThreadInfo[] threadDumps) { - this.dataAggregator = dataAggregator; - this.threadDumps = threadDumps; - } - - @Override - public void run() { - for (ThreadInfo threadInfo : this.threadDumps) { - if (threadInfo.getThreadName() == null || threadInfo.getStackTrace() == null) { - continue; - } - this.dataAggregator.insertData(threadInfo); - } - } - } - - private SamplerData toProto(PlatformInfo platformInfo, CommandSender creator, Comparator<? super Map.Entry<String, ThreadNode>> outputOrder, String comment, MergeMode mergeMode) { - final SamplerMetadata.Builder metadata = SamplerMetadata.newBuilder() - .setPlatform(platformInfo.toData().toProto()) - .setUser(creator.toData().toProto()) - .setStartTime(this.startTime) - .setInterval(this.interval) - .setThreadDumper(this.threadDumper.getMetadata()) - .setDataAggregator(this.dataAggregator.getMetadata()); - - if (comment != null) { - metadata.setComment(comment); - } - - SamplerData.Builder proto = SamplerData.newBuilder(); - proto.setMetadata(metadata.build()); + /** + * Stops the sampler. + */ + void stop(); - List<Map.Entry<String, ThreadNode>> data = new ArrayList<>(this.dataAggregator.getData().entrySet()); - data.sort(outputOrder); + /** + * Gets the time when the sampler started (unix timestamp in millis) + * + * @return the start time + */ + long getStartTime(); - for (Map.Entry<String, ThreadNode> entry : data) { - proto.addThreads(entry.getValue().toProto(mergeMode)); - } + /** + * Gets the time when the sampler should automatically stop (unix timestamp in millis) + * + * @return the end time, or -1 if undefined + */ + long getEndTime(); - return proto.build(); - } + /** + * Gets a future to encapsulate the completion of the sampler + * + * @return a future + */ + CompletableFuture<? extends Sampler> getFuture(); + + // Methods used to export the sampler data to the web viewer. + SamplerData toProto( + PlatformInfo platformInfo, + CommandSender creator, + Comparator<? super Map.Entry<String, ThreadNode>> outputOrder, + String comment, + MergeMode mergeMode + ); - public byte[] formCompressedDataPayload(PlatformInfo platformInfo, CommandSender creator, Comparator<? super Map.Entry<String, ThreadNode>> outputOrder, String comment, MergeMode mergeMode) { + default byte[] formCompressedDataPayload(PlatformInfo platformInfo, CommandSender creator, Comparator<? super Map.Entry<String, ThreadNode>> outputOrder, String comment, MergeMode mergeMode) { SamplerData proto = toProto(platformInfo, creator, outputOrder, comment, mergeMode); ByteArrayOutputStream byteOut = new ByteArrayOutputStream(); 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 8993445..7abe1a7 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 @@ -20,6 +20,9 @@ package me.lucko.spark.common.sampler; +import me.lucko.spark.common.sampler.async.AsyncProfilerAccess; +import me.lucko.spark.common.sampler.async.AsyncSampler; +import me.lucko.spark.common.sampler.java.JavaSampler; import me.lucko.spark.common.sampler.tick.TickHook; import java.util.concurrent.TimeUnit; @@ -32,6 +35,7 @@ public class SamplerBuilder { private double samplingInterval = 4; // milliseconds private boolean ignoreSleeping = false; private boolean ignoreNative = false; + private boolean useAsyncProfiler = true; private long timeout = -1; private ThreadDumper threadDumper = ThreadDumper.ALL; private ThreadGrouper threadGrouper = ThreadGrouper.BY_NAME; @@ -81,14 +85,23 @@ public class SamplerBuilder { return this; } + public SamplerBuilder forceJavaSampler(boolean forceJavaSampler) { + this.useAsyncProfiler = !forceJavaSampler; + return this; + } + public Sampler start() { Sampler sampler; int intervalMicros = (int) (this.samplingInterval * 1000d); if (this.ticksOver == -1 || this.tickHook == null) { - sampler = new Sampler(intervalMicros, this.threadDumper, this.threadGrouper, this.timeout, this.ignoreSleeping, this.ignoreNative); + if (this.useAsyncProfiler && this.timeout == -1 && !(this.threadDumper instanceof ThreadDumper.Regex) && AsyncProfilerAccess.INSTANCE.isSupported()) { + sampler = new AsyncSampler(intervalMicros, this.threadDumper, this.threadGrouper); + } else { + sampler = new JavaSampler(intervalMicros, this.threadDumper, this.threadGrouper, this.timeout, this.ignoreSleeping, this.ignoreNative); + } } else { - sampler = new Sampler(intervalMicros, this.threadDumper, this.threadGrouper, this.timeout, this.ignoreSleeping, this.ignoreNative, this.tickHook, this.ticksOver); + sampler = new JavaSampler(intervalMicros, this.threadDumper, this.threadGrouper, this.timeout, this.ignoreSleeping, this.ignoreNative, this.tickHook, this.ticksOver); } sampler.start(); diff --git a/spark-common/src/main/java/me/lucko/spark/common/sampler/ThreadDumper.java b/spark-common/src/main/java/me/lucko/spark/common/sampler/ThreadDumper.java index 224918e..4863482 100644 --- a/spark-common/src/main/java/me/lucko/spark/common/sampler/ThreadDumper.java +++ b/spark-common/src/main/java/me/lucko/spark/common/sampler/ThreadDumper.java @@ -76,17 +76,38 @@ public interface ThreadDumper { */ final class Specific implements ThreadDumper { private final long[] ids; + private Set<Thread> threads; + private Set<String> threadNamesLowerCase; public Specific(long[] ids) { this.ids = ids; } public Specific(Set<String> names) { - Set<String> namesLower = names.stream().map(String::toLowerCase).collect(Collectors.toSet()); + this.threadNamesLowerCase = names.stream().map(String::toLowerCase).collect(Collectors.toSet()); this.ids = new ThreadFinder().getThreads() - .filter(t -> namesLower.contains(t.getName().toLowerCase())) + .filter(t -> this.threadNamesLowerCase.contains(t.getName().toLowerCase())) .mapToLong(Thread::getId) .toArray(); + Arrays.sort(this.ids); + } + + public Set<Thread> getThreads() { + if (this.threads == null) { + this.threads = new ThreadFinder().getThreads() + .filter(t -> Arrays.binarySearch(this.ids, t.getId()) >= 0) + .collect(Collectors.toSet()); + } + return this.threads; + } + + public Set<String> getThreadNames() { + if (this.threadNamesLowerCase == null) { + this.threadNamesLowerCase = getThreads().stream() + .map(t -> t.getName().toLowerCase()) + .collect(Collectors.toSet()); + } + return this.threadNamesLowerCase; } @Override diff --git a/spark-common/src/main/java/me/lucko/spark/common/sampler/aggregator/AbstractDataAggregator.java b/spark-common/src/main/java/me/lucko/spark/common/sampler/aggregator/AbstractDataAggregator.java index c9a4f37..7640d60 100644 --- a/spark-common/src/main/java/me/lucko/spark/common/sampler/aggregator/AbstractDataAggregator.java +++ b/spark-common/src/main/java/me/lucko/spark/common/sampler/aggregator/AbstractDataAggregator.java @@ -23,37 +23,22 @@ package me.lucko.spark.common.sampler.aggregator; import me.lucko.spark.common.sampler.ThreadGrouper; import me.lucko.spark.common.sampler.node.ThreadNode; -import java.lang.management.ThreadInfo; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ExecutorService; +/** + * Abstract implementation of {@link DataAggregator}. + */ public abstract class AbstractDataAggregator implements DataAggregator { /** A map of root stack nodes for each thread with sampling data */ protected final Map<String, ThreadNode> threadData = new ConcurrentHashMap<>(); - /** The worker pool for inserting stack nodes */ - protected final ExecutorService workerPool; - /** The instance used to group threads together */ protected final ThreadGrouper threadGrouper; - /** The interval to wait between sampling, in microseconds */ - protected final int interval; - - /** If sleeping threads should be ignored */ - private final boolean ignoreSleeping; - - /** If threads executing native code should be ignored */ - private final boolean ignoreNative; - - public AbstractDataAggregator(ExecutorService workerPool, ThreadGrouper threadGrouper, int interval, boolean ignoreSleeping, boolean ignoreNative) { - this.workerPool = workerPool; + protected AbstractDataAggregator(ThreadGrouper threadGrouper) { this.threadGrouper = threadGrouper; - this.interval = interval; - this.ignoreSleeping = ignoreSleeping; - this.ignoreNative = ignoreNative; } protected ThreadNode getNode(String group) { @@ -64,42 +49,8 @@ public abstract class AbstractDataAggregator implements DataAggregator { return this.threadData.computeIfAbsent(group, ThreadNode::new); } - protected void writeData(ThreadInfo threadInfo) { - if (this.ignoreSleeping && isSleeping(threadInfo)) { - return; - } - if (this.ignoreNative && threadInfo.isInNative()) { - return; - } - - try { - ThreadNode node = getNode(this.threadGrouper.getGroup(threadInfo.getThreadId(), threadInfo.getThreadName())); - node.log(threadInfo.getStackTrace(), this.interval); - } catch (Exception e) { - e.printStackTrace(); - } + @Override + public Map<String, ThreadNode> getData() { + return this.threadData; } - - private static boolean isSleeping(ThreadInfo thread) { - if (thread.getThreadState() == Thread.State.WAITING || thread.getThreadState() == Thread.State.TIMED_WAITING) { - return true; - } - - StackTraceElement[] stackTrace = thread.getStackTrace(); - if (stackTrace.length == 0) { - return false; - } - - StackTraceElement call = stackTrace[0]; - String clazz = call.getClassName(); - String method = call.getMethodName(); - - // java.lang.Thread.yield() - // jdk.internal.misc.Unsafe.park() - // sun.misc.Unsafe.park() - return (clazz.equals("java.lang.Thread") && method.equals("yield")) || - (clazz.equals("jdk.internal.misc.Unsafe") && method.equals("park")) || - (clazz.equals("sun.misc.Unsafe") && method.equals("park")); - } - } diff --git a/spark-common/src/main/java/me/lucko/spark/common/sampler/aggregator/DataAggregator.java b/spark-common/src/main/java/me/lucko/spark/common/sampler/aggregator/DataAggregator.java index 8318fbd..a91a998 100644 --- a/spark-common/src/main/java/me/lucko/spark/common/sampler/aggregator/DataAggregator.java +++ b/spark-common/src/main/java/me/lucko/spark/common/sampler/aggregator/DataAggregator.java @@ -39,13 +39,6 @@ public interface DataAggregator { Map<String, ThreadNode> getData(); /** - * Inserts sampling data into this aggregator - * - * @param threadInfo the thread info - */ - void insertData(ThreadInfo threadInfo); - - /** * Gets metadata about the data aggregator instance. */ SamplerMetadata.DataAggregator getMetadata(); diff --git a/spark-common/src/main/java/me/lucko/spark/common/sampler/async/AsyncDataAggregator.java b/spark-common/src/main/java/me/lucko/spark/common/sampler/async/AsyncDataAggregator.java new file mode 100644 index 0000000..cb3bd6f --- /dev/null +++ b/spark-common/src/main/java/me/lucko/spark/common/sampler/async/AsyncDataAggregator.java @@ -0,0 +1,53 @@ +/* + * This file is part of spark. + * + * Copyright (c) lucko (Luck) <luck@lucko.me> + * 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.aggregator.AbstractDataAggregator; +import me.lucko.spark.common.sampler.node.ThreadNode; +import me.lucko.spark.proto.SparkProtos.SamplerMetadata; + +/** + * Data aggregator for {@link AsyncSampler}. + */ +public class AsyncDataAggregator extends AbstractDataAggregator { + protected AsyncDataAggregator(ThreadGrouper threadGrouper) { + super(threadGrouper); + } + + @Override + public SamplerMetadata.DataAggregator getMetadata() { + return SamplerMetadata.DataAggregator.newBuilder() + .setType(SamplerMetadata.DataAggregator.Type.SIMPLE) + .setThreadGrouper(this.threadGrouper.asProto()) + .build(); + } + + public void insertData(ProfileSegment element) { + try { + ThreadNode node = getNode(this.threadGrouper.getGroup(element.getNativeThreadId(), element.getThreadName())); + node.log(element.getStackTrace(), element.getTime()); + } catch (Exception e) { + e.printStackTrace(); + } + } + +} 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 new file mode 100644 index 0000000..3d53903 --- /dev/null +++ b/spark-common/src/main/java/me/lucko/spark/common/sampler/async/AsyncProfilerAccess.java @@ -0,0 +1,117 @@ +/* + * This file is part of spark. + * + * Copyright (c) lucko (Luck) <luck@lucko.me> + * 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 one.profiler.AsyncProfiler; + +import java.io.InputStream; +import java.net.URL; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.StandardCopyOption; + +/** + * Provides a bridge between spark and async-profiler. + */ +public final class AsyncProfilerAccess { + + // Singleton + public static final AsyncProfilerAccess INSTANCE = new AsyncProfilerAccess(); + + // Only support Linux x86_64 + private static final String SUPPORTED_OS = "linux"; + private static final String SUPPORTED_ARCH = "amd64"; + + private static AsyncProfiler load() throws Exception { + // check compatibility + String os = System.getProperty("os.name"); + if (!SUPPORTED_OS.equalsIgnoreCase(os)) { + throw new UnsupportedOperationException("Only supported on Linux x86_64, your OS: " + os); + } + + String arch = System.getProperty("os.arch"); + if (!SUPPORTED_ARCH.equalsIgnoreCase(arch)) { + throw new UnsupportedOperationException("Only supported on Linux x86_64, your arch: " + os); + } + + // extract the profiler binary from the spark jar file + URL profilerResource = AsyncProfilerAccess.class.getClassLoader().getResource("libasyncProfiler.so"); + if (profilerResource == null) { + throw new IllegalStateException("Could not find libasyncProfiler.so in spark jar file"); + } + + Path extractPath = Files.createTempFile("spark-", "-libasyncProfiler.so.tmp"); + extractPath.toFile().deleteOnExit(); + + try (InputStream in = profilerResource.openStream()) { + Files.copy(in, extractPath, StandardCopyOption.REPLACE_EXISTING); + } + + // get an instance of async-profiler + return AsyncProfiler.getInstance(extractPath.toAbsolutePath().toString()); + } + + /** An instance of the async-profiler Java API. */ + private final AsyncProfiler profiler; + + /** If profiler is null, contains the reason why setup failed */ + private final Exception setupException; + + private AsyncProfilerAccess() { + AsyncProfiler profiler; + Exception setupException = null; + + try { + profiler = load(); + ensureCpuEventSupported(profiler); + } catch (Exception e) { + profiler = null; + setupException = e; + } + + this.profiler = profiler; + this.setupException = setupException; + } + + /** + * Checks the {@code profiler} to ensure the CPU event is supported. + * + * @param profiler the profiler instance + * @throws Exception if the event is not supported + */ + private static void ensureCpuEventSupported(AsyncProfiler profiler) throws Exception { + String resp = profiler.execute("check,event=cpu").trim(); + if (!resp.equalsIgnoreCase("ok")) { + throw new UnsupportedOperationException("CPU event is not supported"); + } + } + + public AsyncProfiler getProfiler() { + if (this.profiler == null) { + throw new UnsupportedOperationException("async-profiler not supported", this.setupException); + } + return this.profiler; + } + + public boolean isSupported() { + return this.profiler != null; + } +} diff --git a/spark-common/src/main/java/me/lucko/spark/common/sampler/async/AsyncSampler.java b/spark-common/src/main/java/me/lucko/spark/common/sampler/async/AsyncSampler.java new file mode 100644 index 0000000..a109be7 --- /dev/null +++ b/spark-common/src/main/java/me/lucko/spark/common/sampler/async/AsyncSampler.java @@ -0,0 +1,279 @@ +/* + * This file is part of spark. + * + * Copyright (c) lucko (Luck) <luck@lucko.me> + * 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.command.sender.CommandSender; +import me.lucko.spark.common.platform.PlatformInfo; +import me.lucko.spark.common.sampler.Sampler; +import me.lucko.spark.common.sampler.ThreadDumper; +import me.lucko.spark.common.sampler.ThreadGrouper; +import me.lucko.spark.common.sampler.async.jfr.ClassRef; +import me.lucko.spark.common.sampler.async.jfr.JfrReader; +import me.lucko.spark.common.sampler.async.jfr.MethodRef; +import me.lucko.spark.common.sampler.async.jfr.Sample; +import me.lucko.spark.common.sampler.async.jfr.StackTrace; +import me.lucko.spark.common.sampler.node.MergeMode; +import me.lucko.spark.common.sampler.node.ThreadNode; +import me.lucko.spark.proto.SparkProtos; + +import one.profiler.AsyncProfiler; + +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.Comparator; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; +import java.util.function.Predicate; + +/** + * A sampler implementation using async-profiler. + */ +public class AsyncSampler implements Sampler { + private final AsyncProfiler profiler; + + /** 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 AsyncDataAggregator dataAggregator; + /** Flag to mark if the output has been completed */ + private boolean outputComplete = false; + /** The temporary output file */ + private Path outputFile; + + /** The interval to wait between sampling, in microseconds */ + private final int interval; + /** The time when sampling first began */ + private long startTime = -1; + + public AsyncSampler(int interval, ThreadDumper threadDumper, ThreadGrouper threadGrouper) { + this.profiler = AsyncProfilerAccess.INSTANCE.getProfiler(); + this.threadDumper = threadDumper; + this.dataAggregator = new AsyncDataAggregator(threadGrouper); + this.interval = interval; + } + + @Override + public long getStartTime() { + if (this.startTime == -1) { + throw new IllegalStateException("Not yet started"); + } + return this.startTime; + } + + @Override + public long getEndTime() { + return -1; + } + + @Override + public CompletableFuture<? extends Sampler> getFuture() { + return new CompletableFuture<>(); + } + + /** + * Executes a profiler command. + * + * @param command the command to execute + * @return the response + */ + private String execute(String command) { + try { + return this.profiler.execute(command); + } catch (IOException e) { + throw new RuntimeException("Exception whilst executing profiler command", e); + } + } + + /** + * Starts the profiler. + */ + @Override + public void start() { + this.startTime = System.currentTimeMillis(); + + try { + this.outputFile = Files.createTempFile("spark-profile-", ".jfr.tmp"); + this.outputFile.toFile().deleteOnExit(); + } catch (IOException e) { + throw new RuntimeException("Unable to create temporary output file", e); + } + + String command = "start,event=cpu,interval=" + this.interval + "us,threads,jfr,file=" + this.outputFile.toString(); + if (this.threadDumper instanceof ThreadDumper.Specific) { + command += ",filter"; + } + + String resp = execute(command).trim(); + if (!resp.equalsIgnoreCase("profiling started")) { + throw new RuntimeException("Unexpected response: " + resp); + } + + if (this.threadDumper instanceof ThreadDumper.Specific) { + ThreadDumper.Specific threadDumper = (ThreadDumper.Specific) this.threadDumper; + for (Thread thread : threadDumper.getThreads()) { + this.profiler.addThread(thread); + } + } + } + + /** + * Stops the profiler. + */ + @Override + public void stop() { + this.profiler.stop(); + } + + @Override + public SparkProtos.SamplerData toProto(PlatformInfo platformInfo, CommandSender creator, Comparator<? super Map.Entry<String, ThreadNode>> outputOrder, String comment, MergeMode mergeMode) { + final SparkProtos.SamplerMetadata.Builder metadata = SparkProtos.SamplerMetadata.newBuilder() + .setPlatform(platformInfo.toData().toProto()) + .setUser(creator.toData().toProto()) + .setStartTime(this.startTime) + .setInterval(this.interval) + .setThreadDumper(this.threadDumper.getMetadata()) + .setDataAggregator(this.dataAggregator.getMetadata()); + + if (comment != null) { + metadata.setComment(comment); + } + + SparkProtos.SamplerData.Builder proto = SparkProtos.SamplerData.newBuilder(); + proto.setMetadata(metadata.build()); + + aggregateOutput(); + + List<Map.Entry<String, ThreadNode>> data = new ArrayList<>(this.dataAggregator.getData().entrySet()); + data.sort(outputOrder); + + for (Map.Entry<String, ThreadNode> entry : data) { + proto.addThreads(entry.getValue().toProto(mergeMode)); + } + + return proto.build(); + } + + private void aggregateOutput() { + if (this.outputComplete) { + return; + } + this.outputComplete = true; + + Predicate<String> threadFilter; + if (this.threadDumper instanceof ThreadDumper.Specific) { + ThreadDumper.Specific threadDumper = (ThreadDumper.Specific) this.threadDumper; + threadFilter = n -> threadDumper.getThreadNames().contains(n.toLowerCase()); + } else { + threadFilter = n -> true; + } + + // read the jfr file produced by async-profiler + try (JfrReader reader = new JfrReader(this.outputFile)) { + readSegments(reader, threadFilter); + } catch (IOException e) { + throw new RuntimeException("Read error", e); + } + + // delete the output file after reading + try { + Files.deleteIfExists(this.outputFile); + } catch (IOException e) { + // ignore + } + } + + private void readSegments(JfrReader reader, Predicate<String> threadFilter) { + List<Sample> samples = reader.samples; + for (int i = 0; i < samples.size(); i++) { + Sample sample = samples.get(i); + + long duration; + if (i == 0) { + // we don't really know the duration of the first sample, so just use the sampling + // interval + duration = this.interval; + } else { + // calculate the duration of the sample by calculating the time elapsed since the + // previous sample + duration = TimeUnit.NANOSECONDS.toMicros(sample.time - samples.get(i - 1).time); + } + + String threadName = reader.threads.get(sample.tid); + if (!threadFilter.test(threadName)) { + continue; + } + + // parse the segment and give it to the data aggregator + ProfileSegment segment = parseSegment(reader, sample, threadName, duration); + this.dataAggregator.insertData(segment); + } + } + + private static ProfileSegment parseSegment(JfrReader reader, Sample sample, String threadName, long duration) { + StackTrace stackTrace = reader.stackTraces.get(sample.stackTraceId); + int len = stackTrace.methods.length; + + AsyncStackTraceElement[] stack = new AsyncStackTraceElement[len]; + for (int i = 0; i < len; i++) { + stack[i] = parseStackFrame(reader, stackTrace.methods[i]); + } + + return new ProfileSegment(sample.tid, threadName, stack, duration); + } + + private static AsyncStackTraceElement parseStackFrame(JfrReader reader, long methodId) { + AsyncStackTraceElement result = reader.stackFrames.get(methodId); + if (result != null) { + return result; + } + + MethodRef methodRef = reader.methods.get(methodId); + ClassRef classRef = reader.classes.get(methodRef.cls); + + byte[] className = reader.symbols.get(classRef.name); + byte[] methodName = reader.symbols.get(methodRef.name); + + if (className == null || className.length == 0) { + // native call + result = new AsyncStackTraceElement( + "native", + new String(methodName, StandardCharsets.UTF_8), + null + ); + } else { + // java method + byte[] methodDesc = reader.symbols.get(methodRef.sig); + result = new AsyncStackTraceElement( + new String(className, StandardCharsets.UTF_8).replace('/', '.'), + new String(methodName, StandardCharsets.UTF_8), + new String(methodDesc, StandardCharsets.UTF_8) + ); + } + + reader.stackFrames.put(methodId, result); + return result; + } +} diff --git a/spark-common/src/main/java/me/lucko/spark/common/sampler/async/AsyncStackTraceElement.java b/spark-common/src/main/java/me/lucko/spark/common/sampler/async/AsyncStackTraceElement.java new file mode 100644 index 0000000..cf66ded --- /dev/null +++ b/spark-common/src/main/java/me/lucko/spark/common/sampler/async/AsyncStackTraceElement.java @@ -0,0 +1,52 @@ +/* + * This file is part of spark. + * + * Copyright (c) lucko (Luck) <luck@lucko.me> + * 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; + +/** + * Version of {@link StackTraceElement} for async-profiler output. + */ +public class AsyncStackTraceElement { + + /** The name of the class */ + private final String className; + /** The name of the method */ + private final String methodName; + /** The method description */ + private final String methodDescription; + + public AsyncStackTraceElement(String className, String methodName, String methodDescription) { + this.className = className; + this.methodName = methodName; + this.methodDescription = methodDescription; + } + + public String getClassName() { + return this.className; + } + + public String getMethodName() { + return this.methodName; + } + + public String getMethodDescription() { + return this.methodDescription; + } +} diff --git a/spark-common/src/main/java/me/lucko/spark/common/sampler/async/ProfileSegment.java b/spark-common/src/main/java/me/lucko/spark/common/sampler/async/ProfileSegment.java new file mode 100644 index 0000000..154e6fe --- /dev/null +++ b/spark-common/src/main/java/me/lucko/spark/common/sampler/async/ProfileSegment.java @@ -0,0 +1,61 @@ +/* + * This file is part of spark. + * + * Copyright (c) lucko (Luck) <luck@lucko.me> + * 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; + +/** + * Represents a profile "segment". + * + * <p>async-profiler groups unique stack traces together per-thread in its output.</p> + */ +public class ProfileSegment { + + /** The native thread id (does not correspond to Thread#getId) */ + private final int nativeThreadId; + /** The name of the thread */ + private final String threadName; + /** The stack trace for this segment */ + private final AsyncStackTraceElement[] stackTrace; + /** The time spent executing this segment in microseconds */ + private final long time; + + public ProfileSegment(int nativeThreadId, String threadName, AsyncStackTraceElement[] stackTrace, long time) { + this.nativeThreadId = nativeThreadId; + this.threadName = threadName; + this.stackTrace = stackTrace; + this.time = time; + } + + public int getNativeThreadId() { + return this.nativeThreadId; + } + + public String getThreadName() { + return this.threadName; + } + + public AsyncStackTraceElement[] getStackTrace() { + return this.stackTrace; + } + + public long getTime() { + return this.time; + } +} diff --git a/spark-common/src/main/java/me/lucko/spark/common/sampler/async/jfr/ClassRef.java b/spark-common/src/main/java/me/lucko/spark/common/sampler/async/jfr/ClassRef.java new file mode 100644 index 0000000..2366fa6 --- /dev/null +++ b/spark-common/src/main/java/me/lucko/spark/common/sampler/async/jfr/ClassRef.java @@ -0,0 +1,25 @@ +/* + * Copyright 2020 Andrei Pangin + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package me.lucko.spark.common.sampler.async.jfr; + +public class ClassRef { + public final long name; + + public ClassRef(long name) { + this.name = name; + } +} diff --git a/spark-common/src/main/java/me/lucko/spark/common/sampler/async/jfr/Dictionary.java b/spark-common/src/main/java/me/lucko/spark/common/sampler/async/jfr/Dictionary.java new file mode 100644 index 0000000..55f78ba --- /dev/null +++ b/spark-common/src/main/java/me/lucko/spark/common/sampler/async/jfr/Dictionary.java @@ -0,0 +1,107 @@ +/* + * Copyright 2020 Andrei Pangin + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package me.lucko.spark.common.sampler.async.jfr; + +/** + * Fast and compact long->Object map. + */ +public class Dictionary<T> { + private static final int INITIAL_CAPACITY = 16; + + private long[] keys; + private Object[] values; + private int size; + + public Dictionary() { + this.keys = new long[INITIAL_CAPACITY]; + this.values = new Object[INITIAL_CAPACITY]; + } + + public void put(long key, T value) { + if (key == 0) { + throw new IllegalArgumentException("Zero key not allowed"); + } + + if (++size * 2 > keys.length) { + resize(keys.length * 2); + } + + int mask = keys.length - 1; + int i = hashCode(key) & mask; + while (keys[i] != 0 && keys[i] != key) { + i = (i + 1) & mask; + } + keys[i] = key; + values[i] = value; + } + + @SuppressWarnings("unchecked") + public T get(long key) { + int mask = keys.length - 1; + int i = hashCode(key) & mask; + while (keys[i] != key && keys[i] != 0) { + i = (i + 1) & mask; + } + return (T) values[i]; + } + + @SuppressWarnings("unchecked") + public void forEach(Visitor<T> visitor) { + for (int i = 0; i < keys.length; i++) { + if (keys[i] != 0) { + visitor.visit(keys[i], (T) values[i]); + } + } + } + + public int preallocate(int count) { + int newSize = size + count; + if (newSize * 2 > keys.length) { + resize(Integer.highestOneBit(newSize * 4 - 1)); + } + return count; + } + + private void resize(int newCapacity) { + long[] newKeys = new long[newCapacity]; + Object[] newValues = new Object[newCapacity]; + int mask = newKeys.length - 1; + + for (int i = 0; i < keys.length; i++) { + if (keys[i] != 0) { + for (int j = hashCode(keys[i]) & mask; ; j = (j + 1) & mask) { + if (newKeys[j] == 0) { + newKeys[j] = keys[i]; + newValues[j] = values[i]; + break; + } + } + } + } + + keys = newKeys; + values = newValues; + } + + private static int hashCode(long key) { + return (int) (key ^ (key >>> 32)); + } + + public interface Visitor<T> { + void visit(long key, T value); + } +} diff --git a/spark-common/src/main/java/me/lucko/spark/common/sampler/async/jfr/Element.java b/spark-common/src/main/java/me/lucko/spark/common/sampler/async/jfr/Element.java new file mode 100644 index 0000000..9d6b6c7 --- /dev/null +++ b/spark-common/src/main/java/me/lucko/spark/common/sampler/async/jfr/Element.java @@ -0,0 +1,23 @@ +/* + * Copyright 2020 Andrei Pangin + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package me.lucko.spark.common.sampler.async.jfr; + +class Element { + + void addChild(Element e) { + } +} diff --git a/spark-common/src/main/java/me/lucko/spark/common/sampler/async/jfr/JfrClass.java b/spark-common/src/main/java/me/lucko/spark/common/sampler/async/jfr/JfrClass.java new file mode 100644 index 0000000..a171552 --- /dev/null +++ b/spark-common/src/main/java/me/lucko/spark/common/sampler/async/jfr/JfrClass.java @@ -0,0 +1,49 @@ +/* + * Copyright 2020 Andrei Pangin + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package me.lucko.spark.common.sampler.async.jfr; + +import java.util.ArrayList; +import java.util.List; +import java.util.Map; + +class JfrClass extends Element { + final int id; + final String name; + final List<JfrField> fields; + + JfrClass(Map<String, String> attributes) { + this.id = Integer.parseInt(attributes.get("id")); + this.name = attributes.get("name"); + this.fields = new ArrayList<>(2); + } + + @Override + void addChild(Element e) { + if (e instanceof JfrField) { + fields.add((JfrField) e); + } + } + + JfrField field(String name) { + for (JfrField field : fields) { + if (field.name.equals(name)) { + return field; + } + } + return null; + } +} diff --git a/spark-common/src/main/java/me/lucko/spark/common/sampler/async/jfr/JfrField.java b/spark-common/src/main/java/me/lucko/spark/common/sampler/async/jfr/JfrField.java new file mode 100644 index 0000000..7a78f2c --- /dev/null +++ b/spark-common/src/main/java/me/lucko/spark/common/sampler/async/jfr/JfrField.java @@ -0,0 +1,31 @@ +/* + * Copyright 2020 Andrei Pangin + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package me.lucko.spark.common.sampler.async.jfr; + +import java.util.Map; + +class JfrField extends Element { + final String name; + final int type; + final boolean constantPool; + + JfrField(Map<String, String> attributes) { + this.name = attributes.get("name"); + this.type = Integer.parseInt(attributes.get("class")); + this.constantPool = "true".equals(attributes.get("constantPool")); + } +} 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 new file mode 100644 index 0000000..49a3474 --- /dev/null +++ b/spark-common/src/main/java/me/lucko/spark/common/sampler/async/jfr/JfrReader.java @@ -0,0 +1,389 @@ +/* + * Copyright 2020 Andrei Pangin + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package me.lucko.spark.common.sampler.async.jfr; + +import me.lucko.spark.common.sampler.async.AsyncStackTraceElement; + +import java.io.Closeable; +import java.io.IOException; +import java.nio.ByteBuffer; +import java.nio.channels.FileChannel; +import java.nio.charset.StandardCharsets; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.nio.file.StandardOpenOption; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +/** + * Parses JFR output produced by async-profiler. + */ +public class JfrReader implements Closeable { + private static final int CHUNK_HEADER_SIZE = 68; + private static final int CPOOL_OFFSET = 16; + private static final int META_OFFSET = 24; + + private final FileChannel ch; + private final ByteBuffer buf; + + public final long startNanos; + public final long durationNanos; + public final long startTicks; + public final long ticksPerSec; + + public final Dictionary<JfrClass> types = new Dictionary<>(); + public final Map<String, JfrClass> typesByName = new HashMap<>(); + public final Dictionary<String> threads = new Dictionary<>(); + public final Dictionary<ClassRef> classes = new Dictionary<>(); + public final Dictionary<byte[]> symbols = new Dictionary<>(); + public final Dictionary<MethodRef> methods = new Dictionary<>(); + public final Dictionary<StackTrace> stackTraces = new Dictionary<>(); + public final Dictionary<AsyncStackTraceElement> stackFrames = new Dictionary<>(); // spark + public final Map<Integer, String> frameTypes = new HashMap<>(); + public final Map<Integer, String> threadStates = new HashMap<>(); + public final List<Sample> samples = new ArrayList<>(); + + public JfrReader(Path path) throws IOException { // spark - Path instead of String + this.ch = FileChannel.open(path, StandardOpenOption.READ); // spark - Path instead of String + this.buf = ch.map(FileChannel.MapMode.READ_ONLY, 0, ch.size()); + + if (buf.getInt(0) != 0x464c5200) { + throw new IOException("Not a valid JFR file"); + } + + int version = buf.getInt(4); + if (version < 0x20000 || version > 0x2ffff) { + throw new IOException("Unsupported JFR version: " + (version >>> 16) + "." + (version & 0xffff)); + } + + this.startNanos = buf.getLong(32); + this.durationNanos = buf.getLong(40); + this.startTicks = buf.getLong(48); + this.ticksPerSec = buf.getLong(56); + + readMeta(); + readConstantPool(); + readEvents(); + } + + @Override + public void close() throws IOException { + ch.close(); + } + + private void readMeta() { + buf.position(buf.getInt(META_OFFSET + 4)); + getVarint(); + getVarint(); + getVarlong(); + getVarlong(); + getVarlong(); + + String[] strings = new String[getVarint()]; + for (int i = 0; i < strings.length; i++) { + strings[i] = getString(); + } + readElement(strings); + } + + private Element readElement(String[] strings) { + String name = strings[getVarint()]; + + int attributeCount = getVarint(); + Map<String, String> attributes = new HashMap<>(attributeCount); + for (int i = 0; i < attributeCount; i++) { + attributes.put(strings[getVarint()], strings[getVarint()]); + } + + Element e = createElement(name, attributes); + int childCount = getVarint(); + for (int i = 0; i < childCount; i++) { + e.addChild(readElement(strings)); + } + return e; + } + + private Element createElement(String name, Map<String, String> attributes) { + switch (name) { + case "class": { + JfrClass type = new JfrClass(attributes); + if (!attributes.containsKey("superType")) { + types.put(type.id, type); + } + typesByName.put(type.name, type); + return type; + } + case "field": + return new JfrField(attributes); + default: + return new Element(); + } + } + + private void readConstantPool() { + int offset = buf.getInt(CPOOL_OFFSET + 4); + while (true) { + buf.position(offset); + getVarint(); + getVarint(); + getVarlong(); + getVarlong(); + long delta = getVarlong(); + getVarint(); + + int poolCount = getVarint(); + for (int i = 0; i < poolCount; i++) { + int type = getVarint(); + readConstants(types.get(type)); + } + + if (delta == 0) { + break; + } + offset += delta; + } + } + + private void readConstants(JfrClass type) { + switch (type.name) { + case "jdk.types.ChunkHeader": + buf.position(buf.position() + (CHUNK_HEADER_SIZE + 3)); + break; + case "java.lang.Thread": + readThreads(type.field("group") != null); + break; + case "java.lang.Class": + readClasses(type.field("hidden") != null); + break; + case "jdk.types.Symbol": + readSymbols(); + break; + case "jdk.types.Method": + readMethods(); + break; + case "jdk.types.StackTrace": + readStackTraces(); + break; + case "jdk.types.FrameType": + readMap(frameTypes); + break; + case "jdk.types.ThreadState": + readMap(threadStates); + break; + default: + readOtherConstants(type.fields); + } + } + + private void readThreads(boolean hasGroup) { + int count = threads.preallocate(getVarint()); + for (int i = 0; i < count; i++) { + long id = getVarlong(); + String osName = getString(); + int osThreadId = getVarint(); + String javaName = getString(); + long javaThreadId = getVarlong(); + if (hasGroup) getVarlong(); + threads.put(id, javaName != null ? javaName : osName); + } + } + + private void readClasses(boolean hasHidden) { + int count = classes.preallocate(getVarint()); + for (int i = 0; i < count; i++) { + long id = getVarlong(); + long loader = getVarlong(); + long name = getVarlong(); + long pkg = getVarlong(); + int modifiers = getVarint(); + if (hasHidden) getVarint(); + classes.put(id, new ClassRef(name)); + } + } + + private void readMethods() { + int count = methods.preallocate(getVarint()); + for (int i = 0; i < count; i++) { + long id = getVarlong(); + long cls = getVarlong(); + long name = getVarlong(); + long sig = getVarlong(); + int modifiers = getVarint(); + int hidden = getVarint(); + methods.put(id, new MethodRef(cls, name, sig)); + } + stackFrames.preallocate(count); // spark + } + + private void readStackTraces() { + int count = stackTraces.preallocate(getVarint()); + for (int i = 0; i < count; i++) { + long id = getVarlong(); + int truncated = getVarint(); + StackTrace stackTrace = readStackTrace(); + stackTraces.put(id, stackTrace); + } + } + + private StackTrace readStackTrace() { + int depth = getVarint(); + long[] methods = new long[depth]; + byte[] types = new byte[depth]; + for (int i = 0; i < depth; i++) { + methods[i] = getVarlong(); + int line = getVarint(); + int bci = getVarint(); + types[i] = buf.get(); + } + return new StackTrace(methods, types); + } + + private void readSymbols() { + int count = symbols.preallocate(getVarint()); + for (int i = 0; i < count; i++) { + long id = getVarlong(); + if (buf.get() != 3) { + throw new IllegalArgumentException("Invalid symbol encoding"); + } + symbols.put(id, getBytes()); + } + } + + private void readMap(Map<Integer, String> map) { + int count = getVarint(); + for (int i = 0; i < count; i++) { + map.put(getVarint(), getString()); + } + } + + private void readOtherConstants(List<JfrField> fields) { + int stringType = getTypeId("java.lang.String"); + + boolean[] numeric = new boolean[fields.size()]; + for (int i = 0; i < numeric.length; i++) { + JfrField f = fields.get(i); + numeric[i] = f.constantPool || f.type != stringType; + } + + int count = getVarint(); + for (int i = 0; i < count; i++) { + getVarlong(); + readFields(numeric); + } + } + + private void readFields(boolean[] numeric) { + for (boolean n : numeric) { + if (n) { + getVarlong(); + } else { + getString(); + } + } + } + + private void readEvents() { + int executionSample = getTypeId("jdk.ExecutionSample"); + int nativeMethodSample = getTypeId("jdk.NativeMethodSample"); + + buf.position(CHUNK_HEADER_SIZE); + while (buf.hasRemaining()) { + int position = buf.position(); + int size = getVarint(); + int type = getVarint(); + if (type == executionSample || type == nativeMethodSample) { + readExecutionSample(); + } else { + buf.position(position + size); + } + } + + Collections.sort(samples); + } + + private void readExecutionSample() { + long time = getVarlong(); + int tid = getVarint(); + int stackTraceId = getVarint(); + int threadState = getVarint(); + samples.add(new Sample(time, tid, stackTraceId, threadState)); + + StackTrace stackTrace = stackTraces.get(stackTraceId); + if (stackTrace != null) { + stackTrace.samples++; + } + } + + private int getTypeId(String typeName) { + JfrClass type = typesByName.get(typeName); + return type != null ? type.id : -1; + } + + private int getVarint() { + int result = 0; + for (int shift = 0; ; shift += 7) { + byte b = buf.get(); + result |= (b & 0x7f) << shift; + if (b >= 0) { + return result; + } + } + } + + private long getVarlong() { + long result = 0; + for (int shift = 0; shift < 56; shift += 7) { + byte b = buf.get(); + result |= (b & 0x7fL) << shift; + if (b >= 0) { + return result; + } + } + return result | (buf.get() & 0xffL) << 56; + } + + private String getString() { + switch (buf.get()) { + case 0: + return null; + case 1: + return ""; + case 3: + return new String(getBytes(), StandardCharsets.UTF_8); + case 4: { + char[] chars = new char[getVarint()]; + for (int i = 0; i < chars.length; i++) { + chars[i] = (char) getVarint(); + } + return new String(chars); + } + case 5: + return new String(getBytes(), StandardCharsets.ISO_8859_1); + default: + throw new IllegalArgumentException("Invalid string encoding"); + } + } + + private byte[] getBytes() { + byte[] bytes = new byte[getVarint()]; + buf.get(bytes); + return bytes; + } +} diff --git a/spark-common/src/main/java/me/lucko/spark/common/sampler/async/jfr/MethodRef.java b/spark-common/src/main/java/me/lucko/spark/common/sampler/async/jfr/MethodRef.java new file mode 100644 index 0000000..2f9071e --- /dev/null +++ b/spark-common/src/main/java/me/lucko/spark/common/sampler/async/jfr/MethodRef.java @@ -0,0 +1,29 @@ +/* + * Copyright 2020 Andrei Pangin + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package me.lucko.spark.common.sampler.async.jfr; + +public class MethodRef { + public final long cls; + public final long name; + public final long sig; + + public MethodRef(long cls, long name, long sig) { + this.cls = cls; + this.name = name; + this.sig = sig; + } +} diff --git a/spark-common/src/main/java/me/lucko/spark/common/sampler/async/jfr/Sample.java b/spark-common/src/main/java/me/lucko/spark/common/sampler/async/jfr/Sample.java new file mode 100644 index 0000000..095e261 --- /dev/null +++ b/spark-common/src/main/java/me/lucko/spark/common/sampler/async/jfr/Sample.java @@ -0,0 +1,36 @@ +/* + * Copyright 2020 Andrei Pangin + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package me.lucko.spark.common.sampler.async.jfr; + +public class Sample implements Comparable<Sample> { + public final long time; + public final int tid; + public final int stackTraceId; + public final int threadState; + + public Sample(long time, int tid, int stackTraceId, int threadState) { + this.time = time; + this.tid = tid; + this.stackTraceId = stackTraceId; + this.threadState = threadState; + } + + @Override + public int compareTo(Sample o) { + return Long.compare(time, o.time); + } +} diff --git a/spark-common/src/main/java/me/lucko/spark/common/sampler/async/jfr/StackTrace.java b/spark-common/src/main/java/me/lucko/spark/common/sampler/async/jfr/StackTrace.java new file mode 100644 index 0000000..4c12c5e --- /dev/null +++ b/spark-common/src/main/java/me/lucko/spark/common/sampler/async/jfr/StackTrace.java @@ -0,0 +1,28 @@ +/* + * Copyright 2020 Andrei Pangin + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package me.lucko.spark.common.sampler.async.jfr; + +public class StackTrace { + public final long[] methods; + public final byte[] types; + public long samples; + + public StackTrace(long[] methods, byte[] types) { + this.methods = methods; + this.types = types; + } +} diff --git a/spark-common/src/main/java/me/lucko/spark/common/sampler/java/JavaDataAggregator.java b/spark-common/src/main/java/me/lucko/spark/common/sampler/java/JavaDataAggregator.java new file mode 100644 index 0000000..a81a5c1 --- /dev/null +++ b/spark-common/src/main/java/me/lucko/spark/common/sampler/java/JavaDataAggregator.java @@ -0,0 +1,116 @@ +/* + * This file is part of spark. + * + * Copyright (c) lucko (Luck) <luck@lucko.me> + * 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.java; + +import me.lucko.spark.common.sampler.ThreadGrouper; +import me.lucko.spark.common.sampler.aggregator.AbstractDataAggregator; +import me.lucko.spark.common.sampler.aggregator.DataAggregator; +import me.lucko.spark.common.sampler.node.ThreadNode; + +import java.lang.management.ThreadInfo; +import java.util.Map; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.TimeUnit; + +/** + * Abstract {@link DataAggregator} for the {@link JavaSampler}. + */ +public abstract class JavaDataAggregator extends AbstractDataAggregator { + + /** The worker pool for inserting stack nodes */ + protected final ExecutorService workerPool; + + /** The interval to wait between sampling, in microseconds */ + protected final int interval; + + /** If sleeping threads should be ignored */ + private final boolean ignoreSleeping; + + /** If threads executing native code should be ignored */ + private final boolean ignoreNative; + + public JavaDataAggregator(ExecutorService workerPool, ThreadGrouper threadGrouper, int interval, boolean ignoreSleeping, boolean ignoreNative) { + super(threadGrouper); + this.workerPool = workerPool; + this.interval = interval; + this.ignoreSleeping = ignoreSleeping; + this.ignoreNative = ignoreNative; + } + + /** + * Inserts sampling data into this aggregator + * + * @param threadInfo the thread info + */ + public abstract void insertData(ThreadInfo threadInfo); + + protected void writeData(ThreadInfo threadInfo) { + if (this.ignoreSleeping && isSleeping(threadInfo)) { + return; + } + if (this.ignoreNative && threadInfo.isInNative()) { + return; + } + + try { + ThreadNode node = getNode(this.threadGrouper.getGroup(threadInfo.getThreadId(), threadInfo.getThreadName())); + node.log(threadInfo.getStackTrace(), this.interval); + } catch (Exception e) { + e.printStackTrace(); + } + } + + @Override + public Map<String, ThreadNode> 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 super.getData(); + } + + private static boolean isSleeping(ThreadInfo thread) { + if (thread.getThreadState() == Thread.State.WAITING || thread.getThreadState() == Thread.State.TIMED_WAITING) { + return true; + } + + StackTraceElement[] stackTrace = thread.getStackTrace(); + if (stackTrace.length == 0) { + return false; + } + + StackTraceElement call = stackTrace[0]; + String clazz = call.getClassName(); + String method = call.getMethodName(); + + // java.lang.Thread.yield() + // jdk.internal.misc.Unsafe.park() + // sun.misc.Unsafe.park() + return (clazz.equals("java.lang.Thread") && method.equals("yield")) || + (clazz.equals("jdk.internal.misc.Unsafe") && method.equals("park")) || + (clazz.equals("sun.misc.Unsafe") && method.equals("park")); + } + +} diff --git a/spark-common/src/main/java/me/lucko/spark/common/sampler/java/JavaSampler.java b/spark-common/src/main/java/me/lucko/spark/common/sampler/java/JavaSampler.java new file mode 100644 index 0000000..568609e --- /dev/null +++ b/spark-common/src/main/java/me/lucko/spark/common/sampler/java/JavaSampler.java @@ -0,0 +1,189 @@ +/* + * This file is part of spark. + * + * Copyright (C) Albert Pham <http://www.sk89q.com> + * Copyright (c) lucko (Luck) <luck@lucko.me> + * 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.java; + +import com.google.common.util.concurrent.ThreadFactoryBuilder; +import me.lucko.spark.common.command.sender.CommandSender; +import me.lucko.spark.common.platform.PlatformInfo; +import me.lucko.spark.common.sampler.Sampler; +import me.lucko.spark.common.sampler.ThreadDumper; +import me.lucko.spark.common.sampler.ThreadGrouper; +import me.lucko.spark.common.sampler.node.MergeMode; +import me.lucko.spark.common.sampler.node.ThreadNode; +import me.lucko.spark.common.sampler.tick.TickHook; +import me.lucko.spark.proto.SparkProtos.SamplerData; +import me.lucko.spark.proto.SparkProtos.SamplerMetadata; + +import java.lang.management.ManagementFactory; +import java.lang.management.ThreadInfo; +import java.lang.management.ThreadMXBean; +import java.util.ArrayList; +import java.util.Comparator; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +/** + * A sampler implementation using Java (WarmRoast). + */ +public class JavaSampler implements Sampler, Runnable { + private static final AtomicInteger THREAD_ID = new AtomicInteger(0); + + /** The worker pool for inserting stack nodes */ + private final ScheduledExecutorService workerPool = Executors.newScheduledThreadPool( + 6, new ThreadFactoryBuilder().setNameFormat("spark-worker-" + THREAD_ID.getAndIncrement() + "-%d").build() + ); + + /** The main sampling task */ + private ScheduledFuture<?> task; + + /** 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 JavaDataAggregator dataAggregator; + + /** A future to encapsulate the completion of this sampler instance */ + private final CompletableFuture<JavaSampler> future = new CompletableFuture<>(); + /** The interval to wait between sampling, in microseconds */ + private final int interval; + /** 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 JavaSampler(int interval, ThreadDumper threadDumper, ThreadGrouper threadGrouper, long endTime, boolean ignoreSleeping, boolean ignoreNative) { + this.threadDumper = threadDumper; + this.dataAggregator = new SimpleDataAggregator(this.workerPool, threadGrouper, interval, ignoreSleeping, ignoreNative); + this.interval = interval; + this.endTime = endTime; + } + + public JavaSampler(int interval, ThreadDumper threadDumper, ThreadGrouper threadGrouper, long endTime, boolean ignoreSleeping, boolean ignoreNative, TickHook tickHook, int tickLengthThreshold) { + this.threadDumper = threadDumper; + this.dataAggregator = new TickedDataAggregator(this.workerPool, threadGrouper, interval, ignoreSleeping, ignoreNative, tickHook, tickLengthThreshold); + this.interval = interval; + this.endTime = endTime; + } + + @Override + public void start() { + this.startTime = System.currentTimeMillis(); + this.task = this.workerPool.scheduleAtFixedRate(this, 0, this.interval, TimeUnit.MICROSECONDS); + } + + @Override + public long getStartTime() { + if (this.startTime == -1) { + throw new IllegalStateException("Not yet started"); + } + return this.startTime; + } + + @Override + public long getEndTime() { + return this.endTime; + } + + @Override + public CompletableFuture<JavaSampler> getFuture() { + return this.future; + } + + @Override + public void stop() { + this.task.cancel(false); + } + + @Override + public void run() { + // this is effectively synchronized, the worker pool will not allow this task + // to concurrently execute. + try { + if (this.endTime != -1 && this.endTime <= System.currentTimeMillis()) { + this.future.complete(this); + stop(); + return; + } + + ThreadInfo[] threadDumps = this.threadDumper.dumpThreads(this.threadBean); + this.workerPool.execute(new InsertDataTask(this.dataAggregator, threadDumps)); + } catch (Throwable t) { + this.future.completeExceptionally(t); + stop(); + } + } + + private static final class InsertDataTask implements Runnable { + private final JavaDataAggregator dataAggregator; + private final ThreadInfo[] threadDumps; + + InsertDataTask(JavaDataAggregator dataAggregator, ThreadInfo[] threadDumps) { + this.dataAggregator = dataAggregator; + this.threadDumps = threadDumps; + } + + @Override + public void run() { + for (ThreadInfo threadInfo : this.threadDumps) { + if (threadInfo.getThreadName() == null || threadInfo.getStackTrace() == null) { + continue; + } + this.dataAggregator.insertData(threadInfo); + } + } + } + + @Override + public SamplerData toProto(PlatformInfo platformInfo, CommandSender creator, Comparator<? super Map.Entry<String, ThreadNode>> outputOrder, String comment, MergeMode mergeMode) { + final SamplerMetadata.Builder metadata = SamplerMetadata.newBuilder() + .setPlatform(platformInfo.toData().toProto()) + .setUser(creator.toData().toProto()) + .setStartTime(this.startTime) + .setInterval(this.interval) + .setThreadDumper(this.threadDumper.getMetadata()) + .setDataAggregator(this.dataAggregator.getMetadata()); + + if (comment != null) { + metadata.setComment(comment); + } + + SamplerData.Builder proto = SamplerData.newBuilder(); + proto.setMetadata(metadata.build()); + + List<Map.Entry<String, ThreadNode>> data = new ArrayList<>(this.dataAggregator.getData().entrySet()); + data.sort(outputOrder); + + for (Map.Entry<String, ThreadNode> entry : data) { + proto.addThreads(entry.getValue().toProto(mergeMode)); + } + + return proto.build(); + } + +} diff --git a/spark-common/src/main/java/me/lucko/spark/common/sampler/aggregator/SimpleDataAggregator.java b/spark-common/src/main/java/me/lucko/spark/common/sampler/java/SimpleDataAggregator.java index 8463f98..e7113a1 100644 --- a/spark-common/src/main/java/me/lucko/spark/common/sampler/aggregator/SimpleDataAggregator.java +++ b/spark-common/src/main/java/me/lucko/spark/common/sampler/java/SimpleDataAggregator.java @@ -18,21 +18,19 @@ * along with this program. If not, see <http://www.gnu.org/licenses/>. */ -package me.lucko.spark.common.sampler.aggregator; +package me.lucko.spark.common.sampler.java; import me.lucko.spark.common.sampler.ThreadGrouper; -import me.lucko.spark.common.sampler.node.ThreadNode; +import me.lucko.spark.common.sampler.aggregator.DataAggregator; import me.lucko.spark.proto.SparkProtos.SamplerMetadata; import java.lang.management.ThreadInfo; -import java.util.Map; import java.util.concurrent.ExecutorService; -import java.util.concurrent.TimeUnit; /** * Basic implementation of {@link DataAggregator}. */ -public class SimpleDataAggregator extends AbstractDataAggregator { +public class SimpleDataAggregator extends JavaDataAggregator { public SimpleDataAggregator(ExecutorService workerPool, ThreadGrouper threadGrouper, int interval, boolean ignoreSleeping, boolean ignoreNative) { super(workerPool, threadGrouper, interval, ignoreSleeping, ignoreNative); } @@ -50,16 +48,4 @@ public class SimpleDataAggregator extends AbstractDataAggregator { writeData(threadInfo); } - @Override - public Map<String, ThreadNode> 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; - } } diff --git a/spark-common/src/main/java/me/lucko/spark/common/sampler/aggregator/TickedDataAggregator.java b/spark-common/src/main/java/me/lucko/spark/common/sampler/java/TickedDataAggregator.java index 7ad6781..1a0bcdd 100644 --- a/spark-common/src/main/java/me/lucko/spark/common/sampler/aggregator/TickedDataAggregator.java +++ b/spark-common/src/main/java/me/lucko/spark/common/sampler/java/TickedDataAggregator.java @@ -18,9 +18,10 @@ * along with this program. If not, see <http://www.gnu.org/licenses/>. */ -package me.lucko.spark.common.sampler.aggregator; +package me.lucko.spark.common.sampler.java; import me.lucko.spark.common.sampler.ThreadGrouper; +import me.lucko.spark.common.sampler.aggregator.DataAggregator; import me.lucko.spark.common.sampler.node.ThreadNode; import me.lucko.spark.common.sampler.tick.TickHook; import me.lucko.spark.proto.SparkProtos.SamplerMetadata; @@ -36,7 +37,7 @@ 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 extends AbstractDataAggregator { +public class TickedDataAggregator extends JavaDataAggregator { /** Used to monitor the current "tick" of the server */ private final TickHook tickHook; @@ -107,15 +108,7 @@ public class TickedDataAggregator extends AbstractDataAggregator { pushCurrentTick(); } - // 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; + return super.getData(); } private final class TickList implements Runnable { diff --git a/spark-common/src/main/java/me/lucko/spark/common/sampler/node/AbstractNode.java b/spark-common/src/main/java/me/lucko/spark/common/sampler/node/AbstractNode.java index fe2fd62..2ef06d3 100644 --- a/spark-common/src/main/java/me/lucko/spark/common/sampler/node/AbstractNode.java +++ b/spark-common/src/main/java/me/lucko/spark/common/sampler/node/AbstractNode.java @@ -21,6 +21,8 @@ package me.lucko.spark.common.sampler.node; +import me.lucko.spark.common.sampler.async.AsyncStackTraceElement; + import java.util.ArrayList; import java.util.Collections; import java.util.List; @@ -110,6 +112,34 @@ public abstract class AbstractNode { child.log(elements, offset + 1, time); } + public void log(AsyncStackTraceElement[] elements, long time) { + log(elements, 0, time); + } + + private void log(AsyncStackTraceElement[] elements, int offset, long time) { + this.totalTime.add(time); + + if (offset >= MAX_STACK_DEPTH) { + return; + } + + if (elements.length - offset == 0) { + return; + } + + // the first element in the array is the top of the call stack, and the last is the root + // offset starts at 0. + + // pointer is determined by subtracting the offset from the index of the last element + int pointer = (elements.length - 1) - offset; + AsyncStackTraceElement element = elements[pointer]; + + // resolve a child element within the structure for the element at pointer + AbstractNode child = resolveChild(new StackTraceNode.Description(element.getClassName(), element.getMethodName(), element.getMethodDescription())); + // call the log method on the found child, with an incremented offset. + child.log(elements, offset + 1, time); + } + protected List<StackTraceNode> exportChildren(MergeMode mergeMode) { if (this.children.isEmpty()) { return Collections.emptyList(); diff --git a/spark-common/src/main/java/me/lucko/spark/common/sampler/node/StackTraceNode.java b/spark-common/src/main/java/me/lucko/spark/common/sampler/node/StackTraceNode.java index a73620e..bd731c1 100644 --- a/spark-common/src/main/java/me/lucko/spark/common/sampler/node/StackTraceNode.java +++ b/spark-common/src/main/java/me/lucko/spark/common/sampler/node/StackTraceNode.java @@ -24,6 +24,7 @@ package me.lucko.spark.common.sampler.node; import me.lucko.spark.common.util.MethodDisambiguator; import me.lucko.spark.proto.SparkProtos; +import java.util.Comparator; import java.util.Objects; /** @@ -73,9 +74,13 @@ public final class StackTraceNode extends AbstractNode implements Comparable<Sta proto.setParentLineNumber(this.description.parentLineNumber); } - mergeMode.getMethodDisambiguator().disambiguate(this) - .map(MethodDisambiguator.MethodDescription::getDesc) - .ifPresent(proto::setMethodDesc); + if (this.description.methodDescription != null) { + proto.setMethodDesc(this.description.methodDescription); + } else { + mergeMode.getMethodDisambiguator().disambiguate(this) + .map(MethodDisambiguator.MethodDescription::getDesc) + .ifPresent(proto::setMethodDesc); + } for (StackTraceNode child : exportChildren(mergeMode)) { proto.addChildren(child.toProto(mergeMode)); @@ -104,19 +109,48 @@ public final class StackTraceNode extends AbstractNode implements Comparable<Sta public static final class Description implements Comparable<Description> { private final String className; private final String methodName; + + // async-profiler + private final String methodDescription; + + // Java private final int lineNumber; private final int parentLineNumber; private final int hash; + // Constructor used by the Java sampler public Description(String className, String methodName, int lineNumber, int parentLineNumber) { this.className = className; this.methodName = methodName; + this.methodDescription = null; this.lineNumber = lineNumber; this.parentLineNumber = parentLineNumber; this.hash = Objects.hash(this.className, this.methodName, this.lineNumber, this.parentLineNumber); } + // Constructor used by the async-profiler sampler + public Description(String className, String methodName, String methodDescription) { + this.className = className; + this.methodName = methodName; + this.methodDescription = methodDescription; + this.lineNumber = StackTraceNode.NULL_LINE_NUMBER; + this.parentLineNumber = StackTraceNode.NULL_LINE_NUMBER; + this.hash = Objects.hash(this.className, this.methodName, this.methodDescription); + } + + private static <T extends Comparable<T>> int nullCompare(T a, T b) { + if (a == null && b == null) { + return 0; + } else if (a == null) { + return -1; + } else if (b == null) { + return 1; + } else { + return a.compareTo(b); + } + } + @Override public int compareTo(Description that) { if (this == that) { @@ -133,6 +167,18 @@ public final class StackTraceNode extends AbstractNode implements Comparable<Sta return i; } + i = nullCompare(this.methodDescription, that.methodDescription); + if (i != 0) { + return i; + } + + if (this.methodDescription != null && that.methodDescription != null) { + i = this.methodDescription.compareTo(that.methodDescription); + if (i != 0) { + return i; + } + } + i = Integer.compare(this.lineNumber, that.lineNumber); if (i != 0) { return i; @@ -146,10 +192,12 @@ public final class StackTraceNode extends AbstractNode implements Comparable<Sta if (this == o) return true; if (o == null || getClass() != o.getClass()) return false; Description description = (Description) o; - return this.lineNumber == description.lineNumber && + return this.hash == description.hash && + this.lineNumber == description.lineNumber && this.parentLineNumber == description.parentLineNumber && this.className.equals(description.className) && - this.methodName.equals(description.methodName); + this.methodName.equals(description.methodName) && + Objects.equals(this.methodDescription, description.methodDescription); } @Override |