QPID-541 A large portion of memory was being wasted in 32k ByteBuffers being held by the AMQShortStrings.

Patch submitted by Robert Godfrey to intern() the AMQSSs to reduce memory usage. Current implementation *will* impact performance due to the usage of a static Map for storage. However, a thread local implementation is in the works.


git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/branches/M2@556890 13f79535-47bb-0310-9956-ffa450edef68

5 files changed, 38 insertions, 7 deletions

diff --git a/java/broker/src/main/java/org/apache/qpid/server/handler/BasicConsumeMethodHandler.java b/java/broker/src/main/java/org/apache/qpid/server/handler/BasicConsumeMethodHandler.java
index 56eae279dc..d1e4ae6aa6 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/handler/BasicConsumeMethodHandler.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/handler/BasicConsumeMethodHandler.java
@@ -98,6 +98,12 @@ public class BasicConsumeMethodHandler implements StateAwareMethodListener<Basic
             }
             else
             {
+
+                if (body.consumerTag != null)
+                {
+                    body.consumerTag = body.consumerTag.intern();
+                }
+
                 try
                 {
                     AMQShortString consumerTag = channel.subscribeToQueue(body.consumerTag, queue, session, !body.noAck,
@@ -136,15 +142,15 @@ public class BasicConsumeMethodHandler implements StateAwareMethodListener<Basic
                     // If the above doesn't work then perhaps this is wrong too.
 //                    throw body.getConnectionException(AMQConstant.NOT_ALLOWED,
 //                                                      "Non-unique consumer tag, '" + body.consumerTag + "'");
-                                        // AMQP version change: Hardwire the version to 0-8 (major=8, minor=0)
+                    // AMQP version change: Hardwire the version to 0-8 (major=8, minor=0)
                     // TODO: Connect this to the session version obtained from ProtocolInitiation for this session.
                     // Be aware of possible changes to parameter order as versions change.
                     session.writeFrame(ConnectionCloseBody.createAMQFrame(channelId,
-                        (byte)8, (byte)0,	// AMQP version (major, minor)
-                        BasicConsumeBody.getClazz((byte)8, (byte)0),	// classId
-                        BasicConsumeBody.getMethod((byte)8, (byte)0),	// methodId
-                        AMQConstant.NOT_ALLOWED.getCode(),	// replyCode
-                        msg));	// replyText
+                                                                          (byte) 8, (byte) 0,    // AMQP version (major, minor)
+                                                                          BasicConsumeBody.getClazz((byte) 8, (byte) 0),    // classId
+                                                                          BasicConsumeBody.getMethod((byte) 8, (byte) 0),    // methodId
+                                                                          AMQConstant.NOT_ALLOWED.getCode(),    // replyCode
+                                                                          msg));    // replyText
                 }
                 catch (AMQQueue.ExistingExclusiveSubscription e)
                 {
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 67ade0a744..68d4483df3 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
@@ -67,6 +67,10 @@ public class BasicPublishMethodHandler  implements StateAwareMethodListener<Basi
             body.exchange = ExchangeDefaults.DEFAULT_EXCHANGE_NAME;
 
         }
+        else
+        {
+            body.exchange = body.exchange.intern();
+        }
         VirtualHost vHost = session.getVirtualHost();
         Exchange e = vHost.getExchangeRegistry().getExchange(body.exchange);
         // if the exchange does not exist we raise a channel exception
@@ -86,6 +90,11 @@ public class BasicPublishMethodHandler  implements StateAwareMethodListener<Basi
                 throw body.getChannelNotFoundException(evt.getChannelId());
             }
 
+            if(body.routingKey != null)
+            {
+                body.routingKey = body.routingKey.intern();
+            }
+            
             MessagePublishInfo info = session.getRegistry().getProtocolVersionMethodConverter().convertToInfo(body);
             channel.setPublishFrame(info, session);
         }
@@ -93,3 +102,4 @@ public class BasicPublishMethodHandler  implements StateAwareMethodListener<Basi
 }
 
 
+
diff --git a/java/broker/src/main/java/org/apache/qpid/server/handler/ExchangeDeclareHandler.java b/java/broker/src/main/java/org/apache/qpid/server/handler/ExchangeDeclareHandler.java
index be3ffcc698..8ab8366d9e 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/handler/ExchangeDeclareHandler.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/handler/ExchangeDeclareHandler.java
@@ -83,7 +83,9 @@ public class ExchangeDeclareHandler implements StateAwareMethodListener<Exchange
                     try
                     {
 
-                    exchange = exchangeFactory.createExchange(body.exchange, body.type, body.durable,
+                    exchange = exchangeFactory.createExchange(body.exchange == null ? null : body.exchange.intern(),
+                                                              body.type == null ? null : body.type.intern(), 
+                                                              body.durable,
                                                               body.passive, body.ticket);
                     exchangeRegistry.registerExchange(exchange);
                     }
diff --git a/java/broker/src/main/java/org/apache/qpid/server/handler/QueueBindHandler.java b/java/broker/src/main/java/org/apache/qpid/server/handler/QueueBindHandler.java
index 2b751ab692..3e68069838 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/handler/QueueBindHandler.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/handler/QueueBindHandler.java
@@ -97,6 +97,12 @@ public class QueueBindHandler implements StateAwareMethodListener<QueueBindBody>
         {
             throw body.getChannelException(AMQConstant.NOT_FOUND, "Exchange " + body.exchange + " does not exist.");
         }
+
+        if (body.routingKey != null)
+        {
+            body.routingKey = body.routingKey.intern();
+        }
+
         try
         {
             if (!exch.isBound(body.routingKey, body.arguments, queue))
diff --git a/java/broker/src/main/java/org/apache/qpid/server/handler/QueueDeclareHandler.java b/java/broker/src/main/java/org/apache/qpid/server/handler/QueueDeclareHandler.java
index 2e697d4564..2bc185cfb4 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/handler/QueueDeclareHandler.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/handler/QueueDeclareHandler.java
@@ -91,8 +91,15 @@ public class QueueDeclareHandler implements StateAwareMethodListener<QueueDeclar
         synchronized (queueRegistry)
         {
 
+
+
             if (((queue = queueRegistry.getQueue(body.queue)) == null))
             {
+                if(body.queue != null)
+                {
+                    body.queue = body.queue.intern();
+                }
+
                 if (body.passive)
                 {
                     String msg = "Queue: " + body.queue + " not found on VirtualHost(" + virtualHost + ").";