[jbosscache-commits] JBoss Cache SVN: r8167 - in core/trunk/src/main/java/org/jboss/cache: lock and 1 other directory.

[jbosscache-commits at lists.jboss.org]

Author: manik.surtani at jboss.com
Date: 2009-08-07 06:59:23 -0400 (Fri, 07 Aug 2009)
New Revision: 8167

Modified:
   core/trunk/src/main/java/org/jboss/cache/PessimisticUnversionedNode.java
   core/trunk/src/main/java/org/jboss/cache/lock/PessimisticNodeBasedLockManager.java
Log:
Remove node creation race

Modified: core/trunk/src/main/java/org/jboss/cache/PessimisticUnversionedNode.java
===================================================================
--- core/trunk/src/main/java/org/jboss/cache/PessimisticUnversionedNode.java	2009-08-07 10:28:08 UTC (rev 8166)
+++ core/trunk/src/main/java/org/jboss/cache/PessimisticUnversionedNode.java	2009-08-07 10:59:23 UTC (rev 8167)
@@ -26,6 +26,7 @@
 import org.jboss.cache.commands.legacy.write.CreateNodeCommand;
 import org.jboss.cache.lock.IdentityLock;
 import org.jboss.cache.lock.LockStrategyFactory;
+import org.jboss.cache.lock.PessimisticNodeBasedLockManager;
 import org.jboss.cache.marshall.MarshalledValue;
 import org.jboss.cache.transaction.GlobalTransaction;
 import org.jboss.cache.util.FastCopyHashMap;
@@ -161,7 +162,7 @@
          throw new CacheException("Attempting to add a child [" + child.getFqn() + "] to [" + getFqn() + "].  Can only add direct children.");
    }
 
-   private NodeSPI<K, V> getOrCreateChild(Object childName, GlobalTransaction gtx, boolean createIfNotExists, boolean notify)
+   private NodeSPI<K, V> getOrCreateChild(Object childName, GlobalTransaction gtx, boolean createIfNotExists, boolean notify, PessimisticNodeBasedLockManager.LockAcquirer la)
    {
       NodeSPI<K, V> child;
       if (childName == null)
@@ -190,6 +191,7 @@
             {
                if (notify) cache.getNotifier().notifyNodeCreated(childFqn, true, ctx);
                child = newChild;
+               if (la != null) la.acquire(child);
                children().put(childName, child);
             }
          }
@@ -216,7 +218,7 @@
       if (f.size() == 1)
       {
          GlobalTransaction gtx = cache.getInvocationContext().getGlobalTransaction();
-         return getOrCreateChild(f.getLastElement(), gtx, true, notify);
+         return getOrCreateChild(f.getLastElement(), gtx, true, notify, null);
       }
       else
       {
@@ -228,9 +230,15 @@
    public NodeSPI<K, V> addChildDirect(Object o, boolean notify)
    {
       GlobalTransaction gtx = cache.getInvocationContext().getGlobalTransaction();
-      return getOrCreateChild(o, gtx, true, notify);
+      return getOrCreateChild(o, gtx, true, notify, null);
    }
 
+   public NodeSPI<K, V> addChildAndAcquireLock(Object o, PessimisticNodeBasedLockManager.LockAcquirer la)
+   {
+      GlobalTransaction gtx = cache.getInvocationContext().getGlobalTransaction();
+      return getOrCreateChild(o, gtx, true, false, la);
+   }
+
    @Override
    public NodeSPI<K, V> getChildDirect(Fqn fqn)
    {
@@ -331,7 +339,7 @@
    @Override
    public NodeSPI<K, V> getOrCreateChild(Object childName, GlobalTransaction gtx)
    {
-      return getOrCreateChild(childName, gtx, true, true);
+      return getOrCreateChild(childName, gtx, true, true, null);
    }
 
    @Override

Modified: core/trunk/src/main/java/org/jboss/cache/lock/PessimisticNodeBasedLockManager.java
===================================================================
--- core/trunk/src/main/java/org/jboss/cache/lock/PessimisticNodeBasedLockManager.java	2009-08-07 10:28:08 UTC (rev 8166)
+++ core/trunk/src/main/java/org/jboss/cache/lock/PessimisticNodeBasedLockManager.java	2009-08-07 10:59:23 UTC (rev 8167)
@@ -26,6 +26,7 @@
 import org.jboss.cache.Fqn;
 import org.jboss.cache.InvocationContext;
 import org.jboss.cache.NodeSPI;
+import org.jboss.cache.PessimisticUnversionedNode;
 import org.jboss.cache.commands.CommandsFactory;
 import org.jboss.cache.factories.annotations.Inject;
 import static org.jboss.cache.lock.LockType.WRITE;
@@ -137,12 +138,15 @@
 
       do
       {
+         boolean skipLockAcquire = false;
          if (currentNode == null)
          {
             if (createIfNotExists)
             {
                // if the new node is to be marked as deleted, do not notify!
-               currentNode = parent.addChildDirect(childName, !skipNotification);
+               PessimisticUnversionedNode parentInternalNode = (PessimisticUnversionedNode) parent.getDelegationTarget();
+               currentNode = parentInternalNode.addChildAndAcquireLock(childName, new LockAcquirer(ctx, WRITE, timeout, owner));
+               skipLockAcquire = true;
                if (!created)
                {
                   created = true;
@@ -171,8 +175,12 @@
          
 
          Fqn currentNodeFqn = currentNode.getFqn();
-         // actually acquire the lock we need.  This method blocks.
-         acquireNodeLock(ctx, currentNode, owner, lockTypeRequired, timeout);
+         if (!skipLockAcquire)
+         {
+            // actually acquire the lock we need.  This method blocks.
+            acquireNodeLock(ctx, currentNode, owner, lockTypeRequired, timeout);
+         }
+            
 
          LockUtil.manageReverseRemove(ctx, currentNode, reverseRemoveCheck, createdNodes, commandsFactory);
          // make sure the lock we acquired isn't on a deleted node/is an orphan!!
@@ -274,4 +282,31 @@
       // Record the lock for release on method return or tx commit/rollback
       if (acquired) ctx.addLock(lock);
    }
+
+   public class LockAcquirer
+   {
+      InvocationContext ctx;
+      LockType type;
+      long timeout;
+      Object owner;
+
+      public LockAcquirer(InvocationContext ctx, LockType type, long timeout, Object owner) {
+         this.ctx = ctx;
+         this.type = type;
+         this.timeout = timeout;
+         this.owner = owner;
+      }
+
+      public void acquire(NodeSPI node)
+      {
+         try
+         {
+            acquireNodeLock(ctx, node, owner, type, timeout);
+         }
+         catch (InterruptedException e)
+         {
+            Thread.currentThread().interrupt();
+         }
+      }
+   }
 }