Author: rupertlssmith
Date: Wed Oct 10 08:47:38 2007
New Revision: 583520
URL: http://svn.apache.org/viewvc?rev=583520&view=rev
Log:
Merged revisions 583518 via svnmerge from
https://svn.apache.org/repos/asf/incubator/qpid/branches/M2.1
........
r583518 | rupertlssmith | 2007-10-10 16:45:56 +0100 (Wed, 10 Oct 2007) | 1
line
Changed maxPending to be by message correlation id.
........
Modified:
incubator/qpid/branches/M2/ (props changed)
incubator/qpid/branches/M2/java/perftests/src/main/java/org/apache/qpid/requestreply/PingPongProducer.java
Propchange: incubator/qpid/branches/M2/
------------------------------------------------------------------------------
--- svnmerge-integrated (original)
+++ svnmerge-integrated Wed Oct 10 08:47:38 2007
@@ -1 +1 @@
-/incubator/qpid/branches/M2.1:1-573736,573738-577772,577774-578732,578734,578736-578744,578746-578827,578829-583418,583438-583483
+/incubator/qpid/branches/M2.1:1-573736,573738-577772,577774-578732,578734,578736-578744,578746-578827,578829-583418,583438-583483,583518
Modified:
incubator/qpid/branches/M2/java/perftests/src/main/java/org/apache/qpid/requestreply/PingPongProducer.java
URL:
http://svn.apache.org/viewvc/incubator/qpid/branches/M2/java/perftests/src/main/java/org/apache/qpid/requestreply/PingPongProducer.java?rev=583520&r1=583519&r2=583520&view=diff
==============================================================================
---
incubator/qpid/branches/M2/java/perftests/src/main/java/org/apache/qpid/requestreply/PingPongProducer.java
(original)
+++
incubator/qpid/branches/M2/java/perftests/src/main/java/org/apache/qpid/requestreply/PingPongProducer.java
Wed Oct 10 08:47:38 2007
@@ -419,15 +419,6 @@
*/
protected int _maxPendingSize;
- /**
- * 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 static final Object _sendPauseMonitor = new Object();
-
- /** Keeps a count of the number of message currently sent but not
received. */
- 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);
@@ -898,7 +889,7 @@
if ((_maxPendingSize > 0))
{
// Decrement the count of sent but not yet received
messages.
- int unreceived = _unreceived.decrementAndGet();
+ int unreceived =
perCorrelationId._unreceived.decrementAndGet();
int unreceivedSize =
(unreceived * ((_messageSize == 0) ? 1 :
_messageSize))
/ (_isPubSub ? getConsumersPerDestination() : 1);
@@ -906,11 +897,11 @@
// log.debug("unreceived = " + unreceived);
// log.debug("unreceivedSize = " + unreceivedSize);
- synchronized (_sendPauseMonitor)
+ synchronized (perCorrelationId._sendPauseMonitor)
{
if (unreceivedSize < _maxPendingSize)
{
- _sendPauseMonitor.notifyAll();
+ perCorrelationId._sendPauseMonitor.notify();
}
}
}
@@ -1169,10 +1160,14 @@
// Prompt the user to kill the broker when doing failover testing.
_failBeforeSend = waitForUserToPromptOnFailure(_failBeforeSend);
+ // Get the test setup for the correlation id.
+ String correlationID = message.getJMSCorrelationID();
+ PerCorrelationId perCorrelationId =
perCorrelationIds.get(correlationID);
+
// If necessary, wait until the max pending message size comes
within its limit.
if (_maxPendingSize > 0)
{
- synchronized (_sendPauseMonitor)
+ synchronized (perCorrelationId._sendPauseMonitor)
{
// Used to keep track of the number of times that send has
to wait.
int numWaits = 0;
@@ -1184,7 +1179,7 @@
while (true)
{
// Get the size estimate of sent but not yet received
messages.
- int unreceived = _unreceived.get();
+ int unreceived = perCorrelationId._unreceived.get();
int unreceivedSize =
(unreceived * ((_messageSize == 0) ? 1 :
_messageSize))
/ (_isPubSub ? getConsumersPerDestination() : 1);
@@ -1212,7 +1207,7 @@
try
{
long start = System.nanoTime();
- _sendPauseMonitor.wait(10000);
+ perCorrelationId._sendPauseMonitor.wait(10000);
long end = System.nanoTime();
// Count the wait only if it was for > 99% of
the requested wait time.
@@ -1255,7 +1250,8 @@
// in pub/sub mode.
if (_maxPendingSize > 0)
{
- int newUnreceivedCount = _unreceived.addAndGet(_isPubSub ?
getConsumersPerDestination() : 1);
+ int newUnreceivedCount =
+ perCorrelationId._unreceived.addAndGet(_isPubSub ?
getConsumersPerDestination() : 1);
// log.debug("newUnreceivedCount = " + newUnreceivedCount);
}
@@ -1676,5 +1672,14 @@
/** Holds the last timestamp that the timeout was reset to. */
Long timeOutStart;
+
+ /**
+ * 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.
+ */
+ final Object _sendPauseMonitor = new Object();
+
+ /** Keeps a count of the number of message currently sent but not
received. */
+ AtomicInteger _unreceived = new AtomicInteger(0);
}
}
