QPID-3936: Change Performance Test Framework to support running of a series of test definition files and the production of a separate CSV per test definition.

git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@1329215 13f79535-47bb-0310-9956-ffa450edef68

20 files changed, 384 insertions, 186 deletions

diff --git a/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/ControllerRunner.java b/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/ControllerRunner.java
index 07d78790f0..aa9c582bf8 100644
--- a/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/ControllerRunner.java
+++ b/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/ControllerRunner.java
@@ -19,9 +19,14 @@
  */
 package org.apache.qpid.disttest;
 
+import java.io.File;
 import java.io.FileNotFoundException;
 import java.io.FileWriter;
+import java.io.FilenameFilter;
 import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
 
 import javax.naming.Context;
 
@@ -41,13 +46,11 @@ public class ControllerRunner extends AbstractRunner
 
     public static final String TEST_CONFIG_PROP = "test-config";
     public static final String DISTRIBUTED_PROP = "distributed";
-    public static final String OUTPUT_FILE_PROP = "outputfile";
+    public static final String OUTPUT_DIR_PROP = "outputdir";
 
     private static final String TEST_CONFIG_DEFAULT = "perftests-config.json";
     private static final String DISTRIBUTED_DEFAULT = "false";
-    private static final String OUTPUT_FILE_DEFAULT = "output.csv";
-
-    private Controller _controller;
+    private static final String OUTPUT_DIR_DEFAULT = ".";
 
     private final Aggregator _aggregator = new Aggregator();
 
@@ -56,7 +59,7 @@ public class ControllerRunner extends AbstractRunner
     {
         getCliOptions().put(TEST_CONFIG_PROP, TEST_CONFIG_DEFAULT);
         getCliOptions().put(DISTRIBUTED_PROP, DISTRIBUTED_DEFAULT);
-        getCliOptions().put(OUTPUT_FILE_PROP, OUTPUT_FILE_DEFAULT);
+        getCliOptions().put(OUTPUT_DIR_PROP, OUTPUT_DIR_DEFAULT);
     }
 
     public static void main(String[] args) throws Exception
@@ -84,42 +87,65 @@ public class ControllerRunner extends AbstractRunner
 
     private void runTests(ControllerJmsDelegate jmsDelegate)
     {
+        Controller controller = new Controller(jmsDelegate, DistributedTestConstants.REGISTRATION_TIMEOUT, DistributedTestConstants.COMMAND_RESPONSE_TIMEOUT);
 
-        _controller = new Controller(jmsDelegate,
-                DistributedTestConstants.REGISTRATION_TIMEOUT, DistributedTestConstants.COMMAND_RESPONSE_TIMEOUT);
-
-        Config testConfig = getTestConfig();
-        _controller.setConfig(testConfig);
+        final List<String> testConfigFiles = getTestConfigFiles();
+        createClientsIfNotDistributed(testConfigFiles);
 
-        if(!isDistributed())
+        try
         {
-            //we must create the required test clients, running in single-jvm mode
-            int numClients = testConfig.getTotalNumberOfClients();
-            for (int i = 1; i <= numClients; i++)
+            for (String testConfigFile : testConfigFiles)
             {
-                ClientRunner clientRunner = new ClientRunner();
-                clientRunner.setJndiPropertiesFileLocation(getJndiConfig());
-                clientRunner.runClients();
-            }
-        }
+                final Config testConfig = buildTestConfigFrom(testConfigFile);
+                controller.setConfig(testConfig);
 
-        ResultsForAllTests resultsForAllTests = null;
-        try
-        {
-            _controller.awaitClientRegistrations();
+                controller.awaitClientRegistrations();
 
-            ResultsForAllTests rawResultsForAllTests = _controller.runAllTests();
-            resultsForAllTests = _aggregator.aggregateResults(rawResultsForAllTests);
+                LOGGER.info("Running test : " + testConfigFile);
+                runTest(controller, testConfigFile);
+            }
         }
         finally
         {
-            _controller.stopAllRegisteredClients();
+            controller.stopAllRegisteredClients();
         }
 
-        final String outputFile = getOutputFile();
+    }
+
+    private void runTest(Controller controller, String testConfigFile)
+    {
+        final Config testConfig = buildTestConfigFrom(testConfigFile);
+        controller.setConfig(testConfig);
+
+        ResultsForAllTests rawResultsForAllTests = controller.runAllTests();
+        ResultsForAllTests resultsForAllTests = _aggregator.aggregateResults(rawResultsForAllTests);
+
+        final String outputFile = generateOutputCsvNameFrom(testConfigFile);
         writeResultsToFile(resultsForAllTests, outputFile);
     }
 
+    private void createClientsIfNotDistributed(final List<String> testConfigFiles)
+    {
+        if(!isDistributed())
+        {
+            int maxNumberOfClients = 0;
+            for (String testConfigFile : testConfigFiles)
+            {
+                final Config testConfig = buildTestConfigFrom(testConfigFile);
+                final int numClients = testConfig.getTotalNumberOfClients();
+                maxNumberOfClients = Math.max(numClients, maxNumberOfClients);
+            }
+
+            //we must create the required test clients, running in single-jvm mode
+            for (int i = 1; i <= maxNumberOfClients; i++)
+            {
+                ClientRunner clientRunner = new ClientRunner();
+                clientRunner.setJndiPropertiesFileLocation(getJndiConfig());
+                clientRunner.runClients();
+            }
+        }
+    }
+
     private void writeResultsToFile(ResultsForAllTests resultsForAllTests, String outputFile)
     {
         FileWriter writer = null;
@@ -150,22 +176,55 @@ public class ControllerRunner extends AbstractRunner
         }
     }
 
-    private String getOutputFile()
+    private String generateOutputCsvNameFrom(String testConfigFile)
+    {
+        final String filenameOnlyWithExtension = new File(testConfigFile).getName();
+        final String cvsFile = filenameOnlyWithExtension.replaceFirst(".?\\w*$", ".csv");
+        final String outputDir = String.valueOf(getCliOptions().get(ControllerRunner.OUTPUT_DIR_PROP));
+
+        return new File(outputDir, cvsFile).getAbsolutePath();
+    }
+
+    private List<String> getTestConfigFiles()
     {
-        return String.valueOf(getCliOptions().get(ControllerRunner.OUTPUT_FILE_PROP));
+        final List<String> testConfigFile = new ArrayList<String>();
+        final File configFileOrDirectory = new File(getCliOptions().get(ControllerRunner.TEST_CONFIG_PROP));
+
+        if (configFileOrDirectory.isDirectory())
+        {
+            final String[] configFiles = configFileOrDirectory.list(new FilenameFilter()
+            {
+                @Override
+                public boolean accept(File dir, String name)
+                {
+                    return new File(dir, name).isFile() && name.endsWith(".json");
+                }
+            });
+
+            for (String configFile : configFiles)
+            {
+                testConfigFile.add(new File(configFileOrDirectory, configFile).getAbsolutePath());
+            }
+        }
+        else
+        {
+            testConfigFile.add(configFileOrDirectory.getAbsolutePath());
+        }
+
+        return testConfigFile;
     }
 
-    private Config getTestConfig()
+    private Config buildTestConfigFrom(String testConfigFile)
     {
         ConfigReader configReader = new ConfigReader();
         Config testConfig;
         try
         {
-            testConfig = configReader.getConfigFromFile(getCliOptions().get(ControllerRunner.TEST_CONFIG_PROP));
+            testConfig = configReader.getConfigFromFile(testConfigFile);
         }
         catch (FileNotFoundException e)
         {
-            throw new DistributedTestException("Exception while loading test config", e);
+            throw new DistributedTestException("Exception while loading test config from " + testConfigFile, e);
         }
         return testConfig;
     }
diff --git a/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/client/ParticipantResultFactory.java b/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/client/ParticipantResultFactory.java
index 61b64b8c4f..2e4cda1460 100644
--- a/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/client/ParticipantResultFactory.java
+++ b/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/client/ParticipantResultFactory.java
@@ -42,25 +42,30 @@ public class ParticipantResultFactory
         consumerParticipantResult.setSelector(command.getSelector() != null);
         consumerParticipantResult.setNoLocal(command.isNoLocal());
         consumerParticipantResult.setSynchronousConsumer(command.isSynchronous());
