You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@brooklyn.apache.org by he...@apache.org on 2014/07/01 12:53:50 UTC
[1/4] git commit: FileBasedObjectStore.put: try rename before mv
Repository: incubator-brooklyn
Updated Branches:
refs/heads/master 490ebbd86 -> 9ca8250bd
FileBasedObjectStore.put: try rename before mv
- system call for `mv` is *very* expensive (about 20ms on my machine),
and CPU intensive, versus `file.rename` which is < 1ms.
- This change therefore allows us to persist 1000s of entities per
second, rather than < 50 (because done single-threaded currently).
Project: http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/commit/cbb0ffa0
Tree: http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/tree/cbb0ffa0
Diff: http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/diff/cbb0ffa0
Branch: refs/heads/master
Commit: cbb0ffa0d52e89656c0d596e4dccd4ff91c0e5b4
Parents: b1e27d8
Author: Aled Sage <al...@gmail.com>
Authored: Tue Jul 1 11:06:07 2014 +0100
Committer: Aled Sage <al...@gmail.com>
Committed: Tue Jul 1 11:06:33 2014 +0100
----------------------------------------------------------------------
.../rebind/persister/FileBasedObjectStore.java | 28 ++++++++++----------
1 file changed, 14 insertions(+), 14 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/cbb0ffa0/core/src/main/java/brooklyn/entity/rebind/persister/FileBasedObjectStore.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/brooklyn/entity/rebind/persister/FileBasedObjectStore.java b/core/src/main/java/brooklyn/entity/rebind/persister/FileBasedObjectStore.java
index 0033f0f..ac00946 100644
--- a/core/src/main/java/brooklyn/entity/rebind/persister/FileBasedObjectStore.java
+++ b/core/src/main/java/brooklyn/entity/rebind/persister/FileBasedObjectStore.java
@@ -308,27 +308,27 @@ public class FileBasedObjectStore implements PersistenceObjectStore {
* TODO Java 7 gives an atomic Files.move() which would be preferred.
*/
static void moveFile(File srcFile, File destFile) throws IOException, InterruptedException {
+ // Try rename first - it is a *much* cheaper call than invoking a system call in Java.
+ // However, rename is not guaranteed cross platform to succeed if the destination exists,
+ // and not guaranteed to be atomic, but it usually seems to do the right thing...
+ boolean result;
+ result = srcFile.renameTo(destFile);
+ if (result) {
+ if (log.isTraceEnabled()) log.trace("java rename of {} to {} completed", srcFile, destFile);
+ return;
+ }
+
if (!Os.isMicrosoftWindows()) {
// this command, if it succeeds, is guaranteed to be atomic, and it will usually overwrite
- int result;
String cmd = "mv '"+srcFile.getAbsolutePath()+"' '"+destFile.getAbsolutePath()+"'";
- result = new ProcessTool().execCommands(MutableMap.<String,String>of(), MutableList.of(cmd), null);
+ int exitStatus = new ProcessTool().execCommands(MutableMap.<String,String>of(), MutableList.of(cmd), null);
// prefer the above to the below because it wraps it in the appropriate bash
// Process proc = Runtime.getRuntime().exec(cmd);
// result = proc.waitFor();
- log.trace("FS move of {} to {} completed, code {}", new Object[] { srcFile, destFile, result });
- if (result == 0) return;
- }
-
- boolean result;
- // this is not guaranteed cross platform to succeed if the destination exists,
- // and not guaranteed to be atomic, but it usually seems to do the right thing...
- result = srcFile.renameTo(destFile);
- if (result) {
- log.trace("java rename of {} to {} completed", srcFile, destFile);
- return;
+ if (log.isTraceEnabled()) log.trace("FS move of {} to {} completed, code {}", new Object[] { srcFile, destFile, exitStatus });
+ if (exitStatus == 0) return;
}
// finally try a delete - but explicitly warn this is not going to be atomic
@@ -339,7 +339,7 @@ public class FileBasedObjectStore implements PersistenceObjectStore {
}
destFile.delete();
result = srcFile.renameTo(destFile);
- log.trace("java delete and rename of {} to {} completed, code {}", new Object[] { srcFile, destFile, result });
+ if (log.isTraceEnabled()) log.trace("java delete and rename of {} to {} completed, code {}", new Object[] { srcFile, destFile, result });
if (result)
return;
Files.copy(srcFile, destFile);
[3/4] git commit: Testing performance of entity persistence
Posted by he...@apache.org.
Testing performance of entity persistence
Project: http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/commit/870a1b8f
Tree: http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/tree/870a1b8f
Diff: http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/diff/870a1b8f
Branch: refs/heads/master
Commit: 870a1b8fb66792f45db372f2b3a2752dd8680d09
Parents: b8f1a04
Author: Aled Sage <al...@gmail.com>
Authored: Tue Jun 24 08:19:20 2014 +0100
Committer: Aled Sage <al...@gmail.com>
Committed: Tue Jul 1 11:06:33 2014 +0100
----------------------------------------------------------------------
.../BrooklynMementoPersisterToObjectStore.java | 16 +-
.../rebind/persister/MementoFileWriter.java | 2 +-
.../entity/rebind/RebindTestFixture.java | 12 +-
.../SshMachineLocationPerformanceTest.java | 11 +-
.../qa/performance/AbstractPerformanceTest.java | 10 +-
.../qa/performance/EntityPerformanceTest.java | 2 +-
.../EntityPersistencePerformanceTest.java | 81 +++++++++
.../FilePersistencePerformanceTest.java | 171 +++++++++++++++++++
.../GroovyYardStickPerformanceTest.groovy | 44 +++++
.../JavaYardStickPerformanceTest.java | 4 +-
.../qa/performance/PerformanceTestUtils.java | 81 +++++++++
.../java/brooklyn/test/policy/TestPolicy.java | 5 +
12 files changed, 415 insertions(+), 24 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/870a1b8f/core/src/main/java/brooklyn/entity/rebind/persister/BrooklynMementoPersisterToObjectStore.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/brooklyn/entity/rebind/persister/BrooklynMementoPersisterToObjectStore.java b/core/src/main/java/brooklyn/entity/rebind/persister/BrooklynMementoPersisterToObjectStore.java
index dec2bf7..54f4943 100644
--- a/core/src/main/java/brooklyn/entity/rebind/persister/BrooklynMementoPersisterToObjectStore.java
+++ b/core/src/main/java/brooklyn/entity/rebind/persister/BrooklynMementoPersisterToObjectStore.java
@@ -266,9 +266,10 @@ public class BrooklynMementoPersisterToObjectStore implements BrooklynMementoPer
if (LOG.isDebugEnabled()) LOG.debug("Ignoring checkpointing entire memento, because not running");
return;
}
- if (LOG.isDebugEnabled()) LOG.debug("Checkpointing entire memento");
objectStore.prepareForMasterUse();
+ Stopwatch stopwatch = Stopwatch.createStarted();
+
for (EntityMemento entity : newMemento.getEntityMementos().values()) {
persist("entities", entity);
}
@@ -281,6 +282,8 @@ public class BrooklynMementoPersisterToObjectStore implements BrooklynMementoPer
for (EnricherMemento enricher : newMemento.getEnricherMementos().values()) {
persist("enrichers", enricher);
}
+
+ if (LOG.isDebugEnabled()) LOG.debug("Checkpointed entire memento in {}", Time.makeTimeStringRounded(stopwatch));
}
@Override
@@ -289,12 +292,10 @@ public class BrooklynMementoPersisterToObjectStore implements BrooklynMementoPer
if (LOG.isDebugEnabled()) LOG.debug("Ignoring checkpointed delta of memento, because not running");
return;
}
- if (LOG.isDebugEnabled()) LOG.debug("Checkpointed delta of memento; updating {} entities, {} locations and {} policies; " +
- "removing {} entities, {} locations and {} policies",
- new Object[] {delta.entities(), delta.locations(), delta.policies(),
- delta.removedEntityIds(), delta.removedLocationIds(), delta.removedPolicyIds()});
objectStore.prepareForMasterUse();
+ Stopwatch stopwatch = Stopwatch.createStarted();
+
for (EntityMemento entity : delta.entities()) {
persist("entities", entity);
}
@@ -320,6 +321,11 @@ public class BrooklynMementoPersisterToObjectStore implements BrooklynMementoPer
for (String id : delta.removedEnricherIds()) {
delete("enrichers", id);
}
+
+ if (LOG.isDebugEnabled()) LOG.debug("Checkpointed delta of memento in {}; updated {} entities, {} locations and {} policies; " +
+ "removing {} entities, {} locations and {} policies",
+ new Object[] {Time.makeTimeStringRounded(stopwatch), delta.entities(), delta.locations(), delta.policies(),
+ delta.removedEntityIds(), delta.removedLocationIds(), delta.removedPolicyIds()});
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/870a1b8f/core/src/main/java/brooklyn/entity/rebind/persister/MementoFileWriter.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/brooklyn/entity/rebind/persister/MementoFileWriter.java b/core/src/main/java/brooklyn/entity/rebind/persister/MementoFileWriter.java
index 83af866..959f875 100644
--- a/core/src/main/java/brooklyn/entity/rebind/persister/MementoFileWriter.java
+++ b/core/src/main/java/brooklyn/entity/rebind/persister/MementoFileWriter.java
@@ -32,7 +32,7 @@ import brooklyn.util.time.Time;
* scheduled, we will just rely on the existing one; otherwise we will write now.
*
* @author aled
- * @deprecated used only by now-deprecated {@link BrooklynMementoPersisterToMultiFile},
+ * @deprecated since 0.7.0; used only by now-deprecated {@link BrooklynMementoPersisterToMultiFile},
* impl largely moved to {@link FileBasedStoreObjectAccessor}
*/
@Deprecated
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/870a1b8f/core/src/test/java/brooklyn/entity/rebind/RebindTestFixture.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/brooklyn/entity/rebind/RebindTestFixture.java b/core/src/test/java/brooklyn/entity/rebind/RebindTestFixture.java
index 92effff..89ac7a1 100644
--- a/core/src/test/java/brooklyn/entity/rebind/RebindTestFixture.java
+++ b/core/src/test/java/brooklyn/entity/rebind/RebindTestFixture.java
@@ -2,6 +2,8 @@ package brooklyn.entity.rebind;
import java.io.File;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
@@ -21,6 +23,8 @@ import brooklyn.util.time.Duration;
public abstract class RebindTestFixture<T extends StartableApplication> {
+ private static final Logger LOG = LoggerFactory.getLogger(RebindTestFixture.class);
+
protected static final Duration TIMEOUT_MS = Duration.TEN_SECONDS;
protected ClassLoader classLoader = getClass().getClassLoader();
@@ -40,8 +44,14 @@ public abstract class RebindTestFixture<T extends StartableApplication> {
origEnricherPersistenceEnabled = BrooklynFeatureEnablement.enable(BrooklynFeatureEnablement.FEATURE_ENRICHER_PERSISTENCE_PROPERTY);
mementoDir = Os.newTempDir(getClass());
- origManagementContext = RebindTestUtils.newPersistingManagementContext(mementoDir, classLoader, 1);
+ origManagementContext = RebindTestUtils.newPersistingManagementContext(mementoDir, classLoader, getPersistPeriodMillis());
origApp = createApp();
+
+ LOG.info("Test "+getClass()+" persisting to "+mementoDir);
+ }
+
+ protected int getPersistPeriodMillis() {
+ return 1;
}
/** optionally, create the app as part of every test; can be no-op if tests wish to set origApp themselves */
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/870a1b8f/core/src/test/java/brooklyn/location/basic/SshMachineLocationPerformanceTest.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/brooklyn/location/basic/SshMachineLocationPerformanceTest.java b/core/src/test/java/brooklyn/location/basic/SshMachineLocationPerformanceTest.java
index ee8fa44..f2d40aa 100644
--- a/core/src/test/java/brooklyn/location/basic/SshMachineLocationPerformanceTest.java
+++ b/core/src/test/java/brooklyn/location/basic/SshMachineLocationPerformanceTest.java
@@ -1,22 +1,19 @@
package brooklyn.location.basic;
import java.io.ByteArrayOutputStream;
-import java.lang.management.ManagementFactory;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
-import javax.management.MBeanServer;
-import javax.management.ObjectName;
-
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
+import brooklyn.qa.performance.PerformanceTestUtils;
import brooklyn.util.collections.MutableMap;
import brooklyn.util.internal.ssh.SshTool;
import brooklyn.util.net.Networking;
@@ -125,9 +122,7 @@ public class SshMachineLocationPerformanceTest {
}
private void runMany(final Runnable task, final String context, int concurrentRuns, int iterations) throws Exception {
- MBeanServer mbeanServer = ManagementFactory.getPlatformMBeanServer();
- ObjectName osMBeanName = ObjectName.getInstance(ManagementFactory.OPERATING_SYSTEM_MXBEAN_NAME);
- long preCpuTime = (Long) mbeanServer.getAttribute(osMBeanName, "ProcessCpuTime");
+ long preCpuTime = PerformanceTestUtils.getProcessCpuTime();
Stopwatch stopwatch = Stopwatch.createStarted();
for (int i = 0; i < iterations; i++) {
@@ -145,7 +140,7 @@ public class SshMachineLocationPerformanceTest {
}
Futures.allAsList(futures).get();
- long postCpuTime = (Long) mbeanServer.getAttribute(osMBeanName, "ProcessCpuTime");
+ long postCpuTime = PerformanceTestUtils.getProcessCpuTime();
long elapsedTime = stopwatch.elapsed(TimeUnit.MILLISECONDS);
double fractionCpu = (elapsedTime > 0) ? ((double)postCpuTime-preCpuTime) / TimeUnit.MILLISECONDS.toNanos(elapsedTime) : -1;
LOG.info("Executing {}; completed {}; took {}; fraction cpu {}",
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/870a1b8f/core/src/test/java/brooklyn/qa/performance/AbstractPerformanceTest.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/brooklyn/qa/performance/AbstractPerformanceTest.java b/core/src/test/java/brooklyn/qa/performance/AbstractPerformanceTest.java
index b395e9a..e585b7c 100644
--- a/core/src/test/java/brooklyn/qa/performance/AbstractPerformanceTest.java
+++ b/core/src/test/java/brooklyn/qa/performance/AbstractPerformanceTest.java
@@ -50,14 +50,14 @@ public class AbstractPerformanceTest {
protected SimulatedLocation loc;
@BeforeMethod(alwaysRun=true)
- public void setUp() {
+ public void setUp() throws Exception {
for (int i = 0; i < 5; i++) System.gc();
loc = new SimulatedLocation();
app = ApplicationBuilder.newManagedApp(TestApplication.class);
}
@AfterMethod(alwaysRun=true)
- public void tearDown() {
+ public void tearDown() throws Exception {
if (app != null) Entities.destroyAll(app.getManagementContext());
}
@@ -87,8 +87,7 @@ public class AbstractPerformanceTest {
long nextLogTime = logInterval;
// Give it some warm-up cycles
- Stopwatch warmupWatch = new Stopwatch();
- warmupWatch.start();
+ Stopwatch warmupWatch = Stopwatch.createStarted();
for (int i = 0; i < (numIterations/10); i++) {
if (warmupWatch.elapsed(TimeUnit.MILLISECONDS) >= nextLogTime) {
LOG.info("Warm-up "+prefix+" iteration="+i+" at "+warmupWatch.elapsed(TimeUnit.MILLISECONDS)+"ms");
@@ -97,8 +96,7 @@ public class AbstractPerformanceTest {
r.run();
}
- Stopwatch stopwatch = new Stopwatch();
- stopwatch.start();
+ Stopwatch stopwatch = Stopwatch.createStarted();
nextLogTime = 0;
for (int i = 0; i < numIterations; i++) {
if (stopwatch.elapsed(TimeUnit.MILLISECONDS) >= nextLogTime) {
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/870a1b8f/core/src/test/java/brooklyn/qa/performance/EntityPerformanceTest.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/brooklyn/qa/performance/EntityPerformanceTest.java b/core/src/test/java/brooklyn/qa/performance/EntityPerformanceTest.java
index b5811e6..74ab04b 100644
--- a/core/src/test/java/brooklyn/qa/performance/EntityPerformanceTest.java
+++ b/core/src/test/java/brooklyn/qa/performance/EntityPerformanceTest.java
@@ -34,7 +34,7 @@ public class EntityPerformanceTest extends AbstractPerformanceTest {
@BeforeMethod(alwaysRun=true)
@Override
- public void setUp() {
+ public void setUp() throws Exception {
super.setUp();
entities = Lists.newArrayList();
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/870a1b8f/core/src/test/java/brooklyn/qa/performance/EntityPersistencePerformanceTest.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/brooklyn/qa/performance/EntityPersistencePerformanceTest.java b/core/src/test/java/brooklyn/qa/performance/EntityPersistencePerformanceTest.java
new file mode 100644
index 0000000..3a1a16c
--- /dev/null
+++ b/core/src/test/java/brooklyn/qa/performance/EntityPersistencePerformanceTest.java
@@ -0,0 +1,81 @@
+package brooklyn.qa.performance;
+
+import java.util.List;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+
+import org.testng.annotations.Test;
+
+import brooklyn.entity.proxying.EntitySpec;
+import brooklyn.entity.rebind.RebindTestFixtureWithApp;
+import brooklyn.location.LocationSpec;
+import brooklyn.location.basic.SimulatedLocation;
+import brooklyn.policy.Policy;
+import brooklyn.policy.PolicySpec;
+import brooklyn.test.entity.TestEntity;
+import brooklyn.test.policy.TestPolicy;
+import brooklyn.util.repeat.Repeater;
+import brooklyn.util.time.Duration;
+
+import com.google.common.base.Predicates;
+import com.google.common.collect.Iterables;
+import com.google.common.collect.Lists;
+import com.google.common.util.concurrent.Callables;
+
+public class EntityPersistencePerformanceTest extends RebindTestFixtureWithApp {
+
+ // TODO Not measuring performance per cycle; just looking at CPU usage during test
+
+ protected int getPersistPeriodMillis() {
+ return 1000;
+ }
+
+ @Test(groups="Integration")
+ public void testManyEntities() throws Exception {
+ final int NUM_ENTITIES = 100;
+ final Duration TEST_LENGTH = Duration.of(60, TimeUnit.SECONDS);
+ final Duration REPEAT_EVERY = Duration.of(500, TimeUnit.MILLISECONDS);
+ run(NUM_ENTITIES, TEST_LENGTH, REPEAT_EVERY, "manyEntities");
+ }
+
+ @Test(groups="Integration")
+ public void testRapidChanges() throws Exception {
+ final int NUM_ENTITIES = 10;
+ final Duration TEST_LENGTH = Duration.of(60, TimeUnit.SECONDS);
+ final Duration REPEAT_EVERY = Duration.of(10, TimeUnit.MILLISECONDS);
+ run(NUM_ENTITIES, TEST_LENGTH, REPEAT_EVERY, "rapidChanges");
+ }
+
+ protected void run(int numEntities, Duration testLength, Duration repeatEvery, String loggingContext) throws Exception {
+ final List<TestEntity> entities = Lists.newArrayList();
+ final List<SimulatedLocation> locs = Lists.newArrayList();
+
+ for (int i = 0; i < numEntities; i++) {
+ TestEntity entity = origApp.createAndManageChild(EntitySpec.create(TestEntity.class));
+ entity.addPolicy(PolicySpec.create(TestPolicy.class));
+ SimulatedLocation loc = origManagementContext.getLocationManager().createLocation(LocationSpec.create(SimulatedLocation.class));
+ entities.add(entity);
+ locs.add(loc);
+ }
+
+ Future<?> future = PerformanceTestUtils.sampleProcessCpuTime(Duration.ONE_SECOND, "during "+loggingContext);
+ try {
+ Repeater.create()
+ .every(repeatEvery)
+ .repeat(new Runnable() {
+ int i = 0;
+ public void run() {
+ for (TestEntity entity : entities) {
+ entity.setAttribute(TestEntity.SEQUENCE, i++);
+ Policy policy = Iterables.find(entity.getPolicies(), Predicates.instanceOf(TestPolicy.class));
+ policy.setConfig(TestPolicy.CONF_NAME, "name-"+i);
+ }
+ }})
+ .limitTimeTo(testLength)
+ .until(Callables.returning(false))
+ .run();
+ } finally {
+ future.cancel(true);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/870a1b8f/core/src/test/java/brooklyn/qa/performance/FilePersistencePerformanceTest.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/brooklyn/qa/performance/FilePersistencePerformanceTest.java b/core/src/test/java/brooklyn/qa/performance/FilePersistencePerformanceTest.java
new file mode 100644
index 0000000..eaf81ae
--- /dev/null
+++ b/core/src/test/java/brooklyn/qa/performance/FilePersistencePerformanceTest.java
@@ -0,0 +1,171 @@
+package brooklyn.qa.performance;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+import brooklyn.entity.rebind.persister.FileBasedStoreObjectAccessor;
+import brooklyn.util.collections.MutableList;
+import brooklyn.util.collections.MutableMap;
+import brooklyn.util.exceptions.Exceptions;
+import brooklyn.util.internal.ssh.process.ProcessTool;
+
+import com.google.common.base.Charsets;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Lists;
+import com.google.common.io.Files;
+
+public class FilePersistencePerformanceTest extends AbstractPerformanceTest {
+
+ private static final Logger LOG = LoggerFactory.getLogger(FilePersistencePerformanceTest.class);
+
+ File file;
+ FileBasedStoreObjectAccessor fileAccessor;
+
+ @BeforeMethod(alwaysRun=true)
+ @Override
+ public void setUp() throws Exception {
+ super.setUp();
+
+ file = File.createTempFile("fileBasedStoreObject", ".txt");
+ Files.write("initial", file, Charsets.UTF_8);
+ fileAccessor = new FileBasedStoreObjectAccessor(file, "mytmpextension");
+
+ app.start(ImmutableList.of(loc));
+ }
+
+ @Override
+ public void tearDown() throws Exception {
+ super.tearDown();
+ if (file != null) file.delete();
+ }
+
+ protected int numIterations() {
+ return 100;
+ }
+
+ @Test(groups={"Integration", "Acceptance"})
+ public void testFileBasedStoreObjectPuts() throws Exception {
+ int numIterations = numIterations();
+ double minRatePerSec = 100 * PERFORMANCE_EXPECTATION;
+ final AtomicInteger i = new AtomicInteger();
+
+ measureAndAssert("FileBasedStoreObjectAccessor.put", numIterations, minRatePerSec, new Runnable() {
+ public void run() {
+ fileAccessor.put(""+i.incrementAndGet());
+ }});
+ }
+
+ @Test(groups={"Integration", "Acceptance"})
+ public void testFileBasedStoreObjectGet() throws Exception {
+ // The file system will have done a lot of caching here - we are unlikely to touch the disk more than once.
+ int numIterations = numIterations();
+ double minRatePerSec = 100 * PERFORMANCE_EXPECTATION;
+
+ measureAndAssert("FileBasedStoreObjectAccessor.get", numIterations, minRatePerSec, new Runnable() {
+ public void run() {
+ fileAccessor.get();
+ }});
+ }
+
+ @Test(groups={"Integration", "Acceptance"})
+ public void testFileBasedStoreObjectDelete() throws Exception {
+ int numIterations = numIterations();
+ double minRatePerSec = 100 * PERFORMANCE_EXPECTATION;
+
+ // Will do 10% warm up runs first
+ final List<File> files = Lists.newArrayList();
+ for (int i = 0; i < (numIterations * 1.1 + 1); i++) {
+ file = File.createTempFile("fileBasedStoreObjectDelete-"+i, ".txt");
+ Files.write(""+i, file, Charsets.UTF_8);
+ files.add(file);
+ }
+
+ final AtomicInteger i = new AtomicInteger();
+
+ try {
+ measureAndAssert("FileBasedStoreObjectAccessor.delete", numIterations, minRatePerSec, new Runnable() {
+ public void run() {
+ File file = files.get(i.getAndIncrement());
+ FileBasedStoreObjectAccessor fileAccessor = new FileBasedStoreObjectAccessor(file, "mytmpextension");
+ fileAccessor.delete();
+ }});
+ } finally {
+ for (File file : files) {
+ if (file != null) file.delete();
+ }
+ }
+ }
+
+ // fileAccessor.put() is implemented with an execCommands("mv") so look at performance of just that piece
+ @Test(groups={"Integration", "Acceptance"})
+ public void testProcessToolExecCommand() {
+ int numIterations = numIterations();
+ double minRatePerSec = 10 * PERFORMANCE_EXPECTATION;
+
+ measureAndAssert("ProcessTool.exec", numIterations, minRatePerSec, new Runnable() {
+ public void run() {
+ String cmd = "true";
+ new ProcessTool().execCommands(MutableMap.<String,String>of(), MutableList.of(cmd), null);
+ }});
+ }
+
+ @Test(groups={"Integration", "Acceptance"})
+ public void testJavaUtilFileRenames() {
+ int numIterations = numIterations();
+ double minRatePerSec = 10 * PERFORMANCE_EXPECTATION;
+
+ final File parentDir = file.getParentFile();
+ final AtomicInteger i = new AtomicInteger();
+
+ measureAndAssert("java.util.File.rename", numIterations, minRatePerSec, new Runnable() {
+ public void run() {
+ File newFile = new File(parentDir, "fileRename-"+i.incrementAndGet()+".txt");
+ file.renameTo(newFile);
+ file = newFile;
+ }});
+ }
+
+ @Test(groups={"Integration", "Acceptance"})
+ public void testGuavaFileWrites() {
+ int numIterations = numIterations();
+ double minRatePerSec = 10 * PERFORMANCE_EXPECTATION;
+
+ final AtomicInteger i = new AtomicInteger();
+
+ measureAndAssert("guava.Files.write", numIterations, minRatePerSec, new Runnable() {
+ public void run() {
+ try {
+ Files.write(""+i.incrementAndGet(), file, Charsets.UTF_8);
+ } catch (IOException e) {
+ throw Exceptions.propagate(e);
+ }
+ }});
+ }
+
+ @Test(groups={"Integration", "Acceptance"})
+ public void testGuavaFileMoves() {
+ int numIterations = numIterations();
+ double minRatePerSec = 10 * PERFORMANCE_EXPECTATION;
+
+ final File parentDir = file.getParentFile();
+ final AtomicInteger i = new AtomicInteger();
+
+ measureAndAssert("guava.Files.move", numIterations, minRatePerSec, new Runnable() {
+ public void run() {
+ File newFile = new File(parentDir, "fileRename-"+i.incrementAndGet()+".txt");
+ try {
+ Files.move(file, newFile);
+ } catch (IOException e) {
+ throw Exceptions.propagate(e);
+ }
+ file = newFile;
+ }});
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/870a1b8f/core/src/test/java/brooklyn/qa/performance/GroovyYardStickPerformanceTest.groovy
----------------------------------------------------------------------
diff --git a/core/src/test/java/brooklyn/qa/performance/GroovyYardStickPerformanceTest.groovy b/core/src/test/java/brooklyn/qa/performance/GroovyYardStickPerformanceTest.groovy
new file mode 100644
index 0000000..bb385a3
--- /dev/null
+++ b/core/src/test/java/brooklyn/qa/performance/GroovyYardStickPerformanceTest.groovy
@@ -0,0 +1,44 @@
+package brooklyn.qa.performance;
+
+import static org.testng.Assert.assertTrue
+
+import java.util.concurrent.ExecutorService
+import java.util.concurrent.Executors
+import java.util.concurrent.atomic.AtomicInteger
+
+import org.slf4j.Logger
+import org.slf4j.LoggerFactory
+import org.testng.annotations.AfterMethod
+import org.testng.annotations.BeforeMethod
+import org.testng.annotations.Test
+
+public class GroovyYardStickPerformanceTest extends AbstractPerformanceTest {
+
+ private static final Logger LOG = LoggerFactory.getLogger(GroovyYardStickPerformanceTest.class);
+
+ protected static final long TIMEOUT_MS = 10*1000;
+
+ private ExecutorService executor;
+
+ @BeforeMethod(alwaysRun=true)
+ public void setUp() {
+ super.setUp();
+ executor = Executors.newCachedThreadPool();
+ }
+
+ @AfterMethod(alwaysRun=true)
+ public void tearDown() {
+ super.tearDown();
+ if (executor != null) executor.shutdownNow();
+ }
+
+ @Test(groups=["Integration", "Acceptance"])
+ public void testGroovyNoopToEnsureTestFrameworkIsVeryFast() {
+ int numIterations = 1000000;
+ double minRatePerSec = 1000000 * PERFORMANCE_EXPECTATION;
+ AtomicInteger i = new AtomicInteger();
+
+ measureAndAssert("noop-groovy", numIterations, minRatePerSec, { i.incrementAndGet() });
+ assertTrue(i.get() >= numIterations, "i="+i);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/870a1b8f/core/src/test/java/brooklyn/qa/performance/JavaYardStickPerformanceTest.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/brooklyn/qa/performance/JavaYardStickPerformanceTest.java b/core/src/test/java/brooklyn/qa/performance/JavaYardStickPerformanceTest.java
index 8f4db29..5954b8a 100644
--- a/core/src/test/java/brooklyn/qa/performance/JavaYardStickPerformanceTest.java
+++ b/core/src/test/java/brooklyn/qa/performance/JavaYardStickPerformanceTest.java
@@ -23,13 +23,13 @@ public class JavaYardStickPerformanceTest extends AbstractPerformanceTest {
private ExecutorService executor;
@BeforeMethod(alwaysRun=true)
- public void setUp() {
+ public void setUp() throws Exception {
super.setUp();
executor = Executors.newCachedThreadPool();
}
@AfterMethod(alwaysRun=true)
- public void tearDown() {
+ public void tearDown() throws Exception {
super.tearDown();
if (executor != null) executor.shutdownNow();
}
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/870a1b8f/core/src/test/java/brooklyn/qa/performance/PerformanceTestUtils.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/brooklyn/qa/performance/PerformanceTestUtils.java b/core/src/test/java/brooklyn/qa/performance/PerformanceTestUtils.java
new file mode 100644
index 0000000..984ae30
--- /dev/null
+++ b/core/src/test/java/brooklyn/qa/performance/PerformanceTestUtils.java
@@ -0,0 +1,81 @@
+package brooklyn.qa.performance;
+
+import java.lang.management.ManagementFactory;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
+
+import javax.management.MBeanServer;
+import javax.management.ObjectName;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import brooklyn.util.time.Duration;
+import brooklyn.util.time.Time;
+
+import com.google.common.base.Stopwatch;
+
+public class PerformanceTestUtils {
+
+ private static final Logger LOG = LoggerFactory.getLogger(PerformanceTestUtils.class);
+
+ private static boolean hasLoggedProcessCpuTimeUnavailable;
+
+ public static long getProcessCpuTime() {
+ try {
+ MBeanServer mbeanServer = ManagementFactory.getPlatformMBeanServer();
+ ObjectName osMBeanName = ObjectName.getInstance(ManagementFactory.OPERATING_SYSTEM_MXBEAN_NAME);
+ return (Long) mbeanServer.getAttribute(osMBeanName, "ProcessCpuTime");
+ } catch (Exception e) {
+ if (!hasLoggedProcessCpuTimeUnavailable) {
+ hasLoggedProcessCpuTimeUnavailable = true;
+ LOG.warn("ProcessCPuTime not available in local JVM MXBean "+ManagementFactory.OPERATING_SYSTEM_MXBEAN_NAME+" (only available in sun JVM?)");
+ }
+ return -1;
+ }
+ }
+
+ /**
+ * Creates a background thread that will log.info the CPU fraction usage repeatedly, sampling at the given period.
+ * Callers <em>must</em> cancel the returned future, e.g. {@code future.cancel(true)}, otherwise it will keep
+ * logging until the JVM exits.
+ */
+ public static Future<?> sampleProcessCpuTime(final Duration period, final String loggingContext) {
+ final ExecutorService executor = Executors.newSingleThreadExecutor(new ThreadFactory() {
+ @Override public Thread newThread(Runnable r) {
+ Thread thread = new Thread(r, "brooklyn-sampleProcessCpuTime-"+loggingContext);
+ thread.setDaemon(true); // let the JVM exit
+ return thread;
+ }});
+ Future<?> future = executor.submit(new Runnable() {
+ @Override public void run() {
+ try {
+ long prevCpuTime = getProcessCpuTime();
+ if (prevCpuTime == -1) {
+ LOG.warn("ProcessCPuTime not available; cannot sample; aborting");
+ return;
+ }
+ while (true) {
+ Stopwatch stopwatch = Stopwatch.createStarted();
+ Thread.sleep(period.toMilliseconds());
+ long currentCpuTime = getProcessCpuTime();
+
+ long elapsedTime = stopwatch.elapsed(TimeUnit.MILLISECONDS);
+ double fractionCpu = (elapsedTime > 0) ? ((double)currentCpuTime-prevCpuTime) / TimeUnit.MILLISECONDS.toNanos(elapsedTime) : -1;
+ prevCpuTime = currentCpuTime;
+
+ LOG.info("CPU fraction over last {} was {} ({})", new Object[] {
+ Time.makeTimeStringRounded(elapsedTime), fractionCpu, loggingContext});
+ }
+ } catch (InterruptedException e) {
+ return; // graceful termination
+ } finally {
+ executor.shutdownNow();
+ }
+ }});
+ return future;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/870a1b8f/core/src/test/java/brooklyn/test/policy/TestPolicy.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/brooklyn/test/policy/TestPolicy.java b/core/src/test/java/brooklyn/test/policy/TestPolicy.java
index ca928e8..c5fd202 100644
--- a/core/src/test/java/brooklyn/test/policy/TestPolicy.java
+++ b/core/src/test/java/brooklyn/test/policy/TestPolicy.java
@@ -35,4 +35,9 @@ public class TestPolicy extends AbstractPolicy {
public Map<?, ?> getLeftoverProperties() {
return Collections.unmodifiableMap(leftoverProperties);
}
+
+ @Override
+ protected <T> void doReconfigureConfig(ConfigKey<T> key, T val) {
+ // no-op
+ }
}
[2/4] git commit: Covert perf tests from groovy to java
Posted by he...@apache.org.
Covert perf tests from groovy to java
Project: http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/commit/b1e27d8b
Tree: http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/tree/b1e27d8b
Diff: http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/diff/b1e27d8b
Branch: refs/heads/master
Commit: b1e27d8b1cd9d0745aa7a4c04c51c56de37a9297
Parents: 870a1b8
Author: Aled Sage <al...@gmail.com>
Authored: Tue Jun 24 08:55:56 2014 +0100
Committer: Aled Sage <al...@gmail.com>
Committed: Tue Jul 1 11:06:33 2014 +0100
----------------------------------------------------------------------
.../EntityPerformanceLongevityTest.groovy | 23 ---
.../EntityPerformanceLongevityTest.java | 17 ++
.../qa/performance/EntityPerformanceTest.java | 17 --
.../FilePersistencePerformanceTest.java | 4 -
.../JavaYardStickPerformanceTest.java | 4 -
.../SubscriptionPerformanceTest.groovy | 127 --------------
.../SubscriptionPerformanceTest.java | 150 +++++++++++++++++
.../qa/performance/TaskPerformanceTest.groovy | 121 --------------
.../qa/performance/TaskPerformanceTest.java | 164 +++++++++++++++++++
9 files changed, 331 insertions(+), 296 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/b1e27d8b/core/src/test/java/brooklyn/qa/performance/EntityPerformanceLongevityTest.groovy
----------------------------------------------------------------------
diff --git a/core/src/test/java/brooklyn/qa/performance/EntityPerformanceLongevityTest.groovy b/core/src/test/java/brooklyn/qa/performance/EntityPerformanceLongevityTest.groovy
deleted file mode 100644
index 08a009b..0000000
--- a/core/src/test/java/brooklyn/qa/performance/EntityPerformanceLongevityTest.groovy
+++ /dev/null
@@ -1,23 +0,0 @@
-package brooklyn.qa.performance
-
-import static brooklyn.test.TestUtils.*
-import static org.testng.Assert.*
-
-import org.slf4j.Logger
-import org.slf4j.LoggerFactory
-
-import brooklyn.test.entity.TestEntity
-
-
-public class EntityPerformanceLongevityTest extends EntityPerformanceTest {
-
- private static final Logger LOG = LoggerFactory.getLogger(EntityPerformanceLongevityTest.class)
-
- // TODO enable this to some big number to see what happens when things run for a long time.
- // e.g. will we eventually get OOME when storing all tasks relating to effector calls?
-
-// protected int numIterations() {
-// return 1000000
-// }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/b1e27d8b/core/src/test/java/brooklyn/qa/performance/EntityPerformanceLongevityTest.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/brooklyn/qa/performance/EntityPerformanceLongevityTest.java b/core/src/test/java/brooklyn/qa/performance/EntityPerformanceLongevityTest.java
new file mode 100644
index 0000000..5bb1279
--- /dev/null
+++ b/core/src/test/java/brooklyn/qa/performance/EntityPerformanceLongevityTest.java
@@ -0,0 +1,17 @@
+package brooklyn.qa.performance;
+
+import org.testng.annotations.Test;
+
+
+@Test(groups={"Integration", "Acceptance"})
+public class EntityPerformanceLongevityTest extends EntityPerformanceTest {
+
+ // TODO enable this to some big number to see what happens when things run for a long time.
+ // e.g. will we eventually get OOME when storing all tasks relating to effector calls?
+
+// @Override
+// protected int numIterations() {
+// return 1000000;
+// }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/b1e27d8b/core/src/test/java/brooklyn/qa/performance/EntityPerformanceTest.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/brooklyn/qa/performance/EntityPerformanceTest.java b/core/src/test/java/brooklyn/qa/performance/EntityPerformanceTest.java
index 74ab04b..6a3759a 100644
--- a/core/src/test/java/brooklyn/qa/performance/EntityPerformanceTest.java
+++ b/core/src/test/java/brooklyn/qa/performance/EntityPerformanceTest.java
@@ -5,8 +5,6 @@ import static org.testng.Assert.assertTrue;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
@@ -25,8 +23,6 @@ import com.google.common.collect.Lists;
public class EntityPerformanceTest extends AbstractPerformanceTest {
- private static final Logger LOG = LoggerFactory.getLogger(EntityPerformanceTest.class);
-
private static final long TIMEOUT_MS = 10*1000;
TestEntity entity;
@@ -51,19 +47,6 @@ public class EntityPerformanceTest extends AbstractPerformanceTest {
}
@Test(groups={"Integration", "Acceptance"})
- public void testGroovyNoopToEnsureTestFrameworkIsVeryFast() {
- int numIterations = numIterations();
- double minRatePerSec = 100000 * PERFORMANCE_EXPECTATION;
- final AtomicInteger i = new AtomicInteger();
-
- measureAndAssert("noop-groovy", numIterations, minRatePerSec, new Runnable() {
- public void run() {
- i.incrementAndGet();
- }});
- assertTrue(i.get() >= numIterations, "i="+i);
- }
-
- @Test(groups={"Integration", "Acceptance"})
public void testUpdateAttributeWhenNoListeners() {
int numIterations = numIterations();
double minRatePerSec = 1000 * PERFORMANCE_EXPECTATION;
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/b1e27d8b/core/src/test/java/brooklyn/qa/performance/FilePersistencePerformanceTest.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/brooklyn/qa/performance/FilePersistencePerformanceTest.java b/core/src/test/java/brooklyn/qa/performance/FilePersistencePerformanceTest.java
index eaf81ae..65efdb1 100644
--- a/core/src/test/java/brooklyn/qa/performance/FilePersistencePerformanceTest.java
+++ b/core/src/test/java/brooklyn/qa/performance/FilePersistencePerformanceTest.java
@@ -5,8 +5,6 @@ import java.io.IOException;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
@@ -23,8 +21,6 @@ import com.google.common.io.Files;
public class FilePersistencePerformanceTest extends AbstractPerformanceTest {
- private static final Logger LOG = LoggerFactory.getLogger(FilePersistencePerformanceTest.class);
-
File file;
FileBasedStoreObjectAccessor fileAccessor;
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/b1e27d8b/core/src/test/java/brooklyn/qa/performance/JavaYardStickPerformanceTest.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/brooklyn/qa/performance/JavaYardStickPerformanceTest.java b/core/src/test/java/brooklyn/qa/performance/JavaYardStickPerformanceTest.java
index 5954b8a..02f7073 100644
--- a/core/src/test/java/brooklyn/qa/performance/JavaYardStickPerformanceTest.java
+++ b/core/src/test/java/brooklyn/qa/performance/JavaYardStickPerformanceTest.java
@@ -6,8 +6,6 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
@@ -16,8 +14,6 @@ import com.google.common.base.Throwables;
public class JavaYardStickPerformanceTest extends AbstractPerformanceTest {
- private static final Logger LOG = LoggerFactory.getLogger(JavaYardStickPerformanceTest.class);
-
protected static final long TIMEOUT_MS = 10*1000;
private ExecutorService executor;
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/b1e27d8b/core/src/test/java/brooklyn/qa/performance/SubscriptionPerformanceTest.groovy
----------------------------------------------------------------------
diff --git a/core/src/test/java/brooklyn/qa/performance/SubscriptionPerformanceTest.groovy b/core/src/test/java/brooklyn/qa/performance/SubscriptionPerformanceTest.groovy
deleted file mode 100644
index 5cb2a07..0000000
--- a/core/src/test/java/brooklyn/qa/performance/SubscriptionPerformanceTest.groovy
+++ /dev/null
@@ -1,127 +0,0 @@
-package brooklyn.qa.performance
-
-import static brooklyn.test.TestUtils.*
-import static org.testng.Assert.*
-
-import java.util.concurrent.CountDownLatch
-import java.util.concurrent.TimeUnit
-import java.util.concurrent.atomic.AtomicInteger
-
-import org.slf4j.Logger
-import org.slf4j.LoggerFactory
-import org.testng.annotations.BeforeMethod
-import org.testng.annotations.Test
-
-import brooklyn.entity.proxying.EntitySpec
-import brooklyn.event.SensorEventListener
-import brooklyn.location.basic.SimulatedLocation
-import brooklyn.management.SubscriptionManager
-import brooklyn.test.entity.TestApplication
-import brooklyn.test.entity.TestEntity
-
-public class SubscriptionPerformanceTest extends AbstractPerformanceTest {
-
- private static final Logger LOG = LoggerFactory.getLogger(SubscriptionPerformanceTest.class)
-
- private static final long LONG_TIMEOUT_MS = 30*1000
- private static final int NUM_ITERATIONS = 10000
-
- TestEntity entity
- List<TestEntity> entities
- SubscriptionManager subscriptionManager
-
- @BeforeMethod(alwaysRun=true)
- public void setUp() {
- super.setUp()
-
- entities = []
- for (int i = 0; i < 10; i++) {
- entities += app.createAndManageChild(EntitySpec.create(TestEntity.class));
- }
- entity = entities[0]
- app.start([loc])
-
- subscriptionManager = app.managementContext.subscriptionManager
- }
-
- @Test(groups=["Integration", "Acceptance"])
- public void testManyPublishedOneSubscriber() {
- int numSubscribers = 1
- int numIterations = NUM_ITERATIONS
- double minRatePerSec = 100 * PERFORMANCE_EXPECTATION; // i.e. 100*10 events delivered per sec
- int iter = 0
- int expectedCount = numIterations*numSubscribers
-
- AtomicInteger listenerCount = new AtomicInteger()
- CountDownLatch completionLatch = new CountDownLatch(1)
-
- for (int i = 0; i < numSubscribers; i++) {
- subscriptionManager.subscribe([subscriber:i], entity, TestEntity.SEQUENCE,
- {
- int count = listenerCount.incrementAndGet()
- if (count >= expectedCount) completionLatch.countDown()
- } as SensorEventListener)
- }
-
- measureAndAssert("updateAttributeWithManyPublishedOneSubscriber", numIterations, minRatePerSec,
- { entity.setAttribute(TestEntity.SEQUENCE, (iter++)) },
- { completionLatch.await(LONG_TIMEOUT_MS, TimeUnit.MILLISECONDS); assertTrue(completionLatch.getCount() <= 0) })
- }
-
- // TODO but surely parallel should be much faster?!
- @Test(groups=["Integration", "Acceptance"])
- public void testManyListenersForSensorEvent() {
- int numSubscribers = 10
- int numIterations = NUM_ITERATIONS
- double minRatePerSec = 100 * PERFORMANCE_EXPECTATION; // i.e. 100*10 events delivered per sec
- int iter = 0
- int expectedCount = numIterations*numSubscribers
-
- AtomicInteger listenerCount = new AtomicInteger()
- CountDownLatch completionLatch = new CountDownLatch(1)
-
- for (int i = 0; i < numSubscribers; i++) {
- subscriptionManager.subscribe([subscriber:i], entity, TestEntity.SEQUENCE,
- {
- int count = listenerCount.incrementAndGet()
- if (count >= expectedCount) completionLatch.countDown()
- } as SensorEventListener)
- }
-
- measureAndAssert("updateAttributeWithManyListeners", numIterations, minRatePerSec,
- { entity.setAttribute(TestEntity.SEQUENCE, (iter++)) },
- { completionLatch.await(LONG_TIMEOUT_MS, TimeUnit.MILLISECONDS); assertTrue(completionLatch.getCount() <= 0) })
- }
-
- @Test(groups=["Integration", "Acceptance"])
- public void testUpdateAttributeWithNoListenersButManyUnrelatedListeners() {
- int numUnrelatedSubscribers = 1000
- int numIterations = NUM_ITERATIONS
- double minRatePerSec = 1000 * PERFORMANCE_EXPECTATION;
- int iter = 0
- int lastVal = 0
- Exception exception
-
- for (int i = 0; i < (numUnrelatedSubscribers/2); i++) {
- subscriptionManager.subscribe([subscriber:i], entities[1], TestEntity.SEQUENCE,
- {
- exception = new RuntimeException("Unrelated subscriber called with $it")
- throw exception
- } as SensorEventListener)
- subscriptionManager.subscribe([subscriber:i], entity, TestEntity.MY_NOTIF,
- {
- exception = new RuntimeException("Unrelated subscriber called with $it")
- throw exception
- } as SensorEventListener)
- }
-
- measureAndAssert("updateAttributeWithUnrelatedListeners", numIterations, minRatePerSec) {
- entity.setAttribute(TestEntity.SEQUENCE, (++iter))
- }
-
- if (exception != null) {
- throw exception
- }
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/b1e27d8b/core/src/test/java/brooklyn/qa/performance/SubscriptionPerformanceTest.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/brooklyn/qa/performance/SubscriptionPerformanceTest.java b/core/src/test/java/brooklyn/qa/performance/SubscriptionPerformanceTest.java
new file mode 100644
index 0000000..4b7752f
--- /dev/null
+++ b/core/src/test/java/brooklyn/qa/performance/SubscriptionPerformanceTest.java
@@ -0,0 +1,150 @@
+package brooklyn.qa.performance;
+
+import static org.testng.Assert.assertTrue;
+
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+import brooklyn.entity.proxying.EntitySpec;
+import brooklyn.event.SensorEvent;
+import brooklyn.event.SensorEventListener;
+import brooklyn.management.SubscriptionManager;
+import brooklyn.test.entity.TestEntity;
+import brooklyn.util.collections.MutableMap;
+import brooklyn.util.exceptions.Exceptions;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Lists;
+
+public class SubscriptionPerformanceTest extends AbstractPerformanceTest {
+
+ private static final long LONG_TIMEOUT_MS = 30*1000;
+ private static final int NUM_ITERATIONS = 10000;
+
+ TestEntity entity;
+ List<TestEntity> entities;
+ SubscriptionManager subscriptionManager;
+
+ @BeforeMethod(alwaysRun=true)
+ public void setUp() throws Exception {
+ super.setUp();
+
+ entities = Lists.newArrayList();
+ for (int i = 0; i < 10; i++) {
+ entities.add(app.createAndManageChild(EntitySpec.create(TestEntity.class)));
+ }
+ entity = entities.get(0);
+ app.start(ImmutableList.of(loc));
+
+ subscriptionManager = app.getManagementContext().getSubscriptionManager();
+ }
+
+ @Test(groups={"Integration", "Acceptance"})
+ public void testManyPublishedOneSubscriber() throws Exception {
+ int numSubscribers = 1;
+ int numIterations = NUM_ITERATIONS;
+ double minRatePerSec = 100 * PERFORMANCE_EXPECTATION; // i.e. 100*10 events delivered per sec
+ final AtomicInteger iter = new AtomicInteger();
+ final int expectedCount = numIterations*numSubscribers;
+
+ final AtomicInteger listenerCount = new AtomicInteger();
+ final CountDownLatch completionLatch = new CountDownLatch(1);
+
+ for (int i = 0; i < numSubscribers; i++) {
+ subscriptionManager.subscribe(MutableMap.<String, Object>of("subscriber", i), entity, TestEntity.SEQUENCE, new SensorEventListener<Integer>() {
+ public void onEvent(SensorEvent<Integer> event) {
+ int count = listenerCount.incrementAndGet();
+ if (count >= expectedCount) completionLatch.countDown();
+ }});
+ }
+
+ measureAndAssert("updateAttributeWithManyPublishedOneSubscriber", numIterations, minRatePerSec,
+ new Runnable() {
+ public void run() {
+ entity.setAttribute(TestEntity.SEQUENCE, (iter.getAndIncrement()));
+ }
+ },
+ new Runnable() {
+ public void run() {
+ try {
+ completionLatch.await(LONG_TIMEOUT_MS, TimeUnit.MILLISECONDS);
+ } catch (InterruptedException e) {
+ throw Exceptions.propagate(e);
+ }
+ assertTrue(completionLatch.getCount() <= 0);
+ }
+ });
+ }
+
+ @Test(groups={"Integration", "Acceptance"})
+ public void testManyListenersForSensorEvent() throws Exception {
+ int numSubscribers = 10;
+ int numIterations = NUM_ITERATIONS;
+ double minRatePerSec = 100 * PERFORMANCE_EXPECTATION; // i.e. 100*10 events delivered per sec
+ final AtomicInteger iter = new AtomicInteger();
+ final int expectedCount = numIterations*numSubscribers;
+
+ final AtomicInteger listenerCount = new AtomicInteger();
+ final CountDownLatch completionLatch = new CountDownLatch(1);
+
+ for (int i = 0; i < numSubscribers; i++) {
+ subscriptionManager.subscribe(MutableMap.<String, Object>of("subscriber", i), entity, TestEntity.SEQUENCE, new SensorEventListener<Integer>() {
+ public void onEvent(SensorEvent<Integer> event) {
+ int count = listenerCount.incrementAndGet();
+ if (count >= expectedCount) completionLatch.countDown();
+ }});
+ }
+
+ measureAndAssert("updateAttributeWithManyListeners", numIterations, minRatePerSec,
+ new Runnable() {
+ @Override public void run() {
+ entity.setAttribute(TestEntity.SEQUENCE, (iter.getAndIncrement()));
+ }},
+ new Runnable() {
+ public void run() {
+ try {
+ completionLatch.await(LONG_TIMEOUT_MS, TimeUnit.MILLISECONDS);
+ } catch (InterruptedException e) {
+ throw Exceptions.propagate(e);
+ }
+ assertTrue(completionLatch.getCount() <= 0);
+ }});
+ }
+
+ @Test(groups={"Integration", "Acceptance"})
+ public void testUpdateAttributeWithNoListenersButManyUnrelatedListeners() throws Exception {
+ int numUnrelatedSubscribers = 1000;
+ int numIterations = NUM_ITERATIONS;
+ double minRatePerSec = 1000 * PERFORMANCE_EXPECTATION;
+ final AtomicInteger iter = new AtomicInteger();
+ final AtomicReference<RuntimeException> exception = new AtomicReference<RuntimeException>();
+
+ for (int i = 0; i < (numUnrelatedSubscribers/2); i++) {
+ subscriptionManager.subscribe(MutableMap.<String, Object>of("subscriber", i), entities.get(1), TestEntity.SEQUENCE, new SensorEventListener<Integer>() {
+ public void onEvent(SensorEvent<Integer> event) {
+ exception.set(new RuntimeException("Unrelated subscriber called with "+event));
+ throw exception.get();
+ }});
+ subscriptionManager.subscribe(MutableMap.<String, Object>of("subscriber", i), entity, TestEntity.MY_NOTIF, new SensorEventListener<Integer>() {
+ public void onEvent(SensorEvent<Integer> event) {
+ exception.set(new RuntimeException("Unrelated subscriber called with "+event));
+ throw exception.get();
+ }});
+ }
+
+ measureAndAssert("updateAttributeWithUnrelatedListeners", numIterations, minRatePerSec, new Runnable() {
+ @Override public void run() {
+ entity.setAttribute(TestEntity.SEQUENCE, (iter.incrementAndGet()));
+ }});
+
+ if (exception.get() != null) {
+ throw exception.get();
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/b1e27d8b/core/src/test/java/brooklyn/qa/performance/TaskPerformanceTest.groovy
----------------------------------------------------------------------
diff --git a/core/src/test/java/brooklyn/qa/performance/TaskPerformanceTest.groovy b/core/src/test/java/brooklyn/qa/performance/TaskPerformanceTest.groovy
deleted file mode 100644
index 8d0d2ae..0000000
--- a/core/src/test/java/brooklyn/qa/performance/TaskPerformanceTest.groovy
+++ /dev/null
@@ -1,121 +0,0 @@
-package brooklyn.qa.performance
-
-import brooklyn.management.internal.LocalManagementContext
-import org.testng.annotations.AfterMethod
-
-import static brooklyn.test.TestUtils.*
-import static org.testng.Assert.*
-
-import java.util.concurrent.CopyOnWriteArrayList
-import java.util.concurrent.CountDownLatch
-import java.util.concurrent.TimeUnit
-import java.util.concurrent.atomic.AtomicInteger
-
-import org.slf4j.Logger
-import org.slf4j.LoggerFactory
-import org.testng.annotations.BeforeMethod
-import org.testng.annotations.Test
-
-import brooklyn.util.task.BasicExecutionManager
-import brooklyn.util.task.SingleThreadedScheduler
-
-public class TaskPerformanceTest extends AbstractPerformanceTest {
-
- private static final Logger LOG = LoggerFactory.getLogger(TaskPerformanceTest.class)
-
- private static final long LONG_TIMEOUT_MS = 30*1000
-
- BasicExecutionManager executionManager
-
- @BeforeMethod(alwaysRun=true)
- public void setUp() {
- super.setUp()
-
- app.start([loc])
-
- executionManager = app.managementContext.executionManager
- }
-
- public static final int numIterations = 200000;
-
- @Test(groups=["Integration", "Acceptance"])
- public void testExecuteSimplestRunnable() {
- double minRatePerSec = 1000 * PERFORMANCE_EXPECTATION;
-
- final AtomicInteger counter = new AtomicInteger();
- final CountDownLatch completionLatch = new CountDownLatch(1)
-
- Runnable work = new Runnable() { public void run() {
- int val = counter.incrementAndGet()
- if (val >= numIterations) completionLatch.countDown()
- }}
-
- measureAndAssert("executeSimplestRunnable", numIterations, minRatePerSec,
- { executionManager.submit(work) },
- { completionLatch.await(LONG_TIMEOUT_MS, TimeUnit.MILLISECONDS); assertTrue(completionLatch.getCount() <= 0) })
- }
-
- @Test(groups=["Integration", "Acceptance"])
- public void testExecuteRunnableWithTags() {
- double minRatePerSec = 1000 * PERFORMANCE_EXPECTATION;
-
- final AtomicInteger counter = new AtomicInteger();
- final CountDownLatch completionLatch = new CountDownLatch(1)
-
- Runnable work = new Runnable() { public void run() {
- int val = counter.incrementAndGet()
- if (val >= numIterations) completionLatch.countDown()
- }}
-
- Map flags = [tags:["a","b"]]
-
- measureAndAssert("testExecuteRunnableWithTags", numIterations, minRatePerSec,
- { executionManager.submit(flags, work) },
- { completionLatch.await(LONG_TIMEOUT_MS, TimeUnit.MILLISECONDS); assertTrue(completionLatch.getCount() <= 0) })
- }
-
- @Test(groups=["Integration", "Acceptance"])
- public void testExecuteWithSingleThreadedScheduler() {
- double minRatePerSec = 1000 * PERFORMANCE_EXPECTATION;
-
- executionManager.setTaskSchedulerForTag("singlethreaded", SingleThreadedScheduler.class);
-
- final AtomicInteger concurrentCallCount = new AtomicInteger();
- final AtomicInteger submitCount = new AtomicInteger();
- final AtomicInteger counter = new AtomicInteger();
- final CountDownLatch completionLatch = new CountDownLatch(1)
- final List<Exception> exceptions = new CopyOnWriteArrayList()
-
- Runnable work = new Runnable() { public void run() {
- int numConcurrentCalls = concurrentCallCount.incrementAndGet()
- try {
- if (numConcurrentCalls > 1) throw new IllegalStateException("numConcurrentCalls=$numConcurrentCalls")
- int val = counter.incrementAndGet()
- if (val >= numIterations) completionLatch.countDown()
- } catch (Exception e) {
- exceptions.add(e)
- LOG.warn("Exception in runnable of testExecuteWithSingleThreadedScheduler", e)
- throw e
- } finally {
- concurrentCallCount.decrementAndGet()
- }
- }}
-
- measureAndAssert("testExecuteWithSingleThreadedScheduler", numIterations, minRatePerSec,
- {
- while (submitCount.get() > counter.get() + 5000) {
- LOG.info("delaying because ${submitCount.get()} submitted and only ${counter.get()} run")
- Thread.sleep(500);
- }
- executionManager.submit([tags:["singlethreaded"]], work); submitCount.incrementAndGet(); },
- { completionLatch.await(LONG_TIMEOUT_MS, TimeUnit.MILLISECONDS); assertTrue(completionLatch.getCount() <= 0) })
-
- if (exceptions.size() > 0) throw exceptions.get(0)
- }
-
- public static void main(String[] args) {
- def t = new TaskPerformanceTest();
- t.setUp();
- t.testExecuteWithSingleThreadedScheduler();
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/b1e27d8b/core/src/test/java/brooklyn/qa/performance/TaskPerformanceTest.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/brooklyn/qa/performance/TaskPerformanceTest.java b/core/src/test/java/brooklyn/qa/performance/TaskPerformanceTest.java
new file mode 100644
index 0000000..2b549cc
--- /dev/null
+++ b/core/src/test/java/brooklyn/qa/performance/TaskPerformanceTest.java
@@ -0,0 +1,164 @@
+package brooklyn.qa.performance;
+
+import static org.testng.Assert.assertTrue;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+import brooklyn.util.collections.MutableMap;
+import brooklyn.util.exceptions.Exceptions;
+import brooklyn.util.task.BasicExecutionManager;
+import brooklyn.util.task.SingleThreadedScheduler;
+import brooklyn.util.time.Time;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Lists;
+
+public class TaskPerformanceTest extends AbstractPerformanceTest {
+
+ private static final Logger LOG = LoggerFactory.getLogger(TaskPerformanceTest.class);
+
+ private static final long LONG_TIMEOUT_MS = 30*1000;
+
+ BasicExecutionManager executionManager;
+
+ @BeforeMethod(alwaysRun=true)
+ public void setUp() throws Exception {
+ super.setUp();
+
+ app.start(ImmutableList.of(loc));
+
+ executionManager = (BasicExecutionManager) app.getManagementContext().getExecutionManager();
+ }
+
+ public static final int numIterations = 200000;
+
+ @Test(groups={"Integration", "Acceptance"})
+ public void testExecuteSimplestRunnable() {
+ double minRatePerSec = 1000 * PERFORMANCE_EXPECTATION;
+
+ final AtomicInteger counter = new AtomicInteger();
+ final CountDownLatch completionLatch = new CountDownLatch(1);
+
+ final Runnable work = new Runnable() {
+ public void run() {
+ int val = counter.incrementAndGet();
+ if (val >= numIterations) completionLatch.countDown();
+ }};
+
+ measureAndAssert("executeSimplestRunnable", numIterations, minRatePerSec,
+ new Runnable() {
+ public void run() {
+ executionManager.submit(work);
+ }},
+ new Runnable() {
+ public void run() {
+ try {
+ completionLatch.await(LONG_TIMEOUT_MS, TimeUnit.MILLISECONDS);
+ } catch (InterruptedException e) {
+ throw Exceptions.propagate(e);
+ }
+ assertTrue(completionLatch.getCount() <= 0);
+ }});
+ }
+
+ @Test(groups={"Integration", "Acceptance"})
+ public void testExecuteRunnableWithTags() {
+ double minRatePerSec = 1000 * PERFORMANCE_EXPECTATION;
+
+ final AtomicInteger counter = new AtomicInteger();
+ final CountDownLatch completionLatch = new CountDownLatch(1);
+
+ final Runnable work = new Runnable() { public void run() {
+ int val = counter.incrementAndGet();
+ if (val >= numIterations) completionLatch.countDown();
+ }
+ };
+
+ final Map<String, ?> flags = MutableMap.of("tags", ImmutableList.of("a","b"));
+
+ measureAndAssert("testExecuteRunnableWithTags", numIterations, minRatePerSec,
+ new Runnable() {
+ public void run() {
+ executionManager.submit(flags, work);
+ }},
+ new Runnable() {
+ public void run() {
+ try {
+ completionLatch.await(LONG_TIMEOUT_MS, TimeUnit.MILLISECONDS);
+ } catch (InterruptedException e) {
+ throw Exceptions.propagate(e);
+ }
+ assertTrue(completionLatch.getCount() <= 0);
+ }});
+ }
+
+ @Test(groups={"Integration", "Acceptance"})
+ public void testExecuteWithSingleThreadedScheduler() throws Exception {
+ double minRatePerSec = 1000 * PERFORMANCE_EXPECTATION;
+
+ executionManager.setTaskSchedulerForTag("singlethreaded", SingleThreadedScheduler.class);
+
+ final AtomicInteger concurrentCallCount = new AtomicInteger();
+ final AtomicInteger submitCount = new AtomicInteger();
+ final AtomicInteger counter = new AtomicInteger();
+ final CountDownLatch completionLatch = new CountDownLatch(1);
+ final List<Exception> exceptions = Lists.newCopyOnWriteArrayList();
+
+ final Runnable work = new Runnable() { public void run() {
+ int numConcurrentCalls = concurrentCallCount.incrementAndGet();
+ try {
+ if (numConcurrentCalls > 1) throw new IllegalStateException("numConcurrentCalls="+numConcurrentCalls);
+ int val = counter.incrementAndGet();
+ if (val >= numIterations) completionLatch.countDown();
+ } catch (Exception e) {
+ exceptions.add(e);
+ LOG.warn("Exception in runnable of testExecuteWithSingleThreadedScheduler", e);
+ throw Exceptions.propagate(e);
+ } finally {
+ concurrentCallCount.decrementAndGet();
+ }
+ }
+ };
+
+ measureAndAssert("testExecuteWithSingleThreadedScheduler", numIterations, minRatePerSec,
+ new Runnable() {
+ public void run() {
+ while (submitCount.get() > counter.get() + 5000) {
+ LOG.info("delaying because "+submitCount.get()+" submitted and only "+counter.get()+" run");
+ Time.sleep(500);
+ }
+ executionManager.submit(MutableMap.of("tags", ImmutableList.of("singlethreaded")), work);
+ submitCount.incrementAndGet();
+ }},
+ new Runnable() {
+ public void run() {
+ try {
+ completionLatch.await(LONG_TIMEOUT_MS, TimeUnit.MILLISECONDS);
+ } catch (InterruptedException e) {
+ throw Exceptions.propagate(e);
+ }
+ assertTrue(completionLatch.getCount() <= 0);
+ }});
+
+ if (exceptions.size() > 0) throw exceptions.get(0);
+ }
+
+ public static void main(String[] args) throws Exception {
+ TaskPerformanceTest t = new TaskPerformanceTest();
+ t.setUp();
+ try {
+ t.testExecuteWithSingleThreadedScheduler();
+ } finally {
+ t.tearDown();
+ }
+ }
+}
[4/4] git commit: This closes #10
Posted by he...@apache.org.
This closes #10
Project: http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/commit/9ca8250b
Tree: http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/tree/9ca8250b
Diff: http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/diff/9ca8250b
Branch: refs/heads/master
Commit: 9ca8250bd3e74b95ce6970275de6ef1223b460bc
Parents: 490ebbd cbb0ffa
Author: Alex Heneveld <al...@cloudsoftcorp.com>
Authored: Tue Jul 1 11:43:59 2014 +0100
Committer: Alex Heneveld <al...@cloudsoftcorp.com>
Committed: Tue Jul 1 11:43:59 2014 +0100
----------------------------------------------------------------------
.../BrooklynMementoPersisterToObjectStore.java | 16 +-
.../rebind/persister/FileBasedObjectStore.java | 28 ++--
.../rebind/persister/MementoFileWriter.java | 2 +-
.../entity/rebind/RebindTestFixture.java | 12 +-
.../SshMachineLocationPerformanceTest.java | 11 +-
.../qa/performance/AbstractPerformanceTest.java | 10 +-
.../EntityPerformanceLongevityTest.groovy | 23 ---
.../EntityPerformanceLongevityTest.java | 17 ++
.../qa/performance/EntityPerformanceTest.java | 19 +--
.../EntityPersistencePerformanceTest.java | 81 +++++++++
.../FilePersistencePerformanceTest.java | 167 +++++++++++++++++++
.../GroovyYardStickPerformanceTest.groovy | 44 +++++
.../JavaYardStickPerformanceTest.java | 8 +-
.../qa/performance/PerformanceTestUtils.java | 81 +++++++++
.../SubscriptionPerformanceTest.groovy | 127 --------------
.../SubscriptionPerformanceTest.java | 150 +++++++++++++++++
.../qa/performance/TaskPerformanceTest.groovy | 121 --------------
.../qa/performance/TaskPerformanceTest.java | 164 ++++++++++++++++++
.../java/brooklyn/test/policy/TestPolicy.java | 5 +
19 files changed, 756 insertions(+), 330 deletions(-)
----------------------------------------------------------------------