Author: manik.surtani(a)jboss.com Date: 2008-11-17 18:26:21 -0500 (Mon, 17 Nov 2008) New Revision: 7150 Added: core/branches/flat/src/main/java/org/jboss/starobrno/atomic/ core/branches/flat/src/main/java/org/jboss/starobrno/atomic/AtomicHashMap.java core/branches/flat/src/main/java/org/jboss/starobrno/atomic/AtomicHashMapDelta.java core/branches/flat/src/main/java/org/jboss/starobrno/atomic/AtomicHashMapProxy.java core/branches/flat/src/main/java/org/jboss/starobrno/atomic/AtomicMap.java core/branches/flat/src/main/java/org/jboss/starobrno/atomic/AtomicMapCache.java core/branches/flat/src/main/java/org/jboss/starobrno/atomic/ClearOperation.java core/branches/flat/src/main/java/org/jboss/starobrno/atomic/Delta.java core/branches/flat/src/main/java/org/jboss/starobrno/atomic/DeltaAware.java core/branches/flat/src/main/java/org/jboss/starobrno/atomic/NullDelta.java core/branches/flat/src/main/java/org/jboss/starobrno/atomic/Operation.java core/branches/flat/src/main/java/org/jboss/starobrno/atomic/PutOperation.java core/branches/flat/src/main/java/org/jboss/starobrno/atomic/RemoveOperation.java core/branches/flat/src/main/java/org/jboss/starobrno/batch/AutoBatchSupport.java core/branches/flat/src/test/java/org/jboss/starobrno/atomic/ core/branches/flat/src/test/java/org/jboss/starobrno/atomic/APITest.java Removed: core/branches/flat/src/main/java/org/jboss/starobrno/atomic/Delta.java core/branches/flat/src/main/java/org/jboss/starobrno/atomic/DeltaHashMap.java core/branches/flat/src/main/java/org/jboss/starobrno/delta/ Modified: core/branches/flat/src/main/java/org/jboss/cache/DefaultCacheFactory.java core/branches/flat/src/main/java/org/jboss/starobrno/Cache.java core/branches/flat/src/main/java/org/jboss/starobrno/CacheDelegate.java core/branches/flat/src/main/java/org/jboss/starobrno/batch/BatchContainer.java core/branches/flat/src/main/java/org/jboss/starobrno/commands/write/PutKeyValueCommand.java core/branches/flat/src/main/java/org/jboss/starobrno/config/parsing/RootElementBuilder.java core/branches/flat/src/main/java/org/jboss/starobrno/config/parsing/XmlConfigurationParser.java core/branches/flat/src/main/java/org/jboss/starobrno/marshall/CacheMarshallerStarobrno.java core/branches/flat/src/main/java/org/jboss/starobrno/tree/NodeImpl.java core/branches/flat/src/main/java/org/jboss/starobrno/tree/TreeCacheImpl.java core/branches/flat/src/main/java/org/jboss/starobrno/tree/TreeStructureSupport.java core/branches/flat/src/test/java/org/jboss/starobrno/tree/api/NodeAPITest.java core/branches/flat/src/test/java/org/jboss/starobrno/tree/api/TreeCacheAPITest.java Log: AtomicMap and Tree facade based on AtomicMap Modified: core/branches/flat/src/main/java/org/jboss/cache/DefaultCacheFactory.java =================================================================== --- core/branches/flat/src/main/java/org/jboss/cache/DefaultCacheFactory.java 2008-11-17 23:15:11 UTC (rev 7149) +++ core/branches/flat/src/main/java/org/jboss/cache/DefaultCacheFactory.java 2008-11-17 23:26:21 UTC (rev 7150) @@ -23,9 +23,9 @@ import org.jboss.cache.annotations.Compat; import org.jboss.cache.jmx.PlatformMBeanServerRegistration; +import org.jboss.starobrno.Cache; import org.jboss.starobrno.CacheDelegate; import org.jboss.starobrno.CacheSPI; -import org.jboss.starobrno.Cache; import org.jboss.starobrno.config.Configuration; import org.jboss.starobrno.config.ConfigurationException; import org.jboss.starobrno.config.parsing.XmlConfigurationParser; @@ -137,7 +137,7 @@ /** * Bootstraps this factory with a Configuration and a ComponentRegistry. */ - private void bootstrap( CacheSPI spi, Configuration configuration) + private void bootstrap(CacheSPI spi, Configuration configuration) { // injection bootstrap stuff componentRegistry = new ComponentRegistry(configuration, spi); @@ -166,7 +166,7 @@ return createCache(c); } - public Cache<K, V> createCache(InputStream is, boolean start) throws ConfigurationException + public Cache<K, V> createCache(InputStream is, boolean start) throws ConfigurationException { XmlConfigurationParser parser = new XmlConfigurationParser(); Configuration c = parser.parseStream(is); Modified: core/branches/flat/src/main/java/org/jboss/starobrno/Cache.java =================================================================== --- core/branches/flat/src/main/java/org/jboss/starobrno/Cache.java 2008-11-17 23:15:11 UTC (rev 7149) +++ core/branches/flat/src/main/java/org/jboss/starobrno/Cache.java 2008-11-17 23:26:21 UTC (rev 7150) @@ -53,7 +53,10 @@ CacheStatus getCacheStatus(); - public void startBatch(); + /** + * @return true if a batch was successfully started; false if one was available and already running. + */ + public boolean startBatch(); public void endBatch(boolean successful); Modified: core/branches/flat/src/main/java/org/jboss/starobrno/CacheDelegate.java =================================================================== --- core/branches/flat/src/main/java/org/jboss/starobrno/CacheDelegate.java 2008-11-17 23:15:11 UTC (rev 7149) +++ core/branches/flat/src/main/java/org/jboss/starobrno/CacheDelegate.java 2008-11-17 23:26:21 UTC (rev 7150) @@ -27,6 +27,9 @@ import org.jboss.cache.buddyreplication.GravitateResult; import org.jboss.cache.loader.CacheLoaderManager; import org.jboss.cache.marshall.Marshaller; +import org.jboss.starobrno.atomic.AtomicHashMap; +import org.jboss.starobrno.atomic.AtomicMap; +import org.jboss.starobrno.atomic.AtomicMapCache; import org.jboss.starobrno.batch.BatchContainer; import org.jboss.starobrno.commands.CommandsFactory; import org.jboss.starobrno.commands.read.GetKeyValueCommand; @@ -64,7 +67,7 @@ * @author Mircea.Markus(a)jboss.com */ @NonVolatile -public class CacheDelegate<K, V> implements CacheSPI<K, V> +public class CacheDelegate<K, V> implements CacheSPI<K, V>, AtomicMapCache<K, V> { protected InvocationContextContainer invocationContextContainer; protected CommandsFactory commandsFactory; @@ -357,11 +360,11 @@ return componentRegistry.getState(); } - public void startBatch() + public boolean startBatch() { if (!config.isInvocationBatchingEnabled()) throw new ConfigurationException("Invocation batching not enabled in current configuration! Please use the <invocationBatching /> element."); - batchContainer.startBatch(); + return batchContainer.startBatch(); } public void endBatch(boolean successful) @@ -406,4 +409,16 @@ { return dataContainer == null ? super.toString() : dataContainer.toString(); } + + public AtomicMap getAtomicMap(K key) throws ClassCastException + { + Object value = get(key); + if (value == null) value = AtomicHashMap.newInstance(this, key); + return ((AtomicHashMap) value).getProxy(this, key); + } + + public <AMK, AMV> AtomicMap<AMK, AMV> getAtomicMap(K key, Class<AMK> atomicMapKeyType, Class<AMV> atomicMapValueType) throws ClassCastException + { + return getAtomicMap(key); + } } Copied: core/branches/flat/src/main/java/org/jboss/starobrno/atomic (from rev 7145, core/branches/flat/src/main/java/org/jboss/starobrno/delta) Copied: core/branches/flat/src/main/java/org/jboss/starobrno/atomic/AtomicHashMap.java (from rev 7145, core/branches/flat/src/main/java/org/jboss/starobrno/delta/DeltaHashMap.java) =================================================================== --- core/branches/flat/src/main/java/org/jboss/starobrno/atomic/AtomicHashMap.java (rev 0) +++ core/branches/flat/src/main/java/org/jboss/starobrno/atomic/AtomicHashMap.java 2008-11-17 23:26:21 UTC (rev 7150) @@ -0,0 +1,186 @@ +/* + * JBoss, Home of Professional Open Source + * Copyright 2008, Red Hat Middleware LLC, and individual contributors + * by the @authors tag. See the copyright.txt in the distribution for a + * full listing of individual contributors. + * + * This is free software; you can redistribute it and/or modify it + * under the terms of the GNU Lesser General Public License as + * published by the Free Software Foundation; either version 2.1 of + * the License, or (at your option) any later version. + * + * This software is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this software; if not, write to the Free + * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA + * 02110-1301 USA, or see the FSF site: http://www.fsf.org. + */ +package org.jboss.starobrno.atomic; + +import net.jcip.annotations.NotThreadSafe; +import org.jboss.starobrno.Cache; +import org.jboss.starobrno.util.FastCopyHashMap; + +import java.util.Collection; +import java.util.Map; +import java.util.Set; + +/** + * Note that for replication to work properly, this class <b><i>requires</i></b> that all writes take place within the + * scope of an ongoing transaction or batch. + * <p/> + * + * @param <K> + * @param <V> + */ +@NotThreadSafe +public class AtomicHashMap<K, V> implements AtomicMap<K, V>, DeltaAware +{ + FastCopyHashMap<K, V> delegate; + AtomicHashMapDelta delta = null; + AtomicHashMapProxy proxy; + + /** + * Construction only allowed through this factory method. This factory is intended for use internally by the + * CacheDelegate. User code should use {@link AtomicMapCache#getAtomicMap(Object)}. + */ + public static AtomicHashMap newInstance(Cache cache, Object cacheKey) + { + AtomicHashMap value = new AtomicHashMap(); + Object oldValue = cache.putIfAbsent(cacheKey, value); + if (oldValue != null) value = (AtomicHashMap) oldValue; + return value; + } + + AtomicHashMap() + { + delegate = new FastCopyHashMap<K, V>(); + } + + public void rollback() + { + // replay reversal operations on the changelog, in reverse order + +// ListIterator<AtomicHashMapDelta.Operation> li = delta.changelog.listIterator(delta.changelog.size()); +// while (li.hasPrevious()) li.previous().rollback(delegate); + + // surely a no-op? + } + + public void commit() + { + if (delta != null) delta.changelog.clear(); + } + + public int size() + { + return delegate.size(); + } + + public boolean isEmpty() + { + return delegate.isEmpty(); + } + + public boolean containsKey(Object key) + { + return delegate.containsKey(key); + } + + public boolean containsValue(Object value) + { + return delegate.containsValue(value); + } + + public V get(Object key) + { + return delegate.get(key); + } + + public V put(K key, V value) + { + PutOperation<K, V> op = new PutOperation<K, V>(); + op.key = key; + op.newValue = value; + op.oldValue = delegate.put(key, value); + delta.changelog.add(op); + return op.oldValue; + } + + public V remove(Object key) + { + RemoveOperation<K, V> op = new RemoveOperation<K, V>(); + op.key = (K) key; + op.oldValue = delegate.remove(key); + delta.changelog.add(op); + return op.oldValue; + } + + public void putAll(Map<? extends K, ? extends V> t) + { + // this is crappy - need to do this more efficiently! + for (Entry<? extends K, ? extends V> e : t.entrySet()) put(e.getKey(), e.getValue()); + } + + public void clear() + { + ClearOperation<K, V> op = new ClearOperation<K, V>(); + op.originalEntries = (FastCopyHashMap<K, V>) delegate.clone(); + delta.changelog.add(op); + delegate.clear(); + } + + public Set<K> keySet() + { + return delegate.keySet(); + } + + public Collection<V> values() + { + return delegate.values(); + } + + public Set<Entry<K, V>> entrySet() + { + return delegate.entrySet(); + } + + public AtomicMap getProxy(Cache cache, Object mapKey) + { + // construct the proxy lazily + if (proxy == null) // DCL is OK here since proxy is volatile (and we live in a post-JDK 5 world) + { + synchronized (this) + { + if (proxy == null) proxy = new AtomicHashMapProxy(cache, mapKey); + } + } + return proxy; + } + + public Delta delta() + { + return delta == null ? NullDelta.INSTANCE : delta; + } + + public AtomicHashMap copyForWrite() + { + AtomicHashMap clone = new AtomicHashMap(); + clone.delegate = (FastCopyHashMap) delegate.clone(); + clone.proxy = proxy; + clone.delta = new AtomicHashMapDelta(); + return clone; + } + + @Override + public String toString() + { + return "AtomicHashMap{" + + "delegate=" + delegate + + '}'; + } +} Added: core/branches/flat/src/main/java/org/jboss/starobrno/atomic/AtomicHashMapDelta.java =================================================================== --- core/branches/flat/src/main/java/org/jboss/starobrno/atomic/AtomicHashMapDelta.java (rev 0) +++ core/branches/flat/src/main/java/org/jboss/starobrno/atomic/AtomicHashMapDelta.java 2008-11-17 23:26:21 UTC (rev 7150) @@ -0,0 +1,76 @@ +/* + * JBoss, Home of Professional Open Source + * Copyright 2008, Red Hat Middleware LLC, and individual contributors + * by the @authors tag. See the copyright.txt in the distribution for a + * full listing of individual contributors. + * + * This is free software; you can redistribute it and/or modify it + * under the terms of the GNU Lesser General Public License as + * published by the Free Software Foundation; either version 2.1 of + * the License, or (at your option) any later version. + * + * This software is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this software; if not, write to the Free + * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA + * 02110-1301 USA, or see the FSF site: http://www.fsf.org. + */ +package org.jboss.starobrno.atomic; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +import java.io.IOException; +import java.io.ObjectInput; +import java.io.ObjectOutput; +import java.util.LinkedList; +import java.util.List; + +/** + * Changes that have occured on an AtomicHashMap + * + * @author Manik Surtani (<a href="mailto:manik AT jboss DOT org">manik AT jboss DOT org</a>) + */ +public class AtomicHashMapDelta implements Delta +{ + List<Operation> changelog = new LinkedList<Operation>(); + private static final Log log = LogFactory.getLog(AtomicHashMapDelta.class); + private static final boolean trace = log.isTraceEnabled(); + + public DeltaAware merge(DeltaAware d) + { + AtomicHashMap other; + if (d != null && (d instanceof AtomicHashMap)) + other = (AtomicHashMap) d; + else + other = new AtomicHashMap(); + + for (Operation o : changelog) o.replay(other.delegate); + other.commit(); + return other; + } + + public void writeExternal(ObjectOutput out) throws IOException + { + if (trace) log.trace("Serializing changelog " + changelog); + out.writeObject(changelog); + } + + public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException + { + changelog = (List<Operation>) in.readObject(); + if (trace) log.trace("Deserialized changelog " + changelog); + } + + @Override + public String toString() + { + return "AtomicHashMapDelta{" + + "changelog=" + changelog + + '}'; + } +} \ No newline at end of file Added: core/branches/flat/src/main/java/org/jboss/starobrno/atomic/AtomicHashMapProxy.java =================================================================== --- core/branches/flat/src/main/java/org/jboss/starobrno/atomic/AtomicHashMapProxy.java (rev 0) +++ core/branches/flat/src/main/java/org/jboss/starobrno/atomic/AtomicHashMapProxy.java 2008-11-17 23:26:21 UTC (rev 7150) @@ -0,0 +1,170 @@ +/* + * JBoss, Home of Professional Open Source + * Copyright 2008, Red Hat Middleware LLC, and individual contributors + * by the @authors tag. See the copyright.txt in the distribution for a + * full listing of individual contributors. + * + * This is free software; you can redistribute it and/or modify it + * under the terms of the GNU Lesser General Public License as + * published by the Free Software Foundation; either version 2.1 of + * the License, or (at your option) any later version. + * + * This software is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this software; if not, write to the Free + * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA + * 02110-1301 USA, or see the FSF site: http://www.fsf.org. + */ +package org.jboss.starobrno.atomic; + +import org.jboss.starobrno.Cache; +import org.jboss.starobrno.batch.AutoBatchSupport; + +import java.util.Collection; +import java.util.Map; +import java.util.Set; + +/** + * A layer of indirection around an {@link org.jboss.starobrno.atomic.AtomicHashMap} to provide reader consistency + * + * @author Manik Surtani (<a href="mailto:manik AT jboss DOT org">manik AT jboss DOT org</a>) + */ +public class AtomicHashMapProxy<K, V> extends AutoBatchSupport implements AtomicMap<K, V> +{ + Object deltaMapKey; + + public AtomicHashMapProxy(Cache cache, Object deltaMapKey) + { + this.cache = cache; + this.deltaMapKey = deltaMapKey; + } + + // internal helper, reduces lots of casts. + private AtomicHashMap<K, V> getDeltaMapForRead() + { + return (AtomicHashMap<K, V>) cache.get(deltaMapKey); + } + + private AtomicHashMap<K, V> getDeltaMapForWrite() + { + if (ownsLock()) + { + return (AtomicHashMap<K, V>) cache.get(deltaMapKey); + } + else + { + // acquire WL + cache.getInvocationContext().getOptionOverrides().setForceWriteLock(true); + AtomicHashMap map = getDeltaMapForRead(); + // copy for write + AtomicHashMap copy = map == null ? new AtomicHashMap() : map.copyForWrite(); + cache.put(deltaMapKey, copy); + return copy; + } + } + + private boolean ownsLock() + { + return cache.getInvocationContext().hasLockedKey(deltaMapKey); + } + + // readers + + public Set<K> keySet() + { + return getDeltaMapForRead().keySet(); + } + + public Collection<V> values() + { + return getDeltaMapForRead().values(); + } + + public Set<Entry<K, V>> entrySet() + { + return getDeltaMapForRead().entrySet(); + } + + public int size() + { + return getDeltaMapForRead().size(); + } + + public boolean isEmpty() + { + return getDeltaMapForRead().isEmpty(); + } + + public boolean containsKey(Object key) + { + return getDeltaMapForRead().containsKey(key); + } + + public boolean containsValue(Object value) + { + return getDeltaMapForRead().containsValue(value); + } + + public V get(Object key) + { + return getDeltaMapForRead().get(key); + } + + // writers + + public V put(K key, V value) + { + try + { + startAtomic(); + return getDeltaMapForWrite().put(key, value); + } + finally + { + endAtomic(); + } + } + + public V remove(Object key) + { + try + { + startAtomic(); + return getDeltaMapForWrite().remove(key); + } + finally + { + endAtomic(); + } + } + + public void putAll(Map<? extends K, ? extends V> m) + { + try + { + startAtomic(); + getDeltaMapForWrite().putAll(m); + } + finally + { + endAtomic(); + } + } + + public void clear() + { + try + { + startAtomic(); + getDeltaMapForWrite().clear(); + } + finally + { + endAtomic(); + } + } +} Added: core/branches/flat/src/main/java/org/jboss/starobrno/atomic/AtomicMap.java =================================================================== --- core/branches/flat/src/main/java/org/jboss/starobrno/atomic/AtomicMap.java (rev 0) +++ core/branches/flat/src/main/java/org/jboss/starobrno/atomic/AtomicMap.java 2008-11-17 23:26:21 UTC (rev 7150) @@ -0,0 +1,41 @@ +/* + * JBoss, Home of Professional Open Source + * Copyright 2008, Red Hat Middleware LLC, and individual contributors + * by the @authors tag. See the copyright.txt in the distribution for a + * full listing of individual contributors. + * + * This is free software; you can redistribute it and/or modify it + * under the terms of the GNU Lesser General Public License as + * published by the Free Software Foundation; either version 2.1 of + * the License, or (at your option) any later version. + * + * This software is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this software; if not, write to the Free + * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA + * 02110-1301 USA, or see the FSF site: http://www.fsf.org. + */ +package org.jboss.starobrno.atomic; + +import java.util.Map; + +/** + * This is a special type of Map geared for use in JBoss Cache. This map type supports JBoss Cache atomicizing writes + * on the cache such that a coarse grained locking is used if this map is stored in the cache, such that the entire map + * is locked for writes or is isolated for safe concurrent read. + * <p/> + * This is, for all practical purposes, a marker interface that indicates that Maps of this type will be locked + * atomically in the cache and replicated in a fine grained manner. + * <p/> + * + * @author Manik Surtani (<a href="mailto:manik AT jboss DOT org">manik AT jboss DOT org</a>) + * @see DeltaAware + * @see AtomicHashMap + */ +public interface AtomicMap<K, V> extends Map<K, V> +{ +} Added: core/branches/flat/src/main/java/org/jboss/starobrno/atomic/AtomicMapCache.java =================================================================== --- core/branches/flat/src/main/java/org/jboss/starobrno/atomic/AtomicMapCache.java (rev 0) +++ core/branches/flat/src/main/java/org/jboss/starobrno/atomic/AtomicMapCache.java 2008-11-17 23:26:21 UTC (rev 7150) @@ -0,0 +1,56 @@ +/* + * JBoss, Home of Professional Open Source + * Copyright 2008, Red Hat Middleware LLC, and individual contributors + * by the @authors tag. See the copyright.txt in the distribution for a + * full listing of individual contributors. + * + * This is free software; you can redistribute it and/or modify it + * under the terms of the GNU Lesser General Public License as + * published by the Free Software Foundation; either version 2.1 of + * the License, or (at your option) any later version. + * + * This software is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this software; if not, write to the Free + * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA + * 02110-1301 USA, or see the FSF site: http://www.fsf.org. + */ +package org.jboss.starobrno.atomic; + +import org.jboss.starobrno.Cache; + +/** + * This interface adds the getAtomicMap() method which allows users to get a hold of a map type where operations on its + * elements are all atomic. Refer to the {@link AtomicMap} javadocs for more details. + * + * @author Manik Surtani (<a href="mailto:manik AT jboss DOT org">manik AT jboss DOT org</a>) + * @see AtomicMap + */ +public interface AtomicMapCache<K, V> extends Cache<K, V> +{ + /** + * Returns an atomic map. The classes passed in are used to parameterize the Map returned. + * + * @param key key under which to obtain and store this map in the cache + * @param mapKeyType type of the key used for this map + * @param mapValueType type of the value used for this map. + * @param <X> map keys + * @param <Y> map values + * @return a new or existing atomic map. Never null. + * @throws ClassCastException if there already is a value stored under the given key and the type of value cannot be used as an AtomicMap. + */ + <AMK, AMV> AtomicMap<AMK, AMV> getAtomicMap(K key, Class<AMK> atomicMapKeyType, Class<AMV> atomicMapValueType) throws ClassCastException; + + /** + * Un-parameterized version of {@link #getAtomicMap(Object, Class, Class)} which returns an un-parameterized map. + * + * @param key key under which to obtain and store this map in the cache + * @return a new or existing atomic map. Never null. + * @throws ClassCastException if there already is a value stored under the given key and the type of value cannot be used as an AtomicMap. + */ + AtomicMap getAtomicMap(K key) throws ClassCastException; +} Added: core/branches/flat/src/main/java/org/jboss/starobrno/atomic/ClearOperation.java =================================================================== --- core/branches/flat/src/main/java/org/jboss/starobrno/atomic/ClearOperation.java (rev 0) +++ core/branches/flat/src/main/java/org/jboss/starobrno/atomic/ClearOperation.java 2008-11-17 23:26:21 UTC (rev 7150) @@ -0,0 +1,42 @@ +/* + * JBoss, Home of Professional Open Source + * Copyright 2008, Red Hat Middleware LLC, and individual contributors + * by the @authors tag. See the copyright.txt in the distribution for a + * full listing of individual contributors. + * + * This is free software; you can redistribute it and/or modify it + * under the terms of the GNU Lesser General Public License as + * published by the Free Software Foundation; either version 2.1 of + * the License, or (at your option) any later version. + * + * This software is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this software; if not, write to the Free + * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA + * 02110-1301 USA, or see the FSF site: http://www.fsf.org. + */ +package org.jboss.starobrno.atomic; + +import org.jboss.starobrno.util.FastCopyHashMap; + +import java.util.Map; + + +public class ClearOperation<K, V> extends Operation<K, V> +{ + FastCopyHashMap<K, V> originalEntries; + + public void rollback(Map<K, V> delegate) + { + if (!originalEntries.isEmpty()) delegate.putAll(originalEntries); + } + + public void replay(Map<K, V> delegate) + { + delegate.clear(); + } +} \ No newline at end of file Deleted: core/branches/flat/src/main/java/org/jboss/starobrno/atomic/Delta.java =================================================================== --- core/branches/flat/src/main/java/org/jboss/starobrno/delta/Delta.java 2008-11-17 10:43:05 UTC (rev 7145) +++ core/branches/flat/src/main/java/org/jboss/starobrno/atomic/Delta.java 2008-11-17 23:26:21 UTC (rev 7150) @@ -1,97 +0,0 @@ -/* - * JBoss, Home of Professional Open Source - * Copyright 2008, Red Hat Middleware LLC, and individual contributors - * by the @authors tag. See the copyright.txt in the distribution for a - * full listing of individual contributors. - * - * This is free software; you can redistribute it and/or modify it - * under the terms of the GNU Lesser General Public License as - * published by the Free Software Foundation; either version 2.1 of - * the License, or (at your option) any later version. - * - * This software is distributed in the hope that it will be useful, - * but WITHOUT ANY WARRANTY; without even the implied warranty of - * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU - * Lesser General Public License for more details. - * - * You should have received a copy of the GNU Lesser General Public - * License along with this software; if not, write to the Free - * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA - * 02110-1301 USA, or see the FSF site: http://www.fsf.org. - */ -package org.jboss.starobrno.delta; - -import java.io.Externalizable; - -/** - * This interface allows for implementations to be palced as values in the cache, and report what has changed. This - * allows for finer grained replication messages. - * <p/> - * Basically, the first time a <tt>Delta</tt> object is placed in the cache, it is cached in its entirity. When a - * <tt>Delta</tt> is removed, similarly, it is removed in its entirity. - * <p/> - * Special behavior is seen when a <tt>Delta</tt> object is changed and replicated though. For example, consider the - * following code: - * <code> - * Person p = cache.get("personName"); - * p.setAge(p.getAge() + 1); - * cache.put("personName", p); - * </code> - * <p/> - * where <tt>Person</tt> implements Delta. - * <p/> - * Normally, the put() call would involve serializing and replicating the entire <tt>Person</tt> instance, but since - * <tt>Person</tt> implements <tt>Delta</tt>, instead the marshaller will only expect the changes to be marshalled (when - * calling the {@link Externalizable} interfaces. As such, implementations could safely only marshall modified fields when - * {@link Externalizable#writeExternal(java.io.ObjectOutput)} is called, thereby reducing the time taken serializing - * and replicating changes. - * <p/> - * On the receiving end, changes are applied as such. In a {@link org.jboss.starobrno.commands.write.PutKeyValueCommand}, - * if the value being put is a <tt>Delta</tt> and there is already a <tt>Delta</tt> under the same key, a merge operation - * is called by invoking {@link Delta#merge(Delta)} on the <i>replicated</i> <tt>Delta</tt> value and passing in the - * <i>original</i> <tt>Delta</tt> instance. It would then be up to the implementation to efficiently and correctly merge - * in state, and return a coherent and complete <tt>Delta</tt> instance that could be placed in the cache. - * <p/> - * If the command does not find anything under the key - or emor importantly, finds an object that does not implement - * <tt>Delta</tt> - it would pass a null into the {@link #merge(Delta)} method. - * <p/> - * It is important to note that {@link #merge(Delta)} is <b>only</b> called on remote nodes, after replication when - * merging state. This is never used on the local cache instance. - * <p/> - * In addition to {@link #merge(Delta)}, other important methods on this interface are {@link #rollback()}. If running - * within a JTA transaction scope, and a rollback occurs, no replication takes place so nothing needs to happen on - * remote instances. However, on your local cache instance, you may have changed internal state of a reference - * which already exists in the cache, such as in the above example. To correctly deal with JTA transaction rollbacks, - * all {@link #rollback()} methods on <tt>Delta</tt> instances involved in a transaction will be called and it is - * up to the implementation to reset internal state. There is a corresponding {@link #commit()} method where changes - * should be considered as accepted, are flushed and rollback information can be cleared. - * <p/> - * For an example of a <tt>Delta</tt> implementation, and for typical use cases, please refer to {@link org.jboss.starobrno.delta.DeltaHashMap}. - * - * @author Manik Surtani (<a href="mailto:manik AT jboss DOT org">manik AT jboss DOT org</a>) - * @see org.jboss.starobrno.delta.DeltaHashMap - */ -public interface Delta extends Externalizable -{ - /** - * Invoked if a transaction involving modifying state on this instance has rolled back, and state needs to be reverted. - */ - void rollback(); - - /** - * Invoked if a transaction involving modifying state on this instance completes successfully, and any recorded - * rollback information can safely be discarded. - */ - void commit(); - - /** - * Merges changes. Important to note that this method is always called on remote nodes, after replication, and is - * called on the de-serialized instance read off the wire. What is passed in is an extisting <tt>Delta</tt> instance - * to merge with, or a null. This method should return a fully coherent <tt>Delta</tt> instance than can be placed - * in the cache - * - * @param toMergeInto - * @return a fully coherent <tt>Delta</tt> instance - */ - Delta merge(Delta toMergeInto); -} Added: core/branches/flat/src/main/java/org/jboss/starobrno/atomic/Delta.java =================================================================== --- core/branches/flat/src/main/java/org/jboss/starobrno/atomic/Delta.java (rev 0) +++ core/branches/flat/src/main/java/org/jboss/starobrno/atomic/Delta.java 2008-11-17 23:26:21 UTC (rev 7150) @@ -0,0 +1,44 @@ +/* + * JBoss, Home of Professional Open Source + * Copyright 2008, Red Hat Middleware LLC, and individual contributors + * by the @authors tag. See the copyright.txt in the distribution for a + * full listing of individual contributors. + * + * This is free software; you can redistribute it and/or modify it + * under the terms of the GNU Lesser General Public License as + * published by the Free Software Foundation; either version 2.1 of + * the License, or (at your option) any later version. + * + * This software is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this software; if not, write to the Free + * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA + * 02110-1301 USA, or see the FSF site: http://www.fsf.org. + */ +package org.jboss.starobrno.atomic; + +import java.io.Externalizable; + +/** + * Represents changes made to a {@link DeltaAware} implementation. Should be efficiently externalizable. + * + * @author Manik Surtani (<a href="mailto:manik AT jboss DOT org">manik AT jboss DOT org</a>) + */ +public interface Delta extends Externalizable +{ + /** + * Merge the current set of deltas with a given {@link DeltaAware} instance, and return a coherent and complete + * {@link DeltaAware} instance. Implementations should be able to deal with null values passed in, or values of a + * different type from the expected DeltaAware instance. Usually the approach would be to ignore what is passed in, + * create a new instance of the DeltaAware implementation that the current Delta implementation is written for, + * apply changes and pass it back. + * + * @param d instance to merge with, or null if no merging is needed + * @return a fully coherent and usable instance of DeltaAware which may or may not be the same instance passed in + */ + DeltaAware merge(DeltaAware d); +} Added: core/branches/flat/src/main/java/org/jboss/starobrno/atomic/DeltaAware.java =================================================================== --- core/branches/flat/src/main/java/org/jboss/starobrno/atomic/DeltaAware.java (rev 0) +++ core/branches/flat/src/main/java/org/jboss/starobrno/atomic/DeltaAware.java 2008-11-17 23:26:21 UTC (rev 7150) @@ -0,0 +1,53 @@ +/* + * JBoss, Home of Professional Open Source + * Copyright 2008, Red Hat Middleware LLC, and individual contributors + * by the @authors tag. See the copyright.txt in the distribution for a + * full listing of individual contributors. + * + * This is free software; you can redistribute it and/or modify it + * under the terms of the GNU Lesser General Public License as + * published by the Free Software Foundation; either version 2.1 of + * the License, or (at your option) any later version. + * + * This software is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this software; if not, write to the Free + * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA + * 02110-1301 USA, or see the FSF site: http://www.fsf.org. + */ +package org.jboss.starobrno.atomic; + +/** + * This interface allows the extraction of deltas. + * <p/> + * Implementations would be closely coupled to a corresponding {@link Delta} implementation, since {@link org.jboss.starobrno.atomic.Delta#instantiate()} + * would need to know how to recreate this instance of DeltaAware if needed. + * <p/> + * + * @author Manik Surtani (<a href="mailto:manik AT jboss DOT org">manik AT jboss DOT org</a>) + * @see Delta + */ +public interface DeltaAware +{ + /** + * Extracts changes made to implementations, in an efficient format that can easily and cheaply be serialized and + * deserialized. + * + * @return an instance of Delta + */ + Delta delta(); + + /** + * Indicate that all deltas collected to date can be applied and discarded. + */ + void commit(); + + /** + * Indicate that all deltas collected to date can be discarded. + */ + void rollback(); +} Deleted: core/branches/flat/src/main/java/org/jboss/starobrno/atomic/DeltaHashMap.java =================================================================== --- core/branches/flat/src/main/java/org/jboss/starobrno/delta/DeltaHashMap.java 2008-11-17 10:43:05 UTC (rev 7145) +++ core/branches/flat/src/main/java/org/jboss/starobrno/atomic/DeltaHashMap.java 2008-11-17 23:26:21 UTC (rev 7150) @@ -1,301 +0,0 @@ -/* - * JBoss, Home of Professional Open Source - * Copyright 2008, Red Hat Middleware LLC, and individual contributors - * by the @authors tag. See the copyright.txt in the distribution for a - * full listing of individual contributors. - * - * This is free software; you can redistribute it and/or modify it - * under the terms of the GNU Lesser General Public License as - * published by the Free Software Foundation; either version 2.1 of - * the License, or (at your option) any later version. - * - * This software is distributed in the hope that it will be useful, - * but WITHOUT ANY WARRANTY; without even the implied warranty of - * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU - * Lesser General Public License for more details. - * - * You should have received a copy of the GNU Lesser General Public - * License along with this software; if not, write to the Free - * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA - * 02110-1301 USA, or see the FSF site: http://www.fsf.org. - */ -package org.jboss.starobrno.delta; - -import org.jboss.starobrno.util.FastCopyHashMap; - -import java.io.Externalizable; -import java.io.IOException; -import java.io.ObjectInput; -import java.io.ObjectOutput; -import java.util.Collection; -import java.util.LinkedList; -import java.util.List; -import java.util.ListIterator; -import java.util.Map; -import java.util.Set; - -public class DeltaHashMap<K, V> implements Delta, Map<K, V> -{ - FastCopyHashMap<K, V> delegate; - List<Operation> changelog = new LinkedList<Operation>(); - - /** - * Constructs an empty <tt>HashMap</tt> with the specified initial - * capacity and load factor. - * - * @param initialCapacity The initial capacity. - * @param loadFactor The load factor. - * @throws IllegalArgumentException if the initial capacity is negative - * or the load factor is nonpositive. - */ - public DeltaHashMap(int initialCapacity, float loadFactor) - { - delegate = new FastCopyHashMap(initialCapacity, loadFactor); - } - - /** - * Constructs an empty <tt>HashMap</tt> with the specified initial - * capacity and the default load factor (0.75). - * - * @param initialCapacity the initial capacity. - * @throws IllegalArgumentException if the initial capacity is negative. - */ - public DeltaHashMap(int initialCapacity) - { - delegate = new FastCopyHashMap<K, V>(initialCapacity); - } - - /** - * Constructs an empty <tt>HashMap</tt> with the default initial capacity - * (16) and the default load factor (0.75). - */ - public DeltaHashMap() - { - delegate = new FastCopyHashMap<K, V>(); - } - - /** - * Constructs a new <tt>HashMap</tt> with the same mappings as the - * specified <tt>Map</tt>. The <tt>HashMap</tt> is created with - * default load factor (0.75) and an initial capacity sufficient to - * hold the mappings in the specified <tt>Map</tt>. - * - * @param m the map whose mappings are to be placed in this map. - * @throws NullPointerException if the specified map is null. - */ - public DeltaHashMap(Map<? extends K, ? extends V> m) - { - delegate = new FastCopyHashMap<K, V>(m); - } - - - public void rollback() - { - // replay reversal operations on the changelog, in reverse order - ListIterator<Operation> li = changelog.listIterator(changelog.size()); - while (li.hasPrevious()) li.previous().rollback(delegate); - } - - public void commit() - { - changelog.clear(); - } - - public Delta merge(Delta toMergeInto) - { - if (toMergeInto != null) - { - if (toMergeInto instanceof DeltaHashMap) - { - DeltaHashMap other = (DeltaHashMap) toMergeInto; - for (Operation o : changelog) o.replay(other.delegate); - commit(); - other.commit(); - return toMergeInto; - } - else - { - throw new IllegalArgumentException("This instance of " + getClass().getSimpleName() + " can only merge with other instances of the same type. Don't know how to deal with " + toMergeInto.getClass()); - } - } - else - { - // use the current instance, since there is nothing to merge into. - for (Operation o : changelog) o.replay(delegate); - commit(); - return this; - } - } - - public void writeExternal(ObjectOutput out) throws IOException - { - out.writeObject(changelog); - } - - public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException - { - changelog = (List<Operation>) in.readObject(); - } - - public int size() - { - return delegate.size(); - } - - public boolean isEmpty() - { - return delegate.isEmpty(); - } - - public boolean containsKey(Object key) - { - return delegate.containsKey(key); - } - - public boolean containsValue(Object value) - { - return delegate.containsValue(value); - } - - public V get(Object key) - { - return delegate.get(key); - } - - public V put(K key, V value) - { - PutOperation<K, V> op = new PutOperation<K, V>(); - op.key = key; - op.newValue = value; - op.oldValue = delegate.put(key, value); - changelog.add(op); - return op.oldValue; - } - - public V remove(Object key) - { - RemoveOperation<K, V> op = new RemoveOperation<K, V>(); - op.key = (K) key; - op.oldValue = delegate.remove(key); - changelog.add(op); - return op.oldValue; - } - - public void putAll(Map<? extends K, ? extends V> t) - { - // this is crappy - need to do this more efficiently! - for (Entry<? extends K, ? extends V> e : t.entrySet()) put(e.getKey(), e.getValue()); - } - - public void clear() - { - ClearOperation<K, V> op = new ClearOperation<K, V>(); - op.originalEntries = (FastCopyHashMap<K, V>) delegate.clone(); - changelog.add(op); - delegate.clear(); - } - - public Set<K> keySet() - { - return delegate.keySet(); - } - - public Collection<V> values() - { - return delegate.values(); - } - - public Set<Entry<K, V>> entrySet() - { - return delegate.entrySet(); - } - - private static abstract class Operation<K, V> implements Externalizable - { - abstract void rollback(Map<K, V> delegate); - - abstract void replay(Map<K, V> delegate); - - public void writeExternal(ObjectOutput out) throws IOException - { - } - - public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException - { - } - } - - private static class PutOperation<K, V> extends Operation<K, V> - { - K key; - V oldValue; - V newValue; - - void rollback(Map<K, V> delegate) - { - if (oldValue == null) - delegate.remove(key); - else - delegate.put(key, oldValue); - } - - void replay(Map<K, V> delegate) - { - delegate.put(key, newValue); - } - - public void writeExternal(ObjectOutput out) throws IOException - { - // don't bother writing out the old value since it will never be rolled back - out.writeObject(key); - out.writeObject(newValue); - } - - public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException - { - key = (K) in.readObject(); - newValue = (V) in.readObject(); - } - } - - private static class RemoveOperation<K, V> extends Operation<K, V> - { - K key; - V oldValue; - - void rollback(Map<K, V> delegate) - { - if (oldValue != null) delegate.put(key, oldValue); - } - - void replay(Map<K, V> delegate) - { - delegate.remove(key); - } - - public void writeExternal(ObjectOutput out) throws IOException - { - out.writeObject(key); - } - - public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException - { - key = (K) in.readObject(); - } - } - - private static class ClearOperation<K, V> extends Operation<K, V> - { - FastCopyHashMap<K, V> originalEntries; - - void rollback(Map<K, V> delegate) - { - if (!originalEntries.isEmpty()) delegate.putAll(originalEntries); - } - - void replay(Map<K, V> delegate) - { - delegate.clear(); - } - } -} Added: core/branches/flat/src/main/java/org/jboss/starobrno/atomic/NullDelta.java =================================================================== --- core/branches/flat/src/main/java/org/jboss/starobrno/atomic/NullDelta.java (rev 0) +++ core/branches/flat/src/main/java/org/jboss/starobrno/atomic/NullDelta.java 2008-11-17 23:26:21 UTC (rev 7150) @@ -0,0 +1,52 @@ +/* + * JBoss, Home of Professional Open Source + * Copyright 2008, Red Hat Middleware LLC, and individual contributors + * by the @authors tag. See the copyright.txt in the distribution for a + * full listing of individual contributors. + * + * This is free software; you can redistribute it and/or modify it + * under the terms of the GNU Lesser General Public License as + * published by the Free Software Foundation; either version 2.1 of + * the License, or (at your option) any later version. + * + * This software is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this software; if not, write to the Free + * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA + * 02110-1301 USA, or see the FSF site: http://www.fsf.org. + */ +package org.jboss.starobrno.atomic; + +import java.io.Externalizable; +import java.io.IOException; +import java.io.ObjectInput; +import java.io.ObjectOutput; + +/** + * Represents no changes. + * + * @author Manik Surtani (<a href="mailto:manik AT jboss DOT org">manik AT jboss DOT org</a>) + */ +public class NullDelta implements Externalizable, Delta +{ + static final NullDelta INSTANCE = new NullDelta(); + + public void writeExternal(ObjectOutput out) throws IOException + { + // don't bother writing anything + } + + public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException + { + // nothing to read + } + + public DeltaAware merge(DeltaAware d) + { + return (d != null && d instanceof AtomicHashMap) ? d : new AtomicHashMap(); + } +} \ No newline at end of file Added: core/branches/flat/src/main/java/org/jboss/starobrno/atomic/Operation.java =================================================================== --- core/branches/flat/src/main/java/org/jboss/starobrno/atomic/Operation.java (rev 0) +++ core/branches/flat/src/main/java/org/jboss/starobrno/atomic/Operation.java 2008-11-17 23:26:21 UTC (rev 7150) @@ -0,0 +1,45 @@ +/* + * JBoss, Home of Professional Open Source + * Copyright 2008, Red Hat Middleware LLC, and individual contributors + * by the @authors tag. See the copyright.txt in the distribution for a + * full listing of individual contributors. + * + * This is free software; you can redistribute it and/or modify it + * under the terms of the GNU Lesser General Public License as + * published by the Free Software Foundation; either version 2.1 of + * the License, or (at your option) any later version. + * + * This software is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this software; if not, write to the Free + * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA + * 02110-1301 USA, or see the FSF site: http://www.fsf.org. + */ +package org.jboss.starobrno.atomic; + +import java.io.Externalizable; +import java.io.IOException; +import java.io.ObjectInput; +import java.io.ObjectOutput; +import java.util.Map; + +public abstract class Operation<K, V> implements Externalizable +{ + public abstract void replay(Map<K, V> delegate); + + public abstract void rollback(Map<K, V> delegate); + + public void writeExternal(ObjectOutput out) throws IOException + { + //no op + } + + public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException + { + //no op + } +} Added: core/branches/flat/src/main/java/org/jboss/starobrno/atomic/PutOperation.java =================================================================== --- core/branches/flat/src/main/java/org/jboss/starobrno/atomic/PutOperation.java (rev 0) +++ core/branches/flat/src/main/java/org/jboss/starobrno/atomic/PutOperation.java 2008-11-17 23:26:21 UTC (rev 7150) @@ -0,0 +1,63 @@ +/* + * JBoss, Home of Professional Open Source + * Copyright 2008, Red Hat Middleware LLC, and individual contributors + * by the @authors tag. See the copyright.txt in the distribution for a + * full listing of individual contributors. + * + * This is free software; you can redistribute it and/or modify it + * under the terms of the GNU Lesser General Public License as + * published by the Free Software Foundation; either version 2.1 of + * the License, or (at your option) any later version. + * + * This software is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this software; if not, write to the Free + * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA + * 02110-1301 USA, or see the FSF site: http://www.fsf.org. + */ +package org.jboss.starobrno.atomic; + +import java.io.IOException; +import java.io.ObjectInput; +import java.io.ObjectOutput; +import java.util.Map; + + +public class PutOperation<K, V> extends Operation<K, V> +{ + K key; + V oldValue; + V newValue; + + public void rollback(Map<K, V> delegate) + { + if (oldValue == null) + delegate.remove(key); + else + delegate.put(key, oldValue); + } + + public void replay(Map<K, V> delegate) + { + delegate.put(key, newValue); + } + + @Override + public void writeExternal(ObjectOutput out) throws IOException + { + // don't bother writing out the old value since it will never be rolled back + out.writeObject(key); + out.writeObject(newValue); + } + + @Override + public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException + { + key = (K) in.readObject(); + newValue = (V) in.readObject(); + } +} \ No newline at end of file Added: core/branches/flat/src/main/java/org/jboss/starobrno/atomic/RemoveOperation.java =================================================================== --- core/branches/flat/src/main/java/org/jboss/starobrno/atomic/RemoveOperation.java (rev 0) +++ core/branches/flat/src/main/java/org/jboss/starobrno/atomic/RemoveOperation.java 2008-11-17 23:26:21 UTC (rev 7150) @@ -0,0 +1,56 @@ +/* + * JBoss, Home of Professional Open Source + * Copyright 2008, Red Hat Middleware LLC, and individual contributors + * by the @authors tag. See the copyright.txt in the distribution for a + * full listing of individual contributors. + * + * This is free software; you can redistribute it and/or modify it + * under the terms of the GNU Lesser General Public License as + * published by the Free Software Foundation; either version 2.1 of + * the License, or (at your option) any later version. + * + * This software is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this software; if not, write to the Free + * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA + * 02110-1301 USA, or see the FSF site: http://www.fsf.org. + */ +package org.jboss.starobrno.atomic; + +import java.io.IOException; +import java.io.ObjectInput; +import java.io.ObjectOutput; +import java.util.Map; + + +public class RemoveOperation<K, V> extends Operation<K, V> +{ + K key; + V oldValue; + + public void rollback(Map<K, V> delegate) + { + if (oldValue != null) delegate.put(key, oldValue); + } + + public void replay(Map<K, V> delegate) + { + delegate.remove(key); + } + + @Override + public void writeExternal(ObjectOutput out) throws IOException + { + out.writeObject(key); + } + + @Override + public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException + { + key = (K) in.readObject(); + } +} \ No newline at end of file Added: core/branches/flat/src/main/java/org/jboss/starobrno/batch/AutoBatchSupport.java =================================================================== --- core/branches/flat/src/main/java/org/jboss/starobrno/batch/AutoBatchSupport.java (rev 0) +++ core/branches/flat/src/main/java/org/jboss/starobrno/batch/AutoBatchSupport.java 2008-11-17 23:26:21 UTC (rev 7150) @@ -0,0 +1,45 @@ +/* + * JBoss, Home of Professional Open Source + * Copyright 2008, Red Hat Middleware LLC, and individual contributors + * by the @authors tag. See the copyright.txt in the distribution for a + * full listing of individual contributors. + * + * This is free software; you can redistribute it and/or modify it + * under the terms of the GNU Lesser General Public License as + * published by the Free Software Foundation; either version 2.1 of + * the License, or (at your option) any later version. + * + * This software is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this software; if not, write to the Free + * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA + * 02110-1301 USA, or see the FSF site: http://www.fsf.org. + */ +package org.jboss.starobrno.batch; + +import org.jboss.starobrno.Cache; + +/** + * Enables for automatic batching. + * + * @author Manik Surtani (<a href="mailto:manik AT jboss DOT org">manik AT jboss DOT org</a>) + */ +public abstract class AutoBatchSupport +{ + boolean startedBatch; + protected Cache cache; + + protected void startAtomic() + { + if (!startedBatch) startedBatch = cache.startBatch(); + } + + protected void endAtomic() + { + if (startedBatch) cache.endBatch(true); + } +} Modified: core/branches/flat/src/main/java/org/jboss/starobrno/batch/BatchContainer.java =================================================================== --- core/branches/flat/src/main/java/org/jboss/starobrno/batch/BatchContainer.java 2008-11-17 23:15:11 UTC (rev 7149) +++ core/branches/flat/src/main/java/org/jboss/starobrno/batch/BatchContainer.java 2008-11-17 23:26:21 UTC (rev 7150) @@ -45,16 +45,23 @@ this.transactionManager = transactionManager; } - public void startBatch() throws CacheException + /** + * Starts a batch + * + * @return true if a batch was started; false if one was already available. + * @throws CacheException + */ + public boolean startBatch() throws CacheException { try { - if (transactionManager.getTransaction() != null) return; - if (batchTransactionContainer.get() == null) + if (transactionManager.getTransaction() == null && batchTransactionContainer.get() == null) { transactionManager.begin(); batchTransactionContainer.set(transactionManager.suspend()); + return true; } + return false; } catch (Exception e) { Modified: core/branches/flat/src/main/java/org/jboss/starobrno/commands/write/PutKeyValueCommand.java =================================================================== --- core/branches/flat/src/main/java/org/jboss/starobrno/commands/write/PutKeyValueCommand.java 2008-11-17 23:15:11 UTC (rev 7149) +++ core/branches/flat/src/main/java/org/jboss/starobrno/commands/write/PutKeyValueCommand.java 2008-11-17 23:26:21 UTC (rev 7150) @@ -21,6 +21,8 @@ */ package org.jboss.starobrno.commands.write; +import org.jboss.starobrno.atomic.Delta; +import org.jboss.starobrno.atomic.DeltaAware; import org.jboss.starobrno.commands.Visitor; import org.jboss.starobrno.commands.read.AbstractDataCommand; import org.jboss.starobrno.container.MVCCEntry; @@ -77,7 +79,21 @@ { notifier.notifyCacheEntryModified(key, true, ctx); MVCCEntry e = ctx.lookupEntry(key); - Object o = e.setValue(value); + Object o = null; + if (value instanceof Delta) + { + // magic + Delta dv = (Delta) value; + Object existing = e.getValue(); + DeltaAware toMergeWith = null; + if (existing instanceof DeltaAware) toMergeWith = (DeltaAware) existing; + e.setValue(dv.merge(toMergeWith)); + o = existing; + } + else + { + o = e.setValue(value); + } notifier.notifyCacheEntryModified(key, false, ctx); return o; } Modified: core/branches/flat/src/main/java/org/jboss/starobrno/config/parsing/RootElementBuilder.java =================================================================== --- core/branches/flat/src/main/java/org/jboss/starobrno/config/parsing/RootElementBuilder.java 2008-11-17 23:15:11 UTC (rev 7149) +++ core/branches/flat/src/main/java/org/jboss/starobrno/config/parsing/RootElementBuilder.java 2008-11-17 23:26:21 UTC (rev 7150) @@ -47,8 +47,8 @@ private static final JBossEntityResolver resolver = new JBossEntityResolver(); - public static final String JBOSSCACHE_CORE_NS = "urn:jboss:jbosscache-core:config:3.0"; - public static final String JBOSSCACHE_REPO_NS = "urn:jboss:jbosscache-core:cache-repo:3.0"; + public static final String JBOSSCACHE_CORE_NS = "urn:jboss:starobrno-core:config:1.0"; + public static final String JBOSSCACHE_REPO_NS = "urn:jboss:starobrno-core:cache-repo:1.0"; static { Modified: core/branches/flat/src/main/java/org/jboss/starobrno/config/parsing/XmlConfigurationParser.java =================================================================== --- core/branches/flat/src/main/java/org/jboss/starobrno/config/parsing/XmlConfigurationParser.java 2008-11-17 23:15:11 UTC (rev 7149) +++ core/branches/flat/src/main/java/org/jboss/starobrno/config/parsing/XmlConfigurationParser.java 2008-11-17 23:26:21 UTC (rev 7150) @@ -19,7 +19,7 @@ * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA * 02110-1301 USA, or see the FSF site: http://www.fsf.org. */ -package org.jboss.cache.config.parsing; +package org.jboss.starobrno.config.parsing; import org.jboss.cache.lock.IsolationLevel; import org.jboss.cache.util.FileLookup; @@ -29,9 +29,6 @@ import org.jboss.starobrno.config.Configuration.CacheMode; import org.jboss.starobrno.config.ConfigurationException; import org.jboss.starobrno.config.CustomInterceptorConfig; -import org.jboss.starobrno.config.parsing.JGroupsStackParser; -import org.jboss.starobrno.config.parsing.RootElementBuilder; -import org.jboss.starobrno.config.parsing.XmlParserBase; import org.jboss.starobrno.config.parsing.element.BuddyElementParser; import org.jboss.starobrno.config.parsing.element.CustomInterceptorsElementParser; import org.jboss.starobrno.config.parsing.element.EvictionElementParser; Modified: core/branches/flat/src/main/java/org/jboss/starobrno/marshall/CacheMarshallerStarobrno.java =================================================================== --- core/branches/flat/src/main/java/org/jboss/starobrno/marshall/CacheMarshallerStarobrno.java 2008-11-17 23:15:11 UTC (rev 7149) +++ core/branches/flat/src/main/java/org/jboss/starobrno/marshall/CacheMarshallerStarobrno.java 2008-11-17 23:26:21 UTC (rev 7150) @@ -24,25 +24,40 @@ import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.jboss.starobrno.CacheException; -import org.jboss.starobrno.io.ByteBuffer; -import org.jboss.starobrno.io.ExposedByteArrayOutputStream; +import org.jboss.starobrno.atomic.DeltaAware; import org.jboss.starobrno.commands.CommandsFactory; import org.jboss.starobrno.commands.ReplicableCommand; import org.jboss.starobrno.config.Configuration; -import org.jboss.starobrno.factories.annotations.Inject; +import org.jboss.starobrno.io.ByteBuffer; +import org.jboss.starobrno.io.ExposedByteArrayOutputStream; import org.jboss.starobrno.transaction.GlobalTransaction; import org.jboss.starobrno.util.FastCopyHashMap; import org.jboss.starobrno.util.Immutables; import org.jboss.util.NotImplementedException; import org.jboss.util.stream.MarshalledValueInputStream; -import org.jboss.cache.marshall.Marshaller; import org.jgroups.Address; import org.jgroups.stack.IpAddress; import org.jgroups.util.Buffer; -import java.io.*; +import java.io.ByteArrayInputStream; +import java.io.Externalizable; +import java.io.IOException; +import java.io.InputStream; +import java.io.ObjectInputStream; +import java.io.ObjectOutputStream; +import java.io.Serializable; import java.lang.reflect.Array; -import java.util.*; +import java.util.ArrayList; +import java.util.Collection; +import java.util.HashMap; +import java.util.HashSet; +import java.util.IdentityHashMap; +import java.util.LinkedList; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.TreeMap; +import java.util.TreeSet; /** * Abstract AbstractMarshaller for JBoss Cache. @@ -129,7 +144,8 @@ if (o != null && o.getClass().isArray() && isKnownType(o.getClass().getComponentType())) { marshallArray(o, out, refMap); - } else + } + else { if (o == null) { @@ -163,6 +179,13 @@ out.writeByte(MAGICNUMBER_MARSHALLEDVALUE); ((MarshalledValue) o).writeExternal(out); } + else if (o instanceof DeltaAware) + { + // reading in should be nothing special. + out.writeByte(MAGICNUMBER_SERIALIZABLE); + // only write the delta for these maps. + out.writeObject(((DeltaAware) o).delta()); + } else if (o instanceof GlobalTransaction) { out.writeByte(MAGICNUMBER_GTX); @@ -341,7 +364,8 @@ if (loader == null) { return unmarshallObject(in, refMap); - } else + } + else { Thread currentThread = Thread.currentThread(); ClassLoader old = currentThread.getContextClassLoader(); @@ -372,7 +396,8 @@ { reference = readReference(in); return refMap.getReferencedObject(reference); - } else break; + } + else break; case MAGICNUMBER_SERIALIZABLE: if (useRefs) reference = readReference(in); retVal = in.readObject(); @@ -671,7 +696,8 @@ boolean[] a = new boolean[sz]; for (int i = 0; i < sz; i++) a[i] = in.readBoolean(); return a; - } else + } + else { Boolean[] a = new Boolean[sz]; for (int i = 0; i < sz; i++) a[i] = in.readBoolean(); @@ -686,7 +712,8 @@ int[] a = new int[sz]; for (int i = 0; i < sz; i++) a[i] = in.readInt(); return a; - } else + } + else { Integer[] a = new Integer[sz]; for (int i = 0; i < sz; i++) a[i] = in.readInt(); @@ -701,7 +728,8 @@ long[] a = new long[sz]; for (int i = 0; i < sz; i++) a[i] = in.readLong(); return a; - } else + } + else { Long[] a = new Long[sz]; for (int i = 0; i < sz; i++) a[i] = in.readLong(); @@ -716,7 +744,8 @@ char[] a = new char[sz]; for (int i = 0; i < sz; i++) a[i] = in.readChar(); return a; - } else + } + else { Character[] a = new Character[sz]; for (int i = 0; i < sz; i++) a[i] = in.readChar(); @@ -739,7 +768,8 @@ bytesLeft -= read; } return a; - } else + } + else { Byte[] a = new Byte[sz]; for (int i = 0; i < sz; i++) a[i] = in.readByte(); @@ -754,7 +784,8 @@ short[] a = new short[sz]; for (int i = 0; i < sz; i++) a[i] = in.readShort(); return a; - } else + } + else { Short[] a = new Short[sz]; for (int i = 0; i < sz; i++) a[i] = in.readShort(); @@ -769,7 +800,8 @@ float[] a = new float[sz]; for (int i = 0; i < sz; i++) a[i] = in.readFloat(); return a; - } else + } + else { Float[] a = new Float[sz]; for (int i = 0; i < sz; i++) a[i] = in.readFloat(); @@ -784,7 +816,8 @@ double[] a = new double[sz]; for (int i = 0; i < sz; i++) a[i] = in.readDouble(); return a; - } else + } + else { Double[] a = new Double[sz]; for (int i = 0; i < sz; i++) a[i] = in.readDouble(); @@ -814,7 +847,8 @@ { out.writeByte(MAGICNUMBER_OBJECT); for (int i = 0; i < sz; i++) marshallObject(Array.get(o, i), out, refMap); - } else if (arrayTypeClass.equals(byte.class) || arrayTypeClass.equals(Byte.class)) + } + else if (arrayTypeClass.equals(byte.class) || arrayTypeClass.equals(Byte.class)) { out.writeByte(MAGICNUMBER_BYTE); out.writeBoolean(isPrim); @@ -822,7 +856,8 @@ out.write((byte[]) o); else for (int i = 0; i < sz; i++) out.writeByte((Byte) Array.get(o, i)); - } else if (arrayTypeClass.equals(int.class) || arrayTypeClass.equals(Integer.class)) + } + else if (arrayTypeClass.equals(int.class) || arrayTypeClass.equals(Integer.class)) { out.writeByte(MAGICNUMBER_INTEGER); out.writeBoolean(isPrim); @@ -830,7 +865,8 @@ for (int i = 0; i < sz; i++) out.writeInt(Array.getInt(o, i)); else for (int i = 0; i < sz; i++) out.writeInt((Integer) Array.get(o, i)); - } else if (arrayTypeClass.equals(long.class) || arrayTypeClass.equals(Long.class)) + } + else if (arrayTypeClass.equals(long.class) || arrayTypeClass.equals(Long.class)) { out.writeByte(MAGICNUMBER_LONG); out.writeBoolean(isPrim); @@ -838,7 +874,8 @@ for (int i = 0; i < sz; i++) out.writeLong(Array.getLong(o, i)); else for (int i = 0; i < sz; i++) out.writeLong((Long) Array.get(o, i)); - } else if (arrayTypeClass.equals(boolean.class) || arrayTypeClass.equals(Boolean.class)) + } + else if (arrayTypeClass.equals(boolean.class) || arrayTypeClass.equals(Boolean.class)) { out.writeByte(MAGICNUMBER_BOOLEAN); out.writeBoolean(isPrim); @@ -846,7 +883,8 @@ for (int i = 0; i < sz; i++) out.writeBoolean(Array.getBoolean(o, i)); else for (int i = 0; i < sz; i++) out.writeBoolean((Boolean) Array.get(o, i)); - } else if (arrayTypeClass.equals(char.class) || arrayTypeClass.equals(Character.class)) + } + else if (arrayTypeClass.equals(char.class) || arrayTypeClass.equals(Character.class)) { out.writeByte(MAGICNUMBER_CHAR); out.writeBoolean(isPrim); @@ -854,7 +892,8 @@ for (int i = 0; i < sz; i++) out.writeChar(Array.getChar(o, i)); else for (int i = 0; i < sz; i++) out.writeChar((Character) Array.get(o, i)); - } else if (arrayTypeClass.equals(short.class) || arrayTypeClass.equals(Short.class)) + } + else if (arrayTypeClass.equals(short.class) || arrayTypeClass.equals(Short.class)) { out.writeByte(MAGICNUMBER_SHORT); out.writeBoolean(isPrim); @@ -862,7 +901,8 @@ for (int i = 0; i < sz; i++) out.writeShort(Array.getShort(o, i)); else for (int i = 0; i < sz; i++) out.writeShort((Short) Array.get(o, i)); - } else if (arrayTypeClass.equals(float.class) || arrayTypeClass.equals(Float.class)) + } + else if (arrayTypeClass.equals(float.class) || arrayTypeClass.equals(Float.class)) { out.writeByte(MAGICNUMBER_FLOAT); out.writeBoolean(isPrim); @@ -870,7 +910,8 @@ for (int i = 0; i < sz; i++) out.writeFloat(Array.getFloat(o, i)); else for (int i = 0; i < sz; i++) out.writeFloat((Float) Array.get(o, i)); - } else if (arrayTypeClass.equals(double.class) || arrayTypeClass.equals(Double.class)) + } + else if (arrayTypeClass.equals(double.class) || arrayTypeClass.equals(Double.class)) { out.writeByte(MAGICNUMBER_DOUBLE); out.writeBoolean(isPrim); @@ -878,7 +919,8 @@ for (int i = 0; i < sz; i++) out.writeDouble(Array.getDouble(o, i)); else for (int i = 0; i < sz; i++) out.writeDouble((Double) Array.get(o, i)); - } else throw new CacheException("Unknown array type!"); + } + else throw new CacheException("Unknown array type!"); } protected boolean isKnownType(Class c) Modified: core/branches/flat/src/main/java/org/jboss/starobrno/tree/NodeImpl.java =================================================================== --- core/branches/flat/src/main/java/org/jboss/starobrno/tree/NodeImpl.java 2008-11-17 23:15:11 UTC (rev 7149) +++ core/branches/flat/src/main/java/org/jboss/starobrno/tree/NodeImpl.java 2008-11-17 23:26:21 UTC (rev 7150) @@ -25,7 +25,6 @@ import org.jboss.starobrno.util.Immutables; import org.jboss.starobrno.util.Util; -import java.util.Collections; import java.util.HashSet; import java.util.Map; import java.util.Set; @@ -41,9 +40,9 @@ Fqn fqn; Key dataKey, structureKey; - public NodeImpl(Fqn fqn, Cache cache, ThreadLocal<Boolean> atomicityContainer) + public NodeImpl(Fqn fqn, Cache cache) { - super(atomicityContainer, cache); + super(cache); this.fqn = fqn; dataKey = new Key(fqn, Type.DATA); structureKey = new Key(fqn, Type.STRUCTURE); @@ -52,7 +51,7 @@ public Node<K, V> getParent() { if (fqn.isRoot()) return this; - return new NodeImpl(fqn.getParent(), cache, batchStarted); + return new NodeImpl(fqn.getParent(), cache); } public Set<Node<K, V>> getChildren() @@ -63,7 +62,7 @@ Set set = new HashSet(); for (Fqn f : getStructure().children.values()) { - NodeImpl n = new NodeImpl(f, cache, batchStarted); + NodeImpl n = new NodeImpl(f, cache); set.add(n); } return Immutables.immutableSetWrap(set); @@ -108,7 +107,7 @@ try { Fqn absoluteChildFqn = Fqn.fromRelativeFqn(fqn, f); - NodeImpl child = new NodeImpl(absoluteChildFqn, cache, batchStarted); + NodeImpl child = new NodeImpl(absoluteChildFqn, cache); Structure s = getStructure(); s.children.put(f.getLastElement(), absoluteChildFqn); updateStructure(s); @@ -135,7 +134,7 @@ Fqn childFqn = s.children.remove(childName); if (childFqn != null) { - Node child = new NodeImpl(childFqn, cache, batchStarted); + Node child = new NodeImpl(childFqn, cache); child.removeChildren(); cache.remove(new Key(childFqn, Type.DATA)); cache.remove(new Key(childFqn, Type.STRUCTURE)); @@ -157,7 +156,7 @@ try { if (hasChild(f)) - return new NodeImpl(Fqn.fromRelativeFqn(fqn, f), cache, batchStarted); + return new NodeImpl(Fqn.fromRelativeFqn(fqn, f), cache); else return null; } @@ -173,7 +172,7 @@ try { if (hasChild(name)) - return new NodeImpl(Fqn.fromRelativeElements(fqn, name), cache, batchStarted); + return new NodeImpl(Fqn.fromRelativeElements(fqn, name), cache); else return null; } @@ -225,10 +224,8 @@ startAtomic(); try { - Map data = getData(); - V old = (V) data.put(key, value); - putData(data); - return old; + Map<K, V> data = getDataInternal(); + return data.put(key, value); } finally { @@ -263,7 +260,6 @@ { Map data = getDataInternal(); data.putAll(map); - putData(data); } finally { @@ -273,7 +269,17 @@ public void replaceAll(Map<? extends K, ? extends V> map) { - putData(map); + startAtomic(); + try + { + Map data = getDataInternal(); + data.clear(); + data.putAll(map); + } + finally + { + endAtomic(); + } } public V get(K key) @@ -286,10 +292,8 @@ startAtomic(); try { - Map data = getDataInternal(); - V old = (V) data.remove(key); - putData(data); - return old; + Map<K, V> data = getDataInternal(); + return data.remove(key); } finally { @@ -299,7 +303,7 @@ public void clearData() { - putData(Collections.emptyMap()); + getDataInternal().clear(); } public int dataSize() @@ -359,14 +363,9 @@ } } - void putData(Map m) - { - cache.put(dataKey, m); - } - Map<K, V> getDataInternal() { - return (Map<K, V>) cache.get(dataKey); + return (Map<K, V>) atomicMapCache.getAtomicMap(dataKey); } void updateStructure(Structure s) Modified: core/branches/flat/src/main/java/org/jboss/starobrno/tree/TreeCacheImpl.java =================================================================== --- core/branches/flat/src/main/java/org/jboss/starobrno/tree/TreeCacheImpl.java 2008-11-17 23:15:11 UTC (rev 7149) +++ core/branches/flat/src/main/java/org/jboss/starobrno/tree/TreeCacheImpl.java 2008-11-17 23:26:21 UTC (rev 7150) @@ -42,20 +42,13 @@ { public TreeCacheImpl(Cache<K, V> cache) { - super(new ThreadLocal<Boolean>() - { - @Override - protected Boolean initialValue() - { - return false; - } - }, cache); + super(cache); createRoot(); } public Node<K, V> getRoot() { - return new NodeImpl(Fqn.ROOT, cache, batchStarted); + return new NodeImpl(Fqn.ROOT, cache); } public V put(String fqn, K key, V value) @@ -130,7 +123,7 @@ try { if (exists(fqn)) - return new NodeImpl(fqn, cache, batchStarted); + return new NodeImpl(fqn, cache); else return null; } finally Modified: core/branches/flat/src/main/java/org/jboss/starobrno/tree/TreeStructureSupport.java =================================================================== --- core/branches/flat/src/main/java/org/jboss/starobrno/tree/TreeStructureSupport.java 2008-11-17 23:15:11 UTC (rev 7149) +++ core/branches/flat/src/main/java/org/jboss/starobrno/tree/TreeStructureSupport.java 2008-11-17 23:26:21 UTC (rev 7150) @@ -22,42 +22,23 @@ package org.jboss.starobrno.tree; import org.jboss.starobrno.Cache; +import org.jboss.starobrno.atomic.AtomicMapCache; +import org.jboss.starobrno.batch.AutoBatchSupport; import org.jboss.starobrno.util.Util; import java.util.HashMap; import java.util.Map; -public class TreeStructureSupport +public class TreeStructureSupport extends AutoBatchSupport { - ThreadLocal<Boolean> batchStarted; - Cache cache; - boolean batchOwner; + AtomicMapCache atomicMapCache; - public TreeStructureSupport(ThreadLocal<Boolean> batchStarted, Cache cache) + public TreeStructureSupport(Cache cache) { - this.batchStarted = batchStarted; this.cache = cache; + atomicMapCache = (AtomicMapCache) cache; } - void startAtomic() - { - if (!batchStarted.get()) - { - cache.startBatch(); - batchStarted.set(true); - batchOwner = true; - } - } - - void endAtomic() - { - if (batchOwner) - { - cache.endBatch(true); - batchStarted.set(false); - } - } - public boolean exists(Fqn f) { startAtomic(); @@ -89,7 +70,7 @@ if (cache.containsKey(dataKey) && cache.containsKey(structureKey)) return; Structure s = new Structure(); s.parent = fqn.getParent(); - cache.put(dataKey, new HashMap()); + atomicMapCache.getAtomicMap(dataKey); cache.put(structureKey, s); } finally Added: core/branches/flat/src/test/java/org/jboss/starobrno/atomic/APITest.java =================================================================== --- core/branches/flat/src/test/java/org/jboss/starobrno/atomic/APITest.java (rev 0) +++ core/branches/flat/src/test/java/org/jboss/starobrno/atomic/APITest.java 2008-11-17 23:26:21 UTC (rev 7150) @@ -0,0 +1,254 @@ +/* + * JBoss, Home of Professional Open Source + * Copyright 2008, Red Hat Middleware LLC, and individual contributors + * by the @authors tag. See the copyright.txt in the distribution for a + * full listing of individual contributors. + * + * This is free software; you can redistribute it and/or modify it + * under the terms of the GNU Lesser General Public License as + * published by the Free Software Foundation; either version 2.1 of + * the License, or (at your option) any later version. + * + * This software is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this software; if not, write to the Free + * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA + * 02110-1301 USA, or see the FSF site: http://www.fsf.org. + */ +package org.jboss.starobrno.atomic; + +import org.jboss.cache.DefaultCacheFactory; +import org.jboss.cache.transaction.DummyTransactionManager; +import org.jboss.cache.transaction.DummyTransactionManagerLookup; +import org.jboss.starobrno.config.Configuration; +import org.jboss.starobrno.config.Configuration.CacheMode; +import org.jboss.starobrno.util.TestingUtil; +import org.testng.annotations.Test; + +import javax.transaction.Transaction; +import java.util.Map; + +@Test(groups = "functional") +public class APITest +{ + + private void assertIsEmpty(Map map) + { + assert map.size() == 0; + assert map.get("blah") == null; + assert !map.containsKey("blah"); + } + + private void assertIsEmptyMap(AtomicMapCache cache, Object key) + { + assertIsEmpty(cache.getAtomicMap(key)); + } + + public void testAtomicMap() + { + Configuration c = new Configuration(); + c.setInvocationBatchingEnabled(true); + AtomicMapCache cache = (AtomicMapCache) new DefaultCacheFactory().createCache(c); + try + { + AtomicMap map = cache.getAtomicMap("map"); + + assertIsEmpty(map); + assertIsEmptyMap(cache, "map"); + + map.put("blah", "blah"); + assert map.size() == 1; + assert map.get("blah").equals("blah"); + assert map.containsKey("blah"); + + map.clear(); + + assertIsEmpty(map); + assertIsEmptyMap(cache, "map"); + } + finally + { + cache.stop(); + } + } + + + public void testReadSafetyEmptyCache() throws Exception + { + Configuration c = new Configuration(); + c.setTransactionManagerLookupClass(DummyTransactionManagerLookup.class.getName()); + c.setInvocationBatchingEnabled(true); + AtomicMapCache cache = (AtomicMapCache) new DefaultCacheFactory().createCache(c); + try + { + AtomicMap map = cache.getAtomicMap("map"); + + assertIsEmpty(map); + assertIsEmptyMap(cache, "map"); + + DummyTransactionManager.getInstance().begin(); + map.put("blah", "blah"); + assert map.size() == 1; + assert map.get("blah").equals("blah"); + assert map.containsKey("blah"); + Transaction t = DummyTransactionManager.getInstance().suspend(); + + assertIsEmpty(map); + assertIsEmptyMap(cache, "map"); + + DummyTransactionManager.getInstance().resume(t); + DummyTransactionManager.getInstance().commit(); + + assert map.size() == 1; + assert map.get("blah").equals("blah"); + assert map.containsKey("blah"); + + map.clear(); + + assertIsEmpty(map); + assertIsEmptyMap(cache, "map"); + } + finally + { + cache.stop(); + } + } + + public void testReadSafetyNotEmptyCache() throws Exception + { + Configuration c = new Configuration(); + c.setTransactionManagerLookupClass(DummyTransactionManagerLookup.class.getName()); + c.setInvocationBatchingEnabled(true); + AtomicMapCache cache = (AtomicMapCache) new DefaultCacheFactory().createCache(c); + try + { + AtomicMap map = cache.getAtomicMap("map"); + + DummyTransactionManager.getInstance().begin(); + map.put("blah", "blah"); + assert map.size() == 1; + assert map.get("blah").equals("blah"); + assert map.containsKey("blah"); + Transaction t = DummyTransactionManager.getInstance().suspend(); + + assertIsEmpty(map); + assertIsEmptyMap(cache, "map"); + + DummyTransactionManager.getInstance().resume(t); + DummyTransactionManager.getInstance().commit(); + + assert map.size() == 1; + assert map.get("blah").equals("blah"); + assert map.containsKey("blah"); + + map.clear(); + + assertIsEmpty(map); + assertIsEmptyMap(cache, "map"); + } + finally + { + cache.stop(); + } + } + + public void testReadSafetyRollback() throws Exception + { + Configuration c = new Configuration(); + c.setTransactionManagerLookupClass(DummyTransactionManagerLookup.class.getName()); + c.setInvocationBatchingEnabled(true); + AtomicMapCache cache = (AtomicMapCache) new DefaultCacheFactory().createCache(c); + try + { + AtomicMap map = cache.getAtomicMap("map"); + + DummyTransactionManager.getInstance().begin(); + map.put("blah", "blah"); + assert map.size() == 1; + assert map.get("blah").equals("blah"); + assert map.containsKey("blah"); + Transaction t = DummyTransactionManager.getInstance().suspend(); + + assertIsEmpty(map); + assertIsEmptyMap(cache, "map"); + + DummyTransactionManager.getInstance().resume(t); + DummyTransactionManager.getInstance().rollback(); + + assertIsEmpty(map); + assertIsEmptyMap(cache, "map"); + } + finally + { + cache.stop(); + } + } + + public void testReplicationCommit() throws Exception + { + Configuration c = new Configuration(); + c.setTransactionManagerLookupClass(DummyTransactionManagerLookup.class.getName()); + c.setCacheMode(CacheMode.REPL_SYNC); + c.setInvocationBatchingEnabled(true); + AtomicMapCache cache1 = (AtomicMapCache) new DefaultCacheFactory().createCache(c.clone()); + AtomicMapCache cache2 = (AtomicMapCache) new DefaultCacheFactory().createCache(c.clone()); + + try + { + TestingUtil.blockUntilViewsReceived(20000, cache1, cache2); + AtomicMap map = cache1.getAtomicMap("map"); + + DummyTransactionManager.getInstance().begin(); + map.put("existing", "existing"); + map.put("blah", "blah"); + DummyTransactionManager.getInstance().commit(); + + assert map.size() == 2; + assert map.get("blah").equals("blah"); + assert map.containsKey("blah"); + + System.out.println("Map on cache 2 is " + cache2.getAtomicMap("map")); + + assert cache2.getAtomicMap("map").size() == 2; + assert cache2.getAtomicMap("map").get("blah").equals("blah"); + assert cache2.getAtomicMap("map").containsKey("blah"); + } + finally + { + TestingUtil.killCaches(cache1, cache2); + } + } + + public void testReplicationRollback() throws Exception + { + Configuration c = new Configuration(); + c.setTransactionManagerLookupClass(DummyTransactionManagerLookup.class.getName()); + c.setCacheMode(CacheMode.REPL_SYNC); + c.setInvocationBatchingEnabled(true); + AtomicMapCache cache1 = (AtomicMapCache) new DefaultCacheFactory().createCache(c.clone()); + AtomicMapCache cache2 = (AtomicMapCache) new DefaultCacheFactory().createCache(c.clone()); + + try + { + TestingUtil.blockUntilViewsReceived(20000, cache1, cache2); + AtomicMap map = cache1.getAtomicMap("map"); + + DummyTransactionManager.getInstance().begin(); + map.put("existing", "existing"); + map.put("blah", "blah"); + DummyTransactionManager.getInstance().rollback(); + + assertIsEmpty(map); + assertIsEmptyMap(cache1, "map"); + assertIsEmptyMap(cache2, "map"); + } + finally + { + TestingUtil.killCaches(cache1, cache2); + } + } +} Modified: core/branches/flat/src/test/java/org/jboss/starobrno/tree/api/NodeAPITest.java =================================================================== --- core/branches/flat/src/test/java/org/jboss/starobrno/tree/api/NodeAPITest.java 2008-11-17 23:15:11 UTC (rev 7149) +++ core/branches/flat/src/test/java/org/jboss/starobrno/tree/api/NodeAPITest.java 2008-11-17 23:26:21 UTC (rev 7150) @@ -1,7 +1,10 @@ package org.jboss.cache.api; +import org.jboss.cache.transaction.DummyTransactionManager; +import org.jboss.cache.transaction.DummyTransactionManagerLookup; import org.jboss.starobrno.CacheSPI; import org.jboss.starobrno.UnitTestCacheFactory; +import org.jboss.starobrno.config.Configuration; import org.jboss.starobrno.tree.Fqn; import org.jboss.starobrno.tree.Node; import org.jboss.starobrno.tree.TreeCache; @@ -37,7 +40,9 @@ public void setUp() throws Exception { // start a single cache instance - CacheSPI<Object, Object> cache = (CacheSPI<Object, Object>) new UnitTestCacheFactory<Object, Object>().createCache("configs/local-tx.xml", false); + Configuration c = new Configuration(); + c.setTransactionManagerLookupClass(DummyTransactionManagerLookup.class.getName()); + CacheSPI<Object, Object> cache = (CacheSPI<Object, Object>) new UnitTestCacheFactory<Object, Object>().createCache(c, false); cache.getConfiguration().setInvocationBatchingEnabled(true); cache.start(); cacheTL.set(new TreeCacheImpl(cache)); @@ -355,7 +360,9 @@ { TreeCache<Object, Object> cache = cacheTL.get(); + assert DummyTransactionManager.getInstance().getTransaction() == null; cache.put("/foo/1/2/3", "item", 1); + assert DummyTransactionManager.getInstance().getTransaction() == null; assert 1 == (Integer) cache.get("/foo/1/2/3", "item"); tm.begin(); assert 1 == (Integer) cache.get("/foo/1/2/3", "item"); Modified: core/branches/flat/src/test/java/org/jboss/starobrno/tree/api/TreeCacheAPITest.java =================================================================== --- core/branches/flat/src/test/java/org/jboss/starobrno/tree/api/TreeCacheAPITest.java 2008-11-17 23:15:11 UTC (rev 7149) +++ core/branches/flat/src/test/java/org/jboss/starobrno/tree/api/TreeCacheAPITest.java 2008-11-17 23:26:21 UTC (rev 7150) @@ -1,8 +1,8 @@ package org.jboss.starobrno.tree.api; +import org.jboss.cache.transaction.DummyTransactionManagerLookup; import org.jboss.starobrno.CacheSPI; import org.jboss.starobrno.config.Configuration; -import org.jboss.starobrno.config.parsing.XmlConfigurationParser; import org.jboss.starobrno.manager.CacheManager; import org.jboss.starobrno.tree.Fqn; import org.jboss.starobrno.tree.Node; @@ -33,10 +33,8 @@ public void setUp() throws Exception { // start a single cache instance - XmlConfigurationParser parser = new XmlConfigurationParser(); - Configuration c; - c = parser.parseFile("configs/local-tx.xml"); - c.setEvictionConfig(null); + Configuration c = new Configuration(); + c.setTransactionManagerLookupClass(DummyTransactionManagerLookup.class.getName()); CacheManager cm = new CacheManager(c); cache = cm.createTreeCache(getClass().getSimpleName());