Author: mircea.markus Date: 2009-03-09 18:10:53 -0400 (Mon, 09 Mar 2009) New Revision: 7893 Added: core/branches/flat/src/test/java/org/horizon/atomic/AtomicHashMapConcurrencyTest.java Modified: core/branches/flat/src/main/java/org/horizon/tree/TreeCacheImpl.java core/branches/flat/src/main/java/org/horizon/tree/TreeStructureSupport.java core/branches/flat/src/test/java/org/horizon/api/tree/NodeMoveAPITest.java Log: fixed UT Modified: core/branches/flat/src/main/java/org/horizon/tree/TreeCacheImpl.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/tree/TreeCacheImpl.java 2009-03-09 15:02:01 UTC (rev 7892) +++ core/branches/flat/src/main/java/org/horizon/tree/TreeCacheImpl.java 2009-03-09 22:10:53 UTC (rev 7893) @@ -118,13 +118,17 @@ public boolean removeNode(Fqn fqn) { if (fqn.isRoot()) return false; startAtomic(); + boolean result; try { + if (trace) log.trace("About to remove node " + fqn); Node<K, V> n = getNode(fqn.getParent()); - return n != null && n.removeChild(fqn.getLastElement()); + result = n != null && n.removeChild(fqn.getLastElement()); } finally { endAtomic(); } + if (trace) log.trace("Node successfully removed"); + return result; } public boolean removeNode(Fqn fqn, Options... options) { @@ -202,10 +206,12 @@ return get(fqn, key); } - public void move(Fqn nodeToMove, Fqn newParent) throws NodeNotExistsException { - if (nodeToMove == null || newParent == null) throw new NullPointerException("Cannot accept null parameters!"); + public void move(Fqn nodeToMoveFqn, Fqn newParentFqn) throws NodeNotExistsException { + if (trace) log.trace("Moving node '" + nodeToMoveFqn + "' to '" + newParentFqn + "'"); + if (nodeToMoveFqn == null || newParentFqn == null) throw new NullPointerException("Cannot accept null parameters!"); - if (nodeToMove.getParent().equals(newParent)) { + if (nodeToMoveFqn.getParent().equals(newParentFqn)) { + if (trace) log.trace("Not doing anything as this node is equal with its parent"); // moving onto self! Do nothing! return; } @@ -213,30 +219,35 @@ // Depth first. Lets start with getting the node we want. startAtomic(); try { - Node node = getNode(nodeToMove); - if (node == null) return; // nothing to do here! - if (!exists(newParent)) { + Node nodeToMove = getNode(nodeToMoveFqn, Options.FORCE_WRITE_LOCK); + if (nodeToMove == null) { + if (trace) log.trace("Did not find the node that needs to be moved. Returning..."); + return; // nothing to do here! + } + if (!exists(newParentFqn)) { // then we need to silently create the new parent - createNodeInCache(newParent); + createNodeInCache(newParentFqn); + if (trace) log.trace("The new parent ("+newParentFqn +") did not exists, was created"); } // create an empty node for this new parent - Fqn newFqn = Fqn.fromRelativeElements(newParent, nodeToMove.getLastElement()); + Fqn newFqn = Fqn.fromRelativeElements(newParentFqn, nodeToMoveFqn.getLastElement()); createNodeInCache(newFqn); Node newNode = getNode(newFqn); - Map oldData = node.getData(); + Map oldData = nodeToMove.getData(); if (oldData != null && !oldData.isEmpty()) newNode.putAll(oldData); - for (Object child : node.getChildrenNames()) { + for (Object child : nodeToMove.getChildrenNames()) { // move kids - Fqn oldChildFqn = Fqn.fromRelativeElements(nodeToMove, child); + if (trace) log.trace("Moving child " + child); + Fqn oldChildFqn = Fqn.fromRelativeElements(nodeToMoveFqn, child); move(oldChildFqn, newFqn); } - - removeNode(nodeToMove); + removeNode(nodeToMoveFqn); } finally { endAtomic(); } + log.trace("Successfully moved node '" + nodeToMoveFqn + "' to '" + newParentFqn + "'"); } public void move(Fqn nodeToMove, Fqn newParent, Options... options) throws NodeNotExistsException { Modified: core/branches/flat/src/main/java/org/horizon/tree/TreeStructureSupport.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/tree/TreeStructureSupport.java 2009-03-09 15:02:01 UTC (rev 7892) +++ core/branches/flat/src/main/java/org/horizon/tree/TreeStructureSupport.java 2009-03-09 22:10:53 UTC (rev 7893) @@ -29,8 +29,12 @@ import org.horizon.invocation.InvocationContextContainer; import org.horizon.invocation.Options; import org.horizon.lock.LockManager; +import org.horizon.logging.Log; +import org.horizon.logging.LogFactory; public class TreeStructureSupport extends AutoBatchSupport { + private static Log log = LogFactory.getLog(TreeStructureSupport.class); + AtomicMapCache cache; InvocationContextContainer icc; @@ -70,6 +74,7 @@ } cache.getAtomicMap(structureKey); cache.getAtomicMap(dataKey); + if (log.isTraceEnabled()) log.trace("Created node " + fqn); return true; } finally { Modified: core/branches/flat/src/test/java/org/horizon/api/tree/NodeMoveAPITest.java =================================================================== --- core/branches/flat/src/test/java/org/horizon/api/tree/NodeMoveAPITest.java 2009-03-09 15:02:01 UTC (rev 7892) +++ core/branches/flat/src/test/java/org/horizon/api/tree/NodeMoveAPITest.java 2009-03-09 22:10:53 UTC (rev 7893) @@ -2,6 +2,7 @@ import org.horizon.api.mvcc.LockAssert; import org.horizon.config.Configuration; +import org.horizon.container.DataContainer; import org.horizon.factories.ComponentRegistry; import org.horizon.invocation.InvocationContextContainer; import org.horizon.lock.LockManager; @@ -12,8 +13,6 @@ import org.horizon.test.TestingUtil; import org.horizon.tree.Fqn; import org.horizon.tree.Node; -import org.horizon.tree.NodeNotExistsException; -import org.horizon.tree.TreeCache; import org.horizon.tree.TreeCacheImpl; import org.horizon.tree.TreeStructureSupport; import static org.testng.AssertJUnit.*; @@ -42,8 +41,9 @@ static final Fqn D_B_C = Fqn.fromRelativeFqn(D_B, C); protected static final Object k = "key", vA = "valueA", vB = "valueB", vC = "valueC", vD = "valueD", vE = "valueE"; - TreeCache<Object, Object> treeCache; + TreeCacheImpl<Object, Object> treeCache; TransactionManager tm; + DataContainer dc; protected CacheManager createCacheManager() throws Exception { CacheManager cm = TestingUtil.createLocalCacheManager(); @@ -55,6 +55,7 @@ cache = cm.getCache("test"); tm = TestingUtil.extractComponent(cache, TransactionManager.class); treeCache = new TreeCacheImpl(cache); + dc = TestingUtil.extractComponent(cache, DataContainer.class); return cm; } @@ -329,19 +330,9 @@ for (int counter = 0; counter < loops; counter++) { - try { - treeCache.move(NODE_X.getFqn(), NODES[rnd.nextInt(NODES.length)].getFqn()); - } - catch (NodeNotExistsException e) { - // this may happen ... - } + treeCache.move(NODE_X.getFqn(), NODES[rnd.nextInt(NODES.length)].getFqn()); TestingUtil.sleepRandom(250); - try { - treeCache.move(NODE_Y.getFqn(), NODES[rnd.nextInt(NODES.length)].getFqn()); - } - catch (NodeNotExistsException e) { - // this may happen ... - } + treeCache.move(NODE_Y.getFqn(), NODES[rnd.nextInt(NODES.length)].getFqn()); TestingUtil.sleepRandom(250); } } Added: core/branches/flat/src/test/java/org/horizon/atomic/AtomicHashMapConcurrencyTest.java =================================================================== --- core/branches/flat/src/test/java/org/horizon/atomic/AtomicHashMapConcurrencyTest.java (rev 0) +++ core/branches/flat/src/test/java/org/horizon/atomic/AtomicHashMapConcurrencyTest.java 2009-03-09 22:10:53 UTC (rev 7893) @@ -0,0 +1,140 @@ +package org.horizon.atomic; + +import org.horizon.config.Configuration; +import org.horizon.lock.TimeoutException; +import org.horizon.manager.CacheManager; +import org.horizon.manager.DefaultCacheManager; +import org.horizon.test.TestingUtil; +import org.horizon.transaction.DummyTransactionManagerLookup; +import org.testng.annotations.AfterMethod; +import org.testng.annotations.BeforeMethod; +import org.testng.annotations.Test; + +import javax.transaction.TransactionManager; +import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.BlockingQueue; + +/** + * Tester class for AtomicMapCache. + * + * @author Mircea.Markus(a)jboss.com + */ +@Test(groups = "functional", testName = "atomic.AtomicHashMapConcurrencyTest") +public class AtomicHashMapConcurrencyTest { + + public static final String KEY = "key"; + AtomicMapCache<String, String> cache; + TransactionManager tm; + + enum Operation { + PUT, + COMMIT, + READ + } + + @BeforeMethod + @SuppressWarnings("unchecked") + public void setUp() { + Configuration c = new Configuration(); + c.setLockAcquisitionTimeout(500); + // these 2 need to be set to use the AtomicMapCache + c.setTransactionManagerLookupClass(DummyTransactionManagerLookup.class.getName()); + c.setInvocationBatchingEnabled(true); + CacheManager cm = new DefaultCacheManager(c); + cache = (AtomicMapCache<String, String>) cm.getCache(); + tm = TestingUtil.getTransactionManager(cache); + } + + @AfterMethod + public void tearDown() { + try { + tm.rollback(); + } catch (Exception e) { + } + } + + public void testConcurrentCreate() throws Exception { + tm.begin(); + AtomicMap<Integer, String> atomicMap = cache.getAtomicMap(KEY, Integer.class, String.class); + OtherThread ot = new OtherThread(); + ot.start(); + Object response = ot.response.take(); + assert response instanceof TimeoutException; + } + + public void testConcurrentModifications() throws Exception { + AtomicMap<Integer, String> atomicMap = cache.getAtomicMap(KEY, Integer.class, String.class); + tm.begin(); + atomicMap.put(1,""); + OtherThread ot = new OtherThread(); + ot.start(); + ot.toExecute.put(Operation.PUT); + Object response = ot.response.take(); + assert response instanceof TimeoutException; + } + + public void testReadAfterTxStarted() throws Exception { + AtomicMap<Integer, String> atomicMap = cache.getAtomicMap(KEY, Integer.class, String.class); + atomicMap.put(1, "existing"); + tm.begin(); + atomicMap.put(1,"newVal"); + OtherThread ot = new OtherThread(); + ot.start(); + ot.toExecute.put(Operation.READ); + Object response = ot.response.take(); + assert response.equals("existing"); + tm.commit(); + assert atomicMap.get(1).equals("newVal"); + ot.toExecute.put(Operation.READ); + response = ot.response.take(); + assert response.equals("newVal"); + } + + public class OtherThread extends Thread { + + public OtherThread() { + super("OtherThread"); + } + + BlockingQueue response = new ArrayBlockingQueue(1); + + BlockingQueue<Operation> toExecute = new ArrayBlockingQueue<Operation>(1); + + @Override + public void run() { + try { + tm.begin(); + AtomicMap<Integer, String> atomicMap = cache.getAtomicMap(KEY, Integer.class, String.class); + boolean notCommited = true; + while (notCommited) { + Operation op = toExecute.take(); + switch (op) { + case PUT: { + atomicMap.put(1, "val"); + response.put(new Object()); + break; + } + case READ: { + String val = atomicMap.get(1); + response.put(String.valueOf(val)); + break; + } + case COMMIT: { + tm.commit(); + response.put(new Object()); + notCommited = false; + break; + } + } + } + } catch (Exception e) { + try { + response.put(e); + } catch (InterruptedException e1) { + e1.printStackTrace(); + } + e.printStackTrace(); + } + } + } +} Property changes on: core/branches/flat/src/test/java/org/horizon/atomic/AtomicHashMapConcurrencyTest.java ___________________________________________________________________ Name: svn:keywords + Id Revision Name: svn:eol-style + LF