geode-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From dschnei...@apache.org
Subject [2/2] incubator-geode git commit: GEODE-599: fix clear with concurrent writes
Date Tue, 16 Aug 2016 19:14:37 GMT
GEODE-599: fix clear with concurrent writes

Cache modification lock was being released before operation distribution to other members.
This provided a small window in which an operation from another thread could update the
region prior to the other members receiving notification (i.e. down leveled)

This closes #232


Project: http://git-wip-us.apache.org/repos/asf/incubator-geode/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-geode/commit/e08c1f54
Tree: http://git-wip-us.apache.org/repos/asf/incubator-geode/tree/e08c1f54
Diff: http://git-wip-us.apache.org/repos/asf/incubator-geode/diff/e08c1f54

Branch: refs/heads/develop
Commit: e08c1f54448589c85237c5110d14e81ed679ddc9
Parents: 7e585ea
Author: Scott Jewell <sjewell@pivotal.io>
Authored: Fri Jul 8 15:52:03 2016 -0700
Committer: Darrel Schneider <dschneider@pivotal.io>
Committed: Tue Aug 16 11:43:06 2016 -0700

----------------------------------------------------------------------
 .../internal/cache/AbstractRegionMap.java       | 917 ++++++++++---------
 .../gemfire/internal/cache/BucketRegion.java    |  38 +-
 .../cache/DistributedClearOperation.java        |   2 +-
 .../internal/cache/DistributedRegion.java       |  46 +-
 .../gemfire/internal/cache/LocalRegion.java     | 493 +++++-----
 .../gemfire/internal/cache/ProxyRegionMap.java  |   6 +
 .../gemfire/internal/cache/RegionMap.java       |   4 +
 .../internal/cache/StateFlushOperation.java     |   5 +
 .../cache/versions/RegionVersionVector.java     |  12 +-
 .../internal/cache/ARMLockTestHookAdapter.java  |  38 +
 .../cache/ClearRvvLockingDUnitTest.java         | 667 ++++++++++++++
 11 files changed, 1529 insertions(+), 699 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/e08c1f54/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/AbstractRegionMap.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/AbstractRegionMap.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/AbstractRegionMap.java
index faa8580..81e4d9f 100644
--- a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/AbstractRegionMap.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/AbstractRegionMap.java
@@ -41,7 +41,6 @@ import com.gemstone.gemfire.internal.cache.tier.sockets.HAEventWrapper;
 import com.gemstone.gemfire.internal.cache.versions.*;
 import com.gemstone.gemfire.internal.cache.wan.GatewaySenderEventImpl;
 import com.gemstone.gemfire.internal.concurrent.MapCallbackAdapter;
-import com.gemstone.gemfire.internal.concurrent.MapResult;
 import com.gemstone.gemfire.internal.i18n.LocalizedStrings;
 import com.gemstone.gemfire.internal.logging.LogService;
 import com.gemstone.gemfire.internal.logging.log4j.LocalizedMessage;
@@ -54,6 +53,7 @@ import com.gemstone.gemfire.internal.offheap.annotations.Retained;
 import com.gemstone.gemfire.internal.offheap.annotations.Unretained;
 import com.gemstone.gemfire.internal.sequencelog.EntryLogger;
 import com.gemstone.gemfire.internal.util.concurrent.CustomEntryConcurrentHashMap;
+
 import org.apache.logging.log4j.Logger;
 
 import java.util.*;
