Repository: brooklyn-server Updated Branches: refs/heads/master 0b9717a73 -> 37308909a
Performance tests: support numConcurrentJobs Project: http://git-wip-us.apache.org/repos/asf/brooklyn-server/repo Commit: http://git-wip-us.apache.org/repos/asf/brooklyn-server/commit/3f36a298 Tree: http://git-wip-us.apache.org/repos/asf/brooklyn-server/tree/3f36a298 Diff: http://git-wip-us.apache.org/repos/asf/brooklyn-server/diff/3f36a298 Branch: refs/heads/master Commit: 3f36a298e71027bdb3f92b6ba0425a8c98020c7a Parents: ffd799e Author: Aled Sage <[email protected]> Authored: Tue Jun 6 11:21:57 2017 +0100 Committer: Aled Sage <[email protected]> Committed: Fri Jun 16 13:31:23 2017 +0100 ---------------------------------------------------------------------- .../test/performance/PerformanceMeasurer.java | 34 ++++++++++++++++++-- .../performance/PerformanceTestDescriptor.java | 31 +++++++++++++++++- 2 files changed, 62 insertions(+), 3 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/3f36a298/test-support/src/main/java/org/apache/brooklyn/test/performance/PerformanceMeasurer.java ---------------------------------------------------------------------- diff --git a/test-support/src/main/java/org/apache/brooklyn/test/performance/PerformanceMeasurer.java b/test-support/src/main/java/org/apache/brooklyn/test/performance/PerformanceMeasurer.java index b214c3b..6aef945 100644 --- a/test-support/src/main/java/org/apache/brooklyn/test/performance/PerformanceMeasurer.java +++ b/test-support/src/main/java/org/apache/brooklyn/test/performance/PerformanceMeasurer.java @@ -20,8 +20,11 @@ package org.apache.brooklyn.test.performance; import static org.testng.Assert.fail; +import java.util.ArrayList; import java.util.Date; import java.util.List; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Executors; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; @@ -34,6 +37,10 @@ import org.slf4j.LoggerFactory; import com.google.common.annotations.Beta; import com.google.common.base.Stopwatch; import com.google.common.collect.Lists; +import com.google.common.util.concurrent.Futures; +import com.google.common.util.concurrent.ListenableFuture; +import com.google.common.util.concurrent.ListeningExecutorService; +import com.google.common.util.concurrent.MoreExecutors; /** * For running simplistic performance tests, to measure the number of operations per second. @@ -89,12 +96,17 @@ public class PerformanceMeasurer { sampleCpuFuture = PerformanceTestUtils.sampleProcessCpuTime(options.sampleCpuInterval, options.summary, cpuSampleFractions); } + int numConcurrentJobs = options.numConcurrentJobs; + ListeningExecutorService executorService = null; + if (numConcurrentJobs > 1) { + executorService = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(numConcurrentJobs)); + } try { long preCpuTime = PerformanceTestUtils.getProcessCpuTime(); Stopwatch watch = Stopwatch.createStarted(); while ((options.duration != null) ? options.duration.isLongerThan(watch) : counter < options.iterations) { - if (warmupWatch.elapsed(TimeUnit.MILLISECONDS) >= nextLogTime) { + if (watch.elapsed(TimeUnit.MILLISECONDS) >= nextLogTime) { LOG.info(options.summary+" iteration="+counter+" at "+Time.makeTimeStringRounded(watch)); nextLogTime += options.logInterval.toMilliseconds(); } @@ -106,7 +118,11 @@ public class PerformanceMeasurer { } long before = watch.elapsed(TimeUnit.NANOSECONDS); - options.job.run(); + if (numConcurrentJobs > 1) { + runConcurrentAndBlock(executorService, options.job, numConcurrentJobs); + } else { + options.job.run(); + } if (options.histogram) { histogram.add(watch.elapsed(TimeUnit.NANOSECONDS) - before, TimeUnit.NANOSECONDS); } @@ -162,10 +178,24 @@ public class PerformanceMeasurer { return result; + } catch (InterruptedException | ExecutionException e) { + throw Exceptions.propagate(e); } finally { + if (executorService != null) { + executorService.shutdownNow(); + } if (sampleCpuFuture != null) { sampleCpuFuture.cancel(true); } } } + + protected static void runConcurrentAndBlock(ListeningExecutorService executor, Runnable job, int numConcurrentJobs) throws InterruptedException, ExecutionException { + List<ListenableFuture<?>> futures = new ArrayList<ListenableFuture<?>>(numConcurrentJobs); + for (int i = 0; i < numConcurrentJobs; i++) { + ListenableFuture<?> future = executor.submit(job); + futures.add(future); + Futures.allAsList(futures).get(); + } + } } http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/3f36a298/test-support/src/main/java/org/apache/brooklyn/test/performance/PerformanceTestDescriptor.java ---------------------------------------------------------------------- diff --git a/test-support/src/main/java/org/apache/brooklyn/test/performance/PerformanceTestDescriptor.java b/test-support/src/main/java/org/apache/brooklyn/test/performance/PerformanceTestDescriptor.java index a274fc7..f75763f 100644 --- a/test-support/src/main/java/org/apache/brooklyn/test/performance/PerformanceTestDescriptor.java +++ b/test-support/src/main/java/org/apache/brooklyn/test/performance/PerformanceTestDescriptor.java @@ -22,9 +22,13 @@ import static org.testng.Assert.assertFalse; import static org.testng.Assert.assertNotNull; import static org.testng.Assert.assertTrue; +import static com.google.common.base.Preconditions.checkArgument; + import java.io.File; +import java.util.concurrent.Callable; import java.util.concurrent.CountDownLatch; +import org.apache.brooklyn.util.exceptions.Exceptions; import org.apache.brooklyn.util.time.Duration; import org.apache.commons.io.FileUtils; @@ -52,6 +56,7 @@ public class PerformanceTestDescriptor { public Integer warmupIterations; public Duration duration; public Integer iterations; + public int numConcurrentJobs = 1; public Runnable job; public Runnable preJob; public Runnable postJob; @@ -112,7 +117,17 @@ public class PerformanceTestDescriptor { if (sealed) throw new IllegalStateException("Should not modify after sealed (e.g. after use)"); this.iterations = val; return this; } - + + /** + * The number of concurrent jobs to execute. If used with {@link #preJob(Runnable)} or {@link #postJob(Runnable)}, + * then those pre/post hooks will be called before/after the concurrent jobs are all done. + */ + public PerformanceTestDescriptor numConcurrentJobs(int val) { + if (sealed) throw new IllegalStateException("Should not modify after sealed (e.g. after use)"); + checkArgument(val >= 1, "val (%s) must be one or more", val); + this.numConcurrentJobs = val; return this; + } + /** * The job to be repeatedly executed. */ @@ -122,6 +137,20 @@ public class PerformanceTestDescriptor { } /** + * See {@link #job(Runnable)} + */ + public PerformanceTestDescriptor job(Callable<?> val) { + return job(new Runnable() { + public void run() { + try { + val.call(); + } catch (Exception e) { + throw Exceptions.propagate(e); + } + }}); + } + + /** * To be run each time before the job (pausing the timer while this is run). */ public PerformanceTestDescriptor preJob(Runnable val) {
