From 19d23d9e5882d8a487fa03d561b80513c5c0b71d Mon Sep 17 00:00:00 2001
From: Robert Greig <rgreig@apache.org>
Date: Mon, 22 Jan 2007 17:30:29 +0000
Subject: (Patch submitted by Rupert Smith) Added configurations for all
 performance test setups to the pom. Commented out to not break build. Waiting
 on junit-toolkit-maven-plugin being added to maven repository.

Create a throttle utility class and tests for it.

git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/trunk/qpid@498720 13f79535-47bb-0310-9956-ffa450edef68
---
 java/perftests/pom.xml                             | 72 ++++++++++++++++---
 .../org/apache/qpid/ping/AbstractPingProducer.java |  3 +-
 .../main/java/org/apache/qpid/ping/Throttle.java   | 66 +++++++++++++++++
 .../java/org/apache/qpid/ping/PingTestPerf.java    | 83 ++++++++++++++--------
 .../org/apache/qpid/ping/ThrottleTestPerf.java     | 63 ++++++++++++++++
 .../apache/qpid/requestreply/PingPongTestPerf.java | 58 +++++++++------
 6 files changed, 283 insertions(+), 62 deletions(-)
 create mode 100644 java/perftests/src/main/java/org/apache/qpid/ping/Throttle.java
 create mode 100644 java/perftests/src/test/java/org/apache/qpid/ping/ThrottleTestPerf.java

(limited to 'java/perftests')

diff --git a/java/perftests/pom.xml b/java/perftests/pom.xml
index 1799b30790..e200212a61 100644
--- a/java/perftests/pom.xml
+++ b/java/perftests/pom.xml
@@ -87,7 +87,7 @@
 
 		 To run from the command line (after doing assembly:assembly goal):
 
-		 java -cp target/test_jar-jar-with-dependencies.jar uk.co.thebadgerset.junit.extensions.OldTKTestRunner -s 1 -r 100000 -o target org.apache.qpid.requestreply.PingPongTestPerf
+		 java -cp target/test_jar-jar-with-dependencies.jar uk.co.thebadgerset.junit.extensions.TKTestRunner -s 1 -r 100000 -o target org.apache.qpid.requestreply.PingPongTestPerf
 
 		 To generate the scripts do:
 
@@ -99,23 +99,73 @@
 
 		 These scripts can find everything in the 'all test dependencies' jar created by the assembly:assembly goal.
 		 -->	
-            <!--	 
+	    <!--
             <plugin>
               <groupId>uk.co.thebadgerset</groupId>
               <artifactId>junit-toolkit-maven-plugin</artifactId> 
               <version>0.3</version>
 
               <configuration>
+	        <scriptOutDirectory>target</scriptOutDirectory>
+	        <testJar>${project.build.finalName}-all-test-deps.jar</testJar>
                 <commands>
