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 2010/05/12 16:55:59 UTC
svn commit: r943534 - in
/qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/queue:
QueueEntryImpl.java SimpleQueueEntryList.java
Author: ritchiem
Date: Wed May 12 14:55:59 2010
New Revision: 943534
URL: http://svn.apache.org/viewvc?rev=943534&view=rev
Log:
QPID-2567 : Update the scavenge code based on feedback from Andrew Kennedy and Rob Godfrey.
Modified:
qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/QueueEntryImpl.java
qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleQueueEntryList.java
Modified: qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/QueueEntryImpl.java
URL: http://svn.apache.org/viewvc/qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/QueueEntryImpl.java?rev=943534&r1=943533&r2=943534&view=diff
==============================================================================
--- qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/QueueEntryImpl.java (original)
+++ qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/QueueEntryImpl.java Wed May 12 14:55:59 2010
@@ -391,25 +391,8 @@ public class QueueEntryImpl implements Q
if(state != DELETED_STATE && _stateUpdater.compareAndSet(this,state,DELETED_STATE))
{
- //
- // We can't advanceHead() before the 'if' as we need to check if we
- // are at the head or not. If we are not at the head then we should
- // try scavenge the list.
- //
- // An alternative would be to provide a remove(this) method on
- // the queueEntryList, this would however be less efficient as it
- // would require that we walk the queue to locate this before we
- // could perform the removal.
- //
- if (((QueueEntryImpl) _queueEntryList.getHead()).nextNode() != this)
- {
- _queueEntryList.advanceHead();
- _queueEntryList.scavenge();
- }
- else
- {
- _queueEntryList.advanceHead();
- }
+
+ _queueEntryList.advanceHead();
return true;
}
Modified: qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleQueueEntryList.java
URL: http://svn.apache.org/viewvc/qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleQueueEntryList.java?rev=943534&r1=943533&r2=943534&view=diff
==============================================================================
--- qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleQueueEntryList.java (original)
+++ qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleQueueEntryList.java Wed May 12 14:55:59 2010
@@ -46,7 +46,6 @@ public class SimpleQueueEntryList implem
(QueueEntryImpl.class, QueueEntryImpl.class, "_next");
private AtomicLong _scavenges = new AtomicLong(0L);
- private AtomicReference<Thread> _scavenger = new AtomicReference<Thread>(null);
private static final long SCAVENGE_COUNT = Integer.getInteger("qpid.queue.scavenge_count", 50);
@@ -61,90 +60,29 @@ public class SimpleQueueEntryList implem
void advanceHead()
{
- QueueEntryImpl head = _head.nextNode();
- while(head._next != null && head.isDeleted())
- {
+ QueueEntryImpl next = _head.nextNode();
+ QueueEntryImpl newNext = _head.getNext();
- final QueueEntryImpl newhead = head.nextNode();
- if(newhead != null)
- {
- _nextUpdater.compareAndSet(_head,head, newhead);
- }
- head = _head.nextNode();
- }
- }
-
- void scavenge()
- {
- _scavenges.incrementAndGet();
-
- if (_scavenges.get() < SCAVENGE_COUNT)
+ if (!_nextUpdater.compareAndSet(_head, next, newNext))
{
- return;
+ _scavenges.incrementAndGet();
}
- try
+ if (_scavenges.get() > SCAVENGE_COUNT)
{
+ _scavenges.set(0L);
+ scavenge();
+ }
+ }
- if (_scavenger.compareAndSet(null, Thread.currentThread()))
- {
- // only delete the number of scavenges requested.
- // This should keep things fair when we have multiple consumers
- // using selectors that will be calling this.
- // With multiple consumers this will also be called but
- // advanceHead should take care of it in most instances.
- // Often it will be the ExpiredMessageTask so will not
- // affect throughput.
- long deletesToPerform = _scavenges.getAndSet(0);
-
-
- QueueEntryImpl root = _head;
- QueueEntryImpl next = root.nextNode();
-
- do
- {
-
- while (next._next != null && next.isDeleted())
- {
-
- final QueueEntryImpl newhead = next.nextNode();
- if (newhead != null)
- {
- _nextUpdater.compareAndSet(root, next, newhead);
- }
- next = root.nextNode();
- }
- if (next._next != null)
- {
- if (!next.isDeleted())
- {
- root = next;
- next = root.nextNode();
- deletesToPerform--;
- }
-
- // Limit the number of scavenges performed by this
- // thread. For fairness.
- if (deletesToPerform == 0)
- {
- return;
- }
-
- }
- else
- {
- break;
- }
+ private void scavenge()
+ {
+ QueueEntryImpl next = _head.getNext();
- }
- while (next != null && next._next != null);
- }
- }
- finally
+ while (next != null)
{
- _scavenger.compareAndSet(Thread.currentThread(), null);
+ next = next.getNext();
}
-
}
public AMQQueue getQueue()
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:commits-subscribe@qpid.apache.org