qpid-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From oru...@apache.org
Subject svn commit: r1632318 - in /qpid/trunk/qpid/java/tools: pom.xml src/main/java/org/apache/qpid/tools/StressTestClient.java src/main/resources/ src/main/resources/stress-test-client.properties
Date Thu, 16 Oct 2014 14:13:20 GMT
Author: orudyy
Date: Thu Oct 16 14:13:20 2014
New Revision: 1632318

URL: http://svn.apache.org/r1632318
Log:
QPID-6158: Add sample utility to help to perform stress testing of broker

The utility was originally implemented by Robbie Gemmell

Added:
    qpid/trunk/qpid/java/tools/src/main/java/org/apache/qpid/tools/StressTestClient.java
    qpid/trunk/qpid/java/tools/src/main/resources/
    qpid/trunk/qpid/java/tools/src/main/resources/stress-test-client.properties
Modified:
    qpid/trunk/qpid/java/tools/pom.xml

Modified: qpid/trunk/qpid/java/tools/pom.xml
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/tools/pom.xml?rev=1632318&r1=1632317&r2=1632318&view=diff
==============================================================================
--- qpid/trunk/qpid/java/tools/pom.xml (original)
+++ qpid/trunk/qpid/java/tools/pom.xml Thu Oct 16 14:13:20 2014
@@ -51,7 +51,6 @@
       <groupId>org.apache.geronimo.specs</groupId>
       <artifactId>geronimo-jms_1.1_spec</artifactId>
       <version>${geronimo-jms-1-1-version}</version>
-      <scope>provided</scope>
     </dependency>
   </dependencies>
    

