From dd79a780347a2529ef9ec342669e95a3602e5a95 Mon Sep 17 00:00:00 2001 From: m1a2st Date: Wed, 18 Sep 2024 23:18:52 +0800 Subject: [PATCH 1/9] complete move TestPurgatoryPerformance to java --- checkstyle/import-control-jmh-benchmarks.xml | 1 + .../kafka/TestPurgatoryPerformance.scala | 291 --------------- .../jmh/core/TestPurgatoryPerformance.java | 346 ++++++++++++++++++ 3 files changed, 347 insertions(+), 291 deletions(-) delete mode 100644 core/src/test/scala/other/kafka/TestPurgatoryPerformance.scala create mode 100644 jmh-benchmarks/src/main/java/org/apache/kafka/jmh/core/TestPurgatoryPerformance.java diff --git a/checkstyle/import-control-jmh-benchmarks.xml b/checkstyle/import-control-jmh-benchmarks.xml index 1612477ea00f4..66404bc6697ed 100644 --- a/checkstyle/import-control-jmh-benchmarks.xml +++ b/checkstyle/import-control-jmh-benchmarks.xml @@ -30,6 +30,7 @@ + diff --git a/core/src/test/scala/other/kafka/TestPurgatoryPerformance.scala b/core/src/test/scala/other/kafka/TestPurgatoryPerformance.scala deleted file mode 100644 index 2e36d67f20c72..0000000000000 --- a/core/src/test/scala/other/kafka/TestPurgatoryPerformance.scala +++ /dev/null @@ -1,291 +0,0 @@ -/** - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You 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 kafka - -import java.lang.management.ManagementFactory -import java.lang.management.OperatingSystemMXBean -import java.util.Random -import java.util.concurrent._ -import joptsimple._ -import kafka.server.{DelayedOperation, DelayedOperationPurgatory} -import org.apache.kafka.common.utils.Time -import org.apache.kafka.server.util.{CommandLineUtils, ShutdownableThread} - -import scala.math._ -import scala.jdk.CollectionConverters._ - -/** - * This is a benchmark test of the purgatory. - */ -object TestPurgatoryPerformance { - - def main(args: Array[String]): Unit = { - val parser = new OptionParser(false) - val keySpaceSizeOpt = parser.accepts("key-space-size", "The total number of possible keys") - .withRequiredArg - .describedAs("total_num_possible_keys") - .ofType(classOf[java.lang.Integer]) - .defaultsTo(100) - val numRequestsOpt = parser.accepts("num", "The number of requests") - .withRequiredArg - .describedAs("num_requests") - .ofType(classOf[java.lang.Double]) - val requestRateOpt = parser.accepts("rate", "The request rate per second") - .withRequiredArg - .describedAs("request_per_second") - .ofType(classOf[java.lang.Double]) - val requestDataSizeOpt = parser.accepts("size", "The request data size in bytes") - .withRequiredArg - .describedAs("num_bytes") - .ofType(classOf[java.lang.Long]) - val numKeysOpt = parser.accepts("keys", "The number of keys for each request") - .withRequiredArg - .describedAs("num_keys") - .ofType(classOf[java.lang.Integer]) - .defaultsTo(3) - val timeoutOpt = parser.accepts("timeout", "The request timeout in ms") - .withRequiredArg - .describedAs("timeout_milliseconds") - .ofType(classOf[java.lang.Long]) - val pct75Opt = parser.accepts("pct75", "75th percentile of request latency in ms (log-normal distribution)") - .withRequiredArg - .describedAs("75th_percentile") - .ofType(classOf[java.lang.Double]) - val pct50Opt = parser.accepts("pct50", "50th percentile of request latency in ms (log-normal distribution)") - .withRequiredArg - .describedAs("50th_percentile") - .ofType(classOf[java.lang.Double]) - val verboseOpt = parser.accepts("verbose", "show additional information") - .withRequiredArg - .describedAs("true|false") - .ofType(classOf[java.lang.Boolean]) - .defaultsTo(true) - - val options = parser.parse(args: _*) - - CommandLineUtils.checkRequiredArgs(parser, options, numRequestsOpt, requestRateOpt, requestDataSizeOpt, pct75Opt, pct50Opt) - - val numRequests = options.valueOf(numRequestsOpt).intValue - val requestRate = options.valueOf(requestRateOpt).doubleValue - val requestDataSize = options.valueOf(requestDataSizeOpt).intValue - val numPossibleKeys = options.valueOf(keySpaceSizeOpt).intValue - val numKeys = options.valueOf(numKeysOpt).intValue - val timeout = options.valueOf(timeoutOpt).longValue - val pct75 = options.valueOf(pct75Opt).doubleValue - val pct50 = options.valueOf(pct50Opt).doubleValue - val verbose = options.valueOf(verboseOpt).booleanValue - - val gcMXBeans = ManagementFactory.getGarbageCollectorMXBeans.asScala.sortBy(_.getName) - val osMXBean = ManagementFactory.getOperatingSystemMXBean - val latencySamples = new LatencySamples(1000000, pct75, pct50) - val intervalSamples = new IntervalSamples(1000000, requestRate) - - val purgatory = DelayedOperationPurgatory[FakeOperation]("fake purgatory") - val queue = new CompletionQueue() - - val gcNames = gcMXBeans.map(_.getName) - - val initialCpuTimeNano = getProcessCpuTimeNanos(osMXBean) - val latch = new CountDownLatch(numRequests) - val start = System.currentTimeMillis - val rand = new Random() - val keys = (0 until numKeys).map(_ => "fakeKey%d".format(rand.nextInt(numPossibleKeys))) - @volatile var requestArrivalTime = start - @volatile var end = 0L - val generator = new Runnable { - def run(): Unit = { - var i = numRequests - while (i > 0) { - i -= 1 - val requestArrivalInterval = intervalSamples.next() - val latencyToComplete = latencySamples.next() - val now = System.currentTimeMillis - requestArrivalTime = requestArrivalTime + requestArrivalInterval - - if (requestArrivalTime > now) Thread.sleep(requestArrivalTime - now) - - val request = new FakeOperation(timeout, requestDataSize, latencyToComplete, latch) - if (latencyToComplete < timeout) queue.add(request) - purgatory.tryCompleteElseWatch(request, keys) - } - end = System.currentTimeMillis - } - } - val generatorThread = new Thread(generator) - - generatorThread.start() - generatorThread.join() - latch.await() - val done = System.currentTimeMillis - queue.shutdown() - - if (verbose) { - latencySamples.printStats() - intervalSamples.printStats() - println("# enqueue rate (%d requests):".format(numRequests)) - val gcCountHeader = gcNames.map("<" + _ + " count>").mkString(" ") - val gcTimeHeader = gcNames.map("<" + _ + " time ms>").mkString(" ") - println("# \t\t\t\t%s\t%s".format(gcCountHeader, gcTimeHeader)) - } - - val targetRate = numRequests.toDouble * 1000d / (requestArrivalTime - start).toDouble - val actualRate = numRequests.toDouble * 1000d / (end - start).toDouble - - val cpuTime = getProcessCpuTimeNanos(osMXBean).map(x => (x - initialCpuTimeNano.get) / 1000000L) - val gcCounts = gcMXBeans.map(_.getCollectionCount) - val gcTimes = gcMXBeans.map(_.getCollectionTime) - - println("%d\t%f\t%f\t%d\t%s\t%s".format(done - start, targetRate, actualRate, cpuTime.getOrElse(-1L), gcCounts.mkString(" "), gcTimes.mkString(" "))) - - purgatory.shutdown() - } - - // Use JRE-specific class to get process CPU time - private def getProcessCpuTimeNanos(osMXBean : OperatingSystemMXBean) = { - try { - Some(Class.forName("com.sun.management.OperatingSystemMXBean").getMethod("getProcessCpuTime").invoke(osMXBean).asInstanceOf[Long]) - } catch { - case _: Throwable => try { - Some(Class.forName("com.ibm.lang.management.OperatingSystemMXBean").getMethod("getProcessCpuTimeByNS").invoke(osMXBean).asInstanceOf[Long]) - } catch { - case _: Throwable => None - } - } - } - - // log-normal distribution (http://en.wikipedia.org/wiki/Log-normal_distribution) - // mu: the mean of the underlying normal distribution (not the mean of this log-normal distribution) - // sigma: the standard deviation of the underlying normal distribution (not the stdev of this log-normal distribution) - private class LogNormalDistribution(mu: Double, sigma: Double) { - private val rand = new Random - def next(): Double = { - val n = rand.nextGaussian() * sigma + mu - math.exp(n) - } - } - - // exponential distribution (http://en.wikipedia.org/wiki/Exponential_distribution) - // lambda : the rate parameter of the exponential distribution - private class ExponentialDistribution(lambda: Double) { - private val rand = new Random - def next(): Double = { - math.log(1d - rand.nextDouble()) / (- lambda) - } - } - - // Samples of Latencies to completion - // They are drawn from a log normal distribution. - // A latency value can never be negative. A log-normal distribution is a convenient way to - // model such a random variable. - private class LatencySamples(sampleSize: Int, pct75: Double, pct50: Double) { - private[this] val rand = new Random - private[this] val samples = { - val normalMean = math.log(pct50) - val normalStDev = (math.log(pct75) - normalMean) / 0.674490d // 0.674490 is 75th percentile point in N(0,1) - val dist = new LogNormalDistribution(normalMean, normalStDev) - (0 until sampleSize).map { _ => dist.next().toLong }.toArray - } - def next() = samples(rand.nextInt(sampleSize)) - - def printStats(): Unit = { - val p75 = samples.sorted.apply((sampleSize.toDouble * 0.75d).toInt) - val p50 = samples.sorted.apply((sampleSize.toDouble * 0.5d).toInt) - - println("# latency samples: pct75 = %d, pct50 = %d, min = %d, max = %d".format(p75, p50, samples.min, samples.max)) - } - } - - // Samples of Request arrival intervals - // The request arrival is modeled as a Poisson process. - // So, the internals are drawn from an exponential distribution. - private class IntervalSamples(sampleSize: Int, requestPerSecond: Double) { - private[this] val rand = new Random - private[this] val samples = { - val dist = new ExponentialDistribution(requestPerSecond / 1000d) - var residue = 0.0 - (0 until sampleSize).map { _ => - val interval = dist.next() + residue - val roundedInterval = interval.toLong - residue = interval - roundedInterval.toDouble - roundedInterval - }.toArray - } - - def next() = samples(rand.nextInt(sampleSize)) - - def printStats(): Unit = { - println( - "# interval samples: rate = %f, min = %d, max = %d" - .format(1000d / (samples.map(_.toDouble).sum / sampleSize.toDouble), samples.min, samples.max) - ) - } - } - - private class FakeOperation(delayMs: Long, size: Int, val latencyMs: Long, latch: CountDownLatch) extends DelayedOperation(delayMs) { - val completesAt = System.currentTimeMillis + latencyMs - - def onExpiration(): Unit = {} - - def onComplete(): Unit = { - latch.countDown() - } - - def tryComplete(): Boolean = { - if (System.currentTimeMillis >= completesAt) - forceComplete() - else - false - } - } - - private class CompletionQueue { - private[this] val delayQueue = new DelayQueue[Scheduled]() - private[this] val thread = new ShutdownableThread("completion thread", false) { - override def doWork(): Unit = { - val scheduled = delayQueue.poll(100, TimeUnit.MILLISECONDS) - if (scheduled != null) { - scheduled.operation.forceComplete() - } - } - } - thread.start() - - def add(operation: FakeOperation): Unit = { - delayQueue.offer(new Scheduled(operation)) - } - - def shutdown(): Unit = { - thread.shutdown() - } - - private class Scheduled(val operation: FakeOperation) extends Delayed { - def getDelay(unit: TimeUnit): Long = { - unit.convert(max(operation.completesAt - Time.SYSTEM.milliseconds, 0), TimeUnit.MILLISECONDS) - } - - def compareTo(d: Delayed): Int = { - - val other = d.asInstanceOf[Scheduled] - - if (operation.completesAt < other.operation.completesAt) -1 - else if (operation.completesAt > other.operation.completesAt) 1 - else 0 - } - } - } -} diff --git a/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/core/TestPurgatoryPerformance.java b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/core/TestPurgatoryPerformance.java new file mode 100644 index 0000000000000..616c488042e38 --- /dev/null +++ b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/core/TestPurgatoryPerformance.java @@ -0,0 +1,346 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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 org.apache.kafka.jmh.core; + +import kafka.server.DelayedOperation; +import kafka.server.DelayedOperationPurgatory; + +import org.apache.kafka.server.util.ShutdownableThread; + +import org.openjdk.jmh.annotations.Benchmark; +import org.openjdk.jmh.annotations.BenchmarkMode; +import org.openjdk.jmh.annotations.Fork; +import org.openjdk.jmh.annotations.Level; +import org.openjdk.jmh.annotations.Measurement; +import org.openjdk.jmh.annotations.Mode; +import org.openjdk.jmh.annotations.OutputTimeUnit; +import org.openjdk.jmh.annotations.Param; +import org.openjdk.jmh.annotations.Scope; +import org.openjdk.jmh.annotations.Setup; +import org.openjdk.jmh.annotations.State; +import org.openjdk.jmh.annotations.TearDown; +import org.openjdk.jmh.annotations.Warmup; + +import java.util.ArrayList; +import java.util.List; +import java.util.Random; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.DelayQueue; +import java.util.concurrent.Delayed; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; +import java.util.stream.Collectors; +import java.util.stream.IntStream; + +import scala.Option; +import scala.jdk.javaapi.CollectionConverters; + +@State(Scope.Benchmark) +@Fork(value = 1) +@Warmup(iterations = 5) +@Measurement(iterations = 15) +@BenchmarkMode({Mode.AverageTime, Mode.Throughput}) +@OutputTimeUnit(TimeUnit.NANOSECONDS) +public class TestPurgatoryPerformance { + + /** + * The number of requests + */ + @Param({"100"}) + private Integer numRequests; + /** + * The request rate per second + */ + @Param({"10.0"}) + private Double requestRate; + /** + * The total number of possible keys + */ + @Param({"100"}) + private Integer numPossibleKeys; + /** + * The number of keys for each request + */ + @Param({"3"}) + private Integer numKeys; + /** + * The request timeout in ms + */ + @Param({"1000"}) + private Long timeout; + /** + * 75th percentile of request latency in ms (log-normal distribution) + */ + @Param({"0.75"}) + private Double pct75; + /** + * 50th percentile of request latency in ms (log-normal distribution) + */ + @Param({"0.5"}) + private Double pct50; + + private DelayedOperationPurgatory purgatory; + private Random rand; + + private CompletionQueue queue; + + @Setup(Level.Invocation) + public void setUpInvocation() { + rand = new Random(); + } + + @Setup(Level.Trial) + public void setUpTrial() { + purgatory = DelayedOperationPurgatory.apply("fake purgatory", + 0, 1000, true, true); + queue = new CompletionQueue(); + } + + @TearDown(Level.Trial) + public void tearDownTrial() throws InterruptedException { + queue.shutdown(); + purgatory.shutdown(); + } + + @Benchmark + public void main() throws InterruptedException { + + LatencySamples latencySamples = new LatencySamples(1000000, pct75, pct50); + IntervalSamples intervalSamples = new IntervalSamples(1000000, requestRate); + CountDownLatch latch = new CountDownLatch(numRequests); + + List keys = IntStream.range(0, numRequests) + .mapToObj(i -> String.format("fakeKey%d", rand.nextInt(numPossibleKeys))) + .collect(Collectors.toList()); + + AtomicLong requestArrivalTime = new AtomicLong(System.currentTimeMillis()); + AtomicLong end = new AtomicLong(0); + Runnable generate = () -> runnable(intervalSamples, latencySamples, requestArrivalTime, latch, keys, end); + + Thread generateThread = new Thread(generate); + generateThread.start(); + generateThread.join(); + latch.await(); + } + + private void runnable(IntervalSamples intervalSamples, + LatencySamples latencySamples, + AtomicLong requestArrivalTime, + CountDownLatch latch, + List keys, + AtomicLong end) { + Integer i = numRequests; + while (i > 0) { + i -= 1; + long requestArrivalInterval = intervalSamples.next(); + long latencyToComplete = latencySamples.next(); + long now = System.currentTimeMillis(); + requestArrivalTime.addAndGet(requestArrivalInterval); + + if (requestArrivalTime.get() > now) { + try { + Thread.sleep(requestArrivalTime.get() - now); + } catch (InterruptedException e) { + throw new RuntimeException(e); + } + } + + FakeOperation request = new FakeOperation(timeout, latencyToComplete, latch); + if (latencyToComplete < timeout) { + queue.add(request); + } + + purgatory.tryCompleteElseWatch(request, CollectionConverters.asScala( + keys.stream().map(k -> (Object) k).collect(Collectors.toList()) + ).toSeq()); + } + end.set(System.currentTimeMillis()); + } + + /** + * log-normal distribution (...) + * mu: the mean of the underlying normal distribution (not the mean of this log-normal distribution) + * sigma: the standard deviation of the underlying normal distribution (not the stdev of this log-normal distribution) + */ + private static class LogNormalDistribution { + private final Random random = new Random(); + private final double mu; + private final double sigma; + + private LogNormalDistribution(double mu, double sigma) { + this.mu = mu; + this.sigma = sigma; + } + + public double next() { + double n = random.nextGaussian() * sigma + mu; + return Math.exp(n); + } + } + + /** + * Samples of Latencies to completion + * They are drawn from a log normal distribution. + * A latency value can never be negative. A log-normal distribution is a convenient way to + * model such a random variable. + */ + private static class LatencySamples { + private final Random random = new Random(); + private final List samples; + + public LatencySamples(int sampleSize, double pct75, double pct50) { + this.samples = new ArrayList<>(sampleSize); + double normalMean = Math.log(pct50); + double normalStDev = (Math.log(pct75) - normalMean) / 0.674490d; // 0.674490 is 75th percentile point in N(0,1) + LogNormalDistribution dist = new LogNormalDistribution(normalMean, normalStDev); + for (int i = 0; i < sampleSize; i++) { + samples.add((long) dist.next()); + } + } + + public long next() { + return samples.get(random.nextInt(samples.size())); + } + } + + /** + * Samples of Request arrival intervals + * The request arrival is modeled as a Poisson process. + * So, the internals are drawn from an exponential distribution. + */ + private static class IntervalSamples { + private final Random random = new Random(); + private final List samples; + + public IntervalSamples(int sampleSize, double requestPerSecond) { + this.samples = new ArrayList<>(sampleSize); + ExponentialDistribution dist = new ExponentialDistribution(requestPerSecond / 1000d); + double residue = 0; + for (int i = 0; i < sampleSize; i++) { + double interval = dist.next() + residue; + long roundedInterval = (long) interval; + residue = interval - (double) roundedInterval; + samples.add(roundedInterval); + } + } + + public long next() { + return samples.get(random.nextInt(samples.size())); + } + + } + + /** + * exponential distribution (...) + * lambda : the rate parameter of the exponential distribution + */ + private static class ExponentialDistribution { + private final Random random = new Random(); + private final double lambda; + + private ExponentialDistribution(double lambda) { + this.lambda = lambda; + } + + public double next() { + return Math.log(1d - random.nextDouble()) / (-lambda); + } + } + + private static class CompletionQueue { + private final DelayQueue delayQueue = new DelayQueue<>(); + private final ShutdownableThread thread = new ShutdownableThread("completion thread", false) { + @Override + public void doWork() { + try { + Scheduled scheduled = delayQueue.poll(100, TimeUnit.MILLISECONDS); + if (scheduled != null) { + scheduled.operation.forceComplete(); + } + } catch (InterruptedException e) { + throw new RuntimeException(e); + } + } + }; + + public CompletionQueue() { + thread.start(); + } + + public void add(FakeOperation operation) { + delayQueue.add(new Scheduled(operation)); + } + + public void shutdown() throws InterruptedException { + thread.shutdown(); + } + + } + + private static class Scheduled implements Delayed { + final FakeOperation operation; + + public Scheduled(FakeOperation operation) { + this.operation = operation; + } + + @Override + public long getDelay(TimeUnit unit) { + return unit.convert(Math.max(operation.completesAt - System.currentTimeMillis(), 0), TimeUnit.MILLISECONDS); + } + + @Override + public int compareTo(Delayed o) { + if (o instanceof Scheduled) { + Scheduled other = (Scheduled) o; + if (operation.completesAt < other.operation.completesAt) + return -1; + else if (operation.completesAt > other.operation.completesAt) + return 1; + } + return 0; + } + } + + private static class FakeOperation extends DelayedOperation { + final long completesAt; + final long latencyMs; + final CountDownLatch latch; + + public FakeOperation(long delayMs, long latencyMs, CountDownLatch latch) { + super(delayMs, Option.empty()); + this.latencyMs = latencyMs; + this.latch = latch; + completesAt = System.currentTimeMillis() + delayMs; + } + + @Override + public void onExpiration() { + + } + + @Override + public void onComplete() { + latch.countDown(); + } + + @Override + public boolean tryComplete() { + return System.currentTimeMillis() >= completesAt && forceComplete(); + } + } +} From 969c25dc4adb7336e4ceb9d92be0f9e0129764af Mon Sep 17 00:00:00 2001 From: m1a2st Date: Sat, 21 Sep 2024 23:42:21 +0800 Subject: [PATCH 2/9] rename main and method --- .../kafka/jmh/core/TestPurgatoryPerformance.java | 16 ++++++++-------- 1 file changed, 8 insertions(+), 8 deletions(-) diff --git a/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/core/TestPurgatoryPerformance.java b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/core/TestPurgatoryPerformance.java index 616c488042e38..65e7eb8405017 100644 --- a/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/core/TestPurgatoryPerformance.java +++ b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/core/TestPurgatoryPerformance.java @@ -117,7 +117,7 @@ public void tearDownTrial() throws InterruptedException { } @Benchmark - public void main() throws InterruptedException { + public void testPurgatoryPerformance() throws InterruptedException { LatencySamples latencySamples = new LatencySamples(1000000, pct75, pct50); IntervalSamples intervalSamples = new IntervalSamples(1000000, requestRate); @@ -129,7 +129,7 @@ public void main() throws InterruptedException { AtomicLong requestArrivalTime = new AtomicLong(System.currentTimeMillis()); AtomicLong end = new AtomicLong(0); - Runnable generate = () -> runnable(intervalSamples, latencySamples, requestArrivalTime, latch, keys, end); + Runnable generate = () -> generateTask(intervalSamples, latencySamples, requestArrivalTime, latch, keys, end); Thread generateThread = new Thread(generate); generateThread.start(); @@ -137,12 +137,12 @@ public void main() throws InterruptedException { latch.await(); } - private void runnable(IntervalSamples intervalSamples, - LatencySamples latencySamples, - AtomicLong requestArrivalTime, - CountDownLatch latch, - List keys, - AtomicLong end) { + private void generateTask(IntervalSamples intervalSamples, + LatencySamples latencySamples, + AtomicLong requestArrivalTime, + CountDownLatch latch, + List keys, + AtomicLong end) { Integer i = numRequests; while (i > 0) { i -= 1; From deb7a1347766a6ea658c435df91ea11a74def9ad Mon Sep 17 00:00:00 2001 From: m1a2st Date: Sat, 21 Sep 2024 23:47:27 +0800 Subject: [PATCH 3/9] refactor --- .../org/apache/kafka/jmh/core/TestPurgatoryPerformance.java | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/core/TestPurgatoryPerformance.java b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/core/TestPurgatoryPerformance.java index 65e7eb8405017..8dd559b29528f 100644 --- a/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/core/TestPurgatoryPerformance.java +++ b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/core/TestPurgatoryPerformance.java @@ -93,9 +93,9 @@ public class TestPurgatoryPerformance { @Param({"0.5"}) private Double pct50; - private DelayedOperationPurgatory purgatory; private Random rand; + private DelayedOperationPurgatory purgatory; private CompletionQueue queue; @Setup(Level.Invocation) @@ -118,7 +118,6 @@ public void tearDownTrial() throws InterruptedException { @Benchmark public void testPurgatoryPerformance() throws InterruptedException { - LatencySamples latencySamples = new LatencySamples(1000000, pct75, pct50); IntervalSamples intervalSamples = new IntervalSamples(1000000, requestRate); CountDownLatch latch = new CountDownLatch(numRequests); @@ -126,7 +125,6 @@ public void testPurgatoryPerformance() throws InterruptedException { List keys = IntStream.range(0, numRequests) .mapToObj(i -> String.format("fakeKey%d", rand.nextInt(numPossibleKeys))) .collect(Collectors.toList()); - AtomicLong requestArrivalTime = new AtomicLong(System.currentTimeMillis()); AtomicLong end = new AtomicLong(0); Runnable generate = () -> generateTask(intervalSamples, latencySamples, requestArrivalTime, latch, keys, end); From dd45082d72e2450dc148855c52acc8795bcd621e Mon Sep 17 00:00:00 2001 From: m1a2st Date: Wed, 23 Oct 2024 04:53:55 +0800 Subject: [PATCH 4/9] rewrite by java main function --- .../jmh/core/TestPurgatoryPerformance.java | 307 ++++++++++++------ 1 file changed, 211 insertions(+), 96 deletions(-) diff --git a/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/core/TestPurgatoryPerformance.java b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/core/TestPurgatoryPerformance.java index 8dd559b29528f..fb65aa7f6aebc 100644 --- a/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/core/TestPurgatoryPerformance.java +++ b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/core/TestPurgatoryPerformance.java @@ -16,27 +16,24 @@ */ package org.apache.kafka.jmh.core; +import joptsimple.ArgumentAcceptingOptionSpec; +import joptsimple.OptionParser; +import joptsimple.OptionSet; import kafka.server.DelayedOperation; import kafka.server.DelayedOperationPurgatory; - +import org.apache.kafka.server.util.CommandLineUtils; import org.apache.kafka.server.util.ShutdownableThread; +import scala.Option; +import scala.jdk.javaapi.CollectionConverters; -import org.openjdk.jmh.annotations.Benchmark; -import org.openjdk.jmh.annotations.BenchmarkMode; -import org.openjdk.jmh.annotations.Fork; -import org.openjdk.jmh.annotations.Level; -import org.openjdk.jmh.annotations.Measurement; -import org.openjdk.jmh.annotations.Mode; -import org.openjdk.jmh.annotations.OutputTimeUnit; -import org.openjdk.jmh.annotations.Param; -import org.openjdk.jmh.annotations.Scope; -import org.openjdk.jmh.annotations.Setup; -import org.openjdk.jmh.annotations.State; -import org.openjdk.jmh.annotations.TearDown; -import org.openjdk.jmh.annotations.Warmup; - +import java.lang.management.GarbageCollectorMXBean; +import java.lang.management.ManagementFactory; +import java.lang.management.MemoryManagerMXBean; +import java.lang.management.OperatingSystemMXBean; import java.util.ArrayList; +import java.util.Comparator; import java.util.List; +import java.util.Optional; import java.util.Random; import java.util.concurrent.CountDownLatch; import java.util.concurrent.DelayQueue; @@ -46,102 +43,202 @@ import java.util.stream.Collectors; import java.util.stream.IntStream; -import scala.Option; -import scala.jdk.javaapi.CollectionConverters; +import static java.lang.String.format; -@State(Scope.Benchmark) -@Fork(value = 1) -@Warmup(iterations = 5) -@Measurement(iterations = 15) -@BenchmarkMode({Mode.AverageTime, Mode.Throughput}) -@OutputTimeUnit(TimeUnit.NANOSECONDS) public class TestPurgatoryPerformance { - /** - * The number of requests - */ - @Param({"100"}) - private Integer numRequests; - /** - * The request rate per second - */ - @Param({"10.0"}) - private Double requestRate; - /** - * The total number of possible keys - */ - @Param({"100"}) - private Integer numPossibleKeys; - /** - * The number of keys for each request - */ - @Param({"3"}) - private Integer numKeys; - /** - * The request timeout in ms - */ - @Param({"1000"}) - private Long timeout; - /** - * 75th percentile of request latency in ms (log-normal distribution) - */ - @Param({"0.75"}) - private Double pct75; - /** - * 50th percentile of request latency in ms (log-normal distribution) - */ - @Param({"0.5"}) - private Double pct50; + public static void main(String[] args) throws InterruptedException { + TestArgumentDefinition def = new TestArgumentDefinition(args); + def.checkRequiredArgs(); + + int numRequests = def.numRequests(); + double requestRate = def.requestRate(); + int numPossibleKeys = def.numPossibleKeys(); + int numKeys = def.numKeys(); + long timeout = def.timeout(); + double pct75 = def.pct75(); + double pct50 = def.pct50(); + boolean verbose = def.verbose(); + + List gcMXBeans = ManagementFactory.getGarbageCollectorMXBeans(); + gcMXBeans.sort(Comparator.comparing(MemoryManagerMXBean::getName)); + OperatingSystemMXBean osMXBean = ManagementFactory.getOperatingSystemMXBean(); + LatencySamples latencySamples = new LatencySamples(1000000, pct75, pct50); + IntervalSamples intervalSamples = new IntervalSamples(1000000, requestRate); - private Random rand; + DelayedOperationPurgatory purgatory = + DelayedOperationPurgatory.apply("fake purgatory", 0, 1000, true, true); + CompletionQueue queue = new CompletionQueue(); - private DelayedOperationPurgatory purgatory; - private CompletionQueue queue; + List gcNames = gcMXBeans.stream().map(MemoryManagerMXBean::getName).collect(Collectors.toList()); + CountDownLatch latch = new CountDownLatch(numRequests); + long initialCpuTimeNano = getProcessCpuTimeNanos(osMXBean).orElseThrow(); + long start = System.currentTimeMillis(); + Random rand = new Random(); + List keys = IntStream.range(0, numKeys) + .mapToObj(i -> format("fakeKey%d", rand.nextInt(numPossibleKeys))) + .collect(Collectors.toList()); - @Setup(Level.Invocation) - public void setUpInvocation() { - rand = new Random(); - } + AtomicLong requestArrivalTime = new AtomicLong(start); + AtomicLong end = new AtomicLong(0); + Runnable task = () -> generateTask(numRequests, timeout, purgatory, queue, intervalSamples, + latencySamples, requestArrivalTime, latch, keys, end); - @Setup(Level.Trial) - public void setUpTrial() { - purgatory = DelayedOperationPurgatory.apply("fake purgatory", - 0, 1000, true, true); - queue = new CompletionQueue(); - } + Thread generateThread = new Thread(task); + generateThread.start(); + generateThread.join(); + latch.await(); - @TearDown(Level.Trial) - public void tearDownTrial() throws InterruptedException { + long done = System.currentTimeMillis(); queue.shutdown(); + + if (verbose) { + latencySamples.printStats(); + intervalSamples.printStats(); + System.out.printf("# enqueue rate (%d requests):%n", numRequests); + String gcCountHeader = gcNames.stream().map(gc -> "<" + gc + " count>").collect(Collectors.joining(" ")); + String gcTimeHeader = gcNames.stream().map(gc -> "<" + gc + " time ms>").collect(Collectors.joining(" ")); + System.out.printf("# \t\t\t\t%s\t%s%n", gcCountHeader, gcTimeHeader); + } + + double targetRate = numRequests * 1000d / (requestArrivalTime.get() - start); + double actualRate = numRequests * 1000d / (end.get() - start); + + Optional cpuTime = getProcessCpuTimeNanos(osMXBean).map(x -> (x - initialCpuTimeNano) / 1000000L); + String gcCounts = gcMXBeans.stream() + .map(GarbageCollectorMXBean::getCollectionCount) + .map(String::valueOf) + .collect(Collectors.joining(" ")); + String gcTimes = gcMXBeans.stream() + .map(GarbageCollectorMXBean::getCollectionTime) + .map(String::valueOf) + .collect(Collectors.joining(" ")); + + System.out.printf("%d\t%f\t%f\t%d\t%s\t%s%n", done - start, targetRate, actualRate, cpuTime.orElse(-1L), gcCounts, gcTimes); purgatory.shutdown(); } - @Benchmark - public void testPurgatoryPerformance() throws InterruptedException { - LatencySamples latencySamples = new LatencySamples(1000000, pct75, pct50); - IntervalSamples intervalSamples = new IntervalSamples(1000000, requestRate); - CountDownLatch latch = new CountDownLatch(numRequests); - - List keys = IntStream.range(0, numRequests) - .mapToObj(i -> String.format("fakeKey%d", rand.nextInt(numPossibleKeys))) - .collect(Collectors.toList()); - AtomicLong requestArrivalTime = new AtomicLong(System.currentTimeMillis()); - AtomicLong end = new AtomicLong(0); - Runnable generate = () -> generateTask(intervalSamples, latencySamples, requestArrivalTime, latch, keys, end); + private static Optional getProcessCpuTimeNanos(OperatingSystemMXBean osMXBean) { + try { + return Optional.of(Long.parseLong(Class.forName("com.sun.management.OperatingSystemMXBean") + .getMethod("getProcessCpuTime").invoke(osMXBean).toString())); + } catch (Exception e) { + try { + return Optional.of(Long.parseLong(Class.forName("com.ibm.lang.management.OperatingSystemMXBean") + .getMethod("getProcessCpuTimeByNS").invoke(osMXBean).toString())); + } catch (Exception ex) { + throw new RuntimeException(ex); + } + } + } - Thread generateThread = new Thread(generate); - generateThread.start(); - generateThread.join(); - latch.await(); + private static class TestArgumentDefinition { + final OptionParser parser = new OptionParser(false); + public final ArgumentAcceptingOptionSpec keySpaceSizeOpt; + public final ArgumentAcceptingOptionSpec numRequestsOpt; + public final ArgumentAcceptingOptionSpec requestRateOpt; + public final ArgumentAcceptingOptionSpec numKeysOpt; + public final ArgumentAcceptingOptionSpec timeoutOpt; + public final ArgumentAcceptingOptionSpec pct75Opt; + public final ArgumentAcceptingOptionSpec pct50Opt; + public final ArgumentAcceptingOptionSpec verboseOpt; + public OptionSet options; + + public TestArgumentDefinition(String[] args) { + this.keySpaceSizeOpt = parser + .accepts("key-space-size", "The total number of possible keys") + .withRequiredArg() + .describedAs("total_num_possible_keys") + .ofType(Integer.class) + .defaultsTo(100); + this.numRequestsOpt = parser + .accepts("num", "The number of requests") + .withRequiredArg() + .describedAs("num_requests") + .ofType(Double.class); + this.requestRateOpt = parser + .accepts("rate", "The request rate per second") + .withRequiredArg() + .describedAs("request_per_second") + .ofType(Double.class); + this.numKeysOpt = parser + .accepts("keys", "The number of keys for each request") + .withRequiredArg() + .describedAs("num_keys") + .ofType(Integer.class) + .defaultsTo(3); + this.timeoutOpt = parser + .accepts("timeout", "The request timeout in ms") + .withRequiredArg() + .describedAs("timeout_milliseconds") + .ofType(Long.class); + this.pct75Opt = parser + .accepts("pct75", "75th percentile of request latency in ms (log-normal distribution)") + .withRequiredArg() + .describedAs("75th_percentile") + .ofType(Double.class); + this.pct50Opt = parser + .accepts("pct50", "50th percentile of request latency in ms (log-normal distribution)") + .withRequiredArg() + .describedAs("50th_percentile") + .ofType(Double.class); + this.verboseOpt = parser + .accepts("verbose", "show additional information") + .withRequiredArg() + .describedAs("true|false") + .ofType(Boolean.class) + .defaultsTo(true); + this.options = parser.parse(args); + } + + public void checkRequiredArgs() { + CommandLineUtils.checkRequiredArgs(parser, options, numRequestsOpt, requestRateOpt, pct75Opt, pct50Opt); + } + + public int numRequests() { + return options.valueOf(numRequestsOpt).intValue(); + } + + public double requestRate() { + return options.valueOf(requestRateOpt); + } + + public int numPossibleKeys() { + return options.valueOf(keySpaceSizeOpt); + } + + public int numKeys() { + return options.valueOf(numKeysOpt); + } + + public long timeout() { + return options.valueOf(timeoutOpt); + } + + public double pct75() { + return options.valueOf(pct75Opt); + } + + public double pct50() { + return options.valueOf(pct50Opt); + } + + public boolean verbose() { + return options.valueOf(verboseOpt); + } } - private void generateTask(IntervalSamples intervalSamples, - LatencySamples latencySamples, - AtomicLong requestArrivalTime, - CountDownLatch latch, - List keys, - AtomicLong end) { - Integer i = numRequests; + private static void generateTask(int numRequests, + long timeout, + DelayedOperationPurgatory purgatory, + CompletionQueue queue, + IntervalSamples intervalSamples, + LatencySamples latencySamples, + AtomicLong requestArrivalTime, + CountDownLatch latch, + List keys, + AtomicLong end) { + int i = numRequests; while (i > 0) { i -= 1; long requestArrivalInterval = intervalSamples.next(); @@ -213,6 +310,17 @@ public LatencySamples(int sampleSize, double pct75, double pct50) { public long next() { return samples.get(random.nextInt(samples.size())); } + + public void printStats() { + List samples = this.samples.stream().sorted().collect(Collectors.toList()); + + long p75 = samples.get((int) (samples.size() * 0.75d)); + long p50 = samples.get((int) (samples.size() * 0.5d)); + + System.out.printf("# latency samples: pct75 = %d, pct50 = %d, min = %d, max = %d%n", p75, p50, + samples.stream().min(Comparator.comparingDouble(s -> s)).get(), + samples.stream().min(Comparator.comparingDouble(s -> s)).get()); + } } /** @@ -240,6 +348,13 @@ public long next() { return samples.get(random.nextInt(samples.size())); } + public void printStats() { + System.out.printf( + "# interval samples: rate = %f, min = %d, max = %d%n" + , 1000d / (samples.stream().mapToDouble(s -> s).sum() / samples.size()) + , samples.stream().min(Comparator.comparingDouble(s -> s)).get() + , samples.stream().max(Comparator.comparingDouble(s -> s)).get()); + } } /** From 07c43f16cc12940af241703a50aca59ee4c9171b Mon Sep 17 00:00:00 2001 From: m1a2st Date: Wed, 23 Oct 2024 04:54:56 +0800 Subject: [PATCH 5/9] use spotless apply --- .../kafka/jmh/core/TestPurgatoryPerformance.java | 12 +++++++----- 1 file changed, 7 insertions(+), 5 deletions(-) diff --git a/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/core/TestPurgatoryPerformance.java b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/core/TestPurgatoryPerformance.java index fb65aa7f6aebc..2159a0f04935a 100644 --- a/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/core/TestPurgatoryPerformance.java +++ b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/core/TestPurgatoryPerformance.java @@ -16,15 +16,11 @@ */ package org.apache.kafka.jmh.core; -import joptsimple.ArgumentAcceptingOptionSpec; -import joptsimple.OptionParser; -import joptsimple.OptionSet; import kafka.server.DelayedOperation; import kafka.server.DelayedOperationPurgatory; + import org.apache.kafka.server.util.CommandLineUtils; import org.apache.kafka.server.util.ShutdownableThread; -import scala.Option; -import scala.jdk.javaapi.CollectionConverters; import java.lang.management.GarbageCollectorMXBean; import java.lang.management.ManagementFactory; @@ -43,6 +39,12 @@ import java.util.stream.Collectors; import java.util.stream.IntStream; +import joptsimple.ArgumentAcceptingOptionSpec; +import joptsimple.OptionParser; +import joptsimple.OptionSet; +import scala.Option; +import scala.jdk.javaapi.CollectionConverters; + import static java.lang.String.format; public class TestPurgatoryPerformance { From e22c86f5edc091f06f13c0e84110c77b6a74085c Mon Sep 17 00:00:00 2001 From: m1a2st Date: Wed, 23 Oct 2024 08:36:56 +0800 Subject: [PATCH 6/9] update some error --- .../jmh/core/TestPurgatoryPerformance.java | 23 ++++++++++--------- 1 file changed, 12 insertions(+), 11 deletions(-) diff --git a/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/core/TestPurgatoryPerformance.java b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/core/TestPurgatoryPerformance.java index 2159a0f04935a..6f5def9bc35ad 100644 --- a/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/core/TestPurgatoryPerformance.java +++ b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/core/TestPurgatoryPerformance.java @@ -135,18 +135,19 @@ private static Optional getProcessCpuTimeNanos(OperatingSystemMXBean osMXB } private static class TestArgumentDefinition { - final OptionParser parser = new OptionParser(false); - public final ArgumentAcceptingOptionSpec keySpaceSizeOpt; - public final ArgumentAcceptingOptionSpec numRequestsOpt; - public final ArgumentAcceptingOptionSpec requestRateOpt; - public final ArgumentAcceptingOptionSpec numKeysOpt; - public final ArgumentAcceptingOptionSpec timeoutOpt; - public final ArgumentAcceptingOptionSpec pct75Opt; - public final ArgumentAcceptingOptionSpec pct50Opt; - public final ArgumentAcceptingOptionSpec verboseOpt; - public OptionSet options; + private final OptionParser parser; + private final ArgumentAcceptingOptionSpec keySpaceSizeOpt; + private final ArgumentAcceptingOptionSpec numRequestsOpt; + private final ArgumentAcceptingOptionSpec requestRateOpt; + private final ArgumentAcceptingOptionSpec numKeysOpt; + private final ArgumentAcceptingOptionSpec timeoutOpt; + private final ArgumentAcceptingOptionSpec pct75Opt; + private final ArgumentAcceptingOptionSpec pct50Opt; + private final ArgumentAcceptingOptionSpec verboseOpt; + private final OptionSet options; public TestArgumentDefinition(String[] args) { + this.parser = new OptionParser(false); this.keySpaceSizeOpt = parser .accepts("key-space-size", "The total number of possible keys") .withRequiredArg() @@ -321,7 +322,7 @@ public void printStats() { System.out.printf("# latency samples: pct75 = %d, pct50 = %d, min = %d, max = %d%n", p75, p50, samples.stream().min(Comparator.comparingDouble(s -> s)).get(), - samples.stream().min(Comparator.comparingDouble(s -> s)).get()); + samples.stream().max(Comparator.comparingDouble(s -> s)).get()); } } From ca391ca172ddbf2ecac847ac3cd5e569589bab30 Mon Sep 17 00:00:00 2001 From: m1a2st Date: Wed, 23 Oct 2024 08:46:21 +0800 Subject: [PATCH 7/9] update code format --- .../apache/kafka/jmh/core/TestPurgatoryPerformance.java | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/core/TestPurgatoryPerformance.java b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/core/TestPurgatoryPerformance.java index 6f5def9bc35ad..b847da617f7e9 100644 --- a/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/core/TestPurgatoryPerformance.java +++ b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/core/TestPurgatoryPerformance.java @@ -353,10 +353,10 @@ public long next() { public void printStats() { System.out.printf( - "# interval samples: rate = %f, min = %d, max = %d%n" - , 1000d / (samples.stream().mapToDouble(s -> s).sum() / samples.size()) - , samples.stream().min(Comparator.comparingDouble(s -> s)).get() - , samples.stream().max(Comparator.comparingDouble(s -> s)).get()); + "# interval samples: rate = %f, min = %d, max = %d%n", + 1000d / (samples.stream().mapToDouble(s -> s).sum() / samples.size()), + samples.stream().min(Comparator.comparingDouble(s -> s)).get(), + samples.stream().max(Comparator.comparingDouble(s -> s)).get()); } } From babf6748804016fe409b1f78160ceafc34faa28b Mon Sep 17 00:00:00 2001 From: m1a2st Date: Wed, 23 Oct 2024 14:28:26 +0800 Subject: [PATCH 8/9] update unused import --- checkstyle/import-control-jmh-benchmarks.xml | 1 + 1 file changed, 1 insertion(+) diff --git a/checkstyle/import-control-jmh-benchmarks.xml b/checkstyle/import-control-jmh-benchmarks.xml index 0cebe13eba3ae..6840d786926d3 100644 --- a/checkstyle/import-control-jmh-benchmarks.xml +++ b/checkstyle/import-control-jmh-benchmarks.xml @@ -61,6 +61,7 @@ + From 8adbdd3ff3043645d834a8920356b57c4b89d453 Mon Sep 17 00:00:00 2001 From: m1a2st Date: Wed, 23 Oct 2024 14:46:46 +0800 Subject: [PATCH 9/9] update exception catch --- .../org/apache/kafka/jmh/core/TestPurgatoryPerformance.java | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/core/TestPurgatoryPerformance.java b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/core/TestPurgatoryPerformance.java index b847da617f7e9..4817a29cfe92e 100644 --- a/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/core/TestPurgatoryPerformance.java +++ b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/core/TestPurgatoryPerformance.java @@ -26,6 +26,7 @@ import java.lang.management.ManagementFactory; import java.lang.management.MemoryManagerMXBean; import java.lang.management.OperatingSystemMXBean; +import java.lang.reflect.InvocationTargetException; import java.util.ArrayList; import java.util.Comparator; import java.util.List; @@ -124,11 +125,11 @@ private static Optional getProcessCpuTimeNanos(OperatingSystemMXBean osMXB try { return Optional.of(Long.parseLong(Class.forName("com.sun.management.OperatingSystemMXBean") .getMethod("getProcessCpuTime").invoke(osMXBean).toString())); - } catch (Exception e) { + } catch (ClassNotFoundException | InvocationTargetException | IllegalAccessException | NoSuchMethodException e) { try { return Optional.of(Long.parseLong(Class.forName("com.ibm.lang.management.OperatingSystemMXBean") .getMethod("getProcessCpuTimeByNS").invoke(osMXBean).toString())); - } catch (Exception ex) { + } catch (ClassNotFoundException | InvocationTargetException | IllegalAccessException | NoSuchMethodException ex) { throw new RuntimeException(ex); } }