Author: manik.surtani(a)jboss.com Date: 2008-02-02 09:41:13 -0500 (Sat, 02 Feb 2008) New Revision: 5287 Modified: core/trunk/src/main/docbook/userguide/en/modules/configuration_reference.xml core/trunk/src/main/java/org/jboss/cache/config/Configuration.java core/trunk/src/main/java/org/jboss/cache/marshall/AbstractMarshaller.java core/trunk/src/main/java/org/jboss/cache/marshall/CacheMarshaller200.java core/trunk/src/test/java/org/jboss/cache/marshall/CacheMarshaller200Test.java core/trunk/src/test/java/org/jboss/cache/marshall/CacheMarshallerTestBase.java core/trunk/src/test/java/org/jboss/cache/profiling/AbstractProfileTest.java core/trunk/src/test/java/org/jboss/cache/profiling/ProfileTest.java Log: Improved marshaller performance by removing reference counting (now optional) Modified: core/trunk/src/main/docbook/userguide/en/modules/configuration_reference.xml =================================================================== --- core/trunk/src/main/docbook/userguide/en/modules/configuration_reference.xml 2008-02-01 12:13:53 UTC (rev 5286) +++ core/trunk/src/main/docbook/userguide/en/modules/configuration_reference.xml 2008-02-02 14:41:13 UTC (rev 5287) @@ -706,6 +706,37 @@ </entry> </row> + <row> + <entry> + <para>UseReferenceCounting</para> + </entry> + + <entry> + <para> + An optional parameter that is used to control how identical objects are marshalled. If reference + counting is disabled, they are marshalled again and added to the stream. If reference counting + is enabled, + they are written once and a reference is written in subsequent instances. + </para> + <para> + By default this is set to<literal>false</literal>. Set this to + <literal>true</literal> + if you use + transactions and have a lot of operations per transaction. Naturally, there is no way to + ascertain + a definite threshold after which reference counting should be used. It is usually a tradeoff + between + network speed and larger data streams, and CPU overhead in generating references. It is best to + profile + with this enabled and disabled to determine which is best, but one thing for certain is that if + you do + not use transactions, or have transactions with just one or two operations, reference counting + ought to + be disabled. + </para> + </entry> + </row> + </tbody> </tgroup> </informaltable> Modified: core/trunk/src/main/java/org/jboss/cache/config/Configuration.java =================================================================== --- core/trunk/src/main/java/org/jboss/cache/config/Configuration.java 2008-02-01 12:13:53 UTC (rev 5286) +++ core/trunk/src/main/java/org/jboss/cache/config/Configuration.java 2008-02-02 14:41:13 UTC (rev 5287) @@ -25,10 +25,10 @@ implements Cloneable { private static final long serialVersionUID = 5553791890144997466L; - private int numberOfNotifierThreads = 25; private Marshaller marshaller; + public void setCacheMarshaller(Marshaller instance) { marshaller = instance; @@ -187,6 +187,7 @@ private String marshallerClass; private ShutdownHookBehavior shutdownHookBehavior = ShutdownHookBehavior.DEFAULT; private boolean useLazyDeserialization = true; + private boolean useReferenceCounting = false; // ------------------------------------------------------------------------------------------------------------ // SETTERS - MAKE SURE ALL SETTERS PERFORM testImmutability()!!! @@ -491,6 +492,17 @@ this.useLazyDeserialization = useLazyDeserialization; } + /** + * Sets whether reference counting should be used when marshalling remote calls. + * + * @since 2.1.0 + */ + public void setUseReferenceCounting(boolean b) + { + testImmutability("useReferenceCounting"); + useReferenceCounting = b; + } + // ------------------------------------------------------------------------------------------------------------ // GETTERS // ------------------------------------------------------------------------------------------------------------ @@ -679,16 +691,16 @@ this.marshallerClass = marshallerClass; } - public int getNumberOfNotifierThreads() + /** + * Gets whether reference counting should be used when marshalling remote calls. + * + * @since 2.1.0 + */ + public boolean isUseReferenceCounting() { - return numberOfNotifierThreads; + return useReferenceCounting; } - public void setNumberOfNotifierThreads(int numberOfNotifierThreads) - { - this.numberOfNotifierThreads = numberOfNotifierThreads; - } - /** * Returns a {@link java.net.URL} to a default JGroups configuration file. * @@ -762,7 +774,7 @@ if (lockParentForChildInsertRemove != that.lockParentForChildInsertRemove) return false; - if (numberOfNotifierThreads != that.numberOfNotifierThreads) return false; + if (useReferenceCounting != that.useReferenceCounting) return false; return true; } @@ -797,7 +809,7 @@ result = 29 * result + (runtimeConfig != null ? runtimeConfig.hashCode() : 0); result = 29 * result + (marshallerClass != null ? marshallerClass.hashCode() : 0); result = 29 * result + (lockParentForChildInsertRemove ? 1 : 0); - result = 29 * result + numberOfNotifierThreads; + result = 29 * result + (useReferenceCounting ? 1 : 0); return result; } Modified: core/trunk/src/main/java/org/jboss/cache/marshall/AbstractMarshaller.java =================================================================== --- core/trunk/src/main/java/org/jboss/cache/marshall/AbstractMarshaller.java 2008-02-01 12:13:53 UTC (rev 5286) +++ core/trunk/src/main/java/org/jboss/cache/marshall/AbstractMarshaller.java 2008-02-02 14:41:13 UTC (rev 5287) @@ -45,6 +45,7 @@ private Map<GlobalTransaction, Fqn> transactions = new ConcurrentHashMap<GlobalTransaction, Fqn>(16); protected Configuration configuration; protected ClassLoader defaultClassLoader; + protected boolean useRefs = false; @Inject void injectDependencies(RegionManager regionManager, Configuration configuration, @ComponentName("deployerClassLoader")ClassLoader defaultClassLoader) @@ -59,6 +60,10 @@ { this.useRegionBasedMarshalling = configuration.isUseRegionBasedMarshalling(); this.defaultInactive = configuration.isInactiveOnStartup(); + + // ensure that if we are using CacheMarshaller200 - i.e., using replication version of 2.0.0 - reference counting is + // enabled, since this is always enabled in JBC 2.0.0. + this.useRefs = configuration.isUseReferenceCounting() || getClass().equals(CacheMarshaller200.class); } protected void initLogger() Modified: core/trunk/src/main/java/org/jboss/cache/marshall/CacheMarshaller200.java =================================================================== --- core/trunk/src/main/java/org/jboss/cache/marshall/CacheMarshaller200.java 2008-02-01 12:13:53 UTC (rev 5286) +++ core/trunk/src/main/java/org/jboss/cache/marshall/CacheMarshaller200.java 2008-02-02 14:41:13 UTC (rev 5287) @@ -6,6 +6,7 @@ */ package org.jboss.cache.marshall; +import org.jboss.cache.CacheException; import org.jboss.cache.Fqn; import org.jboss.cache.Region; import static org.jboss.cache.Region.Status; @@ -130,7 +131,7 @@ } else { - Map<Integer, Object> refMap = new HashMap<Integer, Object>(); + UnmarshalledReferences refMap = new UnmarshalledReferences(); Object retValue = unmarshallObject(in, defaultClassLoader, refMap); if (trace) log.trace("Unmarshalled object " + retValue); return retValue; @@ -140,7 +141,7 @@ public void objectToObjectStream(Object o, ObjectOutputStream out, Fqn region) throws Exception { if (trace) log.trace("Marshalling object " + o); - Map<Object, Integer> refMap = new HashMap<Object, Integer>(); + Map<Object, Integer> refMap = useRefs ? new HashMap<Object, Integer>() : null; ClassLoader toUse = defaultClassLoader; Thread current = Thread.currentThread(); ClassLoader old = current.getContextClassLoader(); @@ -171,7 +172,7 @@ protected Object objectFromObjectStreamRegionBased(ObjectInputStream in) throws Exception { - Map<Integer, Object> refMap = new HashMap<Integer, Object>(); + UnmarshalledReferences refMap = useRefs ? new UnmarshalledReferences() : null; Object o = unmarshallObject(in, refMap); Fqn regionFqn = null; if (o == null) @@ -283,7 +284,7 @@ { out.writeByte(MAGICNUMBER_NULL); } - else if (refMap.containsKey(o))// see if this object has been marshalled before. + else if (useRefs && refMap.containsKey(o))// see if this object has been marshalled before. { out.writeByte(MAGICNUMBER_REF); writeReference(out, refMap.get(o)); @@ -311,13 +312,13 @@ else if (o instanceof Fqn) { out.writeByte(MAGICNUMBER_FQN); - writeReference(out, createReference(o, refMap)); + if (useRefs) writeReference(out, createReference(o, refMap)); marshallFqn((Fqn) o, out, refMap); } else if (o instanceof GlobalTransaction) { out.writeByte(MAGICNUMBER_GTX); - writeReference(out, createReference(o, refMap)); + if (useRefs) writeReference(out, createReference(o, refMap)); marshallGlobalTransaction((GlobalTransaction) o, out, refMap); } else if (o instanceof IpAddress) @@ -388,7 +389,7 @@ else if (o instanceof String) { out.writeByte(MAGICNUMBER_STRING); - writeReference(out, createReference(o, refMap)); + if (useRefs) writeReference(out, createReference(o, refMap)); marshallString((String) o, out); } else if (o instanceof NodeDataMarker) @@ -418,7 +419,7 @@ log.trace("Warning: using object serialization for " + o.getClass()); } out.writeByte(MAGICNUMBER_SERIALIZABLE); - writeReference(out, createReference(o, refMap)); + if (useRefs) writeReference(out, createReference(o, refMap)); out.writeObject(o); } else @@ -514,7 +515,7 @@ // --------- Unmarshalling methods - private Object unmarshallObject(ObjectInputStream in, ClassLoader loader, Map<Integer, Object> refMap) throws Exception + private Object unmarshallObject(ObjectInputStream in, ClassLoader loader, UnmarshalledReferences refMap) throws Exception { if (loader == null) { @@ -536,26 +537,31 @@ } } - private Object unmarshallObject(ObjectInputStream in, Map<Integer, Object> refMap) throws Exception + private Object unmarshallObject(ObjectInputStream in, UnmarshalledReferences refMap) throws Exception { byte magicNumber = in.readByte(); - Integer reference; + Integer reference = 0; Object retVal; switch (magicNumber) { case MAGICNUMBER_NULL: return null; case MAGICNUMBER_REF: - reference = readReference(in); - if (!refMap.containsKey(reference)) + if (useRefs) { - throw new IOException("Unable to locate object reference " + reference + " in byte stream!"); + reference = readReference(in); + return refMap.getReferencedObject(reference); +// if (!refMap.containsKey(reference)) +// { +// throw new IOException("Unable to locate object reference " + reference + " in byte stream!"); +// } +// return refMap.get(reference); } - return refMap.get(reference); + else break; case MAGICNUMBER_SERIALIZABLE: - reference = readReference(in); + if (useRefs) reference = readReference(in); retVal = in.readObject(); - refMap.put(reference, retVal); + if (useRefs) refMap.putReferencedObject(reference, retVal); return retVal; case MAGICNUMBER_MARSHALLEDVALUE: MarshalledValue mv = new MarshalledValue(); @@ -565,14 +571,14 @@ retVal = unmarshallMethodCall(in, refMap); return retVal; case MAGICNUMBER_FQN: - reference = readReference(in); + if (useRefs) reference = readReference(in); retVal = unmarshallFqn(in, refMap); - refMap.put(reference, retVal); + if (useRefs) refMap.putReferencedObject(reference, retVal); return retVal; case MAGICNUMBER_GTX: - reference = readReference(in); + if (useRefs) reference = readReference(in); retVal = unmarshallGlobalTransaction(in, refMap); - refMap.put(reference, retVal); + if (useRefs) refMap.putReferencedObject(reference, retVal); return retVal; case MAGICNUMBER_IPADDRESS: retVal = unmarshallIpAddress(in); @@ -603,9 +609,9 @@ case MAGICNUMBER_SHORT: return in.readShort(); case MAGICNUMBER_STRING: - reference = readReference(in); + if (useRefs) reference = readReference(in); retVal = unmarshallString(in); - refMap.put(reference, retVal); + if (useRefs) refMap.putReferencedObject(reference, retVal); return retVal; case MAGICNUMBER_NODEDATA_MARKER: retVal = new NodeDataMarker(); @@ -628,9 +634,10 @@ } throw new Exception("Unknown magic number " + magicNumber); } + throw new Exception("Unknown magic number " + magicNumber); } - private GravitateResult unmarshallGravitateResult(ObjectInputStream in, Map<Integer, Object> refMap) throws Exception + private GravitateResult unmarshallGravitateResult(ObjectInputStream in, UnmarshalledReferences refMap) throws Exception { Boolean found = (Boolean) unmarshallObject(in, refMap); if (found) @@ -651,7 +658,7 @@ return (String) in.readObject(); } - private MethodCall unmarshallMethodCall(ObjectInputStream in, Map<Integer, Object> refMap) throws Exception + private MethodCall unmarshallMethodCall(ObjectInputStream in, UnmarshalledReferences refMap) throws Exception { short methodId = in.readShort(); byte numArgs = in.readByte(); @@ -669,7 +676,7 @@ return MethodCallFactory.create(methodId, args); } - private GlobalTransaction unmarshallGlobalTransaction(ObjectInputStream in, Map<Integer, Object> refMap) throws Exception + private GlobalTransaction unmarshallGlobalTransaction(ObjectInputStream in, UnmarshalledReferences refMap) throws Exception { GlobalTransaction gtx = new GlobalTransaction(); long id = in.readLong(); @@ -679,7 +686,7 @@ return gtx; } - Fqn unmarshallFqn(ObjectInputStream in, Map<Integer, Object> refMap) throws Exception + Fqn unmarshallFqn(ObjectInputStream in, UnmarshalledReferences refMap) throws Exception { boolean isRoot = in.readBoolean(); @@ -692,7 +699,7 @@ { elements.add(unmarshallObject(in, refMap)); } - fqn = new Fqn(elements); + fqn = new Fqn(elements, true); } else { @@ -708,7 +715,7 @@ return ipAddress; } - private List unmarshallArrayList(ObjectInputStream in, Map refMap) throws Exception + private List unmarshallArrayList(ObjectInputStream in, UnmarshalledReferences refMap) throws Exception { int listSize = readUnsignedInt(in); List list = new ArrayList(listSize); @@ -716,49 +723,49 @@ return list; } - private List unmarshallLinkedList(ObjectInputStream in, Map refMap) throws Exception + private List unmarshallLinkedList(ObjectInputStream in, UnmarshalledReferences refMap) throws Exception { List list = new LinkedList(); populateFromStream(in, refMap, list, readUnsignedInt(in)); return list; } - private Map unmarshallHashMap(ObjectInputStream in, Map refMap) throws Exception + private Map unmarshallHashMap(ObjectInputStream in, UnmarshalledReferences refMap) throws Exception { Map map = new HashMap(); populateFromStream(in, refMap, map); return map; } - private Map unmarshallMapCopy(ObjectInputStream in, Map<Integer, Object> refMap) throws Exception + private Map unmarshallMapCopy(ObjectInputStream in, UnmarshalledReferences refMap) throws Exception { // read in as a HashMap first Map m = unmarshallHashMap(in, refMap); return new MapCopy(m); } - private Map unmarshallTreeMap(ObjectInputStream in, Map refMap) throws Exception + private Map unmarshallTreeMap(ObjectInputStream in, UnmarshalledReferences refMap) throws Exception { Map map = new TreeMap(); populateFromStream(in, refMap, map); return map; } - private Set unmarshallHashSet(ObjectInputStream in, Map refMap) throws Exception + private Set unmarshallHashSet(ObjectInputStream in, UnmarshalledReferences refMap) throws Exception { Set set = new HashSet(); populateFromStream(in, refMap, set); return set; } - private Set unmarshallTreeSet(ObjectInputStream in, Map refMap) throws Exception + private Set unmarshallTreeSet(ObjectInputStream in, UnmarshalledReferences refMap) throws Exception { Set set = new TreeSet(); populateFromStream(in, refMap, set); return set; } - private void populateFromStream(ObjectInputStream in, Map refMap, Map mapToPopulate) throws Exception + private void populateFromStream(ObjectInputStream in, UnmarshalledReferences refMap, Map mapToPopulate) throws Exception { int size = readUnsignedInt(in); for (int i = 0; i < size; i++) @@ -767,7 +774,7 @@ } } - private void populateFromStream(ObjectInputStream in, Map refMap, Set setToPopulate) throws Exception + private void populateFromStream(ObjectInputStream in, UnmarshalledReferences refMap, Set setToPopulate) throws Exception { int size = readUnsignedInt(in); for (int i = 0; i < size; i++) @@ -776,7 +783,7 @@ } } - private void populateFromStream(ObjectInputStream in, Map refMap, List listToPopulate, int listSize) throws Exception + private void populateFromStream(ObjectInputStream in, UnmarshalledReferences refMap, List listToPopulate, int listSize) throws Exception { for (int i = 0; i < listSize; i++) { @@ -838,3 +845,48 @@ out.writeLong(i); } } + +/** + * An efficient array-based list of referenced objects, using the reference id as a subscript for the array. + */ +class UnmarshalledReferences +{ + private ArrayList<Object> referencedObjects = new ArrayList<Object>(); + + /** + * Retrieves an object referenced by an id + * + * @param ref reference + * @return object + */ + public Object getReferencedObject(int ref) + { + if (ref >= referencedObjects.size()) + throw new CacheException("Attempting to look up a ref that hasn't been inserted yet"); + return referencedObjects.get(ref); + } + + /** + * Adds a referenced object to the list of references + * + * @param ref reference id + * @param o object + */ + public void putReferencedObject(int ref, Object o) + { + int sz = referencedObjects.size(); + // if we are not adding the object to the end of the list, make sure we use a specific position + if (ref < sz) + { + referencedObjects.add(ref, o); + return; + } + else if (ref > sz) + { + // if we are adding the reference to a position beyond the end of the list, make sure we expand the list first. + // this can happen, weirdly enough, since marshallObject() can be called recursively, such as from marshallFqn(). + for (int i = sz; i < ref; i++) referencedObjects.add(null); + } + referencedObjects.add(o); + } +} Modified: core/trunk/src/test/java/org/jboss/cache/marshall/CacheMarshaller200Test.java =================================================================== --- core/trunk/src/test/java/org/jboss/cache/marshall/CacheMarshaller200Test.java 2008-02-01 12:13:53 UTC (rev 5286) +++ core/trunk/src/test/java/org/jboss/cache/marshall/CacheMarshaller200Test.java 2008-02-02 14:41:13 UTC (rev 5287) @@ -16,7 +16,6 @@ import java.io.ByteArrayOutputStream; import java.io.ObjectInputStream; import java.io.ObjectOutputStream; -import java.util.HashMap; import java.util.List; import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.ExecutorService; @@ -55,7 +54,7 @@ assert magic == CacheMarshaller200.MAGICNUMBER_FQN; // now the chunks of an Fqn - Fqn f = cm200.unmarshallFqn(ois, new HashMap<Integer, Object>()); + Fqn f = cm200.unmarshallFqn(ois, new UnmarshalledReferences()); assert f.equals(Fqn.fromString("/hello")); } @@ -126,7 +125,7 @@ assert in.readByte() == CacheMarshaller200.MAGICNUMBER_FQN : "The stream should start with an Fqn"; // discard the nest refId short in.readShort(); - Fqn f = cm200.unmarshallFqn(in, new HashMap<Integer, Object>()); + Fqn f = cm200.unmarshallFqn(in, new UnmarshalledReferences()); assert region.equals(f) : "Should use the same region for the response as was used for the request!"; } Modified: core/trunk/src/test/java/org/jboss/cache/marshall/CacheMarshallerTestBase.java =================================================================== --- core/trunk/src/test/java/org/jboss/cache/marshall/CacheMarshallerTestBase.java 2008-02-01 12:13:53 UTC (rev 5286) +++ core/trunk/src/test/java/org/jboss/cache/marshall/CacheMarshallerTestBase.java 2008-02-02 14:41:13 UTC (rev 5287) @@ -35,6 +35,7 @@ c.setUseRegionBasedMarshalling(false); c.setInactiveOnStartup(false); c.setReplVersionString(currentVersion); + //c.setUseReferenceCounting(true); marshaller = createVAM(c, regionManager); } Modified: core/trunk/src/test/java/org/jboss/cache/profiling/AbstractProfileTest.java =================================================================== --- core/trunk/src/test/java/org/jboss/cache/profiling/AbstractProfileTest.java 2008-02-01 12:13:53 UTC (rev 5286) +++ core/trunk/src/test/java/org/jboss/cache/profiling/AbstractProfileTest.java 2008-02-02 14:41:13 UTC (rev 5287) @@ -24,6 +24,7 @@ { Configuration cfg = UnitTestCacheConfigurationFactory.createConfiguration(Configuration.CacheMode.REPL_SYNC); cache = new DefaultCacheFactory().createCache(cfg, false); + cfg.setUseReferenceCounting(false); } @AfterTest Modified: core/trunk/src/test/java/org/jboss/cache/profiling/ProfileTest.java =================================================================== --- core/trunk/src/test/java/org/jboss/cache/profiling/ProfileTest.java 2008-02-01 12:13:53 UTC (rev 5286) +++ core/trunk/src/test/java/org/jboss/cache/profiling/ProfileTest.java 2008-02-02 14:41:13 UTC (rev 5287) @@ -36,7 +36,7 @@ /* Test configuration options */ - protected static final long DURATION = 2 * 60 * 1000; + protected static final long DURATION = 10 * 60 * 1000; protected static final int NUM_THREADS = 10; protected static final int MAX_RANDOM_SLEEP_MILLIS = 100; protected static final int MAX_DEPTH = 8; @@ -112,6 +112,9 @@ startup(); warmup(); doTest(); + + // wait for user exit + System.in.read(); } /**