QPID-8430: Reduce memory consumption of AMQP 0-10 messages when rerouted by preventing deep copy of non-persistent messages. AMQP 1.0 memory consumption is unchanged.

Author: Kim van der Riet

src/qpid/broker/Message.cpp

@@ -172,7 +172,7 @@ void Message::addAnnotation(const std::string& key, const qpid::types::Variant&
 
 void Message::annotationsChanged()
 {
-    if (persistentContext) {
+    if (persistentContext && persistentContext->isMergeRequired()) {
         uint64_t id = persistentContext->getPersistenceId();
         persistentContext = persistentContext->merge(getAnnotations());
         persistentContext->setIngressCompletion(sharedState);

src/qpid/broker/PersistableMessage.h

@@ -86,6 +86,7 @@ class PersistableMessage : public Persistable
     virtual void decodeHeader(framing::Buffer& buffer) = 0;
     virtual void decodeContent(framing::Buffer& buffer) = 0;
     virtual uint32_t encodedHeaderSize() const = 0;
+    virtual bool isMergeRequired() const = 0;
     virtual boost::intrusive_ptr<PersistableMessage> merge(const std::map<std::string, qpid::types::Variant>& annotations) const = 0;
 };
 

src/qpid/broker/amqp/Message.cpp

@@ -514,6 +514,10 @@ void Message::decodeHeader(framing::Buffer& buffer)
 }
 void Message::decodeContent(framing::Buffer& /*buffer*/) {}
 
+bool Message::isMergeRequired() const {
+	return true;
+}
+
 boost::intrusive_ptr<PersistableMessage> Message::merge(const std::map<std::string, qpid::types::Variant>& added) const
 {
     //message- or delivery- annotations? would have to determine that from the name, for now assume always message-annotations

src/qpid/broker/amqp/Message.h

@@ -86,6 +86,7 @@ class Message : public qpid::broker::Message::SharedStateImpl, private qpid::amq
     void decodeHeader(framing::Buffer& buffer);
     void decodeContent(framing::Buffer& buffer);
     uint32_t encodedHeaderSize() const;
+    bool isMergeRequired() const;
     boost::intrusive_ptr<PersistableMessage> merge(const std::map<std::string, qpid::types::Variant>& annotations) const;
 
     static const Message& get(const qpid::broker::Message&);

src/qpid/broker/amqp_0_10/MessageTransfer.cpp

@@ -465,6 +465,10 @@ std::string MessageTransfer::getUserId() const
 }
 MessageTransfer::MessageTransfer(const qpid::framing::FrameSet& f) : frames(f), requiredCredit(0) {}
 
+bool MessageTransfer::isMergeRequired() const {
+	return isPersistent();
+}
+
 boost::intrusive_ptr<PersistableMessage> MessageTransfer::merge(const std::map<std::string, qpid::types::Variant>& annotations) const
 {
     boost::intrusive_ptr<MessageTransfer> clone(new MessageTransfer(this->frames));

src/qpid/broker/amqp_0_10/MessageTransfer.h

@@ -114,6 +114,7 @@ class MessageTransfer : public qpid::broker::Message::SharedStateImpl, public qp
      * but other meta data e.g.routing key and exchange)
      */
     uint32_t encodedHeaderSize() const;
+    bool isMergeRequired() const;
     boost::intrusive_ptr<PersistableMessage> merge(const std::map<std::string, qpid::types::Variant>& annotations) const;
 
     QPID_BROKER_EXTERN bool isQMFv2() const;