diff --git a/checkstyle/import-control-jmh-benchmarks.xml b/checkstyle/import-control-jmh-benchmarks.xml index 65bfbb6337353..6840d786926d3 100644 --- a/checkstyle/import-control-jmh-benchmarks.xml +++ b/checkstyle/import-control-jmh-benchmarks.xml @@ -31,6 +31,7 @@ + @@ -60,6 +61,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..4817a29cfe92e --- /dev/null +++ b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/core/TestPurgatoryPerformance.java @@ -0,0 +1,463 @@ +/* + * 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.CommandLineUtils; +import org.apache.kafka.server.util.ShutdownableThread; + +import java.lang.management.GarbageCollectorMXBean; +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; +import java.util.Optional; +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 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 { + + 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); + + DelayedOperationPurgatory purgatory = + DelayedOperationPurgatory.apply("fake purgatory", 0, 1000, true, true); + CompletionQueue queue = new CompletionQueue(); + + 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()); + + AtomicLong requestArrivalTime = new AtomicLong(start); + AtomicLong end = new AtomicLong(0); + Runnable task = () -> generateTask(numRequests, timeout, purgatory, queue, intervalSamples, + latencySamples, requestArrivalTime, latch, keys, end); + + Thread generateThread = new Thread(task); + generateThread.start(); + generateThread.join(); + latch.await(); + + 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(); + } + + private static Optional getProcessCpuTimeNanos(OperatingSystemMXBean osMXBean) { + try { + return Optional.of(Long.parseLong(Class.forName("com.sun.management.OperatingSystemMXBean") + .getMethod("getProcessCpuTime").invoke(osMXBean).toString())); + } 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 (ClassNotFoundException | InvocationTargetException | IllegalAccessException | NoSuchMethodException ex) { + throw new RuntimeException(ex); + } + } + } + + private static class TestArgumentDefinition { + 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() + .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 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(); + 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())); + } + + 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().max(Comparator.comparingDouble(s -> s)).get()); + } + } + + /** + * 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())); + } + + 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()); + } + } + + /** + * 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(); + } + } +}