Made PerfRate a more thread safe.

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

activemq-core/src/test/java/org/apache/activemq/perf/PerfConsumer.java

@@ -51,7 +51,7 @@ public class PerfConsumer implements MessageListener{
     }
     public void start() throws JMSException{
         connection.start();
-        rate.getRate();
+        rate.reset();
     }
     public void stop() throws JMSException{
         connection.stop();

activemq-core/src/test/java/org/apache/activemq/perf/PerfProducer.java

@@ -59,10 +59,10 @@ public class PerfProducer implements Runnable {
     	
 	synchronized public void start() throws JMSException{
 		if( !running ) {
+	        rate.reset();
 			running = true;
 	        connection.start();
 	        new Thread(this).start(); 
-	        rate.reset();
 		}
     }
     public void stop() throws JMSException, InterruptedException{

activemq-core/src/test/java/org/apache/activemq/perf/PerfRate.java

@@ -21,9 +21,11 @@ package org.apache.activemq.perf;
  * @version $Revision: 1.3 $
  */
 public class PerfRate{
+	
     protected int totalCount;
     protected int count;
     protected long startTime=System.currentTimeMillis();
+    
     /**
      * @return Returns the count.
      */
@@ -31,7 +33,7 @@ public class PerfRate{
         return totalCount;
     }
 
-    public void increment(){
+    synchronized public void increment(){
         totalCount++;
         count++;
     }
@@ -43,6 +45,19 @@ public class PerfRate{
         return result;
     }
     
+    /**
+     * Resets the rate sampling.
+     */
+    synchronized public PerfRate cloneAndReset() {
+    	PerfRate rc = new PerfRate();
+    	rc.totalCount = totalCount;
+    	rc.count=count;
+    	rc.startTime=startTime;    	
+        count=0;
+        startTime=System.currentTimeMillis();
+        return rc;
+    }
+    
     /**
      * Resets the rate sampling.
      */

activemq-core/src/test/java/org/apache/activemq/perf/SimpleTopicTest.java

@@ -43,10 +43,10 @@ public class SimpleTopicTest extends TestCase{
     //protected String bindAddress="vm://localhost";
     protected PerfProducer[] producers;
     protected PerfConsumer[] consumers;
-    protected String DESTINATION_NAME=getClass().toString();
+    protected String DESTINATION_NAME=getClass().getName();
     protected int SAMPLE_COUNT = 30;
     protected long SAMPLE_INTERVAL = 2000;
-    protected int NUMBER_OF_CONSUMERS=1;
+    protected int NUMBER_OF_CONSUMERS=10;
     protected int NUMBER_OF_PRODUCERS=1;
     protected int PAYLOAD_SIZE=1024;
     protected byte[] array=null;
@@ -163,27 +163,23 @@ public class SimpleTopicTest extends TestCase{
         int totalRate=0;
         int totalCount=0;
         for(int i=0;i<producers.length;i++){
-            totalRate+=producers[i].getRate().getRate();
+        	PerfRate rate = producers[i].getRate().cloneAndReset();
-            totalCount+=producers[i].getRate().getTotalCount();
+            totalRate+=rate.getRate();
+            totalCount+=rate.getTotalCount();
         }
         int avgRate = totalRate/producers.length;
         log.info("Avg producer rate = "+avgRate+" msg/sec | Total rate = "+totalRate+", sent = "+totalCount);
-        for(int i=0;i<producers.length;i++){
-            producers[i].getRate().reset();
-        }
     }
 
     protected void dumpConsumerRate(){
         int totalRate=0;
         int totalCount=0;
         for(int i=0;i<consumers.length;i++){
-            totalRate+=consumers[i].getRate().getRate();
+        	PerfRate rate = consumers[i].getRate().cloneAndReset();
-            totalCount+=consumers[i].getRate().getTotalCount();
+            totalRate+=rate.getRate();
+            totalCount+=rate.getTotalCount();
         }
         int avgRate = totalRate/consumers.length;
         log.info("Avg consumer rate = "+avgRate+" msg/sec | Total rate = "+totalRate+", received = "+totalCount);
-        for(int i=0;i<consumers.length;i++){
-            consumers[i].getRate().reset();
-        }
     }
 }