You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by ri...@apache.org on 2007/01/23 10:39:57 UTC
svn commit: r498965 - in /incubator/qpid/trunk/qpid/java/perftests:
distribution/ distribution/src/main/assembly/
src/main/java/org/apache/qpid/ping/
src/main/java/org/apache/qpid/requestreply/
src/test/java/org/apache/qpid/ping/ src/test/java/org/apac...
Author: ritchiem
Date: Tue Jan 23 01:39:56 2007
New Revision: 498965
URL: http://svn.apache.org/viewvc?view=rev&rev=498965
Log:
Added ability to cause failover before/after commit/sends
Added batch size ability.
Modified:
incubator/qpid/trunk/qpid/java/perftests/distribution/pom.xml
incubator/qpid/trunk/qpid/java/perftests/distribution/src/main/assembly/performance.xml
incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/ping/AbstractPingClient.java
incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/ping/AbstractPingProducer.java
incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/ping/TestPingClient.java
incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/ping/TestPingItself.java
incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/ping/TestPingProducer.java
incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/requestreply/PingPongProducer.java
incubator/qpid/trunk/qpid/java/perftests/src/test/java/org/apache/qpid/ping/PingTestPerf.java
incubator/qpid/trunk/qpid/java/perftests/src/test/java/org/apache/qpid/requestreply/PingPongTestPerf.java
Modified: incubator/qpid/trunk/qpid/java/perftests/distribution/pom.xml
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/perftests/distribution/pom.xml?view=diff&rev=498965&r1=498964&r2=498965
==============================================================================
--- incubator/qpid/trunk/qpid/java/perftests/distribution/pom.xml (original)
+++ incubator/qpid/trunk/qpid/java/perftests/distribution/pom.xml Tue Jan 23 01:39:56 2007
@@ -44,8 +44,22 @@
<dependency>
<groupId>org.apache.qpid</groupId>
<artifactId>qpid-perftests</artifactId>
+ <type>jar</type>
+ <version>${pom.version}</version>
</dependency>
- </dependencies>
+ <dependency>
+ <groupId>org.apache.qpid</groupId>
+ <artifactId>qpid-perftests</artifactId>
+ <type>test-jar</type>
+ <version>${pom.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>uk.co.thebadgerset</groupId>
+ <artifactId>junit-toolkit</artifactId>
+ <version>0.4</version>
+ <scope>runtime</scope>
+ </dependency>
+ </dependencies>
<build>
<pluginManagement>
Modified: incubator/qpid/trunk/qpid/java/perftests/distribution/src/main/assembly/performance.xml
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/perftests/distribution/src/main/assembly/performance.xml?view=diff&rev=498965&r1=498964&r2=498965
==============================================================================
--- incubator/qpid/trunk/qpid/java/perftests/distribution/src/main/assembly/performance.xml (original)
+++ incubator/qpid/trunk/qpid/java/perftests/distribution/src/main/assembly/performance.xml Tue Jan 23 01:39:56 2007
@@ -74,6 +74,13 @@
<include>**/*.log4j</include>
</includes>
</fileSet>
+ <fileSet>
+ <directory>../src/test</directory>
+ <outputDirectory>qpid-${qpid.version}/src</outputDirectory>
+ <includes>
+ <include>**/*.java</include>
+ </includes>
+ </fileSet>
<!-- Metadata Jar -->
<fileSet>
Modified: incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/ping/AbstractPingClient.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/ping/AbstractPingClient.java?view=diff&rev=498965&r1=498964&r2=498965
==============================================================================
--- incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/ping/AbstractPingClient.java (original)
+++ incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/ping/AbstractPingClient.java Tue Jan 23 01:39:56 2007
@@ -1,6 +1,7 @@
package org.apache.qpid.ping;
import java.text.SimpleDateFormat;
+import java.io.IOException;
import javax.jms.Connection;
import javax.jms.JMSException;
@@ -13,7 +14,7 @@
/**
* Provides functionality common to all ping clients. Provides the ability to manage a session and a convenience method
* to commit on the current transaction.
- *
+ * <p/>
* <p><table id="crc"><caption>CRC Card</caption>
* <tr><th> Responsibilities <th> Collaborations
* <tr><td> Commit the current transcation.
@@ -29,6 +30,9 @@
private static final Logger _logger = Logger.getLogger(TestPingClient.class);
private AMQConnection _connection;
+ protected boolean _failBeforeCommit = false;
+ protected boolean _failAfterCommit = false;
+
public AMQConnection getConnection()
{
return _connection;
@@ -50,7 +54,20 @@
{
try
{
+ if (_failBeforeCommit)
+ {
+ _logger.trace("Failing Before Commit");
+ doFailover();
+ }
+
session.commit();
+
+ if (_failAfterCommit)
+ {
+ _logger.trace("Failing After Commit");
+ doFailover();
+ }
+
_logger.trace("Session Commited.");
}
catch (JMSException e)
@@ -72,4 +89,32 @@
}
}
}
+
+ protected void doFailover(String broker)
+ {
+ System.out.println("Kill Broker " + broker + " now.");
+ try
+ {
+ System.in.read();
+ }
+ catch (IOException e)
+ {
+ }
+ System.out.println("Continuing.");
+ }
+
+ protected void doFailover()
+ {
+ System.out.println("Kill Broker now.");
+ try
+ {
+ System.in.read();
+ }
+ catch (IOException e)
+ {
+ }
+ System.out.println("Continuing.");
+
+ }
+
}
Modified: incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/ping/AbstractPingProducer.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/ping/AbstractPingProducer.java?view=diff&rev=498965&r1=498964&r2=498965
==============================================================================
--- incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/ping/AbstractPingProducer.java (original)
+++ incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/ping/AbstractPingProducer.java Tue Jan 23 01:39:56 2007
@@ -2,17 +2,23 @@
import java.text.DateFormat;
import java.text.SimpleDateFormat;
+import java.io.IOException;
import java.util.List;
import java.util.ArrayList;
import java.util.concurrent.atomic.AtomicInteger;
import javax.jms.*;
+import javax.jms.Connection;
+import javax.jms.Message;
+import javax.jms.MessageProducer;
import org.apache.log4j.Logger;
import org.apache.qpid.client.AMQNoConsumersException;
+import org.apache.qpid.client.BasicMessageProducer;
import org.apache.qpid.client.AMQQueue;
import org.apache.qpid.client.message.TestMessageFactory;
+import org.apache.qpid.jms.*;
import org.apache.qpid.jms.Session;
import org.apache.qpid.framing.AMQShortString;
@@ -20,7 +26,7 @@
* This abstract class captures functionality that is common to all ping producers. It provides functionality to
* manage a session, and a convenience method to commit a transaction on the session. It also provides a framework
* for running a ping loop, and terminating that loop on exceptions or a shutdown handler.
- *
+ * <p/>
* <p><table id="crc"><caption>CRC Card</caption>
* <tr><th> Responsibilities <th> Collaborations
* <tr><td> Manage the connection.
@@ -35,25 +41,49 @@
{
private static final Logger _logger = Logger.getLogger(AbstractPingProducer.class);
- /** Used to format time stamping output. */
+ /**
+ * Used to format time stamping output.
+ */
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. */
+ /**
+ * Used to tell the ping loop when to terminate, it only runs while this is true.
+ */
protected boolean _publish = true;
- /** Holds the connection handle to the broker. */
+ /**
+ * Holds the connection handle to the broker.
+ */
private Connection _connection;
- /** Holds the producer session, need to create test messages. */
+ /**
+ * Holds the producer session, need to create test messages.
+ */
private Session _producerSession;
- /** holds the no of queues the tests will be using to send messages. By default it will be 1 */
- private int _queueCount;
+ /**
+ * holds the no of queues the tests will be using to send messages. By default it will be 1
+ */
+ protected int _queueCount;
private static AtomicInteger _queueSequenceID = new AtomicInteger();
private List<Queue> _queues = new ArrayList<Queue>();
/**
+ * Holds the message producer to send the pings through.
+ */
+ protected org.apache.qpid.jms.MessageProducer _producer;
+
+ protected boolean _failBeforeCommit = false;
+ protected boolean _failAfterCommit = false;
+ protected boolean _failBeforeSend = false;
+ protected boolean _failAfterSend = false;
+
+ protected int _sentMessages = 0;
+ protected int _batchSize = 1;
+
+
+ /**
* Convenience method for a short pause.
*
* @param sleepTime The time in milliseconds to pause for.
@@ -67,7 +97,8 @@
Thread.sleep(sleepTime);
}
catch (InterruptedException ie)
- { }
+ {
+ }
}
}
@@ -78,9 +109,7 @@
*
* @param replyQueue The reply-to destination for the message.
* @param messageSize The desired size of the message in bytes.
- *
* @return A freshly generated test message.
- *
* @throws javax.jms.JMSException All underlying JMSException are allowed to fall through.
*/
public ObjectMessage getTestMessage(Queue replyQueue, int messageSize, boolean persistent) throws JMSException
@@ -153,12 +182,12 @@
public Thread getShutdownHook()
{
return new Thread(new Runnable()
+ {
+ public void run()
{
- public void run()
- {
- stop();
- }
- });
+ stop();
+ }
+ });
}
public Connection getConnection()
@@ -181,6 +210,12 @@
this._producerSession = session;
}
+
+ protected void commitTx() throws JMSException
+ {
+ commitTx(getProducerSession());
+ }
+
public int getQueueCount()
{
return _queueCount;
@@ -194,6 +229,7 @@
/**
* Creates queues dynamically and adds to the queues list. This is when the test is being done with
* multiple queues.
+ *
* @param queueCount
*/
protected void createQueues(int queueCount)
@@ -202,7 +238,7 @@
{
AMQShortString name = new AMQShortString("Queue_" + _queueSequenceID.incrementAndGet() + "_" + System.currentTimeMillis());
AMQQueue queue = new AMQQueue(name, name, false, false, false);
-
+
_queues.add(queue);
}
}
@@ -219,36 +255,107 @@
*/
protected void commitTx(Session session) throws JMSException
{
- if (session.getTransacted())
+ if ((++_sentMessages % _batchSize) == 0)
{
- try
- {
- session.commit();
- _logger.trace("Session Commited.");
- }
- catch (JMSException e)
+ if (session.getTransacted())
{
- _logger.trace("JMSException on commit:" + e.getMessage(), e);
-
- // Warn that the bounce back client is not available.
- if (e.getLinkedException() instanceof AMQNoConsumersException)
- {
- _logger.debug("No consumers on queue.");
- }
-
try
{
- session.rollback();
- _logger.trace("Message rolled back.");
+ if (_failBeforeCommit)
+ {
+ _logger.trace("Failing Before Commit");
+ doFailover();
+ }
+
+ session.commit();
+
+ if (_failAfterCommit)
+ {
+ _logger.trace("Failing After Commit");
+ doFailover();
+ }
+ _logger.trace("Session Commited.");
}
- catch (JMSException jmse)
+ catch (JMSException e)
{
- _logger.trace("JMSE on rollback:" + jmse.getMessage(), jmse);
+ _logger.trace("JMSException on commit:" + e.getMessage(), e);
- // Both commit and rollback failed. Throw the rollback exception.
- throw jmse;
+ // Warn that the bounce back client is not available.
+ if (e.getLinkedException() instanceof AMQNoConsumersException)
+ {
+ _logger.debug("No consumers on queue.");
+ }
+
+ try
+ {
+ session.rollback();
+ _logger.trace("Message rolled back.");
+ }
+ catch (JMSException jmse)
+ {
+ _logger.trace("JMSE on rollback:" + jmse.getMessage(), jmse);
+
+ // Both commit and rollback failed. Throw the rollback exception.
+ throw jmse;
+ }
}
}
}
+ }
+
+ protected void sendMessage(Message message) throws JMSException
+ {
+ sendMessage(null, message);
+ }
+
+ protected void sendMessage(Queue q, Message message) throws JMSException
+ {
+ if (_failBeforeSend)
+ {
+ _logger.trace("Failing Before Send");
+ doFailover();
+ }
+
+ if (q == null)
+ {
+ _producer.send(message);
+ }
+ else
+ {
+ _producer.send(q, message);
+ }
+
+ if (_failAfterSend)
+ {
+ _logger.trace("Failing After Send");
+ doFailover();
+ }
+ }
+
+ protected void doFailover(String broker)
+ {
+ System.out.println("Kill Broker " + broker + " now then press return");
+ try
+ {
+ System.in.read();
+ }
+ catch (IOException e)
+ {
+ }
+ System.out.println("Continuing.");
+ }
+
+ protected void doFailover()
+ {
+ System.out.println("Kill Broker now then press return");
+ try
+ {
+ System.in.read();
+ }
+ catch (IOException e)
+ {
+ }
+ System.out.println("Continuing.");
+
}
}
Modified: incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/ping/TestPingClient.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/ping/TestPingClient.java?view=diff&rev=498965&r1=498964&r2=498965
==============================================================================
--- incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/ping/TestPingClient.java (original)
+++ incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/ping/TestPingClient.java Tue Jan 23 01:39:56 2007
@@ -34,11 +34,11 @@
* PingClient is a message listener that received time stamped ping messages. It can work out how long a ping took,
* provided that its clokc is synchronized to that of the ping producer, or by running it on the same machine (or jvm)
* as the ping producer.
- *
+ * <p/>
* <p/>There is a verbose mode flag which causes information about each ping to be output to the console
* (info level logging, so usually console). This can be helpfull to check the bounce backs are happening but should
* be disabled for real timing tests as writing to the console will slow things down.
- *
+ * <p/>
* <p><table id="crc"><caption>CRC Card</caption>
* <tr><th> Responsibilities <th> Collaborations
* <tr><td> Provide command line invocation to start the ping consumer on a configurable broker url.
@@ -50,10 +50,14 @@
{
private static final Logger _logger = Logger.getLogger(TestPingClient.class);
- /** Used to indicate that the reply generator should log timing info to the console (logger info level). */
+ /**
+ * Used to indicate that the reply generator should log timing info to the console (logger info level).
+ */
private boolean _verbose = false;
- /** The producer session. */
+ /**
+ * The producer session.
+ */
private Session _consumerSession;
/**
@@ -67,11 +71,11 @@
* @param transacted
* @param selector
* @param verbose
- *
- * @throws Exception All underlying exceptions allowed to fall through. This is only test code...
+ * @param afterCommit
+ *@param beforeCommit @throws Exception All underlying exceptions allowed to fall through. This is only test code...
*/
public TestPingClient(String brokerDetails, String username, String password, String queueName, String virtualpath,
- boolean transacted, String selector, boolean verbose) throws Exception
+ boolean transacted, String selector, boolean verbose, boolean afterCommit, boolean beforeCommit) throws Exception
{
// Create a connection to the broker.
InetAddress address = InetAddress.getLocalHost();
@@ -85,10 +89,15 @@
// Connect a consumer to the ping queue and register this to be called back by it.
Queue q = new AMQQueue(queueName);
MessageConsumer consumer = _consumerSession.createConsumer(q, 1, false, false, selector);
+
consumer.setMessageListener(this);
// Hang on to the verbose flag setting.
_verbose = verbose;
+
+ // Set failover interrupts
+ _failAfterCommit = afterCommit;
+ _failBeforeCommit = beforeCommit;
}
/**
@@ -104,7 +113,7 @@
if (args.length < 4)
{
System.out.println(
- "Usage: brokerdetails username password virtual-path [queueName] [verbose] [transacted] [selector]");
+ "Usage: brokerdetails username password virtual-path [queueName] [verbose] [transacted] [selector] [failover:<before|after>:commit]");
System.exit(1);
}
@@ -118,14 +127,38 @@
boolean transacted = (args.length >= 7) ? Boolean.parseBoolean(args[6]) : false;
String selector = (args.length == 8) ? args[7] : null;
+ boolean afterCommit = false;
+ boolean beforeCommit = false;
+
+ for (String arg : args)
+ {
+ if (arg.startsWith("failover:"))
+ {
+ //failover:<before|after>:<send:commit>
+ String[] parts = arg.split(":");
+ if (parts.length == 3)
+ {
+ if (parts[2].equals("commit"))
+ {
+ afterCommit = parts[1].equals("after");
+ beforeCommit = parts[1].equals("before");
+ }
+ }
+ else
+ {
+ System.out.println("Unrecognized failover request:" + arg);
+ }
+ }
+ }
+
// Create the test ping client and set it running.
TestPingClient pingClient =
- new TestPingClient(brokerDetails, username, password, queueName, virtualpath, transacted, selector, verbose);
+ new TestPingClient(brokerDetails, username, password, queueName, virtualpath, transacted, selector, verbose, afterCommit, beforeCommit);
+
pingClient.getConnection().start();
System.out.println("Waiting...");
}
-
/**
* This is a callback method that is notified of all messages for which this has been registered as a message
* listener on a message consumer.
Modified: incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/ping/TestPingItself.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/ping/TestPingItself.java?view=diff&rev=498965&r1=498964&r2=498965
==============================================================================
--- incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/ping/TestPingItself.java (original)
+++ incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/ping/TestPingItself.java Tue Jan 23 01:39:56 2007
@@ -45,18 +45,24 @@
private static final Logger _logger = Logger.getLogger(TestPingItself.class);
public TestPingItself(String brokerDetails, String username, String password, String virtualpath, String queueName,
- String selector, boolean transacted, boolean persistent, int messageSize, boolean verbose)
+ String selector, boolean transacted, boolean persistent, int messageSize, boolean verbose,
+ boolean afterCommit, boolean beforeCommit, boolean afterSend, boolean beforeSend,
+ int batchSize)
throws Exception
{
- super(brokerDetails, username, password, virtualpath, queueName, selector, transacted, persistent, messageSize, verbose);
+ super(brokerDetails, username, password, virtualpath, queueName, selector, transacted, persistent, messageSize,
+ verbose, afterCommit, beforeCommit, afterSend, beforeSend, batchSize, 0);
}
- public TestPingItself(String brokerDetails, String username, String password, String virtualpath, int queueCount,
- String selector, boolean transacted, boolean persistent, int messageSize, boolean verbose)
+ public TestPingItself(String brokerDetails, String username, String password, String virtualpath,
+ String selector, boolean transacted, boolean persistent, int messageSize, boolean verbose,
+ boolean afterCommit, boolean beforeCommit, boolean afterSend, boolean beforeSend,
+ int batchSize, int queueCount)
throws Exception
{
- super(brokerDetails, username, password, virtualpath, transacted);
- setQueueCount(queueCount);
+ super(brokerDetails, username, password, virtualpath, null, null, transacted, persistent, messageSize,
+ verbose, afterCommit, beforeCommit, afterSend, beforeSend, batchSize, queueCount);
+
createQueues(queueCount);
_persistent = persistent;
@@ -82,7 +88,7 @@
/**
* Starts a ping-pong loop running from the command line. The bounce back client {@link org.apache.qpid.requestreply.PingPongBouncer} also needs
* to be started to bounce the pings back again.
- *
+ * <p/>
* <p/>The command line takes from 2 to 4 arguments:
* <p/><table>
* <tr><td>brokerDetails <td> The broker connection string.
@@ -99,7 +105,7 @@
if (args.length < 2)
{
System.err.println("Usage: TestPingPublisher <brokerDetails> <virtual path> [verbose (true/false)] " +
- "[transacted (true/false)] [persistent (true/false)] [message size in bytes]");
+ "[transacted (true/false)] [persistent (true/false)] [message size in bytes]");
System.exit(0);
}
@@ -109,14 +115,50 @@
boolean transacted = (args.length >= 4) ? Boolean.parseBoolean(args[3]) : false;
boolean persistent = (args.length >= 5) ? Boolean.parseBoolean(args[4]) : false;
int messageSize = (args.length >= 6) ? Integer.parseInt(args[5]) : DEFAULT_MESSAGE_SIZE;
+ int batchSize = (args.length >= 7) ? Integer.parseInt(args[6]) : 1;
- String queue = "ping_"+ System.currentTimeMillis();
- _logger.info("Queue:" + queue + ", Transacted:" + transacted + ", persistent:"+ persistent +
+ String queue = "ping_" + System.currentTimeMillis();
+ _logger.info("Queue:" + queue + ", Transacted:" + transacted + ", persistent:" + persistent +
",MessageSize:" + messageSize + " bytes");
+
+ boolean afterCommit = false;
+ boolean beforeCommit = false;
+ boolean afterSend = false;
+ boolean beforeSend = false;
+
+ for (String arg : args)
+ {
+ if (arg.startsWith("failover:"))
+ {
+ //failover:<before|after>:<send:commit>
+ String[] parts = arg.split(":");
+ if (parts.length == 3)
+ {
+ if (parts[2].equals("commit"))
+ {
+ afterCommit = parts[1].equals("after");
+ beforeCommit = parts[1].equals("before");
+ }
+
+ if (parts[2].equals("send"))
+ {
+ afterSend = parts[1].equals("after");
+ beforeSend = parts[1].equals("before");
+ }
+ }
+ else
+ {
+ System.out.println("Unrecognized failover request:" + arg);
+ }
+ }
+ }
+
// Create a ping producer to handle the request/wait/reply cycle.
TestPingItself pingItself = new TestPingItself(brokerDetails, "guest", "guest", virtualpath, queue, null,
- transacted, persistent, messageSize, verbose);
+ transacted, persistent, messageSize, verbose,
+ afterCommit, beforeCommit, afterSend, beforeSend,
+ batchSize);
pingItself.getConnection().start();
// Run a few priming pings to remove warm up time from test results.
Modified: incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/ping/TestPingProducer.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/ping/TestPingProducer.java?view=diff&rev=498965&r1=498964&r2=498965
==============================================================================
--- incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/ping/TestPingProducer.java (original)
+++ incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/ping/TestPingProducer.java Tue Jan 23 01:39:56 2007
@@ -37,12 +37,12 @@
* PingProducer is a client that sends timestamped pings to a queue. It is designed to be run from the command line
* as a stand alone test tool, but it may also be fairly easily instantiated by other code by supplying a session and
* configured message producer.
- *
+ * <p/>
* <p/>This implements the Runnable interface with a run method that implements an infinite ping loop. The ping loop
* does all its work through helper methods, so that code wishing to run a ping cycle is not forced to do so
* by starting a new thread. The command line invocation does take advantage of this ping loop. A shutdown hook is
* also registered to terminate the ping loop cleanly.
- *
+ * <p/>
* <p/><table id="crc"><caption>CRC Card</caption>
* <tr><th> Responsibilities <th> Collaborations
* <tr><td> Provide a ping cycle.
@@ -53,31 +53,42 @@
{
private static final Logger _logger = Logger.getLogger(TestPingProducer.class);
- /** Used to set up a default message size. */
+ /**
+ * Used to set up a default message size.
+ */
private static final int DEFAULT_MESSAGE_SIZE = 0;
- /** Used to define how long to wait between pings. */
+ /**
+ * Used to define how long to wait between pings.
+ */
private static final long SLEEP_TIME = 250;
- /** Holds the name of the queue to send pings on. */
+ /**
+ * Holds the name of the queue to send pings on.
+ */
private static final String PING_QUEUE_NAME = "ping";
private static TestPingProducer _pingProducer;
- /** Holds the message producer to send the pings through. */
- private MessageProducer _producer;
-
- /** Determines whether this producer sends persistent messages from the run method. */
+ /**
+ * Determines whether this producer sends persistent messages from the run method.
+ */
private boolean _persistent = false;
- /** Holds the message size to send, from the run method. */
+ /**
+ * Holds the message size to send, from the run method.
+ */
private int _messageSize = DEFAULT_MESSAGE_SIZE;
- /** Used to indicate that the ping loop should print out whenever it pings. */
+ /**
+ * Used to indicate that the ping loop should print out whenever it pings.
+ */
private boolean _verbose = false;
+
public TestPingProducer(String brokerDetails, String username, String password, String virtualpath, String queueName,
- boolean transacted, boolean persistent, int messageSize, boolean verbose) throws Exception
+ boolean transacted, boolean persistent, int messageSize, boolean verbose, boolean afterCommit,
+ boolean beforeCommit, boolean afterSend, boolean beforeSend,int batchSize) throws Exception
{
// Create a connection to the broker.
InetAddress address = InetAddress.getLocalHost();
@@ -96,6 +107,14 @@
_messageSize = messageSize;
_verbose = verbose;
+
+ // Set failover interrupts
+ _failAfterCommit = afterCommit;
+ _failBeforeCommit = beforeCommit;
+ _failAfterSend = afterSend;
+ _failBeforeSend = beforeSend;
+ _sentMessages = 0;
+ _batchSize = batchSize;
}
/**
@@ -110,7 +129,8 @@
if (args.length < 2)
{
System.err.println(
- "Usage: TestPingPublisher <brokerDetails> <virtual path> [verbose] [transacted] [persistent] [message size in bytes]");
+ "Usage: TestPingPublisher <brokerDetails> <virtual path> "+
+ "[<verbose(true|false)> <transacted(true|false))> <persistent(true|false)> <message size in bytes> <batchsize>");
System.exit(0);
}
@@ -120,10 +140,46 @@
boolean transacted = (args.length >= 4) ? Boolean.parseBoolean(args[3]) : false;
boolean persistent = (args.length >= 5) ? Boolean.parseBoolean(args[4]) : false;
int messageSize = (args.length >= 6) ? Integer.parseInt(args[5]) : DEFAULT_MESSAGE_SIZE;
+ int batchSize = (args.length >= 7) ? Integer.parseInt(args[6]) : 1;
+
+
+ boolean afterCommit = false;
+ boolean beforeCommit = false;
+ boolean afterSend = false;
+ boolean beforeSend = false;
+
+ for (String arg : args)
+ {
+ if (arg.startsWith("failover:"))
+ {
+ //failover:<before|after>:<send:commit>
+ String[] parts = arg.split(":");
+ if (parts.length == 3)
+ {
+ if (parts[2].equals("commit"))
+ {
+ afterCommit = parts[1].equals("after");
+ beforeCommit = parts[1].equals("before");
+ }
+
+ if (parts[2].equals("send"))
+ {
+ afterSend = parts[1].equals("after");
+ beforeSend = parts[1].equals("before");
+ }
+ }
+ else
+ {
+ System.out.println("Unrecognized failover request:" + arg);
+ }
+ }
+ }
// Create a ping producer to generate the pings.
- _pingProducer = new TestPingProducer(brokerDetails, "guest", "guest", virtualpath, PING_QUEUE_NAME, transacted,
- persistent, messageSize, verbose);
+ _pingProducer = new TestPingProducer(brokerDetails, "guest", "guest", virtualpath, PING_QUEUE_NAME,
+ transacted, persistent, messageSize, verbose,
+ afterCommit, beforeCommit, afterSend, beforeSend,
+ batchSize);
// Start the connection running.
_pingProducer.getConnection().start();
@@ -144,19 +200,18 @@
* Sends the specified ping message.
*
* @param message The message to send.
- *
* @throws JMSException All underlying JMSExceptions are allowed to fall through.
*/
public void ping(Message message) throws JMSException
{
- _producer.send(message);
+ sendMessage(message);
// Keep the messageId to correlate with the reply.
String messageId = message.getJMSMessageID();
// Commit the transaction if running in transactional mode. This must happen now, rather than at the end of
- // this method, as the message will not be sent until the transaction is committed.
- commitTx(getProducerSession());
+ // this method, as the message will not be sent until the transaction is committed.
+ commitTx();
}
/**
Modified: incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/requestreply/PingPongProducer.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/requestreply/PingPongProducer.java?view=diff&rev=498965&r1=498964&r2=498965
==============================================================================
--- incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/requestreply/PingPongProducer.java (original)
+++ incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/requestreply/PingPongProducer.java Tue Jan 23 01:39:56 2007
@@ -44,17 +44,17 @@
* client (see {@link PingPongBouncer} for the bounce back client). It is designed to be run from the command line
* as a stand alone test tool, but it may also be fairly easily instantiated by other code by supplying a session,
* message producer and message consumer to run the ping-pong cycle on.
- *
+ * <p/>
* <p/>The pings are sent with a reply-to field set to a single temporary queue, which is the same for all pings.
* This means that this class has to do some work to correlate pings with pongs; it expectes the original message
* id in the ping to be bounced back in the correlation id. If a new temporary queue per ping were used, then
* this correlation would not need to be done.
- *
+ * <p/>
* <p/>This implements the Runnable interface with a run method that implements an infinite ping loop. The ping loop
* does all its work through helper methods, so that code wishing to run a ping-pong cycle is not forced to do so
* by starting a new thread. The command line invocation does take advantage of this ping loop. A shutdown hook is
* also registered to terminate the ping-pong loop cleanly.
- *
+ * <p/>
* <p/><table id="crc"><caption>CRC Card</caption>
* <tr><th> Responsibilities <th> Collaborations
* <tr><td> Provide a ping and wait for response cycle.
@@ -62,65 +62,91 @@
* </table>
*
* @todo Make temp queue per ping a command line option.
- *
* @todo Make the queue name a command line option.
*/
public class PingPongProducer extends AbstractPingProducer implements Runnable, MessageListener, ExceptionListener
{
private static final Logger _logger = Logger.getLogger(PingPongProducer.class);
- /** Used to set up a default message size. */
+ /**
+ * Used to set up a default message size.
+ */
protected static final int DEFAULT_MESSAGE_SIZE = 0;
- /** Used to define how long to wait between pings. */
+ /**
+ * Used to define how long to wait between pings.
+ */
protected static final long SLEEP_TIME = 250;
- /** Used to define how long to wait before assuming that a ping has timed out. */
+ /**
+ * Used to define how long to wait before assuming that a ping has timed out.
+ */
protected static final long TIMEOUT = 9000;
- /** Holds the name of the queue to send pings on. */
+ /**
+ * Holds the name of the queue to send pings on.
+ */
protected static final String PING_QUEUE_NAME = "ping";
- /** The batch size. */
+ /**
+ * The batch size.
+ */
protected static final int BATCH_SIZE = 100;
- /** Keeps track of the ping producer instance used in the run loop. */
+ /**
+ * Keeps track of the ping producer instance used in the run loop.
+ */
private static PingPongProducer _pingProducer;
protected static final int PREFETCH = 100;
protected static final boolean NO_LOCAL = true;
protected static final boolean EXCLUSIVE = false;
- /** The number of priming loops to run. */
+ /**
+ * The number of priming loops to run.
+ */
protected static final int PRIMING_LOOPS = 3;
- /** A source for providing sequential unique correlation ids. */
+ /**
+ * A source for providing sequential unique correlation ids.
+ */
private AtomicLong idGenerator = new AtomicLong(0L);
- /** Holds the message producer to send the pings through. */
- private MessageProducer _producer;
-
- /** Holds the queue to send the ping replies to. */
+ /**
+ * Holds the queue to send the ping replies to.
+ */
private Queue _replyQueue;
- /** Hold the known Queue where the producer will be sending message to*/
+ /**
+ * Hold the known Queue where the producer will be sending message to
+ */
private Queue _pingQueue;
- /** Determines whether this producer sends persistent messages from the run method. */
+ /**
+ * Determines whether this producer sends persistent messages from the run method.
+ */
protected boolean _persistent;
- /** Holds the message size to send, from the run method. */
+ /**
+ * Holds the message size to send, from the run method.
+ */
protected int _messageSize;
- /** Holds a map from message ids to latches on which threads wait for replies. */
+ /**
+ * Holds a map from message ids to latches on which threads wait for replies.
+ */
private Map<String, CountDownLatch> trafficLights = new HashMap<String, CountDownLatch>();
- /** Used to indicate that the ping loop should print out whenever it pings. */
+ /**
+ * Used to indicate that the ping loop should print out whenever it pings.
+ */
protected boolean _verbose = false;
protected Session _consumerSession;
- protected PingPongProducer(String brokerDetails, String username, String password, String virtualpath,
- boolean transacted)
+ private PingPongProducer(String brokerDetails, String username, String password, String virtualpath,
+ boolean transacted, boolean persistent, int messageSize, boolean verbose,
+ boolean afterCommit, boolean beforeCommit, boolean afterSend, boolean beforeSend,
+ int batchSize)
throws Exception
{
// Create a connection to the broker.
@@ -132,6 +158,18 @@
// Create transactional or non-transactional sessions, based on the command line arguments.
setProducerSession((Session) getConnection().createSession(transacted, Session.AUTO_ACKNOWLEDGE));
_consumerSession = (Session) getConnection().createSession(transacted, Session.AUTO_ACKNOWLEDGE);
+
+ _persistent = persistent;
+ _messageSize = messageSize;
+ _verbose = verbose;
+
+ // Set failover interrupts
+ _failAfterCommit = afterCommit;
+ _failBeforeCommit = beforeCommit;
+ _failAfterSend = afterSend;
+ _failBeforeSend = beforeSend;
+ _batchSize = batchSize;
+ _sentMessages = 0;
}
/**
@@ -142,31 +180,39 @@
* @param password
* @param virtualpath
* @param transacted
- * @param persistent
- * @param messageSize
- * @param verbose
- *
* @throws Exception All allowed to fall through. This is only test code...
*/
public PingPongProducer(String brokerDetails, String username, String password, String virtualpath, String queueName,
- String selector, boolean transacted, boolean persistent, int messageSize, boolean verbose)
- throws Exception
+ String selector, boolean transacted, boolean persistent, int messageSize, boolean verbose,
+ boolean afterCommit, boolean beforeCommit, boolean afterSend, boolean beforeSend,
+ int batchSize, int queueCount)
+ throws Exception
{
- this(brokerDetails, username, password, virtualpath, transacted);
+ this(brokerDetails, username, password, virtualpath, transacted, persistent, messageSize, verbose,
+ afterCommit, beforeCommit, afterSend, beforeSend, batchSize);
- _pingQueue = new AMQQueue(queueName);
- _persistent = persistent;
- _messageSize = messageSize;
- _verbose = verbose;
-
- // Create producer and the consumer
- createProducer();
- createConsumer(selector);
+ if (queueName != null)
+ {
+ _pingQueue = new AMQQueue(queueName);
+ // Create producer and the consumer
+ createProducer();
+ createConsumer(selector);
+ }
+ else if (queueCount > 0)
+ {
+ _queueCount = queueCount;
+ }
+ else
+ {
+ _logger.error("Queue Count is zero and no queueName specified. One must be set.");
+ throw new IllegalArgumentException("Queue Count is zero and no queueName specified. One must be set.");
+ }
}
/**
* Creates the producer to send the pings on. If the tests are with nultiple queues, then producer
* is created with null destination, so that any destination can be specified while sending
+ *
* @throws JMSException
*/
public void createProducer() throws JMSException
@@ -189,6 +235,7 @@
/**
* Creates the temporary queue to listen to the responses
+ *
* @param selector
* @throws JMSException
*/
@@ -204,6 +251,7 @@
/**
* Creates consumer instances for each queue. This is used when test is being done with multiple queues.
+ *
* @param selector
* @throws JMSException
*/
@@ -234,7 +282,7 @@
/**
* Starts a ping-pong loop running from the command line. The bounce back client {@link org.apache.qpid.requestreply.PingPongBouncer} also needs
* to be started to bounce the pings back again.
- *
+ * <p/>
* <p/>The command line takes from 2 to 4 arguments:
* <p/><table>
* <tr><td>brokerDetails <td> The broker connection string.
@@ -251,7 +299,7 @@
if (args.length < 2)
{
System.err.println("Usage: TestPingPublisher <brokerDetails> <virtual path> [verbose (true/false)] " +
- "[transacted (true/false)] [persistent (true/false)] [message size in bytes]");
+ "[transacted (true/false)] [persistent (true/false)] [message size in bytes]");
System.exit(0);
}
@@ -261,10 +309,46 @@
boolean transacted = (args.length >= 4) ? Boolean.parseBoolean(args[3]) : false;
boolean persistent = (args.length >= 5) ? Boolean.parseBoolean(args[4]) : false;
int messageSize = (args.length >= 6) ? Integer.parseInt(args[5]) : DEFAULT_MESSAGE_SIZE;
+ int batchSize = (args.length >= 7) ? Integer.parseInt(args[6]) : 1;
+
+ boolean afterCommit = false;
+ boolean beforeCommit = false;
+ boolean afterSend = false;
+ boolean beforeSend = false;
+
+ for (String arg : args)
+ {
+ if (arg.startsWith("failover:"))
+ {
+ //failover:<before|after>:<send:commit>
+ String[] parts = arg.split(":");
+ if (parts.length == 3)
+ {
+ if (parts[2].equals("commit"))
+ {
+ afterCommit = parts[1].equals("after");
+ beforeCommit = parts[1].equals("before");
+ }
+
+ if (parts[2].equals("send"))
+ {
+ afterSend = parts[1].equals("after");
+ beforeSend = parts[1].equals("before");
+ }
+ }
+ else
+ {
+ System.out.println("Unrecognized failover request:" + arg);
+ }
+ }
+ }
// Create a ping producer to handle the request/wait/reply cycle.
_pingProducer = new PingPongProducer(brokerDetails, "guest", "guest", virtualpath, PING_QUEUE_NAME, null, transacted,
- persistent, messageSize, verbose);
+ persistent, messageSize, verbose,
+ afterCommit, beforeCommit, afterSend, beforeSend,
+ batchSize, 0);
+
_pingProducer.getConnection().start();
// Run a few priming pings to remove warm up time from test results.
@@ -287,7 +371,6 @@
* this a few times, in order to prime the JVMs JIT compilation.
*
* @param x The number of priming loops to run.
- *
* @throws JMSException All underlying exceptions are allowed to fall through.
*/
public void prime(int x) throws JMSException
@@ -296,15 +379,18 @@
{
// Create and send a small message.
Message first = getTestMessage(_replyQueue, 0, false);
- _producer.send(first);
- commitTx(getProducerSession());
+
+ sendMessage(first);
+
+ commitTx();
try
{
Thread.sleep(100);
}
catch (InterruptedException ignore)
- { }
+ {
+ }
}
}
@@ -365,10 +451,8 @@
* @param message The message to send.
* @param numPings The number of ping messages to send.
* @param timeout The timeout in milliseconds.
- *
* @return The number of replies received. This may be less than the number sent if the timeout terminated the
* wait for all prematurely.
- *
* @throws JMSException All underlying JMSExceptions are allowed to fall through.
*/
public int pingAndWaitForReply(Message message, int numPings, long timeout) throws JMSException, InterruptedException
@@ -394,13 +478,13 @@
{
// Re-timestamp the message.
message.setLongProperty("timestamp", System.currentTimeMillis());
- _producer.send(message);
+ sendMessage(message);
}
}
// Commit the transaction if running in transactional mode. This must happen now, rather than at the end of
// this method, as the message will not be sent until the transaction is committed.
- commitTx(getProducerSession());
+ commitTx();
// Keep the messageId to correlate with the reply.
//String messageId = message.getJMSMessageID();
@@ -429,7 +513,8 @@
/**
* When the test is being performed with multiple queues, then this method will be used, which has a loop to
- * pick up the next queue from the queues list and sends message to it.
+ * pick up the next queue from the queues list and sends message to it.
+ *
* @param message
* @param numPings
* @throws JMSException
@@ -441,31 +526,30 @@
{
// Re-timestamp the message.
message.setLongProperty("timestamp", System.currentTimeMillis());
- _producer.send(getQueue(queueIndex++), message);
+
+ sendMessage(getQueue(queueIndex++), message);
// reset the counter to get the first queue
- if (queueIndex == getQueueCount() -1)
+ if (queueIndex == getQueueCount() - 1)
{
queueIndex = 0;
}
}
}
-
+
/**
* Sends the specified ping message but does not wait for a correlating reply.
*
* @param message The message to send.
* @param numPings The number of pings to send.
- *
* @return The reply, or null if no reply arrives before the timeout.
- *
* @throws JMSException All underlying JMSExceptions are allowed to fall through.
*/
public void pingNoWaitForReply(Message message, int numPings) throws JMSException, InterruptedException
{
for (int i = 0; i < numPings; i++)
{
- _producer.send(message);
+ sendMessage(message);
if (_verbose)
{
@@ -474,7 +558,7 @@
}
// Commit the transaction if running in transactional mode, to force the send now.
- commitTx(getProducerSession());
+ commitTx();
}
/**
@@ -524,19 +608,21 @@
public static class FailoverNotifier implements ConnectionListener
{
public void bytesSent(long count)
- { }
+ {
+ }
public void bytesReceived(long count)
- { }
+ {
+ }
public boolean preFailover(boolean redirect)
{
- return true;
+ return true; //Allow failover
}
public boolean preResubscribe()
{
- return true;
+ return true; // Allow resubscription
}
public void failoverComplete()
Modified: incubator/qpid/trunk/qpid/java/perftests/src/test/java/org/apache/qpid/ping/PingTestPerf.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/perftests/src/test/java/org/apache/qpid/ping/PingTestPerf.java?view=diff&rev=498965&r1=498964&r2=498965
==============================================================================
--- incubator/qpid/trunk/qpid/java/perftests/src/test/java/org/apache/qpid/ping/PingTestPerf.java (original)
+++ incubator/qpid/trunk/qpid/java/perftests/src/test/java/org/apache/qpid/ping/PingTestPerf.java Tue Jan 23 01:39:56 2007
@@ -13,23 +13,22 @@
import uk.co.thebadgerset.junit.extensions.AsymptoticTestCase;
/**
- *
* PingTestPerf is a ping test, that has been written with the intention of being scaled up to run many times
* simultaneously to simluate many clients/producers/connections.
- *
+ * <p/>
* <p/>A single run of the test using the default JUnit test runner will result in the sending and timing of a single
* full round trip ping. This test may be scaled up using a suitable JUnit test runner.
- *
+ * <p/>
* <p/>The setup/teardown cycle establishes a connection to a broker and sets up a queue to send ping messages to and a
* temporary queue for replies. This setup is only established once for all the test repeats/threads that may be run,
* except if the connection is lost in which case an attempt to re-establish the setup is made.
- *
+ * <p/>
* <p/>The test cycle is: Connects to a queue, creates a temporary queue, creates messages containing a property that
* is the name of the temporary queue, fires off a message on the original queue and waits for a response on the
* temporary queue.
- *
+ * <p/>
* <p/>Configurable test properties: message size, transacted or not, persistent or not. Broker connection details.
- *
+ * <p/>
* <p><table id="crc"><caption>CRC Card</caption>
* <tr><th> Responsibilities <th> Collaborations
* </table>
@@ -40,65 +39,95 @@
{
private static Logger _logger = Logger.getLogger(PingTestPerf.class);
- /** Holds the name of the property to get the test message size from. */
+ /**
+ * Holds the name of the property to get the test message size from.
+ */
private static final String MESSAGE_SIZE_PROPNAME = "messageSize";
- /** Holds the name of the property to get the ping queue name from. */
+ /**
+ * Holds the name of the property to get the ping queue name from.
+ */
private static final String PING_QUEUE_NAME_PROPNAME = "pingQueue";
- /** holds the queue count, if the test is being performed with multiple queues */
+ /**
+ * holds the queue count, if the test is being performed with multiple queues
+ */
private static final String PING_QUEUE_COUNT_PROPNAME = "queues";
- /** Holds the name of the property to get the test delivery mode from. */
+ /**
+ * Holds the name of the property to get the test delivery mode from.
+ */
private static final String PERSISTENT_MODE_PROPNAME = "persistent";
- /** Holds the name of the property to get the test transactional mode from. */
+ /**
+ * Holds the name of the property to get the test transactional mode from.
+ */
private static final String TRANSACTED_PROPNAME = "transacted";
- /** Holds the name of the property to get the test broker url from. */
+ /**
+ * Holds the name of the property to get the test broker url from.
+ */
private static final String BROKER_PROPNAME = "broker";
- /** Holds the name of the property to get the test broker virtual path. */
+ /**
+ * Holds the name of the property to get the test broker virtual path.
+ */
private static final String VIRTUAL_PATH_PROPNAME = "virtualPath";
- /** Holds the waiting timeout for response messages */
+ /**
+ * Holds the waiting timeout for response messages
+ */
private static final String TIMEOUT_PROPNAME = "timeout";
- /** Holds the size of message body to attach to the ping messages. */
+ /**
+ * Holds the size of message body to attach to the ping messages.
+ */
private static final int MESSAGE_SIZE_DEFAULT = 0;
- /** Holds the name of the queue to which pings are sent. */
+ private static final int BATCH_SIZE_DEFAULT = 2;
+
+ /**
+ * Holds the name of the queue to which pings are sent.
+ */
private static final String PING_QUEUE_NAME_DEFAULT = "ping";
- /** Holds the message delivery mode to use for the test. */
+ /**
+ * Holds the message delivery mode to use for the test.
+ */
private static final boolean PERSISTENT_MODE_DEFAULT = false;
- /** Holds the transactional mode to use for the test. */
+ /**
+ * Holds the transactional mode to use for the test.
+ */
private static final boolean TRANSACTED_DEFAULT = false;
- /** Holds the default broker url for the test. */
+ /**
+ * Holds the default broker url for the test.
+ */
private static final String BROKER_DEFAULT = "tcp://localhost:5672";
- /** Holds the default virtual path for the test. */
+ /**
+ * Holds the default virtual path for the test.
+ */
private static final String VIRTUAL_PATH_DEFAULT = "/test";
- /** Sets a default ping timeout. */
+ /**
+ * Sets a default ping timeout.
+ */
private static final long TIMEOUT_DEFAULT = 3000;
- // Sets up the test parameters with defaults.
- static
- {
- setSystemPropertyIfNull(MESSAGE_SIZE_PROPNAME, Integer.toString(MESSAGE_SIZE_DEFAULT));
- setSystemPropertyIfNull(PING_QUEUE_NAME_PROPNAME, PING_QUEUE_NAME_DEFAULT);
- setSystemPropertyIfNull(PERSISTENT_MODE_PROPNAME, Boolean.toString(PERSISTENT_MODE_DEFAULT));
- setSystemPropertyIfNull(TRANSACTED_PROPNAME, Boolean.toString(TRANSACTED_DEFAULT));
- setSystemPropertyIfNull(BROKER_PROPNAME, BROKER_DEFAULT);
- setSystemPropertyIfNull(VIRTUAL_PATH_PROPNAME, VIRTUAL_PATH_DEFAULT);
- setSystemPropertyIfNull(TIMEOUT_PROPNAME, Long.toString(TIMEOUT_DEFAULT));
- setSystemPropertyIfNull(PING_QUEUE_COUNT_PROPNAME, Integer.toString(1));
- }
- /** Thread local to hold the per-thread test setup fields. */
+ private static final String FAIL_AFTER_COMMIT = "FailAfterCommit";
+ private static final String FAIL_BEFORE_COMMIT = "FailBeforeCommit";
+ private static final String FAIL_AFTER_SEND = "FailAfterSend";
+ private static final String FAIL_BEFORE_SEND = "FailBeforeSend";
+ private static final String BATCH_SIZE = "BatchSize";
+
+
+
+ /**
+ * 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
@@ -107,9 +136,27 @@
private Properties testParameters = System.getProperties();
//private Properties testParameters = new ContextualProperties(System.getProperties());
+
public PingTestPerf(String name)
{
super(name);
+ // Sets up the test parameters with defaults.
+
+
+ setSystemPropertyIfNull(FAIL_AFTER_COMMIT, "false");
+ setSystemPropertyIfNull(FAIL_BEFORE_COMMIT, "false");
+ setSystemPropertyIfNull(FAIL_AFTER_SEND, "false");
+ setSystemPropertyIfNull(FAIL_BEFORE_SEND, "false");
+
+ setSystemPropertyIfNull(BATCH_SIZE, Integer.toString(BATCH_SIZE_DEFAULT));
+ setSystemPropertyIfNull(MESSAGE_SIZE_PROPNAME, Integer.toString(MESSAGE_SIZE_DEFAULT));
+ setSystemPropertyIfNull(PING_QUEUE_NAME_PROPNAME, PING_QUEUE_NAME_DEFAULT);
+ setSystemPropertyIfNull(PERSISTENT_MODE_PROPNAME, Boolean.toString(PERSISTENT_MODE_DEFAULT));
+ setSystemPropertyIfNull(TRANSACTED_PROPNAME, Boolean.toString(TRANSACTED_DEFAULT));
+ setSystemPropertyIfNull(BROKER_PROPNAME, BROKER_DEFAULT);
+ setSystemPropertyIfNull(VIRTUAL_PATH_PROPNAME, VIRTUAL_PATH_DEFAULT);
+ setSystemPropertyIfNull(TIMEOUT_PROPNAME, Long.toString(TIMEOUT_DEFAULT));
+ setSystemPropertyIfNull(PING_QUEUE_COUNT_PROPNAME, Integer.toString(1));
}
/**
@@ -124,7 +171,7 @@
suite.addTest(new PingTestPerf("testPingOk"));
return suite;
- //return new junit.framework.TestSuite(PingTestPerf.class);
+ //return new junit.framework.TestSuite(PingTestPerf.class);
}
private static void setSystemPropertyIfNull(String propName, String propValue)
@@ -135,18 +182,28 @@
}
}
+ public void testPing(int jim) throws Exception
+ {
+ testPingOk(1);
+ }
+
public void testPingOk(int numPings) throws Exception
{
// Get the per thread test setup to run the test through.
PerThreadSetup perThreadSetup = threadSetup.get();
+ if (numPings == 0)
+ {
+ _logger.error("Number of pings requested was zero.");
+ }
+
// Generate a sample message. This message is already time stamped and has its reply-to destination set.
ObjectMessage msg =
- perThreadSetup._pingItselfClient.getTestMessage(null,
- Integer.parseInt(testParameters.getProperty(
- MESSAGE_SIZE_PROPNAME)),
- Boolean.parseBoolean(testParameters.getProperty(
- PERSISTENT_MODE_PROPNAME)));
+ 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));
@@ -185,20 +242,31 @@
boolean verbose = false;
int messageSize = Integer.parseInt(testParameters.getProperty(MESSAGE_SIZE_PROPNAME));
+ boolean afterCommit = Boolean.parseBoolean(testParameters.getProperty(FAIL_AFTER_COMMIT));
+ boolean beforeCommit = Boolean.parseBoolean(testParameters.getProperty(FAIL_BEFORE_COMMIT));
+ boolean afterSend = Boolean.parseBoolean(testParameters.getProperty(FAIL_AFTER_SEND));
+ boolean beforeSend = Boolean.parseBoolean(testParameters.getProperty(FAIL_BEFORE_SEND));
+
+ int batchSize = Integer.parseInt(testParameters.getProperty(BATCH_SIZE));
+
// Establish a client to ping a Queue and listen the reply back from same Queue
if (queueCount > 1)
{
// test client with multiple queues
perThreadSetup._pingItselfClient = new TestPingItself(brokerDetails, username, password, virtualpath,
- queueCount, selector, transacted, persistent,
- messageSize, verbose);
+ selector, transacted, persistent,
+ messageSize, verbose,
+ afterCommit, beforeCommit, afterSend, beforeSend,
+ batchSize, queueCount);
}
else
{
// Establish a client to ping a Queue and listen the reply back from same Queue
perThreadSetup._pingItselfClient = new TestPingItself(brokerDetails, username, password, virtualpath,
queueName, selector, transacted, persistent,
- messageSize, verbose);
+ messageSize, verbose,
+ afterCommit, beforeCommit, afterSend, beforeSend,
+ batchSize);
}
// Start the client connection
@@ -228,7 +296,9 @@
private static class PerThreadSetup
{
- /** Holds the test ping client. */
+ /**
+ * Holds the test ping client.
+ */
private TestPingItself _pingItselfClient;
}
}
Modified: incubator/qpid/trunk/qpid/java/perftests/src/test/java/org/apache/qpid/requestreply/PingPongTestPerf.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/perftests/src/test/java/org/apache/qpid/requestreply/PingPongTestPerf.java?view=diff&rev=498965&r1=498964&r2=498965
==============================================================================
--- incubator/qpid/trunk/qpid/java/perftests/src/test/java/org/apache/qpid/requestreply/PingPongTestPerf.java (original)
+++ incubator/qpid/trunk/qpid/java/perftests/src/test/java/org/apache/qpid/requestreply/PingPongTestPerf.java Tue Jan 23 01:39:56 2007
@@ -18,22 +18,22 @@
* PingPongTestPerf is a full round trip ping test, that has been written with the intention of being scaled up to run
* many times simultaneously to simluate many clients/producer/connections. A full round trip ping sends a message from
* a producer to a conumer, then the consumer replies to the message on a temporary queue.
- *
+ * <p/>
* <p/>A single run of the test using the default JUnit test runner will result in the sending and timing of the number
* of pings specified by the test size and time how long it takes for all of these to complete. This test may be scaled
* up using a suitable JUnit test runner. See {@link TKTestRunner} or {@link PPTestRunner} for more information on how
* to do this.
- *
+ * <p/>
* <p/>The setup/teardown cycle establishes a connection to a broker and sets up a queue to send ping messages to and a
* temporary queue for replies. This setup is only established once for all the test repeats, but each test threads
* gets its own connection/producer/consumer, this is only re-established if the connection is lost.
- *
+ * <p/>
* <p/>The test cycle is: Connects to a queue, creates a temporary queue, creates messages containing a property that
* is the name of the temporary queue, fires off many messages on the original queue and waits for them all to come
* back on the temporary queue.
- *
+ * <p/>
* <p/>Configurable test properties: message size, transacted or not, persistent or not. Broker connection details.
- *
+ * <p/>
* <p><table id="crc"><caption>CRC Card</caption>
* <tr><th> Responsibilities <th> Collaborations
* </table>
@@ -44,43 +44,69 @@
{
private static Logger _logger = Logger.getLogger(PingPongTestPerf.class);
- /** Holds the name of the property to get the test message size from. */
+ /**
+ * Holds the name of the property to get the test message size from.
+ */
private static final String MESSAGE_SIZE_PROPNAME = "messageSize";
- /** Holds the name of the property to get the ping queue name from. */
+ /**
+ * Holds the name of the property to get the ping queue name from.
+ */
private static final String PING_QUEUE_NAME_PROPNAME = "pingQueue";
- /** Holds the name of the property to get the test delivery mode from. */
+ /**
+ * Holds the name of the property to get the test delivery mode from.
+ */
private static final String PERSISTENT_MODE_PROPNAME = "persistent";
- /** Holds the name of the property to get the test transactional mode from. */
+ /**
+ * Holds the name of the property to get the test transactional mode from.
+ */
private static final String TRANSACTED_PROPNAME = "transacted";
- /** Holds the name of the property to get the test broker url from. */
+ /**
+ * Holds the name of the property to get the test broker url from.
+ */
private static final String BROKER_PROPNAME = "broker";
- /** Holds the name of the property to get the test broker virtual path. */
+ /**
+ * Holds the name of the property to get the test broker virtual path.
+ */
private static final String VIRTUAL_PATH_PROPNAME = "virtualPath";
- /** Holds the size of message body to attach to the ping messages. */
+ /**
+ * Holds the size of message body to attach to the ping messages.
+ */
private static final int MESSAGE_SIZE_DEFAULT = 0;
- /** Holds the name of the queue to which pings are sent. */
+ /**
+ * Holds the name of the queue to which pings are sent.
+ */
private static final String PING_QUEUE_NAME_DEFAULT = "ping";
- /** Holds the message delivery mode to use for the test. */
+ /**
+ * Holds the message delivery mode to use for the test.
+ */
private static final boolean PERSISTENT_MODE_DEFAULT = false;
- /** Holds the transactional mode to use for the test. */
+ /**
+ * Holds the transactional mode to use for the test.
+ */
private static final boolean TRANSACTED_DEFAULT = false;
- /** Holds the default broker url for the test. */
+ /**
+ * Holds the default broker url for the test.
+ */
private static final String BROKER_DEFAULT = "tcp://localhost:5672";
- /** Holds the default virtual path for the test. */
+ /**
+ * Holds the default virtual path for the test.
+ */
private static final String VIRTUAL_PATH_DEFAULT = "/test";
- /** Sets a default ping timeout. */
+ /**
+ * Sets a default ping timeout.
+ */
private static final long TIMEOUT = 15000;
// Sets up the test parameters with defaults.
@@ -94,7 +120,9 @@
setSystemPropertyIfNull(VIRTUAL_PATH_PROPNAME, VIRTUAL_PATH_DEFAULT);
}
- /** Thread local to hold the per-thread test setup fields. */
+ /**
+ * 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
@@ -103,6 +131,13 @@
private Properties testParameters = System.getProperties();
//private Properties testParameters = new ContextualProperties(System.getProperties());
+ private static final String FAIL_AFTER_COMMIT = "FailAfterCommit";
+ private static final String FAIL_BEFORE_COMMIT = "FailBeforeCommit";
+ private static final String FAIL_AFTER_SEND = "FailAfterSend";
+ private static final String FAIL_BEFORE_SEND = "FailBeforeSend";
+ private static final String BATCH_SIZE = "BatchSize";
+
+
public PingPongTestPerf(String name)
{
super(name);
@@ -137,11 +172,11 @@
// Generate a sample message. This message is already time stamped and has its reply-to destination set.
ObjectMessage msg =
- perThreadSetup._testPingProducer.getTestMessage(perThreadSetup._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.
@@ -181,6 +216,12 @@
boolean verbose = false;
int messageSize = Integer.parseInt(testParameters.getProperty(MESSAGE_SIZE_PROPNAME));
+ boolean afterCommit = Boolean.parseBoolean(testParameters.getProperty(FAIL_AFTER_COMMIT));
+ boolean beforeCommit = Boolean.parseBoolean(testParameters.getProperty(FAIL_BEFORE_COMMIT));
+ boolean afterSend = Boolean.parseBoolean(testParameters.getProperty(FAIL_AFTER_SEND));
+ boolean beforeSend = Boolean.parseBoolean(testParameters.getProperty(FAIL_BEFORE_SEND));
+ int batchSize = Integer.parseInt(testParameters.getProperty(BATCH_SIZE));
+
// Establish a bounce back client on the ping queue to bounce back the pings.
perThreadSetup._testPingBouncer = new PingPongBouncer(brokerDetails, username, password, virtualpath, queueName,
persistent, transacted, selector, verbose);
@@ -191,7 +232,9 @@
// Establish a ping-pong client on the ping queue to send the pings with.
perThreadSetup._testPingProducer = new PingPongProducer(brokerDetails, username, password, virtualpath,
queueName, selector, transacted, persistent, messageSize,
- verbose);
+ verbose,
+ afterCommit, beforeCommit, afterSend, beforeSend,
+ batchSize, 0);
perThreadSetup._testPingProducer.getConnection().start();
@@ -205,14 +248,14 @@
try
{
/**if ((_testPingBouncer != null) && (_testPingBouncer.getConnection() != null))
- {
- _testPingBouncer.getConnection().close();
- }
-
- if ((_testPingProducer != null) && (_testPingProducer.getConnection() != null))
- {
- _testPingProducer.getConnection().close();
- }*/
+ {
+ _testPingBouncer.getConnection().close();
+ }
+
+ if ((_testPingProducer != null) && (_testPingProducer.getConnection() != null))
+ {
+ _testPingProducer.getConnection().close();
+ }*/
}
finally
{
@@ -222,10 +265,14 @@
private static class PerThreadSetup
{
- /** Holds the test ping-pong producer. */
+ /**
+ * Holds the test ping-pong producer.
+ */
private PingPongProducer _testPingProducer;
- /** Holds the test ping client. */
+ /**
+ * Holds the test ping client.
+ */
private PingPongBouncer _testPingBouncer;
}
}