+        consumerParticipantResult.setTotalNumberOfConsumers(1);
+        consumerParticipantResult.setTotalNumberOfProducers(0);
 
         return consumerParticipantResult;
     }
 
     public ProducerParticipantResult createForProducer(String participantName, String clientRegisteredName, CreateProducerCommand command, int numberOfMessagesSent, int payloadSize, long totalPayloadSent, Date start, Date end)
     {
-        final ProducerParticipantResult participantResult = new ProducerParticipantResult();
+        final ProducerParticipantResult producerParticipantResult = new ProducerParticipantResult();
 
-        participantResult.setStartDelay(command.getStartDelay());
-        participantResult.setDeliveryMode(command.getDeliveryMode());
-        participantResult.setPriority(command.getPriority());
-        participantResult.setInterval(command.getInterval());
-        participantResult.setTimeToLive(command.getTimeToLive());
+        producerParticipantResult.setStartDelay(command.getStartDelay());
+        producerParticipantResult.setDeliveryMode(command.getDeliveryMode());
+        producerParticipantResult.setPriority(command.getPriority());
+        producerParticipantResult.setInterval(command.getInterval());
+        producerParticipantResult.setTimeToLive(command.getTimeToLive());
+        producerParticipantResult.setTotalNumberOfConsumers(0);
+        producerParticipantResult.setTotalNumberOfProducers(1);
 
-        setTestProperties(participantResult, command, participantName, clientRegisteredName);
 
-        setTestResultProperties(participantResult, numberOfMessagesSent, payloadSize, totalPayloadSent, start, end);
+        setTestProperties(producerParticipantResult, command, participantName, clientRegisteredName);
 
-        return participantResult;
+        setTestResultProperties(producerParticipantResult, numberOfMessagesSent, payloadSize, totalPayloadSent, start, end);
+
+        return producerParticipantResult;
     }
 
     private void setTestResultProperties(final ParticipantResult participantResult, int numberOfMessagesSent, int payloadSize, long totalPayloadReceived, Date start, Date end)
diff --git a/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/controller/ClientRegistry.java b/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/controller/ClientRegistry.java
index b049da1f84..eaccb54f0e 100644
--- a/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/controller/ClientRegistry.java
+++ b/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/controller/ClientRegistry.java
@@ -33,6 +33,8 @@ public class ClientRegistry
 
     private final Set<String> _registeredClientNames = new ConcurrentSkipListSet<String>();
 
+    private final Object _lock = new Object();
+
     public void registerClient(String clientName)
     {
         final boolean alreadyContainsClient = !_registeredClientNames.add(clientName);
@@ -41,12 +43,54 @@ public class ClientRegistry
             throw new DistributedTestException("Duplicate client name " + clientName);
         }
 
-        LOGGER.info("Client registered: " + clientName);
+        notifyAllWaiters();
+
+        if (LOGGER.isInfoEnabled())
+        {
+            LOGGER.info("Client registered: " + clientName);
+        }
     }
 
+
     public Collection<String> getClients()
     {
         return Collections.unmodifiableSet(_registeredClientNames);
     }
 
+    public int awaitClients(int numberOfClientsToAwait, long timeout)
+    {
+        final long endTime = System.currentTimeMillis() + timeout;
+
+        int numberOfClientsAbsent = numberOfClientsToAwait - _registeredClientNames.size();
+        long remainingTimeout = endTime - System.currentTimeMillis();
+
+        while(numberOfClientsAbsent > 0 && remainingTimeout > 0)
+        {
+            synchronized (_lock)
+            {
+                try
+                {
+                    _lock.wait(remainingTimeout);
+                }
+                catch (InterruptedException e)
+                {
+                   Thread.currentThread().interrupt();
+                }
+            }
+
+            numberOfClientsAbsent = numberOfClientsToAwait - _registeredClientNames.size();
+            remainingTimeout = endTime - System.currentTimeMillis();
+        }
+
+        return numberOfClientsAbsent < 0 ? 0 : numberOfClientsAbsent;
+    }
+
+    private void notifyAllWaiters()
+    {
+        synchronized (_lock)
+        {
+            _lock.notifyAll();
+        }
+    }
+
 }
diff --git a/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/controller/Controller.java b/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/controller/Controller.java
index a5e0933704..7c935065f0 100644
--- a/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/controller/Controller.java
+++ b/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/controller/Controller.java
@@ -44,7 +44,6 @@ public class Controller
 
     private final ControllerJmsDelegate _jmsDelegate;
 
-    private volatile CountDownLatch _clientRegistrationLatch;
     private volatile CountDownLatch _stopClientsResponseLatch = null;
 
     private Config _config;
@@ -60,25 +59,29 @@ public class Controller
         _commandResponseTimeout = commandResponseTimeout;
         _testRunnerFactory = new TestRunnerFactory();
         _clientRegistry = new ClientRegistry();
+
+        _jmsDelegate.addCommandListener(new RegisterClientCommandListener());
+        _jmsDelegate.addCommandListener(new StopClientResponseListener());
+        _jmsDelegate.start();
     }
 
     public void setConfig(Config config)
     {
         _config = config;
         validateConfiguration();
-        int numberOfClients = config.getTotalNumberOfClients();
-        _clientRegistrationLatch = new CountDownLatch(numberOfClients);
-
-        _jmsDelegate.addCommandListener(new RegisterClientCommandListener());
-        _jmsDelegate.addCommandListener(new StopClientResponseListener());
-        _jmsDelegate.start();
     }
 
-
     public void awaitClientRegistrations()
     {
-        LOGGER.info("Awaiting client registration");
-        awaitLatch(_clientRegistrationLatch, _registrationTimeout, "Timed out waiting for registrations. Expecting %d more registrations");
+        LOGGER.info("Awaiting client registrations");
+
+        final int numberOfAbsentClients = _clientRegistry.awaitClients(_config.getTotalNumberOfClients(), _registrationTimeout);
+        if (numberOfAbsentClients > 0)
+        {
+            String formattedMessage = String.format("Timed out waiting for registrations. Expecting %d more registrations", numberOfAbsentClients);
+            throw new DistributedTestException(formattedMessage);
+        }
+
     }
 
     private void validateConfiguration()