-                  <!## Run the ping pong test once. This is just to check toolkit test runner is working. Real tests to follow. ##>
-                  <param>-s 1 -r 1 -o target org.apache.qpid.requestreply.PingPongTestPerf</param>
-                  <!## Add more here... <param> ... </param> ##>
-
-		  <!## A little bit of work needs to be done on TKTestRunner before the following syntax works. See Javadoc for details. ##>
-		  <!## Thread ramp up: <param>-c [1,5,10,50,100] -r 1000 -o target org.apache.qpid.requestreply.PingPongTestPerf</param> ##>
-		  <!## Run for a length of time: <param>-d1H -o target org.apache.qpid.requestreply.PingPongTestPerf</param> ##>
-		  <!## Configure test parameters: <param>pingQueue=myping persistent=true transacted=true messageSize=1000 -d10M ... </param> ##>
-		  <!## And so on. ##>
+                  <!## Run the ping pong test once. This is just to check toolkit test runner is working. Real tests follow. ##>
+                  <PingOnce>-n PingOnce -s [1] -r 1 -t testPingOk -o . org.apache.qpid.ping.PingTestPerf</PingOnce>
+		  <ThrottleTest>-n ThrottleTest -s [10,10000],samples=8,exp -t testThrottle -o . org.apache.qpid.ping.ThrottleTestPerf</ThrottleTest>
+
+		  <!##
+		     Skim Tests.
+		     These are not part of the performance suite. They run quick tests to check that the different combinations of
+		     options that the performance suite uses are going to work. 
+		    ##>
+		  <Skim-Tx>-n Skim-Tx -s [1000] -o . -t testPingOk org.apache.qpid.ping.PingTestPerf transacted=true</Skim-Tx>
+		  <Skim-Size>-n Skim-Size -s [1000] -o . -t testPingOk org.apache.qpid.ping.PingTestPerf messageSize=51200</Skim-Size>
+		  <Skim-Many>-n Skim-Many -s [1] -c [4] -o . -t testPingOk org.apache.qpid.ping.PingTestPerf</Skim-Many>
+		  <Skim-Queues>-n Skim-Queues -s [1000] -o . -t testPingOk org.apache.qpid.ping.PingTestPerf numQueues=10</Skim-Queues>
+		  <Skim-Duration>-n Skim-Duration -s [1000] -d10S -o . -t testPingOk org.apache.qpid.ping.PingTestPerf</Skim-Duration>
+		  <Skim-Rate>-n Skim-Rate -s [1000] -d10S -o . -t testPingOk org.apache.qpid.ping.PingTestPerf numQueues=10 rate=100</Skim-Rate>
+
+		  <!## P2P Volume Tests. ##>
+		  <VQ-Qpid-1>-n VQ-Qpid-1 -s [900000] -o . -t testPingOk org.apache.qpid.ping.PingTestPerf messageSize=1024 transacted=true</VQ-Qpid-1>
+		  <VQ-Qpid-2>-n VQ-Qpid-2 -s [900000] -o . -t testPingOk org.apache.qpid.ping.PingTestPerf messageSize=1024</VQ-Qpid-2>
+		  <VQ-Qpid-3>-n VQ-Qpid-3 -s [10000] -D24H -o . -t testPingOk org.apache.qpid.ping.PingTestPerf messageSize=1024 rate=10000 transacted=true</VQ-Qpid-3>
+		  <VQ-Qpid-4>-n VQ-Qpid-4 -s [10000] -D24H -o . -t testPingOk org.apache.qpid.ping.PingTestPerf messageSize=1024 rate=10000</VQ-Qpid-4>
+
+		  <!## P2P Scalability Tests. ##>
+		  <PQ-Qpid-1>-n PQ-Qpid-1 -s [15000] -o . -t testPingOk org.apache.qpid.ping.PingTestPerf messageSize=1024 transacted=true</PQ-Qpid-1>
+		  <PQ-Qpid-2>-n PQ-Qpid-2 -s [15000] -o . -t testPingOk org.apache.qpid.ping.PingTestPerf messageSize=1024</PQ-Qpid-2>
+
+		  <!## 3 & 4 Need 10 pub/client but 1 queue... ##>
+
+		  <PQ-Qpid-5>-n PQ-Qpid-5 -s [15000] -c[10] -o . -t testPingOk org.apache.qpid.ping.PingTestPerf messageSize=1024 transacted=true</PQ-Qpid-5>
+		  <PQ-Qpid-6>-n PQ-Qpid-6 -s [15000] -c[10] -o . -t testPingOk org.apache.qpid.ping.PingTestPerf messageSize=1024</PQ-Qpid-6>
+		  <PQ-Qpid-7>-n PQ-Qpid-7 -s [15000] -c[10] -o . -t testPingOk org.apache.qpid.ping.PingTestPerf messageSize=1024 transacted=true numQueues=10</PQ-Qpid-7>
+		  <PQ-Qpid-8>-n PQ-Qpid-8 -s [15000] -c[10] -o . -t testPingOk org.apache.qpid.ping.PingTestPerf messageSize=1024 numQueue=10</PQ-Qpid-8>
+		  <PQ-Qpid-9>-n PQ-Qpid-9 -s [15000] -c[100] -o . -t testPingOk org.apache.qpid.ping.PingTestPerf messageSize=1024 transacted=true</PQ-Qpid-9>
+		  <PQ-Qpid-10>-n PQ-Qpid-10 -s [15000] -c[100] -o . -t testPingOk org.apache.qpid.ping.PingTestPerf messageSize=1024</PQ-Qpid-10>
+		  <PQ-Qpid-11>-n PQ-Qpid-11 -s [15000] -c[100] -o . -t testPingOk org.apache.qpid.ping.PingTestPerf messageSize=1024 transacted=true numQueues=10</PQ-Qpid-11>
+		  <PQ-Qpid-12>-n PQ-Qpid-12 -s [15000] -c[100] -o . -t testPingOk org.apache.qpid.ping.PingTestPerf messageSize=1024 numQueues=10</PQ-Qpid-12>
+		  <PQ-Qpid-13>-n PQ-Qpid-13 -s [15000] -c[1000] -D1M -o . -t testPingOk org.apache.qpid.ping.PingTestPerf messageSize=1024 transacted=true</PQ-Qpid-13>
+		  <PQ-Qpid-14>-n PQ-Qpid-14 -s [15000] -c[1000] -D1M -o . -t testPingOk org.apache.qpid.ping.PingTestPerf messageSize=1024</PQ-Qpid-14>
+
+		  <!## Increasing Message Payload Tests. ##>
+		  <LT-Qpid-3-521b>-n LT-Qpid-3-521b -s [900000] -o . -t testPingOk org.apache.qpid.ping.PingTestPerf messageSize=512 transacted=true</LT-Qpid-3-521b>
+		  <LT-Qpid-4-521b>-n LT-Qpid-4-521b -s [900000] -o . -t testPingOk org.apache.qpid.ping.PingTestPerf messageSize=512 transacted=true</LT-Qpid-4-521b>
+		  <LT-Qpid-3-1K>-n LT-Qpid-3-1K -s [900000] -o . -t testPingOk org.apache.qpid.ping.PingTestPerf messageSize=1024 transacted=true</LT-Qpid-3-1K>
+		  <LT-Qpid-4-1K>-n LT-Qpid-4-1K -s [900000] -o . -t testPingOk org.apache.qpid.ping.PingTestPerf messageSize=1024 transacted=true</LT-Qpid-4-1K>
+		  <LT-Qpid-3-5K>-n LT-Qpid-3-5K -s [900000] -o . -t testPingOk org.apache.qpid.ping.PingTestPerf messageSize=5120 transacted=true</LT-Qpid-3-5K>
+		  <LT-Qpid-4-5K>-n LT-Qpid-4-5K -s [900000] -o . -t testPingOk org.apache.qpid.ping.PingTestPerf messageSize=5120 transacted=true</LT-Qpid-4-5K>
+		  <LT-Qpid-3-10K>-n LT-Qpid-3-10K -s [900000] -o . -t testPingOk org.apache.qpid.ping.PingTestPerf messageSize=10240 transacted=true</LT-Qpid-3-10K>
+		  <LT-Qpid-4-10K>-n LT-Qpid-4-10K -s [900000] -o . -t testPingOk org.apache.qpid.ping.PingTestPerf messageSize=10240 transacted=true</LT-Qpid-4-10K>
+		  <LT-Qpid-3-50K>-n LT-Qpid-3-50K -s [900000] -o . -t testPingOk org.apache.qpid.ping.PingTestPerf messageSize=51200 transacted=true</LT-Qpid-3-50K>
+		  <LT-Qpid-4-50K>-n LT-Qpid-4-50K -s [900000] -o . -t testPingOk org.apache.qpid.ping.PingTestPerf messageSize=51200 transacted=true</LT-Qpid-4-50K>
+		  <LT-Qpid-3-1M>-n LT-Qpid-3-1M -s [900000] -o . -t testPingOk org.apache.qpid.ping.PingTestPerf messageSize=1048576 transacted=true</LT-Qpid-3-1M>
+		  <LT-Qpid-4-1M>-n LT-Qpid-4-1M -s [900000] -o . -t testPingOk org.apache.qpid.ping.PingTestPerf messageSize=1048576 transacted=true</LT-Qpid-4-1M>
+
+		  <!## Failover Tests. ##>
+		  <FT-Qpid-1>-n FT-Qpid-1 -s [250000] -o . -t testPingOk org.apache.qpid.ping.PingTestPerf messageSize=1024 transacted=true failover=true</FT-Qpid-1>
+		  <FT-Qpid-2>-n FT-Qpid-2 -s [250000] -o . -t testPingOk org.apache.qpid.ping.PingTestPerf messageSize=1024 transacted=true failover=true</FT-Qpid-2>
+
                 </commands>
               </configuration>
 
