Author: rupertlssmith
Date: Wed Oct 10 04:46:59 2007
New Revision: 583438
URL: http://svn.apache.org/viewvc?rev=583438&view=rev
Log:
Changed tests to use global maxPending limit, rather than per producer.
Modified:
incubator/qpid/branches/M2.1/java/perftests/src/main/java/org/apache/qpid/requestreply/PingPongProducer.java
Modified:
incubator/qpid/branches/M2.1/java/perftests/src/main/java/org/apache/qpid/requestreply/PingPongProducer.java
URL:
http://svn.apache.org/viewvc/incubator/qpid/branches/M2.1/java/perftests/src/main/java/org/apache/qpid/requestreply/PingPongProducer.java?rev=583438&r1=583437&r2=583438&view=diff
==============================================================================
---
incubator/qpid/branches/M2.1/java/perftests/src/main/java/org/apache/qpid/requestreply/PingPongProducer.java
(original)
+++
incubator/qpid/branches/M2.1/java/perftests/src/main/java/org/apache/qpid/requestreply/PingPongProducer.java
Wed Oct 10 04:46:59 2007
@@ -423,10 +423,10 @@
* Holds a monitor which is used to synchronize sender and receivers
threads, where the sender has elected
* to wait until the number of unreceived message is reduced before
continuing to send.
*/
- protected final Object _sendPauseMonitor = new Object();
+ protected static final Object _sendPauseMonitor = new Object();
/** Keeps a count of the number of message currently sent but not
received. */
- protected AtomicInteger _unreceived = new AtomicInteger(0);
+ protected static AtomicInteger _unreceived = new AtomicInteger(0);
/** A source for providing sequential unique correlation ids. These will
be unique within the same JVM. */
private static AtomicLong _correlationIdGenerator = new AtomicLong(0L);
@@ -868,7 +868,7 @@
long timestamp = getTimestamp(message);
long pingTime = now - timestamp;
- // NDC.push("id" + instanceId + "/cons" + consumerNo);
+ NDC.push("id" + instanceId + "/cons" + consumerNo);
// Extract the messages correlation id.
String correlationID = message.getJMSCorrelationID();
@@ -894,9 +894,6 @@
// log.debug("Reply was expected, decrementing the latch
for the id, " + correlationID);
- // log.debug("unreceived = " + unreceived);
- // log.debug("unreceivedSize = " + unreceivedSize);
-
// Release waiting senders if there are some and using
maxPending limit.
if ((_maxPendingSize > 0))
{
@@ -906,11 +903,14 @@
(unreceived * ((_messageSize == 0) ? 1 :
_messageSize))
/ (_isPubSub ? getConsumersPerDestination() : 1);
+ // log.debug("unreceived = " + unreceived);
+ log.debug("unreceivedSize = " + unreceivedSize);
+
synchronized (_sendPauseMonitor)
{
if (unreceivedSize < _maxPendingSize)
{
- _sendPauseMonitor.notify();
+ _sendPauseMonitor.notifyAll();
}
}
}