@@ -111,15 +114,13 @@ public class Controller
     {
         final String clientName = registrationCommand.getClientName();
 
-        _clientRegistry.registerClient(clientName);
         _jmsDelegate.registerClient(registrationCommand);
-
-        _clientRegistrationLatch.countDown();
-        LOGGER.info("Counted down latch for client: " + clientName + " latch count=" + _clientRegistrationLatch.getCount());
+        _clientRegistry.registerClient(clientName);
     }
 
     void processStopClientResponse(final Response response)
     {
+        // TODO clientRegistry should expose a deregisterClient
         _stopClientsResponseLatch.countDown();
         if (response.hasError())
         {
diff --git a/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/message/ParticipantAttribute.java b/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/message/ParticipantAttribute.java
index c2fb38cc96..96bc4183de 100644
--- a/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/message/ParticipantAttribute.java
+++ b/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/message/ParticipantAttribute.java
@@ -29,29 +29,31 @@ import org.apache.qpid.disttest.client.Participant;
  */
 public enum ParticipantAttribute
 {
-    TEST_NAME("Test Name"),
-    ITERATION_NUMBER("Iteration number"),
-    CONFIGURED_CLIENT_NAME("Client Name"),
-    PARTICIPANT_NAME("Participant name"),
-    NUMBER_OF_MESSAGES_PROCESSED("Number of messages"),
-    PAYLOAD_SIZE("Payload size (bytes)"),
-    PRIORITY("Priority"),
-    TIME_TO_LIVE("Time to live (ms)"),
-    DELIVERY_MODE("Delivery mode"),
-    BATCH_SIZE("Batch size"),
-    MAXIMUM_DURATION("Maximum duration (ms)"),
-    PRODUCER_START_DELAY("Producer start delay (ms)"),
-    PRODUCER_INTERVAL("Producer interval (ms)"),
-    IS_TOPIC("Is topic"),
-    IS_DURABLE_SUBSCRIPTION("Is durable subscription"),
-    IS_BROWSIING_SUBSCRIPTION("Is browsing subscription"),
-    IS_SELECTOR("Is selector"),
-    IS_NO_LOCAL("Is no local"),
-    IS_SYNCHRONOUS_CONSUMER("Is synchronous consumer"),
-    TOTAL_PAYLOAD_PROCESSED("Total payload processed (bytes)"),
-    THROUGHPUT("Throughput (kbytes/s)"),
-    TIME_TAKEN("Time taken (ms)"),
-    ERROR_MESSAGE("Error message");
+    TEST_NAME("testName"),
+    ITERATION_NUMBER("iterationNumber"),
+    CONFIGURED_CLIENT_NAME("clientName"),
+    PARTICIPANT_NAME("participantName"),
+    NUMBER_OF_MESSAGES_PROCESSED("numberOfMessages"),
+    PAYLOAD_SIZE("payloadSizeB"),
+    PRIORITY("priority"),
+    TIME_TO_LIVE("timeToLiveMs"),
+    DELIVERY_MODE("deliveryMode"),
+    BATCH_SIZE("batchSize"),
+    MAXIMUM_DURATION("maximumDurationMs"),
+    PRODUCER_START_DELAY("producerStartDelayMs"),
+    PRODUCER_INTERVAL("producerIntervalMs"),
+    IS_TOPIC("isTopic"),
+    IS_DURABLE_SUBSCRIPTION("isDurableSubscription"),
+    IS_BROWSIING_SUBSCRIPTION("isBrowsingSubscription"),
+    IS_SELECTOR("isSelector"),
+    IS_NO_LOCAL("isNoLocal"),
+    IS_SYNCHRONOUS_CONSUMER("isSynchronousConsumer"),
+    TOTAL_NUMBER_OF_CONSUMERS("totalNumberOfConsumers"),
+    TOTAL_NUMBER_OF_PRODUCERS("totalNumberOfProducers"),
+    TOTAL_PAYLOAD_PROCESSED("totalPayloadProcessedB"),
+    THROUGHPUT("throughputKbPerS"),
+    TIME_TAKEN("timeTakenMs"),
+    ERROR_MESSAGE("errorMessage");
 
     private String _displayName;
 
diff --git a/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/message/ParticipantResult.java b/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/message/ParticipantResult.java
index 4550f10c65..576babf7a8 100644
--- a/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/message/ParticipantResult.java
+++ b/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/message/ParticipantResult.java
@@ -49,6 +49,10 @@ public class ParticipantResult extends Response
     private int _payloadSize;
     private double _throughput;
 
+    private int _totalNumberOfConsumers;
+    private int _totalNumberOfProducers;
+
+
     public static final Comparator<? super ParticipantResult> PARTICIPANT_NAME_COMPARATOR = new Comparator<ParticipantResult>()
     {
         @Override
@@ -228,4 +232,26 @@ public class ParticipantResult extends Response
         _throughput = throughput;
     }
 
+    public void setTotalNumberOfConsumers(int totalNumberOfConsumers)
+    {
+        _totalNumberOfConsumers = totalNumberOfConsumers;
+    }
+
+    @OutputAttribute(attribute=ParticipantAttribute.TOTAL_NUMBER_OF_CONSUMERS)
+    public int getTotalNumberOfConsumers()
+    {
+        return _totalNumberOfConsumers;
+    }
+
+    public void setTotalNumberOfProducers(int totalNumberOfProducers)
+    {
+        _totalNumberOfProducers = totalNumberOfProducers;
+    }
+
+    @OutputAttribute(attribute=ParticipantAttribute.TOTAL_NUMBER_OF_PRODUCERS)
+    public int getTotalNumberOfProducers()
+    {
+        return _totalNumberOfProducers;
+    }
+
 }
diff --git a/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/results/aggregation/ParticipantResultAggregator.java b/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/results/aggregation/ParticipantResultAggregator.java
index 6c4e4f87ac..603f475a16 100644
--- a/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/results/aggregation/ParticipantResultAggregator.java
+++ b/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/results/aggregation/ParticipantResultAggregator.java
@@ -35,14 +35,17 @@ public class ParticipantResultAggregator
     private long _numberOfMessagesProcessed = 0;
     private long _totalPayloadProcessed = 0;
 
+    private int _totalNumberOfConsumers = 0;
+    private int _totalNumberOfProducers = 0;
+
     private NavigableSet<Integer> _encounteredPayloadSizes = new TreeSet<Integer>();
     private NavigableSet<Integer> _encounteredIterationNumbers = new TreeSet<Integer>();
     private NavigableSet<String> _encountedTestNames = new TreeSet<String>();
 
-    public ParticipantResultAggregator(Class<? extends ParticipantResult> taregtClass, String aggregateResultName)
+    public ParticipantResultAggregator(Class<? extends ParticipantResult> targetClass, String aggregateResultName)
     {
         _aggregatedResultName = aggregateResultName;
-        _targetClass = taregtClass;
+        _targetClass = targetClass;
     }
 
     public void aggregate(ParticipantResult result)
@@ -73,6 +76,8 @@ public class ParticipantResultAggregator
     {
         _numberOfMessagesProcessed += result.getNumberOfMessagesProcessed();
         _totalPayloadProcessed += result.getTotalPayloadProcessed();
+        _totalNumberOfConsumers += result.getTotalNumberOfConsumers();
+        _totalNumberOfProducers += result.getTotalNumberOfProducers();
         _minStartDate = Math.min(_minStartDate, result.getStartInMillis());
         _maxEndDate = Math.max(_maxEndDate, result.getEndInMillis());
     }
@@ -91,6 +96,8 @@ public class ParticipantResultAggregator
     {
         aggregatedResult.setNumberOfMessagesProcessed(_numberOfMessagesProcessed);
         aggregatedResult.setTotalPayloadProcessed(_totalPayloadProcessed);
+        aggregatedResult.setTotalNumberOfConsumers(_totalNumberOfConsumers);
+        aggregatedResult.setTotalNumberOfProducers(_totalNumberOfProducers);
         aggregatedResult.setStartDate(new Date(_minStartDate));
         aggregatedResult.setEndDate(new Date(_maxEndDate));
         aggregatedResult.setThroughput(calculateThroughputInKiloBytesPerSecond());
diff --git a/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/results/aggregation/TestResultAggregator.java b/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/results/aggregation/TestResultAggregator.java
index 1938add560..ac0e146635 100644
--- a/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/results/aggregation/TestResultAggregator.java
+++ b/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/results/aggregation/TestResultAggregator.java
@@ -79,19 +79,21 @@ public class TestResultAggregator
     }
 
     private ParticipantResult buildAllResultFromOtherAggregatedResults(
-            ParticipantResult aggregatedProducerResult, ParticipantResult aggregaredConsumerResult)
+            ParticipantResult aggregatedProducerResult, ParticipantResult aggregatedConsumerResult)
     {
         ParticipantResult aggregatedAllResult = new ParticipantResult(ALL_PARTICIPANTS_NAME);
         aggregatedAllResult.setStartDate(aggregatedProducerResult.getStartDate());
 
-        aggregatedAllResult.setEndDate(aggregaredConsumerResult.getEndDate());
+        aggregatedAllResult.setEndDate(aggregatedConsumerResult.getEndDate());
 
-        aggregatedAllResult.setIterationNumber(aggregaredConsumerResult.getIterationNumber());
-        aggregatedAllResult.setTestName(aggregaredConsumerResult.getTestName());
-        aggregatedAllResult.setNumberOfMessagesProcessed(aggregaredConsumerResult.getNumberOfMessagesProcessed());
-        aggregatedAllResult.setPayloadSize(aggregaredConsumerResult.getPayloadSize());
-        aggregatedAllResult.setTotalPayloadProcessed(aggregaredConsumerResult.getTotalPayloadProcessed());
-        aggregatedAllResult.setThroughput(aggregaredConsumerResult.getThroughput());
+        aggregatedAllResult.setIterationNumber(aggregatedConsumerResult.getIterationNumber());
+        aggregatedAllResult.setTestName(aggregatedConsumerResult.getTestName());
+        aggregatedAllResult.setNumberOfMessagesProcessed(aggregatedConsumerResult.getNumberOfMessagesProcessed());
+        aggregatedAllResult.setPayloadSize(aggregatedConsumerResult.getPayloadSize());
+        aggregatedAllResult.setTotalPayloadProcessed(aggregatedConsumerResult.getTotalPayloadProcessed());
+        aggregatedAllResult.setThroughput(aggregatedConsumerResult.getThroughput());
+        aggregatedAllResult.setTotalNumberOfProducers(aggregatedProducerResult.getTotalNumberOfProducers());
+        aggregatedAllResult.setTotalNumberOfConsumers(aggregatedConsumerResult.getTotalNumberOfConsumers());
 
         return aggregatedAllResult;
     }
diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/client/ConsumerParticipantTest.java b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/client/ConsumerParticipantTest.java
index 805ebd3be4..8bc7318b0c 100644
--- a/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/client/ConsumerParticipantTest.java
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/client/ConsumerParticipantTest.java
@@ -18,7 +18,7 @@
  */
 package org.apache.qpid.disttest.client;
 
-import static org.apache.qpid.disttest.client.ParticipantTestHelper.assertExpectedResults;
+import static org.apache.qpid.disttest.client.ParticipantTestHelper.assertExpectedConsumerResults;
 import static org.mockito.Matchers.anyLong;
 import static org.mockito.Matchers.anyString;
 import static org.mockito.Mockito.atLeastOnce;
@@ -104,7 +104,7 @@ public class ConsumerParticipantTest extends TestCase
 
         ParticipantResult result = _consumerParticipant.doIt(CLIENT_NAME);
 
-        assertExpectedResults(result, PARTICIPANT_NAME1, CLIENT_NAME, _testStartTime, numberOfMessages, PAYLOAD_SIZE_PER_MESSAGE, totalPayloadSize, null);
+        assertExpectedConsumerResults(result, PARTICIPANT_NAME1, CLIENT_NAME, _testStartTime, numberOfMessages, PAYLOAD_SIZE_PER_MESSAGE, totalPayloadSize, null);
 
         _inOrder.verify(_delegate).consumeMessage(PARTICIPANT_NAME1, RECEIVE_TIMEOUT);
         _inOrder.verify(_delegate).calculatePayloadSizeFrom(_mockMessage);
@@ -118,7 +118,7 @@ public class ConsumerParticipantTest extends TestCase
 
         ParticipantResult result = _consumerParticipant.doIt(CLIENT_NAME);
 
-        assertExpectedResults(result, PARTICIPANT_NAME1, CLIENT_NAME, _testStartTime, null, PAYLOAD_SIZE_PER_MESSAGE, null, duration);
+        assertExpectedConsumerResults(result, PARTICIPANT_NAME1, CLIENT_NAME, _testStartTime, null, PAYLOAD_SIZE_PER_MESSAGE, null, duration);
 
         verify(_delegate, atLeastOnce()).consumeMessage(PARTICIPANT_NAME1, RECEIVE_TIMEOUT);
         verify(_delegate, atLeastOnce()).calculatePayloadSizeFrom(_mockMessage);
@@ -134,7 +134,7 @@ public class ConsumerParticipantTest extends TestCase
 
         ParticipantResult result = _consumerParticipant.doIt(CLIENT_NAME);
 
-        assertExpectedResults(result, PARTICIPANT_NAME1, CLIENT_NAME, _testStartTime, numberOfMessages, PAYLOAD_SIZE_PER_MESSAGE, totalPayloadSize, null);
+        assertExpectedConsumerResults(result, PARTICIPANT_NAME1, CLIENT_NAME, _testStartTime, numberOfMessages, PAYLOAD_SIZE_PER_MESSAGE, totalPayloadSize, null);
 
         verify(_delegate, times(numberOfMessages)).consumeMessage(PARTICIPANT_NAME1, RECEIVE_TIMEOUT);
         verify(_delegate, times(numberOfMessages)).calculatePayloadSizeFrom(_mockMessage);
@@ -157,7 +157,7 @@ public class ConsumerParticipantTest extends TestCase
 
         final int expectedPayloadResultPayloadSize = 0;
         final long totalPayloadSize = firstPayloadSize + secondPayloadSize + thirdPayloadSize;
-        assertExpectedResults(result, PARTICIPANT_NAME1, CLIENT_NAME, _testStartTime, numberOfMessages, expectedPayloadResultPayloadSize, totalPayloadSize, null);
+        assertExpectedConsumerResults(result, PARTICIPANT_NAME1, CLIENT_NAME, _testStartTime, numberOfMessages, expectedPayloadResultPayloadSize, totalPayloadSize, null);
 
         verify(_delegate, times(numberOfMessages)).consumeMessage(PARTICIPANT_NAME1, RECEIVE_TIMEOUT);
         verify(_delegate, times(numberOfMessages)).calculatePayloadSizeFrom(_mockMessage);
diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/client/ParticipantResultFactoryTest.java b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/client/ParticipantResultFactoryTest.java
index 73836f68e5..af906b5098 100644
--- a/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/client/ParticipantResultFactoryTest.java
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/client/ParticipantResultFactoryTest.java
@@ -76,6 +76,9 @@ public class ParticipantResultFactoryTest extends TestCase
         long timeToLive = 60;
         command.setTimeToLive(timeToLive);
 
+        int totalNumberOfConsumers = 0;
+        int totalNumberOfProducers = 1;
+
         ProducerParticipantResult result = _participantResultFactory.createForProducer(PARTICIPANT_NAME,
                                                                                        REGISTERED_CLIENT_NAME,
                                                                                        command,
@@ -92,6 +95,8 @@ public class ParticipantResultFactoryTest extends TestCase
         assertEquals(producerInterval, result.getInterval());
         assertEquals(producerStartDelay, result.getStartDelay());
         assertEquals(timeToLive, result.getTimeToLive());
+        assertEquals(totalNumberOfConsumers, result.getTotalNumberOfConsumers());
+        assertEquals(totalNumberOfProducers, result.getTotalNumberOfProducers());
     }
 
     public void testCreateForConsumer()
@@ -118,6 +123,9 @@ public class ParticipantResultFactoryTest extends TestCase
         boolean synchronousConsumer = true;
         command.setSynchronous(synchronousConsumer);
 
+        int totalNumberOfConsumers = 1;
+        int totalNumberOfProducers = 0;
+
         ConsumerParticipantResult result = _participantResultFactory.createForConsumer(PARTICIPANT_NAME,
                                                                                        REGISTERED_CLIENT_NAME,
                                                                                        command,
@@ -135,6 +143,8 @@ public class ParticipantResultFactoryTest extends TestCase
         assertEquals(isSelector,             result.isSelector());
         assertEquals(noLocal,                result.isNoLocal());
         assertEquals(synchronousConsumer,    result.isSynchronousConsumer());
+        assertEquals(totalNumberOfConsumers, result.getTotalNumberOfConsumers());
+        assertEquals(totalNumberOfProducers, result.getTotalNumberOfProducers());
     }
 
     public void testCreateForError()
diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/client/ParticipantTestHelper.java b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/client/ParticipantTestHelper.java
index b30d5c1c7c..fa071881cc 100644
--- a/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/client/ParticipantTestHelper.java
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/client/ParticipantTestHelper.java
@@ -32,7 +32,21 @@ public class ParticipantTestHelper
         assertTrue(message + " " + actual, actual >= minimumExpected);
     }
 
