Better cache eviction in place.

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

activemq-core/src/main/java/org/apache/activemq/openwire/OpenWireFormat.java

@@ -42,6 +42,8 @@ final public class OpenWireFormat implements WireFormat {
     
     static final byte NULL_TYPE = CommandTypes.NULL;
     private static final int MARSHAL_CACHE_SIZE = Short.MAX_VALUE/2;
+    private static final int MARSHAL_CACHE_PREFERED_SIZE = MARSHAL_CACHE_SIZE-100;
+    
     private DataStreamMarshaller dataMarshallers[];
     private int version;
     private boolean stackTraceEnabled=true;
@@ -52,7 +54,8 @@ final public class OpenWireFormat implements WireFormat {
 
     private HashMap marshallCacheMap = new HashMap();
     private short nextMarshallCacheIndex=0;    
-    private short lasMarshallCacheEvictionIndex=100;    
+    private short nextMarshallCacheEvictionIndex=0;
+    
     private DataStructure marshallCache[] = new DataStructure[MARSHAL_CACHE_SIZE];
     private DataStructure unmarshallCache[] = new DataStructure[MARSHAL_CACHE_SIZE];
     
@@ -92,6 +95,10 @@ final public class OpenWireFormat implements WireFormat {
     
     public Packet marshal(Object command) throws IOException {
         
+        if( cacheEnabled ) {
+            runMarshallCacheEvictionSweep();
+        }
+        
         MarshallAware ma=null;
         // If not using value caching, then the marshaled form is always the same
         if( !cacheEnabled && ((DataStructure)command).isMarshallAware() ) {
@@ -187,6 +194,11 @@ final public class OpenWireFormat implements WireFormat {
     }
     
     public void marshal(Object o, DataOutputStream ds) throws IOException {
+        
+        if( cacheEnabled ) {
+            runMarshallCacheEvictionSweep();
+        }
+        
         int size=1;
         if( o != null) {
             DataStructure c = (DataStructure) o;
@@ -359,26 +371,33 @@ final public class OpenWireFormat implements WireFormat {
         }
     }
 
+    public void runMarshallCacheEvictionSweep() {
+        // Do we need to start evicting??
+        while( marshallCacheMap.size() > MARSHAL_CACHE_PREFERED_SIZE ) {
             
-    public Short getMarshallCacheIndex(Object o) {
+            marshallCacheMap.remove(marshallCache[nextMarshallCacheEvictionIndex]);
+            marshallCache[nextMarshallCacheEvictionIndex]=null;
+
+            nextMarshallCacheEvictionIndex++;
+            if( nextMarshallCacheEvictionIndex >= MARSHAL_CACHE_SIZE ) {
+                nextMarshallCacheEvictionIndex=0;
+            }
+            
+        }
+    }
+    
+    public Short getMarshallCacheIndex(DataStructure o) {
         return (Short) marshallCacheMap.get(o);
     }
     
-    public Short addToMarshallCache(Object o) {
+    public Short addToMarshallCache(DataStructure o) {
-        nextMarshallCacheIndex++;
+        short i = nextMarshallCacheIndex++;
         if( nextMarshallCacheIndex >= MARSHAL_CACHE_SIZE ) {
             nextMarshallCacheIndex=0;
         }
-        lasMarshallCacheEvictionIndex++;
+        
-        if( lasMarshallCacheEvictionIndex >= MARSHAL_CACHE_SIZE ) {
+        marshallCache[i] = o;
-            lasMarshallCacheEvictionIndex=0;
+        Short index = new Short(i);
-        }
-        if( marshallCache[lasMarshallCacheEvictionIndex]!=null ) {
-            marshallCacheMap.remove(marshallCache[lasMarshallCacheEvictionIndex]);
-            marshallCache[lasMarshallCacheEvictionIndex]=null;
-        }
-        marshallCache[nextMarshallCacheIndex] = (DataStructure) o;
-        Short index = new Short(nextMarshallCacheIndex);
         marshallCacheMap.put(o, index);
         return index;
     }

activemq-core/src/main/java/org/apache/activemq/openwire/OpenWireFormatFactory.java

@@ -28,6 +28,8 @@ public class OpenWireFormatFactory implements WireFormatFactory {
     private boolean stackTraceEnabled=true;
     private boolean tcpNoDelayEnabled=false;
     private boolean cacheEnabled=true;
+    private boolean tightEncodingEnabled=true;
+    private boolean prefixPacketSize=true;
 
     public WireFormat createWireFormat() {
         OpenWireFormat format = new OpenWireFormat();
@@ -35,6 +37,8 @@ public class OpenWireFormatFactory implements WireFormatFactory {
         format.setStackTraceEnabled(stackTraceEnabled);
         format.setCacheEnabled(cacheEnabled);
         format.setTcpNoDelayEnabled(tcpNoDelayEnabled);
+        format.setTightEncodingEnabled(tightEncodingEnabled);
+        format.setPrefixPacketSize(prefixPacketSize);
         return format;
     }
 
@@ -69,4 +73,20 @@ public class OpenWireFormatFactory implements WireFormatFactory {
     public void setCacheEnabled(boolean cacheEnabled) {
         this.cacheEnabled = cacheEnabled;
     }
+
+    public boolean isTightEncodingEnabled() {
+        return tightEncodingEnabled;
+    }
+
+    public void setTightEncodingEnabled(boolean tightEncodingEnabled) {
+        this.tightEncodingEnabled = tightEncodingEnabled;
+    }
+
+    public boolean isPrefixPacketSize() {
+        return prefixPacketSize;
+    }
+
+    public void setPrefixPacketSize(boolean prefixPacketSize) {
+        this.prefixPacketSize = prefixPacketSize;
+    }
 }