You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by kw...@apache.org on 2015/09/14 17:29:43 UTC
svn commit: r1702969 [1/3] - in /qpid/java/trunk: perftests/ perftests/etc/
perftests/etc/ha/ perftests/etc/ha/chardefs/ perftests/etc/ha/testdefs/
perftests/etc/testdefs/ perftests/example/
perftests/src/main/java/org/apache/qpid/disttest/ perftests/s...
Author: kwall
Date: Mon Sep 14 15:29:42 2015
New Revision: 1702969
URL: http://svn.apache.org/r1702969
Log:
QPID-6732: [Java Perf Tests] Change PerfTestFrame to use hill climbing algorithm to discover the a maximum sustainable throughput for a given test-case
Work by Lorenz Quack <qu...@gmail.com> and Keith Wall
Added:
qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/controller/AbstractTestRunner.java
- copied, changed from r1702832, qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/controller/TestRunner.java
qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/controller/HillClimber.java
qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/controller/HillClimbingTestRunner.java
qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/controller/ITestRunner.java
- copied, changed from r1702832, qpid/java/trunk/qpid-perftests-systests/src/test/java/org/apache/qpid/systest/disttest/SystemTestConstants.java
qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/controller/OrdinaryTestRunner.java
- copied, changed from r1702832, qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/client/utils/ExecutorWithLimits.java
qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/message/CreateParticipantCommand.java
- copied, changed from r1702832, qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/message/CreateParticpantCommand.java
qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/message/StartDataCollectionCommand.java
- copied, changed from r1702832, qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/client/utils/ExecutorWithNoLimits.java
qpid/java/trunk/perftests/src/main/resources/logback.xml
qpid/java/trunk/perftests/src/test/java/org/apache/qpid/disttest/controller/HillClimberTest.java
qpid/java/trunk/qpid-perftests-systests/src/test/resources/org/apache/qpid/systest/disttest/endtoend/hillclimbing.js
qpid/java/trunk/qpid-perftests-systests/src/test/resources/org/apache/qpid/systest/disttest/endtoend/iterations.json
- copied, changed from r1702832, qpid/java/trunk/qpid-perftests-systests/src/test/resources/org/apache/qpid/systest/disttest/controllerandclient/iteratingFeature.json
qpid/java/trunk/qpid-perftests-systests/src/test/resources/org/apache/qpid/systest/disttest/endtoend/manyparticipants.json
Removed:
qpid/java/trunk/perftests/etc/ha/chardefs/1001-ScalingParticipants.chartdef
qpid/java/trunk/perftests/etc/ha/readme.txt
qpid/java/trunk/perftests/etc/ha/testdefs/ScalingParticipants.js
qpid/java/trunk/perftests/etc/logback.xml
qpid/java/trunk/perftests/etc/run-perftests.sh
qpid/java/trunk/perftests/example/PQBT-AA-Qpid-01.json
qpid/java/trunk/perftests/example/TQBT-TX-Qpid-01.json
qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/Benchmark.java
qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/client/utils/ExecutorWithLimits.java
qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/client/utils/ExecutorWithLimitsFactory.java
qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/client/utils/ExecutorWithNoLimits.java
qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/client/utils/ExecutorWithTimeLimit.java
qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/controller/TestRunner.java
qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/message/CreateParticpantCommand.java
qpid/java/trunk/perftests/src/main/resources/Benchmark.js
qpid/java/trunk/perftests/src/test/java/org/apache/qpid/disttest/client/utils/ExecutorWithNoLimitsTest.java
qpid/java/trunk/perftests/src/test/java/org/apache/qpid/disttest/client/utils/ExecutorWithTimeLimitTest.java
qpid/java/trunk/qpid-perftests-systests/src/test/java/org/apache/qpid/systest/disttest/ConfigFileTestHelper.java
qpid/java/trunk/qpid-perftests-systests/src/test/java/org/apache/qpid/systest/disttest/SystemTestConstants.java
qpid/java/trunk/qpid-perftests-systests/src/test/java/org/apache/qpid/systest/disttest/clientonly/BasicDistributedClientTest.java
qpid/java/trunk/qpid-perftests-systests/src/test/java/org/apache/qpid/systest/disttest/clientonly/ConsumerParticipantTest.java
qpid/java/trunk/qpid-perftests-systests/src/test/java/org/apache/qpid/systest/disttest/clientonly/ControllerQueue.java
qpid/java/trunk/qpid-perftests-systests/src/test/java/org/apache/qpid/systest/disttest/clientonly/DistributedClientTest.java
qpid/java/trunk/qpid-perftests-systests/src/test/java/org/apache/qpid/systest/disttest/clientonly/MessageProviderTest.java
qpid/java/trunk/qpid-perftests-systests/src/test/java/org/apache/qpid/systest/disttest/clientonly/ProducerParticipantTest.java
qpid/java/trunk/qpid-perftests-systests/src/test/java/org/apache/qpid/systest/disttest/controllerandclient/ControllerAndClientTest.java
qpid/java/trunk/qpid-perftests-systests/src/test/java/org/apache/qpid/systest/disttest/controlleronly/DistributedControllerTest.java
qpid/java/trunk/qpid-perftests-systests/src/test/resources/org/apache/qpid/systest/disttest/controllerandclient/iteratingFeature.json
qpid/java/trunk/qpid-perftests-systests/src/test/resources/org/apache/qpid/systest/disttest/controllerandclient/produceClient.json
qpid/java/trunk/qpid-perftests-systests/src/test/resources/org/apache/qpid/systest/disttest/controllerandclient/producerAndConsumerInSeparateClients.json
qpid/java/trunk/qpid-perftests-systests/src/test/resources/org/apache/qpid/systest/disttest/controllerandclient/producerAndThreeConsumersInSeparateClients.json
qpid/java/trunk/qpid-perftests-systests/src/test/resources/org/apache/qpid/systest/disttest/controllerandclient/testWithTwoTests.json
qpid/java/trunk/qpid-perftests-systests/src/test/resources/org/apache/qpid/systest/disttest/controlleronly/distributedControllerTest.json
qpid/java/trunk/qpid-perftests-systests/src/test/resources/org/apache/qpid/systest/disttest/endtoend/endtoend.json
Modified:
qpid/java/trunk/perftests/etc/testdefs/MessageSize.json
qpid/java/trunk/perftests/example/config-one-test.json
qpid/java/trunk/perftests/example/config-stress-testing-manyp-c1-with-psfc.json
qpid/java/trunk/perftests/example/config-stress-testing.json
qpid/java/trunk/perftests/example/config-two-tests.json
qpid/java/trunk/perftests/pom.xml
qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/ControllerRunner.java
qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/client/Client.java
qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/client/ClientCommandVisitor.java
qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/client/ConsumerParticipant.java
qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/client/Participant.java
qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/client/ParticipantExecutor.java
qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/client/ParticipantResultFactory.java
qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/client/ProducerParticipant.java
qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/controller/Controller.java
qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/controller/TestResult.java
qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/controller/TestRunnerFactory.java
qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/controller/config/ConfigReader.java
qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/controller/config/ParticipantConfig.java
qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/controller/config/ProducerConfig.java
qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/controller/config/SessionConfig.java
qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/controller/config/TestInstance.java
qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/db/ResultsDbWriter.java
qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/jms/ClientJmsDelegate.java
qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/jms/ControllerJmsDelegate.java
qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/jms/JmsMessageAdaptor.java
qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/jms/QpidQueueCreator.java
qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/message/CommandType.java
qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/message/CreateConsumerCommand.java
qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/message/CreateProducerCommand.java
qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/message/ParticipantAttribute.java
qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/message/ProducerParticipantResult.java
qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/results/aggregation/TestResultAggregator.java
qpid/java/trunk/perftests/src/test/java/org/apache/qpid/disttest/client/ClientTest.java
qpid/java/trunk/perftests/src/test/java/org/apache/qpid/disttest/client/ConsumerParticipantTest.java
qpid/java/trunk/perftests/src/test/java/org/apache/qpid/disttest/client/ParticipantExecutorTest.java
qpid/java/trunk/perftests/src/test/java/org/apache/qpid/disttest/client/ParticipantResultFactoryTest.java
qpid/java/trunk/perftests/src/test/java/org/apache/qpid/disttest/client/ProducerParticipantTest.java
qpid/java/trunk/perftests/src/test/java/org/apache/qpid/disttest/controller/ControllerTest.java
qpid/java/trunk/perftests/src/test/java/org/apache/qpid/disttest/controller/TestRunnerTest.java
qpid/java/trunk/perftests/src/test/java/org/apache/qpid/disttest/controller/config/ProducerConfigTest.java
qpid/java/trunk/perftests/src/test/java/org/apache/qpid/disttest/controller/config/TestInstanceTest.java
qpid/java/trunk/perftests/src/test/java/org/apache/qpid/disttest/controller/config/sampleConfig.json
qpid/java/trunk/perftests/src/test/java/org/apache/qpid/disttest/message/ParticipantResultTest.java
qpid/java/trunk/perftests/src/test/java/org/apache/qpid/disttest/results/ResultsTestFixture.java
qpid/java/trunk/perftests/src/test/java/org/apache/qpid/disttest/results/formatting/expectedOutput.csv
qpid/java/trunk/qpid-perftests-systests/src/test/java/org/apache/qpid/systest/disttest/DistributedTestSystemTestBase.java
qpid/java/trunk/qpid-perftests-systests/src/test/java/org/apache/qpid/systest/disttest/endtoend/EndToEndTest.java
Modified: qpid/java/trunk/perftests/etc/testdefs/MessageSize.json
URL: http://svn.apache.org/viewvc/qpid/java/trunk/perftests/etc/testdefs/MessageSize.json?rev=1702969&r1=1702968&r2=1702969&view=diff
==============================================================================
--- qpid/java/trunk/perftests/etc/testdefs/MessageSize.json (original)
+++ qpid/java/trunk/perftests/etc/testdefs/MessageSize.json Mon Sep 14 15:29:42 2015
@@ -5,46 +5,11 @@
"_iterations":[
{
"_messageSize": 256
- },
- {
- "_messageSize": 512
- },
- {
- "_messageSize": 1024
- },
- {
- "_messageSize": 2048
- },
- {
- "_messageSize": 4096
- },
- {
- "_messageSize": 8192
- },
- {
- "_messageSize": 16384
- },
- {
- "_messageSize": 32768
- },
- {
- "_messageSize": 65536
- },
- {
- "_messageSize": 131072
- },
- {
- "_messageSize": 262144
}
],
"_queues":[
{
- "_name": "direct://amq.direct//messageSize",
- "_attributes":
- {
- "x-qpid-capacity": 10485760,
- "x-qpid-flow-resume-capacity": 8388608
- }
+ "_name": "direct://amq.direct//messageSize"
}
],
"_clients":[
@@ -88,103 +53,6 @@
"_maximumDuration": 30000
}
]
- }
- ]
- }
- ]
- }
- ]
- },
- {
- "_name": "Message Size - 1P-1C - PERSISTENT",
- "_iterations":[
- {
- "_messageSize": 256
- },
- {
- "_messageSize": 512
- },
- {
- "_messageSize": 1024
- },
- {
- "_messageSize": 2048
- },
- {
- "_messageSize": 4096
- },
- {
- "_messageSize": 8192
- },
- {
- "_messageSize": 16384
- },
- {
- "_messageSize": 32768
- },
- {
- "_messageSize": 65536
- },
- {
- "_messageSize": 131072
- },
- {
- "_messageSize": 262144
- }
- ],
- "_queues":[
- {
- "_name": "direct://amq.direct//messageSize?durable='true'",
- "_durable": true,
- "_attributes":
- {
- "x-qpid-capacity": 10485760,
- "x-qpid-flow-resume-capacity": 8388608
- }
- }
- ],
- "_clients":[
- {
- "_name": "producingClient",
- "_connections":[
- {
- "_name": "connection1",
- "_factory": "connectionfactory",
- "_sessions": [
- {
- "_sessionName": "session1",
- "_acknowledgeMode": 0,
- "_producers": [
- {
- "_name": "Producer1",
- "_destinationName": "direct://amq.direct//messageSize?durable='true'",
- "_deliveryMode": 2,
- "_maximumDuration": 30000
-
- }
- ]
- }
- ]
- }
- ]
- },
- {
- "_name": "consumingClient",
- "_connections":[
- {
- "_name": "connection1",
- "_factory": "connectionfactory",
- "_sessions": [
- {
- "_sessionName": "session1",
- "_acknowledgeMode": 0,
- "_consumers": [
- {
- "_name": "Consumer1",
- "_destinationName": "direct://amq.direct//messageSize?durable='true'",
- "_maximumDuration": 30000
- }
- ]
}
]
}
Modified: qpid/java/trunk/perftests/example/config-one-test.json
URL: http://svn.apache.org/viewvc/qpid/java/trunk/perftests/example/config-one-test.json?rev=1702969&r1=1702968&r2=1702969&view=diff
==============================================================================
--- qpid/java/trunk/perftests/example/config-one-test.json (original)
+++ qpid/java/trunk/perftests/example/config-one-test.json Mon Sep 14 15:29:42 2015
@@ -20,8 +20,7 @@
"_producers": [
{
"_name": "Producer1",
- "_destinationName": "direct://amq.direct//testQueue",
- "_numberOfMessages": 5000
+ "_destinationName": "direct://amq.direct//testQueue"
}
]
}
@@ -41,8 +40,7 @@
"_consumers": [
{
"_name": "Consumer1",
- "_destinationName": "direct://amq.direct//testQueue",
- "_numberOfMessages": 5000
+ "_destinationName": "direct://amq.direct//testQueue"
}
]
}
Modified: qpid/java/trunk/perftests/example/config-stress-testing-manyp-c1-with-psfc.json
URL: http://svn.apache.org/viewvc/qpid/java/trunk/perftests/example/config-stress-testing-manyp-c1-with-psfc.json?rev=1702969&r1=1702968&r2=1702969&view=diff
==============================================================================
--- qpid/java/trunk/perftests/example/config-stress-testing-manyp-c1-with-psfc.json (original)
+++ qpid/java/trunk/perftests/example/config-stress-testing-manyp-c1-with-psfc.json Mon Sep 14 15:29:42 2015
@@ -27,7 +27,6 @@
{
"_name": "Producer1",
"_destinationName": "direct://amq.direct//testQueue",
- "_numberOfMessages": 1000000,
"_deliveryMode": 1,
"_messageSize": 10000
}
@@ -51,7 +50,6 @@
{
"_name": "Producer2",
"_destinationName": "direct://amq.direct//testQueue",
- "_numberOfMessages": 1000000,
"_deliveryMode": 1,
"_messageSize": 10000
}
@@ -75,7 +73,6 @@
{
"_name": "Producer3",
"_destinationName": "direct://amq.direct//testQueue",
- "_numberOfMessages": 1000000,
"_deliveryMode": 1,
"_messageSize": 10000
}
@@ -99,7 +96,6 @@
{
"_name": "Producer4",
"_destinationName": "direct://amq.direct//testQueue",
- "_numberOfMessages": 1000000,
"_deliveryMode": 1,
"_messageSize": 10000
}
@@ -123,7 +119,6 @@
{
"_name": "Producer5",
"_destinationName": "direct://amq.direct//testQueue",
- "_numberOfMessages": 1000000,
"_deliveryMode": 1,
"_messageSize": 10000
}
@@ -146,8 +141,7 @@
"_consumers": [
{
"_name": "Consumer1",
- "_destinationName": "direct://amq.direct//testQueue",
- "_numberOfMessages": 5000000
+ "_destinationName": "direct://amq.direct//testQueue"
}
]
}
Modified: qpid/java/trunk/perftests/example/config-stress-testing.json
URL: http://svn.apache.org/viewvc/qpid/java/trunk/perftests/example/config-stress-testing.json?rev=1702969&r1=1702968&r2=1702969&view=diff
==============================================================================
--- qpid/java/trunk/perftests/example/config-stress-testing.json (original)
+++ qpid/java/trunk/perftests/example/config-stress-testing.json Mon Sep 14 15:29:42 2015
@@ -48,7 +48,6 @@
{
"_name": "Producer1",
"_destinationName": "direct://amq.direct//testQueue",
- "_numberOfMessages": 1000000,
"_deliveryMode": 1
}
]
@@ -70,8 +69,7 @@
"_consumers": [
{
"_name": "Consumer1",
- "_destinationName": "direct://amq.direct//testQueue",
- "_numberOfMessages": 1000000
+ "_destinationName": "direct://amq.direct//testQueue"
}
]
}
Modified: qpid/java/trunk/perftests/example/config-two-tests.json
URL: http://svn.apache.org/viewvc/qpid/java/trunk/perftests/example/config-two-tests.json?rev=1702969&r1=1702968&r2=1702969&view=diff
==============================================================================
--- qpid/java/trunk/perftests/example/config-two-tests.json (original)
+++ qpid/java/trunk/perftests/example/config-two-tests.json Mon Sep 14 15:29:42 2015
@@ -20,8 +20,7 @@
"_producers": [
{
"_name": "Producer1",
- "_destinationName": "direct://amq.direct//testQueue",
- "_numberOfMessages": 5000
+ "_destinationName": "direct://amq.direct//testQueue"
}
]
}
@@ -41,8 +40,7 @@
"_consumers": [
{
"_name": "Consumer1",
- "_destinationName": "direct://amq.direct//testQueue",
- "_numberOfMessages": 5000
+ "_destinationName": "direct://amq.direct//testQueue"
}
]
}
@@ -72,8 +70,7 @@
"_producers": [
{
"_name": "Producer",
- "_destinationName": "direct://amq.direct//testQueue",
- "_numberOfMessages": 5000
+ "_destinationName": "direct://amq.direct//testQueue"
}
]
}
@@ -93,8 +90,7 @@
"_consumers": [
{
"_name": "Consumer1",
- "_destinationName": "direct://amq.direct//testQueue",
- "_numberOfMessages": 2500
+ "_destinationName": "direct://amq.direct//testQueue"
}
]
},
@@ -103,8 +99,7 @@
"_consumers": [
{
"_name": "Consumer2",
- "_destinationName": "direct://amq.direct//testQueue",
- "_numberOfMessages": 2500
+ "_destinationName": "direct://amq.direct//testQueue"
}
]
}
Modified: qpid/java/trunk/perftests/pom.xml
URL: http://svn.apache.org/viewvc/qpid/java/trunk/perftests/pom.xml?rev=1702969&r1=1702968&r2=1702969&view=diff
==============================================================================
--- qpid/java/trunk/perftests/pom.xml (original)
+++ qpid/java/trunk/perftests/pom.xml Mon Sep 14 15:29:42 2015
@@ -31,10 +31,23 @@
<properties>
<dependency-change-verification>true</dependency-change-verification>
- <qpid.disttest.duration>5000</qpid.disttest.duration>
+ <!-- Overrides for system properties understood by Qpid JMS 0-8..0-10 client -->
<qpid.amqp.version>0-9</qpid.amqp.version>
<qpid.dest_syntax>BURL</qpid.dest_syntax>
<qpid.close_timeout>90000</qpid.close_timeout>
+
+ <!-- Overrides for system properties understood by disttest itself -->
+ <qpid.disttest.loglevel>info</qpid.disttest.loglevel>
+ <qpid.disttest.duration>5000</qpid.disttest.duration>
+
+ <!-- The following provide values for the disttest command line -->
+ <qpid.disttest.test_config>${basedir}/etc/</qpid.disttest.test_config>
+ <qpid.disttest.hillclimb>false</qpid.disttest.hillclimb>
+ <qpid.disttest.hillclimb.max_runs>3</qpid.disttest.hillclimb.max_runs>
+ <qpid.disttest.hillclimb.start_target_rate>1025.0</qpid.disttest.hillclimb.start_target_rate>
+ <qpid.disttest.hillclimb.production_to_target_ratio_success_threshold>0.95</qpid.disttest.hillclimb.production_to_target_ratio_success_threshold>
+ <qpid.disttest.hillclimb.consumption_to_production_ratio_success_threshold>0.95</qpid.disttest.hillclimb.consumption_to_production_ratio_success_threshold>
+ <qpid.disttest.hillclimb.minimum_delta>1.0</qpid.disttest.hillclimb.minimum_delta>
</properties>
<dependencies>
@@ -162,13 +175,23 @@
<mainClass>org.apache.qpid.disttest.ControllerRunner</mainClass>
<arguments>
<argument>jndi-config=${basedir}/etc/perftests-jndi.properties</argument>
- <argument>test-config=${basedir}/etc/testdefs</argument>
+ <argument>test-config=${qpid.disttest.test_config}</argument>
<argument>distributed=false</argument>
<argument>writeToDb=false</argument>
+ <argument>hill-climb=${qpid.disttest.hillclimb}</argument>
+ <argument>hill-climber.max-runs=${qpid.disttest.hillclimb.max_runs}</argument>
+ <argument>hill-climber.start-target-rate=${qpid.disttest.hillclimb.start_target_rate}</argument>
+ <argument>hill-climber.production-to-target-ratio-success-threshold=${qpid.disttest.hillclimb.production_to_target_ratio_success_threshold}</argument>
+ <argument>hill-climber.consumption-to-production-ratio-success-threshold=${qpid.disttest.hillclimb.consumption_to_production_ratio_success_threshold}</argument>
+ <argument>hill-climber.minimum-delta=${qpid.disttest.hillclimb.minimum_delta}</argument>
</arguments>
<classpathScope>compile</classpathScope>
<systemProperties>
<systemProperty>
+ <key>qpid.disttest.loglevel</key>
+ <value>${qpid.disttest.loglevel}</value>
+ </systemProperty>
+ <systemProperty>
<key>qpid.disttest.duration</key>
<value>${qpid.disttest.duration}</value>
</systemProperty>
Modified: qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/ControllerRunner.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/ControllerRunner.java?rev=1702969&r1=1702968&r2=1702969&view=diff
==============================================================================
--- qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/ControllerRunner.java (original)
+++ qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/ControllerRunner.java Mon Sep 14 15:29:42 2015
@@ -47,11 +47,23 @@ public class ControllerRunner extends Ab
public static final String OUTPUT_DIR_PROP = "outputdir";
public static final String WRITE_TO_DB = "writeToDb";
public static final String RUN_ID = "runId";
+ public static final String HILL_CLIMB = "hill-climb";
+ public static final String HILL_CLIMBER_MAX_NUMBER_OF_RUNS = "hill-climber.max-runs";
+ public static final String HILL_CLIMBER_START_TARGET_RATE = "hill-climber.start-target-rate";
+ public static final String HILL_CLIMBER_PRODUCTION_TO_TARGET_RATIO_SUCCESS_THRESHOLD = "hill-climber.production-to-target-ratio-success-threshold";
+ public static final String HILL_CLIMBER_CONSUMPTION_TO_PRODUCTION_RATIO_SUCCESS_THRESHOLD = "hill-climber.consumption-to-production-ratio-success-threshold";
+ public static final String HILL_CLIMBER_MINIMUM_DELTA = "hill-climber.minimum-delta";
private static final String TEST_CONFIG_DEFAULT = "perftests-config.json";
private static final String DISTRIBUTED_DEFAULT = "false";
private static final String OUTPUT_DIR_DEFAULT = ".";
- public static final String WRITE_TO_DB_DEFAULT = "false";
+ private static final String WRITE_TO_DB_DEFAULT = "false";
+ private static final String HILL_CLIMB_DEFAULT = "true";
+ private static final String HILL_CLIMBER_MAX_NUMBER_OF_RUNS_DEFAULT = "3";
+ private static final String HILL_CLIMBER_START_TARGET_RATE_DEFAULT = "1025";
+ private static final String HILL_CLIMBER_CONSUMPTION_TO_PRODUCTION_RATIO_SUCCESS_THRESHOLD_DEFAULT = "0.95";
+ private static final String HILL_CLIMBER_PRODUCTION_TO_TARGET_RATIO_SUCCESS_THRESHOLD_DEFAULT = "0.95";
+ private static final String HILL_CLIMBER_MINIMUM_DELTA_DEFAULT = "1";
private final Aggregator _aggregator = new Aggregator();
@@ -68,6 +80,17 @@ public class ControllerRunner extends Ab
getCliOptions().put(OUTPUT_DIR_PROP, OUTPUT_DIR_DEFAULT);
getCliOptions().put(WRITE_TO_DB, WRITE_TO_DB_DEFAULT);
getCliOptions().put(RUN_ID, null);
+ getCliOptions().put(HILL_CLIMB, HILL_CLIMB_DEFAULT);
+ getCliOptions().put(HILL_CLIMBER_MAX_NUMBER_OF_RUNS,
+ HILL_CLIMBER_MAX_NUMBER_OF_RUNS_DEFAULT);
+ getCliOptions().put(HILL_CLIMBER_START_TARGET_RATE,
+ HILL_CLIMBER_START_TARGET_RATE_DEFAULT);
+ getCliOptions().put(HILL_CLIMBER_PRODUCTION_TO_TARGET_RATIO_SUCCESS_THRESHOLD,
+ HILL_CLIMBER_PRODUCTION_TO_TARGET_RATIO_SUCCESS_THRESHOLD_DEFAULT);
+ getCliOptions().put(HILL_CLIMBER_CONSUMPTION_TO_PRODUCTION_RATIO_SUCCESS_THRESHOLD,
+ HILL_CLIMBER_CONSUMPTION_TO_PRODUCTION_RATIO_SUCCESS_THRESHOLD_DEFAULT);
+ getCliOptions().put(HILL_CLIMBER_MINIMUM_DELTA,
+ HILL_CLIMBER_MINIMUM_DELTA_DEFAULT);
}
public static void main(String[] args) throws Exception
@@ -87,7 +110,18 @@ public class ControllerRunner extends Ab
try
{
- runTests(jmsDelegate);
+ Controller controller = new Controller(jmsDelegate,
+ DistributedTestConstants.REGISTRATION_TIMEOUT,
+ DistributedTestConstants.COMMAND_RESPONSE_TIMEOUT,
+ getCliOptions());
+ try
+ {
+ runTests(controller);
+ }
+ finally
+ {
+ controller.stopAllRegisteredClients();
+ }
}
finally
{
@@ -95,6 +129,7 @@ public class ControllerRunner extends Ab
}
}
+
private void setUpResultsDbWriter()
{
String writeToDbStr = getCliOptions().get(WRITE_TO_DB);
@@ -113,9 +148,8 @@ public class ControllerRunner extends Ab
_resultsFileWriter = new ResultsCsvWriter(outputDir);
}
- private void runTests(ControllerJmsDelegate jmsDelegate)
+ private void runTests(Controller controller)
{
- Controller controller = new Controller(jmsDelegate, DistributedTestConstants.REGISTRATION_TIMEOUT, DistributedTestConstants.COMMAND_RESPONSE_TIMEOUT);
String testConfigPath = getCliOptions().get(ControllerRunner.TEST_CONFIG_PROP);
List<String> testConfigFiles = _configFileHelper.getTestConfigFiles(testConfigPath);
@@ -123,7 +157,7 @@ public class ControllerRunner extends Ab
try
{
- List<ResultsForAllTests> results = new ArrayList<ResultsForAllTests>();
+ List<ResultsForAllTests> results = new ArrayList<>();
for (String testConfigFile : testConfigFiles)
{
@@ -143,10 +177,6 @@ public class ControllerRunner extends Ab
{
LOGGER.error("Problem running test", e);
}
- finally
- {
- controller.stopAllRegisteredClients();
- }
}
private ResultsForAllTests runTest(Controller controller, String testConfigFile)
Modified: qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/client/Client.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/client/Client.java?rev=1702969&r1=1702968&r2=1702969&view=diff
==============================================================================
--- qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/client/Client.java (original)
+++ qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/client/Client.java Mon Sep 14 15:29:42 2015
@@ -19,7 +19,12 @@
*/
package org.apache.qpid.disttest.client;
+import java.util.HashSet;
+import java.util.Set;
import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
@@ -45,6 +50,10 @@ public class Client
private Visitor _visitor;
private final AtomicReference<ClientState> _state;
private ParticipantExecutorRegistry _participantRegistry = new ParticipantExecutorRegistry();
+ private Set<Participant> _participants = new HashSet<>();
+
+ private final ExecutorService _executorService = Executors.newCachedThreadPool(new DaemonThreadFactory());
+
public Client(final ClientJmsDelegate delegate) throws NamingException
{
@@ -87,9 +96,11 @@ public class Client
_latch.countDown();
}
- public void addParticipantExecutor(final ParticipantExecutor participant)
+ public void addParticipant(final Participant participant)
{
- _participantRegistry.add(participant);
+ final ParticipantExecutor executor = new ParticipantExecutor(participant, _executorService);
+ _participantRegistry.add(executor);
+ _participants.add(participant);
}
public void waitUntilStopped()
@@ -110,12 +121,16 @@ public class Client
_latch.await(timeout, TimeUnit.MILLISECONDS);
}
}
- catch (final InterruptedException ie)
+ catch (InterruptedException ie)
{
Thread.currentThread().interrupt();
}
+ finally
+ {
+ _clientJmsDelegate.destroy();
+ _executorService.shutdown();
+ }
- _clientJmsDelegate.destroy();
}
public void processInstruction(final Command command)
@@ -203,8 +218,8 @@ public class Client
throw new DistributedTestException("Client '" + _clientJmsDelegate.getClientName() + "' is not in RUNNING_TEST state! Ignoring tearDownTest");
}
-
_participantRegistry.clear();
+ _participants.clear();
}
public void sendResults(ParticipantResult testResult)
@@ -226,4 +241,24 @@ public class Client
_participantRegistry = participantRegistry;
}
+ public void startDataCollection()
+ {
+ for (Participant participant : _participants)
+ {
+ participant.startDataCollection();
+ }
+ }
+
+ private static final class DaemonThreadFactory implements ThreadFactory
+ {
+ @Override
+ public Thread newThread(Runnable r)
+ {
+ Thread thread = new Thread(r);
+ thread.setDaemon(true);
+ return thread;
+ }
+ }
+
+
}
Modified: qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/client/ClientCommandVisitor.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/client/ClientCommandVisitor.java?rev=1702969&r1=1702968&r2=1702969&view=diff
==============================================================================
--- qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/client/ClientCommandVisitor.java (original)
+++ qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/client/ClientCommandVisitor.java Mon Sep 14 15:29:42 2015
@@ -21,15 +21,7 @@ package org.apache.qpid.disttest.client;
import org.apache.qpid.disttest.Visitor;
import org.apache.qpid.disttest.jms.ClientJmsDelegate;
-import org.apache.qpid.disttest.message.CreateConnectionCommand;
-import org.apache.qpid.disttest.message.CreateConsumerCommand;
-import org.apache.qpid.disttest.message.CreateMessageProviderCommand;
-import org.apache.qpid.disttest.message.CreateProducerCommand;
-import org.apache.qpid.disttest.message.CreateSessionCommand;
-import org.apache.qpid.disttest.message.NoOpCommand;
-import org.apache.qpid.disttest.message.StartTestCommand;
-import org.apache.qpid.disttest.message.StopClientCommand;
-import org.apache.qpid.disttest.message.TearDownTestCommand;
+import org.apache.qpid.disttest.message.*;
public class ClientCommandVisitor extends Visitor
{
@@ -69,16 +61,14 @@ public class ClientCommandVisitor extend
final ProducerParticipant participant = new ProducerParticipant(_clientJmsDelegate, command);
_clientJmsDelegate.createProducer(command);
- final ParticipantExecutor executor = new ParticipantExecutor(participant);
- _client.addParticipantExecutor(executor);
+ _client.addParticipant(participant);
}
public void visit(final CreateConsumerCommand command)
{
final ConsumerParticipant participant = new ConsumerParticipant(_clientJmsDelegate, command);
_clientJmsDelegate.createConsumer(command);
- final ParticipantExecutor executor = new ParticipantExecutor(participant);
- _client.addParticipantExecutor(executor);
+ _client.addParticipant(participant);
}
public void visit(final StartTestCommand command)
@@ -96,4 +86,9 @@ public class ClientCommandVisitor extend
_clientJmsDelegate.createMessageProvider(command);
}
+ public void visit(final StartDataCollectionCommand command)
+ {
+ _client.startDataCollection();
+ }
+
}
Modified: qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/client/ConsumerParticipant.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/client/ConsumerParticipant.java?rev=1702969&r1=1702968&r2=1702969&view=diff
==============================================================================
--- qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/client/ConsumerParticipant.java (original)
+++ qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/client/ConsumerParticipant.java Mon Sep 14 15:29:42 2015
@@ -26,6 +26,7 @@ import java.util.List;
import java.util.NavigableSet;
import java.util.concurrent.ConcurrentSkipListSet;
import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
@@ -51,9 +52,14 @@ public class ConsumerParticipant impleme
private final ClientJmsDelegate _jmsDelegate;
private final CreateConsumerCommand _command;
private final ParticipantResultFactory _resultFactory;
+ private final AtomicBoolean _collectingData = new AtomicBoolean();
+ private final boolean _batchEnabled;
+ private final long _expectedNumberOfMessages;
+ private final boolean _evaluateLatency;
+ private final int _batchSize;
+ private final long _maximumDuration;
- private long _startTime;
-
+ private volatile long _startTime;
private volatile Exception _asyncMessageListenerException;
private List<Long> _messageLatencies;
private final long _syncReceiveTimeout;
@@ -63,38 +69,39 @@ public class ConsumerParticipant impleme
_jmsDelegate = delegate;
_command = command;
_syncReceiveTimeout = _command.getReceiveTimeout() > 0 ? _command.getReceiveTimeout() : 50l;
+ _batchSize = _command.getBatchSize();
+ _batchEnabled = _batchSize > 0;
+
+
_resultFactory = new ParticipantResultFactory();
if (command.isEvaluateLatency())
{
- _messageLatencies = new ArrayList<Long>();
+ _messageLatencies = new ArrayList<>();
}
+ _expectedNumberOfMessages = _command.getNumberOfMessages();
+ _evaluateLatency = _command.isEvaluateLatency();
+ _maximumDuration = _command.getMaximumDuration();
}
@Override
public ParticipantResult doIt(String registeredClientName) throws Exception
{
- final Date start = new Date();
final int acknowledgeMode = _jmsDelegate.getAcknowledgeMode(_command.getSessionName());
- if (_command.getMaximumDuration() == 0 && _command.getNumberOfMessages() == 0)
- {
- throw new DistributedTestException("number of messages and duration cannot both be zero");
- }
-
if (_command.isSynchronous())
{
synchronousRun();
}
else
{
- LOGGER.info("Consumer {} registering listener", getName());
+ LOGGER.debug("Consumer {} registering listener", getName());
_jmsDelegate.registerListener(_command.getParticipantName(), new MessageListener(){
@Override
public void onMessage(Message message)
{
- processAsynchMessage(message);
+ processAsyncMessage(message);
}
});
@@ -104,15 +111,13 @@ public class ConsumerParticipant impleme
}
Date end = new Date();
+ final Date start = new Date(_startTime);
int numberOfMessagesReceived = _totalNumberOfMessagesReceived.get();
long totalPayloadSize = _totalPayloadSizeOfAllMessagesReceived.get();
int payloadSize = getPayloadSizeForResultIfConstantOrZeroOtherwise(_allConsumedPayloadSizes);
- if (LOGGER.isInfoEnabled())
- {
- LOGGER.info("Consumer {} finished consuming. Number of messages consumed: {}",
- getName(), numberOfMessagesReceived);
- }
+ LOGGER.info("Consumer {} finished consuming. Number of messages consumed: {}",
+ getName(), numberOfMessagesReceived);
ParticipantResult result = _resultFactory.createForConsumer(
getName(),
@@ -127,17 +132,35 @@ public class ConsumerParticipant impleme
return result;
}
- private void synchronousRun()
+ @Override
+ public void startDataCollection()
{
- LOGGER.info("Consumer {} about to consume messages", getName());
+ _collectingData.set(true);
+ }
- _startTime = System.currentTimeMillis();
+ private void synchronousRun()
+ {
+ LOGGER.debug("Consumer {} about to consume messages", getName());
- Message message;
+ boolean keepLooping = true;
do
{
- message = _jmsDelegate.consumeMessage(_command.getParticipantName(), _syncReceiveTimeout);
- } while (processMessage(message));
+ Message message = _jmsDelegate.consumeMessage(_command.getParticipantName(), _syncReceiveTimeout);
+ if (message != null)
+ {
+ keepLooping = processMessage(message);
+ }
+ else
+ {
+ boolean reachedMaximumDuration = _startTime > 0 && _maximumDuration > 0 && System.currentTimeMillis() - _startTime >= _maximumDuration;
+ if (reachedMaximumDuration)
+ {
+ LOGGER.debug("Consumer participant done due to maximumDuration reached.");
+ keepLooping = false;
+ }
+ }
+ }
+ while (keepLooping);
}
/**
@@ -145,62 +168,58 @@ public class ConsumerParticipant impleme
*/
private boolean processMessage(Message message)
{
- int messageCount = message == null? _totalNumberOfMessagesReceived.get() : _totalNumberOfMessagesReceived.incrementAndGet() ;
- boolean batchEnabled = _command.getBatchSize() > 0;
- boolean batchComplete = batchEnabled && messageCount % _command.getBatchSize() == 0;
- if (message != null)
+ if (!_collectingData.get())
{
- if (LOGGER.isTraceEnabled())
+ // If we are performing a run with fixed number of messages we may receive a message before the startDataCollection command.
+ if (_expectedNumberOfMessages > 0)
{
- LOGGER.trace("Message {} received by {}", message, this);
+ _totalNumberOfMessagesReceived.incrementAndGet();
}
- int messagePayloadSize = _jmsDelegate.calculatePayloadSizeFrom(message);
- _allConsumedPayloadSizes.add(messagePayloadSize);
- _totalPayloadSizeOfAllMessagesReceived.addAndGet(messagePayloadSize);
+ _jmsDelegate.commitOrAcknowledgeMessageIfNecessary(_command.getSessionName(), message);
+ return true;
+ }
- if (_command.isEvaluateLatency())
- {
- long mesageTimestamp;
- try
- {
- mesageTimestamp = message.getJMSTimestamp();
- }
- catch (JMSException e)
- {
- throw new DistributedTestException("Cannot get message timestamp!", e);
- }
- long latency = System.currentTimeMillis() - mesageTimestamp;
- _messageLatencies.add(latency);
- }
+ if (_startTime == 0)
+ {
+ _startTime = System.currentTimeMillis();
+ }
+
+ LOGGER.trace("Message {} received by {}", message, this);
+
+ final int messageCount = _totalNumberOfMessagesReceived.incrementAndGet() ;
+
+ int messagePayloadSize = _jmsDelegate.calculatePayloadSizeFrom(message);
+ _allConsumedPayloadSizes.add(messagePayloadSize);
+ _totalPayloadSizeOfAllMessagesReceived.addAndGet(messagePayloadSize);
+
+ if (_evaluateLatency)
+ {
+ long messageTimestamp = getMessageTimestamp(message);
+ long latency = System.currentTimeMillis() - messageTimestamp;
+ _messageLatencies.add(latency);
+ }
- if (!batchEnabled || batchComplete)
+ boolean batchComplete = (_batchEnabled && (messageCount % _batchSize == 0));
+ if (!_batchEnabled || batchComplete)
+ {
+ if (_batchEnabled)
{
- if (LOGGER.isTraceEnabled() && batchEnabled)
- {
- LOGGER.trace("Committing: batch size {} ", _command.getBatchSize() );
- }
- _jmsDelegate.commitOrAcknowledgeMessageIfNecessary(_command.getSessionName(), message);
+ LOGGER.trace("Committing: batch size {} ", _batchSize);
}
+ _jmsDelegate.commitOrAcknowledgeMessageIfNecessary(_command.getSessionName(), message);
}
- boolean reachedExpectedNumberOfMessages = _command.getNumberOfMessages() > 0 && messageCount >= _command.getNumberOfMessages();
- boolean reachedMaximumDuration = _command.getMaximumDuration() > 0 && System.currentTimeMillis() - _startTime >= _command.getMaximumDuration();
- boolean finishedConsuming = reachedExpectedNumberOfMessages || reachedMaximumDuration;
+ boolean reachedMaximumDuration = _maximumDuration > 0 && System.currentTimeMillis() - _startTime >= _maximumDuration;
+ boolean receivedAllMessages = _expectedNumberOfMessages > 0 && messageCount >= _expectedNumberOfMessages;
- if (finishedConsuming)
+ if (reachedMaximumDuration || receivedAllMessages)
{
- if (LOGGER.isDebugEnabled())
- {
- LOGGER.debug("Message {} reachedMaximumDuration {}", messageCount, reachedMaximumDuration);
- }
+ LOGGER.debug("Message {} reached duration : {} : received all messages : {}",
+ messageCount, reachedMaximumDuration, reachedMaximumDuration);
- if (batchEnabled && !batchComplete)
+ if (_batchEnabled)
{
- if (LOGGER.isTraceEnabled())
- {
- LOGGER.trace("Committing: batch size " + _command.getBatchSize() );
- }
-
+ LOGGER.trace("Committing: final batch");
// commit/acknowledge remaining messages if necessary
_jmsDelegate.commitOrAcknowledgeMessageIfNecessary(_command.getSessionName(), message);
}
@@ -210,27 +229,33 @@ public class ConsumerParticipant impleme
return true;
}
+ private long getMessageTimestamp(final Message message)
+ {
+ try
+ {
+ return message.getJMSTimestamp();
+ }
+ catch (JMSException e)
+ {
+ throw new DistributedTestException("Cannot get message timestamp!", e);
+ }
+ }
+
/**
* Intended to be called from a {@link MessageListener}. Updates {@link #_asyncRunHasFinished} if
* no more messages should be processed, causing {@link #doIt(String)} to exit.
*/
- public void processAsynchMessage(Message message)
+ public void processAsyncMessage(Message message)
{
boolean continueRunning = true;
try
{
- if (_startTime == 0)
- {
- // reset counter and start time on receiving of first message
- _startTime = System.currentTimeMillis();
- }
-
continueRunning = processMessage(message);
}
catch (Exception e)
{
- LOGGER.error("Error occured consuming message " + _totalNumberOfMessagesReceived, e);
+ LOGGER.error("Error occurred consuming message " + _totalNumberOfMessagesReceived, e);
continueRunning = false;
_asyncMessageListenerException = e;
}
Modified: qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/client/Participant.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/client/Participant.java?rev=1702969&r1=1702968&r2=1702969&view=diff
==============================================================================
--- qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/client/Participant.java (original)
+++ qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/client/Participant.java Mon Sep 14 15:29:42 2015
@@ -28,4 +28,5 @@ public interface Participant
String getName();
+ void startDataCollection();
}
Modified: qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/client/ParticipantExecutor.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/client/ParticipantExecutor.java?rev=1702969&r1=1702968&r2=1702969&view=diff
==============================================================================
--- qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/client/ParticipantExecutor.java (original)
+++ qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/client/ParticipantExecutor.java Mon Sep 14 15:29:42 2015
@@ -19,9 +19,6 @@
package org.apache.qpid.disttest.client;
import java.util.concurrent.Executor;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ThreadFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -32,20 +29,17 @@ public class ParticipantExecutor
{
private static final Logger LOGGER = LoggerFactory.getLogger(ParticipantExecutor.class);
- private static final ExecutorService SHARED_UNBOUNDED_THREAD_POOL = Executors.newCachedThreadPool(new DaemonThreadFactory());
-
- private Executor _executor = SHARED_UNBOUNDED_THREAD_POOL;
-
- private Client _client;
-
+ private final Executor _executor;
private final Participant _participant;
-
private final ParticipantResultFactory _factory;
- public ParticipantExecutor(Participant participant)
+ private Client _client;
+
+ public ParticipantExecutor(Participant participant, Executor executor)
{
_participant = participant;
_factory = new ParticipantResultFactory();
+ _executor = executor;
}
/**
@@ -55,7 +49,7 @@ public class ParticipantExecutor
{
_client = client;
- LOGGER.debug("Starting test participant in background thread: " + this);
+ LOGGER.debug("Starting test participant in background thread: {} ", this);
_executor.execute(new ParticipantRunnable());
}
@@ -64,11 +58,6 @@ public class ParticipantExecutor
return _participant.getName();
}
- void setExecutor(Executor executor)
- {
- _executor = executor;
- }
-
private class ParticipantRunnable implements Runnable
{
@Override
@@ -121,17 +110,6 @@ public class ParticipantExecutor
}
}
- private static final class DaemonThreadFactory implements ThreadFactory
- {
- @Override
- public Thread newThread(Runnable r)
- {
- Thread thread = new Thread(r);
- thread.setDaemon(true);
- return thread;
- }
- }
-
@Override
public String toString()
{
Modified: qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/client/ParticipantResultFactory.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/client/ParticipantResultFactory.java?rev=1702969&r1=1702968&r2=1702969&view=diff
==============================================================================
--- qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/client/ParticipantResultFactory.java (original)
+++ qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/client/ParticipantResultFactory.java Mon Sep 14 15:29:42 2015
@@ -23,7 +23,7 @@ import java.util.Date;
import org.apache.qpid.disttest.message.ConsumerParticipantResult;
import org.apache.qpid.disttest.message.CreateConsumerCommand;
-import org.apache.qpid.disttest.message.CreateParticpantCommand;
+import org.apache.qpid.disttest.message.CreateParticipantCommand;
import org.apache.qpid.disttest.message.CreateProducerCommand;
import org.apache.qpid.disttest.message.ParticipantResult;
import org.apache.qpid.disttest.message.ProducerParticipantResult;
@@ -70,7 +70,6 @@ public class ParticipantResultFactory
{
final ProducerParticipantResult producerParticipantResult = new ProducerParticipantResult();
- producerParticipantResult.setStartDelay(command.getStartDelay());
producerParticipantResult.setDeliveryMode(command.getDeliveryMode());
producerParticipantResult.setPriority(command.getPriority());
producerParticipantResult.setInterval(command.getInterval());
@@ -95,7 +94,7 @@ public class ParticipantResultFactory
participantResult.setEndDate(end);
}
- private void setTestProperties(final ParticipantResult participantResult, CreateParticpantCommand command, String participantName, String clientRegisteredName, int acknowledgeMode)
+ private void setTestProperties(final ParticipantResult participantResult, CreateParticipantCommand command, String participantName, String clientRegisteredName, int acknowledgeMode)
{
participantResult.setParticipantName(participantName);
participantResult.setRegisteredClientName(clientRegisteredName);
Modified: qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/client/ProducerParticipant.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/client/ProducerParticipant.java?rev=1702969&r1=1702968&r2=1702969&view=diff
==============================================================================
--- qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/client/ProducerParticipant.java (original)
+++ qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/client/ProducerParticipant.java Mon Sep 14 15:29:42 2015
@@ -22,14 +22,12 @@ package org.apache.qpid.disttest.client;
import java.util.Date;
import java.util.NavigableSet;
import java.util.TreeSet;
-import java.util.concurrent.Callable;
-import java.util.concurrent.CancellationException;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
import javax.jms.Message;
-import org.apache.qpid.disttest.DistributedTestException;
-import org.apache.qpid.disttest.client.utils.ExecutorWithLimits;
-import org.apache.qpid.disttest.client.utils.ExecutorWithLimitsFactory;
+import com.google.common.util.concurrent.RateLimiter;
import org.apache.qpid.disttest.jms.ClientJmsDelegate;
import org.apache.qpid.disttest.message.CreateProducerCommand;
import org.apache.qpid.disttest.message.ParticipantResult;
@@ -46,146 +44,132 @@ public class ProducerParticipant impleme
private final ParticipantResultFactory _resultFactory;
- private ExecutorWithLimits _limiter;
+ private final CountDownLatch _collectingDataLatch = new CountDownLatch(1);
+ private final long _maximumDuration;
+ private final long _numberOfMessages;
+ private final int _batchSize;
+ private final int _acknowledgeMode;
+ private final RateLimiter _rateLimiter;
public ProducerParticipant(final ClientJmsDelegate jmsDelegate, final CreateProducerCommand command)
{
_jmsDelegate = jmsDelegate;
_command = command;
_resultFactory = new ParticipantResultFactory();
+ _maximumDuration = _command.getMaximumDuration();
+ _numberOfMessages = _command.getNumberOfMessages();
+ _batchSize = _command.getBatchSize();
+ _acknowledgeMode = _jmsDelegate.getAcknowledgeMode(_command.getSessionName());
+ final double rate = _command.getRate();
+ _rateLimiter = (rate > 0 ? RateLimiter.create(rate) : null);
}
@Override
public ParticipantResult doIt(String registeredClientName) throws Exception
{
- long numberOfMessages = _command.getNumberOfMessages();
- long maximumDuration = _command.getMaximumDuration();
-
- if (maximumDuration == 0 && numberOfMessages == 0)
- {
- throw new DistributedTestException("number of messages and duration cannot both be zero");
- }
-
- long duration = maximumDuration - _command.getStartDelay();
- if (maximumDuration > 0 && duration <= 0)
- {
- throw new DistributedTestException("Start delay must be less than maximum test duration");
- }
- final long requiredDuration = duration > 0 ? duration : 0;
-
- doSleepForStartDelay();
-
- final int batchSize = _command.getBatchSize();
- final int acknowledgeMode = _jmsDelegate.getAcknowledgeMode(_command.getSessionName());
- final long startTime = System.currentTimeMillis();
-
+ long startTime = 0;
Message lastPublishedMessage = null;
int numberOfMessagesSent = 0;
long totalPayloadSizeOfAllMessagesSent = 0;
- NavigableSet<Integer> allProducedPayloadSizes = new TreeSet<Integer>();
+ NavigableSet<Integer> allProducedPayloadSizes = new TreeSet<>();
- _limiter = ExecutorWithLimitsFactory.createExecutorWithLimit(startTime, requiredDuration);
-
- if (LOGGER.isInfoEnabled())
- {
- LOGGER.info("Producer {} about to send messages. Duration limit: {} ms, Message limit: {}",
- getName(), requiredDuration, numberOfMessages);
- }
+ LOGGER.debug("Producer {} about to send messages. Duration limit: {} ms Message Limit : {}",
+ getName(), _maximumDuration, _numberOfMessages);
while (true)
{
- if (numberOfMessages > 0 && numberOfMessagesSent >= numberOfMessages
- || requiredDuration > 0 && System.currentTimeMillis() - startTime >= requiredDuration)
+ if (_rateLimiter != null)
{
- break;
+ _rateLimiter.acquire();
}
- try
+ if (_collectingDataLatch.getCount() == 0)
{
- lastPublishedMessage = _limiter.execute(new Callable<Message>()
+ if (startTime == 0)
{
- @Override
- public Message call() throws Exception
- {
- return _jmsDelegate.sendNextMessage(_command);
- }
- });
- }
- catch (CancellationException ce)
- {
- LOGGER.debug("Producer send was cancelled due to maximum duration {} ms", requiredDuration);
- break;
- }
+ startTime = System.currentTimeMillis();
+ }
- numberOfMessagesSent++;
+ if ((_maximumDuration > 0 && System.currentTimeMillis() - startTime >= _maximumDuration) ||
+ (_numberOfMessages > 0 && numberOfMessagesSent >= _numberOfMessages))
+ {
+ break;
+ }
- int lastPayloadSize = _jmsDelegate.calculatePayloadSizeFrom(lastPublishedMessage);
- totalPayloadSizeOfAllMessagesSent += lastPayloadSize;
- allProducedPayloadSizes.add(lastPayloadSize);
+ lastPublishedMessage = _jmsDelegate.sendNextMessage(_command);
- if (LOGGER.isTraceEnabled())
- {
- LOGGER.trace("message " + numberOfMessagesSent + " sent by " + this);
- }
+ numberOfMessagesSent++;
- final boolean batchLimitReached = batchSize <= 0
- || numberOfMessagesSent % batchSize == 0;
+ int lastPayloadSize = _jmsDelegate.calculatePayloadSizeFrom(lastPublishedMessage);
+ totalPayloadSizeOfAllMessagesSent += lastPayloadSize;
+ allProducedPayloadSizes.add(lastPayloadSize);
- if (batchLimitReached)
- {
- if (LOGGER.isTraceEnabled() && batchSize > 0)
+ LOGGER.trace("message {} sent by {}", numberOfMessagesSent, this);
+
+ final boolean batchLimitReached = (_batchSize <= 0 || numberOfMessagesSent % _batchSize == 0);
+
+ if (batchLimitReached)
{
- LOGGER.trace("Committing: batch size " + batchSize );
+ if (LOGGER.isTraceEnabled() && _batchSize > 0)
+ {
+ LOGGER.trace("Committing: batch size " + _batchSize);
+ }
+ _jmsDelegate.commitIfNecessary(_command.getSessionName());
+
+ doSleepForInterval();
}
- _jmsDelegate.commitIfNecessary(_command.getSessionName());
+ }
+ else
+ {
+ if (_maximumDuration > 0)
+ {
+ _jmsDelegate.sendNextMessage(_command);
- doSleepForInterval();
+ _jmsDelegate.commitIfNecessary(_command.getSessionName());
+ LOGGER.trace("Pre-message sent by {}", this);
+ }
+ if (_rateLimiter == null && _maximumDuration == 0)
+ {
+ if (!_collectingDataLatch.await(1, TimeUnit.SECONDS))
+ {
+ LOGGER.debug("Producer {} still waiting for collectingData command from coordinator", getName());
+ }
+ }
}
}
// commit the remaining batch messages
- if (batchSize > 0 && numberOfMessagesSent % batchSize != 0)
+ if (_batchSize > 0 && numberOfMessagesSent % _batchSize != 0)
{
- if (LOGGER.isTraceEnabled())
- {
- LOGGER.trace("Committing: batch size " + batchSize );
- }
+ LOGGER.trace("Committing: batch size {}", _batchSize);
_jmsDelegate.commitIfNecessary(_command.getSessionName());
}
+ Date end = new Date();
- if (LOGGER.isInfoEnabled())
- {
- LOGGER.info("Producer {} finished publishing. Number of messages published: {}",
- getName(), numberOfMessagesSent);
- }
+ LOGGER.info("Producer {} finished publishing. Number of messages published: {}",
+ getName(), numberOfMessagesSent);
Date start = new Date(startTime);
- Date end = new Date();
int payloadSize = getPayloadSizeForResultIfConstantOrZeroOtherwise(allProducedPayloadSizes);
return _resultFactory.createForProducer(
getName(),
registeredClientName,
_command,
- acknowledgeMode,
+ _acknowledgeMode,
numberOfMessagesSent,
payloadSize, totalPayloadSizeOfAllMessagesSent, start, end);
}
- private int getPayloadSizeForResultIfConstantOrZeroOtherwise(NavigableSet<Integer> allPayloadSizes)
+ @Override
+ public void startDataCollection()
{
- return allPayloadSizes.size() == 1 ? allPayloadSizes.first() : 0;
+ _collectingDataLatch.countDown();
}
- private void doSleepForStartDelay()
+ private int getPayloadSizeForResultIfConstantOrZeroOtherwise(NavigableSet<Integer> allPayloadSizes)
{
- long sleepTime = _command.getStartDelay();
- if (sleepTime > 0)
- {
- LOGGER.debug("{} sleeping for {} milliseconds before starting", getName(), sleepTime);
- // start delay is specified. Sleeping...
- doSleep(sleepTime);
- }
+ return allPayloadSizes.size() == 1 ? allPayloadSizes.first() : 0;
}
private void doSleepForInterval() throws InterruptedException
@@ -212,10 +196,6 @@ public class ProducerParticipant impleme
@Override
public void releaseResources()
{
- if (_limiter != null)
- {
- _limiter.shutdown();
- }
_jmsDelegate.closeTestProducer(_command.getParticipantName());
}
Copied: qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/controller/AbstractTestRunner.java (from r1702832, qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/controller/TestRunner.java)
URL: http://svn.apache.org/viewvc/qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/controller/AbstractTestRunner.java?p2=qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/controller/AbstractTestRunner.java&p1=qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/controller/TestRunner.java&r1=1702832&r2=1702969&rev=1702969&view=diff
==============================================================================
--- qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/controller/TestRunner.java (original)
+++ qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/controller/AbstractTestRunner.java Mon Sep 14 15:29:42 2015
@@ -1,4 +1,5 @@
/*
+ *
* 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
@@ -15,7 +16,9 @@
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
+ *
*/
+
package org.apache.qpid.disttest.controller;
import java.util.Collection;
@@ -37,33 +40,24 @@ import org.apache.qpid.disttest.message.
import org.apache.qpid.disttest.message.CommandType;
import org.apache.qpid.disttest.message.ParticipantResult;
import org.apache.qpid.disttest.message.Response;
+import org.apache.qpid.disttest.message.StartDataCollectionCommand;
import org.apache.qpid.disttest.message.StartTestCommand;
import org.apache.qpid.disttest.message.TearDownTestCommand;
-public class TestRunner
+public abstract class AbstractTestRunner implements ITestRunner
{
- private static final Logger LOGGER = LoggerFactory.getLogger(TestRunner.class);
-
- private static final long PARTICIPANT_RESULTS_LOG_INTERVAL = 60000;
+ private static final Logger LOGGER = LoggerFactory.getLogger(AbstractTestRunner.class);
public static final long WAIT_FOREVER = -1;
-
- private final long _commandResponseTimeout;
-
- private final Set<CommandType> _setOfResponsesToExpect = Collections.synchronizedSet(new HashSet<CommandType>());
-
- private final ParticipatingClients _participatingClients;
-
- private final TestInstance _testInstance;
- private ControllerJmsDelegate _jmsDelegate;
-
- private volatile CountDownLatch _commandResponseLatch = null;
- private final CountDownLatch _testResultsLatch;
- private final TestResult _testResult;
-
+ private static final long PARTICIPANT_RESULTS_LOG_INTERVAL = 60000;
+ protected final long _commandResponseTimeout;
+ protected final ParticipatingClients _participatingClients;
+ protected final TestInstance _testInstance;
+ protected CountDownLatch _testResultsLatch;
/** Length of time to await test results or {@value #WAIT_FOREVER} */
- private final long _testResultTimeout;
-
- private Thread _removeQueuesShutdownHook = new Thread()
+ protected final long _testResultTimeout;
+ private final Set<CommandType> _setOfResponsesToExpect = Collections.synchronizedSet(new HashSet<CommandType>());
+ protected ControllerJmsDelegate _jmsDelegate;
+ protected Thread _removeQueuesShutdownHook = new Thread()
{
@Override
public void run()
@@ -79,74 +73,19 @@ public class TestRunner
}
}
};
+ private volatile CountDownLatch _commandResponseLatch = null;
- public TestRunner(ParticipatingClients participatingClients, TestInstance testInstance, ControllerJmsDelegate jmsDelegate, long commandResponseTimeout, long testResultTimeout)
+ public AbstractTestRunner(ParticipatingClients participatingClients,
+ long commandResponseTimeout,
+ TestInstance testInstance,
+ ControllerJmsDelegate jmsDelegate,
+ long testResultTimeout)
{
_participatingClients = participatingClients;
+ _commandResponseTimeout = commandResponseTimeout;
_testInstance = testInstance;
_jmsDelegate = jmsDelegate;
- _commandResponseTimeout = commandResponseTimeout;
- _testResultsLatch = new CountDownLatch(testInstance.getTotalNumberOfParticipants());
_testResultTimeout = testResultTimeout;
- _testResult = new TestResult(testInstance.getName());
- }
-
- public TestResult run()
- {
- final ParticipantResultListener participantResultListener = new ParticipantResultListener();
- TestCommandResponseListener testCommandResponseListener = new TestCommandResponseListener();
-
- try
- {
- _jmsDelegate.addCommandListener(testCommandResponseListener);
- _jmsDelegate.addCommandListener(participantResultListener);
-
- runParts();
-
- return _testResult;
- }
- catch(RuntimeException e)
- {
- LOGGER.error("Couldn't run test", e);
- throw e;
- }
- finally
- {
- _jmsDelegate.removeCommandListener(participantResultListener);
- _jmsDelegate.removeCommandListener(testCommandResponseListener);
- }
- }
-
- private void runParts()
- {
- boolean queuesCreated = false;
-
- try
- {
- createQueues();
- queuesCreated = true;
- Runtime.getRuntime().addShutdownHook(_removeQueuesShutdownHook);
-
- sendTestSetupCommands();
- awaitCommandResponses();
- sendCommandToParticipatingClients(new StartTestCommand());
- awaitCommandResponses();
-
- awaitTestResults();
-
- sendCommandToParticipatingClients(new TearDownTestCommand());
- awaitCommandResponses();
- }
- finally
- {
-
- if (queuesCreated)
- {
- deleteQueues();
- }
-
- Runtime.getRuntime().removeShutdownHook(_removeQueuesShutdownHook);
- }
}
void createQueues()
@@ -161,6 +100,7 @@ public class TestRunner
void sendTestSetupCommands()
{
List<CommandForClient> commandsForAllClients = _testInstance.createCommands();
+ validateCommands(commandsForAllClients);
final int numberOfCommandsToSend = commandsForAllClients.size();
_commandResponseLatch = new CountDownLatch(numberOfCommandsToSend);
@@ -179,12 +119,16 @@ public class TestRunner
}
}
+ protected void validateCommands(List<CommandForClient> commandsForAllClients)
+ {
+ // pass
+ }
+
void awaitCommandResponses()
{
awaitLatch(_commandResponseLatch, _commandResponseTimeout, "Timed out waiting for command responses");
}
-
void processCommandResponse(final Response response)
{
if (LOGGER.isDebugEnabled())
@@ -196,7 +140,6 @@ public class TestRunner
checkForResponseError(response);
}
-
void awaitTestResults()
{
long timeout = _testResultTimeout;
@@ -253,11 +196,11 @@ public class TestRunner
}
}
- public void processParticipantResult(ParticipantResult result)
+ void processParticipantResult(ParticipantResult result, TestResult testResult)
{
setOriginalTestDetailsOn(result);
- _testResult.addParticipantResult(result);
+ testResult.addParticipantResult(result);
LOGGER.debug("Received result " + result);
_testResultsLatch.countDown();
@@ -308,8 +251,79 @@ public class TestRunner
}
}
+
+ protected TestResult doIt()
+ {
+ TestResult testResult = new TestResult(_testInstance.getName());
+ _testResultsLatch = new CountDownLatch(_testInstance.getTotalNumberOfParticipants());
+
+ ParticipantResultListener participantResultListener = new ParticipantResultListener(testResult);
+ TestCommandResponseListener testCommandResponseListener = new TestCommandResponseListener();
+
+ try
+ {
+ _jmsDelegate.addCommandListener(testCommandResponseListener);
+ _jmsDelegate.addCommandListener(participantResultListener);
+
+ runParts();
+
+ return testResult;
+ }
+ catch(RuntimeException e)
+ {
+ LOGGER.error("Couldn't run test", e);
+ throw e;
+ }
+ finally
+ {
+ _jmsDelegate.removeCommandListener(participantResultListener);
+ _jmsDelegate.removeCommandListener(testCommandResponseListener);
+ }
+ }
+
+ protected void runParts()
+ {
+ boolean queuesCreated = false;
+
+ try
+ {
+ createQueues();
+ queuesCreated = true;
+ Runtime.getRuntime().addShutdownHook(_removeQueuesShutdownHook);
+
+ sendTestSetupCommands();
+ awaitCommandResponses();
+ sendCommandToParticipatingClients(new StartTestCommand());
+ awaitCommandResponses();
+ sendCommandToParticipatingClients(new StartDataCollectionCommand());
+ awaitCommandResponses();
+
+ awaitTestResults();
+
+ sendCommandToParticipatingClients(new TearDownTestCommand());
+ awaitCommandResponses();
+ }
+ finally
+ {
+
+ if (queuesCreated)
+ {
+ deleteQueues();
+ }
+
+ Runtime.getRuntime().removeShutdownHook(_removeQueuesShutdownHook);
+ }
+ }
+
final class ParticipantResultListener implements CommandListener
{
+ private final TestResult _testResult;
+
+ public ParticipantResultListener(TestResult testResult)
+ {
+ _testResult = testResult;
+ }
+
@Override
public boolean supports(Command command)
{
@@ -319,7 +333,7 @@ public class TestRunner
@Override
public void processCommand(Command command)
{
- processParticipantResult((ParticipantResult) command);
+ processParticipantResult((ParticipantResult) command, _testResult);
}
}
@@ -344,5 +358,4 @@ public class TestRunner
return false;
}
}
-
}
Modified: qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/controller/Controller.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/controller/Controller.java?rev=1702969&r1=1702968&r2=1702969&view=diff
==============================================================================
--- qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/controller/Controller.java (original)
+++ qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/controller/Controller.java Mon Sep 14 15:29:42 2015
@@ -20,6 +20,7 @@
package org.apache.qpid.disttest.controller;
import java.util.Collection;
+import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
@@ -27,11 +28,9 @@ import org.apache.qpid.disttest.Distribu
import org.apache.qpid.disttest.controller.config.Config;
import org.apache.qpid.disttest.controller.config.TestInstance;
import org.apache.qpid.disttest.jms.ControllerJmsDelegate;
-import org.apache.qpid.disttest.message.Command;
-import org.apache.qpid.disttest.message.CommandType;
-import org.apache.qpid.disttest.message.RegisterClientCommand;
-import org.apache.qpid.disttest.message.Response;
-import org.apache.qpid.disttest.message.StopClientCommand;
+import org.apache.qpid.disttest.message.*;
+import org.apache.qpid.disttest.results.aggregation.ITestResult;
+
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -44,21 +43,23 @@ public class Controller
private final ControllerJmsDelegate _jmsDelegate;
- private volatile CountDownLatch _stopClientsResponseLatch = null;
+ private final TestRunnerFactory _testRunnerFactory;
+ private final Map<String, String> _options;
+ private volatile CountDownLatch _stopClientsResponseLatch = null;
private Config _config;
- private TestRunnerFactory _testRunnerFactory;
private ClientRegistry _clientRegistry;
- private long _testResultTimeout = TestRunner.WAIT_FOREVER;
- public Controller(final ControllerJmsDelegate jmsDelegate, long registrationTimeout, long commandResponseTimeout)
+ public Controller(final ControllerJmsDelegate jmsDelegate, long registrationTimeout,
+ long commandResponseTimeout, final Map<String, String> options)
{
_jmsDelegate = jmsDelegate;
_registrationTimeout = registrationTimeout;
_commandResponseTimeout = commandResponseTimeout;
_testRunnerFactory = new TestRunnerFactory();
_clientRegistry = new ClientRegistry();
+ _options = options;
_jmsDelegate.addCommandListener(new RegisterClientCommandListener());
_jmsDelegate.addCommandListener(new StopClientResponseListener());
@@ -158,20 +159,27 @@ public class Controller
for (TestInstance testInstance : _config.getTests())
{
-
ParticipatingClients participatingClients = new ParticipatingClients(_clientRegistry, testInstance.getClientNames());
- LOGGER.info("Running test " + testInstance + ". Participating clients: " + participatingClients.getRegisteredNames());
- TestRunner runner = _testRunnerFactory.createTestRunner(participatingClients,
- testInstance,
- _jmsDelegate,
- _commandResponseTimeout,
- _testResultTimeout);
-
- TestResult testResult = runner.run();
- LOGGER.info("Finished test " + testInstance);
+ ITestRunner runner = _testRunnerFactory.createTestRunner(participatingClients,
+ testInstance,
+ _jmsDelegate,
+ _commandResponseTimeout,
+ AbstractTestRunner.WAIT_FOREVER,
+ _options);
+ LOGGER.info("Running test {}. Participating clients: {}",
+ testInstance, participatingClients.getRegisteredNames());
- resultsForAllTests.add(testResult);
+ ITestResult result = runner.run();
+ if (result != null)
+ {
+ LOGGER.info("Finished test {}, result {}", testInstance, result);
+ resultsForAllTests.add(result);
+ }
+ else
+ {
+ LOGGER.warn("Finished test {} without producing a result.", testInstance);
+ }
}
return resultsForAllTests;
@@ -208,24 +216,9 @@ public class Controller
}
}
- public void setTestResultTimeout(final long testResultTimeout)
- {
- _testResultTimeout = testResultTimeout;
-
- }
-
void setClientRegistry(ClientRegistry clientRegistry)
{
_clientRegistry = clientRegistry;
-
}
- void setTestRunnerFactory(TestRunnerFactory factory)
- {
- if (factory == null)
- {
- throw new IllegalArgumentException("TestRunnerFactory cannot be null!");
- }
- _testRunnerFactory = factory;
- }
}
Added: qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/controller/HillClimber.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/controller/HillClimber.java?rev=1702969&view=auto
==============================================================================
--- qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/controller/HillClimber.java (added)
+++ qpid/java/trunk/perftests/src/main/java/org/apache/qpid/disttest/controller/HillClimber.java Mon Sep 14 15:29:42 2015
@@ -0,0 +1,84 @@
+/*
+ * 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.qpid.disttest.controller;
+
+public class HillClimber
+{
+ private double _value;
+ private double _delta;
+ private boolean _expSearch = true;
+ private int _direction = 0;
+
+ public HillClimber(double initialValue, double initialDelta)
+ {
+ _value = initialValue;
+ _delta = initialDelta / 2.0;
+ }
+
+ public double nextHigher()
+ {
+ if (_direction == 0)
+ {
+ _direction = 1;
+ }
+ else if (_direction == -1)
+ {
+ _expSearch = false;
+ _direction = 1;
+ }
+
+ return step();
+ }
+
+ public double nextLower()
+ {
+ if (_direction == 0)
+ {
+ _direction = -1;
+ }
+ else if (_direction == 1)
+ {
+ _expSearch = false;
+ _direction = -1;
+ }
+
+ return step();
+ }
+
+ private double step()
+ {
+ if (_expSearch)
+ {
+ _delta *= 2;
+ _value += _direction * _delta;
+ }
+ else
+ {
+ _delta /= 2.0;
+ _value += _direction * _delta;
+ }
+ return _value;
+ }
+
+ public double getCurrentDelta()
+ {
+ return _delta;
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org