From 91dfa2865cb9998a379e099ff58e830b4b1ba8a4 Mon Sep 17 00:00:00 2001 From: Robert Godfrey Date: Wed, 13 Feb 2008 18:10:53 +0000 Subject: QPID-790 : Performance Improvements git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/branches/M2.1@627552 13f79535-47bb-0310-9956-ffa450edef68 --- .../java/org/apache/qpid/server/AMQChannel.java | 6 ++- .../qpid/server/exchange/DestWildExchange.java | 2 +- .../server/handler/BasicPublishMethodHandler.java | 3 +- .../amqp0_8/ProtocolOutputConverterImpl.java | 18 +++---- .../amqp0_9/ProtocolOutputConverterImpl.java | 63 +++++++++++++--------- .../org/apache/qpid/server/queue/AMQMessage.java | 17 +++++- .../qpid/server/exchange/DestWildExchangeTest.java | 5 ++ .../qpid/server/queue/AMQQueueAlertTest.java | 5 ++ .../qpid/server/queue/AMQQueueMBeanTest.java | 5 ++ .../apache/qpid/framing/CompositeAMQDataBlock.java | 21 ++++---- .../org/apache/qpid/framing/DeferredDataBlock.java | 50 +++++++++++++++++ .../qpid/framing/SmallCompositeAMQDataBlock.java | 22 ++++---- .../framing/abstraction/MessagePublishInfo.java | 2 + .../qpid/framing/amqp_0_9/MethodConverter_0_9.java | 9 +++- .../qpid/framing/amqp_8_0/MethodConverter_8_0.java | 7 ++- .../java/org/apache/qpid/server/ack/TxAckTest.java | 5 ++ .../java/org/apache/qpid/server/queue/AckTest.java | 5 ++ .../qpid/server/queue/MessageTestHelper.java | 5 ++ .../qpid/server/store/TestReferenceCounting.java | 5 ++ 19 files changed, 192 insertions(+), 63 deletions(-) create mode 100644 java/common/src/main/java/org/apache/qpid/framing/DeferredDataBlock.java diff --git a/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java b/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java index 1bf0cd027a..10184a79e5 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java +++ b/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java @@ -35,6 +35,7 @@ import org.apache.qpid.server.ack.UnacknowledgedMessageMap; import org.apache.qpid.server.ack.UnacknowledgedMessageMapImpl; import org.apache.qpid.server.exchange.MessageRouter; import org.apache.qpid.server.exchange.NoRouteException; +import org.apache.qpid.server.exchange.Exchange; import org.apache.qpid.server.protocol.AMQProtocolSession; import org.apache.qpid.server.queue.*; import org.apache.qpid.server.store.MessageStore; @@ -199,11 +200,12 @@ public class AMQChannel _prefetch_HighWaterMark = prefetchCount; } - public void setPublishFrame(MessagePublishInfo info, AMQProtocolSession publisher) throws AMQException + public void setPublishFrame(MessagePublishInfo info, AMQProtocolSession publisher, final Exchange e) throws AMQException { _currentMessage = new AMQMessage(_messageStore.getNewMessageId(), info, _txnContext); _currentMessage.setPublisher(publisher); + _currentMessage.setExchange(e); } public void publishContentHeader(ContentHeaderBody contentHeaderBody, AMQProtocolSession protocolSession) @@ -285,7 +287,7 @@ public class AMQChannel { try { - _exchanges.routeContent(_currentMessage); + _currentMessage.route(); } catch (NoRouteException e) { diff --git a/java/broker/src/main/java/org/apache/qpid/server/exchange/DestWildExchange.java b/java/broker/src/main/java/org/apache/qpid/server/exchange/DestWildExchange.java index 75be86a387..19172b98f3 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/exchange/DestWildExchange.java +++ b/java/broker/src/main/java/org/apache/qpid/server/exchange/DestWildExchange.java @@ -239,7 +239,7 @@ public class DestWildExchange extends AbstractExchange { MessagePublishInfo info = payload.getMessagePublishInfo(); - final AMQShortString routingKey = normalize(info.getRoutingKey()); + final AMQShortString routingKey = info.getRoutingKey(); List queues = getMatchedQueues(routingKey); // if we have no registered queues we have nothing to do diff --git a/java/broker/src/main/java/org/apache/qpid/server/handler/BasicPublishMethodHandler.java b/java/broker/src/main/java/org/apache/qpid/server/handler/BasicPublishMethodHandler.java index 66afc61751..687ec33ba0 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/handler/BasicPublishMethodHandler.java +++ b/java/broker/src/main/java/org/apache/qpid/server/handler/BasicPublishMethodHandler.java @@ -91,7 +91,8 @@ public class BasicPublishMethodHandler implements StateAwareMethodListener 0; } + public void setExchange(final Exchange exchange) + { + _exchange = exchange; + } + + public void route() throws AMQException + { + _exchange.route(this); + } + /** * Used to iterate through all the body frames associated with this message. Will not keep all the data in memory * therefore is memory-efficient. diff --git a/java/broker/src/test/java/org/apache/qpid/server/exchange/DestWildExchangeTest.java b/java/broker/src/test/java/org/apache/qpid/server/exchange/DestWildExchangeTest.java index 8e5879a51e..7e2d56b460 100644 --- a/java/broker/src/test/java/org/apache/qpid/server/exchange/DestWildExchangeTest.java +++ b/java/broker/src/test/java/org/apache/qpid/server/exchange/DestWildExchangeTest.java @@ -589,6 +589,11 @@ public class DestWildExchangeTest extends TestCase return null; } + public void setExchange(AMQShortString exchange) + { + + } + public boolean isImmediate() { return false; diff --git a/java/broker/src/test/java/org/apache/qpid/server/queue/AMQQueueAlertTest.java b/java/broker/src/test/java/org/apache/qpid/server/queue/AMQQueueAlertTest.java index 81b0ae2213..fbd9e65480 100644 --- a/java/broker/src/test/java/org/apache/qpid/server/queue/AMQQueueAlertTest.java +++ b/java/broker/src/test/java/org/apache/qpid/server/queue/AMQQueueAlertTest.java @@ -242,6 +242,11 @@ public class AMQQueueAlertTest extends TestCase return null; } + public void setExchange(AMQShortString exchange) + { + //To change body of implemented methods use File | Settings | File Templates. + } + public boolean isImmediate() { return immediate; diff --git a/java/broker/src/test/java/org/apache/qpid/server/queue/AMQQueueMBeanTest.java b/java/broker/src/test/java/org/apache/qpid/server/queue/AMQQueueMBeanTest.java index d86c90bdae..e72e1bf1f0 100644 --- a/java/broker/src/test/java/org/apache/qpid/server/queue/AMQQueueMBeanTest.java +++ b/java/broker/src/test/java/org/apache/qpid/server/queue/AMQQueueMBeanTest.java @@ -234,6 +234,11 @@ public class AMQQueueMBeanTest extends TestCase return null; } + public void setExchange(AMQShortString exchange) + { + //To change body of implemented methods use File | Settings | File Templates. + } + public boolean isImmediate() { return immediate; diff --git a/java/common/src/main/java/org/apache/qpid/framing/CompositeAMQDataBlock.java b/java/common/src/main/java/org/apache/qpid/framing/CompositeAMQDataBlock.java index 5ec62ede93..7b6699b783 100644 --- a/java/common/src/main/java/org/apache/qpid/framing/CompositeAMQDataBlock.java +++ b/java/common/src/main/java/org/apache/qpid/framing/CompositeAMQDataBlock.java @@ -24,7 +24,7 @@ import org.apache.mina.common.ByteBuffer; public class CompositeAMQDataBlock extends AMQDataBlock implements EncodableAMQDataBlock { - private ByteBuffer _encodedBlock; + private AMQDataBlock _firstFrame; private AMQDataBlock[] _blocks; @@ -39,10 +39,10 @@ public class CompositeAMQDataBlock extends AMQDataBlock implements EncodableAMQD * @param encodedBlock already-encoded data * @param blocks some blocks to be encoded. */ - public CompositeAMQDataBlock(ByteBuffer encodedBlock, AMQDataBlock[] blocks) + public CompositeAMQDataBlock(AMQDataBlock encodedBlock, AMQDataBlock[] blocks) { this(blocks); - _encodedBlock = encodedBlock; + _firstFrame = encodedBlock; } public AMQDataBlock[] getBlocks() @@ -50,9 +50,9 @@ public class CompositeAMQDataBlock extends AMQDataBlock implements EncodableAMQD return _blocks; } - public ByteBuffer getEncodedBlock() + public AMQDataBlock getFirstFrame() { - return _encodedBlock; + return _firstFrame; } public long getSize() @@ -62,19 +62,18 @@ public class CompositeAMQDataBlock extends AMQDataBlock implements EncodableAMQD { frameSize += _blocks[i].getSize(); } - if (_encodedBlock != null) + if (_firstFrame != null) { - _encodedBlock.rewind(); - frameSize += _encodedBlock.remaining(); + frameSize += _firstFrame.getSize(); } return frameSize; } public void writePayload(ByteBuffer buffer) { - if (_encodedBlock != null) + if (_firstFrame != null) { - buffer.put(_encodedBlock); + _firstFrame.writePayload(buffer); } for (int i = 0; i < _blocks.length; i++) { @@ -91,7 +90,7 @@ public class CompositeAMQDataBlock extends AMQDataBlock implements EncodableAMQD else { StringBuilder buf = new StringBuilder(this.getClass().getName()); - buf.append("{encodedBlock=").append(_encodedBlock); + buf.append("{encodedBlock=").append(_firstFrame); for (int i = 0 ; i < _blocks.length; i++) { buf.append(" ").append(i).append("=[").append(_blocks[i].toString()).append("]"); diff --git a/java/common/src/main/java/org/apache/qpid/framing/DeferredDataBlock.java b/java/common/src/main/java/org/apache/qpid/framing/DeferredDataBlock.java new file mode 100644 index 0000000000..f6795ff200 --- /dev/null +++ b/java/common/src/main/java/org/apache/qpid/framing/DeferredDataBlock.java @@ -0,0 +1,50 @@ +package org.apache.qpid.framing; + +import org.apache.mina.common.ByteBuffer; + +/* +* +* Licensed to the Apache Software Foundation (ASF) under one +* or more contributor license agreements. See the NOTICE file +* distributed with this work for additional information +* regarding copyright ownership. The ASF licenses this file +* to you under the Apache License, Version 2.0 (the +* "License"); you may not use this file except in compliance +* with the License. You may obtain a copy of the License at +* +* http://www.apache.org/licenses/LICENSE-2.0 +* +* Unless required by applicable law or agreed to in writing, +* software distributed under the License is distributed on an +* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +* KIND, either express or implied. See the License for the +* specific language governing permissions and limitations +* under the License. +* +*/ +public abstract class DeferredDataBlock extends AMQDataBlock +{ + private AMQDataBlock _underlyingDataBlock; + + + public long getSize() + { + if(_underlyingDataBlock == null) + { + _underlyingDataBlock = createAMQDataBlock(); + } + return _underlyingDataBlock.getSize(); + } + + public void writePayload(ByteBuffer buffer) + { + if(_underlyingDataBlock == null) + { + _underlyingDataBlock = createAMQDataBlock(); + } + _underlyingDataBlock.writePayload(buffer); + } + + abstract protected AMQDataBlock createAMQDataBlock(); + +} diff --git a/java/common/src/main/java/org/apache/qpid/framing/SmallCompositeAMQDataBlock.java b/java/common/src/main/java/org/apache/qpid/framing/SmallCompositeAMQDataBlock.java index 26c048e34a..f8cf3f3011 100644 --- a/java/common/src/main/java/org/apache/qpid/framing/SmallCompositeAMQDataBlock.java +++ b/java/common/src/main/java/org/apache/qpid/framing/SmallCompositeAMQDataBlock.java @@ -25,7 +25,7 @@ import org.apache.mina.common.ByteBuffer; public class SmallCompositeAMQDataBlock extends AMQDataBlock implements EncodableAMQDataBlock { - private ByteBuffer _encodedBlock; + private AMQDataBlock _firstFrame; private AMQDataBlock _block; @@ -40,10 +40,10 @@ public class SmallCompositeAMQDataBlock extends AMQDataBlock implements Encodabl * @param encodedBlock already-encoded data * @param block a block to be encoded. */ - public SmallCompositeAMQDataBlock(ByteBuffer encodedBlock, AMQDataBlock block) + public SmallCompositeAMQDataBlock(AMQDataBlock encodedBlock, AMQDataBlock block) { this(block); - _encodedBlock = encodedBlock; + _firstFrame = encodedBlock; } public AMQDataBlock getBlock() @@ -51,28 +51,28 @@ public class SmallCompositeAMQDataBlock extends AMQDataBlock implements Encodabl return _block; } - public ByteBuffer getEncodedBlock() + public AMQDataBlock getFirstFrame() { - return _encodedBlock; + return _firstFrame; } public long getSize() { long frameSize = _block.getSize(); - if (_encodedBlock != null) + if (_firstFrame != null) { - _encodedBlock.rewind(); - frameSize += _encodedBlock.remaining(); + + frameSize += _firstFrame.getSize(); } return frameSize; } public void writePayload(ByteBuffer buffer) { - if (_encodedBlock != null) + if (_firstFrame != null) { - buffer.put(_encodedBlock); + _firstFrame.writePayload(buffer); } _block.writePayload(buffer); @@ -87,7 +87,7 @@ public class SmallCompositeAMQDataBlock extends AMQDataBlock implements Encodabl else { StringBuilder buf = new StringBuilder(this.getClass().getName()); - buf.append("{encodedBlock=").append(_encodedBlock); + buf.append("{encodedBlock=").append(_firstFrame); buf.append(" _block=[").append(_block.toString()).append("]"); diff --git a/java/common/src/main/java/org/apache/qpid/framing/abstraction/MessagePublishInfo.java b/java/common/src/main/java/org/apache/qpid/framing/abstraction/MessagePublishInfo.java index 706499c1b0..49c28bb06b 100644 --- a/java/common/src/main/java/org/apache/qpid/framing/abstraction/MessagePublishInfo.java +++ b/java/common/src/main/java/org/apache/qpid/framing/abstraction/MessagePublishInfo.java @@ -27,6 +27,8 @@ public interface MessagePublishInfo public AMQShortString getExchange(); + public void setExchange(AMQShortString exchange); + public boolean isImmediate(); public boolean isMandatory(); diff --git a/java/common/src/main/java/org/apache/qpid/framing/amqp_0_9/MethodConverter_0_9.java b/java/common/src/main/java/org/apache/qpid/framing/amqp_0_9/MethodConverter_0_9.java index de0007c132..d8b6b25b92 100644 --- a/java/common/src/main/java/org/apache/qpid/framing/amqp_0_9/MethodConverter_0_9.java +++ b/java/common/src/main/java/org/apache/qpid/framing/amqp_0_9/MethodConverter_0_9.java @@ -67,7 +67,7 @@ public class MethodConverter_0_9 extends AbstractMethodConverter implements Prot final AMQShortString exchange = publishBody.getExchange(); final AMQShortString routingKey = publishBody.getRoutingKey(); - return new MethodConverter_0_9.MessagePublishInfoImpl(exchange == null ? null : exchange.intern(), + return new MethodConverter_0_9.MessagePublishInfoImpl(exchange, publishBody.getImmediate(), publishBody.getMandatory(), routingKey == null ? null : routingKey.intern()); @@ -87,7 +87,7 @@ public class MethodConverter_0_9 extends AbstractMethodConverter implements Prot private static class MessagePublishInfoImpl implements MessagePublishInfo { - private final AMQShortString _exchange; + private AMQShortString _exchange; private final boolean _immediate; private final boolean _mandatory; private final AMQShortString _routingKey; @@ -108,6 +108,11 @@ public class MethodConverter_0_9 extends AbstractMethodConverter implements Prot return _exchange; } + public void setExchange(AMQShortString exchange) + { + _exchange = exchange; + } + public boolean isImmediate() { return _immediate; diff --git a/java/common/src/main/java/org/apache/qpid/framing/amqp_8_0/MethodConverter_8_0.java b/java/common/src/main/java/org/apache/qpid/framing/amqp_8_0/MethodConverter_8_0.java index 7a13af8a43..b1be49a350 100644 --- a/java/common/src/main/java/org/apache/qpid/framing/amqp_8_0/MethodConverter_8_0.java +++ b/java/common/src/main/java/org/apache/qpid/framing/amqp_8_0/MethodConverter_8_0.java @@ -107,7 +107,7 @@ public class MethodConverter_8_0 extends AbstractMethodConverter implements Prot private static class MessagePublishInfoImpl implements MessagePublishInfo { - private final AMQShortString _exchange; + private AMQShortString _exchange; private final boolean _immediate; private final boolean _mandatory; private final AMQShortString _routingKey; @@ -128,6 +128,11 @@ public class MethodConverter_8_0 extends AbstractMethodConverter implements Prot return _exchange; } + public void setExchange(AMQShortString exchange) + { + _exchange = exchange; + } + public boolean isImmediate() { return _immediate; diff --git a/java/systests/src/main/java/org/apache/qpid/server/ack/TxAckTest.java b/java/systests/src/main/java/org/apache/qpid/server/ack/TxAckTest.java index 10189a8017..42d9cccb4f 100644 --- a/java/systests/src/main/java/org/apache/qpid/server/ack/TxAckTest.java +++ b/java/systests/src/main/java/org/apache/qpid/server/ack/TxAckTest.java @@ -117,6 +117,11 @@ public class TxAckTest extends TestCase return null; } + public void setExchange(AMQShortString exchange) + { + //To change body of implemented methods use File | Settings | File Templates. + } + public boolean isImmediate() { return false; diff --git a/java/systests/src/main/java/org/apache/qpid/server/queue/AckTest.java b/java/systests/src/main/java/org/apache/qpid/server/queue/AckTest.java index 790607e268..96be579d2a 100644 --- a/java/systests/src/main/java/org/apache/qpid/server/queue/AckTest.java +++ b/java/systests/src/main/java/org/apache/qpid/server/queue/AckTest.java @@ -107,6 +107,11 @@ public class AckTest extends TestCase return new AMQShortString("someExchange"); } + public void setExchange(AMQShortString exchange) + { + //To change body of implemented methods use File | Settings | File Templates. + } + public boolean isImmediate() { return false; diff --git a/java/systests/src/main/java/org/apache/qpid/server/queue/MessageTestHelper.java b/java/systests/src/main/java/org/apache/qpid/server/queue/MessageTestHelper.java index 812aec6a5d..521bedeccd 100644 --- a/java/systests/src/main/java/org/apache/qpid/server/queue/MessageTestHelper.java +++ b/java/systests/src/main/java/org/apache/qpid/server/queue/MessageTestHelper.java @@ -70,6 +70,11 @@ class MessageTestHelper extends TestCase return null; } + public void setExchange(AMQShortString exchange) + { + //To change body of implemented methods use File | Settings | File Templates. + } + public boolean isImmediate() { return immediate; diff --git a/java/systests/src/main/java/org/apache/qpid/server/store/TestReferenceCounting.java b/java/systests/src/main/java/org/apache/qpid/server/store/TestReferenceCounting.java index ab6d9742e4..28cc89353b 100644 --- a/java/systests/src/main/java/org/apache/qpid/server/store/TestReferenceCounting.java +++ b/java/systests/src/main/java/org/apache/qpid/server/store/TestReferenceCounting.java @@ -61,6 +61,11 @@ public class TestReferenceCounting extends TestCase return null; } + public void setExchange(AMQShortString exchange) + { + //To change body of implemented methods use File | Settings | File Templates. + } + public boolean isImmediate() { return false; -- cgit v1.2.1