Author: nfilotto Date: 2011-04-15 15:11:33 -0400 (Fri, 15 Apr 2011) New Revision: 4240 Added: jcr/trunk/exo.jcr.component.core/src/test/java/org/exoplatform/services/jcr/impl/dataflow/persistent/TestWorkspaceStorageCacheInClusterMode.java jcr/trunk/exo.jcr.component.core/src/test/java/org/exoplatform/services/jcr/impl/dataflow/persistent/infinispan/ jcr/trunk/exo.jcr.component.core/src/test/java/org/exoplatform/services/jcr/impl/dataflow/persistent/infinispan/TestISPNCacheWorkspaceStorageCacheInClusterMode.java jcr/trunk/exo.jcr.component.core/src/test/java/org/exoplatform/services/jcr/impl/dataflow/persistent/jbosscache/TestJBossCacheWorkspaceStorageCacheInClusterMode.java jcr/trunk/exo.jcr.component.core/src/test/resources/conf/standalone/cluster/test-jbosscache-data-no-mux.xml Modified: jcr/trunk/exo.jcr.component.core/src/main/java/org/exoplatform/services/jcr/impl/dataflow/persistent/infinispan/BufferedISPNCache.java jcr/trunk/exo.jcr.component.core/src/main/java/org/exoplatform/services/jcr/impl/dataflow/persistent/jbosscache/BufferedJBossCache.java jcr/trunk/exo.jcr.component.core/src/test/java/org/exoplatform/services/jcr/impl/dataflow/persistent/TestCacheableWorkspaceDataManager.java Log: EXOJCR-1302: Second part of the patch that ensure the atomicity of any operations in the apply methods Modified: jcr/trunk/exo.jcr.component.core/src/main/java/org/exoplatform/services/jcr/impl/dataflow/persistent/infinispan/BufferedISPNCache.java =================================================================== --- jcr/trunk/exo.jcr.component.core/src/main/java/org/exoplatform/services/jcr/impl/dataflow/persistent/infinispan/BufferedISPNCache.java 2011-04-15 18:02:47 UTC (rev 4239) +++ jcr/trunk/exo.jcr.component.core/src/main/java/org/exoplatform/services/jcr/impl/dataflow/persistent/infinispan/BufferedISPNCache.java 2011-04-15 19:11:33 UTC (rev 4240) @@ -38,6 +38,7 @@ import java.util.Set; import java.util.concurrent.TimeUnit; +import javax.transaction.Status; import javax.transaction.TransactionManager; /** @@ -160,6 +161,11 @@ } public abstract void apply(); + + public boolean isTxRequired() + { + return false; + } } /** @@ -255,6 +261,12 @@ + existingObject.getClass().getName()); } } + + @Override + public boolean isTxRequired() + { + return true; + } } /** @@ -291,6 +303,12 @@ cache.put(key, newSet); } } + + @Override + public boolean isTxRequired() + { + return true; + } } /** @@ -892,12 +910,62 @@ public void commitTransaction() { CompressedISPNChangesBuffer changesContainer = getChangesBufferSafe(); + TransactionManager tm = getTransactionManager(); try { List<ChangesContainer> containers = changesContainer.getSortedList(); for (ChangesContainer cacheChange : containers) { - cacheChange.apply(); + boolean isTxCreated = false; + try + { + if (cacheChange.isTxRequired() && tm != null && tm.getStatus() == Status.STATUS_NO_TRANSACTION) + { + // No tx exists so we create a new tx + if (LOG.isTraceEnabled()) + LOG.trace("No Tx is active we then create a new tx"); + tm.begin(); + isTxCreated = true; + } + } + catch (Exception e) + { + LOG.warn("Could not create a new tx", e); + } + try + { + cacheChange.apply(); + } + catch (RuntimeException e) + { + if (isTxCreated) + { + try + { + if (LOG.isTraceEnabled()) + LOG.trace("An error occurs the tx will be rollbacked"); + tm.rollback(); + } + catch (Exception e1) + { + LOG.warn("Could not rollback the tx", e1); + } + } + throw e; + } + if (isTxCreated) + { + try + { + if (LOG.isTraceEnabled()) + LOG.trace("The tx will be committed"); + tm.commit(); + } + catch (Exception e) + { + LOG.warn("Could not commit the tx", e); + } + } } } finally Modified: jcr/trunk/exo.jcr.component.core/src/main/java/org/exoplatform/services/jcr/impl/dataflow/persistent/jbosscache/BufferedJBossCache.java =================================================================== --- jcr/trunk/exo.jcr.component.core/src/main/java/org/exoplatform/services/jcr/impl/dataflow/persistent/jbosscache/BufferedJBossCache.java 2011-04-15 18:02:47 UTC (rev 4239) +++ jcr/trunk/exo.jcr.component.core/src/main/java/org/exoplatform/services/jcr/impl/dataflow/persistent/jbosscache/BufferedJBossCache.java 2011-04-15 19:11:33 UTC (rev 4240) @@ -43,6 +43,7 @@ import java.util.Map; import java.util.Set; +import javax.transaction.Status; import javax.transaction.TransactionManager; /** @@ -68,12 +69,15 @@ private final long expirationTimeOut; + private final TransactionManager tm; + protected static final Log LOG = ExoLogger.getLogger("org.exoplatform.services.jcr.impl.dataflow.persistent.jbosscache.BufferedJBossCache"); public BufferedJBossCache(Cache<Serializable, Object> parentCache, boolean useExpiration, long expirationTimeOut) { super(); + this.tm = ((CacheSPI<Serializable, Object>)parentCache).getTransactionManager(); this.parentCache = parentCache; this.useExpiration = useExpiration; this.expirationTimeOut = expirationTimeOut; @@ -112,7 +116,53 @@ //log.info("After=" + changesContainer.toString()); for (ChangesContainer cacheChange : containers) { - cacheChange.apply(); + boolean isTxCreated = false; + try + { + if (cacheChange.isTxRequired() && tm != null && tm.getStatus() == Status.STATUS_NO_TRANSACTION) + { + // No tx exists so we create a new tx + if (LOG.isTraceEnabled()) LOG.trace("No Tx is active we then create a new tx"); + tm.begin(); + isTxCreated = true; + } + } + catch (Exception e) + { + LOG.warn("Could not create a new tx", e); + } + try + { + cacheChange.apply(); + } + catch (RuntimeException e) + { + if (isTxCreated) + { + try + { + if (LOG.isTraceEnabled()) LOG.trace("An error occurs the tx will be rollbacked"); + tm.rollback(); + } + catch (Exception e1) + { + LOG.warn("Could not rollback the tx", e1); + } + } + throw e; + } + if (isTxCreated) + { + try + { + if (LOG.isTraceEnabled()) LOG.trace("The tx will be committed"); + tm.commit(); + } + catch (Exception e) + { + LOG.warn("Could not commit the tx", e); + } + } } } finally @@ -818,6 +868,11 @@ } public abstract void apply(); + + public boolean isTxRequired() + { + return false; + } } /** @@ -882,6 +937,12 @@ setCacheLocalMode(); cache.put(fqn, key, value); } + + @Override + public boolean isTxRequired() + { + return true; + } } /** @@ -943,7 +1004,7 @@ // object found by FQN and key; Object existingObject = cache.get(getFqn(), key); Set<Object> newSet = new HashSet<Object>(); - // if set found of null, perform add + // if set found or null, perform add if (existingObject instanceof Set || (existingObject == null && forceModify)) { // set found @@ -967,6 +1028,12 @@ + existingObject.getClass().getName()); } } + + @Override + public boolean isTxRequired() + { + return true; + } } /** @@ -1009,6 +1076,12 @@ cache.put(fqn, key, newSet); } } + + @Override + public boolean isTxRequired() + { + return false; + } } /** Modified: jcr/trunk/exo.jcr.component.core/src/test/java/org/exoplatform/services/jcr/impl/dataflow/persistent/TestCacheableWorkspaceDataManager.java =================================================================== --- jcr/trunk/exo.jcr.component.core/src/test/java/org/exoplatform/services/jcr/impl/dataflow/persistent/TestCacheableWorkspaceDataManager.java 2011-04-15 18:02:47 UTC (rev 4239) +++ jcr/trunk/exo.jcr.component.core/src/test/java/org/exoplatform/services/jcr/impl/dataflow/persistent/TestCacheableWorkspaceDataManager.java 2011-04-15 19:11:33 UTC (rev 4240) @@ -534,7 +534,6 @@ { } - @Override public int getLastOrderNumber(NodeData parent) throws RepositoryException { return -1; @@ -592,7 +591,6 @@ return "0"; } - @Override public String getUniqueName() { return "MyWorkspaceDataContainer"; Added: jcr/trunk/exo.jcr.component.core/src/test/java/org/exoplatform/services/jcr/impl/dataflow/persistent/TestWorkspaceStorageCacheInClusterMode.java =================================================================== --- jcr/trunk/exo.jcr.component.core/src/test/java/org/exoplatform/services/jcr/impl/dataflow/persistent/TestWorkspaceStorageCacheInClusterMode.java (rev 0) +++ jcr/trunk/exo.jcr.component.core/src/test/java/org/exoplatform/services/jcr/impl/dataflow/persistent/TestWorkspaceStorageCacheInClusterMode.java 2011-04-15 19:11:33 UTC (rev 4240) @@ -0,0 +1,647 @@ +/* + * Copyright (C) 2011 eXo Platform SAS. + * + * 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.exoplatform.services.jcr.impl.dataflow.persistent; + +import org.exoplatform.services.jcr.JcrImplBaseTest; +import org.exoplatform.services.jcr.dataflow.ItemState; +import org.exoplatform.services.jcr.dataflow.PlainChangesLog; +import org.exoplatform.services.jcr.dataflow.PlainChangesLogImpl; +import org.exoplatform.services.jcr.dataflow.persistent.PersistedNodeData; +import org.exoplatform.services.jcr.dataflow.persistent.PersistedPropertyData; +import org.exoplatform.services.jcr.dataflow.persistent.WorkspaceStorageCache; +import org.exoplatform.services.jcr.datamodel.InternalQName; +import org.exoplatform.services.jcr.datamodel.ItemData; +import org.exoplatform.services.jcr.datamodel.ItemType; +import org.exoplatform.services.jcr.datamodel.NodeData; +import org.exoplatform.services.jcr.datamodel.PropertyData; +import org.exoplatform.services.jcr.datamodel.QPath; +import org.exoplatform.services.jcr.datamodel.QPathEntry; +import org.exoplatform.services.jcr.datamodel.ValueData; +import org.exoplatform.services.jcr.impl.Constants; +import org.exoplatform.services.jcr.impl.storage.SystemDataContainerHolder; +import org.exoplatform.services.jcr.impl.storage.WorkspaceDataContainerBase; +import org.exoplatform.services.jcr.storage.WorkspaceDataContainer; +import org.exoplatform.services.jcr.storage.WorkspaceStorageConnection; + +import java.io.UnsupportedEncodingException; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.atomic.AtomicReference; + +import javax.jcr.InvalidItemStateException; +import javax.jcr.PropertyType; +import javax.jcr.RepositoryException; + +/** + * @author <a href="mailto:nfilotto@exoplatform.com">Nicolas Filotto</a> + * @version $Id$ + * + */ +public abstract class TestWorkspaceStorageCacheInClusterMode<T extends WorkspaceStorageCache> extends JcrImplBaseTest +{ + public abstract T getCacheImpl() throws Exception; + + public void testRaceConditions() throws Exception + { + T cache1 = null, cache2 = null; + try + { + MyWorkspaceStorageConnection con = new MyWorkspaceStorageConnection(); + WorkspaceDataContainer wdc = new MyWorkspaceDataContainer(con); + CacheableWorkspaceDataManager cwdmNode1 = + new CacheableWorkspaceDataManager(wdc, cache1 = getCacheImpl(), new SystemDataContainerHolder(wdc)); + CacheableWorkspaceDataManager cwdmNode2 = + new CacheableWorkspaceDataManager(wdc, cache2 = getCacheImpl(), new SystemDataContainerHolder(wdc)); + NodeData parentNode = new PersistedNodeData("parent-id", QPath.makeChildPath(Constants.ROOT_PATH, new InternalQName(null, "parent-node")), Constants.ROOT_UUID, 1, 0, + Constants.NT_UNSTRUCTURED, new InternalQName[0], null); + // Test getChildNodesData + Action readAction = new Action(cwdmNode2) + { + public void execute(NodeData parentNode) throws Exception + { + cwdm.getChildNodesData(parentNode); + } + }; + Action writeAction = new Action(cwdmNode1) + { + public void execute(NodeData parentNode) throws Exception + { + PlainChangesLog chlog = new PlainChangesLogImpl(); + cwdm.getChildNodesData(parentNode); + chlog.add(ItemState.createAddedState(new PersistedNodeData("id-node" + parentNode.getIdentifier(), QPath.makeChildPath(parentNode.getQPath(), new InternalQName(null, "node")), parentNode.getIdentifier(), 1, 0, + Constants.NT_UNSTRUCTURED, new InternalQName[0], null))); + cwdm.save(chlog); + } + }; + executeConcurrentReadNWrite(con, readAction, writeAction, Mode.READ_FIRST, parentNode); + assertNotNull(cwdmNode1.getChildNodesData(parentNode)); + assertEquals(2, cwdmNode1.getChildNodesData(parentNode).size()); + assertNotNull(cwdmNode2.getChildNodesData(parentNode)); + assertEquals(2, cwdmNode2.getChildNodesData(parentNode).size()); + parentNode = new PersistedNodeData("parent-id2", QPath.makeChildPath(Constants.ROOT_PATH, new InternalQName(null, "parent-node2")), Constants.ROOT_UUID, 1, 0, + Constants.NT_UNSTRUCTURED, new InternalQName[0], null); + executeConcurrentReadNWrite(con, readAction, writeAction, Mode.WRITE_FIRST, parentNode); + assertNotNull(cwdmNode1.getChildNodesData(parentNode)); + assertEquals(2, cwdmNode1.getChildNodesData(parentNode).size()); + assertNotNull(cwdmNode2.getChildNodesData(parentNode)); + assertEquals(2, cwdmNode2.getChildNodesData(parentNode).size()); + // Test getChildPropertiesData + readAction = new Action(cwdmNode2) + { + public void execute(NodeData parentNode) throws Exception + { + cwdm.getChildPropertiesData(parentNode); + } + }; + writeAction = new Action(cwdmNode1) + { + public void execute(NodeData parentNode) throws Exception + { + PlainChangesLog chlog = new PlainChangesLogImpl(); + cwdm.getChildPropertiesData(parentNode); + chlog.add(ItemState.createAddedState(new PersistedPropertyData("id-property" + parentNode.getIdentifier(), QPath.makeChildPath( + parentNode.getQPath(), new InternalQName(null, "property")), parentNode.getIdentifier(), 0, + PropertyType.STRING, false, Arrays.asList((ValueData)new ByteArrayPersistedValueData(0, "some data".getBytes("UTF-8")))))); + cwdm.save(chlog); + } + }; + parentNode = new PersistedNodeData("parent-id3", QPath.makeChildPath(Constants.ROOT_PATH, new InternalQName(null, "parent-node3")), Constants.ROOT_UUID, 1, 0, + Constants.NT_UNSTRUCTURED, new InternalQName[0], null); + executeConcurrentReadNWrite(con, readAction, writeAction, Mode.READ_FIRST, parentNode); + assertNotNull(cwdmNode1.getChildPropertiesData(parentNode)); + assertEquals(2, cwdmNode1.getChildPropertiesData(parentNode).size()); + assertNotNull(cwdmNode2.getChildPropertiesData(parentNode)); + assertEquals(2, cwdmNode2.getChildPropertiesData(parentNode).size()); + parentNode = new PersistedNodeData("parent-id4", QPath.makeChildPath(Constants.ROOT_PATH, new InternalQName(null, "parent-node4")), Constants.ROOT_UUID, 1, 0, + Constants.NT_UNSTRUCTURED, new InternalQName[0], null); + executeConcurrentReadNWrite(con, readAction, writeAction, Mode.WRITE_FIRST, parentNode); + assertNotNull(cwdmNode1.getChildPropertiesData(parentNode)); + assertEquals(2, cwdmNode1.getChildPropertiesData(parentNode).size()); + assertNotNull(cwdmNode2.getChildPropertiesData(parentNode)); + assertEquals(2, cwdmNode2.getChildPropertiesData(parentNode).size()); + // Test getReferencesData + readAction = new Action(cwdmNode2) + { + public void execute(NodeData parentNode) throws Exception + { + cwdm.getReferencesData(parentNode.getIdentifier(), false); + } + }; + writeAction = new Action(cwdmNode1) + { + public void execute(NodeData parentNode) throws Exception + { + PlainChangesLog chlog = new PlainChangesLogImpl(); + cwdm.getReferencesData(parentNode.getIdentifier(), false); + chlog.add(ItemState.createAddedState(new PersistedPropertyData("id-reference" + parentNode.getIdentifier(), QPath.makeChildPath( + parentNode.getQPath(), new InternalQName(null, "reference")), parentNode.getIdentifier(), 0, + PropertyType.REFERENCE, false, Arrays.asList((ValueData)new ByteArrayPersistedValueData(0, parentNode.getIdentifier().getBytes("UTF-8")))))); + cwdm.save(chlog); + } + }; + parentNode = new PersistedNodeData("parent-id5", QPath.makeChildPath(Constants.ROOT_PATH, new InternalQName(null, "parent-node5")), Constants.ROOT_UUID, 1, 0, + Constants.NT_UNSTRUCTURED, new InternalQName[0], null); + executeConcurrentReadNWrite(con, readAction, writeAction, Mode.READ_FIRST, parentNode); + assertNotNull(cwdmNode1.getReferencesData(parentNode.getIdentifier(), false)); + assertEquals(2, cwdmNode1.getReferencesData(parentNode.getIdentifier(), false).size()); + assertNotNull(cwdmNode2.getReferencesData(parentNode.getIdentifier(), false)); + assertEquals(2, cwdmNode2.getReferencesData(parentNode.getIdentifier(), false).size()); + parentNode = new PersistedNodeData("parent-id6", QPath.makeChildPath(Constants.ROOT_PATH, new InternalQName(null, "parent-node6")), Constants.ROOT_UUID, 1, 0, + Constants.NT_UNSTRUCTURED, new InternalQName[0], null); + executeConcurrentReadNWrite(con, readAction, writeAction, Mode.WRITE_FIRST, parentNode); + assertNotNull(cwdmNode1.getReferencesData(parentNode.getIdentifier(), false)); + assertEquals(2, cwdmNode1.getReferencesData(parentNode.getIdentifier(), false).size()); + assertNotNull(cwdmNode2.getReferencesData(parentNode.getIdentifier(), false)); + assertEquals(2, cwdmNode2.getReferencesData(parentNode.getIdentifier(), false).size()); + + // Test getItemData by Id + readAction = new Action(cwdmNode2) + { + public void execute(NodeData parentNode) throws Exception + { + cwdm.getItemData(parentNode.getIdentifier()); + } + }; + writeAction = new Action(cwdmNode1) + { + public void execute(NodeData parentNode) throws Exception + { + PlainChangesLog chlog = new PlainChangesLogImpl(); + cwdm.getItemData(parentNode.getIdentifier()); + chlog.add(ItemState.createUpdatedState(new PersistedNodeData(parentNode.getIdentifier(), parentNode.getQPath(), Constants.ROOT_UUID, 2, 1, + Constants.NT_UNSTRUCTURED, new InternalQName[0], null))); + cwdm.save(chlog); + } + }; + parentNode = new PersistedNodeData("parent-id7", QPath.makeChildPath(Constants.ROOT_PATH, new InternalQName(null, "parent-node7")), Constants.ROOT_UUID, 1, 0, + Constants.NT_UNSTRUCTURED, new InternalQName[0], null); + executeConcurrentReadNWrite(con, readAction, writeAction, Mode.READ_FIRST, parentNode); + assertNotNull(cwdmNode1.getItemData(parentNode.getIdentifier())); + assertEquals(2, cwdmNode1.getItemData(parentNode.getIdentifier()).getPersistedVersion()); + assertNotNull(cwdmNode2.getItemData(parentNode.getIdentifier())); + assertEquals(2, cwdmNode2.getItemData(parentNode.getIdentifier()).getPersistedVersion()); + parentNode = new PersistedNodeData("parent-id8", QPath.makeChildPath(Constants.ROOT_PATH, new InternalQName(null, "parent-node8")), Constants.ROOT_UUID, 1, 0, + Constants.NT_UNSTRUCTURED, new InternalQName[0], null); + executeConcurrentReadNWrite(con, readAction, writeAction, Mode.WRITE_FIRST, parentNode); + assertNotNull(cwdmNode1.getItemData(parentNode.getIdentifier())); + assertEquals(2, cwdmNode1.getItemData(parentNode.getIdentifier()).getPersistedVersion()); + assertNotNull(cwdmNode2.getItemData(parentNode.getIdentifier())); + assertEquals(2, cwdmNode2.getItemData(parentNode.getIdentifier()).getPersistedVersion()); + + // Test getItemData by Path + final QPathEntry qpe = new QPathEntry(null, "my-property", 1); + readAction = new Action(cwdmNode2) + { + public void execute(NodeData parentNode) throws Exception + { + cwdm.getItemData(parentNode, qpe, ItemType.PROPERTY); + } + }; + writeAction = new Action(cwdmNode1) + { + public void execute(NodeData parentNode) throws Exception + { + PlainChangesLog chlog = new PlainChangesLogImpl(); + cwdm.getItemData(parentNode, qpe, ItemType.PROPERTY); + chlog.add(ItemState.createUpdatedState(new PersistedPropertyData("property-by-path" + + parentNode.getIdentifier(), QPath.makeChildPath(parentNode.getQPath(), qpe), parentNode + .getIdentifier(), 2, PropertyType.STRING, false, Arrays + .asList((ValueData)new ByteArrayPersistedValueData(0, "some new data".getBytes("UTF-8")))))); + cwdm.save(chlog); + } + }; + parentNode = new PersistedNodeData("parent-id9", QPath.makeChildPath(Constants.ROOT_PATH, new InternalQName(null, "parent-node9")), Constants.ROOT_UUID, 1, 0, + Constants.NT_UNSTRUCTURED, new InternalQName[0], null); + executeConcurrentReadNWrite(con, readAction, writeAction, Mode.READ_FIRST, parentNode); + assertNotNull(cwdmNode1.getItemData(parentNode, qpe, ItemType.PROPERTY)); + assertEquals(2, cwdmNode1.getItemData(parentNode, qpe, ItemType.PROPERTY).getPersistedVersion()); + assertNotNull(cwdmNode2.getItemData(parentNode, qpe, ItemType.PROPERTY)); + assertEquals(2, cwdmNode2.getItemData(parentNode, qpe, ItemType.PROPERTY).getPersistedVersion()); + parentNode = new PersistedNodeData("parent-id10", QPath.makeChildPath(Constants.ROOT_PATH, new InternalQName(null, "parent-node10")), Constants.ROOT_UUID, 1, 0, + Constants.NT_UNSTRUCTURED, new InternalQName[0], null); + executeConcurrentReadNWrite(con, readAction, writeAction, Mode.WRITE_FIRST, parentNode); + assertNotNull(cwdmNode1.getItemData(parentNode, qpe, ItemType.PROPERTY)); + assertEquals(2, cwdmNode1.getItemData(parentNode, qpe, ItemType.PROPERTY).getPersistedVersion()); + assertNotNull(cwdmNode2.getItemData(parentNode, qpe, ItemType.PROPERTY)); + assertEquals(2, cwdmNode2.getItemData(parentNode, qpe, ItemType.PROPERTY).getPersistedVersion()); + } + finally + { + if (cache1 != null) + { + try + { + finalize(cache1); + } + catch (Exception e) + { + // ignore me + } + } + if (cache2 != null) + { + try + { + finalize(cache2); + } + catch (Exception e) + { + // ignore me + } + } + } + } + + protected void finalize(T cache) + { + } + + /** + * @param con + * @param cwdm + * @param mode + * @param idNode + * @throws InterruptedException + */ + private void executeConcurrentReadNWrite(final MyWorkspaceStorageConnection con, final Action readAction, + final Action writeAction, final Mode mode, final NodeData parentNode) throws InterruptedException + { + final CountDownLatch goSignal = con.initCountDownLatch(); + con.setParentNode(parentNode); + final AtomicReference<Exception> ex = new AtomicReference<Exception>(); + final CountDownLatch startSignal = new CountDownLatch(1); + final CountDownLatch doneSignal = new CountDownLatch(2); + Thread writer = new Thread() + { + public void run() + { + try + { + startSignal.await(); + con.wait.set(mode != Mode.WRITE_FIRST); + writeAction.execute(parentNode); + } + catch (Exception e) + { + e.printStackTrace(); + ex.set(e); + } + finally + { + if (mode == Mode.WRITE_FIRST) goSignal.countDown(); + doneSignal.countDown(); + } + } + }; + writer.start(); + Thread reader = new Thread() + { + public void run() + { + try + { + startSignal.await(); + con.wait.set(mode != Mode.READ_FIRST); + readAction.execute(parentNode); + } + catch (Exception e) + { + e.printStackTrace(); + ex.set(e); + } + finally + { + if (mode == Mode.READ_FIRST) goSignal.countDown(); + doneSignal.countDown(); + } + } + }; + reader.start(); + startSignal.countDown(); + doneSignal.await(); + assertNull(ex.get()); + } + + private abstract class Action + { + + protected final CacheableWorkspaceDataManager cwdm; + public Action(CacheableWorkspaceDataManager cwdm) + { + this.cwdm = cwdm; + } + protected abstract void execute(NodeData parentNode) throws Exception; + } + private static enum Mode + { + READ_FIRST, WRITE_FIRST; + } + private static class MyWorkspaceStorageConnection implements WorkspaceStorageConnection + { + public ThreadLocal<Boolean> wait = new ThreadLocal<Boolean>(); + private NodeData parentNode; + private CountDownLatch goSignal; + + public CountDownLatch initCountDownLatch() + { + return this.goSignal = new CountDownLatch(1); + } + + public void setParentNode(NodeData parentNode) + { + this.parentNode = parentNode; + } + + public void add(NodeData data) throws RepositoryException, UnsupportedOperationException, + InvalidItemStateException, IllegalStateException + { + } + + public void add(PropertyData data) throws RepositoryException, UnsupportedOperationException, + InvalidItemStateException, IllegalStateException + { + } + + public void close() throws IllegalStateException, RepositoryException + { + } + + public void commit() throws IllegalStateException, RepositoryException + { + if (wait.get() != null && wait.get()) + { + try + { + goSignal.await(); + } + catch (InterruptedException e) + { + Thread.currentThread().interrupt(); + } + } + } + + public void delete(NodeData data) throws RepositoryException, UnsupportedOperationException, + InvalidItemStateException, IllegalStateException + { + } + + public void delete(PropertyData data) throws RepositoryException, UnsupportedOperationException, + InvalidItemStateException, IllegalStateException + { + } + + public int getChildNodesCount(NodeData parent) throws RepositoryException + { + return -1; + } + + public List<NodeData> getChildNodesData(NodeData parent) throws RepositoryException, IllegalStateException + { + if (wait.get() != null && wait.get()) + { + try + { + goSignal.await(); + } + catch (InterruptedException e) + { + Thread.currentThread().interrupt(); + } + } + List<NodeData> children = new ArrayList<NodeData>(); + children.add(new PersistedNodeData("id-node2" + parentNode.getIdentifier(), QPath.makeChildPath(parent.getQPath(), new InternalQName(null, "node2")), parent.getIdentifier(), 1, 0, + Constants.NT_UNSTRUCTURED, new InternalQName[0], null)); + return children; + } + + public List<PropertyData> getChildPropertiesData(NodeData parent) throws RepositoryException, + IllegalStateException + { + if (wait.get() != null && wait.get()) + { + try + { + goSignal.await(); + } + catch (InterruptedException e) + { + Thread.currentThread().interrupt(); + } + } + List<PropertyData> children = new ArrayList<PropertyData>(); + try + { + children.add(new PersistedPropertyData("id-property2" + parentNode.getIdentifier(), QPath.makeChildPath( + parentNode.getQPath(), new InternalQName(null, "property2")), parentNode.getIdentifier(), 0, + PropertyType.STRING, false, Arrays.asList((ValueData)new ByteArrayPersistedValueData(0, "some data".getBytes("UTF-8"))))); + } + catch (UnsupportedEncodingException e) + { + e.printStackTrace(); + } + return children; + } + + /** + * {@inheritDoc} + */ + public ItemData getItemData(NodeData parentData, QPathEntry name) throws RepositoryException, + IllegalStateException + { + return getItemData(parentData, name, ItemType.UNKNOWN); + } + + public ItemData getItemData(NodeData parentData, QPathEntry name, ItemType itemType) throws RepositoryException, + IllegalStateException + { + if (wait.get() != null && wait.get()) + { + try + { + goSignal.await(); + } + catch (InterruptedException e) + { + Thread.currentThread().interrupt(); + } + } + if (itemType == ItemType.NODE) + { + return new PersistedNodeData("my-node" + parentNode.getIdentifier(), QPath.makeChildPath(parentNode.getQPath(), name), Constants.ROOT_UUID, 1, 1, + Constants.NT_UNSTRUCTURED, new InternalQName[0], null); + } + try + { + return new PersistedPropertyData("property-by-path" + + parentNode.getIdentifier(), QPath.makeChildPath(parentNode.getQPath(), name), parentNode + .getIdentifier(), 1, PropertyType.STRING, false, Arrays + .asList((ValueData)new ByteArrayPersistedValueData(0, "some new data".getBytes("UTF-8")))); + } + catch (UnsupportedEncodingException e) + { + e.printStackTrace(); + } + return null; + } + + public ItemData getItemData(String identifier) throws RepositoryException, IllegalStateException + { + if (wait.get() != null && wait.get()) + { + try + { + goSignal.await(); + } + catch (InterruptedException e) + { + Thread.currentThread().interrupt(); + } + } + return parentNode; + } + + public List<PropertyData> getReferencesData(String nodeIdentifier) throws RepositoryException, + IllegalStateException, UnsupportedOperationException + { + if (wait.get() != null && wait.get()) + { + try + { + goSignal.await(); + } + catch (InterruptedException e) + { + Thread.currentThread().interrupt(); + } + } + List<PropertyData> children = new ArrayList<PropertyData>(); + try + { + children.add(new PersistedPropertyData("id-reference2" + parentNode.getIdentifier(), QPath.makeChildPath( + parentNode.getQPath(), new InternalQName(null, "reference2")), parentNode.getIdentifier(), 0, + PropertyType.REFERENCE, false, Arrays.asList((ValueData)new ByteArrayPersistedValueData(0, parentNode.getIdentifier().getBytes("UTF-8"))))); + } + catch (UnsupportedEncodingException e) + { + e.printStackTrace(); + } + return children; + } + + public boolean isOpened() + { + return true; + } + + public List<PropertyData> listChildPropertiesData(NodeData parent) throws RepositoryException, + IllegalStateException + { + return null; + } + + public void rename(NodeData data) throws RepositoryException, UnsupportedOperationException, + InvalidItemStateException, IllegalStateException + { + } + + public void rollback() throws IllegalStateException, RepositoryException + { + } + + public void update(NodeData data) throws RepositoryException, UnsupportedOperationException, + InvalidItemStateException, IllegalStateException + { + } + + public void update(PropertyData data) throws RepositoryException, UnsupportedOperationException, + InvalidItemStateException, IllegalStateException + { + } + + /** + * @see org.exoplatform.services.jcr.storage.WorkspaceStorageConnection#getLastOrderNumber(org.exoplatform.services.jcr.datamodel.NodeData) + */ + public int getLastOrderNumber(NodeData parent) throws RepositoryException + { + return 0; + } + }; + + private static class MyWorkspaceDataContainer extends WorkspaceDataContainerBase + { + + private WorkspaceStorageConnection con; + + public MyWorkspaceDataContainer(WorkspaceStorageConnection con) + { + this.con = con; + } + + public boolean isCheckSNSNewConnection() + { + return false; + } + + public boolean isSame(WorkspaceDataContainer another) + { + return false; + } + + public WorkspaceStorageConnection openConnection() throws RepositoryException + { + return con; + } + + public WorkspaceStorageConnection openConnection(boolean readOnly) throws RepositoryException + { + return con; + } + + public WorkspaceStorageConnection reuseConnection(WorkspaceStorageConnection original) throws RepositoryException + { + return con; + } + + public String getInfo() + { + return "MyWorkspaceDataContainer"; + } + + public String getName() + { + return "MyWorkspaceDataContainer"; + } + + public String getStorageVersion() + { + return "0"; + } + + /** + * @see org.exoplatform.services.jcr.storage.DataContainer#getUniqueName() + */ + public String getUniqueName() + { + return "MyWorkspaceDataContainer"; + } + }; +} Added: jcr/trunk/exo.jcr.component.core/src/test/java/org/exoplatform/services/jcr/impl/dataflow/persistent/infinispan/TestISPNCacheWorkspaceStorageCacheInClusterMode.java =================================================================== --- jcr/trunk/exo.jcr.component.core/src/test/java/org/exoplatform/services/jcr/impl/dataflow/persistent/infinispan/TestISPNCacheWorkspaceStorageCacheInClusterMode.java (rev 0) +++ jcr/trunk/exo.jcr.component.core/src/test/java/org/exoplatform/services/jcr/impl/dataflow/persistent/infinispan/TestISPNCacheWorkspaceStorageCacheInClusterMode.java 2011-04-15 19:11:33 UTC (rev 4240) @@ -0,0 +1,58 @@ +/* + * Copyright (C) 2011 eXo Platform SAS. + * + * 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.exoplatform.services.jcr.impl.dataflow.persistent.infinispan; + +import org.exoplatform.container.configuration.ConfigurationManagerImpl; +import org.exoplatform.services.jcr.config.CacheEntry; +import org.exoplatform.services.jcr.config.SimpleParameterEntry; +import org.exoplatform.services.jcr.config.WorkspaceEntry; +import org.exoplatform.services.jcr.impl.dataflow.persistent.TestWorkspaceStorageCacheInClusterMode; +import org.exoplatform.services.jcr.infinispan.ISPNCacheFactory; + +import java.util.ArrayList; + +/** + * @author <a href="mailto:nfilotto@exoplatform.com">Nicolas Filotto</a> + * @version $Id$ + * + */ +public class TestISPNCacheWorkspaceStorageCacheInClusterMode extends TestWorkspaceStorageCacheInClusterMode<ISPNCacheWorkspaceStorageCache> +{ + + public ISPNCacheWorkspaceStorageCache getCacheImpl() throws Exception + { + ArrayList<SimpleParameterEntry> list = new ArrayList<SimpleParameterEntry>(); + list.add(new SimpleParameterEntry(ISPNCacheFactory.INFINISPAN_CONFIG, + "jar:/conf/standalone/cluster/test-infinispan-config.xml")); + list.add(new SimpleParameterEntry("infinispan-cluster-name", "TestISPNCacheWorkspaceStorageCacheInClusterMode")); + list.add(new SimpleParameterEntry("jgroups-configuration", "jar:/conf/standalone/cluster/udp-mux.xml")); + + CacheEntry entry = new CacheEntry(list); + entry.setEnabled(true); + WorkspaceEntry workspaceEntry = new WorkspaceEntry(); + workspaceEntry.setCache(entry); + workspaceEntry.setUniqueName("MyWorkspace"); + return new ISPNCacheWorkspaceStorageCache(workspaceEntry, new ConfigurationManagerImpl()); + } + + protected void finalize(ISPNCacheWorkspaceStorageCache cache) + { + cache.cache.stop(); + } +} \ No newline at end of file Added: jcr/trunk/exo.jcr.component.core/src/test/java/org/exoplatform/services/jcr/impl/dataflow/persistent/jbosscache/TestJBossCacheWorkspaceStorageCacheInClusterMode.java =================================================================== --- jcr/trunk/exo.jcr.component.core/src/test/java/org/exoplatform/services/jcr/impl/dataflow/persistent/jbosscache/TestJBossCacheWorkspaceStorageCacheInClusterMode.java (rev 0) +++ jcr/trunk/exo.jcr.component.core/src/test/java/org/exoplatform/services/jcr/impl/dataflow/persistent/jbosscache/TestJBossCacheWorkspaceStorageCacheInClusterMode.java 2011-04-15 19:11:33 UTC (rev 4240) @@ -0,0 +1,62 @@ +/* + * Copyright (C) 2011 eXo Platform SAS. + * + * 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.exoplatform.services.jcr.impl.dataflow.persistent.jbosscache; + +import org.exoplatform.container.configuration.ConfigurationManagerImpl; +import org.exoplatform.services.jcr.config.CacheEntry; +import org.exoplatform.services.jcr.config.SimpleParameterEntry; +import org.exoplatform.services.jcr.config.WorkspaceEntry; +import org.exoplatform.services.jcr.impl.dataflow.persistent.TestWorkspaceStorageCacheInClusterMode; +import org.exoplatform.services.transaction.TransactionService; + +import java.util.ArrayList; + +/** + * @author <a href="mailto:nicolas.filotto@exoplatform.com">Nicolas Filotto</a> + * @version $Id$ + */ +public class TestJBossCacheWorkspaceStorageCacheInClusterMode extends TestWorkspaceStorageCacheInClusterMode<JBossCacheWorkspaceStorageCache> +{ + + public JBossCacheWorkspaceStorageCache getCacheImpl() throws Exception + { + TransactionService transactionService = + (TransactionService)container.getComponentInstanceOfType(TransactionService.class); + + ArrayList<SimpleParameterEntry> list = new ArrayList<SimpleParameterEntry>(); + list.add(new SimpleParameterEntry(JBossCacheWorkspaceStorageCache.JBOSSCACHE_CONFIG, + "jar:/conf/standalone/cluster/test-jbosscache-data-no-mux.xml")); + list.add(new SimpleParameterEntry("jbosscache-cluster-name", "TestJBossCacheWorkspaceStorageCacheInClusterMode")); + + CacheEntry entry = new CacheEntry(list); + entry.setEnabled(true); + WorkspaceEntry workspaceEntry = new WorkspaceEntry(); + workspaceEntry.setCache(entry); + workspaceEntry.setUniqueName("MyWorkspace"); + JBossCacheWorkspaceStorageCache cache = new JBossCacheWorkspaceStorageCache(workspaceEntry, + transactionService == null ? null : transactionService, new ConfigurationManagerImpl()); + cache.start(); + return cache; + } + + protected void finalize(JBossCacheWorkspaceStorageCache cache) + { + cache.cache.stop(); + } +} \ No newline at end of file Added: jcr/trunk/exo.jcr.component.core/src/test/resources/conf/standalone/cluster/test-jbosscache-data-no-mux.xml =================================================================== --- jcr/trunk/exo.jcr.component.core/src/test/resources/conf/standalone/cluster/test-jbosscache-data-no-mux.xml (rev 0) +++ jcr/trunk/exo.jcr.component.core/src/test/resources/conf/standalone/cluster/test-jbosscache-data-no-mux.xml 2011-04-15 19:11:33 UTC (rev 4240) @@ -0,0 +1,24 @@ +<?xml version="1.0" encoding="UTF-8"?> +<jbosscache xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns="urn:jboss:jbosscache-core:config:3.1"> + + <locking useLockStriping="false" concurrencyLevel="500" lockParentForChildInsertRemove="false" + lockAcquisitionTimeout="20000" /> + <transaction syncRollbackPhase="true" syncCommitPhase="true" /> + + <clustering mode="replication" clusterName="${jbosscache-cluster-name}"> + <stateRetrieval timeout="20000" fetchInMemoryState="false" /> + <sync replTimeout="20000"/> + <jgroupsConfig configFile="udp.xml"> + </jgroupsConfig> + </clustering> + + <!-- Eviction configuration --> + <eviction wakeUpInterval="5000"> + <default algorithmClass="org.jboss.cache.eviction.ExpirationAlgorithm" + actionPolicyClass="org.exoplatform.services.jcr.impl.dataflow.persistent.jbosscache.ParentNodeEvictionActionPolicy" + eventQueueSize="1000000"> + <property name="maxNodes" value="1000000" /> + <property name="warnNoExpirationKey" value="false" /> + </default> + </eviction> +</jbosscache> \ No newline at end of file