@@ -1031,21 +1031,19 @@ public abstract class AbstractRegionMap implements RegionMap {
     
     boolean retry = true;
     
-    while (retry) {
-      retry = false;
-
-      boolean opCompleted = false;
-      boolean doPart3 = false;
-
-      // We need to acquire the region entry while holding the lock to avoid #45620.
-      // However, we also want to release the lock before distribution to prevent
-      // potential deadlocks.  The outer try/finally ensures that the lock will be
-      // released without fail.  I'm avoiding indenting just to preserve the ability
-      // to track diffs since the code is fairly complex.
-      boolean doUnlock = true;
-      lockForCacheModification(owner, event);
-      try {
+    lockForCacheModification(owner, event);
+    try {
+
+      while (retry) {
+        retry = false;
 
+        boolean opCompleted = false;
+        boolean doPart3 = false;
+
+        // We need to acquire the region entry while holding the lock to avoid #45620.
+        // The outer try/finally ensures that the lock will be released without fail.  
+        // I'm avoiding indenting just to preserve the ability
+        // to track diffs since the code is fairly complex.
 
         RegionEntry re = getOrCreateRegionEntry(owner, event, Token.REMOVED_PHASE1, null, true, true); 
         RegionEntry tombstone = null;
@@ -1498,9 +1496,6 @@ public abstract class AbstractRegionMap implements RegionMap {
           return opCompleted;
         }
         finally {
-          releaseCacheModificationLock(owner, event);
-          doUnlock = false;
-
           try {
             // If concurrency conflict is there and event contains gateway version tag then
             // do NOT distribute.
@@ -1523,12 +1518,10 @@ public abstract class AbstractRegionMap implements RegionMap {
           }
         }
 
-      } finally { // failsafe on the read lock...see comment above
-        if (doUnlock) {
-          releaseCacheModificationLock(owner, event);
-        }
-      }
-    } // retry loop
+      } // retry loop
+    } finally { // failsafe on the read lock...see comment above
+      releaseCacheModificationLock(owner, event);
+    }
     return false;
   }
 
@@ -1840,10 +1833,10 @@ public abstract class AbstractRegionMap implements RegionMap {
   
   public final boolean invalidate(EntryEventImpl event,
       boolean invokeCallbacks, boolean forceNewEntry, boolean forceCallbacks)
-      throws EntryNotFoundException
+          throws EntryNotFoundException
   {
     final boolean isDebugEnabled = logger.isDebugEnabled();
-    
+
     final LocalRegion owner = _getOwner();
     if (owner == null) {
       // "fix" for bug 32440
@@ -1857,377 +1850,381 @@ public abstract class AbstractRegionMap implements RegionMap {
     DiskRegion dr = owner.getDiskRegion();
     boolean ownerIsInitialized = owner.isInitialized();
     try {
-    // Fix for Bug #44431. We do NOT want to update the region and wait
-    // later for index INIT as region.clear() can cause inconsistency if
-    // happened in parallel as it also does index INIT.
-    IndexManager oqlIndexManager = owner.getIndexManager() ; 
-    if (oqlIndexManager != null) {
-      oqlIndexManager.waitForIndexInit();
-    }
-    lockForCacheModification(owner, event);
-    try {
-      if (forceNewEntry || forceCallbacks) {
-        boolean opCompleted = false;
-        RegionEntry newRe = getEntryFactory().createEntry(owner, event.getKey(),
-            Token.REMOVED_PHASE1);
-          synchronized (newRe) {
-            try {
-              RegionEntry oldRe = putEntryIfAbsent(event.getKey(), newRe);
-              
-              while (!opCompleted && oldRe != null) {
-                synchronized (oldRe) {
-                  // if the RE is in phase 2 of removal, it will really be removed
-                  // from the map.  Otherwise, we can use it here and the thread
-                  // that is destroying the RE will see the invalidation and not
-                  // proceed to phase 2 of removal.
-                  if (oldRe.isRemovedPhase2()) {
-                    oldRe = putEntryIfAbsent(event.getKey(), newRe);
-                    if (oldRe != null) {
-                      owner.getCachePerfStats().incRetries();
-                    }
-                  } else {
-                    opCompleted = true;
-                    event.setRegionEntry(oldRe);
-                    if (oldRe.isDestroyed()) {
-                      if (isDebugEnabled) {
-                        logger.debug("mapInvalidate: Found DESTROYED token, not invalidated; key={}", event.getKey());
-                      }
-                    } else if (oldRe.isInvalid()) {
-                    
-                      // was already invalid, do not invoke listeners or increment stat
-                      if (isDebugEnabled) {
-                        logger.debug("mapInvalidate: Entry already invalid: '{}'", event.getKey());
-                      }
-                      processVersionTag(oldRe, event);
-                      try {
-                        oldRe.setValue(owner, oldRe.getValueInVM(owner)); // OFFHEAP noop setting an already invalid to invalid; No need to call prepareValueForCache since it is an invalid token.
-                      } catch (RegionClearedException e) {
-                        // that's okay - when writing an invalid into a disk, the
-                        // region has been cleared (including this token)
+      // Fix for Bug #44431. We do NOT want to update the region and wait
+      // later for index INIT as region.clear() can cause inconsistency if
+      // happened in parallel as it also does index INIT.
+      IndexManager oqlIndexManager = owner.getIndexManager() ; 
+      if (oqlIndexManager != null) {
+        oqlIndexManager.waitForIndexInit();
+      }
+      lockForCacheModification(owner, event);
+      try {
+        try {
+          if (forceNewEntry || forceCallbacks) {
+            boolean opCompleted = false;
+            RegionEntry newRe = getEntryFactory().createEntry(owner, event.getKey(),
+                Token.REMOVED_PHASE1);
+            synchronized (newRe) {
+              try {
+                RegionEntry oldRe = putEntryIfAbsent(event.getKey(), newRe);
+
+                while (!opCompleted && oldRe != null) {
+                  synchronized (oldRe) {
+                    // if the RE is in phase 2 of removal, it will really be removed
+                    // from the map.  Otherwise, we can use it here and the thread
+                    // that is destroying the RE will see the invalidation and not
+                    // proceed to phase 2 of removal.
+                    if (oldRe.isRemovedPhase2()) {
+                      oldRe = putEntryIfAbsent(event.getKey(), newRe);
+                      if (oldRe != null) {
+                        owner.getCachePerfStats().incRetries();
                       }
                     } else {
-                      owner.serverInvalidate(event);
-                      if (owner.concurrencyChecksEnabled && event.noVersionReceivedFromServer()) {
-                        // server did not perform the invalidation, so don't leave an invalid
-                        // entry here
-                        return false;
-                      }
-                      final int oldSize = owner.calculateRegionEntryValueSize(oldRe);
-                      //added for cq which needs old value. rdubey
-                      FilterProfile fp = owner.getFilterProfile();
-                      if (!oldRe.isRemoved() && 
-                          (fp != null && fp.getCqCount() > 0)) {
-                        
-                        Object oldValue = oldRe.getValueInVM(owner); // OFFHEAP EntryEventImpl oldValue
-                        
-                        // this will not fault in the value.
-                        if (oldValue == Token.NOT_AVAILABLE){
-                          event.setOldValue(oldRe.getValueOnDiskOrBuffer(owner));
-                        } else {
-                          event.setOldValue(oldValue);
+                      opCompleted = true;
+                      event.setRegionEntry(oldRe);
+                      if (oldRe.isDestroyed()) {
+                        if (isDebugEnabled) {
+                          logger.debug("mapInvalidate: Found DESTROYED token, not invalidated; key={}", event.getKey());
                         }
-                      }
-                      boolean isCreate = false;
-                      try {
-                        if (oldRe.isRemoved()) {
-                          processVersionTag(oldRe, event);
-                          event.putNewEntry(owner, oldRe);
-                          EntryLogger.logInvalidate(event);
-                          owner.recordEvent(event);
-                          if (!oldRe.isTombstone()) {
-                            owner.updateSizeOnPut(event.getKey(), oldSize, event.getNewValueBucketSize());
+                      } else if (oldRe.isInvalid()) {
+
+                        // was already invalid, do not invoke listeners or increment stat
+                        if (isDebugEnabled) {
+                          logger.debug("mapInvalidate: Entry already invalid: '{}'", event.getKey());
+                        }
+                        processVersionTag(oldRe, event);
+                        try {
+                          oldRe.setValue(owner, oldRe.getValueInVM(owner)); // OFFHEAP noop setting an already invalid to invalid; No need to call prepareValueForCache since it is an invalid token.
+                        } catch (RegionClearedException e) {
+                          // that's okay - when writing an invalid into a disk, the
+                          // region has been cleared (including this token)
+                        }
+                      } else {
+                        owner.serverInvalidate(event);
+                        if (owner.concurrencyChecksEnabled && event.noVersionReceivedFromServer()) {
+                          // server did not perform the invalidation, so don't leave an invalid
+                          // entry here
+                          return false;
+                        }
+                        final int oldSize = owner.calculateRegionEntryValueSize(oldRe);
+                        //added for cq which needs old value. rdubey
+                        FilterProfile fp = owner.getFilterProfile();
+                        if (!oldRe.isRemoved() && 
+                            (fp != null && fp.getCqCount() > 0)) {
+
+                          Object oldValue = oldRe.getValueInVM(owner); // OFFHEAP EntryEventImpl oldValue
+
+                          // this will not fault in the value.
+                          if (oldValue == Token.NOT_AVAILABLE){
+                            event.setOldValue(oldRe.getValueOnDiskOrBuffer(owner));
                           } else {
-                            owner.updateSizeOnCreate(event.getKey(), event.getNewValueBucketSize());
-                            isCreate = true;
+                            event.setOldValue(oldValue);
                           }
-                        } else {
-                          processVersionTag(oldRe, event);
-                          event.putExistingEntry(owner, oldRe);
+                        }
+                        boolean isCreate = false;
+                        try {
+                          if (oldRe.isRemoved()) {
+                            processVersionTag(oldRe, event);
+                            event.putNewEntry(owner, oldRe);
+                            EntryLogger.logInvalidate(event);
+                            owner.recordEvent(event);
+                            if (!oldRe.isTombstone()) {
+                              owner.updateSizeOnPut(event.getKey(), oldSize, event.getNewValueBucketSize());
+                            } else {
+                              owner.updateSizeOnCreate(event.getKey(), event.getNewValueBucketSize());
+                              isCreate = true;
+                            }
+                          } else {
+                            processVersionTag(oldRe, event);
+                            event.putExistingEntry(owner, oldRe);
+                            EntryLogger.logInvalidate(event);
+                            owner.recordEvent(event);
+                            owner.updateSizeOnPut(event.getKey(), oldSize, event.getNewValueBucketSize());
+                          }
+                        }
+                        catch (RegionClearedException e) {
+                          // generate versionTag for the event
                           EntryLogger.logInvalidate(event);
                           owner.recordEvent(event);
-                          owner.updateSizeOnPut(event.getKey(), oldSize, event.getNewValueBucketSize());
+                          clearOccured = true;
                         }
+                        owner.basicInvalidatePart2(oldRe, event,
+                            clearOccured /* conflict with clear */, invokeCallbacks);
+                        if (!clearOccured) {
+                          if (isCreate) {
+                            lruEntryCreate(oldRe);
+                          } else {
+                            lruEntryUpdate(oldRe);
+                          }
+                        }                   
+                        didInvalidate = true;
+                        invalidatedRe = oldRe;
                       }
-                      catch (RegionClearedException e) {
-                        // generate versionTag for the event
-                        EntryLogger.logInvalidate(event);
-                        owner.recordEvent(event);
-                        clearOccured = true;
-                      }
-                      owner.basicInvalidatePart2(oldRe, event,
-                          clearOccured /* conflict with clear */, invokeCallbacks);
-                      if (!clearOccured) {
-                        if (isCreate) {
-                          lruEntryCreate(oldRe);
-                        } else {
-                          lruEntryUpdate(oldRe);
-                        }
-                      }                   
-                      didInvalidate = true;
-                      invalidatedRe = oldRe;
+                    }
+                  } // synchronized oldRe
+                } // while oldRe exists
+
+                if (!opCompleted) {
+                  if (forceNewEntry && event.isFromServer()) {
+                    // don't invoke listeners - we didn't force new entries for
+                    // CCU invalidations before 7.0, and listeners don't care
+                    if (!FORCE_INVALIDATE_EVENT) {
+                      event.inhibitCacheListenerNotification(true);
                     }
                   }
-                } // synchronized oldRe
-              } // while oldRe exists
-              
-              if (!opCompleted) {
-                if (forceNewEntry && event.isFromServer()) {
-                  // don't invoke listeners - we didn't force new entries for
-                  // CCU invalidations before 7.0, and listeners don't care
-                  if (!FORCE_INVALIDATE_EVENT) {
-                    event.inhibitCacheListenerNotification(true);
+                  event.setRegionEntry(newRe);
+                  owner.serverInvalidate(event);
+                  if (!forceNewEntry && event.noVersionReceivedFromServer()) {
+                    // server did not perform the invalidation, so don't leave an invalid
+                    // entry here
+                    return false;
                   }
-                }
-                event.setRegionEntry(newRe);
-                owner.serverInvalidate(event);
-                if (!forceNewEntry && event.noVersionReceivedFromServer()) {
-                  // server did not perform the invalidation, so don't leave an invalid
-                  // entry here
-                  return false;
-                }
-                try {
-                  ownerIsInitialized = owner.isInitialized();
-                  if (!ownerIsInitialized && owner.getDataPolicy().withReplication()) {
-                    final int oldSize = owner.calculateRegionEntryValueSize(newRe);
-                    invalidateEntry(event, newRe, oldSize);
+                  try {
+                    ownerIsInitialized = owner.isInitialized();
+                    if (!ownerIsInitialized && owner.getDataPolicy().withReplication()) {
+                      final int oldSize = owner.calculateRegionEntryValueSize(newRe);
+                      invalidateEntry(event, newRe, oldSize);
+                    }
+                    else {
+                      invalidateNewEntry(event, owner, newRe);
+                    }
                   }
-                  else {
-                    invalidateNewEntry(event, owner, newRe);
+                  catch (RegionClearedException e) {
+                    // TODO: deltaGII: do we even need RegionClearedException?
+                    // generate versionTag for the event
+                    owner.recordEvent(event);
+                    clearOccured = true;
                   }
+                  owner.basicInvalidatePart2(newRe, event, clearOccured /*conflict with clear*/, invokeCallbacks);
+                  if (!clearOccured) {
+                    lruEntryCreate(newRe);
+                    incEntryCount(1);
+                  }            
+                  opCompleted = true;
+                  didInvalidate = true;
+                  invalidatedRe = newRe;
+                  // Don't leave an entry in the cache, if we
+                  // just wanted to force the distribution and events
+                  // for this invalidate
+                  if (!forceNewEntry) {
+                    removeEntry(event.getKey(), newRe, false);
+                  } 
+                } // !opCompleted
+              } catch (ConcurrentCacheModificationException ccme) {
+                VersionTag tag = event.getVersionTag();
+                if (tag != null && tag.isTimeStampUpdated()) {
+                  // Notify gateways of new time-stamp.
+                  owner.notifyTimestampsToGateways(event);
                 }
-                catch (RegionClearedException e) {
-                  // TODO: deltaGII: do we even need RegionClearedException?
-                  // generate versionTag for the event
-                  owner.recordEvent(event);
-                  clearOccured = true;
-                }
-                owner.basicInvalidatePart2(newRe, event, clearOccured /*conflict with clear*/, invokeCallbacks);
-                if (!clearOccured) {
-                  lruEntryCreate(newRe);
-                  incEntryCount(1);
-                }            
-                opCompleted = true;
-                didInvalidate = true;
-                invalidatedRe = newRe;
-                // Don't leave an entry in the cache, if we
-                // just wanted to force the distribution and events
-                // for this invalidate
-                if (!forceNewEntry) {
+                throw ccme;
+              } finally {
+                if (!opCompleted) {
                   removeEntry(event.getKey(), newRe, false);
-                } 
-              } // !opCompleted
-            } catch (ConcurrentCacheModificationException ccme) {
-              VersionTag tag = event.getVersionTag();
-              if (tag != null && tag.isTimeStampUpdated()) {
-                // Notify gateways of new time-stamp.
-                owner.notifyTimestampsToGateways(event);
-              }
-              throw ccme;
-            } finally {
-              if (!opCompleted) {
-                removeEntry(event.getKey(), newRe, false);
-              }
-            }
-          } // synchronized newRe
-      } // forceNewEntry
-      else { // !forceNewEntry
-        boolean retry = true;
-        // RegionEntry retryEntry = null;
-        // int retries = -1;
-        
-        while (retry) {
-          retry = false;
-          boolean entryExisted = false;
-          RegionEntry re = getEntry(event.getKey());
-          RegionEntry tombstone = null;
-          boolean haveTombstone = false;
-          if (re != null && re.isTombstone()) {
-            tombstone = re;
-            haveTombstone = true;
-            re = null;
-          }
-          if (re == null) {
-            ownerIsInitialized = owner.isInitialized();
-            if (!ownerIsInitialized) {
-              // when GII message arrived or processed later than invalidate
-              // message, the entry should be created as placeholder
-              RegionEntry newRe = haveTombstone? tombstone : getEntryFactory().createEntry(owner, event.getKey(),
-                  Token.INVALID);
-              synchronized (newRe) {
-                if (haveTombstone && !tombstone.isTombstone()) {
-                  // state of the tombstone has changed so we need to retry
-                  retry = true;
-                  //retryEntry = tombstone; // leave this in place for debugging
-                  continue;
-                }
-                re = putEntryIfAbsent(event.getKey(), newRe);
-                if (re == tombstone) {
-                  re = null; // pretend we don't have an entry
                 }
               }
-            } else if (owner.getServerProxy() != null) {
-              Object sync = haveTombstone? tombstone : new Object();
-              synchronized(sync) {
-                if (haveTombstone && !tombstone.isTombstone()) { 
-                  // bug 45295: state of the tombstone has changed so we need to retry
-                  retry = true;
-                  //retryEntry = tombstone; // leave this in place for debugging
-                  continue;
-                }
-       
-                // bug #43287 - send event to server even if it's not in the client (LRU may have evicted it)
-                owner.serverInvalidate(event);
-                if (owner.concurrencyChecksEnabled) {
-                  if (event.getVersionTag() == null) {
-                    // server did not perform the invalidation, so don't leave an invalid
-                    // entry here
-                    return false;
-                  } else if (tombstone != null) {
-                    processVersionTag(tombstone, event);
-                    try {
-                      if (!tombstone.isTombstone()) {
-                        if (isDebugEnabled) {
-                          logger.debug("tombstone is no longer a tombstone. {}:event={}", tombstone, event);
+            } // synchronized newRe
+          } // forceNewEntry
+          else { // !forceNewEntry
+            boolean retry = true;
+            // RegionEntry retryEntry = null;
+            // int retries = -1;
+
+            while (retry) {
+              retry = false;
+              boolean entryExisted = false;
+              RegionEntry re = getEntry(event.getKey());
+              RegionEntry tombstone = null;
+              boolean haveTombstone = false;
+              if (re != null && re.isTombstone()) {
+                tombstone = re;
+                haveTombstone = true;
+                re = null;
+              }
+              if (re == null) {
+                ownerIsInitialized = owner.isInitialized();
+                if (!ownerIsInitialized) {
+                  // when GII message arrived or processed later than invalidate
+                  // message, the entry should be created as placeholder
+                  RegionEntry newRe = haveTombstone? tombstone : getEntryFactory().createEntry(owner, event.getKey(),
+                      Token.INVALID);
+                  synchronized (newRe) {
+                    if (haveTombstone && !tombstone.isTombstone()) {
+                      // state of the tombstone has changed so we need to retry
+                      retry = true;
+                      //retryEntry = tombstone; // leave this in place for debugging
+                      continue;
+                    }
+                    re = putEntryIfAbsent(event.getKey(), newRe);
+                    if (re == tombstone) {
+                      re = null; // pretend we don't have an entry
+                    }
+                  }
+                } else if (owner.getServerProxy() != null) {
+                  Object sync = haveTombstone? tombstone : new Object();
+                  synchronized(sync) {
+                    if (haveTombstone && !tombstone.isTombstone()) { 
+                      // bug 45295: state of the tombstone has changed so we need to retry
+                      retry = true;
+                      //retryEntry = tombstone; // leave this in place for debugging
+                      continue;
+                    }
+
+                    // bug #43287 - send event to server even if it's not in the client (LRU may have evicted it)
+                    owner.serverInvalidate(event);
+                    if (owner.concurrencyChecksEnabled) {
+                      if (event.getVersionTag() == null) {
+                        // server did not perform the invalidation, so don't leave an invalid
+                        // entry here
+                        return false;
+                      } else if (tombstone != null) {
+                        processVersionTag(tombstone, event);
+                        try {
+                          if (!tombstone.isTombstone()) {
+                            if (isDebugEnabled) {
+                              logger.debug("tombstone is no longer a tombstone. {}:event={}", tombstone, event);
+                            }
+                          }
+                          tombstone.setValue(owner, Token.TOMBSTONE);
+                        } catch (RegionClearedException e) {
+                          // that's okay - when writing a tombstone into a disk, the
+                          // region has been cleared (including this tombstone)
+                        } catch (ConcurrentCacheModificationException ccme) {
+                          VersionTag tag = event.getVersionTag();
+                          if (tag != null && tag.isTimeStampUpdated()) {
+                            // Notify gateways of new time-stamp.
+                            owner.notifyTimestampsToGateways(event);
+                          }
+                          throw ccme;
                         }
+                        // update the tombstone's version to prevent an older CCU/putAll from overwriting it
+                        owner.rescheduleTombstone(tombstone, event.getVersionTag());
                       }
-                      tombstone.setValue(owner, Token.TOMBSTONE);
-                    } catch (RegionClearedException e) {
-                      // that's okay - when writing a tombstone into a disk, the
-                      // region has been cleared (including this tombstone)
-                    } catch (ConcurrentCacheModificationException ccme) {
-                      VersionTag tag = event.getVersionTag();
-                      if (tag != null && tag.isTimeStampUpdated()) {
-                        // Notify gateways of new time-stamp.
-                        owner.notifyTimestampsToGateways(event);
-                      }
-                      throw ccme;
                     }
-                    // update the tombstone's version to prevent an older CCU/putAll from overwriting it
-                    owner.rescheduleTombstone(tombstone, event.getVersionTag());
                   }
+                  entryExisted = true;
                 }
               }
-              entryExisted = true;
-            }
-          }
-          if (re != null) {
-            // Gester: Race condition in GII
-            // when adding the placeholder for invalidate entry during GII,
-            // if the GII got processed earlier for this entry, then do 
-            // normal invalidate operation
-            synchronized (re) {
-              if (!event.isOriginRemote() && event.getOperation().isExpiration()) {
-                // If this expiration started locally then only do it if the RE is not being used by a tx.
-                if (re.isInUseByTransaction()) {
-                  return false;
-                }
-              }
-              if (re.isTombstone() || (!re.isRemoved() && !re.isDestroyed())) {
-                entryExisted = true;
-                if (re.isInvalid()) {
-                  // was already invalid, do not invoke listeners or increment
-                  // stat
-                  if (isDebugEnabled) {
-                    logger.debug("Invalidate: Entry already invalid: '{}'", event.getKey());
-                  }
-                  if (event.getVersionTag() != null && owner.getVersionVector() != null) {
-                    owner.getVersionVector().recordVersion((InternalDistributedMember) event.getDistributedMember(), event.getVersionTag());
-                  }
-                }
-                else { // previous value not invalid
-                  event.setRegionEntry(re);
-                  owner.serverInvalidate(event);
-                  if (owner.concurrencyChecksEnabled && event.noVersionReceivedFromServer()) {
-                    // server did not perform the invalidation, so don't leave an invalid
-                    // entry here
-                    if (isDebugEnabled) {
-                      logger.debug("returning early because server did not generate a version stamp for this event:{}", event);
-                    }
-                    return false;
-                  }
-             // in case of overflow to disk we need the old value for cqs.
-                  if(owner.getFilterProfile().getCqCount() > 0){
-                    //use to be getValue and can cause dead lock rdubey.
-                    if (re.isValueNull()) {
-                      event.setOldValue(re.getValueOnDiskOrBuffer(owner));
-                    } else {
-                      Object v = re.getValueInVM(owner);
-                      event.setOldValue(v); // OFFHEAP escapes to EntryEventImpl oldValue
+              if (re != null) {
+                // Gester: Race condition in GII
+                // when adding the placeholder for invalidate entry during GII,
+                // if the GII got processed earlier for this entry, then do 
+                // normal invalidate operation
+                synchronized (re) {
+                  if (!event.isOriginRemote() && event.getOperation().isExpiration()) {
+                    // If this expiration started locally then only do it if the RE is not being used by a tx.
+                    if (re.isInUseByTransaction()) {
+                      return false;
                     }
                   }
-                  final boolean oldWasTombstone = re.isTombstone();
-                  final int oldSize = _getOwner().calculateRegionEntryValueSize(re);
-                  try {
-                    invalidateEntry(event, re, oldSize);
-                  }
-                  catch (RegionClearedException rce) {
-                    // generate versionTag for the event
-                    EntryLogger.logInvalidate(event);
-                    _getOwner().recordEvent(event);
-                    clearOccured = true;
-                  } catch (ConcurrentCacheModificationException ccme) {
-                    VersionTag tag = event.getVersionTag();
-                    if (tag != null && tag.isTimeStampUpdated()) {
-                      // Notify gateways of new time-stamp.
-                      owner.notifyTimestampsToGateways(event);
+                  if (re.isTombstone() || (!re.isRemoved() && !re.isDestroyed())) {
+                    entryExisted = true;
+                    if (re.isInvalid()) {
+                      // was already invalid, do not invoke listeners or increment
+                      // stat
+                      if (isDebugEnabled) {
+                        logger.debug("Invalidate: Entry already invalid: '{}'", event.getKey());
+                      }
+                      if (event.getVersionTag() != null && owner.getVersionVector() != null) {
+                        owner.getVersionVector().recordVersion((InternalDistributedMember) event.getDistributedMember(), event.getVersionTag());
+                      }
                     }
-                    throw ccme;
+                    else { // previous value not invalid
+                      event.setRegionEntry(re);
+                      owner.serverInvalidate(event);
+                      if (owner.concurrencyChecksEnabled && event.noVersionReceivedFromServer()) {
+                        // server did not perform the invalidation, so don't leave an invalid
+                        // entry here
+                        if (isDebugEnabled) {
+                          logger.debug("returning early because server did not generate a version stamp for this event:{}", event);
+                        }
+                        return false;
+                      }
+                      // in case of overflow to disk we need the old value for cqs.
+                      if(owner.getFilterProfile().getCqCount() > 0){
+                        //use to be getValue and can cause dead lock rdubey.
+                        if (re.isValueNull()) {
+                          event.setOldValue(re.getValueOnDiskOrBuffer(owner));
+                        } else {
+                          Object v = re.getValueInVM(owner);
+                          event.setOldValue(v); // OFFHEAP escapes to EntryEventImpl oldValue
+                        }
+                      }
+                      final boolean oldWasTombstone = re.isTombstone();
+                      final int oldSize = _getOwner().calculateRegionEntryValueSize(re);
+                      try {
+                        invalidateEntry(event, re, oldSize);
+                      }
+                      catch (RegionClearedException rce) {
+                        // generate versionTag for the event
+                        EntryLogger.logInvalidate(event);
+                        _getOwner().recordEvent(event);
+                        clearOccured = true;
+                      } catch (ConcurrentCacheModificationException ccme) {
+                        VersionTag tag = event.getVersionTag();
+                        if (tag != null && tag.isTimeStampUpdated()) {
+                          // Notify gateways of new time-stamp.
+                          owner.notifyTimestampsToGateways(event);
+                        }
+                        throw ccme;
+                      }
+                      owner.basicInvalidatePart2(re, event,
+                          clearOccured /* conflict with clear */, invokeCallbacks);
+                      if (!clearOccured) {
+                        if (oldWasTombstone) {
+                          lruEntryCreate(re);
+                        } else {
+                          lruEntryUpdate(re);
+                        }
+                      }             
+                      didInvalidate = true;
+                      invalidatedRe = re;
+                    } // previous value not invalid
                   }
-                  owner.basicInvalidatePart2(re, event,
-                      clearOccured /* conflict with clear */, invokeCallbacks);
-                  if (!clearOccured) {
-                    if (oldWasTombstone) {
-                      lruEntryCreate(re);
-                    } else {
-                      lruEntryUpdate(re);
-                    }
-                  }             
-                  didInvalidate = true;
-                  invalidatedRe = re;
-                } // previous value not invalid
+                } // synchronized re
+              } // re != null
+              else {
+                // At this point, either it's not in GII mode, or the placeholder
+                // is in region, do nothing
               }
-            } // synchronized re
-          } // re != null
-          else {
-            // At this point, either it's not in GII mode, or the placeholder
-            // is in region, do nothing
-          }
-          if (!entryExisted) {
-            owner.checkEntryNotFound(event.getKey());
-          }
-        } // while(retry)
-      } // !forceNewEntry
-    } catch( DiskAccessException dae) {
-      invalidatedRe = null;
-      didInvalidate = false;
-      this._getOwner().handleDiskAccessException(dae);
-      throw dae;
-    } finally {
-      releaseCacheModificationLock(owner, event);
-      if (oqlIndexManager != null) {
-        oqlIndexManager.countDownIndexUpdaters();
-      }
-      if (invalidatedRe != null) {
-        owner.basicInvalidatePart3(invalidatedRe, event, invokeCallbacks);
-      }
-      if (didInvalidate && !clearOccured) {
-        try {
-          lruUpdateCallback();
+              if (!entryExisted) {
+                owner.checkEntryNotFound(event.getKey());
+              }
+            } // while(retry)
+          } // !forceNewEntry
         } catch( DiskAccessException dae) {
+          invalidatedRe = null;
+          didInvalidate = false;
           this._getOwner().handleDiskAccessException(dae);
           throw dae;
+        } finally {
+          if (oqlIndexManager != null) {
+            oqlIndexManager.countDownIndexUpdaters();
+          }
+          if (invalidatedRe != null) {
+            owner.basicInvalidatePart3(invalidatedRe, event, invokeCallbacks);
+          }
+          if (didInvalidate && !clearOccured) {
+            try {
+              lruUpdateCallback();
+            } catch( DiskAccessException dae) {
+              this._getOwner().handleDiskAccessException(dae);
+              throw dae;
+            }
+          }
+          else if (!didInvalidate){
+            resetThreadLocals();
+          }
+        }
+        return didInvalidate;
+      } finally {
+        if (ownerIsInitialized) {
+          forceInvalidateEvent(event, owner);
         }
       }
-      else if (!didInvalidate){
-        resetThreadLocals();
-      }
-    }
-    return didInvalidate;
     } finally {
-      if (ownerIsInitialized) {
-        forceInvalidateEvent(event, owner);
-      }
+      releaseCacheModificationLock(owner, event);
     }
+
   }
 
   protected void invalidateNewEntry(EntryEventImpl event,
@@ -2682,27 +2679,30 @@ public abstract class AbstractRegionMap implements RegionMap {
     boolean uninitialized = !owner.isInitialized();
     boolean retrieveOldValueForDelta = event.getDeltaBytes() != null
         && event.getRawNewValue() == null;
-    lockForCacheModification(owner, event);
     IndexManager oqlIndexManager = null;
+    lockForCacheModification(owner, event);
     try {
-      // Fix for Bug #44431. We do NOT want to update the region and wait
-      // later for index INIT as region.clear() can cause inconsistency if
-      // happened in parallel as it also does index INIT.
-      oqlIndexManager = owner.getIndexManager() ; 
-      if (oqlIndexManager != null) {
-        oqlIndexManager.waitForIndexInit();
-      }
+      try {
+        // Fix for Bug #44431. We do NOT want to update the region and wait
+        // later for index INIT as region.clear() can cause inconsistency if
+        // happened in parallel as it also does index INIT.
+        oqlIndexManager = owner.getIndexManager() ; 
+        if (oqlIndexManager != null) {
+          oqlIndexManager.waitForIndexInit();
+        }
 
-      // fix for bug #42169, replace must go to server if entry not on client
-      boolean replaceOnClient = event.getOperation() == Operation.REPLACE
-                && owner.getServerProxy() != null; 
+        // fix for bug #42169, replace must go to server if entry not on client
+        boolean replaceOnClient = event.getOperation() == Operation.REPLACE
+            && owner.getServerProxy() != null; 
         // Rather than having two different blocks for synchronizing oldRe
         // and newRe, have only one block and synchronize re
         RegionEntry re = null;
         boolean eventRecorded = false;
         boolean onlyExisting = ifOld && !replaceOnClient;
-		re = getOrCreateRegionEntry(owner, event, 
-		    Token.REMOVED_PHASE1, null, onlyExisting, false);
+
+        re = getOrCreateRegionEntry(owner, event, 
+
+            Token.REMOVED_PHASE1, null, onlyExisting, false);
         if (re == null) {
           return null;
         }
@@ -2712,9 +2712,8 @@ public abstract class AbstractRegionMap implements RegionMap {
             // from the map. otherwise we can append an event to it
             // and change its state
             if (re.isRemovedPhase2()) {
-                re = getOrCreateRegionEntry(owner, event,
-                    Token.REMOVED_PHASE1, null, onlyExisting, false);
-                _getOwner().getCachePerfStats().incRetries();
+              re = getOrCreateRegionEntry(owner, event, Token.REMOVED_PHASE1, null, onlyExisting, false);
+              _getOwner().getCachePerfStats().incRetries();
               if (re == null) {
                 // this will happen when onlyExisting is true
                 return null;
@@ -2757,48 +2756,48 @@ public abstract class AbstractRegionMap implements RegionMap {
 
                 // notify index of an update
                 notifyIndex(re, true);
+                try {
                   try {
-                    try {
-                      if ((cacheWrite && event.getOperation().isUpdate()) // if there is a cacheWriter, type of event has already been set
-                          || !re.isRemoved()
-                          || replaceOnClient) {
-                        // update
-                        updateEntry(event, requireOldValue, oldValueForDelta, re);
-                      } else {
-                        // create
-                        createEntry(event, owner, re);
-                      }
-                      owner.recordEvent(event);
-                      eventRecorded = true;
-                    } catch (RegionClearedException rce) {
-                      clearOccured = true;
-                      owner.recordEvent(event);
-                    } catch (ConcurrentCacheModificationException ccme) {
-                      VersionTag tag = event.getVersionTag();
-                      if (tag != null && tag.isTimeStampUpdated()) {
-                        // Notify gateways of new time-stamp.
-                        owner.notifyTimestampsToGateways(event);
-                      }
-                      throw ccme;
+                    if ((cacheWrite && event.getOperation().isUpdate()) // if there is a cacheWriter, type of event has already been set
+                        || !re.isRemoved()
+                        || replaceOnClient) {
+                      // update
+                      updateEntry(event, requireOldValue, oldValueForDelta, re);
+                    } else {
+                      // create
+                      createEntry(event, owner, re);
                     }
-                    if (uninitialized) {
-                      event.inhibitCacheListenerNotification(true);
+                    owner.recordEvent(event);
+                    eventRecorded = true;
+                  } catch (RegionClearedException rce) {
+                    clearOccured = true;
+                    owner.recordEvent(event);
+                  } catch (ConcurrentCacheModificationException ccme) {
+                    VersionTag tag = event.getVersionTag();
+                    if (tag != null && tag.isTimeStampUpdated()) {
+                      // Notify gateways of new time-stamp.
+                      owner.notifyTimestampsToGateways(event);
                     }
-                    updateLru(clearOccured, re, event);
-
-                    lastModifiedTime = owner.basicPutPart2(event, re,
-                        !uninitialized, lastModifiedTime, clearOccured);
-                  } finally {
-                    notifyIndex(re, false);
+                    throw ccme;
                   }
-                  result = re;
-                  break;
-                } finally {
-                  OffHeapHelper.release(oldValueForDelta);
-                  if (re != null && !onlyExisting && !isOpComplete(re, event)) {
-                    owner.cleanUpOnIncompleteOp(event, re);
+                  if (uninitialized) {
+                    event.inhibitCacheListenerNotification(true);
                   }
-                  else if (re != null && owner.isUsedForPartitionedRegionBucket()) {
+                  updateLru(clearOccured, re, event);
+
+                  lastModifiedTime = owner.basicPutPart2(event, re,
+                      !uninitialized, lastModifiedTime, clearOccured);
+                } finally {
+                  notifyIndex(re, false);
+                }
+                result = re;
+                break;
+              } finally {
+                OffHeapHelper.release(oldValueForDelta);
+                if (re != null && !onlyExisting && !isOpComplete(re, event)) {
+                  owner.cleanUpOnIncompleteOp(event, re);
+                }
+                else if (re != null && owner.isUsedForPartitionedRegionBucket()) {
                   BucketRegion br = (BucketRegion)owner;
                   CachePerfStats stats = br.getPartitionedRegion().getCachePerfStats();
                 }
@@ -2806,14 +2805,13 @@ public abstract class AbstractRegionMap implements RegionMap {
             }
           } // sync re
         }// end while
-    } catch (DiskAccessException dae) {
-      //Asif:Feel that it is safe to destroy the region here as there appears
-      // to be no chance of deadlock during region destruction      
-      result = null;
-      this._getOwner().handleDiskAccessException(dae);
-      throw dae;
-    } finally {
-        releaseCacheModificationLock(owner, event);
+      } catch (DiskAccessException dae) {
+        //Asif:Feel that it is safe to destroy the region here as there appears
+        // to be no chance of deadlock during region destruction      
+        result = null;
+        this._getOwner().handleDiskAccessException(dae);
+        throw dae;
+      } finally {
         if (oqlIndexManager != null) {
           oqlIndexManager.countDownIndexUpdaters();
         }
@@ -2841,8 +2839,10 @@ public abstract class AbstractRegionMap implements RegionMap {
         } else {
           resetThreadLocals();
         }
-    } // finally
-    
+      } 
+    } finally {
+      releaseCacheModificationLock(owner, event);
+    }
     return result;
   }
 
@@ -3632,45 +3632,69 @@ public abstract class AbstractRegionMap implements RegionMap {
   
 
   /** get version-generation permission from the region's version vector */
-  private void lockForCacheModification(LocalRegion owner, EntryEventImpl event) {
+  void lockForCacheModification(LocalRegion owner, EntryEventImpl event) {
     boolean lockedByBulkOp = event.isBulkOpInProgress() && owner.dataPolicy.withReplication();
-    if (!event.isOriginRemote() && !lockedByBulkOp) {
+    
+    if(armLockTestHook!=null) armLockTestHook.beforeLock(owner, event);
+    
+    if (!event.isOriginRemote() && !lockedByBulkOp && !owner.hasServerProxy()) {
       RegionVersionVector vector = owner.getVersionVector();
       if (vector != null) {
-        vector.lockForCacheModification(owner);
+        vector.lockForCacheModification();
       }
     }
+    
+    if(armLockTestHook!=null) armLockTestHook.afterLock(owner, event);
+
   }
   
   /** release version-generation permission from the region's version vector */
-  private void releaseCacheModificationLock(LocalRegion owner, EntryEventImpl event) {
+  void releaseCacheModificationLock(LocalRegion owner, EntryEventImpl event) {
     boolean lockedByBulkOp = event.isBulkOpInProgress() && owner.dataPolicy.withReplication();
-    if (!event.isOriginRemote() && !lockedByBulkOp) {
+    
+    if(armLockTestHook!=null) armLockTestHook.beforeRelease(owner, event);
+
+    if (!event.isOriginRemote() && !lockedByBulkOp && !owner.hasServerProxy()) {
       RegionVersionVector vector = owner.getVersionVector();
       if (vector != null) {
-        vector.releaseCacheModificationLock(owner);
+        vector.releaseCacheModificationLock();
       }
     }
+    
+    if(armLockTestHook!=null) armLockTestHook.afterRelease(owner, event);
+
   }
   
   /** get version-generation permission from the region's version vector */
   private void lockForTXCacheModification(LocalRegion owner, VersionTag tag) {
+    
+    if(armLockTestHook!=null) armLockTestHook.beforeLock(owner, null);
+    
     if ( !(tag != null && tag.isFromOtherMember()) ) {
       RegionVersionVector vector = owner.getVersionVector();
-      if (vector != null) {
-        vector.lockForCacheModification(owner);
+      if (vector != null && !owner.hasServerProxy()) {
+        vector.lockForCacheModification();
       }
     }
+    
+    if(armLockTestHook!=null) armLockTestHook.afterLock(owner, null);
+
   }
   
   /** release version-generation permission from the region's version vector */
   private void releaseTXCacheModificationLock(LocalRegion owner, VersionTag tag) {
+    
+    if(armLockTestHook!=null) armLockTestHook.beforeRelease(owner, null);
+
     if ( !(tag != null && tag.isFromOtherMember()) ) {
       RegionVersionVector vector = owner.getVersionVector();
-      if (vector != null) {
-        vector.releaseCacheModificationLock(owner);
+      if (vector != null && !owner.hasServerProxy()) {
+       vector.releaseCacheModificationLock();
       }
     }
+    
+    if(armLockTestHook!=null) armLockTestHook.afterRelease(owner, null);
+
   }
 
   /**
@@ -3798,5 +3822,30 @@ public abstract class AbstractRegionMap implements RegionMap {
     }
     return true;
   }
+  
+  public interface ARMLockTestHook {
+    public void beforeBulkLock(LocalRegion region);
+    public void afterBulkLock(LocalRegion region);
+    public void beforeBulkRelease(LocalRegion region);
+    public void afterBulkRelease(LocalRegion region);
+
+    public void beforeLock(LocalRegion region, CacheEvent event);
+    public void afterLock(LocalRegion region, CacheEvent event);
+    public void beforeRelease(LocalRegion region, CacheEvent event);
+    public void afterRelease(LocalRegion region, CacheEvent event);
+
+    public void beforeStateFlushWait();
+  }
+  
+  private ARMLockTestHook armLockTestHook;
+  
+  public ARMLockTestHook getARMLockTestHook() {
+    return armLockTestHook;
+  }
+  
+  public void setARMLockTestHook(ARMLockTestHook theHook) {
+    armLockTestHook = theHook;
+  }
+
 }
 

http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/e08c1f54/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/BucketRegion.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/BucketRegion.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/BucketRegion.java
index f1627d3..4e4c417 100644
--- a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/BucketRegion.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/BucketRegion.java
@@ -1268,26 +1268,32 @@ implements Bucket
     Assert.assertTrue(!isTX());
     Assert.assertTrue(event.getOperation().isDistributed());
 
-    beginLocalWrite(event);
+    LocalRegion lr = event.getLocalRegion();
+    AbstractRegionMap arm = ((AbstractRegionMap) lr.getRegionMap());
     try {
-      
-      if (!hasSeenEvent(event)) {
-        this.entries.updateEntryVersion(event);
-      } else {
-        if (logger.isTraceEnabled(LogMarker.DM)) {
-          logger.trace(LogMarker.DM, "BR.basicUpdateEntryVersion: this cache has already seen this event {}", event);
+      arm.lockForCacheModification(lr, event);
+      beginLocalWrite(event);
+      try {      
+        if (!hasSeenEvent(event)) {
+          this.entries.updateEntryVersion(event);
+        } else {
+          if (logger.isTraceEnabled(LogMarker.DM)) {
+            logger.trace(LogMarker.DM, "BR.basicUpdateEntryVersion: this cache has already seen this event {}", event);
+          }
         }
+        if (!event.isOriginRemote() && getBucketAdvisor().isPrimary()) {
+          // This cache has processed the event, forward operation
+          // and event messages to backup buckets
+          if (!getConcurrencyChecksEnabled() || event.hasValidVersionTag()) {
+            distributeUpdateEntryVersionOperation(event);
+          }
+        }
+        return;
+      } finally {
+        endLocalWrite(event);
       }
-      if (!event.isOriginRemote() && getBucketAdvisor().isPrimary()) {
-        // This cache has processed the event, forward operation
-        // and event messages to backup buckets
-    	if (!getConcurrencyChecksEnabled() || event.hasValidVersionTag()) {
-          distributeUpdateEntryVersionOperation(event);
-    	}
-      }
-      return;
     } finally {
-      endLocalWrite(event);
+      arm.releaseCacheModificationLock(event.getLocalRegion(), event);
     }
   }
 

http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/e08c1f54/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/DistributedClearOperation.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/DistributedClearOperation.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/DistributedClearOperation.java
index 5526ef0..b6e1c35 100644
--- a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/DistributedClearOperation.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/DistributedClearOperation.java
@@ -219,7 +219,7 @@ public class DistributedClearOperation extends DistributedCacheOperation
       case OP_LOCK_FOR_CLEAR:
         if (region.getDataPolicy().withStorage()) {
           DistributedClearOperation.regionLocked(this.getSender(), region.getFullPath(), region);
-          region.lockLocallyForClear(dm, this.getSender());
+          region.lockLocallyForClear(dm, this.getSender(), event);
         }
         this.appliedOperation = true;
         break;

http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/e08c1f54/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/DistributedRegion.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/DistributedRegion.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/DistributedRegion.java
index b42a617..a7b82ac 100755
--- a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/DistributedRegion.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/DistributedRegion.java
@@ -41,6 +41,7 @@ import com.gemstone.gemfire.distributed.internal.locks.DLockService;
 import com.gemstone.gemfire.distributed.internal.membership.InternalDistributedMember;
 import com.gemstone.gemfire.i18n.StringId;
 import com.gemstone.gemfire.internal.Assert;
+import com.gemstone.gemfire.internal.cache.AbstractRegionMap.ARMLockTestHook;
 import com.gemstone.gemfire.internal.cache.CacheDistributionAdvisor.CacheProfile;
 import com.gemstone.gemfire.internal.cache.InitialImageOperation.GIIStatus;
 import com.gemstone.gemfire.internal.cache.RemoteFetchVersionMessage.FetchVersionResponse;
@@ -1966,20 +1967,25 @@ public class DistributedRegion extends LocalRegion implements
     }
   }
 
-  
   @Override
   void basicUpdateEntryVersion(EntryEventImpl event)
       throws EntryNotFoundException {
-
+    LocalRegion lr = event.getLocalRegion();
+    AbstractRegionMap arm = ((AbstractRegionMap) lr.getRegionMap());
     try {
-      if (!hasSeenEvent(event)) {
-        super.basicUpdateEntryVersion(event);
+      arm.lockForCacheModification(lr, event);
+      try {
+        if (!hasSeenEvent(event)) {
+          super.basicUpdateEntryVersion(event);
+        }
+        return;
+      } finally {
+        if (!getConcurrencyChecksEnabled() || event.hasValidVersionTag()) {
+          distributeUpdateEntryVersion(event);
+        }
       }
-      return;
     } finally {
-      if (!getConcurrencyChecksEnabled() || event.hasValidVersionTag()) {
-        distributeUpdateEntryVersion(event);
-      }
+      arm.releaseCacheModificationLock(lr, event);
     }
   }
 
@@ -2110,33 +2116,47 @@ public class DistributedRegion extends LocalRegion implements
    * @param participants 
    **/
   private void obtainWriteLocksForClear(RegionEventImpl regionEvent, Set<InternalDistributedMember> participants) {
-    lockLocallyForClear(getDistributionManager(), getMyId());
+    lockLocallyForClear(getDistributionManager(), getMyId(), regionEvent);
     DistributedClearOperation.lockAndFlushToOthers(regionEvent, participants);
   }
   
   /** pause local operations so that a clear() can be performed and flush comm channels to the given member
   */
-  public void lockLocallyForClear(DM dm, InternalDistributedMember locker) {
+  public void lockLocallyForClear(DM dm, InternalDistributedMember locker, CacheEvent event) {
     RegionVersionVector rvv = getVersionVector();
+    
+    ARMLockTestHook alth = getRegionMap().getARMLockTestHook();
+    if(alth!=null) alth.beforeLock(this, event);
+    
     if (rvv != null) {
       // block new operations from being applied to the region map
       rvv.lockForClear(getFullPath(), dm, locker);
       //Check for region destroyed after we have locked, to make sure
       //we don't continue a clear if the region has been destroyed.
       checkReadiness();
-      // wait for current operations to 
-      if (!locker.equals(dm.getDistributionManagerId())) {
+      // Only need to flush if NOACK at this point
+      if (this.getAttributes().getScope().isDistributedNoAck()) {
         Set<InternalDistributedMember> mbrs = getDistributionAdvisor().adviseCacheOp();
         StateFlushOperation.flushTo(mbrs, this);
-      }
+      }      
     }
+    
+    if(alth!=null) alth.afterLock(this, null);
+
   }
 
   /** releases the locks obtained in obtainWriteLocksForClear 
    * @param participants */
   private void releaseWriteLocksForClear(RegionEventImpl regionEvent, Set<InternalDistributedMember> participants) {
+    
+    ARMLockTestHook alth = getRegionMap().getARMLockTestHook();
+    if(alth!=null) alth.beforeRelease(this, regionEvent);
+    
     getVersionVector().unlockForClear(getMyId());
     DistributedClearOperation.releaseLocks(regionEvent, participants);
+    
+    if(alth!=null) alth.afterRelease(this, regionEvent);
+
   }
   
   /**

http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/e08c1f54/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/LocalRegion.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/LocalRegion.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/LocalRegion.java
index 73dc5ab..304a5ad 100644
--- a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/LocalRegion.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/LocalRegion.java
@@ -157,6 +157,7 @@ import com.gemstone.gemfire.internal.InternalStatisticsDisabledException;
 import com.gemstone.gemfire.internal.NanoTimer;
 import com.gemstone.gemfire.internal.Version;
 import com.gemstone.gemfire.internal.admin.ClientHealthMonitoringRegion;
+import com.gemstone.gemfire.internal.cache.AbstractRegionMap.ARMLockTestHook;
 import com.gemstone.gemfire.internal.cache.CacheDistributionAdvisor.CacheProfile;
 import com.gemstone.gemfire.internal.cache.DiskInitFile.DiskRegionFlag;
 import com.gemstone.gemfire.internal.cache.FilterRoutingInfo.FilterInfo;
@@ -10198,141 +10199,144 @@ public class LocalRegion extends AbstractRegion
     }
     lockRVVForBulkOp();
     try {
+      try {
 
-      final DistributedPutAllOperation dpao = putAllOp;
-      int size = (proxyResult == null)? map.size() : proxyResult.size();
-      
-      if (isDebugEnabled) {
-        logger.debug( "size of put result is {} maps is {} proxyResult is {}", size, map, proxyResult);
-      }
+        final DistributedPutAllOperation dpao = putAllOp;
+        int size = (proxyResult == null)? map.size() : proxyResult.size();
 
-      final PutAllPartialResult partialKeys = new PutAllPartialResult(size);
-      final Iterator iterator;
-      final boolean isVersionedResults;
-      if (proxyResult != null) {
-        iterator = proxyResult.iterator();
-        isVersionedResults = true;
-      } else {
-        iterator = map.entrySet().iterator();
-        isVersionedResults = false;
-      }
-      Runnable r = new Runnable() {
-        public void run() {
-          int offset = 0;
-          VersionTagHolder tagHolder = new VersionTagHolder();
-          while (iterator.hasNext()) {
-            stopper.checkCancelInProgress(null);
-            Map.Entry mapEntry = (Map.Entry)iterator.next();
-            Object key = mapEntry.getKey();
-            VersionTag versionTag = null;
-            tagHolder.setVersionTag(null);
-            final Object value;
-            boolean overwritten = false;
-            if (isVersionedResults) {
-              versionTag = ((VersionedObjectList.Entry)mapEntry).getVersionTag();
-              value = map.get(key);
-              if (isDebugEnabled) {
-                logger.debug("putAll key {} -> {} version={}", key, value, versionTag);
-              }
-              if (versionTag == null && serverIsVersioned && concurrencyChecksEnabled && dataPolicy.withStorage()) {
-                // server was unable to determine the version for this operation.
-                // I'm not sure this can still happen as described below on a pr.
-                // But it can happen on the server if NORMAL or PRELOADED. See bug 51644.
-                // This can happen in a PR with redundancy if there is a bucket
-                // failure or migration during the operation.  We destroy the
-                // entry since we don't know what its state should be (but the server should)
+        if (isDebugEnabled) {
+          logger.debug( "size of put result is {} maps is {} proxyResult is {}", size, map, proxyResult);
+        }
+
+        final PutAllPartialResult partialKeys = new PutAllPartialResult(size);
+        final Iterator iterator;
+        final boolean isVersionedResults;
+        if (proxyResult != null) {
+          iterator = proxyResult.iterator();
+          isVersionedResults = true;
+        } else {
+          iterator = map.entrySet().iterator();
+          isVersionedResults = false;
+        }
+        Runnable r = new Runnable() {
+          public void run() {
+            int offset = 0;
+            VersionTagHolder tagHolder = new VersionTagHolder();
+            while (iterator.hasNext()) {
+              stopper.checkCancelInProgress(null);
+              Map.Entry mapEntry = (Map.Entry)iterator.next();
+              Object key = mapEntry.getKey();
+              VersionTag versionTag = null;
+              tagHolder.setVersionTag(null);
+              final Object value;
+              boolean overwritten = false;
+              if (isVersionedResults) {
+                versionTag = ((VersionedObjectList.Entry)mapEntry).getVersionTag();
+                value = map.get(key);
                 if (isDebugEnabled) {
-                  logger.debug("server returned no version information for {}", key);
+                  logger.debug("putAll key {} -> {} version={}", key, value, versionTag);
                 }
-                localDestroyNoCallbacks(key);
-                // to be consistent we need to fetch the current entry
-                get(key, event.getCallbackArgument(), false, null);
-                overwritten = true;
-              }
-            } else {
-              value = mapEntry.getValue();
-              if (isDebugEnabled) {
-                logger.debug("putAll {} -> {}", key, value);
-              }
-            }
-            try {
-              if (serverIsVersioned) {
+                if (versionTag == null && serverIsVersioned && concurrencyChecksEnabled && dataPolicy.withStorage()) {
+                  // server was unable to determine the version for this operation.
+                  // I'm not sure this can still happen as described below on a pr.
+                  // But it can happen on the server if NORMAL or PRELOADED. See bug 51644.
+                  // This can happen in a PR with redundancy if there is a bucket
+                  // failure or migration during the operation.  We destroy the
+                  // entry since we don't know what its state should be (but the server should)
+                  if (isDebugEnabled) {
+                    logger.debug("server returned no version information for {}", key);
+                  }
+                  localDestroyNoCallbacks(key);
+                  // to be consistent we need to fetch the current entry
+                  get(key, event.getCallbackArgument(), false, null);
+                  overwritten = true;
+                }
+              } else {
+                value = mapEntry.getValue();
                 if (isDebugEnabled) {
-                  logger.debug("associating version tag with {} version={}", key, versionTag);
+                  logger.debug("putAll {} -> {}", key, value);
                 }
-                //If we have received a version tag from a server, add it to the event
-                tagHolder.setVersionTag(versionTag);
-                tagHolder.setFromServer(true);
-              } else if (retryVersions != null && retryVersions.containsKey(key)) {
-                //If this is a retried event, and we have a version tag for the retry,
-                //add it to the event.
-                tagHolder.setVersionTag(retryVersions.get(key));
-              }
-              
-              if (!overwritten) {
-                basicEntryPutAll(key, value, dpao, offset, tagHolder);
               }
-              // now we must check again since the cache may have closed during
-              // distribution (causing this process to not receive and queue the
-              // event for clients
-              stopper.checkCancelInProgress(null);
-              succeeded.addKeyAndVersion(key, tagHolder.getVersionTag());
-            } 
-            catch (Exception ex) {
-              // TODO ask Gester if this debug logging can be removed
-              if (isDebugEnabled) {
-                logger.debug("PutAll operation encountered exception for key {}", key, ex);
+              try {
+                if (serverIsVersioned) {
+                  if (isDebugEnabled) {
+                    logger.debug("associating version tag with {} version={}", key, versionTag);
+                  }
+                  //If we have received a version tag from a server, add it to the event
+                  tagHolder.setVersionTag(versionTag);
+                  tagHolder.setFromServer(true);
+                } else if (retryVersions != null && retryVersions.containsKey(key)) {
+                  //If this is a retried event, and we have a version tag for the retry,
+                  //add it to the event.
+                  tagHolder.setVersionTag(retryVersions.get(key));
+                }
+
+                if (!overwritten) {
+                  basicEntryPutAll(key, value, dpao, offset, tagHolder);
+                }
+                // now we must check again since the cache may have closed during
+                // distribution (causing this process to not receive and queue the
+                // event for clients
+                stopper.checkCancelInProgress(null);
+                succeeded.addKeyAndVersion(key, tagHolder.getVersionTag());
+              } 
+              catch (Exception ex) {
+                // TODO ask Gester if this debug logging can be removed
+                if (isDebugEnabled) {
+                  logger.debug("PutAll operation encountered exception for key {}", key, ex);
+                }
+                partialKeys.saveFailedKey(key, ex);
               }
-              partialKeys.saveFailedKey(key, ex);
+              offset++;
             }
-            offset++;
           }
-        }
-      };
-      this.syncBulkOp(r, eventId);
-      if (partialKeys.hasFailure()) {
-        // Bug 51725: Now succeeded contains an order key list, may be missing the version tags. 
-        // Save reference of succeeded into partialKeys. The succeeded may be modified by
-        // postPutAll() to fill in the version tags. 
-        partialKeys.setSucceededKeysAndVersions(succeeded);
-        logger.info(LocalizedMessage.create(LocalizedStrings.Region_PutAll_Applied_PartialKeys_0_1,
-                new Object[] {getFullPath(), partialKeys}));
-        if (isDebugEnabled) {
-          logger.debug(partialKeys.detailString());
-        }
-        if (e == null) {
-          // if received exception from server first, ignore local exception
-          if (dpao.isBridgeOperation()) {
-            if (partialKeys.getFailure() instanceof CancelException) {
-              e = (CancelException)partialKeys.getFailure(); 
-            } else if (partialKeys.getFailure() instanceof LowMemoryException) {
-              throw partialKeys.getFailure();  // fix for #43589
-            } else {
-              e = new PutAllPartialResultException(partialKeys);
-              if (isDebugEnabled) {
-                logger.debug("basicPutAll:"+partialKeys.detailString());
+        };
+        this.syncBulkOp(r, eventId);
+        if (partialKeys.hasFailure()) {
+          // Bug 51725: Now succeeded contains an order key list, may be missing the version tags. 
+          // Save reference of succeeded into partialKeys. The succeeded may be modified by
+          // postPutAll() to fill in the version tags. 
+          partialKeys.setSucceededKeysAndVersions(succeeded);
+          logger.info(LocalizedMessage.create(LocalizedStrings.Region_PutAll_Applied_PartialKeys_0_1,
+              new Object[] {getFullPath(), partialKeys}));
+          if (isDebugEnabled) {
+            logger.debug(partialKeys.detailString());
+          }
+          if (e == null) {
+            // if received exception from server first, ignore local exception
+            if (dpao.isBridgeOperation()) {
+              if (partialKeys.getFailure() instanceof CancelException) {
+                e = (CancelException)partialKeys.getFailure(); 
+              } else if (partialKeys.getFailure() instanceof LowMemoryException) {
+                throw partialKeys.getFailure();  // fix for #43589
+              } else {
+                e = new PutAllPartialResultException(partialKeys);
+                if (isDebugEnabled) {
+                  logger.debug("basicPutAll:"+partialKeys.detailString());
+                }
               }
+            } else {
+              throw partialKeys.getFailure();
             }
-          } else {
-            throw partialKeys.getFailure();
           }
         }
       }
-    }
-    catch (LowMemoryException lme) {
-      throw lme;
-    }
-    catch (RuntimeException ex) {
-      e = ex;
-    }
-    catch (Exception ex) {
-      e = new RuntimeException(ex);
+      catch (LowMemoryException lme) {
+        throw lme;
+      }
+      catch (RuntimeException ex) {
+        e = ex;
+      }
+      catch (Exception ex) {
+        e = new RuntimeException(ex);
+      } finally {
+        putAllOp.getBaseEvent().release();
+        putAllOp.freeOffHeapResources();
+      }
+      getDataView().postPutAll(putAllOp, succeeded, this);
     } finally {
       unlockRVVForBulkOp();
-      putAllOp.getBaseEvent().release();
-      putAllOp.freeOffHeapResources();
     }
-    getDataView().postPutAll(putAllOp, succeeded, this);
     if (e != null) {
       throw e;
     }
@@ -10400,150 +10404,153 @@ public class LocalRegion extends AbstractRegion
     }
     lockRVVForBulkOp();
     try {
+      try {
 
-      final DistributedRemoveAllOperation op = removeAllOp;
-      int size = (proxyResult == null)? keys.size() : proxyResult.size();
-      
-      if (isInternalRegion()) {
-        if (isTraceEnabled) {
-          logger.trace(
-              "size of removeAll result is {} keys are {} proxyResult is {}", size, keys, proxyResult);
+        final DistributedRemoveAllOperation op = removeAllOp;
+        int size = (proxyResult == null)? keys.size() : proxyResult.size();
+
+        if (isInternalRegion()) {
+          if (isTraceEnabled) {
+            logger.trace(
+                "size of removeAll result is {} keys are {} proxyResult is {}", size, keys, proxyResult);
+          } else {
+            if (isTraceEnabled) {
+              logger.trace(
+                  "size of removeAll result is {} keys are {} proxyResult is {}", size, keys, proxyResult);
+            }
+          }
         } else {
           if (isTraceEnabled) {
             logger.trace(
                 "size of removeAll result is {} keys are {} proxyResult is {}", size, keys, proxyResult);
           }
         }
-      } else {
-        if (isTraceEnabled) {
-          logger.trace(
-              "size of removeAll result is {} keys are {} proxyResult is {}", size, keys, proxyResult);
-        }
-      }
 
-      final PutAllPartialResult partialKeys = new PutAllPartialResult(size);
-      final Iterator iterator;
-      final boolean isVersionedResults;
-      if (proxyResult != null) {
-        iterator = proxyResult.iterator();
-        isVersionedResults = true;
-      } else {
-        iterator = keys.iterator();
-        isVersionedResults = false;
-      }
-      Runnable r = new Runnable() {
-        public void run() {
-          int offset = 0;
-          VersionTagHolder tagHolder = new VersionTagHolder();
-          while (iterator.hasNext()) {
-            stopper.checkCancelInProgress(null);
-            Object key;
-            VersionTag versionTag = null;
-            tagHolder.setVersionTag(null);
-            if (isVersionedResults) {
-              Map.Entry mapEntry = (Map.Entry)iterator.next();
-              key = mapEntry.getKey();
-              versionTag = ((VersionedObjectList.Entry)mapEntry).getVersionTag();
-              if (isDebugEnabled) {
-                logger.debug("removeAll key {} version={}",key, versionTag);
-              }
-              if (versionTag == null) {
+        final PutAllPartialResult partialKeys = new PutAllPartialResult(size);
+        final Iterator iterator;
+        final boolean isVersionedResults;
+        if (proxyResult != null) {
+          iterator = proxyResult.iterator();
+          isVersionedResults = true;
+        } else {
+          iterator = keys.iterator();
+          isVersionedResults = false;
+        }
+        Runnable r = new Runnable() {
+          public void run() {
+            int offset = 0;
+            VersionTagHolder tagHolder = new VersionTagHolder();
+            while (iterator.hasNext()) {
+              stopper.checkCancelInProgress(null);
+              Object key;
+              VersionTag versionTag = null;
+              tagHolder.setVersionTag(null);
+              if (isVersionedResults) {
+                Map.Entry mapEntry = (Map.Entry)iterator.next();
+                key = mapEntry.getKey();
+                versionTag = ((VersionedObjectList.Entry)mapEntry).getVersionTag();
                 if (isDebugEnabled) {
-                  logger.debug("removeAll found invalid version tag, which means the entry is not found at server for key={}.", key);
+                  logger.debug("removeAll key {} version={}",key, versionTag);
                 }
-                succeeded.addKeyAndVersion(key, null);
-                continue;
-              }
-              // No need for special handling here in removeAll.
-              // We can just remove this key from the client with versionTag set to null.
-            } else {
-              key = iterator.next();
-              if (isInternalRegion()) {
-                if (isTraceEnabled) {
-                  logger.trace("removeAll {}", key);
+                if (versionTag == null) {
+                  if (isDebugEnabled) {
+                    logger.debug("removeAll found invalid version tag, which means the entry is not found at server for key={}.", key);
+                  }
+                  succeeded.addKeyAndVersion(key, null);
+                  continue;
                 }
+                // No need for special handling here in removeAll.
+                // We can just remove this key from the client with versionTag set to null.
               } else {
-                if (isTraceEnabled) {
-                  logger.trace("removeAll {}", key);
+                key = iterator.next();
+                if (isInternalRegion()) {
+                  if (isTraceEnabled) {
+                    logger.trace("removeAll {}", key);
+                  }
+                } else {
+                  if (isTraceEnabled) {
+                    logger.trace("removeAll {}", key);
+                  }
                 }
+
               }
+              try {
+                if (serverIsVersioned) {
+                  if (isDebugEnabled) {
+                    logger.debug("associating version tag with {} version={}", key, versionTag);
+                  }
+                  //If we have received a version tag from a server, add it to the event
+                  tagHolder.setVersionTag(versionTag);
+                  tagHolder.setFromServer(true);
+                } else if (retryVersions != null) {
+                  VersionTag vt = retryVersions.get(offset);
+                  if (vt != null) {
+                    //If this is a retried event, and we have a version tag for the retry,
+                    //add it to the event.
+                    tagHolder.setVersionTag(vt);
+                  }
+                }
 
+                basicEntryRemoveAll(key, op, offset, tagHolder);
+                // now we must check again since the cache may have closed during
+                // distribution causing this process to not receive and queue the
+                // event for clients
+                stopper.checkCancelInProgress(null);
+                succeeded.addKeyAndVersion(key, tagHolder.getVersionTag());
+              } 
+              catch (Exception ex) {
+                partialKeys.saveFailedKey(key, ex);
+              }
+              offset++;
             }
-            try {
-              if (serverIsVersioned) {
+          }
+        };
+        syncBulkOp(r, eventId);
+        if (partialKeys.hasFailure()) {
+          // Bug 51725: Now succeeded contains an order key list, may be missing the version tags. 
+          // Save reference of succeeded into partialKeys. The succeeded may be modified by
+          // postRemoveAll() to fill in the version tags.
+          partialKeys.setSucceededKeysAndVersions(succeeded);
+          logger.info(LocalizedMessage.create(LocalizedStrings.Region_RemoveAll_Applied_PartialKeys_0_1,
+              new Object[] {getFullPath(), partialKeys}));
+          if (isDebugEnabled) {
+            logger.debug(partialKeys.detailString());
+          }
+          if (e == null) {
+            // if received exception from server first, ignore local exception
+            if (op.isBridgeOperation()) {
+              if (partialKeys.getFailure() instanceof CancelException) {
+                e = (CancelException)partialKeys.getFailure(); 
+              } else if (partialKeys.getFailure() instanceof LowMemoryException) {
+                throw partialKeys.getFailure();  // fix for #43589
+              } else {
+                e = new PutAllPartialResultException(partialKeys);
                 if (isDebugEnabled) {
-                  logger.debug("associating version tag with {} version={}", key, versionTag);
-                }
-                //If we have received a version tag from a server, add it to the event
-                tagHolder.setVersionTag(versionTag);
-                tagHolder.setFromServer(true);
-              } else if (retryVersions != null) {
-                VersionTag vt = retryVersions.get(offset);
-                if (vt != null) {
-                  //If this is a retried event, and we have a version tag for the retry,
-                  //add it to the event.
-                  tagHolder.setVersionTag(vt);
+                  logger.debug("basicRemoveAll:"+partialKeys.detailString());
                 }
               }
-              
-              basicEntryRemoveAll(key, op, offset, tagHolder);
-              // now we must check again since the cache may have closed during
-              // distribution causing this process to not receive and queue the
-              // event for clients
-              stopper.checkCancelInProgress(null);
-              succeeded.addKeyAndVersion(key, tagHolder.getVersionTag());
-            } 
-            catch (Exception ex) {
-              partialKeys.saveFailedKey(key, ex);
-            }
-            offset++;
-          }
-        }
-      };
-      syncBulkOp(r, eventId);
-      if (partialKeys.hasFailure()) {
-        // Bug 51725: Now succeeded contains an order key list, may be missing the version tags. 
-        // Save reference of succeeded into partialKeys. The succeeded may be modified by
-        // postRemoveAll() to fill in the version tags.
-        partialKeys.setSucceededKeysAndVersions(succeeded);
-        logger.info(LocalizedMessage.create(LocalizedStrings.Region_RemoveAll_Applied_PartialKeys_0_1,
-                new Object[] {getFullPath(), partialKeys}));
-        if (isDebugEnabled) {
-          logger.debug(partialKeys.detailString());
-        }
-        if (e == null) {
-          // if received exception from server first, ignore local exception
-          if (op.isBridgeOperation()) {
-            if (partialKeys.getFailure() instanceof CancelException) {
-              e = (CancelException)partialKeys.getFailure(); 
-            } else if (partialKeys.getFailure() instanceof LowMemoryException) {
-              throw partialKeys.getFailure();  // fix for #43589
             } else {
-              e = new PutAllPartialResultException(partialKeys);
-              if (isDebugEnabled) {
-                logger.debug("basicRemoveAll:"+partialKeys.detailString());
-              }
+              throw partialKeys.getFailure();
             }
-          } else {
-            throw partialKeys.getFailure();
           }
         }
       }
-    }
-    catch (LowMemoryException lme) {
-      throw lme;
-    }
-    catch (RuntimeException ex) {
-      e = ex;
-    }
-    catch (Exception ex) {
-      e = new RuntimeException(ex);
+      catch (LowMemoryException lme) {
+        throw lme;
+      }
+      catch (RuntimeException ex) {
+        e = ex;
+      }
+      catch (Exception ex) {
+        e = new RuntimeException(ex);
+      } finally {
+        removeAllOp.getBaseEvent().release();
+        removeAllOp.freeOffHeapResources();
+      }
+      getDataView().postRemoveAll(removeAllOp, succeeded, this);
     } finally {
       unlockRVVForBulkOp();
-      removeAllOp.getBaseEvent().release();
-      removeAllOp.freeOffHeapResources();
     }
-    getDataView().postRemoveAll(removeAllOp, succeeded, this);
     if (e != null) {
       throw e;
     }
@@ -10558,15 +10565,33 @@ public class LocalRegion extends AbstractRegion
    *  to get a valid version tag.
    */
   private void lockRVVForBulkOp() {
+    ARMLockTestHook alth = getRegionMap().getARMLockTestHook();
+    if(alth!=null) { 
+      alth.beforeBulkLock(this); 
+    }
+    
     if (this.versionVector != null && this.dataPolicy.withReplication()) {
       this.versionVector.lockForCacheModification(this);
     }
+    
+    if(alth!=null) {
+      alth.afterBulkLock(this);
+    }
   }
   
   private void unlockRVVForBulkOp() {
+    ARMLockTestHook alth = getRegionMap().getARMLockTestHook();
+    if(alth!=null) {
+      alth.beforeBulkRelease(this);
+    }
+    
     if (this.versionVector != null && this.dataPolicy.withReplication()) {
       this.versionVector.releaseCacheModificationLock(this);
     }
+    
+    if(alth!=null) {
+      alth.afterBulkRelease(this);
+    }
   }
 
 

http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/e08c1f54/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/ProxyRegionMap.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/ProxyRegionMap.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/ProxyRegionMap.java
index 3ad2cc1..99d96ab 100644
--- a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/ProxyRegionMap.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/ProxyRegionMap.java
@@ -34,6 +34,7 @@ import com.gemstone.gemfire.distributed.internal.DM;
 import com.gemstone.gemfire.distributed.internal.membership.InternalDistributedMember;
 import com.gemstone.gemfire.internal.ByteArrayDataInput;
 import com.gemstone.gemfire.internal.InternalStatisticsDisabledException;
+import com.gemstone.gemfire.internal.cache.AbstractRegionMap.ARMLockTestHook;
 import com.gemstone.gemfire.internal.cache.lru.LRUEntry;
 import com.gemstone.gemfire.internal.cache.lru.NewLRUClockHand;
 import com.gemstone.gemfire.internal.cache.tier.sockets.ClientProxyMembershipID;
@@ -745,4 +746,9 @@ final class ProxyRegionMap implements RegionMap {
   @Override
   public void close() {
   }
+
+  @Override
+  public ARMLockTestHook getARMLockTestHook() {
+    return null;
+  }
 }



Mime
View raw message