Author: manik.surtani(a)jboss.com Date: 2008-06-30 15:07:05 -0400 (Mon, 30 Jun 2008) New Revision: 6129 Added: core/trunk/src/test/java/org/jboss/cache/api/mvcc/repeatable_read/RepeatableReadTestBase.java Modified: core/trunk/src/main/java/org/jboss/cache/DataContainer.java core/trunk/src/main/java/org/jboss/cache/DataContainerImpl.java core/trunk/src/main/java/org/jboss/cache/NodeFactory.java core/trunk/src/main/java/org/jboss/cache/UnversionedNode.java core/trunk/src/main/java/org/jboss/cache/commands/write/EvictCommand.java core/trunk/src/main/java/org/jboss/cache/factories/CommandsFactoryImpl.java core/trunk/src/main/java/org/jboss/cache/factories/PessimisticCommandsFactoryImpl.java core/trunk/src/main/java/org/jboss/cache/interceptors/InvocationContextInterceptor.java core/trunk/src/main/java/org/jboss/cache/interceptors/MVCCLockingInterceptor.java core/trunk/src/main/java/org/jboss/cache/mvcc/ReadCommittedNode.java core/trunk/src/main/java/org/jboss/cache/mvcc/RepeatableReadNode.java core/trunk/src/test/java/org/jboss/cache/api/mvcc/LockTestBase.java core/trunk/src/test/java/org/jboss/cache/api/mvcc/repeatable_read/RepeatableReadNoWriteSkewLockTest.java core/trunk/src/test/java/org/jboss/cache/api/mvcc/repeatable_read/RepeatableReadNoWriteSkewWithParentLockTest.java core/trunk/src/test/java/org/jboss/cache/api/mvcc/repeatable_read/RepeatableReadWriteSkewLockTest.java core/trunk/src/test/java/org/jboss/cache/api/mvcc/repeatable_read/RepeatableReadWriteSkewWithParentLockTest.java core/trunk/src/test/java/org/jboss/cache/commands/write/EvictCommandTest.java Log: More MVCC work Modified: core/trunk/src/main/java/org/jboss/cache/DataContainer.java =================================================================== --- core/trunk/src/main/java/org/jboss/cache/DataContainer.java 2008-06-30 16:49:02 UTC (rev 6128) +++ core/trunk/src/main/java/org/jboss/cache/DataContainer.java 2008-06-30 19:07:05 UTC (rev 6129) @@ -144,7 +144,7 @@ /** * Evicts the given node. If recursive is set to true then all child nodes are recusively evicted. */ - void evict(Fqn fqn, boolean recursive); + void evict(Fqn fqn, boolean recursive); // TODO: See if this is still needed here /** * <pre> @@ -157,7 +157,7 @@ * @return true if the FQN is leaf and was removed; false if is an intermediate FQN and only contained data * is droped. */ - boolean evict(Fqn fqn); + boolean evict(Fqn fqn); // TODO: See if this is still needed here /** * Traverses the tree to the given Fqn, creating nodes if needed. Returns a list of nodes created, as well as a reference to the last node. @@ -177,10 +177,11 @@ /** * Similar to {@link #peek(Fqn)} except that the underlying node is NOT wrapped as a {@link org.jboss.cache.NodeSPI}. * - * @param f fqn to peek + * @param f fqn to peek + * @param includeInvalidNodes if true, invalid nodes will be considered as well. * @return internal node */ - InternalNode peekInternalNode(Fqn f); + InternalNode peekInternalNode(Fqn f, boolean includeInvalidNodes); /** * Sets a new root node Modified: core/trunk/src/main/java/org/jboss/cache/DataContainerImpl.java =================================================================== --- core/trunk/src/main/java/org/jboss/cache/DataContainerImpl.java 2008-06-30 16:49:02 UTC (rev 6128) +++ core/trunk/src/main/java/org/jboss/cache/DataContainerImpl.java 2008-06-30 19:07:05 UTC (rev 6129) @@ -424,7 +424,7 @@ } } - protected void removeData(Fqn fqn) + private void removeData(Fqn fqn) { NodeSPI n = peek(fqn); if (n == null) @@ -465,10 +465,10 @@ return new Object[]{result, n}; } - public InternalNode peekInternalNode(Fqn f) + public InternalNode peekInternalNode(Fqn f, boolean includeInvalidNodes) { // Yuck! - NodeSPI nodeSPI = peek(f); + NodeSPI nodeSPI = peek(f, false, includeInvalidNodes); if (nodeSPI == null) return null; return (InternalNode) ((NodeInvocationDelegate) nodeSPI).getDelegationTarget(); } Modified: core/trunk/src/main/java/org/jboss/cache/NodeFactory.java =================================================================== --- core/trunk/src/main/java/org/jboss/cache/NodeFactory.java 2008-06-30 16:49:02 UTC (rev 6128) +++ core/trunk/src/main/java/org/jboss/cache/NodeFactory.java 2008-06-30 19:07:05 UTC (rev 6129) @@ -15,8 +15,11 @@ import org.jboss.cache.interceptors.InterceptorChain; import org.jboss.cache.invocation.InvocationContextContainer; import org.jboss.cache.invocation.NodeInvocationDelegate; +import org.jboss.cache.lock.IsolationLevel; import org.jboss.cache.lock.LockStrategyFactory; import org.jboss.cache.mvcc.InternalNode; +import org.jboss.cache.mvcc.ReadCommittedNode; +import org.jboss.cache.mvcc.RepeatableReadNode; import org.jboss.cache.optimistic.TransactionWorkspace; import org.jboss.cache.optimistic.WorkspaceNode; import org.jboss.cache.optimistic.WorkspaceNodeImpl; @@ -37,6 +40,7 @@ private InterceptorChain interceptorChain; private CommandsFactory commandsFactory; private LockStrategyFactory lockStrategyFactory; + private boolean useRepeatableRead; @Override protected <T> T construct(Class<T> componentType) @@ -44,6 +48,21 @@ throw new UnsupportedOperationException("Should never be called!"); } + /** + * Creates an MVCC wrapped node - either a {@link org.jboss.cache.mvcc.ReadCommittedNode} or it's subclass, a + * {@link org.jboss.cache.mvcc.RepeatableReadNode} based on cache configuration. + * + * @param node internal node to wrap. + * @return a ReadCommittedNode + */ + public ReadCommittedNode createMvccNode(InternalNode node) + { + ReadCommittedNode rcn = useRepeatableRead ? new RepeatableReadNode(node) : new ReadCommittedNode(node); + rcn.initialize(configuration, invocationContextContainer, componentRegistry, interceptorChain); + rcn.injectDependencies(cache); + return rcn; + } + public enum NodeType { UNVERSIONED_NODE, VERSIONED_NODE, WORKSPACE_NODE @@ -82,6 +101,7 @@ public void init() { useVersionedNode = configuration.getNodeLockingScheme() != NodeLockingScheme.PESSIMISTIC; + useRepeatableRead = configuration.getIsolationLevel() == IsolationLevel.REPEATABLE_READ; } Modified: core/trunk/src/main/java/org/jboss/cache/UnversionedNode.java =================================================================== --- core/trunk/src/main/java/org/jboss/cache/UnversionedNode.java 2008-06-30 16:49:02 UTC (rev 6128) +++ core/trunk/src/main/java/org/jboss/cache/UnversionedNode.java 2008-06-30 19:07:05 UTC (rev 6129) @@ -525,8 +525,13 @@ public void setChildrenMapDirect(Map<Object, Node<K, V>> children) { - this.children().clear(); - this.children.putAll(children); + if (children == null) + this.children = null; + else + { + this.children().clear(); + this.children.putAll(children); + } } public void putAll(Map data) @@ -552,7 +557,8 @@ // versioning - public void setVersion(DataVersion version) + public void setVersion(DataVersion + version) { throw new UnsupportedOperationException("Versioning not supported"); } Modified: core/trunk/src/main/java/org/jboss/cache/commands/write/EvictCommand.java =================================================================== --- core/trunk/src/main/java/org/jboss/cache/commands/write/EvictCommand.java 2008-06-30 16:49:02 UTC (rev 6128) +++ core/trunk/src/main/java/org/jboss/cache/commands/write/EvictCommand.java 2008-06-30 19:07:05 UTC (rev 6129) @@ -1,5 +1,7 @@ package org.jboss.cache.commands.write; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; import org.jboss.cache.DataContainer; import org.jboss.cache.Fqn; import org.jboss.cache.NodeSPI; @@ -8,6 +10,8 @@ import org.jboss.cache.invocation.InvocationContext; import org.jboss.cache.notifications.Notifier; +import java.util.Collection; +import java.util.Collections; import java.util.List; /** @@ -24,6 +28,10 @@ private boolean recursive = false; protected Notifier notifier; + private static final Log log = LogFactory.getLog(EvictCommand.class); + private static final boolean trace = log.isTraceEnabled(); + private boolean isUsingMvcc; + private List<Fqn> nodesToEvict; public EvictCommand(Fqn fqn) { @@ -34,12 +42,23 @@ { } - public void initialize(Notifier notifier, DataContainer dataContainer) + public void initialize(Notifier notifier, DataContainer dataContainer, boolean isUsingMvcc) { super.initialize(dataContainer); this.notifier = notifier; + this.isUsingMvcc = isUsingMvcc; } + public List<Fqn> getNodesToEvict() + { + return nodesToEvict; + } + + public void setNodesToEvict(List<Fqn> nodesToEvict) + { + this.nodesToEvict = nodesToEvict; + } + /** * Evicts a node. * <p/> @@ -50,9 +69,8 @@ */ public Object perform(InvocationContext ctx) { - //NodeSPI node = ctx.lookUpNode(fqn); - // TODO: MVCC - eviction actually needs to look at invalid nodes (tombstones) as well. - NodeSPI node = dataContainer.peek(fqn, false, true); + // TODO: remove this ugly hack + NodeSPI node = lookupForEviction(ctx, fqn); if (node == null) { return false; @@ -63,25 +81,61 @@ } else if (recursive) { - List<Fqn> nodesToEvict = dataContainer.getNodesForEviction(fqn, true); - for (Fqn aFqn : nodesToEvict) + Collection<Fqn> nodesToEvict = getRecursiveEvictionNodes(); + + if (nodesToEvict != null) { - evictNode(aFqn, ctx); + for (Fqn aFqn : nodesToEvict) + { + evictNode(aFqn, ctx, lookupForEviction(ctx, aFqn)); + } } return !nodesToEvict.isEmpty(); } else { - return evictNode(fqn, ctx); + return evictNode(fqn, ctx, node); } } - boolean evictNode(Fqn fqn, InvocationContext ctx) + private Collection<Fqn> getRecursiveEvictionNodes() { + if (isUsingMvcc) + { + Collections.sort(nodesToEvict); + Collections.reverse(nodesToEvict); + return nodesToEvict; + } + else + return dataContainer.getNodesForEviction(fqn, true); + } + + boolean evictNode(Fqn fqn, InvocationContext ctx, NodeSPI node) + { notifier.notifyNodeEvicted(fqn, true, ctx); try { - return dataContainer.evict(fqn); + if (node == null) return true; + if (node.hasChildrenDirect()) + { + if (trace) log.trace("removing DATA as node has children: evict(" + fqn + ")"); + node.clearDataDirect(); + node.setDataLoaded(false); + return false; + } + else + { + if (trace) log.trace("removing NODE as it is a leaf: evict(" + fqn + ")"); + NodeSPI parentNode = lookupForEviction(ctx, fqn.getParent()); + + if (parentNode != null) + { + parentNode.removeChildDirect(fqn.getLastElement()); + parentNode.setChildrenLoaded(false); + } + node.setValid(false, false); + return true; + } } finally { @@ -89,6 +143,19 @@ } } + private NodeSPI lookupForEviction(InvocationContext ctx, Fqn fqn) + { + // TODO: do something about this nasty hack + if (isUsingMvcc) + { + return ctx.lookUpNode(fqn); + } + else + { + return dataContainer.peek(fqn, false, true); + } + } + public Object acceptVisitor(InvocationContext ctx, Visitor visitor) throws Throwable { return visitor.visitEvictFqnCommand(ctx, this); Modified: core/trunk/src/main/java/org/jboss/cache/factories/CommandsFactoryImpl.java =================================================================== --- core/trunk/src/main/java/org/jboss/cache/factories/CommandsFactoryImpl.java 2008-06-30 16:49:02 UTC (rev 6128) +++ core/trunk/src/main/java/org/jboss/cache/factories/CommandsFactoryImpl.java 2008-06-30 19:07:05 UTC (rev 6129) @@ -28,10 +28,21 @@ import org.jboss.cache.commands.tx.OptimisticPrepareCommand; import org.jboss.cache.commands.tx.PrepareCommand; import org.jboss.cache.commands.tx.RollbackCommand; -import org.jboss.cache.commands.write.*; +import org.jboss.cache.commands.write.ClearDataCommand; +import org.jboss.cache.commands.write.CreateNodeCommand; +import org.jboss.cache.commands.write.EvictCommand; +import org.jboss.cache.commands.write.InvalidateCommand; +import org.jboss.cache.commands.write.MoveCommand; +import org.jboss.cache.commands.write.PutDataMapCommand; +import org.jboss.cache.commands.write.PutForExternalReadCommand; +import org.jboss.cache.commands.write.PutKeyValueCommand; +import org.jboss.cache.commands.write.RemoveKeyCommand; +import org.jboss.cache.commands.write.RemoveNodeCommand; +import org.jboss.cache.commands.write.VersionedInvalidateCommand; import org.jboss.cache.config.Configuration; +import org.jboss.cache.config.Configuration.NodeLockingScheme; import org.jboss.cache.factories.annotations.Inject; -import org.jboss.cache.factories.annotations.NonVolatile; +import org.jboss.cache.factories.annotations.Start; import org.jboss.cache.interceptors.InterceptorChain; import org.jboss.cache.notifications.Notifier; import org.jboss.cache.transaction.GlobalTransaction; @@ -46,7 +57,6 @@ /** * This is the implementation to use for most commands and most locking schemes. */ -@NonVolatile public class CommandsFactoryImpl implements CommandsFactory { protected RPCManager rpcManager; @@ -59,6 +69,7 @@ protected Configuration configuration; protected TransactionManager txManager; protected BuddyFqnTransformer buddyFqnTransformer; + private boolean usingMvcc; public CommandsFactoryImpl() { @@ -81,6 +92,12 @@ this.buddyFqnTransformer = buddyFqnTransformer; } + @Start + public void start() + { + usingMvcc = configuration.getNodeLockingScheme() == NodeLockingScheme.MVCC; + } + public PutDataMapCommand buildPutDataMapCommand(GlobalTransaction gtx, Fqn fqn, Map data) { PutDataMapCommand cmd = new PutDataMapCommand(gtx, fqn, data); @@ -148,7 +165,7 @@ public EvictCommand buildEvictFqnCommand(Fqn fqn) { EvictCommand command = new EvictCommand(fqn); - command.initialize(notifier, dataContainer); + command.initialize(notifier, dataContainer, usingMvcc); return command; } Modified: core/trunk/src/main/java/org/jboss/cache/factories/PessimisticCommandsFactoryImpl.java =================================================================== --- core/trunk/src/main/java/org/jboss/cache/factories/PessimisticCommandsFactoryImpl.java 2008-06-30 16:49:02 UTC (rev 6128) +++ core/trunk/src/main/java/org/jboss/cache/factories/PessimisticCommandsFactoryImpl.java 2008-06-30 19:07:05 UTC (rev 6129) @@ -18,7 +18,6 @@ import org.jboss.cache.commands.write.PutKeyValueCommand; import org.jboss.cache.commands.write.RemoveKeyCommand; import org.jboss.cache.commands.write.RemoveNodeCommand; -import org.jboss.cache.factories.annotations.NonVolatile; import org.jboss.cache.transaction.GlobalTransaction; import java.util.Map; @@ -31,7 +30,6 @@ * @see org.jboss.cache.commands.pessimistic.ReversibleCommand * @since 3.0 */ -@NonVolatile public class PessimisticCommandsFactoryImpl extends CommandsFactoryImpl { @Override Modified: core/trunk/src/main/java/org/jboss/cache/interceptors/InvocationContextInterceptor.java =================================================================== --- core/trunk/src/main/java/org/jboss/cache/interceptors/InvocationContextInterceptor.java 2008-06-30 16:49:02 UTC (rev 6128) +++ core/trunk/src/main/java/org/jboss/cache/interceptors/InvocationContextInterceptor.java 2008-06-30 19:07:05 UTC (rev 6129) @@ -176,6 +176,7 @@ } finally { + // TODO: scope upgrading should happen transparently /* * we should scrub txs after every call to prevent race conditions * basically any other call coming in on the same thread and hijacking any running tx's @@ -204,6 +205,9 @@ // reset the context to prevent leakage of internals ctx.setCommand(null); ctx.setMethodCall(null); + + + ctx.reset(); } } Modified: core/trunk/src/main/java/org/jboss/cache/interceptors/MVCCLockingInterceptor.java =================================================================== --- core/trunk/src/main/java/org/jboss/cache/interceptors/MVCCLockingInterceptor.java 2008-06-30 16:49:02 UTC (rev 6128) +++ core/trunk/src/main/java/org/jboss/cache/interceptors/MVCCLockingInterceptor.java 2008-06-30 19:07:05 UTC (rev 6129) @@ -14,22 +14,31 @@ import org.jboss.cache.commands.read.GravitateDataCommand; import org.jboss.cache.commands.tx.CommitCommand; import org.jboss.cache.commands.tx.RollbackCommand; -import org.jboss.cache.commands.write.*; +import org.jboss.cache.commands.write.ClearDataCommand; +import org.jboss.cache.commands.write.CreateNodeCommand; +import org.jboss.cache.commands.write.EvictCommand; +import org.jboss.cache.commands.write.InvalidateCommand; +import org.jboss.cache.commands.write.MoveCommand; +import org.jboss.cache.commands.write.PutDataMapCommand; +import org.jboss.cache.commands.write.PutForExternalReadCommand; +import org.jboss.cache.commands.write.PutKeyValueCommand; +import org.jboss.cache.commands.write.RemoveKeyCommand; +import org.jboss.cache.commands.write.RemoveNodeCommand; import org.jboss.cache.factories.annotations.Inject; import org.jboss.cache.factories.annotations.Start; import org.jboss.cache.interceptors.base.PostProcessingCommandInterceptor; import org.jboss.cache.invocation.InvocationContext; import org.jboss.cache.invocation.NodeInvocationDelegate; -import org.jboss.cache.lock.IsolationLevel; import org.jboss.cache.lock.LockManager; import static org.jboss.cache.lock.LockType.WRITE; import org.jboss.cache.mvcc.InternalNode; import org.jboss.cache.mvcc.ReadCommittedNode; -import org.jboss.cache.mvcc.RepeatableReadNode; import java.util.ArrayList; import java.util.Collections; +import java.util.LinkedList; import java.util.List; +import java.util.Map; /** * Interceptor to implement <a href="http://wiki.jboss.org/wiki/JBossCacheMVCC">MVCC</a> functionality. @@ -48,7 +57,7 @@ // How will wrapper nodes unwrap? - boolean isUsingRepeatableRead, ignoreWriteSkew, lockParentForChildInsertRemove; + boolean allowWriteSkew, lockParentForChildInsertRemove; LockManager lockManager; DataContainer dataContainer; NodeFactory nodeFactory; @@ -64,53 +73,133 @@ @Start public void start() { - isUsingRepeatableRead = configuration.getIsolationLevel() == IsolationLevel.REPEATABLE_READ; // otherwise default to READ_COMMITTED - ignoreWriteSkew = true; // todo this should come from a cfg variable + allowWriteSkew = configuration.isAllowWriteSkew(); lockParentForChildInsertRemove = configuration.isLockParentForChildInsertRemove(); } @Override public Object handlePutDataMapCommand(InvocationContext ctx, PutDataMapCommand command) throws Throwable { - getWrappedNode(ctx, command.getFqn(), true, true); // get the node and stick it in the context. + getWrappedNode(ctx, command.getFqn(), true, true, false); // get the node and stick it in the context. return invokeNextInterceptor(ctx, command); } @Override public Object handlePutKeyValueCommand(InvocationContext ctx, PutKeyValueCommand command) throws Throwable { - getWrappedNode(ctx, command.getFqn(), true, true); // get the node and stick it in the context. + getWrappedNode(ctx, command.getFqn(), true, true, false); // get the node and stick it in the context. return invokeNextInterceptor(ctx, command); } @Override public Object handlePutForExternalReadCommand(InvocationContext ctx, PutForExternalReadCommand command) throws Throwable { - getWrappedNode(ctx, command.getFqn(), true, true); // get the node and stick it in the context. + getWrappedNode(ctx, command.getFqn(), true, true, false); // get the node and stick it in the context. return invokeNextInterceptor(ctx, command); } @Override public Object handleRemoveNodeCommand(InvocationContext ctx, RemoveNodeCommand command) throws Throwable { - // TODO : Handle this properly - return handleWriteCommand(ctx, command, new ArrayList<Fqn>(1), Collections.singletonList(command.getFqn())); + // when removing a node we want to get a lock on the Fqn anyway and return the wrapped node. + Fqn nodeFqn = command.getFqn(); + + if (!nodeFqn.isRoot()) + { + Fqn parentFqn = nodeFqn.getParent(); + // inspect parent + InternalNode parent = dataContainer.peekInternalNode(parentFqn, false); + { + if (lockParentForChildInsertRemove || (parent != null && parent.isLockForChildInsertRemove())) + { + // lock parent + lockManager.lockAndRecord(parentFqn, WRITE, ctx); + + // retrieve again from the dataContainer in case we have a race, and add to the context + getWrappedNode(ctx, parentFqn, true, false, false); + } + } + } + + lockManager.lockAndRecord(nodeFqn, WRITE, ctx); // lock node. + + // now wrap and add to the context + getWrappedNode(ctx, nodeFqn, true, false, false); + + return invokeNextInterceptor(ctx, command); } @Override public Object handleClearDataCommand(InvocationContext ctx, ClearDataCommand command) throws Throwable { - // TODO : Handle this properly - return handleWriteCommand(ctx, command, new ArrayList<Fqn>(1), Collections.singletonList(command.getFqn())); + getWrappedNode(ctx, command.getFqn(), true, false, false); + return invokeNextInterceptor(ctx, command); } @Override public Object handleEvictFqnCommand(InvocationContext ctx, EvictCommand command) throws Throwable { - // TODO : Handle this properly - return handleWriteCommand(ctx, command, new ArrayList<Fqn>(1), Collections.singletonList(command.getFqn())); + // set lock acquisition timeout to 0 - we need to fail fast. + ctx.getOptionOverrides().setLockAcquisitionTimeout(0); + + // similar to remove node, except that we need to take care of the recursive case. + Fqn nodeFqn = command.getFqn(); + + if (!nodeFqn.isRoot()) + { + Fqn parentFqn = nodeFqn.getParent(); + // inspect parent + InternalNode parent = dataContainer.peekInternalNode(parentFqn, true); + { + if (lockParentForChildInsertRemove || (parent != null && parent.isLockForChildInsertRemove())) + { + // lock parent + lockManager.lockAndRecord(parentFqn, WRITE, ctx); + + // retrieve again from the dataContainer in case we have a race, and add to the context + ReadCommittedNode rcn = (ReadCommittedNode) getWrappedNode(ctx, parentFqn, true, false, false); + if (rcn != null) rcn.copyNodeForUpdate(dataContainer, allowWriteSkew); + } + } + } + + List<Fqn> fqnsToEvict = command.isRecursive() ? new LinkedList<Fqn>() : null; + + lockForEviction(nodeFqn, command.isRecursive(), ctx, fqnsToEvict); + + if (fqnsToEvict != null) // add this set to the command + { + command.setNodesToEvict(fqnsToEvict); + } + + return invokeNextInterceptor(ctx, command); } + private void lockForEviction(Fqn fqn, boolean isRecursive, InvocationContext ctx, List<Fqn> fqnsToEvict) throws InterruptedException + { + lockManager.lockAndRecord(fqn, WRITE, ctx); // lock node. + if (fqnsToEvict != null) fqnsToEvict.add(fqn); + + // now wrap and add to the context + ReadCommittedNode rcn = (ReadCommittedNode) getWrappedNode(ctx, fqn, true, false, true); + if (rcn != null) + { + rcn.copyNodeForUpdate(dataContainer, allowWriteSkew); + + if (isRecursive) + { + Map<Object, NodeSPI> children = rcn.getChildrenMapDirect(); + if (children != null) + { + for (NodeSPI child : children.values()) + { + lockForEviction(child.getFqn(), isRecursive, ctx, fqnsToEvict); + } + } + } + } + } + @Override public Object handleInvalidateCommand(InvocationContext ctx, InvalidateCommand command) throws Throwable { @@ -121,8 +210,8 @@ @Override public Object handleRemoveKeyCommand(InvocationContext ctx, RemoveKeyCommand command) throws Throwable { - // TODO : Handle this properly - return handleWriteCommand(ctx, command, new ArrayList<Fqn>(1), Collections.singletonList(command.getFqn())); + getWrappedNode(ctx, command.getFqn(), true, false, false); + return invokeNextInterceptor(ctx, command); } @Override @@ -180,7 +269,7 @@ @Override public Object handleCreateNodeCommand(InvocationContext ctx, CreateNodeCommand command) throws Throwable { - getWrappedNode(ctx, command.getFqn(), true, true); // get the node and stick it in the context. + getWrappedNode(ctx, command.getFqn(), true, true, false); // get the node and stick it in the context. return invokeNextInterceptor(ctx, command); } @@ -232,7 +321,7 @@ { // for each of these, swap refs ReadCommittedNode rcn = (ReadCommittedNode) ctx.lookUpNode(fqnsToUnlock[i]); - rcn.copyNodeForUpdate(dataContainer, ignoreWriteSkew); +// rcn.copyNodeForUpdate(dataContainer, allowWriteSkew); rcn.commitUpdate(dataContainer, nodeFactory); // and then unlock lockManager.unlock(fqnsToUnlock[i], owner); @@ -261,7 +350,7 @@ if (commit) { // for each of these, swap refs - rcn.copyNodeForUpdate(dataContainer, ignoreWriteSkew); +// rcn.copyNodeForUpdate(dataContainer, allowWriteSkew); rcn.commitUpdate(dataContainer, nodeFactory); } else @@ -286,10 +375,10 @@ if (ctx.lookUpNode(f) == null) { // simple implementation. Peek the node, wrap it, put wrapped node in the context. - InternalNode node = dataContainer.peekInternalNode(f); + InternalNode node = dataContainer.peekInternalNode(f, false); if (node != null) { - NodeSPI wrapped = isUsingRepeatableRead ? new RepeatableReadNode(node) : new ReadCommittedNode(node); + NodeSPI wrapped = nodeFactory.createMvccNode(node); ctx.putLookedUpNode(f, wrapped); } } @@ -305,10 +394,10 @@ if (ctx.lookUpNode(f) == null) { // simple implementation. Peek the node, wrap it, put wrapped node in the context. - InternalNode node = dataContainer.peekInternalNode(f); + InternalNode node = dataContainer.peekInternalNode(f, false); if (node != null) { - NodeSPI wrapped = isUsingRepeatableRead ? new RepeatableReadNode(node) : new ReadCommittedNode(node); + NodeSPI wrapped = nodeFactory.createMvccNode(node); ctx.putLookedUpNode(f, wrapped); } } @@ -319,11 +408,11 @@ if (ctx.lookUpNode(f) == null) { // simple implementation. Peek the node, wrap it, put wrapped node in the context. - InternalNode node = dataContainer.peekInternalNode(f); + InternalNode node = dataContainer.peekInternalNode(f, false); if (node != null) { lockManager.lock(f, WRITE, ctx); - NodeSPI wrapped = isUsingRepeatableRead ? new RepeatableReadNode(node) : new ReadCommittedNode(node); + NodeSPI wrapped = nodeFactory.createMvccNode(node); ctx.putLookedUpNode(f, wrapped); } } @@ -339,30 +428,43 @@ * data structure. It will lock the node, and potentially the parent as well, if necessary. If the parent is locked, * it too will be added to the context if it wasn't there already. * - * @param fqn to retrieve - * @param lock if true, a lock will be acquired. - * @param createIfAbsent if true, will be created if absent. + * @param fqn to retrieve + * @param lockForWriting if true, a lock will be acquired. + * @param createIfAbsent if true, will be created if absent. + * @param includeInvalidNodes * @return a NodeSPI or null. */ - protected NodeSPI getWrappedNode(InvocationContext context, Fqn fqn, boolean lock, boolean createIfAbsent) throws InterruptedException + protected NodeSPI getWrappedNode(InvocationContext context, Fqn fqn, boolean lockForWriting, boolean createIfAbsent, boolean includeInvalidNodes) throws InterruptedException { - NodeSPI n = context.lookUpNode(fqn); + ReadCommittedNode n = (ReadCommittedNode) context.lookUpNode(fqn); if (n != null) { // acquire lock if needed - if (lock && !isLocked(context, fqn)) lockManager.lockAndRecord(fqn, WRITE, context); + if (lockForWriting && !isLocked(context, fqn)) + { + lockManager.lockAndRecord(fqn, WRITE, context); + // create a copy of the underlying node + + n.copyNodeForUpdate(dataContainer, allowWriteSkew); + } if (trace) log.trace("Retrieving wrapped node " + fqn); return n; } // else, fetch from dataContainer. - InternalNode in = dataContainer.peekInternalNode(fqn); + InternalNode in = dataContainer.peekInternalNode(fqn, includeInvalidNodes); if (in != null) { // do we need a lock? - if (lock && !isLocked(context, fqn)) lockManager.lockAndRecord(fqn, WRITE, context); - NodeSPI wrapped = isUsingRepeatableRead ? new RepeatableReadNode(in) : new ReadCommittedNode(in); + boolean needToCopy = false; + if (lockForWriting && !isLocked(context, fqn)) + { + lockManager.lockAndRecord(fqn, WRITE, context); + needToCopy = true; + } + ReadCommittedNode wrapped = nodeFactory.createMvccNode(in); context.putLookedUpNode(fqn, wrapped); + if (needToCopy) wrapped.copyNodeForUpdate(dataContainer, allowWriteSkew); return wrapped; } @@ -370,20 +472,26 @@ if (createIfAbsent) { Fqn parentFqn = fqn.getParent(); - NodeSPI parent = getWrappedNode(context, parentFqn, false, createIfAbsent); + NodeSPI parent = getWrappedNode(context, parentFqn, false, createIfAbsent, false); // do we need to lock the parent to create children? if (lockParentForChildInsertRemove || parent.isLockForChildInsertRemove()) { // get a lock on the parent. - if (!isLocked(context, parentFqn)) lockManager.lockAndRecord(parentFqn, WRITE, context); + if (!isLocked(context, parentFqn)) + { + lockManager.lockAndRecord(parentFqn, WRITE, context); + ReadCommittedNode parentRCN = (ReadCommittedNode) context.lookUpNode(parentFqn); + parentRCN.copyNodeForUpdate(dataContainer, allowWriteSkew); + } } // now to lock and create the node. if (!isLocked(context, fqn)) lockManager.lockAndRecord(fqn, WRITE, context); NodeSPI temp = parent.getOrCreateChild(fqn.getLastElement(), context.getGlobalTransaction()); in = (InternalNode) ((NodeInvocationDelegate) temp).getDelegationTarget(); - NodeSPI wrapped = isUsingRepeatableRead ? new RepeatableReadNode(in) : new ReadCommittedNode(in); + ReadCommittedNode wrapped = nodeFactory.createMvccNode(in); context.putLookedUpNode(fqn, wrapped); + wrapped.copyNodeForUpdate(dataContainer, allowWriteSkew); return wrapped; } Modified: core/trunk/src/main/java/org/jboss/cache/mvcc/ReadCommittedNode.java =================================================================== --- core/trunk/src/main/java/org/jboss/cache/mvcc/ReadCommittedNode.java 2008-06-30 16:49:02 UTC (rev 6128) +++ core/trunk/src/main/java/org/jboss/cache/mvcc/ReadCommittedNode.java 2008-06-30 19:07:05 UTC (rev 6129) @@ -47,17 +47,16 @@ // TODO: Deal with removes and moves // TODO: Deal with creating - what if children is null? + log.error("Backup is of type " + backup.getClass().getSimpleName()); + log.error("Node is of type " + node.getClass().getSimpleName()); ((NodeReference) backup).setDelegate(((NodeReference) node).getDelegate()); node = backup; } public void rollbackUpdate() { - if (changed) - { - node = backup; - backup = null; - changed = false; - } + node = backup; + backup = null; + changed = false; } } Modified: core/trunk/src/main/java/org/jboss/cache/mvcc/RepeatableReadNode.java =================================================================== --- core/trunk/src/main/java/org/jboss/cache/mvcc/RepeatableReadNode.java 2008-06-30 16:49:02 UTC (rev 6128) +++ core/trunk/src/main/java/org/jboss/cache/mvcc/RepeatableReadNode.java 2008-06-30 19:07:05 UTC (rev 6129) @@ -1,10 +1,10 @@ package org.jboss.cache.mvcc; -import org.jboss.cache.CacheException; import org.jboss.cache.DataContainer; import org.jboss.cache.NodeFactory; import org.jboss.cache.NodeSPI; import org.jboss.cache.optimistic.DataVersion; +import org.jboss.cache.optimistic.DataVersioningException; import org.jboss.cache.optimistic.DefaultDataVersion; /** @@ -23,21 +23,37 @@ @Override public void copyNodeForUpdate(DataContainer container, boolean ignoreWriteSkew) { + // mark node as changed. changed = true; - backup = node; - // write skew check - DataVersion underlyingNodeVersion = container.peek(getFqn()).getVersion(); - if (ignoreWriteSkew || backup.getVersion().equals(underlyingNodeVersion)) + + // check for write skew. + NodeSPI underlyingNode = container.peek(getFqn(), false, true); // even check for invalid nodes. we should check tombstones too. + DataVersion underlyingNodeVersion = underlyingNode == null ? null : underlyingNode.getVersion(); + if (!ignoreWriteSkew && underlyingNode != null && !node.getVersion().equals(underlyingNodeVersion)) { - node = backup.copy(); - // TODO: Make sure this works with custom versions as well! - DataVersion newVersion = ((DefaultDataVersion) node.getVersion()).increment(); - node.setVersion(newVersion); + throw new DataVersioningException("Detected write skew. Attempting to overwrite version " + node.getVersion() + " but current version has progressed to " + underlyingNodeVersion); } - else - { - throw new CacheException("Detected write skew. Attempting to overwrite version " + backup.getVersion() + " but current version has progressed to " + underlyingNodeVersion); - } + + // make a backup copy + backup = node; + node = backup.copy(); + + // update version on copy + // TODO: Make sure this works with custom versions as well! + DataVersion newVersion = ((DefaultDataVersion) node.getVersion()).increment(); + node.setVersion(newVersion); + + // update parent nodes references - May not be necessary, we could just make sure we don't overwrite child maps when + // updateNode() runs. + +// if (!getFqn().isRoot()) +// { +// RepeatableReadNode parent = (RepeatableReadNode) ctx.lookUpNode(getFqn().getParent()); +// if (parent.changed) +// { +// parent.addChildDirect(); +// } +// } } @Override @@ -51,7 +67,15 @@ else { NodeSPI parent = dataContainer.peek(getFqn().getParent()); - parent.addChildDirect(nf.createNodeInvocationDelegate(node)); + if (parent != null) + { + NodeSPI oldChild = parent.getChildDirect(getFqn().getLastElement()); + if (oldChild != null) + { + node.setChildrenMapDirect(oldChild.getChildrenMapDirect()); + } + parent.addChildDirect(nf.createNodeInvocationDelegate(node)); + } } } } Modified: core/trunk/src/test/java/org/jboss/cache/api/mvcc/LockTestBase.java =================================================================== --- core/trunk/src/test/java/org/jboss/cache/api/mvcc/LockTestBase.java 2008-06-30 16:49:02 UTC (rev 6128) +++ core/trunk/src/test/java/org/jboss/cache/api/mvcc/LockTestBase.java 2008-06-30 19:07:05 UTC (rev 6129) @@ -31,14 +31,14 @@ @Test(groups = {"functional", "mvcc"}) public abstract class LockTestBase { - Cache<String, String> cache; - TransactionManager tm; - Fqn A = Fqn.fromString("/a"); - Fqn AB = Fqn.fromString("/a/b"); - Fqn ABC = Fqn.fromString("/a/b/c"); - Fqn ABCD = Fqn.fromString("/a/b/c/d"); - LockManager lockManager; - InvocationContextContainer icc; + protected Cache<String, String> cache; + protected TransactionManager tm; + protected Fqn A = Fqn.fromString("/a"); + protected Fqn AB = Fqn.fromString("/a/b"); + protected Fqn ABC = Fqn.fromString("/a/b/c"); + protected Fqn ABCD = Fqn.fromString("/a/b/c/d"); + protected LockManager lockManager; + protected InvocationContextContainer icc; protected boolean lockParentForInsertRemove = false; protected boolean repeatableRead = true; protected boolean allowWriteSkew = false; @@ -63,20 +63,20 @@ TestingUtil.killCaches(cache); } - private void assertLocked(Fqn fqn) + protected void assertLocked(Fqn fqn) { assert lockManager.isLocked(fqn) : fqn + " not locked!"; assert icc.get().getLocks().contains(fqn) : "Lock not recorded for " + fqn; } - private void assertNotLocked(Fqn fqn) + protected void assertNotLocked(Fqn fqn) { // can't rely on the negative test since other nodes may share the same lock with lock striping. // assert !lockManager.isLocked(fqn) : fqn + " is locked!"; assert !icc.get().getLocks().contains(fqn) : fqn + " lock recorded!"; } - private void assertNoLocks() + protected void assertNoLocks() { LockContainer lc = (LockContainer) TestingUtil.extractField(lockManager, "lockContainer"); assert lc.getNumLocksHeld() == 0 : "Stale locks exist!" + lockManager.printLockInfo(); @@ -347,7 +347,9 @@ assertNoLocks(); tm.begin(); + assert cache.getNode(A) != null; assert !(cache.getNode(A).getChildrenNames().isEmpty()); + assert cache.getNode(A).getChildrenNames().contains(AB.getLastElement()); assertNoLocks(); tm.commit(); assertNoLocks(); @@ -453,7 +455,7 @@ tm.rollback(); tm.resume(reader); - assert "v".equals(cache.get(AB, "k")); + assert "v".equals(cache.get(AB, "k")) : "Expecting 'v' but was " + cache.get(AB, "k"); tm.commit(); // even after commit @@ -461,6 +463,30 @@ assertNoLocks(); } + public void testRollbacksOnNullNode() throws Exception + { + tm.begin(); + assert null == cache.get(AB, "k"); + assert null == cache.getNode(AB); + Transaction reader = tm.suspend(); + + tm.begin(); + cache.put(AB, "k", "v"); + assert null != cache.getNode(AB); + assert "v".equals(cache.get(AB, "k")); + tm.rollback(); + + tm.resume(reader); + assert null == cache.get(AB, "k") : "Expecting null but was " + cache.get(AB, "k"); + assert null == cache.getNode(AB); + tm.commit(); + + // even after commit + assert null == cache.get(AB, "k"); + assert null == cache.getNode(AB); + assertNoLocks(); + } + public void testWriteSkew() throws Exception { if (repeatableRead) Modified: core/trunk/src/test/java/org/jboss/cache/api/mvcc/repeatable_read/RepeatableReadNoWriteSkewLockTest.java =================================================================== --- core/trunk/src/test/java/org/jboss/cache/api/mvcc/repeatable_read/RepeatableReadNoWriteSkewLockTest.java 2008-06-30 16:49:02 UTC (rev 6128) +++ core/trunk/src/test/java/org/jboss/cache/api/mvcc/repeatable_read/RepeatableReadNoWriteSkewLockTest.java 2008-06-30 19:07:05 UTC (rev 6129) @@ -1,14 +1,12 @@ package org.jboss.cache.api.mvcc.repeatable_read; -import org.jboss.cache.api.mvcc.LockTestBase; import org.testng.annotations.Test; @Test(groups = {"functional", "mvcc"}) -public class RepeatableReadNoWriteSkewLockTest extends LockTestBase +public class RepeatableReadNoWriteSkewLockTest extends RepeatableReadTestBase { public RepeatableReadNoWriteSkewLockTest() { - repeatableRead = true; allowWriteSkew = false; lockParentForInsertRemove = false; } Modified: core/trunk/src/test/java/org/jboss/cache/api/mvcc/repeatable_read/RepeatableReadNoWriteSkewWithParentLockTest.java =================================================================== --- core/trunk/src/test/java/org/jboss/cache/api/mvcc/repeatable_read/RepeatableReadNoWriteSkewWithParentLockTest.java 2008-06-30 16:49:02 UTC (rev 6128) +++ core/trunk/src/test/java/org/jboss/cache/api/mvcc/repeatable_read/RepeatableReadNoWriteSkewWithParentLockTest.java 2008-06-30 19:07:05 UTC (rev 6129) @@ -1,14 +1,12 @@ package org.jboss.cache.api.mvcc.repeatable_read; -import org.jboss.cache.api.mvcc.LockTestBase; import org.testng.annotations.Test; @Test(groups = {"functional", "mvcc"}) -public class RepeatableReadNoWriteSkewWithParentLockTest extends LockTestBase +public class RepeatableReadNoWriteSkewWithParentLockTest extends RepeatableReadTestBase { public RepeatableReadNoWriteSkewWithParentLockTest() { - repeatableRead = true; allowWriteSkew = false; lockParentForInsertRemove = true; } Added: core/trunk/src/test/java/org/jboss/cache/api/mvcc/repeatable_read/RepeatableReadTestBase.java =================================================================== --- core/trunk/src/test/java/org/jboss/cache/api/mvcc/repeatable_read/RepeatableReadTestBase.java (rev 0) +++ core/trunk/src/test/java/org/jboss/cache/api/mvcc/repeatable_read/RepeatableReadTestBase.java 2008-06-30 19:07:05 UTC (rev 6129) @@ -0,0 +1,88 @@ +package org.jboss.cache.api.mvcc.repeatable_read; + +import org.jboss.cache.api.mvcc.LockTestBase; + +import javax.transaction.Transaction; + +public abstract class RepeatableReadTestBase extends LockTestBase +{ + protected RepeatableReadTestBase() + { + repeatableRead = true; + } + + public void testRepeatableReadWithRemove() throws Exception + { + cache.put(AB, "k", "v"); + + tm.begin(); + assert cache.getNode(AB) != null; + Transaction reader = tm.suspend(); + + tm.begin(); + assert cache.removeNode(AB); + assert cache.getNode(AB) == null; + tm.commit(); + + assert cache.getNode(AB) == null; + + tm.resume(reader); + assert cache.getNode(AB) != null; + assert "v".equals(cache.get(AB, "k")); + tm.commit(); + + assert cache.getNode(AB) == null; + assertNoLocks(); + } + + public void testRepeatableReadWithEvict() throws Exception + { + cache.put(AB, "k", "v"); + + tm.begin(); + assert cache.getNode(AB) != null; + Transaction reader = tm.suspend(); + + tm.begin(); + cache.evict(AB); + assert cache.getNode(AB) == null; + tm.commit(); + + assert cache.getNode(AB) == null; + + tm.resume(reader); + assert cache.getNode(AB) != null; + assert "v".equals(cache.get(AB, "k")); + tm.commit(); + + assert cache.getNode(AB) == null; + assertNoLocks(); + } + + public void testRepeatableReadWithNull() throws Exception + { + assert cache.getNode(AB) == null; + + tm.begin(); + assert cache.getNode(AB) == null; + Transaction reader = tm.suspend(); + + tm.begin(); + cache.put(AB, "k", "v"); + assert cache.getNode(AB) != null; + assert "v".equals(cache.get(AB, "k")); + tm.commit(); + + assert cache.getNode(AB) != null; + assert "v".equals(cache.get(AB, "k")); + + tm.resume(reader); + assert cache.getNode(AB) == null; + assert cache.get(AB, "k") == null; + tm.commit(); + + assert cache.getNode(AB) != null; + assert "v".equals(cache.get(AB, "k")); + assertNoLocks(); + } +} Modified: core/trunk/src/test/java/org/jboss/cache/api/mvcc/repeatable_read/RepeatableReadWriteSkewLockTest.java =================================================================== --- core/trunk/src/test/java/org/jboss/cache/api/mvcc/repeatable_read/RepeatableReadWriteSkewLockTest.java 2008-06-30 16:49:02 UTC (rev 6128) +++ core/trunk/src/test/java/org/jboss/cache/api/mvcc/repeatable_read/RepeatableReadWriteSkewLockTest.java 2008-06-30 19:07:05 UTC (rev 6129) @@ -1,14 +1,12 @@ package org.jboss.cache.api.mvcc.repeatable_read; -import org.jboss.cache.api.mvcc.LockTestBase; import org.testng.annotations.Test; @Test(groups = {"functional", "mvcc"}) -public class RepeatableReadWriteSkewLockTest extends LockTestBase +public class RepeatableReadWriteSkewLockTest extends RepeatableReadTestBase { public RepeatableReadWriteSkewLockTest() { - repeatableRead = true; allowWriteSkew = true; lockParentForInsertRemove = false; } Modified: core/trunk/src/test/java/org/jboss/cache/api/mvcc/repeatable_read/RepeatableReadWriteSkewWithParentLockTest.java =================================================================== --- core/trunk/src/test/java/org/jboss/cache/api/mvcc/repeatable_read/RepeatableReadWriteSkewWithParentLockTest.java 2008-06-30 16:49:02 UTC (rev 6128) +++ core/trunk/src/test/java/org/jboss/cache/api/mvcc/repeatable_read/RepeatableReadWriteSkewWithParentLockTest.java 2008-06-30 19:07:05 UTC (rev 6129) @@ -1,14 +1,12 @@ package org.jboss.cache.api.mvcc.repeatable_read; -import org.jboss.cache.api.mvcc.LockTestBase; import org.testng.annotations.Test; @Test(groups = {"functional", "mvcc"}) -public class RepeatableReadWriteSkewWithParentLockTest extends LockTestBase +public class RepeatableReadWriteSkewWithParentLockTest extends RepeatableReadTestBase { public RepeatableReadWriteSkewWithParentLockTest() { - repeatableRead = true; allowWriteSkew = true; lockParentForInsertRemove = true; } Modified: core/trunk/src/test/java/org/jboss/cache/commands/write/EvictCommandTest.java =================================================================== --- core/trunk/src/test/java/org/jboss/cache/commands/write/EvictCommandTest.java 2008-06-30 16:49:02 UTC (rev 6128) +++ core/trunk/src/test/java/org/jboss/cache/commands/write/EvictCommandTest.java 2008-06-30 19:07:05 UTC (rev 6129) @@ -33,7 +33,7 @@ notifier = control.createMock(Notifier.class); container = control.createMock(DataContainer.class); command = new EvictCommand(testFqn); - command.initialize(notifier, container); + command.initialize(notifier, container, false); nodes = new MockNodesFixture(); }