-    public static void assertExpectedResults(ParticipantResult result, String participantName, String registeredClientName, long expectedTestStartTime, Integer expectedNumberOfMessages, Integer expectedPayloadSize, Long expectedTotalPayloadProcessed, Long expectedMinimumExpectedDuration)
+    public static void assertExpectedConsumerResults(ParticipantResult result, String participantName, String registeredClientName, long expectedTestStartTime, Integer expectedNumberOfMessages, Integer expectedPayloadSize, Long expectedTotalPayloadProcessed, Long expectedMinimumExpectedDuration)
+    {
+        assertExpectedResults(result, participantName, registeredClientName, expectedTestStartTime, expectedNumberOfMessages, expectedPayloadSize, expectedTotalPayloadProcessed, expectedMinimumExpectedDuration);
+        assertEquals("Unexpected number of consumers", 1, result.getTotalNumberOfConsumers());
+        assertEquals("Unexpected number of producers", 0, result.getTotalNumberOfProducers());
+    }
+
+    public static void assertExpectedProducerResults(ParticipantResult result, String participantName, String registeredClientName, long expectedTestStartTime, Integer expectedNumberOfMessages, Integer expectedPayloadSize, Long expectedTotalPayloadProcessed, Long expectedMinimumExpectedDuration)
+    {
+        assertExpectedResults(result, participantName, registeredClientName, expectedTestStartTime, expectedNumberOfMessages, expectedPayloadSize, expectedTotalPayloadProcessed, expectedMinimumExpectedDuration);
+        assertEquals("Unexpected number of producers", 1, result.getTotalNumberOfProducers());
+        assertEquals("Unexpected number of consumers", 0, result.getTotalNumberOfConsumers());
+    }
+
+    private static void assertExpectedResults(ParticipantResult result, String participantName, String registeredClientName, long expectedTestStartTime, Integer expectedNumberOfMessages, Integer expectedPayloadSize, Long expectedTotalPayloadProcessed, Long expectedMinimumExpectedDuration)
     {
         assertFalse(result.hasError());
 
diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/client/ProducerParticipantTest.java b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/client/ProducerParticipantTest.java
index 3852948201..c78fc8837c 100644
--- a/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/client/ProducerParticipantTest.java
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/client/ProducerParticipantTest.java
@@ -18,7 +18,7 @@
  */
 package org.apache.qpid.disttest.client;
 
-import static org.apache.qpid.disttest.client.ParticipantTestHelper.assertExpectedResults;
+import static org.apache.qpid.disttest.client.ParticipantTestHelper.assertExpectedProducerResults;
 import static org.mockito.Matchers.isA;
 import static org.mockito.Mockito.atLeastOnce;
 import static org.mockito.Mockito.inOrder;
@@ -87,7 +87,7 @@ public class ProducerParticipantTest extends TestCase
         ParticipantResult result = _producer.doIt(CLIENT_NAME);
 
         long expectedPublishedStartTime = _testStartTime + delay;
-        assertExpectedResults(result, PARTICIPANT_NAME1, CLIENT_NAME, expectedPublishedStartTime, numberOfMessages, PAYLOAD_SIZE_PER_MESSAGE, totalPayloadSize, null);
+        assertExpectedProducerResults(result, PARTICIPANT_NAME1, CLIENT_NAME, expectedPublishedStartTime, numberOfMessages, PAYLOAD_SIZE_PER_MESSAGE, totalPayloadSize, null);
     }
 
 