diff --git a/java/perftests/src/main/java/org/apache/qpid/ping/AbstractPingProducer.java b/java/perftests/src/main/java/org/apache/qpid/ping/AbstractPingProducer.java
index 513e1609aa..1877a23056 100644
--- a/java/perftests/src/main/java/org/apache/qpid/ping/AbstractPingProducer.java
+++ b/java/perftests/src/main/java/org/apache/qpid/ping/AbstractPingProducer.java
@@ -1,5 +1,6 @@
 package org.apache.qpid.ping;
 
+import java.text.DateFormat;
 import java.text.SimpleDateFormat;
 import java.util.List;
 import java.util.ArrayList;
@@ -35,7 +36,7 @@ public abstract class AbstractPingProducer implements Runnable, ExceptionListene
     private static final Logger _logger = Logger.getLogger(AbstractPingProducer.class);
 
     /** Used to format time stamping output. */
-    protected static final SimpleDateFormat timestampFormatter = new SimpleDateFormat("hh:mm:ss:SS");
+    protected static final DateFormat timestampFormatter = new SimpleDateFormat("hh:mm:ss:SS");
 
     /** Used to tell the ping loop when to terminate, it only runs while this is true. */
     protected boolean _publish = true;
diff --git a/java/perftests/src/main/java/org/apache/qpid/ping/Throttle.java b/java/perftests/src/main/java/org/apache/qpid/ping/Throttle.java
new file mode 100644
index 0000000000..9fb637149b
--- /dev/null
+++ b/java/perftests/src/main/java/org/apache/qpid/ping/Throttle.java
@@ -0,0 +1,66 @@
+package org.apache.qpid.ping;
+
+/**
+ * Throttle is a helper class used in situations where a controlled rate of processing is desired. It allows a certain
+ * number of operations-per-second to be defined and supplies a {@link #throttle} method that can only be called at
+ * most at that rate. The first call to the throttle method will return immediately, subsequent calls will introduce
+ * a short pause to fill out the remainder of the current cycle to attain the desired rate. If there is no remainder
+ * left then it will return immediately.
+ *
+ * <p><table id="crc"><caption>CRC Card</caption>
+ * <tr><th> Responsibilities <th> Collaborations
+ * </table>
+ *
+ * @author Rupert Smith
+ */
+public class Throttle
+{
+    /** Holds the length of the cycle in nano seconds. */
+    long cycleLengthNanos = 0L;
+
+    /** Records the nano time of the last call to the throttle method. */
+    long lastTimeNanos = 0L;
+
+    /**
+     * Sets up the desired rate of operation per second that the throttle method should restrict to.
+     *
+     * @param opsPerSecond The maximum number of calls per second that the throttle method will take.
+     */
+    public void setRate(int opsPerSecond)
+    {
+        // Calculate the length of a cycle.
+        cycleLengthNanos = 1000000000 / opsPerSecond;
+    }
+
+    /**
+     * Introduces a short pause to fill out any time left in the cycle since this method was last called, of length
+     * defined by a call to the {@link #setRate} method.
+     */
+    public void throttle()
+    {
+        // Record the time now.
+        long currentTimeNanos = System.nanoTime();
+
+        // Check if there is any time remaining in the current cycle and introduce a short wait to fill out the
+        // remainder of the cycle if needed.
+        long remainingTimeNanos = cycleLengthNanos - (currentTimeNanos - lastTimeNanos);
+
+        if (remainingTimeNanos > 0)
+        {
+            long milliWait = remainingTimeNanos / 1000000;
+            int nanoWait = (int) (remainingTimeNanos % 1000000);
+
+            try
+            {
+                Thread.currentThread().sleep(milliWait, nanoWait);
+            }
+            catch (InterruptedException e)
+            {
+                // Just ignore this?
+            }
+        }
+
+        // Keep the time of the last call to this method to calculate the next cycle.
+        lastTimeNanos = currentTimeNanos;
+    }
+}
diff --git a/java/perftests/src/test/java/org/apache/qpid/ping/PingTestPerf.java b/java/perftests/src/test/java/org/apache/qpid/ping/PingTestPerf.java
index af3accf530..ef34b92265 100644
--- a/java/perftests/src/test/java/org/apache/qpid/ping/PingTestPerf.java
+++ b/java/perftests/src/test/java/org/apache/qpid/ping/PingTestPerf.java
@@ -4,12 +4,12 @@ import java.util.Properties;
 
 import javax.jms.*;
 
