(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

5 files changed, 222 insertions, 51 deletions

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;

+    }

 }