@@ -120,7 +120,7 @@ public class ProducerParticipantTest extends TestCase
         _command.setDeliveryMode(deliveryMode);
 
         ParticipantResult result = (ParticipantResult) _producer.doIt(CLIENT_NAME);
-        assertExpectedResults(result, PARTICIPANT_NAME1, CLIENT_NAME, _testStartTime, numberOfMessages, PAYLOAD_SIZE_PER_MESSAGE, totalPayloadSize, null);
+        assertExpectedProducerResults(result, PARTICIPANT_NAME1, CLIENT_NAME, _testStartTime, numberOfMessages, PAYLOAD_SIZE_PER_MESSAGE, totalPayloadSize, null);
 
         _inOrder.verify(_delegate).sendNextMessage(isA(CreateProducerCommand.class));
         _inOrder.verify(_delegate).calculatePayloadSizeFrom(_mockMessage);
@@ -134,7 +134,7 @@ public class ProducerParticipantTest extends TestCase
         _command.setMaximumDuration(duration);
 
         ParticipantResult result = _producer.doIt(CLIENT_NAME);
-        assertExpectedResults(result, PARTICIPANT_NAME1, CLIENT_NAME, _testStartTime, null, PAYLOAD_SIZE_PER_MESSAGE, null, duration);
+        assertExpectedProducerResults(result, PARTICIPANT_NAME1, CLIENT_NAME, _testStartTime, null, PAYLOAD_SIZE_PER_MESSAGE, null, duration);
 
         verify(_delegate, atLeastOnce()).sendNextMessage(isA(CreateProducerCommand.class));
         verify(_delegate, atLeastOnce()).calculatePayloadSizeFrom(_mockMessage);
@@ -151,7 +151,7 @@ public class ProducerParticipantTest extends TestCase
         _command.setBatchSize(3);
 
         ParticipantResult result = _producer.doIt(CLIENT_NAME);
-        assertExpectedResults(result, PARTICIPANT_NAME1, CLIENT_NAME, _testStartTime, numberOfMessages, PAYLOAD_SIZE_PER_MESSAGE, totalPayloadSize, null);
+        assertExpectedProducerResults(result, PARTICIPANT_NAME1, CLIENT_NAME, _testStartTime, numberOfMessages, PAYLOAD_SIZE_PER_MESSAGE, totalPayloadSize, null);
 
         verify(_delegate, times(numberOfMessages)).sendNextMessage(isA(CreateProducerCommand.class));
         verify(_delegate, times(numberOfMessages)).calculatePayloadSizeFrom(_mockMessage);
@@ -172,7 +172,7 @@ public class ProducerParticipantTest extends TestCase
         _command.setInterval(publishInterval);
 
         ParticipantResult result = _producer.doIt(CLIENT_NAME);
-        assertExpectedResults(result, PARTICIPANT_NAME1, CLIENT_NAME, _testStartTime, numberOfMessages, null, totalPayloadSize, expectedTimeToRunTest);
+        assertExpectedProducerResults(result, PARTICIPANT_NAME1, CLIENT_NAME, _testStartTime, numberOfMessages, null, totalPayloadSize, expectedTimeToRunTest);
 
         verify(_delegate, times(numberOfMessages)).sendNextMessage(isA(CreateProducerCommand.class));
         verify(_delegate, times(numberOfMessages)).calculatePayloadSizeFrom(_mockMessage);
@@ -196,7 +196,7 @@ public class ProducerParticipantTest extends TestCase
         ParticipantResult result = _producer.doIt(CLIENT_NAME);
 
         final int expectedPayloadResultPayloadSize = 0;
-        assertExpectedResults(result, PARTICIPANT_NAME1, CLIENT_NAME, _testStartTime, numberOfMessages, expectedPayloadResultPayloadSize, totalPayloadSize, null);
+        assertExpectedProducerResults(result, PARTICIPANT_NAME1, CLIENT_NAME, _testStartTime, numberOfMessages, expectedPayloadResultPayloadSize, totalPayloadSize, null);
 
         verify(_delegate, times(numberOfMessages)).sendNextMessage(isA(CreateProducerCommand.class));
         verify(_delegate, times(numberOfMessages)).calculatePayloadSizeFrom(_mockMessage);
diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/controller/ClientRegistryTest.java b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/controller/ClientRegistryTest.java
index 2c161012a9..cc969e1ef2 100644
--- a/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/controller/ClientRegistryTest.java
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/controller/ClientRegistryTest.java
@@ -18,6 +18,9 @@
  */
 package org.apache.qpid.disttest.controller;
 
+import java.util.Timer;
+import java.util.TimerTask;
+
 import junit.framework.TestCase;
 
 import org.apache.qpid.disttest.DistributedTestException;
@@ -25,6 +28,8 @@ import org.apache.qpid.disttest.DistributedTestException;
 public class ClientRegistryTest extends TestCase
 {
     private static final String CLIENT1_REGISTERED_NAME = "CLIENT1_REGISTERED_NAME";
+    private static final String CLIENT2_REGISTERED_NAME = "CLIENT2_REGISTERED_NAME";
+    private static final int AWAIT_DELAY = 100;
 
     private ClientRegistry _clientRegistry = new ClientRegistry();
 
@@ -51,7 +56,44 @@ public class ClientRegistryTest extends TestCase
         }
     }
 
+    public void testAwaitOneClientWhenClientNotRegistered()
+    {
+        int numberOfClientsAbsent = _clientRegistry.awaitClients(1, AWAIT_DELAY);
+        assertEquals(1, numberOfClientsAbsent);
+    }
 