+import junit.framework.Assert;
 import junit.framework.Test;
 import junit.framework.TestSuite;
-import junit.framework.Assert;
 
 import org.apache.log4j.Logger;
-import org.apache.log4j.NDC;
+
 import uk.co.thebadgerset.junit.extensions.AsymptoticTestCase;
 
 /**
@@ -98,8 +98,8 @@ public class PingTestPerf extends AsymptoticTestCase //implements TimingControll
         setSystemPropertyIfNull(PING_QUEUE_COUNT_PROPNAME, Integer.toString(1));
     }
 
-    /** Holds the test ping client. */
-    private TestPingItself _pingItselfClient;
+    /** Thread local to hold the per-thread test setup fields. */
+    ThreadLocal<PerThreadSetup> threadSetup = new ThreadLocal<PerThreadSetup>();
 
     // Set up a property reader to extract the test parameters from. Once ContextualProperties is available in
     // the project dependencies, use it to get property overrides for configurable tests and to notify the test runner
@@ -112,6 +112,21 @@ public class PingTestPerf extends AsymptoticTestCase //implements TimingControll
         super(name);
     }
 
+    /**
+     * Compile all the tests into a test suite.
+     */
+    public static Test suite()
+    {
+        // Build a new test suite
+        TestSuite suite = new TestSuite("Ping Performance Tests");
+
+        // Run performance tests in read committed mode.
+        suite.addTest(new PingTestPerf("testPingOk"));
+
+        return suite;
+               //return new junit.framework.TestSuite(PingTestPerf.class);
+    }
+
     private static void setSystemPropertyIfNull(String propName, String propValue)
     {
         if (System.getProperty(propName) == null)
@@ -122,15 +137,23 @@ public class PingTestPerf extends AsymptoticTestCase //implements TimingControll
 
     public void testPingOk(int numPings) throws Exception
     {
+        // Get the per thread test setup to run the test through.
+        PerThreadSetup perThreadSetup = threadSetup.get();
+
         // Generate a sample message. This message is already time stamped and has its reply-to destination set.
-        ObjectMessage msg = _pingItselfClient.getTestMessage(null,
-                                             Integer.parseInt(testParameters.getProperty(MESSAGE_SIZE_PROPNAME)),
-                                             Boolean.parseBoolean(testParameters.getProperty(PERSISTENT_MODE_PROPNAME)));
+        ObjectMessage msg =
+            perThreadSetup._pingItselfClient.getTestMessage(null,
+                                                            Integer.parseInt(testParameters.getProperty(
+                                                                                 MESSAGE_SIZE_PROPNAME)),
+                                                            Boolean.parseBoolean(testParameters.getProperty(
+                                                                                     PERSISTENT_MODE_PROPNAME)));
 
         // start the test
         long timeout = Long.parseLong(testParameters.getProperty(TIMEOUT_PROPNAME));
-        int numReplies = _pingItselfClient.pingAndWaitForReply(msg, numPings, timeout);
+        int numReplies = perThreadSetup._pingItselfClient.pingAndWaitForReply(msg, numPings, timeout);
+
         _logger.info("replies = " + numReplies);
+
         // Fail the test if the timeout was exceeded.
         if (numReplies != numPings)
         {
@@ -141,11 +164,14 @@ public class PingTestPerf extends AsymptoticTestCase //implements TimingControll
     protected void setUp() throws Exception
     {
         // Log4j will propagate the test name as a thread local in all log output.
-        NDC.push(getName());
+        // Carefull when using this, it can cause memory leaks when not cleaned up properly.
+        //NDC.push(getName());
 
-        // Ensure that the connection, session and ping queue are established, if they have not already been.
-        if (_pingItselfClient == null)
+        // Create the test setups on a per thread basis, only if they have not already been created.
+        if (threadSetup.get() == null)
         {
+            PerThreadSetup perThreadSetup = new PerThreadSetup();
+
             // Extract the test set up paramaeters.
             String brokerDetails = testParameters.getProperty(BROKER_PROPNAME);
             String username = "guest";
@@ -159,21 +185,27 @@ public class PingTestPerf extends AsymptoticTestCase //implements TimingControll
             boolean verbose = false;
             int messageSize = Integer.parseInt(testParameters.getProperty(MESSAGE_SIZE_PROPNAME));
 
+            // Establish a client to ping a Queue and listen the reply back from same Queue
             if (queueCount > 1)
             {
                 // test client with multiple queues
-                _pingItselfClient = new TestPingItself(brokerDetails, username, password, virtualpath, queueCount,
-                                                       selector, transacted, persistent, messageSize, verbose);
+                perThreadSetup._pingItselfClient = new TestPingItself(brokerDetails, username, password, virtualpath,
+                                                                      queueCount, selector, transacted, persistent,
+                                                                      messageSize, verbose);
             }
             else
             {
                 // Establish a client to ping a Queue and listen the reply back from same Queue
-                _pingItselfClient = new TestPingItself(brokerDetails, username, password, virtualpath, queueName,
-                                                       selector, transacted, persistent, messageSize, verbose);
+                perThreadSetup._pingItselfClient = new TestPingItself(brokerDetails, username, password, virtualpath,
+                                                                      queueName, selector, transacted, persistent,
+                                                                      messageSize, verbose);
             }
 
             // Start the client connection
-            _pingItselfClient.getConnection().start();
+            perThreadSetup._pingItselfClient.getConnection().start();
+
+            // Attach the per-thread set to the thread.
+            threadSetup.set(perThreadSetup);
         }
     }
 
@@ -190,22 +222,13 @@ public class PingTestPerf extends AsymptoticTestCase //implements TimingControll
         }
         finally
         {
-            NDC.pop();
+            //NDC.pop();
         }
     }
 
-    /**
-     * Compile all the tests into a test suite.
-     */
-    public static Test suite()
-    {      
-        // Build a new test suite
-        TestSuite suite = new TestSuite("Ping Performance Tests");
-
-        // Run performance tests in read committed mode.
-        suite.addTest(new PingTestPerf("testPingOk"));
-
-        return suite;
-        //return new junit.framework.TestSuite(PingTestPerf.class);
+    private static class PerThreadSetup
+    {
+        /** Holds the test ping client. */
+        private TestPingItself _pingItselfClient;
     }
 }
diff --git a/java/perftests/src/test/java/org/apache/qpid/ping/ThrottleTestPerf.java b/java/perftests/src/test/java/org/apache/qpid/ping/ThrottleTestPerf.java
new file mode 100644
index 0000000000..274c8b5fc8
--- /dev/null
+++ b/java/perftests/src/test/java/org/apache/qpid/ping/ThrottleTestPerf.java
@@ -0,0 +1,63 @@
+package org.apache.qpid.ping;
+
+import junit.framework.Test;
+import junit.framework.TestSuite;
+
+import uk.co.thebadgerset.junit.extensions.AsymptoticTestCase;
+
+/**
+ * Tests the {@link Throttle} implementation. Test timings can be taken using this test class to confirm that the
+ * throttle works as it should, and what the maximum rate is that it works reliably.
+ *
+ * <p><table id="crc"><caption>CRC Card</caption>
+ * <tr><th> Responsibilities <th> Collaborations
+ * <tr><td> Enable test timings to be taken to confirm that the throttle works at the correct rate.
+ * </table>
+ *
+ * @author Rupert Smith
+ */
+public class ThrottleTestPerf extends AsymptoticTestCase
+{
+    ThreadLocal<Throttle> threadSetup = new ThreadLocal<Throttle>();
+
+    public ThrottleTestPerf(String name)
+    {
+        super(name);
+    }
+
+    /**
+     * Compile all the tests into a test suite.
+     */
+    public static Test suite()
+    {
+        // Build a new test suite
+        TestSuite suite = new TestSuite("Ping-Pong Performance Tests");
+
+        // Run performance tests in read committed mode.
+        suite.addTest(new ThrottleTestPerf("testThrottle"));
+
+        return suite;
+    }
+
+    public void testThrottle(int opsPerSecond)
+    {
+        Throttle throttle = threadSetup.get();
+
+        // Setting this on every test call won't cause any harm, convenient to use the size parameter for this.
+        throttle.setRate(opsPerSecond);
+
+        // Run the test at the throttled rate, do this for the num of opsPerSecond, then every test should take 1 second.
+        for (int i = 0; i < opsPerSecond; i++)
+        {
+            throttle.throttle();
+        }
+    }
+
+    protected void setUp()
+    {
+        if (threadSetup.get() == null)
+        {
+            threadSetup.set(new Throttle());
+        }
+    }
+}
diff --git a/java/perftests/src/test/java/org/apache/qpid/requestreply/PingPongTestPerf.java b/java/perftests/src/test/java/org/apache/qpid/requestreply/PingPongTestPerf.java
index bd1c1baad4..f553faf302 100644
--- a/java/perftests/src/test/java/org/apache/qpid/requestreply/PingPongTestPerf.java
+++ b/java/perftests/src/test/java/org/apache/qpid/requestreply/PingPongTestPerf.java
@@ -81,7 +81,7 @@ public class PingPongTestPerf extends AsymptoticTestCase //implements TimingCont
     private static final String VIRTUAL_PATH_DEFAULT = "/test";
 
     /** Sets a default ping timeout. */
-    private static final long TIMEOUT = 5000;
+    private static final long TIMEOUT = 15000;
 
     // Sets up the test parameters with defaults.
     static
@@ -94,11 +94,8 @@ public class PingPongTestPerf extends AsymptoticTestCase //implements TimingCont
         setSystemPropertyIfNull(VIRTUAL_PATH_PROPNAME, VIRTUAL_PATH_DEFAULT);
     }
 
