QPID-3028: use a small array and companion HashMap to store the incomplete Methods, rather than pre-allocating a 64000 entry array

git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk/qpid@1071675 13f79535-47bb-0310-9956-ffa450edef68

1 files changed, 42 insertions, 13 deletions

diff --git a/java/common/src/main/java/org/apache/qpid/transport/network/Assembler.java b/java/common/src/main/java/org/apache/qpid/transport/network/Assembler.java
index b3c419959c..1a85ab88a5 100644
--- a/java/common/src/main/java/org/apache/qpid/transport/network/Assembler.java
+++ b/java/common/src/main/java/org/apache/qpid/transport/network/Assembler.java
@@ -20,15 +20,12 @@
  */
 package org.apache.qpid.transport.network;
 
+import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
-import java.nio.ByteBuffer;
-
-import org.apache.qpid.transport.codec.BBDecoder;
-
 import org.apache.qpid.transport.Header;
 import org.apache.qpid.transport.Method;
 import org.apache.qpid.transport.ProtocolError;
@@ -36,19 +33,22 @@ import org.apache.qpid.transport.ProtocolEvent;
 import org.apache.qpid.transport.ProtocolHeader;
 import org.apache.qpid.transport.Receiver;
 import org.apache.qpid.transport.Struct;
-
+import org.apache.qpid.transport.codec.BBDecoder;
 
 /**
  * Assembler
  *
  */
-
 public class Assembler implements Receiver<NetworkEvent>, NetworkDelegate
 {
+    // Use a small array to store incomplete Methods for low-value channels, instead of allocating a huge
+    // array or always boxing the channelId and looking it up in the map. This value must be of the form 2^X - 1.
+    private static final int ARRAY_SIZE = 0xFF;
+    private final Method[] _incompleteMethodArray = new Method[ARRAY_SIZE + 1];
+    private final Map<Integer, Method> _incompleteMethodMap = new HashMap<Integer, Method>();
 
     private final Receiver<ProtocolEvent> receiver;
     private final Map<Integer,List<Frame>> segments;
-    private final Method[] incomplete;
     private static final ThreadLocal<BBDecoder> _decoder = new ThreadLocal<BBDecoder>()
     {
         public BBDecoder initialValue()
@@ -61,7 +61,6 @@ public class Assembler implements Receiver<NetworkEvent>, NetworkDelegate
     {
         this.receiver = receiver;
         segments = new HashMap<Integer,List<Frame>>();
-        incomplete = new Method[64*1024];
     }
 
     private int segmentKey(Frame frame)
@@ -190,7 +189,7 @@ public class Assembler implements Receiver<NetworkEvent>, NetworkDelegate
             command.read(dec);
             if (command.hasPayload())
             {
-                incomplete[channel] = command;
+                setIncompleteCommand(channel, command);
             }
             else
             {
@@ -198,7 +197,7 @@ public class Assembler implements Receiver<NetworkEvent>, NetworkDelegate
             }
             break;
         case HEADER:
-            command = incomplete[channel];
+            command = getIncompleteCommand(channel);
             List<Struct> structs = new ArrayList<Struct>(2);
             while (dec.hasRemaining())
             {
@@ -207,14 +206,14 @@ public class Assembler implements Receiver<NetworkEvent>, NetworkDelegate
             command.setHeader(new Header(structs));
             if (frame.isLastSegment())
             {
-                incomplete[channel] = null;
+                setIncompleteCommand(channel, null);
                 emit(channel, command);
             }
             break;
         case BODY:
-            command = incomplete[channel];
+            command = getIncompleteCommand(channel);
             command.setBody(segment);
-            incomplete[channel] = null;
+            setIncompleteCommand(channel, null);
             emit(channel, command);
             break;
         default:
@@ -224,4 +223,34 @@ public class Assembler implements Receiver<NetworkEvent>, NetworkDelegate
         dec.releaseBuffer();
     }
 
+    private void setIncompleteCommand(int channelId, Method incomplete)
+    {
+        if ((channelId & ARRAY_SIZE) == channelId)
+        {
+            _incompleteMethodArray[channelId] = incomplete;
+        }
+        else
+        {
+            if(incomplete != null)
+            {
+                _incompleteMethodMap.put(channelId, incomplete);
+            }
+            else
+            {
+                _incompleteMethodMap.remove(channelId);
+            }
+        }
+    }
+
+    private Method getIncompleteCommand(int channelId)
+    {
+        if ((channelId & ARRAY_SIZE) == channelId)
+        {
+            return _incompleteMethodArray[channelId];
+        }
+        else
+        {
+            return _incompleteMethodMap.get(channelId);
+        }
+    }
 }