+    public void testAwaitOneClientWhenClientAlreadyRegistered()
+    {
+        _clientRegistry.registerClient(CLIENT1_REGISTERED_NAME);
 
+        int numberOfClientsAbsent = _clientRegistry.awaitClients(1, AWAIT_DELAY);
+        assertEquals(0, numberOfClientsAbsent);
+    }
+
+    public void testAwaitTwoClientWhenClientRegistersWhilstWaiting()
+    {
+        _clientRegistry.registerClient(CLIENT1_REGISTERED_NAME);
+        registerClientLater(CLIENT2_REGISTERED_NAME, 50);
 
+        int numberOfClientsAbsent = _clientRegistry.awaitClients(2, AWAIT_DELAY);
+        assertEquals(0, numberOfClientsAbsent);
+    }
+
+    private void registerClientLater(final String clientName, long delayInMillis)
+    {
+        doLater(new TimerTask()
+        {
+            @Override
+            public void run()
+            {
+                _clientRegistry.registerClient(clientName);
+            }
+        }, delayInMillis);
+    }
+
+    private void doLater(TimerTask task, long delayInMillis)
+    {
+        Timer timer = new Timer();
+        timer.schedule(task, delayInMillis);
+    }
 }
diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/controller/ControllerTest.java b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/controller/ControllerTest.java
index c119656afd..bc58ea41c5 100644
--- a/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/controller/ControllerTest.java
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/controller/ControllerTest.java
@@ -30,8 +30,6 @@ import static org.mockito.Mockito.when;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
-import java.util.Timer;
-import java.util.TimerTask;
 
 import junit.framework.TestCase;
 
@@ -50,9 +48,7 @@ import org.mockito.stubbing.Answer;
 public class ControllerTest extends TestCase
 {
     private static final String CLIENT1_REGISTERED_NAME = "client-uid1";
-    private static final String CLIENT2_REGISTERED_NAME = "client-uid2";
 
-    private static final int DELAY = 100;
     private static final long COMMAND_RESPONSE_TIMEOUT = 1000;
     private static final long REGISTRATION_TIMEOUT = 1000;
 
@@ -104,43 +100,19 @@ public class ControllerTest extends TestCase
         }
     }
 
-    public void testControllerRequiresOneClientRegistration()
-    {
-        Config configWithOneClient = createMockConfig(1);
-
-        _controller.setConfig(configWithOneClient);
-        registerClientAndAwait(CLIENT1_REGISTERED_NAME);
-    }
-
     public void testControllerReceivesTwoExpectedClientRegistrations()
     {
         Config configWithTwoClients = createMockConfig(2);
         _controller.setConfig(configWithTwoClients);
+        when(_clientRegistry.awaitClients(2, REGISTRATION_TIMEOUT)).thenReturn(0);
 
-        registerClientLater(_controller, CLIENT1_REGISTERED_NAME);
-        registerClientLater(_controller, CLIENT2_REGISTERED_NAME);
         _controller.awaitClientRegistrations();
     }
 
     public void testControllerDoesntReceiveAnyRegistrations()
     {
-        try
-        {
-            _controller.awaitClientRegistrations();
-            fail("Exception not thrown");
-        }
-        catch (DistributedTestException e)
-        {
-            // PASS
-        }
-    }
-
-    public void testControllerDoesntReceiveTwoExpectedClientRegistrations()
-    {
-        Config configWithTwoClients = createMockConfig(2);
-        _controller.setConfig(configWithTwoClients);
+        when(_clientRegistry.awaitClients(1, REGISTRATION_TIMEOUT)).thenReturn(1);
 
-        registerClientLater(_controller, CLIENT1_REGISTERED_NAME); // only receives one out of two expected registrations
         try
         {
             _controller.awaitClientRegistrations();
@@ -202,12 +174,6 @@ public class ControllerTest extends TestCase
         return config;
     }
 
-    private void doLater(TimerTask task, long delayInMillis)
-    {
-        Timer timer = new Timer();
-        timer.schedule(task, delayInMillis);
-    }
-
     private Config createMockConfig(int numberOfClients)
     {
         Config config = mock(Config.class);
@@ -215,27 +181,6 @@ public class ControllerTest extends TestCase
         return config;
     }
 
-    private void registerClientAndAwait(String... clientNames)
-    {
-        for (String clientName : clientNames)
-        {
-            registerClientLater(_controller, clientName);
-        }
-        _controller.awaitClientRegistrations();
-    }
-
-    private void registerClientLater(final Controller controller, final String clientName)
-    {
-        doLater(new TimerTask()
-        {
-            @Override
-            public void run()
-            {
-                controller.registerClient(new RegisterClientCommand(clientName, "dummy"));
-            }
-        }, DELAY);
-    }
-
     private TestRunnerFactory createTestFactoryReturningMock()
     {
         TestRunnerFactory testRunnerFactory = mock(TestRunnerFactory.class);
diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/message/ParticipantResultTest.java b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/message/ParticipantResultTest.java
index ee095ce199..4d636054c1 100644
--- a/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/message/ParticipantResultTest.java
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/message/ParticipantResultTest.java
@@ -18,7 +18,7 @@
  */
 package org.apache.qpid.disttest.message;
 
-import static org.apache.qpid.disttest.message.ParticipantAttribute.BATCH_SIZE;
+import static org.apache.qpid.disttest.message.ParticipantAttribute.*;
 import static org.apache.qpid.disttest.message.ParticipantAttribute.CONFIGURED_CLIENT_NAME;
 import static org.apache.qpid.disttest.message.ParticipantAttribute.DELIVERY_MODE;
 import static org.apache.qpid.disttest.message.ParticipantAttribute.ERROR_MESSAGE;
@@ -67,6 +67,9 @@ public class ParticipantResultTest extends TestCase
         long endTime = startTime + timeTaken;
         long maximumDuration = 1000;
 
+        int totalNumberOfConsumers = 1;
+        int totalNumberOfProducers = 1;
+
         result.setParticipantName(participantName);
         result.setTestName(testName);
         result.setIterationNumber(iterationNumber);
@@ -74,12 +77,15 @@ public class ParticipantResultTest extends TestCase
 
         result.setNumberOfMessagesProcessed(numberOfMessages);
         result.setConfiguredClientName(clientConfiguredName);
-        result.setBatchSize(10);
+        result.setBatchSize(batchSize);
 
         result.setStartDate(new Date(startTime));
         result.setEndDate(new Date(endTime));
         result.setMaximumDuration(maximumDuration);
 
+        result.setTotalNumberOfConsumers(totalNumberOfConsumers);
+        result.setTotalNumberOfProducers(totalNumberOfProducers);
+
         result.setErrorMessage(errorMessage);
 
         assertEquals(participantName,      result.getAttributes().get(PARTICIPANT_NAME));
@@ -87,8 +93,12 @@ public class ParticipantResultTest extends TestCase
         assertEquals(clientConfiguredName,   result.getAttributes().get(CONFIGURED_CLIENT_NAME));
         assertEquals(numberOfMessages,       result.getAttributes().get(NUMBER_OF_MESSAGES_PROCESSED));
         assertEquals(timeTaken,              result.getAttributes().get(TIME_TAKEN));
+        assertEquals(timeTaken,              result.getAttributes().get(TIME_TAKEN));
+        assertEquals(timeTaken,              result.getAttributes().get(TIME_TAKEN));
         assertEquals(batchSize,              result.getAttributes().get(BATCH_SIZE));
         assertEquals(maximumDuration,        result.getAttributes().get(MAXIMUM_DURATION));
+        assertEquals(totalNumberOfConsumers, result.getAttributes().get(TOTAL_NUMBER_OF_CONSUMERS));
+        assertEquals(totalNumberOfProducers, result.getAttributes().get(TOTAL_NUMBER_OF_PRODUCERS));
         assertEquals(errorMessage,           result.getAttributes().get(ERROR_MESSAGE));
         assertEquals(iterationNumber,         result.getAttributes().get(ITERATION_NUMBER));
     }
diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/results/aggregation/ParticipantResultAggregatorTest.java b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/results/aggregation/ParticipantResultAggregatorTest.java
index 5cf84c77f1..db30595edd 100644
--- a/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/results/aggregation/ParticipantResultAggregatorTest.java
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/results/aggregation/ParticipantResultAggregatorTest.java
@@ -179,4 +179,27 @@ public class ParticipantResultAggregatorTest extends TestCase
         assertEquals(0, aggregratedResult.getPayloadSize());
     }
 