-    /** Holds the test ping-pong producer. */
-    private PingPongProducer _testPingProducer;
-
-    /** Holds the test ping client. */
-    private PingPongBouncer _testPingBouncer;
+    /** Thread local to hold the per-thread test setup fields. */
+    ThreadLocal<PerThreadSetup> threadSetup = new ThreadLocal<PerThreadSetup>();
 
     // Set up a property reader to extract the test parameters from. Once ContextualProperties is available in
     // the project dependencies, use it to get property overrides for configurable tests and to notify the test runner
@@ -135,11 +132,16 @@ public class PingPongTestPerf extends AsymptoticTestCase //implements TimingCont
 
     public void testPingPongOk(int numPings) throws Exception
     {
+        // Get the per thread test setup to run the test through.
+        PerThreadSetup perThreadSetup = threadSetup.get();
+
         // Generate a sample message. This message is already time stamped and has its reply-to destination set.
         ObjectMessage msg =
-            _testPingProducer.getTestMessage(_testPingProducer.getReplyQueue(),
-                                             Integer.parseInt(testParameters.getProperty(MESSAGE_SIZE_PROPNAME)),
-                                             Boolean.parseBoolean(testParameters.getProperty(PERSISTENT_MODE_PROPNAME)));
+            perThreadSetup._testPingProducer.getTestMessage(perThreadSetup._testPingProducer.getReplyQueue(),
+                                                            Integer.parseInt(testParameters.getProperty(
+                                                                                 MESSAGE_SIZE_PROPNAME)),
+                                                            Boolean.parseBoolean(testParameters.getProperty(
+                                                                                     PERSISTENT_MODE_PROPNAME)));
 
         // Use the test timing controller to reset the test timer now and obtain the current time.
         // This can be used to remove the message creation time from the test.
@@ -147,7 +149,7 @@ public class PingPongTestPerf extends AsymptoticTestCase //implements TimingCont
         //long startTime = timingUtils.restart();
 
         // Send the message and wait for a reply.
-        int numReplies = _testPingProducer.pingAndWaitForReply(msg, numPings, TIMEOUT);
+        int numReplies = perThreadSetup._testPingProducer.pingAndWaitForReply(msg, numPings, TIMEOUT);
 
         // Fail the test if the timeout was exceeded.
         if (numReplies != numPings)
@@ -159,11 +161,14 @@ public class PingPongTestPerf extends AsymptoticTestCase //implements TimingCont
     protected void setUp() throws Exception
     {
         // Log4j will propagate the test name as a thread local in all log output.
-        NDC.push(getName());
+        // Carefull when using this, it can cause memory leaks when not cleaned up properly.
+        //NDC.push(getName());
 
-        // Ensure that the connection, session and ping queue are established, if they have not already been.
-        if (_testPingProducer == null)
+        // Create the test setups on a per thread basis, only if they have not already been created.
+        if (threadSetup.get() == null)
         {
+            PerThreadSetup perThreadSetup = new PerThreadSetup();
+
             // Extract the test set up paramaeters.
             String brokerDetails = testParameters.getProperty(BROKER_PROPNAME);
             String username = "guest";
@@ -177,17 +182,21 @@ public class PingPongTestPerf extends AsymptoticTestCase //implements TimingCont
             int messageSize = Integer.parseInt(testParameters.getProperty(MESSAGE_SIZE_PROPNAME));
 
             // Establish a bounce back client on the ping queue to bounce back the pings.
-            _testPingBouncer = new PingPongBouncer(brokerDetails, username, password, virtualpath, queueName, persistent,
-                                                   transacted, selector, verbose);
+            perThreadSetup._testPingBouncer = new PingPongBouncer(brokerDetails, username, password, virtualpath, queueName,
+                                                                  persistent, transacted, selector, verbose);
 
             // Start the connections for client and producer running.
-            _testPingBouncer.getConnection().start();
+            perThreadSetup._testPingBouncer.getConnection().start();
 
             // Establish a ping-pong client on the ping queue to send the pings with.
-            _testPingProducer = new PingPongProducer(brokerDetails, username, password, virtualpath, queueName, selector,
-                                                     transacted, persistent, messageSize, verbose);
+            perThreadSetup._testPingProducer = new PingPongProducer(brokerDetails, username, password, virtualpath,
+                                                                    queueName, selector, transacted, persistent, messageSize,
+                                                                    verbose);
+
+            perThreadSetup._testPingProducer.getConnection().start();
 
-            _testPingProducer.getConnection().start();
+            // Attach the per-thread set to the thread.
+            threadSetup.set(perThreadSetup);
         }
     }
 
@@ -207,7 +216,16 @@ public class PingPongTestPerf extends AsymptoticTestCase //implements TimingCont
         }
         finally
         {
-            NDC.pop();
+            //NDC.pop();
         }
     }
+
+    private static class PerThreadSetup
+    {
+        /** Holds the test ping-pong producer. */
+        private PingPongProducer _testPingProducer;
+
+        /** Holds the test ping client. */
+        private PingPongBouncer _testPingBouncer;
+    }
 }
-- 
cgit v1.2.1