Added: qpid/trunk/qpid/java/tools/src/main/java/org/apache/qpid/tools/StressTestClient.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/tools/src/main/java/org/apache/qpid/tools/StressTestClient.java?rev=1632318&view=auto
==============================================================================
--- qpid/trunk/qpid/java/tools/src/main/java/org/apache/qpid/tools/StressTestClient.java (added)
+++ qpid/trunk/qpid/java/tools/src/main/java/org/apache/qpid/tools/StressTestClient.java Thu
Oct 16 14:13:20 2014
@@ -0,0 +1,446 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.qpid.tools;
+
+import java.io.InputStream;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Random;
+
+import javax.jms.BytesMessage;
+import javax.jms.Connection;
+import javax.jms.ConnectionFactory;
+import javax.jms.DeliveryMode;
+import javax.jms.Destination;
+import javax.jms.ExceptionListener;
+import javax.jms.JMSException;
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageProducer;
+import javax.jms.Session;
+import javax.naming.Context;
+import javax.naming.InitialContext;
+
+public class StressTestClient
+{
+    private static final String QUEUE_NAME_PREFIX = "BURL:direct://amq.direct//stress-test-queue";
+    private static final String DURABLE_SUFFIX = "?durable='true'";
+
+    public static final String CONNECTIONS_ARG = "connections";
+    public static final String SESSIONS_ARG = "sessions";
+    public static final String CONSUME_IMMEDIATELY_ARG = "consumeImmediately";
+    public static final String CONSUMERS_ARG = "consumers";
+    public static final String CLOSE_CONSUMERS_ARG = "closeconsumers";
+    public static final String PRODUCERS_ARG = "producers";
+    public static final String MESSAGE_COUNT_ARG = "messagecount";
+    public static final String MESSAGE_SIZE_ARG = "size";
+    public static final String SUFFIX_ARG = "suffix";
+    public static final String REPETITIONS_ARG = "repetitions";
+    public static final String PERSISTENT_ARG = "persistent";
+    public static final String RANDOM_ARG = "random";
+    public static final String TIMEOUT_ARG = "timeout";
+    public static final String DELAYCLOSE_ARG = "delayclose";
+    public static final String REPORT_MOD_ARG = "reportmod";
+    public static final String LOW_PREFETCH_ARG = "lowprefetch";
+    public static final String TRANSACTED_ARG = "transacted";
+    public static final String TX_BATCH_ARG = "txbatch";
+
+    public static final String CONNECTIONS_DEFAULT = "1";
+    public static final String SESSIONS_DEFAULT = "1";
+    public static final String CONSUME_IMMEDIATELY_DEFAULT = "true";
+    public static final String CLOSE_CONSUMERS_DEFAULT = "true";
+    public static final String PRODUCERS_DEFAULT = "1";
+    public static final String CONSUMERS_DEFAULT = "1";
+    public static final String MESSAGE_COUNT_DEFAULT = "1";
+    public static final String MESSAGE_SIZE_DEFAULT = "256";
+    public static final String SUFFIX_DEFAULT = "";
+    public static final String REPETITIONS_DEFAULT = "1";
+    public static final String PERSISTENT_DEFAULT = "false";
+    public static final String RANDOM_DEFAULT = "true";
+    public static final String TIMEOUT_DEFAULT = "30000";
+    public static final String DELAYCLOSE_DEFAULT = "0";
+    public static final String REPORT_MOD_DEFAULT = "1";
+    public static final String LOW_PREFETCH_DEFAULT = "false";
+    public static final String TRANSACTED_DEFAULT = "false";
+    public static final String TX_BATCH_DEFAULT = "1";
+
+    private static final String CLASS = "StressTestClient";
+
+    public static void main(String[] args)
+    {
+        Map<String,String> options = new HashMap<>();
+        options.put(CONNECTIONS_ARG, CONNECTIONS_DEFAULT);
+        options.put(SESSIONS_ARG, SESSIONS_DEFAULT);
+        options.put(CONSUME_IMMEDIATELY_ARG, CONSUME_IMMEDIATELY_DEFAULT);
+        options.put(PRODUCERS_ARG, PRODUCERS_DEFAULT);
+        options.put(CONSUMERS_ARG, CONSUMERS_DEFAULT);
+        options.put(CLOSE_CONSUMERS_ARG, CLOSE_CONSUMERS_DEFAULT);
+        options.put(MESSAGE_COUNT_ARG, MESSAGE_COUNT_DEFAULT);
+        options.put(MESSAGE_SIZE_ARG, MESSAGE_SIZE_DEFAULT);
+        options.put(SUFFIX_ARG, SUFFIX_DEFAULT);
+        options.put(REPETITIONS_ARG, REPETITIONS_DEFAULT);
+        options.put(PERSISTENT_ARG, PERSISTENT_DEFAULT);
+        options.put(RANDOM_ARG, RANDOM_DEFAULT);
+        options.put(TIMEOUT_ARG, TIMEOUT_DEFAULT);
+        options.put(DELAYCLOSE_ARG, DELAYCLOSE_DEFAULT);
+        options.put(REPORT_MOD_ARG, REPORT_MOD_DEFAULT);
+        options.put(LOW_PREFETCH_ARG, LOW_PREFETCH_DEFAULT);
+        options.put(TRANSACTED_ARG, TRANSACTED_DEFAULT);
+        options.put(TX_BATCH_ARG, TX_BATCH_DEFAULT);
+
+        if(args.length == 1 &&
+                (args[0].equals("-h") || args[0].equals("--help") || args[0].equals("help")))
+        {
+            System.out.println("arg=value options: \n" + options.keySet());
+            return;
+        }
+
+        parseArgumentsIntoConfig(options, args);
+
+        StressTestClient testClient = new StressTestClient();
+        testClient.runTest(options);
+    }
+
+    public static void parseArgumentsIntoConfig(Map<String, String> initialValues,
String[] args)
+    {
+        for(String arg: args)
+        {
+            String[] splitArg = arg.split("=");
+            if(splitArg.length != 2)
+            {
+                throw new IllegalArgumentException("arguments must have format <name>=<value>:
" + arg);
+            }
+
+            if(initialValues.put(splitArg[0], splitArg[1]) == null)
+            {
+                throw new IllegalArgumentException("not a valid configuration property: "
+ arg);
+            }
+        }
+    }
+
+
+    private void runTest(Map<String,String> options)
+    {
+        int numConnections = Integer.parseInt(options.get(CONNECTIONS_ARG));
+        int numSessions = Integer.parseInt(options.get(SESSIONS_ARG));
+        int numProducers = Integer.parseInt(options.get(PRODUCERS_ARG));
+        int numConsumers = Integer.parseInt(options.get(CONSUMERS_ARG));
+        boolean closeConsumers = Boolean.valueOf(options.get(CLOSE_CONSUMERS_ARG));
+        boolean consumeImmediately = Boolean.valueOf(options.get(CONSUME_IMMEDIATELY_ARG));
+        int numMessage = Integer.parseInt(options.get(MESSAGE_COUNT_ARG));
+        int messageSize = Integer.parseInt(options.get(MESSAGE_SIZE_ARG));
+        int repetitions = Integer.parseInt(options.get(REPETITIONS_ARG));
+        String queueString = QUEUE_NAME_PREFIX + options.get(SUFFIX_ARG) + DURABLE_SUFFIX;
+        int deliveryMode = Boolean.valueOf(options.get(PERSISTENT_ARG)) ? DeliveryMode.PERSISTENT
: DeliveryMode.NON_PERSISTENT;
+        boolean random = Boolean.valueOf(options.get(RANDOM_ARG));
+        long recieveTimeout = Long.parseLong(options.get(TIMEOUT_ARG));
+        long delayClose = Long.parseLong(options.get(DELAYCLOSE_ARG));
+        int reportingMod = Integer.parseInt(options.get(REPORT_MOD_ARG));
+        boolean lowPrefetch = Boolean.valueOf(options.get(LOW_PREFETCH_ARG));
+        boolean transacted = Boolean.valueOf(options.get(TRANSACTED_ARG));
+        int txBatch = Integer.parseInt(options.get(TX_BATCH_ARG));
+
+        System.out.println(CLASS + ": Using options: " + options);
+
+        System.out.println(CLASS + ": Creating message payload of " + messageSize + " (bytes)");
+        byte[] sentBytes = generateMessage(random, messageSize);
+
+        try
+        {
+            // Load JNDI properties
+            Properties properties = new Properties();
+            try(InputStream is = this.getClass().getClassLoader().getResourceAsStream("stress-test-client.properties"))
+            {
+                properties.load(is);
+            }
+            Context ctx = new InitialContext(properties);
+
+            ConnectionFactory conFac;
+            if(lowPrefetch)
+            {
+                System.out.println(CLASS + ": Using lowprefetch connection factory");
+                conFac = (ConnectionFactory)ctx.lookup("qpidConnectionfactoryLowPrefetch");
+            }
+            else
+            {
+                conFac = (ConnectionFactory)ctx.lookup("qpidConnectionfactory");
+            }
+
+            //ensure the queue to be used exists and is bound
+            System.out.println(CLASS + ": Creating queue: " + queueString);
+            Connection startupConn = conFac.createConnection();
+            Session startupSess = startupConn.createSession(false, Session.AUTO_ACKNOWLEDGE);
+            Destination startupDestination = startupSess.createQueue(queueString);
+            MessageConsumer startupConsumer = startupSess.createConsumer(startupDestination);
+            startupConsumer.close();
+            startupSess.close();
+            startupConn.close();
+
+            for(int rep = 1 ; rep <= repetitions; rep++)
+            {
+                ArrayList<Connection> connectionList = new ArrayList<>();
+
+                for (int co= 1; co<= numConnections ; co++)
+                {
+                    if( co % reportingMod == 0)
+                    {
+                        System.out.println(CLASS + ": Creating connection " + co);
+                    }
+                    Connection conn = conFac.createConnection();
+                    conn.setExceptionListener(new ExceptionListener()
+                    {
+                        public void onException(JMSException jmse)
+                        {
+                            System.err.println(CLASS + ": The sample received an exception
through the ExceptionListener");
+                            jmse.printStackTrace();
+                            System.exit(0);
+                        }
+                    });
+
+                    connectionList.add(conn);
+                    conn.start();
+                    for (int se= 1; se<= numSessions ; se++)
+                    {
+                        if( se % reportingMod == 0)
+                        {
+                            System.out.println(CLASS + ": Creating Session " + se);
+                        }
+                        try
+                        {
+                            Session sess;
+                            if(transacted)
+                            {
+                                sess = conn.createSession(true, Session.SESSION_TRANSACTED);
+                            }
+                            else
+                            {
+                                sess = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
+                            }
+
+                            BytesMessage message = sess.createBytesMessage();
+
+                            message.writeBytes(sentBytes);
+
+                            if(!random && numMessage == 1 && numSessions
== 1 && numConnections == 1 && repetitions == 1)
+                            {
+                                //null the array to save memory
+                                sentBytes = null;
+                            }
+
+                            Destination destination = sess.createQueue(queueString);
+
+                            MessageConsumer consumer = null;
+                            for(int cns = 1 ; cns <= numConsumers ; cns++)
+                            {
+                                if( cns % reportingMod == 0)
+                                {
+                                    System.out.println(CLASS + ": Creating Consumer " + cns);
+                                }
+                                consumer = sess.createConsumer(destination);
+                            }
+
+                            for(int pr = 1 ; pr <= numProducers ; pr++)
+                            {
+                                if( pr % reportingMod == 0)
+                                {
+                                    System.out.println(CLASS + ": Creating Producer " + pr);
+                                }
+                                MessageProducer prod = sess.createProducer(destination);
+                                for(int me = 1; me <= numMessage ; me++)
+                                {
+                                    if( me % reportingMod == 0)
+                                    {
+                                        System.out.println(CLASS + ": Sending Message " +
me);
+                                    }
+                                    prod.send(message, deliveryMode,
+                                            Message.DEFAULT_PRIORITY,
+                                            Message.DEFAULT_TIME_TO_LIVE);
+                                    if(transacted && me % txBatch == 0)
+                                    {
+                                        sess.commit();
+                                    }
+                                }
+                            }
+
+                            if(numConsumers > 0 && consumeImmediately)
+                            {
+                                for(int cs = 1 ; cs <= numMessage ; cs++)
+                                {
+                                    if( cs % reportingMod == 0)
+                                    {
+                                        System.out.println(CLASS + ": Consuming Message "
+ cs);
+                                    }
+                                    BytesMessage msg = (BytesMessage) consumer.receive(recieveTimeout);
+
+                                    if(transacted && cs % txBatch == 0)
+                                    {
+                                        sess.commit();
+                                    }
+
+                                    if(msg == null)
+                                    {
+                                        throw new RuntimeException("Expected message not
received in allowed time: " + recieveTimeout);
+                                    }
+
+                                    validateReceivedMessageContent(sentBytes, msg, random,
messageSize);
+                                }
+
+                                if(closeConsumers)
+                                {
+                                    consumer.close();
+                                }
+                            }
+
+                        }
+                        catch (Exception exp)
+                        {
+                            System.err.println(CLASS + ": Caught an Exception: " + exp);
+                            exp.printStackTrace();
+                        }
+
+                    }
+                }
+
+                if(numConsumers == -1 && !consumeImmediately)
+                {
+                    System.out.println(CLASS + ": Consuming left over messages, using recieve
timeout:" + recieveTimeout);
+
+                    Connection conn = conFac.createConnection();
+                    Session sess = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
+                    Destination destination = sess.createQueue(queueString);
+                    MessageConsumer consumer = sess.createConsumer(destination);
+                    conn.start();
+
+                    int count = 0;
+                    while(true)
+                    {
+                        BytesMessage msg = (BytesMessage) consumer.receive(recieveTimeout);
+
+                        if(msg == null)
+                        {
+                            System.out.println(CLASS + ": Received " + count + " messages");
+                            break;
+                        }
+                        else
+                        {
+                            count++;
+                        }
+
+                        validateReceivedMessageContent(sentBytes, msg, random, messageSize);
+                    }
+
+                    consumer.close();
+                    sess.close();
+                    conn.close();
+                }
+
+                if(delayClose > 0)
+                {
+                    System.out.println(CLASS + ": Delaying closing connections: " + delayClose);
+                    Thread.sleep(delayClose);
+                }
+
+                // Close the connections to the server
+                System.out.println(CLASS + ": Closing connections");
+
+                for(int connection = 0 ; connection < connectionList.size() ; connection++)
+                {
+                    if( (connection+1) % reportingMod == 0)
+                    {
+                        System.out.println(CLASS + ": Closing connection " + (connection+1));
+                    }
+                    Connection c = connectionList.get(connection);
+                    c.close();
+                }
+
+                // Close the JNDI reference
+                System.out.println(CLASS + ": Closing JNDI context");
+                ctx.close();
+            }
+        }
+        catch (Exception exp)
+        {
+            System.err.println(CLASS + ": Caught an Exception: " + exp);
+            exp.printStackTrace();
+        }
+    }
+
+
+    private byte[] generateMessage(boolean random, int messageSize)
+    {
+        byte[] sentBytes = new byte[messageSize];
+        if(random)
+        {
+            //fill the array with numbers from 0-9
+            Random rand = new Random(System.currentTimeMillis());
+            for(int r = 0 ; r < messageSize ; r++)
+            {
+                sentBytes[r] = (byte) (48 + rand.nextInt(10));
+            }
+        }
+        else
+        {
+            //use sequential numbers from 0-9
+            for(int r = 0 ; r < messageSize ; r++)
+            {
+                sentBytes[r] = (byte) (48 + (r % 10));
+            }
+        }
+        return sentBytes;
+    }
+
+
+    private void validateReceivedMessageContent(byte[] sentBytes,
+            BytesMessage msg, boolean random, int messageSize) throws JMSException
+    {
+        Long length = msg.getBodyLength();
+
+        if(length != messageSize)
+        {
+            throw new RuntimeException("Incorrect number of bytes received");
+        }
+
+        byte[] recievedBytes = new byte[length.intValue()];
+        msg.readBytes(recievedBytes);
+
+        if(random)
+        {
+            if(!Arrays.equals(sentBytes, recievedBytes))
+            {
+                throw new RuntimeException("Incorrect value of bytes received");
+            }
+        }
+        else
+        {
+            for(int r = 0 ; r < messageSize ; r++)
+            {
+                if(! (recievedBytes[r] == (byte) (48 + (r % 10))))
+                {
+                    throw new RuntimeException("Incorrect value of bytes received");
+                }
+            }
+        }
+    }
+}
+

Added: qpid/trunk/qpid/java/tools/src/main/resources/stress-test-client.properties
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/tools/src/main/resources/stress-test-client.properties?rev=1632318&view=auto
==============================================================================
--- qpid/trunk/qpid/java/tools/src/main/resources/stress-test-client.properties (added)
+++ qpid/trunk/qpid/java/tools/src/main/resources/stress-test-client.properties Thu Oct 16
14:13:20 2014
@@ -0,0 +1,3 @@
+java.naming.factory.initial = org.apache.qpid.jndi.PropertiesFileInitialContextFactory
+connectionfactory.qpidConnectionfactory = amqp://guest:guest@clientid/?brokerlist='tcp://localhost:5672'
+connectionfactory.qpidConnectionfactoryLowPrefetch=amqp://guest:guest@clientid/?brokerlist='tcp://localhost:5672?maxprefetch='10''



---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org


Mime
View raw message