+    public void testSumNumberOfConsumerAndProducers() throws Exception
+    {
+        final int expectedNumberOfProducers = 1;
+        final int expectedNumberOfConsumers = 2;
+
+        ParticipantResult result1 = new ParticipantResult();
+        result1.setTotalNumberOfConsumers(1);
+
+        ParticipantResult result2 = new ParticipantResult();
+        result2.setTotalNumberOfConsumers(1);
+
+        ParticipantResult result3 = new ParticipantResult();
+        result2.setTotalNumberOfProducers(1);
+
+        _aggregator.aggregate(result1);
+        _aggregator.aggregate(result2);
+        _aggregator.aggregate(result3);
+
+        ParticipantResult aggregratedResult = _aggregator.getAggregatedResult();
+        assertEquals(expectedNumberOfConsumers, aggregratedResult.getTotalNumberOfConsumers());
+        assertEquals(expectedNumberOfProducers, aggregratedResult.getTotalNumberOfProducers());
+    }
+
 }
diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/results/aggregation/TestResultAggregatorTest.java b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/results/aggregation/TestResultAggregatorTest.java
index 7cf900ca79..95f01d853a 100644
--- a/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/results/aggregation/TestResultAggregatorTest.java
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/results/aggregation/TestResultAggregatorTest.java
@@ -78,17 +78,17 @@ public class TestResultAggregatorTest extends TestCase
         assertMinimalAggregatedResults(
                 aggregatedTestResult.getAllParticipantResult(),
                 TEST1_NAME, TEST1_ITERATION_NUMBER,
-                NUMBER_OF_MESSAGES_CONSUMED_IN_TOTAL);
+                NUMBER_OF_MESSAGES_CONSUMED_IN_TOTAL, 2, 1);
 
         assertMinimalAggregatedResults(
                 aggregatedTestResult.getAllConsumerParticipantResult(),
                 TEST1_NAME, TEST1_ITERATION_NUMBER,
-                NUMBER_OF_MESSAGES_CONSUMED_IN_TOTAL);
+                NUMBER_OF_MESSAGES_CONSUMED_IN_TOTAL, 2, 0);
 
         assertMinimalAggregatedResults(
                 aggregatedTestResult.getAllProducerParticipantResult(),
                 TEST1_NAME, TEST1_ITERATION_NUMBER,
-                NUMBER_OF_MESSAGES_PRODUCED);
+                NUMBER_OF_MESSAGES_PRODUCED, 0, 1);
     }
 
     public void testAggregateResultsWhenParticipantErrored()
@@ -103,11 +103,13 @@ public class TestResultAggregatorTest extends TestCase
         assertEquals(TestResultAggregator.AGGREGATED_ERROR_MESSAGE, aggregatedTestResult.getAllParticipantResult().getErrorMessage());
     }
 
-    private void assertMinimalAggregatedResults(ParticipantResult result, String expectedTestName, int expectedIterationNumber, long expectedNumberOfMessagesProcessed)
+    private void assertMinimalAggregatedResults(ParticipantResult result, String expectedTestName, int expectedIterationNumber, long expectedNumberOfMessagesProcessed, int expectedTotalNumberOfConsumers, int expectedTotalNumberOfProducers)
     {
         assertEquals("Unexpected test name in " + result.getParticipantName(), expectedTestName, result.getTestName());
         assertEquals("Unexpected iteration number in " + result.getParticipantName(), expectedIterationNumber, result.getIterationNumber());
         assertEquals("Unexpected number of messages processed in " + result.getParticipantName(), expectedNumberOfMessagesProcessed, result.getNumberOfMessagesProcessed());
+        assertEquals("Unexpected total number of consumers " + result.getParticipantName(), expectedTotalNumberOfConsumers, result.getTotalNumberOfConsumers());
+        assertEquals("Unexpected total number of producers " + result.getParticipantName(), expectedTotalNumberOfProducers, result.getTotalNumberOfProducers());
     }
 
     private TestResult createResultsFromTest()
@@ -115,25 +117,27 @@ public class TestResultAggregatorTest extends TestCase
         TestResult testResult = new TestResult(TEST1_NAME);
 
         ConsumerParticipantResult consumerResult1 = new ConsumerParticipantResult();
-        setPropertiesOn(consumerResult1, TEST1_NAME, TEST1_ITERATION_NUMBER, CONSUMER_PARTICIPANT_NAME1, NUMBER_OF_MESSAGES_PROCESSED_PER_CONSUMER, PAYLOAD_SIZE, TOTAL_PAYLOAD_PROCESSED_PER_CONSUMER, CONSUMER1_STARTDATE, CONSUMER1_ENDDATE);
+        setPropertiesOn(consumerResult1, TEST1_NAME, TEST1_ITERATION_NUMBER, CONSUMER_PARTICIPANT_NAME1, NUMBER_OF_MESSAGES_PROCESSED_PER_CONSUMER, PAYLOAD_SIZE, TOTAL_PAYLOAD_PROCESSED_PER_CONSUMER, CONSUMER1_STARTDATE, CONSUMER1_ENDDATE, 1, 0);
         testResult.addParticipantResult(consumerResult1);
 
         ConsumerParticipantResult consumerResult2 = new ConsumerParticipantResult();
-        setPropertiesOn(consumerResult2, TEST1_NAME, TEST1_ITERATION_NUMBER, CONSUMER_PARTICIPANT_NAME2, NUMBER_OF_MESSAGES_PROCESSED_PER_CONSUMER, PAYLOAD_SIZE, TOTAL_PAYLOAD_PROCESSED_PER_CONSUMER, CONSUMER2_STARTDATE, CONSUMER2_ENDDATE);
+        setPropertiesOn(consumerResult2, TEST1_NAME, TEST1_ITERATION_NUMBER, CONSUMER_PARTICIPANT_NAME2, NUMBER_OF_MESSAGES_PROCESSED_PER_CONSUMER, PAYLOAD_SIZE, TOTAL_PAYLOAD_PROCESSED_PER_CONSUMER, CONSUMER2_STARTDATE, CONSUMER2_ENDDATE, 1, 0);
         testResult.addParticipantResult(consumerResult2);
 
         ParticipantResult producerResult = new ProducerParticipantResult();
-        setPropertiesOn(producerResult, TEST1_NAME, TEST1_ITERATION_NUMBER, PRODUCER_PARTICIPANT_NAME, NUMBER_OF_MESSAGES_PRODUCED, PAYLOAD_SIZE, TOTAL_PAYLOAD_PRODUCED_IN_TOTAL, PRODUCER_STARTDATE, PRODUCER_ENDDATE);
+        setPropertiesOn(producerResult, TEST1_NAME, TEST1_ITERATION_NUMBER, PRODUCER_PARTICIPANT_NAME, NUMBER_OF_MESSAGES_PRODUCED, PAYLOAD_SIZE, TOTAL_PAYLOAD_PRODUCED_IN_TOTAL, PRODUCER_STARTDATE, PRODUCER_ENDDATE, 0, 1);
         testResult.addParticipantResult(producerResult);
 
         return testResult;
     }
 
