You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by rg...@apache.org on 2008/05/19 15:26:20 UTC
svn commit: r657827 - in /incubator/qpid/branches/broker-queue-refactor/java:
broker/src/main/java/org/apache/qpid/server/protocol/
broker/src/main/java/org/apache/qpid/server/queue/
systests/src/main/java/org/apache/qpid/server/queue/
Author: rgodfrey
Date: Mon May 19 06:26:19 2008
New Revision: 657827
URL: http://svn.apache.org/viewvc?rev=657827&view=rev
Log:
Refactoring perf. tweaks
Modified:
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQMinaProtocolSession.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolSession.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/queue/AMQMessage.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/queue/AMQMessageHandle.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/queue/AMQPriorityQueue.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/queue/InMemoryMessageHandle.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/queue/IncomingMessage.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/queue/WeakReferenceMessageHandle.java
incubator/qpid/branches/broker-queue-refactor/java/systests/src/main/java/org/apache/qpid/server/queue/MockProtocolSession.java
Modified: incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQMinaProtocolSession.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQMinaProtocolSession.java?rev=657827&r1=657826&r2=657827&view=diff
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQMinaProtocolSession.java (original)
+++ incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQMinaProtocolSession.java Mon May 19 06:26:19 2008
@@ -113,6 +113,7 @@
private ProtocolOutputConverter _protocolOutputConverter;
private Principal _authorizedID;
private MethodDispatcher _dispatcher;
+ private ProtocolSessionIdentifier _sessionIdentifier;
public ManagedObject getManagedObject()
{
@@ -702,6 +703,7 @@
_clientVersion = new AMQShortString(_clientProperties.getString(ClientProperties.version.toString()));
}
}
+ _sessionIdentifier = new ProtocolSessionIdentifier(this);
}
private void setProtocolVersion(ProtocolVersion pv)
@@ -789,6 +791,11 @@
return _dispatcher;
}
+ public ProtocolSessionIdentifier getSessionIdentifier()
+ {
+ return _sessionIdentifier;
+ }
+
public String getClientVersion()
{
return (_clientVersion == null) ? null : _clientVersion.toString();
Modified: incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolSession.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolSession.java?rev=657827&r1=657826&r2=657827&view=diff
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolSession.java (original)
+++ incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolSession.java Mon May 19 06:26:19 2008
@@ -23,6 +23,7 @@
import javax.security.sasl.SaslServer;
import org.apache.qpid.AMQException;
+import org.apache.qpid.common.ClientProperties;
import org.apache.qpid.framing.*;
import org.apache.qpid.protocol.AMQVersionAwareProtocolSession;
import org.apache.qpid.server.AMQChannel;
@@ -35,7 +36,27 @@
public interface AMQProtocolSession extends AMQVersionAwareProtocolSession
{
+ public static final class ProtocolSessionIdentifier
+ {
+ private final Object _sessionIdentifier;
+ private final Object _sessionInstance;
+ ProtocolSessionIdentifier(AMQProtocolSession session)
+ {
+ _sessionIdentifier = session.getClientIdentifier();
+ _sessionInstance = session.getClientProperties() == null ? null : session.getClientProperties().getObject(ClientProperties.instance.toAMQShortString());
+ }
+
+ public Object getSessionIdentifier()
+ {
+ return _sessionIdentifier;
+ }
+
+ public Object getSessionInstance()
+ {
+ return _sessionInstance;
+ }
+ }
public static interface Task
{
@@ -175,5 +196,7 @@
public MethodRegistry getMethodRegistry();
public MethodDispatcher getMethodDispatcher();
+
+ public ProtocolSessionIdentifier getSessionIdentifier();
}
Modified: incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/queue/AMQMessage.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/queue/AMQMessage.java?rev=657827&r1=657826&r2=657827&view=diff
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/queue/AMQMessage.java (original)
+++ incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/queue/AMQMessage.java Mon May 19 06:26:19 2008
@@ -33,8 +33,7 @@
import org.apache.qpid.server.store.MessageStore;
import org.apache.qpid.server.store.StoreContext;
import org.apache.qpid.server.txn.TransactionalContext;
-import org.apache.qpid.server.exchange.Exchange;
-
+
import java.util.Iterator;
import java.util.concurrent.atomic.AtomicInteger;
@@ -54,21 +53,27 @@
/** Holds the transactional context in which this message is being processed. */
private StoreContext _storeContext;
+ /** Flag to indicate that this message requires 'immediate' delivery. */
+
+ private static final byte IMMEDIATE = 0x01;
+
/**
* Flag to indicate whether this message has been delivered to a consumer. Used in implementing return functionality
* for messages published with the 'immediate' flag.
*/
- private boolean _deliveredToConsumer;
- /** Flag to indicate that this message requires 'immediate' delivery. */
- private boolean _immediate;
+ private static final byte DELIVERED_TO_CONSUMER = 0x02;
+
+ private byte _flags = 0;
private long _expiration;
- private Object _publisherClientInstance;
- private Object _publisherIdentifier;
private final long _size;
+ private AMQProtocolSession.ProtocolSessionIdentifier _sessionIdentifier;
+ private static final byte IMMEDIATE_AND_DELIVERED = (byte) (IMMEDIATE | DELIVERED_TO_CONSUMER);
+
+
/**
* Used to iterate through all the body frames associated with this message. Will not keep all the data in memory
@@ -211,7 +216,11 @@
{
_messageHandle = messageHandle;
_storeContext = storeConext;
- _immediate = info.isImmediate();
+
+ if(info.isImmediate())
+ {
+ _flags |= IMMEDIATE;
+ }
_size = messageHandle.getBodySize(storeConext);
}
@@ -221,7 +230,7 @@
{
_messageHandle = msg._messageHandle;
_storeContext = msg._storeContext;
- _deliveredToConsumer = msg._deliveredToConsumer;
+ _flags = msg._flags;
_size = msg._size;
}
@@ -289,10 +298,7 @@
{
return true;
}
- // if (_log.isDebugEnabled())
- // {
- // _log.debug("Ref count on message " + debugIdentity() + " incremented " + Arrays.asList(Thread.currentThread().getStackTrace()).subList(3, 6));
- // }
+
}
/**
@@ -322,11 +328,6 @@
try
{
- // if (_log.isDebugEnabled())
- // {
- // _log.debug("Decremented ref count on message " + debugIdentity() + " is zero; removing message" + Arrays.asList(Thread.currentThread().getStackTrace()).subList(3, 6));
- // }
-
// must check if the handle is null since there may be cases where we decide to throw away a message
// and the handle has not yet been constructed
if (_messageHandle != null)
@@ -359,12 +360,12 @@
*/
public boolean getDeliveredToConsumer()
{
- return _deliveredToConsumer;
+ return (_flags & DELIVERED_TO_CONSUMER) != 0;
}
public boolean isPersistent() throws AMQException
{
- return _messageHandle.isPersistent(getStoreContext());
+ return _messageHandle.isPersistent();
}
/**
@@ -376,7 +377,7 @@
public boolean immediateAndNotDelivered()
{
- return (_immediate && !_deliveredToConsumer);
+ return (_flags & IMMEDIATE_AND_DELIVERED) == IMMEDIATE;
}
@@ -428,7 +429,7 @@
*/
public void setDeliveredToConsumer()
{
- _deliveredToConsumer = true;
+ _flags |= DELIVERED_TO_CONSUMER;
}
@@ -444,26 +445,22 @@
}
- public void setPublisherClientInstance(final Object publisherClientInstance)
- {
- _publisherClientInstance = publisherClientInstance;
- }
-
public Object getPublisherClientInstance()
{
- return _publisherClientInstance;
+ return _sessionIdentifier.getSessionInstance();
}
public Object getPublisherIdentifier()
{
- return _publisherIdentifier;
+ return _sessionIdentifier.getSessionIdentifier();
}
- public void setPublisherIdentifier(final Object publisherIdentifier)
+ public void setClientIdentifier(final AMQProtocolSession.ProtocolSessionIdentifier sessionIdentifier)
{
- _publisherIdentifier = publisherIdentifier;
+ _sessionIdentifier = sessionIdentifier;
}
+
public String toString()
{
// return "Message[" + debugIdentity() + "]: " + _messageId + "; ref count: " + _referenceCount + "; taken : " +
Modified: incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/queue/AMQMessageHandle.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/queue/AMQMessageHandle.java?rev=657827&r1=657826&r2=657827&view=diff
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/queue/AMQMessageHandle.java (original)
+++ incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/queue/AMQMessageHandle.java Mon May 19 06:26:19 2008
@@ -67,7 +67,7 @@
void setRedelivered(boolean redelivered);
- boolean isPersistent(StoreContext context) throws AMQException;
+ boolean isPersistent();
void setPublishAndContentHeaderBody(StoreContext storeContext, MessagePublishInfo messagePublishInfo,
ContentHeaderBody contentHeaderBody)
Modified: incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/queue/AMQPriorityQueue.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/queue/AMQPriorityQueue.java?rev=657827&r1=657826&r2=657827&view=diff
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/queue/AMQPriorityQueue.java (original)
+++ incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/queue/AMQPriorityQueue.java Mon May 19 06:26:19 2008
@@ -22,6 +22,8 @@
import org.apache.qpid.framing.AMQShortString;
import org.apache.qpid.server.virtualhost.VirtualHost;
+import org.apache.qpid.server.subscription.SubscriptionList;
+import org.apache.qpid.server.subscription.Subscription;
import org.apache.qpid.AMQException;
public class AMQPriorityQueue extends SimpleAMQQueue
@@ -37,5 +39,29 @@
super(name, durable, owner, autoDelete, virtualHost, new PriorityQueueList.Factory(priorities));
}
+ @Override
+ protected void checkSubscriptionsNotAheadOfDelivery(final QueueEntry entry)
+ {
+ // check that all subscriptions are not in advance of the entry
+ SubscriptionList.SubscriptionNodeIterator subIter = _subscriptionList.iterator();
+ while(subIter.advance() && !entry.isAcquired())
+ {
+ final Subscription subscription = subIter.getNode().getSubscription();
+ QueueEntry subnode = subscription.getLastSeenEntry();
+ while((entry.compareTo(subnode) < 0) && !entry.isAcquired())
+ {
+ if(subscription.setLastSeenEntry(subnode,entry))
+ {
+ break;
+ }
+ else
+ {
+ subnode = subscription.getLastSeenEntry();
+ }
+ }
+
+ }
+ }
+
}
Modified: incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/queue/InMemoryMessageHandle.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/queue/InMemoryMessageHandle.java?rev=657827&r1=657826&r2=657827&view=diff
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/queue/InMemoryMessageHandle.java (original)
+++ incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/queue/InMemoryMessageHandle.java Mon May 19 06:26:19 2008
@@ -22,10 +22,10 @@
import java.util.LinkedList;
import java.util.List;
+import java.util.Collections;
+import java.util.ArrayList;
import org.apache.qpid.AMQException;
-import org.apache.qpid.framing.BasicContentHeaderProperties;
-import org.apache.qpid.framing.ContentBody;
import org.apache.qpid.framing.ContentHeaderBody;
import org.apache.qpid.framing.abstraction.MessagePublishInfo;
import org.apache.qpid.framing.abstraction.ContentChunk;
@@ -40,7 +40,7 @@
private MessagePublishInfo _messagePublishInfo;
- private List<ContentChunk> _contentBodies = new LinkedList<ContentChunk>();
+ private List<ContentChunk> _contentBodies;
private boolean _redelivered;
@@ -86,7 +86,22 @@
public void addContentBodyFrame(StoreContext storeContext, ContentChunk contentBody, boolean isLastContentBody)
throws AMQException
{
- _contentBodies.add(contentBody);
+ if(_contentBodies == null)
+ {
+ if(isLastContentBody)
+ {
+ _contentBodies = Collections.singletonList(contentBody);
+ }
+ else
+ {
+ _contentBodies = new ArrayList<ContentChunk>();
+ _contentBodies.add(contentBody);
+ }
+ }
+ else
+ {
+ _contentBodies.add(contentBody);
+ }
}
public MessagePublishInfo getMessagePublishInfo(StoreContext context) throws AMQException
@@ -105,12 +120,9 @@
_redelivered = redelivered;
}
- public boolean isPersistent(StoreContext context) throws AMQException
+ public boolean isPersistent()
{
- //todo remove literal values to a constant file such as AMQConstants in common
- ContentHeaderBody chb = getContentHeaderBody(context);
- return chb.properties instanceof BasicContentHeaderProperties &&
- ((BasicContentHeaderProperties) chb.properties).getDeliveryMode() == 2;
+ return false;
}
/**
@@ -125,6 +137,10 @@
{
_messagePublishInfo = messagePublishInfo;
_contentHeaderBody = contentHeaderBody;
+ if(contentHeaderBody.bodySize == 0)
+ {
+ _contentBodies = Collections.EMPTY_LIST;
+ }
_arrivalTime = System.currentTimeMillis();
}
Modified: incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/queue/IncomingMessage.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/queue/IncomingMessage.java?rev=657827&r1=657826&r2=657827&view=diff
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/queue/IncomingMessage.java (original)
+++ incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/queue/IncomingMessage.java Mon May 19 06:26:19 2008
@@ -176,13 +176,10 @@
message = new AMQMessage(_messageHandle,_txnContext.getStoreContext(), _messagePublishInfo);
- message.setPublisherIdentifier(_publisher.getClientIdentifier());
+
message.setExpiration(_expiration);
+ message.setClientIdentifier(_publisher.getSessionIdentifier());
- if (_publisher.getClientProperties() != null)
- {
- message.setPublisherClientInstance(_publisher.getClientProperties().getObject(ClientProperties.instance.toAMQShortString()));
- }
Modified: incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java?rev=657827&r1=657826&r2=657827&view=diff
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java (original)
+++ incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java Mon May 19 06:26:19 2008
@@ -81,7 +81,7 @@
private final AtomicBoolean _quiesced = new AtomicBoolean(false);
- private final SubscriptionList _subscriptionList = new SubscriptionList(this);
+ protected final SubscriptionList _subscriptionList = new SubscriptionList(this);
private final AtomicReference<SubscriptionList.SubscriptionNode> _lastSubscriptionNode = new AtomicReference<SubscriptionList.SubscriptionNode>(_subscriptionList.getHead());
private boolean _exclusiveSubscriber;
@@ -389,13 +389,14 @@
}
else
{
- deliverMessage(sub, entry);
QueueEntry queueEntryNode = sub.getLastSeenEntry();
if(_entries.next(queueEntryNode) == entry)
{
sub.setLastSeenEntry(queueEntryNode,entry);
}
+ deliverMessage(sub, entry);
+
}
}
}
@@ -414,25 +415,8 @@
}
else if(!entry.isAcquired())
{
- // check that all subscriptions are not in advance of the entry
- SubscriptionList.SubscriptionNodeIterator subIter = _subscriptionList.iterator();
- while(subIter.advance() && !entry.isAcquired())
- {
- final Subscription subscription = subIter.getNode().getSubscription();
- QueueEntry subnode = subscription.getLastSeenEntry();
- while((entry.compareTo(subnode) < 0) && !entry.isAcquired())
- {
- if(subscription.setLastSeenEntry(subnode,entry))
- {
- break;
- }
- else
- {
- subnode = subscription.getLastSeenEntry();
- }
- }
+ checkSubscriptionsNotAheadOfDelivery(entry);
- }
deliverAsync();
}
@@ -451,6 +435,11 @@
}
+ protected void checkSubscriptionsNotAheadOfDelivery(final QueueEntry entry)
+ {
+ // This method is only required for queues which mess with ordering
+ }
+
private void incrementQueueSize(final AMQMessage message)
{
getAtomicQueueSize().addAndGet(message.getSize());
@@ -1204,6 +1193,7 @@
throws AMQException
{
QueueEntry node = sub.getLastSeenEntry();
+
while(node.isAcquired() || node.isDeleted() || node.expired())
{
if(!node.isAcquired() && !node.isDeleted() && node.expired())
Modified: incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/queue/WeakReferenceMessageHandle.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/queue/WeakReferenceMessageHandle.java?rev=657827&r1=657826&r2=657827&view=diff
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/queue/WeakReferenceMessageHandle.java (original)
+++ incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/queue/WeakReferenceMessageHandle.java Mon May 19 06:26:19 2008
@@ -26,7 +26,6 @@
import java.util.List;
import org.apache.qpid.AMQException;
-import org.apache.qpid.framing.BasicContentHeaderProperties;
import org.apache.qpid.framing.ContentHeaderBody;
import org.apache.qpid.framing.abstraction.MessagePublishInfo;
import org.apache.qpid.framing.abstraction.ContentChunk;
@@ -50,8 +49,6 @@
private final Long _messageId;
private long _arrivalTime;
- private boolean _persistent;
-
public WeakReferenceMessageHandle(final Long messageId, MessageStore messageStore)
{
@@ -175,9 +172,9 @@
_redelivered = redelivered;
}
- public boolean isPersistent(StoreContext context) throws AMQException
+ public boolean isPersistent()
{
- return _persistent;
+ return true;
}
/**
@@ -203,12 +200,6 @@
MessageMetaData mmd = new MessageMetaData(publishBody, contentHeaderBody, _contentBodies.size(), arrivalTime);
-
-
-
- _persistent = contentHeaderBody.properties instanceof BasicContentHeaderProperties &&
- ((BasicContentHeaderProperties) contentHeaderBody.properties).getDeliveryMode() == 2;
-
_messageStore.storeMessageMetaData(storeContext, _messageId, mmd);
Modified: incubator/qpid/branches/broker-queue-refactor/java/systests/src/main/java/org/apache/qpid/server/queue/MockProtocolSession.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/systests/src/main/java/org/apache/qpid/server/queue/MockProtocolSession.java?rev=657827&r1=657826&r2=657827&view=diff
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/systests/src/main/java/org/apache/qpid/server/queue/MockProtocolSession.java (original)
+++ incubator/qpid/branches/broker-queue-refactor/java/systests/src/main/java/org/apache/qpid/server/queue/MockProtocolSession.java Mon May 19 06:26:19 2008
@@ -215,6 +215,11 @@
return null; //To change body of implemented methods use File | Settings | File Templates.
}
+ public ProtocolSessionIdentifier getSessionIdentifier()
+ {
+ return null;
+ }
+
public byte getProtocolMajorVersion()
{
return getProtocolVersion().getMajorVersion();