-    private void setPropertiesOn(ParticipantResult participantResult, String testName, int iterationNumber, String participantName, long numberOfMessagesProcessed, int payloadSize, long totalPayloadProcessed, long start, long end)
+    private void setPropertiesOn(ParticipantResult participantResult, String testName, int iterationNumber, String participantName, long numberOfMessagesProcessed, int payloadSize, long totalPayloadProcessed, long start, long end, int totalNumberOfConsumers, int totalNumberOfProducers)
     {
         participantResult.setParticipantName(participantName);
         participantResult.setTestName(testName);
         participantResult.setIterationNumber(iterationNumber);
+        participantResult.setTotalNumberOfConsumers(totalNumberOfConsumers);
+        participantResult.setTotalNumberOfProducers(totalNumberOfProducers);
 
         participantResult.setNumberOfMessagesProcessed(numberOfMessagesProcessed);
         participantResult.setPayloadSize(payloadSize);
diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/results/formatting/CSVFormaterTest.java b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/results/formatting/CSVFormaterTest.java
index 061ed6a28d..14fe7179da 100644
--- a/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/results/formatting/CSVFormaterTest.java
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/results/formatting/CSVFormaterTest.java
@@ -30,16 +30,18 @@ import static org.apache.qpid.disttest.message.ParticipantAttribute.IS_SYNCHRONO
 import static org.apache.qpid.disttest.message.ParticipantAttribute.IS_TOPIC;
 import static org.apache.qpid.disttest.message.ParticipantAttribute.ITERATION_NUMBER;
 import static org.apache.qpid.disttest.message.ParticipantAttribute.MAXIMUM_DURATION;
-import static org.apache.qpid.disttest.message.ParticipantAttribute.PAYLOAD_SIZE;
 import static org.apache.qpid.disttest.message.ParticipantAttribute.NUMBER_OF_MESSAGES_PROCESSED;
 import static org.apache.qpid.disttest.message.ParticipantAttribute.PARTICIPANT_NAME;
+import static org.apache.qpid.disttest.message.ParticipantAttribute.PAYLOAD_SIZE;
 import static org.apache.qpid.disttest.message.ParticipantAttribute.PRIORITY;
 import static org.apache.qpid.disttest.message.ParticipantAttribute.PRODUCER_INTERVAL;
 import static org.apache.qpid.disttest.message.ParticipantAttribute.PRODUCER_START_DELAY;
 import static org.apache.qpid.disttest.message.ParticipantAttribute.TEST_NAME;
+import static org.apache.qpid.disttest.message.ParticipantAttribute.THROUGHPUT;
 import static org.apache.qpid.disttest.message.ParticipantAttribute.TIME_TAKEN;
 import static org.apache.qpid.disttest.message.ParticipantAttribute.TIME_TO_LIVE;
-import static org.apache.qpid.disttest.message.ParticipantAttribute.THROUGHPUT;
+import static org.apache.qpid.disttest.message.ParticipantAttribute.TOTAL_NUMBER_OF_CONSUMERS;
+import static org.apache.qpid.disttest.message.ParticipantAttribute.TOTAL_NUMBER_OF_PRODUCERS;
 import static org.apache.qpid.disttest.message.ParticipantAttribute.TOTAL_PAYLOAD_PROCESSED;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
@@ -56,7 +58,6 @@ import org.apache.qpid.disttest.controller.ResultsForAllTests;
 import org.apache.qpid.disttest.controller.TestResult;
 import org.apache.qpid.disttest.message.ParticipantAttribute;
 import org.apache.qpid.disttest.message.ParticipantResult;
-import org.apache.qpid.disttest.results.formatting.CSVFormater;
 
 public class CSVFormaterTest extends TestCase
 {
@@ -110,6 +111,8 @@ public class CSVFormaterTest extends TestCase
         participantAttributes.put(IS_SELECTOR, false);
         participantAttributes.put(IS_NO_LOCAL, true);
         participantAttributes.put(IS_SYNCHRONOUS_CONSUMER, false);
+        participantAttributes.put(TOTAL_NUMBER_OF_CONSUMERS, 1);
+        participantAttributes.put(TOTAL_NUMBER_OF_PRODUCERS, 2);
         participantAttributes.put(TOTAL_PAYLOAD_PROCESSED, 1024);
         participantAttributes.put(THROUGHPUT, 2048);
         participantAttributes.put(TIME_TAKEN, 1000);
diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/results/formatting/expectedOutput.csv b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/results/formatting/expectedOutput.csv
index 08cb2275b5..073fafb68a 100644
--- a/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/results/formatting/expectedOutput.csv
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/results/formatting/expectedOutput.csv
@@ -1,2 +1,2 @@
-Test Name,Iteration number,Client Name,Participant name,Number of messages,Payload size (bytes),Priority,Time to live (ms),Delivery mode,Batch size,Maximum duration (ms),Producer start delay (ms),Producer interval (ms),Is topic,Is durable subscription,Is browsing subscription,Is selector,Is no local,Is synchronous consumer,Total payload processed (bytes),Throughput (kbytes/s),Time taken (ms),Error message
-TEST1,0,CONFIGURED_CLIENT1,PARTICIPANT,1,2,3,4,5,6,7,8,9,true,false,true,false,true,false,1024,2048,1000,error
\ No newline at end of file
+testName,iterationNumber,clientName,participantName,numberOfMessages,payloadSizeB,priority,timeToLiveMs,deliveryMode,batchSize,maximumDurationMs,producerStartDelayMs,producerIntervalMs,isTopic,isDurableSubscription,isBrowsingSubscription,isSelector,isNoLocal,isSynchronousConsumer,totalNumberOfConsumers,totalNumberOfProducers,totalPayloadProcessedB,throughputKbPerS,timeTakenMs,errorMessage
+TEST1,0,CONFIGURED_CLIENT1,PARTICIPANT,1,2,3,4,5,6,7,8,9,true,false,true,false,true,false,1,2,1024,2048,1000,error
\ No newline at end of file
diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/systest/disttest/endtoend/EndToEndTest.java b/qpid/java/perftests/src/test/java/org/apache/qpid/systest/disttest/endtoend/EndToEndTest.java
index c2ec1d6291..63c9b42858 100644
--- a/qpid/java/perftests/src/test/java/org/apache/qpid/systest/disttest/endtoend/EndToEndTest.java
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/systest/disttest/endtoend/EndToEndTest.java
@@ -19,7 +19,7 @@
 package org.apache.qpid.systest.disttest.endtoend;
 
 import static org.apache.qpid.disttest.AbstractRunner.JNDI_CONFIG_PROP;
-import static org.apache.qpid.disttest.ControllerRunner.OUTPUT_FILE_PROP;
+import static org.apache.qpid.disttest.ControllerRunner.OUTPUT_DIR_PROP;
 import static org.apache.qpid.disttest.ControllerRunner.TEST_CONFIG_PROP;
 
 import java.io.File;
@@ -39,18 +39,19 @@ public class EndToEndTest extends QpidBrokerTestCase
 
     public void testRunner() throws Exception
     {
-        File csvOutputFile = createTemporaryCsvFile();
-        assertFalse("CSV output file must exist",csvOutputFile.exists());
+        File csvOutputDir = createTemporaryCsvDirectory();
+        assertTrue("CSV output dir must not exist",csvOutputDir.isDirectory());
 
         final String[] args = new String[] {TEST_CONFIG_PROP + "=" + TEST_CONFIG,
                                             JNDI_CONFIG_PROP + "=" + JNDI_CONFIG_FILE,
-                                            OUTPUT_FILE_PROP  + "=" + csvOutputFile.getAbsolutePath()};
+                                            OUTPUT_DIR_PROP  + "=" + csvOutputDir.getAbsolutePath()};
         _runner = new ControllerRunner();
         _runner.parseArgumentsIntoConfig(args);
         _runner.runController();
 
-        assertTrue("CSV output file must exist", csvOutputFile.exists());
-        final String csvContents = FileUtils.readFileAsString(csvOutputFile);
+        File expectedCsvOutputFile = new File(csvOutputDir, "endtoend.csv");
+        assertTrue("CSV output file must exist", expectedCsvOutputFile.exists());
+        final String csvContents = FileUtils.readFileAsString(expectedCsvOutputFile);
         final String[] csvLines = csvContents.split("\n");
 
         int numberOfHeaders = 1;
@@ -82,13 +83,13 @@ public class EndToEndTest extends QpidBrokerTestCase
 
     }
 
-    private File createTemporaryCsvFile() throws IOException
+    private File createTemporaryCsvDirectory() throws IOException
     {
-        File csvFile = File.createTempFile(getName(), ".csv");
-        csvFile.delete();
-        csvFile.deleteOnExit();
-
-        return csvFile;
+        String tmpDir = System.getProperty("java.io.tmpdir");
+        File csvDir = new File(tmpDir, "csv");
+        csvDir.mkdir();
+        csvDir.deleteOnExit();
+        return csvDir;
     }
 
 }